​WebSocket 金融实时行情推送 API 实战解析：低延迟、高可用架构设计与落地

const WebSocket = require("ws");
const Redis = require("ioredis");
const redis = new Redis({ host: "localhost", port: 6379 }); // 连接Redis存储订阅关系

// WebSocket文档配置（参考iTick的官方文档）
const ITICK_CONFIG = {
  wsUrl: "wss://api.itick.org/stock", // WebSocket接入地址
  apiToken: "your_token", // 替换为你的iTick API Token
  pingInterval: 30000, // 心跳间隔30秒
  reconnectDelay: 3000, // 重连延迟3秒，避免频繁重连
  maxReconnectTimes: 10, // 最大重连次数，避免无限重连
  subscribeTypes: ["tick", "quote", "depth", "kline@1"], // 订阅类型（tick成交、quote报价、depth盘口、klineK线）
};

// 存储iTick WebSocket连接实例
let iTickWs = null;
// 存储前端客户端连接（key=客户端ID，value=WebSocket实例）
const clientMap = new Map();
// 存储客户端订阅关系（key=客户端ID，value=订阅标的数组，格式：AAPL$US）
const clientSubscriptions = new Map();
// 存储客户端订阅类型（key=客户端ID，value=订阅类型数组）
const clientSubscribeTypes = new Map();

