StreamData gRPC Method

Please note that 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.

Parameters

request

object

Loading...

Loading...

Loading...

Returns

stream

stream<SubscribeUpdate>

Loading...

data

StreamResponse

Loading...

data.data - TRADES

array

Loading...

data.data - ORDERS

array

Loading...

data.data - BOOK_UPDATES

array

Loading...

data.data - TWAP

array

Loading...

data.data - EVENTS

array

Loading...

data.data - WRITER_ACTIONS

array

Loading...

data.data - BLOCKS

object

Loading...

data.data - MEMPOOL_TXS

array

Loading...

pong

Pong

Loading...

Request

1
package main
2

3
import (
4
	"context"
5
	"crypto/tls"
6
	"encoding/json"
7
	"fmt"
8
	"io"
9
	"log"
10
	"time"
11

12
	"google.golang.org/grpc"
13
	"google.golang.org/grpc/credentials"
14
	"google.golang.org/grpc/metadata"
15

16
	pb "hyperliquid-grpc-go/pb"
17
)
18

19
const (
20
	// Using Quicknode endpoint
21
	grpcEndpoint = "your-endpoint.hype-mainnet.quiknode.pro:10000"
22
	authToken    = "your-auth-token"
23
)
24

25
// Create gRPC connection with TLS
26
func createConnection() (*grpc.ClientConn, error) {
27
	tlsConfig := &tls.Config{
28
		InsecureSkipVerify: false, // Set to true only for testing
29
	}
30
	creds := credentials.NewTLS(tlsConfig)
31

32
	conn, err := grpc.Dial(
33
		grpcEndpoint,
34
		grpc.WithTransportCredentials(creds),
35
		grpc.WithDefaultCallOptions(
36
			grpc.MaxCallRecvMsgSize(100*1024*1024), // 100MB
37
		),
38
	)
39

40
	return conn, err
41
}
42

43
// Create context with auth metadata
44
func createContext() context.Context {
45
	md := metadata.Pairs("x-token", authToken)
46
	return metadata.NewOutgoingContext(context.Background(), md)
47
}
48

49

50
// Stream trades
51
func streamTrades(client pb.StreamingClient) error {
52
	ctx := createContext()
53

54
	stream, err := client.StreamData(ctx)
55
	if err != nil {
56
		return fmt.Errorf("failed to create stream: %v", err)
57
	}
58

59
	// Send subscription request
60
	// Empty Filters means receive all events for this stream type
61
	err = stream.Send(&pb.SubscribeRequest{
62
		Request: &pb.SubscribeRequest_Subscribe{
63
			Subscribe: &pb.StreamSubscribe{
64
				StreamType: pb.StreamType_TRADES,
65
				StartBlock: 0,
66
				Filters:    map[string]*pb.FilterValues{},
67
			},
68
		},
69
	})
70
	if err != nil {
71
		return fmt.Errorf("failed to subscribe: %v", err)
72
	}
73

74
	// Start ping goroutine
75
	go func() {
76
		ticker := time.NewTicker(30 * time.Second)
77
		defer ticker.Stop()
78

79
		for range ticker.C {
80
			err := stream.Send(&pb.SubscribeRequest{
81
				Request: &pb.SubscribeRequest_Ping{
82
					Ping: &pb.Ping{
83
						Timestamp: time.Now().UnixMilli(),
84
					},
85
				},
86
			})
87
			if err != nil {
88
				log.Printf("Failed to send ping: %v", err)
89
				return
90
			}
91
		}
92
	}()
93

94
	// Receive messages
95
	log.Println("🔄 Starting trades stream...")
96
	for {
97
		msg, err := stream.Recv()
98
		if err == io.EOF {
99
			log.Println("Stream ended")
100
			break
101
		}
102
		if err != nil {
103
			return fmt.Errorf("receive error: %v", err)
104
		}
105

106
		switch update := msg.Update.(type) {
107
		case *pb.SubscribeUpdate_Data:
108
			var data map[string]interface{}
109
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
110
				log.Printf("⚠️  Failed to parse data: %v", err)
111
				continue
112
			}
113

114
			trades, ok := data["trades"].([]interface{})
115
			tradesCount := 0
116
			if ok {
117
				tradesCount = len(trades)
118
			}
119

120
			log.Printf("📈 Block %d: %d trades", update.Data.BlockNumber, tradesCount)
121
			
122
			// Pretty print first trade if available
123
			if tradesCount > 0 {
124
				firstTrade := trades[0]
125
				tradeJSON, _ := json.MarshalIndent(firstTrade, "   ", "  ")
126
				fmt.Printf("   Example trade:\n%s\n", tradeJSON)
127
			}
128

129
		case *pb.SubscribeUpdate_Pong:
130
			log.Printf("🏓 Pong received: %d", update.Pong.Timestamp)
131
		}
132
	}
133

134
	return nil
135
}
136

137
// Stream filtered trades
138
func streamFilteredTrades(client pb.StreamingClient, coins []string) error {
139
	ctx := createContext()
140

141
	stream, err := client.StreamData(ctx)
142
	if err != nil {
143
		return fmt.Errorf("failed to create stream: %v", err)
144
	}
145

146
	// Send filtered subscription
147
	err = stream.Send(&pb.SubscribeRequest{
148
		Request: &pb.SubscribeRequest_Subscribe{
149
			Subscribe: &pb.StreamSubscribe{
150
				StreamType: pb.StreamType_TRADES,
151
				StartBlock: 0,
152
				Filters: map[string]*pb.FilterValues{
153
					"coin": {Values: coins},
154
				},
155
			},
156
		},
157
	})
158
	if err != nil {
159
		return fmt.Errorf("failed to subscribe: %v", err)
160
	}
161

162
	log.Printf("🔄 Starting filtered trades stream for coins: %v", coins)
163

164
	// Receive messages
165
	for {
166
		msg, err := stream.Recv()
167
		if err == io.EOF {
168
			log.Println("Filtered stream ended")
169
			break
170
		}
171
		if err != nil {
172
			return fmt.Errorf("receive error: %v", err)
173
		}
174

175
		switch update := msg.Update.(type) {
176
		case *pb.SubscribeUpdate_Data:
177
			var data map[string]interface{}
178
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
179
				log.Printf("⚠️  Failed to parse data: %v", err)
180
				continue
181
			}
182

183
			if trades, ok := data["trades"].([]interface{}); ok && len(trades) > 0 {
184
				log.Printf("📈 Filtered Block %d: %d trades", update.Data.BlockNumber, len(trades))
185
				prettyJSON, _ := json.MarshalIndent(data, "", "  ")
186
				fmt.Println(string(prettyJSON))
187
			}
188
		}
189
	}
