feat: require explicit benchmarks for strict monitoring

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

Author: Pigbibi

docs/strategy-lifecycle-benchmark-catalog.md

@@ -0,0 +1,32 @@
+# Lifecycle monitoring benchmark catalog
+
+Promotion-grade monitoring must name the passive or unleveraged instrument used
+to judge each strategy. A generic US-equity default such as SPY is not an
+acceptable substitute for a leveraged-sector strategy.
+
+Use a JSON file with this shape:
+
+```json
+{
+  "schema_version": "qsl.strategy-benchmark-catalog.v1",
+  "authority": {"monitoring_only": true, "no_order": true},
+  "bindings": [
+    {
+      "strategy_profile": "soxl_soxx_trend_income",
+      "benchmark_symbol": "buy_hold_SOXX",
+      "benchmark_kind": "unleveraged_underlying",
+      "relative_drawdown_required": true
+    }
+  ]
+}
+```
+
+Run strict monitoring with:
+
+```text
+quant-lifecycle monitor --domain us_equity --benchmark-catalog catalog.json --require-explicit-benchmark
+```
+
+Strict mode refuses to publish a snapshot when either the strategy binding or
+its benchmark return series is absent. The catalog is monitoring-only and
+never grants strategy, broker, or promotion authority.

src/quant_platform_kit/strategy_lifecycle/__init__.py

@@ -14,6 +14,13 @@
     UpdateStage,
     WindowPerformance,
 )
