diff --git a/Makefile b/Makefile index 8ed0bda..aa0373f 100644 --- a/Makefile +++ b/Makefile @@ -2,6 +2,8 @@ # "Enable ActiveX and Socket Clients" on and the port matching config.toml. # Daily flow, 09:31-15:45 ET on a trading day: `make bars` (picks up today's # open), `make plan` to review, `make submit` to send MOC orders filled at the close. +# Only rebalance sessions (every strategy.rebalance_every from strategy.anchor) +# trade; on other days plan and submit say so and do nothing. # With delayed data (market_data_type = 3) today's bar appears ~15 min after the # open, so start after 09:47 ET. RUN = uv run ibkr-quant diff --git a/config.toml b/config.toml index bcc1b2b..a0649c7 100644 --- a/config.toml +++ b/config.toml @@ -12,10 +12,14 @@ universe = "data/universe.csv" bars = "data/bars/daily.csv" bars_duration = "2 Y" # the strategy needs > 252 sessions -# UpCap+MomGap (daily), ported from quant-trading fc_up_cap_mom_gap_daily: -# signal on opens through today's open, equal-weight top_n, filled at today's close. +# overnight_recovery_top10, ported from quant-trading strategies/robust_momentum.py: +# overnight-return momentum (12m, skip 1m) + distance from the 126-day low, signal on +# sessions through yesterday's close, equal-weight top_n filled at today's close. [strategy] top_n = 10 +rebalance_every = 10 # sessions; nothing trades in between and weights drift +buffer = 2.0 # keep a held name while ranked within top_n * (1 + buffer) +anchor = 2026-09-29 # first rebalance session; then every rebalance_every sessions [execution] order_type = "MOC" # MOC (submit 09:30-15:45 ET) | LMT diff --git a/deploy/systemd/ibkr-quant-trade.service b/deploy/systemd/ibkr-quant-trade.service index 5403191..4f09150 100644 --- a/deploy/systemd/ibkr-quant-trade.service +++ b/deploy/systemd/ibkr-quant-trade.service @@ -1,5 +1,5 @@ [Unit] -Description=ibkr-quant daily rebalance: today's bars, then MOC orders +Description=ibkr-quant daily run: today's bars, then MOC orders on rebalance sessions After=ibgateway.service # make gateway restarts a Gateway that has lost IBKR. Retries cover today's bar # arriving late (delayed data) and Gateway restarts; make submit refuses to diff --git a/deploy/systemd/ibkr-quant-trade.timer b/deploy/systemd/ibkr-quant-trade.timer index f3d4de5..286b206 100644 --- a/deploy/systemd/ibkr-quant-trade.timer +++ b/deploy/systemd/ibkr-quant-trade.timer @@ -1,5 +1,5 @@ [Unit] -Description=ibkr-quant daily rebalance, 09:50 New York time on weekdays +Description=ibkr-quant daily run, 09:50 New York time on weekdays [Timer] OnCalendar=Mon..Fri 09:50 America/New_York diff --git a/src/ibkr_quant/backtest.py b/src/ibkr_quant/backtest.py index 47ebfaa..b1e5b73 100644 --- a/src/ibkr_quant/backtest.py +++ b/src/ibkr_quant/backtest.py @@ -1,5 +1,6 @@ -"""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.""" +"""Day-by-day replay of the live loop: on rebalance sessions, targets from the signal +known at day d's open, orders sized at d's open by `plan_orders` and filled at d's +close, commissions charged per order; in between, holdings drift untouched.""" from collections.abc import Callable from dataclasses import dataclass @@ -8,7 +9,7 @@ import numpy as np import pandas as pd from .rebalance import plan_orders -from .strategy import WARMUP, select_top_n +from .strategy import WARMUP, select_with_buffer # A held name without a price for this many sessions (delisted, acquired) is # sold at its last close. @@ -28,14 +29,13 @@ def flat_fee(amount: float) -> Callable[[float, float], float]: class RunConfig: top_n: int rebalance_every: int + buffer: float 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) @@ -49,14 +49,13 @@ class RunResult: 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.""" + """`signal` must be causal: row d uses only data known at d's open.""" 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 @@ -68,33 +67,31 @@ def run( 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 + if i % cfg.rebalance_every == 0: + targets = select_with_buffer(signal.loc[day], set(positions), cfg.top_n, cfg.buffer) + # 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, + ) + 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 diff --git a/src/ibkr_quant/cli.py b/src/ibkr_quant/cli.py index 1f2d86a..2e62ba5 100644 --- a/src/ibkr_quant/cli.py +++ b/src/ibkr_quant/cli.py @@ -12,7 +12,7 @@ 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 .strategy import WARMUP, select_with_buffer, sessions_until_rebalance, signal_overnight_recovery from .universe import fetch_sp500, load_universe # MOC orders must reach NYSE by 15:50 ET; keep a margin. @@ -71,20 +71,30 @@ async def update_bars(cfg: Config) -> None: print(f"no bars ({len(failed)}): {failed}") -def load_opens(path: Path, today: dt.date) -> pd.DataFrame: +def load_bars(path: Path, today: dt.date) -> tuple[pd.DataFrame, pd.DataFrame]: bars = pd.read_csv(path, parse_dates=["date"]) - opens = bars.pivot(index="date", columns="symbol", values="open").sort_index() + opens, closes = (bars.pivot(index="date", columns="symbol", values=v).sort_index() for v in ("open", "close")) 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 + return opens, closes async def rebalance(cfg: Config, submit: bool) -> None: - ex = cfg.execution + ex, st = cfg.execution, cfg.strategy 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)) + opens, closes = load_bars(cfg.data.bars, now.date()) + wait = sessions_until_rebalance(opens.index, st.anchor, st.rebalance_every) + if wait != 0: + after = f"in {wait} sessions" if wait else f"on the first session from {st.anchor}" + print(f"not a rebalance session; holdings drift until the next one {after}") + return + # Today's bar is still forming, so the signal uses completed sessions only. + history = opens.index[:-1] + if len(history) <= WARMUP: + raise SystemExit(f"need more than {WARMUP} completed sessions of bars, have {len(history)}") + members = opens.columns.intersection(sorted(universe)) + score = signal_overnight_recovery(opens.loc[history, members], closes.loc[history, members]).iloc[-1] run_dir = cfg.root / "runs" / now.strftime("%Y%m%d-%H%M%S") async with Broker(cfg.ib) as broker: @@ -92,7 +102,8 @@ async def rebalance(cfg: Config, submit: bool) -> None: 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. + weights = select_with_buffer(score, {h.symbol for h in managed}, st.top_n, st.buffer) + # Sized at today's open, the latest price in the bars; filled at the close. orders = plan_orders( weights, {h.symbol: h.quantity for h in managed}, @@ -107,7 +118,7 @@ async def rebalance(cfg: Config, submit: bool) -> None: 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(f"account {state.account} NAV ${state.nav_usd:,.2f} investable ${investable:,.2f} signal on sessions through {history[-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}") @@ -157,6 +168,7 @@ async def rebalance(cfg: Config, submit: bool) -> None: def run_backtest(cfg: Config, top_ns: list[int], everies: list[int], capitals: list[float], fee: str) -> None: + st = cfg.strategy 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() @@ -166,17 +178,18 @@ def run_backtest(cfg: Config, top_ns: list[int], everies: list[int], capitals: l 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) + # Row d holds the signal from sessions through d - 1, known at d's open. + signal = signal_overnight_recovery(opens, closes, member).shift(1) 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) + run_cfg = backtest.RunConfig(top_n, every, st.buffer, 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. + # Same-universe baseline: equal weight, re-weighted 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" @@ -214,8 +227,8 @@ def main() -> None: 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("--top-n", type=int, nargs="+", help="default: strategy.top_n") + backtest_parser.add_argument("--every", type=int, nargs="+", help="rebalance interval in sessions; default: strategy.rebalance_every") 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() @@ -236,4 +249,4 @@ def main() -> None: case "fills": asyncio.run(fills(cfg)) case "backtest": - run_backtest(cfg, args.top_n, args.every, args.capital, args.fee) + run_backtest(cfg, args.top_n or [cfg.strategy.top_n], args.every or [cfg.strategy.rebalance_every], args.capital, args.fee) diff --git a/src/ibkr_quant/config.py b/src/ibkr_quant/config.py index 18e0e5f..1128a3e 100644 --- a/src/ibkr_quant/config.py +++ b/src/ibkr_quant/config.py @@ -1,3 +1,4 @@ +import datetime as dt import tomllib from dataclasses import dataclass from pathlib import Path @@ -24,6 +25,9 @@ class DataConfig: @dataclass(frozen=True) class StrategyConfig: top_n: int + rebalance_every: int + buffer: float + anchor: dt.date @dataclass(frozen=True) diff --git a/src/ibkr_quant/strategy.py b/src/ibkr_quant/strategy.py index d4ef8e8..358d5c0 100644 --- a/src/ibkr_quant/strategy.py +++ b/src/ibkr_quant/strategy.py @@ -1,76 +1,60 @@ -"""UpCap+MomGap factor combo, ported from quant-trading `strategies/factor_combo.py` -(`fc_up_cap_mom_gap_daily`). +"""overnight_recovery_top10, ported from quant-trading `strategies/robust_momentum.py` +(`RobustMomentumStrategy("overnight_recovery", top_n=10, rebal_freq=10, buffer=2.0)`). -The source runs on pandas 2, whose `pct_change` forward-fills gaps first; -`_pct_change` keeps that behaviour under pandas 3. +Score: mean cross-sectional percentile rank of overnight-return momentum (12 months, +skipping the last one) and distance from the 126-day low. Every `rebalance_every` +sessions the book holds the top_n names equal-weighted, keeping a held name while its +rank is within top_n * (1 + buffer); in between nothing trades and weights drift. """ +import datetime as dt + import numpy as np import pandas as pd -# Rows the source zeroes before trading: 12-1 momentum needs 21 + 231 bars. +# Rows the source zeroes before trading; overnight momentum needs 180 + 21 of them. WARMUP = 252 -def _pct_change(p: pd.DataFrame, periods: int = 1) -> pd.DataFrame: - return p.ffill().pct_change(periods) +def overnight_momentum(opens: pd.DataFrame, closes: pd.DataFrame, window: int = 252, skip: int = 21) -> pd.DataFrame: + """Sum of log overnight returns (open / previous close), Lou, Polk & Skouras (2019).""" + overnight = np.log(opens / closes.ffill().shift(1)) + return overnight.rolling(window, min_periods=180).sum().shift(skip) -def _rank(df: pd.DataFrame) -> pd.DataFrame: - return df.rank(axis=1, pct=True, na_option="keep") +def recovery(closes: pd.DataFrame, window: int = 126) -> pd.DataFrame: + c = closes.ffill() + return c / c.rolling(window, min_periods=window).min() - 1 -def _mom_12_1(p: pd.DataFrame) -> pd.DataFrame: - return _pct_change(p.shift(21), 231) +def signal_overnight_recovery( + opens: pd.DataFrame, closes: pd.DataFrame, member: pd.DataFrame | None = None +) -> pd.DataFrame: + """Row t uses bars through t's close. `member` (dates x symbols) limits the ranks to + each day's index members; factors still use each name's full price history. Live, + every column is a current member. A name without a close on a row is not eligible.""" + factors = [overnight_momentum(opens, closes), recovery(closes)] + if member is not None: + factors = [f.where(member) for f in factors] + score = sum(f.rank(axis=1, pct=True) for f in factors) / len(factors) + return score.where(closes.notna()) -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 select_with_buffer(score: pd.Series, held: set[str], top_n: int, buffer: float) -> dict[str, float]: + """Equal weights on the top_n names of `score`, keeping held names while ranked + within top_n * (1 + buffer). Ties keep column order, as in the source.""" + valid = score.dropna() + if len(valid) < top_n: + raise ValueError(f"only {len(valid)} names have a score, need {top_n}") + order = valid.index[np.argsort(-valid.to_numpy(), kind="stable")] + rank = {symbol: i for i, symbol in enumerate(order)} + chosen = sorted((s for s in held if rank.get(s, np.inf) < top_n * (1 + buffer)), key=rank.__getitem__)[:top_n] + chosen += [s for s in order if s not in chosen][: top_n - len(chosen)] + return {symbol: 1.0 / top_n for symbol in chosen} -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} +def sessions_until_rebalance(sessions: pd.DatetimeIndex, anchor: dt.date, every: int) -> int | None: + """0 when the last of `sessions` is a rebalance session: the first session on or + after `anchor` and every `every` sessions after it. None before the anchor.""" + n = int((sessions.date >= anchor).sum()) - 1 + return None if n < 0 else -n % every diff --git a/tests/test_strategy.py b/tests/test_strategy.py new file mode 100644 index 0000000..2434277 --- /dev/null +++ b/tests/test_strategy.py @@ -0,0 +1,28 @@ +import datetime as dt + +import pandas as pd +import pytest + +from ibkr_quant.strategy import select_with_buffer, sessions_until_rebalance + +SCORE = pd.Series({"A": 0.9, "B": 0.8, "C": 0.7, "D": 0.6, "E": 0.5, "F": None}) + + +def test_picks_top_n_equal_weight_without_holdings(): + assert select_with_buffer(SCORE, set(), top_n=2, buffer=0.0) == {"A": 0.5, "B": 0.5} + + +def test_keeps_held_names_within_the_buffer_and_fills_from_the_top(): + # top_n * (1 + buffer) = 4: D (rank 3) stays, E (rank 4) and unscored F go. + assert select_with_buffer(SCORE, {"D", "E", "F"}, top_n=2, buffer=1.0) == {"D": 0.5, "A": 0.5} + + +def test_refuses_too_few_scored_names(): + with pytest.raises(ValueError, match="only 5 names"): + select_with_buffer(SCORE, set(), top_n=6, buffer=0.0) + + +def test_rebalances_on_the_anchor_session_and_every_n_after(): + sessions = pd.bdate_range("2026-09-24", periods=12) # anchor 09-29 is the 4th + wait = [sessions_until_rebalance(sessions[: i + 1], dt.date(2026, 9, 29), 5) for i in range(len(sessions))] + assert wait == [None, None, None, 0, 4, 3, 2, 1, 0, 4, 3, 2]