refactor(snapshot): remove strategy reverse dependency

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

Author: Pigbibi

docs/us_equity_cross_platform_strategy_spec.md

@@ -178,6 +178,10 @@ stable contract:
 - freshness rule
 - optional manifest/checksum rule
 
+Feature-snapshot manifests and config names must already use canonical profile
+keys. Retired aliases are rejected by shared guards instead of being silently
+resolved inside `QuantPlatformKit`.
+
 The platform runtime owns:
 
 - artifact transport

docs/us_equity_cross_platform_strategy_spec.zh-CN.md

@@ -178,6 +178,10 @@ adapter 可以描述：
 - 新鲜度规则
 - 是否需要 manifest / checksum
 
+`feature_snapshot` 的 manifest 里的 `strategy_profile` / `config_name`
+必须直接使用 canonical profile key。已经退役的 alias 会被共享 guard
+直接拒绝，不再由 `QuantPlatformKit` 静默兜底解析。
+
 平台仓库负责：
 
 - artifact transport

qsl.toml

@@ -1,5 +1,5 @@
 tier = "core"
-ring = 0
+upgrade_ring = "ring_a"
 allow_legacy = true
 legacy_reason = "constraints.txt remains as the pip-compatible mirror for qsl-pins.txt and update-qpk-pin workflow."
 

src/quant_platform_kit/common/feature_snapshot.py

@@ -13,6 +13,8 @@
 
 import pandas as pd
 
+from .strategies import normalize_profile_name
+
 
 DEFAULT_SNAPSHOT_DATE_COLUMNS = ("as_of", "snapshot_date")
 DEFAULT_MAX_SNAPSHOT_MONTH_LAG = 1
@@ -39,17 +41,7 @@
 
 
 def _normalize_strategy_profile_label(value: object) -> str:
-    label = str(value or "").strip()
-    if not label:
-        return ""
-    try:
-        from us_equity_strategies.catalog import resolve_canonical_profile
-    except Exception:
-        return label
-    try:
-        return str(resolve_canonical_profile(label)).strip()
-    except Exception:
-        return label
+    return normalize_profile_name(str(value or "").strip())
 
 
 def _normalize_contract_version_label(value: object) -> str:

tests/test_feature_snapshot.py

@@ -2,45 +2,24 @@
 
 import hashlib
 import json
-import sys
 import tempfile
 import unittest
 from pathlib import Path
-from types import ModuleType
 
 from quant_platform_kit.common.feature_snapshot import load_feature_snapshot_guarded
 
 
 class FeatureSnapshotGuardAliasTests(unittest.TestCase):
-    def test_guard_accepts_legacy_strategy_profile_alias_in_manifest(self) -> None:
-        package = ModuleType("us_equity_strategies")
-        catalog = ModuleType("us_equity_strategies.catalog")
-
-        def resolve_canonical_profile(profile: str | None) -> str:
-            mapping = {
-                "tech_pullback_cash_buffer": "qqq_tech_enhancement",
-                "qqq_tech_enhancement": "qqq_tech_enhancement",
-            }
-            return mapping.get(str(profile), str(profile))
-
-        catalog.resolve_canonical_profile = resolve_canonical_profile
-        sys.modules[package.__name__] = package
-        sys.modules[catalog.__name__] = catalog
-        self.addCleanup(sys.modules.pop, package.__name__, None)
-        self.addCleanup(sys.modules.pop, catalog.__name__, None)
-
+    def test_guard_rejects_retired_strategy_profile_alias_in_manifest(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             tmp_path = Path(tmp_dir)
             snapshot_path = tmp_path / "snapshot.csv"
             manifest_path = tmp_path / "snapshot.csv.manifest.json"
-            config_path = tmp_path / "tech_pullback_cash_buffer.json"
             snapshot_path.write_text(
                 "as_of,symbol,close\n2026-04-01,QQQ,500\n",
                 encoding="utf-8",
             )
-            config_path.write_text('{"name": "tech_pullback_cash_buffer"}', encoding="utf-8")
             snapshot_sha256 = hashlib.sha256(snapshot_path.read_bytes()).hexdigest()
-            config_sha256 = hashlib.sha256(config_path.read_bytes()).hexdigest()
             manifest_path.write_text(
                 json.dumps(
                     {
@@ -49,7 +28,7 @@ def resolve_canonical_profile(profile: str | None) -> str:
                         "config_name": "tech_pullback_cash_buffer",
                         "contract_version": "tech_pullback_cash_buffer.feature_snapshot.v1",
                         "snapshot_sha256": snapshot_sha256,
-                        "config_sha256": config_sha256,
+                        "config_sha256": "abc",
                         "price_as_of": "2026-04-01",
                         "universe_as_of": "2026-03-31",
                         "source_input_status": "fresh",
@@ -70,13 +49,13 @@ def resolve_canonical_profile(profile: str | None) -> str:
                 expected_config_name="qqq_tech_enhancement",
             )
 
-            self.assertIsNotNone(result.frame)
-            self.assertEqual(result.metadata["snapshot_guard_decision"], "proceed")
+            self.assertIsNone(result.frame)
+            self.assertEqual(result.metadata["snapshot_guard_decision"], "fail_closed")
             self.assertEqual(result.metadata["snapshot_manifest_strategy_profile"], "tech_pullback_cash_buffer")
-            self.assertEqual(result.metadata["snapshot_manifest_price_as_of"], "2026-04-01")
-            self.assertEqual(result.metadata["snapshot_manifest_universe_as_of"], "2026-03-31")
-            self.assertEqual(result.metadata["snapshot_manifest_source_input_status"], "fresh")
-            self.assertEqual(result.metadata["snapshot_manifest_source_refresh_run_id"], "12345")
+            self.assertIn(
+                "feature_snapshot_manifest_strategy_profile_mismatch",
+                str(result.metadata["fail_reason"]),
+            )
 
     def test_guard_includes_manifest_diagnostics_when_stale(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:

tests/test_live_return_collector.py

@@ -151,13 +151,17 @@ def test_collect_merges_live_run_returns(self) -> None:
                 {"platform": "schwab", "total_equity": 100.0},
                 domain="us_equity",
             )
+            first_record = store.list_live_run_records("us_equity", strategy_profile="global_etf_rotation")[0]
+            second_recorded_at = (
+                pd.Timestamp(first_record["recorded_at"]).normalize() + pd.Timedelta(days=1, hours=10)
+            ).isoformat()
             # Second record on a later day so pct_change has one observation.
             second = store._local_path("live_runs/us_equity/global_etf_rotation/manual-day2.json")
             second.parent.mkdir(parents=True, exist_ok=True)
             second.write_text(
                 (
                     '{"strategy_profile":"global_etf_rotation","domain":"us_equity",'
-                    '"recorded_at":"2026-07-10T10:00:00+00:00","record_kind":"execution",'
+                    f'"recorded_at":"{second_recorded_at}","record_kind":"execution",'
                     '"execution_result":{"total_equity":102.0}}'
                 ),
                 encoding="utf-8",