ctx.subscribe

订阅WebSocket推送，可订阅品种行情、全市场ticker或账户私有数据流，返回Subscription对象。可订阅的内容包括：单个品种的ticker、最优买卖价、订单簿、逐笔成交、K线，全市场的ticker，以及整个账户的订单回报、成交回报、持仓和余额。symbol传入数组时，一次订阅多个品种，并返回订阅对象数组。事件通过ctx.poll获取，再按事件的channel分别处理。订阅归属于调用它的线程，主线程和threading.Thread()创建的线程均可订阅。

    
    
    
ctx.subscribe(ex, options)

示例

• 使用交易所对象订阅当前交易对的ticker和20档订单簿，同时订阅账户私有流，并按channel分别处理各类事件。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function main() {
      // 第一个参数传入交易所对象；行情频道未指定symbol时，使用该交易所对象当前的交易对
      var tk = ctx.subscribe(exchange, {channel: "ticker"})
      var book = ctx.subscribe(exchange, {channel: "depth", depth: 20})
      // 私有流不指定symbol，覆盖整个账户：订单回报、成交回报、持仓和余额均通过该流推送
      var acct = ctx.subscribe(exchange, {channel: "private"})
      Log("subscribed:", JSON.stringify(tk), JSON.stringify(book), JSON.stringify(acct))
      while (true) {
          var ev = ctx.poll([], 1000)    // []表示本线程订阅的全部事件流，最多等待1秒
          if (!ev) {
              continue
          }
          if (ev.type === "status") {    // 网关状态、初始化状态事件需优先判断，它们的channel可能与数据事件相同
              if (ev.kind === 35 && ev.channel === "private") {
                  Log("account snapshot ready:", ev.prefillStatus === 1, ev.message)
              }
              continue
          }
          if (ev.channel === "ticker") {
              Log("ticker", ev.symbol, "bid:", ev.bid, "ask:", ev.ask, "last:", ev.last)
          } else if (ev.channel === "depth") {
              var b = book.book(1)       // 与ev.book(1)读取的是同一本订单簿
              if (b.bids.length > 0 && b.asks.length > 0) {
                  Log("bid1:", b.bids[0][0], "ask1:", b.asks[0][0])
              }
          } else if (ev.channel === "fills") {
              Log("fill", ev.orderId, ev.side, ev.px, ev.qty, "fee:", ev.fee, ev.feeCcy, ev.liquidity)
          } else if (ev.channel === "orders") {
              Log("order", ev.id, ev.state, "filled:", ev.filledQty, "/", ev.qty)
          }
      }
  }
  

  
      python
      
      
  def main():
      # 第一个参数传入交易所对象；行情频道未指定symbol时，使用该交易所对象当前的交易对
      tk = ctx.subscribe(exchange, {"channel": "ticker"})
      book = ctx.subscribe(exchange, {"channel": "depth", "depth": 20})
      # 私有流不指定symbol，覆盖整个账户：订单回报、成交回报、持仓和余额均通过该流推送
      acct = ctx.subscribe(exchange, {"channel": "private"})
      Log("subscribed:", tk.info(), book.info(), acct.info())
      while True:
          ev = ctx.poll([], 1000)        # []表示本线程订阅的全部事件流，最多等待1秒
          if ev is None:
              continue
          if ev.type == "status":        # 网关状态、初始化状态事件需优先判断
              if ev.kind == 35 and ev.channel == "private":
                  Log("account snapshot ready:", ev.prefillStatus == 1, ev.message)
              continue
          if ev.channel == "ticker":
              Log("ticker", ev.symbol, "bid:", ev.bid, "ask:", ev.ask, "last:", ev.last)
          elif ev.channel == "depth":
              b = book.book(1)           # 与ev.book(1)读取的是同一本订单簿
              if b["bids"] and b["asks"]:
                  Log("bid1:", b["bids"][0][0], "ask1:", b["asks"][0][0])
          elif ev.channel == "fills":
              # 交易所未提供手续费时，事件中不含fee字段
              Log("fill", ev.orderId, ev.side, ev.px, ev.qty, "fee:", ev.get("fee"), ev.feeCcy)
          elif ev.channel == "orders":
              Log("order", ev.id, ev.state, "filled:", ev.filledQty, "/", ev.qty)
  

  
      rust
      
      
  use sdk::ctx::EventExt;
  
  fn main() {
      // 交易所对象作为第一个参数：行情频道未指定symbol时，使用其当前的交易对
      let book = ctx::subscribe(exchange, ctx::SubOpts::depth(20)).unwrap();
      let acct = ctx::subscribe(exchange, ctx::SubOpts::private()).unwrap();
      loop {
          match ctx::poll(&[], Some(1000)) {
              ctx::Polled::Event(ev) => match ev.channel() {
                  "depth" => {
                      if let Ok(b) = book.book(1) {
                          if !b.bids.is_empty() && !b.asks.is_empty() {
                              Log!("bid1:", b.bids[0].0, "ask1:", b.asks[0].0);
                          }
                      }
                  }
                  "fills" => {
                      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);
                      }
                  }
                  "positions" | "balance" => {
                      Log!("positions:", format!("{:?}", acct.positions()));
                  }
                  channel => Log!("event:", channel, "kind:", ev.kind),
              },
              ctx::Polled::Stopped => break,
              _ => {}
          }
      }
  }
  

