ctx.poll

取出本线程订阅的下一条推送事件，可设置等待超时。事件来自本线程通过ctx.subscribe创建的订阅，策略可根据事件的channel（或kind）字段分别处理。

    
    
    
ctx.poll()

    
    
    
ctx.poll(streams)

    
    
    
ctx.poll(streams, timeout)

    
    
    
ctx.poll(options, timeout)

示例

• 订阅1分钟K线，仅在K线收盘时进行处理，并记录网关状态的变化。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function onKline(ev) {
      if (ev.prefill) {
          return                        // 首次订阅时推送的历史K线
      }
      if (ev.closed) {
          Log("closed bar", _D(ev.openTimeMs), "close:", ev.close, "volume:", ev.volume)
      }
  }
  
  function main() {
      var k1m = ctx.subscribe(exchange, {channel: "kline", interval: 60})
      while (true) {
          var ev = ctx.poll([], 5000)
          if (!ev) {
              Log("no event in 5 seconds")
              continue
          }
          if (ev.stream === k1m.id && ev.kind === 5) {
              onKline(ev)
          } else if (ev.stream === k1m.id && ev.kind === 35) {
              if (ev.prefillStatus === 2) {
                  throw new Error("kline subscription failed: " + ev.message)
              }
              Log("kline history ready")
          } else if (ev.kind === 33) {
              Log("gateway status:", ev.status)
          }
      }
  }
  

  
      python
      
      
  def on_kline(ev):
      if ev.prefill:
          return                        # 首次订阅时推送的历史K线
      if ev.closed:
          Log("closed bar", _D(ev.openTimeMs), "close:", ev.close, "volume:", ev.volume)
  
  def main():
      k1m = ctx.subscribe(exchange, {"channel": "kline", "interval": 60})
      while True:
          ev = ctx.poll([], 5000)
          if ev is None:
              Log("no event in 5 seconds")
              continue
          if ev.stream == k1m.id and ev.kind == 5:
              on_kline(ev)
          elif ev.stream == k1m.id and ev.kind == 35:
              if ev.prefillStatus == 2:
                  raise Exception("kline subscription failed: " + ev.message)
              Log("kline history ready")
          elif ev.kind == 33:
              Log("gateway status:", ev.status)
  

  
      rust
      
      
  use sdk::ctx::EventExt;
  
  fn main() {
      let k1m = ctx::subscribe(exchange, ctx::SubOpts::kline(60)).unwrap();
      loop {
          match ctx::poll(&[], Some(5000)) {
              ctx::Polled::Event(ev) if ev.stream() == k1m.id() => {
                  if let Some(k) = ev.kline() {
                      if !k.prefill && k.closed {        // 跳过首次订阅时推送的历史K线
                          Log!("closed bar", _D(k.open_time), "close:", k.close, "volume:", k.volume);
                      }
                  } else if let Some(st) = ev.feed_status() {
                      if st.failed() {
                          panic!("kline subscription failed: {}", st.message);
                      }
                      Log!("kline history ready");
                  }
              }
              ctx::Polled::Event(ev) => {
                  if let Some(g) = ev.gateway() {
                      Log!("gateway status:", g.status);
                  }
              }
              ctx::Polled::Empty => Log!("no event in 5 seconds"),
              ctx::Polled::Stopped => break,
              _ => {}
          }
      }
  }
  

• 订阅账户私有数据流：待账户快照就绪后读取持仓和余额，此后根据推送进行增量更新。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function main() {
      var acct = ctx.subscribe(exchange, {channel: "private"})
      while (true) {
          var ev = ctx.poll([], 1000)
          if (!ev || ev.stream !== acct.id) {
              continue
          }
          if (ev.kind === 35) {
              if (ev.prefillStatus === 1) {
                  Log("balances:", acct.balances(), "positions:", acct.positions())
              } else {
                  Log("account snapshot failed:", ev.message)
              }
          } else if (ev.kind === 17 && !ev.snapshot) {
              Log("position", ev.symbol, ev.posSide, "qty:", ev.qty, "avgPx:", ev.avgPx)
          } else if (ev.kind === 18 && !ev.snapshot) {
              Log("balance", ev.asset, "free:", ev.free, "frozen:", ev.frozen)
          } else if (ev.kind === 16) {
              Log("order", ev.id, ev.state, "filled:", ev.filledQty, "/", ev.qty)
          }
      }
  }
  

  
      python
      
      
  def main():
      acct = ctx.subscribe(exchange, {"channel": "private"})
      while True:
          ev = ctx.poll([], 1000)
          if ev is None or ev.stream != acct.id:
              continue
          if ev.kind == 35:
              if ev.prefillStatus == 1:
                  Log("balances:", acct.balances(), "positions:", acct.positions())
              else:
                  Log("account snapshot failed:", ev.message)
          elif ev.kind == 17 and not ev.snapshot:
              Log("position", ev.symbol, ev.posSide, "qty:", ev.qty, "avgPx:", ev.avgPx)
          elif ev.kind == 18 and not ev.snapshot:
              Log("balance", ev.asset, "free:", ev.free, "frozen:", ev.frozen)
          elif ev.kind == 16:
              Log("order", ev.id, ev.state, "filled:", ev.filledQty, "/", ev.qty)
  

  
      rust
      
      
  use sdk::ctx::EventExt;
  
  fn main() {
      let acct = ctx::subscribe(exchange, ctx::SubOpts::private()).unwrap();
      loop {
          match ctx::poll(&[], Some(1000)) {
              ctx::Polled::Event(ev) if ev.stream() == acct.id() => {
                  if let Some(st) = ev.feed_status() {
                      if st.ready() {
                          Log!("balances:", format!("{:?}", acct.balances()),
                               "positions:", format!("{:?}", acct.positions()));
                      } else {
                          Log!("account snapshot failed:", st.message);
                      }
                  } else if let Some(p) = ev.position().filter(|p| !p.snapshot) {
                      Log!("position", p.symbol, p.pos_side, "qty:", p.qty, "avgPx:", p.avg_px);
                  } else if let Some(b) = ev.balance().filter(|b| !b.snapshot) {
                      Log!("balance", b.asset, "free:", b.free, "frozen:", b.frozen);
                  } else if let Some(o) = ev.order() {
                      Log!("order", o.id, o.state, "filled:", o.filled_qty, "/", o.qty);
                  }
              }
              ctx::Polled::Stopped => break,
              _ => {}
          }
      }
  }
  

