|
|
"""BacktestRunner: main simulation class and CLI entry point."""
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import argparse
|
|
|
import datetime as dt
|
|
|
import subprocess
|
|
|
import sys
|
|
|
from pathlib import Path
|
|
|
from typing import Any
|
|
|
|
|
|
from libs.backtest.allocator import build_planned_order
|
|
|
from libs.backtest.artifacts import create_run_directory, write_all_artifacts
|
|
|
from libs.backtest.domain import (
|
|
|
BacktestConfig,
|
|
|
Candidate,
|
|
|
DailyPortfolioState,
|
|
|
ExperimentManifest,
|
|
|
ExperimentResult,
|
|
|
FilledTrade,
|
|
|
MetricsBundle,
|
|
|
OpenPosition,
|
|
|
PositionStatus,
|
|
|
)
|
|
|
from libs.backtest.execution import (
|
|
|
simulate_entry,
|
|
|
simulate_exit,
|
|
|
simulate_kill_switch_exit,
|
|
|
update_trailing_stop,
|
|
|
)
|
|
|
from libs.backtest.manifests import generate_run_id, load_manifest, resolve_config
|
|
|
from libs.backtest.metrics import build_metrics_bundle
|
|
|
from libs.backtest.selector import select_candidates
|
|
|
from libs.backtest.snapshot_store import SnapshotStore
|
|
|
from libs.common.logging import get_logger
|
|
|
from libs.common.time_utils import utc_now
|
|
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
_KILL_SWITCH_DRAWDOWN_PCT = 25.0
|
|
|
|
|
|
|
|
|
def _get_git_commit_hash() -> str:
|
|
|
try:
|
|
|
result = subprocess.run(
|
|
|
["git", "rev-parse", "--short", "HEAD"],
|
|
|
capture_output=True, text=True, timeout=5,
|
|
|
)
|
|
|
return result.stdout.strip() or "unknown"
|
|
|
except Exception:
|
|
|
return "unknown"
|
|
|
|
|
|
|
|
|
class BacktestRunner:
|
|
|
"""Event-driven backtester simulation engine."""
|
|
|
|
|
|
def __init__(
|
|
|
self,
|
|
|
manifest: ExperimentManifest,
|
|
|
config: BacktestConfig,
|
|
|
store: SnapshotStore,
|
|
|
initial_equity: float = 100_000.0,
|
|
|
) -> None:
|
|
|
self.manifest = manifest
|
|
|
self.config = config
|
|
|
self.store = store
|
|
|
self.initial_equity = initial_equity
|
|
|
|
|
|
# Simulation state
|
|
|
self._equity = initial_equity
|
|
|
self._cash = initial_equity
|
|
|
self._open_positions: list[OpenPosition] = []
|
|
|
self._closed_trades: list[FilledTrade] = []
|
|
|
self._equity_curve: list[DailyPortfolioState] = []
|
|
|
self._candidate_map: dict[str, Candidate] = {} # trade_id → candidate
|
|
|
|
|
|
# Stats
|
|
|
self._total_candidates_seen = 0
|
|
|
self._total_orders_rejected = 0
|
|
|
self._peak_equity = initial_equity
|
|
|
self._realized_pnl = 0.0
|
|
|
self._daily_new_risk_used = 0.0
|
|
|
self._consecutive_losses = 0
|
|
|
self._cooldown_remaining = 0
|
|
|
self._kill_switch_triggered = False
|
|
|
|
|
|
def run(self, output_root: str | Path | None = None) -> ExperimentResult:
|
|
|
"""Execute the full simulation. Returns ExperimentResult."""
|
|
|
started_at = utc_now()
|
|
|
run_id = generate_run_id(self.config)
|
|
|
logger.info("backtest_start", run_id=run_id, strategy=self.config.strategy_name)
|
|
|
|
|
|
exec_dates = self.store.all_execution_dates()
|
|
|
if not exec_dates:
|
|
|
logger.warning("backtest_no_dates", run_id=run_id)
|
|
|
|
|
|
# Iterate ALL trading days (not just candidate days) so stop/target/time
|
|
|
# exits are checked every day, not just on days with new candidates.
|
|
|
all_dates = self.store.all_trading_days()
|
|
|
|
|
|
# Record initial equity state (before any trades)
|
|
|
if all_dates:
|
|
|
self._equity_curve.append(
|
|
|
DailyPortfolioState(
|
|
|
date=all_dates[0],
|
|
|
equity=self.initial_equity,
|
|
|
cash_available=self.initial_equity,
|
|
|
gross_exposure=0.0,
|
|
|
net_exposure=0.0,
|
|
|
reserved_risk_budget=0.0,
|
|
|
unrealized_pnl=0.0,
|
|
|
realized_pnl=0.0,
|
|
|
open_positions=[],
|
|
|
daily_new_risk_used=0.0,
|
|
|
peak_equity=self.initial_equity,
|
|
|
current_drawdown_pct=0.0,
|
|
|
)
|
|
|
)
|
|
|
|
|
|
for date in all_dates:
|
|
|
self._simulate_day(date)
|
|
|
|
|
|
# Force-close any remaining open positions at end of backtest
|
|
|
if self._open_positions:
|
|
|
last_date = all_dates[-1] if all_dates else dt.date.today()
|
|
|
self._force_close_all(last_date, reason="end_of_backtest")
|
|
|
|
|
|
finished_at = utc_now()
|
|
|
metrics = build_metrics_bundle(
|
|
|
self._closed_trades, self._equity_curve, self._candidate_map
|
|
|
)
|
|
|
|
|
|
# Create run directory and write artifacts
|
|
|
run_dir = None
|
|
|
artifact_paths: dict[str, str] = {}
|
|
|
if output_root is not None:
|
|
|
run_dir = create_run_directory(output_root, run_id)
|
|
|
git_hash = _get_git_commit_hash()
|
|
|
artifact_paths = write_all_artifacts(
|
|
|
run_dir=run_dir,
|
|
|
run_id=run_id,
|
|
|
manifest=self.manifest,
|
|
|
config=self.config,
|
|
|
metrics=metrics,
|
|
|
trades=self._closed_trades,
|
|
|
equity_curve=self._equity_curve,
|
|
|
open_positions=self._open_positions,
|
|
|
candidate_map=self._candidate_map,
|
|
|
started_at=started_at,
|
|
|
finished_at=finished_at,
|
|
|
git_hash=git_hash,
|
|
|
total_trading_days=len(self._equity_curve),
|
|
|
total_candidates_seen=self._total_candidates_seen,
|
|
|
total_orders_rejected=self._total_orders_rejected,
|
|
|
)
|
|
|
|
|
|
logger.info(
|
|
|
"backtest_complete",
|
|
|
run_id=run_id,
|
|
|
trades=len(self._closed_trades),
|
|
|
days=len(self._equity_curve),
|
|
|
)
|
|
|
|
|
|
return ExperimentResult(
|
|
|
run_id=run_id,
|
|
|
manifest=self.manifest,
|
|
|
resolved_config=self.config,
|
|
|
metrics=metrics,
|
|
|
artifact_paths=artifact_paths,
|
|
|
started_at=started_at,
|
|
|
finished_at=finished_at,
|
|
|
total_trading_days=len(self._equity_curve),
|
|
|
total_candidates_seen=self._total_candidates_seen,
|
|
|
total_orders_rejected=self._total_orders_rejected,
|
|
|
)
|
|
|
|
|
|
def _simulate_day(self, date: dt.date) -> None:
|
|
|
"""Simulate a single trading day."""
|
|
|
# Reset daily risk tracker
|
|
|
self._daily_new_risk_used = 0.0
|
|
|
|
|
|
# Decrement cooldown
|
|
|
if self._cooldown_remaining > 0:
|
|
|
self._cooldown_remaining -= 1
|
|
|
|
|
|
# Increment days_held for all open positions
|
|
|
for pos in self._open_positions:
|
|
|
pos.days_held += 1
|
|
|
|
|
|
# --- EXITS FIRST (using today's OHLCV) ---
|
|
|
newly_closed: list[FilledTrade] = []
|
|
|
still_open: list[OpenPosition] = []
|
|
|
|
|
|
for pos in self._open_positions:
|
|
|
bar = self.store.get_bar(pos.plan.candidate.symbol, date)
|
|
|
|
|
|
# Kill switch: force close
|
|
|
if self._kill_switch_triggered:
|
|
|
trade = simulate_kill_switch_exit(pos, bar, date, self.config.execution)
|
|
|
newly_closed.append(trade)
|
|
|
continue
|
|
|
|
|
|
if bar is None:
|
|
|
# Missing bar — hold position (do not impute zero)
|
|
|
still_open.append(pos)
|
|
|
continue
|
|
|
|
|
|
# Update trailing stop if configured
|
|
|
if self.config.execution.trailing_model:
|
|
|
update_trailing_stop(pos, bar)
|
|
|
|
|
|
trade = simulate_exit(pos, bar, self.config.execution, date)
|
|
|
if trade is not None:
|
|
|
newly_closed.append(trade)
|
|
|
else:
|
|
|
still_open.append(pos)
|
|
|
|
|
|
# Process closed trades
|
|
|
for trade in newly_closed:
|
|
|
self._closed_trades.append(trade)
|
|
|
self._realized_pnl += trade.net_pnl
|
|
|
self._cash += trade.net_pnl + (trade.entry_price * trade.shares)
|
|
|
# Track consecutive losses for cooldown
|
|
|
if trade.net_pnl < 0:
|
|
|
self._consecutive_losses += 1
|
|
|
else:
|
|
|
self._consecutive_losses = 0
|
|
|
if (
|
|
|
self.config.risk.cooldown_after_loss_streak > 0
|
|
|
and self._consecutive_losses >= self.config.risk.cooldown_after_loss_streak
|
|
|
):
|
|
|
self._cooldown_remaining = self.config.risk.cooldown_days
|
|
|
self._consecutive_losses = 0
|
|
|
|
|
|
self._open_positions = still_open
|
|
|
|
|
|
# --- Compute current equity for kill-switch check ---
|
|
|
market_value = self._compute_positions_market_value(date)
|
|
|
unrealized = market_value - sum(
|
|
|
p.entry_price * p.shares_open for p in self._open_positions
|
|
|
)
|
|
|
self._equity = self._cash + market_value
|
|
|
self._peak_equity = max(self._peak_equity, self._equity)
|
|
|
drawdown_pct = (
|
|
|
(self._peak_equity - self._equity) / self._peak_equity * 100.0
|
|
|
if self._peak_equity > 0
|
|
|
else 0.0
|
|
|
)
|
|
|
|
|
|
if drawdown_pct >= _KILL_SWITCH_DRAWDOWN_PCT and not self._kill_switch_triggered:
|
|
|
logger.warning("kill_switch_triggered", date=str(date), drawdown_pct=drawdown_pct)
|
|
|
self._kill_switch_triggered = True
|
|
|
|
|
|
# --- ENTRIES (only if kill switch not triggered) ---
|
|
|
if not self._kill_switch_triggered:
|
|
|
raw_rows = self.store.get_candidates_for_date(date)
|
|
|
self._total_candidates_seen += len(raw_rows)
|
|
|
portfolio_state = self._build_portfolio_state(date, drawdown_pct, unrealized)
|
|
|
candidates = select_candidates(
|
|
|
raw_rows, self.config.universe, self.config.signal
|
|
|
)
|
|
|
|
|
|
for candidate in candidates:
|
|
|
plan = build_planned_order(
|
|
|
candidate=candidate,
|
|
|
portfolio_state=portfolio_state,
|
|
|
open_positions=self._open_positions,
|
|
|
config=self.config,
|
|
|
cooldown_remaining=self._cooldown_remaining,
|
|
|
)
|
|
|
|
|
|
if plan.skip_reason is not None:
|
|
|
self._total_orders_rejected += 1
|
|
|
logger.debug(
|
|
|
"order_rejected",
|
|
|
symbol=candidate.symbol,
|
|
|
reason=plan.skip_reason,
|
|
|
date=str(date),
|
|
|
)
|
|
|
continue
|
|
|
|
|
|
bar = self.store.get_bar(candidate.symbol, date)
|
|
|
pos = simulate_entry(plan, bar, self.config.execution)
|
|
|
if pos is not None:
|
|
|
self._open_positions.append(pos)
|
|
|
self._cash -= pos.entry_price * pos.shares_total
|
|
|
self._daily_new_risk_used += plan.risk_dollars
|
|
|
# Update equity and portfolio state for next candidate
|
|
|
mv = self._compute_positions_market_value(date)
|
|
|
self._equity = self._cash + mv
|
|
|
ur = mv - sum(
|
|
|
p.entry_price * p.shares_open
|
|
|
for p in self._open_positions
|
|
|
)
|
|
|
portfolio_state = self._build_portfolio_state(
|
|
|
date, drawdown_pct, ur
|
|
|
)
|
|
|
|
|
|
# --- Record daily equity curve snapshot ---
|
|
|
market_value_final = self._compute_positions_market_value(date)
|
|
|
unrealized_final = market_value_final - sum(
|
|
|
p.entry_price * p.shares_open for p in self._open_positions
|
|
|
)
|
|
|
self._equity = self._cash + market_value_final
|
|
|
self._peak_equity = max(self._peak_equity, self._equity)
|
|
|
final_drawdown = (
|
|
|
(self._peak_equity - self._equity) / self._peak_equity * 100.0
|
|
|
if self._peak_equity > 0
|
|
|
else 0.0
|
|
|
)
|
|
|
self._equity_curve.append(
|
|
|
DailyPortfolioState(
|
|
|
date=date,
|
|
|
equity=self._equity,
|
|
|
cash_available=self._cash,
|
|
|
gross_exposure=sum(
|
|
|
p.entry_price * p.shares_open for p in self._open_positions
|
|
|
),
|
|
|
net_exposure=sum(
|
|
|
p.entry_price * p.shares_open for p in self._open_positions
|
|
|
),
|
|
|
reserved_risk_budget=self._daily_new_risk_used,
|
|
|
unrealized_pnl=unrealized_final,
|
|
|
realized_pnl=self._realized_pnl,
|
|
|
open_positions=[p.position_id for p in self._open_positions],
|
|
|
daily_new_risk_used=self._daily_new_risk_used,
|
|
|
peak_equity=self._peak_equity,
|
|
|
current_drawdown_pct=final_drawdown,
|
|
|
)
|
|
|
)
|
|
|
|
|
|
def _compute_positions_market_value(self, date: dt.date) -> float:
|
|
|
"""Market value of all open positions using today's close.
|
|
|
|
|
|
Falls back to entry_price when bar is missing (assumes no change
|
|
|
rather than treating the position as worthless).
|
|
|
"""
|
|
|
total = 0.0
|
|
|
for pos in self._open_positions:
|
|
|
bar = self.store.get_bar(pos.plan.candidate.symbol, date)
|
|
|
if bar and bar.get("close"):
|
|
|
total += float(bar["close"]) * pos.shares_open
|
|
|
else:
|
|
|
total += pos.entry_price * pos.shares_open
|
|
|
return total
|
|
|
|
|
|
def _compute_unrealized_pnl(self, date: dt.date) -> float:
|
|
|
"""Unrealized PnL = market_value − cost_basis."""
|
|
|
market_value = self._compute_positions_market_value(date)
|
|
|
cost_basis = sum(p.entry_price * p.shares_open for p in self._open_positions)
|
|
|
return market_value - cost_basis
|
|
|
|
|
|
def _build_portfolio_state(
|
|
|
self,
|
|
|
date: dt.date,
|
|
|
drawdown_pct: float,
|
|
|
unrealized: float,
|
|
|
) -> DailyPortfolioState:
|
|
|
return DailyPortfolioState(
|
|
|
date=date,
|
|
|
equity=self._equity,
|
|
|
cash_available=self._cash,
|
|
|
gross_exposure=sum(
|
|
|
p.entry_price * p.shares_open for p in self._open_positions
|
|
|
),
|
|
|
net_exposure=sum(
|
|
|
p.entry_price * p.shares_open for p in self._open_positions
|
|
|
),
|
|
|
reserved_risk_budget=self._daily_new_risk_used,
|
|
|
unrealized_pnl=unrealized,
|
|
|
realized_pnl=self._realized_pnl,
|
|
|
open_positions=[p.position_id for p in self._open_positions],
|
|
|
daily_new_risk_used=self._daily_new_risk_used,
|
|
|
peak_equity=self._peak_equity,
|
|
|
current_drawdown_pct=drawdown_pct,
|
|
|
)
|
|
|
|
|
|
def _force_close_all(self, date: dt.date, reason: str = "force_close") -> None:
|
|
|
"""Close all open positions (end of backtest or kill switch)."""
|
|
|
for pos in list(self._open_positions):
|
|
|
bar = self.store.get_bar(pos.plan.candidate.symbol, date)
|
|
|
trade = simulate_kill_switch_exit(pos, bar, date, self.config.execution)
|
|
|
self._closed_trades.append(trade)
|
|
|
self._realized_pnl += trade.net_pnl
|
|
|
self._cash += trade.net_pnl + (trade.entry_price * trade.shares)
|
|
|
self._open_positions = []
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
# CLI
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
def _build_store(
|
|
|
manifest: ExperimentManifest,
|
|
|
config: BacktestConfig,
|
|
|
split_name: str,
|
|
|
snapshot_dir_override: str | None = None,
|
|
|
) -> SnapshotStore:
|
|
|
from libs.common.config import get_settings
|
|
|
|
|
|
s = get_settings()
|
|
|
snapshot_dir = Path(snapshot_dir_override or s.parquet_dir) / config.dataset_snapshot_id
|
|
|
return SnapshotStore.load(
|
|
|
snapshot_dir=snapshot_dir,
|
|
|
split_name=split_name,
|
|
|
oracle_url=s.stock_oracle_url,
|
|
|
db_dsn=s.postgres_dsn,
|
|
|
)
|
|
|
|
|
|
|
|
|
def main() -> None:
|
|
|
parser = argparse.ArgumentParser(description="ACE-F Backtester")
|
|
|
parser.add_argument("--manifest", required=True, help="Path to experiment manifest JSON")
|
|
|
parser.add_argument("--snapshot-id", help="Override dataset_snapshot_id")
|
|
|
parser.add_argument("--snapshot-dir", help="Override snapshot root directory (default: data/parquet/)")
|
|
|
parser.add_argument("--split", default="train", help="Split name (train/valid/test)")
|
|
|
parser.add_argument("--output-root", default="./runs", help="Output root directory")
|
|
|
parser.add_argument("--initial-equity", type=float, default=100_000.0)
|
|
|
parser.add_argument("--config-root", default=".", help="Root dir for resolving config paths")
|
|
|
args = parser.parse_args()
|
|
|
|
|
|
manifest = load_manifest(args.manifest)
|
|
|
config = resolve_config(manifest, config_root=args.config_root, snapshot_id_override=args.snapshot_id)
|
|
|
store = _build_store(manifest, config, args.split, snapshot_dir_override=args.snapshot_dir)
|
|
|
|
|
|
runner = BacktestRunner(
|
|
|
manifest=manifest,
|
|
|
config=config,
|
|
|
store=store,
|
|
|
initial_equity=args.initial_equity,
|
|
|
)
|
|
|
result = runner.run(output_root=args.output_root)
|
|
|
print(f"Run complete: {result.run_id}")
|
|
|
print(f"Trades: {result.metrics.trade_count}")
|
|
|
if result.metrics.total_return_pct is not None:
|
|
|
print(f"Total return: {result.metrics.total_return_pct:.2f}%")
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
main()
|