• 批量订阅三个品种的逐笔成交：调用返回订阅对象数组，可将整个数组直接传给ctx.poll()；统计1分钟内各品种的成交笔数后，一次性全部退订。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function main() {
      // symbol传入数组时，返回与其顺序一致的订阅对象数组；只要有一个品种不合法，整批均不订阅
      var symbols = ["BTC_USDT", "ETH_USDT", "SOL_USDT"]
      var subs = ctx.subscribe(exchange, {channel: "trade", symbol: symbols})
      subs.forEach(function(s) {
          Log(s.symbol, "stream:", s.id)
      })
      var count = {}
      var t0 = Date.now()
      while (Date.now() - t0 < 60000) {
          var ev = ctx.poll(subs, 1000)    // 订阅对象数组可直接传给ctx.poll
          if (!ev) {
              continue
          }
          if (ev.kind === 35 && ev.prefillStatus === 2) {
              Log("subscription failed:", ev.symbol, ev.message)   // 仅该品种订阅失败，其余品种不受影响
          } else if (ev.kind === 4) {
              count[ev.symbol] = (count[ev.symbol] || 0) + 1
          }
      }
      Log("trades in 60 seconds:", count)
      subs.unsubscribe()                   // 一次性退订数组中的全部订阅
  }
  

  
      python
      
      
  import time
  
  def main():
      # symbol传入列表时，返回顺序一致的SubscriptionList；只要有一个品种不合法，整批均不订阅
      symbols = ["BTC_USDT", "ETH_USDT", "SOL_USDT"]
      subs = ctx.subscribe(exchange, {"channel": "trade", "symbol": symbols})
      for s in subs:
          Log(s.symbol, "stream:", s.id)
      count = {}
      t0 = time.time()
      while time.time() - t0 < 60:
          ev = ctx.poll(subs, 1000)          # SubscriptionList可直接传给ctx.poll
          if ev is None:
              continue
          if ev.kind == 35 and ev.prefillStatus == 2:
              Log("subscription failed:", ev.symbol, ev.message)
          elif ev.kind == 4:
              count[ev.symbol] = count.get(ev.symbol, 0) + 1
      Log("trades in 60 seconds:", count)
      subs.unsubscribe()                     # 一次性退订列表中的全部订阅
  

  
      rust
      
      
  use sdk::ctx::EventExt;
  use std::collections::HashMap;
  
  fn main() {
      let symbols = ["BTC_USDT", "ETH_USDT", "SOL_USDT"];
      // 返回与symbols顺序一致的句柄；只要有一个品种不合法，整批返回Err
      let subs = ctx::subscribe_many(exchange, ctx::SubOpts::trade(), &symbols).unwrap();
      let ids: Vec<u32> = subs.iter().map(|s| s.id()).collect();
      let mut count: HashMap<u32, u64> = HashMap::new();
      let t0 = Unix();
      while Unix() - t0 < 60 {
          match ctx::poll(&ids, Some(1000)) {
              ctx::Polled::Event(ev) if ev.kind == EV_MD_TRADE => {
                  *count.entry(ev.stream()).or_insert(0) += 1;
              }
              ctx::Polled::Stopped => return,
              _ => {}
          }
      }
      for (s, symbol) in subs.iter().zip(symbols) {
          Log!(symbol, "trades in 60 seconds:", count.get(&s.id()).copied().unwrap_or(0));
          s.unsubscribe().unwrap();
      }
  }
  

