ctx.poll

Retrieves the next pushed event from the event streams subscribed to by the current thread, with an optional wait timeout. Events come from subscriptions that the current thread created with ctx.subscribe. The strategy can handle each event according to its channel (or kind) field.

    
    
    
ctx.poll()

    
    
    
ctx.poll(streams)

    
    
    
ctx.poll(streams, timeout)

    
    
    
ctx.poll(options, timeout)

Examples

• Subscribe to 1-minute K-lines, process each bar only when it closes, and log gateway status changes.

  • javascript
  • python
  • rust

  
      javascript
      
      
  function onKline(ev) {
      if (ev.prefill) {
          return                        // Skip the historical K-lines pushed on the initial subscription
      }
      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                        # Skip the historical K-lines pushed on the initial subscription
      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 {        // Skip the historical K-lines pushed on the initial subscription
                          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,
              _ => {}
          }
      }
  }
  

• Subscribe to the account's private stream: once the account snapshot is ready, read positions and balances; after that, update the data from pushes and log every fill.

  • 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 when the push carries no value on the same basis
          } 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 when the push carries no value on the same basis
          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 and frozen are Option types: None when the push carries no value on the same basis
                      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,
              _ => {}
          }
      }
  }
  

• Subscribes to tickers for the whole market. Once a minute it logs how many events arrived and the receive-buffer counters from ctx.connectors(). Use this to check whether the strategy processes events fast enough to keep up with the market data feed.

  • 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 {
              // Returns a JSON array with one entry per exchange; its feed field holds the receive-buffer counters
              Log!("tickers/min:", n, "connectors:", ctx::connectors());
              n = 0;
              last = Unix();
          }
      }
  }
  

Returns

Arguments

See Also

Remarks

All events contain the following fields:

Additional fields by event type:

Prices and quantities are all real numbers, already converted according to the contract precision.

How time is populated for each event type (in backtesting, both time and ts are the simulated backtest time):

ev.ts - ev.time can be used to estimate the latency from the exchange to the docker, and is only meaningful when time is the exchange time.

The tradeId of a trade event is always a string, in the same format as the trade ID returned by exchange.GetTrades: numeric trade IDs from the exchange are converted to decimal strings, and other formats are kept as is; when the exchange does not provide a trade ID, it is a synthetic ID starting with syn:; an empty string means the connector did not provide a trade ID.

Depth events have an ev.book(n) method that reads the top n levels of the order book belonging to the event's own subscription (when n is not passed, the subscribed depth is used), and returns {bids: [[price, qty], ...], asks: [[price, qty], ...]}, where bids are sorted by price from high to low and asks from low to high, the same as the book method of the subscription object. ev.book is non-enumerable, so it is not included in the result of JSON.stringify(ev); calling it after the owning subscription has been unsubscribed throws an exception.

Position (kind is 17) and balance (kind is 18) events come from the private subscription. Each event is the complete current state of that position or asset, not a delta: a position qty of 0 means the position is closed (in one-way position mode, side is an empty string after closing), and a balance with both free and frozen equal to 0 means the asset has been fully cleared; if the exchange's balance push does not contain consistently defined available or frozen data, the corresponding field is null (no REST call is made afterwards to fill it in). The fields are consistent with exchange.GetAssets and exchange.GetPositions; profit, margin, liqPx, markPx, leverage, equity and upnl are absent (rather than 0) when the exchange does not push them. snapshot being true indicates an account snapshot pushed when the subscription takes effect or after reconnecting; after the snapshot, a prefill status event with channel set to private is pushed on the same event stream.

Fills (kind is 19, channel is fills) come from the private subscription, are on the same event stream as order updates, and each represents one execution of your own order:

• The fill is pushed first, followed by the order update reflecting it: when a received order update shows an increase in filledQty, all fills making up that increase have already been received; when an order is fully filled, the sum of the qty of its fills equals the order's filledQty.
• orderId is the same as the id of the order update and the return value of ctx.order; tradeId is always a string, and is a synthetic ID starting with syn: when the exchange does not provide a trade ID. side is the order side, not the aggressor side.
• There is no snapshot: historical fills are not replayed when the subscription takes effect or after reconnecting, and fills that occurred during a disconnection are not replayed either (the cumulative filled quantity in order updates remains accurate); for historical fills, query the REST API. A fill that the exchange pushes again after reconnecting is delivered only once. When a buy order and a sell order of the same account match each other, the same tradeId is pushed once for each of the two orders.
• Fills are only available when the exchange's private push includes per-trade executions; see each exchange's page for details. Exchanges that do not support this provide only order updates.

Async order acks (kind is 20, both type and channel are ack) are the results of ctx.orderAsync and ctx.cancelAsync. They are on the built-in event stream ctx.STREAM_ORDER_ACK, are delivered only to the thread that issued the request, and require no subscription: ctx.poll([ctx.STREAM_ORDER_ACK], timeout) retrieves only acks, and ctx.poll([], timeout) retrieves them as well. A status of unknown means the request was sent over WebSocket but no response was received, and the order may already have been submitted to the exchange. Do not resend it; instead, query the order by clientId, or wait for an order update on the private stream to confirm.

