Alpaca API实时数据流报错WebSocket is not open: readyState 0求助

问题描述 #

通过Alpaca API获取实时数据，初始阶段可成功连接并接收数据流，但运行一段时间后出现连接报错。

代码示例 #

const http = require('http');
const server = http.createServer(app);
const io = require('socket.io')(server);

const Alpaca = require('@alpacahq/alpaca-trade-api');
const alpaca = new Alpaca({
  keyId: process.env.ALPACA_API_KEY_ID,
  secretKey: process.env.ALPACA_API_SECRET_KEY,
  paper: true, 
});

io.on('connection', (socket) => {
  console.log(`Client connected with ID: ${socket.id}`);
  let stream = null;

  socket.on('subscribe', (ticker) => {

    console.log(`Subscribing to data for ${ticker}`);

    const stream = alpaca.data_stream_v2;
    stream.onConnect(function () {
        console.log("Connected");
        stream.subscribeForQuotes([(ticker)]);
      });
  
      stream.onStockQuote((quote) => {
        console.log(quote);
        socket.emit('stockData', quote);
      });

      stream.connect();

  });

  socket.on('disconnect', () => {
    console.log(`Client disconnected with ID: ${socket.id}`);
    if (stream) {
      console.log(`Unsubscribing from data for ${stream.ticker}`);
      stream.disconnect();
    }
  });
});

报错信息 #

/Users/vivien/Documents/tradingapp/node_modules/ws/lib/websocket.js:394
      throw new Error('WebSocket is not open: readyState 0 (CONNECTING)');
      ^

Error: WebSocket is not open: readyState 0 (CONNECTING)
    at WebSocket.send (/Users/vivien/Documents/tradingapp/node_modules/ws/lib/websocket.js:394:13)
    at AlpacaStocksClient.authenticate (/Users/vivien/Documents/tradingapp/node_modules/@alpacahq/alpaca-trade-api/dist/resources/datav2/websocket.js:171:19)
    at WebSocket.<anonymous> (/Users/vivien/Documents/tradingapp/node_modules/@alpacahq/alpaca-trade-api/dist/resources/datav2/websocket.js:108:43)
    at Object.onceWrapper (node:events:509:28)
    at WebSocket.emit (node:events:390:28)
    at WebSocket.setSocket (/Users/vivien/Documents/tradingapp/node_modules/ws/lib/websocket.js:225:10)
    at ClientRequest.<anonymous> (/Users/vivien/Documents/tradingapp/node_modules/ws/lib/websocket.js:882:15)
    at ClientRequest.emit (node:events:390:28)
    at TLSSocket.socketOnData (node:_http_client:527:11)
    at TLSSocket.emit (node:events:390:28)

报错原因分析 #

1. 重复连接与事件绑定：每次客户端触发subscribe事件时，都会重新获取全局的alpaca.data_stream_v2实例并调用connect()，多次重复连接会导致WebSocket状态混乱，当连接处于CONNECTING阶段时尝试发送认证请求就会触发报错。
2. 变量作用域错误：subscribe回调内用const stream重新声明变量，导致外部的stream始终为null，客户端断开时无法正确调用disconnect()，旧连接未被清理，积累后引发状态异常。
3. 无重连机制：Alpaca的WebSocket连接可能因网络波动、API超时等原因断开，当前代码未处理断开后的重连逻辑，运行一段时间后连接失效触发报错。

解决方案 #

优化后的代码 #

const http = require('http');
const server = http.createServer(app);
const io = require('socket.io')(server);

const Alpaca = require('@alpacahq/alpaca-trade-api');
const alpaca = new Alpaca({
  keyId: process.env.ALPACA_API_KEY_ID,
  secretKey: process.env.ALPACA_API_SECRET_KEY,
  paper: true, 
});

// 全局复用数据流实例，避免重复创建连接
const globalStream = alpaca.data_stream_v2;
let isStreamConnected = false;
const clientSubscriptions = new Map(); // 记录每个客户端的订阅列表

// 初始化全局流事件监听
globalStream.onConnect(() => {
  console.log("Connected to Alpaca data stream");
  isStreamConnected = true;
  // 重连后恢复所有订阅
  const allTickers = Array.from(clientSubscriptions.values()).flat();
  if (allTickers.length > 0) {
    globalStream.subscribeForQuotes(allTickers);
  }
});

globalStream.onStockQuote((quote) => {
  // 向订阅该标的的客户端推送数据
  clientSubscriptions.forEach((tickers, socketId) => {
    if (tickers.includes(quote.Symbol)) {
      io.to(socketId).emit('stockData', quote);
    }
  });
});

globalStream.onDisconnect(() => {
  console.log("Disconnected from Alpaca data stream");
  isStreamConnected = false;
  // 5秒后自动重连
  setTimeout(() => globalStream.connect(), 5000);
});

globalStream.onError((err) => {
  console.error("Alpaca stream error:", err);
});

io.on('connection', (socket) => {
  console.log(`Client connected with ID: ${socket.id}`);
  clientSubscriptions.set(socket.id, []);

  socket.on('subscribe', (ticker) => {
    console.log(`Client ${socket.id} subscribing to ${ticker}`);
    const clientTickers = clientSubscriptions.get(socket.id);
    if (!clientTickers.includes(ticker)) {
      clientTickers.push(ticker);
      clientSubscriptions.set(socket.id, clientTickers);
      
      // 流已连接则立即订阅，未连接则等待connect时统一订阅
      if (isStreamConnected) {
        globalStream.subscribeForQuotes([ticker]);
      } else {
        globalStream.connect();
      }
    }
  });

  socket.on('disconnect', () => {
    console.log(`Client disconnected with ID: ${socket.id}`);
    const clientTickers = clientSubscriptions.get(socket.id);
    if (clientTickers?.length) {
      // 取消该客户端的订阅
      globalStream.unsubscribeFromQuotes(clientTickers);
      clientSubscriptions.delete(socket.id);
      
      // 无客户端订阅时断开流连接
      if (clientSubscriptions.size === 0) {
        globalStream.disconnect();
        isStreamConnected = false;
      }
    }
  });
});

关键优化点 #

• 全局复用流实例：避免重复创建WebSocket连接，减少资源冲突。
• 订阅状态管理：用Map跟踪每个客户端的订阅列表，精准推送数据并管理订阅生命周期。
• 自动重连机制：连接断开后自动尝试重连，恢复数据接收。
• 修复作用域问题：确保客户端断开时能正确清理订阅与连接资源。

额外注意事项 #

• 确认Alpaca API密钥拥有实时数据访问权限。
• 监听流的错误事件，避免未捕获异常导致程序崩溃。
• 添加详细日志，方便排查连接状态变化与异常情况。

内容的提问来源于stack exchange，提问作者Vivien Richaud