From 1a60fef6e14c7aab14e292ca30e23a21f4a2641f Mon Sep 17 00:00:00 2001 From: ao gong <41768719+ageorge156@users.noreply.github.com> Date: Fri, 21 Aug 2026 23:28:34 +0800 Subject: [PATCH] feat: publish reproducible portfolio risk facts --- src/quant_engine/artifact.py | 77 +++++++++++++++++++++++++++++++++++- src/quant_engine/risk.py | 52 ++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 2 deletions(-) diff --git a/src/quant_engine/artifact.py b/src/quant_engine/artifact.py index 87ec3e2..cae5f40 100644 --- a/src/quant_engine/artifact.py +++ b/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), ) diff --git a/src/quant_engine/risk.py b/src/quant_engine/risk.py index d2328fa..93dda1c 100644 --- a/src/quant_engine/risk.py +++ b/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."""