190

191
	return nil
192
}
193

194
// Stream orders
195
func streamOrders(client pb.StreamingClient) error {
196
	ctx := createContext()
197

198
	stream, err := client.StreamData(ctx)
199
	if err != nil {
200
		return fmt.Errorf("failed to create stream: %v", err)
201
	}
202

203
	// Send subscription request for orders
204
	err = stream.Send(&pb.SubscribeRequest{
205
		Request: &pb.SubscribeRequest_Subscribe{
206
			Subscribe: &pb.StreamSubscribe{
207
				StreamType: pb.StreamType_ORDERS,
208
				StartBlock: 0,
209
				Filters:    map[string]*pb.FilterValues{},
210
			},
211
		},
212
	})
213
	if err != nil {
214
		return fmt.Errorf("failed to subscribe: %v", err)
215
	}
216

217
	log.Println("🔄 Starting orders stream...")
218

219
	// Receive messages
220
	for {
221
		msg, err := stream.Recv()
222
		if err == io.EOF {
223
			log.Println("Orders stream ended")
224
			break
225
		}
226
		if err != nil {
227
			return fmt.Errorf("receive error: %v", err)
228
		}
229

230
		switch update := msg.Update.(type) {
231
		case *pb.SubscribeUpdate_Data:
232
			var data map[string]interface{}
233
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
234
				log.Printf("⚠️  Failed to parse order data: %v", err)
235
				continue
236
			}
237

238
			log.Printf("📋 Orders Block %d:", update.Data.BlockNumber)
239
			prettyJSON, _ := json.MarshalIndent(data, "", "  ")
240
			fmt.Println(string(prettyJSON))
241
		}
242
	}
243

244
	return nil
245
}
246

247

248
// Interactive menu
249
func runInteractiveMenu(conn *grpc.ClientConn) {
250
	client := pb.NewStreamingClient(conn)
251

252
	fmt.Println("\n🎉 Connected successfully!")
253
	fmt.Println("\nAvailable streams:")
254
	fmt.Println("1. All trades")
255
	fmt.Println("2. Filtered trades (BTC, ETH)")
256
	fmt.Println("3. Orders")
257
	fmt.Println("4. Exit")
258

259
	for {
260
		fmt.Print("\nSelect stream (1-4): ")
261
		var choice string
262
		fmt.Scanln(&choice)
263

264
		switch choice {
265
		case "1":
266
			if err := streamTrades(client); err != nil {
267
				log.Printf("❌ Stream error: %v", err)
268
			}
269
		case "2":
270
			if err := streamFilteredTrades(client, []string{"BTC", "ETH"}); err != nil {
271
				log.Printf("❌ Filtered stream error: %v", err)
272
			}
273
		case "3":
274
			if err := streamOrders(client); err != nil {
275
				log.Printf("❌ Orders stream error: %v", err)
276
			}
277
		case "4":
278
			fmt.Println("👋 Goodbye!")
279
			return
280
		default:
281
			fmt.Println("❌ Invalid choice. Please select 1-4.")
282
		}
283
	}
284
}
285

286
func main() {
287
	// Create connection
288
	conn, err := createConnection()
289
	if err != nil {
290
		log.Fatalf("❌ Failed to connect: %v", err)
291
	}
292
	defer conn.Close()
293

294
	fmt.Printf("Connected to: %s\n", grpcEndpoint)
295

296
	// Run interactive menu
297
	runInteractiveMenu(conn)
298
}

1
package main
2

3
import (
4
	"context"
5
	"crypto/tls"
6
	"encoding/json"
7
	"fmt"
8
	"io"
9
	"log"
10
	"time"
11

12
	"google.golang.org/grpc"
13
	"google.golang.org/grpc/credentials"
14
	"google.golang.org/grpc/metadata"
15

16
	pb "hyperliquid-grpc-go/pb"
17
)
18

19
const (
20
	// Using Quicknode endpoint
21
	grpcEndpoint = "your-endpoint.hype-mainnet.quiknode.pro:10000"
22
	authToken    = "your-auth-token"
23
)
24

25
// Create gRPC connection with TLS
26
func createConnection() (*grpc.ClientConn, error) {
27
	tlsConfig := &tls.Config{
28
		InsecureSkipVerify: false, // Set to true only for testing
29
	}
30
	creds := credentials.NewTLS(tlsConfig)
31

32
	conn, err := grpc.Dial(
33
		grpcEndpoint,
34
		grpc.WithTransportCredentials(creds),
35
		grpc.WithDefaultCallOptions(
36
			grpc.MaxCallRecvMsgSize(100*1024*1024), // 100MB
37
		),
38
	)
39

40
	return conn, err
41
}
42

43
// Create context with auth metadata
44
func createContext() context.Context {
45
	md := metadata.Pairs("x-token", authToken)
46
	return metadata.NewOutgoingContext(context.Background(), md)
47
}
48

49

