跳至主要內容

StreamData

更新於
Jul 12, 2026

概覽​

Establishes a bidirectional streaming connection to receive real-time data updates from Hyperliquid. StreamData is the primary method for subscribing to trades, orders, events, and other blockchain data streams.

Key capabilities:

  • Subscribe to multiple stream types (trades, orders, events, blocks, and more)
  • Filter data by coin, user, event type, and other fields to reduce bandwidth
  • Use named filters for complex OR logic across multiple filter sets
  • Start from a specific block number for historical data replay
  • Maintain connection with periodic ping/pong messages
Metered Billing

This method is metered based on data consumption at 0.1 MB = 10 API credits. Concurrent stream caps and filter limits vary by plan - see the Streaming Limits table.

請求​

The client sends SubscribeRequest messages with one of two variants:

  • StreamSubscribe - Subscribe to a stream with optional filters
  • Ping - Keep-alive message with 時間戳記 (int64) to maintain connection

Subscribe Parameters​

Parameter類型Required說明
stream_typeenum是的Stream type to subscribe to (see below)
filtersmap不Field filters - key = field name, value = list of allowed values
start_blockuint64不Block number to start from (omit for latest)
filter_name字串不Name for this filter set (enables OR logic across subscriptions)

Stream Types​

TRADES

Executed trades with price, size, side, and maker/taker info

ORDERS

Order lifecycle events (open, filled, canceled, triggered)

BOOK_UPDATES

Order book changes with bid/ask price level updates

TWAP

TWAP execution status and progress updates

EVENTS

Ledger updates, deposits, withdrawals, funding, liquidations

BLOCKS

Raw HyperCore blockchain data (replica_cmds)

WRITER_ACTIONS

HyperCore to HyperEVM asset transfers

MEMPOOL_TXS

Pending mempool transactions before block confirmation

ORDER_PRIORITY

Normalized priority-fee order events from mempool and blocks

GOSSIP_PRIORITY

Normalized gossip/read-priority bid events from mempool and blocks

回應​

每個 SubscribeUpdate response contains one of:

  • 資料 - Stream data with block_number (uint64), 時間戳記 (uint64, server ingress in ms), and 資料 (JSON string payload)
  • pong - Pong response with 時間戳記 (int64) echoing your ping

該 data.data payload structure varies by stream type - see Data Streams for detailed schemas.

Filtering​

StreamData supports powerful filtering to reduce bandwidth and focus on specific data. Filters use AND logic across fields and OR logic within field values.

Filter by coin:

filters: {
  "coin": FilterValues { values: ["BTC", "ETH"] }
}

Filter by user address:

filters: {
  "user": FilterValues { values: ["0x1234...", "0x5678..."] }
}

Combine multiple filters (AND logic):

filters: {
  "coin": FilterValues { values: ["BTC"] },
  "side": FilterValues { values: ["B"] }
}

For complete filtering documentation including field references for each stream type, see the Stream Filtering Guide.

程式碼範例​

const grpc = require('@grpc/grpc-js');
const protoLoader = require('@grpc/proto-loader');
const path = require('path');

// Configuration
const GRPC_ENDPOINT = 'your-endpoint.hype-mainnet.quiknode.pro:10000';
const AUTH_TOKEN = 'your-auth-token';

// Load proto file
const packageDefinition = protoLoader.loadSync(
  path.join(__dirname, 'proto/streaming.proto'),
  {
    keepCase: true,
    longs: String,
    enums: String,
    defaults: true,
    oneofs: true,
  }
);

const hyperliquidProto = grpc.loadPackageDefinition(packageDefinition).hyperliquid;

// Create gRPC client with TLS
function createClient() {
  const credentials = grpc.credentials.createSsl();

  const client = new hyperliquidProto.Streaming(
    GRPC_ENDPOINT,
    credentials,
    {
      'grpc.max_receive_message_length': 100 * 1024 * 1024, // 100MB
    }
  );

  return client;
}

// Create metadata with auth token
function createMetadata() {
  const metadata = new grpc.Metadata();
  metadata.add('x-token', AUTH_TOKEN);
  return metadata;
}

// Test connectivity with ping
function testPing(client) {
  return new Promise((resolve, reject) => {
    const metadata = createMetadata();

    client.Ping({ count: 1 }, metadata, (error, response) => {
      if (error) {
        reject(new Error(`Ping failed: ${error.message}`));
        return;
      }

      console.log('✅ Ping successful:', response);
      resolve(response);
    });
  });
}

