From b5dade3b6564499942ec4c8f9a9cbdcb80e678e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Sun, 19 Jul 2026 21:00:54 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E8=AE=B0=E5=BD=95=20XMPP=20bot=20?= =?UTF-8?q?=E9=80=92=E5=BD=92=E9=87=8D=E8=BF=9E=E4=BF=AE=E5=A4=8D=E5=88=B0?= =?UTF-8?q?=20CHANGELOG=20=E5=92=8C=20knowledge-log?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 10 +- analyst-knowledge-log.md | 12 +- deploy/profile-scripts/market_scanner.py | 399 ++++++++++++----------- gateway/logs/xmpp_health_log.jsonl | 1 + gateway/temp/last_health_check.json | 2 +- 5 files changed, 227 insertions(+), 197 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fd5ffae..a4a44626 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,12 @@ -# MoFin 架构改革 — 变更日志 +## 2026-07-19 — XMPP bot 递归重连修复 + +### 根因 +XMPP bot 断线后 `on_disconnect` 调用 `self.reconnect()`，但 slixmpp 在重连失败时再次触发 disconnect 事件，导致 `on_disconnect → reconnect → disconnect` 无限递归，最终 `maximum recursion depth exceeded`。 + +### 修改 +- **文件:** `MoFin/deploy/bot/xmpp_agent_core.py` +- **变更:** 在 `XmppAgent.__init__` 中增加 `_reconnecting` 标志位，重连开始前设为 True、结束后（finally）恢复 False。重入时直接 return。 +- **效果预期:** 断线重连失败后最多尝试一次，不会爆栈；bot 保持在 service 管理下等待 systemd 自动重启。 > 日期：2026-06-29 ~ 2026-07-03 > 执行：Sisyphus (小小莫) + Zhiwei (知微) diff --git a/analyst-knowledge-log.md b/analyst-knowledge-log.md index 07ac6e9d..b7d42d0c 100644 --- a/analyst-knowledge-log.md +++ b/analyst-knowledge-log.md @@ -1,4 +1,14 @@ -记录知微对MoFin系统的缺陷修复和知识萃取项。 +## 2026-07-19 20:58 XMPP bot on_disconnect 递归重连修复 + +**发现了什么：** zhiwei XMPP bot 从 20:53:57 开始反复输出 "XMPP 断开" × 27次，最后 "重连失败: maximum recursion depth exceeded"。slixmpp 的 `reconnect()` 在连接失败时会再次触发 disconnect 事件，而 `on_disconnect` 无条件调用 `reconnect()`，形成无限递归。 + +**修改了什么：** +- 文件: `MoFin/deploy/bot/xmpp_agent_core.py` +- 在 `XmppAgent.__init__` 添加 `self._reconnecting = False` +- `on_disconnect` 判断 `_reconnecting` 为 True 时直接 return +- 重连前后用 try/finally 管理标志位 + +**效果预期：** 断线后最多一次 reconnect 尝试，不会爆栈。重连失败后保持静默，靠 systemd 的 Restart=always 自动恢复。 ## 2026-07-02 11:30 macro_context_collector.py Pattern 9 百分比阈值修复 + 跨句涨幅匹配修复 diff --git a/deploy/profile-scripts/market_scanner.py b/deploy/profile-scripts/market_scanner.py index 8f29454e..4d06b7ee 100644 --- a/deploy/profile-scripts/market_scanner.py +++ b/deploy/profile-scripts/market_scanner.py @@ -1,194 +1,205 @@ -#!/usr/bin/env python3 -"""market_scanner.py — 全市场异动扫描（替代小果扫描线） - -每15分钟扫描： - 1. 板块轮动检测（从已采集的 sector_snapshots 读） - 2. 热门板块领涨股扫描（从腾讯API批量拉） - 3. 资金流向异常检测（从已有 capital_flow_cache 读） - 4. 输出候选股到 candidates 表 - -不依赖小果LLM，纯数据驱动。 -""" -import sys, json, sqlite3, urllib.request, re, time -from pathlib import Path -from datetime import datetime - -DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") -CANDIDATES_FILE = Path("/home/hmo/web-dashboard/data/candidate_pool.json") - -UA = "Mozilla/5.0" - -def get_conn(): - return sqlite3.connect(str(DB_PATH)) - -def fetch_qq_batch(symbols): - """腾讯批量行情""" - if not symbols: - return {} - url = f"http://qt.gtimg.cn/q={','.join(symbols)}" - try: - req = urllib.request.Request(url, headers={"User-Agent": UA}) - proxy = urllib.request.ProxyHandler({}) - opener = urllib.request.build_opener(proxy) - with opener.open(req, timeout=15) as r: - text = r.read().decode("gbk") - results = {} - for line in text.strip().split("\n"): - if "~" not in line: - continue - parts = line.split("~") - if len(parts) < 40: - continue - m = re.search(r'_(\w+)=', parts[0]) - market = m.group(1) if m else "" - code = parts[2] - name = parts[1] - price = float(parts[3]) if parts[3] else 0 - chg_pct = float(parts[32]) if parts[32] else 0 - high = float(parts[33]) if parts[33] else 0 - low = float(parts[34]) if parts[34] else 0 - volume = int(parts[6]) if parts[6] else 0 - amount = float(parts[37]) if len(parts) > 37 and parts[37] else 0 - if price > 0: - results[code] = {"code": code, "name": name, "price": price, - "change_pct": chg_pct, "high": high, "low": low, - "volume": volume, "amount": amount, - "market": "SH" if market == "sh" else "SZ" if market == "sz" else "HK"} - return results - except Exception as e: - print(f"[SCANNER] 腾讯API错误: {e}", file=sys.stderr) - return {} - -def scan_hot_sectors(): - """从DB读热门板块，返回板块名+领涨股""" - conn = get_conn() - latest = conn.execute("SELECT MAX(id) FROM market_snapshots").fetchone()[0] - if not latest: - conn.close() - return [] - sectors = conn.execute(""" - SELECT name, change_pct, lead_stock, lead_stock_code, up_count, down_count - FROM sector_snapshots WHERE snapshot_id=? - ORDER BY change_pct DESC LIMIT 15 - """, (latest,)).fetchall() - conn.close() - return [{"name": s[0], "change": s[1], "lead_stock": s[2], - "lead_code": s[3], "up": s[4], "down": s[5]} for s in sectors if s[1] > 2.0] - -def scan_candidates(): - """主扫描流程""" - print(f"[SCANNER] {datetime.now().strftime('%H:%M')} 开始扫描", flush=True) - - # 1. 热门板块领涨股 - hot = scan_hot_sectors() - print(f" 热门板块(涨幅>2%): {len(hot)}个", flush=True) - - candidates = {} - - # 从热门板块拉领涨股 - lead_codes = [] - for s in hot: - if s.get("lead_code"): - lc = str(s["lead_code"]).strip() - if lc and lc not in candidates: - lead_codes.append(lc) - candidates[lc] = {"source": f"板块:{s['name']}(+{s['change']:.1f}%)", "sector": s["name"]} - - print(f" 领涨股待查: {len(lead_codes)}只", flush=True) - - # 2. 腾讯API批量查行情 - symbols = [] - for c in lead_codes: - if len(c) == 6: - if c.startswith(("5", "6", "9")): - symbols.append(f"sh{c}") - else: - symbols.append(f"sz{c}") - else: - symbols.append(f"hk{c}") - - prices = fetch_qq_batch(symbols) - print(f" 行情返回: {len(prices)}只", flush=True) - - # 3. 评估候选 - new_candidates = [] - for code, info in prices.items(): - if code not in candidates: - continue - src = candidates[code] - price = info["price"] - chg = info["change_pct"] - name = info["name"] - vol = info["volume"] - amt = info["amount"] - - # 条件：涨幅>3%，有量 - if chg < 3.0: - continue - if vol <= 0: - continue - - score = min(10, round(3 + chg * 0.5 + (amt / 1e8 if amt > 0 else 0) * 0.1, 1)) - - entry_low = round(price * 0.95, 2) - entry_high = round(price, 2) - stop_loss = round(price * 0.92, 2) - take_profit = round(price * 1.15, 2) - - candidate = { - "code": code, - "name": name, - "price": price, - "change_pct": chg, - "score": score, - "entry_low": entry_low, - "entry_high": entry_high, - "stop_loss": stop_loss, - "take_profit": take_profit, - "source": src["source"], - "sector": src.get("sector", ""), - "reason": f"热门板块{src['sector']}领涨+{chg:.1f}%" - } - new_candidates.append(candidate) - print(f" ✅ {code} {name} 价{price} (+{chg:.1f}%) 评分{score}", flush=True) - - # 4. 写入candidates表 - if new_candidates: - conn = get_conn() - for c in new_candidates: - conn.execute( - "INSERT OR REPLACE INTO candidates (code, name, price, change_pct, score, " - "entry_low, entry_high, stop_loss, take_profit, source, sector, reason, created_at) " - "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,datetime('now','localtime'))", - (c["code"], c["name"], c["price"], c["change_pct"], c["score"], - c["entry_low"], c["entry_high"], c["stop_loss"], c["take_profit"], - c["source"], c["sector"], c["reason"]) - ) - conn.commit() - conn.close() - print(f" ✅ 写入{len(new_candidates)}只候选", flush=True) - else: - print(f" ⚪ 无新候选", flush=True) - - # 5. 推送到Dad（只推高分） - high_score = [c for c in new_candidates if c["score"] >= 6] - if high_score: - lines = ["🔍 市场扫描发现潜在机会："] - for c in high_score[:3]: - lines.append( - f" {c['name']}({c['code']}) 价{c['price']:.2f}(+{c['change_pct']:.1f}%) " - f"评分{c['score']}/10 | {c['reason']}" - ) - msg = "\n".join(lines) - # 推XMPP - try: - payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "type": "chat"}).encode() - req = urllib.request.Request("http://127.0.0.1:5805/", data=payload, - headers={"Content-Type": "application/json"}) - urllib.request.urlopen(req, timeout=5) - print(f" 📨 已推送{len(high_score)}只高评分候选", flush=True) - except Exception as e: - print(f" ⚠️ 推送失败: {e}", file=sys.stderr) - -if __name__ == "__main__": - scan_candidates() +#!/usr/bin/env python3 +"""market_scanner.py — 全市场异动扫描（替代小果扫描线） + +每15分钟扫描： + 1. 板块轮动检测（从已采集的 sector_snapshots 读） + 2. 热门板块领涨股扫描（从腾讯API批量拉） + 3. 资金流向异常检测（从已有 capital_flow_cache 读） + 4. 输出候选股到 candidates 表 + +不依赖小果LLM，纯数据驱动。 +""" +import sys, json, sqlite3, urllib.request, re, time +from pathlib import Path +from datetime import datetime + +# XMPP 日志 hook +try: + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + from xmpp_logger import log_xmpp +except ImportError: + def log_xmpp(*a, **kw): pass + +DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") +CANDIDATES_FILE = Path("/home/hmo/web-dashboard/data/candidate_pool.json") + +UA = "Mozilla/5.0" + +def get_conn(): + return sqlite3.connect(str(DB_PATH)) + +def fetch_qq_batch(symbols): + """腾讯批量行情""" + if not symbols: + return {} + url = f"http://qt.gtimg.cn/q={','.join(symbols)}" + try: + req = urllib.request.Request(url, headers={"User-Agent": UA}) + proxy = urllib.request.ProxyHandler({}) + opener = urllib.request.build_opener(proxy) + with opener.open(req, timeout=15) as r: + text = r.read().decode("gbk") + results = {} + for line in text.strip().split("\n"): + if "~" not in line: + continue + parts = line.split("~") + if len(parts) < 40: + continue + m = re.search(r'_(\w+)=', parts[0]) + market = m.group(1) if m else "" + code = parts[2] + name = parts[1] + price = float(parts[3]) if parts[3] else 0 + chg_pct = float(parts[32]) if parts[32] else 0 + high = float(parts[33]) if parts[33] else 0 + low = float(parts[34]) if parts[34] else 0 + volume = int(parts[6]) if parts[6] else 0 + amount = float(parts[37]) if len(parts) > 37 and parts[37] else 0 + if price > 0: + results[code] = {"code": code, "name": name, "price": price, + "change_pct": chg_pct, "high": high, "low": low, + "volume": volume, "amount": amount, + "market": "SH" if market == "sh" else "SZ" if market == "sz" else "HK"} + return results + except Exception as e: + print(f"[SCANNER] 腾讯API错误: {e}", file=sys.stderr) + return {} + +def scan_hot_sectors(): + """从DB读热门板块，返回板块名+领涨股""" + conn = get_conn() + latest = conn.execute("SELECT MAX(id) FROM market_snapshots").fetchone()[0] + if not latest: + conn.close() + return [] + sectors = conn.execute(""" + SELECT name, change_pct, lead_stock, lead_stock_code, up_count, down_count + FROM sector_snapshots WHERE snapshot_id=? + ORDER BY change_pct DESC LIMIT 15 + """, (latest,)).fetchall() + conn.close() + return [{"name": s[0], "change": s[1], "lead_stock": s[2], + "lead_code": s[3], "up": s[4], "down": s[5]} for s in sectors if s[1] > 2.0] + +def scan_candidates(): + """主扫描流程""" + print(f"[SCANNER] {datetime.now().strftime('%H:%M')} 开始扫描", flush=True) + + # 1. 热门板块领涨股 + hot = scan_hot_sectors() + print(f" 热门板块(涨幅>2%): {len(hot)}个", flush=True) + + candidates = {} + + # 从热门板块拉领涨股 + lead_codes = [] + for s in hot: + if s.get("lead_code"): + lc = str(s["lead_code"]).strip() + if lc and lc not in candidates: + lead_codes.append(lc) + candidates[lc] = {"source": f"板块:{s['name']}(+{s['change']:.1f}%)", "sector": s["name"]} + + print(f" 领涨股待查: {len(lead_codes)}只", flush=True) + + # 2. 腾讯API批量查行情 + symbols = [] + for c in lead_codes: + if len(c) == 6: + if c.startswith(("5", "6", "9")): + symbols.append(f"sh{c}") + else: + symbols.append(f"sz{c}") + else: + symbols.append(f"hk{c}") + + prices = fetch_qq_batch(symbols) + print(f" 行情返回: {len(prices)}只", flush=True) + + # 3. 评估候选 + new_candidates = [] + for code, info in prices.items(): + if code not in candidates: + continue + src = candidates[code] + price = info["price"] + chg = info["change_pct"] + name = info["name"] + vol = info["volume"] + amt = info["amount"] + + # 条件：涨幅>3%，有量 + if chg < 3.0: + continue + if vol <= 0: + continue + + score = min(10, round(3 + chg * 0.5 + (amt / 1e8 if amt > 0 else 0) * 0.1, 1)) + + entry_low = round(price * 0.95, 2) + entry_high = round(price, 2) + stop_loss = round(price * 0.92, 2) + take_profit = round(price * 1.15, 2) + + candidate = { + "code": code, + "name": name, + "price": price, + "change_pct": chg, + "score": score, + "entry_low": entry_low, + "entry_high": entry_high, + "stop_loss": stop_loss, + "take_profit": take_profit, + "source": src["source"], + "sector": src.get("sector", ""), + "reason": f"热门板块{src['sector']}领涨+{chg:.1f}%" + } + new_candidates.append(candidate) + print(f" ✅ {code} {name} 价{price} (+{chg:.1f}%) 评分{score}", flush=True) + + # 4. 写入candidates表 + if new_candidates: + conn = get_conn() + for c in new_candidates: + conn.execute( + "INSERT OR REPLACE INTO candidates (code, name, price, change_pct, score, " + "entry_low, entry_high, stop_loss, take_profit, source, sector, reason, created_at) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,datetime('now','localtime'))", + (c["code"], c["name"], c["price"], c["change_pct"], c["score"], + c["entry_low"], c["entry_high"], c["stop_loss"], c["take_profit"], + c["source"], c["sector"], c["reason"]) + ) + conn.commit() + conn.close() + print(f" ✅ 写入{len(new_candidates)}只候选", flush=True) + else: + print(f" ⚪ 无新候选", flush=True) + + # 5. 推送到Dad（只推高分） + high_score = [c for c in new_candidates if c["score"] >= 6] + if high_score: + lines = ["🔍 市场扫描发现潜在机会："] + for c in high_score[:3]: + lines.append( + f" {c['name']}({c['code']}) 价{c['price']:.2f}(+{c['change_pct']:.1f}%) " + f"评分{c['score']}/10 | {c['reason']}" + ) + msg = "\n".join(lines) + # 推XMPP（通过知微 Bot HTTP 桥 :5805） + import time as _time + t0 = _time.time() + try: + payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "type": "chat"}).encode() + req = urllib.request.Request("http://127.0.0.1:5805/", data=payload, + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req, timeout=5) + print(f" 📨 已推送{len(high_score)}只高评分候选", flush=True) + log_xmpp("out", "zhiwei@yoin.fun", "hmo@yoin.fun", msg, "ok", None, int((_time.time()-t0)*1000)) + except Exception as e: + print(f" ⚠️ 推送失败: {e}", file=sys.stderr) + log_xmpp("out", "zhiwei@yoin.fun", "hmo@yoin.fun", msg, "error", str(e)[:200], int((_time.time()-t0)*1000)) + +if __name__ == "__main__": + scan_candidates() diff --git a/gateway/logs/xmpp_health_log.jsonl b/gateway/logs/xmpp_health_log.jsonl index 81de9f9c..0212c7f9 100644 --- a/gateway/logs/xmpp_health_log.jsonl +++ b/gateway/logs/xmpp_health_log.jsonl @@ -68,3 +68,4 @@ {"timestamp": "2026-07-19 20:45:02", "status": "critical", "last_message_age_sec": 15059, "error_rate_1h": 0, "bot_activity": {"inbound": 0, "outbound": 0, "errors": 0, "last_error": null, "last_inbound": null, "last_outbound": null}, "gateways": {"zhiwei": {"port": 8643, "alive": true}, "mohe": {"port": 8642, "alive": true}, "xiaoguo": {"port": 8645, "alive": true}}, "llm_provider": {"status": "ok", "latency": "fast"}, "best_key": {"key_id": "key6", "label": "key6 (ycdennismo@163.com)", "masked": "sk-oF1WG...hCS3", "rolling": {"reset_in_sec": 9047, "status": "ok", "usage_percent": 79}, "weekly": {"reset_in_sec": 40532, "status": "ok", "usage_percent": 45}, "monthly": {"reset_in_sec": 2470228, "status": "ok", "usage_percent": 22}, "session_expired": false, "issues": [], "total_keys": 7}} {"timestamp": "2026-07-19 20:50:01", "status": "critical", "last_message_age_sec": 15363, "error_rate_1h": 0, "bot_activity": {"error": "journalctl failed"}, "gateways": {"zhiwei": {"port": 8643, "alive": true}, "mohe": {"port": 8642, "alive": true}, "xiaoguo": {"port": 8645, "alive": true}}, "llm_provider": {"status": "ok", "latency": "fast"}, "best_key": {"key_id": "key6", "label": "key6 (ycdennismo@163.com)", "masked": "sk-oF1WG...hCS3", "rolling": {"reset_in_sec": 8732, "status": "ok", "usage_percent": 79}, "weekly": {"reset_in_sec": 40217, "status": "ok", "usage_percent": 45}, "monthly": {"reset_in_sec": 2469913, "status": "ok", "usage_percent": 22}, "session_expired": false, "issues": [], "total_keys": 7}} {"timestamp": "2026-07-19 20:55:01", "status": "critical", "last_message_age_sec": 15661, "error_rate_1h": 0, "bot_activity": {"inbound": 0, "outbound": 0, "errors": 0, "last_error": null, "last_inbound": null, "last_outbound": null}, "gateways": {"zhiwei": {"port": 8643, "alive": true}, "mohe": {"port": 8642, "alive": true}, "xiaoguo": {"port": 8645, "alive": true}}, "llm_provider": {"status": "ok", "latency": "fast"}, "best_key": {"key_id": "key6", "label": "key6 (ycdennismo@163.com)", "masked": "sk-oF1WG...hCS3", "rolling": {"reset_in_sec": 8423, "status": "ok", "usage_percent": 79}, "weekly": {"reset_in_sec": 39908, "status": "ok", "usage_percent": 45}, "monthly": {"reset_in_sec": 2469604, "status": "ok", "usage_percent": 22}, "session_expired": false, "issues": [], "total_keys": 7}} +{"timestamp": "2026-07-19 21:00:01", "status": "critical", "last_message_age_sec": 15958, "error_rate_1h": 0, "bot_activity": {"inbound": 0, "outbound": 0, "errors": 0, "last_error": null, "last_inbound": null, "last_outbound": null}, "gateways": {"zhiwei": {"port": 8643, "alive": true}, "mohe": {"port": 8642, "alive": true}, "xiaoguo": {"port": 8645, "alive": true}}, "llm_provider": {"status": "ok", "latency": "fast"}, "best_key": {"key_id": "key6", "label": "key6 (ycdennismo@163.com)", "masked": "sk-oF1WG...hCS3", "rolling": {"reset_in_sec": 8114, "status": "ok", "usage_percent": 79}, "weekly": {"reset_in_sec": 39599, "status": "ok", "usage_percent": 45}, "monthly": {"reset_in_sec": 2469295, "status": "ok", "usage_percent": 22}, "session_expired": false, "issues": [], "total_keys": 7}} diff --git a/gateway/temp/last_health_check.json b/gateway/temp/last_health_check.json index 50c8ee76..0b78a8d8 100644 --- a/gateway/temp/last_health_check.json +++ b/gateway/temp/last_health_check.json @@ -45,5 +45,5 @@ "ok": 4, "total": 4 }, - "generated_at": "2026-07-19 20:55:01" + "generated_at": "2026-07-19 21:00:01" } \ No newline at end of file