From 4044ec723a62b386c98c6d47ce7e82989d327336 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 02:53:08 +0800 Subject: [PATCH 01/16] fix(ci): align drift baseline store and workflow pin Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 73 ++++--------------------- scripts/run_walk_forward_backtest.py | 73 ++++++++++++++++++++++--- tests/test_drift_workflow_config.py | 15 +++++ tests/test_run_walk_forward_backtest.py | 35 ++++++++++++ 4 files changed, 127 insertions(+), 69 deletions(-) create mode 100644 tests/test_drift_workflow_config.py create mode 100644 tests/test_run_walk_forward_backtest.py diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 459fb53..3125697 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -14,65 +14,14 @@ permissions: jobs: drift: - runs-on: ubuntu-latest - timeout-minutes: 15 - env: - FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" - STRATEGY_DOMAIN: crypto - - steps: - - name: Checkout - uses: actions/checkout@v6 - - - name: Checkout QuantPlatformKit - uses: actions/checkout@v6 - with: - repository: QuantStrategyLab/QuantPlatformKit - ref: main - path: external/QuantPlatformKit - - - name: Set up Python - uses: actions/setup-python@v6 - with: - python-version: "3.11" - - - name: Install dependencies - run: | - set -euo pipefail - python -m pip install --upgrade pip - 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: Sync drift alerts to GitHub Issues - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - GITHUB_REPOSITORY: ${{ github.repository }} - run: python scripts/run_drift_github_issues.py - - - name: Checkout AIAuditBridge - uses: actions/checkout@v6 - with: - repository: QuantStrategyLab/AIAuditBridge - ref: main - path: external/AIAuditBridge - - - 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 + uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289 + 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 + ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }} + secrets: + codex_audit_service_url: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} + snapshot_repository_token: ${{ secrets.SNAPSHOT_REPOSITORY_TOKEN }} diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 3cde4b0..f91e033 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -4,7 +4,10 @@ from __future__ import annotations import argparse +import copy +import hashlib import json +import re from datetime import date from pathlib import Path from typing import Any @@ -45,6 +48,40 @@ def _result_payload(item: Any) -> dict[str, Any]: } +def _baseline_param_set_id(profile: str, params: dict[str, Any]) -> str: + fingerprint = hashlib.sha256(json.dumps(params, sort_keys=True, default=str).encode("utf-8")).hexdigest()[:12] + return f"{profile}_baseline_{fingerprint}" + + +def _current_qpk_pin() -> str: + text = (Path(__file__).resolve().parents[1] / "qsl.toml").read_text(encoding="utf-8") + match = re.search(r"QuantPlatformKit\.git@([0-9a-f]{40})", text) + return match.group(1) if match else "unknown" + + +def _baseline_identity_params( + params: dict[str, Any], + *, + synthetic_days: int, + baseline_result: Any, +) -> dict[str, Any]: + identity = copy.deepcopy(params) + identity["_baseline_start_date"] = baseline_result.start_date.isoformat() if baseline_result.start_date else None + identity["_baseline_end_date"] = baseline_result.end_date.isoformat() if baseline_result.end_date else None + identity["_qpk_pin"] = _current_qpk_pin() + identity["_synthetic_days"] = synthetic_days + return identity + + +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 run_walk_forward( *, profile: str, @@ -61,21 +98,43 @@ 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, + store = PerformanceStore(local_root=store_root) if store_root is not None else PerformanceStore.from_env() + orchestrator = BacktestOrchestrator(store=store) + + baseline_params = copy.deepcopy(params) + runner = _build_runner( + profile=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) + baseline_probe = orchestrator.run( + profile, + domain="crypto", + params=copy.deepcopy(baseline_params), + param_set_id="__discarded__", + start_date=None, + end_date=None, + ) + baseline_store_params = _baseline_identity_params( + baseline_params, + synthetic_days=synthetic_days, + baseline_result=baseline_probe, + ) + baseline = orchestrator.run( + profile, + domain="crypto", + params=baseline_store_params, + param_set_id=_baseline_param_set_id(profile, baseline_store_params), + start_date=None, + end_date=None, + ) + wf_params = copy.deepcopy(params) wf_results = orchestrator.walk_forward( profile, domain="crypto", - params=params, + params=wf_params, windows=windows, param_set_id=f"{profile}_wf", ) diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py new file mode 100644 index 0000000..ca7e0aa --- /dev/null +++ b/tests/test_drift_workflow_config.py @@ -0,0 +1,15 @@ +from pathlib import Path + + +def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: + workflow = (Path(__file__).resolve().parents[1] / ".github" / "workflows" / "drift-check.yml").read_text(encoding="utf-8") + + assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289" in workflow + assert "strategy_domain: crypto" in workflow + assert "caller_event_name: ${{ github.event_name }}" in workflow + assert "caller_pr_head_repository: ${{ github.event.pull_request.head.repo.full_name || '' }}" in workflow + assert "snapshot_repository: QuantStrategyLab/CryptoLivePoolPipelines" in workflow + assert "snapshot_checkout_path: external/CryptoLivePoolPipelines" in workflow + assert "ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }}" 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_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py new file mode 100644 index 0000000..e7e7f2e --- /dev/null +++ b/tests/test_run_walk_forward_backtest.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path + +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)) + +from scripts.run_walk_forward_backtest import 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"]["_qpk_pin"] + assert baseline_records[-1]["params"]["_baseline_end_date"] + assert any("_wf" in record["param_set_id"] for record in records) From 9a700c8f6e6e7644a578c3feac8fab9b60b3122f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 03:02:01 +0800 Subject: [PATCH 02/16] fix(ci): persist validated lifecycle baselines Co-Authored-By: Codex --- scripts/run_walk_forward_backtest.py | 61 ++++++++----------------- tests/test_run_walk_forward_backtest.py | 38 +++++++++++++-- 2 files changed, 55 insertions(+), 44 deletions(-) diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index f91e033..7005210 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -7,7 +7,6 @@ import copy import hashlib import json -import re from datetime import date from pathlib import Path from typing import Any @@ -48,31 +47,15 @@ def _result_payload(item: Any) -> dict[str, Any]: } -def _baseline_param_set_id(profile: str, params: dict[str, Any]) -> str: - fingerprint = hashlib.sha256(json.dumps(params, sort_keys=True, default=str).encode("utf-8")).hexdigest()[:12] +def _baseline_param_set_id(profile: str, params: dict[str, Any], *, synthetic_days: int) -> str: + identity = { + "params": params, + "synthetic_days": synthetic_days, + } + fingerprint = hashlib.sha256(json.dumps(identity, sort_keys=True, default=str).encode("utf-8")).hexdigest()[:12] return f"{profile}_baseline_{fingerprint}" -def _current_qpk_pin() -> str: - text = (Path(__file__).resolve().parents[1] / "qsl.toml").read_text(encoding="utf-8") - match = re.search(r"QuantPlatformKit\.git@([0-9a-f]{40})", text) - return match.group(1) if match else "unknown" - - -def _baseline_identity_params( - params: dict[str, Any], - *, - synthetic_days: int, - baseline_result: Any, -) -> dict[str, Any]: - identity = copy.deepcopy(params) - identity["_baseline_start_date"] = baseline_result.start_date.isoformat() if baseline_result.start_date else None - identity["_baseline_end_date"] = baseline_result.end_date.isoformat() if baseline_result.end_date else None - identity["_qpk_pin"] = _current_qpk_pin() - identity["_synthetic_days"] = synthetic_days - return identity - - def _build_runner(*, profile: str, synthetic_days: int, panel: Any = None, market_history: Any = None): return build_backtest_runner( profile, @@ -98,7 +81,7 @@ 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})) - store = PerformanceStore(local_root=store_root) if store_root is not None else PerformanceStore.from_env() + store = PerformanceStore(local_root=store_root or Path("/tmp/crypto_wf_store")) orchestrator = BacktestOrchestrator(store=store) baseline_params = copy.deepcopy(params) @@ -109,24 +92,9 @@ def run_walk_forward( synthetic_days=synthetic_days, ) orchestrator.register_runner("crypto", runner) - baseline_probe = orchestrator.run( + baseline_raw = runner.run( profile, - domain="crypto", - params=copy.deepcopy(baseline_params), - param_set_id="__discarded__", - start_date=None, - end_date=None, - ) - baseline_store_params = _baseline_identity_params( - baseline_params, - synthetic_days=synthetic_days, - baseline_result=baseline_probe, - ) - baseline = orchestrator.run( - profile, - domain="crypto", - params=baseline_store_params, - param_set_id=_baseline_param_set_id(profile, baseline_store_params), + copy.deepcopy(baseline_params), start_date=None, end_date=None, ) @@ -138,6 +106,17 @@ def run_walk_forward( windows=windows, param_set_id=f"{profile}_wf", ) + baseline = orchestrator.persist_result( + baseline_raw, + strategy_profile=profile, + domain="crypto", + params=baseline_params, + param_set_id=_baseline_param_set_id( + profile, + baseline_params, + synthetic_days=synthetic_days, + ), + ) return { "strategy_profile": profile, "domain": "crypto", diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py index e7e7f2e..c26fdb8 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -4,6 +4,8 @@ import sys from pathlib import Path +import pytest + ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" QPK_SRC = ROOT.parent / "QuantPlatformKit" / "src" @@ -12,7 +14,7 @@ if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) -from scripts.run_walk_forward_backtest import run_walk_forward +from scripts.run_walk_forward_backtest import _baseline_param_set_id, run_walk_forward def test_run_walk_forward_persists_lifecycle_baseline(tmp_path: Path) -> None: @@ -30,6 +32,36 @@ def test_run_walk_forward_persists_lifecycle_baseline(tmp_path: Path) -> None: 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"]["_qpk_pin"] - assert baseline_records[-1]["params"]["_baseline_end_date"] + assert baseline_records[-1]["params"] == {"min_history_days": 120, "top_n": 2, "rebalance_every": 7} assert 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")) From 10ed16a3243f613f1fd590c3c1b64808dbeb85eb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 03:11:33 +0800 Subject: [PATCH 03/16] fix(ci): seed lifecycle backtests before drift checks Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 48 ++++++++++++++++++++++++++++ scripts/run_walk_forward_backtest.py | 22 +++++++++---- tests/test_drift_workflow_config.py | 5 +++ 3 files changed, 69 insertions(+), 6 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 3125697..0de0568 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -13,7 +13,55 @@ permissions: id-token: write jobs: + preflight_backtests: + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" + LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }} + LIFECYCLE_LOCAL_ROOT: ${{ github.workspace }}/data/lifecycle_store + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + set -euo pipefail + python -m pip install --upgrade pip + python -m pip install -e . pandas + + - name: Persist lifecycle backtests + run: | + python - <<'PY' + import json + import subprocess + + profiles = json.loads( + subprocess.check_output( + ["python", "scripts/run_walk_forward_backtest.py", "--list-profiles"], + text=True, + ) + )["profiles"] + for profile in profiles: + subprocess.check_call( + [ + "python", + "scripts/run_walk_forward_backtest.py", + "--profile", + profile, + "--store-root", + "data/lifecycle_store", + ] + ) + PY + drift: + needs: preflight_backtests uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289 with: strategy_domain: crypto diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 7005210..a4af537 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -65,6 +65,21 @@ def _build_runner(*, profile: str, synthetic_days: int, panel: Any = None, marke ) +def _run_baseline(runner: Any, profile: str, params: dict[str, Any]) -> Any: + try: + return runner.run( + profile, + params, + start_date=None, + end_date=None, + ) + except TypeError as exc: + message = str(exc) + if "start_date" not in message and "end_date" not in message: + raise + return runner.run(profile, params) + + def run_walk_forward( *, profile: str, @@ -92,12 +107,7 @@ def run_walk_forward( synthetic_days=synthetic_days, ) orchestrator.register_runner("crypto", runner) - baseline_raw = runner.run( - profile, - copy.deepcopy(baseline_params), - start_date=None, - end_date=None, - ) + baseline_raw = _run_baseline(runner, profile, copy.deepcopy(baseline_params)) wf_params = copy.deepcopy(params) wf_results = orchestrator.walk_forward( profile, diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index ca7e0aa..bcf3344 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -4,6 +4,11 @@ def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> 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 "scripts/run_walk_forward_backtest.py" in workflow + assert '"--list-profiles"' in workflow + assert '"data/lifecycle_store"' in workflow assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289" in workflow assert "strategy_domain: crypto" in workflow assert "caller_event_name: ${{ github.event_name }}" in workflow From 1be0b372cff6795ce92c2916a5f95c1beb4d2641 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 03:41:13 +0800 Subject: [PATCH 04/16] fix(ci): share drift preflight through lifecycle bucket Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 19 ++++++++++++++++--- scripts/run_walk_forward_backtest.py | 2 +- tests/test_drift_workflow_config.py | 6 +++++- 3 files changed, 22 insertions(+), 5 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 0de0568..53c99e2 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -19,11 +19,25 @@ jobs: env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }} - LIFECYCLE_LOCAL_ROOT: ${{ github.workspace }}/data/lifecycle_store steps: - name: Checkout uses: actions/checkout@v6 + - name: Require lifecycle bucket + run: | + set -euo pipefail + if [ -z "${LIFECYCLE_PERFORMANCE_BUCKET:-}" ]; then + echo "::error::LIFECYCLE_PERFORMANCE_BUCKET must be configured for drift preflight" + exit 1 + fi + + - name: Checkout QuantPlatformKit + uses: actions/checkout@v6 + with: + repository: QuantStrategyLab/QuantPlatformKit + ref: 335c7a22bc3f570bd5705427ccc40172eda6b289 + path: external/QuantPlatformKit + - name: Set up Python uses: actions/setup-python@v6 with: @@ -34,6 +48,7 @@ jobs: set -euo pipefail python -m pip install --upgrade pip python -m pip install -e . pandas + python -m pip install --no-deps -e external/QuantPlatformKit - name: Persist lifecycle backtests run: | @@ -54,8 +69,6 @@ jobs: "scripts/run_walk_forward_backtest.py", "--profile", profile, - "--store-root", - "data/lifecycle_store", ] ) PY diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index a4af537..0d80122 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -96,7 +96,7 @@ 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})) - store = PerformanceStore(local_root=store_root or Path("/tmp/crypto_wf_store")) + store = PerformanceStore(local_root=store_root) if store_root is not None else PerformanceStore.from_env() orchestrator = BacktestOrchestrator(store=store) baseline_params = copy.deepcopy(params) diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index bcf3344..f709c39 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -6,9 +6,13 @@ def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: assert "preflight_backtests:" in workflow assert "needs: preflight_backtests" in workflow + assert "Require lifecycle bucket" in workflow + assert "LIFECYCLE_PERFORMANCE_BUCKET must be configured for drift preflight" in workflow + assert "repository: QuantStrategyLab/QuantPlatformKit" in workflow + assert "ref: 335c7a22bc3f570bd5705427ccc40172eda6b289" 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 '"data/lifecycle_store"' in workflow assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289" in workflow assert "strategy_domain: crypto" in workflow assert "caller_event_name: ${{ github.event_name }}" in workflow From 9137f8011f9cf7222e5cee974d470f2b9ecfd6f3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:00:55 +0800 Subject: [PATCH 05/16] fix(ci): harden lifecycle preflight persistence Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 49 ++++++++++++++++++++++++- scripts/run_walk_forward_backtest.py | 3 +- tests/test_drift_workflow_config.py | 5 +++ tests/test_run_walk_forward_backtest.py | 9 +++++ 4 files changed, 63 insertions(+), 3 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 53c99e2..e22e26c 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -10,12 +10,17 @@ on: permissions: contents: read issues: write - id-token: write jobs: preflight_backtests: + if: >- + github.event_name != 'pull_request' || + github.event.pull_request.head.repo.full_name == github.repository runs-on: ubuntu-latest timeout-minutes: 15 + permissions: + contents: read + id-token: write env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }} @@ -50,11 +55,15 @@ jobs: python -m pip install -e . pandas python -m pip install --no-deps -e external/QuantPlatformKit - - name: Persist lifecycle backtests + - name: Build lifecycle backtests in staging + env: + LIFECYCLE_PREFLIGHT_STAGING_ROOT: ${{ runner.temp }}/lifecycle_preflight run: | python - <<'PY' import json + import os import subprocess + from pathlib import Path profiles = json.loads( subprocess.check_output( @@ -62,6 +71,7 @@ jobs: text=True, ) )["profiles"] + staging_root = Path(os.environ["LIFECYCLE_PREFLIGHT_STAGING_ROOT"]) for profile in profiles: subprocess.check_call( [ @@ -69,12 +79,47 @@ jobs: "scripts/run_walk_forward_backtest.py", "--profile", profile, + "--store-root", + str(staging_root), ] ) PY + - name: Promote staged lifecycle backtests + env: + LIFECYCLE_PREFLIGHT_STAGING_ROOT: ${{ runner.temp }}/lifecycle_preflight + run: | + python - <<'PY' + import json + import os + import subprocess + from pathlib import Path + + from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore + + domain = "crypto" + staging_root = Path(os.environ["LIFECYCLE_PREFLIGHT_STAGING_ROOT"]) + source = PerformanceStore(local_root=staging_root) + target = PerformanceStore.from_env() + profiles = json.loads( + subprocess.check_output( + ["python", "scripts/run_walk_forward_backtest.py", "--list-profiles"], + text=True, + ) + )["profiles"] + results = [source.load_latest_backtest(domain, profile) for profile in profiles] + if any(result is None for result in results): + raise RuntimeError("staged lifecycle backtests are incomplete") + for result in results: + target.save_backtest_result(result) + PY + drift: needs: preflight_backtests + permissions: + contents: read + issues: write + id-token: write uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289 with: strategy_domain: crypto diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 0d80122..9313783 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -24,6 +24,7 @@ (date(2023, 6, 1), date(2024, 5, 31)), (date(2024, 6, 1), date(2025, 5, 31)), ) +DEFAULT_STORE_ROOT = Path("/tmp/crypto_wf_store") PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { PROFILE_NAME: {"min_history_days": DEFAULT_MIN_HISTORY_DAYS, "top_n": 2, "rebalance_every": 7}, @@ -96,7 +97,7 @@ 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})) - store = PerformanceStore(local_root=store_root) if store_root is not None else PerformanceStore.from_env() + store = PerformanceStore(local_root=store_root or DEFAULT_STORE_ROOT) orchestrator = BacktestOrchestrator(store=store) baseline_params = copy.deepcopy(params) diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index f709c39..bd30252 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -13,6 +13,11 @@ def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: 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 '"--store-root"' in workflow + assert "LIFECYCLE_PREFLIGHT_STAGING_ROOT" in workflow + assert "Promote staged lifecycle backtests" in workflow + assert "head.repo.full_name == github.repository" in workflow + assert "id-token: write" in workflow assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289" in workflow assert "strategy_domain: crypto" in workflow assert "caller_event_name: ${{ github.event_name }}" in workflow diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py index c26fdb8..48fbf2c 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -14,6 +14,7 @@ if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) +import scripts.run_walk_forward_backtest as walk_forward from scripts.run_walk_forward_backtest import _baseline_param_set_id, run_walk_forward @@ -65,3 +66,11 @@ def _raise(*args, **kwargs): 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")) From d88039507a38b0fd4bb335fcaa7524c9cbcca414 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:03:04 +0800 Subject: [PATCH 06/16] chore(ci): pin caller lifecycle drift workflow Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 3 ++- tests/test_drift_workflow_config.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index e22e26c..cbffdda 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -120,7 +120,7 @@ jobs: contents: read issues: write id-token: write - uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289 + uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@17278db4e7aef0007346d853eb308b6c1bd8c859 with: strategy_domain: crypto caller_event_name: ${{ github.event_name }} @@ -128,6 +128,7 @@ jobs: snapshot_repository: QuantStrategyLab/CryptoLivePoolPipelines snapshot_checkout_path: external/CryptoLivePoolPipelines ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }} + lifecycle_performance_bucket: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET }} secrets: codex_audit_service_url: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} snapshot_repository_token: ${{ secrets.SNAPSHOT_REPOSITORY_TOKEN }} diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index bd30252..3582633 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -18,7 +18,7 @@ def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: assert "Promote staged lifecycle backtests" in workflow assert "head.repo.full_name == github.repository" in workflow assert "id-token: write" in workflow - assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@335c7a22bc3f570bd5705427ccc40172eda6b289" in workflow + assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@17278db4e7aef0007346d853eb308b6c1bd8c859" in workflow assert "strategy_domain: crypto" in workflow assert "caller_event_name: ${{ github.event_name }}" in workflow assert "caller_pr_head_repository: ${{ github.event.pull_request.head.repo.full_name || '' }}" in workflow From ae524989d27a56588e00096ec1ad98beb8897ca5 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:05:59 +0800 Subject: [PATCH 07/16] fix(ci): isolate drift baseline promotion Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 2 ++ tests/test_drift_workflow_config.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index cbffdda..bd5c4d0 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -86,6 +86,7 @@ jobs: PY - name: Promote staged lifecycle backtests + if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) env: LIFECYCLE_PREFLIGHT_STAGING_ROOT: ${{ runner.temp }}/lifecycle_preflight run: | @@ -115,6 +116,7 @@ jobs: PY drift: + if: ${{ always() && (needs.preflight_backtests.result == 'success' || needs.preflight_backtests.result == 'skipped') }} needs: preflight_backtests permissions: contents: read diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index 3582633..fc452a6 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -16,6 +16,8 @@ def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: assert '"--store-root"' in workflow assert "LIFECYCLE_PREFLIGHT_STAGING_ROOT" in workflow assert "Promote staged lifecycle backtests" in workflow + assert "github.ref == format('refs/heads/{0}', github.event.repository.default_branch)" in workflow + assert "needs.preflight_backtests.result == 'skipped'" in workflow assert "head.repo.full_name == github.repository" in workflow assert "id-token: write" in workflow assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@17278db4e7aef0007346d853eb308b6c1bd8c859" in workflow From 57baad81be323ee043a618a5484e940db8dcb812 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:08:53 +0800 Subject: [PATCH 08/16] fix: isolate crypto walk-forward lifecycle records Co-Authored-By: Codex --- scripts/run_walk_forward_backtest.py | 26 +++++++++++++------------ tests/test_run_walk_forward_backtest.py | 2 +- 2 files changed, 15 insertions(+), 13 deletions(-) diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 9313783..7ee440a 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -7,6 +7,7 @@ import copy import hashlib import json +import tempfile from datetime import date from pathlib import Path from typing import Any @@ -97,9 +98,8 @@ 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})) - store = PerformanceStore(local_root=store_root or DEFAULT_STORE_ROOT) - orchestrator = BacktestOrchestrator(store=store) - + target_root = store_root or DEFAULT_STORE_ROOT + target_root.mkdir(parents=True, exist_ok=True) baseline_params = copy.deepcopy(params) runner = _build_runner( profile=profile, @@ -107,16 +107,18 @@ def run_walk_forward( market_history=market_history, synthetic_days=synthetic_days, ) - orchestrator.register_runner("crypto", runner) baseline_raw = _run_baseline(runner, profile, copy.deepcopy(baseline_params)) - wf_params = copy.deepcopy(params) - wf_results = orchestrator.walk_forward( - profile, - domain="crypto", - params=wf_params, - windows=windows, - param_set_id=f"{profile}_wf", - ) + with tempfile.TemporaryDirectory(prefix=f"{profile}_wf_", dir=target_root) as scratch_dir: + scratch_orchestrator = BacktestOrchestrator(store=PerformanceStore(local_root=Path(scratch_dir))) + scratch_orchestrator.register_runner("crypto", runner) + wf_results = scratch_orchestrator.walk_forward( + profile, + domain="crypto", + params=copy.deepcopy(params), + windows=windows, + param_set_id=f"{profile}_wf", + ) + orchestrator = BacktestOrchestrator(store=PerformanceStore(local_root=target_root)) baseline = orchestrator.persist_result( baseline_raw, strategy_profile=profile, diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py index 48fbf2c..1299a1f 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -34,7 +34,7 @@ def test_run_walk_forward_persists_lifecycle_baseline(tmp_path: Path) -> 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 any("_wf" in record["param_set_id"] for record in records) + assert not any("_wf" in record["param_set_id"] for record in records) def test_baseline_param_set_id_tracks_synthetic_days() -> None: From da3de79956460baadc2362f5651a5754dc8b776c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:09:10 +0800 Subject: [PATCH 09/16] fix(ci): remove OIDC from preflight job Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 1 - tests/test_drift_workflow_config.py | 2 ++ 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index bd5c4d0..1286741 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -20,7 +20,6 @@ jobs: timeout-minutes: 15 permissions: contents: read - id-token: write env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }} diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index fc452a6..45dec69 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -20,6 +20,8 @@ def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: assert "needs.preflight_backtests.result == 'skipped'" in workflow assert "head.repo.full_name == github.repository" in workflow assert "id-token: write" in workflow + preflight = workflow.split(" drift:", maxsplit=1)[0] + assert "id-token: write" not in preflight assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@17278db4e7aef0007346d853eb308b6c1bd8c859" in workflow assert "strategy_domain: crypto" in workflow assert "caller_event_name: ${{ github.event_name }}" in workflow From cddce269e535322bc52eb582b132580932c27bda Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:11:56 +0800 Subject: [PATCH 10/16] fix(ci): gate drift on successful preflight Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 2 +- tests/test_drift_workflow_config.py | 3 --- 2 files changed, 1 insertion(+), 4 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 1286741..294e222 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -20,6 +20,7 @@ jobs: timeout-minutes: 15 permissions: contents: read + id-token: write env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }} @@ -115,7 +116,6 @@ jobs: PY drift: - if: ${{ always() && (needs.preflight_backtests.result == 'success' || needs.preflight_backtests.result == 'skipped') }} needs: preflight_backtests permissions: contents: read diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index 45dec69..bf27b9e 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -17,11 +17,8 @@ def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: assert "LIFECYCLE_PREFLIGHT_STAGING_ROOT" in workflow assert "Promote staged lifecycle backtests" in workflow assert "github.ref == format('refs/heads/{0}', github.event.repository.default_branch)" in workflow - assert "needs.preflight_backtests.result == 'skipped'" in workflow assert "head.repo.full_name == github.repository" in workflow assert "id-token: write" in workflow - preflight = workflow.split(" drift:", maxsplit=1)[0] - assert "id-token: write" not in preflight assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@17278db4e7aef0007346d853eb308b6c1bd8c859" in workflow assert "strategy_domain: crypto" in workflow assert "caller_event_name: ${{ github.event_name }}" in workflow From 7135cf571439a0f3292e38e2e44cd33300f172cf Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:49:50 +0800 Subject: [PATCH 11/16] fix: run crypto drift preflight on trusted inputs Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 154 +++++++++----- pyproject.toml | 2 +- scripts/run_walk_forward_backtest.py | 195 ++++++++++++++++-- .../backtest/orchestrator_runner.py | 41 +++- tests/test_drift_workflow_config.py | 31 +-- tests/test_orchestrator_runner.py | 8 + tests/test_run_walk_forward_backtest.py | 51 +++++ uv.lock | 4 +- 8 files changed, 392 insertions(+), 94 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 294e222..2e68267 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -13,34 +13,18 @@ permissions: jobs: preflight_backtests: - if: >- - github.event_name != 'pull_request' || - github.event.pull_request.head.repo.full_name == github.repository + if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) runs-on: ubuntu-latest - timeout-minutes: 15 - permissions: - contents: read - id-token: write - env: - FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" - LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }} + timeout-minutes: 30 steps: - name: Checkout uses: actions/checkout@v6 - - name: Require lifecycle bucket - run: | - set -euo pipefail - if [ -z "${LIFECYCLE_PERFORMANCE_BUCKET:-}" ]; then - echo "::error::LIFECYCLE_PERFORMANCE_BUCKET must be configured for drift preflight" - exit 1 - fi - - name: Checkout QuantPlatformKit uses: actions/checkout@v6 with: repository: QuantStrategyLab/QuantPlatformKit - ref: 335c7a22bc3f570bd5705427ccc40172eda6b289 + ref: 9bb8f31e898ea238a6446472f9f5e58133128d0c path: external/QuantPlatformKit - name: Set up Python @@ -55,10 +39,76 @@ jobs: python -m pip install -e . pandas python -m pip install --no-deps -e external/QuantPlatformKit - - name: Build lifecycle backtests in staging + - name: Download latest trusted lifecycle inputs + 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" + 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()) + artifacts = [item for page in pages for item in page.get("artifacts", [])] + candidates = [ + 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("head_branch") == "main" + ] + if not candidates: + raise SystemExit("no trusted crypto lifecycle input artifact is available") + selected = max(candidates, key=lambda item: item["created_at"]) + 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 + 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: Build lifecycle preflight bundle env: - LIFECYCLE_PREFLIGHT_STAGING_ROOT: ${{ runner.temp }}/lifecycle_preflight + 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 @@ -71,57 +121,52 @@ jobs: text=True, ) )["profiles"] - staging_root = Path(os.environ["LIFECYCLE_PREFLIGHT_STAGING_ROOT"]) + 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(staging_root), + str(store_root), + "--returns-output", + str(returns_output), ] ) PY - - name: Promote staged lifecycle backtests - if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) - env: - LIFECYCLE_PREFLIGHT_STAGING_ROOT: ${{ runner.temp }}/lifecycle_preflight - run: | - python - <<'PY' - import json - import os - import subprocess - from pathlib import Path - - from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore - - domain = "crypto" - staging_root = Path(os.environ["LIFECYCLE_PREFLIGHT_STAGING_ROOT"]) - source = PerformanceStore(local_root=staging_root) - target = PerformanceStore.from_env() - profiles = json.loads( - subprocess.check_output( - ["python", "scripts/run_walk_forward_backtest.py", "--list-profiles"], - text=True, - ) - )["profiles"] - results = [source.load_latest_backtest(domain, profile) for profile in profiles] - if any(result is None for result in results): - raise RuntimeError("staged lifecycle backtests are incomplete") - for result in results: - target.save_backtest_result(result) - PY + - name: Upload lifecycle preflight artifact + uses: actions/upload-artifact@v4 + with: + name: lifecycle-preflight-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/lifecycle-preflight-bundle + if-no-files-found: error 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@17278db4e7aef0007346d853eb308b6c1bd8c859 + uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@9bb8f31e898ea238a6446472f9f5e58133128d0c with: strategy_domain: crypto caller_event_name: ${{ github.event_name }} @@ -129,7 +174,8 @@ jobs: snapshot_repository: QuantStrategyLab/CryptoLivePoolPipelines snapshot_checkout_path: external/CryptoLivePoolPipelines ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }} - lifecycle_performance_bucket: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET }} + quant_platform_kit_ref: 9bb8f31e898ea238a6446472f9f5e58133128d0c + 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/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 7ee440a..75234f1 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -12,6 +12,8 @@ from pathlib import Path from typing import Any +import pandas as pd + from crypto_strategies.backtest.orchestrator_runner import ( COMBO_DEFAULT_MIN_HISTORY_DAYS, DEFAULT_MIN_HISTORY_DAYS, @@ -26,6 +28,7 @@ (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}, @@ -49,10 +52,19 @@ def _result_payload(item: Any) -> dict[str, Any]: } -def _baseline_param_set_id(profile: str, params: dict[str, Any], *, synthetic_days: int) -> str: +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, - "synthetic_days": synthetic_days, + "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}" @@ -67,19 +79,102 @@ def _build_runner(*, profile: str, synthetic_days: int, panel: Any = None, marke ) -def _run_baseline(runner: Any, profile: str, params: dict[str, Any]) -> Any: - try: - return runner.run( - profile, - params, - start_date=None, - end_date=None, - ) - except TypeError as exc: - message = str(exc) - if "start_date" not in message and "end_date" not in message: - raise - return runner.run(profile, params) +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", "final_score"]) + return frame.drop_duplicates(["date", "symbol"], keep="last").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") + return frame.dropna().drop_duplicates(["date", "symbol"], keep="last").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)) & (panel_dates <= pd.Timestamp(full_end)) + ] + 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") + if normalized_panel.groupby(level="date")["in_universe"].sum().max() < 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) + normalized_history = normalized_history.loc[ + (normalized_history["date"] >= lookback_start) + & (normalized_history["date"] <= pd.Timestamp(full_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 run_walk_forward( @@ -90,6 +185,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 @@ -101,16 +197,56 @@ def run_walk_forward( target_root = store_root or DEFAULT_STORE_ROOT target_root.mkdir(parents=True, exist_ok=True) baseline_params = copy.deepcopy(params) - runner = _build_runner( + 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, + ) + return_matrix_runner = _build_runner( profile=profile, - panel=panel, - market_history=market_history, + panel=shared_panel, + market_history=shared_market_history, synthetic_days=synthetic_days, ) - baseline_raw = _run_baseline(runner, profile, copy.deepcopy(baseline_params)) + full_start = min(start for start, _ in windows) + baseline_end = max(end for _, end in windows) + return_matrix_runner.run( + profile, + copy.deepcopy(baseline_params), + start_date=full_start, + end_date=baseline_end, + ) + full_window_returns = return_matrix_runner.last_daily_returns + if len(full_window_returns) < DRIFT_BASELINE_HORIZON_DAYS: + raise ValueError("full-window returns do not cover the 126-day drift baseline") + baseline_start = full_window_returns.index[-DRIFT_BASELINE_HORIZON_DAYS].date() + baseline_runner = _build_runner( + profile=profile, + panel=shared_panel, + market_history=shared_market_history, + synthetic_days=synthetic_days, + ) + baseline_raw = baseline_runner.run( + profile, + copy.deepcopy(baseline_params), + start_date=baseline_start, + end_date=baseline_end, + ) with tempfile.TemporaryDirectory(prefix=f"{profile}_wf_", dir=target_root) as scratch_dir: scratch_orchestrator = BacktestOrchestrator(store=PerformanceStore(local_root=Path(scratch_dir))) - scratch_orchestrator.register_runner("crypto", runner) + scratch_orchestrator.register_runner( + "crypto", + _build_runner( + profile=profile, + panel=shared_panel, + market_history=shared_market_history, + synthetic_days=synthetic_days, + ), + ) wf_results = scratch_orchestrator.walk_forward( profile, domain="crypto", @@ -128,8 +264,19 @@ def run_walk_forward( 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") + _write_return_matrix( + returns_output, + profile=profile, + returns=full_window_returns, + market_history=shared_market_history, + ) return { "strategy_profile": profile, "domain": "crypto", @@ -146,16 +293,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..fa970fd 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,11 @@ 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) + + @property + def last_daily_returns(self) -> pd.Series: + return self._last_daily_returns.copy() def run( self, @@ -161,12 +181,17 @@ 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, + ) 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 +209,11 @@ def __init__( ) -> None: self._market_history = market_history self._synthetic_days = int(synthetic_days) + self._last_daily_returns = pd.Series(dtype=float) + + @property + def last_daily_returns(self) -> pd.Series: + return self._last_daily_returns.copy() def run( self, @@ -225,6 +255,11 @@ 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, + ) elapsed = (datetime.now(timezone.utc) - started).total_seconds() eval_frame = sliced if start_date is not None: @@ -232,7 +267,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 index bf27b9e..28d1424 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -1,30 +1,33 @@ from pathlib import Path -def test_drift_workflow_wires_pipeline_repo_and_lifecycle_env() -> None: +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 "Require lifecycle bucket" in workflow - assert "LIFECYCLE_PERFORMANCE_BUCKET must be configured for drift preflight" in workflow + assert "Download latest trusted lifecycle inputs" in workflow + assert "gh api --paginate --slurp" 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: 335c7a22bc3f570bd5705427ccc40172eda6b289" in workflow + assert "ref: 9bb8f31e898ea238a6446472f9f5e58133128d0c" 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 '"--store-root"' in workflow - assert "LIFECYCLE_PREFLIGHT_STAGING_ROOT" in workflow - assert "Promote staged lifecycle backtests" in workflow - assert "github.ref == format('refs/heads/{0}', github.event.repository.default_branch)" in workflow - assert "head.repo.full_name == github.repository" in workflow - assert "id-token: write" in workflow - assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@17278db4e7aef0007346d853eb308b6c1bd8c859" 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@9bb8f31e898ea238a6446472f9f5e58133128d0c" in workflow assert "strategy_domain: crypto" in workflow - assert "caller_event_name: ${{ github.event_name }}" in workflow - assert "caller_pr_head_repository: ${{ github.event.pull_request.head.repo.full_name || '' }}" in workflow assert "snapshot_repository: QuantStrategyLab/CryptoLivePoolPipelines" in workflow assert "snapshot_checkout_path: external/CryptoLivePoolPipelines" in workflow - assert "ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }}" 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..021eca7 100644 --- a/tests/test_orchestrator_runner.py +++ b/tests/test_orchestrator_runner.py @@ -39,6 +39,10 @@ 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)) def test_walk_forward_produces_one_result_per_window(self) -> None: from pathlib import Path @@ -74,6 +78,10 @@ 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)) 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 index 1299a1f..b8e062d 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -5,6 +5,7 @@ from pathlib import Path import pytest +import pandas as pd ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" @@ -15,6 +16,7 @@ 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_param_set_id, run_walk_forward @@ -74,3 +76,52 @@ def test_run_walk_forward_keeps_local_default_store(tmp_path: Path, monkeypatch: 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", "2024-12-31", 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"] 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" } From ec91e859a1427447f0ca46e955f3da61fc5f01f3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:55:47 +0800 Subject: [PATCH 12/16] fix: derive crypto baseline from exact return tail Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 16 +++++++++-- scripts/run_walk_forward_backtest.py | 38 ++++++++++++++++--------- tests/test_drift_workflow_config.py | 1 + tests/test_run_walk_forward_backtest.py | 19 ++++++++++++- 4 files changed, 56 insertions(+), 18 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 2e68267..09184d0 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -49,18 +49,26 @@ jobs: 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/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_run_ids = {run["id"] for page in run_pages for run in page.get("workflow_runs", [])} candidates = [ 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("head_branch") == "main" + and item.get("workflow_run", {}).get("id") in trusted_run_ids ] if not candidates: raise SystemExit("no trusted crypto lifecycle input artifact is available") @@ -68,7 +76,8 @@ jobs: 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" + gh api "/repos/QuantStrategyLab/CryptoLivePoolPipelines/actions/runs/${workflow_run_id}" \ + > "${RUNNER_TEMP}/snapshot-workflow-run.json" python - <<'PY' import json import os @@ -85,7 +94,8 @@ jobs: if mismatches: raise SystemExit(f"lifecycle input provenance check failed: {mismatches}") PY - gh api "/repos/QuantStrategyLab/CryptoLivePoolPipelines/actions/artifacts/${artifact_id}/zip" > "${RUNNER_TEMP}/snapshot-artifact.zip" + gh api "/repos/QuantStrategyLab/CryptoLivePoolPipelines/actions/artifacts/${artifact_id}/zip" \ + > "${RUNNER_TEMP}/snapshot-artifact.zip" python - <<'PY' import os import zipfile diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 75234f1..cf5341f 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -8,6 +8,7 @@ import hashlib import json import tempfile +from dataclasses import replace from datetime import date from pathlib import Path from typing import Any @@ -21,6 +22,7 @@ SUPPORTED_PROFILES, build_backtest_runner, ) +from crypto_strategies.backtest.live_pool_simulator import _performance_metrics from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = ( @@ -177,6 +179,26 @@ def _write_return_matrix( 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 = _performance_metrics(tail) + max_drawdown = float(metrics["Max Drawdown"]) + cagr = float(metrics["CAGR"]) + return replace( + full_result, + sharpe_ratio=float(metrics["Sharpe"]), + calmar_ratio=abs(cagr / max_drawdown) if max_drawdown else None, + max_drawdown=max_drawdown, + cagr=cagr, + volatility=float(metrics["Annualized Volatility"]), + win_rate=float(metrics["Win Rate"]), + total_return=float(metrics["total_return"]), + start_date=tail.index.min().date(), + end_date=tail.index.max().date(), + observation_count=int(metrics["Trading Days"]), + ) + + def run_walk_forward( *, profile: str, @@ -214,7 +236,7 @@ def run_walk_forward( ) full_start = min(start for start, _ in windows) baseline_end = max(end for _, end in windows) - return_matrix_runner.run( + full_window_raw = return_matrix_runner.run( profile, copy.deepcopy(baseline_params), start_date=full_start, @@ -223,19 +245,7 @@ def run_walk_forward( full_window_returns = return_matrix_runner.last_daily_returns if len(full_window_returns) < DRIFT_BASELINE_HORIZON_DAYS: raise ValueError("full-window returns do not cover the 126-day drift baseline") - baseline_start = full_window_returns.index[-DRIFT_BASELINE_HORIZON_DAYS].date() - baseline_runner = _build_runner( - profile=profile, - panel=shared_panel, - market_history=shared_market_history, - synthetic_days=synthetic_days, - ) - baseline_raw = baseline_runner.run( - profile, - copy.deepcopy(baseline_params), - start_date=baseline_start, - end_date=baseline_end, - ) + baseline_raw = _baseline_from_return_tail(full_window_raw, full_window_returns) with tempfile.TemporaryDirectory(prefix=f"{profile}_wf_", dir=target_root) as scratch_dir: scratch_orchestrator = BacktestOrchestrator(store=PerformanceStore(local_root=Path(scratch_dir))) scratch_orchestrator.register_runner( diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index 28d1424..3af59a2 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -8,6 +8,7 @@ def test_drift_workflow_wires_real_pipeline_inputs_and_preflight_bundle() -> Non assert "needs: preflight_backtests" 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 diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py index b8e062d..4895b49 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -17,7 +17,7 @@ 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_param_set_id, run_walk_forward +from scripts.run_walk_forward_backtest import _baseline_from_return_tail, _baseline_param_set_id, run_walk_forward def test_run_walk_forward_persists_lifecycle_baseline(tmp_path: Path) -> None: @@ -125,3 +125,20 @@ def test_run_walk_forward_uses_real_panel_and_writes_return_matrix( 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"] + + +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) From 6570847d6c0f09a1544485192816a7cc7a5ca6df Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:57:56 +0800 Subject: [PATCH 13/16] fix: fail closed on crypto input corruption Co-Authored-By: Codex --- scripts/run_walk_forward_backtest.py | 11 +++++++--- tests/test_run_walk_forward_backtest.py | 28 ++++++++++++++++++++++++- 2 files changed, 35 insertions(+), 4 deletions(-) diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index cf5341f..a2fe6b0 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -96,7 +96,9 @@ def _normalize_panel(panel: pd.DataFrame) -> pd.DataFrame: 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", "final_score"]) - return frame.drop_duplicates(["date", "symbol"], keep="last").set_index(["date", "symbol"]).sort_index() + 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: @@ -111,7 +113,10 @@ def _normalize_market_history(market_history: pd.DataFrame) -> pd.DataFrame: 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") - return frame.dropna().drop_duplicates(["date", "symbol"], keep="last").sort_values(["date", "symbol"]) + 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: @@ -136,7 +141,7 @@ def _shared_inputs( ] 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") - if normalized_panel.groupby(level="date")["in_universe"].sum().max() < 2: + if normalized_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) diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py index 4895b49..bc3e65c 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -17,7 +17,13 @@ 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, run_walk_forward +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: @@ -142,3 +148,23 @@ def test_baseline_uses_exact_tail_of_full_return_stream() -> None: 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) From b3e81a75881340b5abce317eed1dbcdd0b0d16a6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 08:02:12 +0800 Subject: [PATCH 14/16] chore: align QPK compatibility metadata Co-Authored-By: Codex --- qsl.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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", ] From 68871745cd26d8d811bbb6b6b74ce2c2be446881 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 08:39:12 +0800 Subject: [PATCH 15/16] fix: derive crypto drift baseline from walk-forward returns Co-Authored-By: Codex --- .github/workflows/drift-check.yml | 50 ++++++++++++++----- scripts/run_walk_forward_backtest.py | 35 ++++--------- .../backtest/orchestrator_runner.py | 12 +++++ tests/test_drift_workflow_config.py | 8 ++- tests/test_orchestrator_runner.py | 2 + tests/test_run_walk_forward_backtest.py | 1 + 6 files changed, 69 insertions(+), 39 deletions(-) diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 09184d0..d954d06 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -16,6 +16,8 @@ jobs: if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) runs-on: ubuntu-latest timeout-minutes: 30 + outputs: + snapshot_repository_ref: ${{ steps.snapshot-input.outputs.snapshot_repository_ref }} steps: - name: Checkout uses: actions/checkout@v6 @@ -24,7 +26,7 @@ jobs: uses: actions/checkout@v6 with: repository: QuantStrategyLab/QuantPlatformKit - ref: 9bb8f31e898ea238a6446472f9f5e58133128d0c + ref: bda6afdab0a2dd693c35d14493176829f4da1231 path: external/QuantPlatformKit - name: Set up Python @@ -40,6 +42,7 @@ jobs: python -m pip install --no-deps -e external/QuantPlatformKit - name: Download latest trusted lifecycle inputs + id: snapshot-input env: GH_TOKEN: ${{ secrets.SNAPSHOT_REPOSITORY_TOKEN }} INPUT_ROOT: ${{ runner.temp }}/crypto-lifecycle-inputs @@ -62,17 +65,24 @@ jobs: 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_run_ids = {run["id"] for page in run_pages for run in page.get("workflow_runs", [])} - candidates = [ - 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("head_branch") == "main" - and item.get("workflow_run", {}).get("id") in trusted_run_ids - ] - if not candidates: + 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") - selected = max(candidates, key=lambda item: item["created_at"]) print(selected["id"], selected["workflow_run"]["id"]) PY read -r artifact_id workflow_run_id < "${RUNNER_TEMP}/snapshot-artifact-selection.txt" @@ -94,6 +104,19 @@ jobs: 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' @@ -176,15 +199,16 @@ jobs: contents: read issues: write id-token: write - uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@9bb8f31e898ea238a6446472f9f5e58133128d0c + 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: 9bb8f31e898ea238a6446472f9f5e58133128d0c + 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 }} diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index a2fe6b0..4a4b9d9 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -233,34 +233,17 @@ def run_walk_forward( panel=panel, market_history=market_history, ) - return_matrix_runner = _build_runner( - profile=profile, - panel=shared_panel, - market_history=shared_market_history, - synthetic_days=synthetic_days, - ) - full_start = min(start for start, _ in windows) - baseline_end = max(end for _, end in windows) - full_window_raw = return_matrix_runner.run( - profile, - copy.deepcopy(baseline_params), - start_date=full_start, - end_date=baseline_end, - ) - full_window_returns = return_matrix_runner.last_daily_returns - if len(full_window_returns) < DRIFT_BASELINE_HORIZON_DAYS: - raise ValueError("full-window returns do not cover the 126-day drift baseline") - baseline_raw = _baseline_from_return_tail(full_window_raw, full_window_returns) 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", - _build_runner( - profile=profile, - panel=shared_panel, - market_history=shared_market_history, - synthetic_days=synthetic_days, - ), + walk_forward_runner, ) wf_results = scratch_orchestrator.walk_forward( profile, @@ -269,6 +252,10 @@ def run_walk_forward( 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, diff --git a/src/crypto_strategies/backtest/orchestrator_runner.py b/src/crypto_strategies/backtest/orchestrator_runner.py index fa970fd..91a83fd 100644 --- a/src/crypto_strategies/backtest/orchestrator_runner.py +++ b/src/crypto_strategies/backtest/orchestrator_runner.py @@ -145,11 +145,16 @@ def __init__(self, *, panel: pd.DataFrame | None = None, synthetic_days: int = 1 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, strategy_profile: str, @@ -186,6 +191,7 @@ def run( 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( @@ -210,11 +216,16 @@ def __init__( 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, strategy_profile: str, @@ -260,6 +271,7 @@ def run( 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: diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index 3af59a2..5c1382e 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -6,6 +6,10 @@ def test_drift_workflow_wires_real_pipeline_inputs_and_preflight_bundle() -> Non 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 @@ -15,7 +19,7 @@ def test_drift_workflow_wires_real_pipeline_inputs_and_preflight_bundle() -> Non assert "research_panel.csv.gz" in workflow assert "market_history.csv.gz" in workflow assert "repository: QuantStrategyLab/QuantPlatformKit" in workflow - assert "ref: 9bb8f31e898ea238a6446472f9f5e58133128d0c" 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 @@ -25,7 +29,7 @@ def test_drift_workflow_wires_real_pipeline_inputs_and_preflight_bundle() -> Non 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@9bb8f31e898ea238a6446472f9f5e58133128d0c" in workflow + 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 diff --git a/tests/test_orchestrator_runner.py b/tests/test_orchestrator_runner.py index 021eca7..05957e8 100644 --- a/tests/test_orchestrator_runner.py +++ b/tests/test_orchestrator_runner.py @@ -43,6 +43,7 @@ def test_run_returns_backtest_result(self) -> None: 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 @@ -82,6 +83,7 @@ def test_run_returns_backtest_result(self) -> None: 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 index bc3e65c..ce15e65 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -131,6 +131,7 @@ def test_run_walk_forward_uses_real_panel_and_writes_return_matrix( 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 len(return_matrix) == sum(item["observation_count"] for item in payload["walk_forward_folds"]) def test_baseline_uses_exact_tail_of_full_return_stream() -> None: From 80a352404e52f1aede429f31a11cc0ca5e44fb2d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 08:47:54 +0800 Subject: [PATCH 16/16] fix: separate current drift returns from baseline window Co-Authored-By: Codex --- scripts/run_walk_forward_backtest.py | 55 +++++++++++++++++-------- tests/test_run_walk_forward_backtest.py | 23 ++++++++++- 2 files changed, 59 insertions(+), 19 deletions(-) diff --git a/scripts/run_walk_forward_backtest.py b/scripts/run_walk_forward_backtest.py index 4a4b9d9..c18a000 100644 --- a/scripts/run_walk_forward_backtest.py +++ b/scripts/run_walk_forward_backtest.py @@ -14,6 +14,7 @@ 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, @@ -22,7 +23,6 @@ SUPPORTED_PROFILES, build_backtest_runner, ) -from crypto_strategies.backtest.live_pool_simulator import _performance_metrics from crypto_strategies.strategies.crypto_equity_combo import PROFILE_NAME as CRYPTO_EQUITY_COMBO_PROFILE DEFAULT_WINDOWS: tuple[tuple[date, date], ...] = ( @@ -95,7 +95,7 @@ def _normalize_panel(panel: pd.DataFrame) -> pd.DataFrame: 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", "final_score"]) + 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() @@ -137,18 +137,23 @@ def _shared_inputs( 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)) & (panel_dates <= pd.Timestamp(full_end)) + 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") - if normalized_panel.groupby(level="date")["in_universe"].sum().min() < 2: + 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"] <= pd.Timestamp(full_end)) + & (normalized_history["date"] <= current_end) ].copy() required_symbols = {"BTCUSDT", "ETHUSDT"} missing_symbols = sorted(required_symbols - set(normalized_history["symbol"])) @@ -186,21 +191,21 @@ def _write_return_matrix( def _baseline_from_return_tail(full_result: Any, returns: pd.Series) -> Any: tail = returns.tail(DRIFT_BASELINE_HORIZON_DAYS) - metrics = _performance_metrics(tail) - max_drawdown = float(metrics["Max Drawdown"]) - cagr = float(metrics["CAGR"]) + 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"]), - calmar_ratio=abs(cagr / max_drawdown) if max_drawdown else None, + sharpe_ratio=float(metrics.sharpe_ratio), + calmar_ratio=float(metrics.calmar_ratio), max_drawdown=max_drawdown, cagr=cagr, - volatility=float(metrics["Annualized Volatility"]), - win_rate=float(metrics["Win Rate"]), - total_return=float(metrics["total_return"]), - start_date=tail.index.min().date(), - end_date=tail.index.max().date(), - observation_count=int(metrics["Trading Days"]), + 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, ) @@ -273,10 +278,26 @@ def run_walk_forward( 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=full_window_returns, + returns=current_runner.last_daily_returns, market_history=shared_market_history, ) return { diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py index ce15e65..780132d 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -88,7 +88,7 @@ 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", "2024-12-31", freq="D") + dates = pd.date_range("2022-01-01", "2025-02-28", freq="D") symbols = ("BTCUSDT", "ETHUSDT", "SOLUSDT") rows = [] market_rows = [] @@ -131,7 +131,7 @@ def test_run_walk_forward_uses_real_panel_and_writes_return_matrix( 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 len(return_matrix) == sum(item["observation_count"] for item in payload["walk_forward_folds"]) + assert pd.Timestamp(return_matrix["as_of"].max()) > pd.Timestamp("2024-12-31") def test_baseline_uses_exact_tail_of_full_return_stream() -> None: @@ -169,3 +169,22 @@ def test_external_inputs_reject_duplicate_keys() -> None: _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"])