From 939149e8a0593ba584c3a20000d926aec89fe162 Mon Sep 17 00:00:00 2001 From: shy3130 Date: Thu, 29 Oct 2026 13:09:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(backtest):=20=E5=BC=BA=E5=88=B6=E9=87=8D?= =?UTF-8?q?=E8=B7=91=20=E2=80=94=20force=20=E7=BB=95=E8=BF=87=E5=90=8C?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E4=BB=BB=E5=8A=A1=E7=BB=93=E6=9E=9C=E7=BC=93?= =?UTF-8?q?=E5=AD=98;=20=E4=BF=AE=20cancel=20key=20=E5=AF=B9=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 策略回测按全参数 md5 作 job_key, 已完成任务结果保留 5 分钟 — 同参数再次 运行直接命中缓存返回旧结果 (重连/切页恢复设计的副产物), 用户改完数据后 重跑拿到的仍是旧行为产物, 此前没有绕过手段。 - startBacktest/strategy_stream 新增 force: 掺入 job_key 使其不命中缓存 必然重算; 不动任务复用/并发语义, 重连与取消按同一 qs 重算 key 天然对齐 - 运行回测旁新增「强制重跑」按钮 (RefreshCw, 次要样式, 悬停说明语义), disabled 条件与主按钮一致 - 补算环境并继续改为 handleRun(true): 已失败任务的 error 同样被任务缓存 保留, 补算成功后自动重跑不带 force 会命中同参数旧 job 直接重放补算前的 报错 — 用户实测补算完成后仍提示"缺少前一交易日环境 2019-10-22 等 112 天" 即此因 (实测数据已补齐: 2019-10-22~2026-10-29 共 1703 行无缺口) - 修预存 bug: /strategy/cancel 重算 job_key 漏了 minute_fill 与 regime_filter — 开了分钟K精确回测或市场环境过滤的任务停止按钮失效 验证: test_backtest_stream_force 3 passed (force 改 key 且幂等; cancel 带 force/minute_fill/regime_filter 的 qs 能取消到、缺参取消不到 — 走真实 qs→key 解析路径不启动回测线程); ruff 与 HEAD 对照零新增; pnpm build 通过。 Signed-off-by: shy3130 --- backend/app/api/backtest.py | 13 ++- backend/tests/test_backtest_stream_force.py | 96 +++++++++++++++++++ frontend/src/lib/backtestTask.ts | 3 + .../src/pages/backtest/StrategyBacktest.tsx | 48 ++++++---- 4 files changed, 143 insertions(+), 17 deletions(-) create mode 100644 backend/tests/test_backtest_stream_force.py diff --git a/backend/app/api/backtest.py b/backend/app/api/backtest.py index a746af93..3e130711 100644 --- a/backend/app/api/backtest.py +++ b/backend/app/api/backtest.py @@ -514,8 +514,9 @@ def _make_job_key( regime_filter: str | None = None, position_policy: str | None = None, position_policy_params: str | None = None, + force: bool = False, ) -> str: - raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}|{minute_fill}|{regime_filter}|{position_policy}|{position_policy_params}" + raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}|{minute_fill}|{regime_filter}|{position_policy}|{position_policy_params}|force={force}" return hashlib.md5(raw.encode()).hexdigest()[:12] @@ -546,12 +547,15 @@ async def strategy_stream( asset_type: str = "stock", minute_fill: bool = False, regime_filter: str | None = None, + force: bool = False, ): """SSE 流式策略回测: 实时推送进度, 完成后推送结果, 支持重连 (刷新/切页后恢复)。 - 相同参数的任务只启动一次, 多次连接订阅同一个任务 - 断开连接不会取消任务 (除非显式调用 cancel) - 结果保留 5 分钟供重连 + - force=1: 掺入 job_key 使其不命中已完成任务的缓存, 必然重算 + (重连/取消仍按 qs 重算同一 key, 语义不变) 事件类型: - progress: {day, total, date, equity} @@ -591,6 +595,7 @@ async def strategy_stream( regime_filter=regime_filter, position_policy=position_policy, position_policy_params=position_policy_params, + force=force, ) _cleanup_stale_jobs() @@ -740,6 +745,9 @@ async def strategy_cancel(request: Request): # 可选成本参数: 缺省或空串 → None (与 stream 侧 float | None 口径一致, 保证 job_key 对齐)。 v = _get(key) return float(v) if v else None + def _get_opt_bool(key: str) -> bool: + # 与 FastAPI 查询参数 bool 解析口径一致 ("1"/"true" 为真, 其余为假) + return _get(key).lower() in ("1", "true") job_key = _make_job_key( _get("strategy_id"), _get("symbols") or None, @@ -761,8 +769,11 @@ async def strategy_cancel(request: Request): commission_pct=_get_opt_float("commission_pct"), stamp_tax_pct=_get_opt_float("stamp_tax_pct"), asset_type=_get("asset_type", "stock"), + minute_fill=_get_opt_bool("minute_fill"), + regime_filter=_get("regime_filter") or None, position_policy=_get("position_policy") or None, position_policy_params=_get("position_policy_params") or None, + force=_get("force") == "1", ) # 持锁读任务表: 与 _cleanup_stale_jobs 的 pop、stream 的写入互斥 with _jobs_lock: diff --git a/backend/tests/test_backtest_stream_force.py b/backend/tests/test_backtest_stream_force.py new file mode 100644 index 00000000..da92296b --- /dev/null +++ b/backend/tests/test_backtest_stream_force.py @@ -0,0 +1,96 @@ +"""strategy/stream 的 force 参数 — 绕过同参数任务结果缓存。 + +SSE 回测按全参数 md5 作 job_key, 已完成任务保留 5 分钟内同 key 连接直接复用 +结果 (缓存)。force=1 掺入 job_key 使其不命中缓存必然重算; cancel 从 qs 重算 +同一 key, 必须同步解析 force, 否则强制重跑的任务取消不到。 +""" +from __future__ import annotations + +import threading + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.api import backtest as bt +from app.api.backtest import router + + +def _client() -> TestClient: + app = FastAPI() + app.include_router(router) + return TestClient(app) + + +def test_force_changes_job_key_and_is_stable(): + args = ("ma_cross", None, "2024-01-01", "2024-12-31", "open_t+1", None, None, + 0.0002, 5.0, 10, 1.0, 1_000_000.0, "equal", None, None) + cached = bt._make_job_key(*args) + forced = bt._make_job_key(*args, force=True) + assert cached != forced + # 各自幂等: 同参数重复计算 key 稳定 (重连/取消依赖) + assert bt._make_job_key(*args) == cached + assert bt._make_job_key(*args, force=True) == forced + # force=False 是默认值: 不显式传参与显式 False 同 key + assert bt._make_job_key(*args, force=False) == cached + + +def test_cancel_resolves_force_key_from_qs(monkeypatch): + """带 force=1 的任务, cancel 用带 force=1 的 qs 能取消到; 不带 force 的 qs 取消不到。 + + 手工往任务表塞一个未完成 job (key 按 force 计算), 走真实 cancel 端点的 + qs→key 解析路径, 不启动任何回测线程。 + """ + args = ("ma_cross", None, None, None, "open_t+1", None, None, + 0.0002, 5.0, 10, 1.0, 1_000_000.0, "equal", None, None) + forced_key = bt._make_job_key(*args, force=True) + job = bt._BacktestJob(forced_key) + job.cancel_event = threading.Event() + monkeypatch.setitem(bt._running_jobs, forced_key, job) + + qs_force = "strategy_id=ma_cross&force=1" + resp = _client().post("/api/backtest/strategy/cancel", json={"qs": qs_force}) + assert resp.status_code == 200 + assert resp.json() == {"ok": True} + assert job.cancel_event.is_set() + + # 不带 force 的 qs 算出的是缓存任务的 key, 取消不到 force 任务 + job2 = bt._BacktestJob(forced_key) + job2.cancel_event = threading.Event() + monkeypatch.setitem(bt._running_jobs, forced_key, job2) + qs_plain = "strategy_id=ma_cross" + resp2 = _client().post("/api/backtest/strategy/cancel", json={"qs": qs_plain}) + assert resp2.status_code == 200 + assert resp2.json()["ok"] is False + assert not job2.cancel_event.is_set() + + +def test_cancel_resolves_minute_fill_and_regime_filter(monkeypatch): + """minute_fill / regime_filter 掺在 job_key 里, cancel 必须同步解析, 否则取消不到。 + + 曾为预存 bug: stream 侧 key 含这两个参数, cancel 重算漏掉 —— 开了分钟K + 精确回测或市场环境过滤的任务停止按钮失效。 + """ + from urllib.parse import quote + + args = ("ma_cross", None, None, None, "open_t+1", None, None, + 0.0002, 5.0, 10, 1.0, 1_000_000.0, "equal", None, None) + regime_json = '{"states":["bull","shake"]}' + key = bt._make_job_key(*args, minute_fill=True, regime_filter=regime_json) + job = bt._BacktestJob(key) + job.cancel_event = threading.Event() + monkeypatch.setitem(bt._running_jobs, key, job) + + qs = f"strategy_id=ma_cross&minute_fill=true®ime_filter={quote(regime_json)}" + resp = _client().post("/api/backtest/strategy/cancel", json={"qs": qs}) + assert resp.status_code == 200 + assert resp.json() == {"ok": True} + assert job.cancel_event.is_set() + + # 缺这两个参数的 qs 算出不同 key, 取消不到 + job2 = bt._BacktestJob(key) + job2.cancel_event = threading.Event() + monkeypatch.setitem(bt._running_jobs, key, job2) + resp2 = _client().post("/api/backtest/strategy/cancel", json={"qs": "strategy_id=ma_cross"}) + assert resp2.status_code == 200 + assert resp2.json()["ok"] is False + assert not job2.cancel_event.is_set() diff --git a/frontend/src/lib/backtestTask.ts b/frontend/src/lib/backtestTask.ts index e6c902ae..c2d94707 100644 --- a/frontend/src/lib/backtestTask.ts +++ b/frontend/src/lib/backtestTask.ts @@ -214,6 +214,8 @@ export function startBacktest(params: { asset_type?: 'stock' | 'etf' minute_fill?: boolean regime_filter?: { states?: string[]; min_score?: number } | null + /** 忽略服务端同参数任务的结果缓存 (5 分钟), 必然重新计算; 重连/取消按 qs 对齐同一任务 */ + force?: boolean }): void { // 取消之前的任务状态 if (eventSource) { @@ -248,6 +250,7 @@ export function startBacktest(params: { asset_type: params.asset_type, minute_fill: params.minute_fill, regime_filter: params.regime_filter ? JSON.stringify(params.regime_filter) : undefined, + force: params.force ? 1 : undefined, }) // 存 reconnect 信息 (刷新后用) + 上报注册表 (其他设备可接回) diff --git a/frontend/src/pages/backtest/StrategyBacktest.tsx b/frontend/src/pages/backtest/StrategyBacktest.tsx index c61d2982..742f2c16 100644 --- a/frontend/src/pages/backtest/StrategyBacktest.tsx +++ b/frontend/src/pages/backtest/StrategyBacktest.tsx @@ -1,7 +1,7 @@ import { useState, useMemo, useEffect, useRef, useCallback, memo, type ReactNode } from 'react' import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import { motion, AnimatePresence } from 'framer-motion' -import { Play, FlaskConical, Clock, Loader2, Square, Search, Plus, X, SlidersHorizontal, BarChart3, Gauge, Zap, ListPlus, HelpCircle, ChevronRight, AlertTriangle, Layers, BookmarkPlus, Download, Filter, Star, TrendingUp, CalendarRange } from 'lucide-react' +import { Play, FlaskConical, Clock, Loader2, Square, Search, Plus, X, SlidersHorizontal, BarChart3, Gauge, Zap, ListPlus, HelpCircle, ChevronRight, AlertTriangle, Layers, BookmarkPlus, Download, Filter, Star, TrendingUp, CalendarRange, RefreshCw } from 'lucide-react' import { api, type StrategyBacktestResult, @@ -1242,7 +1242,7 @@ export function StrategyBacktest({ loadCandidate, onLoadConsumed }: { } }, [backtestTask]) - const handleRun = () => { + const handleRun = (force = false) => { if (!selectedStrategy || backtestDataUnavailable) return const requestOverrides = detail ? normalizeStrategyOverrides(detail, overrides) @@ -1277,6 +1277,7 @@ export function StrategyBacktest({ loadCandidate, onLoadConsumed }: { ...(regimeMinScore !== '' ? { min_score: Number(regimeMinScore) } : {}), } : null, + force, }) } @@ -1296,7 +1297,9 @@ export function StrategyBacktest({ loadCandidate, onLoadConsumed }: { mutationFn: () => api.regimeRecompute(regimeGapStart ?? undefined, end || undefined), onSuccess: data => { toast(`环境数据补算完成 (新增 ${data.computed} 天)，自动重新回测`, 'success') - handleRun() + // 数据已变, 必须 force: 已失败任务的 error 同样被 5 分钟任务缓存保留, + // 不带 force 会命中同参数旧 job 直接重放补算前的报错 + handleRun(true) }, onError: e => toast(`环境数据补算失败 · ${String((e as Error)?.message || e)}`, 'error'), }) @@ -2128,19 +2131,32 @@ export function StrategyBacktest({ loadCandidate, onLoadConsumed }: { 停止回测 ) : ( - +
+ + +
)}