StreamOrderbookSnapshots - Real-time Order Book Streaming

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "strings"
    "time"

    pb "hyperliquid-grpc-client/api/v2"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/metadata"
)

type Snapshot struct {
    Block     string          `json:"block"`
    Timestamp int64           `json:"timestamp"`
    Data      json.RawMessage `json:"data"`
}

func streamOrderbookSnapshots() {
    endpoint := os.Getenv("HYPERLIQUID_ENDPOINT")
    apiKey := os.Getenv("API_KEY")

    if endpoint == "" {
        log.Fatal("Error: HYPERLIQUID_ENDPOINT environment variable is required.")
    }

    if apiKey == "" {
        log.Fatal("Error: API_KEY environment variable is required.")
    }

    fmt.Println("Hyperliquid Go gRPC Client - Stream Order Book Snapshots")
    fmt.Println("========================================================")
    fmt.Printf("Endpoint: %s\n\n", endpoint)

    // Set up TLS connection
    creds := credentials.NewTLS(nil)

    opts := []grpc.DialOption{
        grpc.WithTransportCredentials(creds),
        grpc.WithDefaultCallOptions(
            grpc.MaxCallRecvMsgSize(500 * 1024 * 1024), // 500MB max
        ),
    }

    fmt.Println("Connecting to gRPC server...")
    conn, err := grpc.NewClient(endpoint, opts...)
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()

    client := pb.NewHyperliquidL1GatewayClient(conn)
    fmt.Println("Connected successfully!\n")

    // Create context with API key
    ctx := metadata.AppendToOutgoingContext(context.Background(), "x-api-key", apiKey)

    // Create request - empty Position means latest/current snapshots
    request := &pb.Position{}

    fmt.Println("Starting order book snapshots stream...")
    fmt.Println("Press Ctrl+C to stop streaming\n")

    stream, err := client.StreamOrderbookSnapshots(ctx, request)
    if err != nil {
        log.Fatalf("Failed to start stream: %v", err)
    }

    snapshotCount := 0
    for {
        response, err := stream.Recv()
        if err != nil {
            fmt.Printf("Stream ended: %v\n", err)
            break
        }

        snapshotCount++
        fmt.Printf("\n===== SNAPSHOT #%d =====\n", snapshotCount)
        fmt.Printf("Response size: %d bytes\n", len(response.Data))

        processSnapshot(response.Data, snapshotCount)

        fmt.Println("\n" + strings.Repeat("-", 50))
    }

    fmt.Printf("\nTotal snapshots received: %d\n", snapshotCount)
}

func processSnapshot(data []byte, snapshotNum int) {
    var snapshot Snapshot
    if err := json.Unmarshal(data, &snapshot); err != nil {
        fmt.Printf("Failed to parse JSON: %v\n", err)
        return
    }

    fmt.Printf("Block: %s\n", snapshot.Block)

    timestamp := time.UnixMilli(snapshot.Timestamp)
    fmt.Printf("Time: %s\n", timestamp.Format("2006-01-02 15:04:05 UTC"))

    // Parse the market data
    var markets []json.RawMessage
    if err := json.Unmarshal(snapshot.Data, &markets); err != nil {
        fmt.Printf("Failed to parse markets: %v\n", err)
        return
    }

    fmt.Printf("Total markets: %d\n", len(markets))
}

func main() {
    streamOrderbookSnapshots()
}

import grpc
import json
import signal
import sys
import os
from datetime import datetime
from dotenv import load_dotenv
import hyperliquid_pb2
import hyperliquid_pb2_grpc

load_dotenv()

def stream_orderbook_snapshots():
    endpoint = os.getenv('HYPERLIQUID_ENDPOINT')
    api_key = os.getenv('API_KEY')

    if not endpoint:
        print("Error: HYPERLIQUID_ENDPOINT environment variable is required.")
        sys.exit(1)

    if not api_key:
        print("Error: API_KEY environment variable is required.")
        sys.exit(1)

    print('Hyperliquid Python gRPC Client - Stream Order Book Snapshots')
    print('============================================================')
    print(f'Endpoint: {endpoint}\n')

    credentials = grpc.ssl_channel_credentials()

    options = [
        ('grpc.max_receive_message_length', 500 * 1024 * 1024),  # 500MB max
    ]

    metadata = [('x-api-key', api_key)]

    print('Connecting to gRPC server...')
    with grpc.secure_channel(endpoint, credentials, options=options) as channel:
        client = hyperliquid_pb2_grpc.HyperliquidL1GatewayStub(channel)
        print('Connected successfully!\n')

        # Create request - empty Position means latest/current snapshots
        request = hyperliquid_pb2.Position()

        print('Starting order book snapshots stream...')
        print('Press Ctrl+C to stop streaming\n')

        snapshot_count = 0

        def signal_handler(sig, frame):
            print('\nStopping stream...')
            print(f'Total snapshots received: {snapshot_count}')
            sys.exit(0)

        signal.signal(signal.SIGINT, signal_handler)

        try:
            for response in client.StreamOrderbookSnapshots(request, metadata=metadata):
                snapshot_count += 1
                print(f'\n===== SNAPSHOT #{snapshot_count} =====')
                print(f'Response size: {len(response.data)} bytes')

                process_snapshot(response.data, snapshot_count)

                print('\n' + '-' * 50)

        except grpc.RpcError as e:
            print(f'Stream error: {e}')
        except KeyboardInterrupt:
            print('\nStopping stream...')

        print(f'\nTotal snapshots received: {snapshot_count}')

