fix(strategy): 叠加策略单候选子策略改用中性分, 与回测合并同口径

子策略当天只选出一只票时无法排名, 选股合并却按 max(count-1,1) 把它当成
"最优=1", 凭空抬高该票的融合分; 回测合并 (merge_signal_matrices 的
n <= 1 分支) 用的是中性分 0.5。同一天同一标的在选股页和回测里评分与排序
不一致 —— 正是本模块声明要防的口径分裂 (_NEUTRAL_NORM 注释也写明单候选
应取中性分)。
This commit is contained in:
kevin9327
2026-09-10 07:59:56 +09:00
parent d0a14b5c1b
commit be54e11912
2 changed files with 54 additions and 1 deletions
+6 -1
View File
@@ -80,7 +80,12 @@ def merge_results(
ordered = sorted(symbols, key=lambda s: res.scores[s], reverse=True)
count = len(ordered)
for rank, sym in enumerate(ordered, start=1):
norm[sym] = 1 - (rank - 1) / max(count - 1, 1)
# 单候选无法排名, 必须用中性分: 当成"最优=1"会凭空抬高融合分,
# 而回测合并 (merge_signal_matrices 的 n <= 1 分支) 用的是中性分,
# 两条路径同一天同一标的会给出不同评分与排序。
norm[sym] = (
_NEUTRAL_NORM if count <= 1 else 1 - (rank - 1) / (count - 1)
)
else:
# 子策略未产出 score: 命中即中性分, 不奖励也不惩罚。
for row in res.rows:
+48
View File
@@ -399,6 +399,54 @@ def test_composite_no_scores_uses_neutral(tmp_path):
assert all(abs(s - 50.0) < 0.01 for s in merged.scores.values()) # 中性分 0.5*100
def test_composite_single_candidate_child_matches_backtest_merge():
"""子策略当天只选出一只票时, 选股合并与回测合并必须给出同一套评分。
单候选无法排名, 只能用中性分 0.5; 若当成"最优=1"会凭空抬高该票的融合分,
与 merge_signal_matrices (n <= 1 → 中性分) 分叉 —— 同一天同一标的在选股页
和回测里评分与排序都不一样, 正是本模块要防的口径分裂。
"""
from app.backtest.matrix import make_signal_matrix
from app.strategy import composite as composite_mod
from app.strategy.engine import StrategyResult
as_of = date(2026, 1, 2)
child_a = StrategyResult(as_of=as_of, strategy_id="a", scores={"X": 7.0})
child_b = StrategyResult(
as_of=as_of, strategy_id="b", scores={"X": 1.0, "Y": 3.0, "Z": 2.0}
)
merged = composite_mod.merge_results(
[child_a, child_b], [1.0, 1.0], "union", 0, as_of=as_of, strategy_id="blend"
)
shape = (1, 3) # 一个交易日, 三只标的 X/Y/Z
def _sig(entry: list[int], score: list[float]):
return make_signal_matrix(
shape,
entry=np.array([entry], dtype=np.uint8),
exit=np.zeros(shape, dtype=np.uint8),
score=np.array([score], dtype=np.float32),
)
matrix = composite_mod.merge_signal_matrices(
shape,
[_sig([1, 0, 0], [7.0, 0.0, 0.0]), _sig([1, 1, 1], [1.0, 3.0, 2.0])],
[("a", 1.0), ("b", 1.0)],
"union",
0,
max_hold=1,
)
backtest_scores = dict(zip(("X", "Y", "Z"), matrix.score[0], strict=True))
# X 只被单候选子策略 a 命中: 中性分 0.5 与 b 的最差名 0 融合 → 25 分
assert abs(merged.scores["X"] - 25.0) < 0.01
for symbol in ("X", "Y", "Z"):
assert abs(merged.scores[symbol] - float(backtest_scores[symbol])) < 0.01, symbol
# 排序也一致: Y > Z > X
assert merged.scores["Y"] > merged.scores["Z"] > merged.scores["X"]
def test_composite_empty_children_returns_empty(tmp_path):
"""空子结果列表 → 返回空 StrategyResult。"""
from app.strategy import composite as composite_mod