diff --git a/.github/workflows/reusable-drift-check.yml b/.github/workflows/reusable-drift-check.yml index 76f824c..60438dd 100644 --- a/.github/workflows/reusable-drift-check.yml +++ b/.github/workflows/reusable-drift-check.yml @@ -276,7 +276,7 @@ jobs: if [ -n "${LIFECYCLE_LIVE_STREAM_ID:-}" ]; then lifecycle_args+=(--live-stream-id "${LIFECYCLE_LIVE_STREAM_ID}") fi - quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} "${lifecycle_args[@]}" + quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} --source-revision "${GITHUB_SHA}" "${lifecycle_args[@]}" - name: Validate lifecycle prerequisites shell: bash diff --git a/src/quant_platform_kit/strategy_lifecycle/cli.py b/src/quant_platform_kit/strategy_lifecycle/cli.py index f64123f..36ff7b2 100644 --- a/src/quant_platform_kit/strategy_lifecycle/cli.py +++ b/src/quant_platform_kit/strategy_lifecycle/cli.py @@ -42,6 +42,9 @@ def _run_monitor(args: argparse.Namespace) -> int: live_stream_id = getattr(args, "live_stream_id", None) if live_stream_id: kwargs["live_stream_id"] = live_stream_id + source_revision = getattr(args, "source_revision", None) + if source_revision: + kwargs["source_revision"] = source_revision snapshots = run_monitor(**kwargs) _print(f"[monitor] Generated {len(snapshots)} performance snapshots") return 0 @@ -381,6 +384,15 @@ def build_parser() -> argparse.ArgumentParser: default=None, help="Monitor one stable account/runtime telemetry stream; never mix streams.", ) + monitor.add_argument( + "--source-revision", + default=None, + help=( + "Observation provenance written onto daily snapshots " + "(40-char git SHA preferred). Falls back to LIFECYCLE_SOURCE_REVISION " + "or GITHUB_SHA when omitted." + ), + ) monitor.add_argument( "--benchmark-catalog", default=None, diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_monitor.py b/src/quant_platform_kit/strategy_lifecycle/performance_monitor.py index 90b7415..6d07870 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_monitor.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_monitor.py @@ -7,6 +7,9 @@ from __future__ import annotations import logging +import os +import re +from dataclasses import replace from datetime import date, datetime, timezone from pathlib import Path from typing import Any, Mapping, Sequence @@ -28,11 +31,60 @@ resolve_strategy_benchmark, ) +_SOURCE_REVISION_SHA = re.compile(r"^[0-9a-f]{40}$") +_SOURCE_REVISION_SENTINELS = frozenset( + {"", "unavailable", "not_available", "legacy_missing", "unknown", "none", "null"} +) + def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() +def resolve_monitor_source_revision( + explicit: str | None = None, + *, + environ: Mapping[str, str] | None = None, +) -> str: + """Resolve a real observation ``source_revision`` for daily snapshots. + + Order: explicit argument, ``LIFECYCLE_SOURCE_REVISION``, then ``GITHUB_SHA``. + Full 40-character lowercase git SHAs are always accepted. Non-SHA labels are + allowed only from the explicit argument or ``LIFECYCLE_SOURCE_REVISION`` + (existing probe/contract non-empty provenance). ``GITHUB_SHA`` must be a + full 40-character SHA. Missing or sentinel values raise instead of writing + an empty string that would look like a complete observation. + """ + env = os.environ if environ is None else environ + candidates: list[tuple[str, bool]] = [] + if explicit is not None: + candidates.append((str(explicit), False)) + for key, sha_only in ( + ("LIFECYCLE_SOURCE_REVISION", False), + ("GITHUB_SHA", True), + ): + raw = env.get(key) + if raw is None: + continue + candidates.append((str(raw), sha_only)) + + for raw, sha_only in candidates: + value = raw.strip() + if not value or value.lower() in _SOURCE_REVISION_SENTINELS: + continue + lowered = value.lower() + if _SOURCE_REVISION_SHA.fullmatch(lowered): + return lowered + if sha_only: + continue + return value + + raise RuntimeError( + "monitor source_revision is required; pass source_revision=..., " + "set LIFECYCLE_SOURCE_REVISION, or provide GITHUB_SHA (40-char)" + ) + + def resolve_lifecycle_stream_id( explicit_stream_id: str = "", *, @@ -45,8 +97,6 @@ def resolve_lifecycle_stream_id( synthetic equity curve. ``LIFECYCLE_STREAM_ID`` is available for non-Cloud-Run runtimes that need a stable explicit identity. """ - import os - for candidate in ( explicit_stream_id, os.environ.get("LIFECYCLE_STREAM_ID"), @@ -83,6 +133,7 @@ def run_monitor( strategy_benchmarks: Mapping[str, str] | None = None, require_explicit_benchmark: bool = False, live_stream_id: str | None = None, + source_revision: str | None = None, ) -> list[StrategyPerformanceSnapshot]: """Run the performance monitor for the given domain. @@ -102,6 +153,9 @@ def run_monitor( live_stream_id: Optional stable telemetry stream identity. When live account data is used, this prevents independent broker accounts from being merged into one return series. + source_revision: Observation provenance. Required when snapshots would be + written; resolved from env when omitted (see + :func:`resolve_monitor_source_revision`). Returns: List of StrategyPerformanceSnapshot objects generated. @@ -122,6 +176,9 @@ def run_monitor( ) return [] + # Fail closed before any snapshot write when provenance is missing. + resolved_source_revision = resolve_monitor_source_revision(source_revision) + profiles = [strategy_profile] if strategy_profile else sorted(all_returns.keys()) snapshots: list[StrategyPerformanceSnapshot] = [] @@ -160,6 +217,7 @@ def run_monitor( as_of=date.today(), benchmark_symbol=benchmark_symbol, computed_at=_now_iso(), + source_revision=resolved_source_revision, ) # Compute each window @@ -177,28 +235,19 @@ def run_monitor( # Latest return if len(series) > 0: - snapshot = StrategyPerformanceSnapshot( - strategy_profile=snapshot.strategy_profile, - domain=snapshot.domain, - platform=snapshot.platform, - as_of=snapshot.as_of, + snapshot = replace( + snapshot, windows=windows_dict, latest_return=float(series.iloc[-1]), - benchmark_symbol=snapshot.benchmark_symbol, - data_freshness_days=(date.today() - series.index[-1].date()).days if hasattr(series.index[-1], "date") else 0, + data_freshness_days=( + (date.today() - series.index[-1].date()).days + if hasattr(series.index[-1], "date") + else 0 + ), source_artifact_path="", - computed_at=snapshot.computed_at, ) else: - snapshot = StrategyPerformanceSnapshot( - strategy_profile=snapshot.strategy_profile, - domain=snapshot.domain, - platform=snapshot.platform, - as_of=snapshot.as_of, - windows=windows_dict, - benchmark_symbol=snapshot.benchmark_symbol, - computed_at=snapshot.computed_at, - ) + snapshot = replace(snapshot, windows=windows_dict) # Attach drift reference: use 126-day window to compare against backtest ref_window = windows_dict.get(126) or windows_dict.get(252) @@ -206,19 +255,7 @@ def run_monitor( deviations = compare_with_backtest(ref_window, latest_backtest) if deviations: max_dev = max(deviations.values()) - snapshot = StrategyPerformanceSnapshot( - strategy_profile=snapshot.strategy_profile, - domain=snapshot.domain, - platform=snapshot.platform, - as_of=snapshot.as_of, - windows=snapshot.windows, - latest_return=snapshot.latest_return, - benchmark_symbol=snapshot.benchmark_symbol, - drift_score=min(max_dev, 1.0), - data_freshness_days=snapshot.data_freshness_days, - source_artifact_path=snapshot.source_artifact_path, - computed_at=snapshot.computed_at, - ) + snapshot = replace(snapshot, drift_score=min(max_dev, 1.0)) # Persist store.save_snapshot(snapshot) diff --git a/tests/test_lifecycle_performance_monitor.py b/tests/test_lifecycle_performance_monitor.py index 3a73157..3a9b961 100644 --- a/tests/test_lifecycle_performance_monitor.py +++ b/tests/test_lifecycle_performance_monitor.py @@ -11,6 +11,7 @@ from quant_platform_kit.strategy_lifecycle.performance_monitor import ( PerformanceMonitor, infer_strategy_domain, + resolve_monitor_source_revision, run_monitor, try_record_platform_execution, ) @@ -65,15 +66,57 @@ def test_infer_strategy_domain_from_profile_prefix(self) -> None: def test_try_record_platform_execution_swallows_errors(self) -> None: try_record_platform_execution("", {"status": "ok"}) + def test_resolve_monitor_source_revision_requires_real_provenance(self) -> None: + sha = "a" * 40 + self.assertEqual(resolve_monitor_source_revision(sha), sha) + self.assertEqual( + resolve_monitor_source_revision(None, environ={"LIFECYCLE_SOURCE_REVISION": "fixture-rev"}), + "fixture-rev", + ) + self.assertEqual( + resolve_monitor_source_revision(None, environ={"GITHUB_SHA": sha}), + sha, + ) + with self.assertRaisesRegex(RuntimeError, "source_revision is required"): + resolve_monitor_source_revision(None, environ={}) + with self.assertRaisesRegex(RuntimeError, "source_revision is required"): + resolve_monitor_source_revision("", environ={"GITHUB_SHA": "deadbeef"}) + def test_run_monitor_fails_closed_when_no_profiles_found(self) -> None: class EmptyCollector: def collect(self, _domain: str) -> dict[str, pd.Series]: return {} with self.assertRaisesRegex(RuntimeError, "No strategy return series found"): - run_monitor("us_equity", collector=EmptyCollector()) + run_monitor( + "us_equity", + collector=EmptyCollector(), + source_revision="a" * 40, + ) + + def test_run_monitor_fails_closed_without_source_revision(self) -> None: + class OneCollector: + def collect(self, _domain: str) -> dict[str, pd.Series]: + return { + "synthetic_soxl": pd.Series( + [0.01, -0.02], + index=pd.to_datetime(["2026-09-08", "2026-09-09"]), + ) + } + + with patch.dict("os.environ", {}, clear=False): + for key in ("LIFECYCLE_SOURCE_REVISION", "GITHUB_SHA"): + # Ensure missing provenance cannot silently write empty revision. + pass + with patch( + "quant_platform_kit.strategy_lifecycle.performance_monitor.resolve_monitor_source_revision", + side_effect=RuntimeError("monitor source_revision is required"), + ): + with self.assertRaisesRegex(RuntimeError, "source_revision is required"): + run_monitor("us_equity", collector=OneCollector(), windows=(2,), min_observations=2) def test_csv_collector_to_monitor_rejects_duplicate_daily_returns(self) -> None: + revision = "b" * 40 for dates, values in ( (["2026-09-08", "2026-09-08"], [0.01, 0.01]), (["2026-09-08T09:00:00", "2026-09-08T16:00:00"], [0.01, -0.02]), @@ -86,14 +129,30 @@ def test_csv_collector_to_monitor_rejects_duplicate_daily_returns(self) -> None: collector = ReturnCollector(artifact_roots={"us_equity": root}, projects_root=root, store=store) with patch.object(PerformanceStore, "save_snapshot", autospec=True) as save: with self.assertRaisesRegex(RuntimeError, "No strategy return series found"): - run_monitor("us_equity", strategy_profile="synthetic_soxl", collector=collector, - store=store, windows=(2,), min_observations=2) - self.assertEqual(run_monitor( - "us_equity", collector=collector, store=store, min_observations=2, fail_on_empty=False, - ), []) + run_monitor( + "us_equity", + strategy_profile="synthetic_soxl", + collector=collector, + store=store, + windows=(2,), + min_observations=2, + source_revision=revision, + ) + self.assertEqual( + run_monitor( + "us_equity", + collector=collector, + store=store, + min_observations=2, + fail_on_empty=False, + source_revision=revision, + ), + [], + ) save.assert_not_called() def test_csv_collector_to_monitor_preserves_unique_daily_returns(self) -> None: + revision = "c" * 40 with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) pd.DataFrame({ @@ -107,8 +166,10 @@ def test_csv_collector_to_monitor_preserves_unique_daily_returns(self) -> None: "us_equity", strategy_profile="synthetic_soxl", collector=collector, store=store, windows=(2,), min_observations=2, require_explicit_benchmark=True, strategy_benchmarks={"synthetic_soxl": "SPY"}, + source_revision=revision, ) self.assertEqual(len(snapshots), 1) + self.assertEqual(snapshots[0].source_revision, revision) metrics = snapshots[0].windows[2] self.assertEqual(metrics.observation_count, 2) self.assertAlmostEqual(metrics.total_return, 1.01 * 0.98 - 1.0) @@ -133,6 +194,7 @@ def test_csv_collector_to_monitor_rejects_duplicate_required_benchmark(self) -> "us_equity", strategy_profile="synthetic_soxl", collector=collector, store=store, windows=(2,), min_observations=2, require_explicit_benchmark=True, strategy_benchmarks={"synthetic_soxl": "SPY"}, + source_revision="d" * 40, ) save.assert_not_called() diff --git a/tests/test_reusable_drift_workflow.py b/tests/test_reusable_drift_workflow.py index 630d7e7..4d38546 100644 --- a/tests/test_reusable_drift_workflow.py +++ b/tests/test_reusable_drift_workflow.py @@ -40,7 +40,7 @@ def test_reusable_drift_workflow_enforces_lifecycle_preflight() -> None: assert 'lifecycle_args+=(--strategy "${LIFECYCLE_STRATEGY_PROFILE}")' in workflow assert 'lifecycle_args+=(--live-stream-id "${LIFECYCLE_LIVE_STREAM_ID}")' in workflow assert "not paper/live account PnL" in workflow - assert 'quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} "${lifecycle_args[@]}"' in workflow + assert 'quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} --source-revision "${GITHUB_SHA}" "${lifecycle_args[@]}"' in workflow assert 'quant-lifecycle doctor --domain ${{ inputs.strategy_domain }} --require-snapshot --require-backtest --max-freshness-days 7 "${lifecycle_args[@]}"' in workflow assert 'quant-lifecycle drift --domain ${{ inputs.strategy_domain }} --no-alerts "${lifecycle_args[@]}"' in workflow assert 'repository: ${{ inputs.snapshot_repository }}' in workflow