• 订阅全市场ticker，每分钟输出一次收到的事件数以及ctx.connectors()中的接收缓冲计数，用于判断策略的处理速度是否跟得上行情推送。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function main() {
      var all = ctx.subscribe(exchange, {channel: "tickers"})
      var n = 0
      var last = Date.now()
      while (true) {
          var ev = ctx.poll([], 1000)
          if (ev && ev.kind === 1) {
              n++
          } else if (ev && ev.kind === 35 && ev.prefillStatus === 2) {
              throw new Error("tickers subscription failed: " + ev.message)
          }
          if (Date.now() - last >= 60000) {
              var feed = ctx.connectors()[exchange.GetIndex()].feed
              Log("tickers/min:", n, "dropped:", feed.ticksDropped,
                  "backpressure:", feed.backpressure, "ring:", feed.ringCapacity)
              n = 0
              last = Date.now()
          }
      }
  }
  

  
      python
      
      
  import time
  
  def main():
      all_tickers = ctx.subscribe(exchange, {"channel": "tickers"})
      n = 0
      last = time.time()
      while True:
          ev = ctx.poll([], 1000)
          if ev is not None and ev.kind == 1:
              n += 1
          elif ev is not None and ev.kind == 35 and ev.prefillStatus == 2:
              raise Exception("tickers subscription failed: " + ev.message)
          if time.time() - last >= 60:
              feed = ctx.connectors()[all_tickers.ex].feed
              Log("tickers/min:", n, "dropped:", feed.ticksDropped,
                  "backpressure:", feed.backpressure, "ring:", feed.ringCapacity)
              n = 0
              last = time.time()
  

  
      rust
      
      
  use sdk::ctx::EventExt;
  
  fn main() {
      let all = ctx::subscribe(exchange, ctx::SubOpts::tickers()).unwrap();
      let mut n = 0u64;
      let mut last = Unix();
      loop {
          match ctx::poll(&[all.id()], Some(1000)) {
              ctx::Polled::Event(ev) => {
                  if ev.kind == EV_MD_TICK {
                      n += 1;
                  } else if let Some(st) = ev.feed_status().filter(|st| st.failed()) {
                      panic!("tickers subscription failed: {}", st.message);
                  }
              }
              ctx::Polled::Stopped => break,
              _ => {}
          }
          if Unix() - last >= 60 {
              // 返回JSON数组，每个交易所对应一项，其中feed字段为接收缓冲计数
              Log!("tickers/min:", n, "connectors:", ctx::connectors());
              n = 0;
              last = Unix();
          }
      }
  }
  

返回值

参数

参考

备注

所有事件都有以下字段：

各类事件的其他字段：

价格和数量均为实数，已按合约精度换算。

time在各类事件中的取值方式（回测中，time与ts均为回测的模拟时间）：

ev.ts - ev.time可用于估算交易所到托管者的延迟，仅在time为交易所时间时有意义。

成交事件的tradeId始终为字符串，与exchange.GetTrades返回的成交ID格式一致：交易所的数字成交号转为十进制字符串，其他格式原样保留；交易所不提供成交号时，为以syn:开头的合成编号；空字符串表示连接器未提供成交号。

深度事件带有ev.book(n)方法，用于读取该事件所属订阅自身订单簿的前n档（不传n时为订阅的档数），返回{bids: [[price, qty], ...], asks: [[price, qty], ...]}，其中bids按价格从高到低排列，asks按价格从低到高排列，与订阅对象的book方法相同。ev.book不可枚举，JSON.stringify(ev)的结果中不包含它；所属订阅已退订时调用会抛出异常。

