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)   // 推送中无同口径数值时为null
          } else if (ev.kind === 19) {
              Log("fill", ev.orderId, ev.side, ev.px, ev.qty, ev.liquidity, "fee:", ev.fee, ev.feeCcy)
          } 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)   # 推送中无同口径数值时为None
          elif ev.kind == 19:
              Log("fill", ev.orderId, ev.side, ev.px, ev.qty, "fee:", ev.get("fee"), ev.feeCcy)
          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) {
                      // free、frozen为Option类型：推送中无同口径数值时为None
                      let (free, frozen) = (format!("{:?}", b.free), format!("{:?}", b.frozen));
                      Log!("balance", b.asset, "free:", free, "frozen:", frozen);
                  } else if let Some(f) = ev.fill() {
                      let fee = format!("{:?}", f.fee);
                      Log!("fill", f.order_id, f.side, f.px, f.qty, "fee:", fee, f.fee_ccy);
                  } 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表示该资产已清空；若交易所的余额推送中没有口径一致的可用或冻结数据，对应字段为null（推送后不会再调用REST接口补齐）。字段与exchange.GetAssets、exchange.GetPositions口径一致；profit、margin、liqPx、markPx、leverage、equity、upnl在交易所未推送时不出现（而不是为0）。snapshot为true表示订阅生效或断线重连后推送的账户快照，快照之后，同一事件流上会推送一条channel为private的初始化状态事件。

成交回报（kind为19，channel为fills）来自private订阅，与订单回报位于同一个事件流上，表示自己订单的一笔成交：

• 先推送成交，再推送反映该成交的订单回报：当收到的订单回报中filledQty增加时，构成该增量的成交均已收到；订单全部成交时，其各笔成交的qty之和等于订单的filledQty。
• orderId与订单回报的id、ctx.order的返回值相同；tradeId始终为字符串，交易所未提供成交号时为以syn:开头的合成编号。side为订单方向，而非主动成交方向。
• 没有快照：订阅生效和断线重连后不会补发历史成交，断线期间的成交也不会补发（订单回报中的累计成交量仍然准确），如需历史成交，请调用REST接口查询。交易所重连后重复推送的同一笔成交只会推送一次。同一账户的买单与卖单相互成交时，同一个tradeId会按两个订单各推送一条。
• 仅当交易所的私有推送包含逐笔成交时才有成交回报，详见各交易所页面的说明；不支持的交易所只有订单回报。

异步下单应答（kind为20，type与channel均为ack）是ctx.orderAsync、ctx.cancelAsync的执行结果，位于内置事件流ctx.STREAM_ORDER_ACK上，仅投递给发起请求的线程，无需订阅：ctx.poll([ctx.STREAM_ORDER_ACK], timeout)只获取应答，ctx.poll([], timeout)也能获取到。status为unknown表示请求已通过WebSocket发出但未等到应答，订单可能已提交到交易所，请勿重复发送，应按clientId查询订单，或等待私有流的订单回报进行确认。

初始化状态事件（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、ctx.STREAM_ORDER_ACK。

Rust中为ctx::poll(streams: &[u32], timeout_ms: Option<u64>) -> ctx::Polled：streams传入sub.id()或内置的ctx::STREAM_TIMER、ctx::STREAM_STATUS、ctx::STREAM_ORDER_ACK，空切片表示本线程订阅的全部事件流，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.fill()（成交回报，FillView：order_id、trade_id、px、qty、fee（Option<f64>）、fee_ccy、side、liquidity、time）、ev.order_ack()（异步下单应答，OrderAckView：req_id、op、status、ok、via、id、client_id、error、reason、reason_text、exch_time（Option<i64>）、latency_ms、time）、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字符串。