From 6989a8f8b74d724f9bc61468cfc634b8476bb6d8 Mon Sep 17 00:00:00 2001 From: xxm Date: Wed, 19 Aug 2026 11:09:00 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8C=81=E4=BB=93=E5=BC=82=E5=8A=A8?= =?UTF-8?q?=E7=9B=91=E6=8E=A7=E2=80=94=E2=80=94=E6=A3=80=E6=B5=8B=E6=80=A5?= =?UTF-8?q?=E6=B6=A8=E6=80=A5=E8=B7=8C/=E6=94=BE=E9=87=8F/20=E6=97=A5?= =?UTF-8?q?=E7=AA=81=E7=A0=B4/=E6=B7=B1=E5=A5=97=E5=8F=8D=E5=BC=B9,ACTION?= =?UTF-8?q?=E7=BA=A7=E6=8E=A8=E9=80=81(=E8=80=81=E8=8E=AB:=E4=B8=AD?= =?UTF-8?q?=E7=A7=91=E7=94=B5=E6=B0=94=E5=BC=82=E5=8A=A8=E6=97=A0=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E7=9A=84=E6=95=99=E8=AE=AD)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deploy/profile-scripts/anomaly_monitor.py | 171 ++++++++++++++++++++++ 1 file changed, 171 insertions(+) create mode 100644 deploy/profile-scripts/anomaly_monitor.py diff --git a/deploy/profile-scripts/anomaly_monitor.py b/deploy/profile-scripts/anomaly_monitor.py new file mode 100644 index 00000000..461786c9 --- /dev/null +++ b/deploy/profile-scripts/anomaly_monitor.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""anomaly_monitor.py — 持仓股异动监控（2026-08-19 老莫：中科电气放量拉升无通知的教训） +检测持仓股/自选股的异动（价格急变/量能异常/突破），异动时 ACTION 级推送（不被信息淹没）。 + +异动判定（确定性代码，不依赖 LLM）： +1. 单日涨跌幅 ≥ 阈值（默认 ±5%）→ 急涨/急跌 +2. 量能异常：当日成交量 ≥ 5日均量 × 3 → 放量 +3. 价格突破：创 20 日新高（或跌破 20 日新低） +4. 深套股反弹：浮亏 ≤ -20% 的持仓单日涨幅 ≥ 4%（深套异动，老莫关注） + +推送：ACTION 级（alert_helper 直通），只对持仓股 + 自选股。 + +用法：cron 每 5 分钟跑（交易日盘中） + python3 anomaly_monitor.py +""" +import sqlite3, os, sys +from datetime import datetime +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +DB = "/home/hmo/MoFin/data/mofin.db" +THRESHOLD_CHG = 5.0 # 单日涨跌幅阈值 % +THRESHOLD_VOL = 3.0 # 量能倍数（vs 5日均量） +THRESHOLD_20D = True # 20日新高/新低检测 +THRESHOLD_DEEP_REBOUND = 4.0 # 深套股反弹阈值 % + +STATE_FILE = "/home/hmo/MoFin/data/anomaly_state.json" + +def get_conn(): + conn = sqlite3.connect(DB, timeout=30) + conn.execute("PRAGMA busy_timeout=30000") + return conn + +def load_state(): + try: + import json + with open(STATE_FILE, encoding="utf-8") as f: + return json.load(f) + except Exception: + return {"notified": {}} # {code_date: ts} + +def save_state(st): + import json + try: + # 只保留最近 500 条 + if len(st.get("notified", {})) > 500: + st["notified"] = dict(list(st["notified"].items())[-500:]) + with open(STATE_FILE, "w", encoding="utf-8") as f: + json.dump(st, f) + except Exception: + pass + +def push_anomaly(body): + """ACTION 级推送（直通，不限速）""" + try: + from alert_helper import notify, ACTION + notify("持仓异动", body, level=ACTION) + return True + except Exception as e: + print(f" ⚠️ 异动推送失败: {e}", flush=True) + return False + +def get_holdings(): + """持仓 + 自选股列表""" + conn = get_conn() + # 持仓（holdings 表） + holds = conn.execute("SELECT code, name, shares, cost, price FROM holdings WHERE is_active=1 AND shares>0").fetchall() + # 自选（holding_strategies 自选策略） + wl = conn.execute("SELECT code, name FROM holding_strategies WHERE status='active' AND decision_type='自选策略'").fetchall() + conn.close() + codes = set(h[0] for h in holds) | set(w[0] for w in wl) + names = {h[0]: h[1] or h[0] for h in holds} + names.update({w[0]: w[1] or w[0] for w in wl}) + costs = {h[0]: h[2] for h in holds} + return codes, names, costs + +def check_anomalies(): + conn = get_conn() + codes, names, costs = get_holdings() + st = load_state() + notified = st.get("notified", {}) + now = datetime.now().strftime("%Y-%m-%d %H:%M") + today = now[:10] + anomalies = [] + + for code in sorted(codes): + # 最新日线 + 昨收 + 前5日均量 + row = conn.execute(""" + SELECT date, close, volume FROM stock_daily + WHERE code=? ORDER BY date DESC LIMIT 6 + """, (code,)).fetchall() + if len(row) < 2: + continue + latest = row[0] # (date, close, volume) + prev = row[1] + if not latest[1] or not prev[1] or prev[1] <= 0: + continue + # 当日涨跌幅 + chg = (latest[1] - prev[1]) / prev[1] * 100 + # 5日均量（排除当日） + vols = [r[2] for r in row[1:6] if r[2]] + avg_vol = sum(vols) / len(vols) if vols else 0 + vol_ratio = latest[2] / avg_vol if avg_vol > 0 else 0 + # 20日新高/新低 + high20 = conn.execute(""" + SELECT MAX(high), MIN(low) FROM ( + SELECT high, low FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 20 + ) + """, (code,)).fetchone() + is_new_high = bool(high20 and high20[0] and latest[1] >= high20[0]) + is_new_low = bool(high20 and high20[1] and latest[1] <= high20[1]) + # 深套反弹（成本 vs 现价） + cost = costs.get(code) or 0 + pnl = (latest[1] - cost) / cost * 100 if cost > 0 else 0 + is_deep = pnl <= -20 + + # 判定异动 + reasons = [] + if abs(chg) >= THRESHOLD_CHG: + reasons.append(f"{'急涨' if chg > 0 else '急跌'}{abs(chg):.1f}%") + if vol_ratio >= THRESHOLD_VOL: + reasons.append(f"放量{vol_ratio:.1f}倍") + if THRESHOLD_20D and is_new_high: + reasons.append("创20日新高") + if THRESHOLD_20D and is_new_low: + reasons.append("破20日新低") + if is_deep and chg >= THRESHOLD_DEEP_REBOUND: + reasons.append(f"深套反弹{chg:.1f}%") + + if reasons and latest[0] == today: + # 去重：同 code 同原因类型当天只推一次 + key = f"{code}_{'|'.join(sorted(reasons))}_{today}" + if key not in notified or (datetime.now().timestamp() - notified[key] > 3600): + name = names.get(code, code) + cost_s = f" 成本{cost:.2f} 浮盈{pnl:+.1f}%" if cost > 0 else "" + body = (f"⚡ {name}({code}) 异动\n" + f"现价 {latest[1]:.2f} ({chg:+.1f}%) | 量能{vol_ratio:.1f}倍\n" + f"触发: {'、'.join(reasons)}{cost_s}") + anomalies.append(body) + notified[key] = datetime.now().timestamp() + print(f" ⚡ {code}: {'、'.join(reasons)}", flush=True) + + save_state({"notified": notified}) + conn.close() + + if anomalies: + full = f"🔔 持仓/自选异动 {len(anomalies)} 条\n" + "\n\n".join(anomalies) + push_anomaly(full) + print(f"[异动] 推送 {len(anomalies)} 条", flush=True) + else: + print(f"[异动] {now} 无异动", flush=True) + return len(anomalies) + +if __name__ == "__main__": + # 单例守卫（防 cron 重复） + import fcntl + lock_f = open("/tmp/anomaly_monitor.lock", "w") + try: + fcntl.flock(lock_f, fcntl.LOCK_EX | fcntl.LOCK_NB) + except Exception: + print("已有实例在运行，跳过", flush=True) + sys.exit(0) + try: + raise SystemExit(check_anomalies()) + finally: + try: + fcntl.flock(lock_f, fcntl.LOCK_UN) + except Exception: + pass