docs: record covariance estimator decision

README.md

@@ -10,7 +10,7 @@
 
 | 仓库 | 角色 |
 |---|---|
-| `quant_engine` | **纯回测核心**（alpha + execution + indicators + data_adapter + backtest + metrics） |
+| `quant_engine` | **纯研究核心**（alpha + execution + ledger + attribution + risk + metrics） |
 | `research_results` | 业务集成（47 个 proj 调度 + 注册 + 平台对接） |
 | `tushare2db_pro_aoge` | 数据层（行情 ELT） |
 | `research_platform` | 展示层（FastAPI + Next.js） |
@@ -25,10 +25,12 @@
 - `backtest` — weight-based 多日仿真（rebalance_table / compute_nav / compare_to_benchmark）
 - `portfolio_construction` — 多期因子分数 → Top-K → 等权目标权重表
 - `research_pipeline` — 因子日 → 下一真实交易日 → 显式执行价 → 日末估值 → 成本后绩效（防前视编排）
-- `metrics` — 绩效（年化收益 / 波动率 / Sharpe / 最大回撤 / Calmar）
+- `artifact` — 版本化、确定性、存储中立的完整 research run 事实表与 manifest
+- `attribution` — 基于实际成交后持仓的隔夜 / 日内 / 交易成本逐日收益归因与闭合审计
+- `metrics` — 绝对绩效 + 严格日期对齐的 TE / IR / alpha / beta 基准相对绩效
 - `factor_library` — 通用方法（turnover / winsorize / IC / OLS / jb_test）
 - `portfolio_decomp` — 组合分解（risk_parity / mean_variance / 因子归因）
-- `risk` — 风险指标（边际 / 风险贡献）
+- `risk` — ndarray 低层风险公式 + 标签安全、可分组的 Euler 成分风险分解
 - `perf_stats` — 详细绩效（与 metrics 并存）
 - `logging` — 统一 logger（标准库 + 可选 loguru）
 
@@ -123,6 +125,50 @@ print(factor_backtest.returns)
 print(factor_backtest.stats())
 print(factor_backtest.execution.ledger_frame)
 print(factor_backtest.execution.trades_frame)
+print(factor_backtest.position_weights)  # 实际日末资产权重
+print(factor_backtest.cash_weights)
+
+# 所有分析都以实际成交后的 Ledger 为事实源，不直接使用目标权重伪造结果。
+attribution = factor_backtest.return_attribution()
+print(attribution.asset_contributions)
+print(attribution.transaction_cost)
+print(attribution.residual)  # 应接近 0；否则说明贡献未闭合到账本收益
+
+# benchmark_returns 必须与成本后 factor_backtest.returns 使用完全相同的日期索引。
+print(factor_backtest.benchmark_stats(benchmark_returns))
+
+# 下游稳定交付：显式提供代码版本、数据快照和时区，不在核心层写数据库。
+from quant_engine.artifact import build_research_run_artifact
+from quant_engine.risk import estimate_covariance_snapshot
+
+risk_date = factor_backtest.position_weights.index[-1].date()
+risk_snapshot = estimate_covariance_snapshot(
+    daily_asset_returns,
+    as_of_date=risk_date,
+    lookback_sessions=252,
+    min_observations=120,
+    data_snapshot_id="<risk-return-data-snapshot-id>",
+)
+
+artifact = build_research_run_artifact(
+    factor_backtest,
+    run_id="research-run-001",
+    strategy_id="alpha-top20",
+    strategy_name="Alpha Top 20",
+    strategy_version="1.0.0",
+    engine_version="1.2.0",
+    code_revision="<git-sha>",
+    data_snapshot_id="<data-snapshot-id>",
+    calendar="CN-A",
+    timezone="Asia/Shanghai",
+    started_at="2026-08-21T10:00:00+08:00",
+    finished_at="2026-08-21T10:01:00+08:00",
+    parameters={"top_k": 20, "lag_sessions": 1},
+    benchmark_id="000300.SH",
+    benchmark_returns=benchmark_returns,
+    risk_snapshots={risk_date: risk_snapshot},
+)
+print(artifact.manifest())
 
 # run_weight_backtest 是低层算子：只接受收益区间开始前已经生效的持仓权重。
 # 不要把 signal-date 的 factor_scores/decision_weights 直接传给它。

docs/OPEN_SOURCE_REFERENCES.md

