diff --git a/src/quant_engine/backtest.py b/src/quant_engine/backtest.py index 6becddb..e083d36 100644 --- a/src/quant_engine/backtest.py +++ b/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] diff --git a/src/quant_engine/factor_library.py b/src/quant_engine/factor_library.py index 4fa4c59..68f6bad 100644 --- a/src/quant_engine/factor_library.py +++ b/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] diff --git a/src/quant_engine/metrics.py b/src/quant_engine/metrics.py index 270ceea..3f80b5e 100644 --- a/src/quant_engine/metrics.py +++ b/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()) diff --git a/src/quant_engine/risk.py b/src/quant_engine/risk.py index 2ca6a69..ea659e0 100644 --- a/src/quant_engine/risk.py +++ b/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() - cov = np.asarray(cov, dtype=float) + w, cov = _validate_inputs(weights, cov) 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() - cov = np.asarray(cov, dtype=float) + w, cov = _validate_inputs(weights, cov) 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() - cov = np.asarray(cov, dtype=float) + w, cov = _validate_inputs(weights, cov) return w * (cov @ w) # type: ignore[no-any-return]