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)