feat: add label-safe component risk decomposition

src/quant_engine/risk.py

@@ -5,11 +5,48 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import Any
 
 import numpy as np
+import pandas as pd
 from numpy.typing import NDArray
 
+__all__ = [
+    "ComponentRiskResult",
+    "component_var",
+    "labeled_component_risk",
+    "marginal_risk_contribution",
+    "risk_contribution",
+]
+
+
+@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 +96,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",
+        ),
+    )