tick-stock-panel/backend/app/strategy/monitor.py

"""策略实时监控 — 订阅行情更新，检查策略买卖信号。

职责: 接收实时行情 DataFrame → 检查监控中策略的信号 → 推送告警。
不知道: 策略加载逻辑、AI、API、配置持久化、回测。
依赖: 外部调用 on_quote_update() 传入实时数据。

本模块含两个评估器:
  1. StrategyMonitorService — 旧的策略监控 (type=strategy),第二步迁移到 MonitorRuleEngine
  2. MonitorRuleEngine — 通用规则引擎,覆盖 signal/price/market/strategy 四类,
     支持 scope (symbols/all/sector) + 多条件 AND/OR + cooldown 去重
"""
from __future__ import annotations

import datetime as _dt
import logging
import math
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Callable

import polars as pl

from app.market_time import cn_today
from app.strategy import config as _strategy_config
from app.strategy.custom_signals import _OP_BUILDERS  # type: ignore  # 复用运算符构造器
from app.strategy.intraday_signals import INTRADAY_SIGNAL_LABELS, uses_intraday_signals

logger = logging.getLogger(__name__)

# 信号 / 字段中文名映射 — 与前端 lib/signals.ts 对齐, 用于告警 message / 推送文案。
# signal_* 为内置原子信号, 其余为技术指标/行情字段。
_SIGNAL_CN: dict[str, str] = {
    # 内置信号
    "signal_ma_golden_5_20": "MA5上穿MA20", "signal_ma_dead_5_20": "MA5下穿MA20",
    "signal_ma_golden_20_60": "MA20上穿MA60", "signal_macd_golden": "MACD金叉",
    "signal_macd_dead": "MACD死叉", "signal_ma20_breakout": "突破MA20",
    "signal_ma20_breakdown": "跌破MA20", "signal_ma5_breakout": "突破MA5",
    "signal_ma5_breakdown": "跌破MA5", "signal_ma10_breakout": "突破MA10",
    "signal_ma10_breakdown": "跌破MA10", "signal_n_day_high": "60日新高",
    "signal_n_day_low": "60日新低", "signal_boll_breakout_upper": "突破布林上轨",
    "signal_boll_breakdown_lower": "跌破布林下轨", "signal_volume_surge": "放量",
    "signal_limit_up": "涨停", "signal_limit_down": "跌停",
    "signal_limit_down_recovery": "跌停翘板", "signal_broken_limit_up": "炸板",
    **INTRADAY_SIGNAL_LABELS,
    # 行情字段
    "close": "收盘价", "open": "开盘价", "high": "最高价", "low": "最低价",
    "change_pct": "涨跌幅", "change_amount": "涨跌额", "amplitude": "振幅",
        "turnover_rate": "换手率", "volume": "成交量", "amount": "成交额",
        "_volume_delta": "轮询成交量差值(手)", "_sealed_vol": "封单量(手)",
    # 均线
    "ma5": "MA5", "ma10": "MA10", "ma20": "MA20", "ma30": "MA30", "ma60": "MA60",
    "ema5": "EMA5", "ema10": "EMA10", "ema20": "EMA20",
    # MACD / BOLL / KDJ / RSI
    "macd_dif": "MACD-DIF", "macd_dea": "MACD-DEA", "macd_hist": "MACD柱",
    "boll_upper": "布林上轨", "boll_lower": "布林下轨",
    "kdj_k": "KDJ-K", "kdj_d": "KDJ-D", "kdj_j": "KDJ-J",
    "rsi_6": "RSI6", "rsi_14": "RSI14", "rsi_24": "RSI24",
    # 量能 / 动量 / 波动
    "vol_ratio_5d": "5日量比", "vol_ratio_20d": "20日量比",
    "vol_ma5": "5日均量", "vol_ma10": "10日均量",
    "high_60d": "60日最高", "low_60d": "60日最低",
    "momentum_5d": "5日动量", "momentum_20d": "20日动量", "momentum_60d": "60日动量",
    "atr_14": "ATR14", "annual_vol_20d": "20日年化波动",
    "consecutive_limit_ups": "连板数", "consecutive_limit_downs": "跌停连板",
}


def _signal_cn_name(name: str) -> str:
    """返回信号/字段的中文名, 找不到原样返回 (与前端 cnSignal 对齐)。"""
    return _SIGNAL_CN.get(name, name)


@dataclass
class StrategyAlert:
    """策略告警"""
    type: str              # "entry" | "exit"
    strategy_id: str
    symbol: str
    name: str | None
    message: str
    price: float | None = None
    change_pct: float | None = None
    signals: list[str] = field(default_factory=list)


class StrategyMonitorService:
    """策略实时监控服务"""

    def __init__(self, alert_handler: Callable[[StrategyAlert], None] | None = None):
        """
        Args:
            alert_handler: 告警回调 (如推 SSE)
        """
        self._alert_handler = alert_handler
        # strategy_id → 监控配置
        self._watching: dict[str, dict] = {}
        # _watching 跨线程锁: on_quote_update 跑在行情轮询线程迭代 _watching,
        # API 线程同时 start/stop 增删会抛 "dict changed size during iteration"。
        # 增删与迭代前的快照都持此锁 (镜像 MonitorRuleEngine.evaluate 的 list 快照)。
        self._watching_lock = threading.Lock()

    def start(self, strategy_id: str, config: dict) -> None:
        """开始监控一个策略

        config: {
            "entry_signals": ["signal_n_day_high", ...],
            "exit_signals": ["signal_ma20_breakdown", ...],
        }
        """
        with self._watching_lock:
            self._watching[strategy_id] = config
        logger.info("strategy monitor started: %s", strategy_id)

    def stop(self, strategy_id: str) -> None:
        with self._watching_lock:
            self._watching.pop(strategy_id, None)
        logger.info("strategy monitor stopped: %s", strategy_id)

    def stop_all(self) -> None:
        with self._watching_lock:
            self._watching.clear()

    @property
    def watching(self) -> dict[str, dict]:
        with self._watching_lock:
            return dict(self._watching)

    def on_quote_update(self, df: pl.DataFrame) -> list[StrategyAlert]:
        """行情更新后调用。向量化检查所有监控策略。

        Args:
            df: 实时 enriched 数据 (~5500行)
        Returns:
            触发的告警列表
        """
        if not self._watching or df.is_empty():
            return []

        all_alerts: list[StrategyAlert] = []

        # 迭代前持锁快照, 避免行情线程迭代时 API 线程 start/stop 改变字典大小
        with self._watching_lock:
            watching_items = list(self._watching.items())

        for strategy_id, cfg in watching_items:
            # 买入信号
            entry_sigs = cfg.get("entry_signals", [])
            if entry_sigs:
                for sym, name, price, pct, hit_sigs in self._check_signals(df, entry_sigs):
                    alert = StrategyAlert(
                        type="entry",
                        strategy_id=strategy_id,
                        symbol=sym,
                        name=name,
                        message="入场信号触发",
                        price=price,
                        change_pct=pct,
                        signals=hit_sigs,
                    )
                    all_alerts.append(alert)
                    self._emit(alert)

            # 卖出信号
            exit_sigs = cfg.get("exit_signals", [])
            if exit_sigs:
                for sym, name, price, pct, hit_sigs in self._check_signals(df, exit_sigs):
                    alert = StrategyAlert(
                        type="exit",
                        strategy_id=strategy_id,
                        symbol=sym,
                        name=name,
                        message="出场信号触发",
                        price=price,
                        change_pct=pct,
                        signals=hit_sigs,
                    )
                    all_alerts.append(alert)
                    self._emit(alert)

        return all_alerts

    def _emit(self, alert: StrategyAlert) -> None:
        if self._alert_handler:
            try:
                self._alert_handler(alert)
            except Exception as e:
                logger.warning("alert handler failed: %s", e)

    @staticmethod
    def _check_signals(
        df: pl.DataFrame,
        signals: list[str],
    ) -> list[tuple[str, str | None, float | None, float | None, list[str]]]:
        """检查信号列，返回 [(symbol, name, price, change_pct, [hit_signals])]。
        支持内置 signal_ 与自定义 csg_ 前缀。"""
        cols = set(df.columns)
        resolved: list[tuple[str, str]] = []  # (原值, 列名)
        for s in signals:
            col = s if (s.startswith("signal_") or s.startswith("csg_")) else f"signal_{s}"
            if col in cols:
                resolved.append((s, col))
        if not resolved:
            return []

        mask = pl.any_horizontal(pl.col(c).fill_null(False) for _, c in resolved)
        hit_df = df.filter(mask)

        results = []
        for row in hit_df.iter_rows(named=True):
            sym = row.get("symbol", "")
            name = row.get("name")
            price = row.get("close")
            pct = row.get("change_pct")
            hit_sigs = [orig for orig, col in resolved if row.get(col)]
            results.append((sym, name, price, pct, hit_sigs))
        return results