// 1. 初始化iTick WebSocket连接（连接→认证→订阅）
function initITickConnection() {
  // 关闭现有连接，避免多连接冲突
  if (iTickWs) {
    iTickWs.close(1000, "重新初始化连接");
  }

  // 建立与iTick WebSocket的连接（携带token请求头）
  iTickWs = new WebSocket(ITICK_CONFIG.wsUrl, {
    headers: {
      token: ITICK_CONFIG.apiToken, // 通过header传递token完成认证前置，非原auth指令
    },
  });

  // 1.1 连接成功回调（连接成功后先接收连接成功消息，无需主动发送auth指令）
  iTickWs.on("open", () => {
    console.log("已成功连接iTick官方WebSocket服务器（遵循官方规范）");
  });

  // 1.2 接收iTick服务器消息（处理连接结果、认证结果、订阅结果、行情数据）
  iTickWs.on("message", (message) => {
    try {
      const data = JSON.parse(message.toString());
      // 处理连接成功消息（返回格式：code=1，msg=Connected Successfully）
      if (data.code === 1 && data.msg === "Connected Successfully") {
        console.log("iTick WebSocket连接成功，等待系统认证");
      }
      // 处理认证结果（返回格式：resAc=auth，code=1成功，code=0失败）
      else if (data.resAc === "auth") {
        if (data.code === 1) {
          console.log("iTick API认证成功，可开始订阅行情");
          // 认证成功后，推送所有客户端已订阅的标的（批量订阅）
          pushAllClientSubscriptions();
        } else {
          console.error(
            `iTick认证失败：${data.msg}（错误码：${data.code}），请检查API Token（参考官方文档）`
          );
          // 认证失败，直接断开连接，流程终止，此处触发重连重试
          setTimeout(initITickConnection, ITICK_CONFIG.reconnectDelay);
        }
      }
      // 处理订阅结果（返回格式：resAc=subscribe，code=1成功，code=0失败）
      else if (data.resAc === "subscribe") {
        if (data.code === 1) {
          console.log(`iTick订阅成功：${data.msg}`);
        } else {
          // 错误原因参考官方文档（如超出订阅上限、参数错误
          console.error(
            `iTick订阅失败：${data.msg}（错误码：${data.code}）`
          );
        }
      }
      // 处理心跳响应（返回格式：resAc=pong，data包含对应ping的params时间戳）
      else if (data.resAc === "pong") {
        console.log(
          `收到iTick心跳响应，连接正常，时间戳：${data.data.params}`
        );
      }
      // 处理行情数据（返回格式：code=1，data包含标的、类型及对应字段，分tick/quote/depth/kline四类）
      else if (data.code === 1 && data.data) {
        const marketData = data.data;
        const dataType = marketData.type; // 行情类型：tick/quote/depth/kline@1等
        // 解析iTick行情数据
        let formattedData = {};
        switch (dataType) {
          // 成交数据（tick）
          case "tick":
            formattedData = {
              symbol: marketData.s, // 标的编码（如AAPL$US）
              lastDealPrice: marketData.ld, // 最新成交价
              volume: marketData.v, // 成交量
              tradeTime: marketData.t, // 成交时间戳（毫秒）
              type: marketData.type, // 行情类型：tick
            };
            break;
          // 报价数据（quote）
          case "quote":
            formattedData = {
              symbol: marketData.s, // 标的编码
              lastDealPrice: marketData.ld, // 最新成交价
              openPrice: marketData.o, // 开盘价
              highPrice: marketData.h, // 最高价
              lowPrice: marketData.l, // 最低价
              tradeTime: marketData.t, // 时间戳
              volume: marketData.v, // 成交量
              turnover: marketData.tu, // 成交额
              ts: marketData.ts, 
              type: marketData.type, // 行情类型：quote
            };
            break;
          // 盘口数据（depth）
          case "depth":
            formattedData = {
              symbol: marketData.s, // 标的编码
              ask: marketData.a, // 卖盘（字段：a，数组，包含po/p/v/o字段）
              bid: marketData.b, // 买盘（字段：b，数组，包含po/p/v/o字段）
              type: marketData.type, // 数据类型：depth（字段：type）
            };
            break;
          // K线数据（kline@1及其他周期）
          case "kline@1":
          case "kline@2":
          case "kline@3":
          case "kline@4":
          case "kline@5":
          case "kline@8":
          case "kline@9":
          case "kline@10":
            formattedData = {
              symbol: marketData.s, // 标的编码
              region: marketData.r, // 标的地区
              turnover: marketData.tu, // 当前周期总成交额
              closePrice: marketData.c, // 当前周期收盘价
              time: marketData.t, // 周期时间戳（毫秒）
              volume: marketData.v, // 当前周期总成交量
              highPrice: marketData.h, // 当前周期最高价
              lowPrice: marketData.l, // 当前周期最低价
              openPrice: marketData.o, // 当前周期开盘价
              klineCycle: marketData.type, // K线周期（type，如kline@1=1分钟）
              type: "kline", // 统一数据类型标识
            };
            break;
          default:
            formattedData = marketData;
            console.log(`未匹配的行情类型：${dataType}，按原始格式转发`);
        }
        // 将格式化后的行情数据推送给所有订阅该标的的前端客户端
        pushQuotesToClients(formattedData);
      }
      // 处理其他未知消息
      else {
        console.log("收到iTick未知消息：", data);
      }
    } catch (err) {
      console.error(
        "iTick消息解析失败：",
        err.message
      );
    }
  });

  // 1.3 连接关闭回调（官方文档：认证失败会主动断开，其他关闭场景触发重连）
  iTickWs.on("close", (code, reason) => {
    console.log(
      `iTick WebSocket连接关闭（代码：${code}，原因：${reason}），将在${
        ITICK_CONFIG.reconnectDelay / 1000
      }秒后重连`
    );
    if (ITICK_CONFIG.maxReconnectTimes > 0) {
      ITICK_CONFIG.maxReconnectTimes--;
      setTimeout(initITickConnection, ITICK_CONFIG.reconnectDelay);
    } else {
      console.error(
        "iTick WebSocket重连次数耗尽，请检查网络或API Token"
      );
    }
  });

  // 1.4 连接错误回调
  iTickWs.on("error", (err) => {
    console.error(
      "iTick WebSocket连接错误：",
      err.message
    );
  });

  // 1.5 发送心跳（ac=ping，params=时间戳，每30秒一次）
  setInterval(() => {
    if (iTickWs && iTickWs.readyState === WebSocket.OPEN) {
      const pingMsg = {
        ac: "ping",
        params: Date.now().toString()
      };
      iTickWs.send(JSON.stringify(pingMsg));
      console.log(`发送iTick心跳包，时间戳：${pingMsg.params}`);
    }
  }, ITICK_CONFIG.pingInterval);
}