持仓（kind为17）和余额（kind为18）事件来自private订阅，每条事件都是该持仓或该资产当前的完整状态，而非增量：持仓qty为0表示已平仓（单向持仓模式下平仓后side为空字符串），余额free与frozen均为0表示该资产已清空。字段与exchange.GetAssets、exchange.GetPositions口径一致；profit、margin、liqPx、markPx、leverage、equity、upnl在交易所未推送时不会出现（而不是为0）。snapshot为true表示订阅生效或断线重连后推送的账户快照，快照之后，同一事件流上会有一条channel为private的初始化状态事件。

初始化状态事件（kind为35）与其所属订阅共用同一个事件流（stream相同），channel为所属订阅的频道（私有流快照的状态为private），用于通知以下情况：K线订阅已推送完首批历史K线、private订阅已推送完账户快照（prefillStatus为1），以及订阅被交易所或连接器拒绝（prefillStatus为2，message为失败原因，频道可以是任意频道）。在收到就绪状态之前，不要把已收到的少量实时K线当作完整的历史数据。

ctx.poll()只获取本线程订阅的事件：主线程和threading.Thread()创建的线程（Python中为标准库的threading.Thread）各自订阅、各自获取，过滤条件中填写其他线程的订阅对象或事件流ID将永远取不到事件。不同线程订阅了相同内容时，各自会收到一份。网关状态事件会发送给在该交易所有订阅的每个线程，定时器事件只发送给主线程。

使用streams过滤仅表示本次只获取这几路事件：未指定的事件仍留在本线程的队列中，最终需要一次不带过滤条件（[]）的调用将它们取走。订单回报、持仓、余额、网关状态、初始化状态和定时器事件不会因获取不及时而被覆盖，无人获取时会持续积压，每个线程最多积压10240条，超出时丢弃最旧的事件。

等待期间仍会执行Promise、setTimeout回调，以及WebSocket、TCP连接的读写；timeout为0时也会处理一轮这些任务。

在JavaScript和Rust中，同一线程内的ctx.poll()与EventLoop从同一个事件队列获取事件，一个线程只应使用其中一种方式；JavaScript线程可以使用options形式同时等待推送事件、线程消息等其他来源。Python的EventLoop()拥有独立的队列，与ctx.poll()互不争抢。

消费速度诊断：托管者为每个交易所接收推送的缓冲区会根据积压情况从1024条自动扩容到32768条，达到上限后会丢弃行情事件，并让连接器暂缓推送其他事件。ctx.connectors()返回已加载的交易所列表（顺序对应ev.ex），每一项包含ex、label、name、symbol、interval、orderIdScope和feed字段，feed是进程启动以来的累计计数{ticksDropped, backpressure, ringCapacity}，分别为：缓冲区满时丢弃的行情条数、让连接器暂缓推送的次数、缓冲区当前的容量（条）。这些数值增长较快说明策略处理事件过慢，应先检查每条事件的处理耗时，并让全市场tickers这类高频订阅保持默认的仅保留最新模式。

Python中的签名为ctx.poll(streams=None, timeout_ms=None)：streams为None或[]时表示本线程订阅的全部事件流，否则可以是订阅对象、事件流ID、SubscriptionList或它们的（嵌套）列表；timeout_ms为None时一直等待，为0时不等待，为正整数时表示最长等待的毫秒数，超时返回None。事件是字典（dict的子类），字段与JavaScript的事件相同，ev["kind"]与ev.kind两种写法均可；深度事件提供ev.book(n)方法。实盘停止后（包括onexit()执行期间）调用会抛出StopError，正在等待中的调用也会立即抛出。不支持options形式（传入字典会抛出TypeError）：线程消息和exchange.Go()的完成通知需分别通过EventLoop()和各自的对象获取。ctx.connectors()与JavaScript中相同，也可以使用ctx.STREAM_STATUS、ctx.STREAM_TIMER。

Rust中的签名为ctx::poll(streams: &[u32], timeout_ms: Option<u64>) -> ctx::Polled：streams传入sub.id()或内置的ctx::STREAM_TIMER、ctx::STREAM_STATUS，空切片表示本线程订阅的全部事件流，None表示一直等待；返回值为Polled::Event(ev)、Polled::Empty（超时）、Polled::Stopped（实盘已停止）或Polled::Error(rc)。Rust的事件是原始结构体，引入sdk::ctx::EventExt后即可按事件类型读取，无需编写unsafe代码：ev.stream()、ev.channel()、ev.book(n)（深度事件），以及已换算为浮点数的ev.ticker()、ev.trade()、ev.kline()、ev.order()、ev.position()、ev.balance()、ev.gateway()（网关状态）、ev.feed_status()（初始化状态，failed()时message为失败原因）。事件类型不匹配时返回None，字段与JavaScript的事件一一对应；ev.time()、ev.ts()（f64，单位为毫秒）与JavaScript事件的time、ts相同，所有事件均可调用，各类型化结果中的time字段即为ev.time()。Rust的任意线程都可以订阅和poll，每个线程只获取本线程订阅的事件。ctx::connectors()返回内容相同的JSON字符串。