feat: 资金流突变处理——fund_flow_alert(正面突变→选股分析候选池,负面突变持仓→重评+XMPP报告,冷却1小时同_can_push); capital_flow_collector采集完成后调用

deploy/profile-scripts/capital_flow_collector.py

@@ -202,5 +202,12 @@ def main():
     conn.close()
     print(f"[capital_flow] {len(all_flows)}/{len(code_list)}只更新完成")
 
+    # 2026-08-13 资金流突变处理（老莫设计）：正面突变→选股分析，负面突变(持仓)→重评+XMPP报告
+    try:
+        from fund_flow_alert import process_flow_alerts
+        process_flow_alerts(all_flows)
+    except Exception as e:
+        print(f"[资金流突变处理异常] {e}", flush=True)
+
 if __name__ == "__main__":
     main()

deploy/profile-scripts/fund_flow_alert.py

@@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""fund_flow_alert.py — 资金流突变处理（2026-08-13 老莫设计）
+
+核心逻辑（老莫定）：
+- 正面突变（主力连续流入/超大单转入）→ 触发选股分析（入自选流程）
+- 负面突变（超大单转出/单日暴量，持仓股）→ 触发重评 + XMPP 发评估报告（不论结果）
+- 冷却：同 _can_push（同股同类型 1 小时）
+
+在 capital_flow_collector 采集完成后调用。
+"""
+import sys
+import json
+import sqlite3
+import subprocess
+from pathlib import Path
+from datetime import datetime
+
+_SCRIPT_DIR = Path(__file__).resolve().parent
+sys.path.insert(0, str(_SCRIPT_DIR))
+sys.path.insert(0, "/home/hmo/MoFin")
+
+DB = "/home/hmo/MoFin/data/mofin.db"
+XMPP_SEND = "http://127.0.0.1:5805/"
+COOLDOWN_FILE = "/home/hmo/.hermes/.fund_flow_alert_cooldown.json"
+COOLDOWN_SEC = 3600  # 1 小时（同 _can_push）
+
+
+def _load_cooldown():
+    try:
+        if Path(COOLDOWN_FILE).exists():
+            return json.loads(Path(COOLDOWN_FILE).read_text())
+    except Exception:
+        pass
+    return {}
+
+
+def _save_cooldown(cd):
+    try:
+        Path(COOLDOWN_FILE).write_text(json.dumps(cd))
+    except Exception:
+        pass
+
+
+def _can_alert(code, alert_type):
+    """冷却：同股同类型 1 小时内不重复"""
+    cd = _load_cooldown()
+    key = f"{code}_{alert_type}"
+    last = cd.get(key, 0)
+    now = datetime.now().timestamp()
+    if now - last < COOLDOWN_SEC:
+        return False
+    cd[key] = now
+    _save_cooldown(cd)
+    return True
+
+
+def _is_holding(code):
+    """是否持仓股"""
+    conn = sqlite3.connect(DB, timeout=10)
+    r = conn.execute("SELECT shares FROM holdings WHERE code=? AND is_active=1 AND shares>0", (code,)).fetchone()
+    conn.close()
+    return bool(r and r[0] > 0)
+
+
+def _is_watchlist(code):
+    """是否自选股"""
+    conn = sqlite3.connect(DB, timeout=10)
+    r = conn.execute("SELECT code FROM holding_strategies WHERE code=? AND status='active' AND decision_type='自选策略'", (code,)).fetchone()
+    conn.close()
+    return bool(r)
+
+
+def _send_xmpp(text):
+    """XMPP 推送（知微 bridge）"""
+    try:
+        payload = json.dumps({"to": "hmo@yoin.fun", "body": text, "type": "chat"}).encode("utf-8")
+        req = urllib.request.Request(XMPP_SEND, data=payload, headers={"Content-Type": "application/json"})
+        urllib.request.urlopen(req, timeout=5).read()
+    except Exception as e:
+        print(f"[XMPP失败] {e}", file=sys.stderr)
+
+
+def process_flow_alerts(all_flows):
+    """处理资金流突变。all_flows: {code: {flow, analysis}}"""
+    import urllib.request
+    if not all_flows:
+        return
+
+    positive_alerts = []  # 正面突变（选股候选）
+    negative_alerts = []  # 负面突变（持仓重评）
+
+    for code, data in all_flows.items():
+        if not data:
+            continue
+        analysis = data.get("analysis", {})
+        alerts = analysis.get("alerts", [])
+        if not alerts:
+            continue
+
+        name = data.get("name", code)
+        pattern = analysis.get("pattern", "")
+        trend = analysis.get("trend", "")
+
+        for alert in alerts:
+            # 判断正负面
+            is_positive = any(k in alert for k in ["连续3日净流入", "转为净买入", "由出转入"])
+            is_negative = any(k in alert for k in ["转为净卖出", "由入转出", "异常", "出货嫌疑"])
+
+            if is_positive:
+                # 正面突变 → 选股分析候选
+                if _can_alert(code, "positive"):
+                    positive_alerts.append({
+                        "code": code, "name": name, "alert": alert,
+                        "pattern": pattern, "trend": trend,
+                    })
+            elif is_negative:
+                # 负面突变 → 持仓股才重评
+                if _is_holding(code) and _can_alert(code, "negative"):
+                    negative_alerts.append({
+                        "code": code, "name": name, "alert": alert,
+                        "pattern": pattern, "trend": trend,
+                    })
+
+    # 正面突变：触发选股分析（写候选池）
+    if positive_alerts:
+        print(f"[资金流突变] 正面 {len(positive_alerts)} 只 → 触发选股分析", flush=True)
+        _trigger_candidate_analysis(positive_alerts)
+
+    # 负面突变：持仓股触发重评 + XMPP 报告
+    if negative_alerts:
+        print(f"[资金流突变] 负面 {len(negative_alerts)} 只（持仓）→ 触发重评+报告", flush=True)
+        _trigger_holding_reassess(negative_alerts)
+
+
+def _trigger_candidate_analysis(alerts):
+    """正面突变 → 触发选股分析（写候选池，走 candidate_filter 流程）"""
+    conn = sqlite3.connect(DB, timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
+    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    for a in alerts:
+        try:
+            # 写入 candidates 候选池（资金流突变来源）
+            conn.execute(
+                """INSERT OR IGNORE INTO candidates
+                   (code, name, source, reason, created_at)
+                   VALUES (?,?,?,?,?)""",
+                (a["code"], a["name"], "fund_flow_positive",
+                 f"资金流突变: {a['alert']} | {a['pattern']}", now)
+            )
+        except Exception as e:
+            print(f"  候选写入失败 {a['code']}: {e}", file=sys.stderr)
+    conn.commit()
+    conn.close()
+    # XMPP 通知（正面突变候选）
+    codes_str = ", ".join(f"{a['name']}({a['code']})" for a in alerts[:5])
+    _send_xmpp(f"📈 资金流正面突变 {len(alerts)} 只 → 已入候选池待选股分析：{codes_str}")
+
+
+def _trigger_holding_reassess(alerts):
+    """负面突变（持仓股）→ 触发重评 + XMPP 发评估报告"""
+    for a in alerts:
+        code = a["code"]
+        name = a["name"]
+        # 触发重评
+        try:
+            r = subprocess.run(
+                [sys.executable, str(_SCRIPT_DIR / "per_stock_reassess.py"), code],
+                capture_output=True, text=True, timeout=120, cwd=str(_SCRIPT_DIR),
+            )
+            reassess_out = (r.stdout or "").strip()[-500:] if r.stdout else "无输出"
+        except Exception as e:
+            reassess_out = f"重评异常: {e}"
+        # XMPP 报告（不论结果）
+        report = (
+            f"📉 资金流负面突变（持仓）\n"
+            f"{name}({code})\n"
+            f"突变: {a['alert']}\n"
+            f"形态: {a['pattern']} | 趋势: {a['trend']}\n"
+            f"重评: {reassess_out[:300]}"
+        )
+        _send_xmpp(report)
+        print(f"  {name}({code}) 重评+报告已发", flush=True)
+
+
+if __name__ == "__main__":
+    # 测试：读 capital_flow_cache 找突变
+    conn = sqlite3.connect(DB, timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
+    try:
+        row = conn.execute("SELECT cache_json FROM capital_flow_cache ORDER BY updated_at DESC LIMIT 1").fetchone()
+        if row:
+            cache = json.loads(row[0])
+            all_flows = cache.get("stocks", {})
+            print(f"capital_flow_cache 股票数: {len(all_flows)}")
+            process_flow_alerts(all_flows)
+        else:
+            print("capital_flow_cache 无数据")
+    except Exception as e:
+        print(f"读取 capital_flow_cache 失败: {e}", file=sys.stderr)
+    conn.close()