• 在线程中订阅：子线程自行订阅两个品种的订单簿，相关事件只投递给该线程；主线程订阅自己的ticker，二者互不干扰。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function main() {
      var t = threading.Thread(function(symbols, ms) {
          // 在线程中订阅，这两路订阅的事件只投递给本线程
          var subs = ctx.subscribe(0, {channel: "depth", symbol: symbols, depth: 5})
          var n = 0
          var t0 = Date.now()
          while (Date.now() - t0 < ms) {
              var ev = ctx.poll([], 1000)          // 只获取本线程订阅的事件
              if (ev && ev.kind === 3) {
                  var b = ev.book(1)
                  if (b.bids.length > 0 && b.asks.length > 0) {
                      Log("thread", ev.symbol, "bid1:", b.bids[0][0], "ask1:", b.asks[0][0])
                  }
                  n++
              }
          }
          subs.unsubscribe()                       // 只能在创建该订阅的线程中退订
          return n
      }, ["ETH_USDT" , "SOL_USDT"], 30000)
  
      var tk = ctx.subscribe(0, {channel: "ticker" , symbol: "BTC_USDT"})
      var t0 = Date.now()
      while (Date.now() - t0 < 30000) {
          var ev = ctx.poll([], 1000)              // 获取不到子线程订阅的深度事件
          if (ev && ev.kind === 1) {
              Log("main", ev.symbol, "last:", ev.last)
          }
      }
      tk.unsubscribe()
      Log("depth updates in the thread:", t.join().ret)
  }
  

  
      python
      
      
  import threading
  import time
  
  def depth_worker(symbols, seconds):
      # 在线程中订阅，这两路订阅的事件只投递给本线程
      subs = ctx.subscribe(0, {"channel": "depth" , "symbol": symbols, "depth": 5})
      t0 = time.time()
      while time.time() - t0 < seconds:
          ev = ctx.poll([], 1000)              # 只获取本线程订阅的事件
          if ev is not None and ev.kind == 3:
              b = ev.book(1)
              if b["bids"] and b["asks"]:
                  Log("thread" , ev.symbol, "bid1:", b["bids"][0][0], "ask1:", b["asks"][0][0])
      subs.unsubscribe()                       # 线程结束时也会自动退订
  
  def main():
      t = threading.Thread(target=depth_worker, args=(["ETH_USDT", "SOL_USDT" ], 30))
      t.start()
      tk = ctx.subscribe(0, {"channel": "ticker", "symbol": "BTC_USDT"})
      t0 = time.time()
      while time.time() - t0 < 30:
          ev = ctx.poll([], 1000)              # 获取不到子线程订阅的深度事件
          if ev is not None and ev.kind == 1:
              Log("main", ev.symbol, "last:", ev.last)
      tk.unsubscribe()
      t.join()
  

  
      rust
      
      
  fn main() {
      let worker = std::thread::spawn(|| {
          // 在线程中订阅，这两路订阅的事件只投递给本线程
          let symbols = ["ETH_USDT", "SOL_USDT"];
          let subs = ctx::subscribe_many(0u16, ctx::SubOpts::depth(5), &symbols).unwrap();
          let mut n = 0u64;
          let t0 = Unix();
          while Unix() - t0 < 30 {
              match ctx::poll(&[], Some(1000)) {           // 只获取本线程订阅的事件
                  ctx::Polled::Event(ev) if ev.kind == EV_MD_DEPTH5 => n += 1,
                  ctx::Polled::Stopped => break,
                  _ => {}
              }
          }
          for s in subs {
              let _ = s.unsubscribe();                     // 只能在创建该订阅的线程中退订
          }
          n
      });
      let tk = ctx::subscribe(0u16, ctx::SubOpts::ticker().symbol("BTC_USDT")).unwrap();
      let t0 = Unix();
      while Unix() - t0 < 30 {
          match ctx::poll(&[tk.id()], Some(1000)) {
              ctx::Polled::Event(ev) => Log!("main ticker event, seq:", ev.seq),
              ctx::Polled::Stopped => return,
              _ => {}
          }
      }
      let _ = tk.unsubscribe();
      Log!("depth updates in the thread:", worker.join().unwrap_or(0));
  }
  

