From fd44842570fae12ecfee48c9834d8ff833943bde Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Mon, 12 Oct 2026 23:17:11 -0700 Subject: [PATCH] live: IBKR order layer for trend_rider_v10 (guarded MOC orders via IB Gateway) ibkr_trader.py computes the strategy's target for the close the way trader.py's morning phase does, sizes whole-share orders on min(NAV, $10.5k) less a 1% cash buffer against the account's real positions, and sends MOC orders, sells before buys. Guards: the login must manage IBKR_ACCOUNT; a live account needs --live and data/ibkr_live/ARMED; the account may hold only the strategy's universe; MOC whole-share orders only; one submission per session. After the close it records executions and commissions, scores each order against the primary exchange's closing cross (SIP condition 6) and compares NAV with the simulated account, with a preregistered verdict on the first 20 orders. A what-if run on the paper Gateway (no orders) accepted GLD/SHY MOC orders at $1.00 commission and rejected TQQQ/UPRO: the account lacks the Complex or Leveraged ETP permission. The live service is defined but not enabled. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 11 +- deploy/start-ibkr-live.sh | 23 + .../systemd/quant-trading-ibkr-live.service | 18 + ibkr_trader.py | 582 ++++++++++++++++++ pyproject.toml | 1 + tests/test_ibkr_trader.py | 74 +++ uv.lock | 37 ++ 7 files changed, 745 insertions(+), 1 deletion(-) create mode 100755 deploy/start-ibkr-live.sh create mode 100644 deploy/systemd/quant-trading-ibkr-live.service create mode 100644 ibkr_trader.py create mode 100644 tests/test_ibkr_trader.py diff --git a/CLAUDE.md b/CLAUDE.md index b9f8198..04d7051 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -50,6 +50,14 @@ uv run python overnight_paper.py buy|sell --dry-run # show the orders it wou uv run python overnight_paper.py report # paper fills vs official open/close prints, nightly returns uv run python overnight_paper.py daemon # the scheduler (service quant-trading-overnight-paper) +# trend_rider_v10 in an IBKR account: today's target as trader.py morning computes it, whole-share MOC orders +# via IB Gateway, fills scored against the closing cross (IBKR_ACCOUNT/IBKR_PORT/IBKR_CLIENT_ID from .env) +uv run python ibkr_trader.py check # account, positions, open orders (read-only) +uv run python ibkr_trader.py plan [--what-if] # today's orders; --what-if: IBKR validates/prices them, nothing placed +uv run python ibkr_trader.py submit --live # MOC orders, close-30 to close-15 min; live needs data/ibkr_live/ARMED +uv run python ibkr_trader.py report # executions, fills vs closing cross, live vs simulated account +uv run python ibkr_trader.py daemon --live # the scheduler (service quant-trading-ibkr-live, not enabled) + # Precompute the strategy page's backtests (writes data/precomputed/) uv run python precompute_backtest.py @@ -217,7 +225,7 @@ The comparison writes its fixed budget and acceptance rule before running and re ## Server Deployment -Four `systemd --user` services and one timer, defined in `deploy/systemd/`, run from `/home/gahow/projects/quant-trading`: +Four `systemd --user` services (a fifth defined but not enabled) and one timer, defined in `deploy/systemd/`, run from `/home/gahow/projects/quant-trading`: | Service | Script | Role | |---|---|---| @@ -225,6 +233,7 @@ Four `systemd --user` services and one timer, defined in `deploy/systemd/`, run | `quant-trading-api` | `deploy/start-web-api.sh` | Web API on `127.0.0.1:8787` | | `quant-trading-vite` | `deploy/start-web-vite.sh` | Vite on `127.0.0.1:5174`, behind nginx on `:5443` | | `quant-trading-overnight-paper` | `deploy/start-overnight-paper.sh` | `overnight_paper.py daemon`: overnight-only momentum on the Alpaca paper account (from 2026-10-12) | +| `quant-trading-ibkr-live` (not enabled) | `deploy/start-ibkr-live.sh` | `ibkr_trader.py daemon --live`: trend_rider_v10 in the $10k IBKR Pro live account. Needs a live IB Gateway on port 4001 (the paper Gateway on 4002 belongs to ../ibkr-quant, whose strategy owns that paper account), IBKR_ACCOUNT in .env, the Complex or Leveraged ETP permission (paper what-if 2026-10-13 rejected TQQQ/UPRO without it) and data/ibkr_live/ARMED | | `quant-trading-options` (timer, 16:25 New York Mon-Fri) | `deploy/start-options-snapshot.sh` | `research.lab.fetch_options snapshot`: S&P 500 option chains (indicative quotes), OPRA daily bars, per-name IV/skew/spread/volume summary and a quote-quality sample into `data/options/us/` (~15 MB/day); `status` prints the indicative-vs-OPRA verdict after 20 sessions | ```bash diff --git a/deploy/start-ibkr-live.sh b/deploy/start-ibkr-live.sh new file mode 100755 index 0000000..c38cee6 --- /dev/null +++ b/deploy/start-ibkr-live.sh @@ -0,0 +1,23 @@ +#!/usr/bin/env bash +# trend_rider_v10 in the IBKR live account (ibkr_trader.py daemon --live). +# Started by the quant-trading-ibkr-live.service systemd --user unit. +# Needs: a live IB Gateway on IBKR_PORT (default 4001), IBKR_ACCOUNT in .env, +# and data/ibkr_live/ARMED; without ARMED the daemon refuses to send orders. +# Yahoo is refreshed through QUANT_MARKET_PROXY as in start-monitor.sh; +# Alpaca and the local Gateway are reached directly. +set -euo pipefail + +SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +PROJECT_ROOT="$( cd "$SCRIPT_DIR/.." && pwd )" +cd "$PROJECT_ROOT" + +export PATH="/home/gahow/.local/bin:$PATH" +if [[ -f .env ]]; then + set -a; source .env; set +a +fi +if [[ -n "${QUANT_MARKET_PROXY:-}" ]]; then + export http_proxy="$QUANT_MARKET_PROXY" https_proxy="$QUANT_MARKET_PROXY" all_proxy="$QUANT_MARKET_PROXY" + export no_proxy="127.0.0.1,localhost,.alpaca.markets" +fi + +exec uv run python ibkr_trader.py daemon --live diff --git a/deploy/systemd/quant-trading-ibkr-live.service b/deploy/systemd/quant-trading-ibkr-live.service new file mode 100644 index 0000000..2b7b3b0 --- /dev/null +++ b/deploy/systemd/quant-trading-ibkr-live.service @@ -0,0 +1,18 @@ +[Unit] +Description=trend_rider_v10 in the IBKR live account (MOC orders via IB Gateway) +After=network-online.target + +# Not enabled by default. Install only after the live IB Gateway is logged in, +# IBKR_ACCOUNT is set in .env, the account may trade leveraged ETFs, and +# data/ibkr_live/ARMED exists (see ibkr_trader.py). +[Service] +Type=simple +WorkingDirectory=/home/gahow/projects/quant-trading +ExecStart=/home/gahow/projects/quant-trading/deploy/start-ibkr-live.sh +Restart=always +SuccessExitStatus=143 +RestartSec=30 +Environment=PYTHONUNBUFFERED=1 + +[Install] +WantedBy=default.target diff --git a/ibkr_trader.py b/ibkr_trader.py new file mode 100644 index 0000000..7ce167a --- /dev/null +++ b/ibkr_trader.py @@ -0,0 +1,582 @@ +""" +Trade one registered strategy in an Interactive Brokers account with +market-on-close orders (first: trend_rider_v10, $10k, IBKR Pro). Each session +it computes the strategy's target for the close exactly as trader.py's morning +phase does (session_signals on refreshed Yahoo prices, so the target equals +the simulated account in data/trader_us_trend_rider_v10.json), sizes +whole-share orders against the account's actual positions and sends them as +MOC orders through IB Gateway. After the close it records IBKR's executions +and commissions and scores every fill against the primary listing +exchange's closing cross (SIP condition 6, as in research/lab/auction_audit.py). + + uv run python ibkr_trader.py check # account, positions, open orders (read-only) + uv run python ibkr_trader.py plan [--what-if] # today's orders; --what-if has IBKR validate and price + # each order without placing it + uv run python ibkr_trader.py submit [--live] # send the MOC orders (close - 30 to close - 15 min) + uv run python ibkr_trader.py report # executions, fills vs the closing cross, live vs simulated + uv run python ibkr_trader.py daemon [--live] # submit and report on Alpaca's market calendar + +Connection settings come from the environment or .env: IBKR_ACCOUNT +(required), IBKR_HOST (127.0.0.1), IBKR_PORT (4001 = live Gateway; the paper +Gateway of ../ibkr-quant listens on 4002), IBKR_CLIENT_ID (31; ibkr-quant uses +17 and 18). Live account numbers are masked in output. + +Guards, each enforced in code: + account the Gateway login must manage IBKR_ACCOUNT; an account that is not a + paper account (DU/DF) also needs --live and the file data/ibkr_live/ARMED + (delete it to stop sending orders) + holdings submit refuses an account holding anything outside the strategy's + universe (plan only warns, so it can run on a shared paper account) + size orders are sized on min(net liquidation, MAX_EQUITY) less CASH_BUFFER + orders MOC only, whole shares, universe symbols, sells no larger than the + position; sells are sent and accepted before any buy + once one submission per session, none while an order of ours is open + +A buy that IBKR rejects (for example buying power while the sell that funds it +is still pending) leaves the cash idle; the next session's plan buys again. +The live record therefore differs from the simulated account by fills, +whole-share sizing, the cash buffer, commissions and any such delay. + +Boundary: NAV-based returns in the report assume no deposits or withdrawals. + +State and logs: data/ibkr_live/ (state.json, orders.jsonl, runs/.json, +fills.csv, scored.csv, daily.csv, prereg.json, verdict.json). +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import math +import os +import time +from datetime import date, datetime, timedelta, timezone +from zoneinfo import ZoneInfo + +import pandas as pd + +from execution import compute_trades + +OUT = os.path.join("data", "ibkr_live") +NY = ZoneInfo("America/New_York") +ARMED = os.path.join(OUT, "ARMED") +MARKET = "us" +STRATEGY = "trend_rider_v10" +SIM_STATE = os.path.join("data", f"trader_{MARKET}_{STRATEGY}.json") +MAX_EQUITY = 10_500.0 +CASH_BUFFER = 0.01 # MOC buys are sized at 15:30 prices and paid at the close +PAPER_PREFIXES = ("DU", "DF") +REF = "ibkr_trader" +SUBMIT_BEFORE_CLOSE = timedelta(minutes=30) +SUBMIT_CUTOFF = timedelta(minutes=15) # NYSE takes MOC orders until 15:50 ET +REPORT_AFTER_CLOSE = timedelta(minutes=60) # after the monitor's 16:35 evening phase +ACCEPTED = {"PreSubmitted", "Submitted", "Filled"} +DONE = {"Filled", "Cancelled", "ApiCancelled", "Inactive"} +PREREG = { + "written": "before the first live order", + "sample": "the first 20 filled MOC orders of the live account", + "reference": "primary exchange closing cross, largest SIP print with condition 6 in 15:59:59-16:10 ET", + "cost_per_side_bp": "fill vs cross, adverse positive, value-weighted; commission excluded (tracked separately)", + "rules": {"cost_per_side_bp": {"PASS": 1.0, "MEASURED": 2.5}, + "fill_rate": {"PASS": 0.99, "MEASURED": 0.97}, + "cross_match": {"PASS": 0.95, "MEASURED": 0.90, "tolerance_usd": 0.01}}, + "decision": "the worst of the three; STOP means stop sending orders and ask IBKR about the fills", +} + + +# --------------------------------------------------------------------------- +# guards (pure, tested in tests/test_ibkr_trader.py) +# --------------------------------------------------------------------------- + +def mask(account: str) -> str: + return account if account.startswith(PAPER_PREFIXES) else f"{account[:1]}***{account[-3:]}" + + +def check_account(managed: list[str], configured: str, live: bool, armed: bool, trading: bool = True) -> bool: + """True for a live account; SystemExit unless the login (and, to trade, the flags) allow it.""" + if not configured: + raise SystemExit("IBKR_ACCOUNT is not set (environment or .env)") + if configured not in managed: + raise SystemExit(f"the Gateway login manages {[mask(a) for a in managed]}, not {mask(configured)}") + is_live = not configured.startswith(PAPER_PREFIXES) + if not trading: + return is_live + if is_live and not live: + raise SystemExit(f"{mask(configured)} is a live account; pass --live to trade it") + if is_live and not armed: + raise SystemExit(f"{mask(configured)} is a live account and {ARMED} does not exist") + return is_live + + +def size_orders(target: dict[str, float], holdings: dict[str, int], prices: dict[str, float], + nav: float, resize_band: float, cap: float = MAX_EQUITY, + buffer: float = CASH_BUFFER) -> list[dict]: + """Whole-share trades taking ``holdings`` to ``target`` on min(nav, cap) less the buffer.""" + if missing := sorted(s for s in set(target) | set(holdings) if not prices.get(s, 0) > 0): + raise SystemExit(f"no price for {missing}") + equity = min(nav, cap) + cash = equity * (1 - buffer) - sum(q * prices[s] for s, q in holdings.items()) + return compute_trades(holdings, cash, target, prices, min_trade_value=max(50.0, 0.001 * equity), + integer_shares=True, lot_size=1, resize_band=resize_band) + + +def validate_orders(trades: list[dict], universe: set[str], holdings: dict[str, int]) -> None: + for t in trades: + q = t["shares_delta"] + if t["ticker"] not in universe: + raise SystemExit(f"{t['ticker']} is outside the strategy's universe") + if q != int(q) or q == 0: + raise SystemExit(f"{t['ticker']}: {q} is not a whole, nonzero share count") + if q < 0 and -q > holdings.get(t["ticker"], 0): + raise SystemExit(f"{t['ticker']}: selling {-q} of {holdings.get(t['ticker'], 0)} held") + + +def moc_order(delta: int, account: str, ref: str): + from ib_async import Order + if delta == 0 or delta != int(delta): + raise ValueError(f"MOC order needs a whole, nonzero quantity, got {delta}") + order = Order(action="BUY" if delta > 0 else "SELL", totalQuantity=abs(int(delta)), + orderType="MOC", tif="DAY") + order.account = account + order.orderRef = ref + return order + + +# --------------------------------------------------------------------------- +# configuration, state, logs +# --------------------------------------------------------------------------- + +def settings() -> dict: + from overnight_paper import _env + env = _env() + return {"host": env.get("IBKR_HOST", "127.0.0.1"), "port": int(env.get("IBKR_PORT", "4001")), + "client_id": int(env.get("IBKR_CLIENT_ID", "31")), "account": env.get("IBKR_ACCOUNT", "")} + + +def _state() -> dict: + path = os.path.join(OUT, "state.json") + return json.load(open(path)) if os.path.exists(path) else {} + + +def _save_state(st: dict) -> None: + os.makedirs(OUT, exist_ok=True) + tmp = os.path.join(OUT, ".state.json.tmp") + with open(tmp, "w") as fh: + json.dump(st, fh, indent=1) + os.replace(tmp, os.path.join(OUT, "state.json")) + + +def _append_jsonl(name: str, row: dict) -> None: + os.makedirs(OUT, exist_ok=True) + with open(os.path.join(OUT, name), "a") as fh: + fh.write(json.dumps(row, default=str) + "\n") + + +def now_utc() -> datetime: + return datetime.now(timezone.utc) + + +# --------------------------------------------------------------------------- +# target and prices +# --------------------------------------------------------------------------- + +def session_target() -> tuple[date, dict[str, float], list[str]]: + """The strategy's target for the close of the latest session with an open bar, as trader.py morning computes it.""" + import trader + tickers, benchmark = trader.strategy_universe(MARKET, STRATEGY) + close, open_data, close_raw = trader.load_strategy_prices(MARKET, STRATEGY, tickers, benchmark, with_open=True) + tickers = trader.filter_tradable_tickers(close, tickers) + day = open_data.index[-1] + row = trader.session_signals(MARKET, STRATEGY, tickers, pd.DatetimeIndex([day]), close_raw, open_data).loc[day] + return day.date(), {t: float(w) for t, w in row.items() if w > 1e-6}, sorted(trader.ETF_STRATEGY_UNIVERSES[STRATEGY]) + + +def monitor_target(day: date) -> dict[str, float] | None: + """The simulated account's pending target for ``day`` (written by the monitor's 09:45 morning phase).""" + if not os.path.exists(SIM_STATE): + return None + pending = json.load(open(SIM_STATE)).get("pending_trades") or {} + return pending.get("target_weights") if pending.get("date") == day.isoformat() else None + + +def sizing_prices(symbols: list[str]) -> dict[str, float]: + from overnight_paper import Alpaca, latest_prices + return latest_prices(Alpaca(), symbols) + + +# --------------------------------------------------------------------------- +# IB Gateway +# --------------------------------------------------------------------------- + +class Gateway: + def __init__(self, cfg: dict) -> None: + from ib_async import IB + self.cfg, self.ib = cfg, IB() + + async def __aenter__(self) -> "Gateway": + address = f"{self.cfg['host']}:{self.cfg['port']}" + try: + # Without raiseSyncErrors a Gateway that lost IBKR's servers connects with an + # empty portfolio, which would read as no holdings (ibkr-quant, 2026-09-29). + await self.ib.connectAsync(self.cfg["host"], self.cfg["port"], self.cfg["client_id"], timeout=15, + account=self.cfg["account"], raiseSyncErrors=True) + except ConnectionRefusedError: + raise SystemExit(f"nothing listening on {address}; start IB Gateway") from None + except TimeoutError: + raise SystemExit(f"IB Gateway on {address} did not finish the API handshake") from None + except ConnectionError as e: + raise SystemExit(f"IB Gateway on {address} failed {e}; it may have lost IBKR's servers") from None + return self + + async def __aexit__(self, *exc) -> None: + self.ib.disconnect() + + async def account(self, account: str) -> dict: + tags = {v.tag: v for v in await self.ib.accountSummaryAsync(account)} + positions, foreign = {}, [] + for p in self.ib.positions(account): + if not p.position: + continue + c = p.contract + if (c.secType, c.currency) == ("STK", "USD") and p.position == int(p.position): + positions[c.symbol] = int(p.position) + else: + foreign.append(f"{c.symbol} {c.secType} {c.currency} {p.position}") + return {"nav": float(tags["NetLiquidation"].value), "cash": float(tags["TotalCashValue"].value), + "available": float(tags["AvailableFunds"].value), "positions": positions, "foreign": foreign} + + async def our_open_orders(self, account: str) -> list: + trades = await self.ib.reqAllOpenOrdersAsync() + return [t for t in trades if t.order.account in ("", account) and t.order.orderRef.startswith(REF)] + + async def contracts(self, symbols: list[str]) -> dict: + from ib_async import Stock + qualified = await self.ib.qualifyContractsAsync(*[Stock(s, "SMART", "USD") for s in symbols]) + found = {c.symbol: c for c in qualified if c is not None and getattr(c, "conId", 0)} + if missing := sorted(set(symbols) - set(found)): + raise SystemExit(f"IBKR could not qualify {missing}") + return found + + async def confirm(self, trades: list, timeout: float = 30) -> list: + """Wait until each order is accepted or done; return the ones not accepted.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while any(t.orderStatus.status not in ACCEPTED | DONE for t in trades) and loop.time() < deadline: + await asyncio.sleep(0.1) + return [t for t in trades if t.orderStatus.status not in ACCEPTED] + + +def _errors(trade) -> str: + return "; ".join(f"{e.errorCode} {e.message}" for e in trade.log if e.errorCode) + + +def _order_row(kind: str, trade, day: date) -> dict: + return {"logged_at": now_utc().isoformat(), "kind": kind, "session": day.isoformat(), + "symbol": trade.contract.symbol, "action": trade.order.action, "qty": trade.order.totalQuantity, + "order_id": trade.order.orderId, "perm_id": trade.order.permId, "status": trade.orderStatus.status, + "errors": _errors(trade)} + + +# --------------------------------------------------------------------------- +# commands +# --------------------------------------------------------------------------- + +async def check(cfg: dict) -> None: + async with Gateway(cfg) as gw: + managed = gw.ib.managedAccounts() + acct = cfg["account"] or (managed[0] if len(managed) == 1 else "") + if acct not in managed: + raise SystemExit(f"set IBKR_ACCOUNT; the login manages {[mask(a) for a in managed]}") + a = await gw.account(acct) + print(f"account {mask(acct)} NAV ${a['nav']:,.2f} cash ${a['cash']:,.2f} available ${a['available']:,.2f}") + print(f"positions {a['positions'] or 'none'}" + (f"; other {a['foreign']}" if a["foreign"] else "")) + print(f"open orders of ours: {len(await gw.our_open_orders(acct))}") + + +async def plan(cfg: dict, what_if: bool = False, submit: bool = False, live: bool = False) -> dict: + day, target, universe = session_target() + today = datetime.now(NY).date() + if submit and day != today: + raise SystemExit(f"latest open bar is {day}, not today {today}; Yahoo has not printed today's open") + mon = monitor_target(day) + agree = mon is not None and set(mon) == set(target) and all(abs(mon[s] - target[s]) < 1e-6 for s in target) + print(f"{STRATEGY} target for the close of {day}: " + + (", ".join(f"{s} {w:.1%}" for s, w in sorted(target.items())) or "cash") + + ("" if mon is None else f" (monitor's target {'agrees' if agree else 'DIFFERS: ' + str(mon)})")) + + async with Gateway(cfg) as gw: + is_live = check_account(gw.ib.managedAccounts(), cfg["account"], live, os.path.exists(ARMED), trading=submit) + acct = cfg["account"] + a = await gw.account(acct) + held = {s: q for s, q in a["positions"].items() if s in universe} + other = a["foreign"] + [f"{s} {q}" for s, q in a["positions"].items() if s not in universe] + if other: + if submit: + raise SystemExit(f"the account holds positions outside {STRATEGY}'s universe: {other}") + print(f" ignoring positions outside the universe (plan only): {len(other)}") + prices = sizing_prices(sorted(set(target) | set(held))) + trades = size_orders(target, held, prices, a["nav"], resize_band=_resize_band()) + validate_orders(trades, set(universe), held) + print(f"account {mask(acct)}{' (live)' if is_live else ' (paper)'} NAV ${a['nav']:,.2f} " + f"sized on ${min(a['nav'], MAX_EQUITY) * (1 - CASH_BUFFER):,.2f} held {held or 'nothing'}") + for t in trades: + print(f" {t['direction']:4s} {abs(t['shares_delta']):5d} {t['ticker']:5s} ~${t['est_value']:>9,.2f} " + f"@ {t['price']:.2f} -> {t['target_shares']} shares") + if not trades: + print(" no orders") + run = {"session": day.isoformat(), "computed_at": now_utc().isoformat(), "target": target, + "monitor_target": mon, "nav": a["nav"], "held": held, "prices": prices, "trades": trades} + if what_if and trades: + contracts = await gw.contracts([t["ticker"] for t in trades]) + errors: list[str] = [] + on_error = lambda req_id, code, message, contract: errors.append(f"{code} {message}") # noqa: E731 + gw.ib.errorEvent += on_error + try: + for t in trades: + errors.clear() + st = await gw.ib.whatIfOrderAsync(contracts[t["ticker"]], moc_order(t["shares_delta"], acct, REF)) + if not hasattr(st, "initMarginChange"): # a rejected what-if returns no OrderState + t["what_if"] = {"rejected": "; ".join(errors) or "no answer"} + print(f" what-if {t['ticker']:5s}: REJECTED {t['what_if']['rejected'].replace('
', ' ')[:300]}") + continue + t["what_if"] = {k: getattr(st, k, None) for k in ("status", "initMarginChange", "maintMarginChange", + "equityWithLoanAfter", "commission", + "minCommission", "maxCommission", "warningText")} + print(f" what-if {t['ticker']:5s}: accepted, commission " + f"{_money(st.commission, st.minCommission, st.maxCommission)}, init margin change " + f"{st.initMarginChange}{', warning: ' + st.warningText if st.warningText else ''}") + finally: + gw.ib.errorEvent -= on_error + if not submit: + return run + + st = _state() + if st.get("submitted") == day.isoformat(): + raise SystemExit(f"{day}: orders already submitted") + if mine := await gw.our_open_orders(acct): + raise SystemExit(f"{len(mine)} orders of ours are still open") + if is_live and not os.path.exists(os.path.join(OUT, "prereg.json")): + with open(os.path.join(OUT, "prereg.json"), "w") as fh: + json.dump({**PREREG, "written_at": now_utc().isoformat(), "strategy": STRATEGY}, fh, indent=2) + st["submitted"] = day.isoformat() # before sending: a crash must not resend + _save_state(st) + os.makedirs(os.path.join(OUT, "runs"), exist_ok=True) + contracts = await gw.contracts([t["ticker"] for t in trades]) if trades else {} + failed = [] + for side in ("SELL", "BUY"): + batch = [gw.ib.placeOrder(contracts[t["ticker"]], moc_order(t["shares_delta"], acct, f"{REF} {day}")) + for t in trades if t["direction"] == side] + bad = await gw.confirm(batch) if batch else [] + for tr in batch: + _append_jsonl("orders.jsonl", _order_row("moc", tr, day)) + print(f" {tr.order.action:4s} {tr.order.totalQuantity:>5.0f} {tr.contract.symbol:5s} MOC -> " + f"{tr.orderStatus.status}{' (' + _errors(tr) + ')' if _errors(tr) else ''}") + failed += bad + if side == "SELL" and bad: + break # no buys on top of a sell that did not go in + run["orders"] = [o for o in (json.loads(l) for l in open(os.path.join(OUT, "orders.jsonl"))) + if o["session"] == day.isoformat()] if trades else [] + with open(os.path.join(OUT, "runs", f"{day}.json"), "w") as fh: + json.dump(run, fh, indent=1, default=str) + if failed: + raise SystemExit("not accepted: " + "; ".join(f"{t.contract.symbol} {t.orderStatus.status} {_errors(t)}" + for t in failed)) + return run + + +def _resize_band() -> float: + import trader + return trader.strategy_resize_band(STRATEGY) + + +def _money(c, lo, hi) -> str: + ok = lambda x: x is not None and math.isfinite(x) and abs(x) < 1e9 + if ok(c): + return f"${c:.2f}" + return f"${lo:.2f}-${hi:.2f}" if ok(lo) and ok(hi) else "n/a" + + +async def report(cfg: dict) -> None: + from ib_async import ExecutionFilter + async with Gateway(cfg) as gw: + acct = cfg["account"] + if acct not in gw.ib.managedAccounts(): + raise SystemExit(f"the Gateway login does not manage {mask(acct)}") + fills = await gw.ib.reqExecutionsAsync(ExecutionFilter(acctCode=acct)) + a = await gw.account(acct) + rows = [] + for f in fills: + e, cr = f.execution, f.commissionReport + if not e.orderRef.startswith(REF): + continue + commission = cr.commission if cr and math.isfinite(cr.commission) and cr.commission < 1e9 else float("nan") + rows.append({"exec_id": e.execId, "time": e.time.astimezone(NY).isoformat(), + "session": e.time.astimezone(NY).date().isoformat(), "symbol": f.contract.symbol, + "side": "BUY" if e.side == "BOT" else "SELL", "shares": e.shares, "price": e.price, + "exchange": e.exchange, "perm_id": e.permId, "order_ref": e.orderRef, "commission": commission}) + path = os.path.join(OUT, "fills.csv") + old = pd.read_csv(path) if os.path.exists(path) else pd.DataFrame() + F = pd.concat([old, pd.DataFrame(rows)], ignore_index=True) + if len(F): + F = F.drop_duplicates("exec_id", keep="last").sort_values("time") + F.to_csv(path, index=False) + today = datetime.now(NY).date().isoformat() + _append_daily(today, a) + if F.empty: + print("no fills yet") + F = pd.DataFrame(columns=["session", "symbol", "side", "shares", "price", "commission"]) + _score(F) + _compare_with_simulation() + + +def _append_daily(day: str, a: dict) -> None: + path = os.path.join(OUT, "daily.csv") + D = pd.read_csv(path) if os.path.exists(path) else pd.DataFrame() + row = pd.DataFrame([{"date": day, "nav": a["nav"], "cash": a["cash"], "positions": json.dumps(a["positions"])}]) + D = pd.concat([D[D["date"] != day] if len(D) else D, row], ignore_index=True) + D.to_csv(path, index=False) + + +def _score(F: pd.DataFrame) -> None: + """Each accepted MOC order against the closing cross; preregistered verdict on the first 20 orders. + + Orders and fills are matched on (session, symbol, side): a session sends at most + one order per symbol. An accepted order with no fill counts against the fill rate. + """ + from research.lab.auction_audit import crosses + import research.lab.fetch_options as fo + path = os.path.join(OUT, "orders.jsonl") + if not os.path.exists(path): + return + O = pd.DataFrame([json.loads(l) for l in open(path)]) + O = O[O["status"].isin(ACCEPTED)].drop_duplicates(["session", "symbol", "action"], keep="last") + settled = pd.Timestamp.now(tz=NY) - pd.Timedelta(hours=16, minutes=20) # free SIP data lags 15 min + O = O[pd.to_datetime(O["session"]).dt.tz_localize(NY) <= settled] + if O.empty: + return + f = F.assign(value=F["shares"] * F["price"]).groupby(["session", "symbol", "side"]) \ + .agg(filled=("shares", "sum"), value=("value", "sum"), commission=("commission", "sum")) + g = O.join(f, on=["session", "symbol", "action"])[["session", "symbol", "action", "qty", "filled", "value", + "commission"]].fillna({"filled": 0.0}) + g["avg"] = g["value"] / g["filled"] + api, ref = fo.Api(min_interval=0.4), {} + for day, h in g.groupby("session"): + for sym, v in crosses(api, sorted(h["symbol"].unique()), pd.Timestamp(day), "15:59:59", "16:10:00", "6").items(): + ref[(day, sym)] = v["p"] + g["cross"] = [ref.get((d, sym)) for d, sym in zip(g["session"], g["symbol"])] + g["slip_bp"] = g["action"].map({"BUY": 1, "SELL": -1}) * (g["avg"] / g["cross"] - 1) * 1e4 + g["match"] = (g["avg"] - g["cross"]).abs() <= PREREG["rules"]["cross_match"]["tolerance_usd"] + g["commission_bp"] = g["commission"] / g["value"] * 1e4 + g = g.sort_values(["session", "symbol"]).reset_index(drop=True) + g.to_csv(os.path.join(OUT, "scored.csv"), index=False) + print(g.tail(10)[["session", "symbol", "action", "qty", "filled", "avg", "cross", "slip_bp", "commission_bp"]] + .to_string(index=False, float_format=lambda x: f"{x:.3f}")) + + def measure(x: pd.DataFrame) -> dict: + y = x[x["filled"] > 0].dropna(subset=["cross"]) + return {"cost_per_side_bp": (y["slip_bp"] * y["value"]).sum() / y["value"].sum() if len(y) else float("nan"), + "fill_rate": x["filled"].sum() / x["qty"].sum(), + "cross_match": y["match"].mean() if len(y) else float("nan"), + "commission_bp": y["commission"].sum() / y["value"].sum() * 1e4 if len(y) else float("nan")} + + m = measure(g) + print(f"orders {len(g)}: cost {m['cost_per_side_bp']:+.2f} bp a side (ex commission), fill rate " + f"{m['fill_rate']:.1%}, cross match {m['cross_match']:.0%}, commission {m['commission_bp']:.2f} bp a side") + verdict_path = os.path.join(OUT, "verdict.json") + if len(g) < 20 or not os.path.exists(os.path.join(OUT, "prereg.json")) or os.path.exists(verdict_path): + return + v, r = measure(g.head(20)), PREREG["rules"] + grades = { + "cost_per_side_bp": "PASS" if v["cost_per_side_bp"] <= r["cost_per_side_bp"]["PASS"] + else "MEASURED" if v["cost_per_side_bp"] <= r["cost_per_side_bp"]["MEASURED"] else "STOP", + "fill_rate": "PASS" if v["fill_rate"] >= r["fill_rate"]["PASS"] + else "MEASURED" if v["fill_rate"] >= r["fill_rate"]["MEASURED"] else "STOP", + "cross_match": "PASS" if v["cross_match"] >= r["cross_match"]["PASS"] + else "MEASURED" if v["cross_match"] >= r["cross_match"]["MEASURED"] else "STOP", + } + decision = max(grades.values(), key=["PASS", "MEASURED", "STOP"].index) + with open(verdict_path, "w") as fh: + json.dump({"decided_at": now_utc().isoformat(), "orders": 20, "values": v, "grades": grades, + "decision": decision}, fh, indent=2) + print(f"verdict on the first 20 orders: {decision} {grades}") + + +def _compare_with_simulation() -> None: + path = os.path.join(OUT, "daily.csv") + if not (os.path.exists(path) and os.path.exists(SIM_STATE)): + return + live = pd.read_csv(path).set_index("date")["nav"] + sim = pd.Series(json.load(open(SIM_STATE))["daily_equity"]) + both = pd.concat({"live": live, "sim": sim}, axis=1).dropna() + if len(both) < 2: + return + r = both / both.iloc[0] - 1 + print(f"since {both.index[0]}: live {r['live'].iloc[-1]:+.2%}, simulated account {r['sim'].iloc[-1]:+.2%}, " + f"gap {r['live'].iloc[-1] - r['sim'].iloc[-1]:+.2%} over {len(both) - 1} sessions") + + +# --------------------------------------------------------------------------- +# daemon +# --------------------------------------------------------------------------- + +def daemon(cfg: dict, live: bool) -> None: + from overnight_paper import Alpaca, sessions + cl = Alpaca() + print(f"{now_utc():%Y-%m-%d %H:%M} UTC ibkr_trader daemon started ({STRATEGY})", flush=True) + done: set[tuple[str, str]] = set() + while True: + try: + now = now_utc() + cal = sessions(cl, (now - timedelta(days=1)).date(), (now + timedelta(days=7)).date()) + for d, row in cal.iterrows(): + key = d.isoformat() + if ("submit", key) not in done and row["close"] - SUBMIT_BEFORE_CLOSE <= now < row["close"] - SUBMIT_CUTOFF: + # Retried each minute inside the window until the session is marked submitted, + # which happens just before the first order goes out: nothing is ever sent twice. + try: + if _state().get("submitted") != key: + asyncio.run(plan(cfg, submit=True, live=live)) + finally: + if _state().get("submitted") == key: + done.add(("submit", key)) + if ("report", key) not in done and row["close"] + REPORT_AFTER_CLOSE <= now < row["close"] + timedelta(hours=7): + done.add(("report", key)) + asyncio.run(report(cfg)) + except SystemExit as exc: + print(f"{now_utc():%Y-%m-%d %H:%M} UTC stopped: {exc}", flush=True) + except Exception as exc: # noqa: BLE001 - keep the loop alive + print(f"{now_utc():%Y-%m-%d %H:%M} UTC error: {type(exc).__name__}: {str(exc)[:300]}", flush=True) + time.sleep(60) + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("what", choices=["check", "plan", "submit", "report", "daemon"]) + ap.add_argument("--what-if", action="store_true", help="plan: have IBKR validate and price each order") + ap.add_argument("--live", action="store_true", help="allow a live (non-DU) account; also needs data/ibkr_live/ARMED") + a = ap.parse_args() + cfg = settings() + if a.what == "check": + asyncio.run(check(cfg)) + elif a.what == "plan": + asyncio.run(plan(cfg, what_if=a.what_if)) + elif a.what == "submit": + from overnight_paper import Alpaca, sessions + today = datetime.now(NY).date() + cal = sessions(Alpaca(), today, today) + if today not in cal.index: + raise SystemExit(f"{today}: no session") + close = cal.loc[today, "close"] + if not close - SUBMIT_BEFORE_CLOSE <= now_utc() <= close - SUBMIT_CUTOFF: + raise SystemExit(f"outside the MOC window ({(close - SUBMIT_BEFORE_CLOSE).astimezone(NY):%H:%M}-" + f"{(close - SUBMIT_CUTOFF).astimezone(NY):%H:%M} ET)") + asyncio.run(plan(cfg, submit=True, live=a.live)) + elif a.what == "report": + asyncio.run(report(cfg)) + else: + daemon(cfg, a.live) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index 1c7e7fb..a809a2a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,7 @@ description = "Add your description here" readme = "README.md" requires-python = ">=3.12" dependencies = [ + "ib-async>=2.1.0", "lxml>=6.0.2", "matplotlib>=3.10.8", "numpy>=2.4.0", diff --git a/tests/test_ibkr_trader.py b/tests/test_ibkr_trader.py new file mode 100644 index 0000000..e5b9942 --- /dev/null +++ b/tests/test_ibkr_trader.py @@ -0,0 +1,74 @@ +"""Guards of the IBKR order layer (ibkr_trader.py): which accounts it may trade, +how much it sizes, and which orders it will send.""" + +import unittest + +import ibkr_trader as it + + +class AccountGuard(unittest.TestCase): + def test_paper_account_trades_without_flags(self): + self.assertFalse(it.check_account(["DU123"], "DU123", live=False, armed=False)) + + def test_account_must_be_managed_by_the_login(self): + with self.assertRaises(SystemExit): + it.check_account(["DU123"], "U999", live=True, armed=True) + with self.assertRaises(SystemExit): + it.check_account(["U999"], "", live=True, armed=True) + + def test_live_account_needs_flag_and_armed_file(self): + with self.assertRaises(SystemExit): + it.check_account(["U999"], "U999", live=False, armed=True) + with self.assertRaises(SystemExit): + it.check_account(["U999"], "U999", live=True, armed=False) + self.assertTrue(it.check_account(["U999"], "U999", live=True, armed=True)) + + def test_read_only_use_skips_the_trading_flags(self): + self.assertTrue(it.check_account(["U999"], "U999", live=False, armed=False, trading=False)) + + def test_live_account_number_is_masked(self): + self.assertEqual(it.mask("U1234567"), "U***567") + self.assertEqual(it.mask("DU123"), "DU123") + + +class Sizing(unittest.TestCase): + def test_orders_are_sized_on_the_capped_equity(self): + trades = it.size_orders({"TQQQ": 1.0}, {}, {"TQQQ": 100.0}, nav=972_000.0, resize_band=0.1) + self.assertEqual([t["ticker"] for t in trades], ["TQQQ"]) + self.assertLessEqual(trades[0]["shares_delta"] * 100.0, it.MAX_EQUITY * (1 - it.CASH_BUFFER)) + self.assertGreater(trades[0]["shares_delta"], 95) + + def test_a_switch_sells_the_old_leg_and_buys_the_new(self): + trades = it.size_orders({"GLD": 1.0}, {"TQQQ": 100}, {"TQQQ": 100.0, "GLD": 300.0}, nav=10_000.0, + resize_band=0.1) + by = {t["ticker"]: t["shares_delta"] for t in trades} + self.assertEqual(by["TQQQ"], -100) + self.assertLessEqual(by["GLD"] * 300.0, 10_000.0 * (1 - it.CASH_BUFFER)) + self.assertGreaterEqual(by["GLD"], 32) + + def test_missing_price_stops_the_run(self): + with self.assertRaises(SystemExit): + it.size_orders({"TQQQ": 1.0}, {"GLD": 3}, {"TQQQ": 100.0}, nav=10_000.0, resize_band=0.1) + + +class OrderWhitelist(unittest.TestCase): + def test_only_universe_symbols_and_whole_shares(self): + with self.assertRaises(SystemExit): + it.validate_orders([{"ticker": "AAPL", "shares_delta": 1}], {"TQQQ"}, {}) + with self.assertRaises(SystemExit): + it.validate_orders([{"ticker": "TQQQ", "shares_delta": 1.5}], {"TQQQ"}, {}) + + def test_never_sells_more_than_held(self): + with self.assertRaises(SystemExit): + it.validate_orders([{"ticker": "TQQQ", "shares_delta": -5}], {"TQQQ"}, {"TQQQ": 3}) + it.validate_orders([{"ticker": "TQQQ", "shares_delta": -3}], {"TQQQ"}, {"TQQQ": 3}) + + def test_orders_are_market_on_close(self): + o = it.moc_order(-7, "DU123", "ibkr_trader 2026-10-13") + self.assertEqual((o.action, o.totalQuantity, o.orderType, o.tif, o.account), ("SELL", 7, "MOC", "DAY", "DU123")) + with self.assertRaises(ValueError): + it.moc_order(0, "DU123", "x") + + +if __name__ == "__main__": + unittest.main() diff --git a/uv.lock b/uv.lock index 387277b..d5833fd 100644 --- a/uv.lock +++ b/uv.lock @@ -2,6 +2,18 @@ version = 1 revision = 3 requires-python = ">=3.12" +[[package]] +name = "aeventkit" +version = "2.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/5c/8c/c08db1a1910f8d04ec6a524de522edd0bac181bdf94dbb01183f7685cd77/aeventkit-2.1.0.tar.gz", hash = "sha256:4e7d81bb0a67227121da50a23e19e5bbf13eded541a9f4857eeb6b7b857b738a", size = 24703, upload-time = "2025-06-22T15:54:03.961Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8d/8c/2a4b912b1afa201b25bdd0f5bccf96d5a8b5dccb6131316a8dd2d9cabcc1/aeventkit-2.1.0-py3-none-any.whl", hash = "sha256:962d43f79e731ac43527f2d0defeed118e6dbaa85f1487f5667540ebb8f00729", size = 26678, upload-time = "2025-06-22T15:54:02.141Z" }, +] + [[package]] name = "beautifulsoup4" version = "4.14.3" @@ -284,6 +296,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/38/74/f94141b38a51a553efef7f510fc213894161ae49b88bffd037f8d2a7cb2f/frozendict-2.4.7-py3-none-any.whl", hash = "sha256:972af65924ea25cf5b4d9326d549e69a9a4918d8a76a9d3a7cd174d98b237550", size = 16264, upload-time = "2025-11-11T22:40:12.836Z" }, ] +[[package]] +name = "ib-async" +version = "2.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "aeventkit" }, + { name = "nest-asyncio" }, + { name = "tzdata" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/30/4d/dfc1da8224c3ffcdcd668da7283c4e5f14239a07f83ea66af99700296fc3/ib_async-2.1.0.tar.gz", hash = "sha256:6a03a87d6c06acacb0217a5bea60a8a168ecd5b5a7e86e1c73678d5b48cbc796", size = 87678, upload-time = "2025-12-08T01:42:32.004Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/80/e7/8f33801788c66f15e9250957ff7f53a8000843f79af1a3ed7a96def0e96b/ib_async-2.1.0-py3-none-any.whl", hash = "sha256:f6d8b991bdbd6dd38e700c61b3dced06ebe0f14be4e5263e2ef10ba10b88d434", size = 88876, upload-time = "2025-12-08T01:42:30.883Z" }, +] + [[package]] name = "idna" version = "3.11" @@ -505,6 +531,15 @@ version = "0.0.12" source = { registry = "https://pypi.org/simple" } sdist = { url = "https://files.pythonhosted.org/packages/17/0d/74f0293dfd7dcc3837746d0138cbedd60b31701ecc75caec7d3f281feba0/multitasking-0.0.12.tar.gz", hash = "sha256:2fba2fa8ed8c4b85e227c5dd7dc41c7d658de3b6f247927316175a57349b84d1", size = 19984, upload-time = "2025-07-20T21:27:51.636Z" } +[[package]] +name = "nest-asyncio" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/83/f8/51569ac65d696c8ecbee95938f89d4abf00f47d58d48f6fbabfe8f0baefe/nest_asyncio-1.6.0.tar.gz", hash = "sha256:6f172d5449aca15afd6c646851f4e31e02c598d553a667e38cafa997cfec55fe", size = 7418, upload-time = "2024-01-21T14:25:19.227Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a0/c4/c2971a3ba4c6103a3d10c4b0f24f461ddc027f0f09763220cf35ca1401b3/nest_asyncio-1.6.0-py3-none-any.whl", hash = "sha256:87af6efd6b5e897c81050477ef65c62e2b2f35d51703cae01aff2905b1852e1c", size = 5195, upload-time = "2024-01-21T14:25:17.223Z" }, +] + [[package]] name = "numpy" version = "2.4.0" @@ -765,6 +800,7 @@ name = "quant-trading" version = "0.1.0" source = { virtual = "." } dependencies = [ + { name = "ib-async" }, { name = "lxml" }, { name = "matplotlib" }, { name = "numpy" }, @@ -776,6 +812,7 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "ib-async", specifier = ">=2.1.0" }, { name = "lxml", specifier = ">=6.0.2" }, { name = "matplotlib", specifier = ">=3.10.8" }, { name = "numpy", specifier = ">=2.4.0" },