mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 14:24:15 +08:00
Merge pull request #292 from kevin9327/fix/composite-single-candidate-neutral
fix(strategy): 叠加策略单候选子策略改用中性分, 与回测合并同口径
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user