From b227156d6ca8e78243514da4c98cde280a78998e Mon Sep 17 00:00:00 2001 From: hmo Date: Fri, 14 Aug 2026 02:24:56 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=B5=84=E9=87=91=E6=B5=81=E7=AA=81?= =?UTF-8?q?=E5=8F=98=E5=A4=84=E7=90=86=E2=80=94=E2=80=94fund=5Fflow=5Faler?= =?UTF-8?q?t(=E6=AD=A3=E9=9D=A2=E7=AA=81=E5=8F=98=E2=86=92=E9=80=89?= =?UTF-8?q?=E8=82=A1=E5=88=86=E6=9E=90=E5=80=99=E9=80=89=E6=B1=A0,?= =?UTF-8?q?=E8=B4=9F=E9=9D=A2=E7=AA=81=E5=8F=98=E6=8C=81=E4=BB=93=E2=86=92?= =?UTF-8?q?=E9=87=8D=E8=AF=84+XMPP=E6=8A=A5=E5=91=8A,=E5=86=B7=E5=8D=B41?= =?UTF-8?q?=E5=B0=8F=E6=97=B6=E5=90=8C=5Fcan=5Fpush);=20capital=5Fflow=5Fc?= =?UTF-8?q?ollector=E9=87=87=E9=9B=86=E5=AE=8C=E6=88=90=E5=90=8E=E8=B0=83?= =?UTF-8?q?=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../profile-scripts/capital_flow_collector.py | 7 + deploy/profile-scripts/fund_flow_alert.py | 201 ++++++++++++++++++ 2 files changed, 208 insertions(+) create mode 100644 deploy/profile-scripts/fund_flow_alert.py diff --git a/deploy/profile-scripts/capital_flow_collector.py b/deploy/profile-scripts/capital_flow_collector.py index 679ccafb..75914edf 100644 --- a/deploy/profile-scripts/capital_flow_collector.py +++ b/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() diff --git a/deploy/profile-scripts/fund_flow_alert.py b/deploy/profile-scripts/fund_flow_alert.py new file mode 100644 index 00000000..1b77b96f --- /dev/null +++ b/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()