50
// Stream trades
51
func streamTrades(client pb.StreamingClient) error {
52
	ctx := createContext()
53

54
	stream, err := client.StreamData(ctx)
55
	if err != nil {
56
		return fmt.Errorf("failed to create stream: %v", err)
57
	}
58

59
	// Send subscription request
60
	// Empty Filters means receive all events for this stream type
61
	err = stream.Send(&pb.SubscribeRequest{
62
		Request: &pb.SubscribeRequest_Subscribe{
63
			Subscribe: &pb.StreamSubscribe{
64
				StreamType: pb.StreamType_TRADES,
65
				StartBlock: 0,
66
				Filters:    map[string]*pb.FilterValues{},
67
			},
68
		},
69
	})
70
	if err != nil {
71
		return fmt.Errorf("failed to subscribe: %v", err)
72
	}
73

74
	// Start ping goroutine
75
	go func() {
76
		ticker := time.NewTicker(30 * time.Second)
77
		defer ticker.Stop()
78

79
		for range ticker.C {
80
			err := stream.Send(&pb.SubscribeRequest{
81
				Request: &pb.SubscribeRequest_Ping{
82
					Ping: &pb.Ping{
83
						Timestamp: time.Now().UnixMilli(),
84
					},
85
				},
86
			})
87
			if err != nil {
88
				log.Printf("Failed to send ping: %v", err)
89
				return
90
			}
91
		}
92
	}()
93

94
	// Receive messages
95
	log.Println("🔄 Starting trades stream...")
96
	for {
97
		msg, err := stream.Recv()
98
		if err == io.EOF {
99
			log.Println("Stream ended")
100
			break
101
		}
102
		if err != nil {
103
			return fmt.Errorf("receive error: %v", err)
104
		}
105

106
		switch update := msg.Update.(type) {
107
		case *pb.SubscribeUpdate_Data:
108
			var data map[string]interface{}
109
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
110
				log.Printf("⚠️  Failed to parse data: %v", err)
111
				continue
112
			}
113

114
			trades, ok := data["trades"].([]interface{})
115
			tradesCount := 0
116
			if ok {
117
				tradesCount = len(trades)
118
			}
119

120
			log.Printf("📈 Block %d: %d trades", update.Data.BlockNumber, tradesCount)
121
			
122
			// Pretty print first trade if available
123
			if tradesCount > 0 {
124
				firstTrade := trades[0]
125
				tradeJSON, _ := json.MarshalIndent(firstTrade, "   ", "  ")
126
				fmt.Printf("   Example trade:\n%s\n", tradeJSON)
127
			}
128

129
		case *pb.SubscribeUpdate_Pong:
130
			log.Printf("🏓 Pong received: %d", update.Pong.Timestamp)
131
		}
132
	}
133

134
	return nil
135
}
136

137
// Stream filtered trades
138
func streamFilteredTrades(client pb.StreamingClient, coins []string) error {
139
	ctx := createContext()
140

141
	stream, err := client.StreamData(ctx)
142
	if err != nil {
143
		return fmt.Errorf("failed to create stream: %v", err)
144
	}
145

146
	// Send filtered subscription
147
	err = stream.Send(&pb.SubscribeRequest{
148
		Request: &pb.SubscribeRequest_Subscribe{
149
			Subscribe: &pb.StreamSubscribe{
150
				StreamType: pb.StreamType_TRADES,
151
				StartBlock: 0,
152
				Filters: map[string]*pb.FilterValues{
153
					"coin": {Values: coins},
154
				},
155
			},
156
		},
157
	})
158
	if err != nil {
159
		return fmt.Errorf("failed to subscribe: %v", err)
160
	}
161

162
	log.Printf("🔄 Starting filtered trades stream for coins: %v", coins)
163

164
	// Receive messages
165
	for {
166
		msg, err := stream.Recv()
167
		if err == io.EOF {
168
			log.Println("Filtered stream ended")
169
			break
170
		}
171
		if err != nil {
172
			return fmt.Errorf("receive error: %v", err)
173
		}
174

175
		switch update := msg.Update.(type) {
176
		case *pb.SubscribeUpdate_Data:
177
			var data map[string]interface{}
178
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
179
				log.Printf("⚠️  Failed to parse data: %v", err)
180
				continue
181
			}
182

183
			if trades, ok := data["trades"].([]interface{}); ok && len(trades) > 0 {
184
				log.Printf("📈 Filtered Block %d: %d trades", update.Data.BlockNumber, len(trades))
185
				prettyJSON, _ := json.MarshalIndent(data, "", "  ")
186
				fmt.Println(string(prettyJSON))
187
			}
188
		}
189
	}
190

191
	return nil
192
}
193

194
// Stream orders
195
func streamOrders(client pb.StreamingClient) error {
196
	ctx := createContext()
197

198
	stream, err := client.StreamData(ctx)
199
	if err != nil {
200
		return fmt.Errorf("failed to create stream: %v", err)
201
	}
202

203
	// Send subscription request for orders
204
	err = stream.Send(&pb.SubscribeRequest{
205
		Request: &pb.SubscribeRequest_Subscribe{
206
			Subscribe: &pb.StreamSubscribe{
207
				StreamType: pb.StreamType_ORDERS,
208
				StartBlock: 0,
209
				Filters:    map[string]*pb.FilterValues{},
210
			},
211
		},
212
	})
213
	if err != nil {
214
		return fmt.Errorf("failed to subscribe: %v", err)
215
	}
216

217
	log.Println("🔄 Starting orders stream...")
218

219
	// Receive messages
220
	for {
221
		msg, err := stream.Recv()
222
		if err == io.EOF {
223
			log.Println("Orders stream ended")
224
			break
225
		}
226
		if err != nil {
227
			return fmt.Errorf("receive error: %v", err)
228
		}
229

230
		switch update := msg.Update.(type) {
231
		case *pb.SubscribeUpdate_Data:
232
			var data map[string]interface{}
233
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
234
				log.Printf("⚠️  Failed to parse order data: %v", err)
235
				continue
236
			}
237

238
			log.Printf("📋 Orders Block %d:", update.Data.BlockNumber)
239
			prettyJSON, _ := json.MarshalIndent(data, "", "  ")
240
			fmt.Println(string(prettyJSON))
241
		}
242
	}
243

244
	return nil
245
}
246

247