• 订阅全市场ticker：通过一路订阅接收交易所全部品种的ticker，10秒后统计收到的品种数量；若交易所不支持全市场推送，该路订阅会收到一个表示失败的状态事件。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function main() {
      // 不指定symbol：一路订阅即可接收该交易所全部品种的ticker
      var all = ctx.subscribe(exchange, {channel: "tickers"})
      var last = {}
      var t0 = Date.now()
      while (Date.now() - t0 < 10000) {
          var ev = ctx.poll(all, 1000)
          if (!ev) {
              continue
          }
          if (ev.kind === 35 && ev.prefillStatus === 2) {
              Log("tickers not supported:", ev.message)    // 交易所不支持全市场推送
              return
          }
          if (ev.kind === 1) {
              last[ev.symbol] = ev.last                    // symbol为该条ticker所属的品种
          }
      }
      Log("symbols:", Object.keys(last).length, "BTC_USDT last:", last["BTC_USDT"])
      all.unsubscribe()
  }
  

  
      python
      
      
  import time
  
  def main():
      # 不指定symbol：一路订阅即可接收该交易所全部品种的ticker
      all_tickers = ctx.subscribe(exchange, {"channel": "tickers"})
      last = {}
      t0 = time.time()
      while time.time() - t0 < 10:
          ev = ctx.poll(all_tickers, 1000)
          if ev is None:
              continue
          if ev.kind == 35 and ev.prefillStatus == 2:
              Log("tickers not supported:", ev.message)    # 交易所不支持全市场推送
              return
          if ev.kind == 1:
              last[ev.symbol] = ev.last                    # symbol为该条ticker所属的品种
      Log("symbols:", len(last), "BTC_USDT last:", last.get("BTC_USDT"))
      all_tickers.unsubscribe()
  

  
      rust
      
      
  use sdk::ctx::EventExt;
  use std::collections::HashMap;
  
  fn main() {
      // 不指定symbol：一路订阅即可接收该交易所全部品种的ticker
      let all = ctx::subscribe(exchange, ctx::SubOpts::tickers()).unwrap();
      let mut last: HashMap<String, f64> = HashMap::new();
      let t0 = Unix();
      while Unix() - t0 < 10 {
          match ctx::poll(&[all.id()], Some(1000)) {
              ctx::Polled::Event(ev) => {
                  if let Some(st) = ev.feed_status() {
                      if st.failed() {
                          Log!("tickers not supported:", st.message);   // 交易所不支持全市场推送
                          return;
                      }
                  } else if let Some(t) = ev.ticker() {
                      last.insert(t.symbol, t.last);                     // symbol为该条ticker所属的品种
                  }
              }
              ctx::Polled::Stopped => return,
              _ => {}
          }
      }
      Log!("symbols:", last.len(), "BTC_USDT last:", format!("{:?}", last.get("BTC_USDT")));
      let _ = all.unsubscribe();
  }
  

