diff --git a/backend/app/api/screener.py b/backend/app/api/screener.py index 784d105..fe6af72 100644 --- a/backend/app/api/screener.py +++ b/backend/app/api/screener.py @@ -317,7 +317,10 @@ def run_preset(req: PresetRequest, request: Request): raise HTTPException(status_code=status_code, detail=str(e)) from e safe_data = _safe(asdict(result)) - _update_cache_strategy(data_dir, str(as_of), req.strategy_id, safe_data) + # 分钟周期结果不写入盘后缓存 (strategy_cache 是日线语义, as_of/updated_at + # 混入分钟结果会污染页面秒加载路径)。 + if req.timeframe == "1d": + _update_cache_strategy(data_dir, str(as_of), req.strategy_id, safe_data) return _result_with_ext(safe_data, ext_values) @@ -583,8 +586,8 @@ def run_all(request: Request, body: Optional[dict] = None): elapsed = (time.perf_counter() - t_total) * 1000 logger.info("run_all: total took %.1fms (%d strategies)", elapsed, len(all_ids)) - # 写入策略缓存 (供页面秒加载) - if results: + # 写入策略缓存 (供页面秒加载); 分钟周期结果不落盘 (日线语义缓存) + if results and timeframe == "1d": try: strategy_cache.write_cache(data_dir, str(as_of), results) except Exception: # noqa: BLE001 diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index c10009e..b124b3b 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -391,6 +391,10 @@ class MinuteSyncPrefs(BaseModel): minute_sync_days: int = 5 # 单段大小(交易日),None 表示不修改现有值。范围 [5, 30],默认 20。 minute_sync_segment_days: int | None = None + # 盘中分钟增量刷新 (Expert 专有)。None 表示不修改现有值。 + minute_refresh_enabled: bool | None = None + # 刷新间隔(秒),范围 [60, 300]。None 表示不修改现有值。 + minute_refresh_interval: int | None = None class DataProvidersIn(BaseModel): @@ -478,6 +482,8 @@ def get_preferences() -> dict: "minute_sync_enabled": preferences.get_minute_sync_enabled(), "minute_sync_days": preferences.get_minute_sync_days(), "minute_sync_segment_days": preferences.get_minute_sync_segment_days(), + "minute_refresh_enabled": preferences.get_minute_refresh_enabled(), + "minute_refresh_interval": preferences.get_minute_refresh_interval(), "daily_data_provider": preferences.get_daily_data_provider(), "adj_factor_provider": preferences.get_adj_factor_provider(), "minute_data_provider": preferences.get_minute_data_provider(), @@ -833,14 +839,29 @@ def update_minute_sync(req: MinuteSyncPrefs) -> dict: } if req.minute_sync_segment_days is not None: updates["minute_sync_segment_days"] = max(5, min(30, req.minute_sync_segment_days)) + if req.minute_refresh_enabled is not None: + updates["minute_refresh_enabled"] = req.minute_refresh_enabled + if req.minute_refresh_interval is not None: + updates["minute_refresh_interval"] = max(60, min(300, req.minute_refresh_interval)) preferences.save(updates) return { "minute_sync_enabled": req.minute_sync_enabled, "minute_sync_days": days, "minute_sync_segment_days": preferences.get_minute_sync_segment_days(), + "minute_refresh_enabled": preferences.get_minute_refresh_enabled(), + "minute_refresh_interval": preferences.get_minute_refresh_interval(), } +@router.get("/minute-refresh/status") +def minute_refresh_status(request: Request) -> dict: + """盘中分钟增量刷新服务状态 (开关/能力门控/最近一轮/下一轮)。""" + svc = getattr(request.app.state, "minute_refresh", None) + if svc is None: + return {"available": False} + return {"available": True, **svc.status()} + + class RealtimeQuotesPrefs(BaseModel): realtime_quotes_enabled: bool diff --git a/backend/app/main.py b/backend/app/main.py index a65e130..801fa96 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -173,6 +173,16 @@ async def _application_lifespan(app: FastAPI): except Exception as e: # noqa: BLE001 logger.warning("depth_service init failed: %s", e) + # 盘中分钟增量刷新 (Expert 专有): 线程常驻, 开关/时段/能力门控在循环内每轮判断 + try: + from app.services.minute_refresh import MinuteRefreshService + minute_refresh = MinuteRefreshService(repo) + minute_refresh.set_app_state(app.state) + app.state.minute_refresh = minute_refresh + minute_refresh.start() + except Exception as e: + logger.warning("minute_refresh init failed: %s", e) + # 停机缺口自检: 延迟后台扫描, 发现最近交易日的盘中快照/缺口时自动创建 # 修复任务 (盘中停机→次日开实时场景, 不修则坏数据被"只刷今天"分支永久留存) try: @@ -356,6 +366,9 @@ async def _application_lifespan(app: FastAPI): wbot = getattr(app.state, "wecom_bot_service", None) if wbot: wbot.stop() + mrs = getattr(app.state, "minute_refresh", None) + if mrs: + mrs.stop() logger.info("shutdown") diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index 46789c1..2350141 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -828,6 +828,52 @@ def fetch_intraday_monitor_batch( return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame() +def fetch_intraday_full_market_burst( + symbols: list[str], + capset: CapabilitySet | None, + *, + count: int = 300, +) -> tuple[pl.DataFrame, int]: + """全市场当日分钟K并发脉冲拉取 (盘中增量刷新专用, 不落盘)。 + + 与 fetch_intraday_monitor_batch 的区别: + - 监控路径每轮只拉少量标的 (≤ batch 上限, 单请求); + 本函数按 batch_size 把全市场切块后用线程池一次全部打出 + (5546/200 = 28 并发), 配合 >=60s 的固定轮节奏, 任何 60s + 滑动窗口至多一个脉冲 (28 < 48 安全 rpm), 轮内失败不重试。 + + 限流口径: 只用 intraday.batch 独立池 (Cap.INTRADAY_BATCH, Expert 专有), + 不与 kline.minute.batch (盘后分钟同步) 共享配额。 + 返回 (当日全市场分钟K, 请求数)。 + """ + if not symbols: + return (pl.DataFrame(), 0) + limits = capset.limits(Cap.INTRADAY_BATCH) if capset and capset.has(Cap.INTRADAY_BATCH) else None + batch_size = max(1, int(limits.batch) if limits and limits.batch else 200) + chunks = list(chunked(symbols, batch_size)) + if not chunks: + return (pl.DataFrame(), 0) + + from concurrent.futures import ThreadPoolExecutor + + tf = get_client() + + def _fetch(chunk: list[str]) -> list[pl.DataFrame]: + raw = tf.klines.intraday_batch( + chunk, count=count, as_dataframe=True, show_progress=False, + batch_size=len(chunk), + ) + return _normalize_intraday_raw(raw) + + frames: list[pl.DataFrame] = [] + with ThreadPoolExecutor(max_workers=min(len(chunks), 32)) as pool: + for result in pool.map(_fetch, chunks): + frames.extend(result) + if not frames: + return (pl.DataFrame(), len(chunks)) + return (pl.concat(frames, how="diagonal_relaxed"), len(chunks)) + + def fetch_minute_single( symbol: str, trade_date: date, diff --git a/backend/app/services/minute_refresh.py b/backend/app/services/minute_refresh.py new file mode 100644 index 0000000..43035de --- /dev/null +++ b/backend/app/services/minute_refresh.py @@ -0,0 +1,238 @@ +"""盘中分钟K增量落盘服务 (Expert 专有)。 + +每轮用 intraday.batch (日内分时批量, 独立限流池) 并发脉冲拉全市场当日分钟K, +单次合并写入当日 kline_minute 分区, 供分钟策略 (minute_filter) 读到新鲜数据。 + +设计约束 (见 feat/minute-strategy 方案): +- Expert 专有: 能力门控 Cap.INTRADAY_BATCH — 该能力仅 Expert 档具备, 天然排他。 +- 并发脉冲: 全市场按 batch_size 分块 (5546/200 = 28 块), ThreadPoolExecutor 一次 + 打出全部块 (≤28 并发)。任何 60s 滑动窗口至多一个脉冲 (28 < 48 安全 rpm)。 +- 固定节奏: 默认 60s 一轮 (clamp [60, 300]), 下一轮 = max(本轮起点+间隔, 上轮完成), + 不补跑 (missed 轮次直接跳过), 轮内失败不重试。 +- 仅连续竞价时段运行 (9:30-11:30 / 13:00-15:00), 午休/收盘自动暂停与恢复。 +- 不与其他分钟能力冲突: 与 盘后分钟同步 (kline.minute.batch) / 分时监控路径 + (fetch_intraday_monitor_batch) 分属不同限流池; 落盘走 _write_minute_partition + 的 unique(symbol,datetime) 合并, 与盘后同步写同一分区安全幂等。 +- 数据源插件化让位: 配置了自定义分钟源 (minute_data_provider != tickflow) 时 + 服务不启动 — 盘中增量交由插件自管, 本服务不抢占。 + +分层: 本模块只做调度/落盘/状态; TickFlow SDK 调用全部在 kline_sync 边界层 +(fetch_intraday_full_market_burst), 保持插件化边界不泄漏。 +""" +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass, field +from datetime import time as dt_time +from typing import Any + +import polars as pl + +from app.market_time import cn_now +from app.services import preferences + +# 轮询间隔允许范围 (秒): 下限 60s 保证任何滑动窗口 ≤1 个脉冲, 上限防误配。 +REFRESH_INTERVAL_MIN = 60 +REFRESH_INTERVAL_MAX = 300 +# 等待步长 (秒): 循环小步睡眠, 便于快速停止与偏好热生效。 +_LOOP_STEP_S = 2.0 + + +def _in_continuous_session(now=None) -> bool: + """A股连续竞价时段 (北京时间): 9:30-11:30 / 13:00-15:00, 仅工作日。""" + now = now or cn_now() + t = now.time() + morning = dt_time(9, 30) <= t <= dt_time(11, 30) + afternoon = dt_time(13, 0) <= t <= dt_time(15, 0) + return now.weekday() < 5 and (morning or afternoon) + + +@dataclass +class _RefreshState: + """服务运行状态 (status() 的内存镜像, 循环线程内更新)。""" + + rounds: int = 0 + last_round_at: float | None = None # epoch 秒 + last_round_ms: float | None = None # 单轮耗时 + last_rows: int = 0 # 上轮写入行数 (合并后) + last_symbols: int = 0 # 上轮覆盖标的数 + last_requests: int = 0 # 上轮请求数 (分块数) + last_error: str | None = None + next_round_at: float | None = None # epoch 秒 + extra: dict[str, Any] = field(default_factory=dict) + + +class MinuteRefreshService: + """盘中分钟增量刷新: 单实例挂 app.state.minute_refresh, 后台守护线程。""" + + def __init__(self, repo) -> None: + self._repo = repo + self._app_state: Any | None = None + self._thread: threading.Thread | None = None + self._stop = threading.Event() + self._state = _RefreshState() + self._round_lock = threading.Lock() # 同时只允许一轮 (手动触发与定时轮互斥) + + # ------------------------------------------------------------------ + # 生命周期 + # ------------------------------------------------------------------ + + def set_repo(self, repo) -> None: + self._repo = repo + + def set_app_state(self, app_state: Any) -> None: + self._app_state = app_state + + def start(self) -> bool: + """启动后台线程 (幂等)。开关/时段/能力判断都在循环内每轮做, 热生效。""" + if self._thread is not None and self._thread.is_alive(): + return True + self._stop.clear() + self._thread = threading.Thread( + target=self._loop, name="minute-refresh", daemon=True, + ) + self._thread.start() + return True + + def stop(self) -> None: + self._stop.set() + + # ------------------------------------------------------------------ + # 门控 + # ------------------------------------------------------------------ + + def capability_ok(self) -> bool: + """Cap.INTRADAY_BATCH 存在 (Expert)。能力探测结果缓存在 app.state。""" + capset = getattr(self._app_state, "capabilities", None) if self._app_state else None + if capset is None: + return False + try: + from app.tickflow.capabilities import Cap + + return capset.has(Cap.INTRADAY_BATCH) + except Exception: + return False + + def custom_provider_active(self) -> bool: + """配置了自定义分钟源 → 让位插件, 本服务不启动。""" + try: + return preferences.get_minute_data_provider() != "tickflow" + except Exception: + return False + + def _gate_reason(self) -> str | None: + """返回本轮不执行的原因 (None = 放行)。""" + if not preferences.get_minute_refresh_enabled(): + return "disabled" + if self.custom_provider_active(): + return "custom_minute_provider" + if not self.capability_ok(): + return "capability" + if not _in_continuous_session(): + return "outside_trading_hours" + return None + + # ------------------------------------------------------------------ + # 主循环 + # ------------------------------------------------------------------ + + def _loop(self) -> None: + while not self._stop.is_set(): + try: + reason = self._gate_reason() + if reason is None: + interval = preferences.get_minute_refresh_interval() + started = time.time() + self._run_round() + # 固定节奏: 下一轮 = max(本轮起点+间隔, 本轮完成), 不补跑 + finish = time.time() + self._state.next_round_at = max(started + interval, finish) + # 等到下一轮 (小步睡眠保持可停/偏好热切换) + while not self._stop.is_set(): + now = time.time() + gate = self._gate_reason() + if gate is not None: + self._state.next_round_at = None + break # 门控关闭 → 回外层等待重评估 + if now >= self._state.next_round_at: + break + self._stop.wait(min(_LOOP_STEP_S, max(0.0, self._state.next_round_at - now))) + continue + except Exception as e: + self._state.last_error = f"round failed: {e}" + self._stop.wait(_LOOP_STEP_S) + + # ------------------------------------------------------------------ + # 单轮 + # ------------------------------------------------------------------ + + def _run_round(self) -> None: + from app.services import kline_sync + + t0 = time.perf_counter() + symbols = self._universe() + self._state.last_symbols = len(symbols) + if not symbols: + self._state.last_error = "empty universe (instruments 未加载)" + return + + capset = getattr(self._app_state, "capabilities", None) if self._app_state else None + with self._round_lock: + df, requests = kline_sync.fetch_intraday_full_market_burst(symbols, capset) + self._state.last_requests = requests + if df.is_empty(): + self._state.last_error = "intraday burst returned no data" + return + written = kline_sync._write_minute_partition( + df, self._repo.store.data_dir / "kline_minute", + ) + + self._state.rounds += 1 + self._state.last_round_at = time.time() + self._state.last_round_ms = (time.perf_counter() - t0) * 1000 + self._state.last_rows = written + self._state.last_error = None + + def _universe(self) -> list[str]: + """全市场 A 股标的 (instruments 维表, 与盘后分钟同步同一来源)。""" + inst = self._repo.get_instruments() + if inst.is_empty() or "symbol" not in inst.columns: + return [] + return inst["symbol"].cast(pl.Utf8).unique().sort().to_list() + + # ------------------------------------------------------------------ + # 状态 + # ------------------------------------------------------------------ + + def status(self) -> dict[str, Any]: + import contextlib + + with contextlib.suppress(Exception): + enabled = preferences.get_minute_refresh_enabled() + running = self._thread is not None and self._thread.is_alive() + gate = self._gate_reason() + return { + "enabled": enabled, + "running": running, + "interval_seconds": preferences.get_minute_refresh_interval(), + "capability_ok": self.capability_ok(), + "custom_provider_active": self.custom_provider_active(), + "in_trading_hours": _in_continuous_session(), + "gate_reason": gate if (enabled and running) else (gate or "disabled"), + "rounds": self._state.rounds, + "last_round_at": self._state.last_round_at, + "last_round_ms": self._state.last_round_ms, + "last_rows": self._state.last_rows, + "last_symbols": self._state.last_symbols, + "last_requests": self._state.last_requests, + "next_round_at": self._state.next_round_at, + "last_error": self._state.last_error, + } + + def trigger_manual_round(self) -> dict[str, Any]: + """手动触发一轮 (无视时段门控, 但仍受能力/插件门控); 供状态页「立即刷新」。""" + if self.custom_provider_active() or not self.capability_ok(): + return {"ok": False, "reason": self._gate_reason() or "capability"} + threading.Thread(target=self._run_round, daemon=True, name="minute-refresh-manual").start() + return {"ok": True} diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index f5470ff..184c576 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -214,6 +214,25 @@ def get_minute_sync_segment_days() -> int: """ return max(5, min(30, load().get("minute_sync_segment_days", 20))) +# ===== 盘中分钟增量刷新 (Expert 专有, intraday.batch 独立限流池) ===== + +# 下限 60s: 保证任何 60s 滑动窗口至多一个全市场脉冲 (28 并发 < 48 安全 rpm)。 +_MINUTE_REFRESH_INTERVAL_MIN = 60 +_MINUTE_REFRESH_INTERVAL_MAX = 300 + + +def get_minute_refresh_enabled() -> bool: + """盘中分钟K增量落盘开关。默认关闭; 能力门控 (Expert) 在服务层判断。""" + return bool(load().get("minute_refresh_enabled", False)) + + +def get_minute_refresh_interval() -> int: + """盘中分钟增量刷新间隔(秒)。默认 60,范围 [60, 300]。""" + return max( + _MINUTE_REFRESH_INTERVAL_MIN, + min(_MINUTE_REFRESH_INTERVAL_MAX, int(load().get("minute_refresh_interval", 60))), + ) + # ===== 数据源选择 (默认 TickFlow；第一阶段仅日K切换入口) ===== diff --git a/backend/app/services/screener.py b/backend/app/services/screener.py index 1e62e29..134cc90 100644 --- a/backend/app/services/screener.py +++ b/backend/app/services/screener.py @@ -386,6 +386,19 @@ class ScreenerService: if current is None: current = self._load_enriched_for_date(as_of) + if timeframe == "1m": + # 分钟策略数据源是本地当日分钟K分区 (单分区文件直读), 与日线 + # enriched 历史窗口无关, 不走 required_history_bars 日线路径。 + history = self._load_minute_history(as_of, current) + return StrategyDataContext( + asset_type=self.asset_type, + timeframe=timeframe, + as_of=as_of, + current=current, + history=history, + market=None, + cache_key=cache_key, + ) history_bars = engine.required_history_bars( strategy_ids, params_map=params_map, @@ -404,6 +417,31 @@ class ScreenerService: cache_key=cache_key, ) + def _load_minute_history(self, as_of: date, current: pl.DataFrame | None) -> pl.DataFrame: + """分钟策略数据源: 优先 as_of 当日分钟分区, 缺失时回退全市场最近分区。 + + 只按日期直读单个分区文件 (get_minute_by_dates), 与全量 glob 扫描解耦, + 内存只随当日分区大小 (~67万行) 走。标的池限定为 enriched 快照 universe; + 分区与快照的日期差是允许的 (分钟分区可能比 enriched 更新, 行自带时间戳)。 + """ + if self.asset_type != "stock": + raise ValueError("分钟策略当前仅支持 A 股") + symbols: list[str] = [] + if current is not None and not current.is_empty(): + symbols = current["symbol"].cast(pl.Utf8).unique().to_list() + if not symbols: + return pl.DataFrame() + df = self.repo.get_minute_by_dates(symbols, [as_of]) + if df.is_empty(): + fallback = self.repo.latest_minute_date_global() + if fallback is None: + raise ValueError( + "无分钟K数据 — 请先在 数据→分钟K 完成同步, 或开启盘中增量刷新" + ) + if fallback != as_of: + df = self.repo.get_minute_by_dates(symbols, [fallback]) + return df + def latest_date(self) -> date | None: if self.asset_type != "stock": _, d = self.repo.get_enriched_latest_asset(self.asset_type) diff --git a/backend/app/strategy/builtin/minute_red_streak.py b/backend/app/strategy/builtin/minute_red_streak.py new file mode 100644 index 0000000..4de22b6 --- /dev/null +++ b/backend/app/strategy/builtin/minute_red_streak.py @@ -0,0 +1,110 @@ +"""分钟红7 — 最近 N 根分钟K多数收红, 且最高的 top_red 根全红。 + +数据契约: filter_minute_history 接收当日全市场分钟K窗口 +(symbol, datetime, open, high, low, close, volume, amount), +由 ScreenerService.build_strategy_context 的 1m 分支从本地 kline_minute +分区注入; 策略本身不感知数据来源 (本地同步 / 盘中增量刷新对它透明)。 +""" + +import polars as pl + +META = { + "id": "minute_red_streak", + "name": "分钟红7", + "description": "最近7根1分钟K至少5根收红, 且最高的2根(按最高价)都是红K", + "tags": ["分钟", "形态", "短线"], + "asset_types": ["stock"], + "timeframes": ["1m"], + "params": [ + { + "id": "bars", + "label": "检查K线数", + "type": "int", + "default": 7, + "min": 5, + "max": 15, + "step": 1, + }, + { + "id": "min_red", + "label": "最少红K数", + "type": "int", + "default": 5, + "min": 1, + "max": 15, + "step": 1, + }, + { + "id": "top_red", + "label": "最高K需红数", + "type": "int", + "default": 2, + "min": 1, + "max": 3, + "step": 1, + }, + { + "id": "rank_by_close", + "label": "最高K按收盘价排序", + "type": "bool", + "default": False, + }, + ], + "order_by": "red_count", + "descending": True, + "limit": 100, +} + +EXECUTION_BACKEND = "minute_filter" +ENTRY_SIGNALS: list[str] = [] +EXIT_SIGNALS: list[str] = [] + + +def filter_minute_history(df: pl.DataFrame, params: dict) -> pl.DataFrame: + """红K形态过滤: 全向量化, 无逐行 Python 循环。 + + - 每标的按时间取最近 bars 根; 不足 bars 根不触发 + - 红 = close > open; 窗口内红K数 >= min_red + - 按 rank_by (high / close) 降序取前 top_red 根, 同值取时间更晚者, 需全红 + """ + bars = int(params.get("bars") or 7) + min_red = min(int(params.get("min_red") or 5), bars) + top_red = min(int(params.get("top_red") or 2), bars) + rank_by = "close" if params.get("rank_by_close") else "high" + if rank_by not in df.columns: + rank_by = "high" + + tailed = ( + df.sort(["symbol", "datetime"]) + .filter(pl.int_range(pl.len()).over("symbol") >= pl.len().over("symbol") - bars) + .with_columns(_red=(pl.col("close") > pl.col("open")).cast(pl.Int32)) + ) + + window = tailed.group_by("symbol").agg( + bars_checked=pl.len(), + red_count=pl.col("_red").sum(), + last_datetime=pl.col("datetime").max(), + # 输出列名用 close: 基础过滤的股价区间直接作用于最新分钟价 + close=pl.col("close").sort_by("datetime").last(), + window_high=pl.col("high").max(), + window_low=pl.col("low").min(), + window_volume=pl.col("volume").sum(), + window_amount=pl.col("amount").sum(), + ) + + top = ( + tailed.sort([rank_by, "datetime"], descending=[True, True]) + .filter(pl.int_range(pl.len()).over("symbol") < top_red) + .group_by("symbol") + .agg(top_red_count=pl.col("_red").sum()) + ) + + return ( + window.join(top, on="symbol", how="inner") + .filter( + (pl.col("bars_checked") >= bars) + & (pl.col("red_count") >= min_red) + & (pl.col("top_red_count") >= top_red) + ) + .drop("bars_checked") + ) diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index aa7e581..42dd140 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -199,6 +199,8 @@ class StrategyDef: execution_backend: str = "polars_expr" matrix_strategy: Any | None = None composite: CompositeSpec | None = None # 仅 backend=="composite" 时非空 + # 仅 backend=="minute_filter" 时非空: 输入为当日分钟K窗口, 输出为命中标的行 + filter_minute_history_fn: Callable[[pl.DataFrame, dict], pl.DataFrame] | None = None @dataclass @@ -471,6 +473,7 @@ class StrategyEngine: filter_fn = getattr(mod, "filter", None) filter_history_fn = getattr(mod, "filter_history", None) + filter_minute_history_fn = getattr(mod, "filter_minute_history", None) execution_backend = str( getattr( mod, @@ -481,7 +484,7 @@ class StrategyEngine: ), ) ) - valid_backends = {"polars_expr", "matrix_native", "python_history_legacy", "composite"} + valid_backends = {"polars_expr", "matrix_native", "python_history_legacy", "composite", "minute_filter"} if execution_backend not in valid_backends: raise ValueError( f"unsupported execution backend {execution_backend!r}; " @@ -515,6 +518,23 @@ class StrategyEngine: "composite strategy must not declare filter, filter_history or MATRIX_STRATEGY" ) composite_spec = _parse_composite_children(meta.get("children")) + elif execution_backend == "minute_filter": + # 分钟形态策略: 只声明 filter_minute_history; 数据源是本地当日分钟K分区 + # (由 ScreenerService.build_strategy_context 的 1m 分支注入), 因此 timeframes + # 必须且只能是 ["1m"] — 混入 1d 会让日线 context 走错数据路径。 + if ( + filter_minute_history_fn is None + or filter_fn is not None + or filter_history_fn is not None + or matrix_strategy is not None + ): + raise ValueError( + "minute_filter strategy must declare only filter_minute_history" + ) + if meta.get("timeframes") != ["1m"]: + raise ValueError( + "minute_filter strategy must declare timeframes == ['1m']" + ) elif filter_history_fn is None or filter_fn is not None: raise ValueError("python_history_legacy strategy must declare only filter_history") @@ -538,6 +558,7 @@ class StrategyEngine: execution_backend=execution_backend, matrix_strategy=matrix_strategy, composite=composite_spec, + filter_minute_history_fn=filter_minute_history_fn, ) def reload(self) -> None: @@ -885,7 +906,23 @@ class StrategyEngine: exit_signal_hits = self._collect_signal_hits(signal_df, exit_signals) # 普通策略只读目标日期；历史策略读取调用方注入的历史窗口。 - if s.filter_history_fn: + if s.execution_backend == "minute_filter": + # 分钟策略: 读取调用方注入的当日分钟K窗口。无 date 列, 不按 as_of 过滤, + # 每个命中行自带最后K线时间戳 (last_datetime)。 + if history is None: + raise ValueError(f"strategy {strategy_id} requires minute history data") + if history.is_empty(): + return StrategyResult( + as_of=as_of, + strategy_id=strategy_id, + exit_signal_hits=exit_signal_hits, + ) + df = s.filter_minute_history_fn(history, params) + # 基础过滤/展示列 (name/total_shares/change_pct 等) 来自 enriched 快照, + # 在命中结果上事后联表, 避免把 enriched 列铺到全市场分钟行上。 + if current is not None and not current.is_empty(): + df = self._join_basic_columns(df, current) + elif s.filter_history_fn: if history is None: raise ValueError(f"strategy {strategy_id} requires history data") df = history @@ -945,7 +982,9 @@ class StrategyEngine: # Stage 3: 评分 df = self._apply_scoring(df, scoring, scoring_directions) entry_signal_hits = self._collect_signal_hits(df, entry_signals) - if not entry_signals and (s.filter_history_fn or s.filter_fn): + if not entry_signals and ( + s.filter_history_fn or s.filter_fn or s.execution_backend == "minute_filter" + ): entry_signal_hits = [ {"symbol": str(symbol), "signals": []} for symbol in df["symbol"].cast(pl.Utf8).unique().to_list() @@ -1036,7 +1075,8 @@ class StrategyEngine: history_strats = [ (sid, strategy) for sid, strategy in selected - if strategy.filter_history_fn or strategy.execution_backend == "matrix_native" + if strategy.filter_history_fn + or strategy.execution_backend in ("matrix_native", "minute_filter") ] shared_history = context.history if history_strats and shared_history is None: @@ -1466,6 +1506,25 @@ class StrategyEngine: return df.filter(expr) return df + # 分钟策略命中行需要从事后联表补齐的 enriched 列: 基础过滤引用 + 前端展示。 + # close 不在列 — 分钟策略输出的 close 是最后一根分钟K收盘价, 优先于日线快照。 + MINUTE_JOIN_COLUMNS: tuple[str, ...] = ( + "name", "total_shares", "float_shares", "amount", + "turnover_rate", "change_pct", "pre_close", + ) + + @staticmethod + def _join_basic_columns(df: pl.DataFrame, current: pl.DataFrame) -> pl.DataFrame: + """把 enriched 快照列按 symbol 联到分钟策略输出上, 只补 df 缺失的列。""" + cols = [ + c for c in StrategyEngine.MINUTE_JOIN_COLUMNS + if c in current.columns and c not in df.columns + ] + if not cols: + return df + extra = current.select(["symbol", *cols]).unique(subset=["symbol"], keep="last") + return df.join(extra, on="symbol", how="left") + # ================================================================ # 内部: 评分 # ================================================================ diff --git a/backend/tests/backtest/test_matrix_strategy.py b/backend/tests/backtest/test_matrix_strategy.py index c1db050..0f8defd 100644 --- a/backend/tests/backtest/test_matrix_strategy.py +++ b/backend/tests/backtest/test_matrix_strategy.py @@ -319,9 +319,13 @@ def test_builtin_matrix_strategies_use_their_declared_formula_modules(): path for path in strategy_dir.glob("*.py") if path.name != "__init__.py" ) - assert len(strategy_files) == 19 + # 非 matrix 后端的内置策略白名单 (当前仅分钟形态策略) + non_matrix = {"minute_red_streak"} + assert len(strategy_files) == 19 + len(non_matrix) for strategy_path in strategy_files: strategy = StrategyEngine._load_file(strategy_path) + if strategy_path.stem in non_matrix: + continue assert strategy.execution_backend == "matrix_native" assert strategy.matrix_strategy is not None assert strategy.matrix_strategy.__class__.__module__ == strategy_path.stem @@ -784,7 +788,10 @@ def test_registered_builtin_matrix_strategies_share_one_cache_profile(): strategy_dirs=[REPO_ROOT / "backend" / "app" / "strategy" / "builtin"] ) profile = build_matrix_cache_profile(engine, "stock") - strategies = engine.strategy_definitions() + strategies = tuple( + s for s in engine.strategy_definitions() + if s.execution_backend != "minute_filter" + ) assert len(strategies) == 19 assert all(strategy.execution_backend == "matrix_native" for strategy in strategies) diff --git a/backend/tests/test_minute_refresh.py b/backend/tests/test_minute_refresh.py new file mode 100644 index 0000000..21e5cd1 --- /dev/null +++ b/backend/tests/test_minute_refresh.py @@ -0,0 +1,198 @@ +"""盘中分钟增量刷新服务 (minute_refresh) 测试。 + +覆盖: +- 连续竞价时段判定 (含边界) +- 门控链: 开关关闭 / 自定义分钟源让位 / 能力缺失 / 时段外 / 放行 +- 单轮: mock 边界层脉冲 + 落盘, 校验状态字段与 universe 来源 +- 偏好读写: 默认关闭、间隔 clamp [60, 300] +- API: /minute-refresh/status 无服务时 available=false + +不发起真实网络请求: fetch_intraday_full_market_burst 与 _write_minute_partition +均 monkeypatch 替换。 +""" +from __future__ import annotations + +from datetime import datetime + +import polars as pl + +from app.services import minute_refresh, preferences +from app.services.minute_refresh import MinuteRefreshService, _in_continuous_session + + +def _isolated_prefs(tmp_path, monkeypatch): + path = tmp_path / "preferences.json" + monkeypatch.setattr(preferences, "_path", lambda: path) + preferences._invalidate_cache() + return path + + +class _FakeCapSet: + def __init__(self, has_intraday_batch: bool): + self._has = has_intraday_batch + + def has(self, cap) -> bool: + from app.tickflow.capabilities import Cap + + return self._has and cap == Cap.INTRADAY_BATCH + + +class _FakeAppState: + def __init__(self, has_intraday_batch: bool): + self.capabilities = _FakeCapSet(has_intraday_batch) + + +class _FakeRepo: + def __init__(self, symbols: list[str]): + from pathlib import Path + self._inst = pl.DataFrame({"symbol": symbols}) + self.store = type("S", (), {"data_dir": Path(".")})() + + def get_instruments(self) -> pl.DataFrame: + return self._inst + + +# ── 时段判定 ──────────────────────────────────────────────────────── + + +def test_continuous_session_boundaries(): + wk = datetime(2026, 8, 25, 10, 0) # 周二 + assert _in_continuous_session(wk) + assert not _in_continuous_session(datetime(2026, 8, 25, 9, 29)) + assert not _in_continuous_session(datetime(2026, 8, 25, 11, 31)) # 午休 + assert _in_continuous_session(datetime(2026, 8, 25, 13, 0)) # 午后恢复 + assert _in_continuous_session(datetime(2026, 8, 25, 15, 0)) # 收盘瞬时 + assert not _in_continuous_session(datetime(2026, 8, 25, 15, 1)) + assert not _in_continuous_session(datetime(2026, 8, 22, 10, 0)) # 周六 + + +# ── 门控链 ────────────────────────────────────────────────────────── + + +def _svc(tmp_path, monkeypatch, *, enabled=True, custom_provider=False, capability=True, in_hours=True): + _isolated_prefs(tmp_path, monkeypatch) + preferences.save({"minute_refresh_enabled": enabled}) + if custom_provider: + # 模拟已注册的自定义分钟源 (真实注册表在测试环境未加载) + monkeypatch.setattr(preferences, "get_minute_data_provider", lambda: "a-stock-data") + svc = MinuteRefreshService(_FakeRepo(["600000.SH"])) + svc.set_app_state(_FakeAppState(capability)) + monkeypatch.setattr(minute_refresh, "_in_continuous_session", lambda now=None: in_hours) + return svc + + +def test_gate_disabled(tmp_path, monkeypatch): + assert _svc(tmp_path, monkeypatch, enabled=False)._gate_reason() == "disabled" + + +def test_gate_custom_provider_yields(tmp_path, monkeypatch): + svc = _svc(tmp_path, monkeypatch, custom_provider=True) + assert svc._gate_reason() == "custom_minute_provider" + + +def test_gate_capability_missing(tmp_path, monkeypatch): + svc = _svc(tmp_path, monkeypatch, capability=False) + assert svc._gate_reason() == "capability" + + +def test_gate_outside_trading_hours(tmp_path, monkeypatch): + svc = _svc(tmp_path, monkeypatch, in_hours=False) + assert svc._gate_reason() == "outside_trading_hours" + + +def test_gate_pass(tmp_path, monkeypatch): + svc = _svc(tmp_path, monkeypatch) + assert svc._gate_reason() is None + assert svc.capability_ok() and not svc.custom_provider_active() + + +# ── 单轮 ──────────────────────────────────────────────────────────── + + +def test_run_round_writes_partition_and_updates_status(tmp_path, monkeypatch): + svc = _svc(tmp_path, monkeypatch) + minute_df = pl.DataFrame({ + "symbol": ["600000.SH"], + "datetime": [datetime(2026, 8, 25, 1, 30)], + "open": [10.0], "high": [10.5], "low": [9.9], "close": [10.2], + "volume": [1000.0], "amount": [10200.0], + }) + calls: dict = {} + + def fake_burst(symbols, capset, *, count=300): + calls["symbols"] = list(symbols) + return (minute_df, 1) + + def fake_write(df, minute_dir): + calls["dir"] = minute_dir + calls["rows"] = df.height + return df.height + + monkeypatch.setattr( + "app.services.kline_sync.fetch_intraday_full_market_burst", fake_burst + ) + monkeypatch.setattr("app.services.kline_sync._write_minute_partition", fake_write) + + svc._run_round() + + assert calls["symbols"] == ["600000.SH"] + assert calls["rows"] == 1 + st = svc.status() + assert st["rounds"] == 1 + assert st["last_rows"] == 1 + assert st["last_symbols"] == 1 + assert st["last_requests"] == 1 + assert st["last_round_at"] is not None + assert st["last_error"] is None + assert st["capability_ok"] is True + + +def test_run_round_records_error_when_burst_empty(tmp_path, monkeypatch): + svc = _svc(tmp_path, monkeypatch) + monkeypatch.setattr( + "app.services.kline_sync.fetch_intraday_full_market_burst", + lambda symbols, capset, *, count=300: (pl.DataFrame(), 3), + ) + svc._run_round() + st = svc.status() + assert st["rounds"] == 0 + assert "no data" in st["last_error"] + assert st["last_requests"] == 3 + + +def test_status_reports_gate_reason_when_stopped(tmp_path, monkeypatch): + svc = _svc(tmp_path, monkeypatch, enabled=False) + st = svc.status() + assert st["enabled"] is False + assert st["running"] is False + assert st["gate_reason"] == "disabled" + assert st["interval_seconds"] == 60 + + +# ── 偏好 ──────────────────────────────────────────────────────────── + + +def test_refresh_preferences_defaults_and_clamp(tmp_path, monkeypatch): + _isolated_prefs(tmp_path, monkeypatch) + assert preferences.get_minute_refresh_enabled() is False + assert preferences.get_minute_refresh_interval() == 60 + preferences.save({"minute_refresh_interval": 5}) + assert preferences.get_minute_refresh_interval() == 60 # 下限 + preferences.save({"minute_refresh_interval": 999}) + assert preferences.get_minute_refresh_interval() == 300 # 上限 + preferences.save({"minute_refresh_interval": 90}) + assert preferences.get_minute_refresh_interval() == 90 + + +def test_status_endpoint_without_service(): + from fastapi import FastAPI + from fastapi.testclient import TestClient + + from app.api.settings import router + + app = FastAPI() + app.include_router(router) + client = TestClient(app) + resp = client.get("/api/settings/minute-refresh/status") + assert resp.status_code == 200 + assert resp.json() == {"available": False} diff --git a/backend/tests/test_minute_strategy.py b/backend/tests/test_minute_strategy.py new file mode 100644 index 0000000..ca4e209 --- /dev/null +++ b/backend/tests/test_minute_strategy.py @@ -0,0 +1,308 @@ +"""分钟策略 (minute_filter 后端) 测试。 + +覆盖: +- minute_red_streak 形态: 命中 / 不足根数不触发 / 最高K不红 / rank_by 两口径 / + 乱序输入 / 最高价并列取更晚K线 +- 引擎加载校验: 只能声明 filter_minute_history、timeframes 必须且只能是 ["1m"] +- 引擎 1m 运行: enriched 联表基础过滤 (剔除ST / 股价区间)、entry hits、 + 日线 context 拒绝 +- ScreenerService 1m context: 当日分区优先、缺失回退最近分区、空库报错、 + 非股票资产拒绝 +""" +from __future__ import annotations + +import datetime as _dt +from datetime import date, datetime +from pathlib import Path + +import polars as pl + +from app.services.screener import ScreenerService +from app.strategy.builtin import minute_red_streak +from app.strategy.engine import StrategyDataContext, StrategyEngine + + +def _bars(symbol: str, candles: list[tuple[float, float, float]], start_hour: int = 9) -> pl.DataFrame: + """candles: (open, close, high) 序列, 时间从 start_hour:30 起每分钟一根。""" + n = len(candles) + base = datetime(2026, 8, 25, start_hour, 30) + return pl.DataFrame({ + "symbol": [symbol] * n, + "datetime": [base + _dt.timedelta(minutes=i) for i in range(n)], + "open": [float(c[0]) for c in candles], + "high": [float(c[2]) for c in candles], + "low": [float(min(c[0], c[1])) for c in candles], + "close": [float(c[1]) for c in candles], + "volume": [100.0] * n, + "amount": [10000.0] * n, + }) + + +# ── 形态 ──────────────────────────────────────────────────────────── + + +def test_pattern_hits_five_red_of_seven_with_red_top_two(): + # 7根: 5红2绿, 绿K的最高价都压得比红K低 → 最高的两根(10.9/10.7)都是红 + candles = [ + (10.0, 10.2, 10.30), # 红 + (10.2, 10.1, 10.25), # 绿 (低高点) + (10.1, 10.4, 10.50), # 红 + (10.4, 10.6, 10.70), # 红 (次高) + (10.6, 10.5, 10.65), # 绿 (低高点) + (10.5, 10.7, 10.80), # 红 + (10.7, 10.8, 10.90), # 红 (最高) + ] + out = minute_red_streak.filter_minute_history(_bars("600000.SH", candles), {}) + assert out["symbol"].to_list() == ["600000.SH"] + row = out.row(0, named=True) + assert row["red_count"] == 5 + assert row["top_red_count"] == 2 + assert row["close"] == 10.8 + + +def test_pattern_insufficient_bars_never_triggers(): + out = minute_red_streak.filter_minute_history(_bars("600000.SH", [(10.0, 10.2, 10.3)] * 6), {}) + assert out.is_empty() + + +def test_pattern_green_at_top_blocks_hit(): + # 5红, 但最高的一根是绿 (高开回落) → 最高两根不全红, 不触发 + candles = [ + (10.0, 10.2, 10.30), # 红 + (10.1, 10.4, 10.50), # 红 + (10.3, 10.6, 10.70), # 红 + (10.6, 10.5, 10.65), # 绿 (低高点) + (10.4, 10.5, 10.55), # 红 (低高点) + (11.5, 11.0, 12.00), # 绿 (最高) + (11.0, 11.4, 11.90), # 红 (次高) + ] + out = minute_red_streak.filter_minute_history(_bars("600000.SH", candles), {}) + assert out.is_empty() + + +def test_pattern_rank_by_close_uses_close_not_high(): + # high 口径最高两根是绿K冲高; close 口径最高两根是红K → 仅 close 口径命中 + candles = [ + (10.0, 10.5, 10.60), # 红 + (10.5, 10.9, 11.50), # 绿 (high 最高, 并列) + (10.9, 11.2, 11.40), # 红 + (11.2, 11.3, 11.35), # 红 + (11.3, 11.4, 11.45), # 红 (close 次高) + (11.4, 11.1, 11.50), # 绿 (high 最高, 并列) + (11.1, 11.5, 11.55), # 红 (close 最高) + ] + by_high = minute_red_streak.filter_minute_history(_bars("600000.SH", candles), {}) + by_close = minute_red_streak.filter_minute_history( + _bars("600000.SH", candles), {"rank_by_close": True} + ) + assert by_high.is_empty() + assert by_close["symbol"].to_list() == ["600000.SH"] + + +def test_pattern_sorts_unordered_input_by_datetime(): + bars = pl.concat([ + _bars("600000.SH", [(10.0, 10.2, 10.30)]), + _bars("600000.SH", [ + (10.2, 10.1, 10.25), (10.1, 10.4, 10.50), (10.4, 10.6, 10.70), + (10.6, 10.5, 10.65), (10.5, 10.7, 10.80), (10.7, 10.8, 10.90), + ]), + ]).sample(fraction=1.0, shuffle=True, seed=7) + out = minute_red_streak.filter_minute_history(bars, {}) + assert out["symbol"].to_list() == ["600000.SH"] + assert out.row(0, named=True)["close"] == 10.8 # 最后一根(时间最大)的收盘 + + +def test_pattern_three_way_high_tie_prefers_later_bars(): + # 三根 high 并列最高: 更早的绿K应被更晚的两根红K挤出 top2 → 命中 + # (若并列取更早, top2 = {红, 绿} → 不命中; 该测试固定 "同值取更晚" 契约) + candles = [ + (10.0, 10.2, 10.30), # 红 + (10.1, 10.4, 10.50), # 红 + (10.2, 10.1, 10.25), # 绿 (低高点) + (10.3, 10.6, 10.70), # 红 + (10.8, 10.5, 10.90), # 绿 (并列最高, 最早 → 被 top2 排除) + (10.5, 10.6, 10.90), # 红 (并列最高, 中间) + (10.6, 10.8, 10.90), # 红 (并列最高, 最晚) + ] + out = minute_red_streak.filter_minute_history(_bars("600000.SH", candles), {}) + assert out["symbol"].to_list() == ["600000.SH"] + assert out.row(0, named=True)["top_red_count"] == 2 + + +def test_pattern_min_red_threshold_respected(): + # 4红3绿, 最高的两根红 → min_red=5 不命中, min_red=4 命中 + candles = [ + (10.0, 10.2, 10.30), # 红 + (10.2, 10.1, 10.25), # 绿 + (10.1, 10.4, 10.50), # 红 + (10.4, 10.3, 10.45), # 绿 + (10.3, 10.6, 10.70), # 红 + (10.6, 10.5, 10.65), # 绿 + (10.5, 10.8, 10.90), # 红 + ] + bars = _bars("600000.SH", candles) + assert minute_red_streak.filter_minute_history(bars, {"min_red": 5}).is_empty() + assert not minute_red_streak.filter_minute_history(bars, {"min_red": 4}).is_empty() + + +# ── 引擎加载与运行 ────────────────────────────────────────────────── + + +def test_builtin_minute_strategy_loads_with_minute_filter_backend(): + engine = StrategyEngine( + strategy_dirs=[Path(__file__).resolve().parent.parent / "app" / "strategy" / "builtin"] + ) + assert not [e for e in engine.load_errors() if "minute" in e["file"]] + s = engine.get("minute_red_streak") + assert s.execution_backend == "minute_filter" + assert s.filter_minute_history_fn is not None + assert s.meta["timeframes"] == ["1m"] + + +def _minute_code(sid: str, timeframes: str = '["1m"]', extra: str = "") -> str: + return f'''import polars as pl +META = {{"id": "{sid}", "name": "{sid}", "asset_types": ["stock"], "timeframes": {timeframes}}} +EXECUTION_BACKEND = "minute_filter" +{extra} +def filter_minute_history(df, params): + return df.group_by("symbol").agg( + close=pl.col("close").max(), last_datetime=pl.col("datetime").max() + ) +''' + + +def test_minute_filter_backend_validation(tmp_path): + (tmp_path / "ok.py").write_text(_minute_code("m_ok")) + (tmp_path / "bad_filter.py").write_text( + _minute_code("m_bad1", extra="def filter(df, params):\n return pl.lit(True)") + ) + (tmp_path / "bad_tf.py").write_text(_minute_code("m_bad2", timeframes='["1d", "1m"]')) + engine = StrategyEngine(strategy_dirs=[tmp_path]) + ids = {m["id"] for m in engine.list_strategies(include_research=True)} + assert "m_ok" in ids + assert "m_bad1" not in ids + assert "m_bad2" not in ids + assert any("only filter_minute_history" in e["error"] for e in engine.load_errors()) + assert any("timeframes" in e["error"] for e in engine.load_errors()) + + +def test_minute_context_run_applies_enriched_basic_filter(tmp_path): + (tmp_path / "m_basic.py").write_text(_minute_code("m_basic")) + engine = StrategyEngine(strategy_dirs=[tmp_path]) + + hist = pl.concat([ + _bars("600001.SH", [(10.0, 20.0, 25.0)] * 7), # 命中, 收盘 20 + _bars("600002.SH", [(10.0, 20.0, 25.0)] * 7), # 命中但 ST → 剔除 + _bars("600003.SH", [(10.0, 20.0, 25.0)] * 7), # 命中 + _bars("600004.SH", [(100.0, 200.0, 250.0)] * 7), # 命中但收盘 200 → 超上限剔除 + ]) + current = pl.DataFrame({ + "symbol": ["600001.SH", "600002.SH", "600003.SH", "600004.SH"], + "name": ["正常股", "ST垃圾", "正常股2", "高价股"], + "total_shares": [1e8, 1e8, 1e8, 1e8], + "float_shares": [5e7, 5e7, 5e7, 5e7], + "amount": [3e8, 3e8, 3e8, 3e8], + "change_pct": [0.01, 0.01, 0.01, 0.01], + }) + context = StrategyDataContext( + asset_type="stock", + timeframe="1m", + as_of=date(2026, 8, 25), + current=current, + history=hist, + ) + result = engine.run( + "m_basic", context, overrides={"basic_filter": {"price_max": 150.0}} + ) + symbols = {r["symbol"] for r in result.rows} + assert symbols == {"600001.SH", "600003.SH"} + assert all("name" in r for r in result.rows) # enriched 列已联表 + assert {h["symbol"] for h in result.entry_signal_hits} == symbols + + +def test_minute_strategy_rejects_daily_context(tmp_path): + (tmp_path / "m_daily.py").write_text(_minute_code("m_daily")) + engine = StrategyEngine(strategy_dirs=[tmp_path]) + context = StrategyDataContext( + asset_type="stock", + timeframe="1d", + as_of=date(2026, 8, 25), + current=pl.DataFrame({"symbol": ["600001.SH"]}), + ) + try: + engine.run("m_daily", context) + raise AssertionError("expected ValueError") + except ValueError as e: + assert "timeframe" in str(e) + + +# ── ScreenerService 1m context ────────────────────────────────────── + + +class _FakeMinuteRepo: + def __init__(self, partitions: dict[date, pl.DataFrame]): + self.partitions = partitions + + def get_minute_by_dates(self, symbols, dates, asset_type="stock"): + frames = [self.partitions[d] for d in dates if d in self.partitions] + if not frames: + return pl.DataFrame() + return pl.concat(frames).filter(pl.col("symbol").is_in(symbols)) + + def latest_minute_date_global(self): + return max(self.partitions) if self.partitions else None + + +def _svc(partitions: dict[date, pl.DataFrame], asset_type: str = "stock") -> ScreenerService: + return ScreenerService(_FakeMinuteRepo(partitions), asset_type=asset_type) # type: ignore[arg-type] + + +def test_minute_context_prefers_as_of_partition(): + d1, d2 = date(2026, 8, 24), date(2026, 8, 25) + svc = _svc({ + d1: _bars("600001.SH", [(10.0, 10.2, 10.3)] * 3), + d2: _bars("600001.SH", [(10.0, 10.2, 10.3)] * 4), + }) + ctx = svc.build_strategy_context( + None, d1, [], timeframe="1m", + current=pl.DataFrame({"symbol": ["600001.SH"], "name": ["x"]}), + ) + assert ctx.history.height == 3 # as_of 当日分区, 不取更新的 d2 + assert ctx.timeframe == "1m" + + +def test_minute_context_falls_back_to_latest_partition(): + d1, d2 = date(2026, 8, 24), date(2026, 8, 25) + svc = _svc({ + d1: _bars("600001.SH", [(10.0, 10.2, 10.3)] * 3), + d2: _bars("600001.SH", [(10.0, 10.2, 10.3)] * 4), + }) + ctx = svc.build_strategy_context( + None, date(2026, 8, 20), [], timeframe="1m", + current=pl.DataFrame({"symbol": ["600001.SH"]}), + ) + assert ctx.history.height == 4 # 回退到最近分区 d2 + + +def test_minute_context_empty_store_raises_with_guidance(): + svc = _svc({}) + try: + svc.build_strategy_context( + None, date(2026, 8, 25), [], timeframe="1m", + current=pl.DataFrame({"symbol": ["600001.SH"]}), + ) + raise AssertionError("expected ValueError") + except ValueError as e: + assert "分钟K" in str(e) + + +def test_minute_context_rejects_non_stock_asset(): + svc = _svc({date(2026, 8, 25): _bars("510300.SH", [(10.0, 10.2, 10.3)] * 3)}, asset_type="etf") + try: + svc.build_strategy_context( + None, date(2026, 8, 25), [], timeframe="1m", + current=pl.DataFrame({"symbol": ["510300.SH"]}), + ) + raise AssertionError("expected ValueError") + except ValueError as e: + assert "A 股" in str(e) diff --git a/backend/tests/test_screener_etf.py b/backend/tests/test_screener_etf.py index c201f40..32c4de7 100644 --- a/backend/tests/test_screener_etf.py +++ b/backend/tests/test_screener_etf.py @@ -41,18 +41,23 @@ def test_all_builtin_strategies_declare_asset_types_and_timeframes(): assert engine.load_errors() == [] for meta in engine.list_strategies(): assert meta["asset_types"] - assert meta["timeframes"] == ["1d"] + # 分钟策略 timeframes 为 ["1m"], 日线内置策略为 ["1d"] + assert meta["timeframes"] in (["1d"], ["1m"]) def test_all_builtin_strategies_use_matrix_backend_only(): engine = _engine() assert engine.load_errors() == [] strategies = [engine.get(meta["id"]) for meta in engine.list_strategies()] - assert len(strategies) == 18 - assert all(strategy.execution_backend == "matrix_native" for strategy in strategies) - assert all(strategy.matrix_strategy is not None for strategy in strategies) - assert all(strategy.filter_fn is None for strategy in strategies) - assert all(strategy.filter_history_fn is None for strategy in strategies) + matrix_strategies = [s for s in strategies if s.execution_backend == "matrix_native"] + assert len(matrix_strategies) == 18 + assert all(s.matrix_strategy is not None for s in matrix_strategies) + assert all(s.filter_fn is None for s in matrix_strategies) + assert all(s.filter_history_fn is None for s in matrix_strategies) + # 分钟形态策略 (minute_filter) 不参与日线矩阵不变量 + assert [s.meta["id"] for s in strategies if s.execution_backend == "minute_filter"] == [ + "minute_red_streak" + ] def test_all_builtin_matrix_formulas_accept_base_market_matrix(): @@ -79,10 +84,14 @@ def test_all_builtin_matrix_formulas_accept_base_market_matrix(): from app.backtest.matrix import build_market_data_matrix fields = set() - for strategy in (engine.get(meta["id"]) for meta in engine.list_strategies()): + matrix_metas = [ + m for m in engine.list_strategies() + if engine.get(m["id"]).execution_backend == "matrix_native" + ] + for strategy in (engine.get(meta["id"]) for meta in matrix_metas): fields.update(engine._matrix_field_columns(strategy)) market = build_market_data_matrix(panel, field_columns=fields) - for meta in engine.list_strategies(): + for meta in matrix_metas: strategy = engine.get(meta["id"]) signals = strategy.matrix_strategy.compute_signals(market, {}) assert signals.shape == market.shape, meta["id"] diff --git a/frontend/src/components/data/MinuteSyncConfig.tsx b/frontend/src/components/data/MinuteSyncConfig.tsx index e104866..86ce910 100644 --- a/frontend/src/components/data/MinuteSyncConfig.tsx +++ b/frontend/src/components/data/MinuteSyncConfig.tsx @@ -1,6 +1,6 @@ import { useState, useEffect } from 'react' import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' -import { Loader2, Trash2, Download, Calendar } from 'lucide-react' +import { Loader2, Trash2, Download, Calendar, Zap } from 'lucide-react' import { api } from '@/lib/api' import { QK } from '@/lib/queryKeys' import { MissingCapChip } from '@/lib/capability-labels' @@ -12,17 +12,59 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string; queryFn: api.preferences, }) const update = useMutation({ - mutationFn: ({ enabled, days, segmentDays }: { enabled: boolean; days: number; segmentDays?: number }) => - api.updateMinuteSync(enabled, days, segmentDays), - onSuccess: () => qc.invalidateQueries({ queryKey: QK.preferences }), + mutationFn: ({ enabled, days, segmentDays, refresh }: { + enabled: boolean; days: number; segmentDays?: number + refresh?: { enabled?: boolean; interval?: number } + }) => + api.updateMinuteSync(enabled, days, segmentDays, refresh), + onSuccess: () => { + qc.invalidateQueries({ queryKey: QK.preferences }) + qc.invalidateQueries({ queryKey: ['minute-refresh-status'] }) + }, + }) + + // 盘中增量刷新状态 (轮询 15s, 仅弹窗挂载期间) + const refreshStatus = useQuery({ + queryKey: ['minute-refresh-status'], + queryFn: api.minuteRefreshStatus, + refetchInterval: 15000, }) const hasMinuteCap = !!caps?.capabilities?.['kline.minute.batch'] const enabled = prefs.data?.minute_sync_enabled ?? false const days = prefs.data?.minute_sync_days ?? 5 const segmentDays = prefs.data?.minute_sync_segment_days ?? 20 + const refreshEnabled = prefs.data?.minute_refresh_enabled ?? false + const refreshInterval = prefs.data?.minute_refresh_interval ?? 60 const [localDays, setLocalDays] = useState(days) const [localSegment, setLocalSegment] = useState(segmentDays) + const [localRefreshInterval, setLocalRefreshInterval] = useState(refreshInterval) + + useEffect(() => { setLocalDays(days) }, [days]) + useEffect(() => { setLocalSegment(segmentDays) }, [segmentDays]) + useEffect(() => { setLocalRefreshInterval(refreshInterval) }, [refreshInterval]) + + // 盘中增量 = intraday.batch 独立能力 (Expert 专有), 与盘后同步的 minute.batch 分属不同限流池 + const hasIntradayBatchCap = !!caps?.capabilities?.['intraday.batch'] + const rs = refreshStatus.data + const refreshGateText = rs?.custom_provider_active + ? '已配置自定义分钟源, 盘中增量由插件自管' + : rs && rs.available && !rs.capability_ok + ? '需要日内分时批量能力 (Expert)' + : !rs?.in_trading_hours ? '非连续竞价时段, 暂停中' + : rs?.last_error ? `最近错误: ${rs.last_error}` + : null + + const handleRefreshToggle = () => { + if (!hasIntradayBatchCap) return + update.mutate({ enabled, days: localDays, refresh: { enabled: !refreshEnabled } }) + } + + const setRefreshInterval = (v: number) => { + const clamped = Math.max(60, Math.min(300, Math.round(v / 30) * 30)) + setLocalRefreshInterval(clamped) + update.mutate({ enabled, days: localDays, refresh: { interval: clamped } }) + } useEffect(() => { setLocalDays(days) }, [days]) useEffect(() => { setLocalSegment(segmentDays) }, [segmentDays]) @@ -147,6 +189,67 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string; + {/* 区块 A2: 盘中增量刷新 (Expert 专有, intraday.batch 独立限流池) */} +
+
+
+ +
+ + + 盘中增量刷新{refreshEnabled ? '已开启' : '已关闭'} + +
+
+
+
+ +
+ {Math.round(localRefreshInterval / 60) >= 1 && localRefreshInterval % 60 === 0 ? `${localRefreshInterval / 60}m` : `${localRefreshInterval}s`} +
+ +
+ {!hasIntradayBatchCap && } +
+
+
+ 交易时段内用日内分时批量 (独立配额) 每 {Math.round(localRefreshInterval / 60) >= 1 && localRefreshInterval % 60 === 0 ? `${localRefreshInterval / 60} 分钟` : `${localRefreshInterval} 秒`} 全市场脉冲落盘一次, + 分钟策略读到最新K线; 不占用盘后分钟同步的限流配额。 +
+ {/* 运行状态一行: 门控原因 / 下一轮 / 最近一轮 */} + {rs?.available && ( +
+ + ● {rs.running ? '服务运行中' : '服务未运行'} + + {refreshGateText && {refreshGateText}} + {rs.rounds != null && rs.rounds > 0 && ( + 已 {rs.rounds} 轮 · 最近 {rs.last_symbols} 标的 / {rs.last_rows} 行 / {rs.last_requests} 请求{rs.last_round_ms != null ? ` · ${(rs.last_round_ms / 1000).toFixed(1)}s` : ''} + )} +
+ )} +
+ {/* 区块 B: 手动获取 (一次性操作, 独立于上方自动同步开关) */}
diff --git a/frontend/src/components/screener/StrategyCard.tsx b/frontend/src/components/screener/StrategyCard.tsx index f2efbff..7281acb 100644 --- a/frontend/src/components/screener/StrategyCard.tsx +++ b/frontend/src/components/screener/StrategyCard.tsx @@ -92,12 +92,14 @@ interface StrategyCardProps { monitored?: boolean /** 切换策略监控 (点击 RadioTower 图标) */ onToggleMonitor?: () => void + /** 周期徽章 (如 '分钟'); 日线策略不传 */ + timeframeBadge?: string } export function StrategyCard({ name, description, source, active, count, expiredCount, loading, cardSize, - onRun, disabled, onSettings, monitored, onToggleMonitor, + onRun, disabled, onSettings, monitored, onToggleMonitor, timeframeBadge, }: StrategyCardProps) { const cs = CARD_STYLES[cardSize] const activeCls = active @@ -125,6 +127,9 @@ export function StrategyCard({ className="flex flex-col items-start cursor-pointer disabled:opacity-50 disabled:cursor-wait w-full">
{srcLabel} + {timeframeBadge && ( + {timeframeBadge} + )} {name}
{description && ( @@ -164,6 +169,9 @@ export function StrategyCard({ className="flex flex-col items-start cursor-pointer disabled:opacity-50 disabled:cursor-wait min-w-0">
{srcLabel} + {timeframeBadge && ( + {timeframeBadge} + )} {name} {count != null && !loading && ( {count} diff --git a/frontend/src/components/screener/StrategyPoolDialog.tsx b/frontend/src/components/screener/StrategyPoolDialog.tsx index d6dcc21..8f41e09 100644 --- a/frontend/src/components/screener/StrategyPoolDialog.tsx +++ b/frontend/src/components/screener/StrategyPoolDialog.tsx @@ -8,6 +8,8 @@ interface Props { pool: string[] onConfirm: (newPool: string[]) => void onClose: () => void + /** 列表周期: 1d 日线 / 1m 分钟, 与策略页当前周期一致 */ + timeframe?: '1d' | '1m' } const SOURCE_CLS: Record = { @@ -42,7 +44,7 @@ function fileStem(name: string): string { return name.replace(/\.py$/i, '').replace(/[^A-Za-z0-9_-]/g, '_').replace(/^_+|_+$/g, '') } -export function StrategyPoolDialog({ pool, onConfirm, onClose }: Props) { +export function StrategyPoolDialog({ pool, onConfirm, onClose, timeframe = '1d' }: Props) { const backdrop = useDialogBackdrop(onClose) // 草稿状态: 打开时从 pool 复制, 操作只改草稿, 点确定才提交 const [draftPool, setDraftPool] = useState(() => [...pool]) @@ -57,7 +59,7 @@ export function StrategyPoolDialog({ pool, onConfirm, onClose }: Props) { const loadStrategies = useCallback(async () => { setLoading(true) try { - const d = await api.strategyList() + const d = await api.strategyList(undefined, timeframe) setAllStrategies(d.strategies) } catch { setAllStrategies([]) diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 6ae9cc3..de54b86 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -319,6 +319,8 @@ export interface ScreenerStrategy { name: string description: string source?: string + /** 支持的周期, 如 ['1d'] / ['1m'] (分钟策略) */ + timeframes?: string[] } export interface StrategyLoadError { @@ -1457,6 +1459,8 @@ export interface Preferences { minute_sync_enabled: boolean minute_sync_days: number minute_sync_segment_days: number + minute_refresh_enabled: boolean + minute_refresh_interval: number daily_data_provider?: string adj_factor_provider?: string minute_data_provider?: string @@ -1644,15 +1648,38 @@ export const api = { }), }, ), - updateMinuteSync: (enabled: boolean, days: number, segmentDays?: number) => + updateMinuteSync: (enabled: boolean, days: number, segmentDays?: number, refresh?: { enabled?: boolean; interval?: number }) => request('/api/settings/preferences/minute-sync', { method: 'PUT', body: JSON.stringify({ minute_sync_enabled: enabled, minute_sync_days: days, ...(segmentDays != null ? { minute_sync_segment_days: segmentDays } : {}), + ...(refresh?.enabled != null ? { minute_refresh_enabled: refresh.enabled } : {}), + ...(refresh?.interval != null ? { minute_refresh_interval: refresh.interval } : {}), }), }), + + /** 盘中分钟增量刷新服务状态 (Expert 专有) */ + minuteRefreshStatus: () => + request<{ + available: boolean + enabled?: boolean + running?: boolean + interval_seconds?: number + capability_ok?: boolean + custom_provider_active?: boolean + in_trading_hours?: boolean + gate_reason?: string | null + rounds?: number + last_round_at?: number | null + last_round_ms?: number | null + last_rows?: number + last_symbols?: number + last_requests?: number + next_round_at?: number | null + last_error?: string | null + }>('/api/settings/minute-refresh/status'), updatePipelinePullTypes: (cfg: Partial>) => request<{ pipeline_pull_a_share: boolean @@ -2091,16 +2118,16 @@ export const api = { : '/api/watchlist/enriched', ), - screenerStrategies: async (assetType?: 'stock' | 'etf' | 'index') => { + screenerStrategies: async (assetType?: 'stock' | 'etf' | 'index', timeframe: '1d' | '1m' = '1d') => { const data = await request<{ strategies: StrategyDetail[]; load_errors?: StrategyLoadError[] }>( - `/api/strategies?${assetType ? `asset_type=${assetType}&` : ''}timeframe=1d`, + `/api/strategies?${assetType ? `asset_type=${assetType}&` : ''}timeframe=${timeframe}`, ) return { presets: data.strategies, load_errors: data.load_errors } }, - screenerRunPreset: (strategy_id: string, pool?: string[], asOf?: string, extColumns?: string, assetType: 'stock' | 'etf' = 'stock') => + screenerRunPreset: (strategy_id: string, pool?: string[], asOf?: string, extColumns?: string, assetType: 'stock' | 'etf' = 'stock', timeframe: '1d' | '1m' = '1d') => request('/api/screener/run_preset', { method: 'POST', - body: JSON.stringify({ strategy_id, pool, as_of: asOf ?? null, ext_columns: extColumns || null, asset_type: assetType }), + body: JSON.stringify({ strategy_id, pool, as_of: asOf ?? null, ext_columns: extColumns || null, asset_type: assetType, timeframe }), }), screenerRunCustom: (conditions: string[], orderBy?: string, limit = 30, pool?: string[], extColumns?: string, assetType: 'stock' | 'etf' = 'stock') => request('/api/screener/run', { diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index bc2910e..bd1e141 100644 --- a/frontend/src/pages/Screener.tsx +++ b/frontend/src/pages/Screener.tsx @@ -40,6 +40,8 @@ const SHOW_STRATEGY_STORE = false export function Screener() { const [assetType, setAssetType] = useState<'stock' | 'etf'>('stock') + // 周期: 日线 (盘后缓存 + runAll) / 分钟 (本地分钟K分区, 单策略实时跑) + const [timeframe, setTimeframe] = useState<'1d' | '1m'>('1d') const [activeStrategy, setActiveStrategy] = useState(null) const [result, setResult] = useState(null) const [asOf, setAsOf] = useState('') @@ -128,27 +130,28 @@ export function Screener() { const screenerAutoRun = prefs?.screener_auto_run ?? true const strategies = useQuery({ - queryKey: QK.screenerStrategies('all'), - queryFn: () => api.screenerStrategies(), + queryKey: [...QK.screenerStrategies('all'), timeframe], + queryFn: () => api.screenerStrategies(undefined, timeframe), }) // 卡片首屏只读取轻量摘要；明细在点击策略或“全部”时按需加载。 const summaryQuery = useQuery({ queryKey: QK.screenerCachedSummary, queryFn: api.screenerCachedSummary, - enabled: assetType === 'stock', + enabled: assetType === 'stock' && timeframe === '1d', }) const fullCachedQuery = useQuery({ queryKey: QK.screenerCached(asOf, extColumnsParam), queryFn: () => api.screenerCached(extColumnsParam || undefined), - enabled: assetType === 'stock' && showAll, + enabled: assetType === 'stock' && timeframe === '1d' && showAll, }) const singleCachedQuery = useQuery({ queryKey: QK.screenerCachedResult(activeStrategy ?? '', asOf, extColumnsParam), queryFn: () => api.screenerCachedResult(activeStrategy!, extColumnsParam || undefined), enabled: assetType === 'stock' + && timeframe === '1d' && !showAll && !!activeStrategy && summaryQuery.data?.results[activeStrategy]?.as_of === asOf, @@ -204,8 +207,10 @@ export function Screener() { if (strategies.isError) return // 拉取失败: 不 prune if (!strategies.isSuccess) return // 加载中: 不 prune if (allStrategyIds.size === 0) return // 空列表: 不 prune + // 分钟模式的列表只含分钟策略, prune 会误删池中的日线策略 → 仅日线模式清理 + if (timeframe !== '1d') return prune(allStrategyIds) - }, [allStrategyIds, prune, strategies.isError, strategies.isSuccess]) + }, [allStrategyIds, prune, strategies.isError, strategies.isSuccess, timeframe]) // 策略文件加载失败时提示用户(避免"策略静默消失"被误判为正常) const loadErrors = strategies.data?.load_errors ?? [] @@ -445,7 +450,8 @@ export function Screener() { // 缓存命中时秒加载; 未命中时, 仅当 screener_auto_run 开启才自动触发 runAll useEffect(() => { // ETF 模式无股票盘后缓存/ runAll, 单策略走实时单跑, 不触发 runAll - if (assetType !== 'stock') return + // 分钟模式走本地分钟K分区, 同样不触发 runAll (盘后缓存是日线语义) + if (assetType !== 'stock' || timeframe !== '1d') return if (!asOf || strategyPresets.length === 0 || !summaryQuery.isSuccess || runAll.isPending || visiblePool.length === 0) return const runKey = `${asOf}|${visiblePool.join(',')}` if (runAllDateRef.current === runKey) return @@ -458,11 +464,11 @@ export function Screener() { if (!screenerAutoRun) return runAllDateRef.current = runKey requestRunAll({ date: asOf, strategyIds: missingStrategyIds }) - }, [asOf, strategyPresets.length, summaryQuery.isSuccess, visiblePool, cacheCoversPool, missingStrategyIds, screenerAutoRun, assetType, runAll.isPending, requestRunAll]) + }, [asOf, strategyPresets.length, summaryQuery.isSuccess, visiblePool, cacheCoversPool, missingStrategyIds, screenerAutoRun, assetType, timeframe, runAll.isPending, requestRunAll]) const run = useMutation({ mutationFn: ({ id, date }: { id: string; date: string }) => - api.screenerRunPreset(id, undefined, date || undefined, extColumnsParam || undefined, assetType), + api.screenerRunPreset(id, undefined, date || undefined, extColumnsParam || undefined, assetType, timeframe), onSuccess: (data, vars) => { setResult(data) // 同步更新卡片上的命中数 @@ -479,7 +485,7 @@ export function Screener() { if (result?.strategy !== s.id || result.as_of !== asOf) setResult(null) // ETF 模式: 无股票盘后缓存, 始终实时单跑。 // 传空日期让后端用 ETF 自己的最新交易日 (asOf 跟随的是股票 enriched, 两者可能不同日)。 - if (assetType !== 'stock') { + if (assetType !== 'stock' || timeframe !== '1d') { run.mutate({ id: s.id, date: '' }) return } @@ -602,19 +608,47 @@ export function Screener() { subtitle="基于本地 enriched 表 · 毫秒级 SQL" right={
- {/* 资产类型切换: 股票 / ETF */} + {/* 资产类型切换: 股票 / ETF (分钟策略仅支持股票, 1m 模式下 ETF 置灰) */}
- {(['stock', 'etf'] as const).map(t => ( + {(['stock', 'etf'] as const).map(t => { + const disabled = t === 'etf' && timeframe === '1m' + return ( + + ) + })} +
+ {/* 周期切换: 日线 (盘后缓存) / 分钟 (本地分钟K分区实时计算) */} +
+ {(['1d', '1m'] as const).map(tf => ( ))}
@@ -746,6 +780,7 @@ export function Screener() { onSettings={() => setSettingsStrategyId(s.id)} monitored={strategyMonitorMap.has(s.id)} onToggleMonitor={() => toggleStrategyMonitor(s.id, s.name)} + timeframeBadge={s.timeframes?.includes('1m') ? '分钟' : undefined} /> ) })} @@ -1004,6 +1039,7 @@ export function Screener() { {showPoolDialog && ( { reorderPool(newPool) }}