# ================================================================
# 通用监控规则引擎 MonitorRuleEngine
# ================================================================

_SIGNAL_PREFIXES = ("signal_", "csg_")


# ── 自选分组作用域: group_id → 成员集合解析 (进程内缓存) ────
# 缓存按 watchlist 数据版本号失效: 版本不变时零磁盘 IO; 自选页任何增删
# 分组/成员的操作都会 bump 版本号, 下一轮评估立即拿到新成员 (无需等 TTL)。
_group_cache_lock = threading.Lock()
_group_cache: dict[str, Any] = {}
# 已告警过的「分组已删除」(rule_id, group_id), 防止每轮评估刷日志
_warned_missing_groups: set[tuple[str, str]] = set()


def _watchlist_groups_snapshot() -> dict[str, frozenset[str]]:
    """返回 {group_id: 成员symbol集}。读前后版本一致才写缓存, 避免缓存住写竞态下的旧数据。"""
    from app.services import watchlist

    rev_before = watchlist.revision()
    with _group_cache_lock:
        cached = _group_cache.get("groups")
        if cached is not None and _group_cache.get("_rev") == rev_before:
            return cached
    groups: dict[str, set[str]] = {g["id"]: set() for g in watchlist.list_groups()}
    for row in watchlist.list_symbols():
        for gid in row.get("group_ids") or []:
            members = groups.get(gid)
            if members is not None:
                members.add(str(row["symbol"]))
    frozen = {gid: frozenset(syms) for gid, syms in groups.items()}
    if watchlist.revision() == rev_before:
        with _group_cache_lock:
            _group_cache["_rev"] = rev_before
            _group_cache["groups"] = frozen
    return frozen


def _group_members_or_none(rule: dict) -> frozenset[str] | None:
    """解析规则绑定的分组成员; 分组已删除返回 None, 解析异常返回 None 并记日志。"""
    group_id = str(rule.get("group_id") or "")
    try:
        groups = _watchlist_groups_snapshot()
    except Exception as exc:  # noqa: BLE001
        logger.warning("自选分组数据读取失败, 规则 %s 本轮跳过: %s", rule.get("id"), exc)
        return None
    members = groups.get(group_id)
    if members is None:
        key = (str(rule.get("id") or ""), group_id)
        if key not in _warned_missing_groups:
            _warned_missing_groups.add(key)
            logger.warning(
                "监控规则 %s 绑定的自选分组 %s 已删除, 本轮跳过 (fail-closed, 恢复分组后自动生效)",
                rule.get("id"), group_id,
            )
    return members


def _is_signal_field(field: str) -> bool:
    return any(field.startswith(p) for p in _SIGNAL_PREFIXES)


def _build_condition_mask(df: pl.DataFrame, conditions: list[dict], logic: str) -> pl.DataFrame:
    """根据 conditions + logic 构建过滤后的命中 DataFrame。

    conditions: [{"field","op","value"?}] — op=truth 为布尔信号, 否则阈值比较
    logic: "and" | "or"
    返回命中行 (含 symbol/name/close/change_pct + 各信号列)
    """
    cols = set(df.columns)
    parts: list[pl.Expr] = []
    for c in conditions:
        field = c["field"]
        if field not in cols:
            return df.head(0)  # 字段缺失,无法判定 → 空结果
        op = c["op"]
        if op == "truth":
            parts.append(pl.col(field).fill_null(False))
        elif op in _OP_BUILDERS:
            parts.append(_OP_BUILDERS[op](pl.col(field), c["value"]))
        else:
            return df.head(0)
    if not parts:
        return df.head(0)
    if logic == "or":
        mask = pl.any_horizontal(parts)
    else:
        mask = pl.all_horizontal(parts)
    return df.filter(mask)


