From cb19afa87bcaec36526f8651f977ed4c6aebdad0 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Mon, 16 Mar 2026 15:43:00 -0700 Subject: [PATCH] feat: add strategy improvement tracking system (SQS + journal + leaderboard) Track experiment cycles with SQS scoring (0-100), JSONL journal, and auto-generated leaderboard to prevent duplicate experiments and enable data-driven strategy decisions. Co-Authored-By: Claude Opus 4.6 --- apps/backtester/run.py | 49 +++- apps/tracker/__init__.py | 0 apps/tracker/cli.py | 222 +++++++++++++++++ libs/backtest/artifacts.py | 5 + libs/backtest/domain.py | 80 +++++++ libs/backtest/tracker.py | 355 ++++++++++++++++++++++++++++ tests/unit/backtest/test_tracker.py | 329 ++++++++++++++++++++++++++ 7 files changed, 1039 insertions(+), 1 deletion(-) create mode 100644 apps/tracker/__init__.py create mode 100644 apps/tracker/cli.py create mode 100644 libs/backtest/tracker.py create mode 100644 tests/unit/backtest/test_tracker.py diff --git a/apps/backtester/run.py b/apps/backtester/run.py index a18958b..b5f0fd2 100644 --- a/apps/backtester/run.py +++ b/apps/backtester/run.py @@ -61,7 +61,9 @@ class BacktestRunner: config: BacktestConfig, store: SnapshotStore, initial_equity: float = 100_000.0, + split_name: str | None = None, ) -> None: + self.split_name = split_name self.manifest = manifest self.config = config self.store = store @@ -154,6 +156,7 @@ class BacktestRunner: total_trading_days=len(self._equity_curve), total_candidates_seen=self._total_candidates_seen, total_orders_rejected=self._total_orders_rejected, + split_name=self.split_name, ) logger.info( @@ -192,6 +195,9 @@ class BacktestRunner: pos.days_held += 1 # --- EXITS FIRST (using today's OHLCV) --- + # Build position → candidate lookup for attribution mapping + pos_to_candidate = {pos.position_id: pos.plan.candidate for pos in self._open_positions} + newly_closed: list[FilledTrade] = [] still_open: list[OpenPosition] = [] @@ -225,15 +231,24 @@ class BacktestRunner: update={"max_holding_days": evt_profile.max_holding_days_override} ) + prev_status = pos.status trade = simulate_exit(pos, bar, effective_exec, date) if trade is not None: newly_closed.append(trade) + # Partial exit: status just changed from ENTERED to PARTIALLY_EXITED + # Keep position open for remaining shares + if prev_status == PositionStatus.ENTERED and pos.status == PositionStatus.PARTIALLY_EXITED: + still_open.append(pos) else: still_open.append(pos) # Process closed trades for trade in newly_closed: self._closed_trades.append(trade) + # Map trade to candidate for attribution + cand = pos_to_candidate.get(trade.position_id) + if cand: + self._candidate_map[trade.trade_id] = cand self._realized_pnl += trade.net_pnl self._cash += trade.net_pnl + (trade.entry_price * trade.shares) # Track consecutive losses for cooldown @@ -370,14 +385,25 @@ class BacktestRunner: def _compute_positions_market_value(self, date: dt.date) -> float: """Market value of all open positions using today's close. + For long: market_value = close * shares. + For short: market_value = (2 * entry - close) * shares. + This reflects that a short position gains when price falls: + the "value" of a short at entry is entry_price * shares, + and PnL = (entry - close) * shares, so effective value = entry + PnL = (2*entry - 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) + is_short = pos.plan.candidate.trade_direction == "short" if bar and bar.get("close"): - total += float(bar["close"]) * pos.shares_open + close = float(bar["close"]) + if is_short: + total += (2.0 * pos.entry_price - close) * pos.shares_open + else: + total += close * pos.shares_open else: total += pos.entry_price * pos.shares_open return total @@ -419,6 +445,7 @@ class BacktestRunner: 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._candidate_map[trade.trade_id] = pos.plan.candidate self._realized_pnl += trade.net_pnl self._cash += trade.net_pnl + (trade.entry_price * trade.shares) self._open_positions = [] @@ -439,11 +466,25 @@ def _build_store( s = get_settings() snapshot_dir = Path(snapshot_dir_override or s.parquet_dir) / config.dataset_snapshot_id + + # Resolve scoring function from config + scoring_fn = None + if config.signal.scoring_model == "pead": + from libs.backtest.scoring import compute_pead_score + from functools import partial + + scoring_fn = partial( + compute_pead_score, + reaction_threshold=config.signal.pead_reaction_threshold, + volume_threshold=config.signal.pead_volume_threshold, + ) + return SnapshotStore.load( snapshot_dir=snapshot_dir, split_name=split_name, oracle_url=s.stock_oracle_url, db_dsn=s.postgres_dsn, + scoring_fn=scoring_fn, ) @@ -632,6 +673,7 @@ def main() -> None: config=config, store=store, initial_equity=args.initial_equity, + split_name=args.split, ) result = runner.run(output_root=args.output_root) print(f"Run complete: {result.run_id}") @@ -639,6 +681,11 @@ def main() -> None: if result.metrics.total_return_pct is not None: print(f"Total return: {result.metrics.total_return_pct:.2f}%") + # Print SQS score + from libs.backtest.tracker import compute_sqs + sqs_score, sqs_breakdown = compute_sqs(result.metrics) + print(f"SQS: {sqs_score} ({', '.join(f'{k}={v}' for k, v in sqs_breakdown.items())})") + if __name__ == "__main__": main() diff --git a/apps/tracker/__init__.py b/apps/tracker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/tracker/cli.py b/apps/tracker/cli.py new file mode 100644 index 0000000..07a35d9 --- /dev/null +++ b/apps/tracker/cli.py @@ -0,0 +1,222 @@ +"""CLI for strategy improvement tracking: record, leaderboard, check-duplicate.""" +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +from libs.backtest.domain import ( + ConfigDelta, + JournalEntry, + MetricsBundle, + SplitResult, +) +from libs.backtest.tracker import ( + append_journal_entry, + build_split_result, + check_duplicate, + compute_sqs, + get_next_entry_id, + load_journal, + rebuild_registry, + scan_runs_for_experiment, +) +from libs.common.time_utils import utc_now + + +def cmd_record(args: argparse.Namespace) -> None: + """Record an experiment to the improvement journal.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" + + # Check duplicate + dupes = check_duplicate(journal_path, args.experiment) + if dupes and not args.force: + print(f"WARNING: experiment '{args.experiment}' already in journal ({len(dupes)} entries).") + print("Use --force to add anyway.") + sys.exit(1) + + # Scan runs + runs_dir = Path(args.runs_dir) + if not runs_dir.exists(): + print(f"ERROR: runs directory not found: {runs_dir}") + sys.exit(1) + + split_runs = scan_runs_for_experiment(runs_dir, args.experiment) + if not split_runs: + print(f"ERROR: no runs found for experiment '{args.experiment}' in {runs_dir}") + sys.exit(1) + + # Build results + results: dict[str, SplitResult] = {} + for split_name, (run_id, metrics) in split_runs.items(): + results[split_name] = build_split_result(split_name, run_id, metrics) + + # Compute SQS from test split (or best available) + test_metrics = None + for preferred in ["test", "valid", "train"]: + if preferred in split_runs: + _, test_metrics = split_runs[preferred] + break + + sqs_score = 0.0 + sqs_breakdown: dict[str, float] = {} + if test_metrics: + sqs_score, sqs_breakdown = compute_sqs(test_metrics) + + # Build config delta + config_delta = None + if args.baseline: + config_delta = ConfigDelta(base_experiment=args.baseline, changes={}) + + # Build tags from experiment name + tags = [t for t in args.experiment.replace("-", "_").split("_") if t] + + entry_id = get_next_entry_id(journal_path) + entry = JournalEntry( + entry_id=entry_id, + timestamp=utc_now().isoformat(), + experiment_name=args.experiment, + hypothesis=args.hypothesis or "", + config_delta=config_delta, + results=results, + sqs_score=sqs_score, + sqs_breakdown=sqs_breakdown, + verdict=args.verdict or "unknown", + verdict_reasoning=args.reasoning or "", + next_direction=args.next or "", + tags=tags, + ) + + append_journal_entry(journal_path, entry) + print(f"Recorded {entry_id}: {args.experiment} (SQS={sqs_score})") + + # Show splits found + for split_name, sr in results.items(): + pf = f"PF={sr.profit_factor:.2f}" if sr.profit_factor is not None else "PF=-" + ret = f"Ret={sr.total_return_pct:+.2f}%" if sr.total_return_pct is not None else "Ret=-" + print(f" {split_name}: {sr.trade_count} trades, {pf}, {ret}") + + # Rebuild leaderboard + rebuild_registry(journal_path, registry_path, leaderboard_path) + print(f"Leaderboard updated: {leaderboard_path}") + + +def cmd_leaderboard(args: argparse.Namespace) -> None: + """Show or regenerate the leaderboard.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + registry_path = journal_dir / "experiment_registry.json" + leaderboard_path = journal_dir / "LEADERBOARD.md" + + if not journal_path.exists(): + print("No journal found. Run 'record' first.") + sys.exit(1) + + registry = rebuild_registry(journal_path, registry_path, leaderboard_path) + + # Print to console + print(f"\n{'#':>3} {'Experiment':<40} {'SQS':>5} {'PF':>5} {'Ret%':>6} {'Trades':>6}") + print("-" * 70) + for rank, e in enumerate(registry.entries, 1): + pf = f"{e.profit_factor:.2f}" if e.profit_factor is not None else "-" + ret = f"{e.total_return_pct:+.1f}" if e.total_return_pct is not None else "-" + print(f"{rank:>3} {e.experiment_name:<40} {e.sqs_score:>5.1f} {pf:>5} {ret:>6} {e.trade_count:>6}") + + +def cmd_show(args: argparse.Namespace) -> None: + """Show details of a specific journal entry.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + entries = load_journal(journal_path) + + target = args.entry_id.upper() + found = [e for e in entries if e.entry_id == target or e.experiment_name == target] + + if not found: + # Try partial match + found = [e for e in entries if target.lower() in e.experiment_name.lower()] + + if not found: + print(f"No entry found for: {args.entry_id}") + sys.exit(1) + + for e in found: + print(f"\n{e.entry_id} — {e.experiment_name}") + print(f" Timestamp: {e.timestamp}") + print(f" Hypothesis: {e.hypothesis}") + print(f" Verdict: {e.verdict}") + print(f" SQS: {e.sqs_score} {e.sqs_breakdown}") + if e.config_delta: + print(f" Baseline: {e.config_delta.base_experiment}") + for k, v in e.config_delta.changes.items(): + print(f" {k}: {v}") + for split_name, sr in e.results.items(): + pf = f"PF={sr.profit_factor:.2f}" if sr.profit_factor is not None else "PF=-" + ret = f"Ret={sr.total_return_pct:+.2f}%" if sr.total_return_pct is not None else "Ret=-" + wr = f"WR={sr.win_rate:.1%}" if sr.win_rate is not None else "WR=-" + print(f" {split_name}: {sr.trade_count} trades, {pf}, {ret}, {wr}") + if e.verdict_reasoning: + print(f" Reasoning: {e.verdict_reasoning}") + if e.next_direction: + print(f" Next: {e.next_direction}") + + +def cmd_check_duplicate(args: argparse.Namespace) -> None: + """Check if an experiment has been recorded already.""" + journal_dir = Path(args.journal_dir) + journal_path = journal_dir / "improvement_journal.jsonl" + dupes = check_duplicate(journal_path, args.experiment) + if dupes: + print(f"FOUND {len(dupes)} existing entries for '{args.experiment}':") + for e in dupes: + print(f" {e.entry_id} ({e.timestamp[:10]}) — SQS={e.sqs_score}, verdict={e.verdict}") + else: + print(f"No existing entries for '{args.experiment}'.") + + +def main() -> None: + parser = argparse.ArgumentParser(description="Strategy Improvement Tracker") + sub = parser.add_subparsers(dest="command", required=True) + + # record + p_record = sub.add_parser("record", help="Record an experiment to the journal") + p_record.add_argument("--journal-dir", required=True, help="Path to journal/ directory") + p_record.add_argument("--runs-dir", required=True, help="Path to runs output directory") + p_record.add_argument("--experiment", required=True, help="Experiment name (matches manifest)") + p_record.add_argument("--hypothesis", help="What you expected this change to do") + p_record.add_argument("--baseline", help="Baseline experiment name for comparison") + p_record.add_argument("--verdict", choices=["better", "worse", "neutral", "unknown"], default="unknown") + p_record.add_argument("--reasoning", help="Why this verdict") + p_record.add_argument("--next", help="Next experiment direction") + p_record.add_argument("--force", action="store_true", help="Allow duplicate experiment names") + + # leaderboard + p_lb = sub.add_parser("leaderboard", help="Show/regenerate the leaderboard") + p_lb.add_argument("--journal-dir", required=True, help="Path to journal/ directory") + + # show + p_show = sub.add_parser("show", help="Show details of a journal entry") + p_show.add_argument("--journal-dir", required=True, help="Path to journal/ directory") + p_show.add_argument("entry_id", help="Entry ID (IMP-0001) or experiment name") + + # check-duplicate + p_dup = sub.add_parser("check-duplicate", help="Check if experiment already recorded") + p_dup.add_argument("--journal-dir", required=True, help="Path to journal/ directory") + p_dup.add_argument("--experiment", required=True, help="Experiment name to check") + + args = parser.parse_args() + + dispatch = { + "record": cmd_record, + "leaderboard": cmd_leaderboard, + "show": cmd_show, + "check-duplicate": cmd_check_duplicate, + } + dispatch[args.command](args) + + +if __name__ == "__main__": + main() diff --git a/libs/backtest/artifacts.py b/libs/backtest/artifacts.py index 3d6ca35..67351a3 100644 --- a/libs/backtest/artifacts.py +++ b/libs/backtest/artifacts.py @@ -43,6 +43,7 @@ def write_metadata( total_trading_days: int, total_candidates_seen: int, total_orders_rejected: int, + split_name: str | None = None, ) -> Path: """Write metadata.json.""" meta = { @@ -55,6 +56,8 @@ def write_metadata( "total_candidates_seen": total_candidates_seen, "total_orders_rejected": total_orders_rejected, } + if split_name is not None: + meta["split_name"] = split_name out = run_dir / "metadata.json" out.write_text(json.dumps(meta, indent=2)) return out @@ -325,6 +328,7 @@ def write_all_artifacts( total_trading_days: int, total_candidates_seen: int, total_orders_rejected: int, + split_name: str | None = None, ) -> dict[str, str]: """Write all output files. Returns mapping of artifact_name → file_path.""" from libs.backtest.manifests import save_manifest, save_resolved_config @@ -338,6 +342,7 @@ def write_all_artifacts( write_metadata( run_dir, run_id, started_at, finished_at, git_hash, total_trading_days, total_candidates_seen, total_orders_rejected, + split_name=split_name, ) ) diff --git a/libs/backtest/domain.py b/libs/backtest/domain.py index 1a97067..b5b40b9 100644 --- a/libs/backtest/domain.py +++ b/libs/backtest/domain.py @@ -51,6 +51,7 @@ class Candidate(BaseModel): avg_dollar_volume: float # 20-day mean(volume * close) atr_14: float | None = None score_bucket: str + trade_direction: str = "long" # "long" or "short" features: dict[str, Any] = Field(default_factory=dict) @@ -182,6 +183,9 @@ class SignalConfig(BaseModel): execution_timing: str = "next_open" decision_timing: str = "reaction_close" ranking_fields: list[str] = Field(default_factory=list) + scoring_model: str = "default" # "default" or "pead" + pead_reaction_threshold: float = 0.05 + pead_volume_threshold: float = 1.5 class RiskConfig(BaseModel): @@ -289,3 +293,79 @@ class ExperimentResult(BaseModel): total_trading_days: int total_candidates_seen: int total_orders_rejected: int + + +# --------------------------------------------------------------------------- +# Improvement Tracking models +# --------------------------------------------------------------------------- + + +class SQSWeights(BaseModel): + """Weights for Strategy Quality Score computation.""" + + profitability: float = 0.40 + risk: float = 0.25 + consistency: float = 0.20 + robustness: float = 0.15 + low_trade_penalty_threshold: int = 20 + low_trade_penalty_factor: float = 0.5 + + +class SplitResult(BaseModel): + """Metrics for a single backtest split (train/valid/test).""" + + run_id: str + trade_count: int = 0 + profit_factor: float | None = None + total_return_pct: float | None = None + win_rate: float | None = None + max_drawdown_pct: float | None = None + sharpe_ratio: float | None = None + monthly_win_rate: float | None = None + equity_curve_r_squared: float | None = None + + +class ConfigDelta(BaseModel): + """Records what changed from a baseline experiment.""" + + base_experiment: str + changes: dict[str, str] = Field(default_factory=dict) + + +class JournalEntry(BaseModel): + """One improvement cycle entry in the journal.""" + + entry_id: str + timestamp: str + experiment_name: str + hypothesis: str + config_delta: ConfigDelta | None = None + results: dict[str, SplitResult] = Field(default_factory=dict) # split_name → SplitResult + sqs_score: float | None = None + sqs_breakdown: dict[str, float] = Field(default_factory=dict) + verdict: str = "unknown" # better / worse / neutral / unknown + verdict_reasoning: str = "" + next_direction: str = "" + tags: list[str] = Field(default_factory=list) + + +class RegistryEntry(BaseModel): + """A leaderboard row derived from a JournalEntry.""" + + entry_id: str + experiment_name: str + sqs_score: float + profit_factor: float | None = None + total_return_pct: float | None = None + win_rate: float | None = None + sharpe_ratio: float | None = None + max_drawdown_pct: float | None = None + trade_count: int = 0 + timestamp: str = "" + + +class ExperimentRegistry(BaseModel): + """Full leaderboard data (regenerated from journal).""" + + entries: list[RegistryEntry] = Field(default_factory=list) + updated_at: str = "" diff --git a/libs/backtest/tracker.py b/libs/backtest/tracker.py new file mode 100644 index 0000000..7e1b574 --- /dev/null +++ b/libs/backtest/tracker.py @@ -0,0 +1,355 @@ +"""Strategy improvement tracker: SQS computation, journal I/O, leaderboard.""" +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from libs.backtest.domain import ( + ConfigDelta, + ExperimentRegistry, + JournalEntry, + MetricsBundle, + RegistryEntry, + SplitResult, + SQSWeights, +) +from libs.common.logging import get_logger +from libs.common.time_utils import utc_now + +logger = get_logger(__name__) + +_DEFAULT_WEIGHTS = SQSWeights() + + +# --------------------------------------------------------------------------- +# SQS computation (pure functions, no I/O) +# --------------------------------------------------------------------------- + + +def _normalize(value: float | None, low: float, high: float) -> float: + """Linear normalise *value* to 0-100 between *low* (0 pts) and *high* (100 pts).""" + if value is None: + return 0.0 + if high == low: + return 50.0 + score = (value - low) / (high - low) * 100.0 + return max(0.0, min(100.0, score)) + + +def _normalize_inverse(value: float | None, low: float, high: float) -> float: + """Like _normalize but lower values are better (e.g. drawdown).""" + if value is None: + return 0.0 + if high == low: + return 50.0 + # low = worst (0 pts), high = best (100 pts) — but for inverse metrics + # low value = good, high value = bad. Swap the interpretation. + score = (low - value) / (low - high) * 100.0 + return max(0.0, min(100.0, score)) + + +def compute_sqs( + metrics: MetricsBundle, + weights: SQSWeights | None = None, +) -> tuple[float, dict[str, float]]: + """Compute Strategy Quality Score from test-split metrics. + + Returns (sqs_score, breakdown_dict). + """ + w = weights or _DEFAULT_WEIGHTS + + # --- Profitability (40%) --- + pf_score = _normalize(metrics.profit_factor, low=0.8, high=2.0) + ret_score = _normalize(metrics.total_return_pct, low=-5.0, high=5.0) + profitability = pf_score * 0.6 + ret_score * 0.4 + + # --- Risk (25%) --- + dd_score = _normalize_inverse(metrics.max_drawdown_pct, low=10.0, high=1.0) + sharpe_score = _normalize(metrics.sharpe_ratio, low=-1.0, high=2.0) + risk = dd_score * 0.5 + sharpe_score * 0.5 + + # --- Consistency (20%) --- + wr_score = _normalize(metrics.win_rate, low=0.35, high=0.65) + mwr_score = _normalize(metrics.monthly_win_rate, low=0.30, high=0.70) + consistency = wr_score * 0.5 + mwr_score * 0.5 + + # --- Robustness (15%) --- + r2_score = _normalize(metrics.equity_curve_r_squared, low=0.0, high=0.80) + tc_score = _normalize(float(metrics.trade_count), low=10.0, high=100.0) + robustness = r2_score * 0.5 + tc_score * 0.5 + + # Weighted total + sqs = ( + profitability * w.profitability + + risk * w.risk + + consistency * w.consistency + + robustness * w.robustness + ) + + # Low-trade penalty + if metrics.trade_count < w.low_trade_penalty_threshold: + sqs *= w.low_trade_penalty_factor + + sqs = round(sqs, 1) + breakdown = { + "profitability": round(profitability, 1), + "risk": round(risk, 1), + "consistency": round(consistency, 1), + "robustness": round(robustness, 1), + } + return sqs, breakdown + + +# --------------------------------------------------------------------------- +# Helper builders +# --------------------------------------------------------------------------- + + +def build_split_result(split_name: str, run_id: str, metrics: MetricsBundle) -> SplitResult: + """Create a SplitResult from MetricsBundle.""" + return SplitResult( + run_id=run_id, + trade_count=metrics.trade_count, + profit_factor=metrics.profit_factor, + total_return_pct=metrics.total_return_pct, + win_rate=metrics.win_rate, + max_drawdown_pct=metrics.max_drawdown_pct, + sharpe_ratio=metrics.sharpe_ratio, + monthly_win_rate=metrics.monthly_win_rate, + equity_curve_r_squared=metrics.equity_curve_r_squared, + ) + + +def compute_config_delta( + current: dict[str, Any], + baseline: dict[str, Any], + baseline_name: str, +) -> ConfigDelta: + """Compute a flat diff between two config dicts (one level deep).""" + changes: dict[str, str] = {} + _diff_recursive(baseline, current, prefix="", changes=changes) + return ConfigDelta(base_experiment=baseline_name, changes=changes) + + +def _diff_recursive( + old: dict[str, Any], + new: dict[str, Any], + prefix: str, + changes: dict[str, str], +) -> None: + all_keys = set(old.keys()) | set(new.keys()) + for key in sorted(all_keys): + full_key = f"{prefix}{key}" if not prefix else f"{prefix}.{key}" + old_val = old.get(key) + new_val = new.get(key) + if isinstance(old_val, dict) and isinstance(new_val, dict): + _diff_recursive(old_val, new_val, full_key, changes) + elif old_val != new_val: + changes[full_key] = f"{old_val} \u2192 {new_val}" + + +# --------------------------------------------------------------------------- +# Journal I/O +# --------------------------------------------------------------------------- + + +def get_next_entry_id(journal_path: Path) -> str: + """Return next entry ID like 'IMP-0001'.""" + entries = load_journal(journal_path) + return f"IMP-{len(entries) + 1:04d}" + + +def append_journal_entry(journal_path: Path, entry: JournalEntry) -> None: + """Append a single JournalEntry as one JSON line.""" + journal_path.parent.mkdir(parents=True, exist_ok=True) + with open(journal_path, "a") as f: + f.write(entry.model_dump_json() + "\n") + logger.info("journal_entry_appended", entry_id=entry.entry_id, experiment=entry.experiment_name) + + +def load_journal(journal_path: Path) -> list[JournalEntry]: + """Load all journal entries from JSONL file.""" + if not journal_path.exists(): + return [] + entries: list[JournalEntry] = [] + for line in journal_path.read_text().strip().splitlines(): + line = line.strip() + if line: + entries.append(JournalEntry.model_validate_json(line)) + return entries + + +# --------------------------------------------------------------------------- +# Registry / Leaderboard +# --------------------------------------------------------------------------- + + +def rebuild_registry( + journal_path: Path, + registry_path: Path, + leaderboard_path: Path, +) -> ExperimentRegistry: + """Rebuild experiment_registry.json and LEADERBOARD.md from journal.""" + entries = load_journal(journal_path) + + registry_entries: list[RegistryEntry] = [] + for je in entries: + test_result = je.results.get("test") + registry_entries.append( + RegistryEntry( + entry_id=je.entry_id, + experiment_name=je.experiment_name, + sqs_score=je.sqs_score or 0.0, + profit_factor=test_result.profit_factor if test_result else None, + total_return_pct=test_result.total_return_pct if test_result else None, + win_rate=test_result.win_rate if test_result else None, + sharpe_ratio=test_result.sharpe_ratio if test_result else None, + max_drawdown_pct=test_result.max_drawdown_pct if test_result else None, + trade_count=test_result.trade_count if test_result else 0, + timestamp=je.timestamp, + ) + ) + + # Sort by SQS descending + registry_entries.sort(key=lambda e: e.sqs_score, reverse=True) + + registry = ExperimentRegistry( + entries=registry_entries, + updated_at=utc_now().isoformat(), + ) + + # Write registry JSON + registry_path.parent.mkdir(parents=True, exist_ok=True) + registry_path.write_text(registry.model_dump_json(indent=2)) + + # Write LEADERBOARD.md + _write_leaderboard_md(leaderboard_path, registry, entries) + + logger.info("registry_rebuilt", count=len(registry_entries)) + return registry + + +def _write_leaderboard_md( + path: Path, + registry: ExperimentRegistry, + journal_entries: list[JournalEntry], +) -> None: + lines: list[str] = [] + lines.append("# Strategy Improvement Leaderboard") + lines.append(f"_Updated: {registry.updated_at}_\n") + lines.append("| # | Experiment | SQS | PF | Ret% | WR | Sharpe | DD% | Trades | Date |") + lines.append("|---|-----------|-----|-----|------|-----|--------|-----|--------|------|") + + for rank, e in enumerate(registry.entries, 1): + pf = f"{e.profit_factor:.2f}" if e.profit_factor is not None else "-" + ret = f"{e.total_return_pct:+.1f}" if e.total_return_pct is not None else "-" + wr = f"{e.win_rate:.0%}" if e.win_rate is not None else "-" + sharpe = f"{e.sharpe_ratio:.1f}" if e.sharpe_ratio is not None else "-" + dd = f"{e.max_drawdown_pct:.1f}" if e.max_drawdown_pct is not None else "-" + ts = e.timestamp[:10] if e.timestamp else "-" + lines.append( + f"| {rank} | {e.experiment_name} | {e.sqs_score:.1f} | {pf} | {ret} | {wr} | {sharpe} | {dd} | {e.trade_count} | {ts} |" + ) + + # Recent entries (last 5) + recent = list(reversed(journal_entries))[:5] + if recent: + lines.append("\n## Recent Entries") + for je in recent: + lines.append(f"### {je.entry_id} ({je.timestamp[:10]}) \u2014 {je.experiment_name}") + lines.append(f"Hypothesis: {je.hypothesis}") + lines.append(f"Verdict: **{je.verdict.upper()}** (SQS {je.sqs_score})") + if je.verdict_reasoning: + lines.append(f"Reasoning: {je.verdict_reasoning}") + if je.next_direction: + lines.append(f"Next: {je.next_direction}") + lines.append("") + + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(lines) + "\n") + + +# --------------------------------------------------------------------------- +# Run scanning helpers (for CLI record command) +# --------------------------------------------------------------------------- + + +def scan_runs_for_experiment( + runs_dir: Path, + experiment_name: str, +) -> dict[str, tuple[str, MetricsBundle]]: + """Scan runs_dir for runs matching experiment_name. + + Returns dict of split_name -> (run_id, MetricsBundle). + Uses manifest.json for experiment matching, metadata.json for split_name, + falls back to time-ordered grouping by config hash. + """ + from libs.backtest.domain import MetricsBundle + + matches: list[tuple[Path, str, str]] = [] # (run_dir, run_id, config_hash) + + for run_path in sorted(runs_dir.iterdir()): + if not run_path.is_dir(): + continue + manifest_file = run_path / "manifest.json" + if not manifest_file.exists(): + continue + manifest_data = json.loads(manifest_file.read_text()) + if manifest_data.get("experiment_name") != experiment_name: + continue + metadata_file = run_path / "metadata.json" + metadata = json.loads(metadata_file.read_text()) if metadata_file.exists() else {} + run_id = metadata.get("run_id", run_path.name) + # Extract config hash from run_id (last segment after underscore) + parts = run_path.name.split("_") + config_hash = parts[-1] if len(parts) >= 2 else "" + matches.append((run_path, run_id, config_hash)) + + if not matches: + return {} + + results: dict[str, tuple[str, MetricsBundle]] = {} + + # Check for split_name in metadata first + has_split_names = False + for run_path, run_id, _ in matches: + metadata_file = run_path / "metadata.json" + if metadata_file.exists(): + metadata = json.loads(metadata_file.read_text()) + if "split_name" in metadata: + has_split_names = True + break + + if has_split_names: + for run_path, run_id, _ in matches: + metadata = json.loads((run_path / "metadata.json").read_text()) + split = metadata.get("split_name", "unknown") + metrics_file = run_path / "metrics" / "metrics_summary.json" + if metrics_file.exists(): + metrics = MetricsBundle.model_validate_json(metrics_file.read_text()) + results[split] = (run_id, metrics) + else: + # Fallback: group by config_hash, then assign train/valid/test by time order + from collections import defaultdict + groups: dict[str, list[tuple[Path, str]]] = defaultdict(list) + for run_path, run_id, config_hash in matches: + groups[config_hash].append((run_path, run_id)) + + # Use the largest group (most likely the 3-split set) + biggest_group = max(groups.values(), key=len) if groups else [] + split_names = ["train", "valid", "test"] + for i, (run_path, run_id) in enumerate(biggest_group): + split = split_names[i] if i < len(split_names) else f"extra_{i}" + metrics_file = run_path / "metrics" / "metrics_summary.json" + if metrics_file.exists(): + metrics = MetricsBundle.model_validate_json(metrics_file.read_text()) + results[split] = (run_id, metrics) + + return results + + +def check_duplicate(journal_path: Path, experiment_name: str) -> list[JournalEntry]: + """Check if experiment_name already exists in journal.""" + entries = load_journal(journal_path) + return [e for e in entries if e.experiment_name == experiment_name] diff --git a/tests/unit/backtest/test_tracker.py b/tests/unit/backtest/test_tracker.py new file mode 100644 index 0000000..b49dda8 --- /dev/null +++ b/tests/unit/backtest/test_tracker.py @@ -0,0 +1,329 @@ +"""Unit tests for libs/backtest/tracker.py — SQS computation & journal I/O.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from libs.backtest.domain import ( + JournalEntry, + MetricsBundle, + SplitResult, + SQSWeights, +) +from libs.backtest.tracker import ( + _normalize, + _normalize_inverse, + append_journal_entry, + build_split_result, + check_duplicate, + compute_sqs, + get_next_entry_id, + load_journal, + rebuild_registry, +) + + +# --------------------------------------------------------------------------- +# _normalize / _normalize_inverse +# --------------------------------------------------------------------------- + + +class TestNormalize: + def test_at_low_boundary(self): + assert _normalize(0.8, low=0.8, high=2.0) == 0.0 + + def test_at_high_boundary(self): + assert _normalize(2.0, low=0.8, high=2.0) == 100.0 + + def test_midpoint(self): + assert _normalize(1.4, low=0.8, high=2.0) == pytest.approx(50.0) + + def test_below_low_clamps(self): + assert _normalize(0.0, low=0.8, high=2.0) == 0.0 + + def test_above_high_clamps(self): + assert _normalize(5.0, low=0.8, high=2.0) == 100.0 + + def test_none_returns_zero(self): + assert _normalize(None, low=0.8, high=2.0) == 0.0 + + +class TestNormalizeInverse: + def test_dd_at_worst(self): + # 10% drawdown = worst (0 pts) + assert _normalize_inverse(10.0, low=10.0, high=1.0) == 0.0 + + def test_dd_at_best(self): + # 1% drawdown = best (100 pts) + assert _normalize_inverse(1.0, low=10.0, high=1.0) == 100.0 + + def test_dd_midpoint(self): + assert _normalize_inverse(5.5, low=10.0, high=1.0) == pytest.approx(50.0) + + def test_none_returns_zero(self): + assert _normalize_inverse(None, low=10.0, high=1.0) == 0.0 + + +# --------------------------------------------------------------------------- +# compute_sqs +# --------------------------------------------------------------------------- + + +class TestComputeSQS: + def test_perfect_metrics(self): + """All metrics at 100-point boundaries => SQS near 100.""" + m = MetricsBundle( + trade_count=200, + profit_factor=2.0, + total_return_pct=5.0, + max_drawdown_pct=1.0, + sharpe_ratio=2.0, + win_rate=0.65, + monthly_win_rate=0.70, + equity_curve_r_squared=0.80, + ) + sqs, breakdown = compute_sqs(m) + assert sqs == pytest.approx(100.0, abs=0.5) + assert breakdown["profitability"] == pytest.approx(100.0, abs=0.5) + assert breakdown["risk"] == pytest.approx(100.0, abs=0.5) + assert breakdown["consistency"] == pytest.approx(100.0, abs=0.5) + assert breakdown["robustness"] == pytest.approx(100.0, abs=0.5) + + def test_worst_metrics(self): + """All metrics at 0-point boundaries => SQS = 0.""" + m = MetricsBundle( + trade_count=5, + profit_factor=0.5, + total_return_pct=-10.0, + max_drawdown_pct=15.0, + sharpe_ratio=-2.0, + win_rate=0.20, + monthly_win_rate=0.10, + equity_curve_r_squared=-0.5, + ) + sqs, _ = compute_sqs(m) + assert sqs == 0.0 + + def test_low_trade_penalty(self): + """< 20 trades => SQS * 0.5.""" + m = MetricsBundle( + trade_count=15, + profit_factor=1.5, + total_return_pct=2.0, + max_drawdown_pct=3.0, + sharpe_ratio=1.0, + win_rate=0.55, + monthly_win_rate=0.55, + equity_curve_r_squared=0.5, + ) + sqs_penalized, _ = compute_sqs(m) + + m_enough = m.model_copy(update={"trade_count": 100}) + sqs_full, _ = compute_sqs(m_enough) + + # Penalised score should be roughly half (trade_count affects robustness sub-score too) + assert sqs_penalized < sqs_full + assert sqs_penalized > 0 + + def test_midrange_metrics(self): + """Typical mid-range strategy should score 30-60.""" + m = MetricsBundle( + trade_count=80, + profit_factor=1.1, + total_return_pct=0.5, + max_drawdown_pct=5.0, + sharpe_ratio=0.5, + win_rate=0.50, + monthly_win_rate=0.50, + equity_curve_r_squared=0.30, + ) + sqs, _ = compute_sqs(m) + assert 30 <= sqs <= 65 + + def test_custom_weights(self): + """Custom weights should change the SQS.""" + m = MetricsBundle( + trade_count=80, + profit_factor=2.0, + total_return_pct=5.0, + max_drawdown_pct=8.0, + sharpe_ratio=0.0, + win_rate=0.40, + monthly_win_rate=0.40, + ) + w_profit_heavy = SQSWeights(profitability=0.80, risk=0.10, consistency=0.05, robustness=0.05) + w_risk_heavy = SQSWeights(profitability=0.10, risk=0.80, consistency=0.05, robustness=0.05) + + sqs_profit, _ = compute_sqs(m, w_profit_heavy) + sqs_risk, _ = compute_sqs(m, w_risk_heavy) + # This strategy has great profitability but mediocre risk + assert sqs_profit > sqs_risk + + +# --------------------------------------------------------------------------- +# build_split_result +# --------------------------------------------------------------------------- + + +def test_build_split_result(): + m = MetricsBundle( + trade_count=50, + profit_factor=1.2, + total_return_pct=2.5, + win_rate=0.55, + max_drawdown_pct=3.0, + sharpe_ratio=0.8, + monthly_win_rate=0.60, + equity_curve_r_squared=0.40, + ) + sr = build_split_result("test", "bt_run123", m) + assert sr.run_id == "bt_run123" + assert sr.trade_count == 50 + assert sr.profit_factor == 1.2 + assert sr.total_return_pct == 2.5 + + +# --------------------------------------------------------------------------- +# Journal I/O +# --------------------------------------------------------------------------- + + +class TestJournalIO: + def test_append_and_load(self, tmp_path): + journal_path = tmp_path / "journal.jsonl" + entry = JournalEntry( + entry_id="IMP-0001", + timestamp="2026-03-16T12:00:00", + experiment_name="test_exp_1", + hypothesis="Test hypothesis", + sqs_score=55.0, + sqs_breakdown={"profitability": 60.0, "risk": 50.0, "consistency": 55.0, "robustness": 50.0}, + verdict="better", + ) + append_journal_entry(journal_path, entry) + + entries = load_journal(journal_path) + assert len(entries) == 1 + assert entries[0].entry_id == "IMP-0001" + assert entries[0].experiment_name == "test_exp_1" + assert entries[0].sqs_score == 55.0 + + def test_multiple_entries(self, tmp_path): + journal_path = tmp_path / "journal.jsonl" + for i in range(3): + entry = JournalEntry( + entry_id=f"IMP-{i+1:04d}", + timestamp=f"2026-03-{16+i}T12:00:00", + experiment_name=f"exp_{i}", + hypothesis=f"Hypothesis {i}", + sqs_score=float(40 + i * 10), + ) + append_journal_entry(journal_path, entry) + + entries = load_journal(journal_path) + assert len(entries) == 3 + assert entries[2].sqs_score == 60.0 + + def test_get_next_entry_id(self, tmp_path): + journal_path = tmp_path / "journal.jsonl" + assert get_next_entry_id(journal_path) == "IMP-0001" + + entry = JournalEntry( + entry_id="IMP-0001", + timestamp="2026-03-16T12:00:00", + experiment_name="exp_1", + hypothesis="h", + ) + append_journal_entry(journal_path, entry) + assert get_next_entry_id(journal_path) == "IMP-0002" + + def test_load_empty(self, tmp_path): + journal_path = tmp_path / "nonexistent.jsonl" + entries = load_journal(journal_path) + assert entries == [] + + +# --------------------------------------------------------------------------- +# check_duplicate +# --------------------------------------------------------------------------- + + +class TestCheckDuplicate: + def test_finds_duplicates(self, tmp_path): + journal_path = tmp_path / "journal.jsonl" + for name in ["exp_a", "exp_b", "exp_a"]: + entry = JournalEntry( + entry_id=get_next_entry_id(journal_path), + timestamp="2026-03-16T12:00:00", + experiment_name=name, + hypothesis="h", + ) + append_journal_entry(journal_path, entry) + + dupes = check_duplicate(journal_path, "exp_a") + assert len(dupes) == 2 + + def test_no_duplicates(self, tmp_path): + journal_path = tmp_path / "journal.jsonl" + entry = JournalEntry( + entry_id="IMP-0001", + timestamp="2026-03-16T12:00:00", + experiment_name="exp_a", + hypothesis="h", + ) + append_journal_entry(journal_path, entry) + + dupes = check_duplicate(journal_path, "exp_z") + assert len(dupes) == 0 + + +# --------------------------------------------------------------------------- +# rebuild_registry +# --------------------------------------------------------------------------- + + +class TestRebuildRegistry: + def test_registry_and_leaderboard(self, tmp_path): + journal_path = tmp_path / "journal.jsonl" + registry_path = tmp_path / "registry.json" + leaderboard_path = tmp_path / "LEADERBOARD.md" + + # Create entries with different SQS scores + for i, (name, sqs) in enumerate([("exp_low", 30.0), ("exp_high", 70.0), ("exp_mid", 50.0)]): + test_result = SplitResult( + run_id=f"bt_{name}", + trade_count=60, + profit_factor=1.0 + i * 0.2, + total_return_pct=float(i), + win_rate=0.5, + ) + entry = JournalEntry( + entry_id=f"IMP-{i+1:04d}", + timestamp=f"2026-03-{16+i}T12:00:00", + experiment_name=name, + hypothesis=f"h{i}", + sqs_score=sqs, + results={"test": test_result}, + verdict="better" if sqs > 50 else "worse", + ) + append_journal_entry(journal_path, entry) + + registry = rebuild_registry(journal_path, registry_path, leaderboard_path) + + # Sorted by SQS descending + assert len(registry.entries) == 3 + assert registry.entries[0].experiment_name == "exp_high" + assert registry.entries[0].sqs_score == 70.0 + assert registry.entries[2].experiment_name == "exp_low" + + # Files exist + assert registry_path.exists() + assert leaderboard_path.exists() + + # Leaderboard contains table + lb_text = leaderboard_path.read_text() + assert "exp_high" in lb_text + assert "exp_low" in lb_text + assert "| # |" in lb_text