diff --git a/backend/app/api/signals.py b/backend/app/api/signals.py index fec23bb..404f8f3 100644 --- a/backend/app/api/signals.py +++ b/backend/app/api/signals.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel from app.strategy import custom_signals +from app.strategy.intraday_features import INTRADAY_FEATURES router = APIRouter(prefix="/api/custom-signals", tags=["custom-signals"]) @@ -24,7 +25,9 @@ def _invalidate(request: Request) -> None: 信号增删会改变注入列集合: 只清表达式缓存不够, repo 内存缓存 / strategy 磁盘缓存里算好的历史窗口仍不含新 csg_ 列 (或仍含已删列), 需要一并清除, 否则创建信号后立即运行策略仍会报缺列。 + 盘中信号定义缓存(intraday)一并失效, 下一分钟 bucket 即生效。 """ + custom_signals.invalidate_intraday_cache() from app.indicators.pipeline import invalidate_custom_signals invalidate_custom_signals() from app.services import strategy_cache @@ -35,11 +38,11 @@ def _invalidate(request: Request) -> None: class ConditionModel(BaseModel): - left: str # 字段名（须在白名单） - op: str # > >= < <= == != + left: str # 字段名（日线在白名单 / 盘中在特征白名单） + op: str # > >= < <= == != ; 盘中额外: cross_up cross_down right: str # "field:xxx" 或数字字符串 - leftDays: int = 0 # 左字段取几日前 (0=当日, 默认) - rightDays: int = 0 # 右字段取几日前 (仅 right 为字段时有意义) + leftDays: int = 0 # 左字段取几日前 (0=当日, 默认; 盘中信号必须为 0) + rightDays: int = 0 # 右字段取几日前 (仅 right 为字段时有意义; 盘中信号必须为 0) class SignalModel(BaseModel): @@ -48,6 +51,17 @@ class SignalModel(BaseModel): kind: str # entry | exit | both conditions: list[ConditionModel] enabled: bool = True + timeframe: str = "daily" # daily | intraday(分钟K特征, 输出当日条件上升沿) + min_bars: int = 0 # 仅 intraday: 当日最少已完成 bar 数, 不足不触发 + + +class IntradayReplayRequest(BaseModel): + """盘中信号历史回放 — 用本地分钟K重放触发时点, 不消耗盘中数据能力。""" + signal_id: str + start_date: str # YYYY-MM-DD + end_date: str # YYYY-MM-DD + symbols: list[str] + asset_type: str = "stock" class AIGenerateRequest(BaseModel): @@ -115,6 +129,18 @@ def get_options(): {"key": "exit", "label": "出场"}, {"key": "both", "label": "出入通用"}, ], + # 盘中信号(timeframe=intraday): 分钟K特征白名单 + 额外穿越算子 + "intraday": { + "fields": [ + {"key": f, "label": label} + for f, label in sorted(INTRADAY_FEATURES.items()) + ], + "operators": [">", ">=", "<", "<=", "==", "!=", "cross_up", "cross_down"], + }, + "timeframes": [ + {"key": "daily", "label": "日线"}, + {"key": "intraday", "label": "盘中(分钟K)"}, + ], } @@ -189,3 +215,102 @@ def delete_signal(signal_id: str, request: Request): raise HTTPException(status_code=404, detail="信号不存在") _invalidate(request) return {"ok": True} + + +# ── 盘中信号历史回放 ──────────────────────────────────── + + +@router.post("/intraday/replay") +def intraday_replay(req: IntradayReplayRequest, request: Request): + """用本地历史分钟K回放盘中信号的触发时点。 + + 只读本地分钟分区, 不消耗盘中数据能力 — 用户可先在历史区间验证信号, + 再决定是否配置到监控/分钟策略。昨收取自本地日K(无昨日数据的日子该特征降级)。 + """ + from datetime import date, timedelta + + import polars as pl + + from app.strategy.intraday_features import build_feature_frame + + try: + start = date.fromisoformat(req.start_date) + end = date.fromisoformat(req.end_date) + except ValueError as e: + raise HTTPException(status_code=400, detail=f"日期格式错误: {e}") from e + if start > end: + raise HTTPException(status_code=400, detail="start_date 不能晚于 end_date") + if (end - start).days > 60: + raise HTTPException(status_code=400, detail="回放区间最长 60 天") + symbols = [s for s in dict.fromkeys(req.symbols) if s] + if not symbols: + raise HTTPException(status_code=400, detail="symbols 不能为空") + if len(symbols) > 200: + raise HTTPException(status_code=400, detail="单次回放最多 200 只标的") + + # 信号定义必须存在且为盘中类型 + sig = next( + (s for s in custom_signals.load_all(_data_dir(request)) if s.get("id") == req.signal_id), + None, + ) + if sig is None: + raise HTTPException(status_code=404, detail="信号不存在") + if sig.get("timeframe") != custom_signals.TIMEFRAME_INTRADAY: + raise HTTPException(status_code=400, detail="该信号不是盘中(timeframe=intraday)信号") + exprs = custom_signals.build_intraday_expressions([sig]) + col = custom_signals.intraday_column_name(sig["id"]) + if col not in exprs: + raise HTTPException(status_code=400, detail="信号编译失败, 请检查条件字段") + min_bars = int(sig.get("min_bars", 0) or 0) + + repo = request.app.state.repo + # 昨收映射: 一次性取区间(含前置 15 天)日K, 按「严格早于当日」取最近收盘 + daily = repo.get_daily_batch(symbols, start - timedelta(days=15), end, columns=["symbol", "date", "close"]) + close_by_sym_date: dict[str, dict[date, float]] = {} + if not daily.is_empty(): + for row in daily.sort(["symbol", "date"]).iter_rows(named=True): + close_by_sym_date.setdefault(str(row["symbol"]), {})[row["date"]] = float(row["close"]) + + triggers: list[dict] = [] + days_scanned = 0 + bars_scanned = 0 + day = start + while day <= end: + minute_df = repo.get_minute_batch(symbols, day, asset_type=req.asset_type) + if minute_df is not None and not minute_df.is_empty(): + days_scanned += 1 + bars_scanned += minute_df.height + prev_close = { + sym: closes_map[max(d for d in closes_map if d < day)] + for sym, closes_map in close_by_sym_date.items() + if any(d < day for d in closes_map) + } + frame = build_feature_frame(minute_df, prev_close=prev_close) + if not frame.is_empty(): + evaluated = custom_signals.apply_intraday_edges(frame, {col: exprs[col]}).with_columns( + pl.int_range(pl.len()).over(["symbol", "date"]).alias("_bar_idx") + ) + if min_bars > 0: + evaluated = evaluated.with_columns( + pl.when(pl.col("_bar_idx") + 1 >= min_bars) + .then(pl.col(col)) + .otherwise(False) + .alias(col) + ) + for row in evaluated.filter(pl.col(col)).sort(["datetime", "symbol"]).iter_rows(named=True): + triggers.append({ + "date": day.isoformat(), + "time": str(row["datetime"].time()), + "symbol": row["symbol"], + }) + day += timedelta(days=1) + + return { + "signal_id": req.signal_id, + "start_date": req.start_date, + "end_date": req.end_date, + "symbols": symbols, + "days_scanned": days_scanned, + "bars_scanned": bars_scanned, + "triggers": triggers, + } diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 07eb9f5..186c20a 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -1440,9 +1440,20 @@ class QuoteService: prev_close=prev_close, asset_type=asset_type, now=now, + signals=self._load_intraday_signal_defs(), ) return self._intraday_signal_evaluator.inject(enriched, signals) + def _load_intraday_signal_defs(self) -> list[dict]: + """加载自定义盘中信号定义(带指纹缓存); 失败时退化为仅内置 4 信号。""" + try: + from app.strategy import custom_signals + + return custom_signals.load_intraday_all(self._repo.store.data_dir) + except Exception as e: + logger.warning("load intraday signal defs failed: %s", e) + return [] + @staticmethod def _continuous_session_start_ms() -> float: """当前连续竞价时段的起点 (北京时间 9:30 或 13:00) 的 epoch 毫秒。""" diff --git a/backend/app/strategy/custom_signals.py b/backend/app/strategy/custom_signals.py index e78995b..ef5e51c 100644 --- a/backend/app/strategy/custom_signals.py +++ b/backend/app/strategy/custom_signals.py @@ -202,11 +202,17 @@ def validate(sig: dict) -> None: raise ValueError("信号 name 不能为空") if sig.get("kind") not in ("entry", "exit", "both"): raise ValueError("kind 必须是 entry / exit / both") + timeframe = sig.get("timeframe", TIMEFRAME_DAILY) + if timeframe not in (TIMEFRAME_DAILY, TIMEFRAME_INTRADAY): + raise ValueError(f"timeframe 必须是 {TIMEFRAME_DAILY} / {TIMEFRAME_INTRADAY}: {timeframe!r}") conds = sig.get("conditions") if not isinstance(conds, list) or len(conds) == 0: raise ValueError("conditions 不能为空") if len(conds) > 8: raise ValueError("conditions 最多 8 条") + if timeframe == TIMEFRAME_INTRADAY: + _validate_intraday(sig) + return for i, c in enumerate(conds): if not isinstance(c, dict): raise ValueError(f"第 {i+1} 个条件格式错误") @@ -317,3 +323,169 @@ def _expr_root_columns(expr: pl.Expr) -> set[str]: return set(names) except Exception: return set() + + +# ══ 盘中信号(timeframe="intraday")═════════════════════════ +# 与日线自定义信号同一套 left/op/right 条件结构, 但: +# - 字段白名单换成分钟特征(intraday_features.INTRADAY_FEATURES); +# - 运算符额外支持 cross_up / cross_down(序列上穿/下穿 另一序列或阈值); +# - 不支持 leftDays/rightDays 日期偏移; +# - 信号列名前缀 csgi_, 注入对象是分钟特征帧而非日线 enriched。 +# 语义: 信号输出 = 当日条件组合的上升沿(false→true), 首根 bar 不触发。 + +from app.strategy.intraday_features import INTRADAY_FEATURES # noqa: E402 + +TIMEFRAME_DAILY = "daily" +TIMEFRAME_INTRADAY = "intraday" +INTRADAY_PREFIX = "csgi_" +INTRADAY_OPS = OPS | {"cross_up", "cross_down"} +_EDGE_GROUP = ["symbol", "date"] + + +def intraday_column_name(signal_id: str) -> str: + """盘中信号 id → 分钟帧列名(加 csgi_ 前缀)。""" + return f"{INTRADAY_PREFIX}{signal_id}" + + +def _parse_right_intraday(right: object) -> tuple[str, object]: + """盘中条件的右值: ('const', float) 或 ('field', 特征名)。""" + if isinstance(right, (int, float)): + return ("const", float(right)) + if not isinstance(right, str): + raise ValueError(f"非法右值: {right!r}") + if right.startswith("field:"): + col = right[len("field:"):] + if col not in INTRADAY_FEATURES: + raise ValueError(f"盘中右值字段不在白名单: {col}") + return ("field", col) + try: + return ("const", float(right)) + except ValueError: + pass + if right in INTRADAY_FEATURES: + return ("field", right) + raise ValueError(f"非法盘中右值(应为 field:特征 或数字): {right!r}") + + +def _validate_intraday(sig: dict) -> None: + """校验盘中信号定义, 非法抛 ValueError。""" + conds = sig.get("conditions") + for i, c in enumerate(conds): + if not isinstance(c, dict): + raise ValueError(f"第 {i+1} 个条件格式错误") + left = c.get("left", "") + if left not in INTRADAY_FEATURES: + raise ValueError(f"第 {i+1} 个条件: 盘中字段 {left!r} 不在白名单") + if c.get("op") not in INTRADAY_OPS: + raise ValueError(f"第 {i+1} 个条件: 运算符 {c.get('op')!r} 非法(盘中额外支持 cross_up/cross_down)") + _parse_right_intraday(c.get("right")) + if int(c.get("leftDays", 0) or 0) or int(c.get("rightDays", 0) or 0): + raise ValueError(f"第 {i+1} 个条件: 盘中信号不支持日期偏移(leftDays/rightDays)") + min_bars = sig.get("min_bars", 0) + try: + n = int(min_bars) + except (TypeError, ValueError): + raise ValueError(f"min_bars 必须是整数: {min_bars!r}") # noqa: B904 + if n < 0 or n > 240: + raise ValueError(f"min_bars 必须在 0..240 之间: {n}") + + +def build_intraday_expressions(signals: list[dict]) -> dict[str, pl.Expr]: + """把盘中信号编译为特征帧上的「条件」表达式(AND 组合, 未做上升沿)。 + + 表达式在 intraday_features.build_feature_frame 产出的帧上求值; + 上升沿须通过 apply_intraday_edges 在 DataFrame 层两步计算 — + 对已含 .over() 窗口的组合表达式直接 shift().over() 是窗口嵌套, + Polars 会返回全 null。编译失败的信号跳过并告警。 + """ + out: dict[str, pl.Expr] = {} + for sig in signals: + if sig.get("enabled") is False or sig.get("timeframe") != TIMEFRAME_INTRADAY: + continue + try: + parts: list[pl.Expr] = [] + for c in sig["conditions"]: + left = pl.col(c["left"]) + kind, val = _parse_right_intraday(c["right"]) + op = c["op"] + if op == "cross_up": + # 前一根 bar 未满足 且 当前 bar 满足; 右值为常量时不 shift 字面量 + if kind == "field": + prev_ok = left.shift(1).over(_EDGE_GROUP) <= pl.col(val).shift(1).over(_EDGE_GROUP) + cur_ok = left > pl.col(val) + else: + prev_ok = left.shift(1).over(_EDGE_GROUP) <= val + cur_ok = left > val + parts.append(prev_ok & cur_ok) + elif op == "cross_down": + if kind == "field": + prev_ok = left.shift(1).over(_EDGE_GROUP) >= pl.col(val).shift(1).over(_EDGE_GROUP) + cur_ok = left < pl.col(val) + else: + prev_ok = left.shift(1).over(_EDGE_GROUP) >= val + cur_ok = left < val + parts.append(prev_ok & cur_ok) + else: + right = pl.col(val) if kind == "field" else val + parts.append(_OP_BUILDERS[op](left, right)) + combined = parts[0] + for p in parts[1:]: + combined = combined & p + out[intraday_column_name(sig["id"])] = combined + except Exception as e: + logger.warning("intraday signal compile failed %s: %s", sig.get("id"), e) + return out + + +def apply_intraday_edges(frame: pl.DataFrame, exprs: dict[str, pl.Expr]) -> pl.DataFrame: + """对特征帧求值盘中信号: 先算条件列, 再取「当日条件上升沿」为布尔列。 + + 上升沿: 条件 false→true 的那根 bar 为 true; 首根 bar(前值为 null)不触发; + 条件含 null(特征不足)视为 false。四条消费路径(监控/实盘/回测/回放) + 必须共用本函数, 保证口径一致。 + """ + if frame.is_empty() or not exprs: + return frame + df = frame.with_columns([e.fill_null(False).alias(n) for n, e in exprs.items()]) + return df.with_columns([ + ( + pl.col(n) + & ~pl.col(n).shift(1).over(_EDGE_GROUP).fill_null(True) + ).cast(pl.Boolean).alias(n) + for n in exprs + ]) + + +# ── 盘中信号定义加载(带指纹缓存: 引擎/监控高频路径用) ────────── +_intraday_cache: dict[Path, tuple[object, list[dict]]] = {} + + +def _dir_fingerprint(d: Path) -> tuple: + """目录内 *.json 的 (文件名, mtime) 指纹 — 创建/删除/编辑都会变化。""" + try: + return tuple(sorted((f.name, f.stat().st_mtime_ns) for f in d.glob("*.json"))) + except OSError: + return () + + +def load_intraday_all(data_dir: Path) -> list[dict]: + """读取全部启用的盘中信号定义(带缓存)。 + + 盘中评估与引擎注入每分钟执行, 不宜每次全量读盘; save/delete 端点 + 调用 invalidate_intraday_cache() 主动失效。 + """ + d = _dir(data_dir) + fp = _dir_fingerprint(d) + cached = _intraday_cache.get(data_dir) + if cached is not None and cached[0] == fp: + return cached[1] + sigs = [ + s for s in load_all(data_dir) + if s.get("timeframe") == TIMEFRAME_INTRADAY and s.get("enabled") is not False + ] + _intraday_cache[data_dir] = (fp, sigs) + return sigs + + +def invalidate_intraday_cache() -> None: + _intraday_cache.clear() diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index 0636f98..2c28ae7 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -951,6 +951,19 @@ class StrategyEngine: strategy_id=strategy_id, exit_signal_hits=exit_signal_hits, ) + # 盘中信号列注入(csgi_): 实盘扫描与分钟回测共用本路径 — 与监控评估 + # 同一特征构造器, 单点注入保证三处口径一致。 + history = self._inject_intraday_signal_columns(history) + missing_csgi = [ + name for name in s.required_features + if name.startswith("csgi_") and name not in history.columns + ] + if missing_csgi: + raise ValueError( + "策略引用了未定义的盘中信号: " + + ", ".join(sorted(missing_csgi)) + + " — 请先在「自定义信号」中创建(timeframe=intraday)后再运行" + ) if s.minute_daily_bars > 0: df = s.filter_minute_history_fn(history, params, daily=context.daily_history) else: @@ -982,6 +995,15 @@ class StrategyEngine: + ", ".join(sorted(missing_csg)) + " — 请先在「自定义信号」管理中创建对应信号后再运行" ) + missing_csgi = [ + name for name in s.required_features + if name.startswith("csgi_") + ] + if missing_csgi: + raise ValueError( + "盘中信号仅可用于分钟策略(timeframes=['1m']), 日线策略不支持: " + + ", ".join(sorted(missing_csgi)) + ) df = s.filter_history_fn(df, params) if "date" in df.columns: df = df.filter(pl.col("date") == as_of) @@ -1555,6 +1577,48 @@ class StrategyEngine: "turnover_rate", "change_pct", "pre_close", ) + def _user_data_dir(self) -> Path | None: + """从策略目录推导 data_dir(…/strategies/custom → data_dir)。推不出则跳过注入。""" + for d in self._strategy_dirs: + if d.name == "custom" and d.parent.name == "strategies": + return d.parent.parent + return None + + def _inject_intraday_signal_columns(self, minute_df: pl.DataFrame) -> pl.DataFrame: + """向当日分钟K帧注入自定义盘中信号列(csgi_, 当日条件上升沿)。 + + 单点注入: 实盘分钟扫描与分钟回测 worker 共用本方法, 特征计算与 + 监控评估同源(intraday_features), 保证口径一致。无定义/帧为空时原样返回。 + """ + if minute_df is None or minute_df.is_empty() or "datetime" not in minute_df.columns: + return minute_df + data_dir = self._user_data_dir() + if data_dir is None: + return minute_df + try: + from app.strategy import custom_signals + from app.strategy.intraday_features import build_feature_frame + + definitions = custom_signals.load_intraday_all(data_dir) + if not definitions: + return minute_df + exprs = custom_signals.build_intraday_expressions(definitions) + if not exprs: + return minute_df + frame = build_feature_frame(minute_df) + if frame.is_empty(): + return minute_df + evaluated = custom_signals.apply_intraday_edges(frame, exprs).select( + ["symbol", "datetime", *exprs.keys()] + ) + return minute_df.join(evaluated, on=["symbol", "datetime"], how="left").with_columns( + [pl.col(name).fill_null(False).cast(pl.Boolean).alias(name) for name in exprs] + ) + except Exception as e: + # 注入失败不阻断策略执行: 未注入列会由 required_features 校验兜底报错 + logger.warning("intraday signal inject failed: %s", e) + return minute_df + @staticmethod def _join_basic_columns(df: pl.DataFrame, current: pl.DataFrame) -> pl.DataFrame: """把 enriched 快照列按 symbol 联到分钟策略输出上, 只补 df 缺失的列。""" diff --git a/backend/app/strategy/intraday_features.py b/backend/app/strategy/intraday_features.py new file mode 100644 index 0000000..ba3d8a4 --- /dev/null +++ b/backend/app/strategy/intraday_features.py @@ -0,0 +1,178 @@ +"""盘中信号特征帧 — 当日已完成分钟 K → 数值特征序列(每根已完成 bar 一行)。 + +单一口径源: 监控评估(quote_service) / 分钟策略执行(引擎注入) / 分钟回测 / 回放验证 +共用本模块构造特征, 保证四条路径对同一段分钟数据产出完全一致的特征值。 + +设计: + - 会话对齐: 滚动窗口只在本时段(09:30-11:30 / 13:00-15:00)内回看, + 不跨午休、不跨日; 日累计类特征(vwap/当日高低)按交易日分组。 + - null 语义: 窗口不足、基准为零、缺昨收、开盘未满 30 分钟 → 特征为 null, + 任何条件对 null 判 false, 绝不把数据不足伪装成 0。 + - 纯函数: 不做 IO, 分钟帧由调用方传入(cutoff 过滤也由调用方决定)。 +""" +from __future__ import annotations + +from datetime import datetime + +import polars as pl + +from app.market_time import CN_TZ + +# ── 特征白名单(供 custom_signals 校验与 /options 展示) ────────── +# 字段 → 中文标签。数值均为「每根已完成 bar 一个值」的序列。 +INTRADAY_FEATURES: dict[str, str] = { + "price": "现价", + "vwap": "分时均价", + "pct_vs_prev_close": "相对昨收涨跌幅", + "pct_from_open": "相对开盘涨跌幅", + "vol_ratio_1m_today": "1分钟放量比(今日基准)", + "vol_ratio_3m_today": "3分钟放量比(今日基准)", + "vol_ratio_5m_today": "5分钟放量比(今日基准)", + "day_high_dist": "距当日最高价", + "day_low_dist": "距当日最低价", + "open_30m_high_dist": "距开盘30分钟最高价", + "open_30m_low_dist": "距开盘30分钟最低价", +} + +# 滚动窗口特征的窗口长度(字段名后缀 → bar 数) +_VOL_WINDOWS = {1: "vol_ratio_1m_today", 3: "vol_ratio_3m_today", 5: "vol_ratio_5m_today"} + +_REQUIRED_COLS = ("symbol", "datetime", "close", "volume", "amount") +_DAY_KEY = ["symbol", "date"] +_SESSION_KEY = ["symbol", "date", "session"] + + +def _naive(dt: datetime) -> datetime | None: + """统一为北京墙钟 naive(与分钟存储契约一致)。""" + if not isinstance(dt, datetime): + return None + if dt.tzinfo is not None: + return dt.astimezone(CN_TZ).replace(tzinfo=None) + return dt + + +def build_feature_frame( + minute_df: pl.DataFrame, + *, + prev_close: dict[str, float] | None = None, + cutoff: datetime | None = None, +) -> pl.DataFrame: + """把分钟 K 帧编译为特征帧。 + + 参数: + minute_df: 列含 symbol/datetime/open/high/low/close/volume/amount(后四列必需, + OHLC 缺失时相关特征降级为 null)。可包含多个交易日, 特征按日分组。 + prev_close: 映射 symbol → 昨收(已复权口径需与分钟价一致); 缺失标的的 + pct_vs_prev_close 为 null。 + cutoff: 只使用严格早于 cutoff 的 bar(盘中传入当前分钟; 回放/回测传 None)。 + + 返回: symbol/datetime + INTRADAY_FEATURES 全部特征列(Float64, 可 null)。 + """ + empty = pl.DataFrame( + schema={"symbol": pl.Utf8, "datetime": pl.Datetime, "date": pl.Date} + | {name: pl.Float64 for name in INTRADAY_FEATURES} + ) + if minute_df is None or minute_df.is_empty() or not set(_REQUIRED_COLS).issubset(minute_df.columns): + return empty + + df = minute_df + if "symbol" in df.columns: + df = df.with_columns(pl.col("symbol").cast(pl.Utf8)) + dt_expr = pl.col("datetime") + if df.schema["datetime"].time_zone is not None: + dt_expr = dt_expr.dt.convert_time_zone(CN_TZ.key).dt.replace_time_zone(None) + df = df.with_columns(dt_expr.alias("datetime")) + if cutoff is not None: + cut = _naive(cutoff) + if cut is not None: + df = df.filter(pl.col("datetime") < cut) + df = df.drop_nulls("datetime").sort(["symbol", "datetime"]) + if df.is_empty(): + return empty + + df = df.with_columns( + pl.col("datetime").dt.date().alias("date"), + # 会话归属: 13:00 及以后为午后续时段, 滚动窗口不与上午合并 + pl.when(pl.col("datetime").dt.hour() >= 13).then(1).otherwise(0).alias("session"), + ) + df = df.with_columns(pl.int_range(pl.len()).over(_SESSION_KEY).alias("session_idx")) + + cols = {"price": pl.col("close").cast(pl.Float64)} + + # ── 日累计特征(跨上午/下午累计) ── + if {"volume", "amount"}.issubset(df.columns): + cum_vol = pl.col("volume").cast(pl.Float64).cum_sum().over(_DAY_KEY) + cum_amt = pl.col("amount").cast(pl.Float64).cum_sum().over(_DAY_KEY) + cols["vwap"] = pl.when(cum_vol > 0).then(cum_amt / (cum_vol * 100.0)) + else: + cols["vwap"] = pl.lit(None, dtype=pl.Float64) + + if prev_close: + pc = pl.DataFrame( + {"symbol": list(prev_close.keys()), "_prev_close": [float(v) for v in prev_close.values()]} + ) + df = df.join(pc, on="symbol", how="left") + cols["pct_vs_prev_close"] = pl.when( + pl.col("_prev_close").is_not_null() & (pl.col("_prev_close") > 0) + ).then(pl.col("close") / pl.col("_prev_close") - 1.0) + else: + cols["pct_vs_prev_close"] = pl.lit(None, dtype=pl.Float64) + + if "open" in df.columns: + day_open = pl.col("open").cast(pl.Float64).first().over(_DAY_KEY) + cols["pct_from_open"] = pl.when(day_open > 0).then(pl.col("close") / day_open - 1.0) + else: + cols["pct_from_open"] = pl.lit(None, dtype=pl.Float64) + + if "high" in df.columns: + day_high = pl.col("high").cast(pl.Float64).cum_max().over(_DAY_KEY) + cols["day_high_dist"] = pl.when(day_high > 0).then(pl.col("close") / day_high - 1.0) + else: + cols["day_high_dist"] = pl.lit(None, dtype=pl.Float64) + + if "low" in df.columns: + day_low = pl.col("low").cast(pl.Float64).cum_min().over(_DAY_KEY) + cols["day_low_dist"] = pl.when(day_low > 0).then(pl.col("close") / day_low - 1.0) + else: + cols["day_low_dist"] = pl.lit(None, dtype=pl.Float64) + + # ── 滚动放量比(今日基准): 当前 N 根 bar 量和 / 此前 N 根 bar 量和 ── + # 滚动窗口按(symbol, date, session)分组 → 不跨午休、不跨日; 窗口不满自然为 null。 + if "volume" in df.columns: + vol = pl.col("volume").cast(pl.Float64) + for n, name in _VOL_WINDOWS.items(): + win = vol.rolling_sum(n).over(_SESSION_KEY) + prev_win = win.shift(n).over(_SESSION_KEY) + cols[name] = pl.when(prev_win > 0).then(win / prev_win) + else: + for name in _VOL_WINDOWS.values(): + cols[name] = pl.lit(None, dtype=pl.Float64) + + # ── 开盘 30 分钟高低点: 上午时段第 30 根 bar 的累计高/低, 全日广播 ── + if "high" in df.columns: + marker_h = ( + pl.when((pl.col("session") == 0) & (pl.col("session_idx") == 29)) + .then(pl.col("high").cast(pl.Float64).cum_max().over(_DAY_KEY)) + .otherwise(None) + .forward_fill() + .over(_DAY_KEY) + ) + cols["open_30m_high_dist"] = pl.when(marker_h > 0).then(pl.col("close") / marker_h - 1.0) + else: + cols["open_30m_high_dist"] = pl.lit(None, dtype=pl.Float64) + + if "low" in df.columns: + marker_l = ( + pl.when((pl.col("session") == 0) & (pl.col("session_idx") == 29)) + .then(pl.col("low").cast(pl.Float64).cum_min().over(_DAY_KEY)) + .otherwise(None) + .forward_fill() + .over(_DAY_KEY) + ) + cols["open_30m_low_dist"] = pl.when(marker_l > 0).then(pl.col("close") / marker_l - 1.0) + else: + cols["open_30m_low_dist"] = pl.lit(None, dtype=pl.Float64) + + return df.with_columns([expr.cast(pl.Float64).alias(name) for name, expr in cols.items()]).select( + ["symbol", "date", "datetime", *INTRADAY_FEATURES.keys()] + ) diff --git a/backend/app/strategy/intraday_signals.py b/backend/app/strategy/intraday_signals.py index c2ff6a1..79eecbc 100644 --- a/backend/app/strategy/intraday_signals.py +++ b/backend/app/strategy/intraday_signals.py @@ -1,13 +1,25 @@ -"""监控中心专用的日内分时穿越信号。""" +"""监控中心专用的日内分时信号评估器。 + +v2: 特征计算与条件求值统一走 intraday_features 特征帧 + custom_signals 的 +盘中表达式编译 — 与分钟策略执行/分钟回测/回放验证同一条口径。 + +- 内置 4 个分时穿越信号(signal_intraday_*)由同一表达式机制生成, 列名不变, + 存量监控规则零迁移; +- 自定义盘中信号(timeframe="intraday", csgi_ 前缀)与内置信号一并评估注入。 +""" from __future__ import annotations -import math +import logging from datetime import datetime from typing import Any import polars as pl from app.market_time import CN_TZ +from app.strategy import custom_signals +from app.strategy.intraday_features import build_feature_frame + +logger = logging.getLogger(__name__) INTRADAY_SIGNAL_LABELS: dict[str, str] = { "signal_intraday_avg_cross_up": "分时价格上穿均价", @@ -16,22 +28,41 @@ INTRADAY_SIGNAL_LABELS: dict[str, str] = { "signal_intraday_zero_cross_down": "分时价格下穿0轴", } INTRADAY_SIGNAL_FIELDS = frozenset(INTRADAY_SIGNAL_LABELS) +_LEGACY_MIN_BARS = 2 # 旧实现要求至少两根已完成 bar 才判穿越, 语义保持 def uses_intraday_signals(rule: dict) -> bool: + """规则是否引用盘中信号列(内置 4 个或自定义 csgi_)。""" return any( - c.get("op") == "truth" and c.get("field") in INTRADAY_SIGNAL_FIELDS + ( + isinstance(c, dict) + and c.get("op") == "truth" + and (c.get("field") in INTRADAY_SIGNAL_FIELDS or str(c.get("field", "")).startswith(custom_signals.INTRADAY_PREFIX)) + ) for c in rule.get("conditions", []) - if isinstance(c, dict) ) -def _finite(value: Any) -> float | None: - try: - number = float(value) - except (TypeError, ValueError): - return None - return number if math.isfinite(number) else None +def _legacy_builtin_definitions() -> list[dict]: + """内置 4 个分时穿越信号的等价定义(与 v1 逐字节同口径)。 + + v1 语义: 上穿 = 前一根 bar 未满足且当前 bar 满足 —— 与 + build_intraday_expressions 的「条件上升沿」完全一致。 + """ + return [ + {"id": "signal_intraday_avg_cross_up", "timeframe": "intraday", "enabled": True, + "conditions": [{"left": "price", "op": "cross_up", "right": "field:vwap"}], + "min_bars": _LEGACY_MIN_BARS}, + {"id": "signal_intraday_avg_cross_down", "timeframe": "intraday", "enabled": True, + "conditions": [{"left": "price", "op": "cross_down", "right": "field:vwap"}], + "min_bars": _LEGACY_MIN_BARS}, + {"id": "signal_intraday_zero_cross_up", "timeframe": "intraday", "enabled": True, + "conditions": [{"left": "pct_vs_prev_close", "op": "cross_up", "right": 0}], + "min_bars": _LEGACY_MIN_BARS}, + {"id": "signal_intraday_zero_cross_down", "timeframe": "intraday", "enabled": True, + "conditions": [{"left": "pct_vs_prev_close", "op": "cross_down", "right": 0}], + "min_bars": _LEGACY_MIN_BARS}, + ] def _naive_datetime(value: Any) -> datetime | None: @@ -43,7 +74,7 @@ def _naive_datetime(value: Any) -> datetime | None: class IntradaySignalEvaluator: - """按已完成的一分钟 K 线生成边沿触发信号。""" + """按已完成的一分钟 K 线评估盘中信号(边沿触发, 新 bar 出现才可能触发)。""" def __init__(self) -> None: self._last_bar: dict[tuple[str, str], datetime] = {} @@ -56,86 +87,92 @@ class IntradaySignalEvaluator: prev_close: dict[str, float], asset_type: str, now: datetime, + signals: list[dict] | None = None, ) -> list[dict[str, Any]]: + """返回本分钟触发信号的行列表(每 symbol 一行, 仅新出现的 bar 触发)。""" active_keys = {(asset_type, symbol) for symbol in symbols} self._last_bar = { key: value for key, value in self._last_bar.items() if key[0] != asset_type or key in active_keys } - required = {"symbol", "datetime", "close", "volume", "amount"} - if not symbols or minute_df.is_empty() or not required.issubset(minute_df.columns): + definitions = _legacy_builtin_definitions() + list(signals or []) + if not symbols: return [] + frame = build_feature_frame( + minute_df.filter(pl.col("symbol").cast(pl.Utf8).is_in(sorted(symbols))), + prev_close=prev_close, + cutoff=now, + ) + if frame.is_empty(): + return [] + + exprs = custom_signals.build_intraday_expressions(definitions) + if not exprs: + return [] + # 内置 4 信号保留历史列名(不带 csgi_ 前缀) — 存量监控规则零迁移 + for legacy_id in INTRADAY_SIGNAL_FIELDS: + prefixed = custom_signals.intraday_column_name(legacy_id) + if prefixed in exprs: + exprs[legacy_id] = exprs.pop(prefixed) + min_bars_by_col = { + custom_signals.intraday_column_name(d["id"]): int(d.get("min_bars", 0) or 0) + for d in definitions + } + min_bars_by_col.update({ + name: _LEGACY_MIN_BARS for name in INTRADAY_SIGNAL_FIELDS + }) + + evaluated = custom_signals.apply_intraday_edges(frame, exprs) + # min_bars 门槛: 当日已完成 bar 数不足时强制不触发 + evaluated = evaluated.with_columns( + pl.int_range(pl.len()).over(["symbol", "date"]).alias("_bar_idx") + ) + for name, min_bars in min_bars_by_col.items(): + if name in evaluated.columns and min_bars > 0: + evaluated = evaluated.with_columns( + pl.when(pl.col("_bar_idx") + 1 >= min_bars) + .then(pl.col(name)) + .otherwise(False) + .alias(name) + ) + cutoff = _naive_datetime(now) - if cutoff is None: - return [] - cutoff = cutoff.replace(second=0, microsecond=0) - scoped = minute_df.filter(pl.col("symbol").cast(pl.Utf8).is_in(sorted(symbols))) - if scoped.is_empty(): - return [] - results: list[dict[str, Any]] = [] - for part in scoped.partition_by("symbol", maintain_order=False): + signal_cols = [name for name in exprs if name in evaluated.columns] + for part in evaluated.partition_by("symbol", maintain_order=False): part = part.sort("datetime") symbol = str(part["symbol"][0]) - points: list[tuple[datetime, float, float | None]] = [] - cumulative_amount = 0.0 - cumulative_volume = 0.0 - for row in part.iter_rows(named=True): - bar_time = _naive_datetime(row.get("datetime")) - price = _finite(row.get("close")) - volume = _finite(row.get("volume")) - amount = _finite(row.get("amount")) - if bar_time is None or bar_time.date() != cutoff.date() or bar_time >= cutoff or price is None: - continue - if volume is not None and volume > 0 and amount is not None and amount >= 0: - cumulative_volume += volume - cumulative_amount += amount - average = ( - cumulative_amount / (cumulative_volume * 100.0) - if cumulative_volume > 0 and cumulative_amount > 0 - else None - ) - points.append((bar_time, price, average)) - - if not points: + last_time = part["datetime"][-1] + if cutoff is not None and last_time.date() != cutoff.date(): continue - current = points[-1] key = (asset_type, symbol) - last_bar = self._last_bar.get(key) - self._last_bar[key] = current[0] - if last_bar is None or last_bar.date() != current[0].date() or current[0] <= last_bar: + last_seen = self._last_bar.get(key) + self._last_bar[key] = last_time + # 只有出现新 bar 才可能触发; 首次见到该标的只建状态不发信号 + if last_seen is None or last_time <= last_seen or last_time.date() != last_seen.date(): continue - if len(points) < 2: - continue - - previous = points[-2] - baseline = _finite(prev_close.get(symbol)) - avg_up = previous[2] is not None and current[2] is not None and previous[1] <= previous[2] and current[1] > current[2] - avg_down = previous[2] is not None and current[2] is not None and previous[1] >= previous[2] and current[1] < current[2] - zero_up = baseline is not None and baseline > 0 and previous[1] <= baseline and current[1] > baseline - zero_down = baseline is not None and baseline > 0 and previous[1] >= baseline and current[1] < baseline - if avg_up or avg_down or zero_up or zero_down: - results.append({ - "symbol": symbol, - "signal_intraday_avg_cross_up": avg_up, - "signal_intraday_avg_cross_down": avg_down, - "signal_intraday_zero_cross_up": zero_up, - "signal_intraday_zero_cross_down": zero_down, - }) + row = {name: bool(part[name][-1]) for name in signal_cols} + if any(row.values()): + row["symbol"] = symbol + results.append(row) return results @staticmethod def inject(df: pl.DataFrame, signals: list[dict[str, Any]]) -> pl.DataFrame: - existing = [field for field in INTRADAY_SIGNAL_FIELDS if field in df.columns] + """把本分钟触发的信号以布尔列注入 enriched 快照(缺省 False)。""" + fields = sorted(INTRADAY_SIGNAL_FIELDS | {f for s in signals for f in s if f != "symbol"}) + existing = [field for field in fields if field in df.columns] out = df.drop(existing) if existing else df if signals: - out = out.join(pl.DataFrame(signals), on="symbol", how="left") - else: - out = out.with_columns([ - pl.lit(False).alias(field) for field in INTRADAY_SIGNAL_FIELDS - ]) - return out.with_columns([ - pl.col(field).fill_null(False).cast(pl.Boolean).alias(field) - for field in INTRADAY_SIGNAL_FIELDS + cols = sorted({f for s in signals for f in s if f != "symbol"}) + out = out.join(pl.DataFrame(signals).select(["symbol", *cols]), on="symbol", how="left") + out = out.with_columns([ + ( + pl.col(field).fill_null(False).cast(pl.Boolean).alias(field) + if field in out.columns + else pl.lit(False, dtype=pl.Boolean).alias(field) + ) + for field in fields ]) + return out diff --git a/backend/tests/test_intraday_signals.py b/backend/tests/test_intraday_signals.py new file mode 100644 index 0000000..21fa43e --- /dev/null +++ b/backend/tests/test_intraday_signals.py @@ -0,0 +1,371 @@ +"""盘中信号(分钟K)测试 — 特征帧 / 编译 / 旧 4 信号等价 / 引擎注入 / 回放。 + +口径单源的核心保证: 测试内复刻 v1 盘中评估器的累计循环作为黄金参照, +新评估器(v2, 特征帧 + 表达式)必须与它对任意序列产出完全一致的触发集合。 +""" +from __future__ import annotations + +import json +from datetime import date, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace + +import polars as pl +import pytest + +from app.strategy import custom_signals +from app.strategy.intraday_features import build_feature_frame +from app.strategy.intraday_signals import IntradaySignalEvaluator, uses_intraday_signals + +SYMBOL = "000001.SZ" +DAY = date(2026, 9, 4) + + +def _bars( + prices: list[float], + volumes: list[float] | None = None, + start: datetime = datetime(2026, 9, 4, 9, 30), + symbol: str = SYMBOL, +) -> pl.DataFrame: + """构造规范的上午分钟序列: volume 单位为手, amount = close*volume*100(元)。""" + volumes = volumes or [100.0] * len(prices) + # 跳过午休: 09:30+120 根后进入 13:00 + times = [] + for i in range(len(prices)): + if i < 120: + times.append(start + timedelta(minutes=i)) + else: + times.append(datetime(start.year, start.month, start.day, 13, 0) + timedelta(minutes=i - 120)) + return pl.DataFrame({ + "symbol": [symbol] * len(prices), + "datetime": times, + "open": prices, + "high": [p * 1.01 for p in prices], + "low": [p * 0.99 for p in prices], + "close": prices, + "volume": volumes, + "amount": [p * v * 100.0 for p, v in zip(prices, volumes, strict=True)], + }) + + +# ══ 特征帧 ═══════════════════════════════════════════════ + + +def test_feature_frame_vwap_and_pct(): + df = _bars([10.0, 10.2, 9.8]) + frame = build_feature_frame(df, prev_close={SYMBOL: 10.0}) + # vwap = 累计成交额 / (累计量(手)*100) + assert frame["vwap"][2] == pytest.approx((10.0 + 10.2 + 9.8) / 3) + assert frame["pct_vs_prev_close"][2] == pytest.approx(9.8 / 10.0 - 1) + assert frame["pct_from_open"][2] == pytest.approx(9.8 / 10.0 - 1) + assert frame["price"][2] == 9.8 + # 缺昨收 → null + frame2 = build_feature_frame(df) + assert frame2["pct_vs_prev_close"][2] is None + + +def test_feature_frame_vol_ratio_windows_and_nulls(): + # 量比: 当前N根量和 / 此前N根量和; 前 2N-1 根为 null + prices = [10.0] * 12 + volumes = [100.0, 100.0, 100.0, 100.0, 100.0, 300.0, 100.0, 100.0, 900.0, 100.0, 100.0, 100.0] + frame = build_feature_frame(_bars(prices, volumes)) + vr1 = frame["vol_ratio_1m_today"].to_list() + assert vr1[0] is None # 无前值 + assert vr1[1] == pytest.approx(100.0 / 100.0) + assert vr1[5] == pytest.approx(300.0 / 100.0) + assert vr1[8] == pytest.approx(900.0 / 100.0) + vr3 = frame["vol_ratio_3m_today"].to_list() + assert vr3[0] is None and vr3[4] is None # 3+3-1=5 根前不足 + # 第 6 根(idx5): [3,4,5]=100+100+300 vs [0,1,2]=300 + assert vr3[5] == pytest.approx(500.0 / 300.0) + # 前窗口量为 0 → null(不伪装成 0) + zero_start = build_feature_frame(_bars([10.0, 10.0, 10.0], [0.0, 0.0, 500.0])) + assert zero_start["vol_ratio_1m_today"][2] is None + + +def test_feature_frame_rolling_window_never_crosses_lunch(): + # 上午最后一根的 5 分钟窗口不回看跨日; 下午重新预热 + n_am, n_pm = 8, 4 + am = _bars([10.0] * n_am, [100.0] * n_am) # 09:30-09:37 + pm = _bars([10.0] * n_pm, [100.0] * n_pm, start=datetime(2026, 9, 4, 13, 0)) + frame = build_feature_frame(pl.concat([am, pm])) + vr = frame["vol_ratio_3m_today"].to_list() + # 上午 8 根: idx5 起 3/3 窗口成立(idx5,6,7 非 null); 下午 4 根全部 null(窗口不足) + assert vr[5] is not None and vr[7] is not None + assert all(v is None for v in vr[n_am:n_am + n_pm]) + + +def test_feature_frame_open_30m_and_day_extremes(): + frame = build_feature_frame(_bars([10.0 + 0.1 * i for i in range(32)])) + o30h = frame["open_30m_high_dist"].to_list() + assert all(v is None for v in o30h[:29]) # 开盘未满 30 分钟 + # _bars 里 high = price*1.01 → 第 30 根 close/开盘30分钟最高 = 1/1.01 + assert o30h[29] == pytest.approx(1.0 / 1.01 - 1) + assert o30h[31] == pytest.approx(13.1 / (12.9 * 1.01) - 1) + assert frame["day_high_dist"][5] == pytest.approx(1.0 / 1.01 - 1) + # 累计最低 = 首根 low(10.0*0.99), close_5 = 10.5 + assert frame["day_low_dist"][5] == pytest.approx(10.5 / 9.9 - 1) + + +def test_feature_frame_cutoff_drops_incomplete_bar(): + df = _bars([10.0, 10.1, 10.2]) + frame = build_feature_frame(df, cutoff=datetime(2026, 9, 4, 9, 32)) + assert frame.height == 2 + + +# ══ 校验与编译 ═══════════════════════════════════════════ + + +def _intraday_sig(**overrides) -> dict: + sig = { + "id": "test_sig", "name": "测试", "kind": "entry", + "timeframe": "intraday", + "conditions": [{"left": "price", "op": "cross_up", "right": "field:vwap"}], + } + sig.update(overrides) + return sig + + +def test_validate_intraday_accepts_and_rejects(): + custom_signals.validate(_intraday_sig()) + custom_signals.validate(_intraday_sig( + conditions=[{"left": "vol_ratio_3m_today", "op": ">", "right": "3"}], + min_bars=10, + )) + with pytest.raises(ValueError, match="盘中字段"): + custom_signals.validate(_intraday_sig(conditions=[{"left": "ma5", "op": ">", "right": "1"}])) + with pytest.raises(ValueError, match="运算符"): + custom_signals.validate(_intraday_sig(conditions=[{"left": "price", "op": "~", "right": "1"}])) + with pytest.raises(ValueError, match="日期偏移"): + custom_signals.validate(_intraday_sig( + conditions=[{"left": "price", "op": ">", "right": "1", "leftDays": 1}] + )) + with pytest.raises(ValueError, match="min_bars"): + custom_signals.validate(_intraday_sig(min_bars=500)) + with pytest.raises(ValueError, match="timeframe"): + custom_signals.validate(_intraday_sig(timeframe="weekly")) + + +def test_build_intraday_expressions_edge_semantics(): + # 价格下探再上穿 vwap: 上升沿恰好只在上穿那根 bar 为 true + prices = [10.0, 10.0, 10.0, 10.0, 9.8, 9.7, 9.6, 9.6, 9.6, 10.0, 10.2, 10.3] + frame = build_feature_frame(_bars(prices)) + exprs = custom_signals.build_intraday_expressions([_intraday_sig()]) + col = custom_signals.intraday_column_name("test_sig") + assert col == "csgi_test_sig" + out = custom_signals.apply_intraday_edges(frame, exprs) + fired = [i for i, v in enumerate(out[col].to_list()) if v] + assert len(fired) == 1 + # idx8: 9.6 < vwap; idx9: 10.0 > vwap → 上穿在第 9 根(0 起) + assert fired[0] == 9 + + # 比较条件的上升沿: 持续满足只触发一次(本序列条件自首根即为真 → 只在回升的 idx9 触发) + sig2 = _intraday_sig(id="test_state", conditions=[{"left": "price", "op": ">", "right": "9.65"}]) + exprs2 = custom_signals.build_intraday_expressions([sig2]) + out2 = custom_signals.apply_intraday_edges(frame, exprs2) + fired2 = [i for i, v in enumerate(out2["csgi_test_state"].to_list()) if v] + assert fired2 == [9] + + # 振荡序列: 每次 false→true 各触发一次 + osc = build_feature_frame(_bars([10.0, 9.6, 10.0, 9.6, 10.0])) + exprs3 = custom_signals.build_intraday_expressions([sig2]) + out3 = custom_signals.apply_intraday_edges(osc, exprs3) + assert [i for i, v in enumerate(out3["csgi_test_state"].to_list()) if v] == [2, 4] + + +# ══ v1 黄金等价: 旧累计循环算法作参照 ═════════════════════ + + +def _legacy_v1_triggers(prices: list[float], prev_close: float | None) -> dict[str, int]: + """复刻 v1 IntradaySignalEvaluator 的判定(逐 bar 累计, 边沿触发)。""" + cum_vol = cum_amt = 0.0 + fired: dict[str, int] = {} + prev_price = prev_vwap = None + for i, p in enumerate(prices): + cum_vol += 100.0 + cum_amt += p * 100.0 * 100.0 + vwap = cum_amt / (cum_vol * 100.0) + if i >= 1: + if prev_price <= prev_vwap and p > vwap: + fired.setdefault("avg_up", i) + if prev_price >= prev_vwap and p < vwap: + fired.setdefault("avg_down", i) + if prev_close and prev_price <= prev_close and p > prev_close: + fired.setdefault("zero_up", i) + if prev_close and prev_price >= prev_close and p < prev_close: + fired.setdefault("zero_down", i) + prev_price, prev_vwap = p, vwap + return fired + + +@pytest.mark.parametrize("prices,prev_close", [ + ([10.0, 10.0, 10.0, 10.0, 9.8, 9.7, 9.6, 9.6, 9.6, 10.0, 10.2, 10.3], 10.0), + ([9.0, 9.1, 9.2, 9.3, 9.4, 9.3, 9.2, 9.1, 9.0, 8.9, 8.8, 8.7], 9.25), + ([10.0, 9.9, 10.1, 9.8, 10.2, 9.7, 10.3, 9.6, 10.4, 9.5, 10.5, 9.4], 9.95), +]) +def test_evaluator_matches_legacy_v1(prices, prev_close): + """新评估器逐 bar 喂入, 触发时点必须与 v1 算法完全一致。""" + df = _bars(prices) + evaluator = IntradaySignalEvaluator() + got: dict[str, int] = {} + for t in range(1, len(prices) + 1): + now = df["datetime"][t - 1] + timedelta(minutes=1) + rows = evaluator.evaluate( + df.head(t), symbols={SYMBOL}, prev_close={SYMBOL: prev_close}, + asset_type="stock", now=now, + ) + for r in rows: + if r.get("signal_intraday_avg_cross_up"): + got.setdefault("avg_up", t - 1) + if r.get("signal_intraday_avg_cross_down"): + got.setdefault("avg_down", t - 1) + if r.get("signal_intraday_zero_cross_up"): + got.setdefault("zero_up", t - 1) + if r.get("signal_intraday_zero_cross_down"): + got.setdefault("zero_down", t - 1) + assert got == _legacy_v1_triggers(prices, prev_close) + + +def test_evaluator_no_refire_without_new_bar(): + prices = [10.0, 9.0, 10.5] + df = _bars(prices) + evaluator = IntradaySignalEvaluator() + # 逐根喂入: 首轮只建状态; 新 bar 出现才可能触发; 同批 bar 重跑不重复触发 + fired: list[dict] = [] + for t in range(1, len(prices) + 1): + now = df["datetime"][t - 1] + timedelta(minutes=1) + fired += evaluator.evaluate( + df.head(t), symbols={SYMBOL}, prev_close={SYMBOL: 10.0}, asset_type="stock", now=now, + ) + assert fired # 9.0 下穿 / 10.5 上穿均有触发 + now3 = df["datetime"][2] + timedelta(minutes=1) + again = evaluator.evaluate(df, symbols={SYMBOL}, prev_close={SYMBOL: 10.0}, asset_type="stock", now=now3) + assert again == [] # 无新 bar → 不重复触发 + + +def test_evaluator_custom_csgi_signal_and_inject(): + sig = _intraday_sig(id="my_intraday", conditions=[{"left": "price", "op": "cross_up", "right": 10.05}]) + prices = [10.0, 9.9, 9.8, 10.1, 10.2, 10.3] + df = _bars(prices) + evaluator = IntradaySignalEvaluator() + fired_rows = [] + for t in range(1, len(prices) + 1): + now = df["datetime"][t - 1] + timedelta(minutes=1) + fired_rows += evaluator.evaluate( + df.head(t), symbols={SYMBOL}, prev_close={}, asset_type="stock", + now=now, signals=[sig], + ) + # 上穿 10.05 发生在 idx3 (10.1) + assert any(r.get("csgi_my_intraday") for r in fired_rows) + assert len([r for r in fired_rows if r.get("csgi_my_intraday")]) == 1 + + # inject 的契约是"单桶结果": 只传最后一个触发桶的行 + last_bucket = fired_rows[-1:] if fired_rows else [] + enriched = pl.DataFrame({"symbol": [SYMBOL, "999999.SZ"], "close": [10.0, 5.0]}) + injected = evaluator.inject(enriched, last_bucket) + assert injected.height == 2 # 单桶单行, join 不膨胀 + assert injected["csgi_my_intraday"].to_list() == [True, False] # 未触发标的补 False + assert set(injected.columns) >= {"signal_intraday_avg_cross_up", "signal_intraday_avg_cross_down"} + + +def test_uses_intraday_signals_matches_csgi_fields(): + assert uses_intraday_signals({"conditions": [{"op": "truth", "field": "signal_intraday_avg_cross_up"}]}) + assert uses_intraday_signals({"conditions": [{"op": "truth", "field": "csgi_my_intraday"}]}) + assert not uses_intraday_signals({"conditions": [{"op": "truth", "field": "csg_daily_sig"}]}) + assert not uses_intraday_signals({"conditions": [{"op": ">", "field": "close", "value": 1}]}) + + +# ══ 引擎注入 + 加载缓存 ═══════════════════════════════════ + + +def _engine(tmp_path: Path): + from app.strategy.engine import StrategyEngine + custom_dir = tmp_path / "strategies" / "custom" + custom_dir.mkdir(parents=True, exist_ok=True) + return StrategyEngine(strategy_dirs=[custom_dir]) + + +def test_engine_injects_csgi_columns_into_minute_frame(tmp_path: Path): + sig = _intraday_sig(id="eng_sig", conditions=[{"left": "price", "op": "cross_up", "right": "field:vwap"}]) + sig_dir = tmp_path / "user_data" / "custom_signals" + sig_dir.mkdir(parents=True, exist_ok=True) + (sig_dir / "eng_sig.json").write_text(json.dumps(sig), encoding="utf-8") + + engine = _engine(tmp_path) + assert engine._user_data_dir() == tmp_path + prices = [10.0, 10.0, 9.0, 9.0, 10.5, 10.6] + injected = engine._inject_intraday_signal_columns(_bars(prices)) + assert "csgi_eng_sig" in injected.columns + fired = [i for i, v in enumerate(injected["csgi_eng_sig"].to_list()) if v] + assert fired == [4] # idx4 上穿 vwap + # 定义缓存指纹失效: 改文件后引擎立刻读到新定义 + (sig_dir / "eng_sig.json").write_text(json.dumps( + _intraday_sig(id="eng_sig", conditions=[{"left": "price", "op": "<", "right": "9.5"}]), + ), encoding="utf-8") + injected2 = engine._inject_intraday_signal_columns(_bars(prices)) + assert injected2["csgi_eng_sig"].to_list()[2] is True or injected2["csgi_eng_sig"].to_list()[2] == True # noqa: E712 + + +def test_engine_skips_injection_without_definitions(tmp_path: Path): + engine = _engine(tmp_path) + df = _bars([10.0, 10.1]) + out = engine._inject_intraday_signal_columns(df) + assert out is df or out.equals(df) + + +# ══ 回放 API ═════════════════════════════════════════════ + + +def _seed_repo(tmp_path: Path): + from app.tickflow.repository import DataStore, KlineRepository + + def minute(day: str, prices: list[float]): + part = tmp_path / "kline_minute" / f"date={day}" / "part.parquet" + part.parent.mkdir(parents=True, exist_ok=True) + _bars(prices, start=datetime.fromisoformat(f"{day}T09:30:00")).with_columns( + pl.col("datetime").cast(pl.Datetime("us")) + ).write_parquet(part) + + def daily(day: str, closes: dict[str, float]): + part = tmp_path / "kline_daily" / f"date={day}" / "part.parquet" + part.parent.mkdir(parents=True, exist_ok=True) + pl.DataFrame({"symbol": list(closes), "close": list(closes.values())}).write_parquet(part) + + minute("2026-09-03", [10.0, 9.9, 9.8, 9.9, 10.0, 10.1]) + minute("2026-09-04", [10.0, 10.0, 9.0, 9.0, 10.5, 10.6]) + daily("2026-09-02", {SYMBOL: 9.95}) + daily("2026-09-03", {SYMBOL: 10.0}) + return KlineRepository(DataStore(tmp_path)) + + +def test_intraday_replay_endpoint(tmp_path: Path): + from app.api.signals import IntradayReplayRequest, intraday_replay + + repo = _seed_repo(tmp_path) + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(repo=repo))) + + sig = _intraday_sig(id="replay_sig") + sig_dir = tmp_path / "user_data" / "custom_signals" + sig_dir.mkdir(parents=True, exist_ok=True) + (sig_dir / "replay_sig.json").write_text(json.dumps(sig), encoding="utf-8") + daily_sig = { + "id": "daily_sig", "name": "日线信号", "kind": "entry", + "conditions": [{"left": "close", "op": ">", "right": "1"}], + } + (sig_dir / "daily_sig.json").write_text(json.dumps(daily_sig), encoding="utf-8") + + result = intraday_replay(IntradayReplayRequest( + signal_id="replay_sig", start_date="2026-09-03", end_date="2026-09-04", + symbols=[SYMBOL], + ), request) + assert result["days_scanned"] == 2 + times = [(t["date"], t["time"]) for t in result["triggers"]] + # 09-03: 10.0→9.x→10.0 上穿在 idx4(09:34); 09-04: 上穿在 idx4(09:34) + assert times == [("2026-09-03", "09:34:00"), ("2026-09-04", "09:34:00")] + + # 日线信号不可回放 + with pytest.raises(Exception, match="盘中"): + intraday_replay(IntradayReplayRequest( + signal_id="daily_sig", start_date="2026-09-03", end_date="2026-09-04", + symbols=[SYMBOL], + ), request) diff --git a/docs/features.md b/docs/features.md index e47f032..94ba9ae 100644 --- a/docs/features.md +++ b/docs/features.md @@ -89,6 +89,16 @@ --- +## 🚦 盘中信号(分钟K) + +信号库支持 `timeframe=intraday` 的盘中信号:基于当日**已完成**的一分钟 K 计算特征——分时均价、相对昨收/开盘涨跌幅、1/3/5 分钟放量比(今日基准,滚动窗口、每分钟刷新)、距当日与开盘 30 分钟高低点。条件支持数值比较与 `cross_up / cross_down`(上穿/下穿另一特征或阈值);信号输出为**当日条件上升沿**(条件 false→true 的那一根 bar 才为真,首根不触发;窗口不足、基准为零、缺昨收时特征为 null,判 false 绝不误报)。 + +- **口径单源**:监控评估、分钟策略执行、分钟回测、历史回放共用同一特征构造器(`app/strategy/intraday_features.py`),四条路径对同一段分钟数据产出完全一致(含防未来函数:只用已收盘 bar,滚动窗口不跨午休/跨日); +- **消费入口**:监控规则字段引用 `csgi_*` 列(与内置 4 个分时穿越信号并列,旧信号列名不变、存量规则零迁移);分钟策略(`timeframes=['1m']`)在 `filter_minute_history` 帧上直接引用 `csgi_*` 列(引擎单点注入);日线策略引用盘中信号会显式报错,不支持; +- **回放验证**:`POST /api/custom-signals/intraday/replay` 用本地历史分钟 K 重放信号触发时点(区间 ≤60 天、标的 ≤200),先验证再配置监控,不消耗盘中数据能力; +- **能力门槛**:盘中评估复用分时信号监控门控——分钟 K 能力(订阅池,单轮标的上限默认 100)或全量分钟能力(落盘服务健康时读本地分区,覆盖全市场);回放与分钟回测只需本地已有分钟 K 历史。 + + ## 📡 监控中心(Monitor) 统一规则引擎,一个页面管理**四类监控**: