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" },