Daily UpCap+MomGap trading on an IBKR paper account

Connects to IB Gateway through ib_async and trades the S&P 500 with the
UpCap+MomGap factor combo ported from quant-trading's
fc_up_cap_mom_gap_daily: signal on opens through today's open, equal
weight top 10, MOC orders filled at today's close, 10% resize band.

- make universe/bars/check/plan/submit/fills for the daily loop; refuses
  non-paper accounts unless allow_live, open orders, stale bars, and
  orders outside 09:31-15:45 ET
- make backtest replays the live loop day by day with plan_orders and
  IBKR Fixed commissions; selection uses each day's index members
- make bars fails fast when the Gateway's historical farm is down

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/ibkr_quant/backtest.py

@@ -0,0 +1,118 @@
+"""Day-by-day replay of the live loop: targets from opens through day d, orders
+sized at d's open by `plan_orders`, filled at d's close, commissions charged per order."""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+import numpy as np
+import pandas as pd
+
+from .rebalance import plan_orders
+from .strategy import WARMUP, select_top_n
+
+# A held name without a price for this many sessions (delisted, acquired) is
+# sold at its last close.
+DELIST_AFTER = 5
+
+
+def ibkr_fixed_fee(shares: float, price: float) -> float:
+    # IBKR Pro Fixed, US stocks: $0.005/share, min $1.00, max 1% of trade value.
+    return min(max(1.0, 0.005 * shares), 0.01 * shares * price)
+
+
+def flat_fee(amount: float) -> Callable[[float, float], float]:
+    return lambda shares, price: amount
+
+
+@dataclass(frozen=True)
+class RunConfig:
+    top_n: int
+    rebalance_every: int
+    capital: float
+    fee: Callable[[float, float], float]
+    cash_buffer: float = 0.02
+    resize_band: float = 0.10
+    min_trade_value: float = 50.0
+    min_trade_fraction: float = 0.001
+    # Picks target weights from (today's signal, currently held names); default: top_n equal weight.
+    select: Callable[[pd.Series, set[str]], dict[str, float]] | None = None
+
+
+@dataclass(frozen=True)
+class RunResult:
+    equity: pd.Series  # NAV at each close
+    turnover: pd.Series  # traded value / NAV per day
+    fees: float
+    orders: int
+
+
+def run(
+    signal: pd.DataFrame, opens: pd.DataFrame, closes: pd.DataFrame, cfg: RunConfig, start: str | None = None
+) -> RunResult:
+    """`signal` must be causal: row d uses opens through d only."""
+    raw_closes, closes = closes, closes.ffill()
+    days = signal.index[WARMUP:]
+    if start:
+        days = days[days >= start]
+    positions: dict[str, float] = {}
+    cash = cfg.capital
+    targets: dict[str, float] = {}
+    equity, turnover = {}, {}
+    fees, orders = 0.0, 0
+    unpriced = raw_closes.isna().rolling(DELIST_AFTER).sum() >= DELIST_AFTER
+    for i, day in enumerate(days):
+        open_px = opens.loc[day]
+        close_px = closes.loc[day]
+        for s in [s for s in positions if unpriced.at[day, s]]:
+            cost = cfg.fee(abs(positions[s]), close_px[s])
+            cash += positions.pop(s) * close_px[s] - cost
+            fees += cost
+            orders += 1
+        if i % cfg.rebalance_every == 0:
+            targets = (
+                cfg.select(signal.loc[day], set(positions)) if cfg.select else select_top_n(signal.loc[day], cfg.top_n)
+            )
+        # A name without an open today is marked at its last close and not traded.
+        marks = {s: open_px[s] if pd.notna(open_px[s]) else close_px[s] for s in set(targets) | set(positions)}
+        nav_open = cash + sum(q * marks[s] for s, q in positions.items())
+        frozen = {s for s in marks if pd.isna(open_px[s])}
+        planned = plan_orders(
+            {s: w for s, w in targets.items() if s not in frozen},
+            {s: q for s, q in positions.items() if s not in frozen},
+            marks,
+            nav_open * (1 - cfg.cash_buffer),
+            max(cfg.min_trade_value, cfg.min_trade_fraction * nav_open),
+            cfg.resize_band,
+        )
+        traded = 0.0
+        for o in planned:
+            fill = close_px[o.symbol]
+            cost = cfg.fee(abs(o.delta), fill)
+            cash -= o.delta * fill + cost
+            positions[o.symbol] = positions.get(o.symbol, 0.0) + o.delta
+            if positions[o.symbol] == 0:
+                del positions[o.symbol]
+            traded += abs(o.delta) * fill
+            fees += cost
+            orders += 1
+        nav = cash + sum(q * close_px[s] for s, q in positions.items())
+        equity[day] = nav
+        turnover[day] = traded / nav
+    return RunResult(pd.Series(equity), pd.Series(turnover), fees, orders)
+
+
+def metrics(result: RunResult, capital: float) -> dict[str, float]:
+    eq = result.equity
+    ret = eq.pct_change().fillna(eq.iloc[0] / capital - 1)
+    half = len(eq) // 2
+    return {
+        "total": eq.iloc[-1] / capital - 1,
+        "h1": eq.iloc[half - 1] / capital - 1,
+        "h2": eq.iloc[-1] / eq.iloc[half - 1] - 1,
+        "vol": ret.std() * np.sqrt(252),
+        "sharpe": ret.mean() / ret.std() * np.sqrt(252),
+        "max_dd": (eq / eq.cummax() - 1).min(),
+        "turnover_day": result.turnover.mean(),
+        "fees_pct": result.fees / capital,
+        "orders": result.orders,
+    }

