ageorge156/quant_engine
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: estimate deterministic covariance snapshots

Browse Source
This commit is contained in:
ao gong
2026-08-21 23:47:29 +08:00
parent 8ef8e3208e
commit 524fc73852
1 changed files with 205 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/quant_engine/risk.py
+205 -7
View File
@@ -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("<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."""