• 相同的订阅共用同一个事件流：两个订阅对象的id相同，其中一个退订后，另一个仍能继续收到事件。

  • javascript
  • python

  
      javascript
      
      
  function main() {
      var a = ctx.subscribe(0, {channel: "trade", symbol: "BTC_USDT"})
      var b = ctx.subscribe(0, {channel: "trade", symbol: "BTC_USDT"})
      var c = ctx.subscribe(0, {channel: "ticker", symbol: "BTC_USDT"})
      Log("a.id:", a.id, "b.id:", b.id, "c.id:", c.id)   // a.id与b.id相同，c.id不同
  
      a.unsubscribe()    // b仍引用这路订阅，成交推送不会中断
      var n = 0
      while (n < 10) {
          var ev = ctx.poll(b, 5000)
          if (!ev) {
              Log("no trade in 5 seconds")
              continue
          }
          if (ev.kind === 4) {
              Log("trade", ev.tradeId, ev.side, ev.px, ev.qty)
              n++
          }
      }
      b.unsubscribe()    // 释放最后一个引用，此时才真正向交易所退订
      c.unsubscribe()
  }
  

  
      python
      
      
  def main():
      a = ctx.subscribe(0, {"channel": "trade", "symbol": "BTC_USDT"})
      b = ctx.subscribe(0, {"channel": "trade", "symbol": "BTC_USDT"})
      c = ctx.subscribe(0, {"channel": "ticker", "symbol": "BTC_USDT"})
      Log("a.id:", a.id, "b.id:", b.id, "c.id:", c.id)   # a.id与b.id相同，c.id不同
  
      a.unsubscribe()    # b仍引用这路订阅，成交推送不会中断
      n = 0
      while n < 10:
          ev = ctx.poll(b, 5000)
          if ev is None:
              Log("no trade in 5 seconds")
              continue
          if ev.kind == 4:
              Log("trade", ev.tradeId, ev.side, ev.px, ev.qty)
              n += 1
      b.unsubscribe()    # 释放最后一个引用，此时才真正向交易所退订
      c.unsubscribe()
  

• 处理被交易所拒绝的订阅：订阅交易所不支持的K线周期时不会抛出异常，失败原因会通过该订阅自身的事件流上报。

  • javascript
  • python

  
      javascript
      
      
  function main() {
      // 多数交易所不支持7秒周期的K线
      var k = ctx.subscribe(exchange, {channel: "kline", interval: 7})
      while (true) {
          var ev = ctx.poll(k, 3000)
          if (!ev) {
              continue
          }
          if (ev.kind === 35) {
              if (ev.prefillStatus === 2) {
                  Log("kline subscription failed:", ev.message)
                  break
              }
              Log("kline history ready")
          } else if (ev.kind === 5 && ev.closed) {
              Log("closed bar", _D(ev.openTimeMs), ev.close)
          }
      }
  }
  

  
      python
      
      
  def main():
      # 多数交易所不支持7秒周期的K线
      k = ctx.subscribe(exchange, {"channel": "kline", "interval": 7})
      while True:
          ev = ctx.poll(k, 3000)
          if ev is None:
              continue
          if ev.kind == 35:
              if ev.prefillStatus == 2:
                  Log("kline subscription failed:", ev.message)
                  break
              Log("kline history ready")
          elif ev.kind == 5 and ev.closed:
              Log("closed bar", _D(ev.openTimeMs), ev.close)
  