248
// Interactive menu
249
func runInteractiveMenu(conn *grpc.ClientConn) {
250
	client := pb.NewStreamingClient(conn)
251

252
	fmt.Println("\n🎉 Connected successfully!")
253
	fmt.Println("\nAvailable streams:")
254
	fmt.Println("1. All trades")
255
	fmt.Println("2. Filtered trades (BTC, ETH)")
256
	fmt.Println("3. Orders")
257
	fmt.Println("4. Exit")
258

259
	for {
260
		fmt.Print("\nSelect stream (1-4): ")
261
		var choice string
262
		fmt.Scanln(&choice)
263

264
		switch choice {
265
		case "1":
266
			if err := streamTrades(client); err != nil {
267
				log.Printf("❌ Stream error: %v", err)
268
			}
269
		case "2":
270
			if err := streamFilteredTrades(client, []string{"BTC", "ETH"}); err != nil {
271
				log.Printf("❌ Filtered stream error: %v", err)
272
			}
273
		case "3":
274
			if err := streamOrders(client); err != nil {
275
				log.Printf("❌ Orders stream error: %v", err)
276
			}
277
		case "4":
278
			fmt.Println("👋 Goodbye!")
279
			return
280
		default:
281
			fmt.Println("❌ Invalid choice. Please select 1-4.")
282
		}
283
	}
284
}
285

286
func main() {
287
	// Create connection
288
	conn, err := createConnection()
289
	if err != nil {
290
		log.Fatalf("❌ Failed to connect: %v", err)
291
	}
292
	defer conn.Close()
293

294
	fmt.Printf("Connected to: %s\n", grpcEndpoint)
295

296
	// Run interactive menu
297
	runInteractiveMenu(conn)
298
}

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

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

9
// Load proto file
10
const packageDefinition = protoLoader.loadSync(
11
  path.join(__dirname, 'proto/streaming.proto'),
12
  {
13
    keepCase: true,
14
    longs: String,
15
    enums: String,
16
    defaults: true,
17
    oneofs: true,
18
  }
19
);
20

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

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

27
  const client = new hyperliquidProto.Streaming(
28
    GRPC_ENDPOINT,
29
    credentials,
30
    {
31
      'grpc.max_receive_message_length': 100 * 1024 * 1024, // 100MB
32
    }
33
  );
34

35
  return client;
36
}
37

38
// Create metadata with auth token
39
function createMetadata() {
40
  const metadata = new grpc.Metadata();
41
  metadata.add('x-token', AUTH_TOKEN);
42
  return metadata;
43
}
44

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

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

56
      console.log('✅ Ping successful:', response);
57
      resolve(response);
58
    });
59
  });
60
}
61

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

67
    const stream = client.StreamData(metadata);
68

69
    // Handle incoming messages
70
    stream.on('data', (response) => {
71
      if (response.data) {
72
        try {
73
          const data = JSON.parse(response.data.data);
74
          const trades = data.trades || [];
75

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

78
          // Pretty print first trade if available
79
          if (trades.length > 0) {
80
            console.log('   Example trade:', JSON.stringify(trades[0], null, 2));
81
          }
82
        } catch (error) {
83
          console.log('⚠️  Failed to parse data:', error.message);
84
        }
85
      } else if (response.pong) {
86
        console.log('🏓 Pong received:', response.pong.timestamp);
87
      }
88
    });
89

90
    stream.on('error', (error) => {
91
      console.error('❌ Stream error:', error.message);
92
      reject(error);
93
    });
94

95
    stream.on('end', () => {
96
      console.log('Stream ended');
97
      resolve();
98
    });
99

100
    // Send subscription request
101
    // Empty filters means receive all events for this stream type
102
    stream.write({
103
      subscribe: {
104
        stream_type: 'TRADES',
105
        filters: {}
106
      }
107
    });
108

109
    // Start ping interval to keep connection alive
110
    const pingInterval = setInterval(() => {
111
      stream.write({
112
        ping: {
113
          timestamp: Date.now()
114
        }
115
      });
116
    }, 30000);
117

118
    // Clean up on exit
119
    const cleanup = () => {
120
      clearInterval(pingInterval);
121
      stream.end();
122
    };
123

124
    process.on('SIGINT', cleanup);
125
    process.on('SIGTERM', cleanup);
126

127
    console.log('🔄 Starting trades stream...');
128
  });
129
}
130

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

136
    const stream = client.StreamData(metadata);
137

138
    stream.on('data', (response) => {
139
      if (response.data) {
140
        try {
141
          const data = JSON.parse(response.data.data);
142
          console.log(`📋 Orders Block ${response.data.block_number}:`);
143
          console.log(JSON.stringify(data, null, 2));
144
        } catch (error) {
145
          console.log('⚠️  Failed to parse order data:', error.message);
146
        }
147
      }
148
    });
149

150
    stream.on('error', (error) => {
151
      console.error('❌ Stream error:', error.message);
152
      reject(error);
153
    });
154

155
    stream.on('end', () => {
156
      console.log('Orders stream ended');
157
      resolve();
158
    });
159

160
    // Send subscription request for orders
161
    stream.write({
162
      subscribe: {
163
        stream_type: 'ORDERS',
164
        filters: {}
165
      }
166
    });
167

168
    console.log('🔄 Starting orders stream...');
169
  });
170
}
171

172
// Interactive menu
173
function showMenu() {
174
  console.log('\n🎉 Connected successfully!');
175
  console.log('\nAvailable streams:');
176
  console.log('1. All trades');
177
  console.log('2. Filtered trades (BTC, ETH)');
178
  console.log('3. Orders');
179
  console.log('4. Test ping');
180
  console.log('5. Exit');
181

182
  const readline = require('readline');
183
  const rl = readline.createInterface({
184
    input: process.stdin,
185
    output: process.stdout
186
  });
187

188
  return new Promise((resolve) => {
189
    rl.question('\nSelect stream (1-5): ', (choice) => {
190
      rl.close();
191
      resolve(choice);
192
    });
193
  });
194
}
195

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

201
    const stream = client.StreamData(metadata);
202

203
    stream.on('data', (response) => {
204
      if (response.data) {
205
        try {
206
          const data = JSON.parse(response.data.data);
207
          const trades = data.trades || [];
208

209
          if (trades.length > 0) {
210
            console.log(`📈 Filtered Block ${response.data.block_number}: ${trades.length} trades`);
211
            console.log(JSON.stringify(data, null, 2));
212
          }
213
        } catch (error) {
214
          console.log('⚠️  Failed to parse data:', error.message);
215
        }
216
      }
217
    });
