feat(selector): 短债作为动态现金底仓

设计理念变更:
- 短债(BOND大类)永远参与持仓,不受排名过滤影响
- 其他大类冠军需要排名 <= select_num才能入选
- 当其他标动量都不强时,仓位自动集中到短债(现金)
- 永远不会真正空仓,短债作为现金底仓始终存在

测试验证:
- 正常情况:[NDX, 399006.SZ, 931862.CSI](2动量强+1现金)
- 极端情况:[931862.CSI](仓位集中到现金)
This commit is contained in:
2026-05-18 22:42:08 +08:00
parent 6f915e67e1
commit 98597aa369

View File

@@ -193,11 +193,11 @@ class TopNSelector(SignalGenerator):
def _grouped_selection(self, scores: Dict[str, float]) -> List[str]:
"""分组选股先类内竞争每大类选Top1再跨类排序
改进:大类冠军二次过滤
1. min_score过滤负动量
2. 动量排名过滤大类冠军必须在全局Top select_num范围内才有效
- 假设短债排名第4select_num=3则短债被排除
- 这避免持有动量过低(排名靠后)的防御资产
设计理念:短债作为"动态现金"
1. 短债(BOND大类)永远参与持仓,不受排名过滤影响
2. 其他大类冠军需要排名 <= select_num才能入选
3. 当其他标的动量都不强时,仓位自动集中到短债(现金)
4. 这样永远不会真正空仓,短债作为现金底仓始终存在
"""
if not scores:
return []
@@ -211,26 +211,43 @@ class TopNSelector(SignalGenerator):
group_champions[group] = (code, score)
# ⭐ 计算全局动量排名(用于二次过滤)
# 将所有标的按得分排序,计算每个标的的排名
all_sorted = sorted(scores.items(), key=lambda x: x[1], reverse=True)
rank_map = {code: rank + 1 for rank, (code, _) in enumerate(all_sorted)}
# ⭐ 大类冠军二次过滤
# 只保留全局排名 <= select_num 的大类冠军
# ⭐ 大类冠军二次过滤特殊处理BOND大类
valid_champions = []
for group, (code, score) in group_champions.items():
rank = rank_map.get(code, len(all_sorted) + 1) # 未找到则排名为最后
bond_champion = None # 保存短债作为现金底仓
# 过滤条件:
# 1. 得分 >= min_score过滤负动量
# 2. 全局排名 <= select_num过滤排名靠后的冠军
for group, (code, score) in group_champions.items():
rank = rank_map.get(code, len(all_sorted) + 1)
# ⭐ BOND大类短债特殊处理永远保留作为现金底仓
if group == 'BOND':
if score >= self.min_score: # 只需满足min_score不受排名限制
bond_champion = (code, score, rank)
continue
# 其他大类:需要排名 <= select_num 才有效
if score >= self.min_score and rank <= self.select_num:
valid_champions.append((code, score, rank))
# 对有效冠军按得分排序选出Top N
# 持仓数量动态调整最多select_num最少可以是0
# ⭐ 组合构建:动量强的标的 + 短债(现金)
# 对有效冠军按得分排序
sorted_champions = sorted(valid_champions, key=lambda x: x[1], reverse=True)
return [code for code, score, rank in sorted_champions[:self.select_num]]
# 先选动量强的标的最多select_num-1个留1个位置给短债
top_codes = [code for code, score, rank in sorted_champions[:self.select_num - 1]]
# 再加入短债作为现金底仓(如果有)
if bond_champion:
top_codes.append(bond_champion[0])
# 如果没有短债min_score未满足则从其他冠军补满
if not bond_champion and len(top_codes) < self.select_num:
remaining = [code for code, score, rank in sorted_champions[self.select_num - 1:self.select_num]]
top_codes.extend(remaining)
return top_codes
class TrendFollower(SignalGenerator):