• 任意档数的深度订阅：订阅7档时，选用交易所原生档位中不少于7档的最小档位，并截断为7档推送；档数超过255时立即抛出异常；超过交易所支持的上限时，该路订阅会收到失败状态事件。

  • javascript
  • python
  • rust

  
      javascript
      
      
  function main() {
      var d7 = ctx.subscribe(exchange, {channel: "depth", depth: 7})
      try {
          ctx.subscribe(exchange, {channel: "depth", depth: 300})    // 超过255档，抛出异常
      } catch (e) {
          Log("depth 300 rejected:", String(e))
      }
      var n = 0
      while (n < 10) {
          var ev = ctx.poll(d7, 3000)
          if (!ev) {
              continue
          }
          if (ev.kind === 35 && ev.prefillStatus === 2) {
              Log("depth 7 subscription failed:", ev.message)       // 例如交易所不支持该档数
              break
          }
          if (ev.kind === 3) {
              var b = ev.book()          // 不传n时，返回订阅的档数
              Log("levels:", b.bids.length, b.asks.length)
              n++
          }
      }
      d7.unsubscribe()
  }
  

  
      python
      
      
  def main():
      d7 = ctx.subscribe(exchange, {"channel": "depth", "depth": 7})
      try:
          ctx.subscribe(exchange, {"channel": "depth", "depth": 300})   # 超过255档，抛出异常
      except Exception as e:
          Log("depth 300 rejected:", e)
      n = 0
      while n < 10:
          ev = ctx.poll(d7, 3000)
          if ev is None:
              continue
          if ev.kind == 35 and ev.prefillStatus == 2:
              Log("depth 7 subscription failed:", ev.message)      # 例如交易所不支持该档数
              break
          if ev.kind == 3:
              b = ev.book()              # 不传n时，返回订阅的档数
              Log("levels:", len(b["bids"]), len(b["asks"]))
              n += 1
      d7.unsubscribe()
  

  
      rust
      
      
  use sdk::ctx::EventExt;
  
  fn main() {
      let d7 = ctx::subscribe(exchange, ctx::SubOpts::depth(7)).unwrap();
      if let Err(e) = ctx::subscribe(exchange, ctx::SubOpts::depth(300)) {
          Log!("depth 300 rejected:", e);                          // 超过255档，返回Err
      }
      let mut n = 0;
      while n < 10 {
          match ctx::poll(&[d7.id()], Some(3000)) {
              ctx::Polled::Event(ev) => {
                  if let Some(st) = ev.feed_status() {
                      if st.failed() {
                          Log!("depth 7 subscription failed:", st.message);
                          break;
                      }
                  } else if let Ok(b) = ev.book(7) {
                      Log!("levels:", b.bids.len(), b.asks.len());
                      n += 1;
                  }
              }
              ctx::Polled::Stopped => break,
              _ => {}
          }
      }
      let _ = d7.unsubscribe();
  }
  

返回值

参数

参考

备注

options的字段：

品种写法见品种代码。当ex传交易所对象且行情频道未传symbol时，使用调用ctx.subscribe()时该交易所对象的交易对；期货还包含合约类型（例如BTC_USDT.swap）。此后调用exchange.SetCurrency()、exchange.SetContractType()不会改变已建立的订阅。ex传下标时不会自动补全symbol；tickers（整个市场）和private（整个账户）频道始终不带symbol。

symbol传数组时为批量订阅，一次调用即可订阅多个品种：

• 返回与数组顺序一致的订阅对象数组，即使数组只有一个元素也返回数组；symbol为字符串时仍返回单个订阅对象。
• 整批先统一校验：任一品种或选项不合法时，整批抛出异常，不订阅任何品种。
• 校验通过后，每个元素都是一路独立的订阅，拥有各自的id、订单簿和引用计数；交易所拒绝其中某个品种时，只有该路订阅收到失败的状态事件，其余照常推送。
• 返回的数组带有unsubscribe()方法（不可枚举），可一次退订全部元素；该数组也可直接传给ctx.poll()。

订阅的身份由账户、symbol、channel、depth、interval共同决定：

• 同一线程中身份完全相同的订阅共用一路事件流，返回的订阅对象id相同。托管者按引用计数管理：每调用一次ctx.subscribe()增加一个引用，每个订阅对象调用一次unsubscribe()减少一个引用，最后一个引用释放时才真正向交易所退订，因此某处退订不会中断其他地方仍在使用的同一路订阅。
• 任一项不同即为另一路订阅、另一条事件流：同一品种的ticker与bbo是两路；depth为5与20也是两路，各自维护一本订单簿（交易所对不同档数的推送频率可能不同，托管者不会将其合并为最大档数）。
• 事件流ID在一次运行中只增不减、不会复用，全部退订后重新订阅会得到新的ID。

订阅归属于调用ctx.subscribe()的线程。主线程和threading.Thread()创建的线程（Python中为标准库的threading.Thread，Rust中为任意线程）都可以订阅和获取事件：

