Merge pull request #386 from QuantStrategyLab/codex/lifecycle-stream-isolation

feat: isolate lifecycle telemetry account streams

Author: Pigbibi

docs/strategy-lifecycle-benchmark-catalog.md

@@ -24,9 +24,23 @@ Use a JSON file with this shape:
 Run strict monitoring with:
 
 ```text
-quant-lifecycle monitor --domain us_equity --benchmark-catalog catalog.json --require-explicit-benchmark
+quant-lifecycle monitor --domain us_equity --strategy soxl_soxx_trend_income --live-stream-id longbridge-quant-sg-service --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.
+
+## Account-safe live telemetry
+
+Live account equity must be evaluated one account/runtime stream at a time.
+When it runs in Cloud Run, QPK records the built-in `K_SERVICE` identity as
+`lifecycle_stream_id`; other runtimes can set `LIFECYCLE_STREAM_ID`. The
+stream becomes part of the storage path and the live-record deduplication key.
+
+If the same strategy profile has records from more than one stream and no
+stream is selected, the return collector omits that profile rather than
+combining separate broker accounts into a false equity curve. `--live-stream-id`
+is therefore required for promotion-grade monitoring of persisted live data.
+This is still read-only monitoring: it cannot place orders, alter a strategy,
+or promote a candidate.

src/quant_platform_kit/strategy_lifecycle/cli.py

@@ -39,6 +39,9 @@ def _run_monitor(args: argparse.Namespace) -> int:
         kwargs["strategy_benchmarks"] = load_strategy_benchmark_catalog(benchmark_catalog)
     if getattr(args, "require_explicit_benchmark", False):
         kwargs["require_explicit_benchmark"] = True
+    live_stream_id = getattr(args, "live_stream_id", None)
+    if live_stream_id:
+        kwargs["live_stream_id"] = live_stream_id
     snapshots = run_monitor(**kwargs)
     _print(f"[monitor] Generated {len(snapshots)} performance snapshots")
     return 0
@@ -206,6 +209,7 @@ def _run_lifecycle(args: argparse.Namespace) -> int:
             output_dir=None,
             benchmark_catalog=getattr(args, "benchmark_catalog", None),
             require_explicit_benchmark=getattr(args, "require_explicit_benchmark", False),
+            live_stream_id=getattr(args, "live_stream_id", None),
         )
     )
     if monitor_status != 0:
@@ -313,6 +317,11 @@ 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(
+        "--live-stream-id",
+        default=None,
+        help="Monitor one stable account/runtime telemetry stream; never mix streams.",
+    )
     monitor.add_argument(
         "--benchmark-catalog",
         default=None,
@@ -384,6 +393,7 @@ def build_parser() -> argparse.ArgumentParser:
     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")
+    lifecycle.add_argument("--live-stream-id", default=None)
     _add_baseline_options(lifecycle)
     lifecycle.set_defaults(func=_run_lifecycle)
 

src/quant_platform_kit/strategy_lifecycle/performance_monitor.py

@@ -33,6 +33,37 @@ def _now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+def resolve_lifecycle_stream_id(
+    explicit_stream_id: str = "",
+    *,
+    execution_result: Mapping[str, Any] | None = None,
+) -> str:
+    """Return a stable, account-safe telemetry stream identity.
+
+    Cloud Run supplies ``K_SERVICE`` per deployed service, which keeps the
+    same strategy on different broker accounts from being combined into a
+    synthetic equity curve.  ``LIFECYCLE_STREAM_ID`` is available for
+    non-Cloud-Run runtimes that need a stable explicit identity.
+    """
+    import os
+
+    for candidate in (
+        explicit_stream_id,
+        os.environ.get("LIFECYCLE_STREAM_ID"),
+        os.environ.get("K_SERVICE"),
+    ):
+        value = str(candidate or "").strip()
+        if value:
+            return value
+
+    payload = execution_result if isinstance(execution_result, Mapping) else {}
+    platform = str(payload.get("platform") or "").strip()
+    account_scope = str(payload.get("account_scope") or payload.get("service_name") or "").strip()
+    if platform and account_scope:
+        return f"{platform}:{account_scope}"
+    return platform
+
+
 def _is_valid_series(series: object, *, min_observations: int = 10) -> bool:
     if not isinstance(series, pd.Series):
         return False
@@ -51,6 +82,7 @@ def run_monitor(
     collector: ReturnCollector | None = None,
     strategy_benchmarks: Mapping[str, str] | None = None,
     require_explicit_benchmark: bool = False,
+    live_stream_id: str | None = None,
 ) -> list[StrategyPerformanceSnapshot]:
     """Run the performance monitor for the given domain.
 
@@ -67,6 +99,9 @@ def run_monitor(
         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.
+        live_stream_id: Optional stable telemetry stream identity.  When live
+            account data is used, this prevents independent broker accounts
+            from being merged into one return series.
 
     Returns:
         List of StrategyPerformanceSnapshot objects generated.
@@ -75,7 +110,10 @@ def run_monitor(
     collector = collector or ReturnCollector()
 
     # 1. Collect returns
-    all_returns = collector.collect(domain)
+    if isinstance(collector, ReturnCollector):
+        all_returns = collector.collect(domain, live_stream_id=live_stream_id)
+    else:
+        all_returns = collector.collect(domain)
     if not all_returns:
         if fail_on_empty:
             raise RuntimeError(
@@ -118,7 +156,7 @@ def run_monitor(
         snapshot = StrategyPerformanceSnapshot(
             strategy_profile=profile,
             domain=domain,
-            platform="",
+            platform=str(live_stream_id or "").strip(),
             as_of=date.today(),
             benchmark_symbol=benchmark_symbol,
             computed_at=_now_iso(),
@@ -234,6 +272,7 @@ def record(
         execution_result: Mapping[str, Any] | None = None,
         *,
         domain: str = "",
+        stream_id: str = "",
     ) -> dict[str, Any]:
         profile = str(profile_id or "").strip()
         if not profile:
@@ -246,8 +285,21 @@ def record(
             "decision": _serialize_decision(decision),
             "execution_result": dict(execution_result or {}),
         }
-        self._store.save_live_run_record(profile, str(domain or "").strip(), payload)
-        return {"ok": True, "profile": profile, "domain": str(domain or "").strip()}
+        resolved_stream_id = resolve_lifecycle_stream_id(stream_id, execution_result=execution_result)
+        if resolved_stream_id:
+            payload["lifecycle_stream_id"] = resolved_stream_id
+        self._store.save_live_run_record(
+            profile,
+            str(domain or "").strip(),
+            payload,
+            stream_id=resolved_stream_id,
+        )
+        return {
+            "ok": True,
+            "profile": profile,
+            "domain": str(domain or "").strip(),
+            "stream_id": resolved_stream_id,
+        }
 
     def record_execution(
         self,
@@ -256,6 +308,7 @@ def record_execution(
         *,
         domain: str = "",
         decision: Any | None = None,
+        stream_id: str = "",
     ) -> dict[str, Any]:
         """Persist platform-layer execution telemetry after order routing."""
         profile = str(profile_id or "").strip()
@@ -273,8 +326,21 @@ def record_execution(
         }
         if decision is not None:
             payload["decision"] = _serialize_decision(decision)
-        self._store.save_live_run_record(profile, str(domain or "").strip(), payload)
-        return {"ok": True, "profile": profile, "domain": str(domain or "").strip()}
+        resolved_stream_id = resolve_lifecycle_stream_id(stream_id, execution_result=execution_result)
+        if resolved_stream_id:
+            payload["lifecycle_stream_id"] = resolved_stream_id
+        self._store.save_live_run_record(
+            profile,
+            str(domain or "").strip(),
+            payload,
+            stream_id=resolved_stream_id,
+        )
+        return {
+            "ok": True,
+            "profile": profile,
+            "domain": str(domain or "").strip(),
+            "stream_id": resolved_stream_id,
+        }
 
 
 def infer_strategy_domain(profile_id: str, *, explicit_domain: str = "") -> str:
@@ -297,6 +363,7 @@ def try_record_platform_execution(
     *,
     domain: str = "",
     decision: Any | None = None,
+    stream_id: str = "",
 ) -> None:
     """Best-effort execution recorder for platform runtimes; never raises."""
     try:
@@ -308,6 +375,7 @@ def try_record_platform_execution(
             execution_result,
             domain=infer_strategy_domain(profile_id, explicit_domain=domain),
             decision=decision,
+            stream_id=stream_id,
         )
     except Exception as exc:  # pragma: no cover
         logger.warning("PerformanceMonitor.record_execution failed: %s", exc)

src/quant_platform_kit/strategy_lifecycle/performance_store.py

@@ -379,32 +379,55 @@ def list_snapshot_profiles(self, domain: str) -> tuple[str, ...]:
 
     # ── live runs (per-evaluate / per-execution records) ─────────
 
-    def _live_run_key(self, domain: str, strategy_profile: str, recorded_at: str) -> str:
+    def _live_run_key(
+        self,
+        domain: str,
+        strategy_profile: str,
+        recorded_at: str,
+        *,
+        stream_id: str = "",
+    ) -> str:
         safe_time = recorded_at.replace(":", "-")
-        return f"live_runs/{_clean_key(domain)}/{_clean_key(strategy_profile)}/{safe_time}.json"
+        root = f"live_runs/{_clean_key(domain)}/{_clean_key(strategy_profile)}"
+        stream = str(stream_id or "").strip()
+        if stream:
+            return f"{root}/streams/{_clean_key(stream)}/{safe_time}.json"
+        return f"{root}/{safe_time}.json"
 
     def save_live_run_record(
         self,
         strategy_profile: str,
         domain: str,
         payload: Mapping[str, Any],
+        *,
+        stream_id: str = "",
     ) -> None:
         recorded_at = str(payload.get("recorded_at") or _now_iso())
+        stream = str(stream_id or payload.get("lifecycle_stream_id") or "").strip()
+        stored_payload = dict(payload)
+        if stream:
+            stored_payload["lifecycle_stream_id"] = stream
         self._write(
-            self._live_run_key(domain, strategy_profile, recorded_at),
-            {**dict(payload), "schema_version": SCHEMA_VERSION},
+            self._live_run_key(domain, strategy_profile, recorded_at, stream_id=stream),
+            {**stored_payload, "schema_version": SCHEMA_VERSION},
         )
 
     def list_live_run_records(
         self,
         domain: str,
         *,
         strategy_profile: str | None = None,
+        stream_id: str | None = None,
     ) -> list[dict[str, Any]]:
         """Load persisted live evaluation/execution records for a domain."""
         prefix = f"live_runs/{_clean_key(domain)}/"
         if strategy_profile:
             prefix = f"{prefix}{_clean_key(strategy_profile)}/"
+        stream = str(stream_id or "").strip()
+        if stream:
+            if not strategy_profile:
+                raise ValueError("strategy_profile is required when filtering live records by stream_id")
+            prefix = f"{prefix}streams/{_clean_key(stream)}/"
 
         records: list[dict[str, Any]] = []
 
@@ -426,9 +449,12 @@ def list_live_run_records(
 
         deduped: dict[str, dict[str, Any]] = {}
         for record in records:
+            if stream and str(record.get("lifecycle_stream_id") or "").strip() != stream:
+                continue
             dedupe_key = "|".join(
                 [
                     str(record.get("strategy_profile") or ""),
+                    str(record.get("lifecycle_stream_id") or ""),
                     str(record.get("recorded_at") or ""),
                     str(record.get("record_kind") or ""),
                 ]

src/quant_platform_kit/strategy_lifecycle/return_collector.py

@@ -108,15 +108,42 @@ def _store_instance(self) -> PerformanceStore:
             return self._store
         return PerformanceStore.from_env()
 
-    def collect_from_live_runs(self, domain: str) -> Mapping[str, pd.Series]:
-        """Build per-strategy return series from persisted live run equity snapshots."""
+    def collect_from_live_runs(
+        self,
+        domain: str,
+        *,
+        stream_id: str | None = None,
+    ) -> Mapping[str, pd.Series]:
+        """Build per-strategy returns without merging independent account streams.
+
+        A caller that needs live monitoring must supply a specific stream when
+        more than one stream has reported the same strategy profile.  Skipping
+        that ambiguous profile is safer than deriving a false equity curve
+        from separate broker accounts.
+        """
         records = self._store_instance().list_live_run_records(domain)
         grouped = group_live_run_records_by_profile(records)
-        return {
-            profile: live_run_records_to_return_series(profile_records)
-            for profile, profile_records in grouped.items()
-            if live_run_records_to_return_series(profile_records).size > 0
-        }
+        result: dict[str, pd.Series] = {}
+        requested_stream = str(stream_id or "").strip()
+        for profile, profile_records in grouped.items():
+            streams = {
+                str(record.get("lifecycle_stream_id") or "").strip()
+                for record in profile_records
+            }
+            if requested_stream:
+                if requested_stream not in streams:
+                    continue
+                profile_records = [
+                    record
+                    for record in profile_records
+                    if str(record.get("lifecycle_stream_id") or "").strip() == requested_stream
+                ]
+            elif len(streams) > 1:
+                continue
+            series = live_run_records_to_return_series(profile_records)
+            if not series.empty:
+                result[profile] = series
+        return result
 
     def _merge_return_series(
         self,
@@ -140,6 +167,7 @@ def collect(
         *,
         date_column: str = "as_of",
         benchmark_columns: Sequence[str] | None = None,
+        live_stream_id: str | None = None,
     ) -> Mapping[str, pd.Series]:
         """Collect all strategy return series for a domain.
 
@@ -164,7 +192,12 @@ def collect(
                     else:
                         all_strategies[name] = series
 
-        live_series = self.collect_from_live_runs(domain)
+        if live_stream_id:
+            live_series = self.collect_from_live_runs(domain, stream_id=live_stream_id)
+        else:
+            # Keep the original call shape for integrations that replace this
+            # best-effort collector with a compatible one-argument adapter.
+            live_series = self.collect_from_live_runs(domain)
         return self._merge_return_series(all_strategies, live_series)
 
     def collect_benchmark(

tests/test_lifecycle_cli.py

@@ -77,6 +77,30 @@ def fake_run_monitor(**kwargs):
             },
         )
 
+    def test_monitor_command_passes_live_stream_filter_only_when_requested(self) -> None:
+        observed = {}
+
+        def fake_load_callable(_module_name: str, _function_name: str):
+            def fake_run_monitor(**kwargs):
+                observed.update(kwargs)
+                return [object()]
+
+            return fake_run_monitor
+
+        with patch.object(cli, "_load_callable", fake_load_callable):
+            result = cli.main([
+                "monitor",
+                "--domain",
+                "us_equity",
+                "--strategy",
+                "soxl_soxx_trend_income",
+                "--live-stream-id",
+                "longbridge-quant-sg-service",
+            ])
+
+        self.assertEqual(result, 0)
+        self.assertEqual(observed["live_stream_id"], "longbridge-quant-sg-service")
+
     def test_drift_command_counts_status_values(self) -> None:
         statuses = [
             SimpleNamespace(status=SimpleNamespace(value="critical")),