Binance Websocket回调重复触发 如何确保交易逻辑仅执行一次

问题场景 #

使用node-binance-api对接币安Websocket实时行情，结合自定义策略触发买卖操作时，Websocket会持续推送行情数据，同一时间可返回5-6条甚至更多响应。调用接口关闭Websocket存在约3秒延迟，残留的行情回调会导致买卖操作、profitCalculations利润计算逻辑重复触发3-4次，需要保证每轮交易仅执行1次买入或卖出操作。

现有实现代码 #

• 依赖库（运行正常）：node-binance-api

1. Websocket初始化逻辑

binance.websockets.bookTickers('BNBBUSD', (mainTicker) => {
  var priceNow = mainTicker.bestBid;
  liveStrategy(priceNow).then(isSold => {
      // console.log('THIS IS WHEN IS CHECKED ->>>>>>>');
      if (isSold == true) {
          console.log('JUST SOLD -> Setting the count to 10');
          sleep(3000).then(() => { 
              console.log("Countdown finished: ", count);
              profitCalculations(mainCurrency);
              count = 10;
          });
      }
      if (isSold == false) {
          console.log('THIS WAS A BUY ->');
          count = 10;
      }
  }).catch((isSoldErr) => {
      console.warn('Error:', isSoldErr.message);
      dumpError(isSoldErr);
  });
});

2. 现有Websocket关闭逻辑

let endpoints = binance.websockets.subscriptions();
for ( let endpoint in endpoints ) {
    console.log("..terminating websocket: "+endpoint);
    let ws = endpoints[endpoint];
    ws.terminate();
}

3. 现有延迟触发买卖逻辑

if (priceNow > buyPrice) {
terminateAllWebsockets();
return new Promise((resolve) => {
    sleep(5000).then(sleep => {
        marketBuyFromNoLossStrategyAmount(buySymbol,buyAmount);
        resolve(false);
    })
})

}

问题根因 #

• ws.terminate()为异步操作，调用终止接口前已经推送到本地事件队列的行情消息，会继续执行回调进入策略判断流程
• 无交易执行状态拦截，多个并发的策略命中判断会独立触发买卖、利润计算逻辑，导致重复执行

适配现有架构的修复方案 #

改动量最小，不需要重构原有逻辑，仅增加交易执行锁拦截重复回调即可：

1. 定义全局交易锁变量，初始状态为未锁定

// 交易执行锁：true=本轮交易已触发，拦截所有后续回调
let isTradeExecuting = false;

2. 修改Websocket回调入口，最前置位置加锁判断，锁生效时直接丢弃所有行情消息，不进入后续策略逻辑

binance.websockets.bookTickers('BNBBUSD', (mainTicker) => {
  // 锁状态下直接返回，不处理任何后续逻辑
  if (isTradeExecuting) return;
  const priceNow = mainTicker.bestBid;
  liveStrategy(priceNow).then(isSold => {
      if (isSold == true) {
          console.log('JUST SOLD -> Setting the count to 10');
          sleep(3000).then(() => { 
              console.log("Countdown finished: ", count);
              profitCalculations(mainCurrency);
              count = 10;
              // 本轮卖出流程全部走完后解锁，供下一轮交易使用
              isTradeExecuting = false;
          });
      }
      if (isSold == false) {
          console.log('THIS WAS A BUY ->');
          count = 10;
          // 本轮买入流程全部走完后解锁
          isTradeExecuting = false;
      }
  }).catch((isSoldErr) => {
      console.warn('Error:', isSoldErr.message);
      dumpError(isSoldErr);
      // 异常场景必须解锁，避免死锁导致下一轮交易无法触发
      isTradeExecuting = false;
  });
});

3. 修改策略命中分支，触发交易前第一时间上锁，再执行关连接、延迟交易逻辑

if (priceNow > buyPrice) {
  // 最先上锁，第一时间拦截所有后续进入的行情回调
  isTradeExecuting = true;
  terminateAllWebsockets();
  return new Promise((resolve) => {
      // 锁已经拦截重复逻辑，不需要等待5秒，可将sleep缩短到500ms仅做连接终止缓冲，降低滑点
      sleep(500).then(() => {
          marketBuyFromNoLossStrategyAmount(buySymbol,buyAmount);
          resolve(false);
      })
  })
}

额外注意事项 #

• 每轮交易结束、重新启动Websocket开启下一轮行情监听前，需要确认isTradeExecuting为false，避免锁状态异常导致行情无法被处理
• 锁判断必须放在回调逻辑的最外层，不要放在liveStrategy调用之后，否则残留队列的消息还是会进入策略执行流程

内容的提问来源于stack exchange，提问作者Marian Petrisor