From 905ca4fe050433e0e033a301c2073635ee95cc0a Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 25 Sep 2026 20:29:39 +0800 Subject: [PATCH] fix(trader): replay and catch-up trade each day's point-in-time index members Replaying live books over past sessions used today's constituent list for every day, so US books bought names (BE, MRVL, ILMN, FLEX, P) before they joined the S&P 500. session_universes/session_signals now compute each past session on that day's members (S&P change log, CSI 300 snapshots); the latest session keeps the live list. Used by auto/monitor catch-up, replay and simulate. _run_session takes precomputed targets only. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_robust_strategies.py | 17 +++++ trader.py | 111 +++++++++++++++++++++++++------- 2 files changed, 104 insertions(+), 24 deletions(-) diff --git a/tests/test_robust_strategies.py b/tests/test_robust_strategies.py index f9d9c02..791f4b4 100644 --- a/tests/test_robust_strategies.py +++ b/tests/test_robust_strategies.py @@ -174,6 +174,23 @@ class MembershipTests(unittest.TestCase): # Never-member columns (benchmarks) are not masked. self.assertTrue(mask["510300.SS"].all()) + def test_past_sessions_never_hold_a_name_before_it_joined_the_index(self): + bars = _bars() + bars["close"]["S00"] *= np.exp(np.linspace(0, 3, len(bars["close"]))) # the best momentum + bars["open"]["S00"] = bars["close"]["S00"] + days = bars["close"].index[-5:] + others = [c for c in bars["close"].columns if c != "S00"] + members = pd.DataFrame(True, index=days[:-1], columns=bars["close"].columns) + members["S00"] = False # joins the index on the last day + with mock.patch.object(trader, "index_members", lambda m, d: members.loc[d]): + runs = trader.session_universes("us", "momentum", others + ["S00"], days) + weights = trader.session_signals("us", "momentum", others + ["S00"], days, + bars["close"], bars["open"]) + self.assertEqual([len(r) for _, r in runs], [4, 1]) + self.assertNotIn("S00", runs[0][0]) + self.assertTrue((weights.loc[days[:-1], "S00"] == 0).all()) + self.assertGreater(weights.loc[days[-1], "S00"], 0) + def test_changes_table_with_two_level_headers_is_found(self): cols = pd.MultiIndex.from_tuples([("Effective Date", "Effective Date"), ("Added", "Ticker"), ("Removed", "Ticker"), ("Reason", "Reason")]) diff --git a/trader.py b/trader.py index 02218dc..e4db54d 100644 --- a/trader.py +++ b/trader.py @@ -40,6 +40,7 @@ import pandas as pd import yfinance as yf import data_manager +import universe_history from strategies.base import BarStrategy from strategies.buy_and_hold import BuyAndHoldStrategy from strategies.dual_momentum import DualMomentumStrategy @@ -326,6 +327,70 @@ def strategy_universe(market: str, strategy_name: str) -> tuple[list[str], str]: return tickers, universe["benchmark"] +def index_members(market: str, days: pd.DatetimeIndex) -> pd.DataFrame | None: + """Point-in-time index membership (``days`` x every ticker that was a member + on one of them), or None for a market without membership history.""" + if market == "us": + mask = universe_history.membership_mask(days, tickers=universe_history.all_tickers_ever()) + elif market == "cn": + snaps = universe_history.load_csi300_history() + mask = universe_history.csi300_membership_mask( + days, universe_history.all_csi300_tickers_ever(snaps), snaps) + else: + return None + return mask.loc[:, mask.any()] + + +def session_universes(market: str, strategy_name: str, tickers: list[str], + days: pd.DatetimeIndex) -> list[tuple[list[str], pd.DatetimeIndex]]: + """Split ``days`` into runs of sessions that share one stock universe. + + The latest session trades ``tickers`` (today's list). Earlier sessions + trade the index members of their own day: a replay or catch-up over past + sessions with today's list would buy names before they joined the index, + i.e. names that were added because they had risen. ETF strategies keep + their fixed basket. + """ + if strategy_name.removeprefix("sim_") in ETF_STRATEGY_UNIVERSES or len(days) < 2: + return [(list(tickers), days)] + members = index_members(market, days[:-1]) + if members is None: + return [(list(tickers), days)] + extras = set(strategy_extra_tickers(strategy_name)) + runs: list[tuple[list[str], list]] = [] + for day in days: + if day == days[-1]: + universe = sorted(tickers) + else: + universe = sorted(set(members.columns[members.loc[day].to_numpy()]) | extras) + if runs and runs[-1][0] == universe: + runs[-1][1].append(day) + else: + runs.append((universe, [day])) + return [(u, pd.DatetimeIndex(d)) for u, d in runs] + + +def pit_candidates(market: str, strategy_name: str, since: pd.Timestamp | None) -> list[str]: + """Tickers that were index members on any day from ``since`` to today, so a + catch-up or replay has prices for names that have left the index since.""" + if since is None or strategy_name.removeprefix("sim_") in ETF_STRATEGY_UNIVERSES: + return [] + members = index_members(market, pd.bdate_range(since, pd.Timestamp.today().normalize())) + return [] if members is None else list(members.columns) + + +def session_signals(market: str, strategy_name: str, tickers: list[str], days: pd.DatetimeIndex, + close: pd.DataFrame, open_data: pd.DataFrame) -> pd.DataFrame: + """Target weights for each day in ``days``, each computed on that day's + point-in-time universe (see ``session_universes``); columns are the union.""" + parts = [] + for universe, run in session_universes(market, strategy_name, tickers, days): + universe = filter_tradable_tickers(close, universe) + bars = strategy_bars(market, strategy_name, universe, close, open_data) + parts.append(_signals_for_sessions(strategy_name, open_data[universe], run, bars)) + return pd.concat(parts).fillna(0.0) + + def strategy_data_market(market: str, strategy_name: str) -> str: """Return the cache namespace used for a strategy's price data.""" base_name = strategy_name.removeprefix("sim_") @@ -796,22 +861,18 @@ def _signals_for_sessions(strategy_name: str, open_data: pd.DataFrame, return weights.shift(-1).reindex(days).fillna(0.0) -def _run_session(state: dict, strategy_name: str, market: str, day: pd.Timestamp, - tickers: list[str], benchmark: str, close_data: pd.DataFrame, - open_data: pd.DataFrame, close_raw: pd.DataFrame, fees: dict, - integer_shares: bool, signal_row: pd.Series | None = None, - bars: dict | None = None) -> dict: - """Signal on ``day``'s open (a BarStrategy: bars through the previous - session), execute at ``day``'s close, record the day. +def _run_session(state: dict, market: str, day: pd.Timestamp, benchmark: str, + close_data: pd.DataFrame, close_raw: pd.DataFrame, fees: dict, + integer_shares: bool, signal_row: pd.Series) -> dict: + """Execute ``signal_row`` (targets from ``day``'s open; a BarStrategy: bars + through the previous session, see ``session_signals``) at ``day``'s close + and record the day. Tickers without a real closing print on ``day`` (forward-filled price) are - valued but not traded. ``signal_row`` supplies precomputed target weights. + valued but not traded. """ day_str = str(day.date()) lot_size = MARKET_LOT_SIZE.get(market, 1) if integer_shares else 1 - if signal_row is None: - signal_row = _signals_for_sessions(strategy_name, open_data[tickers], - pd.DatetimeIndex([day]), bars).loc[day] target_weights = {t: float(w) for t, w in signal_row.items() if w > 1e-6} all_held = sorted(set(target_weights) | set(state["holdings"])) @@ -1366,16 +1427,17 @@ def cmd_simulate(args): market = args.market strategy_name = args.strategy tickers, benchmark = strategy_universe(market, strategy_name) + start = pd.Timestamp(args.start) + end = pd.Timestamp(args.end) # Load both open and close data + request = sorted(set(tickers) | set(pit_candidates(market, strategy_name, start))) close_data, open_data, _ = load_strategy_prices( - market, strategy_name, tickers, benchmark, with_open=True, + market, strategy_name, request, benchmark, with_open=True, ) tickers = filter_tradable_tickers(close_data, tickers) # Date range - start = pd.Timestamp(args.start) - end = pd.Timestamp(args.end) trading_days = close_data.index[(close_data.index >= start) & (close_data.index <= end)] if len(trading_days) == 0: @@ -1385,10 +1447,10 @@ def cmd_simulate(args): # Fresh state for simulation state = init_state(market, f"sim_{strategy_name}", args.capital) - # Pre-compute strategy weights for the full period (much faster than per-day) - # Append dummy row so shift(1) inside strategy produces valid last-day signal - bars = strategy_bars(market, strategy_name, tickers, close_data, open_data) - full_weights = _signals_for_sessions(strategy_name, open_data[tickers], trading_days, bars) + # Pre-compute strategy weights for the full period (much faster than per-day), + # each day on its point-in-time index members + full_weights = session_signals(market, strategy_name, tickers, trading_days, + close_data, open_data) print(f"\n{'='*70}") print(f" SIMULATION: {strategy_name} | {market.upper()}") @@ -1778,8 +1840,11 @@ def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = Non print(f"[{label}] {strategy_name} does not run in the {market.upper()} book; skipping.") return None tickers, benchmark = strategy_universe(market, strategy_name) + since = start if not state else ( + pd.Timestamp(state["last_evening"]) if state.get("last_evening") else None) + request = sorted(set(tickers) | set(pit_candidates(market, strategy_name, since))) close_data, open_data, close_raw = load_strategy_prices( - market, strategy_name, tickers, benchmark, with_open=True, refresh=refresh, + market, strategy_name, request, benchmark, with_open=True, refresh=refresh, ) tickers = filter_tradable_tickers(close_data, tickers) sessions = completed_sessions(close_raw, strategy_data_market(market, strategy_name), benchmark) @@ -1805,12 +1870,10 @@ def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = Non f"{todo[0].date()} .. {todo[-1].date()}") fees = _strategy_fees(args, market) - bars = strategy_bars(market, strategy_name, tickers, close_data, open_data) - signals = _signals_for_sessions(strategy_name, open_data[tickers], todo, bars) + signals = session_signals(market, strategy_name, tickers, todo, close_data, open_data) for day in todo: - r = _run_session(state, strategy_name, market, day, tickers, benchmark, - close_data, open_data, close_raw, fees, args.integer_shares, - signal_row=signals.loc[day], bars=bars) + r = _run_session(state, market, day, benchmark, close_data, close_raw, fees, + args.integer_shares, signal_row=signals.loc[day]) frozen = f" | no fresh price: {', '.join(r['frozen'])}" if r["frozen"] else "" print(f"[{label}] {r['date']} {strategy_name:<34} value {r['value']:>12,.2f} " f"({r['daily_return']:+.2%}) trades {r['trades']:>2} "