From 3fa6471adfeee693c84b0dff208343ad277dda28 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 06:33:19 +0800 Subject: [PATCH 01/21] feat: support accepted baseline drift stores Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/cli.py | 13 ++++++++++++- .../strategy_lifecycle/drift_detector.py | 6 ++++-- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/cli.py b/src/quant_platform_kit/strategy_lifecycle/cli.py index bfdbde4..4e93eca 100644 --- a/src/quant_platform_kit/strategy_lifecycle/cli.py +++ b/src/quant_platform_kit/strategy_lifecycle/cli.py @@ -39,7 +39,17 @@ def _run_drift(args: argparse.Namespace) -> int: "quant_platform_kit.strategy_lifecycle.drift_detector", "run_drift_detection", ) - results = run_drift_detection(domain=args.domain, strategy_profile=args.strategy) + baseline_store = None + if getattr(args, "baseline_local_root", None): + from pathlib import Path + from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore + + baseline_store = PerformanceStore(local_root=Path(args.baseline_local_root)) + results = run_drift_detection( + domain=args.domain, + strategy_profile=args.strategy, + baseline_store=baseline_store, + ) critical_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "critical") review_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "review") _print(f"[drift] {len(results)} strategies checked, {critical_count} critical, {review_count} review") @@ -233,6 +243,7 @@ def build_parser() -> argparse.ArgumentParser: drift.add_argument("--strategy", default=None) drift.add_argument("--no-alerts", action="store_true") drift.add_argument("--dry-run-alerts", action="store_true") + drift.add_argument("--baseline-local-root", default=None) drift.set_defaults(func=_run_drift) optimize = subparsers.add_parser("optimize", help="Run parameter optimization for one strategy.") diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index eb4d971..ce6d4f4 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -134,9 +134,11 @@ def run_drift_detection( policy: DriftPolicy | None = None, fail_on_empty: bool = True, store: PerformanceStore | None = None, + baseline_store: PerformanceStore | None = None, ) -> list[DriftResult]: - """Run drift detection for all strategies in a domain.""" + """Run drift detection, optionally reading accepted baselines from another store.""" store = store or PerformanceStore.from_env() + baseline_store = baseline_store or store policy = policy or DriftPolicy.load_default() from quant_platform_kit.strategy_lifecycle.return_collector import ReturnCollector @@ -158,7 +160,7 @@ def run_drift_detection( if snapshot is None: missing_snapshots += 1 continue - backtest = store.load_latest_backtest(domain, profile) + backtest = baseline_store.load_latest_backtest(domain, profile) previous = store.load_latest_drift(domain, profile) result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) From 93d0516ccc106c01b6436f8608a51c8f116556b5 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 06:37:08 +0800 Subject: [PATCH 02/21] fix: keep drift status with accepted baseline Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/drift_detector.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index ce6d4f4..96ad35e 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -161,7 +161,7 @@ def run_drift_detection( missing_snapshots += 1 continue backtest = baseline_store.load_latest_backtest(domain, profile) - previous = store.load_latest_drift(domain, profile) + previous = baseline_store.load_latest_drift(domain, profile) result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) store.save_drift_result(result) From ccb0c433d27ba67a6ffe20ae0f9d7dd6cb6cdce7 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 06:42:23 +0800 Subject: [PATCH 03/21] fix: separate drift transition store from baseline store Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/drift_detector.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 96ad35e..ff9a2d0 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -135,10 +135,12 @@ def run_drift_detection( fail_on_empty: bool = True, store: PerformanceStore | None = None, baseline_store: PerformanceStore | None = None, + previous_drift_store: PerformanceStore | None = None, ) -> list[DriftResult]: - """Run drift detection, optionally reading accepted baselines from another store.""" + """Run drift detection with explicit baseline and transition-state stores.""" store = store or PerformanceStore.from_env() baseline_store = baseline_store or store + previous_drift_store = previous_drift_store or store policy = policy or DriftPolicy.load_default() from quant_platform_kit.strategy_lifecycle.return_collector import ReturnCollector @@ -161,7 +163,7 @@ def run_drift_detection( missing_snapshots += 1 continue backtest = baseline_store.load_latest_backtest(domain, profile) - previous = baseline_store.load_latest_drift(domain, profile) + previous = previous_drift_store.load_latest_drift(domain, profile) result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) store.save_drift_result(result) From 47f7fa917594d819f7818a6050948835d3f91617 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 06:50:22 +0800 Subject: [PATCH 04/21] fix: require explicit cross-lineage drift state Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/drift_detector.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index ff9a2d0..ef96fdc 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -140,7 +140,7 @@ def run_drift_detection( """Run drift detection with explicit baseline and transition-state stores.""" store = store or PerformanceStore.from_env() baseline_store = baseline_store or store - previous_drift_store = previous_drift_store or store + read_previous = previous_drift_store or (store if baseline_store is store else None) policy = policy or DriftPolicy.load_default() from quant_platform_kit.strategy_lifecycle.return_collector import ReturnCollector @@ -163,7 +163,7 @@ def run_drift_detection( missing_snapshots += 1 continue backtest = baseline_store.load_latest_backtest(domain, profile) - previous = previous_drift_store.load_latest_drift(domain, profile) + previous = read_previous.load_latest_drift(domain, profile) if read_previous else None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) store.save_drift_result(result) From 2d3c3cc8b4c6968f1a9ce94a9aed45d58520427b Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 06:57:22 +0800 Subject: [PATCH 05/21] fix: track drift baseline lineage Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/contracts.py | 2 ++ .../strategy_lifecycle/drift_detector.py | 8 ++++++-- .../strategy_lifecycle/performance_store.py | 2 ++ 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index c7db299..63288da 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -180,6 +180,7 @@ class DriftResult: status: DriftStatus dimensions: Mapping[str, DriftDimension] = field(default_factory=dict) previous_status: DriftStatus | None = None + baseline_param_set_id: str | None = None escalated: bool = False cooldown_active: bool = False alert_suppressed: bool = False @@ -193,6 +194,7 @@ def to_dict(self) -> dict[str, object]: "status": self.status.value, "dimensions": {k: v.to_dict() for k, v in self.dimensions.items()}, "previous_status": self.previous_status.value if self.previous_status else None, + "baseline_param_set_id": self.baseline_param_set_id, "escalated": self.escalated, "cooldown_active": self.cooldown_active, "alert_suppressed": self.alert_suppressed, diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index ef96fdc..3f91e18 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -123,7 +123,9 @@ def detect_drift( strategy_profile=snapshot.strategy_profile, domain=snapshot.domain, as_of=snapshot.as_of, drift_score=round(drift_score, 4), status=status, dimensions=dimensions, - previous_status=previous_status, escalated=escalated, + previous_status=previous_status, + baseline_param_set_id=backtest.param_set_id if backtest else None, + escalated=escalated, ) @@ -163,7 +165,9 @@ def run_drift_detection( missing_snapshots += 1 continue backtest = baseline_store.load_latest_backtest(domain, profile) - previous = read_previous.load_latest_drift(domain, profile) if read_previous else None + previous = store.load_latest_drift(domain, profile) + if baseline_store is not store and previous and previous.baseline_param_set_id != (backtest.param_set_id if backtest else None): + previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) store.save_drift_result(result) diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index fcc347a..2505d73 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -535,6 +535,8 @@ def _drift_from_dict(data: Mapping[str, Any]) -> DriftResult | None: drift_score=float(data.get("drift_score", 0)), status=DriftStatus(str(data.get("status", "healthy"))), dimensions=dimensions, + previous_status=DriftStatus(str(data["previous_status"])) if data.get("previous_status") else None, + baseline_param_set_id=str(data["baseline_param_set_id"]) if data.get("baseline_param_set_id") else None, ) except Exception: return None From a32919c17f50c186770876e07ffffa684e959709 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 06:58:32 +0800 Subject: [PATCH 06/21] fix: default drift transitions to active store Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/drift_detector.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 3f91e18..2509e05 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -142,7 +142,7 @@ def run_drift_detection( """Run drift detection with explicit baseline and transition-state stores.""" store = store or PerformanceStore.from_env() baseline_store = baseline_store or store - read_previous = previous_drift_store or (store if baseline_store is store else None) + read_previous = previous_drift_store or store policy = policy or DriftPolicy.load_default() from quant_platform_kit.strategy_lifecycle.return_collector import ReturnCollector @@ -165,7 +165,7 @@ def run_drift_detection( missing_snapshots += 1 continue backtest = baseline_store.load_latest_backtest(domain, profile) - previous = store.load_latest_drift(domain, profile) + previous = read_previous.load_latest_drift(domain, profile) if baseline_store is not store and previous and previous.baseline_param_set_id != (backtest.param_set_id if backtest else None): previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, From b54d4f01b640b9eb8e1956f52c2da1ef631f53e0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:04:51 +0800 Subject: [PATCH 07/21] fix: enforce drift baseline lineage Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/contracts.py | 2 +- src/quant_platform_kit/strategy_lifecycle/drift_detector.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index 63288da..5bfe61c 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -180,10 +180,10 @@ class DriftResult: status: DriftStatus dimensions: Mapping[str, DriftDimension] = field(default_factory=dict) previous_status: DriftStatus | None = None - baseline_param_set_id: str | None = None escalated: bool = False cooldown_active: bool = False alert_suppressed: bool = False + baseline_param_set_id: str | None = None def to_dict(self) -> dict[str, object]: return { diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 2509e05..07561ac 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -166,7 +166,7 @@ def run_drift_detection( continue backtest = baseline_store.load_latest_backtest(domain, profile) previous = read_previous.load_latest_drift(domain, profile) - if baseline_store is not store and previous and previous.baseline_param_set_id != (backtest.param_set_id if backtest else None): + if previous and previous.baseline_param_set_id != (backtest.param_set_id if backtest else None): previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) From 4db0efd87d5f4df6a09b7360a6fdbaca76559551 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:26:27 +0800 Subject: [PATCH 08/21] fix: preserve legacy drift transition state Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/drift_detector.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 07561ac..255f45d 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -166,7 +166,8 @@ def run_drift_detection( continue backtest = baseline_store.load_latest_backtest(domain, profile) previous = read_previous.load_latest_drift(domain, profile) - if previous and previous.baseline_param_set_id != (backtest.param_set_id if backtest else None): + current_baseline_id = backtest.param_set_id if backtest else None + if previous and previous.baseline_param_set_id and current_baseline_id and previous.baseline_param_set_id != current_baseline_id: previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) From 68473c958c5c28f10f9a7d74a9ff61ecdac100aa Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:28:48 +0800 Subject: [PATCH 09/21] fix: isolate explicit baseline transitions Co-Authored-By: Codex --- .../strategy_lifecycle/drift_detector.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 255f45d..3b1d0cf 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -141,6 +141,7 @@ def run_drift_detection( ) -> list[DriftResult]: """Run drift detection with explicit baseline and transition-state stores.""" store = store or PerformanceStore.from_env() + explicit_baseline_store = baseline_store is not None baseline_store = baseline_store or store read_previous = previous_drift_store or store policy = policy or DriftPolicy.load_default() @@ -167,8 +168,10 @@ def run_drift_detection( backtest = baseline_store.load_latest_backtest(domain, profile) previous = read_previous.load_latest_drift(domain, profile) current_baseline_id = backtest.param_set_id if backtest else None - if previous and previous.baseline_param_set_id and current_baseline_id and previous.baseline_param_set_id != current_baseline_id: - previous = None + if previous and current_baseline_id: + previous_baseline_id = previous.baseline_param_set_id + if (explicit_baseline_store and not previous_baseline_id) or (previous_baseline_id and previous_baseline_id != current_baseline_id): + previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) store.save_drift_result(result) From 6269a8fc77487278caddeb9bbc0952b96acb76f4 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:31:22 +0800 Subject: [PATCH 10/21] fix: require explicit baseline lineage match Co-Authored-By: Codex --- .../strategy_lifecycle/drift_detector.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 3b1d0cf..adc091a 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -168,9 +168,12 @@ def run_drift_detection( backtest = baseline_store.load_latest_backtest(domain, profile) previous = read_previous.load_latest_drift(domain, profile) current_baseline_id = backtest.param_set_id if backtest else None - if previous and current_baseline_id: + if previous: previous_baseline_id = previous.baseline_param_set_id - if (explicit_baseline_store and not previous_baseline_id) or (previous_baseline_id and previous_baseline_id != current_baseline_id): + if explicit_baseline_store: + if not (previous_baseline_id and current_baseline_id and previous_baseline_id == current_baseline_id): + previous = None + elif previous_baseline_id and current_baseline_id and previous_baseline_id != current_baseline_id: previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) From cb3ce5925d05ec9f8f63cf54171e01fc619665bd Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:35:47 +0800 Subject: [PATCH 11/21] fix: keep lifecycle drift invocation compatible Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/cli.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/quant_platform_kit/strategy_lifecycle/cli.py b/src/quant_platform_kit/strategy_lifecycle/cli.py index 4e93eca..f417350 100644 --- a/src/quant_platform_kit/strategy_lifecycle/cli.py +++ b/src/quant_platform_kit/strategy_lifecycle/cli.py @@ -49,6 +49,7 @@ def _run_drift(args: argparse.Namespace) -> int: domain=args.domain, strategy_profile=args.strategy, baseline_store=baseline_store, + baseline_lineage_policy="strict" if getattr(args, "strict_baseline_lineage", False) else "compatible", ) critical_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "critical") review_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "review") @@ -244,6 +245,7 @@ def build_parser() -> argparse.ArgumentParser: drift.add_argument("--no-alerts", action="store_true") drift.add_argument("--dry-run-alerts", action="store_true") drift.add_argument("--baseline-local-root", default=None) + drift.add_argument("--strict-baseline-lineage", action="store_true") drift.set_defaults(func=_run_drift) optimize = subparsers.add_parser("optimize", help="Run parameter optimization for one strategy.") From a8a8897ee12804debf5224235cf3b2f5cc7c82ac Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:40:51 +0800 Subject: [PATCH 12/21] fix: make drift lineage policy explicit Co-Authored-By: Codex --- .../strategy_lifecycle/drift_detector.py | 6 ++- tests/test_lifecycle_drift.py | 38 +++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index adc091a..892da87 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -138,10 +138,12 @@ def run_drift_detection( store: PerformanceStore | None = None, baseline_store: PerformanceStore | None = None, previous_drift_store: PerformanceStore | None = None, + baseline_lineage_policy: str = "compatible", ) -> list[DriftResult]: """Run drift detection with explicit baseline and transition-state stores.""" store = store or PerformanceStore.from_env() - explicit_baseline_store = baseline_store is not None + if baseline_lineage_policy not in {"compatible", "strict"}: + raise ValueError("baseline_lineage_policy must be compatible or strict") baseline_store = baseline_store or store read_previous = previous_drift_store or store policy = policy or DriftPolicy.load_default() @@ -170,7 +172,7 @@ def run_drift_detection( current_baseline_id = backtest.param_set_id if backtest else None if previous: previous_baseline_id = previous.baseline_param_set_id - if explicit_baseline_store: + if baseline_lineage_policy == "strict": if not (previous_baseline_id and current_baseline_id and previous_baseline_id == current_baseline_id): previous = None elif previous_baseline_id and current_baseline_id and previous_baseline_id != current_baseline_id: diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index 3af5b56..da0aa53 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -2,6 +2,7 @@ from __future__ import annotations +from dataclasses import replace from datetime import date import unittest from unittest import mock @@ -148,6 +149,43 @@ def collect(self, _domain: str) -> dict[str, pd.Series]: collector_factory.assert_called_once_with(store=sentinel_store) + def test_baseline_lineage_policy_handles_legacy_history_explicitly(self) -> None: + snapshot = _make_snapshot() + backtest = _make_backtest() + legacy_previous = replace( + detect_drift(snapshot, backtest=backtest), + baseline_param_set_id=None, + ) + + def run(policy: str, accepted_backtest: BacktestResult | None = backtest): + active_store = mock.Mock() + active_store.load_latest_snapshot.return_value = snapshot + active_store.load_latest_drift.return_value = legacy_previous + baseline_store = mock.Mock() + baseline_store.load_latest_backtest.return_value = accepted_backtest + + collector = mock.Mock() + collector.collect.return_value = {snapshot.strategy_profile: pd.Series([0.01])} + with mock.patch( + "quant_platform_kit.strategy_lifecycle.return_collector.ReturnCollector", + return_value=collector, + ): + result = run_drift_detection( + snapshot.domain, + strategy_profile=snapshot.strategy_profile, + store=active_store, + baseline_store=baseline_store, + baseline_lineage_policy=policy, + )[0] + return result + + self.assertEqual(run("compatible").previous_status, legacy_previous.status) + self.assertIsNone(run("strict").previous_status) + self.assertIsNone(run("strict", accepted_backtest=None).previous_status) + + with self.assertRaisesRegex(ValueError, "baseline_lineage_policy"): + run("unknown") + if __name__ == "__main__": unittest.main() From 6064d705043e825da8b863906a4353e59fa13792 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:44:37 +0800 Subject: [PATCH 13/21] fix: clear missing drift baseline lineage Co-Authored-By: Codex --- .../strategy_lifecycle/drift_detector.py | 2 +- tests/test_lifecycle_drift.py | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 892da87..7777bab 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -175,7 +175,7 @@ def run_drift_detection( if baseline_lineage_policy == "strict": if not (previous_baseline_id and current_baseline_id and previous_baseline_id == current_baseline_id): previous = None - elif previous_baseline_id and current_baseline_id and previous_baseline_id != current_baseline_id: + elif previous_baseline_id and current_baseline_id != previous_baseline_id: previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index da0aa53..5495250 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -157,10 +157,14 @@ def test_baseline_lineage_policy_handles_legacy_history_explicitly(self) -> None baseline_param_set_id=None, ) - def run(policy: str, accepted_backtest: BacktestResult | None = backtest): + def run( + policy: str, + accepted_backtest: BacktestResult | None = backtest, + previous=legacy_previous, + ): active_store = mock.Mock() active_store.load_latest_snapshot.return_value = snapshot - active_store.load_latest_drift.return_value = legacy_previous + active_store.load_latest_drift.return_value = previous baseline_store = mock.Mock() baseline_store.load_latest_backtest.return_value = accepted_backtest @@ -182,6 +186,14 @@ def run(policy: str, accepted_backtest: BacktestResult | None = backtest): self.assertEqual(run("compatible").previous_status, legacy_previous.status) self.assertIsNone(run("strict").previous_status) self.assertIsNone(run("strict", accepted_backtest=None).previous_status) + lineage_previous = detect_drift(snapshot, backtest=backtest) + self.assertIsNone( + run( + "compatible", + accepted_backtest=None, + previous=lineage_previous, + ).previous_status + ) with self.assertRaisesRegex(ValueError, "baseline_lineage_policy"): run("unknown") From 360f9e862a7eb88a2de88671b8723148358f6530 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 14/21] fix: default external drift baselines to strict lineage Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/cli.py | 2 +- .../strategy_lifecycle/drift_detector.py | 11 ++++++++--- tests/test_lifecycle_drift.py | 16 +++++++++++++--- 3 files changed, 22 insertions(+), 7 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/cli.py b/src/quant_platform_kit/strategy_lifecycle/cli.py index f417350..564cf17 100644 --- a/src/quant_platform_kit/strategy_lifecycle/cli.py +++ b/src/quant_platform_kit/strategy_lifecycle/cli.py @@ -49,7 +49,7 @@ def _run_drift(args: argparse.Namespace) -> int: domain=args.domain, strategy_profile=args.strategy, baseline_store=baseline_store, - baseline_lineage_policy="strict" if getattr(args, "strict_baseline_lineage", False) else "compatible", + baseline_lineage_policy="strict" if getattr(args, "strict_baseline_lineage", False) else "auto", ) critical_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "critical") review_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "review") diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 7777bab..4429cdf 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -138,12 +138,17 @@ def run_drift_detection( store: PerformanceStore | None = None, baseline_store: PerformanceStore | None = None, previous_drift_store: PerformanceStore | None = None, - baseline_lineage_policy: str = "compatible", + baseline_lineage_policy: str = "auto", ) -> list[DriftResult]: """Run drift detection with explicit baseline and transition-state stores.""" store = store or PerformanceStore.from_env() - if baseline_lineage_policy not in {"compatible", "strict"}: - raise ValueError("baseline_lineage_policy must be compatible or strict") + explicit_baseline_store = baseline_store is not None + if baseline_lineage_policy not in {"auto", "compatible", "strict"}: + raise ValueError("baseline_lineage_policy must be auto, compatible, or strict") + if explicit_baseline_store and baseline_lineage_policy == "compatible": + raise ValueError("compatible baseline lineage is not allowed with an external baseline store") + if baseline_lineage_policy == "auto": + baseline_lineage_policy = "strict" if explicit_baseline_store else "compatible" baseline_store = baseline_store or store read_previous = previous_drift_store or store policy = policy or DriftPolicy.load_default() diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index 5495250..e3b526e 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -161,10 +161,13 @@ def run( policy: str, accepted_backtest: BacktestResult | None = backtest, previous=legacy_previous, + *, + external_baseline: bool = True, ): active_store = mock.Mock() active_store.load_latest_snapshot.return_value = snapshot active_store.load_latest_drift.return_value = previous + active_store.load_latest_backtest.return_value = accepted_backtest baseline_store = mock.Mock() baseline_store.load_latest_backtest.return_value = accepted_backtest @@ -178,23 +181,30 @@ def run( snapshot.domain, strategy_profile=snapshot.strategy_profile, store=active_store, - baseline_store=baseline_store, + baseline_store=baseline_store if external_baseline else None, baseline_lineage_policy=policy, )[0] return result - self.assertEqual(run("compatible").previous_status, legacy_previous.status) + self.assertEqual( + run("auto", external_baseline=False).previous_status, + legacy_previous.status, + ) + self.assertIsNone(run("auto").previous_status) self.assertIsNone(run("strict").previous_status) self.assertIsNone(run("strict", accepted_backtest=None).previous_status) lineage_previous = detect_drift(snapshot, backtest=backtest) self.assertIsNone( run( - "compatible", + "auto", accepted_backtest=None, previous=lineage_previous, ).previous_status ) + with self.assertRaisesRegex(ValueError, "external baseline store"): + run("compatible") + with self.assertRaisesRegex(ValueError, "baseline_lineage_policy"): run("unknown") From b85478130b4829771cf14c861f420dc6b301b395 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:53:52 +0800 Subject: [PATCH 15/21] test: preserve drift result positional compatibility Co-Authored-By: Codex --- tests/test_lifecycle_contracts.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/tests/test_lifecycle_contracts.py b/tests/test_lifecycle_contracts.py index be51173..d38c8e9 100644 --- a/tests/test_lifecycle_contracts.py +++ b/tests/test_lifecycle_contracts.py @@ -136,6 +136,25 @@ def test_drift_result_escalated(self) -> None: ) self.assertTrue(drift.escalated) + def test_drift_result_preserves_legacy_positional_flags(self) -> None: + drift = DriftResult( + "t", + "us", + date(2026, 6, 1), + 0.6, + DriftStatus.REVIEW, + {}, + DriftStatus.WATCH, + True, + True, + True, + ) + + self.assertTrue(drift.escalated) + self.assertTrue(drift.cooldown_active) + self.assertTrue(drift.alert_suppressed) + self.assertIsNone(drift.baseline_param_set_id) + def test_backtest_result(self) -> None: bt = BacktestResult( strategy_profile="p", domain="us", param_set_id="default", From 1bcf86bb6765d664d9ab29d681e4a1672a4daa70 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:58:36 +0800 Subject: [PATCH 16/21] fix: preserve default drift transition history Co-Authored-By: Codex --- .../strategy_lifecycle/drift_detector.py | 2 -- tests/test_lifecycle_drift.py | 10 ++++++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 4429cdf..a5b1d1b 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -180,8 +180,6 @@ def run_drift_detection( if baseline_lineage_policy == "strict": if not (previous_baseline_id and current_baseline_id and previous_baseline_id == current_baseline_id): previous = None - elif previous_baseline_id and current_baseline_id != previous_baseline_id: - previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) store.save_drift_result(result) diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index e3b526e..645f8c2 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -194,6 +194,16 @@ def run( self.assertIsNone(run("strict").previous_status) self.assertIsNone(run("strict", accepted_backtest=None).previous_status) lineage_previous = detect_drift(snapshot, backtest=backtest) + rotated_backtest = replace(backtest, param_set_id="rotated-baseline") + self.assertEqual( + run( + "auto", + accepted_backtest=rotated_backtest, + previous=lineage_previous, + external_baseline=False, + ).previous_status, + lineage_previous.status, + ) self.assertIsNone( run( "auto", From cd3bf6280467f34326b31a76c527fb428bc741df Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:29:36 +0800 Subject: [PATCH 17/21] fix: preserve drift lineage during baseline migration Co-Authored-By: Codex --- .../strategy_lifecycle/cli.py | 10 +++++- .../strategy_lifecycle/drift_detector.py | 22 +++++++++--- tests/test_lifecycle_cli.py | 27 +++++++++++++++ tests/test_lifecycle_drift.py | 34 +++++++++++++++++++ 4 files changed, 88 insertions(+), 5 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/cli.py b/src/quant_platform_kit/strategy_lifecycle/cli.py index 564cf17..c24cdb0 100644 --- a/src/quant_platform_kit/strategy_lifecycle/cli.py +++ b/src/quant_platform_kit/strategy_lifecycle/cli.py @@ -45,11 +45,14 @@ def _run_drift(args: argparse.Namespace) -> int: from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore baseline_store = PerformanceStore(local_root=Path(args.baseline_local_root)) + baseline_lineage_policy = "migration" if getattr(args, "allow_legacy_baseline_history", False) else ( + "strict" if getattr(args, "strict_baseline_lineage", False) else "auto" + ) results = run_drift_detection( domain=args.domain, strategy_profile=args.strategy, baseline_store=baseline_store, - baseline_lineage_policy="strict" if getattr(args, "strict_baseline_lineage", False) else "auto", + baseline_lineage_policy=baseline_lineage_policy, ) critical_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "critical") review_count = sum(1 for item in results if getattr(getattr(item, "status", None), "value", None) == "review") @@ -246,6 +249,11 @@ def build_parser() -> argparse.ArgumentParser: drift.add_argument("--dry-run-alerts", action="store_true") drift.add_argument("--baseline-local-root", default=None) drift.add_argument("--strict-baseline-lineage", action="store_true") + drift.add_argument( + "--allow-legacy-baseline-history", + action="store_true", + help="One-time migration: reuse untagged prior drift status with an external accepted baseline.", + ) drift.set_defaults(func=_run_drift) optimize = subparsers.add_parser("optimize", help="Run parameter optimization for one strategy.") diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index a5b1d1b..51ba0ea 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -140,13 +140,20 @@ def run_drift_detection( previous_drift_store: PerformanceStore | None = None, baseline_lineage_policy: str = "auto", ) -> list[DriftResult]: - """Run drift detection with explicit baseline and transition-state stores.""" + """Run drift detection with explicit baseline and transition-state stores. + + External baselines are strict by default. ``migration`` is an explicit, + one-run compatibility mode for legacy drift history without a lineage ID; + it writes the accepted baseline ID into the next result. + """ store = store or PerformanceStore.from_env() explicit_baseline_store = baseline_store is not None - if baseline_lineage_policy not in {"auto", "compatible", "strict"}: - raise ValueError("baseline_lineage_policy must be auto, compatible, or strict") + if baseline_lineage_policy not in {"auto", "compatible", "migration", "strict"}: + raise ValueError("baseline_lineage_policy must be auto, compatible, migration, or strict") if explicit_baseline_store and baseline_lineage_policy == "compatible": raise ValueError("compatible baseline lineage is not allowed with an external baseline store") + if baseline_lineage_policy == "migration" and not explicit_baseline_store: + raise ValueError("migration baseline lineage requires an external baseline store") if baseline_lineage_policy == "auto": baseline_lineage_policy = "strict" if explicit_baseline_store else "compatible" baseline_store = baseline_store or store @@ -180,9 +187,16 @@ def run_drift_detection( if baseline_lineage_policy == "strict": if not (previous_baseline_id and current_baseline_id and previous_baseline_id == current_baseline_id): previous = None + elif baseline_lineage_policy == "migration": + if previous_baseline_id is None: + if current_baseline_id is None: + previous = None + elif not (current_baseline_id and previous_baseline_id == current_baseline_id): + previous = None result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) - store.save_drift_result(result) + if backtest is not None: + store.save_drift_result(result) results.append(result) if not results and fail_on_empty: raise RuntimeError( diff --git a/tests/test_lifecycle_cli.py b/tests/test_lifecycle_cli.py index 6b69ab6..e98f205 100644 --- a/tests/test_lifecycle_cli.py +++ b/tests/test_lifecycle_cli.py @@ -103,6 +103,33 @@ def fake_publish(events, **kwargs): self.assertEqual(result, 0) self.assertEqual(len(published["events"]), 2) + def test_drift_command_allows_explicit_legacy_baseline_migration(self) -> None: + observed = {} + + def fake_load_callable(_module_name: str, function_name: str): + if function_name == "run_drift_detection": + def fake_run_drift_detection(**kwargs): + observed.update(kwargs) + return [] + + return fake_run_drift_detection + if function_name == "build_drift_alert": + return lambda _result: None + if function_name == "publish_drift_alerts": + return lambda _events, **_kwargs: {} + raise AssertionError(function_name) + + with patch.object(cli, "_load_callable", fake_load_callable): + result = cli.main([ + "drift", + "--baseline-local-root", + "accepted-baselines", + "--allow-legacy-baseline-history", + ]) + + self.assertEqual(result, 0) + self.assertEqual(observed["baseline_lineage_policy"], "migration") + def test_update_returns_non_zero_for_error_stage(self) -> None: def fake_load_callable(_module_name: str, _function_name: str): return lambda **_kwargs: {"stage": "error", "reason": "missing proposal"} diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index 645f8c2..aff829b 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -193,6 +193,7 @@ def run( self.assertIsNone(run("auto").previous_status) self.assertIsNone(run("strict").previous_status) self.assertIsNone(run("strict", accepted_backtest=None).previous_status) + self.assertEqual(run("migration").previous_status, legacy_previous.status) lineage_previous = detect_drift(snapshot, backtest=backtest) rotated_backtest = replace(backtest, param_set_id="rotated-baseline") self.assertEqual( @@ -211,6 +212,13 @@ def run( previous=lineage_previous, ).previous_status ) + self.assertIsNone( + run( + "migration", + accepted_backtest=rotated_backtest, + previous=lineage_previous, + ).previous_status + ) with self.assertRaisesRegex(ValueError, "external baseline store"): run("compatible") @@ -218,6 +226,32 @@ def run( with self.assertRaisesRegex(ValueError, "baseline_lineage_policy"): run("unknown") + def test_missing_external_baseline_does_not_replace_lineage_history(self) -> None: + snapshot = _make_snapshot() + previous = detect_drift(snapshot, backtest=_make_backtest()) + active_store = mock.Mock() + active_store.load_latest_snapshot.return_value = snapshot + active_store.load_latest_drift.return_value = previous + baseline_store = mock.Mock() + baseline_store.load_latest_backtest.return_value = None + collector = mock.Mock() + collector.collect.return_value = {snapshot.strategy_profile: pd.Series([0.01])} + + with mock.patch( + "quant_platform_kit.strategy_lifecycle.return_collector.ReturnCollector", + return_value=collector, + ): + result = run_drift_detection( + snapshot.domain, + strategy_profile=snapshot.strategy_profile, + store=active_store, + baseline_store=baseline_store, + baseline_lineage_policy="strict", + )[0] + + self.assertIsNone(result.baseline_param_set_id) + active_store.save_drift_result.assert_not_called() + if __name__ == "__main__": unittest.main() From 6a1e27f7c9aa2ed7442ff39be1d9ebaca74fc812 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:36:48 +0800 Subject: [PATCH 18/21] fix: retain drift continuity across baseline gaps Co-Authored-By: Codex --- .../strategy_lifecycle/contracts.py | 2 + .../strategy_lifecycle/drift_detector.py | 41 +++++++++++++++---- .../strategy_lifecycle/performance_store.py | 1 + tests/test_lifecycle_contracts.py | 21 ++++++++++ tests/test_lifecycle_drift.py | 39 ++++++++++++------ 5 files changed, 84 insertions(+), 20 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index 5bfe61c..f9c4b2a 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -184,6 +184,7 @@ class DriftResult: cooldown_active: bool = False alert_suppressed: bool = False baseline_param_set_id: str | None = None + baseline_available: bool = True def to_dict(self) -> dict[str, object]: return { @@ -195,6 +196,7 @@ def to_dict(self) -> dict[str, object]: "dimensions": {k: v.to_dict() for k, v in self.dimensions.items()}, "previous_status": self.previous_status.value if self.previous_status else None, "baseline_param_set_id": self.baseline_param_set_id, + "baseline_available": self.baseline_available, "escalated": self.escalated, "cooldown_active": self.cooldown_active, "alert_suppressed": self.alert_suppressed, diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 51ba0ea..4fe1ab8 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -3,6 +3,8 @@ from __future__ import annotations +from dataclasses import replace + import numpy as np from quant_platform_kit.strategy_lifecycle.contracts import ( @@ -181,6 +183,7 @@ def run_drift_detection( continue backtest = baseline_store.load_latest_backtest(domain, profile) previous = read_previous.load_latest_drift(domain, profile) + previous_before_lineage_check = previous current_baseline_id = backtest.param_set_id if backtest else None if previous: previous_baseline_id = previous.baseline_param_set_id @@ -188,15 +191,37 @@ def run_drift_detection( if not (previous_baseline_id and current_baseline_id and previous_baseline_id == current_baseline_id): previous = None elif baseline_lineage_policy == "migration": - if previous_baseline_id is None: - if current_baseline_id is None: - previous = None - elif not (current_baseline_id and previous_baseline_id == current_baseline_id): + if current_baseline_id is None or ( + previous_baseline_id is not None and previous_baseline_id != current_baseline_id + ): previous = None - result = detect_drift(snapshot, backtest=backtest, policy=policy, - previous_status=previous.status if previous else None) - if backtest is not None: - store.save_drift_result(result) + if backtest is None: + continuity_result = ( + previous_before_lineage_check + if baseline_lineage_policy != "migration" + and previous_before_lineage_check is not None + and previous_before_lineage_check.baseline_param_set_id + else None + ) + if continuity_result is not None: + result = replace( + continuity_result, + as_of=snapshot.as_of, + previous_status=continuity_result.status, + escalated=False, + alert_suppressed=True, + baseline_available=False, + ) + else: + result = replace( + detect_drift(snapshot, backtest=None, policy=policy), + alert_suppressed=True, + baseline_available=False, + ) + else: + result = detect_drift(snapshot, backtest=backtest, policy=policy, + previous_status=previous.status if previous else None) + store.save_drift_result(result) results.append(result) if not results and fail_on_empty: raise RuntimeError( diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index 2505d73..154d338 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -537,6 +537,7 @@ def _drift_from_dict(data: Mapping[str, Any]) -> DriftResult | None: dimensions=dimensions, previous_status=DriftStatus(str(data["previous_status"])) if data.get("previous_status") else None, baseline_param_set_id=str(data["baseline_param_set_id"]) if data.get("baseline_param_set_id") else None, + baseline_available=bool(data.get("baseline_available", True)), ) except Exception: return None diff --git a/tests/test_lifecycle_contracts.py b/tests/test_lifecycle_contracts.py index d38c8e9..9d4a2ac 100644 --- a/tests/test_lifecycle_contracts.py +++ b/tests/test_lifecycle_contracts.py @@ -19,6 +19,7 @@ UpdateStage, WindowPerformance, ) +from quant_platform_kit.strategy_lifecycle.performance_store import _drift_from_dict class ContractsTests(unittest.TestCase): @@ -154,6 +155,26 @@ def test_drift_result_preserves_legacy_positional_flags(self) -> None: self.assertTrue(drift.cooldown_active) self.assertTrue(drift.alert_suppressed) self.assertIsNone(drift.baseline_param_set_id) + self.assertTrue(drift.baseline_available) + self.assertTrue(drift.to_dict()["baseline_available"]) + + def test_drift_result_round_trips_baseline_availability(self) -> None: + drift = DriftResult( + strategy_profile="t", + domain="us", + as_of=date(2026, 6, 1), + drift_score=0.6, + status=DriftStatus.REVIEW, + baseline_param_set_id="accepted-v1", + baseline_available=False, + ) + + restored = _drift_from_dict(drift.to_dict()) + + self.assertIsNotNone(restored) + assert restored is not None + self.assertEqual(restored.baseline_param_set_id, "accepted-v1") + self.assertFalse(restored.baseline_available) def test_backtest_result(self) -> None: bt = BacktestResult( diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index aff829b..74f29f3 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -194,6 +194,7 @@ def run( self.assertIsNone(run("strict").previous_status) self.assertIsNone(run("strict", accepted_backtest=None).previous_status) self.assertEqual(run("migration").previous_status, legacy_previous.status) + self.assertIsNone(run("migration", accepted_backtest=None).previous_status) lineage_previous = detect_drift(snapshot, backtest=backtest) rotated_backtest = replace(backtest, param_set_id="rotated-baseline") self.assertEqual( @@ -205,13 +206,13 @@ def run( ).previous_status, lineage_previous.status, ) - self.assertIsNone( - run( - "auto", - accepted_backtest=None, - previous=lineage_previous, - ).previous_status + missing_baseline_history = run( + "auto", + accepted_backtest=None, + previous=lineage_previous, ) + self.assertEqual(missing_baseline_history.previous_status, lineage_previous.status) + self.assertFalse(missing_baseline_history.baseline_available) self.assertIsNone( run( "migration", @@ -226,14 +227,14 @@ def run( with self.assertRaisesRegex(ValueError, "baseline_lineage_policy"): run("unknown") - def test_missing_external_baseline_does_not_replace_lineage_history(self) -> None: + def test_missing_external_baseline_preserves_lineage_history_without_alerting(self) -> None: snapshot = _make_snapshot() previous = detect_drift(snapshot, backtest=_make_backtest()) active_store = mock.Mock() active_store.load_latest_snapshot.return_value = snapshot - active_store.load_latest_drift.return_value = previous + active_store.load_latest_drift.side_effect = [previous, None] baseline_store = mock.Mock() - baseline_store.load_latest_backtest.return_value = None + baseline_store.load_latest_backtest.side_effect = [None, _make_backtest()] collector = mock.Mock() collector.collect.return_value = {snapshot.strategy_profile: pd.Series([0.01])} @@ -241,7 +242,16 @@ def test_missing_external_baseline_does_not_replace_lineage_history(self) -> Non "quant_platform_kit.strategy_lifecycle.return_collector.ReturnCollector", return_value=collector, ): - result = run_drift_detection( + missing_baseline_result = run_drift_detection( + snapshot.domain, + strategy_profile=snapshot.strategy_profile, + store=active_store, + baseline_store=baseline_store, + baseline_lineage_policy="strict", + )[0] + active_store.load_latest_drift.side_effect = None + active_store.load_latest_drift.return_value = missing_baseline_result + recovered_result = run_drift_detection( snapshot.domain, strategy_profile=snapshot.strategy_profile, store=active_store, @@ -249,8 +259,13 @@ def test_missing_external_baseline_does_not_replace_lineage_history(self) -> Non baseline_lineage_policy="strict", )[0] - self.assertIsNone(result.baseline_param_set_id) - active_store.save_drift_result.assert_not_called() + self.assertEqual(missing_baseline_result.baseline_param_set_id, previous.baseline_param_set_id) + self.assertFalse(missing_baseline_result.baseline_available) + self.assertEqual(missing_baseline_result.status, previous.status) + self.assertTrue(missing_baseline_result.alert_suppressed) + self.assertEqual(recovered_result.previous_status, previous.status) + self.assertTrue(recovered_result.baseline_available) + self.assertEqual(active_store.save_drift_result.call_count, 2) if __name__ == "__main__": From 215308f2d47d80c57666477906fc40488e7ddfc1 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 16:43:38 +0800 Subject: [PATCH 19/21] fix: mark baseline gaps without stale drift metrics Co-Authored-By: Codex --- .../strategy_lifecycle/drift_detector.py | 9 ++++++--- tests/test_lifecycle_drift.py | 2 ++ 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 4fe1ab8..411b805 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -204,12 +204,15 @@ def run_drift_detection( else None ) if continuity_result is not None: - result = replace( - continuity_result, + result = DriftResult( + strategy_profile=snapshot.strategy_profile, + domain=snapshot.domain, as_of=snapshot.as_of, + drift_score=0.0, + status=continuity_result.status, previous_status=continuity_result.status, - escalated=False, alert_suppressed=True, + baseline_param_set_id=continuity_result.baseline_param_set_id, baseline_available=False, ) else: diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index 74f29f3..1b301ac 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -263,6 +263,8 @@ def test_missing_external_baseline_preserves_lineage_history_without_alerting(se self.assertFalse(missing_baseline_result.baseline_available) self.assertEqual(missing_baseline_result.status, previous.status) self.assertTrue(missing_baseline_result.alert_suppressed) + self.assertEqual(missing_baseline_result.drift_score, 0.0) + self.assertEqual(missing_baseline_result.dimensions, {}) self.assertEqual(recovered_result.previous_status, previous.status) self.assertTrue(recovered_result.baseline_available) self.assertEqual(active_store.save_drift_result.call_count, 2) From c5556bf46ec67615890f7f417b4ef144db5291e9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 17:14:49 +0800 Subject: [PATCH 20/21] fix: retain remote baseline and canonical daily drift Co-Authored-By: Codex --- src/quant_platform_kit/strategy_lifecycle/cli.py | 6 +++++- .../strategy_lifecycle/drift_detector.py | 6 +++++- tests/test_lifecycle_cli.py | 12 +++++++++++- tests/test_lifecycle_drift.py | 4 ++-- 4 files changed, 23 insertions(+), 5 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/cli.py b/src/quant_platform_kit/strategy_lifecycle/cli.py index c24cdb0..1aa4678 100644 --- a/src/quant_platform_kit/strategy_lifecycle/cli.py +++ b/src/quant_platform_kit/strategy_lifecycle/cli.py @@ -6,6 +6,7 @@ import importlib import sys from collections.abc import Callable, Sequence +from dataclasses import replace from typing import Any @@ -44,7 +45,10 @@ def _run_drift(args: argparse.Namespace) -> int: from pathlib import Path from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore - baseline_store = PerformanceStore(local_root=Path(args.baseline_local_root)) + baseline_store = replace( + PerformanceStore.from_env(), + local_root=Path(args.baseline_local_root), + ) baseline_lineage_policy = "migration" if getattr(args, "allow_legacy_baseline_history", False) else ( "strict" if getattr(args, "strict_baseline_lineage", False) else "auto" ) diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 411b805..0ec0c17 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -224,7 +224,11 @@ def run_drift_detection( else: result = detect_drift(snapshot, backtest=backtest, policy=policy, previous_status=previous.status if previous else None) - store.save_drift_result(result) + if backtest is not None or not ( + previous_before_lineage_check is not None + and previous_before_lineage_check.as_of == snapshot.as_of + ): + store.save_drift_result(result) results.append(result) if not results and fail_on_empty: raise RuntimeError( diff --git a/tests/test_lifecycle_cli.py b/tests/test_lifecycle_cli.py index e98f205..6540823 100644 --- a/tests/test_lifecycle_cli.py +++ b/tests/test_lifecycle_cli.py @@ -7,6 +7,7 @@ from unittest.mock import patch from quant_platform_kit.strategy_lifecycle import cli +from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore class LifecycleCliTests(unittest.TestCase): @@ -119,7 +120,14 @@ def fake_run_drift_detection(**kwargs): return lambda _events, **_kwargs: {} raise AssertionError(function_name) - with patch.object(cli, "_load_callable", fake_load_callable): + environment_store = PerformanceStore( + cloud_bucket="accepted-bucket", + cloud_prefix="lifecycle", + ) + with ( + patch.object(cli, "_load_callable", fake_load_callable), + patch.object(PerformanceStore, "from_env", return_value=environment_store), + ): result = cli.main([ "drift", "--baseline-local-root", @@ -129,6 +137,8 @@ def fake_run_drift_detection(**kwargs): self.assertEqual(result, 0) self.assertEqual(observed["baseline_lineage_policy"], "migration") + self.assertEqual(observed["baseline_store"].cloud_bucket, "accepted-bucket") + self.assertEqual(observed["baseline_store"].cloud_prefix, "lifecycle") def test_update_returns_non_zero_for_error_stage(self) -> None: def fake_load_callable(_module_name: str, _function_name: str): diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index 1b301ac..6dbeaf0 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -250,7 +250,7 @@ def test_missing_external_baseline_preserves_lineage_history_without_alerting(se baseline_lineage_policy="strict", )[0] active_store.load_latest_drift.side_effect = None - active_store.load_latest_drift.return_value = missing_baseline_result + active_store.load_latest_drift.return_value = previous recovered_result = run_drift_detection( snapshot.domain, strategy_profile=snapshot.strategy_profile, @@ -267,7 +267,7 @@ def test_missing_external_baseline_preserves_lineage_history_without_alerting(se self.assertEqual(missing_baseline_result.dimensions, {}) self.assertEqual(recovered_result.previous_status, previous.status) self.assertTrue(recovered_result.baseline_available) - self.assertEqual(active_store.save_drift_result.call_count, 2) + active_store.save_drift_result.assert_called_once_with(recovered_result) if __name__ == "__main__": From a306ab9466896fac34338227c01b08a7525ca538 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 17:30:43 +0800 Subject: [PATCH 21/21] fix: harden accepted baseline drift handling Co-Authored-By: Codex --- .../strategy_lifecycle/cli.py | 86 ++++++++++++++----- .../strategy_lifecycle/codex_integration.py | 2 +- .../strategy_lifecycle/contracts.py | 4 + .../strategy_lifecycle/drift_detector.py | 44 +++++++++- .../strategy_lifecycle/performance_store.py | 10 +++ tests/test_lifecycle_cli.py | 51 +++++++++-- tests/test_lifecycle_codex_integration.py | 33 +++++++ tests/test_lifecycle_contracts.py | 4 + tests/test_lifecycle_drift.py | 12 +++ 9 files changed, 217 insertions(+), 29 deletions(-) create mode 100644 tests/test_lifecycle_codex_integration.py diff --git a/src/quant_platform_kit/strategy_lifecycle/cli.py b/src/quant_platform_kit/strategy_lifecycle/cli.py index 1aa4678..24705bf 100644 --- a/src/quant_platform_kit/strategy_lifecycle/cli.py +++ b/src/quant_platform_kit/strategy_lifecycle/cli.py @@ -6,7 +6,7 @@ import importlib import sys from collections.abc import Callable, Sequence -from dataclasses import replace +from pathlib import Path from typing import Any @@ -34,24 +34,71 @@ def _run_monitor(args: argparse.Namespace) -> int: return 0 +def _parse_baseline_bucket(value: str) -> tuple[str, str]: + location = value.strip() + if location.startswith("gs://"): + location = location[5:] + bucket, _, prefix = location.partition("/") + if not bucket: + raise ValueError("baseline bucket must include a bucket name") + return bucket, prefix.strip("/") + + +def _baseline_store_from_args(args: argparse.Namespace): + from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore + + local_root = getattr(args, "baseline_local_root", None) + bucket_value = getattr(args, "baseline_bucket", None) + if not local_root and not bucket_value: + return None + if not bucket_value: + return PerformanceStore(local_root=Path(local_root)) + + bucket, prefix = _parse_baseline_bucket(bucket_value) + environment_store = PerformanceStore.from_env() + return PerformanceStore( + cloud_bucket=bucket, + cloud_prefix=prefix, + local_root=Path(local_root) if local_root else None, + project_id=environment_store.project_id, + client_factory=environment_store.client_factory, + ) + + +def _baseline_lineage_policy_from_args(args: argparse.Namespace) -> str: + allow_legacy = getattr(args, "allow_legacy_baseline_history", False) + strict = getattr(args, "strict_baseline_lineage", False) + if allow_legacy and strict: + raise ValueError("baseline lineage flags are mutually exclusive") + if allow_legacy: + return "migration" + return "strict" if strict else "auto" + + +def _add_baseline_options(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--baseline-local-root", default=None) + parser.add_argument( + "--baseline-bucket", + default=None, + help="Accepted-baseline bucket or gs://bucket/prefix URI.", + ) + lineage = parser.add_mutually_exclusive_group() + lineage.add_argument("--strict-baseline-lineage", action="store_true") + lineage.add_argument( + "--allow-legacy-baseline-history", + action="store_true", + help="One-time migration: reuse untagged prior drift status with an external accepted baseline.", + ) + + def _run_drift(args: argparse.Namespace) -> int: _print(f"[drift] Running drift detection for domain={args.domain}") run_drift_detection = _load_callable( "quant_platform_kit.strategy_lifecycle.drift_detector", "run_drift_detection", ) - baseline_store = None - if getattr(args, "baseline_local_root", None): - from pathlib import Path - from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore - - baseline_store = replace( - PerformanceStore.from_env(), - local_root=Path(args.baseline_local_root), - ) - baseline_lineage_policy = "migration" if getattr(args, "allow_legacy_baseline_history", False) else ( - "strict" if getattr(args, "strict_baseline_lineage", False) else "auto" - ) + baseline_store = _baseline_store_from_args(args) + baseline_lineage_policy = _baseline_lineage_policy_from_args(args) results = run_drift_detection( domain=args.domain, strategy_profile=args.strategy, @@ -153,6 +200,10 @@ def _run_lifecycle(args: argparse.Namespace) -> int: strategy=None, no_alerts=args.no_alerts, dry_run_alerts=args.dry_run_alerts, + baseline_local_root=getattr(args, "baseline_local_root", None), + baseline_bucket=getattr(args, "baseline_bucket", None), + strict_baseline_lineage=getattr(args, "strict_baseline_lineage", False), + allow_legacy_baseline_history=getattr(args, "allow_legacy_baseline_history", False), ) ) if drift_status != 0: @@ -251,13 +302,7 @@ def build_parser() -> argparse.ArgumentParser: drift.add_argument("--strategy", default=None) drift.add_argument("--no-alerts", action="store_true") drift.add_argument("--dry-run-alerts", action="store_true") - drift.add_argument("--baseline-local-root", default=None) - drift.add_argument("--strict-baseline-lineage", action="store_true") - drift.add_argument( - "--allow-legacy-baseline-history", - action="store_true", - help="One-time migration: reuse untagged prior drift status with an external accepted baseline.", - ) + _add_baseline_options(drift) drift.set_defaults(func=_run_drift) optimize = subparsers.add_parser("optimize", help="Run parameter optimization for one strategy.") @@ -309,6 +354,7 @@ def build_parser() -> argparse.ArgumentParser: lifecycle.add_argument("--skip-optimization", action="store_true") lifecycle.add_argument("--no-alerts", action="store_true") lifecycle.add_argument("--dry-run-alerts", action="store_true") + _add_baseline_options(lifecycle) lifecycle.set_defaults(func=_run_lifecycle) return parser diff --git a/src/quant_platform_kit/strategy_lifecycle/codex_integration.py b/src/quant_platform_kit/strategy_lifecycle/codex_integration.py index 0351f2f..398f82a 100644 --- a/src/quant_platform_kit/strategy_lifecycle/codex_integration.py +++ b/src/quant_platform_kit/strategy_lifecycle/codex_integration.py @@ -403,7 +403,7 @@ def _run_drift_phase(domain: str, store: PerformanceStore) -> tuple[list, list]: """Phase 2: run drift detection, return (all_drifts, alerting_drifts).""" from quant_platform_kit.strategy_lifecycle.drift_detector import run_drift_detection drifts = run_drift_detection(domain, store=store) - alerts = [d for d in drifts if d.status != DriftStatus.HEALTHY] + alerts = [d for d in drifts if d.status != DriftStatus.HEALTHY and not d.alert_suppressed] return drifts, alerts diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index f9c4b2a..89e81a5 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -185,6 +185,8 @@ class DriftResult: alert_suppressed: bool = False baseline_param_set_id: str | None = None baseline_available: bool = True + baseline_param_version: int | None = None + baseline_artifact_id: str | None = None def to_dict(self) -> dict[str, object]: return { @@ -197,6 +199,8 @@ def to_dict(self) -> dict[str, object]: "previous_status": self.previous_status.value if self.previous_status else None, "baseline_param_set_id": self.baseline_param_set_id, "baseline_available": self.baseline_available, + "baseline_param_version": self.baseline_param_version, + "baseline_artifact_id": self.baseline_artifact_id, "escalated": self.escalated, "cooldown_active": self.cooldown_active, "alert_suppressed": self.alert_suppressed, diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py index 0ec0c17..359a2be 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_detector.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_detector.py @@ -34,6 +34,12 @@ ] +def _baseline_artifact_id(backtest: BacktestResult | None) -> str | None: + if backtest is None: + return None + return backtest.run_id or backtest.computed_at or None + + def _compute_dimension( key: str, metric: str, actual_val: float, expected_val: float, threshold: float, @@ -127,6 +133,8 @@ def detect_drift( status=status, dimensions=dimensions, previous_status=previous_status, baseline_param_set_id=backtest.param_set_id if backtest else None, + baseline_param_version=backtest.param_version if backtest else None, + baseline_artifact_id=_baseline_artifact_id(backtest), escalated=escalated, ) @@ -149,7 +157,7 @@ def run_drift_detection( it writes the accepted baseline ID into the next result. """ store = store or PerformanceStore.from_env() - explicit_baseline_store = baseline_store is not None + explicit_baseline_store = baseline_store is not None and baseline_store is not store if baseline_lineage_policy not in {"auto", "compatible", "migration", "strict"}: raise ValueError("baseline_lineage_policy must be auto, compatible, migration, or strict") if explicit_baseline_store and baseline_lineage_policy == "compatible": @@ -185,14 +193,34 @@ def run_drift_detection( previous = read_previous.load_latest_drift(domain, profile) previous_before_lineage_check = previous current_baseline_id = backtest.param_set_id if backtest else None + current_baseline_version = backtest.param_version if backtest else None + current_baseline_artifact_id = _baseline_artifact_id(backtest) if previous: previous_baseline_id = previous.baseline_param_set_id + previous_baseline_version = previous.baseline_param_version + previous_baseline_artifact_id = previous.baseline_artifact_id if baseline_lineage_policy == "strict": - if not (previous_baseline_id and current_baseline_id and previous_baseline_id == current_baseline_id): + if not ( + previous_baseline_id + and current_baseline_id + and previous_baseline_id == current_baseline_id + and previous_baseline_version is not None + and previous_baseline_version == current_baseline_version + and ( + not current_baseline_artifact_id + or previous_baseline_artifact_id == current_baseline_artifact_id + ) + ): previous = None elif baseline_lineage_policy == "migration": if current_baseline_id is None or ( previous_baseline_id is not None and previous_baseline_id != current_baseline_id + ) or ( + previous_baseline_version is not None + and previous_baseline_version != current_baseline_version + ) or ( + previous_baseline_artifact_id is not None + and previous_baseline_artifact_id != current_baseline_artifact_id ): previous = None if backtest is None: @@ -213,6 +241,18 @@ def run_drift_detection( previous_status=continuity_result.status, alert_suppressed=True, baseline_param_set_id=continuity_result.baseline_param_set_id, + baseline_param_version=continuity_result.baseline_param_version, + baseline_artifact_id=continuity_result.baseline_artifact_id, + baseline_available=False, + ) + elif explicit_baseline_store: + result = DriftResult( + strategy_profile=snapshot.strategy_profile, + domain=snapshot.domain, + as_of=snapshot.as_of, + drift_score=0.0, + status=DriftStatus.REVIEW, + alert_suppressed=True, baseline_available=False, ) else: diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index 154d338..b009811 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -538,6 +538,16 @@ def _drift_from_dict(data: Mapping[str, Any]) -> DriftResult | None: previous_status=DriftStatus(str(data["previous_status"])) if data.get("previous_status") else None, baseline_param_set_id=str(data["baseline_param_set_id"]) if data.get("baseline_param_set_id") else None, baseline_available=bool(data.get("baseline_available", True)), + baseline_param_version=( + int(data["baseline_param_version"]) + if data.get("baseline_param_version") is not None + else None + ), + baseline_artifact_id=( + str(data["baseline_artifact_id"]) + if data.get("baseline_artifact_id") + else None + ), ) except Exception: return None diff --git a/tests/test_lifecycle_cli.py b/tests/test_lifecycle_cli.py index 6540823..d655e93 100644 --- a/tests/test_lifecycle_cli.py +++ b/tests/test_lifecycle_cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import unittest +from pathlib import Path from types import SimpleNamespace from unittest.mock import patch @@ -120,10 +121,7 @@ def fake_run_drift_detection(**kwargs): return lambda _events, **_kwargs: {} raise AssertionError(function_name) - environment_store = PerformanceStore( - cloud_bucket="accepted-bucket", - cloud_prefix="lifecycle", - ) + environment_store = PerformanceStore(cloud_bucket="candidate-bucket", cloud_prefix="candidate") with ( patch.object(cli, "_load_callable", fake_load_callable), patch.object(PerformanceStore, "from_env", return_value=environment_store), @@ -137,8 +135,33 @@ def fake_run_drift_detection(**kwargs): self.assertEqual(result, 0) self.assertEqual(observed["baseline_lineage_policy"], "migration") + self.assertEqual(observed["baseline_store"].local_root, Path("accepted-baselines")) + self.assertEqual(observed["baseline_store"].cloud_bucket, "") + self.assertEqual(observed["baseline_store"].cloud_prefix, "") + + def test_drift_command_uses_explicit_baseline_bucket(self) -> None: + observed = {} + + def fake_load_callable(_module_name: str, function_name: str): + if function_name == "run_drift_detection": + def fake_run_drift_detection(**kwargs): + observed.update(kwargs) + return [] + + return fake_run_drift_detection + if function_name == "build_drift_alert": + return lambda _result: None + if function_name == "publish_drift_alerts": + return lambda _events, **_kwargs: {} + raise AssertionError(function_name) + + with patch.object(cli, "_load_callable", fake_load_callable): + result = cli.main(["drift", "--baseline-bucket", "gs://accepted-bucket/lifecycle"]) + + self.assertEqual(result, 0) self.assertEqual(observed["baseline_store"].cloud_bucket, "accepted-bucket") self.assertEqual(observed["baseline_store"].cloud_prefix, "lifecycle") + self.assertIsNone(observed["baseline_store"].local_root) def test_update_returns_non_zero_for_error_stage(self) -> None: def fake_load_callable(_module_name: str, _function_name: str): @@ -200,6 +223,7 @@ def fake_load_callable(_module_name: str, _function_name: str): def test_lifecycle_command_runs_real_steps(self) -> None: calls = [] + drift_kwargs = {} def fake_load_callable(_module_name: str, function_name: str): if function_name == "run_monitor": @@ -209,8 +233,9 @@ def fake_monitor(**_kwargs): return fake_monitor if function_name == "run_drift_detection": - def fake_drift(**_kwargs): + def fake_drift(**kwargs): calls.append("drift") + drift_kwargs.update(kwargs) return [] return fake_drift @@ -227,10 +252,24 @@ def fake_dashboard(**_kwargs): raise AssertionError(function_name) with patch.object(cli, "_load_callable", fake_load_callable): - result = cli.main(["lifecycle", "--domain", "cn_equity", "--skip-optimization"]) + result = cli.main([ + "lifecycle", + "--domain", + "cn_equity", + "--skip-optimization", + "--baseline-local-root", + "accepted-baselines", + "--strict-baseline-lineage", + ]) self.assertEqual(result, 0) self.assertEqual(calls, ["monitor", "drift", "dashboard"]) + self.assertEqual(drift_kwargs["baseline_store"].local_root, Path("accepted-baselines")) + self.assertEqual(drift_kwargs["baseline_lineage_policy"], "strict") + + def test_drift_rejects_conflicting_baseline_lineage_flags(self) -> None: + with self.assertRaises(SystemExit): + cli.main(["drift", "--strict-baseline-lineage", "--allow-legacy-baseline-history"]) def test_error_returns_non_zero(self) -> None: def fake_load_callable(_module_name: str, _function_name: str): diff --git a/tests/test_lifecycle_codex_integration.py b/tests/test_lifecycle_codex_integration.py new file mode 100644 index 0000000..d2c8569 --- /dev/null +++ b/tests/test_lifecycle_codex_integration.py @@ -0,0 +1,33 @@ +from datetime import date +from unittest.mock import Mock, patch + +from quant_platform_kit.strategy_lifecycle.codex_integration import _run_drift_phase +from quant_platform_kit.strategy_lifecycle.contracts import DriftResult, DriftStatus + + +def test_drift_phase_excludes_suppressed_results_from_automation() -> None: + suppressed = DriftResult( + strategy_profile="missing-baseline", + domain="us_equity", + as_of=date(2026, 7, 11), + drift_score=0.0, + status=DriftStatus.REVIEW, + alert_suppressed=True, + baseline_available=False, + ) + critical = DriftResult( + strategy_profile="active-baseline", + domain="us_equity", + as_of=date(2026, 7, 11), + drift_score=0.8, + status=DriftStatus.CRITICAL, + ) + + with patch( + "quant_platform_kit.strategy_lifecycle.drift_detector.run_drift_detection", + return_value=[suppressed, critical], + ): + drifts, alerts = _run_drift_phase("us_equity", Mock()) + + assert drifts == [suppressed, critical] + assert alerts == [critical] diff --git a/tests/test_lifecycle_contracts.py b/tests/test_lifecycle_contracts.py index 9d4a2ac..50bf126 100644 --- a/tests/test_lifecycle_contracts.py +++ b/tests/test_lifecycle_contracts.py @@ -166,6 +166,8 @@ def test_drift_result_round_trips_baseline_availability(self) -> None: drift_score=0.6, status=DriftStatus.REVIEW, baseline_param_set_id="accepted-v1", + baseline_param_version=2, + baseline_artifact_id="accepted-run-2", baseline_available=False, ) @@ -174,6 +176,8 @@ def test_drift_result_round_trips_baseline_availability(self) -> None: self.assertIsNotNone(restored) assert restored is not None self.assertEqual(restored.baseline_param_set_id, "accepted-v1") + self.assertEqual(restored.baseline_param_version, 2) + self.assertEqual(restored.baseline_artifact_id, "accepted-run-2") self.assertFalse(restored.baseline_available) def test_backtest_result(self) -> None: diff --git a/tests/test_lifecycle_drift.py b/tests/test_lifecycle_drift.py index 6dbeaf0..13bf2d2 100644 --- a/tests/test_lifecycle_drift.py +++ b/tests/test_lifecycle_drift.py @@ -213,6 +213,18 @@ def run( ) self.assertEqual(missing_baseline_history.previous_status, lineage_previous.status) self.assertFalse(missing_baseline_history.baseline_available) + missing_baseline_without_history = run("strict", accepted_backtest=None, previous=None) + self.assertEqual(missing_baseline_without_history.status, DriftStatus.REVIEW) + self.assertTrue(missing_baseline_without_history.alert_suppressed) + self.assertFalse(missing_baseline_without_history.baseline_available) + versioned_backtest = replace(backtest, param_version=2) + self.assertIsNone( + run("strict", accepted_backtest=versioned_backtest, previous=lineage_previous).previous_status + ) + rerun_backtest = replace(backtest, run_id="accepted-rerun") + self.assertIsNone( + run("strict", accepted_backtest=rerun_backtest, previous=lineage_previous).previous_status + ) self.assertIsNone( run( "migration",