Prefill status events (kind is 35) share the same event stream (same stream) as their owning subscription, and channel is the channel of the owning subscription (private for the status event of the private stream snapshot). They indicate that: a K-line subscription has finished pushing the initial batch of historical bars, or a private subscription has finished pushing the account snapshot (prefillStatus is 1); or that the subscription was rejected by the exchange or connector (prefillStatus is 2, message is the failure reason; this can occur on any channel). Until the ready status is received, do not treat the few real-time bars received so far as complete historical data.

ctx.poll() only retrieves events subscribed by the current thread: the main thread and each thread created by threading.Thread() (the standard library threading.Thread in Python) subscribe and retrieve independently. If the filter uses a subscription object or stream ID belonging to another thread, no events will ever be retrieved. When different threads subscribe to the same content, each receives its own copy. Gateway status events are sent to every thread that has subscriptions on that exchange, while timer events are sent only to the main thread.

Filtering with streams only means that this call retrieves events from those streams only: unspecified events remain in the current thread's queue, and a call without filter conditions ([]) is eventually required to drain them. Order report, execution report, position, balance, gateway status, initialization status, timer and async order acknowledgment events are not overwritten when they are not read in time; if nobody reads them, they keep accumulating, up to 10240 events per thread, after which the oldest events are discarded.

While waiting, Promise and setTimeout callbacks, as well as reads and writes on WebSocket and TCP connections, are still executed; when timeout is 0, one round of these tasks is also processed.

In JavaScript and Rust, ctx.poll() and EventLoop within the same thread retrieve events from the same event queue, so a thread should use only one of them; JavaScript threads can use the options form to wait simultaneously on push events, thread messages and other sources. In Python, EventLoop() has its own independent queue and does not compete with ctx.poll().

Consumption speed diagnostics: the push receive buffer the docker allocates for each exchange automatically expands from 1024 to 32768 entries as the backlog grows; once the limit is reached, market data events are discarded and the connector is told to throttle pushing other events. ctx.connectors() returns the list of loaded exchanges (the order corresponds to ev.ex), each item containing the fields ex, label, name, symbol, interval, orderIdScope and feed, where feed is the cumulative count since process start {ticksDropped, backpressure, ringCapacity}: respectively the number of market data entries dropped when the buffer was full, the number of times the connector was told to throttle pushing, and the current buffer capacity (entries). Rapid growth in these values indicates that the strategy is processing events too slowly, and the processing time of each event should be checked first; high-frequency subscriptions such as market-wide tickers should keep the default latest-only mode.

In Python it is ctx.poll(streams=None, timeout_ms=None): when streams is None or [], it means all event streams subscribed by the current thread; otherwise it is a subscription object, stream ID, SubscriptionList or a (nested) list of them. When timeout_ms is None, it waits indefinitely; when 0, it does not wait; when a positive integer, it is the maximum wait time in milliseconds, returning None on timeout. Events are dictionaries (subclasses of dict) with the same fields as JavaScript events, and both ev["kind"] and ev.kind notations work; depth events provide the ev.book(n) method. After live trading stops (including while onexit() is executing), calls raise StopError, and calls that are waiting also raise immediately. The options form is not supported (passing a dictionary raises TypeError): thread messages and exchange.Go() completion notifications are obtained via EventLoop() and the corresponding objects respectively. ctx.connectors() is the same as in JavaScript, and ctx.STREAM_STATUS, ctx.STREAM_TIMER and ctx.STREAM_ORDER_ACK are also available.

In Rust, the signature is ctx::poll(streams: &[u32], timeout_ms: Option<u64>) -> ctx::Polled. For streams, pass sub.id() or one of the built-in streams ctx::STREAM_TIMER, ctx::STREAM_STATUS or ctx::STREAM_ORDER_ACK. An empty slice means all event streams this thread subscribes to, and None means wait with no timeout. The return value is Polled::Event(ev), Polled::Empty (timeout), Polled::Stopped (live trading stopped) or Polled::Error(rc). In Rust, events are raw structs. Import sdk::ctx::EventExt to read them by event type without writing unsafe code: ev.stream(), ev.channel(), ev.book(n) (order book depth events), and the following accessors that convert values to floating-point numbers: ev.ticker(), ev.trade(), ev.kline(), ev.order(), ev.position(), ev.balance(), ev.fill() (fill report, FillView: order_id, trade_id, px, qty, fee (Option<f64>), fee_ccy, side, liquidity, time), ev.order_ack() (async order acknowledgement, OrderAckView: req_id, op, status, ok, via, id, client_id, error, reason, reason_text, exch_time (Option<i64>), latency_ms, time), ev.gateway() (gateway status) and ev.feed_status() (initialization status; when failed() is true, message holds the failure reason). If the event type does not match, these return None. Their fields match the fields of JavaScript events one to one. ev.time() and ev.ts() (f64 milliseconds) are the same as time and ts on JavaScript events and can be called on any event. The time field in each typed result is the same value as ev.time(). Any Rust thread can subscribe and call poll, and each thread receives only the events it subscribed to. ctx::connectors() returns the same content as a JSON string.