• 某路订阅的事件只投递给订阅它的线程，ctx.poll()只获取本线程自己订阅的事件。如需多个线程分别接收各自的事件，应在每个线程中分别订阅所需的几路。
• 不同线程订阅身份相同的内容时，交易所只订阅一次，订单簿、K线缓存和账户状态也只有一份；但每个线程获得各自的订阅对象（id不同）和独立的一份事件，投递模式相互独立，某个线程取得慢不会影响其他线程。所有线程都退订后才真正向交易所退订。
• 后订阅的线程订阅一路已在推送的K线时，会先补收已缓存的K线（prefill为true）和最近一条初始化状态事件；私有流同理，若账户快照已就绪，会立即补收全部快照和就绪事件，positions()、balances()可立即读取。其他频道从下一条推送开始接收。
• 只有订阅所属的线程才能退订：其他线程调用该订阅对象的unsubscribe()会抛出异常，订阅保持有效。线程结束时，其订阅会全部自动退订，其他线程共用的同一路订阅不受影响。
• 网关状态事件发送给在该交易所上有订阅的每个线程；定时器事件只发送给主线程。

tickers频道订阅整个市场的ticker：无需传symbol，一路订阅即可收到该交易所全部品种的ticker。事件格式与ticker事件相同（kind为1），channel为tickers，symbol为各自对应的品种。

• 仅当交易所提供全市场推送时才支持，各交易所的支持情况见能力矩阵中行情分组的「推送」列以及各交易所页面。不支持的交易所会在该路订阅上收到一条失败的状态事件（message以unsupported channel: tickers开头），不会自动改为逐个品种订阅。
• 部分交易所的全市场推送不含买一价和卖一价，此时bid、ask为0；如需买卖价，请按品种订阅ticker或bbo。
• 默认每个品种只保留最新一条：即使取得慢，也能拿到每个品种最新的ticker。

private频道是账户的私有流，不传symbol，覆盖整个账户，每个账户只有一路。订单回报、成交回报、持仓、余额都在这一路事件流上推送，对应事件的channel分别为orders、fills、positions、balance：

• 订阅生效后（以及每次断线重连后），先推送一份账户快照（持仓、余额，snapshot为true），再推送一条kind为35、channel为private的状态事件：prefillStatus为1表示快照就绪，为2表示快照失败（message为失败原因，托管者会自动退避重试）。此后才是实时推送（snapshot为false）。订单回报、成交回报没有快照。
• 持仓、余额的每一条推送都是该持仓或该资产当前的完整状态，而非增量：持仓数量为0表示已平仓，可用与冻结均为0表示该资产已清空。
• 成交回报（kind为19，channel为fills）是自身订单的每一笔成交，包含订单号、成交号、该笔成交的价格、数量、手续费，以及是挂单还是吃单；先推送成交回报，再推送反映该成交的订单回报。订阅生效及断线重连后不补发历史成交；仅当交易所的私有推送包含逐笔成交时才会推送，字段说明见ctx.poll。
• 订阅对象的positions、balances方法返回托管者根据推送维护的最新持仓、余额。
• 托管者不会在推送之后再请求REST接口补齐数据：快照之后只推送交易所推来的内容。交易所不推送的变化（例如部分交易所下单、撤单时不推送余额，部分交易所不推送持仓、只提供快照）不会出现在推送中；若余额推送中没有与exchange.GetAssets同口径的可用、冻结数值，对应字段为null。需要最新值时，请调用exchange.GetAssets、exchange.GetPositions。实时推送覆盖的内容见各交易所页面的说明。

订阅是惰性的：ctx.subscribe()立即返回并登记订阅，首次调用ctx.poll()时才与交易所建立连接并发送订阅请求。连接状态的变化通过网关状态事件（kind为33，channel为gateway）通知。

depth频道的档数可以是0到255之间的任意整数：

• 不传或传0表示使用交易所的默认档数，传1到255表示所需的档数。每个档数都是一路独立的订阅，同一品种depth为7与10是两路订阅，对应两本订单簿。
• 托管者选用交易所原生档位中能覆盖所需档数的最小一档，再截断至所需档数后推送。例如交易所只提供5、10、20档快照时，depth为7的订阅使用10档快照，只保留前7档；同一品种的多路订阅可以共用一路原生订阅，各自截断。
• 档数超过255或不是整数时，ctx.subscribe()抛出异常；超过该交易所支持的最大档数时不抛出异常，而是这路订阅收到一条失败的状态事件，message以unsupported depth开头。各交易所的原生档位及上限见交易所页面。