// 2. 推送所有客户端的订阅请求到iTick服务器（ac=subscribe，params=标的，types=类型）
function pushAllClientSubscriptions() {
  for (const [clientId, symbols] of clientSubscriptions.entries()) {
    if (symbols.length > 0) {
      // 获取该客户端的订阅类型
      const types =
        clientSubscribeTypes.get(clientId) || ITICK_CONFIG.subscribeTypes;
      const subscribeMsg = {
        ac: "subscribe", 
        params: symbols.join(","), // 标的格式：多标的用逗号分隔，如AAPL$US,TSLA$US
        types: types.join(","), // 订阅类型：多类型用逗号分隔，如tick,quote,depth
      };
      iTickWs.send(JSON.stringify(subscribeMsg));
      console.log(
        `向iTick发送订阅请求：标的=${subscribeMsg.params}，类型=${subscribeMsg.types}`
      );
    }
  }
}

// 3. 将iTick行情数据推送给对应订阅的前端客户端
function pushQuotesToClients(quote) {
  const targetSymbol = quote.symbol;
  // 遍历所有客户端，匹配订阅该标的的客户端并推送数据
  for (const [clientId, symbols] of clientSubscriptions.entries()) {
    if (symbols.includes(targetSymbol)) {
      const clientWs = clientMap.get(clientId);
      if (clientWs && clientWs.readyState === WebSocket.OPEN) {
        // 向前端推送格式化后的行情数据
        clientWs.send(
          JSON.stringify({
            type: "stock_quote",
            data: quote,
            timestamp: Date.now(),
          })
        );
      }
    }
  }
}

// 4. 启动前端客户端WebSocket服务（监听8080端口，供前端连接）
const wss = new WebSocket.Server({ port: 8080 });
console.log("前端客户端WebSocket服务已启动，监听端口：8080");