class MonitorRuleEngine:
    """通用监控规则引擎 — 接收实时行情 DataFrame,评估所有规则,返回 AlertEvent。

    与 StrategyMonitorService 的区别:
      - 规则来自 monitor_rules 存储 (用户可配), 而非写死的 strategy config
      - 支持 scope (symbols/all/sector) 过滤作用域
      - 支持 conditions + logic (AND/OR) 任意组合
      - ★ cooldown 去重: 同一 (rule_id, symbol, event_type) 在冷却期内不重复触发
    """

    def __init__(self, alert_handler: Callable[[dict], None] | None = None):
        self._alert_handler = alert_handler
        self._rules: dict[str, dict] = {}  # rule_id → rule
        # (rule_id, symbol, event_type) → 上次触发时间戳(秒)。用于 cooldown 去重。
        self._last_fire: dict[tuple[str, str, str], float] = {}
        self._strategy_engine = None  # 延迟注入, type=strategy 规则用它跑选股
        # symbol → 股票名 (enriched DataFrame 已 drop name 列, 触发时从此映射回填)
        self._name_map: dict[str, str] = {}
        # 策略选股池状态: (rule_id, strategy_id, asset_type) → 上期选股符号集合
        self._strategy_pools: dict[tuple[str, str, str], set[str]] = {}
        # 策略信号状态: (rule_id, strategy_id, asset_type, event_type) → (K线日期, 命中集合)
        self._strategy_signal_state: dict[tuple[str, str, str, str], tuple[str, set[str]]] = {}
        # 同一根 K 线的信号即使盘中回落后再次命中也只通知一次。
        self._strategy_signal_seen: dict[tuple[str, str, str, str, str], str] = {}
        # 数据目录 (用于加载策略 overrides)
        self._data_dir = None
        # 历史窗口加载器: (target_date, lookback_days) → 多日 enriched DataFrame。
        # 用于声明 filter_history 的策略 (如反包), 实时监控时拼历史窗口 + 今日行情跑选股。
        # 为 None 时, filter_history 策略仍会被跳过 (保持旧行为, 不破坏无历史场景)。
        self._history_loader: Callable[[_dt.date, int], "pl.DataFrame"] | None = None
        # ETF 版历史窗口加载器 (asset_type=etf 的规则用)。为 None 时 ETF filter_history 策略跳过。
        self._history_loader_etf: Callable[[_dt.date, int], "pl.DataFrame"] | None = None
        self._active_matrix_snapshots: dict[str, Any] = {}
        # 本轮 evaluate() 产出的策略选股结果: strategy_id → {rows, total, as_of}
        # 供策略页实时回显复用 (/api/screener/cached 端点直接读取, 避免重跑)。
        # 注意: 始终是「完整」的 dict —— evaluate 重算时先写到 _building_strategy_results,
        # 算完后整体替换此属性, 保证 /cached 并发读取永远拿到完整结果, 不会读到空中间态。
        self._latest_strategy_results: dict[str, dict] = {}
        # 本轮重算的临时容器 (_match_strategy 写入它); reset 轮开始时初始化为空 dict,
        # evaluate 结束后一次性替换 _latest_strategy_results。
        self._building_strategy_results: dict[str, dict] = {}
        # 本轮成功写入股票策略实时结果的策略 ID, 供 QuoteService 在计算完成后精确通知策略页。
        self._latest_strategy_result_ids: set[str] = set()
        self._sector_monitor_service = None
        self._sector_condition_state: dict[tuple[str, str], bool] = {}
        # abnormal 规则边缘触发状态: (rule_id, symbol) → 上一轮是否已达阈值。
        # 只在 False → True 跳变时告警 (首轮观测不触发, 防止新建规则瞬间刷屏)。
        self._abnormal_condition_state: dict[tuple[str, str], bool] = {}

    def set_strategy_engine(self, engine) -> None:
        """注入 StrategyEngine, type=strategy 规则据此跑选股。"""
        self._strategy_engine = engine

    def set_data_dir(self, data_dir) -> None:
        """注入数据目录, 用于加载策略的用户覆盖配置。"""
        self._data_dir = data_dir

    def set_sector_monitor_service(self, service) -> None:
        self._sector_monitor_service = service

    def invalidate_strategy_state(self) -> None:
        """策略注册表变更后清除选股池、结果和矩阵快照。"""
        self._strategy_pools.clear()
        self._strategy_signal_state.clear()
        self._strategy_signal_seen.clear()
        self._latest_strategy_results = {}
        self._building_strategy_results = {}
        self._latest_strategy_result_ids.clear()
        self._active_matrix_snapshots.clear()

    def set_history_loader(self, fn) -> None:
        """注入历史窗口加载器, 用于声明 filter_history 的策略跑实时监控。

        loader 签名: (target_date, lookback_days) → 多日 enriched DataFrame。
        复用 ScreenerService._load_enriched_history (三级缓存, 命中 ~0ms)。
        为 None 时 filter_history 策略退回到跳过逻辑 (不破坏无历史场景)。
        """
        self._history_loader = fn

    def set_history_loader_etf(self, fn) -> None:
        """注入 ETF 版历史窗口加载器 (asset_type=etf 的 strategy 型规则用)。

        签名同 set_history_loader; 复用 ScreenerService(asset_type='etf')._load_enriched_history。
        为 None 时 ETF filter_history 策略退回到跳过逻辑。
        """
        self._history_loader_etf = fn

    def _history_loader_for(self, rule: dict):
        """按规则的 asset_type 选历史加载器。etf → ETF 加载器, 否则股票加载器。"""
        if rule.get("asset_type") == "etf":
            return self._history_loader_etf
        return self._history_loader

    def set_name_map(self, name_map: dict[str, str]) -> None:
        """注入 symbol → 股票名 映射, 用于在告警事件里回填 name 字段。

        enriched DataFrame 在 pipeline 计算后不含 name 列 (见 indicators/pipeline.py),
        触发时从 instruments 表预构建此映射, 保证 AlertEvent.name 有值。
        """
        self._name_map = name_map or {}

    # ── 规则管理 ───────────────────────────────────────
    @staticmethod
    def _rule_state_signature(rule: dict) -> tuple[Any, ...]:
        return (
            rule.get("type"),
            rule.get("strategy_id"),
            rule.get("score_min"),
            rule.get("score_max"),
            rule.get("asset_type", "stock"),
            rule.get("scope", "symbols"),
            tuple(sorted(str(symbol) for symbol in rule.get("symbols", []))),
            rule.get("sector"),
            rule.get("sector_kind"),
            tuple(sorted(str(target.get("key")) for target in rule.get("sector_targets", []))),
            rule.get("sector_trigger"),
            rule.get("direction"),
            rule.get("threshold_pct"),
            rule.get("window_minutes"),
            rule.get("abnormal_window"),
        )

    def set_rules(self, rules: list[dict]) -> None:
        """批量设置规则 (覆盖)。用于启动时 reload。

        先构建完整 dict 再原子替换 self._rules, 避免评估线程 (行情轮询)
        读到装载了一半的规则表。
        """
        new_rules: dict[str, dict] = {}
        for r in rules:
            if r.get("enabled") is not False:
                new_rules[r["id"]] = r
        changed_ids = {
            rule_id
            for rule_id, rule in new_rules.items()
            if rule_id in self._rules
            and self._rule_state_signature(self._rules[rule_id])
            != self._rule_state_signature(rule)
        }
        self._rules = new_rules
        active_ids = set(new_rules) - changed_ids
        self._last_fire = {
            key: value for key, value in list(self._last_fire.items()) if key[0] in active_ids
        }
        self._strategy_pools = {
            key: value for key, value in list(self._strategy_pools.items()) if key[0] in active_ids
        }
        self._strategy_signal_state = {
            key: value
            for key, value in list(self._strategy_signal_state.items())
            if key[0] in active_ids
        }
        self._strategy_signal_seen = {
            key: value
            for key, value in list(self._strategy_signal_seen.items())
            if key[0] in active_ids
        }
        self._sector_condition_state = {
            key: value
            for key, value in list(self._sector_condition_state.items())
            if key[0] in active_ids
        }
        self._abnormal_condition_state = {
            key: value
            for key, value in list(self._abnormal_condition_state.items())
            if key[0] in active_ids
        }
        logger.info("MonitorRuleEngine: 装载 %d 条规则", len(self._rules))

    def add_rule(self, rule: dict) -> None:
        if rule.get("enabled") is not False:
            self._rules[rule["id"]] = rule
        else:
            self._rules.pop(rule["id"], None)

    def remove_rule(self, rule_id: str) -> None:
        self._rules.pop(rule_id, None)
        self._last_fire = {k: v for k, v in list(self._last_fire.items()) if k[0] != rule_id}
        self._strategy_pools = {
            k: v for k, v in list(self._strategy_pools.items()) if k[0] != rule_id
        }
        self._strategy_signal_state = {
            k: v for k, v in list(self._strategy_signal_state.items()) if k[0] != rule_id
        }
        self._strategy_signal_seen = {
            k: v for k, v in list(self._strategy_signal_seen.items()) if k[0] != rule_id
        }
        self._sector_condition_state = {
            k: v for k, v in self._sector_condition_state.items() if k[0] != rule_id
        }

    def clear(self) -> None:
        self._rules.clear()
        self._last_fire.clear()
        self._strategy_pools.clear()
        self._strategy_signal_state.clear()
        self._strategy_signal_seen.clear()
        self._sector_condition_state.clear()

    @property
    def rules(self) -> dict[str, dict]:
        return dict(self._rules)

    @property
    def rule_count(self) -> int:
        return len(self._rules)

    def latest_strategy_results(self) -> dict[str, dict]:
        """返回本轮 evaluate() 产出的策略选股结果 (strategy_id → {rows, total, as_of})。

        供策略页实时回显复用: /api/screener/cached 端点直接读取此内存结果,
        避免对被监控的策略重跑第二遍。无 type=strategy 规则时返回空 dict。
        """
        return self._latest_strategy_results

    def consume_strategy_result_updates(self) -> bool:
        """返回并清除本轮成功写入的股票策略实时结果标记。"""
        updated = bool(self._latest_strategy_result_ids)
        self._latest_strategy_result_ids.clear()
        return updated

    def has_rule_type(self, rtype: str) -> bool:
        """是否存在指定类型的 (已启用) 规则。供 quote_service 判断是否需要注入特殊数据。"""
        if not self._rules:
            return False
        # list() 快照: API 线程可能并发增删规则, 直接迭代 dict 会抛 RuntimeError
        return any(
            r.get("enabled", True) and r.get("type") == rtype
            for r in list(self._rules.values())
        )

    def intraday_signal_symbols(self, asset_type: str) -> set[str]:
        """返回启用的分时信号规则所需标的并集。"""
        symbols: set[str] = set()
        for rule in list(self._rules.values()):
            if (
                rule.get("enabled", True)
                and rule.get("asset_type", "stock") == asset_type
                and rule.get("scope") == "symbols"
                and uses_intraday_signals(rule)
            ):
                symbols.update(str(symbol) for symbol in rule.get("symbols", []) if symbol)
        return symbols

    # ── 评估 ───────────────────────────────────────────
    def has_asset_rules(self, asset_type: str) -> bool:
        """是否存在指定资产类型的 (已启用) 规则。供 quote_service 判断是否需要 ETF 评估轮。"""
        if not self._rules:
            return False
        return any(
            r.get("enabled", True) and r.get("asset_type", "stock") == asset_type
            for r in list(self._rules.values())
        )

    def evaluate(self, df: pl.DataFrame, asset_type: str = "stock",
                 reset_strategy_results: bool = True) -> list[dict]:
        """行情更新后评估规则。

        按 asset_type 只评估匹配资产类型的规则; ETF 规则应传 ETF enriched 快照。
        股票/ETF 分两轮评估时, 仅股票轮重置 _latest_strategy_results (它供股票策略页
        /cached 回显; ETF 策略页走实时单跑, 不依赖它)。

        Args:
            df: 实时 enriched 数据 (含 signal_/csg_/指标列)
            asset_type: 只评估该资产类型的规则 (默认 stock, 向后兼容)
            reset_strategy_results: 是否重置策略结果缓存 (多轮评估时仅首轮 True)
        Returns:
            触发的 AlertEvent dict 列表 (含 ts/rule_id/source/type/symbol/...)
        """
        if not self._rules or df.is_empty():
            return []

        now = time.time()
        events: list[dict] = []
        # 原子化: reset 轮 (股票轮) 时先把本轮结果写到临时容器, 算完后一次性替换
        # _latest_strategy_results。这样 /cached 并发读取永远拿到完整结果,
        # 不会在「清空 → 逐个回填」窗口里读到空中间态 (曾导致策略页闪烁)。
        # 非 reset 轮 (ETF 轮) 继续往同一临时容器追加 (_match_strategy 仅写 stock, 实际不追加)。
        if reset_strategy_results:
            self._building_strategy_results = {}
            self._latest_strategy_result_ids.clear()

        matrix_rules: list[dict] = []
        params_map: dict[str, dict] = {}
        overrides_map: dict[str, dict] = {}
        if self._strategy_engine is not None:
            for rule in list(self._rules.values()):
                if (
                    not rule.get("enabled", True)
                    or rule.get("type") != "strategy"
                    or rule.get("asset_type", "stock") != asset_type
                ):
                    continue
                sid = rule.get("strategy_id")
                if not sid:
                    continue
                try:
                    strategy = self._strategy_engine.get(sid)
                except Exception:
                    continue
                if getattr(strategy, "execution_backend", "polars_expr") != "matrix_native":
                    continue
                overrides = {}
                if self._data_dir:
                    overrides = _strategy_config.load_override(self._data_dir, sid)
                matrix_rules.append(rule)
                overrides_map[sid] = overrides
                params_map[sid] = dict(overrides.get("params") or {})
        if matrix_rules:
            try:
                history_loader = self._history_loader_for(matrix_rules[0])
                if history_loader is None:
                    raise ValueError("matrix strategy monitor requires history loader")
                matrix_ids = [str(rule["strategy_id"]) for rule in matrix_rules]
                history_bars = self._strategy_engine.required_history_bars(
                    matrix_ids,
                    params_map=params_map,
                    overrides_map=overrides_map,
                )
                from app.strategy.engine import StrategyDataContext
                context = StrategyDataContext(
                    asset_type=asset_type,
                    timeframe="1d",
                    as_of=cn_today(),
                    current=df,
                    cache_key=f"monitor:{asset_type}",
                )
                try:
                    snapshot = self._strategy_engine.prepare_realtime_matrix(
                        context,
                        matrix_ids,
                        params_map=params_map,
                        overrides_map=overrides_map,
                    )
                except ValueError as exc:
                    if "requires history data" not in str(exc):
                        raise
                    history = history_loader(cn_today(), history_bars)
                    snapshot = self._strategy_engine.prepare_realtime_matrix(
                        StrategyDataContext(
                            asset_type=asset_type,
                            timeframe="1d",
                            as_of=cn_today(),
                            current=df,
                            history=history,
                            cache_key=f"monitor:{asset_type}",
                        ),
                        matrix_ids,
                        params_map=params_map,
                        overrides_map=overrides_map,
                    )
                self._active_matrix_snapshots[asset_type] = snapshot
            except Exception as e:
                self._active_matrix_snapshots.pop(asset_type, None)
                logger.warning("%s 矩阵策略实时缓存准备失败: %s", asset_type, e)

        # list() 快照: 本方法跑在行情轮询线程, API 线程同时 add/remove 规则
        # 会触发 "dictionary changed size during iteration", 整轮告警丢失
        for rule_id, rule in list(self._rules.items()):
            if rule.get("asset_type", "stock") != asset_type:
                continue
            if rule.get("type") in ("sector", "abnormal"):
                continue
            try:
                events.extend(self._evaluate_rule(df, rule, now))
            except Exception as e:
                logger.warning("规则评估失败 %s: %s", rule_id, e)

        # 一次性提交本轮结果 (原子替换): /cached 读方要么拿到上一轮完整结果,
        # 要么拿到本轮完整结果, 不会读到空中间态。
        self._latest_strategy_results = self._building_strategy_results
        self._active_matrix_snapshots.pop(asset_type, None)

        return events

    def evaluate_sectors(
        self,
        stock_df: pl.DataFrame,
        index_df: pl.DataFrame,
        *,
        now: float | None = None,
    ) -> list[dict]:
        """按板块聚合快照评估 type=sector 规则。"""
        if self._sector_monitor_service is None:
            return []
        rules = [
            rule for rule in list(self._rules.values())
            if rule.get("enabled", True) and rule.get("type") == "sector"
        ]
        if not rules:
            return []

        targets_by_key: dict[str, dict] = {}
        windows: set[int] = set()
        for rule in rules:
            for target in rule.get("sector_targets", []):
                if target.get("key"):
                    targets_by_key[str(target["key"])] = target
            if rule.get("sector_trigger") == "momentum":
                windows.add(int(rule.get("window_minutes", 5)))

        timestamp = time.time() if now is None else now
        snapshots = self._sector_monitor_service.build_snapshots(
            stock_df,
            index_df,
            list(targets_by_key.values()),
            windows,
            now=timestamp,
        )
        events: list[dict] = []
        for rule in rules:
            try:
                events.extend(self._evaluate_sector_rule(rule, snapshots, timestamp))
            except Exception as exc:  # noqa: BLE001
                logger.warning("板块规则评估失败 %s: %s", rule.get("id"), exc)
        return events

    def _evaluate_sector_rule(self, rule: dict, snapshots: dict[str, dict], now: float) -> list[dict]:
        events: list[dict] = []
        direction = rule.get("direction", "up")
        trigger = rule.get("sector_trigger", "change_pct")
        threshold = float(rule.get("threshold_pct", 1.0)) / 100
        window = int(rule.get("window_minutes", 5))

        for target in rule.get("sector_targets", []):
            target_key = str(target.get("key") or "")
            snapshot = snapshots.get(target_key)
            if not snapshot or not snapshot.get("valid"):
                continue
            value = (
                snapshot.get("change_pct")
                if trigger == "change_pct"
                else snapshot.get("window_changes", {}).get(window)
            )
            condition = value is not None and (
                value >= threshold if direction == "up" else value <= -threshold
            )
            state_key = (rule["id"], target_key)
            previous = self._sector_condition_state.get(state_key)
            self._sector_condition_state[state_key] = condition
            if previous is None or previous or not condition:
                continue

            event_type = f"sector_{trigger}_{direction}"
            cooldown_key = (rule["id"], target_key, event_type)
            last = self._last_fire.get(cooldown_key)
            cooldown = int(rule.get("cooldown_seconds", 3600))
            if last is not None and now - last < cooldown:
                continue
            self._last_fire[cooldown_key] = now
            message = rule.get("message", "") or self._sector_message(
                snapshot, trigger, direction, threshold, window, value,
            )
            event = {
                "ts": int(now * 1000),
                "rule_id": rule["id"],
                "rule_name": rule.get("name", ""),
                "strategy_id": None,
                "source": "sector",
                "type": event_type,
                "symbol": snapshot.get("symbol") if snapshot.get("kind") == "index" else "",
                "name": snapshot.get("name"),
                "message": message,
                "price": snapshot.get("price"),
                "change_pct": snapshot.get("change_pct"),
                "window_change_pct": value if trigger == "momentum" else None,
                "signals": [],
                "severity": rule.get("severity", "info"),
                "conditions": [],
                "logic": "and",
                "sector_kind": snapshot.get("kind"),
                "sector_key": target_key,
                "sector_name": snapshot.get("name"),
                "sector_source_field": snapshot.get("source_field"),
                "sector_value": snapshot.get("value"),
                "sector_level": snapshot.get("level"),
                "coverage_ratio": snapshot.get("coverage_ratio"),
                "valid_count": snapshot.get("valid_count"),
                "total_count": snapshot.get("total_count"),
                "up_count": snapshot.get("up_count"),
                "down_count": snapshot.get("down_count"),
                "leader": snapshot.get("leader"),
            }
            events.append(event)
            if self._alert_handler:
                try:
                    self._alert_handler(event)
                except Exception as exc:  # noqa: BLE001
                    logger.warning("alert handler failed: %s", exc)
        return events

    @staticmethod
    def _sector_message(
        snapshot: dict,
        trigger: str,
        direction: str,
        threshold: float,
        window: int,
        value: float | None,
    ) -> str:
        kind_label = {
            "index": "指数", "concept": "概念", "industry": "行业",
        }.get(snapshot.get("kind"), "板块")
        current = float(snapshot.get("change_pct") or 0)
        if trigger == "momentum":
            action = "快速拉升" if direction == "up" else "快速下跌"
            head = (
                f"{kind_label}「{snapshot.get('name')}」{window}分钟{action} "
                f"{float(value or 0) * 100:+.2f}%"
            )
        else:
            action = "涨幅上穿" if direction == "up" else "跌幅下穿"
            head = f"{kind_label}「{snapshot.get('name')}」{action} {threshold * 100:.2f}%"
        parts = [head, f"当前 {current * 100:+.2f}%"]
        if snapshot.get("kind") != "index":
            parts.append(f"上涨 {snapshot.get('up_count', 0)}/{snapshot.get('valid_count', 0)}")
            parts.append(f"覆盖 {float(snapshot.get('coverage_ratio') or 0) * 100:.0f}%")
            leader = snapshot.get("leader") or {}
            if leader.get("name") or leader.get("symbol"):
                parts.append(
                    f"领涨 {leader.get('name') or leader.get('symbol')} "
                    f"{float(leader.get('change_pct') or 0) * 100:+.2f}%"
                )
        return "｜".join(parts)

    def min_abnormal_closeness(self) -> float:
        """启用的 abnormal 规则中最小的接近度阈值 (小数)。

        供调用方 (quote_service) 构建异动快照时预过滤, 不必按最高阈值拉全量。
        """
        thresholds = [
            float(r.get("threshold_pct", 70)) / 100
            for r in list(self._rules.values())
            if r.get("enabled", True) and r.get("type") == "abnormal"
        ]
        return min(thresholds) if thresholds else 1.0

    def evaluate_abnormal(self, rows: list[dict], *, now: float | None = None) -> list[dict]:
        """按异动边缘快照评估 type=abnormal 规则。

        rows 为 abnormal_moves.build_overview 的 rows (调用方已按
        min_abnormal_closeness 预过滤)。rows 为空也照常评估 —— 用于把
        已消失标的的边缘状态清理回 False。
        """
        rules = [
            rule for rule in list(self._rules.values())
            if rule.get("enabled", True) and rule.get("type") == "abnormal"
        ]
        if not rules:
            return []
        timestamp = time.time() if now is None else now
        events: list[dict] = []
        for rule in rules:
            try:
                events.extend(self._evaluate_abnormal_rule(rule, rows, timestamp))
            except Exception as exc:  # noqa: BLE001
                logger.warning("异动规则评估失败 %s: %s", rule.get("id"), exc)
        return events

    def _evaluate_abnormal_rule(self, rule: dict, rows: list[dict], now: float) -> list[dict]:
        events: list[dict] = []
        threshold = float(rule.get("threshold_pct", 70)) / 100
        if not 0 < threshold <= 1.5:
            threshold = 0.7
        direction = rule.get("direction", "both")
        window_filter = str(rule.get("abnormal_window", "any"))
        if rule.get("scope") == "symbols":
            scope_symbols = {str(s) for s in rule.get("symbols", []) if s}
        elif rule.get("scope") == "watchlist_group":
            # 异动规则同样支持动态分组; 分组已删除返回 None → 本轮整体跳过
            members = _group_members_or_none(rule)
            if members is None:
                return events
            scope_symbols = set(members)
        else:
            scope_symbols = None

        seen: set[str] = set()
        for row in rows:
            symbol = str(row.get("symbol") or "")
            if not symbol or (scope_symbols is not None and symbol not in scope_symbols):
                continue
            seen.add(symbol)
            # 方向/窗口过滤后取接近度最高的窗口作为代表
            best: tuple[str, float, float, float] | None = None  # (窗口, 接近度, 偏离值, 阈值)
            for key, win in (row.get("windows") or {}).items():
                if window_filter != "any" and key != window_filter:
                    continue
                value = win.get("value")
                if value is None:
                    continue
                if direction == "up" and value <= 0:
                    continue
                if direction == "down" and value >= 0:
                    continue
                closeness = float(win.get("closeness") or 0)
                if best is None or closeness > best[1]:
                    best = (key, closeness, float(value), float(win.get("threshold") or 0))
            condition = best is not None and best[1] >= threshold
            state_key = (rule["id"], symbol)
            previous = self._abnormal_condition_state.get(state_key)
            self._abnormal_condition_state[state_key] = condition
            if previous is None or previous or not condition:
                continue

            event_type = f"abnormal_{'up' if best[2] > 0 else 'down'}"
            cooldown_key = (rule["id"], symbol, event_type)
            last = self._last_fire.get(cooldown_key)
            cooldown = int(rule.get("cooldown_seconds", 3600))
            if last is not None and now - last < cooldown:
                continue
            self._last_fire[cooldown_key] = now
            event = {
                "ts": int(now * 1000),
                "rule_id": rule["id"],
                "rule_name": rule.get("name", ""),
                "strategy_id": None,
                "source": "abnormal",
                "type": event_type,
                "symbol": symbol,
                "name": row.get("name"),
                "message": rule.get("message", "") or self._abnormal_message(row, best),
                "price": row.get("close"),
                "change_pct": row.get("rt_pct"),
                "signals": [],
                "severity": rule.get("severity", "info"),
                "conditions": [],
                "logic": "and",
                "abnormal_window": best[0],
                "abnormal_value": round(best[2], 4),
                "abnormal_threshold": best[3],
                "abnormal_closeness": round(best[1], 4),
            }
            events.append(event)
            if self._alert_handler:
                try:
                    self._alert_handler(event)
                except Exception as exc:  # noqa: BLE001
                    logger.warning("alert handler failed: %s", exc)
        # 本轮未出现的标的 (跌出预过滤区间) 状态置 False 而非删除:
        # 删除会被当成「首轮观测」而不触发, 置 False 才能在回升穿过阈值时再次告警。
        for key, value in list(self._abnormal_condition_state.items()):
            if key[0] == rule["id"] and key[1] not in seen and value:
                self._abnormal_condition_state[key] = False
        return events

    @staticmethod
    def _abnormal_message(row: dict, best: tuple[str, float, float, float]) -> str:
        window, closeness, value, threshold = best
        board = row.get("board") or ""
        tag = f"{board}{'·ST' if row.get('st') else ''}"
        state = "已达异常波动阈值" if closeness >= 1 else "接近异常波动阈值"
        return (
            f"{row.get('name') or row.get('symbol')} {window}偏离值 "
            f"{value * 100:+.2f}%/阈值{threshold * 100:.0f}% ({tag}) "
            f"接近度{closeness * 100:.0f}%, {state}"
        )

    def _evaluate_rule(self, df: pl.DataFrame, rule: dict, now: float) -> list[dict]:
        """评估单条规则,返回触发的 events。"""
        # 1. 按 scope 过滤作用域
        scoped = self._apply_scope(df, rule)
        if scoped.is_empty():
            return []

        # 2. 根据 type 构建命中集
        #    元组格式: (event_type, symbol, name, price, pct, signals)
        hit_rows: list[tuple[str, str, Any, Any, Any, list[str]]] = []

        rtype = rule.get("type", "signal")
        if rtype == "strategy":
            # 策略类型: 跑策略选股, 同时产出所选的信号和结果池变更事件
            hit_rows = self._match_strategy(scoped, rule)
        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):
                hit_rows.append((rtype, sym, name, price, pct, hit_sigs))

        if not hit_rows:
            return []

        # 3. cooldown 去重 + 生成 events
        cooldown = rule.get("cooldown_seconds", 3600)
        severity = rule.get("severity", "info")
        source = rtype

        events: list[dict] = []
        for ev_type, sym, name, price, pct, hit_sigs in hit_rows:
            # cooldown 键包含事件类型, 同股不同策略事件互不压制。
            is_batch = sym == "_batch"
            key_symbol = f"_{ev_type}_batch" if is_batch else sym
            key = (rule["id"], key_symbol, ev_type)
            last = self._last_fire.get(key)
            if last is not None and (now - last) < cooldown:
                continue  # 冷却期内, 跳过
            self._last_fire[key] = now

            # 批量事件: name 存放预构建的消息文本
            if is_batch:
                resolved_name = ""
                message = name  # name 字段即批量消息
            else:
                resolved_name = name if name else self._name_map.get(sym)
                message = rule.get("message", "") or self._default_message(
                    rule, ev_type=ev_type, sym=sym, name=resolved_name,
                    pct=pct, price=price,
                    conditions=list(rule.get("conditions", [])) if rule.get("type") != "strategy" else None,
                )

            ev = {
                "ts": int(now * 1000),
                "rule_id": rule["id"],
                "rule_name": rule.get("name", ""),
                "strategy_id": rule.get("strategy_id") if rtype == "strategy" else None,
                "source": source,
                "type": ev_type,
                "symbol": "" if is_batch else sym,
                "name": resolved_name,
                "message": message,
                "price": price,
                "change_pct": pct,
                "signals": hit_sigs,
                "severity": severity,
                # 触发条件快照 (signal/price/market 类型): 用于触发记录展示
                # 「命中了什么条件」。strategy 类型靠策略选股池 diff, 不写条件。
                "conditions": list(rule.get("conditions", [])) if rtype != "strategy" else [],
                "logic": rule.get("logic", "and") if rtype != "strategy" else "and",
            }
            events.append(ev)
            if self._alert_handler:
                try:
                    self._alert_handler(ev)
                except Exception as e:
                    logger.warning("alert handler failed: %s", e)

        return events

    @staticmethod
    def _apply_scope(df: pl.DataFrame, rule: dict) -> pl.DataFrame:
        """按 scope 过滤 DataFrame。"""
        scope = rule.get("scope", "symbols")
        if scope == "all":
            return df
        if scope == "symbols":
            syms = rule.get("symbols", [])
            if not syms:
                return df.head(0)
            return df.filter(pl.col("symbol").is_in(syms))
        if scope == "watchlist_group":
            # 动态绑定自选分组: 每轮评估按分组当前成员过滤 (带版本号缓存)。
            # 分组已删除/暂时为空 → fail-closed 返回空, 绝不退化为全市场。
            members = _group_members_or_none(rule)
            if not members:
                return df.head(0)
            return df.filter(pl.col("symbol").is_in(list(members)))
        if scope == "sector":
            # sector 过滤需 df 含板块列 (后续接入 ext_data JOIN)。在 JOIN 落地前
            # fail-closed 返回空 —— 绝不退化为「全市场」误触发 (旧行为 return df 会让
            # 一条板块规则对全市场每只命中都告警)。新建 sector 规则已在 validate 拦截,
            # 此处兜底任何历史遗留的 sector 规则。
            logger.warning("scope=sector 规则 %s 暂不支持(板块 JOIN 未实现), 本轮跳过",
                           rule.get("id"))
            return df.head(0)
        return df

    def _match_strategy(
        self, df: pl.DataFrame, rule: dict,
    ) -> list[tuple[str, str, Any, Any, Any, list[str]]]:
        """策略类型评估: 一次执行同时产出交易信号和结果池变更事件。

        返回 [(event_type, symbol, name, price, pct, signals)]
        event_type: buy_signal | sell_signal | pool_entry | pool_exit
        同类事件超过 5 只时合并为一条批量事件 (symbol="_batch")
        """
        if self._strategy_engine is None:
            return []
        sid = rule.get("strategy_id")
        if not sid:
            return []
        at = rule.get("asset_type", "stock")
        pool_key = (str(rule.get("id", sid)), sid, at)
        try:
            s = self._strategy_engine.get(sid)
        except Exception:
            return []
        if s is None:
            return []

        # 运行策略选股: 复用当前 enriched DataFrame 跳过数据加载
        overrides = {}
        if self._data_dir:
            try:
                overrides = _strategy_config.load_override(self._data_dir, sid)
            except Exception:
                pass

        # 声明 filter_history 的策略 (如反包) 需要多日历史窗口才能判定形态。
        # 旧实现因"实时监控不支持 history loader"直接跳过 → 反包等策略盘中永不触发。
        # 现接入 history_loader, 拼历史窗口 + 今日实时行情, 经 precomputed_history 喂给引擎。
        # loader 为 None (未装配) 时退回跳过, 保持旧行为, 不破坏无历史场景。
        from app.strategy.engine import StrategyDataContext
        current_context = StrategyDataContext(
            asset_type=at,
            timeframe="1d",
            as_of=cn_today(),
            current=df,
        )
        if getattr(s, "execution_backend", "polars_expr") == "composite":
            # 叠加策略首版不支持实时监控: 各子策略需独立预热实时矩阵, 热路径成本为 N 倍,
            # 违反"实时热路径不得随历史数据量线性增长"约束。fail-closed 跳过本轮,
            # /cached 端点回退到盘后 strategy_cache.json 的批量结果。
            logger.debug("叠加策略 %s 暂不支持实时监控, 跳过", sid)
            return []
        if getattr(s, "execution_backend", "polars_expr") == "matrix_native":
            matrix = self._active_matrix_snapshots.get(at)
            if matrix is None:
                logger.debug("策略 %s 缺少本轮实时矩阵快照, 跳过", sid)
                return []
            current_context = StrategyDataContext(
                asset_type=at,
                timeframe="1d",
                as_of=cn_today(),
                current=df,
                market=matrix,
            )
        required_history_bars = 1
        history_resolver = getattr(self._strategy_engine, "required_history_bars", None)
        if callable(history_resolver):
            required_history_bars = history_resolver(
                [sid],
                overrides_map={sid: overrides},
            )
        if getattr(s, "execution_backend", "polars_expr") not in {"composite", "matrix_native"} and (
            s.filter_history_fn or required_history_bars > 1
        ):
            history_loader = self._history_loader_for(rule)
            if history_loader is None:
                logger.debug("策略 %s 需要历史数据但未注入 history_loader (asset_type=%s), 跳过实时监控",
                             sid, rule.get("asset_type", "stock"))
                return []
            try:
                today = cn_today()
                lookback = max(1, getattr(s, "lookback_days", 1), required_history_bars)
                hist_df = history_loader(today, lookback)
                if hist_df is None or hist_df.is_empty():
                    logger.debug("策略 %s 历史数据为空, 跳过本轮实时监控", sid)
                    return []
                # 历史窗口可能与今日已落盘数据重叠: 排掉 hist_df 中 date==today 的行,
                # 今日行情始终以实时 df 为准 (盘中逐轮更新, 最接近收盘真相)。
                # 否则 today 行重复会污染 filter_history 的 .over("symbol") 窗口判定。
                if "date" in hist_df.columns:
                    hist_df = hist_df.filter(pl.col("date") != today)
                # 拼接历史窗口 + 今日实时行情 (filter_history 用 .over("symbol") 窗口, 多日天然可用)
                current_context = StrategyDataContext(
                    asset_type=at,
                    timeframe="1d",
                    as_of=today,
                    current=df,
                    history=pl.concat(
                        [hist_df, df], how="diagonal_relaxed"
                    ),
                )
            except Exception as e:
                logger.warning("策略 %s 加载历史窗口失败, 跳过: %s", sid, e)
                return []
        try:
            result = self._strategy_engine.run(
                sid,
                current_context,
                pool=(df["symbol"].cast(pl.Utf8).to_list()
                      if getattr(s, "execution_backend", "polars_expr") == "matrix_native"
                      else None),
                overrides=overrides,
                params=dict(overrides.get("params") or {}),
            )
        except Exception as e:
            logger.warning("策略 %s 选股执行失败: %s", sid, e)
            return []

        # 记录本轮完整选股结果 (供策略页实时回显: /cached 端点直接读取, 不落盘)。
        # 与下面的事件无关, 无论是否产生通知结果都用于策略页实时回显。
        # 策略结果缓存仅用于股票策略页 /cached 回显; ETF 策略页走实时单跑, 不写入。
        # 写到 evaluate 提供的临时容器 (_building_strategy_results), 算完后整体替换,
        # 避免并发读到半填充状态。
        if at == "stock":
            try:
                self._building_strategy_results[sid] = {
                    "total": result.total,
                    "as_of": str(cn_today()),
                    "rows": [
                        {k: (None if isinstance(v, float) and not math.isfinite(v) else v)
                         for k, v in row.items()}
                        for row in result.rows
                    ],
                }
                self._latest_strategy_result_ids.add(sid)
            except Exception:  # noqa: BLE001
                pass

        score_min = rule.get("score_min")
        score_max = rule.get("score_max")
        score_filter_enabled = score_min is not None or score_max is not None
        eligible_symbols: set[str] = set()
        if score_filter_enabled:
            for row in result.rows:
                symbol = str(row.get("symbol", ""))
                score = row.get("score", result.scores.get(symbol))
                if isinstance(score, bool) or not isinstance(score, (int, float)):
                    continue
                if not math.isfinite(score):
                    continue
                if score_min is not None and score < score_min:
                    continue
                if score_max is not None and score > score_max:
                    continue
                eligible_symbols.add(symbol)
        else:
            eligible_symbols = {str(row["symbol"]) for row in result.rows}

        current_pool = eligible_symbols
        prev_pool = self._strategy_pools.get(pool_key)
        self._strategy_pools[pool_key] = current_pool

        notify_events = set(rule.get("notify_events") or ("pool_entry", "pool_exit"))
        sname = s.meta.get("name", "") or s.meta.get("id", sid)
        row_map: dict[str, dict] = {r["symbol"]: r for r in result.rows}
        try:
            for row in df.iter_rows(named=True):
                row_map.setdefault(str(row.get("symbol", "")), row)
        except Exception:
            pass

        entry_signal_hits = result.entry_signal_hits
        if score_filter_enabled:
            entry_signal_hits = [
                hit for hit in entry_signal_hits
                if str(hit.get("symbol", "")) in eligible_symbols
            ]
        changes: dict[str, set[str]] = {
            "buy_signal": self._new_strategy_signals(
                pool_key, "buy_signal", result.as_of, entry_signal_hits,
            ),
            "sell_signal": self._new_strategy_signals(
                pool_key, "sell_signal", result.as_of, result.exit_signal_hits,
            ),
            "pool_entry": set() if prev_pool is None else current_pool - prev_pool,
            "pool_exit": set() if prev_pool is None else prev_pool - current_pool,
        }

        results: list[tuple[str, str, Any, Any, Any, list[str]]] = []
        signal_map = {
            "buy_signal": {
                str(hit["symbol"]): list(hit.get("signals") or [])
                for hit in entry_signal_hits
            },
            "sell_signal": {
                str(hit["symbol"]): list(hit.get("signals") or [])
                for hit in result.exit_signal_hits
            },
        }
        action_labels = {
            "buy_signal": "买入信号",
            "sell_signal": "卖出信号",
            "pool_entry": "进入选股结果",
            "pool_exit": "移出选股结果",
        }
        for event_type, symbols in changes.items():
            if event_type not in notify_events or not symbols:
                continue
            symbol_list = sorted(symbols)
            if len(symbol_list) > 5:
                names = [
                    str(row_map.get(symbol, {}).get("name") or self._name_map.get(symbol, symbol))
                    for symbol in symbol_list
                ]
                message = (
                    f"策略「{sname}」{action_labels[event_type]} {len(symbol_list)} 只: "
                    f"{'、'.join(names)}"
                )
                hit_signals = sorted({
                    signal
                    for symbol in symbol_list
                    for signal in signal_map.get(event_type, {}).get(symbol, [])
                })
                results.append((event_type, "_batch", message, None, None, hit_signals))
                continue
            for symbol in symbol_list:
                row = row_map.get(symbol, {})
                name = row.get("name") or self._name_map.get(symbol, symbol)
                results.append((
                    event_type,
                    symbol,
                    name,
                    row.get("close"),
                    row.get("change_pct"),
                    signal_map.get(event_type, {}).get(symbol, []),
                ))

        return results

    def _new_strategy_signals(
        self,
        pool_key: tuple[str, str, str],
        event_type: str,
        as_of: Any,
        hits: list[dict],
    ) -> set[str]:
        rule_id, strategy_id, asset_type = pool_key
        state_key = (rule_id, strategy_id, asset_type, event_type)
        date_key = str(as_of)
        current = {str(hit["symbol"]) for hit in hits}
        previous = self._strategy_signal_state.get(state_key)
        self._strategy_signal_state[state_key] = (date_key, current)

        if previous is None:
            for symbol in current:
                self._strategy_signal_seen[(*state_key, symbol)] = date_key
            return set()

        previous_date, previous_symbols = previous
        candidates = current if previous_date != date_key else current - previous_symbols
        fresh = {
            symbol
            for symbol in candidates
            if self._strategy_signal_seen.get((*state_key, symbol)) != date_key
        }
        for symbol in fresh:
            self._strategy_signal_seen[(*state_key, symbol)] = date_key
        return fresh

    @staticmethod
    def _match_conditions(
        df: pl.DataFrame, rule: dict,
    ) -> list[tuple[str, Any, Any, Any, list[str]]]:
        """按 conditions + logic 匹配,返回命中行 [(symbol,name,price,pct,signals)]。"""
        conditions = rule.get("conditions", [])
        logic = rule.get("logic", "and")
        if not conditions:
            return []
        hit_df = _build_condition_mask(df, conditions, logic)
        results = []
        for row in hit_df.iter_rows(named=True):
            sym = row.get("symbol", "")
            name = row.get("name")
            price = row.get("close")
            pct = row.get("change_pct")
            # 收集命中的信号列名 (仅 op=truth 且为真的)
            hit_sigs = [
                c["field"] for c in conditions
                if c.get("op") == "truth" and row.get(c["field"])
            ]
            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]:
        """评估连板梯队封单监控规则。

        封单量从注入的临时列 _sealed_vol (手) 读取 (由 quote_service 评估前注入)。
        命中条件: 封单比较值 <= threshold (且封单 > 0, 排除无 depth 数据的股票)。
        涨停(direction=up) → 炸板预警; 跌停(direction=down) → 翘板预警。
        """
        if "_sealed_vol" not in scoped.columns:
            return []  # 无封单数据 (depth 未拉取), 安全降级

        metric = rule.get("metric", "sealed_vol")
        threshold = rule.get("threshold", 0)
        direction = rule.get("direction", "up")
        cooldown = rule.get("cooldown_seconds", 600)
        severity = rule.get("severity", "warn")

        # 比较值: sealed_vol 直接用 (手), sealed_amount = 手 × 100股 × close
        if metric == "sealed_amount":
            cmp_expr = pl.col("_sealed_vol") * 100 * pl.col("close")
            unit = "元"
        else:
            cmp_expr = pl.col("_sealed_vol")
            unit = "手"

        # 命中: 封单 > 0 (有数据) 且 比较值 <= 阈值
        hit = scoped.filter(
            pl.col("_sealed_vol").is_not_null()
            & (pl.col("_sealed_vol") > 0)
            & (cmp_expr <= threshold)
        )
        if hit.is_empty():
            return []

        warn_label = "炸板预警" if direction == "up" else "翘板预警"
        events: list[dict] = []
        for row in hit.iter_rows(named=True):
            sym = row.get("symbol", "")
            key = (rule["id"], sym, "ladder")
            last = self._last_fire.get(key)
            if last is not None and (now - last) < cooldown:
                continue
            self._last_fire[key] = now

            name = row.get("name") or self._name_map.get(sym) or sym
            price = row.get("close")
            pct = row.get("change_pct")
            sealed_vol = row.get("_sealed_vol")
            # 预警封单值 (展示用)
            sealed_value = sealed_vol * 100 * (price or 0) if metric == "sealed_amount" else sealed_vol

            # message 体现预警封单量 + 阈值
            if metric == "sealed_amount":
                sv_text = f"{sealed_value / 1e4:.0f}万{unit}"
                th_text = f"{threshold / 1e4:.0f}万{unit}"
            else:
                sv_text = f"{sealed_value:,.0f} {unit}"
                th_text = f"{threshold:,.0f} {unit}"
            message = f"{warn_label} · 封单 {sv_text} ≤ {th_text}"

            events.append({
                "ts": int(now * 1000),
                "rule_id": rule["id"],
                "rule_name": rule.get("name", ""),
                "source": "ladder",
                "type": warn_label,
                "symbol": sym,
                "name": name,
                "message": message,
                "price": price,
                "change_pct": pct,
                "signals": [],
                "severity": severity,
                "conditions": [],
                "logic": "and",
                "sealed_value": sealed_value,   # 预警封单量/额 (飞书+记录展示)
                "sealed_metric": metric,
            })
        return events

    def _default_message(self, rule: dict, ev_type: str = "", sym: str = "",
                          name: str = "", pct: Any = None, price: Any = None,
                          conditions: list[dict] | None = None) -> str:
        """生成默认 message。

        - strategy: 按变更方向生成 (进入/移出 + 涨跌幅)
        - signal/price/market: 条件摘要 + 现价 + 涨跌幅 (避免笼统的「信号触发」)
        """
        rtype = rule.get("type", "signal")
        if rtype == "strategy":
            # 从 StrategyEngine 取策略名; 失败则退化为 rule_name 里截取的部分
            sname = ""
            sid = rule.get("strategy_id")
            if sid and self._strategy_engine is not None:
                try:
                    s = self._strategy_engine.get(sid)
                    sname = s.meta.get("name", "") or s.meta.get("id", "")
                except Exception:  # noqa: BLE001
                    sname = ""
            if not sname:
                rn = rule.get("name", "")
                sname = rn.split(" · ", 1)[1] if " · " in rn else (rn or "策略")

            action = {
                "buy_signal": "买入信号",
                "sell_signal": "卖出信号",
                "pool_entry": "进入选股结果",
                "pool_exit": "移出选股结果",
                "new_entry": "进入选股结果",
                "dropped": "移出选股结果",
            }.get(ev_type)
            if action:
                pct_text = ""
                if pct is not None:
                    sign = "+" if pct >= 0 else ""
                    pct_text = f" {sign}{pct * 100:.1f}%"
                return f"策略「{sname}」{action} {name}{pct_text}"
            return f"策略「{sname}」事件"

        # signal / price / market: 条件摘要 + 现价 + 涨跌幅
        # 条件摘要: 把 conditions (truth/比较) 拼成可读串, 如 "MA20金叉 且 量比>2"
        cond_text = self._format_conditions_text(rule, conditions)
        price_text = f"现价 {price}" if price is not None else ""
        pct_text = ""
        if pct is not None:
            sign = "+" if pct >= 0 else ""
            pct_text = f"{sign}{pct * 100:.1f}%"
        tail = " · ".join(s for s in (price_text, pct_text) if s)
        if cond_text and tail:
            return f"{cond_text} · {tail}"
        return cond_text or tail or "监控触发"

    @staticmethod
    def _format_conditions_text(rule: dict, conditions: list[dict] | None) -> str:
        """把 rule.conditions 拼成可读文本 (用于 message / 推送)。

        op=truth: 直接用信号中文名 (如 "MA20金叉")
        op=比较: 字段中文名 + 操作符 + 值 (如 "涨跌幅≥5")
        logic: and → "且", or → "或"
        """
        conds = conditions if conditions is not None else list(rule.get("conditions", []))
        if not conds:
            return ""
        logic_word = "且" if rule.get("logic", "and") == "and" else "或"
        parts: list[str] = []
        for c in conds:
            field = c.get("field", "")
            op = c.get("op", "truth")
            value = c.get("value")
            label = _signal_cn_name(field) or field
            if op == "truth":
                parts.append(label)
            else:
                op_map = {"gte": "≥", "lte": "≤", "gt": ">", "lt": "<", "eq": "="}
                parts.append(f"{label}{op_map.get(op, op)}{value}")
        return f" {logic_word} ".join(parts)