// Stream trades data
function streamTrades(client) {
  return new Promise((resolve, reject) => {
    const metadata = createMetadata();

    const stream = client.StreamData(metadata);

    // Handle incoming messages
    stream.on('data', (response) => {
      if (response.data) {
        try {
          const data = JSON.parse(response.data.data);
          const trades = data.trades || [];

          console.log(`📈 Block ${response.data.block_number}: ${trades.length} trades`);

          // Pretty print first trade if available
          if (trades.length > 0) {
            console.log('   Example trade:', JSON.stringify(trades[0], null, 2));
          }
        } catch (error) {
          console.log('⚠️  Failed to parse data:', error.message);
        }
      } else if (response.pong) {
        console.log('🏓 Pong received:', response.pong.timestamp);
      }
    });

    stream.on('error', (error) => {
      console.error('❌ Stream error:', error.message);
      reject(error);
    });

    stream.on('end', () => {
      console.log('Stream ended');
      resolve();
    });

    // Send subscription request
    // Empty filters means receive all events for this stream type
    stream.write({
      subscribe: {
        stream_type: 'TRADES',
        filters: {}
      }
    });

    // Start ping interval to keep connection alive
    const pingInterval = setInterval(() => {
      stream.write({
        ping: {
          timestamp: Date.now()
        }
      });
    }, 30000);

    // Clean up on exit
    const cleanup = () => {
      clearInterval(pingInterval);
      stream.end();
    };

    process.on('SIGINT', cleanup);
    process.on('SIGTERM', cleanup);

    console.log('🔄 Starting trades stream...');
  });
}

// Stream orders data
function streamOrders(client) {
  return new Promise((resolve, reject) => {
    const metadata = createMetadata();

    const stream = client.StreamData(metadata);

    stream.on('data', (response) => {
      if (response.data) {
        try {
          const data = JSON.parse(response.data.data);
          console.log(`📋 Orders Block ${response.data.block_number}:`);
          console.log(JSON.stringify(data, null, 2));
        } catch (error) {
          console.log('⚠️  Failed to parse order data:', error.message);
        }
      }
    });

    stream.on('error', (error) => {
      console.error('❌ Stream error:', error.message);
      reject(error);
    });

    stream.on('end', () => {
      console.log('Orders stream ended');
      resolve();
    });

    // Send subscription request for orders
    stream.write({
      subscribe: {
        stream_type: 'ORDERS',
        filters: {}
      }
    });

    console.log('🔄 Starting orders stream...');
  });
}

// Interactive menu
function showMenu() {
  console.log('\n🎉 Connected successfully!');
  console.log('\nAvailable streams:');
  console.log('1. All trades');
  console.log('2. Filtered trades (BTC, ETH)');
  console.log('3. Orders');
  console.log('4. Test ping');
  console.log('5. Exit');

  const readline = require('readline');
  const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout
  });

  return new Promise((resolve) => {
    rl.question('\nSelect stream (1-5): ', (choice) => {
      rl.close();
      resolve(choice);
    });
  });
}

// Stream filtered trades
function streamFilteredTrades(client, coins) {
  return new Promise((resolve, reject) => {
    const metadata = createMetadata();

    const stream = client.StreamData(metadata);

    stream.on('data', (response) => {
      if (response.data) {
        try {
          const data = JSON.parse(response.data.data);
          const trades = data.trades || [];

          if (trades.length > 0) {
            console.log(`📈 Filtered Block ${response.data.block_number}: ${trades.length} trades`);
            console.log(JSON.stringify(data, null, 2));
          }
        } catch (error) {
          console.log('⚠️  Failed to parse data:', error.message);
        }
      }
    });

    stream.on('error', (error) => {
      console.error('❌ Stream error:', error.message);
      reject(error);
    });

    stream.on('end', () => {
      console.log('Filtered stream ended');
      resolve();
    });

    // Send filtered subscription
    stream.write({
      subscribe: {
        stream_type: 'TRADES',
        filters: {
          coin: { values: coins }
        }
      }
    });

    console.log(`🔄 Starting filtered trades stream for coins: ${coins.join(', ')}`);
  });
}

// Main function
async function main() {
  try {
    console.log(`Testing connection to: ${GRPC_ENDPOINT}`);

    // Create client
    const client = createClient();

    // Test initial connectivity
    await testPing(client);

    // Show interactive menu
    while (true) {
      const choice = await showMenu();

      switch (choice) {
        case '1':
          await streamTrades(client);
          break;
        case '2':
          await streamFilteredTrades(client, ['BTC', 'ETH']);
          break;
        case '3':
          await streamOrders(client);
          break;
        case '4':
          await testPing(client);
          break;
        case '5':
          console.log('👋 Goodbye!');
          client.close();
          return;
        default:
          console.log('❌ Invalid choice. Please select 1-5.');
      }
    }

  } catch (error) {
    console.error('❌ Error:', error.message);
    process.exit(1);
  }
}

// Run the application
main();
最後更新時間 於
分享這份文件