src/ibkr_quant/broker.py

@@ -0,0 +1,177 @@
+import asyncio
+import math
+from dataclasses import dataclass
+from zoneinfo import ZoneInfo
+
+import pandas as pd
+from ib_async import IB, Contract, LimitOrder, Order, Stock, Trade
+
+from .config import IBConfig
+
+NEW_YORK = ZoneInfo("America/New_York")
+PAPER_ACCOUNT_PREFIXES = ("DU", "DF")
+# TWS allows 50 simultaneous historical requests; stay well below it.
+MAX_CONCURRENT_HISTORICAL = 20
+
+
+@dataclass(frozen=True)
+class Holding:
+    symbol: str
+    sec_type: str
+    currency: str
+    quantity: float
+    market_price: float
+    market_value_usd: float
+
+
+@dataclass(frozen=True)
+class AccountState:
+    account: str
+    nav_usd: float
+    cash_usd: float
+    holdings: list[Holding]
+
+
+class Broker:
+    def __init__(self, cfg: IBConfig):
+        self.cfg = cfg
+        self.ib = IB()
+        self.account = ""
+
+    async def __aenter__(self) -> "Broker":
+        try:
+            await self.ib.connectAsync(
+                self.cfg.host, self.cfg.port, self.cfg.client_id, timeout=15, account=self.cfg.account
+            )
+        except ConnectionRefusedError:
+            raise SystemExit(
+                f"nothing listening on {self.cfg.host}:{self.cfg.port}; start IB Gateway/TWS and enable the API port"
+            ) from None
+        accounts = self.ib.managedAccounts()
+        if self.cfg.account:
+            self.account = self.cfg.account
+        elif len(accounts) == 1:
+            self.account = accounts[0]
+        else:
+            self.ib.disconnect()
+            raise RuntimeError(f"login manages {accounts}; set ib.account in config.toml")
+        if not self.account.startswith(PAPER_ACCOUNT_PREFIXES) and not self.cfg.allow_live:
+            self.ib.disconnect()
+            raise RuntimeError(f"{self.account} is not a paper account; set ib.allow_live = true to trade it")
+        self.ib.reqMarketDataType(self.cfg.market_data_type)
+        return self
+
+    async def __aexit__(self, *exc) -> None:
+        self.ib.disconnect()
+
+    def _usd_per_unit(self, base_currency: str) -> dict[str, float]:
+        # ExchangeRate tags are "base currency per 1 unit of <currency>".
+        rates = {
+            v.currency: float(v.value)
+            for v in self.ib.accountValues(self.account)
+            if v.tag == "ExchangeRate" and v.currency not in ("", "BASE")
+        }
+        rates.setdefault(base_currency, 1.0)
+        if "USD" not in rates:
+            raise RuntimeError(f"no USD exchange rate for base currency {base_currency}")
+        return {currency: rate / rates["USD"] for currency, rate in rates.items()}
+
+    async def account_state(self) -> AccountState:
+        summary = {v.tag: v for v in await self.ib.accountSummaryAsync(self.account)}
+        nav = summary["NetLiquidation"]
+        cash = summary["TotalCashValue"]
+        to_usd = self._usd_per_unit(nav.currency)
+        holdings = [
+            Holding(
+                symbol=item.contract.symbol,
+                sec_type=item.contract.secType,
+                currency=item.contract.currency,
+                quantity=item.position,
+                market_price=item.marketPrice,
+                market_value_usd=item.marketValue * to_usd[item.contract.currency],
+            )
+            for item in self.ib.portfolio(self.account)
+            if item.position
+        ]
+        return AccountState(
+            account=self.account,
+            nav_usd=float(nav.value) * to_usd[nav.currency],
+            cash_usd=float(cash.value) * to_usd[cash.currency],
+            holdings=holdings,
+        )
+
+    async def qualify_stocks(self, symbols: list[str]) -> tuple[dict[str, Contract], list[str]]:
+        contracts = [Stock(s, "SMART", "USD") for s in symbols]
+        qualified = await self.ib.qualifyContractsAsync(*contracts)
+        found = {s: c for s, c in zip(symbols, qualified) if isinstance(c, Contract)}
+        return found, [s for s in symbols if s not in found]
+
+    async def daily_bars(
+        self, contracts: dict[str, Contract], duration: str, timeout: float = 120
+    ) -> tuple[pd.DataFrame, list[str]]:
+        """Split/dividend-adjusted RTH daily bars; during a session the last bar is today's, still forming."""
+        semaphore = asyncio.Semaphore(MAX_CONCURRENT_HISTORICAL)
+
+        async def fetch(symbol: str, contract: Contract):
+            async with semaphore:
+                bars = await self.ib.reqHistoricalDataAsync(
+                    contract, "", duration, "1 day", "ADJUSTED_LAST", True, timeout=timeout
+                )
+            return symbol, bars
+
+        results = await asyncio.gather(*(fetch(s, c) for s, c in contracts.items()))
+        rows, failed = [], []
+        for symbol, bars in results:
+            if not bars:
+                failed.append(symbol)
+                continue
+            for bar in bars:
+                rows.append((bar.date, symbol, bar.open, bar.high, bar.low, bar.close, bar.volume))
+        frame = pd.DataFrame(rows, columns=["date", "symbol", "open", "high", "low", "close", "volume"])
+        return frame, sorted(failed)
+
+    async def snapshot_prices(self, contracts: dict[str, Contract]) -> dict[str, float]:
+        tickers = await self.ib.reqTickersAsync(*contracts.values())
+        prices = {}
+        for symbol, ticker in zip(contracts, tickers):
+            price = ticker.marketPrice()
+            if math.isnan(price):
+                price = ticker.close
+            if price > 0:
+                prices[symbol] = price
+        return prices
+
+    async def open_trades(self) -> list[Trade]:
+        trades = await self.ib.reqAllOpenOrdersAsync()
+        return [t for t in trades if t.order.account in ("", self.account)]
+
+    def place(self, contract: Contract, delta: float, order_type: str, limit_price: float | None, ref: str) -> Trade:
+        action = "BUY" if delta > 0 else "SELL"
+        quantity = abs(delta)
+        if order_type == "MOC":
+            order = Order(action=action, totalQuantity=quantity, orderType="MOC", tif="DAY")
+        else:
+            order = LimitOrder(action, quantity, limit_price, tif="DAY")
+        order.account = self.account
+        order.orderRef = ref
+        return self.ib.placeOrder(contract, order)
+
+    async def executions(self) -> pd.DataFrame:
+        fills = await self.ib.reqExecutionsAsync()
+        return pd.DataFrame(
+            [
+                {
+                    "time": f.execution.time.astimezone(NEW_YORK),
+                    "symbol": f.contract.symbol,
+                    "side": f.execution.side,
+                    "shares": f.execution.shares,
+                    "price": f.execution.price,
+                    "commission": f.commissionReport.commission,
+                    "order_id": f.execution.orderId,
+                    "perm_id": f.execution.permId,
+                    "order_ref": f.execution.orderRef,
+                }
+                for f in fills
+                if f.execution.acctNumber == self.account
+            ]
+        )