218

219
    stream.on('error', (error) => {
220
      console.error('❌ Stream error:', error.message);
221
      reject(error);
222
    });
223

224
    stream.on('end', () => {
225
      console.log('Filtered stream ended');
226
      resolve();
227
    });
228

229
    // Send filtered subscription
230
    stream.write({
231
      subscribe: {
232
        stream_type: 'TRADES',
233
        filters: {
234
          coin: { values: coins }
235
        }
236
      }
237
    });
238

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

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

248
    // Create client
249
    const client = createClient();
250

251
    // Test initial connectivity
252
    await testPing(client);
253

254
    // Show interactive menu
255
    while (true) {
256
      const choice = await showMenu();
257

258
      switch (choice) {
259
        case '1':
260
          await streamTrades(client);
261
          break;
262
        case '2':
263
          await streamFilteredTrades(client, ['BTC', 'ETH']);
264
          break;
265
        case '3':
266
          await streamOrders(client);
267
          break;
268
        case '4':
269
          await testPing(client);
270
          break;
271
        case '5':
272
          console.log('👋 Goodbye!');
273
          client.close();
274
          return;
275
        default:
276
          console.log('❌ Invalid choice. Please select 1-5.');
277
      }
278
    }
279

280
  } catch (error) {
281
    console.error('❌ Error:', error.message);
282
    process.exit(1);
283
  }
284
}
285

286
// Run the application
287
main();

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

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

9
// Load proto file
10
const packageDefinition = protoLoader.loadSync(
11
  path.join(__dirname, 'proto/streaming.proto'),
12
  {
13
    keepCase: true,
14
    longs: String,
15
    enums: String,
16
    defaults: true,
17
    oneofs: true,
18
  }
19
);
20

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

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

27
  const client = new hyperliquidProto.Streaming(
28
    GRPC_ENDPOINT,
29
    credentials,
30
    {
31
      'grpc.max_receive_message_length': 100 * 1024 * 1024, // 100MB
32
    }
33
  );
34

35
  return client;
36
}
37

38
// Create metadata with auth token
39
function createMetadata() {
40
  const metadata = new grpc.Metadata();
41
  metadata.add('x-token', AUTH_TOKEN);
42
  return metadata;
43
}
44

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

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

56
      console.log('✅ Ping successful:', response);
57
      resolve(response);
58
    });
59
  });
60
}
61

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

67
    const stream = client.StreamData(metadata);
68

69
    // Handle incoming messages
70
    stream.on('data', (response) => {
71
      if (response.data) {
72
        try {
73
          const data = JSON.parse(response.data.data);
74
          const trades = data.trades || [];
75

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

78
          // Pretty print first trade if available
79
          if (trades.length > 0) {
80
            console.log('   Example trade:', JSON.stringify(trades[0], null, 2));
81
          }
82
        } catch (error) {
83
          console.log('⚠️  Failed to parse data:', error.message);
84
        }
85
      } else if (response.pong) {
86
        console.log('🏓 Pong received:', response.pong.timestamp);
87
      }
88
    });
89

90
    stream.on('error', (error) => {
91
      console.error('❌ Stream error:', error.message);
92
      reject(error);
93
    });
94

95
    stream.on('end', () => {
96
      console.log('Stream ended');
97
      resolve();
98
    });
99

100
    // Send subscription request
101
    // Empty filters means receive all events for this stream type
102
    stream.write({
103
      subscribe: {
104
        stream_type: 'TRADES',
105
        filters: {}
106
      }
107
    });
108

109
    // Start ping interval to keep connection alive
110
    const pingInterval = setInterval(() => {
111
      stream.write({
112
        ping: {
113
          timestamp: Date.now()
114
        }
115
      });
116
    }, 30000);
117

118
    // Clean up on exit
119
    const cleanup = () => {
120
      clearInterval(pingInterval);
121
      stream.end();
122
    };
123

124
    process.on('SIGINT', cleanup);
125
    process.on('SIGTERM', cleanup);
126

127
    console.log('🔄 Starting trades stream...');
128
  });
129
}
130

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

136
    const stream = client.StreamData(metadata);
137

138
    stream.on('data', (response) => {
139
      if (response.data) {
140
        try {
141
          const data = JSON.parse(response.data.data);
142
          console.log(`📋 Orders Block ${response.data.block_number}:`);
143
          console.log(JSON.stringify(data, null, 2));
144
        } catch (error) {
145
          console.log('⚠️  Failed to parse order data:', error.message);
146
        }
147
      }
148
    });
149

150
    stream.on('error', (error) => {
151
      console.error('❌ Stream error:', error.message);
152
      reject(error);
153
    });
154

155
    stream.on('end', () => {
156
      console.log('Orders stream ended');
157
      resolve();
158
    });
159

160
    // Send subscription request for orders
161
    stream.write({
162
      subscribe: {
163
        stream_type: 'ORDERS',
164
        filters: {}
165
      }
166
    });
167

168
    console.log('🔄 Starting orders stream...');
169
  });
170
}
171

172
// Interactive menu
173
function showMenu() {
174
  console.log('\n🎉 Connected successfully!');
175
  console.log('\nAvailable streams:');
176
  console.log('1. All trades');
177
  console.log('2. Filtered trades (BTC, ETH)');
178
  console.log('3. Orders');
179
  console.log('4. Test ping');
180
  console.log('5. Exit');
181

182
  const readline = require('readline');
183
  const rl = readline.createInterface({
184
    input: process.stdin,
185
    output: process.stdout
186
  });
187

188
  return new Promise((resolve) => {
189
    rl.question('\nSelect stream (1-5): ', (choice) => {
190
      rl.close();
191
      resolve(choice);
192
    });
193
  });
194
}
195

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

201
    const stream = client.StreamData(metadata);
202

