feat: build target weights from factor scores

src/quant_engine/portfolio_construction.py

@@ -0,0 +1,100 @@
+"""因子分数到目标权重的轻量组合构建闭环。"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+from pandas.api.types import is_numeric_dtype
+
+__all__ = [
+    "select_top_k",
+    "equal_weight",
+    "scores_to_target_weights",
+    "scores_to_weight_table",
+]
+
+
+def _validate_top_k(top_k: int) -> None:
+    if isinstance(top_k, bool) or not isinstance(top_k, int) or top_k <= 0:
+        raise ValueError("top_k must be positive")
+
+
+def _validate_gross_exposure(gross_exposure: float) -> None:
+    if not np.isfinite(gross_exposure) or gross_exposure < 0:
+        raise ValueError("gross_exposure must be finite and non-negative")
+
+
+def _validate_score_series(scores: pd.Series) -> None:
+    if not isinstance(scores, pd.Series):
+        raise TypeError(f"scores must be a pandas Series, got {type(scores).__name__}")
+    if not scores.index.is_unique:
+        raise ValueError("scores must contain unique asset labels")
+    if not is_numeric_dtype(scores.dtype):
+        raise TypeError("scores must contain numeric values")
+
+
+def select_top_k(scores: pd.Series, top_k: int, *, largest: bool = True) -> pd.Index:
+    """稳定选择最高或最低的 K 个有效因子分数。"""
+    _validate_top_k(top_k)
+    _validate_score_series(scores)
+    valid_scores = scores.dropna()
+    ordered = valid_scores.sort_values(ascending=not largest, kind="mergesort")
+    return ordered.iloc[:top_k].index.copy()
+
+
+def equal_weight(assets: pd.Index, *, gross_exposure: float = 1.0) -> pd.Series:
+    """在已选资产间等权分配指定总敞口。"""
+    _validate_gross_exposure(gross_exposure)
+    if not assets.is_unique:
+        raise ValueError("assets must contain unique asset labels")
+    if assets.empty:
+        return pd.Series(index=assets.copy(), dtype=float, name="weight")
+    weight = gross_exposure / len(assets)
+    return pd.Series(weight, index=assets.copy(), dtype=float, name="weight")
+
+
+def scores_to_target_weights(
+    scores: pd.Series,
+    top_k: int,
+    *,
+    gross_exposure: float = 1.0,
+    largest: bool = True,
+) -> pd.Series:
+    """把单期因子分数转换为完整股票池目标权重。"""
+    _validate_score_series(scores)
+    selected = select_top_k(scores, top_k, largest=largest)
+    selected_weights = equal_weight(selected, gross_exposure=gross_exposure)
+    result = pd.Series(0.0, index=scores.index.copy(), dtype=float, name="weight")
+    result.loc[selected_weights.index] = selected_weights
+    return result
+
+
+def scores_to_weight_table(
+    scores: pd.DataFrame,
+    top_k: int,
+    *,
+    gross_exposure: float = 1.0,
+    largest: bool = True,
+) -> pd.DataFrame:
+    """逐调仓日独立构建目标权重表，避免使用未来分数。"""
+    if not isinstance(scores, pd.DataFrame):
+        raise TypeError(f"scores must be a pandas DataFrame, got {type(scores).__name__}")
+    _validate_top_k(top_k)
+    _validate_gross_exposure(gross_exposure)
+    if not scores.columns.is_unique:
+        raise ValueError("scores must contain unique asset labels")
+    if not all(is_numeric_dtype(dtype) for dtype in scores.dtypes):
+        raise TypeError("scores must contain numeric values")
+    if scores.empty:
+        return pd.DataFrame(index=scores.index.copy(), columns=scores.columns.copy(), dtype=float)
+
+    rows = [
+        scores_to_target_weights(
+            row,
+            top_k,
+            gross_exposure=gross_exposure,
+            largest=largest,
+        ).to_numpy()
+        for _, row in scores.iterrows()
+    ]
+    return pd.DataFrame(rows, index=scores.index.copy(), columns=scores.columns.copy(), dtype=float)