docs: record reproducible risk artifact contract

README.md

@@ -139,6 +139,16 @@ print(factor_backtest.benchmark_stats(benchmark_returns))
 
 # 下游稳定交付：显式提供代码版本、数据快照和时区，不在核心层写数据库。
 from quant_engine.artifact import build_research_run_artifact
+from quant_engine.risk import CovarianceSnapshot
+
+risk_date = factor_backtest.position_weights.index[-1].date()
+risk_snapshot = CovarianceSnapshot(
+    snapshot_id="<covariance-snapshot-id>",
+    as_of_date=risk_date,
+    covariance=daily_return_covariance,
+    return_frequency="1d",
+    periods_per_year=252,
+)
 
 artifact = build_research_run_artifact(
     factor_backtest,
@@ -156,6 +166,7 @@ artifact = build_research_run_artifact(
     parameters={"top_k": 20, "lag_sessions": 1},
     benchmark_id="000300.SH",
     benchmark_returns=benchmark_returns,
+    risk_snapshots={risk_date: risk_snapshot},
 )
 print(artifact.manifest())
 

docs/OPEN_SOURCE_REFERENCES.md

@@ -27,6 +27,12 @@
 因此 `ResearchRunArtifact` 使用显式 `schema_version`、`config_hash`、代码版本和数据
 快照身份，并提供确定性 JSON / SHA-256 manifest；核心层仍不写数据库或 artifact store。
 
+schema `1.1.0` 将 Qlib 的独立 risk-analysis artifact 思路与 Riskfolio-Lib 的 Euler
+component-risk 语义结合，但只保留本项目需要的轻量合同：协方差快照必须声明
+`snapshot_id`、`as_of_date`、收益频率和年化期数；风险从成交后的实际日末持仓计算，
+component risk 闭合到年化组合波动，percentage contribution 闭合到 1。未来日期、资产
+标签不完整和零方差组合都直接失败，不以默认值伪造结果。
+
 ## hikyuu 的定位
 
 [hikyuu](https://github.com/fasiondog/hikyuu) 的 SG / MM / CN / PG 部件化思想、

docs/handoff/2026-08-21-research-artifact-contract.md

@@ -11,7 +11,7 @@
 - trades / realized positions / cash；
 - asset and daily return attribution；
 - performance including Sortino / TE / IR / alpha / beta；
-- reserved risk snapshot table；
+- reproducible covariance snapshots and annualized Euler component-risk facts；
 - canonical JSON / SHA-256 manifest。
 
 ## Branch stack
@@ -24,14 +24,22 @@
 
 ## Verification
 
-- `pytest -q --cov=src --cov-report=term-missing`: 520 passed，9 个既有 SciPy warning；
-- total coverage 91%，`artifact.py` 91%；
+- `pytest -q`: 526 passed，9 个既有 SciPy warning；
+- risk/artifact focused coverage 87.85%；
 - `mypy --strict src/`: 16 source files passed；
 - changed-scope Ruff: passed；
 - no runtime dependency added；
 - no database, network, broker or filesystem write side effect in artifact builder。
 
+## Current risk contract
+
+- artifact schema：`1.1.0`；
+- `CovarianceSnapshot` 对输入矩阵深拷贝并显式记录截至日、频率和年化期数；
+- `risk_snapshots` 按研究交易日映射，可只生成需要的风险观察日；
+- 使用成交后实际持仓，不包含现金风险资产；协方差资产标签必须与研究资产全集一致；
+- `covariance_as_of_date` 不得晚于 `trade_date`；无正组合方差时拒绝产物。
+
 ## Next action
 
-在 `research_results` 新建独立分支，实现只接受 `ResearchRunArtifact.table_frames()` 的
-ClickHouse / artifact-store adapter；先以 mock writer 做契约测试，不接触真实数据库。
+保持 Draft PR #5，不绕过堆叠顺序合并；下游 `research_results` / `research_platform`
+已在各自 Draft 分支兼容 1.0.0 / 1.1.0，下一阶段以真实数据适配器提供可复现协方差快照。

src/quant_engine/artifact.py

@@ -20,8 +20,9 @@ import numpy as np
 import pandas as pd
 
 from quant_engine.research_pipeline import FactorBacktestResult
+from quant_engine.risk import CovarianceSnapshot, labeled_component_risk
 
-RESEARCH_ARTIFACT_SCHEMA_VERSION = "1.0.0"
+RESEARCH_ARTIFACT_SCHEMA_VERSION = "1.1.0"
 
 RISK_COLUMNS = [
     "run_id",
@@ -32,6 +33,10 @@ RISK_COLUMNS = [
     "component_risk",
     "risk_contribution",
     "covariance_snapshot_id",
+    "covariance_as_of_date",
+    "risk_measure",
+    "return_frequency",
+    "periods_per_year",
 ]
 
 __all__ = [
@@ -372,6 +377,73 @@ def _build_attribution(
     return pd.DataFrame(rows), daily
 
 
+def _risk_trade_date(value: object) -> date:
+    try:
+        timestamp = pd.Timestamp(value)
+    except (TypeError, ValueError) as error:
+        raise ValueError("risk snapshot keys must be valid trade dates") from error
+    if pd.isna(timestamp):
+        raise ValueError("risk snapshot keys must be valid trade dates")
+    return date(int(timestamp.year), int(timestamp.month), int(timestamp.day))
+
+
+def _build_risk(
+    result: FactorBacktestResult,
+    run_id: str,
+    risk_snapshots: Mapping[object, CovarianceSnapshot] | None,
+) -> pd.DataFrame:
+    if risk_snapshots is None:
+        return pd.DataFrame(columns=RISK_COLUMNS)
+    if not isinstance(risk_snapshots, Mapping):
+        raise TypeError("risk_snapshots must be a mapping")
+
+    session_by_date = {
+        pd.Timestamp(session).date(): session for session in result.position_weights.index
+    }
+    normalized: dict[date, CovarianceSnapshot] = {}
+    for raw_trade_date, snapshot in risk_snapshots.items():
+        trade_date = _risk_trade_date(raw_trade_date)
+        if trade_date in normalized:
+            raise ValueError(f"duplicate risk snapshot trade date: {trade_date}")
+        if trade_date not in session_by_date:
+            raise ValueError(f"risk snapshot trade date {trade_date} must be a result session")
+        if not isinstance(snapshot, CovarianceSnapshot):
+            raise TypeError("risk snapshot values must be CovarianceSnapshot instances")
+        if snapshot.as_of_date > trade_date:
+            raise ValueError(
+                f"covariance as_of_date {snapshot.as_of_date} must not be after trade date "
+                f"{trade_date}"
+            )
+        normalized[trade_date] = snapshot
+
+    weights_by_date = result.position_weights
+    rows: list[dict[str, object]] = []
+    for trade_date in sorted(normalized):
+        snapshot = normalized[trade_date]
+        session = session_by_date[trade_date]
+        weights = weights_by_date.loc[session].astype(float, copy=True)
+        annualized_covariance = snapshot.covariance * snapshot.periods_per_year
+        decomposition = labeled_component_risk(weights, annualized_covariance)
+        for asset_id in weights.index:
+            rows.append(
+                {
+                    "run_id": run_id,
+                    "trade_date": trade_date,
+                    "asset_id": asset_id,
+                    "weight": float(weights.loc[asset_id]),
+                    "marginal_risk": float(decomposition.marginal.loc[asset_id]),
+                    "component_risk": float(decomposition.component.loc[asset_id]),
+                    "risk_contribution": float(decomposition.percentage.loc[asset_id]),
+                    "covariance_snapshot_id": snapshot.snapshot_id,
+                    "covariance_as_of_date": snapshot.as_of_date,
+                    "risk_measure": "annualized_volatility",
+                    "return_frequency": snapshot.return_frequency,
+                    "periods_per_year": snapshot.periods_per_year,
+                }
+            )
+    return pd.DataFrame(rows, columns=RISK_COLUMNS)
+
+
 def _build_performance(
     result: FactorBacktestResult,
     run_id: str,
@@ -429,6 +501,7 @@ def build_research_run_artifact(
     parameters: Mapping[str, object],
     benchmark_id: str | None = None,
     benchmark_returns: pd.Series | None = None,
+    risk_snapshots: Mapping[object, CovarianceSnapshot] | None = None,
 ) -> ResearchRunArtifact:
     """Snapshot one successful factor backtest into schema-versioned fact tables."""
     if not isinstance(result, FactorBacktestResult):
@@ -499,6 +572,6 @@ def build_research_run_artifact(
         _positions=_build_positions(result, normalized_run_id),
         _attribution=attribution,
         _attribution_daily=attribution_daily,
-        _risk=pd.DataFrame(columns=RISK_COLUMNS),
+        _risk=_build_risk(result, normalized_run_id, risk_snapshots),
         _performance=_build_performance(result, normalized_run_id, benchmark_returns),
     )

src/quant_engine/risk.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import date
 from typing import Any
 
 import numpy as np
@@ -14,6 +15,7 @@ from numpy.typing import NDArray
 
 __all__ = [
     "ComponentRiskResult",
+    "CovarianceSnapshot",
     "component_var",
     "labeled_component_risk",
     "marginal_risk_contribution",
@@ -21,6 +23,56 @@ __all__ = [
 ]
 
 
+@dataclass(frozen=True, slots=True, init=False, eq=False)
+class CovarianceSnapshot:
+    """Immutable-by-interface covariance input with explicit time semantics."""
+
+    snapshot_id: str
+    as_of_date: date
+    _covariance: pd.DataFrame
+    return_frequency: str
+    periods_per_year: int
+
+    def __init__(
+        self,
+        *,
+        snapshot_id: str,
+        as_of_date: str | date | pd.Timestamp,
+        covariance: pd.DataFrame,
+        return_frequency: str,
+        periods_per_year: int,
+    ) -> None:
+        if not isinstance(snapshot_id, str) or not snapshot_id.strip():
+            raise ValueError("snapshot_id must be non-empty")
+        if not isinstance(return_frequency, str) or not return_frequency.strip():
+            raise ValueError("return_frequency must be non-empty")
+        if isinstance(periods_per_year, bool) or not isinstance(periods_per_year, int):
+            raise TypeError("periods_per_year must be an integer")
+        if periods_per_year <= 0:
+            raise ValueError("periods_per_year must be positive")
+        if not isinstance(covariance, pd.DataFrame):
+            raise TypeError("covariance must be a pandas DataFrame")
+        if covariance.empty:
+            raise ValueError("covariance must contain at least one asset")
+        try:
+            normalized_as_of = pd.Timestamp(as_of_date)
+        except (TypeError, ValueError) as error:
+            raise ValueError("as_of_date must be a valid date") from error
+        if pd.isna(normalized_as_of):
+            raise ValueError("as_of_date must be a valid date")
+
+        object.__setattr__(self, "snapshot_id", snapshot_id.strip())
+        object.__setattr__(self, "as_of_date", normalized_as_of.date())
+        object.__setattr__(self, "_covariance", covariance.copy(deep=True))
+        object.__setattr__(self, "return_frequency", return_frequency.strip())
+        object.__setattr__(self, "periods_per_year", periods_per_year)
+
+    @property
+    def covariance(self) -> pd.DataFrame:
+        """Return an isolated copy so callers cannot mutate the snapshot."""
+        return self._covariance.copy(deep=True)
+
+
 @dataclass(frozen=True, slots=True, eq=False)
 class ComponentRiskResult:
     """Label-preserving Euler decomposition of portfolio volatility."""

tests/test_artifact.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from datetime import date
 
 import pandas as pd
 import pytest
@@ -14,6 +15,7 @@ from quant_engine.artifact import (
 )
 from quant_engine.execution import ExecutionConfig
 from quant_engine.research_pipeline import FactorBacktestResult, run_factor_backtest_research
+from quant_engine.risk import CovarianceSnapshot
 
 
 def _backtest_result() -> FactorBacktestResult:
@@ -51,6 +53,7 @@ def _build(
     result: FactorBacktestResult,
     *,
     parameters: dict[str, object] | None = None,
+    risk_snapshots: dict[date, CovarianceSnapshot] | None = None,
 ) -> ResearchRunArtifact:
     benchmark = pd.Series(
         [0.0, 0.01, -0.01, 0.02],
@@ -73,6 +76,7 @@ def _build(
         parameters=parameters or {"top_k": 1, "lag_sessions": 1},
         benchmark_id="000300.SH",
         benchmark_returns=benchmark,
+        risk_snapshots=risk_snapshots,
     )
 
 
@@ -132,6 +136,10 @@ def test_research_artifact_projects_versioned_queryable_fact_tables() -> None:
         "component_risk",
         "risk_contribution",
         "covariance_snapshot_id",
+        "covariance_as_of_date",
+        "risk_measure",
+        "return_frequency",
+        "periods_per_year",
     ]
     assert artifact.performance.loc[0, "n_trades"] == len(artifact.trades)
     assert artifact.performance.loc[0, "ir"] == pytest.approx(
@@ -142,6 +150,76 @@ def test_research_artifact_projects_versioned_queryable_fact_tables() -> None:
     assert "sortino" in artifact.performance.columns
 
 
+def test_research_artifact_projects_annualized_risk_from_actual_positions() -> None:
+    result = _backtest_result()
+    trade_date = result.position_weights.index[-1].date()
+    covariance = pd.DataFrame(
+        [[0.0001, 0.00002], [0.00002, 0.0004]],
+        index=["A", "B"],
+        columns=["A", "B"],
+    )
+    snapshot = CovarianceSnapshot(
+        snapshot_id="cov-20260107-v1",
+        as_of_date="2026-01-07",
+        covariance=covariance,
+        return_frequency="1d",
+        periods_per_year=252,
+    )
+
+    artifact = _build(result, risk_snapshots={trade_date: snapshot})
+
+    risk = artifact.risk.set_index("asset_id")
+    expected_weights = result.position_weights.loc[pd.Timestamp(trade_date)]
+    assert artifact.schema_version == "1.1.0"
+    assert risk.index.tolist() == ["A", "B"]
+    assert risk["weight"].tolist() == pytest.approx(expected_weights.tolist())
+    assert risk["covariance_snapshot_id"].unique().tolist() == ["cov-20260107-v1"]
+    assert risk["covariance_as_of_date"].unique().tolist() == [date(2026, 1, 7)]
+    assert risk["risk_measure"].unique().tolist() == ["annualized_volatility"]
+    assert risk["return_frequency"].unique().tolist() == ["1d"]
+    assert risk["periods_per_year"].unique().tolist() == [252]
+    assert risk["component_risk"].sum() == pytest.approx((0.0004 * 252) ** 0.5)
+    assert risk["risk_contribution"].sum() == pytest.approx(1.0)
+
+
+def test_research_artifact_rejects_future_or_misaligned_risk_snapshots() -> None:
+    result = _backtest_result()
+    trade_date = result.position_weights.index[-1].date()
+    covariance = pd.DataFrame(
+        [[0.0001, 0.0], [0.0, 0.0004]],
+        index=["A", "B"],
+        columns=["A", "B"],
+    )
+
+    with pytest.raises(ValueError, match="must not be after trade date"):
+        _build(
+            result,
+            risk_snapshots={
+                trade_date: CovarianceSnapshot(
+                    snapshot_id="future-covariance",
+                    as_of_date="2026-01-09",
+                    covariance=covariance,
+                    return_frequency="1d",
+                    periods_per_year=252,
+                )
+            },
+        )
+
+    with pytest.raises(ValueError, match="same asset labels"):
+        _build(
+            result,
+            risk_snapshots={
+                trade_date: CovarianceSnapshot(
+                    snapshot_id="incomplete-universe",
+                    as_of_date="2026-01-07",
+                    covariance=covariance.loc[["B"], ["B"]],
+                    return_frequency="1d",
+                    periods_per_year=252,
+                )
+            },
+        )
+
+
 def test_research_artifact_serialization_and_hashes_are_deterministic() -> None:
     result = _backtest_result()
     first = _build(result, parameters={"top_k": 1, "lag_sessions": 1})

tests/test_risk.py

@@ -8,6 +8,7 @@ import pytest
 
 from quant_engine.risk import (
     ComponentRiskResult,
+    CovarianceSnapshot,
     component_var,
     labeled_component_risk,
     marginal_risk_contribution,
@@ -15,6 +16,54 @@ from quant_engine.risk import (
 )
 
 
+def test_covariance_snapshot_is_validated_and_immutable_by_interface() -> None:
+    covariance = pd.DataFrame(
+        [[0.04, 0.01], [0.01, 0.09]],
+        index=["A", "B"],
+        columns=["A", "B"],
+    )
+    snapshot = CovarianceSnapshot(
+        snapshot_id="cov-20260107-v1",
+        as_of_date="2026-01-07",
+        covariance=covariance,
+        return_frequency="1d",
+        periods_per_year=252,
+    )
+
+    covariance.loc["A", "A"] = 999.0
+    leaked_copy = snapshot.covariance
+    leaked_copy.loc["B", "B"] = 999.0
+
+    assert snapshot.as_of_date == pd.Timestamp("2026-01-07").date()
+    assert snapshot.covariance.loc["A", "A"] == pytest.approx(0.04)
+    assert snapshot.covariance.loc["B", "B"] == pytest.approx(0.09)
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "message"),
+    [
+        ({"snapshot_id": ""}, "snapshot_id"),
+        ({"return_frequency": ""}, "return_frequency"),
+        ({"periods_per_year": 0}, "periods_per_year"),
+    ],
+)
+def test_covariance_snapshot_rejects_incomplete_identity(
+    kwargs: dict[str, object],
+    message: str,
+) -> None:
+    values: dict[str, object] = {
+        "snapshot_id": "cov-20260107-v1",
+        "as_of_date": "2026-01-07",
+        "covariance": pd.DataFrame([[0.04]], index=["A"], columns=["A"]),
+        "return_frequency": "1d",
+        "periods_per_year": 252,
+    }
+    values.update(kwargs)
+
+    with pytest.raises((TypeError, ValueError), match=message):
+        CovarianceSnapshot(**values)
+
+
 def test_risk_contribution_sums_to_one_for_positive_portfolio_variance() -> None:
     weights = np.array([0.5, 0.5])
     covariance = np.diag([1.0, 4.0])