@@ -0,0 +1,54 @@
+# Open-source design references
+
+本项目采用“借鉴稳定语义、保留轻量实现”的策略。引入新量化能力前先检查成熟
+开源案例；除非维护成本和许可证收益明确优于本地小型实现，否则不增加框架级依赖。
+
+## 2026-08-21：成交后归因与相对绩效
+
+| 项目 | 借鉴内容 | 当前决策 |
+|---|---|---|
+| [Qlib](https://github.com/microsoft/qlib) | 信号时间与交易时间分离、成本前后超额收益分开报告 | 借鉴语义；不引入完整框架 |
+| [Zipline](https://github.com/quantopian/zipline) | Ledger / transaction / portfolio value 状态模型 | 以现有 `ExecutionSimulationResult` 承担事实源 |
+| [empyrical](https://github.com/quantopian/empyrical) | beta 协方差口径、alpha 几何年化、年化因子 | 移植小型公式；不增加老旧运行时依赖 |
+| [Riskfolio-Lib](https://github.com/dcajasn/Riskfolio-Lib) | Euler component risk 与分组/因子风险贡献 | 只实现当前需要的 pandas/numpy 标签安全封装 |
+| [PyPortfolioOpt](https://github.com/PyPortfolio/PyPortfolioOpt) | 协方差估计与优化器解耦 | 留作未来风险模型适配器参考 |
+
+当前核心不新增依赖。逐日收益归因必须从实际换仓前后持仓、成交记录、执行价和
+收盘估值推导；因子分数与目标权重只是意图，不能作为成交后归因事实源。
+
+## 2026-08-21：研究运行工件
+
+- 借鉴 [Qlib Recorder / RecordTemplate](https://github.com/microsoft/qlib/blob/main/qlib/workflow/record_temp.py)
+  将 signal、portfolio analysis 和 risk analysis 分成稳定事实，但不引入 Qlib 运行时；
+- 借鉴 [MLflow Tracking](https://mlflow.org/docs/latest/tracking/) 的 run / params /
+  metrics / artifacts 分层，但 MLflow 只保留为未来可选 exporter；
+- HTML、PNG 和 tearsheet 是可再生展示物，不能替代 NAV、成交、持仓、归因和绩效事实。
+
+因此 `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。未来日期、资产
+标签不完整和零方差组合都直接失败，不以默认值伪造结果。
+
+## 2026-08-21：协方差快照估计
+
+| 项目 | 借鉴内容 | 当前决策 |
+|---|---|---|
+| [PyPortfolioOpt risk models](https://github.com/PyPortfolio/PyPortfolioOpt/blob/main/pypfopt/risk_models.py) | 将收益输入、协方差估计器和组合优化解耦；sample / EWM / shrinkage 使用统一标签输出 | 借鉴可替换估计器边界，不引入完整包 |
+| [scikit-learn covariance](https://github.com/scikit-learn/scikit-learn/blob/main/sklearn/covariance/_shrunk_covariance.py) | 维护成熟的 Ledoit–Wolf / OAS shrinkage 实现 | 未来作为可选 adapter；不复制统计公式 |
+| [Qlib structured risk model](https://github.com/microsoft/qlib/blob/main/qlib/model/riskmodel/structured.py) | PCA/FA 结构化协方差和固定随机状态 | 留作因子风险模型阶段，不进入当前 baseline |
+
+当前 `estimate_covariance_snapshot` 只编排 pandas 的 sample covariance：先按 `as_of_date`
+截断，再取固定 session 窗口，使用 complete-case 行并拒绝历史不足；禁止 pandas 默认的
+pairwise 样本集合产生含义不一致的矩阵。snapshot ID 对窗口数据、缺失掩码、上游数据
+快照身份和估计参数做 SHA-256，追加未来数据不会改变历史快照。
+
+## hikyuu 的定位
+
+[hikyuu](https://github.com/fasiondog/hikyuu) 的 SG / MM / CN / PG 部件化思想、
+A 股交易约束和系统组合方式仍有借鉴价值；但其完整 C++/Python 运行时、对象模型和
+数据体系不适合作为本项目核心依赖。当前原则是按真实研究链路吸收边界设计，不复制
+其框架层级，也不为了“架构完整”预先建设尚无端到端需求的抽象。

docs/handoff/2026-08-21-ledger-attribution.md

@@ -0,0 +1,42 @@
+# Ledger-backed attribution handoff
+
+## Goal
+
+在 `ExecutionSimulationResult` 日频 Ledger 之上增加轻量、可审计的成交后分析层：
+
+- 逐日隔夜 / 日内资产收益贡献；
+- 佣金、印花税、滑点成本独立贡献；
+- 贡献闭合到成本后日收益并显式暴露 residual；
+- 严格日期对齐的 TE / IR / alpha / beta；
+- 标签安全且可分组的 Euler component risk。
+- 从 Ledger 股数和收盘估值投影的实际资产 / 现金权重。
+
+## Branch stack
+
+- 当前：`codex/ledger-attribution-20260821`
+- 基线：`codex/post-execution-ledger-20260821`
+- 再下层：`codex/core-contracts-20260821`（PR #2，尚待用户确认合并）
+
+本分支不得直接合并到 `main`。应按上述顺序逐层审阅；未经用户明确确认，不得合并
+L2 PR。
+
+## Open-source decision
+
+调研结论记录在 `docs/OPEN_SOURCE_REFERENCES.md`。Qlib、Zipline、empyrical、
+Riskfolio-Lib 和 PyPortfolioOpt 只作为时间语义、Ledger、相对指标与 Euler 风险贡献
+的设计参考；本阶段没有新增运行时依赖。
+
+## Verification
+
+- `pytest -q --cov=src --cov-report=term-missing`: 514 passed，9 个既有 SciPy warning，91% coverage；
+- `mypy --strict src/`: 15 source files passed；
+- 变更范围 `ruff check`: passed；
+- 全仓 Ruff：仅 13 个既有 `tests/governance/*` PT009；
+- workspace verify/status：passed，预期提示 quant_engine 非 main；
+- global Gitea workflow check：passed，23 个无关仓库 warning。
+
+## Next action
+
+先按堆叠顺序审阅 PR。基础 Ledger 分支完成后，再将本分支 rebase 到其最终提交，
+运行唯一一次 `ship --ready`；随后将稳定输出适配到 `research_results` 与
+`research_platform`，不要在核心层直接写数据库。

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

@@ -0,0 +1,48 @@
+# Research artifact contract handoff
+
+## Goal
+
+把完整可信研究链固化成存储中立、版本化、确定性的 `ResearchRunArtifact`，供
+`research_results` 持久化和 `research_platform` 查询：
+
+- run identity / schema version / config hash / code revision / data snapshot；
+- signal scores / decision weights / signal-to-execution mapping；
+- NAV / returns / benchmark / costs；
+- trades / realized positions / cash；
+- asset and daily return attribution；
+- performance including Sortino / TE / IR / alpha / beta；
+- reproducible covariance snapshots and annualized Euler component-risk facts；
+- canonical JSON / SHA-256 manifest。
+
+## Branch stack
+
+- 当前：`codex/research-artifact-contract-20260821`
+- 基线：`codex/ledger-attribution-20260821`（Draft PR #4）
+- 下层：Draft PR #3 → Ready PR #2 → `main`
+
+不得绕过堆叠顺序直接合并到 `main`。
+
+## Verification
+
+- `pytest -q`: 529 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`；无正组合方差时拒绝产物。
+- `estimate_covariance_snapshot` 从显式数据快照的日收益生成无前视、complete-case、
+  SHA-256 可复现的 per-period sample covariance；不包含 I/O 或未来行。
+
+## Next action
+
+保持 Draft PR #5，不绕过堆叠顺序合并；下游 `research_results` / `research_platform`
+已在各自 Draft 分支兼容 1.0.0 / 1.1.0，下一阶段让市场数据适配器提供带稳定
+`data_snapshot_id` 的资产日收益，再评估可选 scikit-learn shrinkage adapter。

src/quant_engine/artifact.py

@@ -0,0 +1,577 @@
+"""Versioned, deterministic research-run artifacts for downstream adapters.
+
+This module is deliberately storage-neutral.  It snapshots a completed
+``FactorBacktestResult`` into queryable fact tables but never writes a database,
+starts a service, or talks to a broker.  ``research_results`` owns persistence;
+``research_platform`` owns read models and presentation.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from collections.abc import Mapping
+from dataclasses import dataclass
+from datetime import date, datetime
+from typing import Any
+
+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.1.0"
+
+RISK_COLUMNS = [
+    "run_id",
+    "trade_date",
+    "asset_id",
+    "weight",
+    "marginal_risk",
+    "component_risk",
+    "risk_contribution",
+    "covariance_snapshot_id",
+    "covariance_as_of_date",
+    "risk_measure",
+    "return_frequency",
+    "periods_per_year",
+]
+
+__all__ = [
+    "RESEARCH_ARTIFACT_SCHEMA_VERSION",
+    "ResearchRunArtifact",
+    "build_research_run_artifact",
+]
+
+
+def _frame_copy(frame: pd.DataFrame) -> pd.DataFrame:
+    return frame.copy(deep=True)
+
+
+@dataclass(frozen=True, slots=True, eq=False)
+class ResearchRunArtifact:
+    """Immutable-by-interface snapshot of one completed research run."""
+
+    schema_version: str
+    _run: pd.DataFrame
+    _signals: pd.DataFrame
+    _nav: pd.DataFrame
+    _trades: pd.DataFrame
+    _positions: pd.DataFrame
+    _attribution: pd.DataFrame
+    _attribution_daily: pd.DataFrame
+    _risk: pd.DataFrame
+    _performance: pd.DataFrame
+
+    @property
+    def run(self) -> pd.DataFrame:
+        return _frame_copy(self._run)
+
+    @property
+    def nav(self) -> pd.DataFrame:
+        return _frame_copy(self._nav)
+
+    @property
+    def signals(self) -> pd.DataFrame:
+        return _frame_copy(self._signals)
+
+    @property
+    def trades(self) -> pd.DataFrame:
+        return _frame_copy(self._trades)
+
+    @property
+    def positions(self) -> pd.DataFrame:
+        return _frame_copy(self._positions)
+
+    @property
+    def attribution(self) -> pd.DataFrame:
+        return _frame_copy(self._attribution)
+
+    @property
+    def attribution_daily(self) -> pd.DataFrame:
+        return _frame_copy(self._attribution_daily)
+
+    @property
+    def risk(self) -> pd.DataFrame:
+        return _frame_copy(self._risk)
+
+    @property
+    def performance(self) -> pd.DataFrame:
+        return _frame_copy(self._performance)
+
+    def table_frames(self) -> Mapping[str, pd.DataFrame]:
+        """Return isolated table snapshots keyed by stable logical table name."""
+        return {
+            "run": self.run,
+            "signals": self.signals,
+            "nav": self.nav,
+            "trades": self.trades,
+            "positions": self.positions,
+            "attribution": self.attribution,
+            "attribution_daily": self.attribution_daily,
+            "risk": self.risk,
+            "performance": self.performance,
+        }
+
+    def canonical_json(self) -> str:
+        """Serialize tables deterministically for checksums and artifact storage."""
+        payload = {
+            "schema_version": self.schema_version,
+            "tables": {
+                name: _frame_records(frame)
+                for name, frame in self._internal_table_frames().items()
+            },
+        }
+        return json.dumps(
+            payload,
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+
+    @property
+    def content_sha256(self) -> str:
+        return hashlib.sha256(self.canonical_json().encode("utf-8")).hexdigest()
+
+    def manifest(self) -> Mapping[str, object]:
+        """Return a compact immutable identity and row-count manifest."""
+        return {
+            "schema_version": self.schema_version,
+            "run_id": str(self._run.at[0, "run_id"]),
+            "config_hash": str(self._run.at[0, "config_hash"]),
+            "content_sha256": self.content_sha256,
+            "tables": {
+                name: len(frame) for name, frame in self._internal_table_frames().items()
+            },
+        }
+
+    def _internal_table_frames(self) -> Mapping[str, pd.DataFrame]:
+        return {
+            "run": self._run,
+            "signals": self._signals,
+            "nav": self._nav,
+            "trades": self._trades,
+            "positions": self._positions,
+            "attribution": self._attribution,
+            "attribution_daily": self._attribution_daily,
+            "risk": self._risk,
+            "performance": self._performance,
+        }
+
+
+def _required_text(value: str, name: str, *, max_length: int | None = None) -> str:
+    normalized = value.strip()
+    if not normalized:
+        raise ValueError(f"{name} must be non-empty")
+    if max_length is not None and len(normalized) > max_length:
+        raise ValueError(f"{name} must contain at most {max_length} characters")
+    return normalized
+
+
+def _aware_timestamp(value: str | pd.Timestamp, name: str) -> pd.Timestamp:
+    try:
+        timestamp = pd.Timestamp(value)
+    except (TypeError, ValueError) as error:
+        raise ValueError(f"{name} must be a valid timestamp") from error
+    if timestamp.tzinfo is None:
+        raise ValueError(f"{name} must include a timezone")
+    return timestamp
+
+
+def _json_value(value: object) -> object:
+    if value is None or isinstance(value, str | bool | int):
+        return value
+    if isinstance(value, float):
+        return value if math.isfinite(value) else None
+    if isinstance(value, np.generic):
+        return _json_value(value.item())
+    if isinstance(value, pd.Timestamp):
+        return value.isoformat()
+    if isinstance(value, datetime):
+        return value.isoformat()
+    if isinstance(value, date):
+        return value.isoformat()
+    if isinstance(value, Mapping):
+        return {
+            str(key): _json_value(item)
+            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
+        }
+    if isinstance(value, list | tuple):
+        return [_json_value(item) for item in value]
+    raise TypeError(f"value of type {type(value).__name__} is not JSON serializable")
+
+
+def _canonical_mapping_json(values: Mapping[str, object]) -> str:
+    normalized = _json_value(values)
+    return json.dumps(
+        normalized,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    )
+
+
+def _frame_records(frame: pd.DataFrame) -> list[dict[str, object]]:
+    return [
+        {str(key): _json_value(value) for key, value in row.items()}
+        for row in frame.to_dict(orient="records")
+    ]
+
+
+def _build_nav(
+    result: FactorBacktestResult,
+    run_id: str,
+    benchmark_returns: pd.Series | None,
+) -> pd.DataFrame:
+    nav = result.execution.ledger_frame.copy(deep=True)
+    nav.insert(0, "run_id", run_id)
+    nav["trade_date"] = pd.to_datetime(nav["trade_date"]).dt.date
+    nav["total_cost"] = [
+        sum(execution.total_cost for execution in daily.executions)
+        for daily in result.execution.daily_executions
+    ]
+    if benchmark_returns is None:
+        nav["benchmark_nav"] = np.nan
+        nav["benchmark_return"] = np.nan
+        nav["excess_ret"] = np.nan
+    else:
+        benchmark = benchmark_returns.astype(float, copy=True)
+        nav["benchmark_nav"] = (1.0 + benchmark).cumprod().to_numpy()
+        nav["benchmark_return"] = benchmark.to_numpy()
+        nav["excess_ret"] = result.returns.to_numpy() - benchmark.to_numpy()
+    return nav
+
+
+def _build_signals(result: FactorBacktestResult, run_id: str) -> pd.DataFrame:
+    columns = [
+        "run_id",
+        "signal_date",
+        "execution_date",
+        "asset_id",
+        "factor_score",
+        "target_weight",
+    ]
+    rows: list[dict[str, object]] = []
+    for signal_date, scores in result.factor_scores.iterrows():
+        execution_date = pd.Timestamp(result.schedule.signal_to_execution.at[signal_date]).date()
+        for asset, score in scores.items():
+            rows.append(
+                {
+                    "run_id": run_id,
+                    "signal_date": pd.Timestamp(signal_date).date(),
+                    "execution_date": execution_date,
+                    "asset_id": asset,
+                    "factor_score": float(score),
+                    "target_weight": float(
+                        result.schedule.decision_weights.at[signal_date, asset]
+                    ),
+                }
+            )
+    return pd.DataFrame(rows, columns=columns)
+
+
+def _build_trades(result: FactorBacktestResult, run_id: str) -> pd.DataFrame:
+    trades = result.execution.trades_frame.copy(deep=True)
+    trades.insert(0, "run_id", run_id)
+    trades["trade_date"] = pd.to_datetime(trades["trade_date"]).dt.date
+    trades.insert(
+        1,
+        "trade_id",
+        [f"{run_id}:{sequence:08d}" for sequence in range(1, len(trades) + 1)],
+    )
+    signal_by_execution = {
+        pd.Timestamp(execution_date).date(): pd.Timestamp(signal_date).date()
+        for signal_date, execution_date in result.schedule.signal_to_execution.items()
+    }
+    trades["signal_id"] = [
+        f"{run_id}:signal:{signal_by_execution[trade_date].isoformat()}"
+        for trade_date in trades["trade_date"]
+    ]
+    trades["total_cost"] = trades["fee"] + trades["slippage"]
+    return trades
+
+
+def _build_positions(result: FactorBacktestResult, run_id: str) -> pd.DataFrame:
+    columns = [
+        "run_id",
+        "trade_date",
+        "asset_id",
+        "asset_type",
+        "quantity",
+        "mark_price",
+        "market_value",
+        "weight",
+    ]
+    rows: list[dict[str, object]] = []
+    weights = result.position_weights
+    cash_weights = result.cash_weights
+    for date_value, position in zip(
+        result.valuation_prices.index,
+        result.execution.positions,
+        strict=True,
+    ):
+        session_date = pd.Timestamp(date_value).date()
+        for asset, quantity in position.holdings.items():
+            mark_price = float(result.valuation_prices.at[date_value, asset])
+            rows.append(
+                {
+                    "run_id": run_id,
+                    "trade_date": session_date,
+                    "asset_id": asset,
+                    "asset_type": "security",
+                    "quantity": quantity,
+                    "mark_price": mark_price,
+                    "market_value": quantity * mark_price,
+                    "weight": float(weights.at[date_value, asset]),
+                }
+            )
+        rows.append(
+            {
+                "run_id": run_id,
+                "trade_date": session_date,
+                "asset_id": "CASH",
+                "asset_type": "cash",
+                "quantity": position.cash,
+                "mark_price": 1.0,
+                "market_value": position.cash,
+                "weight": float(cash_weights.at[date_value]),
+            }
+        )
+    return pd.DataFrame(rows, columns=columns)
+
+
+def _build_attribution(
+    result: FactorBacktestResult,
+    run_id: str,
+) -> tuple[pd.DataFrame, pd.DataFrame]:
+    contribution = result.return_attribution()
+    rows: list[dict[str, object]] = []
+    for date_value in contribution.overnight.index:
+        for asset in contribution.overnight.columns:
+            overnight = float(contribution.overnight.at[date_value, asset])
+            intraday = float(contribution.intraday.at[date_value, asset])
+            rows.append(
+                {
+                    "run_id": run_id,
+                    "trade_date": pd.Timestamp(date_value).date(),
+                    "asset_id": asset,
+                    "overnight": overnight,
+                    "intraday": intraday,
+                    "asset_total": overnight + intraday,
+                }
+            )
+    daily = pd.DataFrame(
+        {
+            "run_id": run_id,
+            "trade_date": contribution.total_return.index.date,
+            "transaction_cost": contribution.transaction_cost.to_numpy(),
+            "explained_return": contribution.explained_return.to_numpy(),
+            "residual": contribution.residual.to_numpy(),
+            "total_return": contribution.total_return.to_numpy(),
+        }
+    )
+    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,
+    benchmark_returns: pd.Series | None,
+) -> pd.DataFrame:
+    stats = result.stats()
+    relative = (
+        result.benchmark_stats(benchmark_returns)
+        if benchmark_returns is not None
+        else {
+            "tracking_error": float("nan"),
+            "information_ratio": float("nan"),
+            "alpha": float("nan"),
+            "beta": float("nan"),
+        }
+    )
+    total_return = float(result.nav.iloc[-1] - 1.0)
+    return pd.DataFrame(
+        [
+            {
+                "run_id": run_id,
+                "total_ret": total_return,
+                "ann_ret": stats["ann_return"],
+                "ann_volatility": stats["ann_volatility"],
+                "sharpe": stats["sharpe"],
+                "sortino": stats["sortino"],
+                "max_dd": stats["max_drawdown"],
+                "calmar": stats["calmar"],
+                "win_rate": stats["win_rate"],
+                "tracking_error": relative["tracking_error"],
+                "ir": relative["information_ratio"],
+                "alpha": relative["alpha"],
+                "beta": relative["beta"],
+                "n_trades": len(result.execution.trades_frame),
+                "n_days": len(result.returns),
+            }
+        ]
+    )
+
+
+def build_research_run_artifact(
+    result: FactorBacktestResult,
+    *,
+    run_id: str,
+    strategy_id: str,
+    strategy_name: str,
+    strategy_version: str,
+    engine_version: str,
+    code_revision: str,
+    data_snapshot_id: str,
+    calendar: str,
+    timezone: str,
+    started_at: str | pd.Timestamp,
+    finished_at: str | pd.Timestamp,
+    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):
+        raise TypeError("result must be a FactorBacktestResult")
+    if result.nav.empty:
+        raise ValueError("result must contain at least one research session")
+    normalized_run_id = _required_text(run_id, "run_id", max_length=64)
+    normalized_strategy_id = _required_text(strategy_id, "strategy_id")
+    normalized_strategy_name = _required_text(strategy_name, "strategy_name")
+    normalized_strategy_version = _required_text(strategy_version, "strategy_version")
+    normalized_engine_version = _required_text(engine_version, "engine_version")
+    normalized_code_revision = _required_text(code_revision, "code_revision")
+    normalized_snapshot = _required_text(data_snapshot_id, "data_snapshot_id")
+    normalized_calendar = _required_text(calendar, "calendar")
+    normalized_timezone = _required_text(timezone, "timezone")
+    if not isinstance(parameters, Mapping):
+        raise TypeError("parameters must be a mapping")
+
+    started = _aware_timestamp(started_at, "started_at")
+    finished = _aware_timestamp(finished_at, "finished_at")
+    if finished < started:
+        raise ValueError("finished_at must not precede started_at")
+    if (benchmark_id is None) != (benchmark_returns is None):
+        raise ValueError("benchmark_id and benchmark_returns must be provided together")
+    normalized_benchmark = ""
+    if benchmark_id is not None:
+        normalized_benchmark = _required_text(benchmark_id, "benchmark_id")
+        result.benchmark_stats(benchmark_returns)
+
+    params_json = _canonical_mapping_json(parameters)
+    config_hash = hashlib.sha256(params_json.encode("utf-8")).hexdigest()
+    run = pd.DataFrame(
+        [
+            {
+                "schema_version": RESEARCH_ARTIFACT_SCHEMA_VERSION,
+                "run_id": normalized_run_id,
+                "strategy_id": normalized_strategy_id,
+                "strategy_name": normalized_strategy_name,
+                "strategy_version": normalized_strategy_version,
+                "engine_version": normalized_engine_version,
+                "code_revision": normalized_code_revision,
+                "config_hash": config_hash,
+                "data_snapshot_id": normalized_snapshot,
+                "benchmark_id": normalized_benchmark,
+                "benchmark_alignment_policy": (
+                    "exact_session_index" if benchmark_returns is not None else "none"
+                ),
+                "frequency": "1d",
+                "calendar": normalized_calendar,
+                "timezone": normalized_timezone,
+                "initial_capital": result.execution.initial_cash,
+                "start_date": result.nav.index[0].date(),
+                "end_date": result.nav.index[-1].date(),
+                "status": "success",
+                "started_at": started,
+                "finished_at": finished,
+                "params_json": params_json,
+            }
+        ]
+    )
+    attribution, attribution_daily = _build_attribution(result, normalized_run_id)
+    return ResearchRunArtifact(
+        schema_version=RESEARCH_ARTIFACT_SCHEMA_VERSION,
+        _run=run,
+        _signals=_build_signals(result, normalized_run_id),
+        _nav=_build_nav(result, normalized_run_id, benchmark_returns),
+        _trades=_build_trades(result, normalized_run_id),
+        _positions=_build_positions(result, normalized_run_id),
+        _attribution=attribution,
+        _attribution_daily=attribution_daily,
+        _risk=_build_risk(result, normalized_run_id, risk_snapshots),
+        _performance=_build_performance(result, normalized_run_id, benchmark_returns),
+    )

src/quant_engine/attribution.py

@@ -0,0 +1,141 @@
+"""Post-execution daily return attribution derived from the portfolio ledger.
+
+The ledger is the source of truth: previous-close holdings explain overnight
+PnL, current-close holdings explain intraday PnL, and actual execution costs
+remain a separate contribution.  Target weights and factor scores are not
+accepted here because they are intentions rather than realized positions.
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+
+import pandas as pd
+
+from quant_engine.execution import ExecutionSimulationResult
+
+__all__ = ["DailyReturnAttribution", "compute_daily_return_attribution"]
+
+
+@dataclass(frozen=True, slots=True, eq=False)
+class DailyReturnAttribution:
+    """Auditable decomposition of each net portfolio return."""
+
+    overnight: pd.DataFrame
+    intraday: pd.DataFrame
+    transaction_cost: pd.Series
+    residual: pd.Series
+    total_return: pd.Series
+
+    @property
+    def asset_contributions(self) -> pd.DataFrame:
+        """Return the combined overnight and intraday contribution by asset."""
+        return self.overnight + self.intraday
+
+    @property
+    def explained_return(self) -> pd.Series:
+        """Return asset contributions plus execution costs, before residual."""
+        explained = self.asset_contributions.sum(axis=1) + self.transaction_cost
+        return explained.rename("explained_return")
+
+
+def _validate_prices(
+    execution: ExecutionSimulationResult,
+    execution_prices: pd.DataFrame,
+    valuation_prices: pd.DataFrame,
+) -> pd.DatetimeIndex:
+    if not isinstance(execution_prices, pd.DataFrame):
+        raise TypeError("execution_prices must be a pandas DataFrame")
+    if not isinstance(valuation_prices, pd.DataFrame):
+        raise TypeError("valuation_prices must be a pandas DataFrame")
+    if not isinstance(execution_prices.index, pd.DatetimeIndex):
+        raise TypeError("execution_prices must use a DatetimeIndex")
+    if not execution_prices.index.equals(valuation_prices.index):
+        raise ValueError("execution and valuation prices must use matching trading calendars")
+    if not execution_prices.columns.equals(valuation_prices.columns):
+        raise ValueError("execution and valuation prices must use matching asset labels")
+
+    ledger_index = pd.DatetimeIndex(pd.Timestamp(position.date) for position in execution.positions)
+    if not ledger_index.equals(execution_prices.index):
+        raise ValueError("ledger and price histories must use matching trading calendars")
+    if len(execution.positions) != len(execution.daily_executions):
+        raise ValueError("ledger positions and executions must have matching lengths")
+    return execution_prices.index.copy()
+
+
+def _price_for_held_asset(
+    prices: pd.DataFrame,
+    date: pd.Timestamp,
+    asset: str,
+    stage: str,
+) -> float:
+    if asset not in prices.columns:
+        raise ValueError(f"missing {stage} price for held asset {asset} on {date}")
+    price = float(prices.at[date, asset])
+    if not math.isfinite(price) or price <= 0:
+        raise ValueError(f"invalid {stage} price for held asset {asset} on {date}")
+    return price
+
+
+def compute_daily_return_attribution(
+    execution: ExecutionSimulationResult,
+    execution_prices: pd.DataFrame,
+    valuation_prices: pd.DataFrame,
+) -> DailyReturnAttribution:
+    """Decompose net daily returns using realized pre/post-execution holdings.
+
+    For each session, previous-close shares earn the move from the previous
+    close to the current execution price; current-close shares earn the move
+    from execution price to current close.  Actual commissions, stamp tax and
+    slippage are divided by the same previous NAV denominator.  ``residual``
+    exposes any failure of those components to close to the ledger return.
+    """
+    index = _validate_prices(execution, execution_prices, valuation_prices)
+    columns = execution_prices.columns.copy()
+    overnight = pd.DataFrame(0.0, index=index.copy(), columns=columns)
+    intraday = pd.DataFrame(0.0, index=index.copy(), columns=columns)
+    cost = pd.Series(0.0, index=index.copy(), name="transaction_cost")
+
+    previous_holdings: dict[str, float] = {}
+    previous_nav = execution.initial_cash
+    for row_number, (date, position, daily) in enumerate(
+        zip(index, execution.positions, execution.daily_executions, strict=True)
+    ):
+        if previous_nav <= 0 or not math.isfinite(previous_nav):
+            raise ValueError(f"previous portfolio value must be positive and finite on {date}")
+
+        for asset, shares in previous_holdings.items():
+            execution_price = _price_for_held_asset(
+                execution_prices, date, asset, "execution"
+            )
+            previous_close = _price_for_held_asset(
+                valuation_prices, index[row_number - 1], asset, "previous valuation"
+            )
+            overnight.at[date, asset] = shares * (execution_price - previous_close) / previous_nav
+
+        for asset, shares in position.holdings.items():
+            execution_price = _price_for_held_asset(
+                execution_prices, date, asset, "execution"
+            )
+            close_price = _price_for_held_asset(valuation_prices, date, asset, "valuation")
+            intraday.at[date, asset] = shares * (close_price - execution_price) / previous_nav
+
+        cost.at[date] = -sum(item.total_cost for item in daily.executions) / previous_nav
+        previous_holdings = position.holdings
+        previous_nav = position.portfolio_value
+
+    total_return = pd.Series(
+        execution.daily_returns.to_numpy(copy=True),
+        index=index.copy(),
+        name="total_return",
+    )
+    explained = (overnight + intraday).sum(axis=1) + cost
+    residual = (total_return - explained).rename("residual")
+    return DailyReturnAttribution(
+        overnight=overnight,
+        intraday=intraday,
+        transaction_cost=cost,
+        residual=residual,
+        total_return=total_return,
+    )

src/quant_engine/metrics.py

@@ -65,6 +65,20 @@ def sharpe_ratio(r: pd.Series, rf: float = 0.0) -> float:
     return (annualized_return(r) - rf) / vol
 
 
+def sortino_ratio(r: pd.Series, rf: float = 0.0) -> float:
+    """Sortino = (年化收益 - rf) / 年化下行偏差。"""
+    r = _clean(r)
+    if len(r) < 2:
+        return 0.0
+    downside = np.minimum(r.to_numpy(dtype=float), 0.0)
+    downside_deviation = float(
+        np.sqrt(np.mean(np.square(downside))) * np.sqrt(TRADING_DAYS_PER_YEAR)
+    )
+    if downside_deviation == 0:
+        return 0.0
+    return (annualized_return(r) - rf) / downside_deviation
+
+
 def max_drawdown(r: pd.Series) -> float:
     """最大回撤（负数）。例如 -0.2 表示最大亏 20%。"""
     r = _clean(r)
@@ -120,6 +134,7 @@ def summary(r: pd.Series, rf: float = 0.0) -> Mapping[str, float]:
         "ann_return": ann_ret,
         "ann_volatility": ann_vol,
         "sharpe": sharpe_ratio(r, rf),
+        "sortino": sortino_ratio(r, rf),
         "max_drawdown": mdd,
         "calmar": calmar_ratio(r),
         "win_rate": win_rate(r),
@@ -130,6 +145,62 @@ def summary(r: pd.Series, rf: float = 0.0) -> Mapping[str, float]:
     }
 
 
+def benchmark_summary(
+    portfolio_returns: pd.Series,
+    benchmark_returns: pd.Series,
+    *,
+    risk_free_daily: float = 0.0,
+    annualization: int = TRADING_DAYS_PER_YEAR,
+) -> Mapping[str, float]:
+    """计算成本后组合相对基准的严格对齐绩效。
+
+    与通用 ``summary`` 不同，本函数拒绝静默清洗或日期 inner join。alpha
+    使用日频回归截距的几何年化；基准方差不足时 alpha/beta 为 NaN，明确
+    表示回归不可估计。
+    """
+    portfolio, benchmark = _validate_benchmark_inputs(
+        portfolio_returns,
+        benchmark_returns,
+    )
+    if isinstance(annualization, bool) or not isinstance(annualization, int):
+        raise TypeError("annualization must be an integer")
+    if annualization <= 0:
+        raise ValueError("annualization must be positive")
+    if not np.isfinite(risk_free_daily):
+        raise ValueError("risk_free_daily must be finite")
+
+    active = portfolio - benchmark
+    active_std = float(active.std())
+    tracking_error = active_std * float(np.sqrt(annualization))
+    information_ratio = (
+        float(active.mean()) / active_std * float(np.sqrt(annualization))
+        if active_std >= 1e-30
+        else float("nan")
+    )
+
+    adjusted_portfolio = portfolio - risk_free_daily
+    adjusted_benchmark = benchmark - risk_free_daily
+    benchmark_variance = float(adjusted_benchmark.var())
+    if benchmark_variance < 1e-30:
+        beta = float("nan")
+        alpha = float("nan")
+    else:
+        beta = float(adjusted_portfolio.cov(adjusted_benchmark) / benchmark_variance)
+        alpha_daily = float((adjusted_portfolio - beta * adjusted_benchmark).mean())
+        alpha = (
+            float((1.0 + alpha_daily) ** annualization - 1.0)
+            if alpha_daily > -1.0
+            else float("nan")
+        )
+    return {
+        "n_observations": len(portfolio),
+        "tracking_error": tracking_error,
+        "information_ratio": information_ratio,
+        "alpha": alpha,
+        "beta": beta,
+    }
+
+
 # ── 内部 ──────────────────────────────────────
 
 
@@ -138,3 +209,29 @@ def _clean(r: pd.Series) -> pd.Series:
     if not isinstance(r, pd.Series):
         raise TypeError(f"expected pd.Series, got {type(r).__name__}")
     return r.replace([np.inf, -np.inf], np.nan).dropna()
+
+
+def _validate_benchmark_inputs(
+    portfolio_returns: pd.Series,
+    benchmark_returns: pd.Series,
+) -> tuple[pd.Series, pd.Series]:
+    if not isinstance(portfolio_returns, pd.Series):
+        raise TypeError("portfolio_returns must be a pandas Series")
+    if not isinstance(benchmark_returns, pd.Series):
+        raise TypeError("benchmark_returns must be a pandas Series")
+    if not portfolio_returns.index.equals(benchmark_returns.index):
+        raise ValueError("portfolio and benchmark returns must use matching indexes")
+    if not portfolio_returns.index.is_unique:
+        raise ValueError("portfolio and benchmark indexes must be unique")
+    if len(portfolio_returns) < 2:
+        raise ValueError("benchmark metrics require at least two observations")
+
+    portfolio = portfolio_returns.astype(float, copy=True)
+    benchmark = benchmark_returns.astype(float, copy=True)
+    if not np.isfinite(portfolio.to_numpy()).all() or not np.isfinite(
+        benchmark.to_numpy()
+    ).all():
+        raise ValueError("portfolio and benchmark returns must be finite")
+    if (portfolio < -1.0).any() or (benchmark < -1.0).any():
+        raise ValueError("simple returns cannot be less than -1")
+    return portfolio, benchmark

src/quant_engine/research_pipeline.py

@@ -16,13 +16,14 @@ import numpy as np
 import pandas as pd
 from pandas.api.types import is_numeric_dtype
 
+from quant_engine.attribution import DailyReturnAttribution, compute_daily_return_attribution
 from quant_engine.execution import (
     ExecutionConfig,
     ExecutionSimulationResult,
     simulate_daily_ledger_with_audit,
     simulate_multi_day_with_audit,
 )
-from quant_engine.metrics import summary as metrics_summary
+from quant_engine.metrics import benchmark_summary, summary as metrics_summary
 from quant_engine.portfolio_construction import scores_to_weight_table
 
 __all__ = [
@@ -86,10 +87,63 @@ class FactorBacktestResult:
             name="returns",
         )
 
+    @property
+    def position_weights(self) -> pd.DataFrame:
+        """按日末实际股数、收盘估值和账本 NAV 投影资产权重。"""
+        weights = pd.DataFrame(
+            0.0,
+            index=self.valuation_prices.index.copy(),
+            columns=self.valuation_prices.columns.copy(),
+        )
+        for date, position in zip(
+            self.valuation_prices.index,
+            self.execution.positions,
+            strict=True,
+        ):
+            if position.portfolio_value <= 0:
+                raise ValueError(f"portfolio value must be positive on {date}")
+            for asset, shares in position.holdings.items():
+                weights.at[date, asset] = (
+                    shares * float(self.valuation_prices.at[date, asset])
+                    / position.portfolio_value
+                )
+        return weights
+
+    @property
+    def cash_weights(self) -> pd.Series:
+        """返回与实际资产权重使用同一日末 NAV 分母的现金权重。"""
+        values = []
+        for date, position in zip(
+            self.valuation_prices.index,
+            self.execution.positions,
+            strict=True,
+        ):
+            if position.portfolio_value <= 0:
+                raise ValueError(f"portfolio value must be positive on {date}")
+            values.append(position.cash / position.portfolio_value)
+        return pd.Series(
+            values,
+            index=self.valuation_prices.index.copy(),
+            dtype=float,
+            name="cash_weight",
+        )
+
     def stats(self, rf: float = 0.0) -> Mapping[str, float]:
         """复用标准绩效口径计算指标。"""
         return metrics_summary(self.returns, rf)
 
+    def return_attribution(self) -> DailyReturnAttribution:
+        """从实际成交后持仓与账本生成逐日净收益归因。"""
+        return compute_daily_return_attribution(
+            self.execution,
+            self.execution_prices,
+            self.valuation_prices,
+        )
+
+    def benchmark_stats(self, benchmark_returns: pd.Series) -> Mapping[str, float]:
+        """计算成本后日收益相对同日基准的 TE、IR、alpha 与 beta。"""
+        return benchmark_summary(self.returns, benchmark_returns)
+
 
 def _validate_datetime_index(index: pd.Index, name: str) -> pd.DatetimeIndex:
     if not isinstance(index, pd.DatetimeIndex):

src/quant_engine/risk.py

@@ -5,11 +5,298 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
+from dataclasses import dataclass
+from datetime import date
 from typing import Any
 
 import numpy as np
+import pandas as pd
 from numpy.typing import NDArray
 
+__all__ = [
+    "ComponentRiskResult",
+    "CovarianceSnapshot",
+    "component_var",
+    "estimate_covariance_snapshot",
+    "labeled_component_risk",
+    "marginal_risk_contribution",
+    "risk_contribution",
+]
+
+
+@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
+    method: str
+    window_start_date: date | None
+    window_end_date: date | None
+    observations: int | None
+    lookback_sessions: int | None
+    missing_policy: str
+    data_snapshot_id: str
+    input_sha256: str
+
+    def __init__(
+        self,
+        *,
+        snapshot_id: str,
+        as_of_date: str | date | pd.Timestamp,
+        covariance: pd.DataFrame,
+        return_frequency: str,
+        periods_per_year: int,
+        method: str = "provided",
+        window_start_date: str | date | pd.Timestamp | None = None,
+        window_end_date: str | date | pd.Timestamp | None = None,
+        observations: int | None = None,
+        lookback_sessions: int | None = None,
+        missing_policy: str = "provided",
+        data_snapshot_id: str = "",
+        input_sha256: str = "",
+    ) -> 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")
+        if not isinstance(method, str) or not method.strip():
+            raise ValueError("method must be non-empty")
+        if not isinstance(missing_policy, str) or not missing_policy.strip():
+            raise ValueError("missing_policy must be non-empty")
+        for value, name in (
+            (observations, "observations"),
+            (lookback_sessions, "lookback_sessions"),
+        ):
+            if value is not None and (
+                isinstance(value, bool) or not isinstance(value, int) or value <= 0
+            ):
+                raise ValueError(f"{name} must be a positive integer when provided")
+        if input_sha256 and (
+            len(input_sha256) != 64
+            or any(character not in "0123456789abcdef" for character in input_sha256)
+        ):
+            raise ValueError("input_sha256 must be a lowercase SHA-256 digest")
+
+        normalized_as_of = _normalized_date(as_of_date, "as_of_date")
+        normalized_window_start = (
+            None
+            if window_start_date is None
+            else _normalized_date(window_start_date, "window_start_date")
+        )
+        normalized_window_end = (
+            None
+            if window_end_date is None
+            else _normalized_date(window_end_date, "window_end_date")
+        )
+        if (normalized_window_start is None) != (normalized_window_end is None):
+            raise ValueError("window_start_date and window_end_date must be provided together")
+        if (
+            normalized_window_start is not None
+            and normalized_window_end is not None
+            and normalized_window_start > normalized_window_end
+        ):
+            raise ValueError("window_start_date must not be after window_end_date")
+        if normalized_window_end is not None and normalized_window_end > normalized_as_of:
+            raise ValueError("window_end_date must not be after as_of_date")
+
+        object.__setattr__(self, "snapshot_id", snapshot_id.strip())
+        object.__setattr__(self, "as_of_date", normalized_as_of)
+        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)
+        object.__setattr__(self, "method", method.strip())
+        object.__setattr__(self, "window_start_date", normalized_window_start)
+        object.__setattr__(self, "window_end_date", normalized_window_end)
+        object.__setattr__(self, "observations", observations)
+        object.__setattr__(self, "lookback_sessions", lookback_sessions)
+        object.__setattr__(self, "missing_policy", missing_policy.strip())
+        object.__setattr__(self, "data_snapshot_id", data_snapshot_id.strip())
+        object.__setattr__(self, "input_sha256", input_sha256)
+
+    @property
+    def covariance(self) -> pd.DataFrame:
+        """Return an isolated copy so callers cannot mutate the snapshot."""
+        return self._covariance.copy(deep=True)
+
+
+def _normalized_date(value: object, name: str) -> date:
+    try:
+        timestamp = pd.Timestamp(value)
+    except (TypeError, ValueError) as error:
+        raise ValueError(f"{name} must be a valid date") from error
+    if pd.isna(timestamp):
+        raise ValueError(f"{name} must be a valid date")
+    return date(int(timestamp.year), int(timestamp.month), int(timestamp.day))
+
+
+def _positive_integer(value: int, name: str, *, minimum: int = 1) -> int:
+    if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
+        raise ValueError(f"{name} must be an integer of at least {minimum}")
+    return value
+
+
+def _input_fingerprint(window: pd.DataFrame, session_dates: list[date]) -> str:
+    values = window.to_numpy(dtype=float, copy=True)
+    missing = np.isnan(values)
+    normalized = np.where(missing, 0.0, values).astype("<f8", copy=False)
+    metadata = {
+        "assets": [str(asset) for asset in window.columns],
+        "sessions": [session.isoformat() for session in session_dates],
+        "shape": list(values.shape),
+    }
+    digest = hashlib.sha256(
+        json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    )
+    digest.update(missing.astype(np.uint8, copy=False).tobytes(order="C"))
+    digest.update(normalized.tobytes(order="C"))
+    return digest.hexdigest()
+
+
+def estimate_covariance_snapshot(
+    asset_returns: pd.DataFrame,
+    *,
+    as_of_date: str | date | pd.Timestamp,
+    lookback_sessions: int,
+    min_observations: int,
+    data_snapshot_id: str,
+    return_frequency: str = "1d",
+    periods_per_year: int = 252,
+) -> CovarianceSnapshot:
+    """Estimate a deterministic per-period sample covariance without look-ahead.
+
+    The selected lookback window is truncated at ``as_of_date`` before any
+    calculation. Rows containing a missing asset return are removed as complete
+    cases, preventing pairwise sample sets from producing an ambiguous matrix.
+    """
+    if not isinstance(asset_returns, pd.DataFrame):
+        raise TypeError("asset_returns must be a pandas DataFrame")
+    if asset_returns.empty or asset_returns.shape[1] == 0:
+        raise ValueError("asset_returns must contain observations and assets")
+    if not isinstance(asset_returns.index, pd.DatetimeIndex):
+        raise TypeError("asset_returns index must be a DatetimeIndex")
+    if not asset_returns.index.is_unique or not asset_returns.index.is_monotonic_increasing:
+        raise ValueError("asset_returns index must be unique and strictly increasing")
+    if not asset_returns.columns.is_unique:
+        raise ValueError("asset_returns must contain unique asset labels")
+    if any(not isinstance(asset, str) or not asset.strip() for asset in asset_returns.columns):
+        raise ValueError("asset_returns asset labels must be non-empty strings")
+
+    lookback = _positive_integer(lookback_sessions, "lookback_sessions")
+    minimum = _positive_integer(min_observations, "min_observations", minimum=2)
+    if minimum > lookback:
+        raise ValueError("min_observations must not exceed lookback_sessions")
+    normalized_data_snapshot_id = data_snapshot_id.strip()
+    if not normalized_data_snapshot_id:
+        raise ValueError("data_snapshot_id must be non-empty")
+    normalized_as_of = _normalized_date(as_of_date, "as_of_date")
+
+    returns = asset_returns.astype(float, copy=True)
+    values = returns.to_numpy()
+    if np.isinf(values).any():
+        raise ValueError("asset_returns must not contain infinite values")
+    session_dates = [
+        _normalized_date(index_value, "asset_returns index") for index_value in returns.index
+    ]
+    if len(set(session_dates)) != len(session_dates):
+        raise ValueError("asset_returns must contain at most one observation per session date")
+    historical_mask = [session <= normalized_as_of for session in session_dates]
+    window = returns.loc[historical_mask].tail(lookback)
+    if window.empty:
+        raise ValueError("asset_returns contain no observations on or before as_of_date")
+    window_dates = [
+        _normalized_date(index_value, "asset_returns index") for index_value in window.index
+    ]
+    complete = window.dropna(axis=0, how="any")
+    if len(complete) < minimum:
+        raise ValueError(
+            f"complete observations must be at least {minimum}; received {len(complete)}"
+        )
+
+    covariance = complete.cov(ddof=1)
+    covariance_values = covariance.to_numpy()
+    if not np.isfinite(covariance_values).all():
+        raise ValueError("sample covariance must be finite")
+    input_sha256 = _input_fingerprint(window, window_dates)
+    identity = {
+        "as_of_date": normalized_as_of.isoformat(),
+        "assets": list(returns.columns),
+        "data_snapshot_id": normalized_data_snapshot_id,
+        "estimator": "sample-cov-v1",
+        "input_sha256": input_sha256,
+        "lookback_sessions": lookback,
+        "min_observations": minimum,
+        "missing_policy": "complete_case",
+        "observations": len(complete),
+        "periods_per_year": periods_per_year,
+        "return_frequency": return_frequency,
+        "window_end_date": window_dates[-1].isoformat(),
+        "window_start_date": window_dates[0].isoformat(),
+    }
+    identity_bytes = json.dumps(
+        identity,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    digest = hashlib.sha256(identity_bytes)
+    digest.update(covariance_values.astype("<f8", copy=False).tobytes(order="C"))
+    snapshot_id = f"sample-cov-v1:{digest.hexdigest()}"
+    return CovarianceSnapshot(
+        snapshot_id=snapshot_id,
+        as_of_date=normalized_as_of,
+        covariance=covariance,
+        return_frequency=return_frequency,
+        periods_per_year=periods_per_year,
+        method="sample",
+        window_start_date=window_dates[0],
+        window_end_date=window_dates[-1],
+        observations=len(complete),
+        lookback_sessions=lookback,
+        missing_policy="complete_case",
+        data_snapshot_id=normalized_data_snapshot_id,
+        input_sha256=input_sha256,
+    )
+
+
+@dataclass(frozen=True, slots=True, eq=False)
+class ComponentRiskResult:
+    """Label-preserving Euler decomposition of portfolio volatility."""
+
+    portfolio_volatility: float
+    marginal: pd.Series
+    component: pd.Series
+    percentage: pd.Series
+
+    def grouped_component(self, groups: pd.Series) -> pd.Series:
+        """Aggregate asset component risk by an explicitly aligned label series."""
+        if not isinstance(groups, pd.Series):
+            raise TypeError("groups must be a pandas Series")
+        if not groups.index.is_unique:
+            raise ValueError("groups must contain unique asset labels")
+        if not self.component.index.difference(groups.index).empty or not groups.index.difference(
+            self.component.index
+        ).empty:
+            raise ValueError("groups and component risk must use the same asset labels")
+        aligned = groups.reindex(self.component.index)
+        if aligned.isna().any():
+            raise ValueError("groups must contain a non-missing label for every asset")
+        grouped = self.component.groupby(aligned, sort=True).sum()
+        grouped.name = "component_risk"
+        return grouped
+
 
 def _validate_inputs(weights: NDArray[Any], cov: NDArray[Any]) -> tuple[NDArray[Any], NDArray[Any]]:
     """Normalize a portfolio vector and its covariance matrix."""
@@ -59,3 +346,70 @@ def component_var(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     """成分方差: w_i · (Σw)_i; 与 RC 的关系 RC_i = CV_i / w'Σw。"""
     w, cov = _validate_inputs(weights, cov)
     return w * (cov @ w)  # type: ignore[no-any-return]
+
+
+def labeled_component_risk(
+    weights: pd.Series,
+    covariance: pd.DataFrame,
+) -> ComponentRiskResult:
+    """Return a label-safe Euler decomposition that sums to portfolio volatility.
+
+    The covariance matrix may use a different asset order, but its row and
+    column label sets must exactly match ``weights``.  Invalid or indefinite
+    covariance input is rejected instead of silently producing misleading risk
+    percentages.
+    """
+    if not isinstance(weights, pd.Series):
+        raise TypeError("weights must be a pandas Series")
+    if not isinstance(covariance, pd.DataFrame):
+        raise TypeError("covariance must be a pandas DataFrame")
+    if weights.empty:
+        raise ValueError("weights must contain at least one asset")
+    if not weights.index.is_unique:
+        raise ValueError("weights must contain unique asset labels")
+    if not covariance.index.is_unique or not covariance.columns.is_unique:
+        raise ValueError("covariance must contain unique asset labels")
+    if not weights.index.difference(covariance.index).empty or not covariance.index.difference(
+        weights.index
+    ).empty:
+        raise ValueError("weights and covariance must use the same asset labels")
+    if not weights.index.difference(covariance.columns).empty or not covariance.columns.difference(
+        weights.index
+    ).empty:
+        raise ValueError("weights and covariance must use the same asset labels")
+
+    aligned_weights = weights.astype(float, copy=True)
+    aligned_covariance = covariance.reindex(
+        index=weights.index,
+        columns=weights.index,
+    ).astype(float, copy=True)
+    weight_values = aligned_weights.to_numpy()
+    covariance_values = aligned_covariance.to_numpy()
+    if not np.isfinite(weight_values).all():
+        raise ValueError("weights must be finite")
+    if not np.isfinite(covariance_values).all():
+        raise ValueError("covariance must be finite")
+    if not np.allclose(covariance_values, covariance_values.T, rtol=1e-10, atol=1e-12):
+        raise ValueError("covariance must be symmetric")
+    eigenvalues = np.linalg.eigvalsh(covariance_values)
+    scale = max(1.0, float(np.max(np.abs(eigenvalues))))
+    if float(eigenvalues.min()) < -1e-10 * scale:
+        raise ValueError("covariance must be positive semidefinite")
+
+    portfolio_variance = float(weight_values @ covariance_values @ weight_values)
+    if portfolio_variance <= 0 or not np.isfinite(portfolio_variance):
+        raise ValueError("weights and covariance must produce positive portfolio variance")
+    portfolio_volatility = float(np.sqrt(portfolio_variance))
+    marginal_values = covariance_values @ weight_values / portfolio_volatility
+    component_values = weight_values * marginal_values
+    percentage_values = component_values / portfolio_volatility
+    return ComponentRiskResult(
+        portfolio_volatility=portfolio_volatility,
+        marginal=pd.Series(marginal_values, index=weights.index.copy(), name="marginal_risk"),
+        component=pd.Series(component_values, index=weights.index.copy(), name="component_risk"),
+        percentage=pd.Series(
+            percentage_values,
+            index=weights.index.copy(),
+            name="risk_contribution",
+        ),
+    )

tests/test_artifact.py

@@ -0,0 +1,281 @@
+"""Stable research-run artifact contracts for downstream persistence."""
+
+from __future__ import annotations
+
+import json
+from datetime import date
+
+import pandas as pd
+import pytest
+
+from quant_engine.artifact import (
+    RESEARCH_ARTIFACT_SCHEMA_VERSION,
+    ResearchRunArtifact,
+    build_research_run_artifact,
+)
+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:
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    scores = pd.DataFrame(
+        {"A": [2.0, 0.0], "B": [1.0, 3.0]},
+        index=dates[:2],
+    )
+    opens = pd.DataFrame(
+        {"A": [10.0, 10.0, 15.0, 15.0], "B": [20.0, 20.0, 20.0, 21.0]},
+        index=dates,
+    )
+    closes = pd.DataFrame(
+        {"A": [10.0, 12.0, 15.0, 15.0], "B": [20.0, 20.0, 18.0, 21.0]},
+        index=dates,
+    )
+    return run_factor_backtest_research(
+        scores,
+        opens,
+        closes,
+        top_k=1,
+        execution_price_field="open",
+        valuation_price_field="close",
+        initial_cash=1_000.0,
+        config=ExecutionConfig(
+            commission_bps=0,
+            stamp_tax_bps=0,
+            slippage_bps=0,
+            min_trade_amount=0,
+        ),
+    )
+
+
+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],
+        index=result.returns.index,
+        name="benchmark_return",
+    )
+    return build_research_run_artifact(
+        result,
+        run_id="run-20260105-a",
+        strategy_id="alpha-top1",
+        strategy_name="Alpha Top 1",
+        strategy_version="1.0.0",
+        engine_version="1.2.0",
+        code_revision="3b1ad07",
+        data_snapshot_id="qtdb-pro-20260108-v1",
+        calendar="CN-A",
+        timezone="Asia/Shanghai",
+        started_at="2026-01-08T10:00:00+08:00",
+        finished_at="2026-01-08T10:01:00+08:00",
+        parameters=parameters or {"top_k": 1, "lag_sessions": 1},
+        benchmark_id="000300.SH",
+        benchmark_returns=benchmark,
+        risk_snapshots=risk_snapshots,
+    )
+
+
+def test_research_artifact_projects_versioned_queryable_fact_tables() -> None:
+    result = _backtest_result()
+
+    artifact = _build(result)
+
+    assert artifact.schema_version == RESEARCH_ARTIFACT_SCHEMA_VERSION
+    assert artifact.run.loc[0, "run_id"] == "run-20260105-a"
+    assert artifact.run.loc[0, "benchmark_alignment_policy"] == "exact_session_index"
+    assert artifact.nav["run_id"].unique().tolist() == ["run-20260105-a"]
+    assert artifact.nav["pnl_pct"].tolist() == pytest.approx(result.returns.tolist())
+    assert artifact.nav["benchmark_return"].tolist() == pytest.approx(
+        [0.0, 0.01, -0.01, 0.02]
+    )
+    assert artifact.signals.columns.tolist() == [
+        "run_id",
+        "signal_date",
+        "execution_date",
+        "asset_id",
+        "factor_score",
+        "target_weight",
+    ]
+    first_signal = artifact.signals[
+        artifact.signals["signal_date"] == result.factor_scores.index[0].date()
+    ]
+    assert first_signal.set_index("asset_id").loc["A", "factor_score"] == 2.0
+    assert first_signal.set_index("asset_id").loc["A", "target_weight"] == 1.0
+    assert first_signal["execution_date"].unique().tolist() == [
+        result.schedule.signal_to_execution.iloc[0].date()
+    ]
+    assert set(artifact.trades["side"]) == {"buy", "sell"}
+    assert artifact.trades["trade_id"].is_unique
+    assert artifact.trades["trade_id"].str.startswith("run-20260105-a:").all()
+    assert artifact.trades["signal_id"].str.startswith("run-20260105-a:signal:").all()
+    assert {"security", "cash"}.issubset(set(artifact.positions["asset_type"]))
+    assert artifact.positions.groupby("trade_date")["weight"].sum().tolist() == pytest.approx(
+        [1.0, 1.0, 1.0, 1.0]
+    )
+    assert set(artifact.attribution.columns) == {
+        "run_id",
+        "trade_date",
+        "asset_id",
+        "overnight",
+        "intraday",
+        "asset_total",
+    }
+    assert artifact.attribution_daily["residual"].abs().max() < 1e-12
+    assert artifact.risk.empty
+    assert artifact.risk.columns.tolist() == [
+        "run_id",
+        "trade_date",
+        "asset_id",
+        "weight",
+        "marginal_risk",
+        "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(
+        result.benchmark_stats(pd.Series([0.0, 0.01, -0.01, 0.02], index=result.returns.index))[
+            "information_ratio"
+        ]
+    )
+    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})
+    second = _build(result, parameters={"lag_sessions": 1, "top_k": 1})
+
+    assert first.run.loc[0, "config_hash"] == second.run.loc[0, "config_hash"]
+    assert first.content_sha256 == second.content_sha256
+    assert first.manifest() == second.manifest()
+    decoded = json.loads(first.canonical_json())
+    assert decoded["schema_version"] == RESEARCH_ARTIFACT_SCHEMA_VERSION
+    assert decoded["tables"]["nav"][0]["trade_date"] == "2026-01-05"
+
+    leaked_copy = first.nav
+    leaked_copy.loc[0, "nav"] = -999.0
+    assert first.nav.loc[0, "nav"] != -999.0
+    assert first.content_sha256 == second.content_sha256
+
+
+def test_research_artifact_requires_complete_reproducibility_identity() -> None:
+    result = _backtest_result()
+
+    with pytest.raises(ValueError, match="code_revision"):
+        build_research_run_artifact(
+            result,
+            run_id="run-1",
+            strategy_id="alpha-top1",
+            strategy_name="Alpha Top 1",
+            strategy_version="1.0.0",
+            engine_version="1.2.0",
+            code_revision="",
+            data_snapshot_id="snapshot-1",
+            calendar="CN-A",
+            timezone="Asia/Shanghai",
+            started_at="2026-01-08T10:00:00+08:00",
+            finished_at="2026-01-08T10:01:00+08:00",
+            parameters={},
+        )
+
+
+def test_research_artifact_requires_benchmark_identity_and_returns_together() -> None:
+    result = _backtest_result()
+
+    with pytest.raises(ValueError, match="benchmark_id and benchmark_returns"):
+        build_research_run_artifact(
+            result,
+            run_id="run-1",
+            strategy_id="alpha-top1",
+            strategy_name="Alpha Top 1",
+            strategy_version="1.0.0",
+            engine_version="1.2.0",
+            code_revision="3b1ad07",
+            data_snapshot_id="snapshot-1",
+            calendar="CN-A",
+            timezone="Asia/Shanghai",
+            started_at="2026-01-08T10:00:00+08:00",
+            finished_at="2026-01-08T10:01:00+08:00",
+            parameters={},
+            benchmark_id="000300.SH",
+        )

tests/test_attribution.py

@@ -0,0 +1,118 @@
+"""Post-execution return attribution contracts."""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+from quant_engine.attribution import DailyReturnAttribution
+from quant_engine.execution import ExecutionConfig
+from quant_engine.research_pipeline import run_factor_backtest_research
+
+
+def _zero_cost_config() -> ExecutionConfig:
+    return ExecutionConfig(
+        commission_bps=0,
+        stamp_tax_bps=0,
+        slippage_bps=0,
+        min_trade_amount=0,
+    )
+
+
+def test_daily_attribution_closes_across_rebalance_and_holding_days() -> None:
+    """开盘换仓时，隔夜和日内贡献必须来自实际换仓前后持仓。"""
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    scores = pd.DataFrame(
+        {"A": [2.0, 0.0], "B": [1.0, 3.0]},
+        index=dates[:2],
+    )
+    opens = pd.DataFrame(
+        {"A": [10.0, 10.0, 15.0, 15.0], "B": [20.0, 20.0, 20.0, 21.0]},
+        index=dates,
+    )
+    closes = pd.DataFrame(
+        {"A": [10.0, 12.0, 15.0, 15.0], "B": [20.0, 20.0, 18.0, 21.0]},
+        index=dates,
+    )
+    result = run_factor_backtest_research(
+        scores,
+        opens,
+        closes,
+        top_k=1,
+        execution_price_field="open",
+        valuation_price_field="close",
+        initial_cash=1_000.0,
+        config=_zero_cost_config(),
+    )
+
+    attribution = result.return_attribution()
+
+    assert isinstance(attribution, DailyReturnAttribution)
+    assert attribution.overnight.loc[dates[2], "A"] == pytest.approx(0.25)
+    assert attribution.intraday.loc[dates[2], "B"] == pytest.approx(-0.125)
+    assert attribution.asset_contributions.loc[dates[3], "B"] == pytest.approx(1 / 6)
+    pd.testing.assert_series_equal(
+        attribution.total_return,
+        result.returns.rename("total_return"),
+    )
+    pd.testing.assert_series_equal(
+        attribution.explained_return + attribution.residual,
+        attribution.total_return,
+        check_names=False,
+    )
+    assert attribution.residual.abs().max() < 1e-12
+
+
+def test_daily_attribution_reports_execution_cost_separately() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    scores = pd.DataFrame({"A": [1.0]}, index=dates[:1])
+    prices = pd.DataFrame({"A": [10.0, 10.0, 10.0]}, index=dates)
+    config = ExecutionConfig(
+        commission_bps=10,
+        stamp_tax_bps=0,
+        slippage_bps=10,
+        min_trade_amount=0,
+    )
+    result = run_factor_backtest_research(
+        scores,
+        prices,
+        prices,
+        top_k=1,
+        gross_exposure=0.5,
+        execution_price_field="open",
+        valuation_price_field="close",
+        initial_cash=1_000.0,
+        config=config,
+    )
+
+    attribution = result.return_attribution()
+    execution = result.execution.daily_executions[1].executions[0]
+
+    assert attribution.asset_contributions.loc[dates[1], "A"] == 0.0
+    assert attribution.transaction_cost.loc[dates[1]] == pytest.approx(
+        -execution.total_cost / 1_000.0
+    )
+    assert attribution.total_return.loc[dates[1]] == pytest.approx(
+        attribution.transaction_cost.loc[dates[1]]
+    )
+    assert attribution.residual.loc[dates[1]] == pytest.approx(0.0, abs=1e-12)
+
+
+def test_return_attribution_is_empty_for_empty_research_result() -> None:
+    dates = pd.date_range("2026-01-05", periods=3, freq="B")
+    scores = pd.DataFrame(columns=["A"], index=pd.DatetimeIndex([]), dtype=float)
+    prices = pd.DataFrame({"A": [10.0, 10.0, 10.0]}, index=dates)
+    result = run_factor_backtest_research(
+        scores,
+        prices,
+        prices,
+        top_k=1,
+        execution_price_field="open",
+        valuation_price_field="close",
+    )
+
+    attribution = result.return_attribution()
+
+    assert attribution.overnight.empty
+    assert attribution.intraday.empty
+    assert attribution.total_return.empty

tests/test_metrics.py

@@ -10,9 +10,11 @@ from quant_engine.metrics import (
     TRADING_DAYS_PER_YEAR,
     annualized_return,
     annualized_volatility,
+    benchmark_summary,
     calmar_ratio,
     max_drawdown,
     sharpe_ratio,
+    sortino_ratio,
     summary,
     win_rate,
 )
@@ -47,9 +49,22 @@ def test_zero_volatility_metrics_return_zero() -> None:
     returns = pd.Series([0.0, 0.0, 0.0])
 
     assert sharpe_ratio(returns) == 0.0
+    assert sortino_ratio(returns) == 0.0
     assert calmar_ratio(returns) == 0.0
 
 
+def test_sortino_ratio_uses_all_sessions_for_downside_deviation() -> None:
+    returns = pd.Series([0.02, -0.01, 0.0, -0.03])
+    downside = np.minimum(returns.to_numpy(), 0.0)
+    downside_deviation = np.sqrt(np.mean(np.square(downside))) * np.sqrt(
+        TRADING_DAYS_PER_YEAR
+    )
+
+    assert sortino_ratio(returns) == pytest.approx(
+        annualized_return(returns) / downside_deviation
+    )
+
+
 def test_max_drawdown_includes_loss_from_initial_capital() -> None:
     returns = pd.Series([-0.20, 0.0])
 
@@ -79,7 +94,15 @@ def test_summary_aliases_match_canonical_fields() -> None:
 
 @pytest.mark.parametrize(
     "metric",
-    [annualized_return, annualized_volatility, sharpe_ratio, max_drawdown, calmar_ratio, win_rate],
+    [
+        annualized_return,
+        annualized_volatility,
+        sharpe_ratio,
+        sortino_ratio,
+        max_drawdown,
+        calmar_ratio,
+        win_rate,
+    ],
 )
 def test_metrics_reject_non_series_input(metric) -> None:
     with pytest.raises(TypeError, match=r"expected pd\.Series"):
@@ -91,3 +114,50 @@ def test_short_and_empty_series_return_zero() -> None:
     assert annualized_volatility(pd.Series([0.01])) == 0.0
     assert max_drawdown(pd.Series([0.01])) == 0.0
     assert win_rate(pd.Series(dtype=float)) == 0.0
+
+
+def test_benchmark_summary_uses_aligned_active_returns_and_regression() -> None:
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    benchmark = pd.Series([-0.01, 0.0, 0.01, 0.02], index=dates)
+    portfolio = 0.001 + 1.5 * benchmark
+    active = portfolio - benchmark
+
+    result = benchmark_summary(portfolio, benchmark)
+
+    assert result["n_observations"] == 4
+    assert result["tracking_error"] == pytest.approx(
+        active.std() * np.sqrt(TRADING_DAYS_PER_YEAR)
+    )
+    assert result["information_ratio"] == pytest.approx(
+        active.mean() / active.std() * np.sqrt(TRADING_DAYS_PER_YEAR)
+    )
+    assert result["beta"] == pytest.approx(1.5)
+    assert result["alpha"] == pytest.approx(1.001**TRADING_DAYS_PER_YEAR - 1.0)
+
+
+def test_benchmark_summary_rejects_silent_calendar_alignment() -> None:
+    portfolio = pd.Series([0.01, 0.02], index=pd.date_range("2026-01-05", periods=2))
+    benchmark = pd.Series([0.01, 0.02], index=pd.date_range("2026-01-06", periods=2))
+
+    with pytest.raises(ValueError, match="matching indexes"):
+        benchmark_summary(portfolio, benchmark)
+
+
+def test_benchmark_summary_rejects_missing_observations() -> None:
+    dates = pd.date_range("2026-01-05", periods=2)
+    portfolio = pd.Series([0.01, np.nan], index=dates)
+    benchmark = pd.Series([0.0, 0.01], index=dates)
+
+    with pytest.raises(ValueError, match="finite"):
+        benchmark_summary(portfolio, benchmark)
+
+
+def test_benchmark_summary_marks_constant_benchmark_regression_unestimable() -> None:
+    dates = pd.date_range("2026-01-05", periods=3)
+    portfolio = pd.Series([0.01, -0.01, 0.02], index=dates)
+    benchmark = pd.Series([0.0, 0.0, 0.0], index=dates)
+
+    result = benchmark_summary(portfolio, benchmark)
+
+    assert np.isnan(result["alpha"])
+    assert np.isnan(result["beta"])

tests/test_research_pipeline.py

@@ -265,3 +265,72 @@ def test_factor_backtest_starts_at_first_signal_instead_of_price_warmup() -> Non
         pd.Series([1.0, 1.1, 1.2], index=dates[2:], name="nav"),
     )
     assert result.stats()["n_days"] == 3
+
+
+def test_factor_backtest_exposes_net_benchmark_metrics() -> None:
+    dates = _calendar()
+    scores = pd.DataFrame({"A": [1.0]}, index=dates[:1])
+    prices = pd.DataFrame({"A": [10.0, 10.0, 11.0, 11.0]}, index=dates)
+    result = run_factor_backtest_research(
+        scores,
+        prices,
+        prices,
+        top_k=1,
+        execution_price_field="open",
+        valuation_price_field="close",
+        config=ExecutionConfig(
+            commission_bps=0,
+            stamp_tax_bps=0,
+            slippage_bps=0,
+            min_trade_amount=0,
+        ),
+    )
+    benchmark = pd.Series([0.0, 0.01, -0.01, 0.0], index=dates)
+
+    relative = result.benchmark_stats(benchmark)
+
+    assert relative["n_observations"] == len(result.returns)
+    assert relative["tracking_error"] > 0
+
+
+def test_factor_backtest_projects_actual_close_weights_from_ledger() -> None:
+    dates = _calendar()
+    scores = pd.DataFrame({"A": [1.0], "B": [0.0]}, index=dates[:1])
+    opens = pd.DataFrame(
+        {"A": [10.0, 10.0, 10.0, 10.0], "B": [20.0, 20.0, 20.0, 20.0]},
+        index=dates,
+    )
+    closes = pd.DataFrame(
+        {"A": [10.0, 11.0, 12.0, 12.0], "B": [20.0, 20.0, 20.0, 20.0]},
+        index=dates,
+    )
+    result = run_factor_backtest_research(
+        scores,
+        opens,
+        closes,
+        top_k=1,
+        gross_exposure=0.5,
+        execution_price_field="open",
+        valuation_price_field="close",
+        initial_cash=1_000.0,
+        config=ExecutionConfig(
+            commission_bps=0,
+            stamp_tax_bps=0,
+            slippage_bps=0,
+            min_trade_amount=0,
+        ),
+    )
+
+    weights = result.position_weights
+    cash = result.cash_weights
+
+    assert weights.index.equals(result.nav.index)
+    assert weights.columns.tolist() == ["A", "B"]
+    assert weights.loc[dates[0]].sum() == 0.0
+    assert cash.loc[dates[0]] == 1.0
+    assert weights.loc[dates[1], "A"] == pytest.approx(550.0 / 1_050.0)
+    pd.testing.assert_series_equal(
+        weights.sum(axis=1) + cash,
+        pd.Series(1.0, index=dates),
+        check_names=False,
+    )

tests/test_risk.py

@@ -3,9 +3,167 @@
 from __future__ import annotations
 
 import numpy as np
+import pandas as pd
 import pytest
 
-from quant_engine.risk import component_var, marginal_risk_contribution, risk_contribution
+from quant_engine.risk import (
+    ComponentRiskResult,
+    CovarianceSnapshot,
+    component_var,
+    estimate_covariance_snapshot,
+    labeled_component_risk,
+    marginal_risk_contribution,
+    risk_contribution,
+)
+
+
+def test_estimate_covariance_snapshot_is_complete_case_and_reproducible() -> None:
+    dates = pd.date_range("2026-01-05", periods=6, freq="B")
+    returns = pd.DataFrame(
+        {
+            "A": [0.01, 0.02, 0.03, 0.04, 0.05, 99.0],
+            "B": [0.02, 0.01, np.nan, 0.03, 0.04, -99.0],
+        },
+        index=dates,
+    )
+    as_of = dates[4]
+
+    snapshot = estimate_covariance_snapshot(
+        returns,
+        as_of_date=as_of,
+        lookback_sessions=4,
+        min_observations=3,
+        data_snapshot_id="market-returns-20260109-v1",
+        return_frequency="1d",
+        periods_per_year=252,
+    )
+
+    expected_window = returns.loc[:as_of].tail(4)
+    expected = expected_window.dropna(how="any").cov()
+    pd.testing.assert_frame_equal(snapshot.covariance, expected)
+    assert snapshot.snapshot_id.startswith("sample-cov-v1:")
+    assert snapshot.as_of_date == as_of.date()
+    assert snapshot.method == "sample"
+    assert snapshot.window_start_date == expected_window.index[0].date()
+    assert snapshot.window_end_date == as_of.date()
+    assert snapshot.observations == 3
+    assert snapshot.lookback_sessions == 4
+    assert snapshot.missing_policy == "complete_case"
+    assert snapshot.data_snapshot_id == "market-returns-20260109-v1"
+    assert len(snapshot.input_sha256) == 64
+
+    future_changed = returns.copy()
+    future_changed.loc[dates[-1], :] = [1_000_000.0, -1_000_000.0]
+    repeated = estimate_covariance_snapshot(
+        future_changed,
+        as_of_date=as_of,
+        lookback_sessions=4,
+        min_observations=3,
+        data_snapshot_id="market-returns-20260109-v1",
+        return_frequency="1d",
+        periods_per_year=252,
+    )
+    assert repeated.snapshot_id == snapshot.snapshot_id
+    pd.testing.assert_frame_equal(repeated.covariance, snapshot.covariance)
+
+
+def test_covariance_snapshot_identity_captures_data_and_estimator_contract() -> None:
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    returns = pd.DataFrame(
+        {"A": [0.01, 0.02, -0.01, 0.03], "B": [0.02, -0.01, 0.01, 0.04]},
+        index=dates,
+    )
+    base = estimate_covariance_snapshot(
+        returns,
+        as_of_date=dates[-1],
+        lookback_sessions=4,
+        min_observations=3,
+        data_snapshot_id="snapshot-a",
+    )
+    different_source = estimate_covariance_snapshot(
+        returns,
+        as_of_date=dates[-1],
+        lookback_sessions=4,
+        min_observations=3,
+        data_snapshot_id="snapshot-b",
+    )
+
+    assert base.snapshot_id != different_source.snapshot_id
+    assert base.covariance.equals(different_source.covariance)
+
+
+def test_estimate_covariance_snapshot_rejects_ambiguous_or_insufficient_history() -> None:
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    returns = pd.DataFrame(
+        {"A": [0.01, np.nan, 0.03, 0.04], "B": [0.02, 0.01, np.nan, 0.03]},
+        index=dates,
+    )
+
+    with pytest.raises(ValueError, match="complete observations"):
+        estimate_covariance_snapshot(
+            returns,
+            as_of_date=dates[-1],
+            lookback_sessions=4,
+            min_observations=3,
+            data_snapshot_id="snapshot-a",
+        )
+
+    with pytest.raises(ValueError, match="strictly increasing"):
+        estimate_covariance_snapshot(
+            returns.iloc[::-1],
+            as_of_date=dates[-1],
+            lookback_sessions=4,
+            min_observations=2,
+            data_snapshot_id="snapshot-a",
+        )
+
+
+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:
@@ -52,3 +210,61 @@ def test_risk_functions_reject_covariance_shape_mismatch(function) -> None:
 def test_risk_functions_reject_empty_portfolio(function) -> None:
     with pytest.raises(ValueError, match="at least one asset"):
         function(np.array([]), np.empty((0, 0)))
+
+
+def test_labeled_component_risk_aligns_covariance_and_closes_to_volatility() -> None:
+    weights = pd.Series({"A": 0.25, "B": 0.75}, name="weight")
+    covariance = pd.DataFrame(
+        [[0.09, 0.01], [0.01, 0.04]],
+        index=["B", "A"],
+        columns=["B", "A"],
+    )
+
+    result = labeled_component_risk(weights, covariance)
+
+    aligned = covariance.reindex(index=weights.index, columns=weights.index)
+    expected_volatility = float(np.sqrt(weights @ aligned @ weights))
+    assert isinstance(result, ComponentRiskResult)
+    assert result.component.index.tolist() == ["A", "B"]
+    assert result.portfolio_volatility == pytest.approx(expected_volatility)
+    assert result.component.sum() == pytest.approx(expected_volatility)
+    assert result.percentage.sum() == pytest.approx(1.0)
+
+
+def test_component_risk_groups_actual_asset_contributions_by_label() -> None:
+    weights = pd.Series({"A": 0.2, "B": 0.3, "C": 0.5})
+    covariance = pd.DataFrame(np.diag([0.04, 0.09, 0.16]), index=weights.index, columns=weights.index)
+    groups = pd.Series({"C": "growth", "A": "value", "B": "value"})
+
+    result = labeled_component_risk(weights, covariance)
+    grouped = result.grouped_component(groups)
+
+    assert grouped.index.tolist() == ["growth", "value"]
+    assert grouped.loc["value"] == pytest.approx(
+        result.component.loc["A"] + result.component.loc["B"]
+    )
+    assert grouped.sum() == pytest.approx(result.portfolio_volatility)
+
+
+def test_labeled_component_risk_rejects_asset_label_mismatch() -> None:
+    weights = pd.Series({"A": 0.5, "B": 0.5})
+    covariance = pd.DataFrame(np.eye(2), index=["A", "C"], columns=["A", "C"])
+
+    with pytest.raises(ValueError, match="same asset labels"):
+        labeled_component_risk(weights, covariance)
+
+
+def test_labeled_component_risk_rejects_invalid_covariance() -> None:
+    weights = pd.Series({"A": 0.5, "B": 0.5})
+    asymmetric = pd.DataFrame([[1.0, 0.2], [0.1, 1.0]], index=weights.index, columns=weights.index)
+
+    with pytest.raises(ValueError, match="symmetric"):
+        labeled_component_risk(weights, asymmetric)
+
+
+def test_labeled_component_risk_rejects_zero_variance_portfolio() -> None:
+    weights = pd.Series({"A": 0.5, "B": 0.5})
+    covariance = pd.DataFrame(np.zeros((2, 2)), index=weights.index, columns=weights.index)
+
+    with pytest.raises(ValueError, match="positive portfolio variance"):
+        labeled_component_risk(weights, covariance)