feat: 短债动态阈值仓位分配机制

设计理念:
- 每份仓位 = 1/select_num
- 每个选中标的持有基础份额 1/select_num
- 被排除标的的份额归短债(BOND)继承

信号生成:
- generate()返回signal和signal_ranks
- _grouped_selection_with_ranks()返回标的和排名

仓位分配:
- DynamicThresholdAllocator.allocate()计算权重
- 短债继承被排除标的的份额

示例(短债排名2,select_num=3):
- NDX排名1 → 1/3(基础)
- 短债排名2 → 1/3(基础)+ 1/3(继承)= 2/3
- 排名3的份额被短债继承
This commit is contained in:
2026-05-18 23:07:21 +08:00
parent 2c69c6136a
commit c15e418ead
3 changed files with 200 additions and 12 deletions

View File

@@ -0,0 +1,10 @@
"""
仓位分配模块
提供多种仓位分配策略:
- DynamicThresholdAllocator: 短债动态阈值仓位分配
"""
from .dynamic_threshold import DynamicThresholdAllocator
__all__ = ['DynamicThresholdAllocator']

View File

@@ -0,0 +1,112 @@
"""
短债动态阈值仓位分配
设计理念:
1. 每份仓位 = 1/select_num
2. 每个选中标的持有基础份额 1/select_num
3. 被排除标的的份额归短债(BOND)继承
示例select_num=3, 短债排名2
- NDX排名1 → 1/3基础份额
- 短债排名2 → 1/3基础+ 1/3继承排名3的份额= 2/3
"""
from typing import Dict, List, Tuple
class DynamicThresholdAllocator:
"""短债动态阈值仓位分配器"""
def __init__(self, select_num: int, group_mapping: Dict[str, str] = None):
self.select_num = select_num
self.group_mapping = group_mapping or {}
def allocate(self, selected: List[str], ranks: List[int]) -> Dict[str, float]:
"""
计算仓位分配
Args:
selected: 选中标的列表
ranks: 对应排名列表
Returns:
{code: weight} 权重字典
"""
if not selected:
return {}
# 每份仓位
unit_weight = 1.0 / self.select_num
# 基础仓位:每个选中标的持有 1/select_num
weights = {code: unit_weight for code in selected}
# 找出短债(BOND大类)
bond_code = None
for code in selected:
if self.group_mapping.get(code) == 'BOND':
bond_code = code
break
# 计算被排除的份额数量
# effective_threshold = min(短债排名, select_num)
# 被排除份额 = select_num - effective_threshold
if bond_code and ranks:
bond_rank = ranks[selected.index(bond_code)]
effective_threshold = min(bond_rank, self.select_num)
excluded_slots = self.select_num - effective_threshold
# 被排除的份额归短债继承
if excluded_slots > 0:
weights[bond_code] += excluded_slots * unit_weight
# 确保权重总和为100%
total_weight = sum(weights.values())
if total_weight > 1.0:
# 如果超过100%,需要调整(极端情况)
for code in weights:
weights[code] /= total_weight
return weights
def get_position_info(self, selected: List[str], ranks: List[int]) -> Dict:
"""
获取详细仓位信息(用于报告)
Returns:
{
'weights': {code: weight},
'unit_weight': float,
'effective_threshold': int,
'excluded_slots': int,
'bond_inherited': float,
}
"""
weights = self.allocate(selected, ranks)
unit_weight = 1.0 / self.select_num
# 找短债
bond_code = None
bond_rank = None
for code in selected:
if self.group_mapping.get(code) == 'BOND':
bond_code = code
bond_rank = ranks[selected.index(code)]
break
info = {
'weights': weights,
'unit_weight': unit_weight,
'effective_threshold': self.select_num,
'excluded_slots': 0,
'bond_inherited': 0,
}
if bond_code and bond_rank:
effective_threshold = min(bond_rank, self.select_num)
excluded_slots = self.select_num - effective_threshold
info['effective_threshold'] = effective_threshold
info['excluded_slots'] = excluded_slots
info['bond_inherited'] = excluded_slots * unit_weight
return info