diff --git a/backend/app/api/abnormal.py b/backend/app/api/abnormal.py new file mode 100644 index 0000000..e4f6027 --- /dev/null +++ b/backend/app/api/abnormal.py @@ -0,0 +1,23 @@ +"""异动边缘监控 API — 按交易所异动规则口径统计接近触发的个股。""" +from __future__ import annotations + +from fastapi import APIRouter, Query, Request + +from app.services.abnormal_moves import build_overview + +router = APIRouter(prefix="/api/abnormal", tags=["abnormal"]) + + +@router.get("/overview") +def abnormal_overview( + request: Request, + min_closeness: float = Query(0.5, ge=0.0, le=1.0), + limit: int = Query(200, ge=1, le=1000), +): + """异动边缘总览: 规则表 + 各窗口实时偏离 + 接近度排序。 + + min_closeness: 0.5=观察 / 0.7=边缘 / 1.0=已触发。 + """ + repo = request.app.state.repo + quote_service = getattr(request.app.state, "quote_service", None) + return build_overview(repo, quote_service, min_closeness=min_closeness, limit=limit) diff --git a/backend/app/api/monitor_rules.py b/backend/app/api/monitor_rules.py index eec1f3b..0456ec7 100644 --- a/backend/app/api/monitor_rules.py +++ b/backend/app/api/monitor_rules.py @@ -77,7 +77,7 @@ class RuleModel(BaseModel): id: str name: str enabled: bool = True - type: str # strategy | signal | price | market | sector + type: str # strategy | signal | price | market | sector | abnormal asset_type: str = "stock" # stock | etf (etf: strategy 型走 ETF 历史加载器) scope: str = "symbols" # symbols | all | sector symbols: list[str] = [] @@ -88,7 +88,7 @@ class RuleModel(BaseModel): threshold_pct: float = 1.0 window_minutes: int = 5 strategy_id: str | None = None - direction: str = "entry" # entry | exit | both + direction: str = "entry" # entry | exit | both | (sector/ladder/abnormal: up|down|both) notify_events: list[str] | None = None score_min: float | None = None score_max: float | None = None @@ -100,6 +100,8 @@ class RuleModel(BaseModel): webhook_enabled: bool = False # 兼容老规则 (已由 webhook_channels 取代, 仅做向后兼容读) webhook_channels: list[str] = [] # 命中时推送的外部渠道 (合法值 'feishu' | 'wecom') message: str = "" + # abnormal 专属 (异动边缘监控): any | 3d | 10d | 30d + abnormal_window: str = "any" # ladder 专属 (连板梯队封单监控) metric: str = "sealed_vol" # sealed_vol=封单量(手) | sealed_amount=封单额(元) threshold: float = 0 # 封单 <= 此值时报警 (原始单位: 量=手, 额=元) @@ -154,6 +156,7 @@ def get_options(request: Request): {"key": "price", "label": "价格/涨跌"}, {"key": "market", "label": "市场异动"}, {"key": "strategy", "label": "策略监控"}, + {"key": "abnormal", "label": "异动监控"}, {"key": "sector", "label": "板块监控"}, ], "scopes": [ diff --git a/backend/app/api/watchlist.py b/backend/app/api/watchlist.py index 5db19a1..afd5ed3 100644 --- a/backend/app/api/watchlist.py +++ b/backend/app/api/watchlist.py @@ -264,6 +264,7 @@ _WATCHLIST_COLS = [ "boll_upper", "boll_lower", "atr_14", "momentum_5d", "momentum_10d", "momentum_20d", "momentum_30d", "momentum_60d", + "deviate_3d", "deviate_10d", "deviate_30d", "consecutive_limit_ups", "consecutive_limit_downs", "signal_limit_up", "signal_limit_down", "signal_volume_surge", "signal_ma_golden_5_20", "signal_macd_golden", "signal_n_day_high", diff --git a/backend/app/enriched_generation.py b/backend/app/enriched_generation.py index 87fa81c..66b6b11 100644 --- a/backend/app/enriched_generation.py +++ b/backend/app/enriched_generation.py @@ -137,7 +137,11 @@ def _process_is_alive(pid: Any) -> bool: os.kill(pid, 0) except ProcessLookupError: return False - except (OSError, PermissionError): + except (OSError, PermissionError) as exc: + # Windows 对不存在的 pid 返回 WinError 87 (ERROR_INVALID_PARAMETER), + # 不会映射为 ProcessLookupError; 按存活处理会让孤儿发布锁永远无法恢复。 + if getattr(exc, "winerror", None) == 87: + return False return True return True diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index 99ea33f..f7d9575 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -164,6 +164,10 @@ ENRICHED_COLUMNS: dict[str, dict[str, str]] = { "momentum_20d": "20日动量", "momentum_30d": "30日动量", "momentum_60d": "60日动量", + # ── 异动偏离 (运行时由 repository 附着, 不落盘) ──────── + "deviate_3d": "3日涨跌幅偏离值(vs对应指数, 小数)", + "deviate_10d": "10日涨跌幅偏离值", + "deviate_30d": "30日涨跌幅偏离值", # ── 波动率 ─────────────────────────────────────────── "annual_vol_20d": "20日年化波动率", # ── RSI ────────────────────────────────────────────── @@ -210,6 +214,7 @@ ENRICHED_COLUMNS_BY_CATEGORY: dict[str, list[str]] = { "volume": ["vol_ma5", "vol_ma10", "vol_ratio_5d"], "extremes": ["high_60d", "low_60d"], "momentum": ["momentum_5d", "momentum_10d", "momentum_20d", "momentum_30d", "momentum_60d"], + "deviation": ["deviate_3d", "deviate_10d", "deviate_30d"], "volatility": ["annual_vol_20d"], "rsi": ["rsi_6", "rsi_14", "rsi_24"], "signals": [k for k in ENRICHED_COLUMNS if k.startswith("signal_")], @@ -1029,6 +1034,140 @@ def _select_storage_cols(df: pl.DataFrame) -> pl.DataFrame: return df.select(cols) +# ================================================================ +# 异动偏离列 (deviate_3d/10d/30d) +# +# N 日涨跌幅偏离值 = 个股 N 日累计涨跌幅 - 对应指数同期涨跌幅, +# 是交易所「异常波动 / 严重异常波动」规则的量化口径 (如主板 3日±20%, +# 10日+100%, 30日+200%)。不属于 compute_indicators 的纯函数范围 +# (需要指数数据), 因此在 repository 读取路径上附着, 不随 parquet 落盘。 +# ================================================================ + +DEVIATION_WINDOWS: tuple[int, ...] = (3, 10, 30) + +# 各交易所基准指数 (偏离值规则的「对应指数」近似): 优先分类指数, 缺失时回退 +_BENCHMARK_PREFERENCE: dict[str, list[str]] = { + "SH": ["000002.SH", "000001.SH"], # 上证A指 → 上证指数 + "SZ": ["399107.SZ", "399001.SZ"], # 深证A指 → 深证成指 + "BJ": ["899050.BJ", "000001.SH"], # 北证50 → 上证指数 +} + +_benchmark_cache: dict[str, tuple[float, pl.DataFrame | None]] = {} +_BENCHMARK_CACHE_TTL = 600.0 + + +def load_benchmark_momentum(data_dir: Path) -> pl.DataFrame | None: + """读取指数日K, 计算各基准指数的滚动 N 日涨跌幅。 + + 返回长表: date, bench_exchange, bench_mom3d, bench_mom10d, bench_mom30d。 + 无可用指数数据时返回 None (偏离列置 null, 不阻塞主流程)。 + 进程内按 data_dir 缓存 (TTL 10 分钟)。 + """ + import time as _time + + now = _time.monotonic() + key = str(Path(data_dir).resolve()) + cached = _benchmark_cache.get(key) + if cached is not None and now - cached[0] < _BENCHMARK_CACHE_TTL: + return cached[1] + + frame: pl.DataFrame | None = None + try: + index_glob = str(Path(data_dir) / "kline_index_daily" / "**" / "*.parquet") + wanted: list[str] = [] + bench_of: dict[str, str] = {} + for exchange, candidates in _BENCHMARK_PREFERENCE.items(): + for sym in candidates: + if sym not in bench_of: + wanted.append(sym) + bench_of[sym] = exchange + lf = scan_daily_parquet( + index_glob, cast_options=pl.ScanCastOptions(integer_cast="allow-float") + ) + df_idx = ( + lf.filter(pl.col("symbol").is_in(wanted)) + .select(["symbol", "date", "close"]) + .sort(["symbol", "date"]) + .collect() + ) + if not df_idx.is_empty(): + available = set(df_idx["symbol"].to_list()) + picked = [s for s in wanted if s in available] + # 每个交易所取优先级最高的可用基准; 全缺时回退到任一可用基准。 + # 同一基准可服务多个交易所 (如北证50 缺失时北交所回退上证指数)。 + pairs: list[tuple[str, str]] = [] + for exchange, candidates in _BENCHMARK_PREFERENCE.items(): + hit = next((s for s in candidates if s in available), None) + if hit is None and picked: + hit = picked[0] + if hit is not None: + pairs.append((hit, exchange)) + df_bench = df_idx.filter(pl.col("symbol").is_in([p[0] for p in pairs])) + if not df_bench.is_empty(): + df_bench = df_bench.with_columns( + pl.col("close").cast(pl.Float64, strict=False) + ).with_columns([ + (pl.col("close") / pl.col("close").shift(n).over("symbol") - 1).alias(f"_bm{n}") + for n in DEVIATION_WINDOWS + ]).rename({f"_bm{n}": f"bench_mom{n}d" for n in DEVIATION_WINDOWS}) + exchange_map = pl.DataFrame({ + "symbol": [p[0] for p in pairs], + "bench_exchange": [p[1] for p in pairs], + }) + frame = ( + df_bench.join(exchange_map, on="symbol", how="inner") + .select(["date", "bench_exchange", *[f"bench_mom{n}d" for n in DEVIATION_WINDOWS]]) + .unique(subset=["date", "bench_exchange"]) + ) + except Exception as exc: # noqa: BLE001 + logger.warning("基准指数偏离数据加载失败: %s", exc) + frame = None + + _benchmark_cache[key] = (now, frame) + return frame + + +def attach_deviation_columns(df: pl.DataFrame, data_dir: Path) -> pl.DataFrame: + """为已含 momentum_Nd 的 enriched 帧附着 deviate_Nd 偏离列。 + + 缺失的动量列 (如 momentum_3d 不在指标全集里) 就地按 close 补算, + 与 compute_indicators 在同一帧上的 shift 语义一致。 + 基准按 symbol 后缀分交易所匹配, join 不上的行 (新上市/基准缺失) 置 null。 + """ + if df.is_empty(): + return df + bench = load_benchmark_momentum(data_dir) + dev_cols = [f"deviate_{n}d" for n in DEVIATION_WINDOWS] + if bench is None or bench.is_empty(): + return df.with_columns([pl.lit(None, dtype=pl.Float64).alias(c) for c in dev_cols]) + if "close" not in df.columns: + logger.warning("偏离列附着跳过: 缺少 close 列") + return df.with_columns([pl.lit(None, dtype=pl.Float64).alias(c) for c in dev_cols]) + missing = [n for n in DEVIATION_WINDOWS if f"momentum_{n}d" not in df.columns] + if missing: + df = df.sort(["symbol", "date"]).with_columns([ + (pl.col("close") / pl.col("close").shift(n).over("symbol") - 1).alias(f"momentum_{n}d") + for n in missing + ]) + bench_exchange = ( + pl.col("symbol").str.slice(-2).str.to_uppercase().replace( + {ex: ex for ex in _BENCHMARK_PREFERENCE}, + default=None, + return_dtype=pl.Utf8, + ) + ) + out = ( + df.with_columns(bench_exchange.alias("_bench_ex")) + .join(bench, left_on=["_bench_ex", "date"], right_on=["bench_exchange", "date"], how="left") + .with_columns([ + (pl.col(f"momentum_{n}d") - pl.col(f"bench_mom{n}d")).alias(f"deviate_{n}d") + for n in DEVIATION_WINDOWS + ]) + .drop(["_bench_ex", *[f"bench_mom{n}d" for n in DEVIATION_WINDOWS]]) + ) + return out + + def run_pipeline(data_dir: Path | None = None, symbols: list[str] | None = None, new_dates_only: bool = False, diff --git a/backend/app/main.py b/backend/app/main.py index 2beb09f..6fe86b0 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -13,6 +13,7 @@ from fastapi.staticfiles import StaticFiles from app import __version__ from app.api import ( + abnormal, alerts, analysis, backtest, @@ -431,6 +432,7 @@ app.include_router(mining.router) app.include_router(intraday.router) app.include_router(indices.router) app.include_router(overview.router) +app.include_router(abnormal.router) app.include_router(regime.router) app.include_router(analysis.router) app.include_router(pipeline.router) diff --git a/backend/app/services/abnormal_moves.py b/backend/app/services/abnormal_moves.py new file mode 100644 index 0000000..68e0cf5 --- /dev/null +++ b/backend/app/services/abnormal_moves.py @@ -0,0 +1,217 @@ +"""异动边缘统计 — 按交易所异动规则口径实时计算个股接近度。 + +规则 (近似口径, 与交易所《交易规则》的异常波动/严重异常波动披露阈值对齐): +- 主板: 连续3日收盘价涨跌幅偏离值累计 ±20% (风险警示 ±15%) +- 创业板/科创板: 3日 ±30% +- 北交所: 3日 ±40% +- 严重异常波动: 10日累计偏离 +100% (风险警示 +50%), 30日 +200% (风险警示 +100%) + +偏离值 = 个股 N 日累计涨跌幅 - 对应指数同期涨跌幅 (enriched 运行时列 deviate_Nd)。 +「接近度」= |实时偏离| / 阈值: ≥1 已触发, ≥0.7 边缘, ≥0.5 观察。 +盘中实时叠加: 历史偏离 (已完成交易日) + 今日实时涨跌 - 基准指数今日涨跌。 +""" + +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass +from datetime import date +from typing import Any + +import polars as pl + +from app.indicators.pipeline import DEVIATION_WINDOWS + +# ── 规则表 ──────────────────────────────────────────────── + +@dataclass(frozen=True) +class AbnormalRule: + board: str + st: bool + # 各窗口阈值 (小数): {3: 0.20, 10: 1.00, 30: 2.00} + thresholds: dict[int, float] + + +_MAIN = {3: 0.20, 10: 1.00, 30: 2.00} +_MAIN_ST = {3: 0.15, 10: 0.50, 30: 1.00} +_GEM_STAR = {3: 0.30, 10: 1.00, 30: 2.00} +_BSE = {3: 0.40, 10: 1.00, 30: 2.00} + +RULES_META: list[dict[str, Any]] = [ + {"board": "主板", "st": False, "thresholds": {f"{k}d": v for k, v in _MAIN.items()}, + "note": "3日±20% 异常波动; 10日+100%/30日+200% 严重异常波动"}, + {"board": "主板", "st": True, "thresholds": {f"{k}d": v for k, v in _MAIN_ST.items()}, + "note": "风险警示股票 (ST/*ST) 阈值从严"}, + {"board": "创业板/科创板", "st": False, "thresholds": {f"{k}d": v for k, v in _GEM_STAR.items()}, + "note": "20%涨跌幅板块, 3日±30%"}, + {"board": "北交所", "st": False, "thresholds": {f"{k}d": v for k, v in _BSE.items()}, + "note": "30%涨跌幅板块, 3日±40%"}, +] + +_BENCH_RT_CANDIDATES = ["000002.SH", "000001.SH", "399107.SZ", "399001.SZ", "899050.BJ"] + + +def board_of(symbol: str) -> str: + """按代码前缀判定板块。""" + code = symbol.split(".")[0] + if symbol.endswith(".BJ") or code[:2] in {"43", "83", "87", "92"}: + return "北交所" + if code.startswith("68"): + return "科创板" + if code.startswith(("30", "301")): + return "创业板" + return "主板" + + +def is_st_name(name: str | None) -> bool: + return bool(name) and "ST" in str(name).upper() + + +def rule_for(symbol: str, name: str | None) -> AbnormalRule: + board = board_of(symbol) + st = is_st_name(name) + if board == "北交所": + return AbnormalRule(board, st, _BSE) + if board in ("创业板", "科创板"): + return AbnormalRule(board, st, _GEM_STAR) + return AbnormalRule(board, st, _MAIN_ST if st else _MAIN) + + +# ── 快照计算 ────────────────────────────────────────────── + +_hist_cache_lock = threading.Lock() +_hist_cache: dict[str, Any] = {} +_HIST_CACHE_TTL = 60.0 + +_STATUS_TRIGGERED = "triggered" +_STATUS_EDGE = "edge" +_STATUS_WATCH = "watch" + + +def _status_of(closeness: float) -> str: + if closeness >= 1.0: + return _STATUS_TRIGGERED + if closeness >= 0.7: + return _STATUS_EDGE + return _STATUS_WATCH + + +def _hist_snapshot(repo: Any) -> dict[str, Any]: + """enriched 最新日的偏离列快照 (60s 进程内缓存)。""" + now = time.monotonic() + with _hist_cache_lock: + cached = _hist_cache.get("data") + if cached is not None and now - cached["_ts"] < _HIST_CACHE_TTL: + return cached + + df, cache_date = repo.get_enriched_latest() + rows: dict[str, dict[str, Any]] = {} + if not df.is_empty() and "symbol" in df.columns: + cols = ["symbol", *[c for c in ("name", "close", "change_pct", + "deviate_3d", "deviate_10d", "deviate_30d") if c in df.columns]] + df = df.select(cols) + for r in df.iter_rows(named=True): + rows[str(r["symbol"])] = { + "name": r.get("name"), + "close": r.get("close"), + "rt_pct": r.get("change_pct"), + "deviate_3d": r.get("deviate_3d"), + "deviate_10d": r.get("deviate_10d"), + "deviate_30d": r.get("deviate_30d"), + } + payload = {"_ts": now, "rows": rows, "cache_date": cache_date.isoformat() if cache_date else None} + with _hist_cache_lock: + _hist_cache["data"] = payload + return payload + + +def _bench_rt_pct(quote_service: Any) -> float: + """基准指数今日实时涨跌 (各候选均值, 缺数据时 0)。""" + try: + df = quote_service.get_index_quotes() + except Exception: + return 0.0 + if df is None or df.is_empty(): + return 0.0 + df = df.filter(pl.col("symbol").is_in(_BENCH_RT_CANDIDATES)) + if df.is_empty(): + return 0.0 + for col in ("change_pct", "pct", "pct_change"): + if col in df.columns: + vals = df[col].drop_nulls() + if vals.len() > 0: + return float(vals.mean()) + if {"close", "prev_close"} <= set(df.columns): + sub = df.select(["close", "prev_close"]).drop_nulls() + if sub.height > 0: + return float((sub["close"] / sub["prev_close"] - 1).mean()) + return 0.0 + + +def build_overview( + repo: Any, + quote_service: Any = None, + *, + min_closeness: float = 0.5, + limit: int = 200, +) -> dict[str, Any]: + """返回异动边缘总览: 规则表 + 按接近度排序的个股列表。""" + hist = _hist_snapshot(repo) + cache_date = hist.get("cache_date") + hist_rows: dict[str, dict[str, Any]] = hist["rows"] + + bench_rt = _bench_rt_pct(quote_service) if quote_service is not None else 0.0 + # enriched 已含今日收盘 (盘后已同步) 时, 今日涨跌已计入历史偏离, 不再叠加 + includes_today = cache_date is not None and cache_date >= date.today().isoformat() + + out_rows: list[dict[str, Any]] = [] + for symbol, base in hist_rows.items(): + rule = rule_for(symbol, base.get("name")) + rt_pct = base.get("rt_pct") + rt_delta = 0.0 if includes_today else ((rt_pct or 0.0) - bench_rt) + + windows: dict[str, dict[str, Any]] = {} + max_closeness = 0.0 + for n in DEVIATION_WINDOWS: + hist_dev = base.get(f"deviate_{n}d") + if hist_dev is None: + continue + live = hist_dev + rt_delta + threshold = rule.thresholds[n] + closeness = abs(live) / threshold if threshold > 0 else 0.0 + windows[f"{n}d"] = { + "value": round(live, 4), + "threshold": threshold, + "closeness": round(closeness, 4), + } + max_closeness = max(max_closeness, closeness) + if not windows or max_closeness < min_closeness: + continue + out_rows.append({ + "symbol": symbol, + "name": base.get("name"), + "board": rule.board, + "st": rule.st, + "close": base.get("close"), + "rt_pct": rt_pct, + "windows": windows, + "max_closeness": round(max_closeness, 4), + "status": _status_of(max_closeness), + }) + + out_rows.sort(key=lambda r: r["max_closeness"], reverse=True) + counts = { + _STATUS_TRIGGERED: sum(1 for r in out_rows if r["status"] == _STATUS_TRIGGERED), + _STATUS_EDGE: sum(1 for r in out_rows if r["status"] == _STATUS_EDGE), + _STATUS_WATCH: sum(1 for r in out_rows if r["status"] == _STATUS_WATCH), + } + return { + "asof": time.time(), + "cache_date": cache_date, + "bench_rt_pct": round(bench_rt, 4), + "includes_today": includes_today, + "rules": RULES_META, + "counts": counts, + "rows": out_rows[:limit], + } diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 38f2232..817bddd 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -195,6 +195,9 @@ class QuoteService: self._subscribers: set[QuoteSubscriber] = set() self._strategy_monitor = None # 延迟注入 self._app_state = None # 延迟注入 (FastAPI app.state) + # 异动边缘规则上次评估时间戳 (秒)。异动快照历史部分有 60s 缓存, + # 但每次构建仍有全市场循环, 轮询线程里限频到 30s 一次。 + self._abnormal_last_eval = 0.0 # 拉取元信息 (给 SSE / status 用) self._fetch_time: float = 0.0 # perf_counter (用于计算 quote_age_ms) @@ -1123,6 +1126,23 @@ class QuoteService: enriched_today if stock_ready else pl.DataFrame(), self.get_index_quotes(), ) + # 异动边缘规则轮: 快照 (enriched 偏离列 + 实时叠加) 由 + # abnormal_moves.build_overview 统一构建, 引擎只做边缘触发判定。 + # 30s 限频 —— 快照历史部分 60s 缓存, 无需跟行情轮询同频重算。 + if engine.has_rule_type("abnormal") and self._repo is not None: + _now_ts = time.time() + if _now_ts - self._abnormal_last_eval >= 30.0: + self._abnormal_last_eval = _now_ts + try: + from app.services import abnormal_moves + _overview = abnormal_moves.build_overview( + self._repo, self, + min_closeness=engine.min_abnormal_closeness(), + limit=1000, + ) + rule_events += engine.evaluate_abnormal(_overview.get("rows") or []) + except Exception as e: # noqa: BLE001 + logger.warning("异动监控规则评估失败 (不影响其他告警): %s", e) # ETF 规则轮: 股票快照不含 ETF, 用 ETF enriched 快照单独评估。 # 独立 try —— ETF 轮任何异常都不得丢弃本轮已算出的股票告警。 # refresh=False —— 不在轮询线程上触发 ETF 冷缓存的同步重算 (缓存由 ETF 实时 @@ -1182,6 +1202,8 @@ class QuoteService: "sector_source_field", "sector_value", "sector_level", "window_change_pct", "coverage_ratio", "valid_count", "total_count", "up_count", "down_count", "leader", + "abnormal_window", "abnormal_value", "abnormal_threshold", + "abnormal_closeness", ): if key in ev: alert[key] = ev[key] diff --git a/backend/app/strategy/custom_signals.py b/backend/app/strategy/custom_signals.py index eb957c4..8852b0e 100644 --- a/backend/app/strategy/custom_signals.py +++ b/backend/app/strategy/custom_signals.py @@ -51,6 +51,8 @@ ALLOWED_FIELDS: frozenset[str] = frozenset({ "momentum_5d", "momentum_10d", "momentum_20d", "momentum_30d", "momentum_60d", "annual_vol_20d", "rsi_6", "rsi_14", "rsi_24", + # 异动偏离 (交易所异动规则口径, 运行时列) + "deviate_3d", "deviate_10d", "deviate_30d", }) # 运算符 → Polars 表达式构造器（输入 col_expr, value） diff --git a/backend/app/strategy/monitor.py b/backend/app/strategy/monitor.py index 896e377..84d3bc2 100644 --- a/backend/app/strategy/monitor.py +++ b/backend/app/strategy/monitor.py @@ -300,6 +300,9 @@ class MonitorRuleEngine: 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 规则据此跑选股。""" @@ -371,6 +374,7 @@ class MonitorRuleEngine: rule.get("direction"), rule.get("threshold_pct"), rule.get("window_minutes"), + rule.get("abnormal_window"), ) def set_rules(self, rules: list[dict]) -> None: @@ -413,6 +417,11 @@ class MonitorRuleEngine: 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: @@ -607,7 +616,7 @@ class MonitorRuleEngine: for rule_id, rule in list(self._rules.items()): if rule.get("asset_type", "stock") != asset_type: continue - if rule.get("type") == "sector": + if rule.get("type") in ("sector", "abnormal"): continue try: events.extend(self._evaluate_rule(df, rule, now)) @@ -771,6 +780,133 @@ class MonitorRuleEngine: ) 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")) + scope_symbols = ( + {str(s) for s in rule.get("symbols", []) if s} + if rule.get("scope") == "symbols" else 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 过滤作用域 diff --git a/backend/app/strategy/monitor_rules.py b/backend/app/strategy/monitor_rules.py index 0e16b28..18fe4f6 100644 --- a/backend/app/strategy/monitor_rules.py +++ b/backend/app/strategy/monitor_rules.py @@ -28,7 +28,7 @@ logger = logging.getLogger(__name__) # ── 常量 ──────────────────────────────────────────────── ID_RE = re.compile(r"^[a-z0-9_]{1,40}$") -RULE_TYPES = {"strategy", "signal", "price", "market", "ladder", "sector"} +RULE_TYPES = {"strategy", "signal", "price", "market", "ladder", "sector", "abnormal"} SCOPES = {"symbols", "all", "sector"} LOGICS = {"and", "or"} DIRECTIONS = {"entry", "exit", "both"} @@ -42,6 +42,9 @@ LADDER_DIRECTIONS = {"up", "down"} SECTOR_KINDS = {"index", "concept", "industry"} SECTOR_TRIGGERS = {"change_pct", "momentum"} SECTOR_WINDOWS = {1, 3, 5, 10, 15} +# abnormal 规则 (异动边缘): 接近度方向 / 关注窗口 +ABNORMAL_DIRECTIONS = {"up", "down", "both"} +ABNORMAL_WINDOWS = {"any", "3d", "10d", "30d"} # 布尔信号列前缀 (op=truth 时 field 取这些) _SIGNAL_PREFIXES = ("signal_", "csg_") @@ -176,6 +179,17 @@ def validate(rule: dict) -> None: raise ValueError("板块监控阈值必须大于 0 且不超过 20%") if rule.get("sector_trigger") == "momentum" and rule.get("window_minutes") not in SECTOR_WINDOWS: raise ValueError(f"板块异动窗口必须是 {sorted(SECTOR_WINDOWS)} 分钟之一") + elif rule.get("type") == "abnormal": + # 异动边缘监控: threshold_pct = 接近度阈值% (|偏离值|/规则阈值), 不用 conditions + if rule.get("asset_type", "stock") != "stock": + raise ValueError("异动监控仅支持个股 (偏离值仅对个股计算)") + if rule.get("direction", "both") not in ABNORMAL_DIRECTIONS: + raise ValueError(f"异动监控 direction 必须是 {ABNORMAL_DIRECTIONS} 之一") + if rule.get("abnormal_window", "any") not in ABNORMAL_WINDOWS: + raise ValueError(f"异动监控窗口必须是 {sorted(ABNORMAL_WINDOWS)} 之一") + threshold_pct = rule.get("threshold_pct") + if not isinstance(threshold_pct, (int, float)) or not 1 <= threshold_pct <= 150: + raise ValueError("异动接近度阈值必须是 1 到 150 之间的百分比数字") else: # 信号/价格/市场类型: 需要 conditions conds = rule.get("conditions") @@ -231,17 +245,21 @@ def normalize(rule: dict) -> dict: r = dict(rule) r.setdefault("enabled", True) r.setdefault("asset_type", "stock") - r.setdefault("scope", "symbols") + # sector/abnormal 默认全市场 (sector 随后强制 all; abnormal 支持指定标的) + r.setdefault("scope", "all" if r.get("type") in {"sector", "abnormal"} else "symbols") r.setdefault("symbols", []) r.setdefault("sector", None) r.setdefault("sector_kind", None) r.setdefault("sector_targets", []) r.setdefault("sector_trigger", "change_pct") - r.setdefault("threshold_pct", 1.0) + r.setdefault("threshold_pct", 70.0 if r.get("type") == "abnormal" else 1.0) r.setdefault("window_minutes", 5) r.setdefault("strategy_id", None) - # direction 默认值: ladder/sector 用 "up", 其余用 "entry" - r.setdefault("direction", "up" if r.get("type") in {"ladder", "sector"} else "entry") + # direction 默认值: ladder/sector 用 "up", abnormal 用 "both", 其余用 "entry" + r.setdefault( + "direction", + "up" if r.get("type") in {"ladder", "sector"} else "both" if r.get("type") == "abnormal" else "entry", + ) if r.get("type") == "strategy": r.setdefault("score_min", None) r.setdefault("score_max", None) @@ -261,6 +279,8 @@ def normalize(rule: dict) -> dict: if r.get("type") == "sector": r["scope"] = "all" r["symbols"] = [] + # abnormal 专属默认字段 (异动边缘监控) + r.setdefault("abnormal_window", "any") r.setdefault("logic", "and") r.setdefault("cooldown_seconds", 3600) r.setdefault("severity", "info") diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index 7f6705c..5e57d0f 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -578,6 +578,10 @@ class KlineRepository: df_full = compute_indicators(df_hist) logger.info("enriched refresh step done: compute indicators rows=%d (%.2fs)", len(df_full), time.perf_counter() - step) + # 异动偏离列 (deviate_Nd = 个股动量 - 基准指数动量), 运行时附着 + from app.indicators.pipeline import attach_deviation_columns + df_full = attach_deviation_columns(df_full, self.store.data_dir) + step = time.perf_counter() logger.info("enriched refresh step start: compute signals") df_full = compute_signals(df_full) diff --git a/backend/tests/test_abnormal_moves.py b/backend/tests/test_abnormal_moves.py new file mode 100644 index 0000000..b6b1d44 --- /dev/null +++ b/backend/tests/test_abnormal_moves.py @@ -0,0 +1,296 @@ +"""异动边缘统计测试 — 偏离列附着 + 规则口径 + 快照接近度。""" +from __future__ import annotations + +from datetime import date + +import polars as pl + +from app.indicators.pipeline import attach_deviation_columns, load_benchmark_momentum +from app.services.abnormal_moves import ( + _hist_cache, + _hist_cache_lock, + board_of, + build_overview, + is_st_name, + rule_for, +) + + +def _write_index_daily(tmp_path, rows: list[tuple[str, date, float]]) -> None: + df = pl.DataFrame( + { + "symbol": [r[0] for r in rows], + "date": [r[1] for r in rows], + "close": [r[2] for r in rows], + } + ) + for dt in sorted({r[1] for r in rows}): + target = tmp_path / "kline_index_daily" / f"date={dt.isoformat()}" + target.mkdir(parents=True, exist_ok=True) + df.filter(pl.col("date") == dt).write_parquet(target / "part.parquet") + + +def test_attach_deviation_columns_math(tmp_path) -> None: + # 上证指数 4 天等差 +1: 3日动量 = 13/10-1 = 0.30 + # 个股 close 与指数同序列 → momentum_3d 缺失时按 close 就地补算, 偏离 = 0 + days = [date(2026, 8, 13), date(2026, 8, 14), date(2026, 8, 15), date(2026, 8, 18)] + index_rows = [("000001.SH", d, 10.0 + i) for i, d in enumerate(days)] + _write_index_daily(tmp_path, index_rows) + + stock = pl.DataFrame( + { + "symbol": ["600000.SH"] * len(days), + "date": days, + "close": [10.0 + i for i in range(len(days))], + # 10/30 日窗口已有动量列 → 直接使用 + "momentum_10d": [None] * 4, + "momentum_30d": [None] * 4, + } + ) + out = attach_deviation_columns(stock, tmp_path) + assert "deviate_3d" in out.columns + assert "momentum_3d" in out.columns # 就地补算 + last = out.sort("date").row(-1, named=True) + assert abs(last["deviate_3d"] - 0.0) < 1e-9 + + +def test_attach_deviation_columns_missing_benchmark(tmp_path) -> None: + # 无指数数据: 偏离列为 null, 不抛异常 + stock = pl.DataFrame( + { + "symbol": ["600000.SH"], + "date": [date(2026, 8, 18)], + "momentum_3d": [0.2], + "momentum_10d": [0.5], + "momentum_30d": [1.0], + } + ) + out = attach_deviation_columns(stock, tmp_path) + assert out["deviate_3d"][0] is None + + +def test_board_and_st_rules() -> None: + assert board_of("600000.SH") == "主板" + assert board_of("000001.SZ") == "主板" + assert board_of("301123.SZ") == "创业板" + assert board_of("688123.SH") == "科创板" + assert board_of("920001.BJ") == "北交所" + assert is_st_name("*ST 某某") is True + assert is_st_name("正常股") is False + + main = rule_for("600000.SH", "正常股") + assert main.thresholds == {3: 0.20, 10: 1.00, 30: 2.00} + st = rule_for("600000.SH", "ST 某某") + assert st.thresholds == {3: 0.15, 10: 0.50, 30: 1.00} + gem = rule_for("301123.SZ", "正常股") + assert gem.thresholds[3] == 0.30 + bse = rule_for("920001.BJ", "正常股") + assert bse.thresholds[3] == 0.40 + + +class _FakeRepo: + """最小 repo: get_enriched_latest 返回构造帧。""" + + def __init__(self, df: pl.DataFrame) -> None: + self._df = df + + def get_enriched_latest(self): + return self._df, date(2026, 8, 19) + + +class _FakeQuotes: + def get_index_quotes(self): + return pl.DataFrame( + {"symbol": ["000001.SH"], "close": [3300.0], "prev_close": [3270.0]} + ) + + +def test_build_overview_closeness_and_status() -> None: + with _hist_cache_lock: + _hist_cache.clear() + df = pl.DataFrame( + { + "symbol": ["600000.SH", "300001.SZ", "000002.SZ"], + "name": ["股A", "股B", "股C"], + "close": [10.0, 20.0, 30.0], + "change_pct": [0.05, 0.02, 0.01], + "deviate_3d": [0.19, 0.35, 0.05], + "deviate_10d": [0.99, 0.40, 0.20], + "deviate_30d": [1.95, 2.10, 0.60], + } + ) + result = build_overview(_FakeRepo(df), _FakeQuotes(), min_closeness=0.5, limit=10) + + by_symbol = {r["symbol"]: r for r in result["rows"]} + # 主板: 3d阈值0.2 → 0.19/0.2=0.95 边缘; 指数实时 +30/3270≈0.00917 叠加后略增 + a = by_symbol["600000.SH"] + assert a["status"] in ("edge", "triggered") + # 创业板: 30日 2.10/2.00 ≥ 1 → triggered + b = by_symbol["300001.SZ"] + assert b["status"] == "triggered" + # 000002: 3d 0.05/0.2=0.25, 10d 0.2/1=0.2, 30d 0.6/2=0.3 → 全部 < 0.5 被过滤 + assert "000002.SZ" not in by_symbol + # 排序按接近度降序 + closeness = [r["max_closeness"] for r in result["rows"]] + assert closeness == sorted(closeness, reverse=True) + assert result["counts"]["triggered"] >= 1 + + +def test_build_overview_cache_date_today_no_double_count() -> None: + """cache_date >= 今天时不再叠加实时涨跌 (避免重复计入)。""" + with _hist_cache_lock: + _hist_cache.clear() + + class _TodayRepo(_FakeRepo): + def get_enriched_latest(self): + return self._df, date.today() + + df = pl.DataFrame( + { + "symbol": ["600000.SH"], + "name": ["股A"], + "close": [10.0], + "change_pct": [0.05], + "deviate_3d": [0.19], + "deviate_10d": [None], + "deviate_30d": [None], + } + ) + result = build_overview(_TodayRepo(df), _FakeQuotes(), min_closeness=0.5) + row = result["rows"][0] + assert abs(row["windows"]["3d"]["value"] - 0.19) < 1e-9 + + +# ── 监控规则接入 (type=abnormal) ──────────────────────── + +import pytest + +from app.strategy import monitor_rules +from app.strategy.monitor import MonitorRuleEngine + + +def _ab_rule(**overrides) -> dict: + rule = { + "id": "r_ab", + "name": "异动边缘", + "type": "abnormal", + "scope": "all", + "symbols": [], + "threshold_pct": 70, + "direction": "both", + "abnormal_window": "any", + "cooldown_seconds": 0, + "severity": "warn", + } + rule.update(overrides) + return rule + + +def _row(symbol: str, *wins: tuple[str, float], name: str = "股A", + board: str = "主板", rt_pct: float = 0.05) -> dict: + # wins: (窗口, 偏离值) — 阈值按交易所口径: 主板 3d=0.2, 10d=1.0, 30d=2.0 + thresholds = {"3d": 0.2, "10d": 1.0, "30d": 2.0} + windows = { + key: {"value": value, "threshold": thresholds[key], + "closeness": round(abs(value) / thresholds[key], 4)} + for key, value in wins + } + return {"symbol": symbol, "name": name, "board": board, "st": False, + "close": 10.0, "rt_pct": rt_pct, "windows": windows} + + +def test_abnormal_rule_validation_and_defaults() -> None: + rule = monitor_rules.normalize({"id": "r1", "name": "n", "type": "abnormal"}) + assert rule["direction"] == "both" + assert rule["threshold_pct"] == 70.0 + assert rule["abnormal_window"] == "any" + monitor_rules.validate(rule) + + monitor_rules.validate(_ab_rule(threshold_pct=100, direction="up", abnormal_window="3d")) + + with pytest.raises(ValueError): + monitor_rules.validate(_ab_rule(abnormal_window="5d")) + with pytest.raises(ValueError): + monitor_rules.validate(_ab_rule(threshold_pct=0.5)) + with pytest.raises(ValueError): + monitor_rules.validate(_ab_rule(asset_type="etf")) + with pytest.raises(ValueError): + monitor_rules.validate(_ab_rule(direction="entry")) + + +def test_engine_abnormal_edge_trigger_and_cooldown() -> None: + engine = MonitorRuleEngine() + engine.set_rules([_ab_rule()]) + assert engine.min_abnormal_closeness() == pytest.approx(0.7) + + # 首轮观测不触发 (防新建规则刷屏); 0.10/0.2 = 50% 接近度, 低于阈值 + assert engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.10))], now=1000.0) == [] + # 上穿 70% → 触发 (0.16/0.2 = 80%) + events = engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.16))], now=1006.0) + assert len(events) == 1 + ev = events[0] + assert ev["source"] == "abnormal" + assert ev["type"] == "abnormal_up" + assert ev["symbol"] == "600000.SH" + assert ev["abnormal_window"] == "3d" + assert ev["abnormal_closeness"] == pytest.approx(0.8) + assert "接近" in ev["message"] or "已达" in ev["message"] + # 持续高于阈值: 不重复触发 (边缘语义) + assert engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.18))], now=1012.0) == [] + # 回落再上穿: cooldown=0 时再次触发 + engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.10))], now=1018.0) + assert len(engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.17))], now=1024.0)) == 1 + + # cooldown 内的上穿被抑制 + engine_cd = MonitorRuleEngine() + engine_cd.set_rules([_ab_rule(cooldown_seconds=3600)]) + engine_cd.evaluate_abnormal([_row("600000.SH", ("3d", 0.10))], now=1000.0) + engine_cd.evaluate_abnormal([_row("600000.SH", ("3d", 0.16))], now=1006.0) + engine_cd.evaluate_abnormal([_row("600000.SH", ("3d", 0.10))], now=1012.0) + assert engine_cd.evaluate_abnormal([_row("600000.SH", ("3d", 0.16))], now=1018.0) == [] + + +def test_engine_abnormal_stale_symbol_state_cleared() -> None: + """标的跌出快照后状态应清回 False, 回升穿过阈值时可再次触发。""" + engine = MonitorRuleEngine() + engine.set_rules([_ab_rule()]) + engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.10))], now=1000.0) # 首轮 False + assert len(engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.18))], now=1006.0)) == 1 + # 跌出预过滤区间 (快照中消失) + engine.evaluate_abnormal([], now=1012.0) + # 重新出现且超阈值 → 重新触发 + assert len(engine.evaluate_abnormal([_row("600000.SH", ("3d", 0.18))], now=1018.0)) == 1 + + +def test_engine_abnormal_direction_window_scope_filters() -> None: + # 方向: 只报上涨偏离 + engine = MonitorRuleEngine() + engine.set_rules([_ab_rule(direction="up")]) + engine.evaluate_abnormal([_row("600000.SH", ("3d", -0.16))], now=1000.0) + assert engine.evaluate_abnormal([_row("600000.SH", ("3d", -0.19))], now=1006.0) == [] + + # 窗口: 只看 3d (10d/30d 的偏离不参与) + engine = MonitorRuleEngine() + engine.set_rules([_ab_rule(abnormal_window="3d")]) + engine.evaluate_abnormal([_row("600000.SH", ("10d", 0.98))], now=1000.0) + assert engine.evaluate_abnormal([_row("600000.SH", ("10d", 0.99))], now=1006.0) == [] + + # 作用域: 只监控指定标的 + engine = MonitorRuleEngine() + engine.set_rules([_ab_rule(scope="symbols", symbols=["600000.SH"])]) + engine.evaluate_abnormal( + [_row("600000.SH", ("3d", 0.10)), _row("000001.SZ", ("3d", 0.10))], now=1000.0, + ) + events = engine.evaluate_abnormal( + [_row("600000.SH", ("3d", 0.16)), _row("000001.SZ", ("3d", 0.19))], now=1006.0, + ) + assert [ev["symbol"] for ev in events] == ["600000.SH"] + + +def test_engine_abnormal_down_direction_event_type() -> None: + engine = MonitorRuleEngine() + engine.set_rules([_ab_rule(direction="down")]) + engine.evaluate_abnormal([_row("600000.SH", ("3d", -0.10))], now=1000.0) + events = engine.evaluate_abnormal([_row("600000.SH", ("3d", -0.16))], now=1006.0) + assert len(events) == 1 + assert events[0]["type"] == "abnormal_down" diff --git a/backend/tests/test_enriched_generation.py b/backend/tests/test_enriched_generation.py index a48ccfb..f9b9130 100644 --- a/backend/tests/test_enriched_generation.py +++ b/backend/tests/test_enriched_generation.py @@ -86,6 +86,41 @@ def test_recovery_replaces_stale_publication_but_not_active_owner(tmp_path) -> N assert pl.read_parquet(out)["close"].item() == pytest.approx(12.0) +def test_recovery_takes_over_when_owner_pid_is_dead_on_windows(tmp_path, monkeypatch) -> None: + # 跨进程孤儿锁: 属主进程已死, 但 Windows 的 os.kill(pid, 0) 对不存在的 pid + # 抛 WinError 87 (ERROR_INVALID_PARAMETER) 而非 ProcessLookupError, + # 存活探测若把它当"存活", recover 将永远报 another publication is active。 + stale = { + "state": "publishing", + "generation": "stale-generation", + "publication_id": "stale-publication", + "owner_pid": 12345, + "updated_at_ns": 0, + } + (tmp_path / ".matrix_generation_stock.json").write_text( + json.dumps(stale), encoding="utf-8" + ) + + def probe(_pid: int, _sig: int) -> None: + error = OSError() + error.winerror = 87 + raise error + + monkeypatch.setattr("app.enriched_generation.os.kill", probe) + + out = tmp_path / "kline_daily_enriched" / "date=2026-08-14" / "part.parquet" + recovered = EnrichedPublication(tmp_path, recover=True) + recovered.write_parquet(_frame(10.0), out) + recovered.commit() + + marker = json.loads( + (tmp_path / ".matrix_generation_stock.json").read_text(encoding="utf-8") + ) + assert marker["state"] == "ready" + assert get_enriched_generation(tmp_path, "stock") == marker["generation"] + assert pl.read_parquet(out)["close"].item() == pytest.approx(10.0) + + def test_panel_cache_generation_change_forces_recompute() -> None: cache = PanelCache() calls: list[int] = [] diff --git a/frontend/src/components/AlertToast.tsx b/frontend/src/components/AlertToast.tsx index 4a0a4c6..511bf7c 100644 --- a/frontend/src/components/AlertToast.tsx +++ b/frontend/src/components/AlertToast.tsx @@ -89,6 +89,7 @@ const SOURCE_BADGE: Record = { price: { label: '价格', cls: 'bg-emerald-400/15 text-emerald-400' }, market: { label: '异动', cls: 'bg-purple-500/15 text-purple-400' }, sector: { label: '板块', cls: 'bg-cyan-500/15 text-cyan-700 dark:text-cyan-300' }, + abnormal: { label: '异动边缘', cls: 'bg-orange-500/15 text-orange-500 dark:text-orange-400' }, pool_entry: { label: '进入', cls: 'bg-danger/15 text-danger' }, pool_exit: { label: '移出', cls: 'bg-bear/15 text-bear' }, buy_signal: { label: '买入', cls: 'bg-danger/15 text-danger' }, diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index 67ec645..5257051 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -22,6 +22,7 @@ import { import { QK } from '@/lib/queryKeys' import { tierRank } from '@/lib/capability-labels' import { + Siren, Star, ScanSearch, History, @@ -90,6 +91,7 @@ const nav = [ { to: '/financials', label: '财务分析', icon: FileText }, { to: '/monitor', label: '监控中心', icon: RadioTower }, { to: '/regime', label: '市场环境', icon: Gauge }, + { to: '/abnormal', label: '异动监控', icon: Siren }, { to: '/review', label: '复盘', icon: BookOpenCheck }, { to: '/indices', label: '指数', icon: BarChart3 }, { to: '/data', label: '数据', icon: Database }, diff --git a/frontend/src/components/monitor/RuleEditor.tsx b/frontend/src/components/monitor/RuleEditor.tsx index bc582d5..7198148 100644 --- a/frontend/src/components/monitor/RuleEditor.tsx +++ b/frontend/src/components/monitor/RuleEditor.tsx @@ -1,7 +1,7 @@ import { useEffect, useMemo, useRef, useState } from 'react' import { Link } from 'react-router-dom' import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' -import { Activity, Building2, ChartNoAxesCombined, Check, ChevronDown, ChevronUp, Eraser, Layers3, ListPlus, Plus, RadioTower, Save, Search, Tags, TrendingUp, Waypoints, X } from 'lucide-react' +import { Activity, Building2, ChartNoAxesCombined, Check, ChevronDown, ChevronUp, Eraser, Layers3, ListPlus, Plus, RadioTower, Save, Search, Siren, Tags, TrendingUp, Waypoints, X } from 'lucide-react' import { api, genRuleId, type MonitorRule, type MonitorCondition, type SectorKind, type SectorMonitorTarget, type StrategyNotifyEvent } from '@/lib/api' import { DEFAULT_STRATEGY_NOTIFY_EVENTS, LEGACY_STRATEGY_NOTIFY_EVENTS, STRATEGY_NOTIFY_EVENT_OPTIONS } from '@/lib/strategyMonitorEvents' import { QK } from '@/lib/queryKeys' @@ -23,7 +23,7 @@ interface Props { } const TYPE_DEFAULT_NAME: Record = { - signal: '信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控', sector: '板块监控', + signal: '信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控', sector: '板块监控', abnormal: '异动监控', } const TYPE_ICONS = { @@ -32,6 +32,7 @@ const TYPE_ICONS = { market: RadioTower, strategy: Waypoints, sector: Layers3, + abnormal: Siren, } const SECTOR_KIND_OPTIONS: Array<{ key: SectorKind; label: string; icon: typeof ChartNoAxesCombined }> = [ @@ -61,6 +62,7 @@ const emptyRule = (preset?: Partial): MonitorRule => ({ sector_trigger: 'change_pct', threshold_pct: 1, window_minutes: 5, + abnormal_window: 'any', strategy_id: null, score_min: null, score_max: null, @@ -157,6 +159,8 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { const base = TYPE_DEFAULT_NAME[d.type] ?? '监控规则' d.name = d.type === 'sector' && d.sector_targets?.length ? `${base} · ${d.sector_targets[0].name}${d.sector_targets.length > 1 ? ` 等${d.sector_targets.length}个` : ''}` + : d.type === 'abnormal' + ? `${base} · 接近度≥${d.threshold_pct ?? 70}%${d.abnormal_window && d.abnormal_window !== 'any' ? ` (${d.abnormal_window.toUpperCase()})` : ''}` : d.scope === 'symbols' && d.symbols.length > 0 ? `${base} · ${d.symbols[0]}${d.symbols.length > 1 ? ` 等${d.symbols.length}只` : ''}` : base @@ -181,6 +185,14 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { delete d.notify_events if (!d.sector_targets?.length) throw new Error('请选择至少一个监控对象') if ((d.threshold_pct ?? 0) <= 0 || (d.threshold_pct ?? 0) > 20) throw new Error('阈值必须大于 0 且不超过 20%') + } else if (d.type === 'abnormal') { + delete d.score_min + delete d.score_max + d.conditions = [] + delete d.notify_events + if ((d.threshold_pct ?? 0) < 1 || (d.threshold_pct ?? 0) > 150) { + throw new Error('接近度阈值必须在 1 到 150 之间 (70=边缘, 100=已触发)') + } } else { delete d.score_min delete d.score_max @@ -474,8 +486,8 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { - {/* 资产类型: 股票 / ETF / 指数 (个股极简模式不显示) */} - {!simple && draft.type !== 'sector' && ( + {/* 资产类型: 股票 / ETF / 指数 (个股极简模式不显示; 板块/异动仅个股) */} + {!simple && draft.type !== 'sector' && draft.type !== 'abnormal' && (
资产类型
@@ -510,7 +522,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { {/* 监控类型 */}
监控类型 -
+
{visibleTypes.map(t => { const Icon = TYPE_ICONS[t.key as keyof typeof TYPE_ICONS] ?? Activity const active = draft.type === t.key @@ -527,10 +539,16 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) { notify_events: type === 'strategy' ? [...(d.notify_events ?? DEFAULT_STRATEGY_NOTIFY_EVENTS)] : undefined, - scope: type === 'sector' + scope: type === 'sector' || type === 'abnormal' ? 'all' : type === 'strategy' && d.scope === 'symbols' && d.symbols.length === 0 ? 'all' : d.scope, - direction: type === 'sector' ? 'up' : d.type === 'sector' ? 'entry' : d.direction, + direction: type === 'sector' ? 'up' + : type === 'abnormal' ? 'both' + : d.type === 'sector' || d.type === 'abnormal' ? 'entry' : d.direction, + // 异动规则复用 threshold_pct 存接近度阈值%, 其他类型为涨跌幅% + threshold_pct: type === 'abnormal' && d.type !== 'abnormal' ? 70 + : type !== 'abnormal' && d.type === 'abnormal' ? 1 + : d.threshold_pct, } })} 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 ${ @@ -748,6 +766,80 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
)} + {draft.type === 'abnormal' && ( +
+
+ +
+ 方向 +
+ {([ + ['both', '全部'], + ['up', '涨势偏离'], + ['down', '跌势偏离'], + ] as const).map(([key, label]) => ( + + ))} +
+
+
+
+ 关注窗口 +
+ {([ + ['any', '全部'], + ['3d', '3日 (异常波动)'], + ['10d', '10日 (严重)'], + ['30d', '30日 (严重)'], + ] as const).map(([key, label]) => ( + + ))} +
+
+
+ 按交易所异动规则口径 (3日±20%/30%… 10日+100%、30日+200% 等按板块) 计算 + 个股涨跌幅偏离值的接近度, 上穿阈值时告警; 冷却期内同一标的不重复提醒。 +
+
+ )} + {/* 作用范围 */} {draft.type !== 'sector' &&
作用范围 @@ -878,7 +970,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
} {/* 触发条件 (非 strategy) */} - {draft.type !== 'strategy' && draft.type !== 'sector' && ( + {draft.type !== 'strategy' && draft.type !== 'sector' && draft.type !== 'abnormal' && (
触发条件 diff --git a/frontend/src/components/stock-table/primitives.tsx b/frontend/src/components/stock-table/primitives.tsx index 71d4610..38f6780 100644 --- a/frontend/src/components/stock-table/primitives.tsx +++ b/frontend/src/components/stock-table/primitives.tsx @@ -114,6 +114,9 @@ export function renderBuiltinDataCell(r: any, col: ColumnConfig): ReactNode | nu case 'momentum_20d': return {fmtPct(r.momentum_20d)} case 'momentum_30d': return {fmtPct(r.momentum_30d)} case 'momentum_60d': return {fmtPct(r.momentum_60d)} + case 'deviate_3d': return {fmtPct(r.deviate_3d)} + case 'deviate_10d': return {fmtPct(r.deviate_10d)} + case 'deviate_30d': return {fmtPct(r.deviate_30d)} // 连板 case 'limit_ups': return ( diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index b715743..cf1a52b 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -735,11 +735,49 @@ export interface SectorMonitorTarget { member_count: number } +export interface AbnormalWindowInfo { + /** 实时偏离值 (小数) */ + value: number + /** 该窗口阈值 (小数) */ + threshold: number + /** 接近度 |value|/threshold */ + closeness: number +} + +export type AbnormalStatus = 'triggered' | 'edge' | 'watch' + +export interface AbnormalRow { + symbol: string + name: string | null + board: string + st: boolean + close: number | null + rt_pct: number | null + windows: Record + max_closeness: number + status: AbnormalStatus +} + +export interface AbnormalOverview { + asof: number + cache_date: string | null + bench_rt_pct: number + includes_today: boolean + rules: Array<{ + board: string + st: boolean + thresholds: Record + note: string + }> + counts: { triggered: number; edge: number; watch: number } + rows: AbnormalRow[] +} + export interface MonitorRule { id: string name: string enabled: boolean - type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder' | 'sector' + type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder' | 'sector' | 'abnormal' asset_type?: 'stock' | 'etf' | 'index' scope: 'symbols' | 'all' | 'sector' symbols: string[] @@ -749,6 +787,8 @@ export interface MonitorRule { sector_trigger?: 'change_pct' | 'momentum' threshold_pct?: number window_minutes?: 1 | 3 | 5 | 10 | 15 + /** abnormal 专属: 关注窗口 (any=全部) */ + abnormal_window?: 'any' | '3d' | '10d' | '30d' strategy_id?: string | null direction: 'entry' | 'exit' | 'both' | 'up' | 'down' notify_events?: StrategyNotifyEvent[] @@ -817,6 +857,11 @@ export interface AlertEvent { up_count?: number down_count?: number leader?: { symbol?: string; name?: string; change_pct?: number } | null + /** 异动边缘告警 (source=abnormal) 附加字段 */ + abnormal_window?: string + abnormal_value?: number + abnormal_threshold?: number + abnormal_closeness?: number /** ext 富化字段 (行业/概念等), 键为 "{configId}__{fieldName}" */ [key: string]: unknown } @@ -2780,6 +2825,12 @@ export const api = { customSignalDelete: (id: string) => request<{ ok: boolean }>(`/api/custom-signals/${encodeURIComponent(id)}`, { method: 'DELETE' }), + // ===== Abnormal Moves (异动边缘) ===== + abnormalOverview: (minCloseness = 0.5, limit = 200) => + request( + `/api/abnormal/overview?min_closeness=${minCloseness}&limit=${limit}`, + ), + // ===== Monitor Rules (监控规则) ===== monitorRulesList: () => request<{ rules: MonitorRule[] }>('/api/monitor-rules'), diff --git a/frontend/src/lib/queryKeys.ts b/frontend/src/lib/queryKeys.ts index 1e048a4..4f7f140 100644 --- a/frontend/src/lib/queryKeys.ts +++ b/frontend/src/lib/queryKeys.ts @@ -26,6 +26,8 @@ export const QK = { watchlistGroups: ['watchlist-groups'] as const, watchlistQuotes: ['watchlist-quotes'] as const, watchlistEnriched: (ext?: string) => ['watchlist-enriched', ext] as const, + // 异动边缘总览 (开启监控时才查询, 参数为 min_closeness/limit) + abnormalOverview: (minCloseness: number, limit: number) => ['abnormal-overview', minCloseness, limit] as const, // 不用 watchlist- 前缀: 日K历史盘中几乎不变, 若被 SSE quotes_updated 高频失效 // (expert 1s) 会导致全自选日K每秒重拉, staleTime 形同虚设。 // 刷新点: staleTime 过期 + Watchlist 增删自选/改蜡烛天数时的手动失效; diff --git a/frontend/src/lib/screener-columns.ts b/frontend/src/lib/screener-columns.ts index 24f0f3d..4492e53 100644 --- a/frontend/src/lib/screener-columns.ts +++ b/frontend/src/lib/screener-columns.ts @@ -65,6 +65,10 @@ export const SCREENER_BUILTIN_COLUMNS: ColumnConfig[] = [ { id: 'builtin:momentum_20d', source: { type: 'builtin', key: 'momentum_20d' }, label: '20D 动量', visible: false, align: 'right' }, { id: 'builtin:momentum_30d', source: { type: 'builtin', key: 'momentum_30d' }, label: '30D 动量', visible: false, align: 'right' }, { id: 'builtin:momentum_60d', source: { type: 'builtin', key: 'momentum_60d' }, label: '60D 动量', visible: true, align: 'right' }, + // 异动偏离 (交易所异动规则口径, 运行时列; 默认隐藏, 在列组「异动」中开启) + { id: 'builtin:deviate_3d', source: { type: 'builtin', key: 'deviate_3d' }, label: '3D 偏离', visible: false, align: 'right' }, + { id: 'builtin:deviate_10d', source: { type: 'builtin', key: 'deviate_10d' }, label: '10D 偏离', visible: false, align: 'right' }, + { id: 'builtin:deviate_30d', source: { type: 'builtin', key: 'deviate_30d' }, label: '30D 偏离', visible: false, align: 'right' }, // 连板 { id: 'builtin:limit_ups', source: { type: 'builtin', key: 'limit_ups' }, label: '连板', visible: true, align: 'center' }, { id: 'builtin:limit_downs', source: { type: 'builtin', key: 'limit_downs' }, label: '连跌', visible: false, align: 'center' }, @@ -93,6 +97,7 @@ export const SCREENER_COLUMN_GROUPS: ColumnGroup[] = [ { id: 'range', label: '区间', icon: '📏', keys: ['high_60d', 'low_60d'] }, { id: 'tech', label: '技术指标', icon: '🔬', keys: ['rsi6', 'rsi14', 'rsi24', 'macd_dif', 'macd_dea', 'macd_hist', 'kdj_k', 'kdj_d', 'kdj_j', 'boll_upper', 'boll_lower', 'atr14', 'vol_ma5', 'vol_ma10'] }, { id: 'momentum', label: '动量', icon: '🚀', keys: ['momentum_5d', 'momentum_10d', 'momentum_20d', 'momentum_30d', 'momentum_60d'] }, + { id: 'abnormal', label: '异动', icon: '⚡', keys: ['deviate_3d', 'deviate_10d', 'deviate_30d'] }, { id: 'limit', label: '连板', icon: '🔥', keys: ['limit_ups', 'limit_downs'] }, { id: 'signal', label: '信号', icon: '📡', keys: ['signals', 'candle', 'intraday'] }, { id: 'finance', label: '财务', icon: '📋', keys: ['eps', 'bps', 'roe', 'pe_ttm', 'pb', 'gross_margin', 'net_margin', 'revenue_yoy', 'net_income_yoy', 'debt_ratio'] }, diff --git a/frontend/src/lib/stock-table.ts b/frontend/src/lib/stock-table.ts index 8953a12..54852b8 100644 --- a/frontend/src/lib/stock-table.ts +++ b/frontend/src/lib/stock-table.ts @@ -96,6 +96,9 @@ export function getSortValue(r: any, col: ColumnConfig): any { case 'momentum_20d': return r.momentum_20d case 'momentum_30d': return r.momentum_30d case 'momentum_60d': return r.momentum_60d + case 'deviate_3d': return r.deviate_3d + case 'deviate_10d': return r.deviate_10d + case 'deviate_30d': return r.deviate_30d case 'limit_ups': return r.consecutive_limit_ups ?? 0 case 'limit_downs': return r.consecutive_limit_downs ?? 0 case 'score': return r.score diff --git a/frontend/src/lib/storage.ts b/frontend/src/lib/storage.ts index 85d9d77..191e92d 100644 --- a/frontend/src/lib/storage.ts +++ b/frontend/src/lib/storage.ts @@ -63,6 +63,12 @@ export const storage = { /** 自选分组统计条配置 (metric: 统计指标, sort: 排序方式, card*: 分组卡片显示项) */ watchlistGroupStats: kv<{ metric: string; sort: string; cardTopN?: number; cardColorBar?: boolean; cardRank?: boolean }>('watchlist_groupStats'), + /** 异动监控: 主开关 (默认关, 开启后才轮询计算; 告警走监控中心规则) */ + abnormalEnabled: kv('abnormal_enabled'), + + /** 异动监控: 上次计算结果 (关闭开关后仍展示, 含 asof 计算时间戳) */ + abnormalLastResult: kv('abnormal_last_result'), + /** Screener 卡片尺寸 */ screenerCardSize: kv('screener-card-size'), diff --git a/frontend/src/pages/AbnormalMoves.tsx b/frontend/src/pages/AbnormalMoves.tsx new file mode 100644 index 0000000..6a97e64 --- /dev/null +++ b/frontend/src/pages/AbnormalMoves.tsx @@ -0,0 +1,552 @@ +import { useEffect, useMemo, useState } from 'react' +import { Link } from 'react-router-dom' +import { useQuery } from '@tanstack/react-query' +import { FlaskConical, HelpCircle, History, Power, RefreshCw, Search, Settings2 } from 'lucide-react' +import { api, type AbnormalOverview, type AbnormalRow, type AbnormalStatus } from '@/lib/api' +import { QK } from '@/lib/queryKeys' +import { storage } from '@/lib/storage' +import { fmtPrice, fmtPct, priceColorClass } from '@/lib/format' +import { boardTag } from '@/components/stock-table/primitives' +import { PageHeader } from '@/components/PageHeader' +import { StockPreviewDialog } from '@/components/StockPreviewDialog' + +/** + * 异动监控 — 按交易所异动规则口径 (3日±20%/±30%/±40%, 10日+100%, 30日+200%) + * 实时计算个股「偏离值/阈值」接近度, 找出处于异动边缘的标的。 + * + * 计算量可控: 主开关默认关闭, 开启后才发起轮询 (每 60s 一次); 关闭后不再计算, + * 但保留展示上次计算结果 (含计算时间, 取自 localStorage)。 + * 规则口径通过标题栏「?」展开查看。告警走系统监控体系: 在「监控中心」创建 + * 异动监控规则后由后端持续评估, 统一触发记录/站内通知/飞书·企微推送。 + */ + +const WINDOW_KEYS = ['3d', '10d', '30d'] as const +type WindowKey = (typeof WINDOW_KEYS)[number] + +const WINDOW_LABELS: Record = { + '3d': '3日偏离', + '10d': '10日偏离', + '30d': '30日偏离', +} + +const STATUS_META: Record = { + triggered: { label: '已触发', cls: 'bg-danger/15 text-danger', bar: 'bg-danger' }, + edge: { label: '异动边缘', cls: 'bg-warning/15 text-warning', bar: 'bg-warning' }, + watch: { label: '观察', cls: 'bg-elevated text-secondary', bar: 'bg-muted' }, +} + +const BOARDS = ['主板', '创业板', '科创板', '北交所'] as const + +const REFRESH_MS = 60_000 + +export function AbnormalMoves() { + // 主开关: 默认关闭, 开启后才轮询计算 (仅控制本页计算, 后台告警由监控规则驱动) + const [enabled, setEnabled] = useState(() => storage.abnormalEnabled.get(false)) + // 规则口径面板 (标题栏「?」) + const [rulesOpen, setRulesOpen] = useState(false) + // 上次计算结果: 开启时每次成功计算都落本地, 关闭后仍展示 + const [lastResult, setLastResult] = useState( + () => (storage.abnormalLastResult.get(null) as AbnormalOverview | null) ?? null, + ) + const [windowFilter, setWindowFilter] = useState<'all' | WindowKey>('all') + const [direction, setDirection] = useState<'both' | 'up' | 'down'>('both') + const [boardFilter, setBoardFilter] = useState<'all' | (typeof BOARDS)[number]>('all') + const [minCloseness, setMinCloseness] = useState(0.5) + const [query, setQuery] = useState('') + const [watchlistOnly, setWatchlistOnly] = useState(false) + const [preview, setPreview] = useState<{ symbol: string; name: string } | null>(null) + + const overview = useQuery({ + queryKey: QK.abnormalOverview(minCloseness, 300), + queryFn: () => api.abnormalOverview(minCloseness, 300), + enabled, // 关闭时零计算 + refetchInterval: enabled ? REFRESH_MS : false, + }) + // 自选过滤在关闭 (查看上次结果) 时也可用: 自选列表是轻量接口, 不涉及全市场计算 + const watchlist = useQuery({ + queryKey: QK.watchlist, + queryFn: api.watchlistList, + enabled: watchlistOnly, + }) + + const toggleEnabled = (v: boolean) => { + setEnabled(v) + storage.abnormalEnabled.set(v) + if (v) { + overview.refetch() + } + } + + const data = overview.data + useEffect(() => { + if (!data) return + setLastResult(data) + storage.abnormalLastResult.set(data) + }, [data]) + + // 展示数据源: 开启 → 实时结果; 关闭 → 上次计算结果 (可能为空) + const view = enabled ? data : lastResult + const stale = !enabled && lastResult != null + + const watchSymbols = useMemo(() => { + const set = new Set((watchlist.data?.symbols ?? []).map(e => e.symbol)) + return set + }, [watchlist.data]) + + const rows = useMemo(() => { + let list = view?.rows ?? [] + if (windowFilter !== 'all') { + list = list.filter(r => { + const w = r.windows[windowFilter] + return w != null && w.closeness >= minCloseness + }) + } + if (direction !== 'both') { + list = list.filter(r => { + const w = windowFilter !== 'all' ? r.windows[windowFilter] : dominantWindow(r) + const v = w?.value ?? 0 + return direction === 'up' ? v > 0 : v < 0 + }) + } + if (boardFilter !== 'all') list = list.filter(r => r.board === boardFilter) + if (watchlistOnly) list = list.filter(r => watchSymbols.has(r.symbol)) + const q = query.trim().toLowerCase() + if (q) { + list = list.filter(r => `${r.symbol} ${r.name ?? ''}`.toLowerCase().includes(q)) + } + return list + }, [view, windowFilter, direction, boardFilter, watchlistOnly, watchSymbols, query, minCloseness]) + + const counts = view?.counts + const updating = overview.isFetching + + return ( + // 整页占满视口: 头部/筛选固定, 只有表格列表区滚动 +
+
+ + + {enabled && ( + + )} + + + 告警规则 + + {/* 主开关: 开启后才开始轮询计算 */} + +
+ } + /> +
+
+ + {/* 规则口径面板 (标题栏「?」展开) */} + {rulesOpen && ( +
+
+ {ruleChips()} +
+

+ 口径说明: 偏离值 = 个股 N 日累计涨跌幅 − 对应指数同期涨跌幅 (沪: 上证A指/上证指数, + 深: 深证A指/深证成指, 北: 北证50)。阈值为交易所异常波动披露标准的近似值, 仅供风险提示, + 不构成监管认定。每只股票在 3日/10日/30日 三档各算一个接近度 (|偏离值| ÷ 该档阈值, + 阈值随板块与 ST 身份不同), 表格「接近度」列与状态取三档中的最高值, + 来源窗口的偏离值颜色加重显示、其余窗口淡化; ≥100% 已触发、≥70% 边缘、≥50% 观察。 + 偏离列亦可在自选/选股的「异动」列组中启用, 并可作为监控规则与自定义信号的阈值字段。 +

+
+ )} + + {/* 未开启且无历史结果: 说明 + 开启入口 (有上次结果时直接展示数据, 见下方 stale 横幅) */} + {!enabled && !stale ? ( +
+
+ +
监控未开启
+

+ 开启后按交易所异动规则实时计算全市场个股的涨跌幅偏离值 (个股 N 日累计涨跌 − + 对应指数同期), 找出接近触发「异常波动 / 严重异常波动」的标的。 + 计算量较大, 默认关闭; 每次计算的结果会保留, 关闭后仍可查看 (不再实时更新)。 +

+

+ 需要告警推送时, 在监控中心 + 新建「异动监控」规则 —— 后台持续评估, 触发时统一走触发记录 / 站内通知 / 飞书·企微推送, + 与本页开关互不影响。 +

+ +
+
+ ) : ( + <> + {/* 关闭后展示上次计算结果 */} + {stale && ( +
+ + 已暂停计算 · 展示上次结果 + + 上次计算 {fmtCalcTime(lastResult.asof)} · 数据截至 {lastResult.cache_date ?? '—'} + {lastResult.includes_today ? ' (含今日收盘)' : ''} + + +
+ )} + + {/* 统计 + 筛选 */} +
+ + + + {enabled && ( + + 数据截至 {data?.cache_date ?? '—'} + {data?.includes_today ? ' (含今日收盘)' : ' · 已叠加今日实时涨跌'} + {data ? ` · 基准指数今日 ${(data.bench_rt_pct * 100).toFixed(2)}%` : ''} + + )} +
+ +
+ setWindowFilter(v)} + options={[ + { value: 'all', label: '全部窗口' }, + ...WINDOW_KEYS.map(w => ({ value: w, label: WINDOW_LABELS[w] })), + ]} + /> + setDirection(v)} + options={[ + { value: 'both', label: '双向' }, + { value: 'up', label: '正向' }, + { value: 'down', label: '负向' }, + ]} + /> + setBoardFilter(v)} + options={[ + { value: 'all' as const, label: '全板块' }, + ...BOARDS.map(b => ({ value: b, label: b })), + ]} + /> + + +
+ + setQuery(e.target.value)} + placeholder="搜索代码/名称" + className="h-7 w-40 rounded border border-border bg-base pl-7 pr-2 text-[11px] text-foreground" + /> +
+
+ + {/* 主表: 剩余空间内滚动 (页面本身不滚动) */} +
+ + + + + + + + {WINDOW_KEYS.map(w => ( + + ))} + + + + + + {overview.isLoading ? ( + + + + ) : rows.length === 0 ? ( + + + + ) : ( + rows.map((r, i) => ( + setPreview({ symbol: r.symbol, name: r.name ?? r.symbol })} + /> + )) + )} + +
#代码 / 名称现价今日 + {WINDOW_LABELS[w]} + (阈值) + + 接近度 + (最高档) + 状态
+ 正在计算全市场偏离值… +
+ {view ? '当前没有满足条件的标的' : '暂无数据'} +
+
+ + )} + +
+ {preview && ( + setPreview(null)} + /> + )} +
+ ) + + function ruleChips() { + return (view?.rules ?? FALLBACK_RULES).map((rule, i) => { + const thr = WINDOW_KEYS.map(w => `${w.replace('d', '日')}±${fmtThreshold(rule.thresholds[w])}`).join(' / ') + return ( +
+
+ {rule.board} + {rule.st && ST} +
+
{thr}
+
+ ) + }) + } +} + +/** 上次计算时间 (服务端 asof 秒级时间戳 → 本地日期时间) */ +function fmtCalcTime(asofSec: number): string { + const d = new Date(asofSec * 1000) + const pad = (n: number) => String(n).padStart(2, '0') + return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())} ${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}` +} + +function fmtThreshold(v: number | undefined): string { + if (v == null) return '—' + return `${(v * 100).toFixed(0)}%` +} + +/** 全窗口里接近度最高的窗口 */ +function dominantWindow(r: AbnormalRow): { key: WindowKey; value: number; threshold: number; closeness: number } | undefined { + let best: { key: WindowKey; value: number; threshold: number; closeness: number } | undefined + for (const w of WINDOW_KEYS) { + const info = r.windows[w] + if (info && (!best || info.closeness > best.closeness)) best = { key: w, ...info } + } + return best +} + +function AbnormalRowView({ row, rank, onPreview }: { + row: AbnormalRow + rank: number + onPreview: () => void +}) { + const board = boardTag(row.symbol) + const dominant = dominantWindow(row) + const meta = STATUS_META[row.status] + return ( + + {rank} + + {/* 仅代码/名称可点击打开详情 (与自选列表一致), 其余单元格不可点 */} + + + {fmtPrice(row.close)} + + {fmtPct(row.rt_pct)} + + {WINDOW_KEYS.map(w => { + const info = row.windows[w] + // 接近度取最高档: 来源窗口颜色加重 (加粗), 其余窗口淡化, 以此区分「哪一档」 + const isDominant = dominant?.key === w + return ( + + {info ? ( + + {fmtPct(info.value)} + /{fmtThreshold(info.threshold)} + + ) : ( + — + )} + + ) + })} + +
+
+
+
+ + {((dominant?.closeness ?? 0) * 100).toFixed(0)}% + +
+ + + {meta.label} + + + ) +} + +function StatusChip({ label, count, tone }: { label: string; count?: number; tone: 'danger' | 'warning' | 'muted' }) { + const toneCls = + tone === 'danger' + ? 'border-danger/30 bg-danger/8 text-danger' + : tone === 'warning' + ? 'border-warning/30 bg-warning/8 text-warning' + : 'border-border bg-elevated text-secondary' + return ( + + {count ?? '—'} + {label} + + ) +} + +function SegmentedControl({ value, onChange, options }: { + value: T + onChange: (v: T) => void + options: Array<{ value: T; label: string }> +}) { + return ( +
+ {options.map(o => ( + + ))} +
+ ) +} + +/** 后端数据未到时的规则表兜底 (与后端 RULES_META 同步维护) */ +const FALLBACK_RULES: Array<{ board: string; st: boolean; thresholds: Record; note: string }> = [ + { board: '主板', st: false, thresholds: { '3d': 0.2, '10d': 1.0, '30d': 2.0 }, note: '' }, + { board: '主板', st: true, thresholds: { '3d': 0.15, '10d': 0.5, '30d': 1.0 }, note: '' }, + { board: '创业板/科创板', st: false, thresholds: { '3d': 0.3, '10d': 1.0, '30d': 2.0 }, note: '' }, + { board: '北交所', st: false, thresholds: { '3d': 0.4, '10d': 1.0, '30d': 2.0 }, note: '' }, +] diff --git a/frontend/src/pages/Monitor.tsx b/frontend/src/pages/Monitor.tsx index 69a4cf2..270cfe5 100644 --- a/frontend/src/pages/Monitor.tsx +++ b/frontend/src/pages/Monitor.tsx @@ -1,5 +1,5 @@ import { useState, useRef, useEffect, useMemo } from 'react' -import { useNavigate } from 'react-router-dom' +import { useNavigate, useSearchParams } from 'react-router-dom' import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query' import { motion, AnimatePresence } from 'framer-motion' import { AlertTriangle, RadioTower, Plus, Trash2, Settings2, Zap, Bell, ListChecks, BellRing, TrendingUp, TrendingDown, Flame, Tags } from 'lucide-react' @@ -22,6 +22,7 @@ import { usePreferences } from '@/lib/useSharedQueries' const TYPE_LABEL: Record = { signal: '信号', price: '价格/涨跌', market: '市场异动', strategy: '策略监控', sector: '板块监控', + abnormal: '异动监控', } /** 严重级别 → 左侧色条 + 图标 */ @@ -36,6 +37,7 @@ const SOURCE_BADGE_STYLE: Record = { price: 'bg-emerald-400/10 text-emerald-400 border-emerald-400/20', market: 'bg-purple-500/10 text-purple-400 border-purple-500/20', sector: 'bg-cyan-500/10 text-cyan-700 border-cyan-500/20 dark:text-cyan-300', + abnormal: 'bg-orange-500/10 text-orange-500 border-orange-500/20 dark:text-orange-400', } /** @@ -114,9 +116,22 @@ export function Monitor() { const qc = useQueryClient() const [editorOpen, setEditorOpen] = useState(false) const [editingRule, setEditingRule] = useState(null) + const [editorPreset, setEditorPreset] = useState | null>(null) + + // 深链: /monitor?new=abnormal (异动监控页「告警规则」入口) → 直接弹出预置类型的编辑器 + const [searchParams, setSearchParams] = useSearchParams() + useEffect(() => { + const kind = searchParams.get('new') + if (kind === 'abnormal') { + setEditingRule(null) + setEditorPreset({ type: 'abnormal', threshold_pct: 70, direction: 'both', abnormal_window: 'any', scope: 'all' }) + setEditorOpen(true) + setSearchParams({}, { replace: true }) + } + }, [searchParams, setSearchParams]) // 触发记录: 过滤 + 统计 (提升到主组件, 供 header 行使用) - const [filter, setFilter] = useState<'all' | 'strategy' | 'signal' | 'price' | 'market' | 'sector'>('all') + const [filter, setFilter] = useState<'all' | 'strategy' | 'signal' | 'price' | 'market' | 'sector' | 'abnormal'>('all') const [confirmClear, setConfirmClear] = useState(false) const [confirmClearRules, setConfirmClearRules] = useState(false) @@ -177,7 +192,7 @@ export function Monitor() { {/* 过滤标签 */}
- {(['all', 'strategy', 'signal', 'price', 'market', 'sector'] as const).map(f => ( + {(['all', 'strategy', 'signal', 'price', 'market', 'sector', 'abnormal'] as const).map(f => (
+ ) : r.type === 'abnormal' ? ( +
+ + 接近度 ≥ {r.threshold_pct ?? 70}% + + + {r.abnormal_window && r.abnormal_window !== 'any' ? `${r.abnormal_window.toUpperCase()} 窗口` : '全部窗口'} + + + {r.direction === 'up' ? '涨势偏离' : r.direction === 'down' ? '跌势偏离' : '涨跌双向'} + +
) : r.type === 'strategy' && r.strategy_id ? (
{(r.score_min != null || r.score_max != null) && ( @@ -827,7 +855,12 @@ function RulesList({ rulesQuery, onEdit }: { } // ── 规则编辑对话框 ──────────────────────────────────── -function RuleEditorDialog({ open, rule, onClose }: { open: boolean; rule: MonitorRule | null; onClose: () => void }) { +function RuleEditorDialog({ open, rule, preset, onClose }: { + open: boolean + rule: MonitorRule | null + preset?: Partial | null + onClose: () => void +}) { const backdrop = useDialogBackdrop(onClose) return ( @@ -849,6 +882,7 @@ function RuleEditorDialog({ open, rule, onClose }: { open: boolean; rule: Monito > diff --git a/frontend/src/pages/settings/MenuSettings.tsx b/frontend/src/pages/settings/MenuSettings.tsx index 2127d18..4f40668 100644 --- a/frontend/src/pages/settings/MenuSettings.tsx +++ b/frontend/src/pages/settings/MenuSettings.tsx @@ -41,6 +41,7 @@ const BUILTIN_PAGES: NavEntry[] = [ { id: '/industry-analysis', label: '行业分析', type: 'builtin', visible: true }, { id: '/stock-analysis', label: '个股分析', type: 'builtin', visible: true }, { id: '/regime', label: '市场环境', type: 'builtin', visible: true }, + { id: '/abnormal', label: '异动监控', type: 'builtin', visible: true }, { id: '/review', label: '复盘', type: 'builtin', visible: true }, { id: '/financials', label: '财务分析', type: 'builtin', visible: true }, { id: '/indices', label: '指数', type: 'builtin', visible: true }, diff --git a/frontend/src/router.tsx b/frontend/src/router.tsx index 8fb0297..36456f0 100644 --- a/frontend/src/router.tsx +++ b/frontend/src/router.tsx @@ -33,6 +33,7 @@ const Branding = lazy(() => import('./pages/Branding').then(m => ({ default: m.B const Settings = lazy(() => import('./pages/Settings').then(m => ({ default: m.Settings }))) const Indices = lazy(() => import('./pages/Indices').then(m => ({ default: m.Indices }))) const Regime = lazy(() => import('./pages/Regime').then(m => ({ default: m.Regime }))) +const AbnormalMoves = lazy(() => import('./pages/AbnormalMoves').then(m => ({ default: m.AbnormalMoves }))) const Dev = lazy(() => import('./pages/Dev').then(m => ({ default: m.Dev }))) const CORE_ROUTE_PATHS = new Set([ @@ -56,6 +57,7 @@ const CORE_ROUTE_PATHS = new Set([ '/limit-ladder', '/indices', '/regime', + '/abnormal', '/branding', '/settings', '/dev', @@ -128,6 +130,7 @@ export const router = createBrowserRouter([ { path: 'limit-ladder', element: }, { path: 'indices', element: }, { path: 'regime', element: }, + { path: 'abnormal', element: }, { path: 'branding', element: }, { path: 'settings', element: }, // 隐藏路由：开发者工具（不暴露在菜单，仅供调试）