// 4.1 监听前端客户端连接
wss.on("connection", (ws, req) => {
  // 生成唯一客户端ID（用于区分不同客户端）
  const clientId = `client_${Math.random().toString(36).slice(2)}`;
  clientMap.set(clientId, ws);
  clientSubscriptions.set(clientId, []); // 初始化订阅关系（空数组）
  clientSubscribeTypes.set(clientId, []); // 初始化订阅类型（空数组）
  console.log(`前端客户端${clientId}连接成功，当前在线：${clientMap.size}个`);

  // 4.2 监听前端客户端消息（订阅/取消订阅指令，格式匹配iTick官方文档）
  ws.on("message", (message) => {
    try {
      const data = JSON.parse(message.toString());
      const { action, symbols, types } = data;
      // 订阅行情（前端发送指令格式参考iTick官方文档，与后端向iTick发送的格式一致）
      if (action === "subscribe") {
        // 校验标的格式（需为数组，标的编码符合官方规范：如AAPL$US）
        if (
          !symbols ||
          !Array.isArray(symbols) ||
          symbols.some((s) => !s.includes("$"))
        ) {
          ws.send(
            JSON.stringify({
              type: "error",
              msg: "订阅失败",
            })
          );
          return;
        }
        // 校验订阅类型
        if (
          !types ||
          !Array.isArray(types) ||
          types.some((t) => !ITICK_CONFIG.subscribeTypes.includes(t))
        ) {
          ws.send(
            JSON.stringify({
              type: "error",
              msg: `订阅失败：类型错误（需为数组，支持${ITICK_CONFIG.subscribeTypes.join(
                ","
              )}）`,
            })
          );
          return;
        }
        // 更新客户端订阅关系和订阅类型（去重，避免重复订阅）
        const currentSubs = clientSubscriptions.get(clientId);
        const newSubs = [...new Set([...currentSubs, ...symbols])];
        clientSubscriptions.set(clientId, newSubs);

        const currentTypes = clientSubscribeTypes.get(clientId);
        const newTypes = [...new Set([...currentTypes, ...types])];
        clientSubscribeTypes.set(clientId, newTypes);

        // 向iTick服务器发送订阅请求（ac=subscribe，params=标的，types=类型）
        if (iTickWs && iTickWs.readyState === WebSocket.OPEN) {
          const subscribeMsg = {
            ac: "subscribe",
            params: newSubs.join(","),
            types: newTypes.join(","),
          };
          iTickWs.send(JSON.stringify(subscribeMsg));
        }
        ws.send(
          JSON.stringify({
            type: "success",
            msg: `成功订阅：标的=${newSubs.join(",")}，类型=${newTypes.join(
              ","
            )}`,
          })
        );
      }
      // 取消订阅（前端发送指令）
      else if (action === "unsubscribe") {
        if (!symbols || !Array.isArray(symbols)) {
          ws.send(
            JSON.stringify({
              type: "error",
              msg: "取消订阅失败",
            })
          );
          return;
        }
        // 更新客户端订阅关系
        const currentSubs = clientSubscriptions.get(clientId);
        const newSubs = currentSubs.filter((sym) => !symbols.includes(sym));
        clientSubscriptions.set(clientId, newSubs);
        // 向iTick服务器发送取消订阅请求
        if (iTickWs && iTickWs.readyState === WebSocket.OPEN) {
          const unsubscribeMsg = {
            ac: "subscribe", 
            params: newSubs.join(","),
            types: clientSubscribeTypes.get(clientId).join(","),
          };
          iTickWs.send(JSON.stringify(unsubscribeMsg));
        }
        ws.send(
          JSON.stringify({
            type: "success",
            msg: `成功取消订阅：${symbols.join(",")}`,
          })
        );
      }
      // 查询订阅（查询当前客户端的订阅标的和类型）
      else if (action === "query_subscribe") {
        const currentSubs = clientSubscriptions.get(clientId);
        const currentTypes = clientSubscribeTypes.get(clientId) || [];
        ws.send(
          JSON.stringify({
            type: "subscribe_list",
            data: {
              symbols: currentSubs,
              types: currentTypes,
            },
            msg: "当前订阅标的查询成功",
          })
        );
      } else {
        ws.send(
          JSON.stringify({
            type: "error",
            msg: `无效指令：${action}`,
          })
        );
      }
    } catch (err) {
      ws.send(
        JSON.stringify({
          type: "error",
          msg: "消息格式错误，需为JSON格式",
        })
      );
    }
  });

  // 4.3 监听前端客户端断开连接（清理订阅关系，避免iTick无效订阅）
  ws.on("close", () => {
    const currentSubs = clientSubscriptions.get(clientId);
    // 客户端断开后，向iTick发送重新订阅（仅剩余标的），实现取消该客户端订阅的效果
    if (iTickWs && iTickWs.readyState === WebSocket.OPEN) {
      // 收集所有其他客户端的订阅标的，去重后重新订阅
      const allSubs = [];
      clientSubscriptions.forEach((subs, id) => {
        if (id !== clientId) allSubs.push(...subs);
      });
      const uniqueSubs = [...new Set(allSubs)];
      const commonTypes = ITICK_CONFIG.subscribeTypes;
      const unsubscribeMsg = {
        ac: "subscribe",
        params: uniqueSubs.join(","),
        types: commonTypes.join(","),
      };
      iTickWs.send(JSON.stringify(unsubscribeMsg));
    }
    clientMap.delete(clientId);
    clientSubscriptions.delete(clientId);
    clientSubscribeTypes.delete(clientId);
    console.log(
      `前端客户端${clientId}断开连接，当前在线：${clientMap.size}个（已清理订阅关系）`
    );
  });

  // 4.4 前端客户端错误处理
  ws.on("error", (err) => {
    console.error(`前端客户端${clientId}连接错误：`, err.message);
  });
});

// 初始化iTick WebSocket连接（程序启动时执行）
initITickConnection();