fix: deploy scripts/ files properly (correct directory)

scripts/branch_scanner.py

@@ -23,6 +23,10 @@ SCANNER_STATE = "/home/hmo/web-dashboard/data/scanner_state.json"
 
 
 def get_price(code):
+    # DB 优先
+    try: from mofin_db import get_price_from_db; p, _ = get_price_from_db(code); return p if p else 0
+    except: pass
+    # Fallback: 腾讯
     mkt = "sh" if code.startswith("6") or code.startswith("5") else "sz"
     url = f"http://qt.gtimg.cn/q={mkt}{code}"
     req = Request(url, headers={"User-Agent": "Mozilla/5.0"})

scripts/capital_flow_collector.py

@@ -6,14 +6,39 @@
 
 API: push2his.eastmoney.com 个股资金流日线
 """
-import json, os, sys, time
+import json, os, sys, time, urllib.request
 from datetime import datetime
-from urllib.request import urlopen
+from urllib.request import urlopen, Request
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from threading import Semaphore
 
 DATA_DIR = "/home/hmo/web-dashboard/data"
 DECISIONS_PATH = f"{DATA_DIR}/decisions.json"
 CACHE_PATH = f"{DATA_DIR}/capital_flow_cache.json"
 
+UA = "Mozilla/5.0"
+# 限速器：最多5个并发，每请求后强制间隔0.3s
+RATE_LIMIT = Semaphore(5)
+MIN_INTERVAL = 0.3
+_last_req = 0
+
+def _rate_limited_request(url):
+    """带速率限制的HTTP GET，用Semaphore控制并发数"""
+    global _last_req
+    with RATE_LIMIT:
+        elapsed = time.time() - _last_req
+        if elapsed < MIN_INTERVAL:
+            time.sleep(MIN_INTERVAL - elapsed)
+        proxy_handler = urllib.request.ProxyHandler({})
+        opener = urllib.request.build_opener(proxy_handler)
+        req = Request(url, headers={"User-Agent": UA, "Referer": "https://data.eastmoney.com/"})
+        try:
+            resp = opener.open(req, timeout=8)
+            _last_req = time.time()
+            return json.loads(resp.read().decode("utf-8"))
+        except Exception:
+            return None
+
 # eastmoney secid: 1=上海 0=深圳
 def secid(code):
     code = str(code).strip()
@@ -22,12 +47,12 @@ def secid(code):
     return f"0.{code}"
 
 def fetch_flow(code, days=5):
-    """拉取个股近N日资金流"""
+    """拉取个股近N日资金流（带限速+代理绕过）"""
     sid = secid(code)
     url = f"http://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get?secid={sid}&fields1=f1,f2,f3,f7&fields2=f51,f52,f53,f54,f55,f56,f57&lmt={days}"
-    try:
-        resp = urlopen(url, timeout=5)
-        data = json.loads(resp.read().decode("utf-8"))
+    data = _rate_limited_request(url)
+    if not data:
+        return None
     klines = data.get("data", {}).get("klines", [])
     if not klines:
         return None
@@ -44,8 +69,6 @@ def fetch_flow(code, days=5):
                 "small": float(p[5]),          # 小单净流入(元)
             })
     return result
-    except Exception as e:
-        return None
 
 def fetch_flow_intraday(code):
     """拉取当日分时资金流（用于盘中判断）"""
@@ -122,16 +145,29 @@ def main():
 
     all_flows = {}
 
-    for code in sorted(codes):
+    # 并行抓取：ThreadPoolExecutor + 内置限速器（Semaphore 5 + 0.3s间隔）
+    code_list = sorted(codes)
+    if not code_list:
+        print("[capital_flow] 无代码需要采集")
+        return
+
+    def fetch_one(code):
         flow = fetch_flow(code, days=5)
         if flow:
             analysis = analyze_flow(flow)
-            all_flows[code] = {
+            return (code, {
                 "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
                 "flow": flow,
                 "analysis": analysis,
-            }
-        time.sleep(0.3)  # API限流
+            })
+        return (code, None)
+
+    with ThreadPoolExecutor(max_workers=5) as pool:
+        futures = {pool.submit(fetch_one, c): c for c in code_list}
+        for f in as_completed(futures):
+            code, result = f.result()
+            if result:
+                all_flows[code] = result
 
     # 写缓存
     cache = {
@@ -139,7 +175,7 @@ def main():
         "stocks": all_flows,
     }
     json.dump(cache, open(CACHE_PATH, "w"), indent=2, ensure_ascii=False)
-    print(f"[capital_flow] {len(all_flows)}只更新完成")
+    print(f"[capital_flow] {len(all_flows)}/{len(code_list)}只更新完成")
 
 if __name__ == "__main__":
     main()

scripts/divergence_detector.py

@@ -41,7 +41,7 @@ DIVERGENCE_MODERATE = 3.0 # >3% → moderate信号
 STREAK_DAYS = 3           # 连涨/连跌3天 → 信号
 
 def fetch_indices():
-    """获取所有指数实时数据"""
+    """获取所有指数实时数据（指数无 DB 缓存，腾讯 API 是唯一源）"""
     symbols = list(INDEX_CODES.values())
     url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
     try:

scripts/hardcode_scanner.py

@@ -41,7 +41,7 @@ SUSPICIOUS_NUMBERS = [
     (r'1手\s*[:=]\s*\d{3,}', '可能的每手股数硬编码'),
         (r'[><=]\s*0\.[0-9]+', '可能的百分比阈值硬编码'),
         (r'仓位\s*[:=]\s*\d{3,}', '可能的仓位金额硬编码'),
-        (r'['\"](?!http|~|\./|\.\./)/home/[^'\"]+['\"]', '可能的文件路径硬编码（应使用环境变量或配置）'),
+        (r"['\"](?!http|~|\./|\.\./)/home/[^'\"]+['\"]", '可能的文件路径硬编码（应使用环境变量或配置）'),
     # 扩展点 — meta_growth 在此追加新规则
 ]
 

scripts/macro_context_collector.py

@@ -29,34 +29,35 @@ STATE_PATH = DATA_DIR / "macro_risk_state.json"
 # HIGH: 任何一条匹配 → 立即 HIGH 预警
 HIGH_PATTERNS = [
     # 全球巨头+核心产业
-    r"苹果.*(?:涨价|降价|推迟|取消|禁|制裁|调查|召回|大跌|暴跌)",
-    r"openai.*(?:推迟|取消|风险|调查|起诉|倒闭|ipo)",
-    r"英伟达|nvidia.*(?:跌|调查|制裁|推迟|禁令)",
-    r"台积电.*(?:跌|推迟|取消|地震|火灾|禁)",
+    r"苹果[^。]*(?:涨价|降价|推迟|取消|禁|制裁|调查|召回|大跌|暴跌)",
+    r"openai[^。]*(?:推迟|取消|风险|调查|起诉|倒闭|ipo)",
+    r"(?:英伟达|nvidia)[^。]*(?:跌|调查|制裁|推迟|禁令)",
+    r"台积电[^。]*(?:跌|推迟|取消|地震|火灾|禁)",
     r"特斯拉.*(?:暴跌|召回|调查|破产|禁)",
     # 美联储/央行意外
     r"美联储.*(?:意外|紧急|缩表|风暴|警告|超预期|加息\s*50|降息\s*50|紧急\s*(?:会议|声明))",
     r"美联储.*(?:利率|决议).*(?:超预期|意外|紧急)",
     r"fed.*(?:emergency|unexpected|surprise|hike|cut)",
-    # 指数暴跌
-    r"指数.*(?:跌幅|暴跌|熔断|闪崩|重挫)",
+    # 指数暴跌（需 ≥2% 跌幅或使用更强范围词）
+    r"指数[^。]*?(?:暴跌|熔断|闪崩|重挫)",
+    r"指数[^。]*?(?:跌幅|下跌)[^。]*?[2-9]%",
     r"(?:暴跌|重挫|熔断).*[5-9]%",
     r"熔断|闪崩",
     # 地缘+贸易
     r"关税.*(?:升级|新|报复|制裁)",
     r"制裁.*(?:新|升级|全面)",
-    r"战争|开战|入侵|核|导弹.*发射",
+    r"战争|开战|入侵|核(?:威胁|武器|弹头|试验|攻击|冲突|导弹|战争|潜艇|问题|危机|设施)|导弹[^。]*(?:发射|袭击|攻击)",
     # 系统性能源
     r"原油.*(?:跌破|暴跌|崩盘|断供)",
     r"石油.*(?:禁运|制裁|断供)",
     r"能源危机|粮食危机",
     # 系统金融
     r"银行.*(?:倒闭|挤兑|破产|接管|危机)",
-    r"金融危机|债务危机|违约潮|系统性",
+    r"金融危机(?:风险|爆发|蔓延|冲击|预警|警示|逼近|担忧|席卷|升级|恐慌|进入|出现|形成|即将|来袭|警报|当前|新一轮|全面|全球性)|债务危机|违约潮|系统性(?:风险|危机)",
     # AI/科技板块重挫
     r"半导体.*(?:暴跌|熔断|崩盘|跌幅)",
     r"科技股.*(?:暴跌|熔断|崩盘|重挫)",
-    r"费城半导体|sox.*(?:跌|崩)",
+    r"(?:费城半导体|sox)[^。]*?(?:跌|崩)",
 ]
 
 # MEDIUM: 累计匹配2条以上 → MEDIUM 预警
@@ -75,6 +76,39 @@ MEDIUM_PATTERNS = [
     r"黑天鹅|灰犀牛",
 ]
 
+# ── 模式完整性校验（防 .pyc 缓存/版本回退） ──
+# 直接检查关键特征字符串是否存在于模式中
+# 原理: 旧版有 standalone |核| , 新版有 核(?:威胁
+# 旧版有 银行.*倒闭|挤兑|破产 , 新版有 银行.*(?:倒闭|挤兑|破产
+_PATTERN_CHECKS = {
+    8: ["暴跌|熔断|闪崩|重挫"],  # index pattern must use strong crash words, not "跌幅"
+    9: ["[2-9]%"],  # 指数+跌幅 requires ≥2%
+    14: ["核(?:威胁", "核威胁|武器|弹头"],  # must NOT have standalone 核
+    18: ["倒闭|挤兑|破产"],  # bank pattern must have crisis keywords
+    19: ["金融危机(?:风险", "危机|债务危机"],  # must NOT have standalone 金融危机
+}
+_KNOWN_BAD_SIGS = {
+    # Known stale .pyc signature fragments that indicate wrong version
+    "指数.*跌幅": "旧版用 .* 跨句匹配且无 ≥2% 阈值",
+    "|核|": "旧版有独立单字核",
+    "英伟达|nvidia.*跌": "旧版 alternation 分组错误",
+    "导弹.*发射": "旧版只匹配发射不匹配袭击",
+    "|金融危机|": "旧版 standalone 金融危机匹配历史参照",
+}
+for idx, required_list in _PATTERN_CHECKS.items():
+    if not any(req in HIGH_PATTERNS[idx] for req in required_list):
+        print(f"[MACRO-安全] ⚠️ HIGH_PATTERNS[{idx}] 签名不匹配！")
+        print(f"[MACRO-安全]   当前: {HIGH_PATTERNS[idx][:100]}")
+        print(f"[MACRO-安全]   预期应包含: {required_list[0]}")
+        print(f"[MACRO-安全]   可能原因: .pyc 缓存过期 / 回退到旧版本")
+
+# 额外扫描：检查是否有已知的旧版签名残留
+_all_patterns_text = "\n".join(HIGH_PATTERNS)
+for bad_sig, reason in _KNOWN_BAD_SIGS.items():
+    if bad_sig in _all_patterns_text:
+        print(f"[MACRO-安全] ⚠️ 检测到旧版模式签名 '{bad_sig}' ({reason})")
+        print(f"[MACRO-安全]    .pyc 缓存可能未刷新，当前 HIGH_PATTERNS 可能仍为旧版本")
+
 def ensure_tables(conn):
     conn.execute("""
         CREATE TABLE IF NOT EXISTS macro_raw_news (

scripts/macro_signal_consumer.py

@@ -11,7 +11,7 @@ no_agent 模式：有HIGH风险→输出风险摘要 | 无→静默
   ↓
   如果 HIGH → 推送到 Dad
 """
-import sqlite3, json, os
+import sqlite3, json, os, sys, time
 from pathlib import Path
 from datetime import datetime
 
@@ -20,8 +20,25 @@ DATA = BASE / "data"
 DB_PATH = DATA / "mofin.db"
 STATE_PATH = DATA / "macro_risk_state.json"
 
+def db_update(conn, sql, params, max_retries=3):
+    """幂等DB更新，遇到锁自动重试"""
+    for attempt in range(max_retries):
+        try:
+            conn.execute(sql, params)
+            conn.commit()
+            return True
+        except sqlite3.OperationalError as e:
+            if "locked" in str(e).lower():
+                if attempt < max_retries - 1:
+                    time.sleep(1)
+                continue
+            raise
+    print(f"[MACRO-CONSUMER] DB更新失败（持续锁）: {sql}", file=sys.stderr)
+    return False
+
 def main():
-    conn = sqlite3.connect(str(DB_PATH))
+    conn = sqlite3.connect(str(DB_PATH), timeout=10)
+    conn.execute("PRAGMA busy_timeout=5000")
     conn.row_factory = sqlite3.Row
 
     # 读取未处理的 macro_watch 信号
@@ -43,14 +60,28 @@ def main():
         conn.close()
         return  # SILENT
 
-    # 聚合风险等级
+    # 聚合风险等级（考虑修正覆盖信号）
     levels = {"宏观-WATCH_HIGH": "high", "宏观-WATCH_MEDIUM": "medium", "宏观-WATCH_INFO": "info"}
     highest = "info"
     all_summaries = []
     
+    def _effective_level(sentiment, summary):
+        """修正覆盖信号取修正后的级别，不按原始sentiment算"""
+        if "修正覆盖" in (summary or ""):
+            s = (summary or "").lower()
+            # 明确说零风险/正面/利好 → info
+            if any(kw in s for kw in ["零风险", "正面利好", "正面进展", "非风险", "利好"]):
+                return "info"
+            # 明确说MEDIUM → medium
+            if "medium" in s or "中风险" in s:
+                return "medium"
+            # 修正覆盖HIGH → 默认降为medium（不保留原始HIGH）
+            return "medium"
+        return levels.get(sentiment, "info")
+    
     for r in rows:
         sentiment = r["overall_sentiment"]
-        lv = levels.get(sentiment, "info")
+        lv = _effective_level(sentiment, r["summary"])
         if lv == "high":
             highest = "high"
         elif lv == "medium" and highest != "high":
@@ -72,10 +103,9 @@ def main():
     }
     STATE_PATH.write_text(json.dumps(state, ensure_ascii=False, indent=2))
 
-    # 标记为已处理
+    # 标记为已处理（含重试）
     for r in rows:
-        conn.execute("UPDATE signal_news SET processed=1 WHERE id=?", (r["id"],))
-    conn.commit()
+        db_update(conn, "UPDATE signal_news SET processed=1 WHERE id=?", (r["id"],))
     conn.close()
 
     # no_agent 输出（有 HIGH 才主动出声）

scripts/mofin_collect.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""mofin_collect.py — MoFin 数据采集链
+
+每轮盯盘 cron 前运行，顺序执行：
+  1. market_watch — 拉90个行业板块数据（9:30前跳过，市场未开）
+  2. trend_detector — 检测17种信号（依赖板块数据，同跳）
+  3. mofin_news — 搜新闻+小果分析
+  4. stock_quote — 所有持仓最新行情（CRITICAL: LLM唯一价格源）
+"""
+
+import subprocess, sys, time
+from pathlib import Path
+from datetime import datetime
+
+BASE = Path(__file__).parent.parent if "hermes" in str(Path(__file__).resolve()) else Path(__file__).parent
+
+now = datetime.now()
+market_open = (now.hour >= 9 and now.minute >= 30) or now.hour >= 10
+
+# 步骤1-3: 行业/新闻数据
+SCRIPTS = []
+if market_open:
+    SCRIPTS.append(("market_watch.py", 60))
+    SCRIPTS.append(("trend_detector.py", 60))
+else:
+    print(f"[{now.strftime('%H:%M')}] 市场未开盘(9:30)，跳过板块采集", flush=True)
+
+SCRIPTS.append(("mofin_news.py", 50))
+
+for script, timeout in SCRIPTS:
+    path = BASE / script
+    if not path.exists():
+        path = Path("/home/hmo/MoFin") / script
+    print(f"--- {script} ---", flush=True)
+    start = time.time()
+    try:
+        result = subprocess.run(
+            [sys.executable, str(path)],
+            capture_output=True, text=True, timeout=timeout
+        )
+        elapsed = time.time() - start
+        if result.returncode == 0:
+            print(f"OK ({elapsed:.0f}s)", flush=True)
+            if result.stdout.strip():
+                for line in result.stdout.strip().split("\n")[-3:]:
+                    print(f"  {line}", flush=True)
+        else:
+            print(f"FAIL ({elapsed:.0f}s): {result.stderr[:200]}", flush=True)
+    except subprocess.TimeoutExpired:
+        print(f"TIMEOUT ({timeout}s)", flush=True)
+    except Exception as e:
+        print(f"ERROR: {e}", flush=True)
+
+# ── 步骤4: 个股行情注入（唯一权威价格源）──
+# 所有持仓最新行情，注入到 LLM context
+# LLM 禁止自行调用原始API解析价格
+PRICE_SCRIPT = BASE / "stock_quote.py"
+if not PRICE_SCRIPT.exists():
+    PRICE_SCRIPT = Path("/home/hmo/MoFin/scripts/stock_quote.py")
+if PRICE_SCRIPT.exists():
+    print("--- stock_quote.py ---", flush=True)
+    try:
+        result = subprocess.run(
+            [sys.executable, str(PRICE_SCRIPT), "--all-holdings"],
+            capture_output=True, text=True, timeout=30
+        )
+        if result.returncode == 0 and result.stdout.strip():
+            lines = [l for l in result.stdout.strip().split("\n") if l.strip()]
+            print(f"OK ({len(lines)}只持仓)", flush=True)
+            for line in lines[:50]:
+                print(f"  {line}", flush=True)
+        else:
+            print(f"WARN: stock_quote stderr={result.stderr[:100]}", flush=True)
+    except Exception as e:
+        print(f"WARN: stock_quote skipped ({e})", flush=True)
+else:
+    print("WARN: stock_quote.py not found", flush=True)
+
+print("采集链完成", flush=True)

scripts/price_monitor.py

@@ -0,0 +1,788 @@
+#!/usr/bin/env python3
+"""price_monitor.py — 高频价格监控脚本（批量版）
+规则：进入区间报一次，离开区间报一次，中间不重复。
+每次运行时一次性刷新所有持仓+自选股的实时价。
+"""
+import json
+import urllib.request
+import os
+import sys
+import time
+from datetime import datetime
+
+# ── MoFin unified model ──────────────────────────────────────────────
+sys.path.insert(0, "/home/hmo/MoFin")
+from mo_models import is_hk_stock, get_hk_rate, calc_total_assets, calc_total_mv, calc_position_pct
+from mofin_db import get_conn, write_holdings_batch, write_portfolio_summary, write_price_event, write_watchlist_stock
+
+DECISIONS_PATH = "/home/hmo/web-dashboard/data/decisions.json"
+PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"
+WATCHLIST_PATH = "/home/hmo/web-dashboard/data/watchlist.json"
+BREACH_PATH = "/home/hmo/.hermes/zone_breach.json"
+STATE_PATH = os.path.expanduser("~/.hermes/price_trigger_state.json")
+EVENTS_PATH = "/home/hmo/web-dashboard/data/price_events.json"
+
+# 策略重评依赖（技术面驱动，非机械百分比）
+sys.path.insert(0, "/home/hmo/web-dashboard")
+try:
+    from strategy_lifecycle import reassess_strategy
+    HAS_REASSESS = True
+except ImportError:
+    HAS_REASSESS = False
+
+try:
+    HK_RATE = get_hk_rate()
+except Exception:
+    HK_RATE = 0.87  # ultimate fallback
+
+# 分支系统与情景检测
+try:
+    sys.path.insert(0, '/home/hmo/MoFin')
+    from strategy_tree import detect_scenario, evaluate_branches
+    HAS_TREE = True
+except Exception:
+    HAS_TREE = False
+    def detect_scenario(): return {}
+    def evaluate_branches(*a, **kw): return []
+
+# 情景缓存（每次run_once刷新）
+_SCENARIO_CACHE = {}
+_BRANCH_CACHE = {}  # code -> branches list
+
+UA = "Mozilla/5.0"
+
+# ── 批量拉取价格 ──────────────────────────────────────────────────────────
+
+def fetch_all_prices(codes):
+    """腾讯批量行情API：仅用于A股（沪市/深市）
+    A股：sh600110 / sz000001
+    港股已迁移至 fetch_hk_eastmoney()（东方财富实时行情）
+    返回 {code: (price, change, change_pct)}
+    """
+    if not codes:
+        return {}
+
+    # 只处理A股（6位代码），港股走东方财富
+    a_codes = [c for c in codes if len(str(c).strip()) == 6]
+    if not a_codes:
+        return {}
+
+    symbols = []
+    code_map = {}
+    for code in a_codes:
+        code_s = str(code).strip()
+        if code_s.startswith(('5', '6', '9')):
+            sym = f"sh{code_s}"
+        else:
+            sym = f"sz{code_s}"
+        symbols.append(sym)
+        code_map[sym] = code_s
+
+    url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
+    try:
+        req = urllib.request.Request(url, headers={"User-Agent": UA})
+        with urllib.request.urlopen(req, timeout=10) as r:
+            text = r.read().decode("gbk")
+    except Exception as e:
+        print(f"⚠️ 腾讯A股拉取失败: {e}", file=sys.stderr)
+        return {}
+
+    results = {}
+    for line in text.strip().split("\n"):
+        line = line.strip()
+        if not line or "=" not in line:
+            continue
+        try:
+            raw_value = line.split("=", 1)[1].strip().strip('"').strip(";")
+            fields = raw_value.split("~")
+            if len(fields) < 6:
+                continue
+            sym = line.split("=", 1)[0].strip().lstrip("v_")
+            orig_code = code_map.get(sym)
+            if not orig_code:
+                continue
+            price = float(fields[3]) if fields[3] else 0
+            prev_close = float(fields[4]) if fields[4] else 0
+            change = price - prev_close if prev_close > 0 else 0
+            change_pct = fields[32] if len(fields) > 32 and fields[32] else "0"
+            results[orig_code] = (price, change, change_pct)
+        except (ValueError, IndexError):
+            continue
+
+    return results
+
+
+# ── 港股实时行情（新浪财经批量版，实时，无延迟）─────────────────────────────
+
+def fetch_hk_sina_batch(codes):
+    """新浪财经港股批量实时行情 — 一次HTTP请求获取全部港股。
+
+    新浪港股API（hq.sinajs.cn）支持批量查询，返回实时数据。
+    对比东财逐股查询（0.2s间隔×17只=3.4s），新浪1次请求搞定。
+
+    API: https://hq.sinajs.cn/list=hk00700,hk09988
+    格式: hq_str_hk00700="TENCENT,腾讯控股,当前价,昨收,开盘,最高,最低,涨跌额,涨跌幅,..."
+
+    返回 {code: (price, change, change_pct)}
+    """
+    if not codes:
+        return {}
+
+    hk_codes = [str(c).strip() for c in codes if len(str(c).strip()) <= 5]
+    if not hk_codes:
+        return {}
+
+    symbols = [f"hk{c}" for c in hk_codes]
+    url = f"https://hq.sinajs.cn/list={','.join(symbols)}"
+
+    try:
+        # 新浪要求有 Referer，且需绕过系统代理（某些环境下东财/新浪走代理会断连）
+        proxy_handler = urllib.request.ProxyHandler({})
+        opener = urllib.request.build_opener(proxy_handler)
+        req = urllib.request.Request(url, headers={
+            "User-Agent": "Mozilla/5.0",
+            "Referer": "https://finance.sina.com.cn",
+        })
+        with opener.open(req, timeout=10) as r:
+            text = r.read().decode("gbk")
+    except Exception as e:
+        print(f"⚠️ 新浪港股批量拉取失败: {e}", file=sys.stderr)
+        return {}
+
+    results = {}
+    for line in text.strip().split("\n"):
+        line = line.strip()
+        if "=" not in line:
+            continue
+        try:
+            code = line.split("=", 1)[0].replace("hq_str_hk", "").replace("var ", "").strip()
+            raw = line.split("=", 1)[1].strip().strip('"').strip(";")
+            fields = raw.split(",")
+            if len(fields) < 9:
+                continue
+            price = float(fields[2]) if fields[2] else 0
+            prev_close = float(fields[3]) if fields[3] else 0
+            change_amt = float(fields[7]) if fields[7] else 0
+            change_pct = fields[8] if fields[8] else "0"
+            # 新浪 field[2] 可能非实时最新价，用 prev_close + change 计算更准确
+            if prev_close > 0 and abs(change_amt) > 0:
+                price = round(prev_close + change_amt, 2)
+            change = round(change_amt, 2)
+            if price > 0:
+                results[code] = (price, change, change_pct)
+        except (ValueError, IndexError):
+            continue
+
+    return results
+
+
+# ── 港股备用通道（东方财富逐股 + 腾讯15min延迟）───────────────────────────
+
+def fetch_hk_eastmoney_fallback(codes):
+    """东方财富港股实时行情（备用通道），逐股查询、间隔1秒避免限流。
+
+    FTP 说明：港股限流严重，不适合主通道，降级为备用。
+    建议用上面的 fetch_hk_sina_batch() 做主通道。
+
+    返回 {code: (price, change, change_pct)}
+    Fallback: 仍失败时回退到腾讯 qt.gtimg.cn（15分钟延迟）
+    """
+    if not codes:
+        return {}
+
+    hk_codes = [str(c).strip() for c in codes if len(str(c).strip()) <= 5]
+    if not hk_codes:
+        return {}
+
+    results = {}
+
+    # 东方财富逐股查询，1秒间隔避免限流
+    for code in hk_codes:
+        try:
+            url = (f"https://push2.eastmoney.com/api/qt/stock/get"
+                   f"?secid=116.{code}"
+                   f"&fields=f43,f170,f60,f57,f58"
+                   f"&fltt=2")
+            proxy_handler = urllib.request.ProxyHandler({})
+            opener = urllib.request.build_opener(proxy_handler)
+            req = urllib.request.Request(url, headers={
+                "User-Agent": UA,
+                "Referer": "https://quote.eastmoney.com/",
+            })
+            with opener.open(req, timeout=5) as r:
+                resp = json.loads(r.read().decode("utf-8"))
+
+            if resp.get("rc") != 0:
+                continue
+            item = resp.get("data", {})
+            if not item:
+                continue
+            price = float(item.get("f43", 0)) if item.get("f43") else 0
+            prev_close = float(item.get("f60", 0)) if item.get("f60") else 0
+            change = round(price - prev_close, 2) if prev_close > 0 else 0
+            change_pct = str(item.get("f170", "0"))
+            if price > 0:
+                results[code] = (price, change, change_pct)
+            time.sleep(1.0)  # 1秒间隔，大幅降低限流概率
+        except Exception as e:
+            print(f"  [东财备用 {code}] {e}", file=sys.stderr)
+            continue
+
+    # Fallback: 腾讯 qt.gtimg.cn（15分钟延迟）
+    missing = [c for c in hk_codes if c not in results]
+    if missing:
+        try:
+            fallback = _fetch_hk_tencent_fallback(missing)
+            results.update(fallback)
+        except Exception:
+            pass
+
+    return results
+
+
+def _fetch_hk_tencent_fallback(codes):
+    """腾讯港股行情（15分钟延迟，仅作 fallback）"""
+    symbols = [f"hk{c}" for c in codes]
+    url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
+    req = urllib.request.Request(url, headers={"User-Agent": UA})
+    with urllib.request.urlopen(req, timeout=10) as r:
+        text = r.read().decode("gbk")
+
+    code_map = {f"hk{c}": c for c in codes}
+    results = {}
+    for line in text.strip().split("\n"):
+        if "=" not in line:
+            continue
+        try:
+            raw = line.split("=", 1)[1].strip().strip('"').strip(";")
+            fields = raw.split("~")
+            if len(fields) < 6:
+                continue
+            sym = line.split("=", 1)[0].strip().lstrip("v_")
+            orig = code_map.get(sym)
+            if not orig:
+                continue
+            price = float(fields[3]) if fields[3] else 0
+            prev_close = float(fields[4]) if fields[4] else 0
+            change = price - prev_close if prev_close > 0 else 0
+            change_pct = fields[32] if len(fields) > 32 and fields[32] else "0"
+            results[orig] = (price, change, change_pct)
+        except (ValueError, IndexError):
+            continue
+    return results
+
+
+def refresh_data_prices():
+    """一次性刷新portfolio.json和watchlist.json的所有实时价"""
+    all_codes = set()
+
+    # 收集所有需要拉取的代码
+    try:
+        pf = json.load(open(PORTFOLIO_PATH))
+        for s in pf.get('holdings', []):
+            all_codes.add(s['code'])
+    except:
+        pf = {"holdings": []}
+
+    try:
+        wl = json.load(open(WATCHLIST_PATH))
+        for s in wl.get('stocks', []):
+            all_codes.add(s['code'])
+    except:
+        wl = {"stocks": []}
+
+    if not all_codes:
+        return 0
+
+    # 分批拉取：A股走腾讯（实时） + 港股走新浪批量（实时，无限流）
+    all_list = list(all_codes)
+    prices = fetch_all_prices(all_list)          # A股（腾讯，实时）
+    hk_prices = fetch_hk_sina_batch(all_list)    # 港股（新浪批量，实时）
+    # 新浪未覆盖的走备用通道（东财逐股→腾讯15min延迟）
+    hk_codes_missing = [c for c in all_list if len(str(c).strip()) <= 5 and c not in hk_prices]
+    if hk_codes_missing:
+        fallback = fetch_hk_eastmoney_fallback(hk_codes_missing)
+        hk_prices.update(fallback)
+    prices.update(hk_prices)
+    updated = 0
+
+    # 保存全量实时价快照（供报告管道消费，确保分析用最新数据）
+    try:
+        live = {"updated_at": datetime.now().isoformat(), "prices": {}}
+        for code in all_codes:
+            if code in prices:
+                p, c, chg = prices[code]
+                live["prices"][code] = {"price": p, "change_pct": chg}
+        json.dump(live, open("/home/hmo/web-dashboard/data/live_prices.json", "w"), indent=2)
+    except Exception:
+        pass
+
+    # 更新portfolio（只在价格变化时写入，避免触发文件变更通知）
+    changed = False
+    for s in pf.get('holdings', []):
+        if s['code'] in prices:
+            price, _, change_pct = prices[s['code']]
+            if price > 0:
+                # 港股：API返回HKD，需转RMB
+                if is_hk_stock(s['code']):
+                    price = round(price * HK_RATE, 2)
+                old = s.get('price', 0)
+                if abs(old - price) > 0.001:
+                    s['price'] = round(price, 2)
+                    s['change_pct'] = float(change_pct) if change_pct else 0
+                    updated += 1
+                    changed = True
+    if changed:
+        pf['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M')
+        pf['total_mv'] = calc_total_mv(pf.get('holdings', []))
+        pf['total_assets'] = calc_total_assets(pf)
+        pf['position_pct'] = calc_position_pct(pf)
+        # DB 写入（替代 json.dump，强制币种约束）
+        try:
+            conn = get_conn()
+            write_holdings_batch(conn, pf['holdings'])
+            write_portfolio_summary(conn, pf)
+            conn.close()
+        except Exception as e:
+            print(f"  [DB写入失败] {e}", flush=True)
+        # 保留 JSON 副本作为冷备
+        json.dump(pf, open(PORTFOLIO_PATH, 'w'), ensure_ascii=False, indent=2)
+    elif pf.get('updated_at'):
+        try:
+            last_ts = datetime.strptime(pf['updated_at'], '%Y-%m-%d %H:%M')
+            if (datetime.now() - last_ts).total_seconds() > 600:
+                pf['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M')
+                json.dump(pf, open(PORTFOLIO_PATH, 'w'), ensure_ascii=False, indent=2)
+        except:
+            pass
+
+    # 更新watchlist（只在价格变化时写入）
+    changed = False
+    for s in wl.get('stocks', []):
+        if s['code'] in prices:
+            price, _, change_pct = prices[s['code']]
+            if price > 0:
+                # 港股：API返回HKD，需转RMB
+                if is_hk_stock(s['code']):
+                    price = round(price * HK_RATE, 2)
+                old = s.get('price', 0)
+                if abs(old - price) > 0.001:
+                    s['price'] = round(price, 2)
+                    s['change_pct'] = float(change_pct) if change_pct else 0
+                    updated += 1
+                    changed = True
+    if changed:
+        wl['updated_at'] = datetime.now().isoformat()
+        # DB 写入（替代 json.dump）
+        try:
+            conn = get_conn()
+            for s in wl.get('stocks', []):
+                s['currency'] = 'CNY'  # 自选股价格统一CNY
+                write_watchlist_stock(conn, s)
+            conn.close()
+        except Exception as e:
+            print(f"  [DB watchlist写入失败] {e}", flush=True)
+        # 保留 JSON 冷备
+        json.dump(wl, open(WATCHLIST_PATH, 'w'), ensure_ascii=False, indent=2)
+
+    # --- 汇总值重算（使用 mo_models 唯一公式）---
+    try:
+        live_market_value = calc_total_mv(pf.get('holdings', []))
+        old_mv = pf.get('total_mv', 0)
+        
+        if abs(old_mv - live_market_value) > 0.01:
+            pf['total_mv'] = round(live_market_value, 2)
+        
+        pf['total_assets'] = calc_total_assets(pf)
+        if pf['total_assets'] > 0:
+            pf['position_pct'] = calc_position_pct(pf)
+        pf['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M')
+        # DB 写入
+        try:
+            conn = get_conn()
+            write_portfolio_summary(conn, pf)
+            conn.close()
+        except Exception as e:
+            print(f"  [DB汇总写入失败] {e}", flush=True)
+        # JSON 冷备
+        json.dump(pf, open(PORTFOLIO_PATH, 'w'), ensure_ascii=False, indent=2)
+    except Exception as e:
+        print(f"  [汇总重算失败] {e}", flush=True)
+    # --- 结束汇总重算 ---
+
+    return updated
+
+
+# ── 分支系统辅助函数 ──────────────────────────────────────────────────────
+
+def _branch_alert_suffix(code, price, shares=0, cost=0):
+    """返回分支信息后缀：「 | 情景→动作」"""
+    if not HAS_TREE or not _SCENARIO_CACHE.get('id'):
+        return ""
+    try:
+        sc_id = _SCENARIO_CACHE['id']
+        results = evaluate_branches(code, sc_id, price, shares, cost)
+        for r in results:
+            if r.get('applicable'):
+                _record_branch_trigger(code, r.get('branch_id',''), price)
+                branch_action = r.get('action_type', r.get('action', 'hold'))
+                return f" | {sc_id}→{branch_action}"
+    except Exception:
+        pass
+    return ""
+
+
+def _record_branch_trigger(code, branch_id, price):
+    """记录分支触发事件（自成长：trigger_count+1）"""
+    try:
+        raw = json.load(open(DECISIONS_PATH))
+        for d in raw.get('decisions', []):
+            if d.get('code') == code and d.get('strategy_tree',{}).get('branches'):
+                for b in d['strategy_tree']['branches']:
+                    if b['id'] == branch_id:
+                        b.setdefault('trigger_count', 0)
+                        b['trigger_count'] += 1
+                        b['last_trigger_price'] = round(price, 2)
+                        b['last_triggered'] = datetime.now().isoformat()
+                        break
+        json.dump(raw, open(DECISIONS_PATH, 'w'), ensure_ascii=False, indent=2)
+    except Exception:
+        pass
+
+
+# ── 区间偏离检测 ──────────────────────────────────────────────────────────
+
+def load_state():
+    try:
+        with open(STATE_PATH) as f:
+            return json.load(f)
+    except:
+        return {}
+
+def save_state(state):
+    os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
+    with open(STATE_PATH, 'w') as f:
+        json.dump(state, f, ensure_ascii=False, indent=2)
+
+def load_breaches():
+    try:
+        with open(BREACH_PATH) as f:
+            return json.load(f)
+    except:
+        return {}
+
+def save_breaches(data):
+    os.makedirs(os.path.dirname(BREACH_PATH), exist_ok=True)
+    with open(BREACH_PATH, 'w') as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+
+
+def load_events():
+    try:
+        with open(EVENTS_PATH) as f:
+            return json.load(f)
+    except:
+        return {"events": []}
+
+
+def save_events(events):
+    os.makedirs(os.path.dirname(EVENTS_PATH), exist_ok=True)
+    with open(EVENTS_PATH, 'w') as f:
+        json.dump(events, f, ensure_ascii=False, indent=2)
+
+
+def record_event(code, name, event_type, price, trigger_value, event_label=""):
+    """记录一次价格触发事件到 price_events.json + SQLite"""
+    events = load_events()
+    now = datetime.now().isoformat()
+    events["events"].append({
+        "code": code,
+        "name": name,
+        "event_type": event_type,  # entry_zone, stop_loss, take_profit, exit_zone
+        "price": round(price, 2),
+        "trigger_value": trigger_value,
+        "event_label": event_label,
+        "timestamp": now,
+        "date": datetime.now().strftime("%Y-%m-%d"),
+    })
+    # 保留最近10000条
+    events["events"] = events["events"][-10000:]
+    save_events(events)
+
+    # ── SQLite 双写 ──
+    try:
+        from mofin_db import get_conn, init_all_tables, write_price_event
+        conn = get_conn()
+        init_all_tables(conn)
+        write_price_event(conn, code, name, event_type, price, trigger_value, event_label)
+        conn.close()
+    except Exception:
+        pass  # SQLite 写入失败不影响主流程
+
+
+def get_trigger_zones(d):
+    """返回该decision所有可监控的区间列表，从顶层字段读取"""
+    zones = []
+    is_holding = d.get('shares', 0) > 0
+    # 买入区间（自选和持仓都监控）
+    el = d.get("entry_low", 0)
+    eh = d.get("entry_high", 0)
+    if el and eh and float(el) > 0 and float(eh) > 0:
+        try:
+            zones.append(("entry_zone", "买入区间", float(el), float(eh)))
+        except:
+            pass
+    # 止损+止盈（只有持仓才监控，自选无意义）
+    if is_holding:
+        sl = d.get("stop_loss", 0)
+        if sl and float(sl) > 0:
+            try:
+                zones.append(("stop_loss", "止损", 0, float(sl)))
+            except:
+                pass
+        tp = d.get("take_profit", 0)
+        if tp and float(tp) > 0:
+            try:
+                zones.append(("take_profit_zone", "止盈区间", 0, float(tp)))
+            except:
+                pass
+    return zones
+
+
+def run_once(round_label=""):
+    """执行一轮完整的监控流程"""
+    global _SCENARIO_CACHE, _BRANCH_CACHE
+    label = f" [{round_label}]" if round_label else ""
+    start = time.time()
+
+    # 刷新情景与分支缓存（每轮更新）
+    _SCENARIO_CACHE = detect_scenario() if HAS_TREE else {}
+    _BRANCH_CACHE = {}
+    try:
+        raw = json.load(open(DECISIONS_PATH))
+        for d in raw.get('decisions', []):
+            tree = d.get('strategy_tree', {})
+            if tree and tree.get('branches'):
+                _BRANCH_CACHE[d['code']] = tree['branches']
+    except Exception:
+        pass
+
+    # === 第一步：一次性刷新所有价格 ===
+    refreshed = refresh_data_prices()
+
+    # === 第二步：检查触发条件 ===
+    try:
+        with open(DECISIONS_PATH) as f:
+            dec = json.load(f)
+    except:
+        print(f"❌{label} 无法读取decisions.json", file=sys.stderr)
+        return
+
+    active = [d for d in dec.get("decisions", []) if d.get("status") in ("active", "updated")]
+    state = load_state()
+    outputs = []
+    state_updated = False
+
+    # 收集所有需要检查的代码
+    check_codes = set()
+    for d in active:
+        if get_trigger_zones(d):
+            check_codes.add(d["code"])
+
+    # 批量拉取这些股票的价格
+    prices = fetch_all_prices(list(check_codes))
+
+    for d in active:
+        code = d["code"]
+
+        zones = get_trigger_zones(d)
+        if not zones:
+            continue
+
+        price_info = prices.get(code)
+        if not price_info:
+            continue
+        price, _, _ = price_info
+        if price == 0:
+            continue
+
+        name = d.get("name", code)
+        if code not in state:
+            state[code] = {}
+
+        for key, label, lo, hi in zones:
+            in_zone = lo <= price <= hi
+            prev_in_zone = state[code].get(key, None)
+
+            if in_zone and prev_in_zone != True:
+                if key == "stop_loss":
+                    branch_sfx = _branch_alert_suffix(code, price, d.get('shares',0), d.get('cost',0))
+                    outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！{branch_sfx}")
+                    record_event(code, name, "stop_loss", price, str(hi))
+                else:
+                    extra = ""
+                    if "_price" in key:
+                        batch_shares = d.get(key.replace("_price", "_shares"), "")
+                        action = d.get(key.replace("_price", "_action"), "")
+                        if batch_shares:
+                            extra = f" {action}{batch_shares}股" if action else f" {batch_shares}股"
+                    elif key in ("take_profit_zone",):
+                        act = d.get("take_profit_action", "")
+                        if act:
+                            extra = f"（{act}）"
+                    branch_sfx = _branch_alert_suffix(code, price, d.get('shares',0), d.get('cost',0))
+                    outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}{branch_sfx}")
+                    record_event(code, name, "entry_zone", price, f"{lo}~{hi}", label)
+                state[code][key] = True
+                state_updated = True
+
+            elif not in_zone and prev_in_zone == True:
+                if key != "stop_loss":
+                    outputs.append(f"📌 {name}({code}) {price} → 离开{label}{lo}~{hi}")
+                state[code][key] = False
+                state_updated = True
+
+    # === 第三步：买入区偏离检测 + 自动重评 ===
+    reassesed_codes = []
+    for d in active:
+        code = d["code"]
+        name = d.get("name", code)
+        price_info = prices.get(code)
+        if not price_info:
+            continue
+        price, _, _ = price_info
+        if price == 0:
+            continue
+
+        # 从 decisions.json 中读取 analysis 的买入区
+        entry_low = d.get("entry_low", 0)
+        entry_high = d.get("entry_high", 0)
+        if not entry_low or not entry_high:
+            continue
+
+        in_buy_zone = entry_low <= price <= entry_high
+        prev_in_buy_zone = state.get(code, {}).get("__buy_zone", None)
+
+        # 状态变化时才触发：True→False离区 或 False→True进区
+        # [2026-07-01 fix] prev_in_buy_zone is None（新加自选首次检测）
+        # 也要触发——否则新自选全程不走重评，timing_signal卡在初始值
+        if in_buy_zone and (prev_in_buy_zone == False or prev_in_buy_zone is None):
+            # 进入买入区 → 触发技术面重评，更新止损/止盈/信号
+            outputs.append(f"🔄 {name}({code}) {price} → 重新进入买入区{entry_low}~{entry_high}，触发技术面重评")
+            do_reassess = True
+        elif not in_buy_zone and prev_in_buy_zone == True:
+            # 离开买入区 → 立即重评，更新止损/止盈/区间
+            outputs.append(f"🔄 {name}({code}) {price} → 离开买入区{entry_low}~{entry_high}，立即技术面重评")
+            do_reassess = True
+        else:
+            do_reassess = False
+
+        if do_reassess and HAS_REASSESS:
+            try:
+                cost = d.get("cost", 0) or 0
+                shares = d.get("shares", 0) or 0
+                profit_pct = (price - cost) / cost * 100 if cost else 0
+                is_deep_loss = profit_pct < -20
+                sentiment = "neutral"
+                if d.get("tech_snapshot"):
+                    if "bearish" in d["tech_snapshot"]:
+                        sentiment = "bearish"
+                    elif "bullish" in d["tech_snapshot"]:
+                        sentiment = "bullish"
+
+                # 调用技术面驱动重评（非机械百分比）
+                result = reassess_strategy(
+                    code, name, price, cost, shares,
+                    current_action=d.get("action", ""),
+                    volume_signal="中性", sentiment=sentiment,
+                )
+                outputs.append(f"  📊 新策略: 损{result['stop_loss']} 盈{result['take_profit']} 区{result['entry_low']}~{result['entry_high']} RR={result['rr_ratio']}")
+                reassesed_codes.append(code)
+            except Exception as e:
+                outputs.append(f"  ⚠️ 重评失败: {e}")
+
+        # 更新买入区状态
+        if "__buy_zone" not in state.get(code, {}):
+            if code not in state:
+                state[code] = {}
+        state[code]["__buy_zone"] = in_buy_zone
+        state_updated = True
+
+    # 如果有重评过的股票，更新 decisions.json
+    if reassesed_codes and HAS_REASSESS:
+        try:
+            # 重新 regenerate_all 只针对受影响的股票效率太低
+            # 直接全量重评（regenerate_all 内部会批量拉价格、做技术分析）
+            from strategy_lifecycle import regenerate_all
+            r = regenerate_all(stdout=False)
+            outputs.append(f"  ✅ 策略已全量重评: {r.get('ok',0)}/{r.get('total',0)}成功")
+            outputs.append(f"  📌 触发股票: {', '.join(reassesed_codes)}")
+        except Exception as e:
+            outputs.append(f"  ⚠️ 全量重评失败: {e}")
+
+    # === 3.5 资金流异常检测（2026-06-27 新增）===
+    try:
+        cf = json.load(open("/home/hmo/web-dashboard/data/capital_flow_cache.json"))
+        # 检查所有 active decision 中的资金流异常
+        for d in active:
+            code = d["code"]
+            stock_cf = cf.get("stocks", {}).get(code, {})
+            analysis = stock_cf.get("analysis", {})
+            alerts = analysis.get("alerts", [])
+            if alerts:
+                name = d.get("name", code)
+                for a in alerts:
+                    outputs.append(f"  💰 {name}({code}) {a}")
+    except Exception:
+        pass
+
+    # === 第四步：情景变化检测 + 输出 → 直接推XMPP ===
+    now_str = datetime.now().strftime("%H:%M:%S")
+    elapsed = time.time() - start
+
+    # 情景变化检测（跨轮对比）
+    if HAS_TREE and _SCENARIO_CACHE.get('id'):
+        prev_scenario = state.get('_system', {}).get('last_scenario', '')
+        curr_scenario = _SCENARIO_CACHE['id']
+        if prev_scenario and curr_scenario != prev_scenario:
+            combo = _SCENARIO_CACHE.get('combo_action', '')
+            outputs.insert(0, f"🌀 情景切换: {prev_scenario}→{curr_scenario} | {combo}")
+            if outputs:
+                state.setdefault('_system', {})['last_scenario'] = curr_scenario
+                state_updated = True
+        elif not prev_scenario:
+            state.setdefault('_system', {})['last_scenario'] = curr_scenario
+            state_updated = True
+
+    if outputs:
+        # 简短一行一个触发
+        for o in outputs:
+            print(o)
+        # 推送XMPP（只推关键事件：止损跌破+情景切换+资金流异动，不推买入区进出/重评等操作细节）
+        critical = [o for o in outputs if o.startswith(("⚠️", "🌀", "💰"))]
+        if critical:
+            try:
+                body = "\n".join([f"{now_str}"] + critical)
+                payload = json.dumps({
+                    "to": "hmo@yoin.fun", "body": body, "type": "chat",
+                }).encode("utf-8")
+                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
+    # else: SILENT — 无触发，无输出，不推
+
+    if state_updated:
+        save_state(state)
+
+
+def main():
+    """每cron触发跑一轮"""
+    run_once()
+
+
+if __name__ == "__main__":
+    main()