+from quant_platform_kit.strategy_lifecycle.benchmark_catalog import (
+    STRATEGY_BENCHMARK_CATALOG_SCHEMA,
+    StrategyBenchmarkBinding,
+    StrategyBenchmarkCatalogError,
+    build_strategy_benchmark_catalog,
+    load_strategy_benchmark_catalog,
+)
 from quant_platform_kit.strategy_spec import (
     OPTIMIZATION_SPEC_SCHEMA_VERSION,
     RESEARCH_SPEC_SCHEMA_VERSION,
@@ -80,6 +87,9 @@
 
 __all__ = [
     "BacktestResult",
+    "STRATEGY_BENCHMARK_CATALOG_SCHEMA",
+    "StrategyBenchmarkBinding",
+    "StrategyBenchmarkCatalogError",
     "DriftDimension",
     "DriftResult",
     "DriftStatus",
@@ -110,9 +120,11 @@
     "RESEARCH_DRIVER_SCHEMA_VERSION",
     "RESEARCH_DRIVER_TERMINAL_STATUSES",
     "load_evidence_package",
+    "load_strategy_benchmark_catalog",
     "canonical_evidence_package_v2_bytes",
     "read_evidence_package_v2_json",
     "build_live_candidate_notification",
+    "build_strategy_benchmark_catalog",
     "assess_strategy_release_readiness",
     "build_forward_risk_terminal_artifact",
     "build_nonready_forward_risk_stage",

src/quant_platform_kit/strategy_lifecycle/benchmark_catalog.py

@@ -0,0 +1,115 @@
+"""Validated, read-only benchmark bindings for lifecycle monitoring.
+
+The catalog deliberately maps a *strategy profile* to the passive or
+unleveraged instrument used to judge it. It supplies monitoring context only:
+it cannot change a strategy, rebalance an account, or grant execution rights.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+
+
+STRATEGY_BENCHMARK_CATALOG_SCHEMA = "qsl.strategy-benchmark-catalog.v1"
+_BENCHMARK_KINDS = frozenset({"passive", "unleveraged_underlying"})
+
+
+class StrategyBenchmarkCatalogError(ValueError):
+    """Raised when a benchmark catalog cannot be safely used for monitoring."""
+
+
+def _nonblank(value: object, label: str) -> str:
+    if not isinstance(value, str) or not value or value != value.strip():
+        raise StrategyBenchmarkCatalogError(f"{label} must be a non-empty canonical string")
+    return value
+
+
+@dataclass(frozen=True)
+class StrategyBenchmarkBinding:
+    """One explicit, no-authority performance benchmark binding."""
+
+    strategy_profile: str
+    benchmark_symbol: str
+    benchmark_kind: str = "passive"
+    relative_drawdown_required: bool = True
+
+    def __post_init__(self) -> None:
+        _nonblank(self.strategy_profile, "strategy profile")
+        _nonblank(self.benchmark_symbol, "benchmark symbol")
+        if self.benchmark_kind not in _BENCHMARK_KINDS:
+            raise StrategyBenchmarkCatalogError("benchmark kind is not supported")
+        if type(self.relative_drawdown_required) is not bool:
+            raise StrategyBenchmarkCatalogError("relative drawdown requirement must be boolean")
+
+
+def build_strategy_benchmark_catalog(
+    bindings: Sequence[StrategyBenchmarkBinding],
+) -> dict[str, object]:
+    """Return a validated, JSON-ready catalog without execution authority."""
+    entries = tuple(bindings)
+    if not entries or any(type(entry) is not StrategyBenchmarkBinding for entry in entries):
+        raise StrategyBenchmarkCatalogError("catalog must contain immutable benchmark bindings")
+    profiles = [entry.strategy_profile for entry in entries]
+    if len(set(profiles)) != len(profiles):
+        raise StrategyBenchmarkCatalogError("strategy profiles must be unique")
+    return {
+        "schema_version": STRATEGY_BENCHMARK_CATALOG_SCHEMA,
+        "authority": {"monitoring_only": True, "no_order": True},
+        "bindings": [
+            {
+                "strategy_profile": entry.strategy_profile,
+                "benchmark_symbol": entry.benchmark_symbol,
+                "benchmark_kind": entry.benchmark_kind,
+                "relative_drawdown_required": entry.relative_drawdown_required,
+            }
+            for entry in entries
+        ],
+    }
+
+
+def load_strategy_benchmark_catalog(path: str | Path) -> dict[str, str]:
+    """Load explicit profile-to-benchmark bindings from a validated JSON catalog."""
+    try:
+        payload = json.loads(Path(path).read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise StrategyBenchmarkCatalogError("benchmark catalog could not be read as JSON") from exc
+    if not isinstance(payload, Mapping):
+        raise StrategyBenchmarkCatalogError("benchmark catalog must be an object")
+    if payload.get("schema_version") != STRATEGY_BENCHMARK_CATALOG_SCHEMA:
+        raise StrategyBenchmarkCatalogError("benchmark catalog schema version is not supported")
+    authority = payload.get("authority")
+    if authority != {"monitoring_only": True, "no_order": True}:
+        raise StrategyBenchmarkCatalogError("benchmark catalog must declare monitoring-only no-order authority")
+    raw_bindings = payload.get("bindings")
+    if not isinstance(raw_bindings, list):
+        raise StrategyBenchmarkCatalogError("benchmark catalog bindings must be a list")
+    bindings: list[StrategyBenchmarkBinding] = []
+    for raw in raw_bindings:
+        if not isinstance(raw, Mapping):
+            raise StrategyBenchmarkCatalogError("benchmark catalog binding must be an object")
+        bindings.append(
+            StrategyBenchmarkBinding(
+                strategy_profile=raw.get("strategy_profile"),
+                benchmark_symbol=raw.get("benchmark_symbol"),
+                benchmark_kind=raw.get("benchmark_kind", "passive"),
+                relative_drawdown_required=raw.get("relative_drawdown_required", True),
+            )
+        )
+    catalog = build_strategy_benchmark_catalog(bindings)
+    return {
+        str(item["strategy_profile"]): str(item["benchmark_symbol"])
+        for item in catalog["bindings"]
+        if isinstance(item, Mapping)
+    }
+
+
+__all__ = [
+    "STRATEGY_BENCHMARK_CATALOG_SCHEMA",
+    "StrategyBenchmarkBinding",
+    "StrategyBenchmarkCatalogError",
+    "build_strategy_benchmark_catalog",
+    "load_strategy_benchmark_catalog",
+]

src/quant_platform_kit/strategy_lifecycle/cli.py

@@ -25,11 +25,21 @@ def _run_monitor(args: argparse.Namespace) -> int:
         "quant_platform_kit.strategy_lifecycle.performance_monitor",
         "run_monitor",
     )
-    snapshots = run_monitor(
-        domain=args.domain,
-        strategy_profile=args.strategy,
-        output_dir=args.output_dir,
-    )
+    kwargs: dict[str, Any] = {
+        "domain": args.domain,
+        "strategy_profile": args.strategy,
+        "output_dir": args.output_dir,
+    }
+    benchmark_catalog = getattr(args, "benchmark_catalog", None)
+    if benchmark_catalog:
+        load_strategy_benchmark_catalog = _load_callable(
+            "quant_platform_kit.strategy_lifecycle.benchmark_catalog",
+            "load_strategy_benchmark_catalog",
+        )
+        kwargs["strategy_benchmarks"] = load_strategy_benchmark_catalog(benchmark_catalog)
+    if getattr(args, "require_explicit_benchmark", False):
+        kwargs["require_explicit_benchmark"] = True
+    snapshots = run_monitor(**kwargs)
     _print(f"[monitor] Generated {len(snapshots)} performance snapshots")
     return 0
 
@@ -189,7 +199,15 @@ def _run_autopilot(args: argparse.Namespace) -> int:
 def _run_lifecycle(args: argparse.Namespace) -> int:
     _print(f"[lifecycle] Running full lifecycle for domain={args.domain}")
     _print("[lifecycle] Step: monitor")
-    monitor_status = _run_monitor(argparse.Namespace(domain=args.domain, strategy=None, output_dir=None))
+    monitor_status = _run_monitor(
+        argparse.Namespace(
+            domain=args.domain,
+            strategy=None,
+            output_dir=None,
+            benchmark_catalog=getattr(args, "benchmark_catalog", None),
+            require_explicit_benchmark=getattr(args, "require_explicit_benchmark", False),
+        )
+    )
     if monitor_status != 0:
         return monitor_status
 
@@ -295,6 +313,16 @@ def build_parser() -> argparse.ArgumentParser:
     monitor.add_argument("--domain", default="us_equity")
     monitor.add_argument("--strategy", default=None)
     monitor.add_argument("--output-dir", default=None)
+    monitor.add_argument(
+        "--benchmark-catalog",
+        default=None,
+        help="Validated JSON mapping strategy profiles to their monitoring benchmarks.",
+    )
+    monitor.add_argument(
+        "--require-explicit-benchmark",
+        action="store_true",
+        help="Fail closed if a strategy binding or its benchmark data is unavailable.",
+    )
     monitor.set_defaults(func=_run_monitor)
 
     drift = subparsers.add_parser("drift", help="Run drift detection and publish drift alerts.")
@@ -354,6 +382,8 @@ 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")
+    lifecycle.add_argument("--benchmark-catalog", default=None)
+    lifecycle.add_argument("--require-explicit-benchmark", action="store_true")
     _add_baseline_options(lifecycle)
     lifecycle.set_defaults(func=_run_lifecycle)
 

src/quant_platform_kit/strategy_lifecycle/performance_monitor.py

@@ -49,6 +49,8 @@ def run_monitor(
     fail_on_empty: bool = True,
     store: PerformanceStore | None = None,
     collector: ReturnCollector | None = None,
+    strategy_benchmarks: Mapping[str, str] | None = None,
+    require_explicit_benchmark: bool = False,
 ) -> list[StrategyPerformanceSnapshot]:
     """Run the performance monitor for the given domain.
 
@@ -61,6 +63,10 @@ def run_monitor(
         fail_on_empty: Raise when no usable return series can be monitored.
         store: PerformanceStore instance; auto-created from env if None.
         collector: ReturnCollector instance; auto-created if None.
+        strategy_benchmarks: Explicit strategy-profile to benchmark bindings.
+        require_explicit_benchmark: Refuse to monitor profiles without a binding
+            or without the declared benchmark return series. This is the
+            promotion-grade setting for leveraged strategies.
 
     Returns:
         List of StrategyPerformanceSnapshot objects generated.
@@ -89,8 +95,20 @@ def run_monitor(
         series = normalize_return_series(returns)
 
         # Resolve benchmark
-        benchmark_symbol = resolve_strategy_benchmark(profile, domain)
+        benchmark_symbol = resolve_strategy_benchmark(
+            profile,
+            domain,
+            catalog_benchmarks=strategy_benchmarks,
+            require_explicit=require_explicit_benchmark,
+        )
         benchmark_series = collector.collect_benchmark(domain, benchmark_symbol)
+        if require_explicit_benchmark and not _is_valid_series(
+            benchmark_series, min_observations=min_observations
+        ):
+            raise RuntimeError(
+                f"explicit benchmark data is unavailable or insufficient for "
+                f"strategy_profile={profile!r}, benchmark={benchmark_symbol!r}"
+            )
         benchmark_returns = normalize_return_series(benchmark_series) if benchmark_series is not None else None
 
         # Load backtest reference for comparison

src/quant_platform_kit/strategy_lifecycle/return_collector.py

@@ -25,6 +25,10 @@
 _RETURN_MATRIX_FILENAME = "portfolio_and_tracker_returns.csv"
 
 
+class MissingStrategyBenchmarkError(ValueError):
+    """Raised when strict monitoring has no explicit benchmark binding."""
+
+
 class ReturnCollector:
     """Discover and read return matrices from market pipeline artifact directories.
 
@@ -188,13 +192,28 @@ def resolve_strategy_benchmark(
     domain: str,
     *,
     catalog_benchmarks: Mapping[str, str] | None = None,
+    require_explicit: bool = False,
 ) -> str:
     """Resolve the benchmark symbol for a strategy.
 
-    Falls back through: catalog metadata → domain defaults.
+    In normal compatibility mode this falls back through catalog metadata then
+    domain defaults. Strict mode is intended for promotion-grade or leveraged
+    monitoring: every profile must have a catalog binding, preventing a silent
+    and potentially inappropriate fallback to SPY.
     """
-    if catalog_benchmarks and strategy_profile in catalog_benchmarks:
-        return catalog_benchmarks[strategy_profile]
+    profile = str(strategy_profile or "").strip()
+    if catalog_benchmarks and profile in catalog_benchmarks:
+        benchmark = str(catalog_benchmarks[profile] or "").strip()
+        if benchmark:
+            return benchmark
+        raise MissingStrategyBenchmarkError(
+            f"explicit benchmark for strategy_profile={profile!r} is blank"
+        )
+    if require_explicit:
+        raise MissingStrategyBenchmarkError(
+            f"no explicit benchmark binding for strategy_profile={profile!r}; "
+            "provide a validated strategy benchmark catalog"
+        )
 
     # Domain defaults
     defaults = {