From signal to broker: live execution at IBKR

$ uv add multiedge-relay ib_async   # or: pip install multiedge-relay ib_async
$ export MULTIEDGE_API_KEY="mesk_..."   # subscriber-scoped key from the portal
# Plus a running IB Gateway or TWS of YOUR OWN, with API connections
# enabled (Configure -> Settings -> API -> Enable ActiveX and Socket
# Clients). Your broker credentials stay in that session — they never
# touch the relay or this SDK.

from ib_async import IB

# Your broker session's API ports — paper first, ALWAYS:
#   TWS         7497 (paper)   7496 (live)
#   IB Gateway  4002 (paper)   4001 (live)
# Going live later is a one-character change: the port.

ib = IB()
ib.connect("127.0.0.1", 4002, clientId=17)   # paper Gateway; clientId unique per script
print(ib.isConnected())
ib.disconnect()

import os

from multiedge_relay import SignalSubscriber, SqliteStateStore

store = SqliteStateStore()   # one local SQLite file: cursor + dedup together

subscriber = SignalSubscriber(
    api_key=os.environ["MULTIEDGE_API_KEY"],
    strategy_id="str_the_feed_you_follow",
    # exactly_once() never re-runs a signal_id the handler completed. But
    # order placement is an EXTERNAL side effect: a crash in the instant
    # between handler return and commit re-runs it once. That is why the
    # handler trades to TARGETS (idempotent — re-running computes zero
    # deltas), never replays raw orders.
    on_signal=store.exactly_once(execute_rebalance),
    cursor_store=store,
    on_error=lambda exc: log.warning("relay transport: %s", exc),
)
subscriber.run()   # blocks: catch-up in sequence order, then live polling

from ib_async import IB, Stock

MIN_ORDER_NOTIONAL = 200.0   # rebalance band: skipping dust beats churn

def compute_orders(ib: IB, targets: dict[str, float]) -> list[tuple[Stock, int]]:
    """Share deltas that move the LIVE account onto the signal's weights."""
    net_liq = float(next(
        v.value for v in ib.accountSummary() if v.tag == "NetLiquidation"
    ))
    held = {p.contract.symbol: int(p.position) for p in ib.positions()}

    ib.reqMarketDataType(3)   # delayed quotes size a daily rebalance fine
    tickers = sorted(set(held) | set(targets))
    contracts = {t: Stock(t, "SMART", "USD") for t in tickers}
    ib.qualifyContracts(*contracts.values())

    orders: list[tuple[Stock, int]] = []
    for ticker in tickers:
        price = ib.reqTickers(contracts[ticker])[0].marketPrice()
        if not price or price != price:   # 0 or NaN: no sane quote, no trade
            raise RuntimeError(f"{ticker}: no sane price — refusing to size")
        # Absent from a non-empty book => weight 0 => LIQUIDATION. HOLD
        # lines carry their unchanged weight => delta ~0 => no order.
        desired = round(net_liq * targets.get(ticker, 0.0) / price)
        delta = desired - held.get(ticker, 0)
        if abs(delta) * price < MIN_ORDER_NOTIONAL:
            continue
        orders.append((contracts[ticker], delta))
    # Idempotent by construction: re-running against an already-rebalanced
    # account computes zero deltas and places nothing.
    return orders

from ib_async import IB, MarketOrder
from multiedge_relay import ReceivedSignal, SignalMeta

def execute_rebalance(signal: ReceivedSignal, meta: SignalMeta) -> None:
    """One complete portfolio in — orders at YOUR broker out."""
    payload = signal.payload

    # 1. Heartbeat: an explicit no-action day. Return before any trading
    #    logic — it never trades and never liquidates.
    if not payload["positions"]:
        return

    # 2. Corrections: the HIGHEST sequence per signal_date wins; a stale
    #    revision is dropped in one comparison.
    if applied_sequence_for(payload["signal_date"]) >= signal.sequence:
        return

    # 3. The weights are the instruction; the verbs annotate it.
    targets = {
        p["ticker"]: p["signal_portfolio_weight"] for p in payload["positions"]
    }

    # 4. Connect lazily. subscriber.run() owns this thread, and a
    #    persistently connected IB() starves without its event loop being
    #    pumped — at daily cadence, connect -> trade -> disconnect is both
    #    simpler and safer.
    ib = IB()
    ib.connect("127.0.0.1", 4002, clientId=17)   # paper
    try:
        for contract, delta in compute_orders(ib, targets):
            side = "BUY" if delta > 0 else "SELL"
            trade = ib.placeOrder(contract, MarketOrder(side, abs(delta)))
            while not trade.isDone():
                ib.sleep(0.25)   # pumps ib_async's event loop while waiting
            if trade.orderStatus.status in ("Cancelled", "Inactive"):
                raise RuntimeError(f"{contract.symbol}: {side} {abs(delta)} rejected")
        mark_applied(payload["signal_date"], signal.sequence)
    finally:
        ib.disconnect()

# Two legs, two failure domains — keep them separate.

# RELAY LEG: transient transport errors go to on_error while the SDK
# retries with jittered backoff. You log; you fix nothing.
def log_transport_error(exc: Exception) -> None:
    log.warning("relay transport: %s", exc)

