#!/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表（对齐真实表结构：表无 price/change_pct/score/entry_low/entry_high/ # take_profit/source 列，原 INSERT 必失败——2026-07-23 审计实测确认。 # UPSERT 只更新扫描器自有列，保留 score_*/pass_*/promoted/log 等计算列） if new_candidates: conn = get_conn() for c in new_candidates: conn.execute( "INSERT INTO candidates (code, name, sector, reason, " "entry_range, stop_loss, target, created_at) " "VALUES (?,?,?,?,?,?,?,datetime('now','localtime')) " "ON CONFLICT(code) DO UPDATE SET " "name=excluded.name, sector=excluded.sector, reason=excluded.reason, " "entry_range=excluded.entry_range, stop_loss=excluded.stop_loss, target=excluded.target", (c["code"], c["name"], c.get("sector") or c.get("source", "market_scanner"), f"{c['reason']} 价{c['price']}(+{c['change_pct']:.1f}%) 评分{c['score']}", f"{c['entry_low']}~{c['entry_high']}", c["stop_loss"], c["take_profit"]) ) 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()