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) <noreply@anthropic.com>
This commit is contained in:
66
trader.py
66
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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user