def process_snapshot(data, snapshot_num):
    try:
        snapshot = json.loads(data.decode('utf-8'))

        print(f'Block: {snapshot["block"]}')

        dt = datetime.fromtimestamp(snapshot['timestamp'] / 1000)
        print(f'Time: {dt.strftime("%Y-%m-%d %H:%M:%S UTC")}')

        markets = snapshot['data']
        print(f'Total markets: {len(markets)}')

        # Count orders across all markets
        total_bids = 0
        total_asks = 0
        for market in markets:
            orders = market[1]
            total_bids += len(orders[0])
            total_asks += len(orders[1])

        print(f'Total bid orders: {total_bids:,}')
        print(f'Total ask orders: {total_asks:,}')
        print(f'Total orders: {total_bids + total_asks:,}')

    except json.JSONDecodeError as e:
        print(f'Failed to parse JSON: {e}')
    except Exception as e:
        print(f'Error processing snapshot: {e}')

if __name__ == '__main__':
    stream_orderbook_snapshots()

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

const PROTO_PATH = path.join(__dirname, 'v2.proto');

const packageDefinition = protoLoader.loadSync(PROTO_PATH, {
    keepCase: true,
    longs: String,
    enums: String,
    defaults: true,
    oneofs: true
});

const proto = grpc.loadPackageDefinition(packageDefinition);

async function streamOrderbookSnapshots() {
    const endpoint = process.env.HYPERLIQUID_ENDPOINT;
    const apiKey = process.env.API_KEY;

    if (!endpoint) {
        console.error('Error: HYPERLIQUID_ENDPOINT environment variable is required.');
        process.exit(1);
    }

    if (!apiKey) {
        console.error('Error: API_KEY environment variable is required.');
        process.exit(1);
    }

    console.log('Hyperliquid Node.js gRPC Client - Stream Order Book Snapshots');
    console.log('=============================================================');
    console.log(`Endpoint: ${endpoint}\n`);

    const metadata = new grpc.Metadata();
    metadata.add('x-api-key', apiKey);

    // 500MB max to handle the large snapshots
    const client = new proto.hyperliquid_l1_gateway.v2.HyperliquidL1Gateway(
        endpoint,
        grpc.credentials.createSsl(),
        {
            'grpc.max_receive_message_length': 500 * 1024 * 1024
        }
    );

    console.log('Starting order book snapshots stream...');
    console.log('Press Ctrl+C to stop streaming\n');

    // Make the gRPC call - empty Position for latest
    const stream = client.StreamOrderbookSnapshots({}, metadata);

    let snapshotCount = 0;

    stream.on('data', (data) => {
        snapshotCount++;

        try {
            const snapshot = JSON.parse(data.data);
            console.log(`\n===== SNAPSHOT #${snapshotCount} =====`);
            console.log(`Response size: ${data.data.length} bytes`);

            console.log(`Block: ${snapshot.block}`);
            const date = new Date(snapshot.timestamp);
            console.log(`Time: ${date.toISOString()}`);
            console.log(`Total markets: ${snapshot.data.length}`);

            // Count orders across all markets
            let totalBids = 0;
            let totalAsks = 0;
            for (const market of snapshot.data) {
                totalBids += market[1][0].length;
                totalAsks += market[1][1].length;
            }

            console.log(`Total orders: ${(totalBids + totalAsks).toLocaleString()}`);
            console.log('\n' + '-'.repeat(50));

        } catch (error) {
            console.error(`Failed to parse message #${snapshotCount}:`, error.message);
        }
    });

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

    stream.on('end', () => {
        console.log('Stream ended');
        console.log(`\nTotal snapshots received: ${snapshotCount}`);
    });
}

streamOrderbookSnapshots();