fix: harden accepted baseline drift handling

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

Author: Pigbibi

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

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
 
 

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,

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:

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

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):

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]