feat(weight): 实现 Kelly 仓位权重模式

- config_loader.py: WeightType 枚举新增 KELLY
- simple_rotation.py: compute_position_weights 新增 kelly 分支
  - 公式: w_i = max(score_i, 0) / sum(max(score_j, 0))
  - 负分自动排除 (Kelly: 不下注负期望)
  - 全负分时 fallback 到等权
- _generate_signals 传递 scores 给 kelly 模式
- config_simple.yaml: weight 改为 kelly
- 新增策略总结文档: kelly_weight.md

回测对比 (2020-2026):
- equal: 年化 19.88%, 夏普 1.13, 回撤 -14.65%
- rank:  年化 22.90%, 夏普 1.12, 回撤 -16.27%
- kelly: 年化 30.13%, 夏普 1.15, 回撤 -20.44%
This commit is contained in:
2026-06-08 23:04:41 +08:00
parent 844e609ff7
commit 8b7bcf206a
4 changed files with 239 additions and 4 deletions

View File

@@ -53,6 +53,7 @@ class WeightType(str, Enum):
"""仓位加权模式"""
EQUAL = "equal" # 等权
RANK = "rank" # 按排名加权 (slot i gets (N-i)/triangular(N))
KELLY = "kelly" # Kelly准则近似 (score-proportional weighting)
class DataSourceType(str, Enum):

View File

@@ -108,7 +108,7 @@ rebalance:
rotation:
diversified: true
select_num: 3
weight: rank
weight: kelly
threshold:
dynamic:
fallback_enabled: true

View File

@@ -171,13 +171,15 @@ def momentum_score(prices: np.ndarray) -> float:
def compute_position_weights(
ranked_holdings: List[str],
weight_type: str = 'equal',
scores: Dict[str, float] = None,
) -> Dict[str, float]:
"""Compute position weights from ranked slot list.
Args:
ranked_holdings: Ordered list of signal codes, best first.
May contain duplicates (e.g. bond fills).
weight_type: 'equal' or 'rank'.
weight_type: 'equal', 'rank', or 'kelly'.
scores: Required for 'kelly'. Dict mapping code -> momentum score.
Returns:
Dict mapping each unique code to its total weight (sum of slots).
@@ -186,6 +188,9 @@ def compute_position_weights(
equal: each slot = 1/N, duplicates summed.
rank: slot i (0-indexed) = (N-i) / triangular(N), duplicates summed.
For N=3: [3/6, 2/6, 1/6] = [50%, 33%, 17%].
kelly: w_i = max(score_i, 0) / sum(max(score_j, 0)).
Score-proportional weighting as Kelly criterion proxy.
Negative scores excluded (Kelly: don't bet on negative edge).
"""
N = len(ranked_holdings)
if N == 0:
@@ -193,7 +198,23 @@ def compute_position_weights(
weights: Dict[str, float] = {}
if weight_type == 'rank':
if weight_type == 'kelly':
if not scores:
raise ValueError("Kelly weighting requires 'scores' parameter")
# Kelly proxy: weight proportional to positive scores
positive_scores = {c: max(scores.get(c, 0.0), 0.0) for c in set(ranked_holdings)}
total = sum(positive_scores.values())
if total <= 0:
# Fallback to equal if all scores non-positive
w = 1.0 / len(positive_scores)
for code in positive_scores:
weights[code] = w
else:
for code in ranked_holdings:
w = positive_scores.get(code, 0.0) / total
weights[code] = weights.get(code, 0.0) + w
elif weight_type == 'rank':
triangular = N * (N + 1) / 2
for i, code in enumerate(ranked_holdings):
w = (N - i) / triangular
@@ -587,7 +608,7 @@ class SimpleRotationStrategy:
# These are *pending* weights; the caller (run) locks them in
# only when an actual rebalance occurs.
self._pending_weights = compute_position_weights(
ranked_holdings, self.weight_type,
ranked_holdings, self.weight_type, scores=factors,
)
return sorted(ranked_holdings), factors, bond_momentum