feat(strategy): load validated preflight panel

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

scripts/export_lifecycle_preflight_inputs.py

@@ -74,6 +74,7 @@ def export_lifecycle_inputs(panel: pd.DataFrame, output_dir: Path) -> dict[str,
     market_history.to_csv(market_path, index=False, compression="gzip")
     manifest = {
         "contract_version": "crypto.lifecycle_preflight.v1",
+        "strategy_profile": "crypto_live_pool_rotation",
         "panel_rows": int(len(lifecycle_panel)),
         "panel_symbols": sorted(lifecycle_panel["symbol"].unique().tolist()),
         "market_rows": int(len(market_history)),

src/strategy_lifecycle/backtest_wrapper.py

@@ -3,19 +3,53 @@
 from __future__ import annotations
 
 from datetime import date
+import json
+import os
+from pathlib import Path
 from typing import Any, Mapping
 
 import pandas as pd
 
 from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
 
-from .orchestrator_runner import CryptoLivePoolBacktestRunner
+from .orchestrator_runner import CryptoLivePoolBacktestRunner, PROFILE_NAME
 
 
 class InsufficientEvidenceError(RuntimeError):
     """Raised when lifecycle wiring does not provide a real market panel."""
 
 
+PREFLIGHT_CONTRACT_VERSION = "crypto.lifecycle_preflight.v1"
+PREFLIGHT_ENV = "CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"
+
+
+def load_preflight_panel() -> pd.DataFrame:
+    configured = os.environ.get(PREFLIGHT_ENV) or os.environ.get("PREFLIGHT_BUNDLE_ROOT")
+    if not configured:
+        raise InsufficientEvidenceError(f"{PREFLIGHT_ENV} is required for no-arg lifecycle registration")
+    root = Path(configured).expanduser().resolve()
+    if not (root / "research_panel.csv.gz").exists():
+        raise InsufficientEvidenceError(f"preflight bundle missing research_panel.csv.gz: {root}")
+    try:
+        manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
+        panel = pd.read_csv(root / "research_panel.csv.gz", compression="gzip")
+    except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
+        raise InsufficientEvidenceError(f"invalid lifecycle preflight bundle: {root}") from exc
+    if manifest.get("contract_version") != PREFLIGHT_CONTRACT_VERSION or manifest.get("strategy_profile") != PROFILE_NAME:
+        raise InsufficientEvidenceError("lifecycle preflight manifest mismatch")
+    required = {"date", "symbol", "in_universe", "open", "final_score"}
+    if not required.issubset(panel.columns):
+        raise InsufficientEvidenceError("research_panel.csv.gz missing required columns")
+    panel["date"] = pd.to_datetime(panel["date"], errors="coerce")
+    panel["open"] = pd.to_numeric(panel["open"], errors="coerce")
+    panel["final_score"] = pd.to_numeric(panel["final_score"], errors="coerce")
+    if panel.empty or panel["date"].isna().any() or panel["open"].isna().any() or panel["final_score"].isna().any():
+        raise InsufficientEvidenceError("research_panel.csv.gz contains invalid numeric/date content")
+    if (date.today() - panel["date"].dt.normalize().max().date()).days > 3:
+        raise InsufficientEvidenceError("research panel preflight artifact is stale")
+    return panel.set_index(["date", "symbol"]).sort_index()
+
+
 class CryptoBacktestRunner:
     """Expose the real crypto backtest engine through the lifecycle contract.
 
@@ -25,8 +59,8 @@ class CryptoBacktestRunner:
     runner used by tests/research fixtures.
     """
 
-    def __init__(self, *, panel: pd.DataFrame) -> None:
-        self._runner = CryptoLivePoolBacktestRunner(panel=panel)
+    def __init__(self, *, panel: pd.DataFrame | None = None) -> None:
+        self._runner = CryptoLivePoolBacktestRunner(panel=panel if panel is not None else load_preflight_panel())
 
     def run(
         self,
@@ -40,6 +74,4 @@ def run(
 
 def build_backtest_runner(*, panel: pd.DataFrame | None = None) -> CryptoBacktestRunner:
     """Build the real adapter; lifecycle wiring must inject a prepared panel."""
-    if panel is None:
-        raise InsufficientEvidenceError("build_backtest_runner requires a real prepared market panel")
     return CryptoBacktestRunner(panel=panel)

tests/test_export_lifecycle_preflight_inputs.py

@@ -40,6 +40,7 @@ def test_export_lifecycle_inputs_writes_real_panel_contract(self) -> None:
         self.assertEqual(set(market_history["symbol"]), {"BTCUSDT", "ETHUSDT"})
         self.assertEqual(manifest, persisted_manifest)
         self.assertEqual(manifest["contract_version"], "crypto.lifecycle_preflight.v1")
+        self.assertEqual(manifest["strategy_profile"], "crypto_live_pool_rotation")
 
     def test_export_rejects_scored_date_without_universe(self) -> None:
         panel = self._valid_panel()

tests/test_orchestrator_runner.py

@@ -3,9 +3,13 @@
 import sys
 import tempfile
 import unittest
+import json
+import os
 from datetime import date
 from pathlib import Path
 
+import pandas as pd
+
 PROJECT_ROOT = Path(__file__).resolve().parents[1]
 if str(PROJECT_ROOT) not in sys.path:
     sys.path.insert(0, str(PROJECT_ROOT))
@@ -24,8 +28,37 @@ def test_production_wrapper_requires_real_panel(self) -> None:
         self.assertIsNotNone(build_backtest_runner)
         from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError
 
-        with self.assertRaises(InsufficientEvidenceError):
-            build_backtest_runner()
+        old = os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None)
+        try:
+            with self.assertRaises(InsufficientEvidenceError):
+                build_backtest_runner()
+        finally:
+            if old is not None:
+                os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old
+
+    def test_no_arg_factory_loads_valid_preflight_bundle(self) -> None:
+        from src.strategy_lifecycle.backtest_wrapper import build_backtest_runner
+
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            dates = pd.date_range(end=pd.Timestamp.today().normalize(), periods=3, freq="D")
+            pd.DataFrame({
+                "date": dates,
+                "symbol": ["BTCUSDT"] * 3,
+                "in_universe": [True] * 3,
+                "open": [100.0, 101.0, 102.0],
+                "final_score": [0.1, 0.2, 0.3],
+            }).to_csv(root / "research_panel.csv.gz", index=False, compression="gzip")
+            (root / "manifest.json").write_text(json.dumps({"contract_version": "crypto.lifecycle_preflight.v1", "strategy_profile": "crypto_live_pool_rotation"}), encoding="utf-8")
+            old = os.environ.get("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT")
+            os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = str(root)
+            try:
+                self.assertIsNotNone(build_backtest_runner())
+            finally:
+                if old is None:
+                    os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None)
+                else:
+                    os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old
 
     def test_supported_profile(self) -> None:
         self.assertIn(PROFILE_NAME, SUPPORTED_PROFILES)