From b43b177899b1fcefb7c183be5f4c3323ce234ef4 Mon Sep 17 00:00:00 2001 From: shy3130 Date: Sun, 26 Jul 2026 16:08:42 +0800 Subject: [PATCH] feat(monitor): add strategy signal event controls --- backend/app/api/monitor_rules.py | 1 + backend/app/services/quote_service.py | 2 +- backend/app/strategy/engine.py | 106 +++++- backend/app/strategy/monitor.py | 257 ++++++++----- backend/app/strategy/monitor_rules.py | 20 +- backend/tests/test_strategy_monitor_events.py | 358 ++++++++++++++++++ frontend/src/components/AlertToast.tsx | 56 ++- .../src/components/monitor/RuleEditor.tsx | 112 ++++-- frontend/src/lib/api.ts | 3 + frontend/src/lib/strategyMonitorEvents.ts | 73 ++++ frontend/src/lib/voiceBroadcast.ts | 12 +- frontend/src/pages/Dashboard.tsx | 52 ++- frontend/src/pages/Monitor.tsx | 49 ++- frontend/src/pages/Screener.tsx | 2 + 14 files changed, 912 insertions(+), 191 deletions(-) create mode 100644 backend/tests/test_strategy_monitor_events.py create mode 100644 frontend/src/lib/strategyMonitorEvents.ts diff --git a/backend/app/api/monitor_rules.py b/backend/app/api/monitor_rules.py index 26e614d..9cd0c30 100644 --- a/backend/app/api/monitor_rules.py +++ b/backend/app/api/monitor_rules.py @@ -46,6 +46,7 @@ class RuleModel(BaseModel): sector: str | None = None strategy_id: str | None = None direction: str = "entry" # entry | exit | both + notify_events: list[str] | None = None conditions: list[ConditionModel] = [] logic: str = "and" # and | or cooldown_seconds: int = 3600 diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 1dbbed8..eaf29aa 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -1059,7 +1059,7 @@ class QuoteService: "source": ev["source"], "type": ev["type"], "rule_id": ev.get("rule_id"), - "strategy_id": ev.get("rule_id") if ev["source"] == "strategy" else None, + "strategy_id": ev.get("strategy_id") if ev["source"] == "strategy" else None, "symbol": ev["symbol"], "name": ev["name"], "message": ev["message"], diff --git a/backend/app/strategy/engine.py b/backend/app/strategy/engine.py index fa0fbe9..ecbcc71 100644 --- a/backend/app/strategy/engine.py +++ b/backend/app/strategy/engine.py @@ -143,6 +143,8 @@ class StrategyResult: total: int = 0 elapsed_ms: float = 0.0 scores: dict[str, float] = field(default_factory=dict) + entry_signal_hits: list[dict] = field(default_factory=list) + exit_signal_hits: list[dict] = field(default_factory=list) @dataclass @@ -629,6 +631,8 @@ class StrategyEngine: as_of = context.as_of overrides = overrides or {} params = self.resolve_params(s, params, overrides) + entry_signals = self._effective_signals(overrides, "entry_signals", s.entry_signals) + exit_signals = self._effective_signals(overrides, "exit_signals", s.exit_signals) if s.execution_backend == "matrix_native": return self._run_matrix_strategy( @@ -642,24 +646,40 @@ class StrategyEngine: started_at=t0, ) + signal_df = context.current if context.current is not None else context.history + if signal_df is None: + signal_df = pl.DataFrame() + if not signal_df.is_empty() and "date" in signal_df.columns: + signal_df = signal_df.filter(pl.col("date") == as_of) + if pool and not signal_df.is_empty(): + signal_df = signal_df.filter(pl.col("symbol").is_in(pool)) + exit_signal_hits = self._collect_signal_hits(signal_df, exit_signals) + # 普通策略只读目标日期；历史策略读取调用方注入的历史窗口。 if s.filter_history_fn: if context.history is None: raise ValueError(f"strategy {strategy_id} requires history data") df = context.history if df.is_empty(): - return StrategyResult(as_of=as_of, strategy_id=strategy_id) + return StrategyResult( + as_of=as_of, + strategy_id=strategy_id, + exit_signal_hits=exit_signal_hits, + ) df = s.filter_history_fn(df, params) - if df.is_empty(): - return StrategyResult(as_of=as_of, strategy_id=strategy_id) if "date" in df.columns: df = df.filter(pl.col("date") == as_of) else: if context.current is None: raise ValueError(f"strategy {strategy_id} requires current data") df = context.current - if df.is_empty(): - return StrategyResult(as_of=as_of, strategy_id=strategy_id) + + if df.is_empty(): + return StrategyResult( + as_of=as_of, + strategy_id=strategy_id, + exit_signal_hits=exit_signal_hits, + ) # 基础过滤: 策略默认 basic_filter 兜底, 用户 override 优先覆盖。 # 这样策略文件里写的 exclude_st/price_min 等默认值即使前端没保存也能生效。 @@ -686,6 +706,12 @@ class StrategyEngine: if scoring_overrides: scoring = {**scoring, **scoring_overrides} df = self._apply_scoring(df, scoring) + entry_signal_hits = self._collect_signal_hits(df, entry_signals) + if not entry_signals and (s.filter_history_fn or s.filter_fn): + entry_signal_hits = [ + {"symbol": str(symbol), "signals": []} + for symbol in df["symbol"].cast(pl.Utf8).unique().to_list() + ] # 排序 + 限制 limit = self._result_limit(s, overrides) @@ -715,8 +741,41 @@ class StrategyEngine: total=len(rows), elapsed_ms=elapsed, scores=scores, + entry_signal_hits=entry_signal_hits, + exit_signal_hits=exit_signal_hits, ) + @staticmethod + def _effective_signals(overrides: dict, key: str, default: list[str]) -> list[str]: + value = overrides.get(key) + if isinstance(value, list): + return [str(signal) for signal in value if signal] + return list(default or []) + + @staticmethod + def _collect_signal_hits(df: pl.DataFrame, signals: list[str]) -> list[dict]: + if df.is_empty() or not signals or "symbol" not in df.columns: + return [] + resolved = [ + signal if signal.startswith(("signal_", "csg_")) else f"signal_{signal}" + for signal in signals + ] + available = [ + (signal, column) + for signal, column in zip(signals, resolved, strict=True) + if column in df.columns + ] + if not available: + return [] + hit_df = df.filter(pl.any_horizontal(pl.col(column).fill_null(False) for _, column in available)) + return [ + { + "symbol": str(row["symbol"]), + "signals": [signal for signal, column in available if row.get(column)], + } + for row in hit_df.iter_rows(named=True) + ] + def run_all( self, context: StrategyDataContext, @@ -880,12 +939,31 @@ class StrategyEngine: if not target_ids: return StrategyResult(as_of=as_of, strategy_id=strategy_id) target_time = target_ids[-1] - selected_assets = np.flatnonzero(signals.entry[target_time] != 0) + entry_active = signals.entry[target_time] + exit_active = signals.exit[target_time] + if asset_mask is not None: + entry_active = entry_active & asset_mask + exit_active = exit_active & asset_mask + entry_signal_hits = self._matrix_signal_hits( + entry_active, + signals.entry_signal_code[target_time], + signals.entry_signal_ids, + market.symbols, + ) + exit_signal_hits = self._matrix_signal_hits( + exit_active, + signals.exit_signal_code[target_time], + signals.exit_signal_ids, + market.symbols, + ) + selected_assets = np.flatnonzero(entry_active != 0) if selected_assets.size == 0: return StrategyResult( as_of=as_of, strategy_id=strategy_id, elapsed_ms=(time.perf_counter() - started_at) * 1000, + entry_signal_hits=entry_signal_hits, + exit_signal_hits=exit_signal_hits, ) target_frame = self._matrix_target_frame(source_panel, as_of) @@ -916,8 +994,24 @@ class StrategyEngine: total=len(rows), elapsed_ms=(time.perf_counter() - started_at) * 1000, scores=scores, + entry_signal_hits=entry_signal_hits, + exit_signal_hits=exit_signal_hits, ) + @staticmethod + def _matrix_signal_hits( + active: np.ndarray, + codes: np.ndarray, + signal_ids: tuple[str, ...], + symbols: tuple[str, ...], + ) -> list[dict]: + hits = [] + for asset_id in np.flatnonzero(active != 0): + code = int(codes[int(asset_id)]) + signals = [signal_ids[code]] if 0 <= code < len(signal_ids) else [] + hits.append({"symbol": symbols[int(asset_id)], "signals": signals}) + return hits + @staticmethod def _matrix_target_frame(panel: pl.DataFrame, as_of: date) -> pl.DataFrame: if "datetime" in panel.columns: diff --git a/backend/app/strategy/monitor.py b/backend/app/strategy/monitor.py index 61ce1c6..ed49b78 100644 --- a/backend/app/strategy/monitor.py +++ b/backend/app/strategy/monitor.py @@ -317,19 +317,23 @@ class MonitorRuleEngine: - 规则来自 monitor_rules 存储 (用户可配), 而非写死的 strategy config - 支持 scope (symbols/all/sector) 过滤作用域 - 支持 conditions + logic (AND/OR) 任意组合 - - ★ cooldown 去重: 同一 (rule_id, symbol) 在冷却期内不重复触发 + - ★ 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) → 上次触发时间戳(秒)。用于 cooldown 去重。 - self._last_fire: dict[tuple[str, str], float] = {} + # (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] = {} - # 策略选股池状态: strategy_id → 上期选股符号集合 (用于 diff 变更) + # 策略选股池状态: (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。 @@ -361,6 +365,8 @@ class MonitorRuleEngine: 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() @@ -398,6 +404,17 @@ class MonitorRuleEngine: 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("asset_type", "stock"), + rule.get("scope", "symbols"), + tuple(sorted(str(symbol) for symbol in rule.get("symbols", []))), + rule.get("sector"), + ) + def set_rules(self, rules: list[dict]) -> None: """批量设置规则 (覆盖)。用于启动时 reload。 @@ -408,7 +425,31 @@ class MonitorRuleEngine: 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 + } logger.info("MonitorRuleEngine: 装载 %d 条规则", len(self._rules)) def add_rule(self, rule: dict) -> None: @@ -419,12 +460,23 @@ class MonitorRuleEngine: def remove_rule(self, rule_id: str) -> None: self._rules.pop(rule_id, None) - # 清理对应的 cooldown 记录 (list 快照: 评估线程可能并发写 _last_fire) 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 + } 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() @property def rules(self) -> dict[str, dict]: @@ -613,7 +665,7 @@ class MonitorRuleEngine: rtype = rule.get("type", "signal") if rtype == "strategy": - # 策略类型: 跑策略选股 → 对比上期选股池 → 产出 new_entry/dropped 事件 + # 策略类型: 跑策略选股, 同时产出所选的信号和结果池变更事件 hit_rows = self._match_strategy(scoped, rule) elif rtype == "ladder": # 连板梯队封单监控: 独立处理 (需带预警封单值, 走专属 message) @@ -633,12 +685,10 @@ class MonitorRuleEngine: events: list[dict] = [] for ev_type, sym, name, price, pct, hit_sigs in hit_rows: - # cooldown 键: 批量事件用特殊键, 单只事件用 (rule_id, symbol) + # cooldown 键包含事件类型, 同股不同策略事件互不压制。 is_batch = sym == "_batch" - if is_batch: - key = (rule["id"], f"_{ev_type}_batch") - else: - key = (rule["id"], sym) + 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 # 冷却期内, 跳过 @@ -660,6 +710,7 @@ class MonitorRuleEngine: "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, @@ -707,11 +758,11 @@ class MonitorRuleEngine: 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: "new_entry" (新入选) | "dropped" (已移出) - 单只变更逐只返回; 同一策略 >5 只合并为一条批量事件 (symbol="_batch") + event_type: buy_signal | sell_signal | pool_entry | pool_exit + 同类事件超过 5 只时合并为一条批量事件 (symbol="_batch") """ if self._strategy_engine is None: return [] @@ -804,7 +855,7 @@ class MonitorRuleEngine: return [] # 记录本轮完整选股结果 (供策略页实时回显: /cached 端点直接读取, 不落盘)。 - # 与下面的 diff 事件无关 — 无论是否产生 new_entry/dropped, 结果都该可用于回显。 + # 与下面的事件无关, 无论是否产生通知结果都用于策略页实时回显。 # 策略结果缓存仅用于股票策略页 /cached 回显; ETF 策略页走实时单跑, 不写入。 # 写到 evaluate 提供的临时容器 (_building_strategy_results), 算完后整体替换, # 避免并发读到半填充状态。 @@ -826,75 +877,109 @@ class MonitorRuleEngine: current_pool: set[str] = {r["symbol"] for r in result.rows} prev_pool = self._strategy_pools.get(pool_key) - - # 首次运行: 仅记录当前选股池, 不产生事件 - if prev_pool is None: - self._strategy_pools[pool_key] = current_pool - return [] - - new_entries = current_pool - prev_pool - dropped = prev_pool - current_pool - - # 无变更 - if not new_entries and not dropped: - return [] - - # 更新存储 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) - - # 构建查找表 (新入选股票可在 result.rows 中找到; 移出股票需从 df 找) row_map: dict[str, dict] = {r["symbol"]: r for r in result.rows} - dropped_map: dict[str, dict] = {} - if dropped: - try: - _dd = df.filter(pl.col("symbol").is_in(list(dropped))) - for row in _dd.iter_rows(named=True): - dropped_map[row["symbol"]] = row - except Exception: - pass + try: + for row in df.iter_rows(named=True): + row_map.setdefault(str(row.get("symbol", "")), row) + except Exception: + pass + + changes: dict[str, set[str]] = { + "buy_signal": self._new_strategy_signals( + pool_key, "buy_signal", result.as_of, result.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]]] = [] - - # ── 新入选 ── - new_list = sorted(new_entries) - if len(new_list) > 5: - names: list[str] = [] - for sym in new_list: - row = row_map.get(sym, {}) - name = row.get("name") or self._name_map.get(sym, sym) - names.append(str(name)) - message = f"策略「{sname}」进入 {len(new_entries)} 只：{'、'.join(names)}" - results.append(("new_entry", "_batch", message, None, None, [])) - else: - for sym in new_list: - row = row_map.get(sym, {}) - name = row.get("name") or self._name_map.get(sym, sym) - price = row.get("close") - pct = row.get("change_pct") - results.append(("new_entry", sym, name, price, pct, [])) - - # ── 已移出 ── - dropped_list = sorted(dropped) - if len(dropped_list) > 5: - names = [] - for sym in dropped_list: - row = dropped_map.get(sym, {}) - name = row.get("name") or self._name_map.get(sym, sym) - names.append(str(name)) - message = f"策略「{sname}」移出 {len(dropped)} 只：{'、'.join(names)}" - results.append(("dropped", "_batch", message, None, None, [])) - else: - for sym in dropped_list: - row = dropped_map.get(sym, {}) - name = row.get("name") or self._name_map.get(sym, sym) - price = row.get("close") - pct = row.get("change_pct") - results.append(("dropped", sym, name, price, pct, [])) + signal_map = { + "buy_signal": { + str(hit["symbol"]): list(hit.get("signals") or []) + for hit in result.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, @@ -956,7 +1041,7 @@ class MonitorRuleEngine: events: list[dict] = [] for row in hit.iter_rows(named=True): sym = row.get("symbol", "") - key = (rule["id"], sym) + key = (rule["id"], sym, "ladder") last = self._last_fire.get(key) if last is not None and (now - last) < cooldown: continue @@ -1021,19 +1106,21 @@ class MonitorRuleEngine: rn = rule.get("name", "") sname = rn.split(" · ", 1)[1] if " · " in rn else (rn or "策略") - if ev_type == "new_entry": + 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}」进入 {name}{pct_text}" - elif ev_type == "dropped": - pct_text = "" - if pct is not None: - sign = "+" if pct >= 0 else "" - pct_text = f" {sign}{pct * 100:.1f}%" - return f"策略「{sname}」移出 {name}{pct_text}" - return f"策略「{sname}」变更" + return f"策略「{sname}」{action} {name}{pct_text}" + return f"策略「{sname}」事件" # signal / price / market: 条件摘要 + 现价 + 涨跌幅 # 条件摘要: 把 conditions (truth/比较) 拼成可读串, 如 "MA20金叉 且 量比>2" diff --git a/backend/app/strategy/monitor_rules.py b/backend/app/strategy/monitor_rules.py index 6b3a359..dd2d2a6 100644 --- a/backend/app/strategy/monitor_rules.py +++ b/backend/app/strategy/monitor_rules.py @@ -31,6 +31,7 @@ RULE_TYPES = {"strategy", "signal", "price", "market", "ladder"} SCOPES = {"symbols", "all", "sector"} LOGICS = {"and", "or"} DIRECTIONS = {"entry", "exit", "both"} +STRATEGY_NOTIFY_EVENTS = {"buy_signal", "sell_signal", "pool_entry", "pool_exit"} SEVERITIES = {"info", "warn", "critical"} OPS = {">", ">=", "<", "<=", "==", "!="} # ladder 规则: 封单监控的指标 (量=手, 额=元) @@ -59,7 +60,7 @@ def load_all(data_dir: Path) -> list[dict]: out: list[dict] = [] for f in sorted(d.glob("*.json")): try: - out.append(json.loads(f.read_text(encoding="utf-8"))) + out.append(normalize(json.loads(f.read_text(encoding="utf-8")))) except Exception as e: logger.warning("monitor rule load failed %s: %s", f.name, e) return out @@ -70,7 +71,7 @@ def load_one(data_dir: Path, rule_id: str) -> dict | None: if not p.exists(): return None try: - return json.loads(p.read_text(encoding="utf-8")) + return normalize(json.loads(p.read_text(encoding="utf-8"))) except Exception as e: logger.warning("monitor rule load failed %s: %s", rule_id, e) return None @@ -112,6 +113,12 @@ def validate(rule: dict) -> None: raise ValueError("策略类型规则必须指定 strategy_id") if rule.get("direction", "entry") not in DIRECTIONS: raise ValueError(f"direction 必须是 {DIRECTIONS} 之一") + notify_events = rule.get("notify_events") + if not isinstance(notify_events, list) or not notify_events: + raise ValueError("策略类型规则至少选择一个通知事件") + invalid_events = set(notify_events) - STRATEGY_NOTIFY_EVENTS + if invalid_events: + raise ValueError(f"notify_events 包含非法事件: {sorted(invalid_events)}") elif rule.get("type") == "ladder": # 连板梯队封单监控: 需 metric + threshold + direction(up/down), 不用 conditions if rule.get("metric", "sealed_vol") not in LADDER_METRICS: @@ -182,6 +189,14 @@ def normalize(rule: dict) -> dict: r.setdefault("strategy_id", None) # direction 默认值: ladder 用 "up", 其余用 "entry" r.setdefault("direction", "up" if r.get("type") == "ladder" else "entry") + if r.get("type") == "strategy": + if r.get("notify_events") is None: + # 兼容统一监控上线后的旧规则: 当时实际行为是同时通知进入和移出。 + r["notify_events"] = ["pool_entry", "pool_exit"] + else: + r["notify_events"] = list(dict.fromkeys(r["notify_events"])) + else: + r.pop("notify_events", None) r.setdefault("conditions", []) # ladder 专属默认字段 r.setdefault("metric", "sealed_vol") @@ -252,6 +267,7 @@ def migrate_strategy_monitors(data_dir: Path, strategy_ids: list[str], strategy_ "scope": "all", "strategy_id": sid, "direction": "entry", + "notify_events": ["pool_entry", "pool_exit"], "conditions": [], "cooldown_seconds": 3600, "enabled": True, diff --git a/backend/tests/test_strategy_monitor_events.py b/backend/tests/test_strategy_monitor_events.py new file mode 100644 index 0000000..6c6af6f --- /dev/null +++ b/backend/tests/test_strategy_monitor_events.py @@ -0,0 +1,358 @@ +from __future__ import annotations + +from datetime import date +from types import SimpleNamespace +from unittest.mock import patch + +import numpy as np +import polars as pl +import pytest + +from app.backtest.matrix import build_market_data_matrix, make_signal_matrix +from app.services import alert_store, preferences, quote_service +from app.services.quote_service import QuoteService +from app.strategy import monitor_rules +from app.strategy.engine import StrategyDataContext, StrategyDef, StrategyEngine, StrategyResult +from app.strategy.monitor import MonitorRuleEngine + + +def _rule(*events: str, **overrides) -> dict: + return monitor_rules.normalize({ + "id": "strategy_rule", + "name": "策略监控", + "type": "strategy", + "asset_type": "stock", + "scope": "all", + "symbols": [], + "strategy_id": "demo", + "notify_events": list(events), + "conditions": [], + "cooldown_seconds": 3600, + **overrides, + }) + + +class _SequenceStrategyEngine: + def __init__(self, results: list[StrategyResult]): + self.results = list(results) + self.strategy = SimpleNamespace( + meta={"id": "demo", "name": "示例策略"}, + execution_backend="polars_expr", + filter_history_fn=None, + ) + + def get(self, strategy_id: str): + assert strategy_id == "demo" + return self.strategy + + def run(self, strategy_id: str, context, **kwargs): + assert strategy_id == "demo" + return self.results.pop(0) + + +def _result( + as_of: date, + pool: tuple[str, ...] = (), + buys: tuple[str, ...] = (), + sells: tuple[str, ...] = (), +) -> StrategyResult: + return StrategyResult( + as_of=as_of, + strategy_id="demo", + rows=[{"symbol": symbol, "close": 10.0, "change_pct": 0.01} for symbol in pool], + total=len(pool), + entry_signal_hits=[{"symbol": symbol, "signals": ["signal_buy"]} for symbol in buys], + exit_signal_hits=[{"symbol": symbol, "signals": ["signal_sell"]} for symbol in sells], + ) + + +def _quotes() -> pl.DataFrame: + return pl.DataFrame({ + "symbol": ["A", "B"], + "close": [10.0, 20.0], + "change_pct": [0.01, -0.02], + }) + + +def test_strategy_rule_compatibility_and_validation(tmp_path): + legacy = { + "id": "legacy", "name": "旧规则", "type": "strategy", + "scope": "all", "strategy_id": "demo", + } + monitor_rules.save_one(tmp_path, legacy) + + loaded = monitor_rules.load_one(tmp_path, "legacy") + assert loaded is not None + assert loaded["notify_events"] == ["pool_entry", "pool_exit"] + assert monitor_rules.load_all(tmp_path)[0]["notify_events"] == ["pool_entry", "pool_exit"] + + with pytest.raises(ValueError, match="至少选择一个通知事件"): + monitor_rules.validate(_rule()) + with pytest.raises(ValueError, match="非法事件"): + monitor_rules.validate(_rule("unknown")) + monitor_rules.validate(_rule("buy_signal", "pool_exit")) + + +def test_strategy_events_baseline_dedupe_and_next_day_replay(): + day1 = date(2026, 7, 24) + day2 = date(2026, 7, 25) + engine = MonitorRuleEngine() + engine.set_strategy_engine(_SequenceStrategyEngine([ + _result(day1, pool=("A",), buys=("A",)), + _result(day1, pool=("A", "B"), buys=("A", "B")), + _result(day1, pool=("A", "B")), + _result(day1, pool=("A", "B"), buys=("B",)), + _result(day2, pool=("A", "B"), buys=("B",)), + ])) + engine.set_rules([_rule("buy_signal", "pool_entry")]) + + with patch("app.strategy.monitor.time.time", side_effect=[100, 101, 102, 103, 4000]): + assert engine.evaluate(_quotes()) == [] + events = engine.evaluate(_quotes()) + assert {(event["type"], event["symbol"]) for event in events} == { + ("buy_signal", "B"), + ("pool_entry", "B"), + } + assert all(event["strategy_id"] == "demo" for event in events) + assert engine.evaluate(_quotes()) == [] + assert engine.evaluate(_quotes()) == [] + next_day = engine.evaluate(_quotes()) + assert [(event["type"], event["symbol"]) for event in next_day] == [("buy_signal", "B")] + + +def test_strategy_sell_and_pool_exit_are_independent_events(): + day = date(2026, 7, 24) + engine = MonitorRuleEngine() + engine.set_strategy_engine(_SequenceStrategyEngine([ + _result(day, pool=("A", "B")), + _result(day, pool=("A",), sells=("B",)), + ])) + engine.set_rules([_rule("sell_signal", "pool_exit")]) + + with patch("app.strategy.monitor.time.time", side_effect=[100, 101]): + assert engine.evaluate(_quotes()) == [] + events = engine.evaluate(_quotes()) + + assert {(event["type"], event["symbol"]) for event in events} == { + ("sell_signal", "B"), + ("pool_exit", "B"), + } + + +def test_strategy_rule_reload_preserves_state_and_semantic_edit_resets_it(): + day = date(2026, 7, 24) + engine = MonitorRuleEngine() + engine.set_strategy_engine(_SequenceStrategyEngine([ + _result(day, buys=("A",)), + _result(day, buys=("A",)), + _result(day, buys=("A",)), + _result(day, buys=("A",)), + ])) + rule = _rule("buy_signal") + engine.set_rules([rule]) + assert engine.evaluate(_quotes()) == [] + + engine.set_rules([{**rule, "message": "新文案"}]) + assert engine.evaluate(_quotes()) == [] + + engine.set_rules([{**rule, "scope": "symbols", "symbols": ["A"]}]) + assert engine.evaluate(_quotes()) == [] + + engine.set_rules([]) + assert not engine._strategy_signal_state + engine.set_rules([rule]) + assert engine.evaluate(_quotes()) == [] + + +def test_matrix_signal_hits_map_codes_and_keep_unlabelled_hits(): + mapped = StrategyEngine._matrix_signal_hits( + np.array([1, 1, 0], dtype=np.uint8), + np.array([1, -1, -1], dtype=np.int16), + ("signal_a", "signal_b"), + ("A", "B", "C"), + ) + assert mapped == [ + {"symbol": "A", "signals": ["signal_b"]}, + {"symbol": "B", "signals": []}, + ] + + +def test_matrix_strategy_pool_masks_rows_and_both_signal_directions(): + day = date(2026, 7, 24) + panel = pl.DataFrame({ + "symbol": ["A", "B"], + "date": [day, day], + "open": [10.0, 20.0], + "high": [10.0, 20.0], + "low": [10.0, 20.0], + "close": [10.0, 20.0], + "volume": [100.0, 100.0], + "amount": [1000.0, 2000.0], + }) + calls = 0 + + class _AllSignals: + def required_fields(self): + return frozenset({"close"}) + + def required_warmup_bars(self, params): + return 1 + + def compute_signals(self, market, params): + nonlocal calls + calls += 1 + active = np.ones(market.shape, dtype=np.uint8) + codes = np.zeros(market.shape, dtype=np.int16) + return make_signal_matrix( + market.shape, + entry=active, + exit=active, + entry_signal_code=codes, + exit_signal_code=codes, + entry_signal_ids=("signal_buy",), + exit_signal_ids=("signal_sell",), + ) + + strategy = StrategyDef( + meta={"id": "matrix", "scoring": {}, "limit": 100}, + basic_filter={"enabled": False}, + entry_signals=[], + exit_signals=[], + stop_loss=None, + trailing_stop=None, + trailing_take_profit_activate=None, + trailing_take_profit_drawdown=None, + max_hold_days=None, + alerts=[], + filter_fn=None, + filter_history_fn=None, + lookback_days=1, + source="custom", + execution_backend="matrix_native", + matrix_strategy=_AllSignals(), + ) + engine = StrategyEngine(strategy_dirs=[]) + engine._strategies["matrix"] = strategy + result = engine.run( + "matrix", + StrategyDataContext( + "stock", + "1d", + day, + current=panel, + market=build_market_data_matrix(panel), + ), + pool=["A"], + ) + + assert calls == 1 + assert [row["symbol"] for row in result.rows] == ["A"] + assert result.entry_signal_hits == [{"symbol": "A", "signals": ["signal_buy"]}] + assert result.exit_signal_hits == [{"symbol": "A", "signals": ["signal_sell"]}] + + +def test_ordinary_strategy_uses_signal_overrides_and_ignores_malformed_values(): + day = date(2026, 7, 24) + quotes = pl.DataFrame({ + "symbol": ["A", "B"], + "signal_default_buy": [True, False], + "signal_override_buy": [False, True], + "signal_default_sell": [False, True], + "signal_override_sell": [True, False], + }) + strategy = StrategyDef( + meta={"id": "ordinary", "scoring": {}, "limit": 100}, + basic_filter={"enabled": False}, + entry_signals=["signal_default_buy"], + exit_signals=["signal_default_sell"], + stop_loss=None, + trailing_stop=None, + trailing_take_profit_activate=None, + trailing_take_profit_drawdown=None, + max_hold_days=None, + alerts=[], + filter_fn=None, + filter_history_fn=None, + lookback_days=1, + source="custom", + ) + engine = StrategyEngine(strategy_dirs=[]) + engine._strategies["ordinary"] = strategy + context = StrategyDataContext("stock", "1d", day, current=quotes) + + result = engine.run( + "ordinary", + context, + overrides={ + "entry_signals": ["signal_override_buy"], + "exit_signals": ["signal_override_sell"], + }, + ) + assert result.entry_signal_hits == [{"symbol": "B", "signals": ["signal_override_buy"]}] + assert result.exit_signal_hits == [{"symbol": "A", "signals": ["signal_override_sell"]}] + + fallback = engine.run( + "ordinary", + context, + overrides={"entry_signals": None, "exit_signals": "signal_override_sell"}, + ) + assert fallback.entry_signal_hits == [{"symbol": "A", "signals": ["signal_default_buy"]}] + assert fallback.exit_signal_hits == [{"symbol": "B", "signals": ["signal_default_sell"]}] + + +def test_quote_service_forwards_real_strategy_id(monkeypatch, tmp_path): + event = { + "ts": 1, + "rule_id": "strategy_rule", + "strategy_id": "demo", + "source": "strategy", + "type": "buy_signal", + "symbol": "A", + "name": "测试股票", + "message": "策略买入信号", + "price": 10.0, + "change_pct": 0.01, + "signals": ["signal_buy"], + "severity": "info", + } + + class _Engine: + rule_count = 1 + + def __init__(self): + self.rules = {"strategy_rule": {"webhook_channels": []}} + + def set_name_map(self, name_map): + pass + + def has_rule_type(self, rtype: str) -> bool: + return False + + def has_asset_rules(self, asset_type: str) -> bool: + return False + + def evaluate(self, df, asset_type: str): + return [event] + + def consume_strategy_result_updates(self) -> bool: + return False + + class _Repo: + store = SimpleNamespace(data_dir=tmp_path) + + @staticmethod + def get_instruments(): + return pl.DataFrame({"symbol": ["A"], "name": ["测试股票"]}) + + monkeypatch.setattr(alert_store, "append_many", lambda *args: None) + monkeypatch.setattr(preferences, "get_system_notify_enabled", lambda: False) + service = QuoteService() + subscriber = service.subscribe() + service.set_app_state(SimpleNamespace(monitor_engine=_Engine(), repo=_Repo())) + service._repo = _Repo() + service.get_enriched_today = lambda: (_quotes(), quote_service.cn_today()) + + with patch.object(QuoteService, "_is_continuous_trading", return_value=True): + service._evaluate_monitors(pl.DataFrame(), None) + + assert subscriber.pop()["alerts"][0]["strategy_id"] == "demo" diff --git a/frontend/src/components/AlertToast.tsx b/frontend/src/components/AlertToast.tsx index 3714207..9d25e6d 100644 --- a/frontend/src/components/AlertToast.tsx +++ b/frontend/src/components/AlertToast.tsx @@ -4,10 +4,12 @@ import { motion, AnimatePresence } from 'framer-motion' import { Bell, TrendingUp, TrendingDown, X } from 'lucide-react' import type { AlertEvent } from '@/lib/api' import { fmtPct, fmtPrice } from '@/lib/format' +import { cnSignal } from '@/lib/signals' import { cn } from '@/lib/cn' import { playNotificationSound } from '@/lib/notificationSound' import { speakAlerts } from '@/lib/voiceBroadcast' import { usePreferences } from '@/lib/useSharedQueries' +import { strategyEventMeta, strategyName } from '@/lib/strategyMonitorEvents' /** 通知渠道分发 — 所有副作用渠道在此汇合, 新增渠道只改这里 */ function dispatchSideEffects(alerts: AlertEvent[]) { @@ -86,8 +88,12 @@ const SOURCE_BADGE: Record = { signal: { label: '信号', cls: 'bg-accent/15 text-accent' }, price: { label: '价格', cls: 'bg-emerald-400/15 text-emerald-400' }, market: { label: '异动', cls: 'bg-purple-500/15 text-purple-400' }, - new_entry: { label: '进入', cls: 'bg-emerald-400/15 text-emerald-400' }, - dropped: { label: '移出', cls: 'bg-danger/15 text-danger' }, + pool_entry: { label: '进入', cls: 'bg-emerald-400/15 text-emerald-400' }, + pool_exit: { label: '移出', cls: 'bg-warning/15 text-warning' }, + buy_signal: { label: '买入', cls: 'bg-danger/15 text-danger' }, + sell_signal: { label: '卖出', cls: 'bg-bear/15 text-bear' }, + new_entry: { label: '进入', cls: 'bg-emerald-400/15 text-emerald-400' }, + dropped: { label: '移出', cls: 'bg-warning/15 text-warning' }, } // ===== 容器 — 挂在 Layout ===== @@ -122,18 +128,15 @@ export function AlertToastContainer() { className="fixed bottom-4 right-4 z-[9999] flex flex-col gap-2 w-[320px] pointer-events-none" > - {items - .filter(item => !(item.alert.source === 'strategy' && !item.alert.symbol)) - .map(item => { + {items.map(item => { const ev = item.alert const sev = SEVERITY_BAR[ev.severity ?? 'info'] ?? SEVERITY_BAR.info const badgeKey = (ev.source === 'strategy' && ev.type) ? ev.type : ev.source const badge = SOURCE_BADGE[badgeKey] ?? { label: badgeKey, cls: 'bg-elevated text-muted' } const pct = ev.change_pct ?? 0 const isStrategy = ev.source === 'strategy' - const sm = isStrategy ? ev.message?.match(/策略「([^」]+)」/) : null - const sname = sm ? sm[1] : '' - const isNew = ev.type === 'new_entry' + const sname = isStrategy ? strategyName(ev.message ?? '') : '' + const eventMeta = strategyEventMeta(ev.type) return ( - - - {isNew ? '进入' : '移出'} - - 策略 - 「{sname}」 - - {ev.price != null && {fmtPrice(ev.price)}} - + <> + {ev.symbol ? ( +
+ + + {eventMeta.action} + + {sname + ? 「{sname}」 + : ev.message && {ev.message}} + + {ev.price != null && {fmtPrice(ev.price)}} +
+ ) : ( +
+ + {ev.message} +
+ )} + {ev.signals && ev.signals.length > 0 && ( +
+ {ev.signals.map(signal => ( + {cnSignal(signal)} + ))} +
+ )} + ) : (
diff --git a/frontend/src/components/monitor/RuleEditor.tsx b/frontend/src/components/monitor/RuleEditor.tsx index 045af41..12f8b72 100644 --- a/frontend/src/components/monitor/RuleEditor.tsx +++ b/frontend/src/components/monitor/RuleEditor.tsx @@ -2,7 +2,8 @@ import { useState } from 'react' import { Link } from 'react-router-dom' import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import { Activity, Check, Plus, RadioTower, Save, Search, TrendingUp, Waypoints, X } from 'lucide-react' -import { api, genRuleId, type MonitorRule, type MonitorCondition } from '@/lib/api' +import { api, genRuleId, type MonitorRule, type MonitorCondition, 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' import { SignalPicker } from '@/components/screener/SignalPicker' import { MONITOR_INTRADAY_SIGNAL_OPTIONS, SIGNAL_OPTIONS, cnSignal } from '@/lib/signals' @@ -64,14 +65,25 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { const [editing] = useState(!!rule) // 新建规则: 预填全局「默认推送渠道」(多选数组), preset 显式指定时以 preset 为准。 // 编辑规则: 完全沿用规则自身配置, 不受默认值影响。 - const [draft, setDraft] = useState( - rule - ? { ...rule, conditions: rule.conditions.map(c => ({ ...c })) } - : { - ...emptyRule(preset), - webhook_channels: preset?.webhook_channels ?? (prefs?.webhook_default_channels ?? []), - }, - ) + const [draft, setDraft] = useState(() => { + if (rule) { + return { + ...rule, + notify_events: rule.type === 'strategy' + ? [...(rule.notify_events ?? LEGACY_STRATEGY_NOTIFY_EVENTS)] + : undefined, + conditions: rule.conditions.map(c => ({ ...c })), + } + } + const initial = { + ...emptyRule(preset), + webhook_channels: preset?.webhook_channels ?? (prefs?.webhook_default_channels ?? []), + } + if (initial.type === 'strategy' && !initial.notify_events) { + initial.notify_events = [...DEFAULT_STRATEGY_NOTIFY_EVENTS] + } + return initial + }) const assetType = draft.asset_type ?? 'stock' // 策略列表跟随资产类型: ETF 只列技术类策略。 const strategies = useQuery({ @@ -103,7 +115,9 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { } if (d.type === 'strategy') { if (!d.strategy_id) throw new Error('策略监控必须选择一个策略') + if (!d.notify_events?.length) throw new Error('至少选择一个通知事件') } else { + delete d.notify_events if (d.conditions.length === 0) throw new Error('至少选择一个触发条件') for (const c of d.conditions) { if (!c.field || !c.op) throw new Error('条件填写不完整') @@ -149,6 +163,17 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { return { ...d, webhook_channels: cur.includes(ch) ? cur.filter(c => c !== ch) : [...cur, ch] } }) + const toggleStrategyEvent = (event: StrategyNotifyEvent) => + setDraft(d => { + const current = d.notify_events ?? LEGACY_STRATEGY_NOTIFY_EVENTS + return { + ...d, + notify_events: current.includes(event) + ? current.filter(item => item !== event) + : [...current, event], + } + }) + const thresholdFields = options.data?.threshold_fields ?? [] const operators = options.data?.operators ?? ['>', '>=', '<', '<=', '==', '!='] const selectedSignals = draft.conditions.filter(c => c.op === 'truth').map(c => c.field) @@ -160,7 +185,6 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { ] const thresholdConds = draft.conditions.filter(c => c.op !== 'truth') const strategyPresets = strategies.data?.presets ?? [] - const selectedStrategy = strategyPresets.find(strategy => strategy.id === draft.strategy_id) const normalizedStrategyQuery = strategyQuery.trim().toLowerCase() const visibleStrategies = strategyPresets.filter(strategy => { if (strategyCategory !== 'all' && strategy.source !== strategyCategory) return false @@ -325,11 +349,17 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { key={t.key} type="button" aria-pressed={active} - onClick={() => setDraft(d => ({ - ...d, - type: t.key as MonitorRule['type'], - scope: t.key === 'strategy' && d.scope === 'symbols' && d.symbols.length === 0 ? 'all' : d.scope, - }))} + onClick={() => setDraft(d => { + const type = t.key as MonitorRule['type'] + return { + ...d, + type, + notify_events: type === 'strategy' + ? [...(d.notify_events ?? DEFAULT_STRATEGY_NOTIFY_EVENTS)] + : undefined, + scope: type === 'strategy' && d.scope === 'symbols' && d.symbols.length === 0 ? 'all' : d.scope, + } + })} className={`inline-flex h-9 items-center justify-center gap-1.5 rounded-btn border px-2 text-xs font-medium transition-colors cursor-pointer ${ active ? 'border-accent/40 bg-accent/12 text-accent' @@ -538,32 +568,36 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { })}
-
-
- {selectedStrategy ? ( - <>已选择 {selectedStrategy.name} - ) : '尚未选择策略'} +
+
+ 通知事件 + 至少选择一项
-
- 触发方向 -
- {(options.data?.directions ?? []).map(direction => ( - - ))} -
+
+ {(['signal', 'pool'] as const).map(group => ( +
+
+ {group === 'signal' ? '交易信号' : '选股结果'} +
+
+ {STRATEGY_NOTIFY_EVENT_OPTIONS.filter(option => option.group === group).map(option => ( + + ))} +
+
+ ))}
+ {(draft.notify_events ?? LEGACY_STRATEGY_NOTIFY_EVENTS).length === 0 && ( +
至少选择一个通知事件
+ )}
)} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 948e6d6..c87906a 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -509,6 +509,8 @@ export interface MonitorCondition { value?: number | null // op 非 truth 时必填 } +export type StrategyNotifyEvent = 'buy_signal' | 'sell_signal' | 'pool_entry' | 'pool_exit' + export interface MonitorRule { id: string name: string @@ -520,6 +522,7 @@ export interface MonitorRule { sector?: string | null strategy_id?: string | null direction: 'entry' | 'exit' | 'both' | 'up' | 'down' + notify_events?: StrategyNotifyEvent[] conditions: MonitorCondition[] logic: 'and' | 'or' cooldown_seconds: number diff --git a/frontend/src/lib/strategyMonitorEvents.ts b/frontend/src/lib/strategyMonitorEvents.ts new file mode 100644 index 0000000..db960e6 --- /dev/null +++ b/frontend/src/lib/strategyMonitorEvents.ts @@ -0,0 +1,73 @@ +import type { StrategyNotifyEvent } from './api' + +export interface StrategyEventMeta { + label: string + action: string + className: string +} + +const META: Record = { + buy_signal: { + label: '买入', + action: '买入信号', + className: 'text-danger', + }, + sell_signal: { + label: '卖出', + action: '卖出信号', + className: 'text-bear', + }, + pool_entry: { + label: '进入', + action: '进入选股结果', + className: 'text-emerald-400', + }, + pool_exit: { + label: '移出', + action: '移出选股结果', + className: 'text-warning', + }, + new_entry: { + label: '进入', + action: '进入选股结果', + className: 'text-emerald-400', + }, + dropped: { + label: '移出', + action: '移出选股结果', + className: 'text-warning', + }, +} + +export const DEFAULT_STRATEGY_NOTIFY_EVENTS: StrategyNotifyEvent[] = [ + 'buy_signal', + 'sell_signal', +] + +export const LEGACY_STRATEGY_NOTIFY_EVENTS: StrategyNotifyEvent[] = [ + 'pool_entry', + 'pool_exit', +] + +export const STRATEGY_NOTIFY_EVENT_OPTIONS: { + key: StrategyNotifyEvent + label: string + group: 'signal' | 'pool' +}[] = [ + { key: 'buy_signal', label: '买入信号', group: 'signal' }, + { key: 'sell_signal', label: '卖出信号', group: 'signal' }, + { key: 'pool_entry', label: '进入选股结果', group: 'pool' }, + { key: 'pool_exit', label: '移出选股结果', group: 'pool' }, +] + +export function strategyEventMeta(type: string): StrategyEventMeta { + return META[type] ?? { + label: '事件', + action: '策略事件', + className: 'text-secondary', + } +} + +export function strategyName(message: string): string { + return message.match(/策略「([^」]+)」/)?.[1] ?? '' +} diff --git a/frontend/src/lib/voiceBroadcast.ts b/frontend/src/lib/voiceBroadcast.ts index a1918ca..d699b47 100644 --- a/frontend/src/lib/voiceBroadcast.ts +++ b/frontend/src/lib/voiceBroadcast.ts @@ -12,6 +12,7 @@ */ import type { AlertEvent } from './api' +import { strategyEventMeta, strategyName } from './strategyMonitorEvents' const LS = { enabled: 'voice_broadcast_enabled', // '1'/'0', 默认关 @@ -121,13 +122,10 @@ function buildSingleText(a: AlertEvent): string { if (!a.symbol || a.symbol === '_batch') { return a.message || name } - // 单条事件: 从 message 提取策略名 (格式 "策略「X」" 或直接是策略名) - const sm = a.message?.match(/策略「([^」]+)」/) - const sname = sm ? sm[1] : a.message || '' - const action = a.type === 'new_entry' ? '进入' : a.type === 'dropped' ? '移出' : '' - const parts = [name] - if (action) parts.push(action) - if (sname) parts.push(`策略「${sname}」`) + const sname = strategyName(a.message ?? '') + const parts = [name, strategyEventMeta(a.type).action] + const strategyText = sname ? `策略「${sname}」` : (a.message || a.rule_name || '') + if (strategyText) parts.push(strategyText) if (pctText) parts.push(pctText) return parts.join(' ') } diff --git a/frontend/src/pages/Dashboard.tsx b/frontend/src/pages/Dashboard.tsx index 8d0a96d..7cbc647 100644 --- a/frontend/src/pages/Dashboard.tsx +++ b/frontend/src/pages/Dashboard.tsx @@ -14,6 +14,7 @@ import { SettingsModal } from '@/components/data/SettingsModal' import { STAGE_LABELS } from '@/components/data/ActiveJobCard' import { cn } from '@/lib/cn' import { cnSignal } from '@/lib/signals' +import { strategyEventMeta, strategyName } from '@/lib/strategyMonitorEvents' import { boardTag } from '@/components/stock-table/primitives' function n(v: number | null | undefined) { @@ -112,15 +113,12 @@ function MonitorWidget({ onStockClick }: { onStockClick: (event: AlertEvent) => return ( <>
- {events - .filter((ev: AlertEvent) => !(ev.source === 'strategy' && !ev.symbol)) - .map((ev, i) => { + {events.map((ev, i) => { const sev = _SEVERITY_BAR[ev.severity ?? 'info'] ?? _SEVERITY_BAR.info const pct = ev.change_pct ?? 0 const isStrategy = ev.source === 'strategy' - const sm = isStrategy ? ev.message?.match(/策略「([^」]+)」/) : null - const sname = sm ? sm[1] : '' - const isNew = ev.type === 'new_entry' + const sname = isStrategy ? strategyName(ev.message ?? '') : '' + const eventMeta = strategyEventMeta(ev.type) return (
{/* 第二行: 策略类型走新格式, 其他走旧格式 */} {isStrategy ? ( -
- - {isNew ? '进入' : '移出'} - - 策略 - 「{sname}」 - - - {ev.ts ? new Date(ev.ts).toLocaleString('zh-CN', { month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit' }) : ''} - -
+ <> + {ev.symbol ? ( +
+ + {eventMeta.action} + + {sname + ? 「{sname}」 + : ev.message && {ev.message}} + + + {ev.ts ? new Date(ev.ts).toLocaleString('zh-CN', { month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit' }) : ''} + +
+ ) : ( +
+ {ev.message} + + + {ev.ts ? new Date(ev.ts).toLocaleString('zh-CN', { month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit' }) : ''} + +
+ )} + {ev.signals && ev.signals.length > 0 && ( +
+ {ev.signals.map(signal => ( + {cnSignal(signal)} + ))} +
+ )} + ) : ( <>
diff --git a/frontend/src/pages/Monitor.tsx b/frontend/src/pages/Monitor.tsx index 952f7fe..7e79c71 100644 --- a/frontend/src/pages/Monitor.tsx +++ b/frontend/src/pages/Monitor.tsx @@ -10,6 +10,7 @@ import { QK } from '@/lib/queryKeys' import { fmtPrice, fmtPct } from '@/lib/format' import { cn } from '@/lib/cn' import { cnSignal } from '@/lib/signals' +import { LEGACY_STRATEGY_NOTIFY_EVENTS, STRATEGY_NOTIFY_EVENT_OPTIONS, strategyEventMeta, strategyName } from '@/lib/strategyMonitorEvents' import { boardTag } from '@/components/stock-table/primitives' import { markSeen, resetBadge, leaveMonitorPage } from '@/lib/monitorBadge' import { RuleEditor } from '@/components/monitor/RuleEditor' @@ -340,9 +341,7 @@ function AlertsList({ alertsQuery, confirmClear, setConfirmClear, total, enterTs /> ) : (
- {events - .filter((ev: any) => !(ev.source === 'strategy' && !ev.symbol)) - .map((ev: any, i: number) => { + {events.map((ev: any, i: number) => { const sev = SEVERITY_CONFIG[ev.severity ?? 'info'] ?? SEVERITY_CONFIG.info const SevIcon = sev.icon const isNew = ev.ts > enterTs @@ -367,9 +366,8 @@ function AlertsList({ alertsQuery, confirmClear, setConfirmClear, total, enterTs
{ev.source === 'strategy' ? (() => { - const sm = ev.message?.match(/策略「([^」]+)」/) - const sname = sm ? sm[1] : '' - const isNew = ev.type === 'new_entry' + const sname = strategyName(ev.message ?? '') + const eventMeta = strategyEventMeta(ev.type) const _pct = ev.change_pct ?? 0 return ( <> @@ -408,13 +406,25 @@ function AlertsList({ alertsQuery, confirmClear, setConfirmClear, total, enterTs {sname}
-
- - {isNew ? '进入' : '移出'} - - 策略 - 「{sname}」 -
+ {ev.symbol ? ( +
+ + {eventMeta.action} + + {sname + ? 「{sname}」 + : ev.message && {ev.message}} +
+ ) : ( +
{ev.message}
+ )} + {ev.signals && ev.signals.length > 0 && ( +
+ {ev.signals.map((signal: string) => ( + {cnSignal(signal)} + ))} +
+ )} ) })() : ( @@ -723,10 +733,17 @@ function RulesList({ rulesQuery, onEdit }: {
)} - {/* 第二行: 策略类型显示选股池变更监控 */} + {/* 第二行: 策略类型显示通知事件 */} {r.type === 'strategy' && r.strategy_id ? ( -
- 选股池变更监控 +
+ {(r.notify_events ?? LEGACY_STRATEGY_NOTIFY_EVENTS).map(event => { + const option = STRATEGY_NOTIFY_EVENT_OPTIONS.find(item => item.key === event) + return option ? ( + + {option.label} + + ) : null + })}
) : r.conditions.length > 0 && (
diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index a0166a2..a2cc5cf 100644 --- a/frontend/src/pages/Screener.tsx +++ b/frontend/src/pages/Screener.tsx @@ -3,6 +3,7 @@ import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query' import { motion } from 'framer-motion' import { ScanSearch, Clock, TrendingUp, Star, Filter, Layers, Network, Sparkles, RefreshCw, Settings2, Store, RotateCcw, X } from 'lucide-react' import { api, genRuleId, type ScreenerStrategy, type ScreenerResult } from '@/lib/api' +import { DEFAULT_STRATEGY_NOTIFY_EVENTS } from '@/lib/strategyMonitorEvents' import { toast } from '@/components/Toast' import { useDataStatus, usePreferences, useCapabilities, useQuoteStatus } from '@/lib/useSharedQueries' import { useWatchlistBatchAdd } from '@/lib/useSharedMutations' @@ -536,6 +537,7 @@ export function Screener() { sector: null, strategy_id: strategyId, direction: 'entry', + notify_events: [...DEFAULT_STRATEGY_NOTIFY_EVENTS], conditions: [], logic: 'or', cooldown_seconds: 3600,