203
    stream.on('data', (response) => {
204
      if (response.data) {
205
        try {
206
          const data = JSON.parse(response.data.data);
207
          const trades = data.trades || [];
208

209
          if (trades.length > 0) {
210
            console.log(`📈 Filtered Block ${response.data.block_number}: ${trades.length} trades`);
211
            console.log(JSON.stringify(data, null, 2));
212
          }
213
        } catch (error) {
214
          console.log('⚠️  Failed to parse data:', error.message);
215
        }
216
      }
217
    });
218

219
    stream.on('error', (error) => {
220
      console.error('❌ Stream error:', error.message);
221
      reject(error);
222
    });
223

224
    stream.on('end', () => {
225
      console.log('Filtered stream ended');
226
      resolve();
227
    });
228

229
    // Send filtered subscription
230
    stream.write({
231
      subscribe: {
232
        stream_type: 'TRADES',
233
        filters: {
234
          coin: { values: coins }
235
        }
236
      }
237
    });
238

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

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

248
    // Create client
249
    const client = createClient();
250

251
    // Test initial connectivity
252
    await testPing(client);
253

254
    // Show interactive menu
255
    while (true) {
256
      const choice = await showMenu();
257

258
      switch (choice) {
259
        case '1':
260
          await streamTrades(client);
261
          break;
262
        case '2':
263
          await streamFilteredTrades(client, ['BTC', 'ETH']);
264
          break;
265
        case '3':
266
          await streamOrders(client);
267
          break;
268
        case '4':
269
          await testPing(client);
270
          break;
271
        case '5':
272
          console.log('👋 Goodbye!');
273
          client.close();
274
          return;
275
        default:
276
          console.log('❌ Invalid choice. Please select 1-5.');
277
      }
278
    }
279

280
  } catch (error) {
281
    console.error('❌ Error:', error.message);
282
    process.exit(1);
283
  }
284
}
285

286
// Run the application
287
main();

1
import grpc
2
import json
3
import time
4
import signal
5
import sys
6
import threading
7
from pb import streaming_pb2, streaming_pb2_grpc
8

9
# Configuration
10
GRPC_ENDPOINT = 'your-endpoint.hype-mainnet.quiknode.pro:10000'
11
AUTH_TOKEN = 'your-auth-token'
12

13
def create_channel():
14
    """Create gRPC channel with TLS"""
15
    credentials = grpc.ssl_channel_credentials()
16
    channel = grpc.secure_channel(
17
        GRPC_ENDPOINT,
18
        credentials,
19
        options=[
20
            ('grpc.max_receive_message_length', 100 * 1024 * 1024),  # 100MB
21
            ('grpc.max_send_message_length', 100 * 1024 * 1024),     # 100MB
22
        ]
23
    )
24
    return channel
25

26
def create_metadata():
27
    """Create metadata with auth token"""
28
    return [('x-token', AUTH_TOKEN)]
29

30
def test_ping(stub):
31
    """Test connectivity with ping"""
32
    try:
33
        metadata = create_metadata()
34
        request = streaming_pb2.PingRequest(count=1)
35

36
        response = stub.Ping(request, metadata=metadata)
37
        print(f"✅ Ping successful: {response}")
38
        return True
39
    except grpc.RpcError as e:
40
        print(f"❌ Ping failed: {e.details()}")
41
        return False
42

43
def stream_trades(stub):
44
    """Stream trades data"""
45
    try:
46
        metadata = create_metadata()
47

48
        print("🔄 Starting trades stream...")
49

50
        # Create request generator for bidirectional stream
51
        def request_generator():
52
            # Send initial subscription request
53
            subscribe_request = streaming_pb2.SubscribeRequest()
54
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
55
            # An empty filters map means receive all events for this stream type
56
            yield subscribe_request
57

58
            # Keep sending pings to maintain connection
59
            while True:
60
                time.sleep(30)
61
                ping_request = streaming_pb2.SubscribeRequest()
62
                ping_request.ping.timestamp = int(time.time() * 1000)
63
                yield ping_request
64

65
        # Create bidirectional stream
66
        stream = stub.StreamData(request_generator(), metadata=metadata)
67

68
        # Handle stream responses
69
        for response in stream:
70
            try:
71
                if response.HasField('data'):
72
                    data = json.loads(response.data.data)
73
                    trades = data.get('trades', [])
74

75
                    print(f"📈 Block {response.data.block_number}: {len(trades)} trades")
76

77
                    # Pretty print first trade if available
78
                    if trades:
79
                        print(f"   Example trade: {json.dumps(trades[0], indent=2)}")
80

81
                elif response.HasField('pong'):
82
                    print(f"🏓 Pong received: {response.pong.timestamp}")
83

84
            except json.JSONDecodeError as e:
85
                print(f"⚠️  Failed to parse data: {e}")
86
            except KeyboardInterrupt:
87
                print("\n⏹️  Stream interrupted by user")
88
                break
89

90
    except grpc.RpcError as e:
91
        print(f"❌ Stream error: {e.details()}")
92
    except KeyboardInterrupt:
93
        print("\n⏹️  Stream interrupted by user")
94

95
def stream_orders(stub):
96
    """Stream orders data"""
97
    try:
98
        metadata = create_metadata()
99

100
        print("🔄 Starting orders stream...")
101

102
        # Create request generator for bidirectional stream
103
        def request_generator():
104
            # Send initial subscription request for orders
105
            subscribe_request = streaming_pb2.SubscribeRequest()
106
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.ORDERS
107
            yield subscribe_request
108

109
            # Keep sending pings to maintain connection
110
            while True:
111
                time.sleep(30)
112
                ping_request = streaming_pb2.SubscribeRequest()
113
                ping_request.ping.timestamp = int(time.time() * 1000)
114
                yield ping_request
115

116
        # Create bidirectional stream
117
        stream = stub.StreamData(request_generator(), metadata=metadata)
118

119
        # Handle stream responses
120
        for response in stream:
121
            try:
122
                if response.HasField('data'):
123
                    data = json.loads(response.data.data)
124
                    print(f"📋 Orders Block {response.data.block_number}:")
125
                    print(json.dumps(data, indent=2))
126

127
            except json.JSONDecodeError as e:
128
                print(f"⚠️  Failed to parse order data: {e}")
129
            except KeyboardInterrupt:
