From 0559632de648accf31af7e21f701e80ffe1985f0 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Sun, 30 Aug 2026 19:05:16 +0800 Subject: [PATCH] =?UTF-8?q?feat(backtest):=20=E5=88=86=E9=92=9F=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E5=9B=9E=E6=B5=8B=20v1=20=E2=80=94=20=E9=80=90?= =?UTF-8?q?=E4=BA=A4=E6=98=93=E6=97=A5=E5=9B=9E=E6=94=BE=E4=BF=A1=E5=8F=B7?= =?UTF-8?q?=E5=88=86=E9=92=9F=E6=94=B6=E7=9B=98=E5=85=A5=E5=9C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 MinuteSignalReplayer: 枚举区间内分钟分区日, 逐日组装 StrategyDataContext (timeframe=1m, 日线窗口严格止于 T-1) 并复用 strategy_engine.run, 与实盘选股同路径 - 缺分区日显式跳过并记入 skipped_days, 不回退最近分区 - 涨停拒买: 信号分钟收盘 >= 当日涨停价(T-1 raw_close + 板块幅度)剔除并计数 - MarketMatrix 新增 entry_price 覆盖矩阵, 引擎在有限值处优先于 open/close 惯例 - repository.list_minute_dates 按目录名枚举分钟分区日, 零 parquet 扫描 - strategy.run 增加 minute_filter 分支: 入场 delay 0/离场沿用日K matcher, trades.entry_date 补全为 YYYY-MM-DD HH:MM (北京时间) --- backend/app/backtest/engine.py | 12 +- backend/app/backtest/matrix.py | 11 + backend/app/backtest/minute_replay.py | 286 +++++++++++++++++++++++ backend/app/backtest/strategy.py | 323 +++++++++++++++++++++++++- backend/app/tickflow/repository.py | 23 ++ 5 files changed, 652 insertions(+), 3 deletions(-) create mode 100644 backend/app/backtest/minute_replay.py diff --git a/backend/app/backtest/engine.py b/backend/app/backtest/engine.py index b464462..6501fff 100644 --- a/backend/app/backtest/engine.py +++ b/backend/app/backtest/engine.py @@ -813,6 +813,14 @@ class BacktestEngine: matrix, raw_candidates, config, progress_cb, cancel_event, ) + @staticmethod + def _resolve_entry_prices(matrix: "MarketMatrix", config: "MatcherConfig") -> np.ndarray: + """入场价矩阵: 分钟策略的逐格覆盖有限值处优先, 否则按 open/close 惯例。""" + base = matrix.open if config.entry_fill == "open_t+1" else matrix.close + if matrix.entry_price is None: + return base + return np.where(np.isfinite(matrix.entry_price), matrix.entry_price, base) + def _simulate_independent_matrix( self, matrix: MarketMatrix, @@ -823,7 +831,7 @@ class BacktestEngine: options: SimulationOptions | None = None, ) -> SimResult: options = options or SimulationOptions() - entry_prices = matrix.open if config.entry_fill == "open_t+1" else matrix.close + entry_prices = self._resolve_entry_prices(matrix, config) exit_prices = matrix.open if config.exit_fill == "open_t+1" else matrix.close buy_cost_pct = config.buy_cost_pct() sell_cost_pct = config.sell_cost_pct() @@ -1722,7 +1730,7 @@ class BacktestEngine: ) -> SimResult: options = options or SimulationOptions() time_count, asset_count = matrix.shape - entry_prices = matrix.open if config.entry_fill == "open_t+1" else matrix.close + entry_prices = self._resolve_entry_prices(matrix, config) exit_prices = matrix.open if config.exit_fill == "open_t+1" else matrix.close buy_cost_pct = config.buy_cost_pct() sell_cost_pct = config.sell_cost_pct() diff --git a/backend/app/backtest/matrix.py b/backend/app/backtest/matrix.py index d9cfda8..a70c69f 100644 --- a/backend/app/backtest/matrix.py +++ b/backend/app/backtest/matrix.py @@ -551,6 +551,9 @@ class MarketMatrix: exit_signal_code: np.ndarray entry_signal_ids: tuple[str, ...] exit_signal_ids: tuple[str, ...] + # 逐格入场价覆盖 (time x asset, NaN=回退 open/close 惯例)。分钟策略回测用: + # 信号在盘中第 m 根触发, 入场价 = 触发分钟收盘价, 而非当日开盘/收盘。 + entry_price: np.ndarray | None = None @property def shape(self) -> tuple[int, int]: @@ -2302,11 +2305,14 @@ def build_market_matrix_from_signals( exit_delay_bars: int = 0, reference_price: np.ndarray | None = None, minute_exit_trigger: bool = False, + entry_price_override: np.ndarray | None = None, ) -> MarketMatrix: """Combine base data and strategy signals into the matcher input matrix.""" if entry_delay_bars not in (0, 1) or exit_delay_bars not in (0, 1): raise ValueError("phase-two MarketMatrix supports only zero or one bar delay") validate_signal_matrix(signals, market.shape) + if entry_price_override is not None and entry_price_override.shape != market.shape: + raise ValueError("entry_price_override shape does not match MarketDataMatrix") present = _present_matrix(market.open, market.high, market.low, market.close, market.volume) entry, entry_signal_time, entry_signal_code = _delay_signal_matrix( @@ -2379,6 +2385,11 @@ def build_market_matrix_from_signals( exit_signal_code=exit_signal_code, entry_signal_ids=signals.entry_signal_ids, exit_signal_ids=signals.exit_signal_ids, + entry_price=( + np.array(entry_price_override, dtype=np.float32, copy=True) + if entry_price_override is not None + else None + ), ) diff --git a/backend/app/backtest/minute_replay.py b/backend/app/backtest/minute_replay.py new file mode 100644 index 0000000..2e3dd89 --- /dev/null +++ b/backend/app/backtest/minute_replay.py @@ -0,0 +1,286 @@ +"""分钟策略回测回放器 — 逐交易日回放 filter_minute_history 产生入场信号。 + +与实盘选股 (ScreenerService 1m context) 走同一条 StrategyEngine.run 执行路径, +消除回测/实盘偏差。语义铁律: + +- 分钟侧: 传入当日全量分钟分区, 策略函数自身因果 (第 m 根只用 <=m 的K线); +- 日线侧: T 日的日线条件窗口只含 T-1 及更早的完成态日K — 与实盘盘中行为一致 + (当日成形K不进窗口), 杜绝未来函数; +- 按交易日精确对日: 缺分钟分区的日子显式跳过, 不做"回退最近分区" (那是实盘语义)。 +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import date, timedelta + +import polars as pl + +from app.price_limits import is_risk_warning_name, price_limit_pct +from app.strategy.engine import StrategyDataContext, StrategyDef, StrategyEngine + +# 日线面板列: 基础行情 + 涨停/炸板信号 (策略日线窗口契约) + 基础过滤/展示列。 +# raw_close 用于涨停价计算 (分钟价是未复权真实价, 涨停规则定义在原始价上)。 +MINUTE_DAILY_PANEL_COLUMNS = frozenset({ + "open", "high", "low", "close", "volume", "amount", + "raw_close", "raw_high", "raw_low", + "turnover_rate", + "signal_limit_up", "signal_limit_down", "signal_broken_limit_up", +}) +MINUTE_INSTRUMENT_COLUMNS = frozenset({"name", "total_shares", "float_shares"}) + + +def minute_replay_feature_plan(daily_bars: int): + """分钟回测的日线面板加载计划。 + + execution_backend 用 polars_expr 走"按需计算信号"路径: enriched 分区只落 + 基础列, 涨停/炸板信号由 load_panel_for_backtest 的 compute_limit_signals + 按 signal_columns 需求现算 (matrix_native 路径会跳过通用信号计算)。 + """ + # 函数级导入规避与 strategy.py 的循环依赖 (strategy 顶层导入本模块)。 + from app.backtest.strategy import ResolvedFeaturePlan + + return ResolvedFeaturePlan( + base_columns=MINUTE_DAILY_PANEL_COLUMNS, + intermediate_columns=frozenset(), + indicator_columns=frozenset(), + signal_columns=frozenset({ + "signal_limit_up", "signal_limit_down", "signal_broken_limit_up", + }), + matrix_columns=frozenset(), + instrument_columns=MINUTE_INSTRUMENT_COLUMNS, + warmup_bars=max(daily_bars, 1), + full_feature_fallback=False, + execution_backend="polars_expr", + ) + + +def minute_panel_start(start: date, daily_bars: int) -> date: + """日线面板加载起点: 覆盖首个回测日的 daily_bars 交易日窗口。 + + N 个交易日约需 N*2 自然日 (周末/节假日), 再留 warmup 余量。 + """ + calendar_days = max(daily_bars, 1) * 2 + 30 + return start - timedelta(days=calendar_days) + + +def _trigger_hhmm(value) -> str: + """从 last_datetime 提取北京时间 "HH:MM" 触发分钟。 + + 分区 datetime 为 UTC 存储 (tz-aware 或 naive-UTC), 统一折算到北京时区。 + """ + from app.market_time import CN_TZ + + if hasattr(value, "astimezone"): + if value.tzinfo is None: + from datetime import timezone + + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(CN_TZ).strftime("%H:%M") + text = str(value or "") + if len(text) >= 16 and text[13] == ":": + return text[11:16] + return text[-5:] if text else "" + + +def _scalar_limit_up_price(prev_close: float, limit_pct: float) -> float: + """与 polars_limit_price 同口径的标量涨停价 (整数分半进位)。""" + cents = int(prev_close * 100 + 0.5) + numerator = round((1 + limit_pct) * 100) + return ((cents * numerator + 50) // 100) / 100 + + +@dataclass +class MinuteReplayHit: + """一个盘中入场信号: 触发分钟收盘买入。""" + + trade_date: date + symbol: str + # 已按当日 复权close/原始close 比例折算到复权价系的入场价, 与日线出场价同尺度。 + entry_price: float + trigger_time: str # "HH:MM" — 触发分钟K的时间戳 + score: float = 0.0 + + +@dataclass +class MinuteReplayResult: + hits: list[MinuteReplayHit] = field(default_factory=list) + skipped_days: list[date] = field(default_factory=list) + replayed_days: int = 0 + strategy_matches: int = 0 + buy_limit_up: int = 0 + elapsed_ms: float = 0.0 + + +class MinuteSignalReplayer: + """逐交易日回放分钟策略, 产出与实盘选股同源的入场命中。""" + + def __init__(self, engine, strategy_engine: StrategyEngine) -> None: + # engine: BacktestEngine — 只用其 repo (分钟分区读取)。 + self.engine = engine + self.strategy_engine = strategy_engine + + def replay( + self, + strategy: StrategyDef, + *, + panel: pl.DataFrame, + start: date, + end: date, + params: dict, + overrides: dict, + pool: list[str] | None = None, + symbols: list[str] | None = None, + progress_cb: Callable[[dict], None] | None = None, + cancel_event=None, + ) -> MinuteReplayResult: + t0 = time.perf_counter() + result = MinuteReplayResult() + repo = self.engine.repo + if panel.is_empty(): + return result + + universe = symbols if symbols else panel.get_column("symbol").unique().to_list() + daily_bars = int(strategy.minute_daily_bars or 0) + + # 面板交易日序列 (升序) — 日线窗口切片与缺分区日判定的基准。 + panel_dates = panel.get_column("date").unique().sort().to_list() + date_to_window: dict[date, tuple[date, date]] = {} + for i, day in enumerate(panel_dates): + window_start = panel_dates[max(0, i - daily_bars)] if daily_bars > 0 else day + date_to_window[day] = (window_start, day) + + # 逐分区日回放: 只回放 [start, end] 内有分钟分区的交易日。 + minute_days = repo.list_minute_dates(start, end, "stock") + minute_day_set = set(minute_days) + replay_days = [day for day in panel_dates if start <= day <= end] + result.skipped_days = [day for day in replay_days if day not in minute_day_set] + total = len(minute_days) + + # 逐标的的 T-1 原始收盘/复权收盘查表 (涨停价与复权折算用)。 + prev_raw_close: dict[str, float] = {} + prev_name: dict[str, str] = {} + adj_factor: dict[str, float] = {} + + for i, day in enumerate(minute_days): + if cancel_event is not None and cancel_event.is_set(): + break + if progress_cb is not None: + progress_cb({ + "day": i + 1, + "total": max(total, 1), + "date": str(day), + }) + + history = repo.get_minute_by_dates(universe, [day], "stock") + if history.is_empty(): + result.skipped_days.append(day) + continue + + # 日线窗口: 截至 T-1 的完成态日K (index of last panel date < day)。 + prior = [d for d in panel_dates if d < day] + if not prior: + # 面板起点之前的分区日 (窗口数据不足), 策略按数据不足自然不命中。 + daily_history = pl.DataFrame() + current = pl.DataFrame() + else: + last_prior = prior[-1] + window_start, _ = date_to_window[last_prior] + daily_history = panel.filter( + (pl.col("date") >= window_start) & (pl.col("date") <= last_prior) + ) if daily_bars > 0 else pl.DataFrame() + current = panel.filter(pl.col("date") == last_prior) + + # T-1 收盘/名称 + T 日复权因子查表。 + _refresh_day_lookups(prev_raw_close, prev_name, current, prior) + day_rows = panel.filter(pl.col("date") == day).select( + "symbol", "close", "raw_close", + ) + adj_factor.clear() + adj_factor.update(_adj_factors(day_rows)) + + context = StrategyDataContext( + asset_type="stock", + timeframe="1m", + as_of=day, + current=current if not current.is_empty() else None, + history=history, + daily_history=daily_history if not daily_history.is_empty() else None, + ) + try: + run_result = self.strategy_engine.run( + strategy.meta.get("id", ""), + context, + pool, + params, + overrides, + ) + except ValueError: + # 单日执行失败 (如窗口缺列) 记为跳过, 不中断整个回放。 + result.skipped_days.append(day) + continue + + result.replayed_days += 1 + result.strategy_matches += len(run_result.rows) + for row in run_result.rows: + symbol = row.get("symbol") + close = row.get("close") + if not symbol or close is None or float(close) <= 0: + continue + raw_close = float(close) + name = prev_name.get(str(symbol), "") + prev = prev_raw_close.get(str(symbol)) + # 涨停拒买: 触发分钟收盘已达当日涨停价 (按 T-1 原始收盘 + 板块规则)。 + if prev is not None and prev > 0: + limit_up = _scalar_limit_up_price( + prev, price_limit_pct(str(symbol), day, is_risk_warning=is_risk_warning_name(name)), + ) + if raw_close >= limit_up - 1e-9: + result.buy_limit_up += 1 + continue + trigger = row.get("last_datetime") + trigger_time = _trigger_hhmm(trigger) + result.hits.append(MinuteReplayHit( + trade_date=day, + symbol=str(symbol), + entry_price=raw_close * adj_factor.get(str(symbol), 1.0), + trigger_time=trigger_time, + score=float(run_result.scores.get(str(symbol), 0.0) or 0.0), + )) + + result.elapsed_ms = round((time.perf_counter() - t0) * 1000, 1) + return result + + +def _refresh_day_lookups( + prev_raw_close: dict[str, float], + prev_name: dict[str, str], + prior_snapshot: pl.DataFrame, + prior: list[date], +) -> None: + """从 T-1 快照刷新逐标的原始收盘与名称查表 (涨停价/ST 判定用)。""" + if prior_snapshot.is_empty(): + return + frame = prior_snapshot + if "raw_close" not in frame.columns: + frame = frame.with_columns(pl.col("close").alias("raw_close")) + if "name" not in frame.columns: + frame = frame.with_columns(pl.lit("").alias("name")) + prev_raw_close.clear() + prev_name.clear() + for symbol, raw_close, name in frame.select("symbol", "raw_close", "name").iter_rows(): + prev_raw_close[str(symbol)] = float(raw_close) if raw_close is not None else 0.0 + prev_name[str(symbol)] = str(name or "") + + +def _adj_factors(day_rows: pl.DataFrame) -> dict[str, float]: + """T 日 复权close/原始close 比例: 把分钟原始价折算到复权价系。""" + factors: dict[str, float] = {} + if day_rows.is_empty() or "raw_close" not in day_rows.columns: + return factors + for symbol, close, raw_close in day_rows.select("symbol", "close", "raw_close").iter_rows(): + if close and raw_close and float(raw_close) > 0: + factors[str(symbol)] = float(close) / float(raw_close) + return factors diff --git a/backend/app/backtest/strategy.py b/backend/app/backtest/strategy.py index 9ae6803..7a5d83d 100644 --- a/backend/app/backtest/strategy.py +++ b/backend/app/backtest/strategy.py @@ -28,13 +28,20 @@ from app.backtest.matrix import ( MatrixPipelineConfig, MatrixPrewarmCancelledError, MatrixStrategyPipeline, + SignalMatrix, apply_time_masks, + build_market_data_matrix, build_market_matrix, build_market_matrix_from_signals, rolling_mean, slice_market_data_matrix, slice_signal_matrix, ) +from app.backtest.minute_replay import ( + MinuteSignalReplayer, + minute_panel_start, + minute_replay_feature_plan, +) from app.backtest.minute_trigger import unsupported_minute_exit_signals from app.config import settings from app.indicators.pipeline import ( @@ -995,7 +1002,7 @@ class StrategyBacktestService: s, StrategyDataContext( asset_type=config.asset_type, - timeframe="1d", + timeframe="1m" if s.execution_backend == "minute_filter" else "1d", as_of=config.end, ), ) @@ -1044,6 +1051,26 @@ class StrategyBacktestService: overrides.get("score_max"), ) + if s.execution_backend == "minute_filter": + # 分钟策略回测: 逐交易日回放 filter_minute_history (与实盘选股同源), + # 信号分钟收盘价入场, 之后复用日K矩阵模拟的离场与组合管理。 + return self._run_minute_backtest( + config, s, params, overrides, + stop_loss=stop_loss, + take_profit=take_profit, + trailing_stop=trailing_stop, + trailing_take_profit_activate=trailing_take_profit_activate, + trailing_take_profit_drawdown=trailing_take_profit_drawdown, + max_hold_days=max_hold_days, + score_min=score_min, + score_max=score_max, + progress_cb=progress_cb, + cancel_event=cancel_event, + result_policy=result_policy, + run_id=run_id, + t0=t0, + ) + try: if s.execution_backend == "composite": # composite 回测: 子策略必须全为 matrix_native(否则 fail-closed), @@ -1629,6 +1656,300 @@ class StrategyBacktestService: elapsed_ms=round(elapsed, 1), ) + # ── 分钟策略回测: 逐日回放入场 + 日K矩阵离场 ── + + def _run_minute_backtest( + self, + config: StrategyBacktestConfig, + s: StrategyDef, + params: dict, + overrides: dict, + *, + stop_loss, + take_profit, + trailing_stop, + trailing_take_profit_activate, + trailing_take_profit_drawdown, + max_hold_days, + score_min, + score_max, + progress_cb, + cancel_event, + result_policy: BacktestResultPolicy, + run_id: str, + t0: float, + ) -> StrategyBacktestResult: + def _err(msg: str) -> StrategyBacktestResult: + return StrategyBacktestResult( + run_id=run_id, + config=self._config_to_dict(config), + error=msg, + elapsed_ms=(time.perf_counter() - t0) * 1000, + ) + + if config.asset_type != "stock": + return _err("分钟策略回测当前仅支持 A 股 (stock)") + if config.exit_fill == "signal_next_minute": + return _err("分钟策略回测暂不支持「信号触发卖出」离场口径") + + minute_days = self.engine.repo.list_minute_dates(config.start, config.end, "stock") + if not minute_days: + earliest = self.engine.repo.earliest_minute_date() + hint = f"本地分钟K最早到 {earliest}, " if earliest else "本地无分钟K数据, " + return _err( + f"回测区间内无分钟K数据: {hint}请先用「扩展分钟K历史」拉取, 或开启盘中分钟增量" + ) + + # 日线面板一次加载: 覆盖首个回测日的日线窗口 + 模拟区间 (含 full 模式尾部)。 + daily_bars = int(s.minute_daily_bars or 0) + feature_plan = minute_replay_feature_plan(daily_bars) + load_start = minute_panel_start(config.start, daily_bars) + full_horizon_days = int(max_hold_days or config.holding_days or 5) + load_end = config.end + if config.mode == "full": + load_end = config.end + timedelta(days=(full_horizon_days + 5) * 2) + sim_end = load_end if config.mode == "full" else config.end + + timing_ms: dict[str, float] = {} + t_load = time.perf_counter() + try: + panel = self.engine.load_panel_for_backtest( + config.symbols, + load_start, + load_end, + feature_plan, + asset_type="stock", + ) + except (ValueError, OSError, pl.exceptions.PolarsError) as e: + return _err(f"回测特征准备失败: {e}") + timing_ms["load_panel"] = round((time.perf_counter() - t_load) * 1000, 1) + if panel.is_empty(): + return _err("无日线数据, 请检查日期范围或先运行盘后管道") + + replayer = MinuteSignalReplayer(self.engine, self.strategy_engine) + replay = replayer.replay( + s, + panel=panel, + start=config.start, + end=config.end, + params=params, + overrides=overrides, + symbols=config.symbols, + progress_cb=progress_cb, + cancel_event=cancel_event, + ) + timing_ms["minute_replay"] = replay.elapsed_ms + if cancel_event is not None and cancel_event.is_set(): + return StrategyBacktestResult( + run_id=run_id, + config=self._config_to_dict(config), + error="cancelled", + elapsed_ms=round((time.perf_counter() - t0) * 1000, 1), + ) + if not replay.hits: + skipped_hint = ( + f" (区间内 {len(replay.skipped_days)} 个交易日缺分钟K分区被跳过)" + if replay.skipped_days else "" + ) + return _err("在指定区间内未产生买入信号" + skipped_hint) + + # 日频信号网格: 正式区间面板 → time x asset 矩阵, 命中格写入入场价覆盖。 + sim_panel = panel.filter( + (pl.col("date") >= config.start) & (pl.col("date") <= sim_end) + ) + if sim_panel.is_empty(): + return _err("正式回测区间内无数据") + axis_dates = sim_panel.get_column("date").unique().sort().to_list() + # 轴顺序必须与 build_market_data_matrix 的 _encode_axes 一致 (unique().sort()), + # 否则 (time, asset) 下标指向错误的标的。 + axis_symbols = sim_panel.get_column("symbol").cast(pl.Utf8).unique().sort().to_list() + time_index = {day: i for i, day in enumerate(axis_dates)} + asset_index = {sym: i for i, sym in enumerate(axis_symbols)} + shape = (len(axis_dates), len(axis_symbols)) + + entry = np.zeros(shape, dtype=np.uint8) + score = np.zeros(shape, dtype=np.float32) + entry_price_override = np.full(shape, np.nan, dtype=np.float32) + trigger_times: dict[tuple[str, date], str] = {} + dropped_axis_hits = 0 + for hit in replay.hits: + time_id = time_index.get(hit.trade_date) + asset_id = asset_index.get(hit.symbol) + if time_id is None or asset_id is None: + dropped_axis_hits += 1 + continue + entry[time_id, asset_id] = 1 + score[time_id, asset_id] = hit.score + entry_price_override[time_id, asset_id] = hit.entry_price + trigger_times[(hit.symbol, hit.trade_date)] = hit.trigger_time + raw_candidates = int(entry.sum()) + entry.setflags(write=False) + score.setflags(write=False) + entry_price_override.setflags(write=False) + exit_mask = np.zeros(shape, dtype=np.uint8) + exit_mask.setflags(write=False) + codes = np.zeros(shape, dtype=np.int16) + codes.setflags(write=False) + signals = SignalMatrix( + entry=entry, + exit=exit_mask, + score=score, + entry_signal_code=codes, + exit_signal_code=codes, + entry_signal_ids=(), + exit_signal_ids=(), + ) + + matcher_config = MatcherConfig( + matching=config.matching, + entry_fill="close_t", + exit_fill=config.exit_fill, + fees_pct=config.fees_pct, + commission_pct=config.commission_pct, + stamp_tax_pct=config.stamp_tax_pct, + slippage_bps=config.slippage_bps, + stop_loss_pct=stop_loss, + take_profit_pct=take_profit, + trailing_stop_pct=trailing_stop, + trailing_take_profit_activate_pct=trailing_take_profit_activate, + trailing_take_profit_drawdown_pct=trailing_take_profit_drawdown, + max_hold_days=max_hold_days, + max_positions=config.max_positions, + max_exposure_pct=config.max_exposure_pct, + score_min=score_min, + score_max=score_max, + initial_capital=config.initial_capital, + position_sizing=config.position_sizing, + # 分钟策略的成交价由 entry_price_override 提供 (触发分钟收盘), + # 不再叠加日线口径的分钟成交细化。 + minute_fill=False, + ) + + t_matrix = time.perf_counter() + market_data = build_market_data_matrix(sim_panel) + market_matrix = build_market_matrix_from_signals( + market_data, + signals, + # 入场即信号日盘中 (分钟价覆盖), 离场沿用日K口径。 + entry_delay_bars=0, + exit_delay_bars=1 if matcher_config.exit_fill == "open_t+1" else 0, + entry_price_override=entry_price_override, + ) + timing_ms["matrix_build"] = round((time.perf_counter() - t_matrix) * 1000, 1) + del sim_panel, market_data + + t_sim = time.perf_counter() + if config.mode == "full": + result = self.engine.simulate_independent_market_matrix( + market_matrix, + raw_candidates, + matcher_config, + progress_cb, + cancel_event, + result_policy.simulation_options(), + ) + else: + result = self.engine.simulate_market_matrix( + market_matrix, + matcher_config, + progress_cb, + cancel_event, + result_policy.simulation_options(), + ) + timing_ms["simulate"] = round((time.perf_counter() - t_sim) * 1000, 1) + timing_ms["statistics"] = float(result.stats.pop("statistics_ms", 0.0)) + + if cancel_event is not None and cancel_event.is_set(): + return StrategyBacktestResult( + run_id=run_id, + config=self._config_to_dict(config), + error="cancelled", + elapsed_ms=round((time.perf_counter() - t0) * 1000, 1), + ) + if result.stats.get("error"): + return _err(result.stats["error"]) + + execution = result.stats.get("execution") or {} + execution["buy_limit_up"] = int(execution.get("buy_limit_up", 0)) + replay.buy_limit_up + result.stats["execution"] = execution + timing_ms["total"] = round((time.perf_counter() - t0) * 1000, 1) + result.stats["timing_ms"] = timing_ms + result.stats["panel_rows"] = int(len(axis_dates) * len(axis_symbols)) + result.stats["panel_columns"] = 0 + result.stats["feature_columns"] = 0 + result.stats["execution_backend"] = s.execution_backend + result.stats["selection"] = { + "strategy_matches": replay.strategy_matches, + "entry_candidates": raw_candidates, + "entry_trigger_filtered": max(replay.strategy_matches - raw_candidates, 0), + "entry_trigger_enabled": False, + } + result.stats["minute_replay"] = { + "replayed_days": replay.replayed_days, + "skipped_days": [str(day) for day in replay.skipped_days[:50]], + "skipped_day_count": len(replay.skipped_days), + "dropped_axis_hits": dropped_axis_hits, + } + + benchmark_curve = ( + self._build_benchmark_curve(config.start, config.end) + if result_policy.include_benchmark + else [] + ) + strategy_info = { + "id": s.meta.get("id", config.strategy_id), + "name": s.meta.get("name", config.strategy_id), + "description": s.meta.get("description", ""), + "entry_signals": [], + "exit_signals": [], + "stop_loss": stop_loss, + "take_profit": take_profit, + "trailing_stop": trailing_stop, + "trailing_take_profit_activate": trailing_take_profit_activate, + "trailing_take_profit_drawdown": trailing_take_profit_drawdown, + "max_hold_days": max_hold_days, + "full_horizon_days": full_horizon_days, + "score_min": score_min, + "score_max": score_max, + "source": s.source, + "execution_backend": s.execution_backend, + } if result_policy.include_strategy_info else {} + + trades = ( + [self._trade_to_dict(t) for t in result.trades] + if result_policy.include_trades + else [] + ) + # 入场时间戳补分钟: 交易记录携带触发分钟 (HH:MM), 与日线回测的纯日期区分。 + for trade in trades: + entry_text = str(trade.get("entry_date") or "") + try: + key = (str(trade.get("symbol")), date.fromisoformat(entry_text[:10])) + except ValueError: + continue + trigger = trigger_times.get(key) + if trigger: + trade["entry_date"] = f"{entry_text[:10]} {trigger}" + + selected_stats = result_policy.select_stats(result.stats) + elapsed = (time.perf_counter() - t0) * 1000 + return StrategyBacktestResult( + run_id=run_id, + config=self._config_to_dict(config), + stats=selected_stats, + equity_curve=result.equity_curve if result_policy.include_curves else [], + drawdown_curve=result.drawdown_curve if result_policy.include_curves else [], + benchmark_curve=benchmark_curve, + trades=trades, + per_symbol_stats=( + result.per_symbol_stats + if result_policy.include_per_symbol_stats + else [] + ), + strategy_info=strategy_info, + elapsed_ms=round(elapsed, 1), + ) + # ── 全量模拟 (选股能力统计, 不建组合不算净值) ── def _run_full_simulation( diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index 64f89b7..81aae5e 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -1866,6 +1866,29 @@ class KlineRepository: return None return None + def list_minute_dates(self, start: date, end: date, asset_type: str = "stock") -> list[date]: + """枚举 [start, end] 内存在的分钟K分区日 (目录名直读, 零 parquet 扫描)。 + + 分钟回测按交易日精确对日: 缺分区的日子由调用方显式跳过, + 不做"回退最近分区" (那是实盘选股的语义, 回放会串日)。 + """ + dirname = "kline_minute" if asset_type == "stock" else f"kline_{asset_type}_minute" + minute_dir = self.store.data_dir / dirname + if not minute_dir.exists(): + return [] + out: list[date] = [] + for entry in minute_dir.iterdir(): + if not (entry.is_dir() and entry.name.startswith("date=")): + continue + try: + day = date.fromisoformat(entry.name[5:]) + except ValueError: + continue + if start <= day <= end: + out.append(day) + out.sort() + return out + def latest_daily_date(self) -> date | None: """本地日K数据的最新日期。""" try: