mmr Live-Tick-Pipeline Bug Report

[uc0ne]

mmr Live-Tick-Pipeline Bug Report

Author: uc1 (uc1)
Date: 2026-05-01
Setup: Docker mode (./docker.sh -g), gnzsnz/ib-gateway:latest, IBKR CapTrader Paper Account (DUP203686)
mmr commit: (latest of 9600dev/mmr cloned 2026-05-01 11:02 UTC)

Issue Summary

mmr listen <symbol> and mmr snapshot <symbol> time out indefinitely. Strategy successfully subscribes (streaming: 1 in mmr status), subscribe_contract_direct runs successfully, reqMktData returns OK without exception, but pendingTickersEvent never fires for the trader_service's IB connection. ZMQ PubSub port 42002 receives 0 messages.

Reproduction

1. Standard Docker setup: ./docker.sh -g → enter creds → containers start
2. mmr universe create test && mmr universe add test AAPL (resolves AAPL conId=265598 NASDAQ)
3. Add a no-op subscriber strategy:

   # strategies/probe_subscribe.py
   from trader.trading.strategy import Strategy
   class ProbeSubscribe(Strategy):
       def on_prices(self, prices): return None

   # ~/.config/mmr/strategy_runtime.yaml
   - name: probe
     conids: [265598]
     bar_size: "1 min"
     historical_days_prior: 1
     module: strategies/probe_subscribe.py
     class_name: ProbeSubscribe

4. Restart trader_service (or container)
5. mmr status → streaming: 1 ✓
6. mmr listen AAPL --exchange NASDAQ → timeout, 0 ticks
7. mmr snapshot AAPL --exchange NASDAQ → timeout after 30s

Counter-Test (proves the issue is mmr-specific)