130
                print("\n⏹️  Stream interrupted by user")
131
                break
132

133
    except grpc.RpcError as e:
134
        print(f"❌ Stream error: {e.details()}")
135
    except KeyboardInterrupt:
136
        print("\n⏹️  Stream interrupted by user")
137

138
def stream_filtered_trades(stub, coins):
139
    """Stream filtered trades for specific coins"""
140
    try:
141
        metadata = create_metadata()
142

143
        print(f"🔄 Starting filtered trades stream for coins: {', '.join(coins)}")
144

145
        # Create request generator for bidirectional stream
146
        def request_generator():
147
            # Send filtered subscription request
148
            subscribe_request = streaming_pb2.SubscribeRequest()
149
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
150
            subscribe_request.subscribe.filters["coin"].values.extend(coins)
151
            yield subscribe_request
152

153
            # Keep sending pings to maintain connection
154
            while True:
155
                time.sleep(30)
156
                ping_request = streaming_pb2.SubscribeRequest()
157
                ping_request.ping.timestamp = int(time.time() * 1000)
158
                yield ping_request
159

160
        # Create bidirectional stream
161
        stream = stub.StreamData(request_generator(), metadata=metadata)
162

163
        # Handle stream responses
164
        for response in stream:
165
            try:
166
                if response.HasField('data'):
167
                    data = json.loads(response.data.data)
168
                    trades = data.get('trades', [])
169

170
                    if trades:
171
                        print(f"📈 Filtered Block {response.data.block_number}: {len(trades)} trades")
172
                        print(json.dumps(data, indent=2))
173

174
            except json.JSONDecodeError as e:
175
                print(f"⚠️  Failed to parse data: {e}")
176
            except KeyboardInterrupt:
177
                print("\n⏹️  Stream interrupted by user")
178
                break
179

180
    except grpc.RpcError as e:
181
        print(f"❌ Stream error: {e.details()}")
182
    except KeyboardInterrupt:
183
        print("\n⏹️  Stream interrupted by user")
184

185

186
def show_menu():
187
    """Display interactive menu"""
188
    print("\n🎉 Connected successfully!")
189
    print("\nAvailable streams:")
190
    print("1. All trades")
191
    print("2. Filtered trades (BTC, ETH)")
192
    print("3. Orders")
193
    print("4. Test ping")
194
    print("5. Exit")
195

196
    while True:
197
        try:
198
            choice = input("\nSelect stream (1-5): ").strip()
199
            if choice in ['1', '2', '3', '4', '5']:
200
                return choice
201
            else:
202
                print("❌ Invalid choice. Please select 1-5.")
203
        except KeyboardInterrupt:
204
            return '5'
205

206
def main():
207
    """Main function"""
208
    try:
209
        print(f"Testing connection to: {GRPC_ENDPOINT}")
210

211
        # Create channel and stub
212
        channel = create_channel()
213
        streaming_stub = streaming_pb2_grpc.StreamingStub(channel)
214

215
        # Test initial connectivity
216
        if not test_ping(streaming_stub):
217
            print("❌ Connection test failed!")
218
            return
219

220
        # Interactive menu loop
221
        while True:
222
            choice = show_menu()
223

224
            if choice == '1':
225
                stream_trades(streaming_stub)
226
            elif choice == '2':
227
                stream_filtered_trades(streaming_stub, ['BTC', 'ETH'])
228
            elif choice == '3':
229
                stream_orders(streaming_stub)
230
            elif choice == '4':
231
                test_ping(streaming_stub)
232
            elif choice == '5':
233
                print("👋 Goodbye!")
234
                break
235

236
        # Close channel
237
        channel.close()
238

239
    except KeyboardInterrupt:
240
        print("\n👋 Goodbye!")
241
    except Exception as e:
242
        print(f"❌ Error: {e}")
243
        sys.exit(1)
244

245
if __name__ == "__main__":
246
    main()

1
import grpc
2
import json
3
import time
4
import signal
5
import sys
6
import threading
7
from pb import streaming_pb2, streaming_pb2_grpc
8

9
# Configuration
10
GRPC_ENDPOINT = 'your-endpoint.hype-mainnet.quiknode.pro:10000'
11
AUTH_TOKEN = 'your-auth-token'
12

13
def create_channel():
14
    """Create gRPC channel with TLS"""
15
    credentials = grpc.ssl_channel_credentials()
16
    channel = grpc.secure_channel(
17
        GRPC_ENDPOINT,
18
        credentials,
19
        options=[
20
            ('grpc.max_receive_message_length', 100 * 1024 * 1024),  # 100MB
21
            ('grpc.max_send_message_length', 100 * 1024 * 1024),     # 100MB
22
        ]
23
    )
24
    return channel
25

26
def create_metadata():
27
    """Create metadata with auth token"""
28
    return [('x-token', AUTH_TOKEN)]
29

30
def test_ping(stub):
31
    """Test connectivity with ping"""
32
    try:
33
        metadata = create_metadata()
34
        request = streaming_pb2.PingRequest(count=1)
35

36
        response = stub.Ping(request, metadata=metadata)
37
        print(f"✅ Ping successful: {response}")
38
        return True
39
    except grpc.RpcError as e:
40
        print(f"❌ Ping failed: {e.details()}")
41
        return False
42

43
def stream_trades(stub):
44
    """Stream trades data"""
45
    try:
46
        metadata = create_metadata()
47

48
        print("🔄 Starting trades stream...")
49

50
        # Create request generator for bidirectional stream
51
        def request_generator():
52
            # Send initial subscription request
53
            subscribe_request = streaming_pb2.SubscribeRequest()
54
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
55
            # An empty filters map means receive all events for this stream type
56
            yield subscribe_request
57

58
            # Keep sending pings to maintain connection
59
            while True:
60
                time.sleep(30)
61
                ping_request = streaming_pb2.SubscribeRequest()
62
                ping_request.ping.timestamp = int(time.time() * 1000)
63
                yield ping_request
64

65
        # Create bidirectional stream
66
        stream = stub.StreamData(request_generator(), metadata=metadata)
67

68
        # Handle stream responses
69
        for response in stream:
