From d92ef178519b8199a4f08df66045ad4caa8d663e Mon Sep 17 00:00:00 2001 From: kevin9327 Date: Sun, 30 Aug 2026 18:47:08 +0900 Subject: [PATCH] =?UTF-8?q?fix(backtest):=20max=5Fhold=5Fdays=20=E5=BC=BA?= =?UTF-8?q?=E5=88=B6=E9=80=80=E5=87=BA=E4=BF=A1=E5=8F=B7=E7=9C=9F=E6=AD=A3?= =?UTF-8?q?=E7=94=9F=E6=95=88=20(#198)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题: /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 --- backend/app/services/backtest.py | 39 ++++++++---- backend/tests/backtest/test_max_hold_exits.py | 59 +++++++++++++++++++ 2 files changed, 88 insertions(+), 10 deletions(-) create mode 100644 backend/tests/backtest/test_max_hold_exits.py diff --git a/backend/app/services/backtest.py b/backend/app/services/backtest.py index d3c9640..4f4e942 100644 --- a/backend/app/services/backtest.py +++ b/backend/app/services/backtest.py @@ -118,6 +118,31 @@ _SIGNAL_COLS: dict[SignalKind, str] = { } +def _build_max_hold_exits(entries: pd.DataFrame, max_hold_days: int) -> pd.DataFrame: + """为每个入场信号在 max_hold_days 个交易日后生成一个强制退出信号。 + + 返回与 entries 同形状的布尔矩阵, 仅在「入场位之后第 max_hold_days 个交易日」置 + True(不含入场位本身), 供调用方与用户 exits 做 OR。 + + 两处易错点(见 #198): + - 必须从全 False 起步。若用 `entries.copy()` 起步会把入场位当成退出位, + 导致入场当日即被强制平仓。 + - 用单步定位写入 `iloc[row, col_loc]`。链式 `iloc[row][col] = True` 写入的是 + 临时行副本, 在 pandas Copy-on-Write 语义下不会落到原矩阵(pandas 3.x 直接报错), + 强制退出信号会静默丢失。 + """ + out = pd.DataFrame(False, index=entries.index, columns=entries.columns) + n = len(entries) + for col in entries.columns: + col_loc = out.columns.get_loc(col) + entry_rows = np.where(entries[col].to_numpy())[0] + for i in entry_rows: + end_i = min(int(i) + max_hold_days, n - 1) + if end_i > i: + out.iloc[end_i, col_loc] = True + return out + + class BacktestService: def __init__(self, repo: KlineRepository) -> None: self.repo = repo @@ -270,16 +295,10 @@ class BacktestService: if config.stop_loss_pct is not None: pf_kwargs["sl_stop"] = abs(config.stop_loss_pct) if config.max_hold_days is not None: - # vectorbt 没有内置 max-hold;用时间退出近似: - # 在 max_hold_days 后强制 exit - exits_idx = entries.copy() - for col in entries.columns: - entry_rows = np.where(entries[col].values)[0] - for i in entry_rows: - end_i = min(i + config.max_hold_days, len(entries) - 1) - if end_i > i: - exits_idx.iloc[end_i][col] = True - pf_kwargs["exits"] = (exits | exits_idx).astype(bool) + # vectorbt 没有内置 max-hold;用时间退出近似:入场后第 max_hold_days + # 个交易日强制 exit, 与用户 exits 做 OR(保留原有信号退出)。 + forced_exits = _build_max_hold_exits(entries, config.max_hold_days) + pf_kwargs["exits"] = (exits | forced_exits).astype(bool) pf = vbt.Portfolio.from_signals(**pf_kwargs) except Exception as e: # noqa: BLE001 diff --git a/backend/tests/backtest/test_max_hold_exits.py b/backend/tests/backtest/test_max_hold_exits.py new file mode 100644 index 0000000..f1264f3 --- /dev/null +++ b/backend/tests/backtest/test_max_hold_exits.py @@ -0,0 +1,59 @@ +"""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)