In the same container, fresh ib_async.IB() connection with clientId=98 (different from trader_service's cid=5):

import asyncio
from ib_async import IB, Stock

async def test():
    ib = IB()
    await ib.connectAsync("ib-gateway", 4004, clientId=98, timeout=10)
    contract = Stock("AAPL", "SMART", "USD", primaryExchange="NASDAQ")
    qualified = await ib.qualifyContractsAsync(contract)
    ib.reqMarketDataType(1)  # LIVE
    ticker = ib.reqMktData(qualified[0], "", False, False)

    raw_count = [0]
    def on_pending(tickers): raw_count[0] += 1
    ib.pendingTickersEvent += on_pending

    await asyncio.sleep(12)
    print(f"raw events: {raw_count[0]}")  # → 38

asyncio.run(test())

This delivers 38 ticks in 12s for AAPL during pre-market. mmr's EventSubject + flat_map(mapper) + subscribe pattern (replicated in same script) emits all 38 — pipeline works in isolation.

Diagnostic instrumentation

Temporarily added logging.error() to:

• trader/listeners/ibreactive.py:subscribe_contract_direct ENTRY
• trader/listeners/ibreactive.py:subscribe_contract_direct after call_event_subscriber_sync(reqMktData)
• trader/common/reactivex.py:EventSubject.on_eventkit_update (raw event)
• trader/trading/trading_runtime.py:publish_contract.on_next (ticker → ZMQ)

Trader_service log timeline:

07:34:30,127  subscribe_contract_direct ENTRY  conId=265598 ... delayed=False snap=False
07:34:30,129  reqMktData call OK reqId=3  _contracts_source_id=126868632077200
... 4+ minutes elapsed ...
[NO] on_eventkit_update fired
[NO] publish_contract on_next

pendingTickersEvent does not fire in trader_service's IB connection despite successful reqMktData. The same pendingTickersEvent mechanism works in fresh ib_async.IB() instances with different clientId in the same Python process / container.

Hypothesis

trader_service.py:main() calls trader.connect() synchronously, which internally calls multiple self.run(coro) (= ib.run(coro)) consecutively:

self.run(self.zmq_messagebus.start())
self.run(self.zmq_strategy_client.connect())
self.run(self.zmq_rpc_server.serve())  # never returns

ib.run() blocks until coroutine completes. serve() is long-running, so this blocks connect() forever. The asyncio loop is being driven by serve(), but somehow ib_async's internal event handlers (specifically pendingTickersEvent) don't get callbacks from this loop driver.

Possible root causes:

• mmr's loop-driving via ib.run(serve()) doesn't properly process ib_async's internal recv/event queue
• The EventSubject's weakref to itself is being collected (unlikely — self._contracts_source keeps strong ref via self.client)
• ib_async requires IB.run() to be the only loop entry point, and nesting via ib.run(other_coro) breaks event delivery

Suggested investigation paths

1. Verify ib_async's recv loop is being driven (e.g. ib.client._reqIdSeq increases? await ib.reqHistoricalDataAsync(...) returns? Yes for the latter — historical fetch works in trader_service via strategy_runtime.get_historical_data())
2. Add explicit asyncio.sleep(0) yields in the trader_service main loop
3. Test against a paper/live account with a different cid pre-occupied
4. Add logging.debug to EventSubject.on_eventkit_update to confirm it's never called

Workaround (none for the user)

Currently no clean workaround. Strategies can use historical_days_prior to load past bars during init (get_historical_data works fine — that's a synchronous IB call), but cannot react to live ticks. propose → approve pipeline works without live ticks but is decoupled from market reality.

Environment

• Host: Ubuntu 24.04, Linux 6.17.0-23-generic, Docker 29.4.1, Compose v5.1.3
• Container: Debian bookworm, Python 3.11.2
• ib_async, reactivex, eventkit (latest as of mmr requirements.txt)
• IB Gateway 10.46.1d (CapTrader / IBKR Paper)

[uc0ne]

---

Update — root cause found and fixed locally.

The pipeline-bug is in _snapshot_event_handlers (trader/listeners/ibreactive.py).

Root cause

IBAIORx.__init__ registers EventSubject.on_eventkit_update as a bound method on self.ib.pendingTickersEvent via event += self.method. eventkit stores this as:

Slot(obj=None, weakref=weakref(EventSubject), func=method.__func__)

IBAIORx.connect() then replaces self.ib = IB() (fresh instance) and calls _snapshot_event_handlers(old_ib) + _restore_event_handlers(new_ib, snap) to preserve handlers across the IB-replace.

The bug is in _snapshot_event_handlers:

obj = getattr(slot, 'obj', None)
...
if obj is not None:
    handler = getattr(obj, func.__name__)   # rebuild bound method
else:
    handler = func   # ← BUG: misses weakref-pattern bound methods

For weakref-registered bound methods, slot.obj is None and slot.weakref holds the target. The current snapshot only checks slot.obj, so it captures the bare unbound function. On restore, event += func registers a plain function. When pendingTickersEvent.emit() fires, eventkit calls func(tickers) without self — EventSubject.on_eventkit_update(self, e, *args) raises TypeError: missing self, silently swallowed by the eventkit error path.

→ _contracts_source never gets ticker data → ZMQ PubSub stays silent → mmr listen and mmr snapshot time out.

Verification

Repro script (run in container):

from ib_async import IB
from trader.common.reactivex import EventSubject
from trader.listeners.ibreactive import _snapshot_event_handlers, _restore_event_handlers

ib1 = IB()
es = EventSubject(eventkit_event=ib1.pendingTickersEvent)
snap = _snapshot_event_handlers(ib1)
captured = snap['pendingTickersEvent'][0]
print(f"is_bound_method: {hasattr(captured, '__self__')}")
# Before fix: False  (bare function)
# After fix:  True   (bound method to EventSubject)

Live verification after fix:

mmr listen AAPL --exchange NASDAQ
# AAPL  bid=280.71  ask=280.82  last=280.7308
# AAPL  bid=280.71  ask=280.82  last=280.7173
# ...continuous updates

Fix (diff against main HEAD as of 2026-05-01)

--- a/trader/listeners/ibreactive.py
+++ b/trader/listeners/ibreactive.py
@@ -90,14 +90,36 @@ def _snapshot_event_handlers(ib):
                 if func is None:
                     continue
                 # Reconstruct what was originally registered: a bound method
-                # if obj was present, otherwise the bare function.
+                # if obj was present, otherwise dereference weakref (eventkit
+                # stores bound methods as obj=None + weakref(target)). For
+                # plain functions both obj and weakref are None.
+                handler = func
                 if obj is not None:
                     try:
                         handler = getattr(obj, func.__name__)
                     except Exception:
                         handler = func
                 else:
-                    handler = func
+                    wref = getattr(slot, 'weakref', None)
+                    if wref is not None:
+                        try:
+                            target = wref()
+                        except Exception:
+                            target = None
+                        if target is not None:
+                            try:
+                                handler = getattr(target, func.__name__)
+                            except Exception:
+                                handler = func
                 captured.append(handler)
         except Exception:
             pass

Happy to open a PR if useful.

Closing this issue as fixed locally.