You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
223 lines
8.5 KiB
Python
223 lines
8.5 KiB
Python
"""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()
|