docs: add cross asset strategy inventory

docs/cross_asset_integration_plan.zh-CN.md

@@ -0,0 +1,14 @@
+# 跨资产主线接入计划
+
+此前生命周期矩阵主要覆盖美股策略。本清单补上中国股票、港股和加密策略的统一入口，但它是 inventory，不把各仓库已有的 `runtime_enabled` 字样误当成完整 P0-P6 证据。
+
+接入顺序：
+
+1. 将各仓库 catalog 导出到统一 inventory。
+2. 为每个策略绑定 owner、数据 manifest、evidence package 和 RiskSnapshot。
+3. 先接入 research/shadow，每日记录运行结果。
+4. 再按策略自身证据推进 P4/P5。
+5. live 权限继续由原有 broker gate 控制，自动系统不能扩大权限。
+
+当前覆盖：CN equity 8 条、HK equity 3 条、crypto 5 条；详见
+`docs/registry/cross_asset_strategy_inventory.json`。

docs/registry/cross_asset_strategy_inventory.json

@@ -0,0 +1,25 @@
+{
+  "schema_version": "cross_asset_strategy_inventory.v1",
+  "generated_at": "2026-08-23",
+  "inventory_only": true,
+  "permission_effect": "none",
+  "source_policy": "Repository catalog inventory; entries do not assert P0-P6 evidence or grant trading authority.",
+  "entries": [
+    {"id": "cn_industry_etf_rotation", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "runtime_enabled", "canonical_status": "live_candidate", "next_action": "接入统一 lifecycle matrix、每日 risk snapshot 和 shadow evidence。"},
+    {"id": "cn_industry_etf_rotation_aggressive", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "live_candidate", "canonical_status": "live_candidate", "next_action": "补齐组合级风险边界与统一 shadow 记录，不能自动扩大 live 权限。"},
+    {"id": "cn_index_etf_tactical_rotation", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "接入统一插件 catalog 和 P0-P3 evidence package。"},
+    {"id": "cn_chinext_growth_momentum_quality", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "核对真实数据证据与 Risk Gate 消费。"},
+    {"id": "cn_dividend_quality_snapshot", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "接入 snapshot freshness、组合风险和 shadow 记录。"},
+    {"id": "cn_chinext_tactical_rotation", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "建立统一 P3 evidence package。"},
+    {"id": "cn_star_growth_momentum_quality", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "建立统一 P3 evidence package。"},
+    {"id": "cn_equity_combo", "domain": "cn_equity", "owner_repo": "CnEquityStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "作为组合 orchestrator 接入统一风险层，不直接获得 live 权限。"},
+    {"id": "hk_global_etf_tactical_rotation", "domain": "hk_equity", "owner_repo": "HkEquityStrategies", "catalog_status": "runtime_enabled", "canonical_status": "live_candidate", "next_action": "接入统一 lifecycle matrix 与每日 shadow/risk 记录。"},
+    {"id": "hk_low_vol_dividend_quality_snapshot", "domain": "hk_equity", "owner_repo": "HkEquityStrategies", "catalog_status": "runtime_enabled", "canonical_status": "live_candidate", "next_action": "接入 snapshot freshness 与组合级风险聚合。"},
+    {"id": "hk_equity_combo", "domain": "hk_equity", "owner_repo": "HkEquityStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "保持 research_active，补证据后再评估 shadow。"},
+    {"id": "crypto_live_pool_rotation", "domain": "crypto", "owner_repo": "CryptoStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "接入 pool artifact、统一 Risk Gate 和每日运行记录，再评估 shadow。"},
+    {"id": "crypto_leader_rotation", "domain": "crypto", "owner_repo": "CryptoStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "作为 live_pool alias 补齐统一 P0-P3 evidence 与组合风险记录。"},
+    {"id": "crypto_btc_dca", "domain": "crypto", "owner_repo": "CryptoStrategies", "catalog_status": "shadow_candidate", "canonical_status": "shadow_active", "next_action": "接入统一 shadow/forward 与资金风险快照。"},
+    {"id": "crypto_trend_rotation", "domain": "crypto", "owner_repo": "CryptoStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "保持 research_active，完成证据和风险重构。"},
+    {"id": "crypto_equity_combo", "domain": "crypto", "owner_repo": "CryptoStrategies", "catalog_status": "research_backtest_only", "canonical_status": "research_active", "next_action": "作为 shadow orchestrator 研究，不直接进入 live。"}
+  ]
+}

src/quant_platform_kit/risk/__init__.py

@@ -30,6 +30,7 @@
     extract_portfolio_risk_diagnostics,
 )
 from quant_platform_kit.risk.snapshot import RiskSnapshot, build_risk_snapshot
+from quant_platform_kit.risk.cross_asset_snapshot import build_cross_asset_snapshot
 from quant_platform_kit.risk.research_consumer import (
     DEFAULT_RESEARCH_STRATEGIES,
     ResearchRiskObservation,
@@ -55,6 +56,7 @@
     "build_risk_engine",
     "RiskSnapshot",
     "build_risk_snapshot",
+    "build_cross_asset_snapshot",
     "DEFAULT_RESEARCH_STRATEGIES",
     "ResearchRiskObservation",
     "consume_research_risk",

src/quant_platform_kit/risk/cross_asset_snapshot.py

@@ -0,0 +1,51 @@
+"""Cross-asset daily risk observation envelope.
+
+This is a research/shadow aggregation contract only.  It deliberately does
+not produce target weights, order intents, or broker instructions.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from quant_platform_kit.risk.snapshot import RiskSnapshot
+
+
+def build_cross_asset_snapshot(
+    snapshots: Mapping[str, RiskSnapshot],
+    *,
+    as_of: str,
+    run_mode: str = "research_active",
+) -> dict[str, Any]:
+    """Build a deterministic CN/HK/US/Crypto daily observation envelope.
+
+    Individual asset snapshots retain their own provenance and expiry.  The
+    envelope reports partial readiness instead of treating missing assets as
+    zero risk.  ``no_order`` is fixed true by contract.
+    """
+    if not isinstance(as_of, str) or not as_of.strip():
+        raise ValueError("as_of is required")
+    if run_mode not in {"research_active", "shadow_active"}:
+        raise ValueError("run_mode must be research_active or shadow_active")
+    if not isinstance(snapshots, Mapping) or not snapshots:
+        raise ValueError("at least one asset snapshot is required")
+    if any(not isinstance(asset, str) or not asset.strip() for asset in snapshots):
+        raise ValueError("asset keys must be non-empty strings")
+    if any(not isinstance(snapshot, RiskSnapshot) for snapshot in snapshots.values()):
+        raise ValueError("snapshots must contain RiskSnapshot values")
+
+    ready = {asset: snapshot for asset, snapshot in snapshots.items() if snapshot.is_usable}
+    parked = sorted(asset for asset, snapshot in snapshots.items() if not snapshot.is_usable)
+    return {
+        "contract_version": "cross_asset_risk_snapshot.v1",
+        "as_of": as_of.strip(),
+        "run_mode": run_mode,
+        "no_order": True,
+        "status": "READY" if len(ready) == len(snapshots) else "PARTIAL",
+        "asset_count": len(snapshots),
+        "ready_asset_count": len(ready),
+        "parked_assets": parked,
+        "effective_exposure": sum(snapshot.effective_exposure for snapshot in ready.values()),
+        "max_loss_estimate": sum(snapshot.max_loss_estimate for snapshot in ready.values()),
+        "assets": {asset: snapshot.to_dict() for asset, snapshot in sorted(snapshots.items())},
+    }

tests/test_cross_asset_snapshot.py

@@ -0,0 +1,46 @@
+import unittest
+
+from quant_platform_kit.risk import build_cross_asset_snapshot, build_risk_snapshot
+
+
+def _snapshot(**overrides):
+    values = {
+        "account_equity": 10_000.0,
+        "risk_budget": 0.01,
+        "effective_exposure": 0.20,
+        "max_loss_estimate": 0.01,
+        "drawdown_scalar": 1.0,
+        "kelly_fraction": 0.20,
+        "applied_fraction": 0.10,
+        "circuit_state": "ACTIVE",
+        "evidence_package_id": "sha256:abc",
+        "expires_at": "2026-08-24T00:00:00Z",
+    }
+    values.update(overrides)
+    return build_risk_snapshot(values)
+
+
+class CrossAssetSnapshotTests(unittest.TestCase):
+    def test_ready_envelope_is_no_order_and_deterministic(self):
+        result = build_cross_asset_snapshot(
+            {"crypto": _snapshot(), "us_equity": _snapshot()},
+            as_of="2026-08-23T00:00:00Z",
+            run_mode="shadow_active",
+        )
+        self.assertEqual(result["status"], "READY")
+        self.assertTrue(result["no_order"])
+        self.assertEqual(list(result["assets"]), ["crypto", "us_equity"])
+        self.assertEqual(result["effective_exposure"], 0.4)
+
+    def test_missing_or_parked_asset_is_partial_not_zero_risk(self):
+        result = build_cross_asset_snapshot(
+            {"cn_equity": _snapshot(), "hk_equity": _snapshot(circuit_state="TRIPPED")},
+            as_of="2026-08-23",
+        )
+        self.assertEqual(result["status"], "PARTIAL")
+        self.assertEqual(result["parked_assets"], ["hk_equity"])
+        self.assertEqual(result["ready_asset_count"], 1)
+
+    def test_rejects_live_mode(self):
+        with self.assertRaises(ValueError):
+            build_cross_asset_snapshot({"us_equity": _snapshot()}, as_of="today", run_mode="live")

tests/test_cross_asset_strategy_inventory.py

@@ -0,0 +1,42 @@
+import json
+from pathlib import Path
+
+from quant_platform_kit.strategy_lifecycle.lifecycle_status import (
+    catalog_status_grants_execution_permission,
+    normalize_catalog_lifecycle_status,
+)
+
+
+ROOT = Path(__file__).parents[1]
+
+
+def _inventory():
+    return json.loads(
+        (ROOT / "docs/registry/cross_asset_strategy_inventory.json").read_text()
+    )
+
+
+def test_inventory_is_metadata_only_and_covers_non_us_domains():
+    data = _inventory()
+    assert data["inventory_only"] is True
+    assert "trading authority" in data["source_policy"]
+    domains = {entry["domain"] for entry in data["entries"]}
+    assert {"cn_equity", "hk_equity", "crypto"} <= domains
+
+
+def test_inventory_entries_have_explicit_next_actions_and_no_live_grant():
+    inventory = _inventory()
+    assert inventory["permission_effect"] == "none"
+    entries = inventory["entries"]
+    assert entries
+    ids = [entry["id"] for entry in entries]
+    assert len(ids) == len(set(ids))
+    for entry in entries:
+        assert entry["owner_repo"]
+        assert entry["catalog_status"]
+        assert entry["canonical_status"] == normalize_catalog_lifecycle_status(
+            entry["catalog_status"]
+        )
+        assert catalog_status_grants_execution_permission(entry["catalog_status"]) is False
+        assert entry["next_action"]
+        assert "live" not in entry.get("authority", "").lower()