mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 23:44:16 +08:00
kevin9327andClaude Opus 4.8
d92ef17851
问题: /api/backtest/run 设置 max_hold_days 时, 强制平仓完全不生效, 持仓不会 按最大持有天数退出, 导致交易记录、持仓周期和收益统计失真。 根因(services/backtest.py 的 max_hold 内联逻辑, 两处缺陷叠加): 1. `exits_idx.iloc[end_i][col] = True` 是链式索引: iloc[end_i] 先取出一行副本, 再对副本赋值, 在 pandas Copy-on-Write 语义下写入不会落到原矩阵(pandas 3.x 直接抛 ChainedAssignmentError)。强制退出单元格始终为 False。 2. `exits_idx = entries.copy()` 以入场矩阵起步, 使强制退出矩阵天然带上所有入场位; 即便修好第 1 点, 也会在入场当日就强制平仓, 而非 max_hold_days 之后。 修复: 抽出纯函数 _build_max_hold_exits(entries, max_hold_days), 从全 False 起步, 用单步定位 iloc[row, col_loc] 写入入场后第 max_hold_days 个交易日的强制退出位; run() 中改为与用户 exits 做 OR, 保留原有信号退出。抽出后该逻辑不依赖 vectorbt, 可独立回归(run() 整体仍需 vectorbt 可选依赖)。 边界: end_i 越界 clamp 到最后一根 K; 入场即最后一根 K 时 end_i==i 不产生退出。 验证: 新增 tests/backtest/test_max_hold_exits.py 覆盖强制退出落位、不误标入场位、 越界 clamp、末根不退出、多入场、多列独立 6 例, 修复后全过; 并以复刻旧内联逻辑的 脚本在 pandas 3.0.5 上确认修复前目标退出单元格恒为 False(ChainedAssignmentError)。 ruff 通过。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
"""max_hold_days 强制退出矩阵回归测试(issue #198)。
|
|
|
|
_build_max_hold_exits 是 /api/backtest/run 里 max_hold_days 强制平仓的纯逻辑,
|
|
不依赖 vectorbt, 可独立断言。覆盖两处历史缺陷:
|
|
1. 链式 `iloc[row][col] = True` 在 pandas CoW 下写入丢失 → 强制退出信号从不生效。
|
|
2. 以 `entries.copy()` 起步 → 把入场位当退出位, 入场当日即被平仓。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pandas as pd
|
|
|
|
from app.services.backtest import _build_max_hold_exits
|
|
|
|
|
|
def _entries(data: dict, n: int) -> pd.DataFrame:
|
|
return pd.DataFrame(data, index=pd.RangeIndex(n)).astype(bool)
|
|
|
|
|
|
def test_forced_exit_placed_max_hold_days_after_entry():
|
|
"""入场后第 max_hold_days 个交易日置强制退出(核心: 该单元格必须真的被写入)。"""
|
|
entries = _entries({"A": [True, False, False, False, False]}, 5)
|
|
out = _build_max_hold_exits(entries, 2)
|
|
assert out["A"].tolist() == [False, False, True, False, False]
|
|
|
|
|
|
def test_does_not_mark_entry_bar_as_exit():
|
|
"""回归: 强制退出矩阵不得包含入场位本身。"""
|
|
entries = _entries({"A": [True, False, False]}, 3)
|
|
out = _build_max_hold_exits(entries, 1)
|
|
assert out["A"].tolist() == [False, True, False]
|
|
|
|
|
|
def test_end_index_clamped_to_last_row():
|
|
"""入场后越界时 clamp 到最后一根 K。"""
|
|
entries = _entries({"A": [False, False, False, True, False]}, 5)
|
|
out = _build_max_hold_exits(entries, 5) # 3+5 越界 → clamp 到 4
|
|
assert out["A"].tolist() == [False, False, False, False, True]
|
|
|
|
|
|
def test_entry_on_last_row_produces_no_exit():
|
|
"""入场即最后一根 K 时 end_i == i, 不产生退出(避免同根自相矛盾)。"""
|
|
entries = _entries({"A": [False, False, True]}, 3)
|
|
out = _build_max_hold_exits(entries, 2)
|
|
assert out["A"].tolist() == [False, False, False]
|
|
|
|
|
|
def test_multiple_entries_single_column():
|
|
entries = _entries({"A": [True, False, True, False, False]}, 5)
|
|
out = _build_max_hold_exits(entries, 1)
|
|
assert out["A"].tolist() == [False, True, False, True, False]
|
|
|
|
|
|
def test_multiple_columns_independent():
|
|
entries = _entries({"A": [True, False, False], "B": [False, True, False]}, 3)
|
|
out = _build_max_hold_exits(entries, 1)
|
|
assert out["A"].tolist() == [False, True, False]
|
|
assert out["B"].tolist() == [False, False, True]
|
|
assert list(out.columns) == ["A", "B"]
|
|
assert out.index.equals(entries.index)
|