fix(selector): 短债作为动态过滤阈值(修正逻辑)

修正之前的误解,正确实现用户意图:
- 短债正常参与动量排序,没有任何特殊处理
- 短债排名 <= select_num → 短债被选中,比短债弱的标的被排除
- 短债排名 > select_num → 短债被排除(有更好的选择)
- effective_threshold = min(短债排名, select_num)

验证结果:
- 场景1:短债排名3 → 选[NDX, A, 短债],排名>3排除
- 场景2:短债排名2 → 选[NDX, 短债],排名>2排除(只持2只)
- 场景3:短债排名4 → 选Top3正常,短债被排除
This commit is contained in:
2026-05-18 22:54:09 +08:00
parent 98597aa369
commit 2c69c6136a

View File

@@ -193,11 +193,11 @@ class TopNSelector(SignalGenerator):
def _grouped_selection(self, scores: Dict[str, float]) -> List[str]: def _grouped_selection(self, scores: Dict[str, float]) -> List[str]:
"""分组选股先类内竞争每大类选Top1再跨类排序 """分组选股先类内竞争每大类选Top1再跨类排序
设计理念:短债作为"动态现金" 设计理念:短债作为"动态过滤阈值"
1. 短债(BOND大类)永远参与持仓,不受排名过滤影响 1. 短债正常参与动量排序,没有任何特殊处理
2. 其他大类冠军需要排名 <= select_num才能入选 2. 短债排名 <= select_num → 短债被选中,比短债弱的标的被排除
3. 当其他标的动量都不强时,仓位自动集中到短债(现金 3. 短债排名 > select_num → 短债被排除(有更好的选择
4. 这样永远不会真正空仓,短债作为现金底仓始终存在 4. 实际持仓数量 = min(短债排名, select_num),动态调整
""" """
if not scores: if not scores:
return [] return []
@@ -205,49 +205,46 @@ class TopNSelector(SignalGenerator):
# 建立 group -> (code, score) 的映射 # 建立 group -> (code, score) 的映射
group_champions = {} group_champions = {}
for code, score in scores.items(): for code, score in scores.items():
# 从group_mapping获取分组
group = self.group_mapping.get(code, 'default') group = self.group_mapping.get(code, 'default')
if group not in group_champions or score > group_champions[group][1]: if group not in group_champions or score > group_champions[group][1]:
group_champions[group] = (code, score) group_champions[group] = (code, score)
# 计算全局动量排名(用于二次过滤) # 计算全局动量排名
all_sorted = sorted(scores.items(), key=lambda x: x[1], reverse=True) all_sorted = sorted(scores.items(), key=lambda x: x[1], reverse=True)
rank_map = {code: rank + 1 for rank, (code, _) in enumerate(all_sorted)} rank_map = {code: rank + 1 for rank, (code, _) in enumerate(all_sorted)}
# ⭐ 大类冠军二次过滤特殊处理BOND大类 # ⭐ 找出短债(BOND大类)的排名位置
valid_champions = [] bond_rank = None
bond_champion = None # 保存短债作为现金底仓 for group, (code, score) in group_champions.items():
if group == 'BOND':
bond_rank = rank_map.get(code, len(all_sorted) + 1)
break
# ⭐ 确定有效排名阈值
# 如果短债排名在select_num内则以短债排名为阈值
# 如果短债排名超过select_num则以select_num为阈值
if bond_rank is not None and bond_rank <= self.select_num:
# 短债在Top select_num内以短债排名为阈值
# 比短债弱的标的(排名 > 短债排名)被排除
effective_threshold = bond_rank
else:
# 短债不在Top select_num内使用正常select_num阈值
effective_threshold = self.select_num
# ⭐ 大类冠军过滤
valid_champions = []
for group, (code, score) in group_champions.items(): for group, (code, score) in group_champions.items():
rank = rank_map.get(code, len(all_sorted) + 1) rank = rank_map.get(code, len(all_sorted) + 1)
# ⭐ BOND大类短债特殊处理永远保留作为现金底仓 # 过滤条件:
if group == 'BOND': # 1. 得分 >= min_score过滤负动量
if score >= self.min_score: # 只需满足min_score不受排名限制 # 2. 排名 <= effective_threshold动态阈值
bond_champion = (code, score, rank) if score >= self.min_score and rank <= effective_threshold:
continue
# 其他大类:需要排名 <= select_num 才有效
if score >= self.min_score and rank <= self.select_num:
valid_champions.append((code, score, rank)) valid_champions.append((code, score, rank))
# ⭐ 组合构建:动量强的标的 + 短债(现金) # 对有效冠军按得分排序选出Top N
# 对有效冠军按得分排序
sorted_champions = sorted(valid_champions, key=lambda x: x[1], reverse=True) 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): class TrendFollower(SignalGenerator):