from __future__ import annotations import argparse import json from dataclasses import dataclass from pathlib import Path from statistics import mean from typing import Any, Sequence from config import BotConfig, RuntimePaths from signal_correlation import _pair_closed_trades from strategy_registry import load_strategy_presets @dataclass(frozen=True) class ExperimentPaths: strategy_id: str root: Path paper_state: Path trade_journal: Path metrics: Path @dataclass(frozen=True) class TournamentScore: strategy_id: str closed_trades: int win_rate: float avg_pnl_usd: float total_pnl_usd: float max_drawdown_usd: float profit_factor: float avg_time_in_trade_minutes: float score: float sample_warning: bool def build_experiment_paths(runtime_dir: str | Path, strategy_id: str) -> ExperimentPaths: root = Path(runtime_dir).expanduser() / "experiments" / strategy_id return ExperimentPaths( strategy_id=strategy_id, root=root, paper_state=root / "paper_state.json", trade_journal=root / "Tradeanalyse" / "trade_journal.jsonl", metrics=root / "metrics.json", ) def ensure_experiment_dirs(runtime_dir: str | Path, strategy_ids: Sequence[str]) -> list[ExperimentPaths]: paths = [build_experiment_paths(runtime_dir, strategy_id) for strategy_id in strategy_ids] for item in paths: item.root.mkdir(parents=True, exist_ok=True) item.trade_journal.parent.mkdir(parents=True, exist_ok=True) return paths def _read_jsonl(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] rows = [] for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if line: rows.append(json.loads(line)) return rows def _max_drawdown(pnls: list[float]) -> float: equity = 0.0 peak = 0.0 max_dd = 0.0 for pnl in pnls: equity += pnl peak = max(peak, equity) max_dd = min(max_dd, equity - peak) return round(max_dd, 8) def _profit_factor(pnls: list[float]) -> float: gains = sum(p for p in pnls if p > 0) losses = abs(sum(p for p in pnls if p < 0)) if losses == 0: return round(gains, 8) if gains else 0.0 return round(gains / losses, 8) def score_journal(strategy_id: str, journal_path: str | Path, *, min_samples: int = 5) -> TournamentScore: events = _read_jsonl(Path(journal_path)) pairs, _ = _pair_closed_trades(events) pnls = [float(exit_event.get("realized_pnl_usd", 0.0)) for _, exit_event in pairs] durations = [] for entry, exit_event in pairs: try: from signal_correlation import _parse_ts durations.append(int((_parse_ts(exit_event["ts"]) - _parse_ts(entry["ts"])).total_seconds() // 60)) except Exception: continue closed = len(pnls) avg_pnl = round(mean(pnls), 8) if pnls else 0.0 total_pnl = round(sum(pnls), 8) win_rate = round(sum(1 for pnl in pnls if pnl > 0) / closed, 4) if closed else 0.0 max_dd = _max_drawdown(pnls) pf = _profit_factor(pnls) avg_time = round(mean(durations), 4) if durations else 0.0 sample_warning = closed < min_samples sample_penalty = 2.0 if sample_warning else 0.0 drawdown_penalty = abs(max_dd) * 0.35 overfit_penalty = 0.25 if closed == 1 and total_pnl > 0 else 0.0 score = round(avg_pnl + (win_rate * 0.5) + min(pf, 5.0) * 0.1 - drawdown_penalty - sample_penalty - overfit_penalty, 8) return TournamentScore( strategy_id=strategy_id, closed_trades=closed, win_rate=win_rate, avg_pnl_usd=avg_pnl, total_pnl_usd=total_pnl, max_drawdown_usd=max_dd, profit_factor=pf, avg_time_in_trade_minutes=avg_time, score=score, sample_warning=sample_warning, ) def rank_strategies_from_journals(runtime_dir: str | Path, strategy_ids: Sequence[str], *, min_samples: int = 5) -> list[TournamentScore]: scores = [] for strategy_id in strategy_ids: paths = build_experiment_paths(runtime_dir, strategy_id) scores.append(score_journal(strategy_id, paths.trade_journal, min_samples=min_samples)) return sorted(scores, key=lambda item: (item.score, item.total_pnl_usd, item.closed_trades), reverse=True) def write_metrics(runtime_dir: str | Path, scores: Sequence[TournamentScore]) -> None: for score in scores: paths = build_experiment_paths(runtime_dir, score.strategy_id) paths.root.mkdir(parents=True, exist_ok=True) paths.metrics.write_text(json.dumps(score.__dict__, indent=2, sort_keys=True), encoding="utf-8") def render_tournament_report(scores: Sequence[TournamentScore]) -> str: lines = ["Strategy Tournament Report"] if not scores: lines.append("No strategies found.") return "\n".join(lines) for idx, score in enumerate(scores, start=1): warning = " observation-only" if score.sample_warning else "" lines.append( f"{idx}. {score.strategy_id}: score={score.score:+.4f}, n={score.closed_trades}, " f"win={score.win_rate:.2f}, avg={score.avg_pnl_usd:+.4f}, total={score.total_pnl_usd:+.4f}, " f"max_dd={score.max_drawdown_usd:+.4f}, pf={score.profit_factor:.2f}, avg_time={score.avg_time_in_trade_minutes:.1f}m{warning}" ) return "\n".join(lines) def main(argv: list[str] | None = None) -> int: cfg = BotConfig.from_file() paths = RuntimePaths.from_config(cfg) parser = argparse.ArgumentParser(description="Rank isolated paper-trading strategy experiments.") parser.add_argument("--runtime-dir", default=str(paths.runtime_dir)) parser.add_argument("--min-samples", type=int, default=5) parser.add_argument("--init", action="store_true", help="Create experiment directories for all registered strategies.") parser.add_argument("--strategies", nargs="*", default=None) args = parser.parse_args(argv) presets = load_strategy_presets() strategy_ids = args.strategies or list(presets.keys()) if args.init: ensure_experiment_dirs(args.runtime_dir, strategy_ids) scores = rank_strategies_from_journals(args.runtime_dir, strategy_ids, min_samples=args.min_samples) write_metrics(args.runtime_dir, scores) print(render_tournament_report(scores)) return 0 if __name__ == "__main__": raise SystemExit(main())