feat: publish reproducible portfolio risk facts

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."""