diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 459fb53..d954d06 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -10,16 +10,14 @@ on: permissions: contents: read issues: write - id-token: write jobs: - drift: + preflight_backtests: + if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) runs-on: ubuntu-latest - timeout-minutes: 15 - env: - FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" - STRATEGY_DOMAIN: crypto - + timeout-minutes: 30 + outputs: + snapshot_repository_ref: ${{ steps.snapshot-input.outputs.snapshot_repository_ref }} steps: - name: Checkout uses: actions/checkout@v6 @@ -28,7 +26,7 @@ jobs: uses: actions/checkout@v6 with: repository: QuantStrategyLab/QuantPlatformKit - ref: main + ref: bda6afdab0a2dd693c35d14493176829f4da1231 path: external/QuantPlatformKit - name: Set up Python @@ -43,36 +41,175 @@ jobs: python -m pip install -e . pandas python -m pip install --no-deps -e external/QuantPlatformKit - - name: Run drift detection - run: quant-lifecycle drift --domain crypto --no-alerts + - name: Download latest trusted lifecycle inputs + id: snapshot-input + env: + GH_TOKEN: ${{ secrets.SNAPSHOT_REPOSITORY_TOKEN }} + INPUT_ROOT: ${{ runner.temp }}/crypto-lifecycle-inputs + run: | + set -euo pipefail + if [ -z "${GH_TOKEN:-}" ]; then + echo "::error::SNAPSHOT_REPOSITORY_TOKEN is required for lifecycle input artifact access" + exit 1 + fi + gh api --paginate --slurp \ + "/repos/QuantStrategyLab/CryptoLivePoolPipelines/actions/artifacts?per_page=100" \ + > "${RUNNER_TEMP}/snapshot-artifacts.json" + gh api --paginate --slurp \ + "/repos/QuantStrategyLab/CryptoLivePoolPipelines/actions/workflows/publish-lifecycle-inputs.yml/runs?branch=main&status=success&per_page=100" \ + > "${RUNNER_TEMP}/trusted-snapshot-runs.json" + python - <<'PY' > "${RUNNER_TEMP}/snapshot-artifact-selection.txt" + import json + import os + from pathlib import Path + pages = json.loads((Path(os.environ["RUNNER_TEMP"]) / "snapshot-artifacts.json").read_text()) + run_pages = json.loads((Path(os.environ["RUNNER_TEMP"]) / "trusted-snapshot-runs.json").read_text()) + artifacts = [item for page in pages for item in page.get("artifacts", [])] + trusted_runs = [run for page in run_pages for run in page.get("workflow_runs", [])] + selected = None + for run in sorted( + trusted_runs, + key=lambda item: (item.get("run_number", 0), item.get("run_attempt", 0)), + reverse=True, + ): + matches = [ + item for item in artifacts + if not item.get("expired") + and str(item.get("name", "")).startswith("crypto-lifecycle-inputs-") + and item.get("workflow_run", {}).get("id") == run["id"] + ] + if matches: + selected = max(matches, key=lambda item: item["created_at"]) + break + if selected is None: + raise SystemExit("no trusted crypto lifecycle input artifact is available") + print(selected["id"], selected["workflow_run"]["id"]) + PY + read -r artifact_id workflow_run_id < "${RUNNER_TEMP}/snapshot-artifact-selection.txt" + gh api "/repos/QuantStrategyLab/CryptoLivePoolPipelines/actions/runs/${workflow_run_id}" \ + > "${RUNNER_TEMP}/snapshot-workflow-run.json" + python - <<'PY' + import json + import os + from pathlib import Path + run = json.loads((Path(os.environ["RUNNER_TEMP"]) / "snapshot-workflow-run.json").read_text()) + expected = { + "conclusion": "success", + "head_branch": "main", + "path": ".github/workflows/publish-lifecycle-inputs.yml", + } + mismatches = {key: run.get(key) for key, value in expected.items() if run.get(key) != value} + if run.get("head_repository", {}).get("full_name") != "QuantStrategyLab/CryptoLivePoolPipelines": + mismatches["head_repository"] = run.get("head_repository", {}).get("full_name") + if mismatches: + raise SystemExit(f"lifecycle input provenance check failed: {mismatches}") + PY + snapshot_repository_ref="$(python - <<'PY' + import json + import os + from pathlib import Path + run = json.loads((Path(os.environ["RUNNER_TEMP"]) / "snapshot-workflow-run.json").read_text()) + print(run["head_sha"]) + PY + )" + if [[ ! "${snapshot_repository_ref}" =~ ^[0-9a-f]{40}$ ]]; then + echo "::error::Invalid snapshot producer head SHA" + exit 1 + fi + echo "snapshot_repository_ref=${snapshot_repository_ref}" >> "${GITHUB_OUTPUT}" + gh api "/repos/QuantStrategyLab/CryptoLivePoolPipelines/actions/artifacts/${artifact_id}/zip" \ + > "${RUNNER_TEMP}/snapshot-artifact.zip" + python - <<'PY' + import os + import zipfile + from pathlib import Path + archive = Path(os.environ["RUNNER_TEMP"]) / "snapshot-artifact.zip" + target_root = Path(os.environ["INPUT_ROOT"]) + required = {"research_panel.csv.gz", "market_history.csv.gz", "manifest.json"} + with zipfile.ZipFile(archive) as bundle: + by_name = {Path(name).name: name for name in bundle.namelist() if Path(name).name in required} + missing = sorted(required - set(by_name)) + if missing: + raise SystemExit(f"crypto lifecycle artifact is missing: {', '.join(missing)}") + target_root.mkdir(parents=True, exist_ok=True) + for name, member in by_name.items(): + (target_root / name).write_bytes(bundle.read(member)) + PY - - name: Sync drift alerts to GitHub Issues + - name: Build lifecycle preflight bundle env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - GITHUB_REPOSITORY: ${{ github.repository }} - run: python scripts/run_drift_github_issues.py + INPUT_ROOT: ${{ runner.temp }}/crypto-lifecycle-inputs + LIFECYCLE_PREFLIGHT_BUNDLE_ROOT: ${{ runner.temp }}/lifecycle-preflight-bundle + run: | + set -euo pipefail + python - <<'PY' + import json + import os + import subprocess + from pathlib import Path - - name: Checkout AIAuditBridge - uses: actions/checkout@v6 + profiles = json.loads( + subprocess.check_output( + ["python", "scripts/run_walk_forward_backtest.py", "--list-profiles"], + text=True, + ) + )["profiles"] + input_root = Path(os.environ["INPUT_ROOT"]) + bundle_root = Path(os.environ["LIFECYCLE_PREFLIGHT_BUNDLE_ROOT"]) + store_root = bundle_root / "data" / "lifecycle_store" + for profile in profiles: + returns_output = ( + bundle_root + / "external" + / "CryptoLivePoolPipelines" + / "data" + / "output" + / profile + / "portfolio_and_tracker_returns.csv" + ) + subprocess.check_call( + [ + "python", + "scripts/run_walk_forward_backtest.py", + "--profile", + profile, + "--panel", + str(input_root / "research_panel.csv.gz"), + "--market-history", + str(input_root / "market_history.csv.gz"), + "--store-root", + str(store_root), + "--returns-output", + str(returns_output), + ] + ) + PY + + - name: Upload lifecycle preflight artifact + uses: actions/upload-artifact@v4 with: - repository: QuantStrategyLab/AIAuditBridge - ref: main - path: external/AIAuditBridge + name: lifecycle-preflight-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/lifecycle-preflight-bundle + if-no-files-found: error - - name: Dual-review critical drift - env: - AIAUDIT_BRIDGE_ROOT: external/AIAuditBridge - CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} - AI_GATEWAY_SERVICE_URL: ${{ vars.AI_GATEWAY_SERVICE_URL }} - run: | - script="external/AIAuditBridge/scripts/run_drift_dual_review.py" - if [ ! -f "$script" ]; then - echo "::notice::dual-review scripts unavailable; skipping until AIAuditBridge is merged" - exit 0 - fi - if [ -z "${CODEX_AUDIT_SERVICE_URL:-}" ]; then - echo "::notice::CODEX_AUDIT_SERVICE_URL not configured; skipping dual-review dispatch" - exit 0 - fi - PYTHONPATH=external/AIAuditBridge python "$script" \ - --domain "${STRATEGY_DOMAIN}" --dispatch + drift: + if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) + needs: preflight_backtests + permissions: + contents: read + issues: write + id-token: write + uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@bda6afdab0a2dd693c35d14493176829f4da1231 + with: + strategy_domain: crypto + caller_event_name: ${{ github.event_name }} + caller_pr_head_repository: ${{ github.event.pull_request.head.repo.full_name || '' }} + snapshot_repository: QuantStrategyLab/CryptoLivePoolPipelines + snapshot_checkout_path: external/CryptoLivePoolPipelines + snapshot_repository_ref: ${{ needs.preflight_backtests.outputs.snapshot_repository_ref }} + ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }} + quant_platform_kit_ref: bda6afdab0a2dd693c35d14493176829f4da1231 + lifecycle_preflight_artifact: lifecycle-preflight-${{ github.run_id }}-${{ github.run_attempt }} + secrets: + codex_audit_service_url: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} + snapshot_repository_token: ${{ secrets.SNAPSHOT_REPOSITORY_TOKEN }} diff --git a/pyproject.toml b/pyproject.toml index 742d1e7..9634c5b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ description = "Shared crypto strategy catalog and implementations" readme = "README.md" requires-python = ">=3.11" dependencies = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@335c7a22bc3f570bd5705427ccc40172eda6b289", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@9bb8f31e898ea238a6446472f9f5e58133128d0c", ] [tool.setuptools] diff --git a/qsl.toml b/qsl.toml index 79e5d3b..58a5b82 100644 --- a/qsl.toml +++ b/qsl.toml @@ -4,5 +4,5 @@ upgrade_ring = "ring_b" [compat] bundle = "2026.07.3" requires = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@335c7a22bc3f570bd5705427ccc40172eda6b289", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@9bb8f31e898ea238a6446472f9f5e58133128d0c", ] diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 3cde4b0..c18a000 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -4,11 +4,18 @@ from __future__ import annotations import argparse +import copy +import hashlib import json +import tempfile +from dataclasses import replace from datetime import date from pathlib import Path from typing import Any +import pandas as pd +from quant_platform_kit.strategy_lifecycle.performance_metrics import compute_window_metrics + from crypto_strategies.backtest.orchestrator_runner import ( COMBO_DEFAULT_MIN_HISTORY_DAYS, DEFAULT_MIN_HISTORY_DAYS, @@ -22,6 +29,8 @@ (date(2023, 6, 1), date(2024, 5, 31)), (date(2024, 6, 1), date(2025, 5, 31)), ) +DEFAULT_STORE_ROOT = Path("/tmp/crypto_wf_store") +DRIFT_BASELINE_HORIZON_DAYS = 126 PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { PROFILE_NAME: {"min_history_days": DEFAULT_MIN_HISTORY_DAYS, "top_n": 2, "rebalance_every": 7}, @@ -45,6 +54,161 @@ def _result_payload(item: Any) -> dict[str, Any]: } +def _baseline_param_set_id( + profile: str, + params: dict[str, Any], + *, + synthetic_days: int, + windows: tuple[tuple[date, date], ...] = DEFAULT_WINDOWS, + data_fingerprint: str = "", +) -> str: + identity = { + "params": params, + "data_fingerprint": data_fingerprint or f"synthetic:{synthetic_days}", + "windows": [(start.isoformat(), end.isoformat()) for start, end in windows], + "drift_baseline_horizon_days": DRIFT_BASELINE_HORIZON_DAYS, + } + fingerprint = hashlib.sha256(json.dumps(identity, sort_keys=True, default=str).encode("utf-8")).hexdigest()[:12] + return f"{profile}_baseline_{fingerprint}" + + +def _build_runner(*, profile: str, synthetic_days: int, panel: Any = None, market_history: Any = None): + return build_backtest_runner( + profile, + panel=panel, + market_history=market_history, + synthetic_days=synthetic_days, + ) + + +def _normalize_panel(panel: pd.DataFrame) -> pd.DataFrame: + frame = pd.DataFrame(panel).copy() + if isinstance(panel.index, pd.MultiIndex) and list(panel.index.names) == ["date", "symbol"]: + frame = panel.reset_index() + required = {"date", "symbol", "in_universe", "open", "final_score"} + missing = sorted(required - set(frame.columns)) + if missing: + raise ValueError(f"research panel is missing columns: {', '.join(missing)}") + frame = frame[["date", "symbol", "in_universe", "open", "final_score"]].copy() + frame["date"] = pd.to_datetime(frame["date"], errors="coerce").dt.tz_localize(None).dt.normalize() + frame["symbol"] = frame["symbol"].astype(str).str.strip().str.upper() + frame["open"] = pd.to_numeric(frame["open"], errors="coerce") + frame["final_score"] = pd.to_numeric(frame["final_score"], errors="coerce") + frame["in_universe"] = frame["in_universe"].astype(str).str.lower().isin({"true", "1"}) + frame = frame.dropna(subset=["date", "symbol", "open"]) + if frame.duplicated(["date", "symbol"]).any(): + raise ValueError("research panel contains duplicate date/symbol rows") + return frame.set_index(["date", "symbol"]).sort_index() + + +def _normalize_market_history(market_history: pd.DataFrame) -> pd.DataFrame: + frame = pd.DataFrame(market_history).copy() + if "date" not in frame.columns and "as_of" in frame.columns: + frame = frame.rename(columns={"as_of": "date"}) + required = {"date", "symbol", "close"} + missing = sorted(required - set(frame.columns)) + if missing: + raise ValueError(f"market history is missing columns: {', '.join(missing)}") + frame = frame[["date", "symbol", "close"]].copy() + frame["date"] = pd.to_datetime(frame["date"], errors="coerce").dt.tz_localize(None).dt.normalize() + frame["symbol"] = frame["symbol"].astype(str).str.strip().str.upper() + frame["close"] = pd.to_numeric(frame["close"], errors="coerce") + frame = frame.dropna() + if frame.duplicated(["date", "symbol"]).any(): + raise ValueError("market history contains duplicate date/symbol rows") + return frame.sort_values(["date", "symbol"]) + + +def _fingerprint(*frames: pd.DataFrame) -> str: + digest = hashlib.sha256() + for frame in frames: + digest.update(pd.util.hash_pandas_object(frame, index=True).values.tobytes()) + return digest.hexdigest()[:16] + + +def _shared_inputs( + *, + windows: tuple[tuple[date, date], ...], + panel: pd.DataFrame, + market_history: pd.DataFrame, +) -> tuple[pd.DataFrame, pd.DataFrame, str]: + full_start = min(start for start, _ in windows) + full_end = max(end for _, end in windows) + normalized_panel = _normalize_panel(panel) + panel_dates = normalized_panel.index.get_level_values("date") + normalized_panel = normalized_panel.loc[ + panel_dates >= pd.Timestamp(full_start) + ] + if normalized_panel.empty or normalized_panel.index.get_level_values("date").max() < pd.Timestamp(full_end) - pd.Timedelta(days=2): + raise ValueError("research panel does not cover the latest walk-forward window") + scored_panel = normalized_panel.dropna(subset=["final_score"]) + if scored_panel.groupby(level="date")["in_universe"].sum().min() < 2: + raise ValueError("research panel requires at least two in-universe symbols") + + normalized_history = _normalize_market_history(market_history) + lookback_start = pd.Timestamp(full_start) - pd.Timedelta(days=COMBO_DEFAULT_MIN_HISTORY_DAYS + 5) + current_end = min( + normalized_panel.index.get_level_values("date").max(), + normalized_history["date"].max(), + ) + normalized_history = normalized_history.loc[ + (normalized_history["date"] >= lookback_start) + & (normalized_history["date"] <= current_end) + ].copy() + required_symbols = {"BTCUSDT", "ETHUSDT"} + missing_symbols = sorted(required_symbols - set(normalized_history["symbol"])) + if missing_symbols: + raise ValueError(f"market history is missing required symbols: {', '.join(missing_symbols)}") + reference_dates = set(normalized_history.loc[normalized_history["symbol"] == "BTCUSDT", "date"]) + for symbol in sorted(required_symbols): + symbol_dates = set(normalized_history.loc[normalized_history["symbol"] == symbol, "date"]) + if ( + len(symbol_dates & reference_dates) / len(reference_dates) < 0.99 + or min(symbol_dates) > min(reference_dates) + or max(symbol_dates) < max(reference_dates) + ): + raise ValueError(f"market history has incomplete symbol coverage: {symbol}") + if max(reference_dates) < pd.Timestamp(full_end) - pd.Timedelta(days=2): + raise ValueError("market history does not cover the latest walk-forward window") + return normalized_panel, normalized_history, _fingerprint(normalized_panel, normalized_history) + + +def _write_return_matrix( + output_path: Path, + *, + profile: str, + returns: pd.Series, + market_history: pd.DataFrame, +) -> None: + frame = returns.rename(profile).to_frame() + benchmark = _normalize_market_history(market_history) + benchmark = benchmark.loc[benchmark["symbol"] == "BTCUSDT"].set_index("date")["close"].pct_change() + frame["buy_hold_BTC"] = benchmark.reindex(frame.index) + frame.index.name = "as_of" + output_path.parent.mkdir(parents=True, exist_ok=True) + frame.reset_index().to_csv(output_path, index=False) + + +def _baseline_from_return_tail(full_result: Any, returns: pd.Series) -> Any: + tail = returns.tail(DRIFT_BASELINE_HORIZON_DAYS) + metrics = compute_window_metrics(tail, window_days=DRIFT_BASELINE_HORIZON_DAYS) + max_drawdown = float(metrics.max_drawdown) + cagr = float(metrics.cagr) + return replace( + full_result, + sharpe_ratio=float(metrics.sharpe_ratio), + calmar_ratio=float(metrics.calmar_ratio), + max_drawdown=max_drawdown, + cagr=cagr, + volatility=float(metrics.volatility), + win_rate=float(metrics.win_rate), + total_return=float(metrics.total_return), + start_date=metrics.start_date, + end_date=metrics.end_date, + observation_count=metrics.observation_count, + ) + + def run_walk_forward( *, profile: str, @@ -53,6 +217,7 @@ def run_walk_forward( store_root: Path | None = None, panel: Any = None, market_history: Any = None, + returns_output: Path | None = None, ) -> dict[str, Any]: from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore @@ -61,24 +226,80 @@ def run_walk_forward( raise ValueError(f"unsupported profile={profile!r}; supported={sorted(SUPPORTED_PROFILES)}") params = dict(PROFILE_DEFAULTS.get(profile, {"min_history_days": DEFAULT_MIN_HISTORY_DAYS})) - runner = build_backtest_runner( - profile, - panel=panel, - market_history=market_history, - synthetic_days=synthetic_days, - ) - store = PerformanceStore(local_root=store_root or Path("/tmp/crypto_wf_store")) - orchestrator = BacktestOrchestrator(store=store) - orchestrator.register_runner("crypto", runner) - - baseline = runner.run(profile, params) - wf_results = orchestrator.walk_forward( - profile, + target_root = store_root or DEFAULT_STORE_ROOT + target_root.mkdir(parents=True, exist_ok=True) + baseline_params = copy.deepcopy(params) + data_fingerprint = f"synthetic:{synthetic_days}" + shared_panel = panel + shared_market_history = market_history + if panel is not None and market_history is not None: + shared_panel, shared_market_history, data_fingerprint = _shared_inputs( + windows=windows, + panel=panel, + market_history=market_history, + ) + with tempfile.TemporaryDirectory(prefix=f"{profile}_wf_", dir=target_root) as scratch_dir: + scratch_orchestrator = BacktestOrchestrator(store=PerformanceStore(local_root=Path(scratch_dir))) + walk_forward_runner = _build_runner( + profile=profile, + panel=shared_panel, + market_history=shared_market_history, + synthetic_days=synthetic_days, + ) + scratch_orchestrator.register_runner( + "crypto", + walk_forward_runner, + ) + wf_results = scratch_orchestrator.walk_forward( + profile, + domain="crypto", + params=copy.deepcopy(params), + windows=windows, + param_set_id=f"{profile}_wf", + ) + full_window_returns = pd.concat(walk_forward_runner.run_return_history).sort_index() + if len(full_window_returns) < DRIFT_BASELINE_HORIZON_DAYS: + raise ValueError("walk-forward returns do not cover the 126-day drift baseline") + baseline_raw = _baseline_from_return_tail(wf_results[-1], full_window_returns) + orchestrator = BacktestOrchestrator(store=PerformanceStore(local_root=target_root)) + baseline = orchestrator.persist_result( + baseline_raw, + strategy_profile=profile, domain="crypto", - params=params, - windows=windows, - param_set_id=f"{profile}_wf", + params=baseline_params, + param_set_id=_baseline_param_set_id( + profile, + baseline_params, + synthetic_days=synthetic_days, + windows=windows, + data_fingerprint=data_fingerprint, + ), ) + if returns_output is not None: + if shared_market_history is None: + raise ValueError("returns_output requires market_history") + current_end = min( + shared_panel.index.get_level_values("date").max(), + shared_market_history["date"].max(), + ).date() + current_runner = _build_runner( + profile=profile, + panel=shared_panel, + market_history=shared_market_history, + synthetic_days=synthetic_days, + ) + current_runner.run( + profile, + copy.deepcopy(params), + start_date=min(start for start, _ in windows), + end_date=current_end, + ) + _write_return_matrix( + returns_output, + profile=profile, + returns=current_runner.last_daily_returns, + market_history=shared_market_history, + ) return { "strategy_profile": profile, "domain": "crypto", @@ -95,16 +316,24 @@ def main() -> int: parser.add_argument("--json-output", type=Path) parser.add_argument("--synthetic-days", type=int, default=1600) parser.add_argument("--store-root", type=Path) + parser.add_argument("--panel", type=Path) + parser.add_argument("--market-history", type=Path) + parser.add_argument("--returns-output", type=Path) args = parser.parse_args() if args.list_profiles: print(json.dumps({"profiles": sorted(SUPPORTED_PROFILES)}, indent=2)) return 0 + panel = pd.read_csv(args.panel, compression="infer") if args.panel else None + market_history = pd.read_csv(args.market_history, compression="infer") if args.market_history else None payload = run_walk_forward( profile=args.profile, synthetic_days=args.synthetic_days, store_root=args.store_root, + panel=panel, + market_history=market_history, + returns_output=args.returns_output, ) text = json.dumps(payload, indent=2, sort_keys=True, default=str) if args.json_output: diff --git a/src/crypto_strategies/backtest/orchestrator_runner.py b/src/crypto_strategies/backtest/orchestrator_runner.py index 2ae9e5a..91a83fd 100644 --- a/src/crypto_strategies/backtest/orchestrator_runner.py +++ b/src/crypto_strategies/backtest/orchestrator_runner.py @@ -9,7 +9,7 @@ import pandas as pd from crypto_strategies.backtest.combo_simulator import ComboMode, CryptoComboBacktestConfig, run_combo_backtest -from crypto_strategies.backtest.live_pool_simulator import run_live_pool_rotation_backtest +from crypto_strategies.backtest.live_pool_simulator import _performance_metrics, run_live_pool_rotation_backtest from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE try: @@ -88,6 +88,21 @@ def _slice_history( return frame.sort_values(["date", "symbol"]).reset_index(drop=True) +def _slice_daily_returns( + returns: pd.Series, + *, + start_date: date | None, + end_date: date | None, +) -> pd.Series: + sliced = returns.copy() + sliced.index = pd.to_datetime(sliced.index, utc=False).tz_localize(None).normalize() + if start_date is not None: + sliced = sliced.loc[sliced.index >= pd.Timestamp(start_date)] + if end_date is not None: + sliced = sliced.loc[sliced.index <= pd.Timestamp(end_date)] + return sliced + + def _metrics_to_result( *, strategy_profile: str, @@ -129,6 +144,16 @@ class CryptoLivePoolBacktestRunner: def __init__(self, *, panel: pd.DataFrame | None = None, synthetic_days: int = 1600) -> None: self._panel = panel self._synthetic_days = int(synthetic_days) + self._last_daily_returns = pd.Series(dtype=float) + self._run_return_history: list[pd.Series] = [] + + @property + def last_daily_returns(self) -> pd.Series: + return self._last_daily_returns.copy() + + @property + def run_return_history(self) -> tuple[pd.Series, ...]: + return tuple(item.copy() for item in self._run_return_history) def run( self, @@ -161,12 +186,18 @@ def run( top_n=int(params.get("top_n", 2)), rebalance_every=int(params.get("rebalance_every", 7)), ) + self._last_daily_returns = _slice_daily_returns( + result.returns, + start_date=start_date, + end_date=end_date, + ) + self._run_return_history.append(self._last_daily_returns.copy()) elapsed = (datetime.now(timezone.utc) - started).total_seconds() eval_dates = sliced.index.get_level_values("date") return _metrics_to_result( strategy_profile=strategy_profile, params=params, - metrics=result.metrics, + metrics=_performance_metrics(self._last_daily_returns), start_date=start_date or eval_dates.min().date(), end_date=end_date or eval_dates.max().date(), run_duration_seconds=elapsed, @@ -184,6 +215,16 @@ def __init__( ) -> None: self._market_history = market_history self._synthetic_days = int(synthetic_days) + self._last_daily_returns = pd.Series(dtype=float) + self._run_return_history: list[pd.Series] = [] + + @property + def last_daily_returns(self) -> pd.Series: + return self._last_daily_returns.copy() + + @property + def run_return_history(self) -> tuple[pd.Series, ...]: + return tuple(item.copy() for item in self._run_return_history) def run( self, @@ -225,6 +266,12 @@ def run( min_history_days=min_history_days, ), ) + self._last_daily_returns = _slice_daily_returns( + result.returns, + start_date=start_date, + end_date=end_date, + ) + self._run_return_history.append(self._last_daily_returns.copy()) elapsed = (datetime.now(timezone.utc) - started).total_seconds() eval_frame = sliced if start_date is not None: @@ -232,7 +279,7 @@ def run( return _metrics_to_result( strategy_profile=strategy_profile, params=params, - metrics=result.metrics, + metrics=_performance_metrics(self._last_daily_returns), start_date=start_date or (eval_frame["date"].min().date() if not eval_frame.empty else None), end_date=end_date or (eval_frame["date"].max().date() if not eval_frame.empty else None), run_duration_seconds=elapsed, diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py new file mode 100644 index 0000000..5c1382e --- /dev/null +++ b/tests/test_drift_workflow_config.py @@ -0,0 +1,38 @@ +from pathlib import Path + + +def test_drift_workflow_wires_real_pipeline_inputs_and_preflight_bundle() -> None: + workflow = (Path(__file__).resolve().parents[1] / ".github" / "workflows" / "drift-check.yml").read_text(encoding="utf-8") + + assert "preflight_backtests:" in workflow + assert "needs: preflight_backtests" in workflow + assert "snapshot_repository_ref: ${{ steps.snapshot-input.outputs.snapshot_repository_ref }}" in workflow + assert "id: snapshot-input" in workflow + assert 'print(run["head_sha"])' in workflow + assert "snapshot_repository_ref: ${{ needs.preflight_backtests.outputs.snapshot_repository_ref }}" in workflow + assert "Download latest trusted lifecycle inputs" in workflow + assert "gh api --paginate --slurp" in workflow + assert "trusted-snapshot-runs.json" in workflow + assert "crypto-lifecycle-inputs-" in workflow + assert '"path": ".github/workflows/publish-lifecycle-inputs.yml"' in workflow + assert '"conclusion": "success"' in workflow + assert "research_panel.csv.gz" in workflow + assert "market_history.csv.gz" in workflow + assert "repository: QuantStrategyLab/QuantPlatformKit" in workflow + assert "ref: bda6afdab0a2dd693c35d14493176829f4da1231" in workflow + assert "python -m pip install --no-deps -e external/QuantPlatformKit" in workflow + assert "scripts/run_walk_forward_backtest.py" in workflow + assert '"--list-profiles"' in workflow + assert '"--panel"' in workflow + assert '"--market-history"' in workflow + assert '"--returns-output"' in workflow + assert "Upload lifecycle preflight artifact" in workflow + assert "lifecycle-preflight-${{ github.run_id }}-${{ github.run_attempt }}" in workflow + assert workflow.count("github.ref == format('refs/heads/{0}', github.event.repository.default_branch)") == 2 + assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@bda6afdab0a2dd693c35d14493176829f4da1231" in workflow + assert "strategy_domain: crypto" in workflow + assert "snapshot_repository: QuantStrategyLab/CryptoLivePoolPipelines" in workflow + assert "snapshot_checkout_path: external/CryptoLivePoolPipelines" in workflow + assert "lifecycle_preflight_artifact: lifecycle-preflight-${{ github.run_id }}-${{ github.run_attempt }}" in workflow + assert "codex_audit_service_url: ${{ secrets.CODEX_AUDIT_SERVICE_URL }}" in workflow + assert "snapshot_repository_token: ${{ secrets.SNAPSHOT_REPOSITORY_TOKEN }}" in workflow diff --git a/tests/test_orchestrator_runner.py b/tests/test_orchestrator_runner.py index ab8e16b..05957e8 100644 --- a/tests/test_orchestrator_runner.py +++ b/tests/test_orchestrator_runner.py @@ -39,6 +39,11 @@ def test_run_returns_backtest_result(self) -> None: self.assertEqual(result.strategy_profile, PROFILE_NAME) self.assertEqual(result.domain, "crypto") self.assertGreater(result.observation_count, 0) + self.assertFalse(runner.last_daily_returns.empty) + self.assertGreaterEqual(runner.last_daily_returns.index.min().date(), date(2023, 6, 1)) + self.assertLessEqual(runner.last_daily_returns.index.max().date(), date(2024, 6, 1)) + self.assertEqual(result.observation_count, len(runner.last_daily_returns)) + self.assertEqual(len(runner.run_return_history), 1) def test_walk_forward_produces_one_result_per_window(self) -> None: from pathlib import Path @@ -74,6 +79,11 @@ def test_run_returns_backtest_result(self) -> None: self.assertEqual(result.strategy_profile, CRYPTO_EQUITY_COMBO_PROFILE) self.assertEqual(result.domain, "crypto") self.assertGreater(result.observation_count, 0) + self.assertFalse(runner.last_daily_returns.empty) + self.assertGreaterEqual(runner.last_daily_returns.index.min().date(), date(2023, 6, 1)) + self.assertLessEqual(runner.last_daily_returns.index.max().date(), date(2024, 6, 1)) + self.assertEqual(result.observation_count, len(runner.last_daily_returns)) + self.assertEqual(len(runner.run_return_history), 1) def test_invalid_combo_mode_raises(self) -> None: runner = CryptoEquityComboBacktestRunner(synthetic_days=1600) diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py new file mode 100644 index 0000000..780132d --- /dev/null +++ b/tests/test_run_walk_forward_backtest.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest +import pandas as pd + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +QPK_SRC = ROOT.parent / "QuantPlatformKit" / "src" +if str(QPK_SRC) not in sys.path: + sys.path.insert(0, str(QPK_SRC)) +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +import scripts.run_walk_forward_backtest as walk_forward +import crypto_strategies.backtest.orchestrator_runner as orchestrator_runner +from scripts.run_walk_forward_backtest import ( + _baseline_from_return_tail, + _baseline_param_set_id, + _normalize_market_history, + _normalize_panel, + run_walk_forward, +) + + +def test_run_walk_forward_persists_lifecycle_baseline(tmp_path: Path) -> None: + payload = run_walk_forward( + profile="crypto_live_pool_rotation", + synthetic_days=2200, + store_root=tmp_path, + ) + + records = [ + json.loads(path.read_text(encoding="utf-8")) + for path in (tmp_path / "backtest" / "crypto" / "crypto_live_pool_rotation").glob("*.json") + ] + + assert payload["baseline"]["sharpe_ratio"] is not None + baseline_records = [record for record in records if "_baseline_" in record["param_set_id"]] + assert baseline_records + assert baseline_records[-1]["params"] == {"min_history_days": 120, "top_n": 2, "rebalance_every": 7} + assert not any("_wf" in record["param_set_id"] for record in records) + + +def test_baseline_param_set_id_tracks_synthetic_days() -> None: + first = _baseline_param_set_id( + "crypto_live_pool_rotation", + {"min_history_days": 120, "top_n": 2, "rebalance_every": 7}, + synthetic_days=2200, + ) + second = _baseline_param_set_id( + "crypto_live_pool_rotation", + {"min_history_days": 120, "top_n": 2, "rebalance_every": 7}, + synthetic_days=2600, + ) + + assert first != second + + +def test_run_walk_forward_does_not_persist_partial_results_on_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator + + def _raise(*args, **kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(BacktestOrchestrator, "walk_forward", _raise) + with pytest.raises(RuntimeError, match="boom"): + run_walk_forward( + profile="crypto_live_pool_rotation", + synthetic_days=2200, + store_root=tmp_path, + ) + assert not list(tmp_path.rglob("*.json")) + + +def test_run_walk_forward_keeps_local_default_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(walk_forward, "DEFAULT_STORE_ROOT", tmp_path) + + run_walk_forward(profile="crypto_live_pool_rotation", synthetic_days=2200) + + assert list(tmp_path.rglob("*.json")) + + +def test_run_walk_forward_uses_real_panel_and_writes_return_matrix( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + dates = pd.date_range("2022-01-01", "2025-02-28", freq="D") + symbols = ("BTCUSDT", "ETHUSDT", "SOLUSDT") + rows = [] + market_rows = [] + for symbol_index, symbol in enumerate(symbols): + for day_index, day in enumerate(dates): + price = 100.0 + symbol_index * 10 + day_index * (0.1 + symbol_index / 100) + rows.append( + { + "date": day, + "symbol": symbol, + "in_universe": True, + "open": price, + "final_score": float((day_index + symbol_index) % 10) / 10, + } + ) + if symbol in {"BTCUSDT", "ETHUSDT"}: + market_rows.append({"date": day, "symbol": symbol, "close": price}) + panel = pd.DataFrame(rows) + market_history = pd.DataFrame(market_rows) + monkeypatch.setattr( + orchestrator_runner, + "_synthetic_panel", + lambda **kwargs: (_ for _ in ()).throw(AssertionError("synthetic panel must not be used")), + ) + returns_output = tmp_path / "returns" / "portfolio_and_tracker_returns.csv" + + payload = run_walk_forward( + profile="crypto_live_pool_rotation", + windows=( + (pd.Timestamp("2024-01-01").date(), pd.Timestamp("2024-06-30").date()), + (pd.Timestamp("2024-07-01").date(), pd.Timestamp("2024-12-31").date()), + ), + store_root=tmp_path / "store", + panel=panel, + market_history=market_history, + returns_output=returns_output, + ) + + return_matrix = pd.read_csv(returns_output) + assert payload["baseline"]["observation_count"] == 126 + assert {"as_of", "crypto_live_pool_rotation", "buy_hold_BTC"} <= set(return_matrix.columns) + assert len(return_matrix) > payload["baseline"]["observation_count"] + assert pd.Timestamp(return_matrix["as_of"].max()) > pd.Timestamp("2024-12-31") + + +def test_baseline_uses_exact_tail_of_full_return_stream() -> None: + from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult + + index = pd.date_range("2024-01-01", periods=200, freq="D") + returns = pd.Series(range(200), index=index, dtype=float) / 100000 + full_result = BacktestResult( + strategy_profile="crypto_live_pool_rotation", domain="crypto", param_set_id="", params={} + ) + + baseline = _baseline_from_return_tail(full_result, returns) + + expected = returns.tail(126) + assert baseline.start_date == expected.index.min().date() + assert baseline.end_date == expected.index.max().date() + assert baseline.observation_count == len(expected) + + +def test_external_inputs_reject_duplicate_keys() -> None: + duplicate_panel = pd.DataFrame( + [ + {"date": "2024-01-01", "symbol": "BTCUSDT", "in_universe": True, "open": 1, "final_score": 1}, + {"date": "2024-01-01", "symbol": "BTCUSDT", "in_universe": True, "open": 2, "final_score": 2}, + ] + ) + duplicate_history = pd.DataFrame( + [ + {"date": "2024-01-01", "symbol": "BTCUSDT", "close": 1}, + {"date": "2024-01-01", "symbol": "BTCUSDT", "close": 2}, + ] + ) + + with pytest.raises(ValueError, match="research panel contains duplicate"): + _normalize_panel(duplicate_panel) + with pytest.raises(ValueError, match="market history contains duplicate"): + _normalize_market_history(duplicate_history) + + +def test_normalized_panel_preserves_unscored_open_rows() -> None: + panel = pd.DataFrame( + [ + { + "date": "2024-01-01", + "symbol": "BTCUSDT", + "in_universe": False, + "open": 100.0, + "final_score": None, + } + ] + ) + + normalized = _normalize_panel(panel) + + assert len(normalized) == 1 + assert pd.isna(normalized.iloc[0]["final_score"]) diff --git a/uv.lock b/uv.lock index ec1c485..7eaef84 100644 --- a/uv.lock +++ b/uv.lock @@ -11,9 +11,9 @@ dependencies = [ ] [package.metadata] -requires-dist = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=335c7a22bc3f570bd5705427ccc40172eda6b289" }] +requires-dist = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=9bb8f31e898ea238a6446472f9f5e58133128d0c" }] [[package]] name = "quant-platform-kit" version = "0.10.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=335c7a22bc3f570bd5705427ccc40172eda6b289#335c7a22bc3f570bd5705427ccc40172eda6b289" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=9bb8f31e898ea238a6446472f9f5e58133128d0c#9bb8f31e898ea238a6446472f9f5e58133128d0c" }