diff --git a/backend/app/api/monitor_rules.py b/backend/app/api/monitor_rules.py index 0d9286d..139296e 100644 --- a/backend/app/api/monitor_rules.py +++ b/backend/app/api/monitor_rules.py @@ -107,6 +107,11 @@ class RuleModel(BaseModel): # ladder 专属 (连板梯队封单监控) metric: str = "sealed_vol" # sealed_vol=封单量(手) | sealed_amount=封单额(元) threshold: float = 0 # 封单 <= 此值时报警 (原始单位: 量=手, 额=元) + # volume_delta 专属 (轮询放量监控): 相邻两次全市场快照的成交量增量 + threshold_volume: float = 9000 # 单轮增量 >= 此值(手)时报警 + threshold_amount: float = 1e6 # metric=amount 时: 单轮增量 >= 此值(元)时报警 + # 基础过滤 (与策略 basic_filter 语义对齐): 值为 null 表示不过滤 + basic_filter: dict = {} # ── 字段选项 ───────────────────────────────────────────── @@ -160,6 +165,7 @@ def get_options(request: Request): {"key": "strategy", "label": "策略监控"}, {"key": "abnormal", "label": "异动监控"}, {"key": "sector", "label": "板块监控"}, + {"key": "volume_delta", "label": "轮询放量"}, ], "scopes": [ {"key": "symbols", "label": "指定标的"}, diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 48f94f8..fade3c4 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -28,7 +28,7 @@ import threading import time from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager -from datetime import date, time as dt_time +from datetime import date, datetime, time as dt_time import polars as pl @@ -221,6 +221,16 @@ class QuoteService: self._final_sync_done: set[tuple[date, str]] = set() self._final_sync_failed: dict[tuple[date, str], str] = {} self._holiday_active = False # 交易日探针当前是否判休市 (日志去重) + # 轮询放量 (volume_delta 规则): 上一轮全市场股票快照的 (累计成交量[手], 累计成交额[元])。 + # 每轮全量快照后更新 (含非连续竞价时段, 保证 13:00 恢复时 prev 是 12:59 + # 而非 11:30); 跨交易日清空; cur < prev (数据源重置) 时丢弃该轮差值。 + self._prev_stock_volume: dict[str, tuple[float, float]] | None = None + self._prev_volume_fetched_at: float | None = None # epoch 毫秒 + self._prev_volume_date: date | None = None + # 最近一轮的有效差值 (vol_delta[手], amt_delta[元]) - 仅连续竞价时段内、 + # prev 不早于本时段开盘时计算 + self._volume_delta: dict[str, tuple[float, float]] = {} + self._volume_delta_span_s: float = 0.0 # ================================================================ # 生命周期 @@ -725,6 +735,9 @@ class QuoteService: _persist_last_fetch(fetched_at) logger.info("行情刷新: %d 只股票, %d 只ETF, %d 只指数, 耗时 %.0fms", len(stock_records), len(etf_records), len(index_records), fetch_ms) + # 轮询放量状态更新 (volume_delta 规则的差值来源) + self._update_volume_delta(stock_records, fetched_at) + # ---- 写 kline_daily (不复权原始价格, 只有 OHLCV) ---- daily_df = self._build_daily(stock_records) if not daily_df.is_empty() and self._repo: @@ -1005,6 +1018,8 @@ class QuoteService: eval_df = enriched_today if engine.has_rule_type("ladder"): eval_df = self._inject_sealed_vol(enriched_today, enriched_date) + if engine.has_rule_type("volume_delta"): + eval_df = self._inject_volume_delta(eval_df) eval_df = self._inject_intraday_signals(eval_df, engine, "stock") rule_events = engine.evaluate(eval_df, asset_type="stock") if engine.consume_strategy_result_updates(): @@ -1091,7 +1106,8 @@ class QuoteService: "window_change_pct", "coverage_ratio", "valid_count", "total_count", "up_count", "down_count", "leader", "abnormal_window", "abnormal_value", "abnormal_threshold", - "abnormal_closeness", + "abnormal_closeness", "volume_delta", "volume_delta_span", + "volume_delta_amount", ): if key in ev: alert[key] = ev[key] @@ -1240,6 +1256,88 @@ class QuoteService: ) return self._intraday_signal_evaluator.inject(enriched, signals) + @staticmethod + def _continuous_session_start_ms() -> float: + """当前连续竞价时段的起点 (北京时间 9:30 或 13:00) 的 epoch 毫秒。""" + now = cn_now() + start_time = dt_time(13, 0) if now.time() >= dt_time(13, 0) else dt_time(9, 30) + return datetime.combine(now.date(), start_time, tzinfo=now.tzinfo).timestamp() * 1000.0 + + def _update_volume_delta(self, stock_records: list[dict], fetched_at_ms: float) -> None: + """全市场相邻两次快照的股票累计成交量差值 (手), 供 volume_delta 规则。 + + - prev 每轮都更新 (含非连续竞价时段); 差值只在连续竞价时段内计算 + - 开盘保护: prev 早于本时段起点 (9:30/13:00) 时本轮差值无效 -- 避免 + 9:25 集合竞价撮合量 / 午休缺口被当成"突然放量" + - cur < prev (数据源重置/口径跳变) 的个股丢弃差值; 跨交易日清空 + """ + today = cn_today() + if self._prev_volume_date != today: + self._prev_stock_volume = None + self._prev_volume_fetched_at = None + self._prev_volume_date = today + self._volume_delta = {} + + cur: dict[str, tuple[float, float]] = {} + for r in stock_records: + sym = r.get("symbol") + vol = r.get("volume") + amt = r.get("amount") + if not sym or not isinstance(vol, (int, float)): + continue + cur[str(sym)] = ( + float(vol), + float(amt) if isinstance(amt, (int, float)) else 0.0, + ) + + prev = self._prev_stock_volume + prev_ts = self._prev_volume_fetched_at + if ( + prev is not None + and prev_ts is not None + and self._is_continuous_trading() + and prev_ts >= self._continuous_session_start_ms() + ): + delta = { + sym: (v - prev[sym][0], a - prev[sym][1]) + for sym, (v, a) in cur.items() + if sym in prev and v >= prev[sym][0] and a >= prev[sym][1] and v - prev[sym][0] > 0 + } + self._volume_delta = delta + self._volume_delta_span_s = max((fetched_at_ms - prev_ts) / 1000.0, 0.001) + else: + self._volume_delta = {} + + self._prev_stock_volume = cur + self._prev_volume_fetched_at = fetched_at_ms + + def _inject_volume_delta(self, enriched_today: pl.DataFrame) -> pl.DataFrame: + """把最近一轮快照差值作为临时列注入 enriched 副本。 + + _volume_delta (手) / _volume_delta_amount (元) / _volume_delta_span (秒, 快照间隔)。 + 无有效差值 (首轮/开盘保护/暂停后恢复) 时返回原 df, 规则安全降级不触发。 + """ + try: + delta = self._volume_delta + if not delta: + return enriched_today + span = self._volume_delta_span_s + delta_df = pl.DataFrame({ + "symbol": list(delta.keys()), + "_volume_delta": [v for v, _ in delta.values()], + "_volume_delta_amount": [a for _, a in delta.values()], + "_volume_delta_span": [span] * len(delta), + }) + drop_cols = [ + c for c in ("_volume_delta", "_volume_delta_amount", "_volume_delta_span") + if c in enriched_today.columns + ] + df = enriched_today.drop(drop_cols) if drop_cols else enriched_today + return df.join(delta_df, on="symbol", how="left") + except Exception as e: # noqa: BLE001 + logger.debug("快照差值注入失败 (volume_delta 规则将不触发): %s", e) + return enriched_today + def _inject_sealed_vol(self, enriched_today: pl.DataFrame, enriched_date) -> pl.DataFrame: """从 depth_service 取封单量, 作为临时列 _sealed_vol 注入 enriched 副本。 @@ -1302,7 +1400,7 @@ class QuoteService: source_labels = { "strategy": "策略", "signal": "信号", "price": "价格", "market": "异动", "ladder": "连板梯队", - "sector": "板块", + "sector": "板块", "volume_delta": "放量", } rules = engine.rules if engine is not None else {} enqueued = 0 diff --git a/backend/app/strategy/monitor.py b/backend/app/strategy/monitor.py index 162b324..a6afdfd 100644 --- a/backend/app/strategy/monitor.py +++ b/backend/app/strategy/monitor.py @@ -46,7 +46,8 @@ _SIGNAL_CN: dict[str, str] = { # 行情字段 "close": "收盘价", "open": "开盘价", "high": "最高价", "low": "最低价", "change_pct": "涨跌幅", "change_amount": "涨跌额", "amplitude": "振幅", - "turnover_rate": "换手率", "volume": "成交量", "amount": "成交额", + "turnover_rate": "换手率", "volume": "成交量", "amount": "成交额", + "_volume_delta": "轮询成交量差值(手)", "_sealed_vol": "封单量(手)", # 均线 "ma5": "MA5", "ma10": "MA10", "ma20": "MA20", "ma30": "MA30", "ma60": "MA60", "ema5": "EMA5", "ema10": "EMA10", "ema20": "EMA20", @@ -983,6 +984,9 @@ class MonitorRuleEngine: elif rtype == "ladder": # 连板梯队封单监控: 独立处理 (需带预警封单值, 走专属 message) return self._evaluate_ladder(scoped, rule, now) + elif rtype == "volume_delta": + # 轮询放量监控: 相邻两次全市场快照的成交量差值, 独立处理走专属 message + return self._evaluate_volume_delta(scoped, rule, now) else: # signal / price / market: 通用条件匹配 for sym, name, price, pct, hit_sigs in self._match_conditions(scoped, rule): @@ -1364,6 +1368,140 @@ class MonitorRuleEngine: results.append((sym, name, price, pct, hit_sigs)) return results + @staticmethod + def _volume_delta_basic_mask(df: pl.DataFrame, bf: dict, name_map: dict[str, str]) -> pl.Expr | None: + """轮询放量基础过滤掩码 (与策略 basic_filter 语义对齐, 字段缺失时该项跳过)。 + + 支持: price_min/max (收盘价), market_cap_min (总市值=close x total_shares), + float_cap_min/max (流通市值), amount_min (当日累计成交额), exclude_st (名称含 ST)。 + """ + masks: list[pl.Expr] = [] + if bf.get("price_min") is not None: + masks.append(pl.col("close") >= float(bf["price_min"])) + if bf.get("price_max") is not None: + masks.append(pl.col("close") <= float(bf["price_max"])) + if bf.get("amount_min") is not None and "amount" in df.columns: + masks.append(pl.col("amount") >= float(bf["amount_min"])) + if bf.get("market_cap_min") is not None and "total_shares" in df.columns: + masks.append((pl.col("close") * pl.col("total_shares")) >= float(bf["market_cap_min"])) + if bf.get("float_cap_min") is not None and "float_shares" in df.columns: + masks.append((pl.col("close") * pl.col("float_shares")) >= float(bf["float_cap_min"])) + if bf.get("float_cap_max") is not None and "float_shares" in df.columns: + masks.append((pl.col("close") * pl.col("float_shares")) <= float(bf["float_cap_max"])) + if bf.get("exclude_st") and name_map: + st_symbols = [ + sym for sym, name in name_map.items() + if name and "ST" in str(name).upper() + ] + if st_symbols: + masks.append(~pl.col("symbol").is_in(st_symbols)) + if not masks: + return None + return pl.all_horizontal(masks) + + def _evaluate_volume_delta(self, scoped: pl.DataFrame, rule: dict, now: float) -> list[dict]: + """评估轮询放量监控: 相邻两次全市场快照的成交量/成交额差值。 + + 差值列 _volume_delta(手)/_volume_delta_amount(元)/间隔列 _volume_delta_span + 由 quote_service 评估前注入。metric=volume 按手数、amount 按金额比较阈值; + basic_filter 先行过滤 (股价/市值/成交额/ST, 与策略 basic_filter 语义对齐)。 + 命中 >5 只时合并为一条批量事件防刷屏。 + """ + if "_volume_delta" not in scoped.columns: + return [] # 无差值数据 (首轮/开盘保护/非全市场轮询), 安全降级 + + metric = rule.get("metric", "volume") + if metric == "amount" and "_volume_delta_amount" in scoped.columns: + cmp_col, threshold = "_volume_delta_amount", rule.get("threshold_amount", 1e6) + th_text = f"{threshold / 1e4:,.0f} 万元" + else: + cmp_col, threshold = "_volume_delta", rule.get("threshold_volume", 9000) + th_text = f"{threshold:,.0f} 手" + + cooldown = rule.get("cooldown_seconds", 300) + severity = rule.get("severity", "warn") + span_s = 0.0 + if "_volume_delta_span" in scoped.columns and scoped.height > 0: + v = scoped["_volume_delta_span"][0] + span_s = float(v) if v is not None else 0.0 + span_text = f" (间隔 {span_s:.0f}s)" if span_s > 0 else "" + + candidate = scoped + bf = rule.get("basic_filter") or {} + if bf: + mask = self._volume_delta_basic_mask(candidate, bf, self._name_map) + if mask is not None: + candidate = candidate.filter(mask) + + hit = candidate.filter( + pl.col(cmp_col).is_not_null() & (pl.col(cmp_col) >= threshold) + ).sort(cmp_col, descending=True) + if hit.is_empty(): + return [] + hit_rows = list(hit.iter_rows(named=True)) + + def _name_of(row: dict) -> str: + sym = row.get("symbol", "") + return row.get("name") or self._name_map.get(sym) or sym + + def _fmt(v) -> str: + if metric == "amount": + return f"{v / 1e4:,.0f} 万元" + return f"{v:,.0f} 手" + + def _event(symbol: str, name: str, message: str, *, delta=None, price=None, pct=None) -> dict: + ev = { + "ts": int(now * 1000), + "rule_id": rule["id"], + "rule_name": rule.get("name", ""), + "source": "volume_delta", + "type": "轮询放量", + "symbol": symbol, + "name": name, + "message": message, + "price": price, + "change_pct": pct, + "signals": [], + "severity": severity, + "conditions": [], + "logic": "and", + "volume_delta": delta, + "volume_delta_span": round(span_s, 1), + } + if metric == "amount": + ev["volume_delta_amount"] = delta + return ev + + if len(hit_rows) > 5: + top = "、".join(_name_of(r) for r in hit_rows[:8]) + suffix = "等" if len(hit_rows) > 8 else "" + message = ( + f"放量 · 单轮增量 >= {th_text}{span_text} · " + f"共 {len(hit_rows)} 只: {top}{suffix}" + ) + key = (rule["id"], "_volume_delta_batch", "volume_delta") + last = self._last_fire.get(key) + if last is not None and (now - last) < cooldown: + return [] + self._last_fire[key] = now + return [_event("", "", message)] + + events: list[dict] = [] + for row in hit_rows: + sym = row.get("symbol", "") + key = (rule["id"], sym, "volume_delta") + last = self._last_fire.get(key) + if last is not None and (now - last) < cooldown: + continue + self._last_fire[key] = now + delta = row.get(cmp_col) + message = f"放量 · 单轮增量 {_fmt(delta)} >= {th_text}{span_text}" + events.append(_event( + sym, _name_of(row), message, + delta=delta, price=row.get("close"), pct=row.get("change_pct"), + )) + return events + def _evaluate_ladder(self, scoped: pl.DataFrame, rule: dict, now: float) -> list[dict]: """评估连板梯队封单监控规则。 diff --git a/backend/app/strategy/monitor_rules.py b/backend/app/strategy/monitor_rules.py index 35cd0bc..2d6571b 100644 --- a/backend/app/strategy/monitor_rules.py +++ b/backend/app/strategy/monitor_rules.py @@ -28,7 +28,7 @@ logger = logging.getLogger(__name__) # ── 常量 ──────────────────────────────────────────────── ID_RE = re.compile(r"^[a-z0-9_]{1,40}$") -RULE_TYPES = {"strategy", "signal", "price", "market", "ladder", "sector", "abnormal"} +RULE_TYPES = {"strategy", "signal", "price", "market", "ladder", "sector", "abnormal", "volume_delta"} SCOPES = {"symbols", "all", "sector", "watchlist_group"} LOGICS = {"and", "or"} DIRECTIONS = {"entry", "exit", "both"} @@ -45,6 +45,19 @@ SECTOR_WINDOWS = {1, 3, 5, 10, 15} # abnormal 规则 (异动边缘): 接近度方向 / 关注窗口 ABNORMAL_DIRECTIONS = {"up", "down", "both"} ABNORMAL_WINDOWS = {"any", "3d", "10d", "30d"} +# volume_delta 规则 (轮询放量): 阈值口径 (手数 / 成交额) +VD_METRICS = {"volume", "amount"} +# volume_delta 基础过滤默认值 (与策略 DEFAULT_BASIC_FILTER 核心子集对齐: +# 价格 3-300 元, 总市值 >=10 亿, 当日成交额 >=2000 万, 剔除 ST) +VD_BASIC_FILTER_DEFAULTS: dict = { + "price_min": 3, + "price_max": 300, + "market_cap_min": 10e8, + "float_cap_min": None, + "float_cap_max": None, + "amount_min": 0.2e8, + "exclude_st": True, +} # 布尔信号列前缀 (op=truth 时 field 取这些) _SIGNAL_PREFIXES = ("signal_", "csg_") @@ -190,6 +203,39 @@ def validate(rule: dict) -> None: threshold_pct = rule.get("threshold_pct") if not isinstance(threshold_pct, (int, float)) or not 1 <= threshold_pct <= 150: raise ValueError("异动接近度阈值必须是 1 到 150 之间的百分比数字") + elif rule.get("type") == "volume_delta": + # 轮询放量监控: 相邻两次全市场快照的成交量/成交额差值, 不用 conditions + if rule.get("asset_type", "stock") != "stock": + raise ValueError("轮询放量监控仅支持个股 (依赖全市场股票快照)") + if rule.get("scope", "all") == "sector": + raise ValueError("轮询放量监控不支持板块作用域") + if rule.get("metric", "volume") not in VD_METRICS: + raise ValueError(f"metric 必须是 {VD_METRICS} 之一 (volume=手数, amount=金额)") + if rule.get("metric", "volume") == "amount": + thr = rule.get("threshold_amount") + if isinstance(thr, bool) or not isinstance(thr, (int, float)) or not math.isfinite(thr) or thr < 1: + raise ValueError("threshold_amount 必须是 >=1 的数字 (单轮成交额增量, 单位元)") + else: + thr = rule.get("threshold_volume") + if isinstance(thr, bool) or not isinstance(thr, (int, float)) or not math.isfinite(thr) or thr < 1: + raise ValueError("threshold_volume 必须是 >=1 的数字 (单轮成交量增量, 单位手)") + bf = rule.get("basic_filter") + if bf is not None: + if not isinstance(bf, dict): + raise ValueError("basic_filter 必须是对象") + for key, value in bf.items(): + if key == "exclude_st": + if not isinstance(value, bool): + raise ValueError("basic_filter.exclude_st 必须是布尔值") + elif key in ("price_min", "price_max", "market_cap_min", "float_cap_min", + "float_cap_max", "amount_min"): + if value is not None and ( + isinstance(value, bool) or not isinstance(value, (int, float)) + or not math.isfinite(value) or value <= 0 + ): + raise ValueError(f"basic_filter.{key} 必须是正数字或 null") + else: + raise ValueError(f"basic_filter 不支持字段: {key}") else: # 信号/价格/市场类型: 需要 conditions conds = rule.get("conditions") @@ -254,7 +300,7 @@ def normalize(rule: dict) -> dict: r.setdefault("enabled", True) r.setdefault("asset_type", "stock") # sector/abnormal 默认全市场 (sector 随后强制 all; abnormal 支持指定标的) - r.setdefault("scope", "all" if r.get("type") in {"sector", "abnormal"} else "symbols") + r.setdefault("scope", "all" if r.get("type") in {"sector", "abnormal", "volume_delta"} else "symbols") r.setdefault("symbols", []) r.setdefault("group_id", None) # watchlist_group 作用域: 成员动态来自分组, symbols 不参与; 其他作用域清掉残留 group_id @@ -290,6 +336,15 @@ def normalize(rule: dict) -> dict: # ladder 专属默认字段 r.setdefault("metric", "sealed_vol") r.setdefault("threshold", 0) + # volume_delta 专属默认字段 (轮询放量): 冷却期默认 300s 而非 3600s -- + # 持续放量会连续多轮达标, 1 小时只提醒一次太迟钝。 + if r.get("type") == "volume_delta": + if r.get("cooldown_seconds") is None: + r["cooldown_seconds"] = 300 + r["metric"] = r["metric"] if r.get("metric") in VD_METRICS else "volume" + r.setdefault("threshold_volume", 9000) + r.setdefault("threshold_amount", 1e6) + r["basic_filter"] = {**VD_BASIC_FILTER_DEFAULTS, **(r.get("basic_filter") or {})} if r.get("type") == "sector": r["scope"] = "all" r["symbols"] = [] diff --git a/backend/tests/test_volume_delta_monitor.py b/backend/tests/test_volume_delta_monitor.py new file mode 100644 index 0000000..df1a36d --- /dev/null +++ b/backend/tests/test_volume_delta_monitor.py @@ -0,0 +1,274 @@ +"""轮询放量监控 (volume_delta) 测试: 引擎命中/冷却/批量合并 + 基础过滤 + 快照差值边界。""" +from __future__ import annotations + +from datetime import date + +import polars as pl +import pytest + +from app.strategy import monitor_rules +from app.strategy.monitor import MonitorRuleEngine + + +def _df(rows: list[dict]) -> pl.DataFrame: + """rows 每项: symbol/_volume_delta 必填, 其余可选 (close/amount/total_shares/float_shares)。""" + base = { + "symbol": [], "close": [], "change_pct": [], + "_volume_delta": [], "_volume_delta_amount": [], "_volume_delta_span": [], + } + optional = ["amount", "total_shares", "float_shares"] + for r in rows: + base["symbol"].append(r["symbol"]) + base["close"].append(r.get("close", 10.0)) + base["change_pct"].append(0.01) + base["_volume_delta"].append(r["_volume_delta"]) + base["_volume_delta_amount"].append(r.get("_volume_delta_amount", r["_volume_delta"] * 1000.0)) + base["_volume_delta_span"].append(6.0) + data = {k: v for k, v in base.items()} + for col in optional: + vals = [r.get(col) for r in rows] + if any(v is not None for v in vals): + data[col] = [v if v is not None else 0.0 for v in vals] + return pl.DataFrame(data) + + +def _rule(**kw): + r = { + "id": "vd1", "name": "轮询放量", "type": "volume_delta", + "asset_type": "stock", "scope": "all", "enabled": True, + "threshold_volume": 9000, "cooldown_seconds": 300, + "severity": "warn", + } + r.update(kw) + return r + + +def test_volume_delta_hits_above_threshold(): + eng = MonitorRuleEngine() + eng.set_rules([_rule()]) + events = eng.evaluate(_df([ + {"symbol": "S1.SH", "_volume_delta": 9500.0, "close": 10.0}, + {"symbol": "S2.SH", "_volume_delta": 8999.0, "close": 20.0}, + ])) + assert [e["symbol"] for e in events] == ["S1.SH"] + ev = events[0] + assert ev["source"] == "volume_delta" + assert "9,500" in ev["message"] and "9,000" in ev["message"] and "间隔 6s" in ev["message"] + assert ev["volume_delta"] == 9500.0 + + +def test_volume_delta_no_column_degrades_silently(): + eng = MonitorRuleEngine() + eng.set_rules([_rule()]) + plain = pl.DataFrame({"symbol": ["S1.SH"], "close": [10.0]}) + assert eng.evaluate(plain) == [] + + +def test_volume_delta_cooldown_suppresses_repeat(): + eng = MonitorRuleEngine() + eng.set_rules([_rule(cooldown=300)]) + df = _df([{"symbol": "S1.SH", "_volume_delta": 12000.0}]) + assert len(eng.evaluate(df)) == 1 + assert eng.evaluate(df) == [] + + +def test_volume_delta_batch_merge_over_five(): + eng = MonitorRuleEngine() + eng.set_rules([_rule()]) + rows = [{"symbol": f"S{i}.SH", "_volume_delta": 20000.0 + i} for i in range(8)] + events = eng.evaluate(_df(rows)) + assert len(events) == 1 + assert events[0]["symbol"] == "" + assert "共 8 只" in events[0]["message"] + + +def test_volume_delta_scope_filters(): + eng = MonitorRuleEngine() + eng.set_rules([_rule(scope="symbols", symbols=["S2.SH"])]) + events = eng.evaluate(_df([ + {"symbol": "S1.SH", "_volume_delta": 9500.0}, + {"symbol": "S2.SH", "_volume_delta": 9500.0}, + ])) + assert [e["symbol"] for e in events] == ["S2.SH"] + + +def test_volume_delta_metric_amount(): + eng = MonitorRuleEngine() + eng.set_rules([_rule(metric="amount", threshold_amount=5e6)]) + events = eng.evaluate(_df([ + {"symbol": "S1.SH", "_volume_delta": 100.0, "_volume_delta_amount": 6e6}, + {"symbol": "S2.SH", "_volume_delta": 20000.0, "_volume_delta_amount": 4.9e6}, + ])) + assert [e["symbol"] for e in events] == ["S1.SH"] + assert "万元" in events[0]["message"] + + +def test_volume_delta_basic_filter_price_and_amount(): + eng = MonitorRuleEngine() + eng.set_rules([_rule(basic_filter={ + "price_min": 5, "price_max": 100, "amount_min": 1e8, "exclude_st": False, + })]) + events = eng.evaluate(_df([ + # 价低被滤 + {"symbol": "LOW.SH", "_volume_delta": 20000.0, "close": 3.0, "amount": 5e8}, + # 价过高被滤 + {"symbol": "HIGH.SH", "_volume_delta": 20000.0, "close": 200.0, "amount": 5e8}, + # 成交额不足被滤 + {"symbol": "THIN.SH", "_volume_delta": 20000.0, "close": 10.0, "amount": 5e7}, + # 通过 + {"symbol": "OK.SH", "_volume_delta": 20000.0, "close": 10.0, "amount": 5e8}, + ])) + assert [e["symbol"] for e in events] == ["OK.SH"] + + +def test_volume_delta_basic_filter_market_cap(): + eng = MonitorRuleEngine() + eng.set_rules([_rule(basic_filter={ + "market_cap_min": 20e8, "price_min": None, "price_max": None, + "amount_min": None, "exclude_st": False, + })]) + # close × total_shares: BIG 10×3e8=30亿 通过; SMALL 10×1e8=10亿 被滤 + events = eng.evaluate(_df([ + {"symbol": "BIG.SH", "_volume_delta": 20000.0, "total_shares": 3e8}, + {"symbol": "SMALL.SH", "_volume_delta": 20000.0, "total_shares": 1e8}, + ])) + assert [e["symbol"] for e in events] == ["BIG.SH"] + + +def test_volume_delta_basic_filter_exclude_st(): + eng = MonitorRuleEngine() + eng.set_name_map({"STOCK.SH": "平安银行", "STK.SH": "ST 某某"}) + eng.set_rules([_rule(basic_filter={ + "price_min": None, "price_max": None, "amount_min": None, "exclude_st": True, + })]) + events = eng.evaluate(_df([ + {"symbol": "STOCK.SH", "_volume_delta": 20000.0}, + {"symbol": "STK.SH", "_volume_delta": 20000.0}, + ])) + assert [e["symbol"] for e in events] == ["STOCK.SH"] + + +def test_validate_and_normalize_defaults(): + r = monitor_rules.normalize({"id": "vd2", "type": "volume_delta"}) + assert r["threshold_volume"] == 9000 + assert r["scope"] == "all" + assert r["cooldown_seconds"] == 300 + assert r["metric"] == "volume" + assert r["basic_filter"]["price_min"] == 3 + assert r["basic_filter"]["exclude_st"] is True + # 用户字段覆盖默认 + r2 = monitor_rules.normalize({"id": "vd5", "type": "volume_delta", "basic_filter": {"price_min": 1, "exclude_st": False}}) + assert r2["basic_filter"]["price_min"] == 1 + assert r2["basic_filter"]["exclude_st"] is False + assert r2["basic_filter"]["price_max"] == 300 # 未覆盖项保留默认 + monitor_rules.validate({"id": "vd2", "name": "n", "type": "volume_delta", "threshold_volume": 1}) + with pytest.raises(ValueError): + monitor_rules.validate({"id": "vd3", "name": "n", "type": "volume_delta", "threshold_volume": 0}) + with pytest.raises(ValueError): + monitor_rules.validate({"id": "vd4", "name": "n", "type": "volume_delta", "asset_type": "etf"}) + with pytest.raises(ValueError): + monitor_rules.validate({"id": "vd6", "name": "n", "type": "volume_delta", + "metric": "amount", "threshold_amount": 0}) + with pytest.raises(ValueError): + monitor_rules.validate({"id": "vd7", "name": "n", "type": "volume_delta", + "basic_filter": {"price_min": -1}}) + with pytest.raises(ValueError): + monitor_rules.validate({"id": "vd8", "name": "n", "type": "volume_delta", + "basic_filter": {"unknown_field": 1}}) + + +# ── 快照差值状态 (QuoteService) ────────────────────────── + +def _qs(monkeypatch, *, continuous=True): + from app.services.quote_service import QuoteService + qs = QuoteService.__new__(QuoteService) + qs._prev_stock_volume = None + qs._prev_volume_fetched_at = None + qs._prev_volume_date = None + qs._volume_delta = {} + qs._volume_delta_span_s = 0.0 + monkeypatch.setattr(QuoteService, "_is_continuous_trading", lambda self: continuous) + monkeypatch.setattr( + QuoteService, "_continuous_session_start_ms", + staticmethod(lambda: 0.0), + ) + monkeypatch.setattr("app.services.quote_service.cn_today", lambda: date(2026, 8, 25)) + return qs + + +def test_delta_computed_and_prev_updated(monkeypatch): + qs = _qs(monkeypatch) + t0 = 1_000_000.0 + qs._update_volume_delta( + [{"symbol": "S1.SH", "volume": 10000, "amount": 5e6}, + {"symbol": "S2.SH", "volume": 500, "amount": 1e6}], t0, + ) + assert qs._volume_delta == {} # 首轮无 prev + qs._update_volume_delta( + [{"symbol": "S1.SH", "volume": 19500, "amount": 9.5e6}, + {"symbol": "S2.SH", "volume": 400, "amount": 2e6}], t0 + 6000, + ) + # S2 volume cur < prev (重置) → 丢弃; S1 差值 (9500 手, 450 万元) + assert qs._volume_delta == {"S1.SH": (9500.0, 4.5e6)} + assert qs._volume_delta_span_s == 6.0 + + +def test_delta_cross_day_reset(monkeypatch): + import app.services.quote_service as qsm + qs = _qs(monkeypatch) + qs._update_volume_delta([{"symbol": "S1.SH", "volume": 10000}], 1000.0) + assert qs._prev_volume_date == date(2026, 8, 25) + monkeypatch.setattr(qsm, "cn_today", lambda: date(2026, 8, 26)) + qs._update_volume_delta([{"symbol": "S1.SH", "volume": 20000}], 2000.0) + assert qs._volume_delta == {} + assert qs._prev_volume_date == date(2026, 8, 26) + + +def test_delta_open_protection(monkeypatch): + qs = _qs(monkeypatch) + # 9:29 的 prev (早于 9:30 时段起点) → 9:31 本轮不触发 + session_start = 1_000_000.0 + monkeypatch.setattr( + type(qs), "_continuous_session_start_ms", + staticmethod(lambda: session_start), + ) + qs._update_volume_delta([{"symbol": "S1.SH", "volume": 10000}], session_start - 60_000) + qs._update_volume_delta([{"symbol": "S1.SH", "volume": 99999}], session_start + 60_000) + assert qs._volume_delta == {} + # 之后一轮 prev 已在时段内 → 恢复计算 + qs._update_volume_delta([{"symbol": "S1.SH", "volume": 109999}], session_start + 66_000) + assert qs._volume_delta == {"S1.SH": (10000.0, 0.0)} + + +def test_delta_not_continuous_trading(monkeypatch): + qs = _qs(monkeypatch, continuous=False) + qs._update_volume_delta([{"symbol": "S1.SH", "volume": 10000}], 1000.0) + qs._update_volume_delta([{"symbol": "S1.SH", "volume": 99999}], 7000.0) + # 非连续竞价 (如午休) 不产差值, 但 prev 持续更新 + assert qs._volume_delta == {} + assert qs._prev_stock_volume == {"S1.SH": (99999.0, 0.0)} + + +def test_inject_volume_delta_join(): + from app.services.quote_service import QuoteService + qs = QuoteService.__new__(QuoteService) + qs._volume_delta = {"S1.SH": (900.0, 9e5), "S9.SH": (500.0, 5e5)} + qs._volume_delta_span_s = 6.0 + base = pl.DataFrame({"symbol": ["S1.SH", "S2.SH"], "close": [10.0, 20.0]}) + out = qs._inject_volume_delta(base) + assert out.filter(pl.col("symbol") == "S1.SH")["_volume_delta"][0] == 900.0 + assert out.filter(pl.col("symbol") == "S1.SH")["_volume_delta_amount"][0] == 9e5 + # 未命中股票为 null (不触发) + assert out.filter(pl.col("symbol") == "S2.SH")["_volume_delta"][0] is None + # 空差值原样返回 + qs._volume_delta = {} + assert qs._inject_volume_delta(base).columns == ["symbol", "close"] + + +def test_session_start_ms_matches_clock(): + from app.services.quote_service import QuoteService + from datetime import datetime, time as dt_time, timedelta, timezone + + now = QuoteService._continuous_session_start_ms() / 1000.0 + start_dt = datetime.fromtimestamp(now, tz=timezone(timedelta(hours=8))) + assert start_dt.time() in (dt_time(9, 30), dt_time(13, 0)) diff --git a/frontend/src/components/monitor/RuleEditor.tsx b/frontend/src/components/monitor/RuleEditor.tsx index f4a6a13..a22f4a5 100644 --- a/frontend/src/components/monitor/RuleEditor.tsx +++ b/frontend/src/components/monitor/RuleEditor.tsx @@ -1,7 +1,7 @@ import { useEffect, useMemo, useRef, useState } from 'react' import { Link } from 'react-router-dom' import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' -import { Activity, Building2, ChartNoAxesCombined, Check, ChevronDown, ChevronUp, Eraser, Layers3, ListPlus, Plus, RadioTower, Save, Search, Siren, Tags, TrendingUp, Waypoints, X } from 'lucide-react' +import { Activity, BarChart3, Building2, ChartNoAxesCombined, Check, ChevronDown, ChevronUp, Eraser, Layers3, ListPlus, Plus, RadioTower, Save, Search, Siren, Tags, TrendingUp, Waypoints, X } from 'lucide-react' import { api, genRuleId, type MonitorRule, type MonitorCondition, type SectorKind, type SectorMonitorTarget, type StrategyNotifyEvent } from '@/lib/api' import { DEFAULT_STRATEGY_NOTIFY_EVENTS, LEGACY_STRATEGY_NOTIFY_EVENTS, STRATEGY_NOTIFY_EVENT_OPTIONS } from '@/lib/strategyMonitorEvents' import { QK } from '@/lib/queryKeys' @@ -9,7 +9,7 @@ import { boardTag } from '@/components/stock-table/primitives' import { resolveWatchlistGroupColor } from '@/lib/watchlist-group-colors' import { SignalPicker } from '@/components/screener/SignalPicker' import { MONITOR_INTRADAY_SIGNAL_OPTIONS, SIGNAL_OPTIONS, cnSignal } from '@/lib/signals' -import { usePreferences } from '@/lib/useSharedQueries' +import { usePreferences, useQuoteStatus } from '@/lib/useSharedQueries' interface Props { /** 编辑现有规则;null=新建 */ @@ -23,7 +23,7 @@ interface Props { } const TYPE_DEFAULT_NAME: Record = { - signal: '信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控', sector: '板块监控', abnormal: '异动监控', + signal: '信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控', sector: '板块监控', abnormal: '异动监控', volume_delta: '轮询放量监控', } const TYPE_ICONS = { @@ -33,6 +33,7 @@ const TYPE_ICONS = { strategy: Waypoints, sector: Layers3, abnormal: Siren, + volume_delta: BarChart3, } const SECTOR_KIND_OPTIONS: Array<{ key: SectorKind; label: string; icon: typeof ChartNoAxesCombined }> = [ @@ -73,6 +74,7 @@ const emptyRule = (preset?: Partial): MonitorRule => ({ cooldown_seconds: 3600, severity: 'info', message: '', + threshold_volume: 9000, ...preset, }) @@ -80,6 +82,8 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { const qc = useQueryClient() const options = useQuery({ queryKey: QK.monitorRuleOptions, queryFn: api.monitorRuleOptions }) const { data: prefs } = usePreferences() + const { data: quoteStatus } = useQuoteStatus() + const quoteInterval = quoteStatus?.interval_s const feishuConfigured = !!(prefs?.feishu_webhook_url) const wecomConfigured = !!(prefs?.wecom_webhook_url) const [editing] = useState(!!rule) @@ -211,6 +215,18 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { if ((d.threshold_pct ?? 0) < 1 || (d.threshold_pct ?? 0) > 150) { throw new Error('接近度阈值必须在 1 到 150 之间 (70=边缘, 100=已触发)') } + } else if (d.type === 'volume_delta') { + delete d.score_min + delete d.score_max + d.conditions = [] + delete d.notify_events + if (d.metric === 'amount') { + if (!Number.isFinite(d.threshold_amount) || (d.threshold_amount ?? 0) < 1) { + throw new Error('金额阈值必须是 ≥1 的数字 (万元)') + } + } else if (!Number.isFinite(d.threshold_volume) || (d.threshold_volume ?? 0) < 1) { + throw new Error('单轮放量阈值必须是 ≥1 的手数') + } } else { delete d.score_min delete d.score_max @@ -591,12 +607,24 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { return { ...d, type, + // 轮询放量依赖全市场股票快照, 仅支持个股 + asset_type: type === 'volume_delta' ? 'stock' : d.asset_type, notify_events: type === 'strategy' ? [...(d.notify_events ?? DEFAULT_STRATEGY_NOTIFY_EVENTS)] : undefined, - scope: type === 'sector' || type === 'abnormal' + scope: type === 'sector' || type === 'abnormal' || type === 'volume_delta' ? 'all' : type === 'strategy' && d.scope === 'symbols' && d.symbols.length === 0 ? 'all' : d.scope, + // 轮询放量: 冷却期默认 300s (持续放量会连续多轮达标); 切走时还原 3600 + cooldown_seconds: type === 'volume_delta' && d.type !== 'volume_delta' ? 300 + : type !== 'volume_delta' && d.type === 'volume_delta' ? 3600 + : d.cooldown_seconds, + // 轮询放量: metric / 金额阈值 / 基础过滤默认 (与策略 basic_filter 对齐) + metric: type === 'volume_delta' && d.type !== 'volume_delta' ? 'volume' : d.metric, + threshold_amount: type === 'volume_delta' && d.type !== 'volume_delta' ? 1e6 : d.threshold_amount, + basic_filter: type === 'volume_delta' && d.type !== 'volume_delta' + ? { price_min: 3, price_max: 300, market_cap_min: 10e8, float_cap_min: null, float_cap_max: null, amount_min: 0.2e8, exclude_st: true } + : d.basic_filter, direction: type === 'sector' ? 'up' : type === 'abnormal' ? 'both' : d.type === 'sector' || d.type === 'abnormal' ? 'entry' : d.direction, @@ -895,6 +923,119 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { )} + {draft.type === 'volume_delta' && ( +
+
+
+ 阈值口径 +
+ {([['volume', '按手数'], ['amount', '按金额']] as const).map(([key, label]) => ( + + ))} +
+
+ +
+ +
+ 基础过滤 (与策略选股口径对齐, 留空不过滤) +
+ + + + + + +
+
+ +
+ 捕捉单次轮询间隔内的突发放量 (大单连续扫货)。开盘首轮与暂停恢复后的第一轮不触发, + 防止集合竞价撮合量误报; 冷却期内同一标的不重复提醒, 命中超过 5 只时合并为一条批量通知。 +
+
+ )} + {/* 作用范围 */} {draft.type !== 'sector' &&
作用范围 @@ -1106,7 +1247,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
} {/* 触发条件 (非 strategy) */} - {draft.type !== 'strategy' && draft.type !== 'sector' && draft.type !== 'abnormal' && ( + {draft.type !== 'strategy' && draft.type !== 'sector' && draft.type !== 'abnormal' && draft.type !== 'volume_delta' && (
触发条件 diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index b95f8b1..a960d9f 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -868,7 +868,7 @@ export interface MonitorRule { id: string name: string enabled: boolean - type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder' | 'sector' | 'abnormal' + type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder' | 'sector' | 'abnormal' | 'volume_delta' asset_type?: 'stock' | 'etf' | 'index' scope: 'symbols' | 'all' | 'sector' | 'watchlist_group' symbols: string[] @@ -897,9 +897,23 @@ export interface MonitorRule { webhook_channels?: string[] // 命中时推送的外部渠道 (合法值 'feishu' | 'wecom') created_at?: string runtime_warning?: string - // ladder 专属: 封单监控 - metric?: 'sealed_vol' | 'sealed_amount' // 量(手) / 额(元) + // ladder 专属: 封单监控; volume_delta 复用 metric 表示阈值口径 (volume=手数, amount=金额) + metric?: 'sealed_vol' | 'sealed_amount' | 'volume' | 'amount' threshold?: number // 封单 <= 此值时报警 + // volume_delta 专属 (轮询放量): 相邻两次全市场快照的成交量增量(手) + threshold_volume?: number // 单轮增量 >= 此值时报警 + threshold_amount?: number // metric=amount 时: 单轮增量 >= 此值(元)时报警 + basic_filter?: VDBasicFilter // 基础过滤 (与策略 basic_filter 语义对齐) +} + +export interface VDBasicFilter { + price_min?: number | null // 股价下限 (元) + price_max?: number | null // 股价上限 (元) + market_cap_min?: number | null // 总市值下限 (元) + float_cap_min?: number | null // 流通市值下限 (元) + float_cap_max?: number | null // 流通市值上限 (元) + amount_min?: number | null // 当日成交额下限 (元) + exclude_st?: boolean // 剔除 ST } export interface MonitorRuleOptions { diff --git a/frontend/src/pages/LimitUpLadder.tsx b/frontend/src/pages/LimitUpLadder.tsx index d267741..41de236 100644 --- a/frontend/src/pages/LimitUpLadder.tsx +++ b/frontend/src/pages/LimitUpLadder.tsx @@ -421,7 +421,9 @@ function MonitorMenu({ stock, direction, sealMode, monitorRule, anchorRect, hasD { key: '100000000', label: '亿元', mult: 100000000 }, ] - const [metric, setMetric] = useState<'sealed_vol' | 'sealed_amount'>(existing?.metric ?? (sealMode === 'amount' ? 'sealed_amount' : 'sealed_vol')) + const [metric, setMetric] = useState<'sealed_vol' | 'sealed_amount'>( + existing?.metric === 'sealed_amount' || (!existing && sealMode === 'amount') ? 'sealed_amount' : 'sealed_vol' + ) const units = metric === 'sealed_amount' ? AMT_UNITS : VOL_UNITS // 已有规则: 反算到最大便捷单位 (选能整除的最大倍率); 新建: 额默认亿元, 量默认万手 const initUnit = (() => { diff --git a/frontend/src/pages/Monitor.tsx b/frontend/src/pages/Monitor.tsx index a6c7873..a7e1a19 100644 --- a/frontend/src/pages/Monitor.tsx +++ b/frontend/src/pages/Monitor.tsx @@ -23,7 +23,7 @@ import { usePreferences, useQuoteStatus } from '@/lib/useSharedQueries' const TYPE_LABEL: Record = { signal: '信号', price: '价格/涨跌', market: '市场异动', strategy: '策略监控', sector: '板块监控', - abnormal: '异动监控', + abnormal: '异动监控', volume_delta: '轮询放量', } /** 严重级别 → 左侧色条 + 图标 */ @@ -39,6 +39,7 @@ const SOURCE_BADGE_STYLE: Record = { market: 'bg-purple-500/10 text-purple-400 border-purple-500/20', sector: 'bg-cyan-500/10 text-cyan-700 border-cyan-500/20 dark:text-cyan-300', abnormal: 'bg-orange-500/10 text-orange-500 border-orange-500/20 dark:text-orange-400', + volume_delta: 'bg-rose-500/10 text-rose-400 border-rose-500/20 dark:text-rose-300', } /** @@ -132,7 +133,7 @@ export function Monitor() { }, [searchParams, setSearchParams]) // 触发记录: 过滤 + 统计 (提升到主组件, 供 header 行使用) - const [filter, setFilter] = useState<'all' | 'strategy' | 'signal' | 'price' | 'market' | 'sector' | 'abnormal'>('all') + const [filter, setFilter] = useState<'all' | 'strategy' | 'signal' | 'price' | 'market' | 'sector' | 'abnormal' | 'volume_delta'>('all') const [confirmClear, setConfirmClear] = useState(false) const [confirmClearRules, setConfirmClearRules] = useState(false) @@ -213,7 +214,7 @@ export function Monitor() { {/* 过滤标签 */}
- {(['all', 'strategy', 'signal', 'price', 'market', 'sector', 'abnormal'] as const).map(f => ( + {(['all', 'strategy', 'signal', 'price', 'market', 'sector', 'abnormal', 'volume_delta'] as const).map(f => (
+ ) : r.type === 'volume_delta' ? ( +
+ + {r.metric === 'amount' + ? `单轮增量 ≥ ${Math.round((r.threshold_amount ?? 1e6) / 1e4).toLocaleString()} 万元` + : `单轮增量 ≥ ${(r.threshold_volume ?? 9000).toLocaleString()} 手`} + + + 冷却 {Math.round((r.cooldown_seconds ?? 300) / 60)} 分钟 + + {r.basic_filter && Object.values(r.basic_filter).some(v => v !== null && v !== false) && ( + + 基础过滤{r.basic_filter.exclude_st ? ' · 剔除ST' : ''} + + )} +
) : r.type === 'strategy' && r.strategy_id ? (
{(r.score_min != null || r.score_max != null) && (