From 1697bc16411bc4cb598fa493e9ae341941722f03 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 25 Sep 2026 01:16:33 +0800 Subject: [PATCH] fix(report): correct periodic returns, injection metrics and live view - Monthly/yearly returns chain from the previous period's close (the first trading day of every period was dropped; up to ~9pp/yr error). - With cash injections, metrics/drawdowns/periodic returns are time-weighted on the no-injection curve; capitalFlows reports invested/profit separately. - Benchmarks cover exactly the strategies' dates. - CAGR uses calendar time, so gaps in live records do not inflate it. - Backtests skip the per-trade fee for rebalances below the live minimum order and keep a ruined portfolio at zero. - live: exact strategy match, ineligible strategies hidden, currency, startDate, same-window benchmark, stale-price fallback, equity anchored at the initial capital. - PIT membership falls back to the cached history when the Wikipedia change log cannot be parsed (the table is no longer on the page). Co-Authored-By: Claude Opus 5.5 (1M context) --- bridge.py | 301 +++++++++++++++++++----------------- main.py | 26 +++- metrics.py | 12 +- tests/test_live_pipeline.py | 121 +++++++++++++++ universe_history.py | 19 ++- 5 files changed, 333 insertions(+), 146 deletions(-) create mode 100644 tests/test_live_pipeline.py diff --git a/bridge.py b/bridge.py index 1e127b7..7173af2 100644 --- a/bridge.py +++ b/bridge.py @@ -25,12 +25,21 @@ from main import backtest from research.permanent_yearly import equity_with_cashflows from trader import ( ETF_STRATEGY_UNIVERSES, + LIVE_EXCLUDED_STRATEGIES, + MARKET_FEE_TIERED, STRATEGY_REGISTRY, - strategy_universe, + convert_quote_currency, + price_request, + strategy_supports_market, strategy_data_market, - load_state, filter_tradable_tickers, ) + +MARKET_CURRENCY = {"us": "USD", "cn": "CNY"} + +# Rebalances below this size are not charged the per-trade fee in backtests, +# matching the live trader's minimum order size. +MIN_TRADE_VALUE = 50.0 from universe import UNIVERSES @@ -567,27 +576,37 @@ def compute_rebalance_win_rate(equity: pd.Series, trade_events: list[dict]) -> f def compute_periodic_returns(equity: pd.Series) -> tuple[dict, dict]: - """Compute monthly and yearly returns from an equity curve.""" - monthly = {} - yearly = {} + """Monthly and yearly returns from an equity curve. - # Monthly returns - for (year, month), group in equity.groupby([equity.index.year, equity.index.month]): - if len(group) < 2: - continue - ret = group.iloc[-1] / group.iloc[0] - 1 - monthly[f"{year}-{month:02d}"] = round(float(ret), 4) - - # Yearly returns - for year, group in equity.groupby(equity.index.year): - if len(group) < 2: - continue - ret = group.iloc[-1] / group.iloc[0] - 1 - yearly[str(year)] = round(float(ret), 4) + Each period is measured from the previous period's last value (the first + period from the curve's first value), so the first trading day's move of + every period is included. + """ + def by_period(keys) -> dict: + ends = equity.groupby(keys).last() + out = {} + prev = float(equity.iloc[0]) + for key, end in ends.items(): + if prev > 0: + out[key] = round(float(end) / prev - 1, 4) + prev = float(end) + return out + idx = equity.index + monthly = by_period([f"{d.year}-{d.month:02d}" for d in idx]) + yearly = by_period([str(d.year) for d in idx]) return monthly, yearly +def benchmark_curve(prices: pd.Series, dates: pd.DatetimeIndex, capital: float) -> list[dict]: + """Benchmark value on ``dates``, rebased to ``capital`` on the first date it has a price.""" + series = prices.reindex(dates).ffill().dropna() + if series.empty: + return [] + values = series / series.iloc[0] * capital + return [{"date": d.strftime("%Y-%m-%d"), "value": round(float(v), 2)} for d, v in values.items()] + + def build_injection_schedule(equity: pd.Series, schedule: dict) -> tuple[pd.Series, list[dict]]: """Build a contributions Series from injection schedule config. @@ -772,17 +791,17 @@ def cmd_backtest(args: argparse.Namespace) -> None: for name in stock_strategies: _log(f" Running: {name}") - strategy_fn = STRATEGY_REGISTRY[name] - # Derive top_n from strategy metadata (respects per-strategy defaults) - meta = STRATEGY_META.get(name, {}) - top_n = int(meta.get("params", {}).get("top_n", {}).get("default", 10)) - strategy = strategy_fn(top_n=top_n) + # Registry defaults are the single source of strategy parameters, + # shared with the live trader. + strategy = STRATEGY_REGISTRY[name]() equity = backtest( strategy, data[tickers], initial_capital=capital, transaction_cost=tx_cost, fixed_fee=fixed_fee, + min_trade_value=MIN_TRADE_VALUE, + **MARKET_FEE_TIERED.get(market, {}), ) # Get weight signals for trade extraction @@ -795,14 +814,8 @@ def cmd_backtest(args: argparse.Namespace) -> None: ) results.append(result_entry) - # Benchmark - bench = data[benchmark_ticker].dropna() - if years: - cutoff = bench.index[-1] - pd.DateOffset(years=years) - bench = bench[bench.index >= cutoff] - bench_equity = (bench / bench.iloc[0]) * capital - bench_curve = [{"date": d.strftime("%Y-%m-%d"), "value": round(float(v), 2)} - for d, v in bench_equity.items()] + # Benchmark over exactly the strategies' dates + bench_curve = benchmark_curve(data[benchmark_ticker], data.index, capital) else: benchmark_label = "SPY" bench_curve = [] @@ -816,16 +829,16 @@ def cmd_backtest(args: argparse.Namespace) -> None: all_etf = sorted(set(etf_tickers + ["SPY"])) with _redirect_stdout_to_stderr(): - etf_data = data_manager.update("etfs", all_etf, with_open=False) + etf_data = data_manager.update("etfs", price_request(market, all_etf), with_open=False) if isinstance(etf_data, tuple): etf_data = etf_data[0] + etf_data = convert_quote_currency(etf_data) if years: cutoff = etf_data.index[-1] - pd.DateOffset(years=years) etf_data = etf_data[etf_data.index >= cutoff] - strategy_fn = STRATEGY_REGISTRY[name] - strategy = strategy_fn() + strategy = STRATEGY_REGISTRY[name]() tradable = filter_tradable_tickers(etf_data, etf_tickers) equity = backtest( @@ -833,6 +846,8 @@ def cmd_backtest(args: argparse.Namespace) -> None: initial_capital=capital, transaction_cost=tx_cost, fixed_fee=fixed_fee, + min_trade_value=MIN_TRADE_VALUE, + **MARKET_FEE_TIERED.get(market, {}), ) weights = strategy.generate_signals(etf_data[tradable]) @@ -855,10 +870,7 @@ def cmd_backtest(args: argparse.Namespace) -> None: cutoff = spy_data.index[-1] - pd.DateOffset(years=years) spy_data = spy_data[spy_data.index >= cutoff] if "SPY" in spy_data.columns: - bench = spy_data["SPY"].dropna() - bench_equity = (bench / bench.iloc[0]) * capital - bench_curve = [{"date": d.strftime("%Y-%m-%d"), "value": round(float(v), 2)} - for d, v in bench_equity.items()] + bench_curve = benchmark_curve(spy_data["SPY"], spy_data.index, capital) output = { "meta": { @@ -888,7 +900,13 @@ def _build_strategy_result( injection_schedule: dict | None, initial_capital: float, ) -> dict: - """Build a strategy result dict with optional injection applied.""" + """Build a strategy result dict with optional injection applied. + + Return and risk metrics, drawdowns and periodic returns are time-weighted + (computed on the curve without injections), so injected cash never counts + as performance. ``capitalFlows`` reports the money view: total invested, + final value and profit of the injected curve. + """ injection_events = [] equity_with_inj = None @@ -897,8 +915,10 @@ def _build_strategy_result( contributions, injection_events = build_injection_schedule(equity, injection_schedule) equity_with_inj = equity_with_cashflows(returns, contributions, initial_capital) - # Use injected equity for metrics/curves if available + # Curve shown to the user: the money curve when cash is injected. final_equity = equity_with_inj if equity_with_inj is not None else equity + invested = initial_capital + sum(ev["amount"] for ev in injection_events) + final_value = float(final_equity.iloc[-1]) # Equity curve curve = [{"date": d.strftime("%Y-%m-%d"), "value": round(float(v), 2)} @@ -910,17 +930,14 @@ def _build_strategy_result( base_curve = [{"date": d.strftime("%Y-%m-%d"), "value": round(float(v), 2)} for d, v in equity.items()] - # Metrics - strat_metrics = metrics.raw_summary(final_equity) + # Time-weighted metrics on the curve without injections + strat_metrics = metrics.raw_summary(equity) strat_metrics["rebalanceWinRate"] = compute_rebalance_win_rate( - final_equity, trade_events, + equity, trade_events, ) - # Drawdown periods - drawdowns = compute_drawdown_periods(final_equity) - - # Monthly/yearly returns - monthly_ret, yearly_ret = compute_periodic_returns(final_equity) + drawdowns = compute_drawdown_periods(equity) + monthly_ret, yearly_ret = compute_periodic_returns(equity) meta = STRATEGY_META.get(name, {"label": name, "category": "other"}) @@ -933,6 +950,13 @@ def _build_strategy_result( "drawdowns": drawdowns, "trades": trade_events, "injections": injection_events, + "capitalFlows": { + "invested": round(invested, 2), + "finalValue": round(final_value, 2), + "profit": round(final_value - invested, 2), + "returnOnInvested": final_value / invested - 1 if invested else None, + }, + "ruined": bool((equity <= 0).any()), "monthlyReturns": monthly_ret, "yearlyReturns": yearly_ret, } @@ -942,81 +966,81 @@ def _build_strategy_result( # Subcommand: live # --------------------------------------------------------------------------- -def _load_latest_prices(markets_needed: set[str]) -> dict[str, dict]: - """Pull the latest close + as-of date for each cached market panel. +LIVE_BENCHMARK = {"us": ("SPY", "SPY"), "cn": ("510300.SS", "CSI 300 ETF 510300")} - Returns ``{market: {"asof": str|None, "prices": {ticker: float}}}``. - We read the on-disk CSVs directly so this stays fast even when called - against a freshly initialized live state with no network. + +def _load_panels(markets_needed: set[str]) -> dict[str, pd.DataFrame | None]: + """Load each cached close panel once (forward-filled, HK tickers in USD). + + Reads the on-disk CSVs directly so the live view stays fast and works + without network access. """ - import data_manager # local import to avoid heavyweight import at module load - out: dict[str, dict] = {} + panels: dict[str, pd.DataFrame | None] = {} for mk in markets_needed: try: df = data_manager.load(mk) except Exception as e: # noqa: BLE001 — never fatal - _log(f" load_latest_prices: cannot load {mk}: {e}") - out[mk] = {"asof": None, "prices": {}} - continue - if df is None or df.empty: - out[mk] = {"asof": None, "prices": {}} - continue - last_row = df.iloc[-1] - out[mk] = { - "asof": df.index[-1].strftime("%Y-%m-%d"), - "prices": {col: float(v) for col, v in last_row.items() - if v is not None and not (isinstance(v, float) and np.isnan(v))}, - } + _log(f" load panel {mk}: {e}") + df = None + panels[mk] = convert_quote_currency(df) if df is not None and not df.empty else None + return panels + + +def _latest_prices(panel: pd.DataFrame | None) -> dict: + if panel is None: + return {"asof": None, "prices": {}} + last_row = panel.iloc[-1] + return { + "asof": panel.index[-1].strftime("%Y-%m-%d"), + "prices": {col: float(v) for col, v in last_row.items() if pd.notna(v)}, + } + + +def _last_recorded_prices(state: dict) -> dict[str, float]: + """Most recent price recorded in the daily log for each ticker.""" + out: dict[str, float] = {} + for entry in reversed(state.get("daily_log", []) or []): + for ticker, pos in (entry.get("holdings") or {}).items(): + if ticker not in out and pos.get("price"): + out[ticker] = float(pos["price"]) return out -def _strategy_data_market_safe(strategy_name: str, default: str) -> str: - """Return the price-data market a strategy reads from. - - Mirrors trader.strategy_data_market without importing trader (to keep - bridge import-cheap when the CLI is hit). Tactical-allocation strategies - trade ETFs, so they pull from the 'etfs' panel; stock pickers stay on - their nominal market. - """ - try: - from trader import ETF_STRATEGY_UNIVERSES # noqa: WPS433 - except Exception: - return default - base = strategy_name.removeprefix("sim_") - return "etfs" if base in ETF_STRATEGY_UNIVERSES else default - - def cmd_live(args: argparse.Namespace) -> None: """Read live trader state files and output normalized JSON. Each strategy's record contains: - - Current cash, holdings, marked-to-market value of each holding (using - the latest cached close price), and total assets. - - Daily equity curve from the trader's persisted state. + - Current cash, holdings, marked-to-market value of each holding (latest + cached price, falling back to the last recorded price), and total assets. + - Daily equity curve starting from the initial capital, plus the market + benchmark rebased to the same capital over the same dates. - Full trade log (one entry per ticker per fill) + day-aggregated view. - Metrics, drawdowns, monthly/yearly returns. """ market = args.market - pattern = os.path.join("data", f"trader_{market}_*.json") - state_files = sorted(glob(pattern)) + prefix = f"trader_{market}_" + state_files = sorted(glob(os.path.join("data", f"{prefix}*.json"))) + wanted = args.strategy - if args.strategy: - state_files = [f for f in state_files if args.strategy in f] - - _log(f"Found {len(state_files)} state files for market={market}") - - # Pre-load price panels once. Tactical/ETF strategies read from 'etfs', - # everything else from the nominal market. - markets_needed = {market, "etfs"} - latest_prices = _load_latest_prices(markets_needed) + panels = _load_panels({market, "etfs"}) + bench_ticker, bench_label = LIVE_BENCHMARK[market] + bench_panel = panels.get(market) + bench_prices = ( + bench_panel[bench_ticker] + if bench_panel is not None and bench_ticker in bench_panel.columns + else pd.Series(dtype=float) + ) results = [] for path in state_files: - basename = os.path.basename(path) - parts = basename.replace(".json", "").split("_", 2) - if len(parts) < 3: + strategy_name = os.path.basename(path)[len(prefix):-len(".json")] + if strategy_name.startswith("sim_"): + continue + if wanted and strategy_name != wanted: + continue + if strategy_name in LIVE_EXCLUDED_STRATEGIES or not strategy_supports_market(strategy_name, market): + _log(f" Skipping {strategy_name}: not a live strategy in the {market.upper()} book") continue - strategy_name = parts[2] try: with open(path) as f: @@ -1025,19 +1049,22 @@ def cmd_live(args: argparse.Namespace) -> None: _log(f" Skipping {path}: {e}") continue - data_market = _strategy_data_market_safe(strategy_name, market) - price_book = latest_prices.get(data_market, {"asof": None, "prices": {}}) + data_market = strategy_data_market(market, strategy_name) + price_book = _latest_prices(panels.get(data_market)) + recorded = _last_recorded_prices(state) holdings_raw = state.get("holdings", {}) or {} cash = float(state.get("cash", 0) or 0) - # Per-holding marked-to-market table holdings_detail = [] market_value = 0.0 for ticker, shares in holdings_raw.items(): shares_f = float(shares) price = price_book["prices"].get(ticker) - value = shares_f * float(price) if price is not None else None + stale = price is None + if stale: + price = recorded.get(ticker) + value = shares_f * price if price is not None else None if value is not None: market_value += value holdings_detail.append({ @@ -1045,28 +1072,28 @@ def cmd_live(args: argparse.Namespace) -> None: "shares": shares_f, "lastPrice": float(price) if price is not None else None, "marketValue": round(value, 2) if value is not None else None, + "priceStale": stale, }) - holdings_detail.sort( - key=lambda h: (-(h["marketValue"] or 0.0), h["ticker"]), - ) + holdings_detail.sort(key=lambda h: (-(h["marketValue"] or 0.0), h["ticker"])) - # Equity curve from persisted state + initial_capital = float(state.get("initial_capital", 0) or 0) daily_equity = state.get("daily_equity", {}) or {} dates = sorted(daily_equity.keys()) - equity_curve = [ - {"date": d, "value": round(float(daily_equity[d]), 2)} - for d in dates - ] equity_series = pd.Series( - {pd.Timestamp(d): float(daily_equity[d]) for d in dates} - ).sort_index() if dates else pd.Series(dtype=float) + {pd.Timestamp(d): float(daily_equity[d]) for d in dates}, dtype=float, + ).sort_index() + if len(equity_series) and initial_capital: + # Anchor at the initial capital one day before the first session so + # the first day's P&L is part of every metric. + anchor = equity_series.index[0] - pd.Timedelta(days=1) + equity_series = pd.concat([pd.Series({anchor: initial_capital}), equity_series]) + equity_curve = [ + {"date": d.strftime("%Y-%m-%d"), "value": round(float(v), 2)} + for d, v in equity_series.items() + ] - strat_metrics = ( - metrics.raw_summary(equity_series) if len(equity_series) >= 2 else None - ) + strat_metrics = metrics.raw_summary(equity_series) if len(equity_series) >= 2 else None - # Trade-by-day aggregation (kept for backwards compatibility) plus - # full granular log. trade_log_raw = state.get("trade_log", []) or [] trade_by_date: dict[str, dict] = {} trade_log = [] @@ -1097,29 +1124,24 @@ def cmd_live(args: argparse.Namespace) -> None: agg["sellTickers"].append(entry["ticker"]) trades = [{"date": d, **v} for d, v in sorted(trade_by_date.items())] - drawdowns = ( - compute_drawdown_periods(equity_series) if len(equity_series) >= 2 else [] - ) + drawdowns = compute_drawdown_periods(equity_series) if len(equity_series) >= 2 else [] monthly_ret, yearly_ret = ( - compute_periodic_returns(equity_series) - if len(equity_series) >= 2 else ({}, {}) + compute_periodic_returns(equity_series) if len(equity_series) >= 2 else ({}, {}) ) - meta = STRATEGY_META.get( - strategy_name, {"label": strategy_name, "category": "other"}, - ) + meta = STRATEGY_META.get(strategy_name, {"label": strategy_name, "category": "other"}) - # Current value: prefer live MTM; fall back to last persisted equity. - live_value = cash + market_value if holdings_detail else cash - if not holdings_detail and equity_curve: - live_value = equity_curve[-1]["value"] + # Current value: live mark-to-market of holdings plus cash. + live_value = cash + market_value + if not holdings_detail and dates: + live_value = float(daily_equity[dates[-1]]) - initial_capital = float(state.get("initial_capital", 0) or 0) - unrealized_pnl = ( - live_value - initial_capital if initial_capital else 0.0 - ) - unrealized_pnl_pct = ( - unrealized_pnl / initial_capital if initial_capital else 0.0 + unrealized_pnl = live_value - initial_capital if initial_capital else 0.0 + unrealized_pnl_pct = unrealized_pnl / initial_capital if initial_capital else 0.0 + + bench_curve = ( + benchmark_curve(bench_prices, pd.DatetimeIndex(equity_series.index[1:]), initial_capital) + if len(equity_series) >= 2 and len(bench_prices) else [] ) results.append({ @@ -1127,7 +1149,9 @@ def cmd_live(args: argparse.Namespace) -> None: "label": meta.get("label", strategy_name), "category": meta.get("category", "other"), "market": market, + "currency": MARKET_CURRENCY.get(market, "USD"), "initialCapital": initial_capital, + "startDate": dates[0] if dates else None, "currentValue": round(live_value, 2), "cash": round(cash, 2), "marketValue": round(market_value, 2), @@ -1137,6 +1161,7 @@ def cmd_live(args: argparse.Namespace) -> None: "holdingsDetail": holdings_detail, "priceAsOf": price_book["asof"], "equityCurve": equity_curve, + "benchmark": {"name": bench_label, "equityCurve": bench_curve}, "metrics": strat_metrics, "drawdowns": drawdowns, "trades": trades, diff --git a/main.py b/main.py index fbfb5bf..0d7b611 100644 --- a/main.py +++ b/main.py @@ -34,6 +34,7 @@ def backtest( fee_base: float = 0.0, fee_per_share: float = 0.0, open_data: pd.DataFrame | None = None, + min_trade_value: float = 0.0, ) -> pd.Series: """ Vectorized backtest. @@ -60,6 +61,11 @@ def backtest( IBKR-style schedule: max(2, 1.88 + 0.009 * shares). open_data : pd.DataFrame, optional Open prices. When provided, enables open-to-close execution mode. + min_trade_value : float + Per-ticker rebalances smaller than ``max(min_trade_value, 0.1% of + equity)`` are not charged the per-trade fee, mirroring the live + trader, which never places such orders. 0 (default) charges every + weight change. Returns ------- @@ -98,6 +104,11 @@ def backtest( weight_changes = positions.diff().fillna(0.0) equity_running = (1 + portfolio_returns).cumprod() * initial_capital eq_prev = equity_running.shift(1).fillna(initial_capital) + if min_trade_value > 0: + threshold = np.maximum(min_trade_value, 0.001 * eq_prev.abs()) + charged = weight_changes.abs().mul(eq_prev.abs(), axis=0).ge(threshold, axis=0) + else: + charged = weight_changes.abs() > 1e-8 if fee_per_share > 0: # Convert per-ticker weight change into share count traded. @@ -107,20 +118,27 @@ def backtest( shares_traded = dollar_traded.div(data).replace( [np.inf, -np.inf], 0.0, ).fillna(0.0) - per_trade_fee = (fee_base + fee_per_share * shares_traded).clip( + # The per-share schedule applies to US-listed shares only. + per_share = pd.Series(fee_per_share, index=data.columns) + per_share[[c for c in data.columns if str(c).endswith(".HK")]] = 0.0 + per_trade_fee = (fee_base + shares_traded.mul(per_share, axis=1)).clip( lower=fixed_fee, ) - trade_mask = weight_changes.abs() > 1e-8 - per_trade_fee = per_trade_fee.where(trade_mask, 0.0) + per_trade_fee = per_trade_fee.where(charged, 0.0) daily_fee = per_trade_fee.sum(axis=1) else: - n_trades = (weight_changes.abs() > 1e-8).sum(axis=1) + n_trades = charged.sum(axis=1) daily_fee = n_trades * fixed_fee fee_impact = daily_fee / eq_prev portfolio_returns -= fee_impact equity = (1 + portfolio_returns).cumprod() * initial_capital + # A portfolio that loses everything stays at zero; compounding negative + # equity would otherwise produce meaningless values and metrics. + ruined = (equity <= 0).cummax() + if ruined.any(): + equity = equity.where(~ruined, 0.0) return equity diff --git a/metrics.py b/metrics.py index c306285..0d45756 100644 --- a/metrics.py +++ b/metrics.py @@ -7,8 +7,16 @@ def total_return(equity: pd.Series) -> float: def annualized_return(equity: pd.Series, periods: int = 252) -> float: - n = len(equity) - return (1 + total_return(equity)) ** (periods / n) - 1 + """CAGR. Uses elapsed calendar time when the index is dated, so gaps in a + live record (missed days) do not inflate the annualized figure.""" + growth = 1 + total_return(equity) + if growth <= 0: + return -1.0 + if isinstance(equity.index, pd.DatetimeIndex) and len(equity) > 1: + years = (equity.index[-1] - equity.index[0]).days / 365.25 + if years > 0: + return growth ** (1 / years) - 1 + return growth ** (periods / len(equity)) - 1 def annualized_volatility(returns: pd.Series, periods: int = 252) -> float: diff --git a/tests/test_live_pipeline.py b/tests/test_live_pipeline.py new file mode 100644 index 0000000..afa06d1 --- /dev/null +++ b/tests/test_live_pipeline.py @@ -0,0 +1,121 @@ +"""Contracts of the live pipeline: price panel refresh, session gating, +order sizing, and how results are reported.""" + +import unittest +from datetime import datetime, timezone + +import numpy as np +import pandas as pd + +import bridge +import data_manager +import trader +from main import backtest + + +def _panel(rows: dict[str, list], start: str = "2026-09-14") -> pd.DataFrame: + idx = pd.bdate_range(start, periods=len(next(iter(rows.values())))) + return pd.DataFrame(rows, index=idx, dtype=float) + + +class PanelRefreshTests(unittest.TestCase): + def test_refresh_overwrites_stale_window_and_keeps_missing_bars_raw(self): + stored = _panel({"AAA": [10, 11, 11, 11, 12], "BBB": [5, 5, 5, 5, 5]}) + window_start = stored.index[2] + fresh = pd.DataFrame( + {"AAA": [11.5, np.nan, 12.2]}, index=stored.index[2:], + ) + merged = data_manager._merge_refresh(stored, fresh, window_start) + self.assertEqual(merged.loc[stored.index[2], "AAA"], 11.5) + self.assertTrue(np.isnan(merged.loc[stored.index[3], "AAA"])) + self.assertEqual(merged.loc[stored.index[4], "AAA"], 12.2) + # Tickers not in the refresh keep their stored values. + self.assertEqual(merged.loc[stored.index[3], "BBB"], 5) + + def test_refresh_rescales_history_after_split_adjustment(self): + stored = _panel({"AAA": [100, 102, 104, 106, 108, 110, 112]}) + window_start = stored.index[2] + # Yahoo now reports the whole window divided by 2 (2:1 split adjusted). + fresh = stored.loc[window_start:, ["AAA"]] / 2 + merged = data_manager._merge_refresh(stored, fresh, window_start) + returns = merged["AAA"].pct_change().dropna() + self.assertTrue((returns > 0).all(), returns) + self.assertAlmostEqual(merged["AAA"].iloc[0], 50.0) + + def test_rows_printed_by_a_minority_of_tickers_are_dropped(self): + panel = _panel({ + "SPY": [1, 1, np.nan, 1, 1], + "UPRO": [1, 1, np.nan, 1, 1], + "GLD": [1, 1, np.nan, 1, 1], + "7200.HK": [1, 1, 1, 1, 1], + }) + cleaned = data_manager._drop_sparse_rows(panel) + self.assertNotIn(panel.index[2], cleaned.index) + self.assertEqual(len(cleaned), 4) + + +class SessionAndOrderTests(unittest.TestCase): + def test_todays_bar_is_not_a_close_until_the_session_settles(self): + raw = _panel({"SPY": [1, 1, 1]}, start="2026-09-22") + today = raw.index[-1] + intraday = datetime(2026, 9, 24, 17, 0, tzinfo=timezone.utc) # 13:00 New York + settled = datetime(2026, 9, 24, 20, 30, tzinfo=timezone.utc) # 16:30 New York + self.assertNotIn(today, trader.completed_sessions(raw, "us", "SPY", now=intraday)) + self.assertIn(today, trader.completed_sessions(raw, "us", "SPY", now=settled)) + + def test_days_without_a_benchmark_print_are_not_sessions(self): + raw = _panel({"SPY": [1, np.nan, 1], "7200.HK": [1, 1, 1]}, start="2026-09-02") + later = datetime(2026, 10, 1, tzinfo=timezone.utc) + sessions = trader.completed_sessions(raw, "etfs", "SPY", now=later) + self.assertNotIn(raw.index[1], sessions) + + def test_a_share_orders_use_board_lots_and_exit_odd_lots(self): + trades = trader.compute_trades( + holdings={"600000.SS": 150}, cash=100_000.0, + target_weights={"600519.SS": 0.5}, prices={"600519.SS": 170.0, "600000.SS": 10.0}, + min_trade_value=50, integer_shares=True, lot_size=100, + ) + by_ticker = {t["ticker"]: t["shares_delta"] for t in trades} + self.assertEqual(by_ticker["600519.SS"] % 100, 0) + self.assertEqual(by_ticker["600000.SS"], -150) + + def test_tickers_without_a_fresh_price_are_not_traded(self): + trades = trader.compute_trades( + holdings={"UPRO": 10}, cash=5_000.0, target_weights={"TQQQ": 1.0}, + prices={"UPRO": 150.0, "TQQQ": 80.0}, frozen={"UPRO"}, + ) + self.assertEqual([t["ticker"] for t in trades], ["TQQQ"]) + + +class ReportingTests(unittest.TestCase): + def test_periodic_returns_include_the_first_day_of_each_period(self): + idx = pd.to_datetime(["2026-01-30", "2026-02-02", "2026-02-27", "2026-03-02"]) + equity = pd.Series([100.0, 110.0, 110.0, 121.0], index=idx) + monthly, yearly = bridge.compute_periodic_returns(equity) + self.assertAlmostEqual(monthly["2026-02"], 0.10) + self.assertAlmostEqual(monthly["2026-03"], 0.10) + self.assertAlmostEqual(yearly["2026"], 0.21) + + def test_injected_cash_is_not_reported_as_performance(self): + idx = pd.bdate_range("2024-01-02", periods=400) + equity = pd.Series(10_000 * 1.0003 ** np.arange(400), index=idx) + schedule = {"amount": 5_000, "frequency": "yearly_months", "months": [2, 8], "startYear": 1} + plain = bridge._build_strategy_result("x", equity, [], None, 10_000) + injected = bridge._build_strategy_result("x", equity, [], schedule, 10_000) + self.assertAlmostEqual(injected["metrics"]["totalReturn"], plain["metrics"]["totalReturn"]) + flows = injected["capitalFlows"] + self.assertEqual(flows["invested"], 10_000 + 5_000 * len(injected["injections"])) + self.assertAlmostEqual(flows["profit"], flows["finalValue"] - flows["invested"], places=2) + + def test_ruined_portfolio_stays_at_zero(self): + class TwiceLevered: + def generate_signals(self, data): + return pd.DataFrame(2.0, index=data.index, columns=data.columns) + + prices = pd.DataFrame({"X": [10.0, 10.0, 4.0, 5.0]}, index=pd.bdate_range("2026-01-01", periods=4)) + equity = backtest(TwiceLevered(), prices, initial_capital=100) + self.assertTrue((equity.iloc[2:] == 0).all(), equity) + + +if __name__ == "__main__": + unittest.main() diff --git a/universe_history.py b/universe_history.py index b8fc916..b04df1a 100644 --- a/universe_history.py +++ b/universe_history.py @@ -38,7 +38,13 @@ def _fetch_sp500_tables() -> tuple[pd.DataFrame, pd.DataFrame]: html = resp.read().decode("utf-8") tables = pd.read_html(io.StringIO(html)) current = tables[0] - changes = tables[1] + change_tables = [ + t for t in tables[1:] + if any("Added" in "_".join(map(str, c if isinstance(c, tuple) else (c,))) for c in t.columns) + ] + if not change_tables: + raise ValueError("S&P 500 change log table not found on the Wikipedia page") + changes = change_tables[0] changes.columns = [ "_".join(c).strip() if isinstance(c, tuple) else c for c in changes.columns @@ -161,7 +167,16 @@ def load_sp500_history(force_refresh: bool = False) -> dict[str, list[list[str | except Exception: pass print("--- Rebuilding S&P 500 membership history from Wikipedia ---") - intervals = build_sp500_history() + try: + intervals = build_sp500_history() + except Exception as exc: # noqa: BLE001 — page layout changes must not break backtests + if not os.path.exists(path): + raise + with open(path) as f: + cached = json.load(f) + print(f"--- WARNING: rebuild failed ({exc}); using membership history " + f"cached on {cached.get('date')} ---") + return cached["intervals"] os.makedirs(CACHE_DIR, exist_ok=True) with open(path, "w") as f: json.dump({"date": str(date.today()), "intervals": intervals}, f)