70
            try:
71
                if response.HasField('data'):
72
                    data = json.loads(response.data.data)
73
                    trades = data.get('trades', [])
74

75
                    print(f"📈 Block {response.data.block_number}: {len(trades)} trades")
76

77
                    # Pretty print first trade if available
78
                    if trades:
79
                        print(f"   Example trade: {json.dumps(trades[0], indent=2)}")
80

81
                elif response.HasField('pong'):
82
                    print(f"🏓 Pong received: {response.pong.timestamp}")
83

84
            except json.JSONDecodeError as e:
85
                print(f"⚠️  Failed to parse data: {e}")
86
            except KeyboardInterrupt:
87
                print("\n⏹️  Stream interrupted by user")
88
                break
89

90
    except grpc.RpcError as e:
91
        print(f"❌ Stream error: {e.details()}")
92
    except KeyboardInterrupt:
93
        print("\n⏹️  Stream interrupted by user")
94

95
def stream_orders(stub):
96
    """Stream orders data"""
97
    try:
98
        metadata = create_metadata()
99

100
        print("🔄 Starting orders stream...")
101

102
        # Create request generator for bidirectional stream
103
        def request_generator():
104
            # Send initial subscription request for orders
105
            subscribe_request = streaming_pb2.SubscribeRequest()
106
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.ORDERS
107
            yield subscribe_request
108

109
            # Keep sending pings to maintain connection
110
            while True:
111
                time.sleep(30)
112
                ping_request = streaming_pb2.SubscribeRequest()
113
                ping_request.ping.timestamp = int(time.time() * 1000)
114
                yield ping_request
115

116
        # Create bidirectional stream
117
        stream = stub.StreamData(request_generator(), metadata=metadata)
118

119
        # Handle stream responses
120
        for response in stream:
121
            try:
122
                if response.HasField('data'):
123
                    data = json.loads(response.data.data)
124
                    print(f"📋 Orders Block {response.data.block_number}:")
125
                    print(json.dumps(data, indent=2))
126

127
            except json.JSONDecodeError as e:
128
                print(f"⚠️  Failed to parse order data: {e}")
129
            except KeyboardInterrupt:
130
                print("\n⏹️  Stream interrupted by user")
131
                break
132

133
    except grpc.RpcError as e:
134
        print(f"❌ Stream error: {e.details()}")
135
    except KeyboardInterrupt:
136
        print("\n⏹️  Stream interrupted by user")
137

138
def stream_filtered_trades(stub, coins):
139
    """Stream filtered trades for specific coins"""
140
    try:
141
        metadata = create_metadata()
142

143
        print(f"🔄 Starting filtered trades stream for coins: {', '.join(coins)}")
144

145
        # Create request generator for bidirectional stream
146
        def request_generator():
147
            # Send filtered subscription request
148
            subscribe_request = streaming_pb2.SubscribeRequest()
149
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
150
            subscribe_request.subscribe.filters["coin"].values.extend(coins)
151
            yield subscribe_request
152

153
            # Keep sending pings to maintain connection
154
            while True:
155
                time.sleep(30)
156
                ping_request = streaming_pb2.SubscribeRequest()
157
                ping_request.ping.timestamp = int(time.time() * 1000)
158
                yield ping_request
159

160
        # Create bidirectional stream
161
        stream = stub.StreamData(request_generator(), metadata=metadata)
162

163
        # Handle stream responses
164
        for response in stream:
165
            try:
166
                if response.HasField('data'):
167
                    data = json.loads(response.data.data)
168
                    trades = data.get('trades', [])
169

170
                    if trades:
171
                        print(f"📈 Filtered Block {response.data.block_number}: {len(trades)} trades")
172
                        print(json.dumps(data, indent=2))
173

174
            except json.JSONDecodeError as e:
175
                print(f"⚠️  Failed to parse data: {e}")
176
            except KeyboardInterrupt:
177
                print("\n⏹️  Stream interrupted by user")
178
                break
179

180
    except grpc.RpcError as e:
181
        print(f"❌ Stream error: {e.details()}")
182
    except KeyboardInterrupt:
183
        print("\n⏹️  Stream interrupted by user")
184

185

186
def show_menu():
187
    """Display interactive menu"""
188
    print("\n🎉 Connected successfully!")
189
    print("\nAvailable streams:")
190
    print("1. All trades")
191
    print("2. Filtered trades (BTC, ETH)")
192
    print("3. Orders")
193
    print("4. Test ping")
194
    print("5. Exit")
195

196
    while True:
197
        try:
198
            choice = input("\nSelect stream (1-5): ").strip()
199
            if choice in ['1', '2', '3', '4', '5']:
200
                return choice
201
            else:
202
                print("❌ Invalid choice. Please select 1-5.")
203
        except KeyboardInterrupt:
204
            return '5'
205

206
def main():
207
    """Main function"""
208
    try:
209
        print(f"Testing connection to: {GRPC_ENDPOINT}")
210

211
        # Create channel and stub
212
        channel = create_channel()
213
        streaming_stub = streaming_pb2_grpc.StreamingStub(channel)
214

215
        # Test initial connectivity
216
        if not test_ping(streaming_stub):
217
            print("❌ Connection test failed!")
218
            return
219

220
        # Interactive menu loop
221
        while True:
222
            choice = show_menu()
223

224
            if choice == '1':
225
                stream_trades(streaming_stub)
226
            elif choice == '2':
227
                stream_filtered_trades(streaming_stub, ['BTC', 'ETH'])
228
            elif choice == '3':
229
                stream_orders(streaming_stub)
230
            elif choice == '4':
231
                test_ping(streaming_stub)
232
            elif choice == '5':
233
                print("👋 Goodbye!")
234
                break
235

236
        # Close channel
237
        channel.close()
238

239
    except KeyboardInterrupt:
240
        print("\n👋 Goodbye!")
241
    except Exception as e:
242
        print(f"❌ Error: {e}")
243
        sys.exit(1)
244

245
if __name__ == "__main__":
246
    main()

Don't have an account yet?

Create your Quicknode endpoint in seconds and start building

Get started for free