From a36dff2b913a752bbfe7351dd6cc6fb6b89522fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Thu, 9 Jul 2026 17:09:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=85=A8=E5=B8=82=E5=9C=BA=E5=BC=82?= =?UTF-8?q?=E5=8A=A8=E6=89=AB=E6=8F=8F(=E6=9B=BF=E4=BB=A3=E5=B0=8F?= =?UTF-8?q?=E6=9E=9C)+=E8=87=AA=E5=8A=A8=E6=8F=90=E6=8B=94=E5=85=A5?= =?UTF-8?q?=E8=87=AA=E9=80=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/market_scanner.py | 194 ++++++++++++++++++++++++ scripts/promote_candidates.py | 274 ++++++++++------------------------ 2 files changed, 273 insertions(+), 195 deletions(-) create mode 100644 scripts/market_scanner.py diff --git a/scripts/market_scanner.py b/scripts/market_scanner.py new file mode 100644 index 00000000..8f29454e --- /dev/null +++ b/scripts/market_scanner.py @@ -0,0 +1,194 @@ +#!/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() diff --git a/scripts/promote_candidates.py b/scripts/promote_candidates.py index 266dd75b..d8ab8cc8 100644 --- a/scripts/promote_candidates.py +++ b/scripts/promote_candidates.py @@ -1,207 +1,91 @@ #!/usr/bin/env python3 -"""promote_candidates.py — 将候选股自动提拔入自选 +"""promote_candidates.py — 自动提拔候选股入自选 -管道位置： - candidates 表（候选池）→ 本脚本（价格验证+写入策略）→ holding_strategies（自选股） - -运行时机： - 每30分钟，交易日 9:30~15:00 - -no_agent模式：有提拔→输出，无→静默 - -流程： - 1. 从 candidates 读 promoted=0 AND dropped=0 - 2. 用 stock_quote.py 验证实时价 - 3. 检查是否已在持仓/自选（避免重复） - 4. 写入 holding_strategies (decision_type='自选策略', status='active') - 5. 标记 candidates.promoted=1 +从 candidates 表读未提拔的候选，评估后自动加入 holding_strategies。 """ - -import json, os, sqlite3, subprocess, sys, time +import sys, json, sqlite3 from pathlib import Path from datetime import datetime -BASE = Path("/home/hmo/MoFin") -DATA = BASE / "data" -DB_PATH = DATA / "mofin.db" - -sys.path.insert(0, str(BASE / "scripts")) -from mofin_db import get_conn, write_holding_strategy - - -def get_quote(code): - """用 stock_quote.py 获取实时行情""" - try: - r = subprocess.run( - [sys.executable, str(BASE / "scripts/stock_quote.py"), str(code)], - capture_output=True, text=True, timeout=15 - ) - if r.returncode != 0: - return None - for line in r.stdout.strip().split("\n"): - if line.startswith("{"): - return json.loads(line) - except Exception: - return None - return None - - -def already_in_system(conn, code): - """检查是否已在持仓或自选中""" - # holdings - cur = conn.execute("SELECT COUNT(*) FROM holdings WHERE code=? AND is_active=1", (code,)) - if cur.fetchone()[0] > 0: - return "持仓" - # holding_strategies active - cur = conn.execute( - "SELECT COUNT(*) FROM holding_strategies WHERE code=? AND status='active'", - (code,) - ) - if cur.fetchone()[0] > 0: - return "自选" - return None - - -def promote_one(conn, cand): - """将一只候选股提拔为自选股（带策略）""" - code = cand["code"] - name = cand.get("name", "") - - # 1. 检查是否已在系统内 - exist = already_in_system(conn, code) - if exist: - mark_promoted(conn, code, f"已在{exist}") - return None - - # 2. 获取实时行情 - quote = get_quote(code) - if not quote or not quote.get("price"): - return (code, name, f"取价失败，跳过") - - price = quote["price"] - change_pct = quote.get("change_pct", 0) - - # 3. 解析 entry_range - entry_str = cand.get("entry_range", "") - entry_low = 0 - entry_high = 0 - if entry_str and "~" in entry_str: - try: - parts = entry_str.split("~") - entry_low = float(parts[0].strip()) - entry_high = float(parts[1].strip()) - except (ValueError, IndexError): - pass - - # 如果没entry_range，用价格±3%作为默认区间 - if entry_low <= 0 or entry_high <= 0: - entry_low = round(price * 0.97, 2) - entry_high = round(price * 1.03, 2) - - stop_loss = cand.get("stop_loss") or round(entry_low * 0.93, 2) - target = cand.get("target") or round(entry_high * 1.15, 2) - reason = cand.get("reason", "系统挖掘候选") - - # 4. 检测市场（A股/港股） - code_str = str(code).strip() - currency = "HKD" if (len(code_str) <= 5 or code_str.startswith("0")) and len(code_str) < 6 else "CNY" - # 更精确的检测：港股5位 - if len(code_str) <= 5: - currency = "HKD" - elif code_str.startswith("0") and len(code_str) <= 5: - currency = "HKD" - else: - currency = "CNY" - - sector = cand.get("sector", "") - - # 5. 写入 holding_strategies - strategy_data = { - "version": 1, - "price": price, - "cost": price, - "entry_low": entry_low, - "entry_high": entry_high, - "stop_loss": stop_loss, - "take_profit": target, - "currency": currency, - "strategy_type": "watch", - "action": "观望", - "timing_signal": "中性", - "rr_ratio": round((target - entry_low) / (entry_low - stop_loss), 2) if stop_loss and (entry_low - stop_loss) > 0 else 0, - "stock_category": sector or "未分类", - "sector_context": sector or "", - "status": "active", - "source": "candidates", - "reason": reason, - "type": "自选策略", - "decision_type": "自选策略", - "time_horizon": "中线", - "note": f"候选股自动推广 {datetime.now().strftime('%Y-%m-%d %H:%M')}", - } - - ok, msg = write_holding_strategy(conn, code, name, strategy_data) - if ok: - mark_promoted(conn, code) - return (code, name, f"已加自选 {entry_low}~{entry_high} 止损{stop_loss}") - else: - return (code, name, f"写入失败: {msg}") - - -def mark_promoted(conn, code, reason=None): - """标记候选为已推广""" - now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - if reason: - conn.execute( - "UPDATE candidates SET promoted=1, promoted_at=?, dropped=1, drop_reason=? WHERE code=?", - (now, reason, code) - ) - else: - conn.execute( - "UPDATE candidates SET promoted=1, promoted_at=? WHERE code=?", - (now, code) - ) - conn.commit() - +DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") def main(): - start = time.time() - today = datetime.now().strftime("%Y-%m-%d %H:%M") - - conn = get_conn() - if not conn: - print("[SILENT] 无法连接DB", flush=True) - return - - try: - # 读待推广候选股 - rows = conn.execute( - "SELECT * FROM candidates WHERE promoted=0 AND dropped=0 ORDER BY created_at ASC" - ).fetchall() - - if not rows: - print("[SILENT] 无待推广候选", flush=True) - return - - results = [] - for r in rows: - res = promote_one(conn, dict(r)) - if res: - results.append(res) - - conn.commit() - elapsed = time.time() - start - - if results: - print(f"候选股自动推广 | {today} | {len(results)}只处理 ({elapsed:.0f}s)") - for code, name, msg in results: - print(f" {code} {name}: {msg}") - else: - print("[SILENT] 候选推广结束（无结果）", flush=True) - - finally: + conn = sqlite3.connect(str(DB_PATH)) + conn.row_factory = sqlite3.Row + + # 读未提拔候选（按评分降序） + rows = conn.execute(""" + SELECT * FROM candidates + WHERE (promoted IS NULL OR promoted = 0) + AND (dropped IS NULL OR dropped = 0) + AND score >= 6 + ORDER BY score DESC + """).fetchall() + + if not rows: + print("[PROMOTE] 无待提拔候选") conn.close() - + return + + promoted = 0 + for r in rows: + code = str(r["code"]) + name = r["name"] or code + price = r["price"] or 0 + el = r["entry_low"] or 0 + eh = r["entry_high"] or 0 + sl = r["stop_loss"] or 0 + tp = r["take_profit"] or 0 + score = r["score"] or 0 + sector = r["sector"] or "" + reason = r["reason"] or "" + + # 查是否已在 holding_strategies + exists = conn.execute( + "SELECT id FROM holding_strategies WHERE code=? AND status='active'", + (code,) + ).fetchone() + if exists: + # 标记已提拔但不重复加 + conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,)) + print(f" ⏭ {code} {name} 已在自选中，标记promoted") + continue + + # 构建策略 + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + timing_signal = "买入" if score >= 7 else "关注" + action = f"市场扫描发现({reason})" if reason else "市场扫描发现" + + conn.execute(""" + INSERT INTO holding_strategies + (code, name, price, entry_low, entry_high, stop_loss, take_profit, + timing_signal, action, decision_type, strategy_type, status, + rr_ratio, stock_category, created_at, updated_at, + sector_context, quality_check) + VALUES (?,?,?,?,?,?,?,?,?,'自选策略','scan', + 'active',0,'关注',?,?,'', 'pending') + """, (code, name, price, el, eh, sl, tp, timing_signal, action, now, now)) + + conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,)) + promoted += 1 + print(f" ✅ {code} {name} 评分{score} → 已加入自选({timing_signal})", flush=True) + + conn.commit() + print(f"\n[PROMOTE] 本次提拔{promoted}只", flush=True) + + # 推XMPP + if promoted > 0: + try: + import urllib.request + msg = f"📈 自动提拔{promoted}只候选入自选" + 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) + except Exception: + pass + + conn.close() if __name__ == "__main__": main()