# BROKER LEG: raising OUT of the handler is the safety mechanism. The
# cursor commits only after the handler returns, so an exception means
# the signal is REDELIVERED once the fault is fixed — never skipped, and
# never marked done half-executed. Redelivery is safe because the next
# run recomputes deltas from the live account: only the residual trades.
def execute_rebalance(signal, meta):
    ib = IB()
    ib.connect("127.0.0.1", 4002, clientId=17)
    if not ib.isConnected():
        raise RuntimeError("IB Gateway down — leaving the signal uncommitted")
    ...
    # A reject is a loud stop, not a shrug: reconcile before the signal
    # can ever be marked done.
    if trade.orderStatus.status in ("Cancelled", "Inactive"):
        raise RuntimeError(f"rejected: {trade.log[-1].message}")

from pathlib import Path

MAX_ORDER_NOTIONAL = 50_000.0        # hard per-order cap — size it to YOUR book
KILL_SWITCH = Path("~/.trading/HALT").expanduser()

def guard_order(ticker: str, delta: int, price: float) -> None:
    """Hard guardrails, checked BEFORE any order leaves the process."""
    if KILL_SWITCH.exists():
        # touch ~/.trading/HALT from any shell to stop all trading NOW;
        # the uncommitted signal redelivers after the file is removed.
        raise RuntimeError("kill switch present — no orders")
    if abs(delta) * price > MAX_ORDER_NOTIONAL:
        raise RuntimeError(f"{ticker}: notional cap breached — refusing")

# And the cheapest guardrail of all: stay on the paper port (4002) until
# a full week of signals — including a heartbeat and a correction — has
# executed cleanly end to end.

"""Relay signal in, IBKR orders out — your own execution layer, complete.

The relay delivers the target book; this process owns execution. Paper
account first (Gateway port 4002); the only change for live is the port.
"""
import logging
import os
from pathlib import Path

from ib_async import IB, MarketOrder, Stock
from multiedge_relay import (
    ReceivedSignal,
    SignalMeta,
    SignalSubscriber,
    SqliteStateStore,
)

log = logging.getLogger("execution")

IB_HOST, IB_PORT, IB_CLIENT_ID = "127.0.0.1", 4002, 17   # 4002 = Gateway paper
MIN_ORDER_NOTIONAL = 200.0
MAX_ORDER_NOTIONAL = 50_000.0
KILL_SWITCH = Path("~/.trading/HALT").expanduser()

# signal_date -> highest applied sequence. In-process is enough for the
# supersede rule at runtime (exactly_once already blocks replays across
# restarts); persist it next to your book when you outgrow one process.
applied: dict[str, int] = {}


def compute_orders(ib: IB, targets: dict[str, float]) -> list[tuple[Stock, int]]:
    """Share deltas that move the live account onto the signal's weights."""
    net_liq = float(next(
        v.value for v in ib.accountSummary() if v.tag == "NetLiquidation"
    ))
    held = {p.contract.symbol: int(p.position) for p in ib.positions()}
    ib.reqMarketDataType(3)   # delayed quotes size a daily rebalance fine
    tickers = sorted(set(held) | set(targets))
    contracts = {t: Stock(t, "SMART", "USD") for t in tickers}
    ib.qualifyContracts(*contracts.values())
    orders: list[tuple[Stock, int]] = []
    for ticker in tickers:
        price = ib.reqTickers(contracts[ticker])[0].marketPrice()
        if not price or price != price:
            raise RuntimeError(f"{ticker}: no sane price — refusing to size")
        desired = round(net_liq * targets.get(ticker, 0.0) / price)  # absent => liquidate
        delta = desired - held.get(ticker, 0)
        if abs(delta) * price < MIN_ORDER_NOTIONAL:
            continue   # HOLD lines and dust land here: no order
        if abs(delta) * price > MAX_ORDER_NOTIONAL:
            raise RuntimeError(f"{ticker}: notional cap breached — refusing")
        orders.append((contracts[ticker], delta))
    return orders   # idempotent: an already-rebalanced book yields []


def execute_rebalance(signal: ReceivedSignal, meta: SignalMeta) -> None:
    payload = signal.payload
    if not payload["positions"]:
        return   # heartbeat: no action, nothing liquidated
    date = payload["signal_date"]
    if applied.get(date, -1) >= signal.sequence:
        return   # a later correction for this date already applied
    if KILL_SWITCH.exists():
        raise RuntimeError("kill switch present — leaving the signal uncommitted")
    targets = {
        p["ticker"]: p["signal_portfolio_weight"] for p in payload["positions"]
    }
    ib = IB()
    ib.connect(IB_HOST, IB_PORT, clientId=IB_CLIENT_ID)
    try:
        for contract, delta in compute_orders(ib, targets):
            side = "BUY" if delta > 0 else "SELL"
            trade = ib.placeOrder(contract, MarketOrder(side, abs(delta)))
            while not trade.isDone():
                ib.sleep(0.25)
            if trade.orderStatus.status in ("Cancelled", "Inactive"):
                raise RuntimeError(f"{contract.symbol}: {side} {abs(delta)} rejected")
            log.info("%s %s %d filled", contract.symbol, side, abs(delta))
        applied[date] = signal.sequence
    finally:
        ib.disconnect()


def main() -> None:
    logging.basicConfig(level=logging.INFO)
    store = SqliteStateStore()
    subscriber = SignalSubscriber(
        api_key=os.environ["MULTIEDGE_API_KEY"],
        strategy_id="str_the_feed_you_follow",
        on_signal=store.exactly_once(execute_rebalance),
        cursor_store=store,
        on_error=lambda exc: log.warning("relay transport: %s", exc),
    )
    try:
        subscriber.run()
    except KeyboardInterrupt:
        subscriber.stop()
    finally:
        subscriber.close()
        store.close()


if __name__ == "__main__":
    main()