From 524fc73852d8d32527ec833029857dc8d1401640 Mon Sep 17 00:00:00 2001 From: ao gong <41768719+ageorge156@users.noreply.github.com> Date: Fri, 21 Aug 2026 23:47:29 +0800 Subject: [PATCH] feat: estimate deterministic covariance snapshots --- src/quant_engine/risk.py | 212 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 205 insertions(+), 7 deletions(-) diff --git a/src/quant_engine/risk.py b/src/quant_engine/risk.py index 93dda1c..748ad36 100644 --- a/src/quant_engine/risk.py +++ b/src/quant_engine/risk.py @@ -5,6 +5,8 @@ from __future__ import annotations +import hashlib +import json from dataclasses import dataclass from datetime import date from typing import Any @@ -17,6 +19,7 @@ __all__ = [ "ComponentRiskResult", "CovarianceSnapshot", "component_var", + "estimate_covariance_snapshot", "labeled_component_risk", "marginal_risk_contribution", "risk_contribution", @@ -32,6 +35,14 @@ class CovarianceSnapshot: _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, @@ -41,6 +52,14 @@ class CovarianceSnapshot: 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") @@ -54,18 +73,59 @@ class CovarianceSnapshot: 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") + 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.date()) + 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: @@ -73,6 +133,144 @@ class CovarianceSnapshot: 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(" 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("