src/ibkr_quant/cli.py

@@ -0,0 +1,229 @@
+import argparse
+import asyncio
+import dataclasses
+import datetime as dt
+import json
+import logging
+from pathlib import Path
+
+import pandas as pd
+
+from .broker import NEW_YORK, Broker
+from .config import Config, load_config
+from .rebalance import plan_orders
+from . import backtest
+from .strategy import signal_up_cap_mom_gap, up_cap_mom_gap_weights
+from .universe import fetch_sp500, load_universe
+
+# MOC orders must reach NYSE by 15:50 ET; keep a margin.
+ORDER_WINDOW = (dt.time(9, 31), dt.time(15, 45))
+
+
+def update_universe(cfg: Config) -> None:
+    table = fetch_sp500()
+    cfg.data.universe.parent.mkdir(parents=True, exist_ok=True)
+    table.to_csv(cfg.data.universe, index=False)
+    print(f"{len(table)} symbols -> {cfg.data.universe}")
+
+
+async def check(cfg: Config) -> None:
+    async with Broker(cfg.ib) as broker:
+        state = await broker.account_state()
+        print(f"account {state.account}  NAV ${state.nav_usd:,.2f}  cash ${state.cash_usd:,.2f}")
+        holdings = pd.DataFrame([dataclasses.asdict(h) for h in state.holdings])
+        print(holdings.to_string(index=False) if len(holdings) else "no positions")
+        trades = await broker.open_trades()
+        print(f"{len(trades)} open orders")
+        for t in trades:
+            print(f"  {t.contract.symbol} {t.order.action} {t.order.totalQuantity} {t.order.orderType} {t.orderStatus.status}")
+        contracts, _ = await broker.qualify_stocks(["SPY"])
+        prices = await broker.snapshot_prices(contracts)
+        if "SPY" in prices:
+            print(f"SPY snapshot {prices['SPY']:.2f} (market_data_type={cfg.ib.market_data_type})")
+        else:
+            print("no SPY price: check the market data subscription or set ib.market_data_type = 3")
+
+
+async def update_bars(cfg: Config) -> None:
+    symbols = load_universe(cfg.data.universe)
+    async with Broker(cfg.ib) as broker:
+        contracts, unknown = await broker.qualify_stocks(symbols)
+        # Fail fast when the historical farm is down instead of timing out every symbol.
+        probe = dict([next(iter(contracts.items()))])
+        if (await broker.daily_bars(probe, "5 D", timeout=30))[0].empty:
+            raise SystemExit(f"no historical data for {next(iter(probe))}; check the Gateway's HMDS farm status")
+        bars, failed = await broker.daily_bars(contracts, cfg.data.bars_duration)
+    if bars.empty:
+        raise SystemExit(f"no bars for any of {len(contracts)} symbols; kept existing {cfg.data.bars}")
+    cfg.data.bars.parent.mkdir(parents=True, exist_ok=True)
+    bars.to_csv(cfg.data.bars, index=False)
+    print(f"{bars['symbol'].nunique()} symbols, {bars['date'].min()} .. {bars['date'].max()} -> {cfg.data.bars}")
+    if unknown:
+        print(f"unqualified contracts ({len(unknown)}): {unknown}")
+    if failed:
+        print(f"no bars ({len(failed)}): {failed}")
+
+
+def load_opens(path: Path, today: dt.date) -> pd.DataFrame:
+    bars = pd.read_csv(path, parse_dates=["date"])
+    opens = bars.pivot(index="date", columns="symbol", values="open").sort_index()
+    if opens.index[-1].date() != today:
+        raise SystemExit(f"bars end {opens.index[-1].date()}, need today's open ({today}); run `make bars` after 09:30 ET")
+    return opens
+
+
+async def rebalance(cfg: Config, submit: bool) -> None:
+    ex = cfg.execution
+    now = dt.datetime.now(NEW_YORK)
+    opens = load_opens(cfg.data.bars, now.date())
+    weights = up_cap_mom_gap_weights(opens, cfg.strategy.top_n)
+    universe = set(load_universe(cfg.data.universe))
+    run_dir = cfg.root / "runs" / now.strftime("%Y%m%d-%H%M%S")
+
+    async with Broker(cfg.ib) as broker:
+        state = await broker.account_state()
+        managed = [h for h in state.holdings if h.sec_type == "STK" and h.currency == "USD" and h.symbol in universe]
+        unmanaged_value = sum(h.market_value_usd for h in state.holdings if h not in managed)
+        investable = state.nav_usd * (1 - ex.cash_buffer) - unmanaged_value
+        # Sized at today's open, as the source strategy does; filled at the close.
+        orders = plan_orders(
+            weights,
+            {h.symbol: h.quantity for h in managed},
+            opens.iloc[-1].dropna().to_dict(),
+            investable,
+            max(ex.min_trade_value, ex.min_trade_fraction * state.nav_usd),
+            ex.resize_band,
+        )
+
+        run_dir.mkdir(parents=True)
+        plan = pd.DataFrame([dataclasses.asdict(o) for o in orders], columns=["symbol", "current", "target", "delta", "price"])
+        plan["weight"] = plan["symbol"].map(weights).fillna(0.0)
+        plan.to_csv(run_dir / "plan.csv", index=False)
+        (run_dir / "account.json").write_text(json.dumps(dataclasses.asdict(state), indent=2))
+        print(f"account {state.account}  NAV ${state.nav_usd:,.2f}  investable ${investable:,.2f}  signal on opens through {opens.index[-1].date()}")
+        print(plan.to_string(index=False) if len(plan) else "no orders")
+        if not submit or not orders:
+            print(f"dry run -> {run_dir}")
+            return
+        if not ORDER_WINDOW[0] <= now.time() <= ORDER_WINDOW[1]:
+            raise SystemExit(f"{now:%H:%M} ET is outside the order window {ORDER_WINDOW[0]}-{ORDER_WINDOW[1]}")
+
+        if open_trades := await broker.open_trades():
+            raise SystemExit(f"{len(open_trades)} open orders already exist; cancel or let them finish first")
+        contracts, unknown = await broker.qualify_stocks([o.symbol for o in orders])
+        if unknown:
+            raise SystemExit(f"unqualified contracts: {unknown}")
+        limits: dict[str, float] = {}
+        if ex.order_type == "LMT":
+            snapshot = await broker.snapshot_prices(contracts)
+            if missing := [o.symbol for o in orders if o.symbol not in snapshot]:
+                raise SystemExit(f"no snapshot price for {missing}")
+            offset = ex.limit_offset_bps / 1e4
+            limits = {
+                o.symbol: round(snapshot[o.symbol] * (1 + offset if o.delta > 0 else 1 - offset), 2)
+                for o in orders
+            }
+
+        trades = [
+            broker.place(contracts[o.symbol], o.delta, ex.order_type, limits.get(o.symbol), f"ibkr-quant {run_dir.name}")
+            for o in orders
+        ]
+        await asyncio.sleep(3)  # let TWS acknowledge before recording status
+        submitted = pd.DataFrame(
+            [
+                {
+                    "symbol": t.contract.symbol,
+                    "action": t.order.action,
+                    "quantity": t.order.totalQuantity,
+                    "order_type": ex.order_type,
+                    "limit": limits.get(t.contract.symbol),
+                    "order_id": t.order.orderId,
+                    "perm_id": t.order.permId,
+                    "status": t.orderStatus.status,
+                }
+                for t in trades
+            ]
+        )
+        submitted.to_csv(run_dir / "orders.csv", index=False)
+        print(submitted.to_string(index=False))
+        print(f"submitted {len(trades)} orders -> {run_dir}")
+
+
+def run_backtest(cfg: Config, top_ns: list[int], everies: list[int], capitals: list[float], fee: str) -> None:
+    bars = pd.read_csv(cfg.data.bars, parse_dates=["date"])
+    opens = bars.pivot(index="date", columns="symbol", values="open").sort_index()
+    closes = bars.pivot(index="date", columns="symbol", values="close").sort_index()
+    # As live: pick among the index members of each day, scoring them on their
+    # full price history. Only today's members are known, so names removed from
+    # the index are missing (survivorship bias remains).
+    added = pd.read_csv(cfg.data.universe, parse_dates=["added"]).set_index("symbol")["added"]
+    member = pd.DataFrame(opens.index.values[:, None] >= added.reindex(opens.columns).values[None, :],
+                          index=opens.index, columns=opens.columns)
+    signal = signal_up_cap_mom_gap(opens, member)
+    fee_model = backtest.ibkr_fixed_fee if fee == "ibkr-fixed" else backtest.flat_fee(2.0)
+    rows = []
+    for capital in capitals:
+        for every in everies:
+            for top_n in top_ns:
+                run_cfg = backtest.RunConfig(top_n, every, capital, fee_model)
+                result = backtest.run(signal, opens, closes, run_cfg)
+                rows.append({"capital": capital, "every": every, "top_n": top_n} | backtest.metrics(result, capital))
+    days = signal.index[backtest.WARMUP:]
+    # Same-universe baseline: equal weight, rebalanced at every close, no costs.
+    ew = closes.ffill().pct_change().shift(-1).loc[days[:-1]].mean(axis=1)
+    table = pd.DataFrame(rows)
+    path = cfg.root / "runs" / f"backtest-{dt.datetime.now(NEW_YORK):%Y%m%d-%H%M%S}.csv"
+    path.parent.mkdir(exist_ok=True)
+    table.to_csv(path, index=False)
+    pct = ["total", "h1", "h2", "vol", "max_dd", "turnover_day", "fees_pct"]
+    print(f"{days[0].date()} .. {days[-1].date()} ({len(days)} sessions), fee={fee}")
+    print(f"equal-weight universe: total {((1 + ew).prod() - 1) * 100:.1f}%, sharpe {ew.mean() / ew.std() * 252 ** 0.5:.2f}")
+    print(table.assign(**{c: (table[c] * 100).round(1) for c in pct}).round({"sharpe": 2}).to_string(index=False))
+    print(f"-> {path}")
+
+
+async def fills(cfg: Config) -> None:
+    async with Broker(cfg.ib) as broker:
+        executions = await broker.executions()
+    if executions.empty:
+        print("no executions")
+        return
+    path = cfg.root / "runs" / f"fills-{dt.datetime.now(NEW_YORK):%Y%m%d}.csv"
+    path.parent.mkdir(exist_ok=True)
+    executions.to_csv(path, index=False)
+    print(executions.to_string(index=False))
+    print(f"-> {path}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(prog="ibkr-quant")
+    parser.add_argument("--config", type=Path, default=Path("config.toml"))
+    commands = parser.add_subparsers(dest="command", required=True)
+    commands.add_parser("universe", help="download S&P 500 constituents")
+    commands.add_parser("check", help="connect and show account, positions, open orders, SPY price")
+    commands.add_parser("bars", help="download adjusted daily bars for the universe")
+    rebalance_parser = commands.add_parser("rebalance", help="plan orders (dry run unless --submit)")
+    rebalance_parser.add_argument("--submit", action="store_true")
+    commands.add_parser("fills", help="show today's executions")
+    backtest_parser = commands.add_parser("backtest", help="replay the strategy on the downloaded bars")
+    backtest_parser.add_argument("--top-n", type=int, nargs="+", default=[10])
+    backtest_parser.add_argument("--every", type=int, nargs="+", default=[1], help="rebalance interval in sessions")
+    backtest_parser.add_argument("--capital", type=float, nargs="+", default=[1_000_000])
+    backtest_parser.add_argument("--fee", choices=["ibkr-fixed", "flat2"], default="ibkr-fixed")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.WARNING, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    cfg = load_config(args.config)
+    match args.command:
+        case "universe":
+            update_universe(cfg)
+        case "check":
+            asyncio.run(check(cfg))
+        case "bars":
+            asyncio.run(update_bars(cfg))
+        case "rebalance":
+            asyncio.run(rebalance(cfg, args.submit))
+        case "fills":
+            asyncio.run(fills(cfg))
+        case "backtest":
+            run_backtest(cfg, args.top_n, args.every, args.capital, args.fee)

src/ibkr_quant/config.py

@@ -0,0 +1,64 @@
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class IBConfig:
+    host: str
+    port: int
+    client_id: int
+    account: str
+    market_data_type: int
+    allow_live: bool
+
+
+@dataclass(frozen=True)
+class DataConfig:
+    universe: Path
+    bars: Path
+    bars_duration: str
+
+
+@dataclass(frozen=True)
+class StrategyConfig:
+    top_n: int
+
+
+@dataclass(frozen=True)
+class ExecutionConfig:
+    order_type: str
+    limit_offset_bps: float
+    cash_buffer: float
+    resize_band: float
+    min_trade_value: float
+    min_trade_fraction: float
+
+
+@dataclass(frozen=True)
+class Config:
+    root: Path
+    ib: IBConfig
+    data: DataConfig
+    strategy: StrategyConfig
+    execution: ExecutionConfig
+
+
+def load_config(path: Path) -> Config:
+    raw = tomllib.loads(path.read_text())
+    root = path.resolve().parent
+    data = raw["data"]
+    execution = ExecutionConfig(**raw["execution"])
+    if execution.order_type not in ("MOC", "LMT"):
+        raise ValueError(f"unknown order_type {execution.order_type!r}")
+    return Config(
+        root=root,
+        ib=IBConfig(**raw["ib"]),
+        data=DataConfig(
+            universe=root / data["universe"],
+            bars=root / data["bars"],
+            bars_duration=data["bars_duration"],
+        ),
+        strategy=StrategyConfig(**raw["strategy"]),
+        execution=execution,
+    )

src/ibkr_quant/rebalance.py

@@ -0,0 +1,51 @@
+import math
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class PlannedOrder:
+    symbol: str
+    current: float
+    target: float
+    delta: float  # > 0 buy, < 0 sell
+    price: float
+
+
+def plan_orders(
+    target_weights: dict[str, float],
+    positions: dict[str, float],
+    prices: dict[str, float],
+    investable: float,
+    min_trade_value: float,
+    resize_band: float = 0.0,
+) -> list[PlannedOrder]:
+    """Turn target weights into whole-share orders, sells first.
+
+    `positions` must contain only positions this strategy manages; a managed
+    position absent from `target_weights` is closed. A held position with a
+    nonzero target is left alone while within `resize_band` (fraction of its
+    target value) of the target.
+    """
+    if any(w < 0 for w in target_weights.values()) or sum(target_weights.values()) > 1 + 1e-9:
+        raise ValueError("target weights must be non-negative and sum to <= 1")
+    symbols = sorted(set(target_weights) | {s for s, q in positions.items() if q})
+    missing = [s for s in symbols if not prices.get(s, 0) > 0]
+    if missing:
+        raise ValueError(f"no price for {missing}")
+
+    orders = []
+    for symbol in symbols:
+        price = prices[symbol]
+        current = positions.get(symbol, 0.0)
+        target_value = investable * target_weights.get(symbol, 0.0)
+        if current > 0 and target_value > 0 and abs(target_value - current * price) < resize_band * target_value:
+            continue
+        target = math.floor(target_value / price)
+        delta = target - current
+        if delta == 0:
+            continue
+        if target != 0 and abs(delta) * price < min_trade_value:
+            continue
+        orders.append(PlannedOrder(symbol, current, target, delta, price))
+    orders.sort(key=lambda o: (o.delta > 0, o.symbol))
+    return orders

src/ibkr_quant/strategy.py

@@ -0,0 +1,76 @@
+"""UpCap+MomGap factor combo, ported from quant-trading `strategies/factor_combo.py`
+(`fc_up_cap_mom_gap_daily`).
+
+The source runs on pandas 2, whose `pct_change` forward-fills gaps first;
+`_pct_change` keeps that behaviour under pandas 3.
+"""
+
+import numpy as np
+import pandas as pd
+
+# Rows the source zeroes before trading: 12-1 momentum needs 21 + 231 bars.
+WARMUP = 252
+
+
+def _pct_change(p: pd.DataFrame, periods: int = 1) -> pd.DataFrame:
+    return p.ffill().pct_change(periods)
+
+
+def _rank(df: pd.DataFrame) -> pd.DataFrame:
+    return df.rank(axis=1, pct=True, na_option="keep")
+
+
+def _mom_12_1(p: pd.DataFrame) -> pd.DataFrame:
+    return _pct_change(p.shift(21), 231)
+
+
+def _gap_up_freq(p: pd.DataFrame) -> pd.DataFrame:
+    ret = _pct_change(p)
+    return (ret > 0.01).astype(float).rolling(60, min_periods=40).mean()
+
+
+def _members(df: pd.DataFrame, member: pd.DataFrame | None) -> pd.DataFrame:
+    return df if member is None else df.where(member)
+
+
+def _up_capture(p: pd.DataFrame, member: pd.DataFrame | None = None) -> pd.DataFrame:
+    ret = _pct_change(p)
+    mkt = _members(ret, member).mean(axis=1)
+    up_mkt = mkt > 0
+    arr = ret.values.copy()
+    arr[~up_mkt.values, :] = np.nan
+    stock_up = pd.DataFrame(arr, index=ret.index, columns=ret.columns)
+    mkt_up_vals = mkt.where(up_mkt, np.nan)
+    stock_avg = stock_up.rolling(60, min_periods=20).mean()
+    mkt_avg = mkt_up_vals.rolling(60, min_periods=20).mean()
+    return stock_avg.div(mkt_avg, axis=0)
+
+
+def _mom_x_gap(p: pd.DataFrame, member: pd.DataFrame | None = None) -> pd.DataFrame:
+    return _rank(_members(_mom_12_1(p), member)) * _rank(_members(_gap_up_freq(p), member))
+
+
+def signal_up_cap_mom_gap(p: pd.DataFrame, member: pd.DataFrame | None = None) -> pd.DataFrame:
+    """`member` (dates x symbols) limits the cross-section (ranks, market return) to each
+    day's index members; factors still use each name's full price history. Live, every
+    column is a current member."""
+    return 0.5 * _rank(_members(_up_capture(p, member), member)) + 0.5 * _rank(_mom_x_gap(p, member))
+
+
+def up_cap_mom_gap_weights(opens: pd.DataFrame, top_n: int) -> dict[str, float]:
+    """Equal-weight top_n names for trading at the close of the last row's session.
+
+    `opens` holds opening prices (dates x symbols) up to and including today's open,
+    matching the source's live timing: signal from opens, fills at the same day's close.
+    """
+    if len(opens) <= WARMUP:
+        raise ValueError(f"need more than {WARMUP} sessions of opens, have {len(opens)}")
+    return select_top_n(signal_up_cap_mom_gap(opens).iloc[-1], top_n)
+
+
+def select_top_n(sig: pd.Series, top_n: int) -> dict[str, float]:
+    if sig.notna().sum() < top_n:
+        return {}
+    rank = sig.rank(ascending=False, na_option="bottom")
+    winners = rank.index[rank <= top_n]
+    return {symbol: 1.0 / len(winners) for symbol in winners}

src/ibkr_quant/universe.py

@@ -0,0 +1,31 @@
+import urllib.request
+from io import StringIO
+from pathlib import Path
+
+import pandas as pd
+
+SP500_URL = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies"
+
+
+def to_ib_symbol(ticker: str) -> str:
+    # IBKR writes share classes with a space: BRK.B -> "BRK B".
+    return ticker.replace(".", " ")
+
+
+def fetch_sp500() -> pd.DataFrame:
+    request = urllib.request.Request(SP500_URL, headers={"User-Agent": "ibkr-quant/0.1"})
+    with urllib.request.urlopen(request, timeout=30) as response:
+        html = response.read().decode()
+    table = pd.read_html(StringIO(html), attrs={"id": "constituents"})[0]
+    return pd.DataFrame(
+        {
+            "symbol": table["Symbol"].map(to_ib_symbol),
+            "name": table["Security"],
+            "sector": table["GICS Sector"],
+            "added": pd.to_datetime(table["Date added"], errors="coerce").dt.date,
+        }
+    ).sort_values("symbol", ignore_index=True)
+
+
+def load_universe(path: Path) -> list[str]:
+    return pd.read_csv(path)["symbol"].tolist()