fix: harden quant core calculation boundaries

src/quant_engine/backtest.py

@@ -131,6 +131,9 @@ def rebalance_periodic(
     Returns:
         调仓表 DataFrame（all_dates × 股票代码）
     """
+    if all_dates.empty:
+        return pd.DataFrame(index=all_dates, columns=target_weights.index, dtype=float)
+
     table = pd.DataFrame(0.0, index=all_dates, columns=target_weights.index)
     for date in rebalance_dates:
         if date not in all_dates:
@@ -201,6 +204,8 @@ def compare_to_benchmark(
     """
     # 对齐 index
     common = strategy_nav.index.intersection(benchmark_nav.index)
+    if common.empty:
+        raise ValueError("strategy and benchmark must have overlapping dates")
     s = strategy_nav.loc[common]
     b = benchmark_nav.loc[common]
 

src/quant_engine/factor_library.py

@@ -313,7 +313,7 @@ def ols_regress(
     r_sq = 1.0 - ss_res / ss_tot if ss_tot > 0 else np.nan
     sigma2 = ss_res / max(n - k, 1)
     # 协方差矩阵 = sigma2 * (X'X)^-1
-    xtx_inv = np.linalg.inv(x_arr.T @ x_arr) if sigma2 > 0 else np.full((k, k), np.nan)
+    xtx_inv = np.linalg.pinv(x_arr.T @ x_arr) if sigma2 > 0 else np.full((k, k), np.nan)
     se = np.sqrt(np.diag(xtx_inv) * sigma2)
     t_vals = coef / se if sigma2 > 0 else np.full_like(coef, np.nan)
     if add_constant:
@@ -513,6 +513,8 @@ def apply_factor_direction(
     Returns:
         方向调整后的因子（同向 = 越大越好）
     """
+    if direction not in {"auto", "forward", "reverse"}:
+        raise ValueError(f"direction={direction!r} not supported (auto / forward / reverse)")
     if factor.empty:
         return factor.copy()
     if direction == "auto":
@@ -541,6 +543,8 @@ def cross_sectional_rank_with_direction(
     Returns:
         pd.Series（百分位排名 [0, 1]，越大越优）
     """
+    if direction not in {"auto", "forward", "reverse"}:
+        raise ValueError(f"direction={direction!r} not supported (auto / forward / reverse)")
     if df.empty or factor_col not in df.columns:
         return pd.Series(dtype=float)
     factor = df[factor_col]

src/quant_engine/metrics.py

@@ -71,7 +71,8 @@ def max_drawdown(r: pd.Series) -> float:
     if len(r) < 2:
         return 0.0
     nav = (1 + r).cumprod()
-    peak = nav.cummax()
+    # 初始资金净值为 1；否则首个观测日的亏损会被误当成新的历史高点。
+    peak = nav.cummax().clip(lower=1.0)
     drawdown = (nav - peak) / peak
     return float(drawdown.min())
 

src/quant_engine/risk.py

@@ -10,6 +10,22 @@ from numpy.typing import NDArray
 from typing import Any
 
 
+def _validate_inputs(
+    weights: NDArray[Any], cov: NDArray[Any]
+) -> tuple[NDArray[Any], NDArray[Any]]:
+    """Normalize a portfolio vector and its covariance matrix."""
+    w = np.asarray(weights, dtype=float).ravel()
+    covariance = np.asarray(cov, dtype=float)
+    k = w.size
+    if k == 0:
+        raise ValueError("weights must contain at least one asset")
+    if covariance.shape != (k, k):
+        raise ValueError(
+            f"cov shape {covariance.shape} does not match weights length {k}"
+        )
+    return w, covariance
+
+
 def risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     """风险贡献率 (RC_i): w_i * (Σw)_i / w'Σw。
 
@@ -25,11 +41,8 @@ def risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     Returns:
         RC: 风险贡献向量 (k,), Σ=1
     """
-    w = np.asarray(weights, dtype=float).ravel()
+    w, cov = _validate_inputs(weights, cov)
-    cov = np.asarray(cov, dtype=float)
     k = w.size
-    if cov.shape != (k, k):
-        raise ValueError(f"cov 形状 {cov.shape} 与 weights 长度 {k} 不匹配")
 
     port_var = float(w @ cov @ w)
     if port_var <= 0:
@@ -41,13 +54,11 @@ def risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
 
 def marginal_risk_contribution(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     """边际风险贡献 (MRC_i): (Σw)_i。"""
-    w = np.asarray(weights, dtype=float).ravel()
+    w, cov = _validate_inputs(weights, cov)
-    cov = np.asarray(cov, dtype=float)
     return cov @ w  # type: ignore[no-any-return]
 
 
 def component_var(weights: NDArray[Any], cov: NDArray[Any]) -> NDArray[Any]:
     """成分方差: w_i · (Σw)_i; 与 RC 的关系 RC_i = CV_i / w'Σw。"""
-    w = np.asarray(weights, dtype=float).ravel()
+    w, cov = _validate_inputs(weights, cov)
-    cov = np.asarray(cov, dtype=float)
     return w * (cov @ w)  # type: ignore[no-any-return]