scripts/stock_quote.py

@@ -0,0 +1,434 @@
+#!/usr/bin/env python3
+"""
+stock_quote.py — 统一股票行情查询工具（唯一权威价格源）
+
+用法：
+  python3 stock_quote.py 688411              # 单个A股
+  python3 stock_quote.py 688411 01211 300750 # 批量
+  python3 stock_quote.py --all-holdings       # 所有持仓
+
+输出：每只股票一行JSON，格式统一、无歧义。
+LLM 禁止直接解析新浪/腾讯原始CSV，只能读本脚本输出。
+
+数据来源（按优先级降序）：
+  A股: 东财push2 → 新浪hq → 腾讯qt
+  港股: 东财并行限速(5 workers) → 新浪批量 → 腾讯15min延迟(兜底)
+
+验证规则：
+  - price 必须在 [low, high] 范围内
+  - change_pct 必须与 (price - prev_close) / prev_close 一致（±0.1%容差）
+  - 任一验证失败 → 该数据源降级，尝试下个源
+"""
+
+import json, sys, re, time, urllib.request
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from threading import Semaphore
+from pathlib import Path
+from datetime import datetime
+
+UA = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"
+DATA_DIR = Path("/home/hmo/MoFin/data")
+
+# ── 工具 ──
+
+def _http_get(url, headers=None, timeout=10):
+    """带代理绕过的HTTP GET"""
+    req = urllib.request.Request(url, headers=headers or {"User-Agent": UA})
+    proxy_handler = urllib.request.ProxyHandler({})
+    opener = urllib.request.build_opener(proxy_handler)
+    try:
+        with opener.open(req, timeout=timeout) as r:
+            return r.read().decode("utf-8", errors="replace")
+    except Exception:
+        return None
+
+
+def _detect_market(code):
+    """自动识别A股/港股"""
+    code = str(code).strip()
+    if code.startswith("0") or code.startswith("3") or len(code) == 6:
+        return "ashare"
+    if len(code) <= 5:
+        return "hk"
+    return "unknown"
+
+
+def _sym(code, market):
+    """转成API可用的symbol"""
+    if market == "ashare":
+        pre = "sh" if code.startswith(("5", "6", "9")) else "sz"
+        return pre + code, code
+    return code, code  # HK codes used directly
+
+
+# ── 数据源：东方财富（A股+港股通用） ──
+
+def _em_ashare(code):
+    """东财A股API"""
+    url = f"https://push2.eastmoney.com/api/qt/stock/get?secid=1.{code}&fields=f43,f44,f45,f46,f47,f50,f57,f58,f60,f170,f169&fltt=2"
+    raw = _http_get(url, headers={"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"})
+    if not raw:
+        return None
+    try:
+        resp = json.loads(raw)
+        if resp.get("rc") != 0:
+            return None
+        d = resp.get("data")
+        if not d:
+            return None
+        price = d.get("f43")
+        if price is None or float(price) <= 0:
+            return None
+        return {
+            "code": code,
+            "name": None,  # 东财不返名称
+            "price": float(price),
+            "change_pct": float(d.get("f170", 0)) if d.get("f170") is not None else None,
+            "high": float(d.get("f44", 0)) if d.get("f44") else None,
+            "low": float(d.get("f45", 0)) if d.get("f45") else None,
+            "open": float(d.get("f46", 0)) if d.get("f46") else None,
+            "prev_close": float(d.get("f60", 0)) if d.get("f60") else None,
+            "volume": d.get("f47"),
+            "amount": d.get("f50"),
+            "source": "eastmoney",
+        }
+    except (json.JSONDecodeError, ValueError, TypeError):
+        return None
+
+
+def _em_hk(code):
+    """东财港股API（单股）"""
+    url = f"https://push2.eastmoney.com/api/qt/stock/get?secid=116.{code}&fields=f43,f44,f45,f46,f47,f50,f57,f58,f60,f170,f169&fltt=2"
+    raw = _http_get(url, headers={"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"})
+    if not raw:
+        return None
+    try:
+        resp = json.loads(raw)
+        if resp.get("rc") != 0:
+            return None
+        d = resp.get("data")
+        if not d:
+            return None
+        price = d.get("f43")
+        if price is None or float(price) <= 0:
+            return None
+        return {
+            "code": code,
+            "name": None,
+            "price": float(price),
+            "change_pct": float(d.get("f170", 0)) if d.get("f170") is not None else None,
+            "high": float(d.get("f44", 0)) if d.get("f44") else None,
+            "low": float(d.get("f45", 0)) if d.get("f45") else None,
+            "open": float(d.get("f46", 0)) if d.get("f46") else None,
+            "prev_close": float(d.get("f60", 0)) if d.get("f60") else None,
+            "volume": d.get("f47"),
+            "amount": d.get("f50"),
+            "source": "eastmoney",
+        }
+    except (json.JSONDecodeError, ValueError, TypeError):
+        return None
+
+
+# ── 数据源：新浪（A股） ──
+
+def _sina_ashare(code):
+    """新浪A股API（批量）"""
+    pre = "sh" if code.startswith(("5", "6", "9")) else "sz"
+    url = f"https://hq.sinajs.cn/list={pre}{code}"
+    raw = _http_get(url, headers={
+        "User-Agent": UA,
+        "Referer": "https://finance.sina.com.cn",
+    })
+    if not raw:
+        return None
+    try:
+        fields = raw.split("\"")[1].split(",")
+        # 新浪格式: name,open,prev_close,current,high,low,buy,sell,volume,amount
+        if len(fields) < 8:
+            return None
+        name = fields[0].strip()
+        price = float(fields[3]) if fields[3] else 0
+        prev_close = float(fields[2]) if fields[2] else 0
+        if price <= 0:
+            return None
+        change_pct = round((price - prev_close) / prev_close * 100, 2) if prev_close > 0 else None
+        return {
+            "code": code,
+            "name": name,
+            "price": price,
+            "change_pct": change_pct,
+            "high": float(fields[4]) if fields[4] else None,
+            "low": float(fields[5]) if fields[5] else None,
+            "open": float(fields[1]) if fields[1] else None,
+            "prev_close": prev_close,
+            "volume": int(fields[8]) if len(fields) > 8 and fields[8] else None,
+            "amount": float(fields[9]) if len(fields) > 9 and fields[9] else None,
+            "source": "sina",
+        }
+    except (ValueError, IndexError):
+        return None
+
+
+# ── 数据源：腾讯（A股兜底） ──
+
+def _tencent_ashare(code):
+    """腾讯A股API（批量，15min延迟兜底）"""
+    pre = "sh" if code.startswith(("5", "6", "9")) else "sz"
+    url = f"https://qt.gtimg.cn/q={pre}{code}"
+    raw = _http_get(url, headers={"User-Agent": UA})
+    if not raw:
+        return None
+    try:
+        fields = raw.split("~")
+        # Tencent格式: ~分隔
+        if len(fields) < 10:
+            return None
+        name = fields[1]
+        price = float(fields[3]) if fields[3] else 0
+        prev_close = float(fields[4]) if fields[4] else 0
+        if price <= 0:
+            return None
+        change_pct = round((price - prev_close) / prev_close * 100, 2) if prev_close > 0 else None
+        return {
+            "code": code,
+            "name": name,
+            "price": price,
+            "change_pct": change_pct,
+            "high": float(fields[33]) if len(fields) > 33 and fields[33] else None,
+            "low": float(fields[34]) if len(fields) > 34 and fields[34] else None,
+            "open": float(fields[5]) if fields[5] else None,
+            "prev_close": prev_close,
+            "volume": int(fields[6]) if fields[6] else None,
+            "amount": float(fields[37]) if len(fields) > 37 and fields[37] else None,
+            "source": "tencent",
+        }
+    except (ValueError, IndexError):
+        return None
+
+
+# ── 数据验证 ──
+
+def _validate(q):
+    """验证行情数据自洽性。返回 (is_valid, reason)"""
+    if q is None:
+        return False, "no_data"
+    if q["price"] <= 0:
+        return False, "price_zero"
+    # price 必须在 [low, high] 范围内（如果low/high存在）
+    if q.get("high") and q.get("low"):
+        if q["price"] < q["low"] or q["price"] > q["high"]:
+            return False, f"price_out_of_range: {q['price']} not in [{q['low']},{q['high']}]"
+    # change_pct一致性（如果prev_close存在）
+    if q.get("prev_close") and q["prev_close"] > 0 and q.get("change_pct") is not None:
+        expected = round((q["price"] - q["prev_close"]) / q["prev_close"] * 100, 2)
+        if abs(expected - q["change_pct"]) > 0.5:
+            return False, f"change_pct_mismatch: reported={q['change_pct']} expected={expected}"
+    return True, "ok"
+
+
+# ── 主查询逻辑 ──
+
+def get_quote(code):
+    """
+    获取单只股票行情，按优先级尝试多个数据源。
+    返回统一dict，失败返回None。
+    """
+    market = _detect_market(code)
+    result = None
+
+    if market == "ashare":
+        # A股: 东财 → 新浪 → 腾讯
+        for fetcher in [_em_ashare, _sina_ashare, _tencent_ashare]:
+            q = fetcher(code)
+            valid, reason = _validate(q)
+            if valid:
+                q["market"] = "A股"
+                if q.get("name") is None:
+                    q["name"] = _get_name_from_cache(code)
+                return q
+            if q is not None:
+                result = q  # keep last attempt
+
+    elif market == "hk":
+        for fetcher in [_em_hk, _sina_hk, _tencent_hk]:
+            q = fetcher(code)
+            valid, reason = _validate(q)
+            if valid:
+                q["market"] = "港股"
+                q["code"] = code
+                if q.get("name") is None:
+                    q["name"] = _get_name_from_cache(code)
+                return q
+            if q is not None:
+                result = q
+
+    return result
+
+
+def get_quotes_batch(codes, max_workers=5):
+    """
+    批量获取，并行执行。
+    返回 {code: quote_dict or None}
+    """
+    if max_workers == 1:
+        return {c: get_quote(c) for c in codes}
+    results = {}
+    with ThreadPoolExecutor(max_workers=max_workers) as ex:
+        fut_map = {ex.submit(get_quote, c): c for c in codes}
+        for fut in as_completed(fut_map):
+            c = fut_map[fut]
+            try:
+                results[c] = fut.result()
+            except Exception:
+                results[c] = None
+    return results
+
+
+# ── 名称缓存（从portfolio.json/decisions.json补充） ──
+
+def _get_name_from_cache(code):
+    """从本地数据文件补充股票名称（东财API不返回名称时用）"""
+    try:
+        pf = DATA_DIR / "portfolio.json"
+        if pf.exists():
+            d = json.loads(pf.read_text())
+            for h in d.get("holdings", []):
+                if str(h.get("code", "")) == str(code):
+                    return h.get("name", "")
+    except Exception:
+        pass
+    try:
+        wl = DATA_DIR / "watchlist.json"
+        if wl.exists():
+            d = json.loads(wl.read_text())
+            for item in d if isinstance(d, list) else d.get("stocks", []):
+                if str(item.get("code", "")) == str(code):
+                    return item.get("name", "")
+    except Exception:
+        pass
+    return ""
+
+
+# ── 港股备用数据源 ──
+
+def _sina_hk(code):
+    """新浪港股API（批量友好，单股也支持）"""
+    url = f"https://hq.sinajs.cn/list=hk{code}"
+    raw = _http_get(url, headers={
+        "User-Agent": UA,
+        "Referer": "https://finance.sina.com.cn",
+    })
+    if not raw:
+        return None
+    try:
+        fields = raw.split("\"")[1].split(",")
+        if len(fields) < 9:
+            return None
+        name = fields[1]
+        price = float(fields[2]) if fields[2] else 0
+        prev_close = float(fields[3]) if fields[3] else 0
+        if price <= 0:
+            return None
+        change_amt = float(fields[7]) if fields[7] else 0
+        # 更可靠的price计算
+        if prev_close > 0 and abs(change_amt) > 0:
+            price = round(prev_close + change_amt, 2)
+        change_pct = float(fields[8]) if fields[8] else 0
+        return {
+            "code": code,
+            "name": name,
+            "price": price,
+            "change_pct": change_pct,
+            "high": float(fields[5]) if fields[5] else None,
+            "low": float(fields[6]) if fields[6] else None,
+            "open": float(fields[4]) if fields[4] else None,
+            "prev_close": prev_close,
+            "volume": None,
+            "amount": None,
+            "source": "sina",
+        }
+    except (ValueError, IndexError):
+        return None
+
+
+def _tencent_hk(code):
+    """腾讯港股（15min延迟，兜底）"""
+    url = f"https://qt.gtimg.cn/q=hk{code}"
+    raw = _http_get(url, headers={"User-Agent": UA})
+    if not raw:
+        return None
+    try:
+        fields = raw.split("~")
+        if len(fields) < 10:
+            return None
+        name = fields[1]
+        price = float(fields[3]) if fields[3] else 0
+        prev_close = float(fields[4]) if fields[4] else 0
+        if price <= 0:
+            return None
+        change_pct = float(fields[7]) if len(fields) > 7 and fields[7] else 0
+        return {
+            "code": code,
+            "name": name,
+            "price": price,
+            "change_pct": change_pct,
+            "high": float(fields[33]) if len(fields) > 33 and fields[33] else None,
+            "low": float(fields[34]) if len(fields) > 34 and fields[34] else None,
+            "open": float(fields[5]) if fields[5] else None,
+            "prev_close": prev_close,
+            "volume": None,
+            "amount": None,
+            "source": "tencent",
+        }
+    except (ValueError, IndexError):
+        return None
+
+
+# ── 从portfolio.json读取所有持仓代码 ──
+
+def get_holding_codes():
+    """从portfolio.json提取所有持仓代码"""
+    try:
+        pf = DATA_DIR / "portfolio.json"
+        d = json.loads(pf.read_text())
+        return [h["code"] for h in d.get("holdings", []) if h.get("code")]
+    except Exception:
+        return []
+
+
+# ── CLI入口 ──
+
+def main():
+    if len(sys.argv) < 2:
+        print("用法: python3 stock_quote.py <code1> [code2 ...]", file=sys.stderr)
+        print("       python3 stock_quote.py --all-holdings", file=sys.stderr)
+        sys.exit(1)
+
+    codes = []
+    if sys.argv[1] == "--all-holdings":
+        codes = get_holding_codes()
+        if not codes:
+            print(json.dumps({"error": "无法读取持仓列表", "timestamp": datetime.now().isoformat()}))
+            sys.exit(1)
+    else:
+        codes = [c.strip() for c in sys.argv[1:] if c.strip()]
+
+    results = get_quotes_batch(codes)
+    timestamp = datetime.now().isoformat()
+
+    # 输出：每行一个JSON（方便批量处理）
+    for code in codes:
+        q = results.get(code)
+        if q:
+            q["fetched_at"] = timestamp
+            print(json.dumps(q, ensure_ascii=False, default=str))
+        else:
+            print(json.dumps({
+                "code": code,
+                "error": "无法获取行情",
+                "fetched_at": timestamp,
+            }, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

scripts/strategy_review.py

@@ -38,6 +38,10 @@ SIGNAL_FAILURES = {
 
 
 def fetch_price(code):
+    # DB 优先
+    try: from mofin_db import get_price_from_db; p, _ = get_price_from_db(code); return p if p else 0
+    except: pass
+    # Fallback: 腾讯 API
     try:
         prefix = "sh" if code.startswith(('60','68','51','56','50')) else "sz" if code.startswith(('00','30','15')) else "hk"
         url = f"http://qt.gtimg.cn/q={prefix}{code}"

scripts/xiaoguo_scanner.py

@@ -15,7 +15,7 @@ try:
 except ImportError:
     HAS_AKSHARE = False
 
-DATA_DIR = Path(__file__).parent / "data"
+DATA_DIR = Path("/home/hmo/MoFin/data")
 DB_PATH = DATA_DIR / "mofin.db"
 XIAOGUO_API = "http://node122:18003/v1/chat/completions"
 XIAOGUO_MODEL = "Qwen3.6-27B-MTPLX-Optimized-Speed"

scripts/xiaoguo_signal_consumer.py

@@ -30,7 +30,11 @@ def clean_proxy():
 
 
 def fetch_quote(code):
-    """拉腾讯行情，返回 dict"""
+    """拉行情。DB 优先，腾讯 fallback"""
+    # DB 优先
+    try: from mofin_db import get_price_from_db; p, chg = get_price_from_db(code); return {"name":"", "code":code, "price":p, "change_pct":chg or 0} if p else None
+    except: pass
+    # Fallback: 腾讯
     try:
         prefix = "sh" if code.startswith(('60','68','51','56','50')) else "sz" if code.startswith(('00','30','15')) else "hk"
         url = f"http://qt.gtimg.cn/q={prefix}{code}"