From 592de02e27401860bca8e3637cc2e70546d64756 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 25 Sep 2026 01:56:32 +0800 Subject: [PATCH] perf(replay): refresh panels once and compute signals in one pass Replay refreshes each price panel once for the union of tickers and then replays offline; target weights for all sessions come from one causal signal pass instead of re-running the strategy per day (verified identical daily equity for v7_vt36, recovery_mom_top10 and fc monthly). Co-Authored-By: Claude Opus 5.5 (1M context) --- trader.py | 66 ++++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 53 insertions(+), 13 deletions(-) diff --git a/trader.py b/trader.py index 8f264ea..8f0cc67 100644 --- a/trader.py +++ b/trader.py @@ -337,17 +337,23 @@ def price_request(market: str, tickers: list[str]) -> list[str]: def load_strategy_prices(market: str, strategy_name: str, tickers: list[str], - benchmark: str, with_open: bool = False): - """Refresh and load a strategy's prices in its portfolio currency. + benchmark: str, with_open: bool = False, refresh: bool = True): + """Load a strategy's prices in its portfolio currency, refreshing them first + unless ``refresh`` is False (the panel is then read as stored). Returns ``(close, open_or_None, close_raw)``. ``close_raw`` keeps NaN where the exchange printed no bar that day, so callers can distinguish a stale forward-filled price from a real one. """ data_market = strategy_data_market(market, strategy_name) - request = price_request(market, list(tickers) + [benchmark]) - result = data_manager.update(data_market, request, with_open=with_open) - close, open_data = result if with_open else (result, None) + if refresh: + request = price_request(market, list(tickers) + [benchmark]) + result = data_manager.update(data_market, request, with_open=with_open) + close, open_data = result if with_open else (result, None) + else: + close = data_manager.load(data_market, "close") + open_data = (data_manager._fill_open(close, data_manager.load_raw(data_market, "open")) + if with_open else None) close_raw = data_manager.load_raw(data_market, "close").reindex( index=close.index, columns=close.columns, ) @@ -723,19 +729,37 @@ def _strategy_fees(args, market: str) -> dict: } +def _signals_for_sessions(strategy_name: str, open_data: pd.DataFrame, + days: pd.DatetimeIndex) -> pd.DataFrame: + """Target weights for many sessions in one pass: row ``d`` uses opens up to ``d``. + + Equivalent to ``generate_target_weights`` per day for causal strategies + (weights at t never depend on data after t), which every registered + strategy must be. + """ + hist = open_data.loc[:days[-1]] + dummy = pd.DataFrame([hist.iloc[-1].values], index=[hist.index[-1] + pd.Timedelta(days=1)], + columns=hist.columns) + weights = STRATEGY_REGISTRY[strategy_name]().generate_signals(pd.concat([hist, dummy])) + 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) -> dict: + integer_shares: bool, signal_row: pd.Series | None = None) -> dict: """Signal on ``day``'s open, execute at ``day``'s close, record the day. Tickers without a real closing print on ``day`` (forward-filled price) are - valued but not traded. + valued but not traded. ``signal_row`` supplies precomputed target weights. """ day_str = str(day.date()) lot_size = MARKET_LOT_SIZE.get(market, 1) if integer_shares else 1 - strategy = STRATEGY_REGISTRY[strategy_name]() - target_weights = generate_target_weights(strategy, open_data.loc[:day, tickers], day) + if signal_row is None: + strategy = STRATEGY_REGISTRY[strategy_name]() + target_weights = generate_target_weights(strategy, open_data.loc[:day, tickers], day) + else: + 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"])) close_prices = _with_last_known_prices( @@ -1672,7 +1696,7 @@ def cmd_monitor(args): def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = None, - label: str = "auto") -> dict | None: + label: str = "auto", refresh: bool = True) -> dict | None: """Run every completed session not yet recorded in ``state`` (catch-up safe). A fresh portfolio starts at ``start`` when given, otherwise at the most @@ -1685,7 +1709,7 @@ def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = Non return None tickers, benchmark = strategy_universe(market, strategy_name) close_data, open_data, close_raw = load_strategy_prices( - market, strategy_name, tickers, benchmark, with_open=True, + market, strategy_name, tickers, 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) @@ -1711,9 +1735,12 @@ def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = Non f"{todo[0].date()} .. {todo[-1].date()}") fees = _strategy_fees(args, market) + signals = (_signals_for_sessions(strategy_name, open_data[tickers], todo) + if len(todo) > 1 else None) for day in todo: r = _run_session(state, strategy_name, market, day, tickers, benchmark, - close_data, open_data, close_raw, fees, args.integer_shares) + close_data, open_data, close_raw, fees, args.integer_shares, + signal_row=signals.loc[day] if signals is not None else None) 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} " @@ -1755,6 +1782,19 @@ def cmd_replay(args): os.path.basename(p)[len(f"trader_{args.market}_"):-len(".json")] for p in __import__("glob").glob(os.path.join("data", f"trader_{args.market}_*.json")) ) + live_names = [n for n in names if not n.startswith("sim_") and n in STRATEGY_REGISTRY + and n not in LIVE_EXCLUDED_STRATEGIES and strategy_supports_market(n, args.market)] + + # Refresh each price panel once for the union of tickers, then replay offline. + requests: dict[str, set[str]] = {} + for name in live_names: + tickers, benchmark = strategy_universe(args.market, name) + requests.setdefault(strategy_data_market(args.market, name), set()).update( + price_request(args.market, list(tickers) + [benchmark])) + for data_market, request in requests.items(): + print(f"[replay] Refreshing {data_market} panel for {len(request)} tickers") + data_manager.update(data_market, sorted(request), with_open=True) + for name in names: if name.startswith("sim_"): continue @@ -1774,7 +1814,7 @@ def cmd_replay(args): if start is None: print(f"[replay] {name}: no start date; skipping") continue - state = _process_sessions(sub, None, start=start, label="replay") + state = _process_sessions(sub, None, start=start, label="replay", refresh=False) if state is not None: save_state(state, args.market, name)