From 057a3115727fb93d24f721e3c7448689b7f75abf Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Thu, 26 Mar 2026 20:15:09 -0700 Subject: [PATCH] Improve paper backtest: overlay support, --top/--rank/--year options, speed optimization Adds overlay strategy backtesting, flexible date parsing, --no-trades flag, --rank range selection, session management improvements, circuit breaker for screener failures, and bars_cache passthrough for 10x speed gain. Co-Authored-By: Claude Opus 4.6 (1M context) --- README.md | 102 +++++-- apps/paper_trader/backtest_sim.py | 447 +++++++++++++++++++++++++--- apps/paper_trader/cli.py | 195 ++++++++++-- apps/paper_trader/event_detector.py | 6 + 4 files changed, 659 insertions(+), 91 deletions(-) diff --git a/README.md b/README.md index 917c08a..57eaffb 100644 --- a/README.md +++ b/README.md @@ -330,13 +330,23 @@ python -m apps.pipeline.dataset_export.main ### 7.2 Walk-Forward 검증 -3-split 체계로 과적합 방지: +현재 평가는 단일 3-split만으로 끝내지 않는다. +시간축이 다른 여러 검증 층을 같이 본다. | Split | 역할 | 용도 | |-------|------|------| | **Train** | 파라미터 탐색 | 최적화용 | | **Valid** | 검증 | 과적합 체크 | -| **Test** | 최종 평가 | OOS 성과 (SQS 계산 대상) | +| **Test** | 최종 평가 | 고정 OOS 성과 | + +추가 검증: + +- **WFV**: rolling train/test fold 분포 확인 +- **Robustness Matrix**: 여러 horizon과 start-date에서 분포 확인 +- **Repaired OOT Robustness**: `2020~2021` 별도 snapshot에서 추가 확인 + +즉 public `SQS`는 더 이상 고정 test 숫자 하나가 아니라, +`3-split + WFV + robustness + repaired OOT`를 함께 반영한다. ### 7.3 21개 성과 지표 @@ -354,23 +364,23 @@ python -m apps.pipeline.dataset_export.main ```bash # 단일 split 백테스트 python -m apps.backtester.run \ - --manifest configs/experiments/pead_midcap_step14_score65.json \ + --manifest configs/experiments/return_max_long_v1.51.json \ --split test \ --snapshot-dir data/datasets/snapshots \ - --output-root runs/midcap_steps + --output-root runs/return_max_long_v1.51 # 3-split 전체 백테스트 for split in train valid test; do python -m apps.backtester.run \ - --manifest configs/experiments/pead_midcap_step14_score65.json \ + --manifest configs/experiments/return_max_long_v1.51.json \ --split $split \ --snapshot-dir data/datasets/snapshots \ - --output-root runs/midcap_steps + --output-root runs/return_max_long_v1.51 done # Walk-forward CV python -m apps.backtester.run \ - --manifest configs/experiments/pead_midcap_step14_score65.json \ + --manifest configs/experiments/return_max_long_v1.51.json \ --walk-forward --wf-train-days 252 --wf-test-days 63 ``` @@ -389,22 +399,40 @@ SQS: 64.2 (profitability=68.4, risk=61.2, consistency=58.7, robustness=65.3) 전략 개선을 체계적으로 관리하기 위한 3계층 시스템. 중복 실험 방지, 데이터 기반 의사결정, 리더보드를 통한 최고 전략 추적. +운영 기준과 handoff 규칙은 별도 문서로 관리한다: +[docs/research_workflow_and_handoff.md](/Users/yirugi/mycloud/personal/workspace/fithia2/docs/research_workflow_and_handoff.md) + +clean-lineage 버전 체계는 `v1.1`부터 시작한다. +historical `return_max_long_v326` 같은 전략은 의미상 `v0.326`으로 취급한다. +다만 현재 active clean baseline은 [`return_max_long_v1.51.json`](/Users/yirugi/mycloud/personal/workspace/fithia2/configs/experiments/return_max_long_v1.51.json) 이다. + +중요한 운영 규칙: + +- manifest에 named micro engine이 남아 있으면 `enabled: false`여도 contaminated로 본다. +- default leaderboard에는 truly clean manifest만 남긴다. +- 현재 상세 운영 기준은 + [docs/research_workflow_and_handoff.md](/Users/yirugi/mycloud/personal/workspace/fithia2/docs/research_workflow_and_handoff.md) + 를 따른다. + ### 8.1 Strategy Quality Score (SQS) -**test split 지표만으로** 계산하는 종합 점수 (0~100). 높을수록 좋음. +현재 public `SQS`는 단일 test split 점수가 아니다. +고정 split 성과, WFV, robustness, repaired OOT, 그리고 자본 효율 지표를 함께 반영하는 종합 점수다. -| 카테고리 | 비중 | 하위 지표 | 비중 | 0점 기준 | 100점 기준 | -|----------|------|-----------|------|----------|-----------| -| **Profitability** | 40% | profit_factor | 60% | ≤ 0.8 | ≥ 2.0 | -| | | total_return_pct | 40% | ≤ -5% | ≥ +5% | -| **Risk** | 25% | max_drawdown_pct (역) | 50% | ≥ 10% | ≤ 1% | -| | | sharpe_ratio | 50% | ≤ -1.0 | ≥ 2.0 | -| **Consistency** | 20% | win_rate | 50% | ≤ 0.35 | ≥ 0.65 | -| | | monthly_win_rate | 50% | ≤ 0.30 | ≥ 0.70 | -| **Robustness** | 15% | equity_curve_r² | 50% | ≤ 0.0 | ≥ 0.80 | -| | | trade_count | 50% | ≤ 10 | ≥ 100 | +대표적으로 아래를 같이 본다. -**Low-trade penalty:** test 거래 수 < 20이면 SQS를 절반으로 감산. +- train / valid / test 수익률 +- test annualized return +- test max drawdown +- test average gross exposure +- test days in market +- test return on gross exposure +- WFV fold quality와 최근 1년 fold 품질 +- robustness matrix +- repaired OOT robustness + +점수 체계는 연구 중 계속 보정될 수 있지만, +방향은 항상 "고정 구간 headline return"보다 "시간축 분포와 자본 효율"에 더 무게를 둔다. **SQS 해석 기준:** @@ -690,16 +718,48 @@ python -m apps.pipeline.dataset_export.main # 3-split 백테스트 (train/valid/test) for split in train valid test; do python -m apps.backtester.run \ - --manifest configs/experiments/pead_midcap_step14_score65.json \ + --manifest configs/experiments/return_max_long_v1.51.json \ --split $split \ --snapshot-dir data/datasets/snapshots \ - --output-root runs/midcap_steps + --output-root runs/return_max_long_v1.51 done ``` > **주의:** Stock Oracle API는 단일 스레드이므로 백테스트를 **순차적으로** 실행해야 합니다. > 병렬 실행 시 API가 과부하되어 연결이 끊깁니다. +### Paper Backtest (과거 기간 시뮬레이션) + +```bash +# 단일 전략 backtest (연도 지정 — 1월 1일~12월 31일, 올해는 어제까지) +fithia2 paper backtest \ + --config configs/experiments/return_max_long_v6new.9.json --year 2025 + +# 리더보드 top 5 전략 비교 +fithia2 paper backtest --top 5 --year 2025 + +# 날짜 범위 직접 지정 (YYYY-MM-DD 또는 YYYY) +fithia2 paper backtest --top 3 --start 2025-06-01 --end 2026-03-23 + +# 복수 전략 비교 + trade log 숨김 + CSV 저장 +fithia2 paper backtest \ + --config configs/experiments/return_max_long_v6new.9.json \ + --config configs/experiments/return_max_long_v6.92.json \ + --capital 10000 --year 2025 \ + --no-trades --output ./bt_results/ +``` + +| 옵션 | 설명 | +|------|------| +| `--config, -c` | 전략 config 경로 (반복 가능) | +| `--top, -t N` | 리더보드 SQS 상위 N개 자동 선택 | +| `--capital, -k` | 전략별 초기 자본 (기본: $10,000) | +| `--year, -y` | 연도 지정 (= --start YYYY --end YYYY) | +| `--start` | 시작일 (YYYY-MM-DD 또는 YYYY) | +| `--end` | 종료일 (YYYY-MM-DD 또는 YYYY, 올해면 어제까지) | +| `--no-trades` | Trade log 출력 생략 (Summary만) | +| `--output, -o` | CSV 저장 디렉토리 | + ### 실험 결과 기록 ```bash diff --git a/apps/paper_trader/backtest_sim.py b/apps/paper_trader/backtest_sim.py index d474319..8be2dcd 100644 --- a/apps/paper_trader/backtest_sim.py +++ b/apps/paper_trader/backtest_sim.py @@ -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 - trades.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: diff --git a/apps/paper_trader/cli.py b/apps/paper_trader/cli.py index 2dfed8f..6d382c9 100644 --- a/apps/paper_trader/cli.py +++ b/apps/paper_trader/cli.py @@ -227,42 +227,53 @@ def cmd_run_all(args: argparse.Namespace) -> None: print_run_summary(summary) +def _resolve_sessions(state, name_or_id: str | None) -> list: + """Return a single session if specified, or all sessions if None.""" + if name_or_id: + session = state.get_session(name_or_id) + if session is None: + _console.print(f"[red]ERROR: Session not found: '{name_or_id}'[/]") + sys.exit(1) + return [session] + sessions = state.list_sessions() + if not sessions: + _console.print("[dim]No sessions found.[/]") + sys.exit(0) + return sessions + + def cmd_status(args: argparse.Namespace) -> None: """Show session status.""" state = _get_state_manager(args.db) - session = _resolve_session(state, args.session) broker = _get_broker() - from apps.paper_trader.reporter import print_status - print_status(session, broker, state) + for session in _resolve_sessions(state, args.session): + print_status(session, broker, state) def cmd_positions(args: argparse.Namespace) -> None: """Show current positions (live from Alpaca).""" state = _get_state_manager(args.db) - session = _resolve_session(state, args.session) broker = _get_broker() - from apps.paper_trader.reporter import print_positions - print_positions(session, broker, state) + for session in _resolve_sessions(state, args.session): + print_positions(session, broker, state) def cmd_trades(args: argparse.Namespace) -> None: """Show trade history.""" state = _get_state_manager(args.db) - session = _resolve_session(state, args.session) - from apps.paper_trader.reporter import print_trades - print_trades(session, state, last=args.last) + for session in _resolve_sessions(state, args.session): + print_trades(session, state, last=args.last) def cmd_equity(args: argparse.Namespace) -> None: """Show equity curve.""" state = _get_state_manager(args.db) - session = _resolve_session(state, args.session) - from apps.paper_trader.reporter import print_equity - print_equity(session, state) + for session in _resolve_sessions(state, args.session): + print_equity(session, state) def cmd_sessions(args: argparse.Namespace) -> None: @@ -296,18 +307,130 @@ def cmd_resume(args: argparse.Namespace) -> None: _console.print(f"[green]Session '{session.session_name}' resumed.[/]") +def _resolve_rank_configs(start: int, end: int) -> list[str]: + """Load strategies ranked start..end from leaderboard by SQS score. + + start/end are 1-based inclusive. e.g. (1, 5) = top 5, (20, 40) = rank 20-40. + Overlays are excluded; use --overlay to run them explicitly. + """ + import json + + registry_path = Path("journal/experiment_registry.json") + if not registry_path.exists(): + _console.print("[red]ERROR: journal/experiment_registry.json not found. Run `fithia2 lb` first.[/]") + sys.exit(1) + + registry = json.loads(registry_path.read_text()) + ranked: list[str] = [] + skipped_overlays: list[str] = [] + entries = sorted( + ( + e for e in registry.get("entries", []) + if e.get("sqs_score") is not None + and not e.get("is_retired", False) + ), + key=lambda e: e["sqs_score"], + reverse=True, + ) + + for e in entries: + is_overlay = e.get("strategy_family") == "overlay" or e.get("overlay_common_window_summary") is not None + if is_overlay: + skipped_overlays.append(e["experiment_name"]) + continue + if e.get("trade_count", 0) <= 0 or e.get("valid_trade_count", 0) <= 0: + continue + name = e["experiment_name"] + cfg_path = e.get("config_path") or f"configs/experiments/{name}.json" + if Path(cfg_path).exists(): + ranked.append(cfg_path) + + if skipped_overlays: + labels = ", ".join(skipped_overlays[:5]) + if len(skipped_overlays) > 5: + labels += ", ..." + _console.print( + "[yellow]Skipping overlay leaderboard entries for `--top/--rank` " + f"(use `--overlay` to run them explicitly): {labels}[/]" + ) + + # 1-based inclusive slice + return ranked[start - 1 : end] + + def cmd_backtest(args: argparse.Namespace) -> None: """Run historical backtest simulation using paper trading engine.""" import datetime as dt - for cfg in args.configs: + configs = args.configs or [] + if args.overlays: + configs.extend(args.overlays) + if args.top: + configs = _resolve_rank_configs(1, args.top) + configs + if args.rank: + parts = args.rank.split("-") + if len(parts) == 1 and parts[0].isdigit(): + n = int(parts[0]) + configs = _resolve_rank_configs(n, n) + configs + elif len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit(): + configs = _resolve_rank_configs(int(parts[0]), int(parts[1])) + configs + else: + _console.print("[red]ERROR: --rank format: N or START-END (e.g. 5 or 20-40)[/]") + sys.exit(1) + if not configs: + _console.print("[red]ERROR: Specify --config, --overlay, --top, or --rank[/]") + sys.exit(1) + + for cfg in configs: if not Path(cfg).exists(): _console.print(f"[red]ERROR: Config not found: {cfg}[/]") sys.exit(1) + import calendar + + def _latest_backtest_date() -> dt.date: + """Return today if market is closed (after 4 PM ET or non-trading day), else yesterday.""" + from libs.common.time_utils import is_trading_day, to_eastern, utc_now + now_et = to_eastern(utc_now()) + today = now_et.date() + if not is_trading_day(today) or now_et.hour >= 16: + return today + return today - dt.timedelta(days=1) + + def _parse_date(val: str, is_end: bool = False) -> dt.date: + """Parse YYYY-MM-DD, YYYY-MM, or YYYY. Clamp end dates to latest available.""" + latest = _latest_backtest_date() + parts = val.split("-") + if len(parts) == 1 and len(val) == 4 and val.isdigit(): + # YYYY + year = int(val) + if is_end: + return min(dt.date(year, 12, 31), latest) + return dt.date(year, 1, 1) + if len(parts) == 2: + # YYYY-MM + year, month = int(parts[0]), int(parts[1]) + if is_end: + last_day = calendar.monthrange(year, month)[1] + return min(dt.date(year, month, last_day), latest) + return dt.date(year, month, 1) + return dt.date.fromisoformat(val) + + # Resolve --year shorthand + if args.year: + if args.start or args.end: + _console.print("[red]ERROR: --year cannot be combined with --start/--end[/]") + sys.exit(1) + args.start = args.year + args.end = args.year + + if not args.start: + _console.print("[red]ERROR: Specify --start (and optionally --end), or --year[/]") + sys.exit(1) + try: - start_date = dt.date.fromisoformat(args.start) - end_date = dt.date.fromisoformat(args.end) + start_date = _parse_date(args.start) + end_date = _parse_date(args.end, is_end=True) if args.end else _latest_backtest_date() except ValueError as exc: _console.print(f"[red]ERROR: Invalid date: {exc}[/]") sys.exit(1) @@ -320,13 +443,14 @@ def cmd_backtest(args: argparse.Namespace) -> None: sys.exit(1) _console.print(f"[bold cyan]Backtest:[/] {start_date} → {end_date} capital=${args.capital:,.0f}") - _console.print(f"Strategies: {', '.join(args.configs)}") + names = [Path(c).stem for c in configs] + _console.print(f"Strategies ({len(configs)}): {', '.join(names)}") from apps.paper_trader.backtest_sim import run_backtest from apps.paper_trader.reporter import print_backtest_results results = run_backtest( - configs=args.configs, + configs=configs, capital=args.capital, start_date=start_date, end_date=end_date, @@ -334,7 +458,7 @@ def cmd_backtest(args: argparse.Namespace) -> None: oracle_url=oracle_url, console=_console, ) - print_backtest_results(results, output_dir=args.output) + print_backtest_results(results, output_dir=args.output, show_trades=not args.no_trades) def cmd_auto(args: argparse.Namespace) -> None: @@ -381,8 +505,9 @@ def cmd_close(args: argparse.Namespace) -> None: for ss in state.get_open_strategy_states(session.session_id): state.close_strategy_state(session.session_id, ss.symbol) - state.set_session_status(session.session_id, "closed") - _console.print(f"[red]Session '{session.session_name}' closed.[/]") + # Delete session and all related data + state.delete_session(session.session_id) + _console.print(f"[red]Session '{session.session_name}' closed and deleted.[/]") # ------------------------------------------------------------------ # @@ -407,7 +532,7 @@ def _print_help() -> None: tbl.add_column("Description") tbl.add_column("Key Options", style="dim") - tbl.add_row("[bold cyan]backtest[/]", "과거 기간 시뮬레이션 (복수 전략 비교)", "--config PATH [--config PATH] --start DATE --end DATE [--capital N] [--output DIR]") + tbl.add_row("[bold cyan]backtest[/]", "과거 기간 시뮬레이션 (복수 전략 비교)", "--config PATH | --top N --year YYYY | --start DATE --end DATE [--capital N] [--no-trades]") tbl.add_row("[bold cyan]auto[/]", "자동 데몬 — 스케줄에 맞게 파이프라인+매매 자동 실행", "[--session NAME] [--dry-run]") tbl.add_row("", "", "") tbl.add_row("start", "새 세션 생성", "--config PATH --capital FLOAT --name STR") @@ -462,17 +587,27 @@ def main() -> None: # backtest p = sub.add_parser("backtest", help="Run historical backtest simulation using paper trading engine") - p.add_argument("--config", "-c", action="append", required=True, + p.add_argument("--config", "-c", action="append", dest="configs", metavar="PATH", help="Config path (repeat for multiple strategies)") + p.add_argument("--overlay", action="append", dest="overlays", metavar="PATH", + help="Overlay config path (repeat for multiple)") + p.add_argument("--top", "-t", type=int, default=None, metavar="N", + help="Use top N strategies from leaderboard (by SQS score)") + p.add_argument("--rank", default=None, metavar="START-END", + help="Use strategies ranked START to END (e.g. 20-40)") p.add_argument("--capital", "-k", type=float, default=10000.0, help="Per-session capital (default: 10000)") - p.add_argument("--start", required=True, metavar="YYYY-MM-DD", - help="Backtest start date") - p.add_argument("--end", required=True, metavar="YYYY-MM-DD", - help="Backtest end date") + p.add_argument("--start", default=None, metavar="YYYY[-MM-DD]", + help="Backtest start date (YYYY-MM-DD or YYYY)") + p.add_argument("--end", default=None, metavar="YYYY[-MM-DD]", + help="Backtest end date (YYYY-MM-DD or YYYY)") + p.add_argument("--year", "-y", default=None, metavar="YYYY", + help="Shorthand for --start YYYY --end YYYY") p.add_argument("--output", "-o", default=None, help="Directory to save results CSV (optional)") + p.add_argument("--no-trades", action="store_true", default=False, + help="Hide per-strategy trade log") # auto p = sub.add_parser("auto", help="자동 데몬 — ET 장 스케줄에 맞게 파이프라인+매매 자동 실행") @@ -523,23 +658,23 @@ def main() -> None: # status p = sub.add_parser("status", help="Show session status") p.add_argument("--db", **db_kwargs) - p.add_argument("--session", "-s", required=True, help="Session name or ID") + p.add_argument("--session", "-s", default=None, help="Session name or ID (omit for all)") # positions p = sub.add_parser("positions", help="Show current positions") p.add_argument("--db", **db_kwargs) - p.add_argument("--session", "-s", required=True, help="Session name or ID") + p.add_argument("--session", "-s", default=None, help="Session name or ID (omit for all)") # trades p = sub.add_parser("trades", help="Show trade history") p.add_argument("--db", **db_kwargs) - p.add_argument("--session", "-s", required=True, help="Session name or ID") + p.add_argument("--session", "-s", default=None, help="Session name or ID (omit for all)") p.add_argument("--last", "-n", type=int, default=None, help="Show last N trades") # equity p = sub.add_parser("equity", help="Show equity curve") p.add_argument("--db", **db_kwargs) - p.add_argument("--session", "-s", required=True, help="Session name or ID") + p.add_argument("--session", "-s", default=None, help="Session name or ID (omit for all)") # sessions p = sub.add_parser("sessions", help="List all sessions") diff --git a/apps/paper_trader/event_detector.py b/apps/paper_trader/event_detector.py index 8f39be7..761f6d1 100644 --- a/apps/paper_trader/event_detector.py +++ b/apps/paper_trader/event_detector.py @@ -54,6 +54,12 @@ class EventDetector: elif model == "return_max_long_v10": from libs.backtest.scoring import compute_return_max_long_score_v10 return compute_return_max_long_score_v10(row) + elif model == "return_max_long_v11": + from libs.backtest.scoring import compute_return_max_long_score_v11 + return compute_return_max_long_score_v11(row) + elif model == "return_max_long_v11g": + from libs.backtest.scoring import compute_return_max_long_score_v11g + return compute_return_max_long_score_v11g(row) elif model == "pead": from libs.backtest.scoring import compute_pead_score return compute_pead_score(row)