订阅失败分为两种情况：

• 参数有误时抛出异常，例如：参数个数不是2个、第二个参数不是对象、depth超过255、options中有未知字段（例如拼错的intervel）、未知的channel、行情频道未传symbol、tickers或private频道传了symbol、同时传入mode与buffer、symbol数组中有不合法的元素（此时整批均不订阅）。
• 交易所或连接器拒绝订阅时（例如不支持的频道、K线周期或深度档数），ctx.subscribe()不抛出异常，而是在这路订阅自身的事件流上收到一条kind为35、prefillStatus为2的状态事件，message为失败原因。此后这路订阅作废，调用unsubscribe()不执行任何操作。

未及时取出事件时的处理（投递模式）：ticker、bbo、depth默认只保留最新一条，tickers默认每个品种只保留最新一条；trade、kline默认缓存最近1024条，溢出时丢弃最旧的事件；可通过mode或buffer（二选一）修改。订单回报、成交回报、持仓、余额、网关状态、初始化状态事件不会被丢弃，因此订阅后需持续调用ctx.poll()。订单簿在托管者内持续更新，不受取事件快慢的影响，任何时候通过book()读取的都是最新的完整订单簿。托管者接收推送的缓冲是否积压，可查看ctx.poll中的ctx.connectors()计数。

推送与REST接口完全独立：订阅只影响ctx.poll()收到的事件。exchange.GetTicker、exchange.GetDepth、exchange.GetTrades、exchange.GetRecords、exchange.GetAccount等函数每次都直接请求交易所的REST接口，不读取推送数据；是否订阅以及订阅了什么，都不会改变它们的结果。

在JavaScript、Rust策略中，同一线程内的ctx.poll()与EventLoop从同一个事件队列中取事件，因此一个线程只应使用其中一种方式取事件。Python的EventLoop()拥有独立的队列（仅包含柜台类交易所自动订阅的行情和订单回报，以及exchange.Go()等并发任务的完成通知），与ctx.poll()互不争抢。各交易所是否支持推送以及支持哪些频道，请参见能力矩阵及各交易所页面的说明。

Python策略的写法与JavaScript相同：ctx.subscribe(ex, options)只接受这两个位置参数（多传、少传或使用关键字参数时均抛出TypeError），options为字典，字段与校验规则均相同，传入的字典不会被修改。options["symbol"]为列表（或元组）时返回SubscriptionList（list的子类，元素顺序与输入一致），调用其unsubscribe()可一次性退订全部元素。订阅对象的属性为只读，info()返回包含相同字段的字典。通过threading.Thread创建的线程在结束时会自动退订其订阅；通过_thread.start_new_thread()等方式绕过threading.Thread创建的线程同样可以订阅，但结束时不会自动退订，需要自行调用unsubscribe()。

Rust策略使用ctx模块中的同名函数：ctx::subscribe(ex, opts: ctx::SubOpts) -> Result<ctx::Sub, String>，其中ex可以是交易所对象（exchange、exchanges[i]）或u16下标；失败时返回Err，不会panic。订阅选项通过ctx::SubOpts::ticker()、bbo()、trade()、depth(n)、kline(秒数)、tickers()、private()构造，再链式调用.symbol("BTC_USDT")、.latest()、.buffer(n)。批量订阅使用ctx::subscribe_many(ex, opts, &["BTC_USDT", "ETH_USDT"]) -> Result<Vec<ctx::Sub>, String>，同样会先对整批订阅进行校验。返回的ctx::Sub可以复制，其方法包括id()（传给ctx::poll(&[sub.id()], ...)）、book(n)、positions()、balances()、unsubscribe()；引用计数按ctx::subscribe()的调用次数计算，每调用一次都需对应调用一次unsubscribe()。Rust的任意线程均可订阅，订阅同样归属于发起订阅的线程。