@ -18,6 +18,7 @@ import tempfile
from pathlib import Path
from typing import Any
from libs . common . config import get_settings
from libs . common . logging import get_logger
logger = get_logger ( __name__ )
@ -124,7 +125,7 @@ def _convert_from_runner(
pnl_pct = float ( row . get ( " pnl_pct " , 0.0 ) )
pnl_dollar = pnl_pct * float ( entry_px or 0 ) * shares if entry_px else 0.0
trade s. append ( {
trade = {
" symbol " : str ( row . get ( " symbol " , " " ) ) ,
" entry_date " : str ( row . get ( " entry_date " , " - " ) ) ,
" exit_date " : str ( row . get ( " exit_date " , " - " ) ) ,
@ -136,7 +137,11 @@ def _convert_from_runner(
" event_type " : str ( row . get ( " event_type " , " - " ) ) ,
" score " : float ( row . get ( " score " , 0.0 ) ) ,
" engine_id " : str ( row . get ( " engine_id " , " " ) ) ,
} )
}
# Skip same-day KILL_SWITCH — backtest period end artifact
if trade [ " entry_date " ] == trade [ " exit_date " ] and trade [ " reason " ] == " KILL_SWITCH " :
continue
trades . append ( trade )
except Exception as exc :
logger . warning ( " backtest_sim_artifact_load_failed " , error = str ( exc ) )
@ -192,41 +197,78 @@ def _convert_from_runner(
def _snapshot_needs_refresh (
snapshot_id : str ,
end_date : dt . date ,
snapshot_dir : str = " data/datasets/snapshots " ,
snapshot_dir : str | None = None ,
) - > bool :
""" Check if the Parquet snapshot is stale (doesn' t cover end_date) ."""
import json
manifest_path = Path ( snapshot_dir ) / snapshot_id / " manifest.json "
if not manifest_path . exists ( ) :
return True
""" Refresh only when no existing snapshot covers the requested end date ."""
return not _snapshot_has_required_coverage (
snapshot_id = snapshot_id ,
end_date = end_date ,
snapshot_dir = snapshot_dir ,
)
try :
manifest = json . loads ( manifest_path . read_text ( ) )
created = manifest . get ( " created_at_utc " , " " ) [ : 10 ]
if created and dt . date . fromisoformat ( created ) < end_date - dt . timedelta ( days = 7 ) :
return True
except Exception :
return True
# Check if the latest event_date in the data covers end_date
train_path = Path ( snapshot_dir ) / snapshot_id / " train.parquet "
test_path = Path ( snapshot_dir ) / snapshot_id / " test.parquet "
latest_path = test_path if test_path . exists ( ) else train_path
if not latest_path . exists ( ) :
return True
def _snapshot_has_required_coverage (
snapshot_id : str ,
end_date : dt . date ,
snapshot_dir : str | None = None ,
) - > bool :
""" Return True when an existing snapshot already covers the requested date. """
snapshot_path = _resolve_snapshot_path ( snapshot_id , snapshot_dir = snapshot_dir )
if snapshot_path is None :
return False
train_path = snapshot_path / " train.parquet "
valid_path = snapshot_path / " valid.parquet "
test_path = snapshot_path / " test.parquet "
parquet_paths = [ path for path in ( test_path , valid_path , train_path ) if path . exists ( ) ]
if not parquet_paths :
return False
try :
import pyarrow . parquet as pq
table = pq . read_table ( str ( latest_path ) , columns = [ " event_date " ] )
dates = table . column ( " event_date " ) . to_pylist ( )
max_date = max ( dates ) if dates else " "
if isinstance ( max_date , str ) :
max_date = dt . date . fromisoformat ( max_date [ : 10 ] )
# Stale if snapshot's latest event is more than 14 days before end_date
return max_date < end_date - dt . timedelta ( days = 14 )
max_date : dt . date | None = None
for parquet_path in parquet_paths :
table = pq . read_table ( str ( parquet_path ) , columns = [ " event_date " ] )
dates = table . column ( " event_date " ) . to_pylist ( )
if not dates :
continue
candidate = max ( dates )
if isinstance ( candidate , str ) :
candidate = dt . date . fromisoformat ( candidate [ : 10 ] )
if isinstance ( candidate , dt . datetime ) :
candidate = candidate . date ( )
if isinstance ( candidate , dt . date ) and ( max_date is None or candidate > max_date ) :
max_date = candidate
if max_date is None :
return False
return max_date > = end_date - dt . timedelta ( days = 14 )
except Exception :
return True
return False
def _resolve_snapshot_path (
snapshot_id : str ,
snapshot_dir : str | None = None ,
) - > Path | None :
""" Resolve the on-disk snapshot directory using the same fallback order as the runner. """
candidates : list [ Path ] = [ ]
if snapshot_dir is not None :
candidates . append ( Path ( snapshot_dir ) / snapshot_id )
else :
settings = get_settings ( )
candidates . append ( Path ( settings . parquet_dir ) / snapshot_id )
candidates . append ( Path ( " data/datasets/snapshots " ) / snapshot_id )
seen : set [ Path ] = set ( )
for candidate in candidates :
candidate = candidate . resolve ( )
if candidate in seen :
continue
seen . add ( candidate )
if candidate . exists ( ) :
return candidate
return None
async def _refresh_snapshot (
@ -309,6 +351,310 @@ async def _refresh_snapshot(
raise
# ── Overlay backtest support ──────────────────────────────────────────
def _is_overlay_config ( config_path : str ) - > bool :
""" Return True if config_path is an overlay spec (has ' books ' key). """
import json
try :
data = json . loads ( Path ( config_path ) . read_text ( ) )
return " books " in data and " allocations " in data
except Exception :
return False
def _resolve_book_experiment_config ( book : dict ) - > str | None :
""" Resolve the experiment config path for an overlay book entry. """
# Explicit field
explicit = book . get ( " experiment_config " )
if explicit and Path ( explicit ) . exists ( ) :
return explicit
# Infer from equity_csv filename
csv_path = book . get ( " equity_csv " , " " )
if csv_path :
name = Path ( csv_path ) . stem # e.g. "return_max_long_v6.221_equity"
# Strip common suffixes
for suffix in ( " _equity " , " _train " , " _valid " , " _test " ) :
if name . endswith ( suffix ) :
name = name [ : - len ( suffix ) ]
break
candidate = f " configs/experiments/ { name } .json "
if Path ( candidate ) . exists ( ) :
return candidate
return None
def _overlay_books_are_runnable ( overlay_config_path : str ) - > bool :
""" Check if all books in an overlay config have resolvable experiment configs. """
import json
try :
spec = json . loads ( Path ( overlay_config_path ) . read_text ( ) )
for book in spec . get ( " books " , [ ] ) :
csv_path = book . get ( " equity_csv " )
if csv_path and Path ( csv_path ) . exists ( ) :
continue
if _resolve_book_experiment_config ( book ) is None :
return False
return True
except Exception :
return False
def _rebase_equity_slice (
df ,
* ,
initial_equity : float ,
) :
""" Recompute equity within a requested window so the first kept day starts flat. """
df = df . sort_values ( " date " ) . copy ( )
df [ " daily_return " ] = df [ " equity " ] . astype ( float ) . pct_change ( ) . fillna ( 0.0 )
equity = float ( initial_equity )
rebased : list [ float ] = [ ]
for ret in df [ " daily_return " ] . astype ( float ) :
equity * = 1.0 + float ( ret )
rebased . append ( equity )
df [ " equity " ] = rebased
return df [ [ " date " , " equity " , " daily_return " ] ]
def _summarize_book_curve ( df , * , initial_equity : float ) - > dict [ str , float ] :
""" Return a paper-backtest-like summary from a rebased equity curve. """
returns = df [ " daily_return " ] . astype ( float )
final_equity = float ( df [ " equity " ] . iloc [ - 1 ] )
return_pct = ( final_equity / float ( initial_equity ) - 1.0 ) * 100.0
peak = float ( initial_equity )
max_dd_pct = 0.0
for equity in df [ " equity " ] . astype ( float ) :
peak = max ( peak , float ( equity ) )
drawdown_pct = ( peak - float ( equity ) ) / peak * 100.0 if peak > 0 else 0.0
max_dd_pct = max ( max_dd_pct , drawdown_pct )
if len ( returns ) > = 2 and float ( returns . std ( ) ) > 0 :
sharpe = float ( returns . mean ( ) / returns . std ( ) * math . sqrt ( 252.0 ) )
else :
sharpe = 0.0
return {
" return_pct " : return_pct ,
" final_equity " : final_equity ,
" max_dd_pct " : max_dd_pct ,
" trade_count " : 0 ,
" win_rate " : 0.0 ,
" sharpe " : sharpe ,
}
def _load_overlay_book_curve_from_spec (
book : dict ,
* ,
capital : float ,
start_date : dt . date ,
end_date : dt . date ,
) :
""" Load a frozen overlay input curve from equity_csv and rebase it to the requested window. """
from libs . backtest . overlay import load_equity_curve_csv
csv_path = book . get ( " equity_csv " )
if not csv_path or not Path ( csv_path ) . exists ( ) :
return None
df = load_equity_curve_csv ( csv_path )
df = df [ ( df [ " date " ] > = start_date ) & ( df [ " date " ] < = end_date ) ] . copy ( )
if df . empty :
return None
rebased = _rebase_equity_slice ( df , initial_equity = capital )
summary = _summarize_book_curve ( rebased , initial_equity = capital )
return {
" curve " : rebased ,
" summary " : summary ,
" source " : " equity_csv " ,
}
def run_overlay_backtest_sync (
overlay_config_path : str ,
capital : float ,
start_date : dt . date ,
end_date : dt . date ,
console = None ,
) - > dict [ str , Any ] :
""" Run an overlay backtest: execute each book strategy, then combine by regime. """
import json
import pandas as pd
from libs . backtest . overlay import build_overlay_curve , summarize_overlay_curve
spec = json . loads ( Path ( overlay_config_path ) . read_text ( ) )
overlay_name = spec . get ( " overlay_name " , Path ( overlay_config_path ) . stem )
allocations = spec [ " allocations " ]
# ── Run each book strategy ────────────────────────────────────────
book_results : list [ dict [ str , Any ] ] = [ ]
curves : dict [ str , pd . DataFrame ] = { }
replay_mode = " frozen_equity_csv "
for book in spec [ " books " ] :
label = book [ " label " ]
loaded = _load_overlay_book_curve_from_spec (
book ,
capital = capital ,
start_date = start_date ,
end_date = end_date ,
)
if loaded is not None :
if console :
source_name = Path ( book [ " equity_csv " ] ) . stem
console . print ( f " [dim]Book ' { label } ' :[/] { source_name } [dim](frozen equity_csv)[/] " )
curves [ label ] = loaded [ " curve " ]
book_results . append (
{
" label " : label ,
" result " : {
" session_name " : f " { overlay_name } __ { label } " ,
" summary " : loaded [ " summary " ] ,
" equity_curve " : [
{ " date " : row . date , " equity " : row . equity }
for row in loaded [ " curve " ] . itertuples ( index = False )
] ,
" trades " : [ ] ,
} ,
" source " : loaded [ " source " ] ,
}
)
continue
replay_mode = " rerun_books "
exp_config = _resolve_book_experiment_config ( book )
if exp_config is None :
raise ValueError (
f " Overlay ' { overlay_name } ' : book ' { label } ' has neither a usable equity_csv nor a resolvable experiment config. "
f " Add ' equity_csv ' or ' experiment_config ' to the book entry. "
)
if console :
console . print ( f " [dim]Book ' { label } ' :[/] { Path ( exp_config ) . stem } [dim](rerun)[/] " )
result = run_backtest_session_sync (
session_name = f " { overlay_name } __ { label } " ,
config_path = exp_config ,
initial_equity = capital ,
start_date = start_date ,
end_date = end_date ,
)
book_results . append ( { " label " : label , " result " : result , " source " : " rerun " } )
eq = result . get ( " equity_curve " , [ ] )
if eq :
df = pd . DataFrame ( eq )
df [ " date " ] = pd . to_datetime ( df [ " date " ] ) . dt . date
df [ " equity " ] = df [ " equity " ] . astype ( float )
curves [ label ] = _rebase_equity_slice ( df [ [ " date " , " equity " ] ] , initial_equity = capital )
if not curves :
raise ValueError ( f " Overlay ' { overlay_name } ' : no book produced equity curves " )
# ── Compute regime for each trading day ───────────────────────────
regimes = _compute_overlay_regimes ( spec , start_date , end_date )
# ── Combine using overlay logic ───────────────────────────────────
overlay_curve = build_overlay_curve (
curves = curves ,
allocations = allocations ,
regimes_by_date = regimes ,
initial_equity = capital ,
)
summary = summarize_overlay_curve ( overlay_curve , initial_equity = capital )
# Convert overlay equity curve to standard format
equity_curve = [
{ " date " : row . date , " equity " : row . overlay_equity }
for row in overlay_curve . itertuples ( index = False )
]
# Aggregate trade count across books
total_trades = sum (
br [ " result " ] [ " summary " ] [ " trade_count " ] for br in book_results
)
return {
" session_name " : overlay_name ,
" config_path " : overlay_config_path ,
" initial_equity " : capital ,
" is_overlay " : True ,
" overlay_replay_mode " : replay_mode ,
" equity_curve " : equity_curve ,
" trades " : [ ] ,
" book_results " : book_results ,
" allocations " : allocations ,
" regime_day_counts " : summary . get ( " regime_day_counts " , { } ) ,
" summary " : {
" return_pct " : summary [ " return_pct " ] ,
" final_equity " : summary [ " final_equity " ] ,
" max_dd_pct " : summary [ " max_dd_pct " ] ,
" trade_count " : total_trades ,
" win_rate " : 0.0 ,
" sharpe " : summary [ " sharpe " ] ,
} ,
}
def _compute_overlay_regimes (
spec : dict ,
start_date : dt . date ,
end_date : dt . date ,
) - > dict [ dt . date , str ] :
""" Compute macro regime for each trading day using the regime_source config.
Uses _build_merged_snapshot_store to get a full - period store with macro data ,
covering the paper backtest date range ( not just the original snapshot period ) .
"""
from apps . backtester . run import _build_merged_snapshot_store , load_manifest , resolve_config
from libs . backtest . allocator import _macro_regime_state
from libs . backtest . overlay import load_merged_store_from_snapshot_dir
from libs . common . config import get_settings
regime_source = spec . get ( " regime_source " , { } )
config_path = regime_source . get ( " config_path " )
if not config_path :
return { }
manifest = load_manifest ( config_path )
config = resolve_config ( manifest )
raw_snapshot_dir = regime_source . get ( " snapshot_dir " )
if raw_snapshot_dir and (
( Path ( raw_snapshot_dir ) / " train.parquet " ) . exists ( )
or ( Path ( raw_snapshot_dir ) / " test.parquet " ) . exists ( )
) :
settings = get_settings ( )
store = load_merged_store_from_snapshot_dir (
raw_snapshot_dir ,
oracle_url = settings . stock_oracle_url ,
db_dsn = settings . postgres_dsn ,
)
else :
try :
store = _build_merged_snapshot_store (
manifest ,
config ,
snapshot_dir_override = raw_snapshot_dir ,
)
except FileNotFoundError :
store = _build_merged_snapshot_store ( manifest , config , snapshot_dir_override = None )
store = store . slice_by_date_range ( start_date , end_date )
regimes : dict [ dt . date , str ] = { }
for date in store . all_trading_days ( ) :
regimes [ date ] = _macro_regime_state ( config , store . get_macro_for_date ( date ) )
return regimes
def run_backtest (
configs : list [ str ] ,
capital : float ,
@ -345,6 +691,8 @@ def run_backtest(
# Check if snapshots need refresh (async pipeline, run before sync backtest)
for config_path in configs :
if _is_overlay_config ( config_path ) :
continue # overlay books handle their own snapshots
from apps . backtester . run import load_manifest , resolve_config
manifest = load_manifest ( config_path )
config = resolve_config ( manifest )
@ -361,23 +709,42 @@ def run_backtest(
if console :
console . print ( f " \n [bold yellow]Snapshot ' { snapshot_id } ' is stale — refreshing...[/] " )
asyncio . run ( _refresh_snapshot ( snapshot_id , universe_profile , console = console ) )
try :
asyncio . run ( _refresh_snapshot ( snapshot_id , universe_profile , console = console ) )
except Exception :
if _snapshot_has_required_coverage ( snapshot_id , end_date ) :
if console :
console . print ( " [yellow]Refresh failed, but existing snapshot still covers the requested period. Using current snapshot.[/] " )
else :
raise
configure_logging ( " WARNING " )
results = [ ]
for config_path in configs :
session_name = Path ( config_path ) . stem
if console :
console . print ( f " \n [bold cyan]Running:[/] { session_name } " )
result = run_backtest_session_sync (
session_name = session_name ,
config_path = config_path ,
initial_equity = capital ,
start_date = start_date ,
end_date = end_date ,
)
if _is_overlay_config ( config_path ) :
if console :
console . print ( f " \n [bold magenta]Running overlay:[/] { session_name } " )
result = run_overlay_backtest_sync (
overlay_config_path = config_path ,
capital = capital ,
start_date = start_date ,
end_date = end_date ,
console = console ,
)
else :
if console :
console . print ( f " \n [bold cyan]Running:[/] { session_name } " )
result = run_backtest_session_sync (
session_name = session_name ,
config_path = config_path ,
initial_equity = capital ,
start_date = start_date ,
end_date = end_date ,
)
results . append ( result )
if console and result [ " summary " ] [ " trade_count " ] > 0 :
I Luk Kim
5 months ago