diff --git a/.github/workflows/publish-lifecycle-inputs.yml b/.github/workflows/publish-lifecycle-inputs.yml index cf9094a..20338c9 100644 --- a/.github/workflows/publish-lifecycle-inputs.yml +++ b/.github/workflows/publish-lifecycle-inputs.yml @@ -40,6 +40,7 @@ jobs: run: | set -euo pipefail completed_date="$(python -c 'from datetime import datetime, timedelta, timezone; print((datetime.now(timezone.utc) - timedelta(days=1)).date())')" + echo "CRYPTO_LIFECYCLE_EXPECTED_END_DATE=${completed_date}" >> "$GITHUB_ENV" python scripts/download_history.py --top-liquid "${DOWNLOAD_TOP_LIQUID}" \ --end-date "${completed_date}" \ --force-exchange-info @@ -52,6 +53,32 @@ jobs: --universe-mode broad_liquid \ --output-dir "${RUNNER_TEMP}/crypto-lifecycle-inputs" + - name: Inject and verify lifecycle preflight runtime wiring + env: + CRYPTO_LIFECYCLE_PREFLIGHT_ROOT: ${{ runner.temp }}/crypto-lifecycle-inputs + run: | + set -euo pipefail + echo "CRYPTO_LIFECYCLE_PREFLIGHT_ROOT=${CRYPTO_LIFECYCLE_PREFLIGHT_ROOT}" >> "$GITHUB_ENV" + python - <<'PY' + import json + import os + from datetime import date + from src.strategy_lifecycle.backtest_wrapper import build_backtest_runner + + with open(os.path.join(os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"], "manifest.json"), encoding="utf-8") as handle: + manifest = json.load(handle) + runner = build_backtest_runner() + expected_end_date = date.fromisoformat(os.environ["CRYPTO_LIFECYCLE_EXPECTED_END_DATE"]) + result = runner.run( + "crypto_live_pool_rotation", + {}, + end_date=expected_end_date, + ) + if result.observation_count <= 0: + raise SystemExit("real lifecycle preflight produced no observations") + print(f"verified real lifecycle runner observations={result.observation_count}") + PY + - name: Upload lifecycle inputs uses: actions/upload-artifact@v7 with: diff --git a/scripts/export_lifecycle_preflight_inputs.py b/scripts/export_lifecycle_preflight_inputs.py index 5b1a17e..3a78497 100644 --- a/scripts/export_lifecycle_preflight_inputs.py +++ b/scripts/export_lifecycle_preflight_inputs.py @@ -38,11 +38,12 @@ def export_lifecycle_inputs(panel: pd.DataFrame, output_dir: Path) -> dict[str, scored_panel = lifecycle_panel.dropna(subset=["final_score"]) if scored_panel.empty: raise ValueError("research panel has no scored lifecycle rows") - panel_dates = pd.DatetimeIndex(sorted(scored_panel["date"].unique())) - if len(panel_dates) < MIN_PANEL_DAYS: + # v2 bounds are the scored_panel bounds; warm-up/unscored rows remain in the CSV. + scored_dates = pd.DatetimeIndex(sorted(scored_panel["date"].unique())) + if len(scored_dates) < MIN_PANEL_DAYS: raise ValueError(f"research panel requires at least {MIN_PANEL_DAYS} scored dates") in_universe_counts = scored_panel.loc[scored_panel["in_universe"]].groupby("date")["symbol"].nunique() - in_universe_counts = in_universe_counts.reindex(panel_dates, fill_value=0) + in_universe_counts = in_universe_counts.reindex(scored_dates, fill_value=0) if int(in_universe_counts.min()) < 2: raise ValueError("research panel requires at least two in-universe symbols per scored date") @@ -61,7 +62,7 @@ def export_lifecycle_inputs(panel: pd.DataFrame, output_dir: Path) -> dict[str, or max(symbol_dates) < max(reference_dates) ): raise ValueError(f"market history has incomplete symbol coverage: {symbol}") - if today - panel_dates.max() > pd.Timedelta(days=MAX_FRESHNESS_DAYS): + if today - scored_dates.max() > pd.Timedelta(days=MAX_FRESHNESS_DAYS): raise ValueError("research panel is stale") if today - max(reference_dates) > pd.Timedelta(days=MAX_FRESHNESS_DAYS): raise ValueError("BTC/ETH market history is stale") @@ -73,13 +74,18 @@ def export_lifecycle_inputs(panel: pd.DataFrame, output_dir: Path) -> dict[str, lifecycle_panel.to_csv(panel_path, index=False, compression="gzip") market_history.to_csv(market_path, index=False, compression="gzip") manifest = { - "contract_version": "crypto.lifecycle_preflight.v1", + "contract_version": "crypto.lifecycle_preflight.v2", + "domain": "crypto", + "producer": "export_lifecycle_preflight_inputs.py", + "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)), "market_symbols": sorted(market_history["symbol"].unique().tolist()), - "start_date": panel_dates.min().date().isoformat(), - "end_date": panel_dates.max().date().isoformat(), + "start_date": scored_dates.min().date().isoformat(), + "end_date": scored_dates.max().date().isoformat(), + "market_start_date": market_history["date"].min().date().isoformat(), + "market_end_date": market_history["date"].max().date().isoformat(), } manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") return manifest diff --git a/scripts/run_crypto_live_pool_baseline_review.py b/scripts/run_crypto_live_pool_baseline_review.py new file mode 100644 index 0000000..9cc7400 --- /dev/null +++ b/scripts/run_crypto_live_pool_baseline_review.py @@ -0,0 +1,197 @@ +#!/usr/bin/env python3 +"""Produce a fail-closed S2 baseline review for crypto live-pool rotation.""" + +from __future__ import annotations + +import argparse +import csv +import json +import math +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +PROFILE = "crypto_live_pool_rotation" +GATE_NAMES = { + "H1": "falsifiable_hypothesis", + "H2": "data_provenance", + "H3": "sample_adequacy", + "H4": "benchmark_comparison", + "H5": "cost_model", + "H6": "cost_stress", + "H7": "oos_folds", + "H8": "purge_embargo", + "H9": "leakage_control", + "H10": "parameter_stability", + "H11": "risk_metrics", + "H12": "reproducibility", +} + + +def _gate(gate_id: str, reason: str, *, status: str = "insufficient_evidence", refs: list[str] | None = None) -> dict[str, Any]: + return { + "id": gate_id, + "name": GATE_NAMES[gate_id], + "status": status, + "reason_codes": [reason], + "evidence_refs": refs or [], + } + + +def build_review(*, performance_summary: Path, walkforward_summary: Path) -> dict[str, Any]: + missing = [] + if not _usable_performance_csv(performance_summary): + missing.append(str(performance_summary)) + if not _usable_walkforward_csv(walkforward_summary): + missing.append(str(walkforward_summary)) + reason = "MISSING_OR_INVALID_REAL_PERFORMANCE_ARTIFACT" if missing else "BASELINE_REVIEW_NOT_YET_FROZEN" + gates = [_gate(gate_id, reason) for gate_id in GATE_NAMES] + return { + "schema_version": "strategy_review.v1", + "profile": PROFILE, + "decision": "insufficient_evidence", + "promotion_allowed": False, + "score": 0, + "hard_gates": gates, + "scorecard": {"total": 0, "max": 100, "scored_gates": 0}, + "blocking_reason_codes": [reason], + "evidence": { + "metrics_kind": "performance", + "data_source": "unavailable" if missing else "unfrozen", + "sample_count": 0, + "oos_folds": 0, + "placeholder_metrics": False, + "provenance": { + "snapshot": {"source_revision": "unavailable", "cost_model": "unavailable", "data_timestamp": "unavailable", "status": "unavailable"}, + "backtest": {"source_revision": "unavailable", "cost_model": "unavailable", "data_timestamp": "unavailable", "status": "unavailable"}, + }, + "missing_artifacts": missing, + "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + }, + "decision_packet": { + "strategy_what": "按评分选择加密资产并进行 live-pool rotation；当前仅评审流程，不执行交易。", + "return_source": "真实 performance artifacts 未提供，收益来源无法确认。", + "loss_scenarios": "未完成真实回测，主要亏损场景无法确认；不得用 placeholder 指标替代。", + "max_risk": "证据不足，最大回撤、容量和流动性风险均不可确认。", + "evidence_sufficiency": "insufficient_evidence", + "version_change": "新增 fail-closed 基线评审与人工决策 packet；未改变 live 参数。", + "system_recommendation": "insufficient_evidence", + "technical_evidence_refs": missing, + "automation_boundary": { + "research_auto_after_hard_gates": True, + "shadow_auto_after_hard_gates": True, + "canary_mode": "bounded_preapproved_only", + "canary_limits": {"max_capital": 1000.0, "capital_currency": "USD", "max_duration_days": 14, "max_drawdown_fraction": 0.05, "max_leverage": 1.0, "max_concurrency": 1}, + "auto_scale_allowed": False, + "normal_live_requires_human": True, + "funding_leverage_risk_override_requires_human": True, + "hard_risk_auto_pause_rollback": True, + }, + "allowed_human_decisions": ["approve_research", "reject_rollback"], + }, + } + + +def _read_csv(path: Path) -> tuple[list[str], list[list[str]]] | None: + if not path.exists() or not path.is_file(): + return None + try: + with path.open(newline="", encoding="utf-8") as handle: + reader = csv.reader(handle) + header = next(reader, None) + data_rows = list(reader) + if not header or not data_rows: + return None + return ["".join(ch for ch in cell.lower() if ch.isalnum() or ch == "_") for cell in header], data_rows + except (OSError, UnicodeError, csv.Error): + return None + + +def _has_numeric_value(header: list[str], rows: list[list[str]], candidates: set[str]) -> bool: + for row in rows: + for index, name in enumerate(header): + if name not in candidates or index >= len(row): + continue + try: + value = float(row[index]) + if math.isfinite(value): + return True + except (TypeError, ValueError): + continue + return False + + +def _usable_performance_csv(path: Path) -> bool: + parsed = _read_csv(path) + if parsed is None: + return False + header, rows = parsed + metrics = {"cagr", "sharpe", "calmar", "maxdrawdown", "max_dd", "annualizedvolatility"} + if "strategy" in header: + strategy_index = header.index("strategy") + if not any(len(row) > strategy_index and row[strategy_index].strip().lower() == "final_score" for row in rows): + return False + return bool(set(header).intersection(metrics)) and _has_numeric_value(header, rows, metrics) + + +def _usable_walkforward_csv(path: Path) -> bool: + parsed = _read_csv(path) + if parsed is None: + return False + header, rows = parsed + required = {"window_id", "test_start", "test_end"} + metrics = {"window_cagr", "window_sharpe", "window_max_drawdown", "h30_precision", "h60_precision", "h90_precision"} + if not required.issubset(set(header)) or not set(header).intersection(metrics): + return False + positions = {name: header.index(name) for name in required} + for row in rows: + try: + if any(not row[index].strip() for index in positions.values()): + continue + int(row[positions["window_id"]]) + datetime.fromisoformat(row[positions["test_start"]].replace("Z", "+00:00")) + datetime.fromisoformat(row[positions["test_end"]].replace("Z", "+00:00")) + except (IndexError, TypeError, ValueError): + continue + if _has_numeric_value(header, [row], metrics): + return True + return False + + +def render_markdown(review: dict[str, Any]) -> str: + lines = [ + f"# Baseline review: `{review['profile']}`", + "", + f"- Decision: **{review['decision']}**", + f"- Promotion allowed: **{review['promotion_allowed']}**", + f"- Score: **{review['score']}/100**", + "", + "This report is fail-closed. No real performance artifact is promoted by this command.", + "", + "| Gate | Status | Reason |", + "|---|---|---|", + ] + for gate in review["hard_gates"]: + lines.append(f"| {gate['id']} {gate['name']} | {gate['status']} | {', '.join(gate['reason_codes'])} |") + lines.extend(["", "## Blocking reasons", "", *[f"- `{reason}`" for reason in review["blocking_reason_codes"]]]) + return "\n".join(lines) + "\n" + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--performance-summary", type=Path, default=Path("data/reports/performance_summary.csv")) + parser.add_argument("--walkforward-summary", type=Path, default=Path("data/reports/walkforward_validation_summary.csv")) + parser.add_argument("--output-json", type=Path, required=True) + parser.add_argument("--output-md", type=Path, required=True) + args = parser.parse_args(argv) + review = build_review(performance_summary=args.performance_summary, walkforward_summary=args.walkforward_summary) + args.output_json.parent.mkdir(parents=True, exist_ok=True) + args.output_md.parent.mkdir(parents=True, exist_ok=True) + args.output_json.write_text(json.dumps(review, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + args.output_md.write_text(render_markdown(review), encoding="utf-8") + print(json.dumps({"decision": review["decision"], "promotion_allowed": False, "output": str(args.output_json)})) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/strategy_lifecycle/backtest_wrapper.py b/src/strategy_lifecycle/backtest_wrapper.py index 2ece8bd..f7d5a66 100644 --- a/src/strategy_lifecycle/backtest_wrapper.py +++ b/src/strategy_lifecycle/backtest_wrapper.py @@ -1,19 +1,184 @@ -"""Crypto BacktestRunner — wraps BinancePlatform backtest for the lifecycle system.""" +"""Lifecycle backtest adapter for the crypto live-pool rotation strategy.""" from __future__ import annotations from datetime import date +import json +import math +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, PROFILE_NAME + + +class InsufficientEvidenceError(RuntimeError): + """Raised when lifecycle wiring does not provide a real market panel.""" + + +PREFLIGHT_V1 = "crypto.lifecycle_preflight.v1" +PREFLIGHT_V2 = "crypto.lifecycle_preflight.v2" +PREFLIGHT_ENV = "CRYPTO_LIFECYCLE_PREFLIGHT_ROOT" +MIN_PANEL_DAYS = 730 +MIN_MARKET_DAYS = 900 +MARKET_SYMBOLS = ("BTCUSDT", "ETHUSDT") + + +def load_preflight_panel(expected_start_date: date | None = None, expected_end_date: date | None = None) -> pd.DataFrame: + configured = os.environ.get(PREFLIGHT_ENV) + 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 not isinstance(manifest, dict): + raise InsufficientEvidenceError("lifecycle preflight manifest must be a JSON object") + contract_version = manifest.get("contract_version") + if contract_version not in {PREFLIGHT_V1, PREFLIGHT_V2}: + raise InsufficientEvidenceError("lifecycle preflight manifest mismatch") + if contract_version == PREFLIGHT_V2: + required_manifest = { + "contract_version", "domain", "producer", "strategy_profile", "panel_rows", "panel_symbols", + "market_rows", "market_symbols", "start_date", "end_date", "market_start_date", "market_end_date", + } + if not required_manifest.issubset(manifest): + raise InsufficientEvidenceError("v2 lifecycle preflight manifest fields are incomplete") + if manifest["domain"] != "crypto" or manifest["producer"] != "export_lifecycle_preflight_inputs.py" or manifest["strategy_profile"] != PROFILE_NAME: + raise InsufficientEvidenceError("v2 lifecycle preflight identity mismatch") + else: + if manifest.get("producer") not in {None, "export_lifecycle_preflight_inputs.py"}: + raise InsufficientEvidenceError("v1 lifecycle preflight producer mismatch") + if manifest.get("strategy_profile") not in {None, PROFILE_NAME}: + raise InsufficientEvidenceError("v1 lifecycle preflight strategy_profile mismatch") + for field in ("panel_symbols", "market_symbols"): + value = manifest.get(field) + if field in manifest and (not isinstance(value, list) or not all(isinstance(symbol, str) for symbol in value)): + raise InsufficientEvidenceError(f"{field} must be a list of symbols") + 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").dt.normalize() + panel["open"] = pd.to_numeric(panel["open"], errors="coerce") + panel["final_score"] = pd.to_numeric(panel["final_score"], errors="coerce") + if panel.duplicated(["date", "symbol"]).any(): + raise InsufficientEvidenceError("research_panel.csv.gz contains duplicate date/symbol rows") + universe_values = panel["in_universe"].map( + lambda value: value if isinstance(value, bool) else { + "true": True, "1": True, "false": False, "0": False, + }.get(str(value).strip().lower()) + ) + if universe_values.isna().any(): + raise InsufficientEvidenceError("research_panel.csv.gz contains invalid in_universe values") + panel["in_universe"] = universe_values.astype(bool) + if panel.empty or panel["date"].isna().any() or panel["open"].isna().any() or not panel["open"].map(math.isfinite).all(): + raise InsufficientEvidenceError("research_panel.csv.gz contains invalid numeric/date content") + # v2/v1 optional bounds are compared with scored rows, not warm-up/unscored CSV rows. + scored_dates = panel.loc[panel["final_score"].notna(), "date"] + if not panel["final_score"].dropna().map(math.isfinite).all(): + raise InsufficientEvidenceError("research_panel.csv.gz contains non-finite scores") + market_path = root / "market_history.csv.gz" + if not market_path.exists(): + raise InsufficientEvidenceError("preflight bundle missing market_history.csv.gz") + try: + market = pd.read_csv(market_path, usecols=["date", "symbol", "close"], compression="gzip") + except (OSError, UnicodeError, ValueError) as exc: + raise InsufficientEvidenceError("invalid market_history.csv.gz") from exc + if not {"date", "symbol", "close"}.issubset(market.columns): + raise InsufficientEvidenceError("market_history.csv.gz missing required columns") + market["date"] = pd.to_datetime(market["date"], errors="coerce") + market["close"] = pd.to_numeric(market["close"], errors="coerce") + if market.duplicated(["date", "symbol"]).any(): + raise InsufficientEvidenceError("market_history.csv.gz contains duplicate date/symbol rows") + if market.empty or market["date"].isna().any() or market["close"].isna().any() or not market["close"].map(math.isfinite).all(): + raise InsufficientEvidenceError("market_history.csv.gz contains invalid content") + market_dates = market["date"].dt.normalize() + if market_dates.nunique() < MIN_MARKET_DAYS: + raise InsufficientEvidenceError("market history has insufficient date coverage") + reference_dates = set(market.loc[market["symbol"] == MARKET_SYMBOLS[0], "date"].dt.normalize()) + if not reference_dates: + raise InsufficientEvidenceError("market history is missing BTCUSDT coverage") + for symbol in MARKET_SYMBOLS: + symbol_dates = set(market.loc[market["symbol"] == symbol, "date"].dt.normalize()) + if not symbol_dates or len(symbol_dates & reference_dates) / len(reference_dates) < 0.99: + raise InsufficientEvidenceError(f"market history has incomplete {symbol} coverage") + if manifest.get("panel_rows") is not None and (manifest["panel_rows"] != len(panel) or sorted(manifest.get("panel_symbols", [])) != sorted(panel["symbol"].dropna().unique().tolist())): + raise InsufficientEvidenceError("research panel does not match manifest counts or symbols") + if manifest.get("market_rows") is not None and (manifest["market_rows"] != len(market) or sorted(manifest.get("market_symbols", [])) != sorted(market["symbol"].dropna().unique().tolist())): + raise InsufficientEvidenceError("market history does not match manifest counts or symbols") + if panel["final_score"].notna().sum() == 0: + raise InsufficientEvidenceError("research_panel.csv.gz has no valid scored rows") + in_universe = panel["in_universe"] + if panel.loc[in_universe, "final_score"].isna().any(): + raise InsufficientEvidenceError("research_panel.csv.gz has malformed scores for in-universe rows") + scored_panel = panel.loc[panel["final_score"].notna()].copy() + scored_panel["date"] = scored_panel["date"].dt.normalize() + scored_dates = scored_panel["date"] + if scored_dates.nunique() < MIN_PANEL_DAYS: + raise InsufficientEvidenceError("research panel has insufficient scored date coverage") + in_universe_counts = scored_panel.loc[scored_panel["in_universe"]].groupby("date")["symbol"].nunique() + if in_universe_counts.empty or in_universe_counts.reindex(scored_dates.unique(), fill_value=0).min() < 2: + raise InsufficientEvidenceError("research panel has insufficient in-universe coverage") + panel_end_date = scored_dates.dt.normalize().max().date() if not scored_dates.empty else panel["date"].dt.normalize().max().date() + panel_start_date = scored_dates.dt.normalize().min().date() if not scored_dates.empty else panel["date"].dt.normalize().min().date() + today = date.today() + market_start_date = market["date"].dt.normalize().min().date() + market_end_date = market["date"].dt.normalize().max().date() + if market_start_date > panel_start_date or market_end_date < panel_end_date: + raise InsufficientEvidenceError("market history does not cover scored panel date range") + if panel_end_date > today or market_end_date > today: + raise InsufficientEvidenceError("preflight bundle contains future-dated data") + if (today - market_end_date).days > 3: + raise InsufficientEvidenceError("market history preflight artifact is stale") + if expected_start_date is not None and panel_start_date > expected_start_date: + raise InsufficientEvidenceError("research panel starts after requested evaluation window") + if expected_end_date is not None and panel_end_date < expected_end_date: + raise InsufficientEvidenceError("research panel ends before requested evaluation window") + freshness_reference = date.today() + if (freshness_reference - panel_end_date).days > 3: + raise InsufficientEvidenceError("research panel preflight artifact is stale") + if contract_version == PREFLIGHT_V2: + if manifest["start_date"] != scored_dates.dt.normalize().min().date().isoformat() or manifest["end_date"] != panel_end_date.isoformat(): + raise InsufficientEvidenceError("v2 panel date range does not match manifest") + market_start = market["date"].dt.normalize().min().date().isoformat() + market_end = market_end_date.isoformat() + if manifest["market_start_date"] != market_start or manifest["market_end_date"] != market_end: + raise InsufficientEvidenceError("v2 market date range does not match manifest") + else: + panel_start = panel_start_date.isoformat() + panel_end = panel_end_date.isoformat() + if any(key in manifest for key in ("start_date", "end_date")): + if manifest.get("start_date") != panel_start or manifest.get("end_date") != panel_end: + raise InsufficientEvidenceError("v1 panel date range does not match manifest") + market_start = market["date"].dt.normalize().min().date().isoformat() + market_end = market_end_date.isoformat() + if any(key in manifest for key in ("market_start_date", "market_end_date")): + if manifest.get("market_start_date") != market_start or manifest.get("market_end_date") != market_end: + raise InsufficientEvidenceError("v1 market date range does not match manifest") + return panel.set_index(["date", "symbol"]).sort_index() + class CryptoBacktestRunner: - """BacktestRunner for Crypto strategies. + """Expose the real crypto backtest engine through the lifecycle contract. - Wraps BinancePlatform/research/backtest.py and CryptoLivePoolPipelines scripts. + A market panel is required deliberately: returning hard-coded metrics would + make a lifecycle snapshot look like performance evidence without a data + source. Synthetic panels remain available only through the explicit pilot + runner used by tests/research fixtures. """ + def __init__(self, *, panel: pd.DataFrame | None = None) -> None: + self._panel = panel + self._runner: CryptoLivePoolBacktestRunner | None = None + def run( self, strategy_profile: str, @@ -21,27 +186,13 @@ def run( start_date: date | None = None, end_date: date | None = None, ) -> BacktestResult: - """Run backtest for a crypto strategy.""" - return BacktestResult( - strategy_profile=strategy_profile, - domain="crypto", - param_set_id=f"crypto_{strategy_profile}_1", - params=dict(params), - param_version=1, - sharpe_ratio=1.5, - calmar_ratio=1.1, - max_drawdown=-0.20, - cagr=0.35, - volatility=0.45, - win_rate=0.55, - start_date=start_date or date(2020, 1, 1), - end_date=end_date or date.today(), - observation_count=2000, - benchmark_symbol="buy_hold_BTC", - source_script="CryptoLivePoolPipelines/scripts/run_research_backtest.py", - ) - - -def build_backtest_runner() -> CryptoBacktestRunner: - """Factory for the Crypto backtest runner.""" - return CryptoBacktestRunner() + if self._panel is None: + self._runner = CryptoLivePoolBacktestRunner(panel=load_preflight_panel(start_date, end_date)) + elif self._runner is None: + self._runner = CryptoLivePoolBacktestRunner(panel=self._panel) + return self._runner.run(strategy_profile, params, start_date=start_date, end_date=end_date) + + +def build_backtest_runner(*, panel: pd.DataFrame | None = None) -> CryptoBacktestRunner: + """Build the real adapter; lifecycle wiring must inject a prepared panel.""" + return CryptoBacktestRunner(panel=panel) diff --git a/src/strategy_lifecycle/orchestrator_runner.py b/src/strategy_lifecycle/orchestrator_runner.py index c002388..7d60ac9 100644 --- a/src/strategy_lifecycle/orchestrator_runner.py +++ b/src/strategy_lifecycle/orchestrator_runner.py @@ -2,6 +2,7 @@ from __future__ import annotations +from copy import deepcopy from datetime import date, datetime, timezone from typing import Any, Mapping @@ -77,7 +78,7 @@ def _metrics_to_qpk_result( raise ImportError("quant_platform_kit is required to build BacktestResult") cagr = float(metrics.get("CAGR") or 0.0) max_drawdown = float(metrics.get("Max Drawdown") or 0.0) - calmar = abs(cagr / max_drawdown) if max_drawdown else None + calmar = cagr / abs(max_drawdown) if max_drawdown else None return QpkBacktestResult( strategy_profile=strategy_profile, domain="crypto", @@ -126,7 +127,13 @@ def run( raise ValueError("No panel rows for requested window") started = datetime.now(timezone.utc) - result = run_single_backtest(sliced, "final_score", DEFAULT_BACKTEST_CONFIG) + config = deepcopy(DEFAULT_BACKTEST_CONFIG) + strategy_params = params.get("strategy", params) + if isinstance(strategy_params, Mapping): + for key in config["strategy"]: + if key in strategy_params: + config["strategy"][key] = strategy_params[key] + result = run_single_backtest(sliced, "final_score", config) elapsed = (datetime.now(timezone.utc) - started).total_seconds() eval_dates = sliced.index.get_level_values("date") metrics = dict(result.metrics) @@ -134,7 +141,7 @@ def run( return _metrics_to_qpk_result( strategy_profile=strategy_profile, params=params, - metrics=result.metrics, + metrics=metrics, start_date=start_date or eval_dates.min().date(), end_date=end_date or eval_dates.max().date(), run_duration_seconds=elapsed, diff --git a/tests/test_baseline_review.py b/tests/test_baseline_review.py new file mode 100644 index 0000000..a44c4b0 --- /dev/null +++ b/tests/test_baseline_review.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import importlib.util +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location("baseline_review", ROOT / "scripts/run_crypto_live_pool_baseline_review.py") +MODULE = importlib.util.module_from_spec(SPEC) +assert SPEC and SPEC.loader +SPEC.loader.exec_module(MODULE) + + +class BaselineReviewTests(unittest.TestCase): + def test_walkforward_schema_is_checked_separately_from_performance_metrics(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + performance = root / "performance.csv" + walkforward = root / "walkforward.csv" + performance.write_text("CAGR,Sharpe\n0.2,1.1\n", encoding="utf-8") + walkforward.write_text( + "window_id,test_start,test_end,window_cagr,window_sharpe\n" + "0,2025-01-01,2025-03-31,0.1,0.8\n", + encoding="utf-8", + ) + + review = MODULE.build_review(performance_summary=performance, walkforward_summary=walkforward) + + self.assertEqual(review["blocking_reason_codes"], ["BASELINE_REVIEW_NOT_YET_FROZEN"]) + + def test_missing_real_artifacts_is_insufficient_and_not_promotable(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + review = MODULE.build_review( + performance_summary=root / "performance.csv", + walkforward_summary=root / "walkforward.csv", + ) + self.assertEqual(review["decision"], "insufficient_evidence") + self.assertFalse(review["promotion_allowed"]) + self.assertEqual(len(review["hard_gates"]), 12) + self.assertIn("MISSING_OR_INVALID_REAL_PERFORMANCE_ARTIFACT", review["blocking_reason_codes"]) + self.assertFalse(review["evidence"]["placeholder_metrics"]) + packet = review["decision_packet"] + self.assertEqual(packet["system_recommendation"], "insufficient_evidence") + self.assertEqual(packet["evidence_sufficiency"], "insufficient_evidence") + self.assertEqual(packet["allowed_human_decisions"], ["approve_research", "reject_rollback"]) + + def test_performance_requires_final_score_when_strategy_column_exists(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + performance = root / "performance.csv" + walkforward = root / "walkforward.csv" + performance.write_text("strategy,CAGR\nrule_score,0.2\n", encoding="utf-8") + walkforward.write_text( + "window_id,test_start,test_end,window_cagr\n0,2025-01-01,2025-03-31,0.1\n", + encoding="utf-8", + ) + review = MODULE.build_review(performance_summary=performance, walkforward_summary=walkforward) + self.assertEqual(review["blocking_reason_codes"], ["MISSING_OR_INVALID_REAL_PERFORMANCE_ARTIFACT"]) + + def test_walkforward_requires_populated_parseable_window_fields(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + performance = root / "performance.csv" + walkforward = root / "walkforward.csv" + performance.write_text("CAGR,Sharpe\n0.2,1.1\n", encoding="utf-8") + walkforward.write_text( + "window_id,test_start,test_end,window_cagr\n,,not-a-date,0.1\n", + encoding="utf-8", + ) + review = MODULE.build_review(performance_summary=performance, walkforward_summary=walkforward) + self.assertEqual(review["blocking_reason_codes"], ["MISSING_OR_INVALID_REAL_PERFORMANCE_ARTIFACT"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_export_lifecycle_preflight_inputs.py b/tests/test_export_lifecycle_preflight_inputs.py index d6180a5..a4d0ff6 100644 --- a/tests/test_export_lifecycle_preflight_inputs.py +++ b/tests/test_export_lifecycle_preflight_inputs.py @@ -39,11 +39,16 @@ 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["contract_version"], "crypto.lifecycle_preflight.v2") + self.assertEqual(manifest["domain"], "crypto") + self.assertIn("market_start_date", manifest) + self.assertIn("market_end_date", manifest) + self.assertEqual(manifest["strategy_profile"], "crypto_live_pool_rotation") + self.assertEqual(manifest["producer"], "export_lifecycle_preflight_inputs.py") def test_export_rejects_scored_date_without_universe(self) -> None: panel = self._valid_panel() - latest_completed_date = panel.index.get_level_values("date").unique()[-2] + latest_completed_date = panel.index.get_level_values("date").unique()[-3] panel.loc[(latest_completed_date, slice(None)), "in_universe"] = False with tempfile.TemporaryDirectory() as tmpdir, self.assertRaisesRegex( @@ -69,6 +74,22 @@ def test_export_preserves_open_rows_without_scores(self) -> None: self.assertEqual(len(row), 1) self.assertTrue(pd.isna(row.iloc[0]["final_score"])) + def test_manifest_bounds_use_scored_rows_with_unscored_boundaries(self) -> None: + panel = self._valid_panel() + dates = panel.index.get_level_values("date").unique() + for boundary in (dates[0], dates[-2]): + panel.loc[(boundary, slice(None)), ["in_universe", "final_score"]] = [False, pd.NA] + + with tempfile.TemporaryDirectory() as tmpdir: + output_dir = Path(tmpdir) + manifest = export_lifecycle_inputs(panel, output_dir) + exported = pd.read_csv(output_dir / "research_panel.csv.gz") + + scored = exported.dropna(subset=["final_score"]) + self.assertEqual(manifest["start_date"], scored["date"].min()) + self.assertEqual(manifest["end_date"], scored["date"].max()) + self.assertNotEqual(manifest["start_date"], exported["date"].min()) + def test_export_rejects_stale_combo_history_even_when_panel_is_fresh(self) -> None: panel = self._valid_panel() cutoff = panel.index.get_level_values("date").max() - pd.Timedelta(days=10) diff --git a/tests/test_orchestrator_runner.py b/tests/test_orchestrator_runner.py index a99ffb4..23d9eb6 100644 --- a/tests/test_orchestrator_runner.py +++ b/tests/test_orchestrator_runner.py @@ -3,8 +3,13 @@ import sys import tempfile import unittest -from datetime import date +import json +import os +from datetime import date, timedelta from pathlib import Path +from unittest.mock import patch + +import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: @@ -18,6 +23,244 @@ class CryptoOrchestratorRunnerTests(unittest.TestCase): + @staticmethod + def _write_bundle(root: Path, *, version: str = "v1", days: int = 900, age_days: int = 0, non_finite: bool = False) -> None: + from src.strategy_lifecycle.orchestrator_runner import _synthetic_panel + + panel = _synthetic_panel(days=days).reset_index() + panel["date"] = pd.to_datetime(panel["date"]) + panel["date"] += pd.Timestamp.today().normalize() - pd.Timedelta(days=age_days) - panel["date"].max() + if non_finite: + panel.loc[0, "open"] = float("inf") + panel.to_csv(root / "research_panel.csv.gz", index=False, compression="gzip") + market = panel[["date", "symbol", "open"]].rename(columns={"open": "close"}) + market.to_csv(root / "market_history.csv.gz", index=False, compression="gzip") + manifest = {"contract_version": f"crypto.lifecycle_preflight.{version}"} + if version == "v2": + manifest.update({ + "domain": "crypto", "producer": "export_lifecycle_preflight_inputs.py", + "strategy_profile": PROFILE_NAME, "panel_rows": len(panel), + "panel_symbols": sorted(panel["symbol"].unique().tolist()), + "market_rows": len(market), "market_symbols": sorted(market["symbol"].unique().tolist()), + "start_date": panel["date"].min().date().isoformat(), + "end_date": panel["date"].max().date().isoformat(), + "market_start_date": market["date"].min().date().isoformat(), + "market_end_date": market["date"].max().date().isoformat(), + }) + (root / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + + def test_production_wrapper_requires_real_panel(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import build_backtest_runner + + self.assertIsNotNone(build_backtest_runner) + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError + + old = os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None) + try: + runner = build_backtest_runner() + with self.assertRaises(InsufficientEvidenceError): + runner.run(PROFILE_NAME, {}) + finally: + if old is not None: + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old + + def test_no_arg_factory_ignores_legacy_preflight_env(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, build_backtest_runner + + old = os.environ.get("PREFLIGHT_BUNDLE_ROOT") + os.environ["PREFLIGHT_BUNDLE_ROOT"] = "/tmp/unrelated-preflight-bundle" + os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None) + try: + with self.assertRaises(InsufficientEvidenceError): + build_backtest_runner().run(PROFILE_NAME, {}) + finally: + if old is None: + os.environ.pop("PREFLIGHT_BUNDLE_ROOT", None) + else: + os.environ["PREFLIGHT_BUNDLE_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) + self._write_bundle(root) + old = os.environ.get("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT") + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = str(root) + try: + runner = build_backtest_runner() + self.assertIsNotNone(runner) + self.assertEqual(runner._runner, None) + result = runner.run(PROFILE_NAME, {}) + self.assertEqual(result.strategy_profile, PROFILE_NAME) + self.assertIsNotNone(runner._runner) + finally: + if old is None: + os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None) + else: + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old + + def test_no_arg_runner_revalidates_second_window(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, build_backtest_runner + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root) + old = os.environ.get("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT") + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = str(root) + try: + runner = build_backtest_runner() + runner.run(PROFILE_NAME, {}, start_date=date.today() - timedelta(days=100)) + with self.assertRaises(InsufficientEvidenceError): + runner.run(PROFILE_NAME, {}, start_date=date.today() - timedelta(days=1000)) + finally: + if old is None: + os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None) + else: + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old + + def test_v1_bundle_rejects_inconsistent_optional_date_metadata(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, load_preflight_panel + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root) + manifest_path = root / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["start_date"] = "2000-01-01" + manifest["end_date"] = "2000-01-02" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + with self.assertRaises(InsufficientEvidenceError): + load_preflight_panel(root) + + def test_v2_manifest_is_strictly_loaded(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import build_backtest_runner + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, version="v2") + old = os.environ.get("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT") + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = str(root) + try: + self.assertEqual(build_backtest_runner().run(PROFILE_NAME, {}).strategy_profile, PROFILE_NAME) + finally: + if old is None: + os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None) + else: + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old + + def test_unknown_stale_and_non_finite_bundles_fail_closed(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, build_backtest_runner + + for kwargs in ({"version": "v3"}, {"age_days": 10}, {"non_finite": True}): + with self.subTest(kwargs=kwargs), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, **kwargs) + old = os.environ.get("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT") + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = str(root) + try: + with self.assertRaises(InsufficientEvidenceError): + build_backtest_runner().run(PROFILE_NAME, {}) + finally: + if old is None: + os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None) + else: + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old + + def test_malformed_manifest_and_future_bundle_fail_closed(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, build_backtest_runner + + for malformed in (True, False): + with self.subTest(malformed=malformed), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, age_days=-1 if not malformed else 0) + if malformed: + (root / "manifest.json").write_text("[]", encoding="utf-8") + old = os.environ.get("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT") + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = str(root) + try: + with self.assertRaises(InsufficientEvidenceError): + build_backtest_runner().run(PROFILE_NAME, {}) + finally: + if old is None: + os.environ.pop("CRYPTO_LIFECYCLE_PREFLIGHT_ROOT", None) + else: + os.environ["CRYPTO_LIFECYCLE_PREFLIGHT_ROOT"] = old + + def test_malformed_manifest_symbol_lists_fail_closed(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, load_preflight_panel + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, version="v2") + manifest_path = root / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["panel_symbols"] = None + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + with self.assertRaises(InsufficientEvidenceError): + load_preflight_panel(root) + + def test_undersized_bundle_fails_closed(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, load_preflight_panel + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, version="v2", days=100) + with self.assertRaises(InsufficientEvidenceError): + load_preflight_panel(root) + + def test_stale_market_history_fails_closed(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, load_preflight_panel + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, version="v2") + market_path = root / "market_history.csv.gz" + market = pd.read_csv(market_path, compression="gzip") + market["date"] = pd.to_datetime(market["date"]) - pd.Timedelta(days=10) + market.to_csv(market_path, index=False, compression="gzip") + manifest_path = root / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["market_start_date"] = market["date"].min().date().isoformat() + manifest["market_end_date"] = market["date"].max().date().isoformat() + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + with self.assertRaises(InsufficientEvidenceError): + load_preflight_panel(root) + + def test_duplicate_preflight_rows_fail_closed(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, load_preflight_panel + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, version="v2") + panel_path = root / "research_panel.csv.gz" + panel = pd.read_csv(panel_path, compression="gzip") + panel = pd.concat([panel, panel.iloc[[0]]], ignore_index=True) + panel.to_csv(panel_path, index=False, compression="gzip") + with self.assertRaises(InsufficientEvidenceError): + load_preflight_panel(root) + + def test_market_history_must_cover_scored_panel_window(self) -> None: + from src.strategy_lifecycle.backtest_wrapper import InsufficientEvidenceError, load_preflight_panel + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self._write_bundle(root, version="v2", days=1000) + market_path = root / "market_history.csv.gz" + market = pd.read_csv(market_path, compression="gzip") + panel = pd.read_csv(root / "research_panel.csv.gz", compression="gzip") + panel_start = pd.to_datetime(panel["date"]).min() + market = market[pd.to_datetime(market["date"]) > panel_start + pd.Timedelta(days=10)] + market.to_csv(market_path, index=False, compression="gzip") + manifest_path = root / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["market_rows"] = len(market) + manifest["market_start_date"] = market["date"].min() + manifest["market_end_date"] = market["date"].max() + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + with self.assertRaises(InsufficientEvidenceError): + load_preflight_panel(root) + def test_supported_profile(self) -> None: self.assertIn(PROFILE_NAME, SUPPORTED_PROFILES) @@ -30,6 +273,35 @@ def test_run_returns_backtest_result(self) -> None: end_date=date(2024, 3, 1), ) self.assertEqual(result.strategy_profile, PROFILE_NAME) + + def test_real_runner_applies_params_and_preserves_metrics(self) -> None: + from src.backtest import BacktestResult + + captured: dict[str, object] = {} + + def fake_backtest(panel, score_column, config): + captured["config"] = config + returns = pd.Series([0.01, -0.02]) + return BacktestResult( + name=score_column, + returns=returns, + equity_curve=(1 + returns).cumprod(), + holdings=pd.DataFrame(), + trades=pd.DataFrame(), + turnover=pd.Series([0.0, 0.0]), + metrics={"CAGR": -0.1, "Max Drawdown": -0.2, "Sharpe": -0.1}, + ) + + with patch("src.strategy_lifecycle.orchestrator_runner.run_single_backtest", side_effect=fake_backtest): + result = CryptoLivePoolBacktestRunner(synthetic_days=1600).run( + PROFILE_NAME, + {"top_n": 1, "fee_bps": 30}, + ) + + self.assertEqual(captured["config"]["strategy"]["top_n"], 1) + self.assertEqual(captured["config"]["strategy"]["fee_bps"], 30) + self.assertEqual(result.observation_count, 2) + self.assertLess(result.calmar_ratio, 0) self.assertEqual(result.domain, "crypto") self.assertIsNotNone(result.sharpe_ratio) diff --git a/tests/test_publish_lifecycle_inputs_workflow.py b/tests/test_publish_lifecycle_inputs_workflow.py index a039b74..c422c9f 100644 --- a/tests/test_publish_lifecycle_inputs_workflow.py +++ b/tests/test_publish_lifecycle_inputs_workflow.py @@ -14,3 +14,8 @@ def test_publish_lifecycle_inputs_workflow_uses_real_research_pipeline() -> None assert "--universe-mode broad_liquid" in workflow assert "crypto-lifecycle-inputs-${{ github.run_id }}-${{ github.run_attempt }}" in workflow assert "if-no-files-found: error" in workflow + assert "Inject and verify lifecycle preflight runtime wiring" in workflow + assert 'echo "CRYPTO_LIFECYCLE_PREFLIGHT_ROOT=${CRYPTO_LIFECYCLE_PREFLIGHT_ROOT}" >> "$GITHUB_ENV"' in workflow + assert 'CRYPTO_LIFECYCLE_EXPECTED_END_DATE=${completed_date}' in workflow + assert 'end_date=expected_end_date' in workflow + assert "runner = build_backtest_runner()" in workflow