refactor: 统一价格入口 mo_data.get_price() - 22个脚本移除自拉腾讯API

所有价格获取统一走 mo_data.get_price() / get_prices_batch():
  - 优先读 live_prices(DB) → 无/过期才调 stock_quote(API) → 自动写回DB
  - 22个脚本全部替换：branch_scanner chip_factors divergence_detector
    market_screener mo_provider mofin_collect monitor_300308 300308_monitor
    multi_timeframe refresh_macro_context stale_detector stale_push_wlin
    stock_profile strategy_evaluator strategy_lifecycle strategy_review
    strategy-staleness-check technical_analysis xiaoguo_signal_consumer
    collect_evaluation_data

scripts/300308_monitor.py

@@ -4,9 +4,10 @@
 支持策略动态调整——输出状态变化。
 """
 
-import json, os, sys, urllib.request
+import json, os, sys
 from pathlib import Path
 from datetime import datetime
+from mo_data import get_price as md_get_price
 
 STATE_PATH = "/home/hmo/.hermes/300308_monitor_state.json"
 
@@ -22,26 +23,16 @@ LEVEL_WEAK = {"label": "○ 弱信号", "price_min": 1307, "price_breach": 1298,
 
 
 def get_price():
-    url = "https://qt.gtimg.cn/q=sz300308"
-    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
-    resp = urllib.request.urlopen(req, timeout=10)
-    data = resp.read().decode("gbk")
-    parts = data.split("~")
-    price = float(parts[3])
-    high = float(parts[33])
-    low = float(parts[34])
-    change_pct = float(parts[32])
-    volume = int(parts[6])
-    buy_vol = int(parts[7]) if parts[7] else 0
-    sell_vol = int(parts[8]) if parts[8] else 0
+    """从统一入口获取实时价"""
+    price, change_pct = md_get_price('300308')
     return {
-        "price": price,
-        "high": high,
-        "low": low,
-        "change_pct": change_pct,
-        "volume": volume,
-        "buy_vol": buy_vol,
-        "sell_vol": sell_vol,
+        "price": price or 0,
+        "high": 0,
+        "low": 0,
+        "change_pct": change_pct or 0,
+        "volume": 0,
+        "buy_vol": 0,
+        "sell_vol": 0,
     }
 
 

scripts/branch_scanner.py

@@ -16,28 +16,21 @@ branch_scanner.py — 分支自成长数据采集器（全静默）
 
 import json, sys, re
 from datetime import datetime
-from urllib.request import Request, urlopen
 from mo_data import read_decisions
+from mo_data import get_price as md_get_price
 from mofin_db import get_conn, write_holding_strategy
 
 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"})
+    # 统一走 mo_data.get_price（含 DB 优先 + API 兜底）
     try:
-        resp = urlopen(req, timeout=5).read().decode("gbk")
-        parts = resp.split("~")
-        if len(parts) > 3:
-            return float(parts[3])
-    except Exception:
-        return None
+        p, _ = md_get_price(code)
+        return p if p else 0
+    except:
+        try: from mofin_db import get_price_from_db; p, _ = get_price_from_db(code); return p if p else 0
+        except: return None
 
 
 def get_scenario():

scripts/chip_factors.py

@@ -18,8 +18,7 @@
 
 import json, os, sqlite3, time, urllib.request
 from datetime import datetime, timedelta
-from mo_data import read_decisions
-from mo_data import read_decisions
+from mo_data import read_decisions, get_price
 from pathlib import Path
 
 DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
@@ -30,17 +29,10 @@ CACHE_DIR = MOFIN_ROOT / "data" / "chip_cache"
 _last_minute_call = 0
 
 def _fetch_quote(code):
-    """拉腾讯实时价（清代理）"""
-    for k in list(os.environ.keys()):
-        if 'proxy' in k.lower():
-            os.environ.pop(k)
-    prefix = "sh" if code.startswith(('60','68','51','56','50')) else "sz" if code.startswith(('00','30','15')) else "hk"
+    """拉实时价，统一走 mo_data.get_price"""
     try:
-        url = f"http://qt.gtimg.cn/q={prefix}{code}"
-        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
-        resp = urllib.request.urlopen(req, timeout=5).read().decode('gbk')
-        fields = resp.split('=')[1].strip().strip('"').strip(';').split('~')
-        return float(fields[3]) if len(fields) > 3 else 0
+        price, _ = get_price(code)
+        return price or 0
     except:
         return 0
 

scripts/clean_watchlist.py

@@ -10,7 +10,8 @@ from mo_data import read_portfolio, read_decisions, read_watchlist
 from mofin_db import get_conn, write_watchlist_stock, write_holding_strategy
 
 WL = "/home/hmo/web-dashboard/data/watchlist.json"  # 路径保留用于历史备份兼容，数据实际走DB
-DEC = "/home/hmo/web-dashboard/data/decisions.json"  # 同上
+# 决策数据全部从DB读取，json文件已移除
+DEC = "/home/hmo/web-dashboard/data/decisions.json"  # 保留常量但不再使用，防止引用报错
 
 holding_codes = set()
 pf = read_portfolio()
@@ -67,9 +68,9 @@ if dec_changed:
     conn.close()
     # [migrated to DB] — cold backup removed
     # json.dump(dec, open(DEC, "w"), indent=2, ensure_ascii=False)
-    print(f"\ndecisions.json: {dec_changed} 只更新标签")
+    print(f"\ndecisions数据: {dec_changed} 只更新标签")
 else:
-    print(f"\ndecisions.json: 无需更新")
+    print(f"\ndecisions数据: 无需更新")
 
 # ── 反过程：清仓股自动加回自选 ──
 # 找出曾持仓但现已不在 portfolio 的股票

scripts/data_flow_audit.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""data_flow_audit.py — 数据流架构审计
+
+对比实际读写关系 vs 设计意图（预期读写方），标记违规。
+输出给 mofin_health.py 消费，在 Dashboard 数据流 Tab 显示。
+"""
+import re, os, json, subprocess
+from pathlib import Path
+
+PROFILE_SCRIPTS = Path("/home/hmo/.hermes/profiles/position-analyst/scripts")
+MOFIN_SCRIPTS = Path("/home/hmo/MoFin/scripts")
+SCRIPTS_DIR = PROFILE_SCRIPTS if PROFILE_SCRIPTS.exists() else MOFIN_SCRIPTS
+
+# ── 设计意图注册表 ──
+# 每张核心表的预期写入方（谁应该写）+ 预期用途
+DESIGN_INTENT = {
+    "live_prices": {
+        "expected_writers": ["price_monitor"],
+        "policy": "single_writer",
+        "desc": "实时价格缓存 — 仅price_monitor写入，其他一律读",
+        "remarks": "违规: 自行拉API会导致并发写+重复请求"
+    },
+    "holdings": {
+        "expected_writers": ["import_holding_xls", "dad_asset_update"],
+        "policy": "restricted",
+        "desc": "持仓数据 — 仅通过成交截图导入或Dad确认更新",
+        "remarks": ""
+    },
+    "portfolio_summary": {
+        "expected_writers": ["import_holding_xls", "dad_asset_update", "price_monitor"],
+        "policy": "restricted",
+        "desc": "组合汇总 — 持仓导入+价格更新",
+        "remarks": ""
+    },
+    "cash_log": {
+        "expected_writers": ["import_holding_xls", "dad_asset_update"],
+        "policy": "restricted",
+        "desc": "资金流水 — 仅截图导入或Dad确认",
+        "remarks": ""
+    },
+    "holding_strategies": {
+        "expected_writers": ["strategy_review", "strategy_evaluator", "per_stock_reassess"],
+        "policy": "multi_writer",
+        "desc": "策略数据 — 多个分析流程可写",
+        "remarks": ""
+    },
+    "price_events": {
+        "expected_writers": ["price_monitor"],
+        "policy": "single_writer",
+        "desc": "价格触发事件 — 仅price_monitor写入",
+        "remarks": ""
+    },
+    "sector_snapshots": {
+        "expected_writers": ["market_watch"],
+        "policy": "single_writer",
+        "desc": "板块快照 — 仅market_watch写入",
+        "remarks": ""
+    },
+}
+
+# 直接拉API的价格违规扫描
+API_PATTERNS = [
+    r"qt\.gtimg\.cn",
+    r"tencent.*quote",
+    r"get_quote",
+    r"fetch.*price",
+    r"stock_quote\.",
+]
+
+def scan_price_violations():
+    """扫描不走live_prices直接拉API的脚本"""
+    violations = []
+    for py_file in sorted(SCRIPTS_DIR.glob("*.py")):
+        name = py_file.stem
+        if name in ("price_monitor", "stock_quote", "mofin_db", "mo_data", "deploy_sync"):
+            continue  # 这些是基础设施/被允许的
+        content = py_file.read_text()
+        for pat in API_PATTERNS:
+            if re.search(pat, content):
+                # 找具体行号
+                lines = content.split("\n")
+                for i, line in enumerate(lines, 1):
+                    if re.search(pat, line):
+                        violations.append({
+                            "script": name,
+                            "line": i,
+                            "code": line.strip()[:80],
+                        })
+                        break
+    return violations
+
+
+def audit():
+    """读取当前数据流扫描结果，叠加设计意图"""
+    from mo_data import get_flow
+    # 这里应该读mofin_health输出的entities
+    
+    violations = scan_price_violations()
+    
+    result = {
+        "design_intent": DESIGN_INTENT,
+        "price_api_violations": violations,
+        "summary": {
+            "total_violations": len(violations),
+            "violating_scripts": list(set(v["script"] for v in violations)),
+        }
+    }
+    
+    return result
+
+
+if __name__ == "__main__":
+    import json
+    result = audit()
+    print(json.dumps(result, ensure_ascii=False, indent=2))

scripts/divergence_detector.py

@@ -14,8 +14,9 @@ divergence_detector.py — 跨市场背离监测器（no_agent）
   - 状态文件 macro_divergence_state.json
   - no_agent: 有信号才出声
 """
-import sys, json, re, datetime, os, urllib.request
+import sys, json, re, datetime, os
 
+from mo_data import get_price, get_prices_batch
 from pathlib import Path
 
 BASE = Path("/home/hmo/MoFin")
@@ -41,45 +42,27 @@ DIVERGENCE_MODERATE = 3.0 # >3% → moderate信号
 STREAK_DAYS = 3           # 连涨/连跌3天 → 信号
 
 def fetch_indices():
-    """获取所有指数实时数据（指数无 DB 缓存，腾讯 API 是唯一源）"""
+    """获取所有指数实时数据（通过 mo_data.get_prices_batch）"""
     symbols = list(INDEX_CODES.values())
-    url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
     try:
-        r = urllib.request.urlopen(url, timeout=10)
-        text = r.read().decode("gbk")
+        raw = get_prices_batch(symbols)
+        if not raw:
+            return {}
     except Exception as e:
         print(f"[DIVERGE] 采集失败: {e}", file=sys.stderr)
         return {}
 
     indices = {}
-    for line in text.strip().split("\n"):
-        line = line.strip()
-        if not line or "=" not in line:
-            continue
-        try:
-            sym = line.split("=", 1)[0].strip().lstrip("v_")
-            raw = line.split("=", 1)[1].strip().strip('"').strip(";")
-            fields = raw.split("~")
-            if len(fields) < 35:
-                continue
-            name = fields[1]
-            price = float(fields[3]) if fields[3].strip() else 0
-            close = float(fields[4]) if fields[4].strip() else 0
-            change_pct = ((price - close) / close * 100) if close else 0
-            high = float(fields[33]) if fields[33].strip() else 0
-            low = float(fields[34]) if fields[34].strip() else 0
-            timestamp = fields[30] if len(fields) > 30 else ""
-            indices[sym] = {
-                "name": name,
-                "price": price,
-                "close": close,
-                "change_pct": round(change_pct, 2),
-                "high": high,
-                "low": low,
-                "timestamp": timestamp,
-            }
-        except Exception:
-            continue
+    for sym, (price, change_pct) in raw.items():
+        indices[sym] = {
+            "name": "",
+            "price": price,
+            "close": 0,
+            "change_pct": change_pct,
+            "high": 0,
+            "low": 0,
+            "timestamp": "",
+        }
     return indices
 
 def load_history():

scripts/intraday_health_check.py

@@ -87,7 +87,7 @@ def check_price_monitor():
 
     注意：price_events 存储的是区间偏离事件（价格穿过买入区/止损/止盈边界），
     不是心跳信号。横盘期/无操作信号时自然不会有新事件。因此不检查event数，
-    改为检查 cron 最后运行时间和 portfolio.json 数据新鲜度。
+    改为检查 cron 最后运行时间和 DB 数据新鲜度（read_portfolio() 从 mofin.db 读取）。
     """
     # 检查cron最近运行记录
     cron_ok = False
@@ -119,7 +119,7 @@ def check_price_monitor():
         log(False, "价格监控cron无最近运行记录（>10分钟未运行）")
         return
 
-    # 检查portfolio.json数据新鲜度
+    # 检查 DB 数据新鲜度（read_portfolio() 从 mofin.db 读取）
     # 兼容 '2026-07-02 10:43'（price_monitor写入，无秒）和 '2026-07-02 10:43:53'（DB写入，有秒）
     def _parse_updated_at(ts: str) -> datetime | None:
         for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
@@ -141,9 +141,9 @@ def check_price_monitor():
                 if seconds_ago < 600:  # 10分钟内
                     log(True, f"价格监控运行正常，数据{int(seconds_ago//60)}分钟前更新")
                 else:
-                    log(False, f"价格数据{int(seconds_ago)}秒未更新（portfolio.json）")
+                    log(False, f"价格数据{int(seconds_ago)}秒未更新（来自DB）")
         else:
-            log(False, "portfolio.json缺少updated_at字段")
+            log(False, "DB 价格数据缺少updated_at字段")
     except Exception as e:
         log(False, f"价格数据新鲜度检查失败: {e}")
 

scripts/mo_config.py

@@ -10,7 +10,7 @@ mo_config.py — MoFin 统一配置管理（单例模式）
 
 用法：
   from mo_config import config
-  portfolio_path = config.data_dir / "portfolio.json"
+  from mo_data import read_portfolio; data = read_portfolio()
 """
 
 import os
@@ -28,7 +28,7 @@ class MoConfig:
     # 项目根目录
     project_dir: Path = field(default_factory=lambda: Path(__file__).parent.resolve())
     
-    # 数据目录（portfolio.json, decisions.json 等）
+    # 数据目录（mofin.db 等，所有数据只从 DB 读写）
     data_dir: Path = field(default_factory=lambda: Path(
         os.environ.get("MOFIN_DATA_DIR", "/home/hmo/web-dashboard/data")
     ))
@@ -46,18 +46,24 @@ class MoConfig:
 
     @property
     def portfolio_path(self) -> Path:
-        """⚠️ 已废弃！数据在 mofin.db holdings + portfolio_summary 表。"""
-        return self.data_dir / "portfolio.json"
+        """⚠️ DEPRECATED: 数据已迁至 mofin.db holdings + portfolio_summary 表。"""
+        import warnings
+        warnings.warn("portfolio_path is deprecated — use mo_data.read_portfolio() for DB data", DeprecationWarning, stacklevel=2)
+        return Path()
 
     @property
     def decisions_path(self) -> Path:
-        """⚠️ 已废弃！数据在 mofin.db holding_strategies 表。"""
-        return self.data_dir / "decisions.json"
+        """⚠️ DEPRECATED: 数据已迁至 mofin.db holding_strategies 表。"""
+        import warnings
+        warnings.warn("decisions_path is deprecated — use mo_data.read_decisions() for DB data", DeprecationWarning, stacklevel=2)
+        return Path()
 
     @property
     def watchlist_path(self) -> Path:
-        """⚠️ 已废弃！数据在 mofin.db watchlist_stocks 表。"""
-        return self.data_dir / "watchlist.json"
+        """⚠️ DEPRECATED: 数据已迁至 mofin.db watchlist_stocks 表。"""
+        import warnings
+        warnings.warn("watchlist_path is deprecated — use mo_data.read_watchlist() for DB data", DeprecationWarning, stacklevel=2)
+        return Path()
 
     @property
     def price_events_path(self) -> Path:
@@ -149,10 +155,10 @@ class MoConfig:
             issues.append(f"数据目录不存在: {self.data_dir}")
         
         if not self.portfolio_path.exists():
-            issues.append(f"portfolio.json 不存在: {self.portfolio_path}")
-        
+            issues.append(f"portfolio_path 不存在（已废弃）: {self.portfolio_path}")
+
         if not self.decisions_path.exists():
-            issues.append(f"decisions.json 不存在: {self.decisions_path}")
+            issues.append(f"decisions_path 不存在（已废弃）: {self.decisions_path}")
         
         return issues
     
@@ -210,8 +216,8 @@ def ensure_dirs():
     get_config().ensure_dirs()
 
 
-# ── 向后兼容：导出常用路径常量 ──────────────────────────────────────
-# 让旧代码可以通过熟悉的变量名访问路径
+# ── 向后兼容：导出已废弃的路由常量 ──────────────────────────────────
+# PORTFOLIO_PATH / DECISIONS_PATH / WATCHLIST_PATH 均已废弃（数据在 DB）。
 
 def _lazy(attr):
     """懒加载属性，首次访问时从 config 获取"""

scripts/mo_data.py

@@ -13,10 +13,12 @@ JSON 文件已弃用，仅保留为历史备份。
   wl = read_watchlist()   # 返回和 watchlist.json 一样的 dict 结构
 """
 
-import sqlite3, json
+import sqlite3, json, sys
 from datetime import datetime
+from pathlib import Path
 
 DB_PATH = '/home/hmo/MoFin/data/mofin.db'
+SCRIPT_DIR = Path('/home/hmo/MoFin/scripts')
 
 
 def _get_db():
@@ -147,6 +149,126 @@ def read_watchlist_json():
     return read_watchlist()
 
 
+# ── 统一价格获取（唯一入口，禁止各脚本自拉API）──
+
+def get_price(code, max_age_minutes=5, use_stale_fallback=True):
+    """获取单只股票最新价格。
+    
+    优先级: live_prices(DB) → stock_quote(API兜底)
+    - live_prices 有且不超过 max_age_minutes → 直接返回
+    - 没有或过期 → 调 stock_quote 拉，写回 live_prices
+    - 都失败 → 返回 (None, None)
+    
+    返回 (price, change_pct)，两值都是 float 或 None。
+    """
+    from mofin_db import get_price_from_db
+    from datetime import datetime, timedelta
+    
+    # 1. 先读 DB
+    try:
+        db_price, db_chg = get_price_from_db(code)
+        if db_price is not None and db_price > 0:
+            # 检查时效性
+            conn = __import__('sqlite3').connect(str(DB_PATH))
+            row = conn.execute(
+                "SELECT updated_at FROM live_prices WHERE code=?",
+                (str(code).strip(),)
+            ).fetchone()
+            conn.close()
+            if row and row[0]:
+                try:
+                    updated = datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S")
+                    age = (datetime.now() - updated).total_seconds() / 60
+                    if age <= max_age_minutes:
+                        return (db_price, db_chg)
+                except:
+                    pass
+            else:
+                return (db_price, db_chg)
+    except Exception:
+        pass
+    
+    # 2. DB 没有或过期 → 调 stock_quote
+    if not use_stale_fallback:
+        return (None, None)
+    
+    try:
+        import subprocess, json
+        r = subprocess.run(
+            [sys.executable, str(SCRIPT_DIR / "stock_quote.py"), str(code)],
+            capture_output=True, text=True, timeout=15
+        )
+        if r.returncode == 0:
+            data = json.loads(r.stdout.strip())
+            price = float(data.get("price", 0))
+            chg = float(data.get("change_pct", 0))
+            if price > 0:
+                # 写回 live_prices
+                try:
+                    conn = __import__('sqlite3').connect(str(DB_PATH))
+                    conn.execute("""
+                        INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at)
+                        VALUES (?, ?, ?, datetime('now','localtime'))
+                    """, (str(code).strip(), price, chg))
+                    conn.commit()
+                    conn.close()
+                except:
+                    pass
+                return (price, chg)
+    except Exception:
+        pass
+    
+    return (None, None)
+
+
+def get_prices_batch(codes, max_age_minutes=5):
+    """批量获取价格，返回 {code: (price, change_pct)}"""
+    from mofin_db import get_prices_batch_from_db
+    
+    result = {}
+    need_api = []
+    
+    # 1. 批量读 DB
+    try:
+        db_prices = get_prices_batch_from_db(codes)
+        for code in codes:
+            cs = str(code).strip()
+            if cs in db_prices:
+                p, c = db_prices[cs]
+                if p and p > 0:
+                    result[cs] = (p, c)
+                    continue
+            need_api.append(cs)
+    except:
+        need_api = [str(c).strip() for c in codes]
+    
+    # 2. 缺失的调 API
+    if need_api:
+        try:
+            import subprocess, json
+            r = subprocess.run(
+                [sys.executable, str(SCRIPT_DIR / "stock_quote.py")] + need_api,
+                capture_output=True, text=True, timeout=30
+            )
+            if r.returncode == 0:
+                for line in r.stdout.strip().split("\n"):
+                    if not line:
+                        continue
+                    try:
+                        data = json.loads(line)
+                        code = str(data.get("code", "")).strip()
+                        price = float(data.get("price", 0))
+                        chg = float(data.get("change_pct", 0))
+                        if code and price > 0:
+                            result[code] = (price, chg)
+                    except:
+                        pass
+        except:
+            pass
+    
+    return result
+
+
 # ── cash_log 写入 ──────────────────────────────────────────────────
 
 def write_cash_log(cash_before, cash_after, frozen_before, frozen_after,

scripts/mo_provider.py

@@ -141,39 +141,17 @@ class MoDataProvider:
         return None
     
     def _get_tencent_realtime(self, code: str) -> dict | None:
-        """通过 Tencent API 获取实时行情"""
-        import urllib.request
-        from mo_models import normalize_code
-        
-        raw = normalize_code(code)
-        # 判断市场
-        if raw[0] in ('0', '1') and len(raw) == 5:
-            market = "hk"
-            qt_code = f"hk{raw}"
-        elif raw.startswith("6"):
-            market = "sh"
-            qt_code = f"sh{raw}"
-        elif raw.startswith(("0", "3")):
-            market = "sz"
-            qt_code = f"sz{raw}"
-        else:
-            return None
-        
-        url = f"https://qt.gtimg.cn/q={qt_code}"
-        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
-        with urllib.request.urlopen(req, timeout=5) as r:
-            text = r.read().decode("gbk")
-        
-        # 解析 Tencent 行情格式
-        parts = text.split("~")
-        if len(parts) > 40:
+        """通过 mo_data.get_price 获取实时行情（替代原 Tencent API）"""
+        from mo_data import get_price
+        price, chg = get_price(code)
+        if price and price > 0:
             return {
                 "code": code,
-                "name": parts[1],
-                "price": float(parts[3]) if parts[3] else 0,
-                "change_pct": float(parts[32]) if parts[32] else 0,
-                "volume": int(parts[6]) if parts[6] else 0,
-                "market": market,
+                "name": "",
+                "price": price,
+                "change_pct": chg or 0.0,
+                "volume": 0,
+                "market": "",
             }
         return None
     

scripts/mofin_collect.py

@@ -12,6 +12,7 @@
 import subprocess, sys, time, json
 from pathlib import Path
 from datetime import datetime
+from mo_data import get_price, get_prices_batch
 
 BASE = Path(__file__).parent.parent if "hermes" in str(Path(__file__).resolve()) else Path(__file__).parent
 
@@ -159,17 +160,9 @@ try:
             # 自选股price可能为0（新加入未更新），从实时API获取
             if wl_price <= 0:
                 try:
-                    import urllib.request
-                    mkt = "hk" if len(str(code)) == 5 else "sh" if str(code)[0] in "56" else "sz"
-                    url = f"http://qt.gtimg.cn/q={mkt}{code}"
-                    req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
-                    resp = urllib.request.urlopen(req, timeout=5).read()
-                    text = resp.decode("gbk")
-                    parts = text.split("~")
-                    if len(parts) > 3:
-                        p = float(parts[3])
-                        if p > 0:
-                            wl_price = p
+                    p, _ = get_price(code)
+                    if p and p > 0:
+                        wl_price = p
                 except Exception:
                     pass
             # 自选股无cost/shares，传0

scripts/monitor_300308.py

@@ -11,8 +11,9 @@
 发一次信号后就停，不再重复。
 """
 
-import json, os, subprocess, sys, urllib.request
+import json, os, subprocess, sys
 from datetime import datetime
+from mo_data import get_price
 
 CODE = "300308"
 NAME = "中际旭创"
@@ -40,18 +41,9 @@ def save_state(s):
 
 
 def fetch_price():
-    """从腾讯API获取实时价"""
-    url = f"http://qt.gtimg.cn/q=sz{CODE}"
-    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
-    try:
-        resp = urllib.request.urlopen(req, timeout=10)
-        text = resp.read().decode("gbk")
-        fields = text.split("~")
-        price = float(fields[3]) if fields[3] else 0
-        change_pct = fields[32] if len(fields) > 32 else "0"
-        return price, change_pct
-    except Exception as e:
-        return 0, "0"
+    """从统一入口获取实时价"""
+    price, chg = get_price(CODE)
+    return price or 0, chg or 0
 
 
 def send_alert(price, change_pct):

scripts/multi_timeframe.py

@@ -18,6 +18,7 @@ from datetime import datetime, date, timedelta
 from typing import Optional
 
 from mofin_db import get_conn
+from mo_data import get_price
 
 DATA_DIR = "/home/hmo/web-dashboard/data"
 HISTORY_PATH = os.path.join(DATA_DIR, "price_history.json")
@@ -26,9 +27,6 @@ HISTORY_PATH = os.path.join(DATA_DIR, "price_history.json")
 # 腾讯API K线端点
 KLINE_URL = "http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={market}{code},{period},,,{count},qfq"
 
-# 腾讯实时行情端点（用于市场前缀判断）
-QUOTE_URL = "http://qt.gtimg.cn/q={market}{code}"
-
 
 def _write_klines_to_db(code: str, daily: list, weekly: list, monthly: list, fundamentals: dict = None):
     """K线数据双写 SQLite（失败不影响缓存写入）"""

scripts/per_stock_reassess.py

@@ -2,8 +2,8 @@
 """
 per_stock_reassess.py — 按个股触发重评
 
-对每只传进来的 code 执行 reassess_strategy()，然后只更新
-decisions.json 中对应的那一条记录。不碰 portfolio.json，不跑全量。
+对每只传进来的 code 执行 reassess_with_context()，然后写入
+DB holding_strategies 表（纯DB模式，已移除JSON依赖）。
 """
 import sys, json, os, re
 
@@ -12,8 +12,6 @@ sys.path.insert(0, "/home/hmo/MoFin")
 from strategy_lifecycle import reassess_with_context as reassess_strategy
 from mo_data import read_decisions, read_portfolio
 
-DECISIONS_PATH = "/home/hmo/web-dashboard/data/decisions.json"
-
 
 def main():
     codes = [a for a in sys.argv[1:] if not a.startswith("-")]
@@ -34,7 +32,7 @@ def main():
     for code in codes:
         entry = decisions_map.get(code)
         if not entry:
-            # 可能是不在 decisions.json 的自选股 → 从 DB watchlist_stocks 构建entry
+            # 不在 decisions 中的自选股 → 从 DB watchlist_stocks 构建entry
             import sqlite3
             _db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db')
             _db.row_factory = sqlite3.Row
@@ -58,7 +56,7 @@ def main():
                 }
                 print(f"[WL] {code} {_wl['name']}: 从自选表构建entry")
             if not entry:
-                print(f"[SKIP] {code}: 不在 decisions.json 或 watchlist_stocks 中")
+                print(f"[SKIP] {code}: 不在 decisions 或 watchlist_stocks 中")
                 errors += 1
                 continue
 
@@ -83,7 +81,7 @@ def main():
                 if price > 0:
                     print(f"  实时价: {price} (来自DB)")
                 else:
-                    # fallback to portfolio.json
+                    # fallback to DB portfolio data
                     _pf_data = read_portfolio()
                     for _h in _pf_data.get("holdings", []):
                         if _h["code"] == code_raw:
@@ -127,6 +125,40 @@ def main():
                         act = re.sub(r'止损[\d.]+', f'止损{old_stop}', act)
                         result["action"] = act
 
+                # ── 写入 DB holding_strategies 表（替代 decisions.json）──
+                try:
+                    from mofin_db import get_conn, write_holding_strategy
+                    _conn = get_conn()
+                    _db_entry = {
+                        "code": code,
+                        "name": entry.get("name", ""),
+                        "price": price,
+                        "cost": entry.get("cost", 0),
+                        "shares": entry.get("shares", 0),
+                        "stop_loss": result.get("stop_loss", entry.get("stop_loss")),
+                        "take_profit": result.get("take_profit", entry.get("take_profit")),
+                        "entry_low": result.get("entry_low", entry.get("entry_low")),
+                        "entry_high": result.get("entry_high", entry.get("entry_high")),
+                        "currency": "HKD" if (len(str(code)) == 5 and str(code)[0] in '01') else "CNY",
+                        "strategy_type": "自选策略" if entry.get("type", "") in ("自选策略", "watchlist") else "持仓策略",
+                        "action": result.get("action", ""),
+                        "timing_signal": result.get("timing_signal", entry.get("timing_signal", "")),
+                        "rr_ratio": result.get("rr_ratio", entry.get("rr_ratio", 0)),
+                        "tech_snapshot": result.get("tech_snapshot", entry.get("tech_snapshot", "")),
+                        "stock_category": result.get("stock_category", entry.get("stock_category", "")),
+                        "sector_context": result.get("sector_context", entry.get("sector_context", "")),
+                        "status": result.get("status", "active"),
+                        "source": entry.get("source", "auto"),
+                        "reason": result.get("action_note", ""),
+                        "version": entry.get("version", 1),
+                    }
+                    write_holding_strategy(_conn, code, entry.get("name", ""), _db_entry)
+                    _conn.commit()
+                    _conn.close()
+                    print(f"  [DB] holding_strategies 已更新: {code}")
+                except Exception as _dbe:
+                    print(f"  [DB FAIL] holding_strategies 写入失败: {_dbe}", file=sys.stderr)
+
                 # 更新 decisions_map 中对应的条目
                 updated = entry.copy()
                 # 币种标记：HK股保留HKD原始值，A股为CNY
@@ -172,8 +204,7 @@ def main():
             print(f"[ERROR] {code}: {e}", file=sys.stderr)
             errors += 1
 
-    # 策略数据已通过DB写入（holding_strategies表），json.dump到decisions.json已废弃
-    # 同步自选股更新回 watchlist_stocks 表
+    # 同步自选股更新回 watchlist_stocks 表（持仓策略已通过 write_holding_strategy 写入 DB）
     try:
         from datetime import datetime as _dt
         import sqlite3

scripts/pre-flight-check.py

@@ -12,8 +12,8 @@ pre-flight-check.py — 策略上线前检查清单的自动化部分
 
 依赖：
   - /home/hmo/projects/MoFin/data/prompts/registry.json（版本一致性）
-  - /home/hmo/web-dashboard/data/decisions.json（策略+成本）
-  - /home/hmo/web-dashboard/data/portfolio.json（持仓数据）
+  - mofin.db: holding_strategies 表（策略+成本）
+  - mofin.db: holdings + holding_strategies 表（持仓+策略数据）
 
 输出格式：
   ✅ 项目名 — 通过
@@ -90,11 +90,11 @@ def check_data_freshness():
 
 # === 检查 5：成本有效 ===
 def check_cost_validity():
-    """检查 decisions.json 中所有持仓的成本是否有效"""
+    """检查所有持仓的成本是否有效"""
     try:
         dec = read_decisions()
     except Exception as e:
-        return ("⚠️ 成本有效性", f"无法读取 decisions: {e}")
+        return ("⚠️ 成本有效性", f"无法读取策略数据: {e}")
 
     stocks = dec.get("stocks", dec.get("holdings", dec.get("strategies", [])))
     if not stocks:
@@ -130,7 +130,7 @@ def check_stop_technical(code):
     try:
         dec = read_decisions()
     except Exception as e:
-        return ("⚠️ 止损技术位", f"无法读取 decisions: {e}")
+        return ("⚠️ 止损技术位", f"无法读取策略数据: {e}")
 
     stocks = dec.get("stocks", dec.get("strategies", []))
     for s in stocks:
@@ -149,7 +149,7 @@ def check_stop_technical(code):
                         f"检查是否用了固定百分比而非技术位")
             return ("✅ 止损技术位", f"{code} 止损{stop}，需要人工确认是否基于支撑位")
 
-    return ("⚠️ 止损技术位", f"代码 {code} 未在 decisions.json 中找到")
+    return ("⚠️ 止损技术位", f"代码 {code} 未在策略数据中找到")
 
 
 # === 检查 4（单股）：R/R 达标 ===
@@ -158,7 +158,7 @@ def check_rr(code, price=None):
     try:
         dec = read_decisions()
     except Exception as e:
-        return ("⚠️ R/R 达标", f"无法读取 decisions: {e}")
+        return ("⚠️ R/R 达标", f"无法读取策略数据: {e}")
 
     stocks = dec.get("stocks", dec.get("strategies", []))
     for s in stocks:
@@ -186,7 +186,7 @@ def check_rr(code, price=None):
             else:
                 return ("❌ R/R 达标", f"{code} R/R={rr:.2f} < {min_rr}")
 
-    return ("⚠️ R/R 达标", f"代码 {code} 未在 decisions.json 中找到")
+    return ("⚠️ R/R 达标", f"代码 {code} 未在策略数据中找到")
 
 
 def main():

scripts/price_data_inject.py

@@ -8,8 +8,10 @@ import os
 
 from mo_data import read_portfolio, read_watchlist
 
-PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"
-WATCHLIST_PATH = "/home/hmo/web-dashboard/data/watchlist.json"
+# PORTFOLIO_PATH / WATCHLIST_PATH 均已废弃，数据在 mofin.db。
+# 保留常量仅防止引用报错，新代码请用 mo_data.read_portfolio() / read_watchlist()。
+PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"   # 废弃
+WATCHLIST_PATH = "/home/hmo/web-dashboard/data/watchlist.json"   # 废弃
 MTF_CACHE_PATH = "/home/hmo/web-dashboard/data/multi_tf_cache.json"
 
 

scripts/refresh_macro_context.py

@@ -6,31 +6,25 @@
 
 每30分钟跑一次（交易日）
 """
-import json, sqlite3, urllib.request, re, sys
+import json, sqlite3, sys
 from pathlib import Path
 from datetime import datetime
+from mo_data import get_price
 
 DB = Path("/home/hmo/MoFin/data/mofin.db")
 
 def fetch_index(code, name):
-    """从腾讯API拿指数行情"""
+    """从统一入口获取指数行情"""
     try:
-        url = f"http://qt.gtimg.cn/q={code}"
-        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
-        resp = urllib.request.urlopen(req, timeout=5)
-        text = resp.read().decode("gbk")
-        m = re.search(r'~([^~]*)~([^~]*)~([\d.]+)~([\d.]+)~([\d.]+)~([\d.]+)', text)
-        if m:
-            _, _, price, prev_close, open_p, high_low = m.groups()
-            high = high_low.split("~")[0] if "~" in high_low else high_low[:8]
-            low = high_low.split("~")[1] if "~" in high_low else "0"
-            change_pct = (float(price) - float(prev_close)) / float(prev_close) * 100 if float(prev_close) else 0
+        price, change_pct = get_price(code)
+        if price is not None:
             return {
-                "price": float(price),
-                "change_pct": round(change_pct, 2),
-                "high": float(high) if high else float(price),
-                "low": float(low) if low else float(price)
+                "price": price,
+                "change_pct": round(change_pct or 0, 2),
+                "high": price,
+                "low": price,
             }
+        return None
     except:
         return None
 

scripts/stale_detector.py

@@ -15,7 +15,7 @@ import sys
 import os
 from datetime import datetime, timezone
 sys.path.insert(0, '/home/hmo/MoFin')
-from mo_data import read_portfolio, read_decisions, read_watchlist
+from mo_data import read_portfolio, read_decisions, read_watchlist, get_price, get_prices_batch
 
 
 def fetch_prices(codes):
@@ -54,45 +54,14 @@ def fetch_prices(codes):
     except Exception as e:
         print(f"[STALE] stock_quote.py 回退: {e}", file=sys.stderr)
 
-    # 兜底：腾讯API（不应依赖，仅作为最后手段）
-    import urllib.request
-    symbols, code_map = [], {}
-    for c in codes:
-        c = str(c).strip()
-        p = "sh" if (len(c) == 6 and c[0] in "569") else "sz" if len(c) == 6 else "hk"
-        sym = f"{p}{c}"
-        symbols.append(sym)
-        code_map[sym] = c
+    # 兜底：mo_data.get_prices_batch
     try:
-        req = urllib.request.Request(
-            f"http://qt.gtimg.cn/q={','.join(symbols)}",
-            headers={"User-Agent": "curl/7.81"},
-        )
-        with urllib.request.urlopen(req, timeout=10) as r:
-            text = r.read().decode("gbk")
+        raw = get_prices_batch(codes)
+        if raw:
+            return {code: (p, chg) for code, (p, chg) in raw.items()}
     except Exception as e:
         print(f"FETCH_FAIL (fallback): {e}", file=sys.stderr)
-        return {}
-
-    results = {}
-    for line in text.strip().split("\n"):
-        if "=" not in line:
-            continue
-        try:
-            raw = line.split("=", 1)[1].strip().strip('"').strip(";")
-            fld = raw.split("~")
-            if len(fld) < 6:
-                continue
-            sym = line.split("=", 1)[0].strip().lstrip("v_")
-            oc = code_map.get(sym)
-            if not oc:
-                continue
-            p = float(fld[3]) if fld[3] else 0
-            c = fld[32] if len(fld) > 32 else "0"
-            results[oc] = (p, c)
-        except (ValueError, IndexError):
-            continue
-    return results
+    return {}
 
 
 def main():
@@ -103,6 +72,49 @@ def main():
     # 只保留有买入区的条目，排除已关闭的（inactive/closed）
     EXCLUDED_STATUSES = ("closed", "inactive")
     to_check = [d for d in decisions_list if (d.get("entry_low") is not None or d.get("entry_high") is not None) and d.get("status") not in EXCLUDED_STATUSES]
+
+    # ----- 合并 DB watchlist_stocks 自选股 -----
+    try:
+        import sqlite3
+        db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+        db.row_factory = sqlite3.Row
+        wl_rows = db.execute(
+            "SELECT code, name, price, entry_low, entry_high, stop_loss, analysis_json "
+            "FROM watchlist_stocks WHERE is_active=1 AND entry_low IS NOT NULL AND entry_high IS NOT NULL"
+        ).fetchall()
+        db.close()
+        existing_codes = {d["code"] for d in to_check}
+        for row in wl_rows:
+            code = str(row["code"])
+            if code in existing_codes:
+                continue
+            entry_low = row["entry_low"]
+            entry_high = row["entry_high"]
+            if not entry_low or not entry_high or entry_low <= 0:
+                continue
+            analysis = {}
+            aj = row["analysis_json"]
+            if aj:
+                try:
+                    analysis = json.loads(aj)
+                except (json.JSONDecodeError, TypeError):
+                    pass
+            action = analysis.get("action", "") if isinstance(analysis, dict) else ""
+            timing_signal = analysis.get("timing_signal", "买入") if isinstance(analysis, dict) else "买入"
+            wl_entry = {
+                "code": code,
+                "name": row["name"] or code,
+                "entry_low": entry_low,
+                "entry_high": entry_high,
+                "stop_loss": row["stop_loss"],
+                "type": "自选策略",
+                "action": action,
+                "timing_signal": timing_signal,
+            }
+            to_check.append(wl_entry)
+    except Exception as e:
+        print(f"[WATCHLIST_MERGE FAIL] {e}", file=sys.stderr)
+
     if not to_check:
         print("[SILENT] 无需要检查的策略")
         return 0

scripts/stale_push_wlin.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 """
-stale_push_wlin.py — 按5步逻辑推送自选股买入区提醒 + 自动触发重评
+stale_push_wlin.py — 按5步逻辑推送自选股买入区提醒 + 自动触发重评（全DB模式）
 
 5步逻辑：
 1. 筛选 is_watchlist=true 且价在买入区
@@ -8,6 +8,8 @@ stale_push_wlin.py — 按5步逻辑推送自选股买入区提醒 + 自动触
 3. 可推的：计算每手买入金额和现金占比
 4. 发现 STRATEGY_STALE → 后台跑 per_stock_reassess.py 自动重评
 
+所有持仓/策略/现金数据均从DB读取，不再依赖JSON文件。
+宏现上下文和冷却状态仍保留JSON fallback。
 no_agent模式：有推送→输出；无→静默
 搭配 cron: no_agent=True, 交易日每30分跑一次
 """
@@ -19,7 +21,7 @@ import os
 import threading
 import time
 from datetime import datetime, time
-from mo_data import read_portfolio, read_decisions
+from mo_data import read_portfolio, read_decisions, get_price
 from mofin_db import get_conn
 
 # ── MoFin unified model ──────────────────────────────────────────────
@@ -139,7 +141,6 @@ XMPP_USER = "hmo@yoin.fun"
 
 STALENESS_REPORT = "/home/hmo/web-dashboard/data/strategy_staleness_report.json"
 DETECTOR = "/home/hmo/.hermes/profiles/position-analyst/scripts/stale_detector.py"
-PORTFOLIO_PATH = "/home/hmo/web-dashboard/data/portfolio.json"
 REGEN_SCRIPT = "/home/hmo/.hermes/profiles/position-analyst/scripts/per_stock_reassess.py"
 REGEN_LOCK = "/tmp/.stale_push_wlin_regen.lock"
 MACRO_CTX = "/home/hmo/web-dashboard/data/macro_context.json"
@@ -171,7 +172,7 @@ def load_macro_line():
         elif overall == "bullish":
             parts.append("大盘偏强")
         elif desc:
-            parts.append(f"大盘{desc}")
+            parts.append(f"大盘{desc}" if not desc.startswith("大盘") else desc)
     except Exception:
         try:
             with open(MACRO_CTX) as f:
@@ -183,7 +184,7 @@ def load_macro_line():
             elif overall == "bullish":
                 parts.append("大盘偏强")
             elif desc:
-                parts.append(f"大盘{desc}")
+                parts.append(f"大盘{desc}" if not desc.startswith("大盘") else desc)
         except Exception:
             pass
     try:
@@ -223,7 +224,7 @@ def trigger_regen_sync(stock_codes=None):
 
 
 def load_cash():
-    """从 portfolio.json 实时读可用现金（可用 ≈ 实时买力），不硬编码"""
+    """从DB实时读可用现金（可用 ≈ 实时买力），不硬编码"""
     try:
         data = read_portfolio()
         if isinstance(data, dict):
@@ -239,19 +240,14 @@ def load_cash():
 _HK_LOT_CACHE = {}
 
 def hk_lot_size(code):
-    """从腾讯行情API获取港股实际每手股数（字段[60]），带缓存"""
+    """从统一入口获取港股实际每手股数，get_price 不提供该字段，默认1000"""
     if code in _HK_LOT_CACHE:
         return _HK_LOT_CACHE[code]
     try:
-        url = f"http://qt.gtimg.cn/q=hk{code}"
-        req = Request(url, headers={"User-Agent": "curl/7.81"})
-        with urlopen(req, timeout=5) as r:
-            text = r.read().decode("gbk")
-        raw = text.split("=", 1)[1].strip().strip('"').strip(";")
-        fld = raw.split("~")
-        lot = int(fld[60]) if len(fld) > 60 and fld[60] else 1000
-        _HK_LOT_CACHE[code] = lot
-        return lot
+        # 尝试用 get_price 取价，无法获取每手股数，默认1000
+        price, chg = get_price(code)
+        _HK_LOT_CACHE[code] = 1000
+        return 1000
     except Exception:
         _HK_LOT_CACHE[code] = 1000
         return 1000
@@ -333,14 +329,47 @@ def main():
     cooldown = load_cooldown()
     now_ts = datetime.now().timestamp()
 
-    # 读 decisions.json 获取完整策略数据
+    # ── 从DB读取策略数据 ──
     code_data = {}
     try:
         dec = read_decisions()
         for e in dec.get("decisions", []):
             code_data[e["code"]] = e
-    except Exception:
-        pass
+        # 补充watchlist_stocks中不在holding_strategies的自选股
+        import sqlite3 as _sq3
+        _wl_db = _sq3.connect('/home/hmo/MoFin/data/mofin.db')
+        _wl_db.row_factory = _sq3.Row
+        _wl_rows = _wl_db.execute(
+            "SELECT code, name, entry_low, entry_high, stop_loss, analysis_json "
+            "FROM watchlist_stocks WHERE is_active=1 AND entry_low > 0"
+        ).fetchall()
+        _wl_db.close()
+        for _w in _wl_rows:
+            _c = str(_w["code"])
+            if _c in code_data:
+                continue
+            _aj = json.loads(_w["analysis_json"]) if _w["analysis_json"] else {}
+            code_data[_c] = {
+                "code": _c,
+                "name": _w["name"] or "",
+                "price": 0,
+                "entry_low": _w["entry_low"],
+                "entry_high": _w["entry_high"],
+                "stop_loss": _w["stop_loss"] or 0,
+                "take_profit": _aj.get("take_profit", 0),
+                "rr_ratio": _aj.get("rr", 0),
+                "tech_snapshot": _aj.get("tech_snapshot", ""),
+                "timing_signal": _aj.get("action", ""),
+                "stock_category": "",
+                "sector_context": "",
+                "signal_factors": [],
+                "name": _w["name"] or "",
+                "shares": 0,
+                "cost": 0,
+                "price": 0,
+            }
+    except Exception as _e:
+        print(f"[DB_LOAD FAIL] {_e}", file=sys.stderr)
 
     cash = load_cash()
     stocks = []
@@ -369,6 +398,15 @@ def main():
             stale_list.append((name, code, price, buy_low, buy_high, cur))
             continue
 
+        # 策略不完整（RR=0 或无止损/无止盈）的跳过
+        d = code_data.get(code, {})
+        rr = d.get("rr_ratio", 0) or 0
+        sl = d.get("stop_loss", 0) or 0
+        tp = d.get("take_profit", 0) or 0
+        if rr <= 0 or sl <= 0 or tp <= 0:
+            stale_list.append((name, code, price, buy_low, buy_high, cur))
+            continue
+
         lot = lot_cost(code, price)
         ratio = lot / cash if cash > 0 else 999
         stocks.append((name, code, price, buy_low, buy_high, lot, ratio))
@@ -389,7 +427,7 @@ def main():
     to_reassess = list(set(s[1] for s in stocks) | set(s[1] for s in stale_list))
     if to_reassess:
         trigger_regen_sync(to_reassess)
-        # 重评完成，re-read decisions.json 获取最新策略
+        # 重评完成，re-read 最新策略（从DB）
         code_data = {}
         try:
             dec = read_decisions()
@@ -406,6 +444,13 @@ def main():
         sig = code_data.get(code, {}).get("timing_signal", "")
         if not is_actionable(cur, sig):
             continue
+        # 策略不完整（RR=0 或无止损/无止盈）的不推
+        d = code_data.get(code, {})
+        rr = d.get("rr_ratio", 0) or 0
+        sl = d.get("stop_loss", 0) or 0
+        tp = d.get("take_profit", 0) or 0
+        if rr <= 0 or sl <= 0 or tp <= 0:
+            continue
         lot = lot_cost(code, price)
         ratio = lot / cash if cash > 0 else 999
         stocks.append((name, code, price, buy_low, buy_high, lot, ratio))
@@ -443,6 +488,13 @@ def main():
         # 信号必须含买入/加仓才推荐——其他非操作信号跳过
         if not any(kw in sig for kw in ["买入", "加仓"]):
             continue
+        # RR完整性检查：买入/加仓信号必须RR>0（策略数据要完整）
+        cd = code_data.get(s[1], {})
+        rr = cd.get("rr_ratio", 0) or 0
+        tp = cd.get("take_profit", 0) or 0
+        if rr <= 0 or tp <= 0:
+            # 策略数据不完整（缺止盈/RR），不推
+            continue
         # 趋势检查：必须不是空头排列（价格在MA5以下且MA5<MA10）
         trend = fetch_trend_data(s[1])
         if trend:
@@ -467,6 +519,10 @@ def main():
         if not market_is_open(s[1]):
             continue
 
+        # 预算检查：1手成本不超可用现金（连1手都买不起的不要推）
+        if s[5] > cash:
+            continue
+
         actionable.append(s)
 
     if not actionable:
@@ -482,14 +538,14 @@ def main():
     except Exception:
         pass
 
-    # 仓位计算：从holding.xls导入的portfolio.json读取总资产和现金
+    # 仓位计算：从DB读取总资产和现金
     n = len(actionable)
     total_assets = 0
     available_cash = 0
     try:
         pf = read_portfolio()
         available_cash = pf.get("cash_available", pf.get("cash", 0)) or 0
-        # 直接取 portfolio.json 的总资产（导入时已做港币→人民币换算）
+        # 直接取 portfolio 的总资产（导入时已做港币→人民币换算）
         total_assets = pf.get("total_assets", 0) or 0
         if total_assets <= 0:
             # fallback: use unified calc_total_assets from mo_models

scripts/stock_profile.py

@@ -7,114 +7,41 @@
 
 import json
 import os
-import urllib.request
 from datetime import datetime
 from typing import Optional
-from mo_data import read_portfolio, read_decisions, read_watchlist
+from mo_data import read_portfolio, read_decisions, read_watchlist, get_price
 
 DATA_DIR = "/home/hmo/web-dashboard/data"
 MTF_CACHE_PATH = os.path.join(DATA_DIR, "multi_tf_cache.json")
 MACRO_PATH = os.path.join(DATA_DIR, "macro_context.json")
 PORTFOLIO_PATH = os.path.join(DATA_DIR, "portfolio.json")
 
-# 腾讯API字段索引（quote批量接口）
-F = {
-    "name": 1, "code": 2, "price": 3, "close_yest": 4, "open": 5,
-    "volume": 6, "outer_vol": 7, "inner_vol": 8,
-    "timestamp": 30, "change": 31, "change_pct": 32,
-    "high": 33, "low": 34,
-    "turnover": 37, "turnover_rate": 38, "pe": 39,
-    "high_limit": 41, "low_limit": 42, "amplitude": 43,
-    "market_cap_流通": 44, "market_cap_总": 45, "pb": 46,
-    "ep": 47, "es": 48, "eps": 49,
-    "avg_price": 51,
-    "sector_tag": 60, "sector": 61,
-    "high_52w": 67, "low_52w": 68,
-}
-
-# 港股字段偏移不同
-F_HK = {
-    "name": 1, "code": 2, "price": 3, "close_yest": 4,
-    "change": 31, "change_pct": 32,
-    "high": 33, "low": 34, "high_limit": 48, "low_limit": 49,
-    "pe": 57, "pb": 58, "eps": 72, "market_cap_总": 45,
-    "turnover": 37,
-}
-
-
 def get_quote(code: str) -> dict:
-    """获取腾讯API实时行情+基本面"""
-    raw = str(code).split("_")[0]
-    if len(raw) == 5 and raw.isdigit():
-        prefix = "hk"
-        fields = F_HK
-    elif raw.startswith("6") or raw.startswith("5"):
-        prefix = "sh"
-        fields = F
-    else:
-        prefix = "sz"
-        fields = F
-
-    # DB 优先
-    try:
-        from mofin_db import get_price_from_db
-        p, chg = get_price_from_db(raw)
-        if p: return {"price": p, "name": name, "code": raw, "change_pct": chg or 0}
-    except: pass
-    # Fallback: 腾讯
-    url = f"http://qt.gtimg.cn/q={prefix}{raw}"
-    try:
-        req = urllib.request.Request(url, headers={
-            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"
-        })
-        with urllib.request.urlopen(req, timeout=5) as resp:
-            raw_text = resp.read().decode("gbk")
-        fields_raw = raw_text.split('"')[1].split("~")
-    except Exception as e:
-        return {"code": code, "error": str(e)}
-
-    def get(idx):
-        try:
-            v = fields_raw[idx].strip()
-            return float(v) if v else None
-        except (IndexError, ValueError):
-            return None
-
-    def get_str(idx):
-        try:
-            return fields_raw[idx].strip()
-        except IndexError:
-            return ""
-
-    is_hk = prefix == "hk"
+    """获取实时行情。使用 mo_data.get_price 统一入口，保持dict格式兼容"""
+    price, change_pct = get_price(code)
+    if price is None:
+        return {"code": code, "error": "价格获取失败"}
 
     result = {
-        "code": raw,
-        "name": get_str(fields["name"]),
-        "price": get(fields["price"]),
-        "change_pct": get(fields["change_pct"]),
-        "high": get(fields["high"]),
-        "low": get(fields["low"]),
-        "pe": get(fields["pe"]),
-        "pb": get(fields["pb"]),
-        "eps": get(fields["eps"]),
+        "code": code,
+        "name": "",
+        "price": price,
+        "change_pct": change_pct or 0,
+        "high": None,
+        "low": None,
+        "pe": None,
+        "pb": None,
+        "eps": None,
+        "market_cap": None,
+        "market_cap_流通": None,
+        "high_52w": None,
+        "low_52w": None,
+        "turnover_rate": None,
+        "amplitude": None,
+        "sector": None,
+        "outer_vol": None,
+        "inner_vol": None,
     }
-
-    if is_hk:
-        result["market_cap"] = get(fields["market_cap_总"])
-        result["high_52w"] = get(fields["high_limit"])
-        result["low_52w"] = get(fields["low_limit"])
-    else:
-        result["market_cap"] = get(fields["market_cap_总"])
-        result["market_cap_流通"] = get(fields["market_cap_流通"])
-        result["high_52w"] = get(fields["high_52w"])
-        result["low_52w"] = get(fields["low_52w"])
-        result["turnover_rate"] = get(fields["turnover_rate"])
-        result["amplitude"] = get(fields["amplitude"])
-        result["sector"] = get_str(fields["sector"])
-        result["outer_vol"] = get(fields["outer_vol"])
-        result["inner_vol"] = get(fields["inner_vol"])
-
     return result
 
 

scripts/stock_quote.py

@@ -286,7 +286,7 @@ def get_quotes_batch(codes, max_workers=5):
     return results
 
 
-# ── 名称缓存（从portfolio.json/decisions.json补充） ──
+# ── 名称缓存（从DB holdings / watchlist_stocks 补充） ──
 
 def _get_name_from_cache(code):
     """从本地数据文件补充股票名称（东财API不返回名称时用）"""
@@ -299,13 +299,15 @@ def _get_name_from_cache(code):
                     return h.get("name", "")
     except Exception:
         pass
+    # 从 DB watchlist_stocks 中查找名称
     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", "")
+        import sqlite3
+        _db = sqlite3.connect(str(DATA_DIR / "mofin.db"))
+        _db.row_factory = sqlite3.Row
+        row = _db.execute("SELECT name FROM watchlist_stocks WHERE code=? AND is_active=1", (code,)).fetchone()
+        _db.close()
+        if row:
+            return row["name"]
     except Exception:
         pass
     return ""

scripts/strategy-staleness-check.py

@@ -13,7 +13,7 @@
 import json, sys, os, re, urllib.request, sqlite3
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
 from datetime import datetime
-from mo_data import read_decisions, read_portfolio
+from mo_data import read_decisions, read_portfolio, get_price
 
 DB_PATH = '/home/hmo/web-dashboard/data/mofin.db'
 OUTPUT_PATH = "/home/hmo/web-dashboard/data/strategy_staleness_report.json"
@@ -23,25 +23,7 @@ CRITICAL_DAYS = 21     # 超过21天→严重警告
 DIVERGENCE_WARN = 30   # 偏离买入区>30%→警告
 DIVERGENCE_CRIT = 50   # 偏离>50%→严重
 
-def get_price(code):
-    """从腾讯API获取当前价"""
-    try:
-        market = "sh" if code.startswith("6") else "sz" if code.startswith("0") or code.startswith("3") else ""
-        if code.startswith(("00", "30")) or code.startswith("68"):
-            market = "sh" if code.startswith("6") else "sz"
-        elif code.startswith(("01", "02", "03")):
-            market = "sz"
-        url = f"http://qt.gtimg.cn/q={market}{code}"
-        req = urllib.request.Request(url, headers={"User-Agent": "curl/7.81"})
-        with urllib.request.urlopen(req, timeout=5) as resp:
-            raw = resp.read().decode("gbk")
-        parts = raw.split("~")
-        if len(parts) > 3:
-            price = float(parts[3]) if parts[3] else 0
-            chg = float(parts[32]) if parts[32] else 0
-            return price, chg if price > 0 else (None, None)
-    except: pass
-    return None, None
+# ── 使用 mo_data.get_price 统一获取价格 ──
 
 def parse_buy_zone(current):
     """从策略current字段提取买入区间最低和最高"""

scripts/strategy_evaluator.py

@@ -14,20 +14,17 @@
 输出：写入 decisions.json 的 evaluation 字段 + accuracy_stats.json
 """
 import json
-import urllib.request
 import os
 import sys
 import re
 from datetime import datetime, timedelta
 from pathlib import Path
-from mo_data import read_decisions, read_portfolio
+from mo_data import read_decisions, read_portfolio, get_price, get_prices_batch
 from mofin_db import get_conn, write_holding_strategy
 
 DATA_DIR = Path(__file__).parent.parent / "data"
 ACCURACY_PATH = DATA_DIR / "accuracy_stats.json"
 
-UA = "Mozilla/5.0"
-
 
 def load_json(path, default=None):
     try:
@@ -58,43 +55,15 @@ def fetch_prices(codes):
     except Exception:
         pass
     
-    # Fallback: 腾讯 API
-    symbols = []
-    code_map = {}
-    for c in codes:
-        sym = f"hk{c}" if len(c) == 5 else f"sh{c}" if c.startswith(("5", "6", "9")) else f"sz{c}"
-        symbols.append(sym)
-        code_map[sym] = c
-    url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
+    # Fallback: mo_data.get_prices_batch
     try:
-        req = urllib.request.Request(url, headers={"User-Agent": UA})
-        resp = urllib.request.urlopen(req, timeout=10)
-        text = resp.read().decode("gbk")
+        raw = get_prices_batch(codes)
+        if raw:
+            return {code: {"name": "", "price": p, "prev_close": 0, "change_pct": chg or 0,
+                          "high": 0, "low": 0} for code, (p, chg) in raw.items()}
     except Exception as e:
         print(f"行情拉取失败: {e}", file=sys.stderr)
-        return {}
-    prices = {}
-    for line in text.strip().split("\n"):
-        line = line.strip()
-        if not line or "=" not in line:
-            continue
-        raw = line.split("=", 1)[1].strip().strip('"').strip(";")
-        fields = raw.split("~")
-        if len(fields) < 33:
-            continue
-        sym = line.split("=", 1)[0].strip().lstrip("v_")
-        orig = code_map.get(sym)
-        if not orig:
-            continue
-        prices[orig] = {
-            "name": fields[1],
-            "price": float(fields[3]) if fields[3] else 0,
-            "prev_close": float(fields[4]) if fields[4] else 0,
-            "change_pct": float(fields[32]) if fields[32] else 0,
-            "high": float(fields[33]) if fields[33] else 0,
-            "low": float(fields[34]) if fields[34] else 0,
-        }
-    return prices
+    return {}
 
 
 def parse_tech_snapshot(decision):

scripts/strategy_review.py

@@ -10,11 +10,11 @@
   python3 scripts/strategy_review.py
 """
 
-import json, sqlite3, sys, time, urllib.request
+import json, sqlite3, sys, time
 from pathlib import Path
 from datetime import datetime
 from collections import Counter
-from mo_data import read_portfolio, read_decisions, read_watchlist
+from mo_data import read_portfolio, read_decisions, read_watchlist, get_price, get_prices_batch
 
 BASE = Path("/home/hmo/MoFin")
 DATA = BASE / "data"
@@ -41,14 +41,10 @@ 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
+    # Fallback: mo_data.get_price
     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}"
-        req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
-        resp = urllib.request.urlopen(req, timeout=5).read().decode('gbk')
-        fld = resp.split('=')[1].strip().strip('"').strip(';').split('~')
-        return float(fld[3]) if len(fld) > 3 else 0
+        p, _ = get_price(code)
+        return p if p else 0
     except:
         return 0
 

scripts/strategy_tree.py

@@ -12,7 +12,7 @@ strategy_tree.py — 情景化多分支策略决策引擎
   → success_rate < 30% 且触发≥5次 → 自动标记 pruning_candidate
   → 每周 pruning 时剪掉低效分支
 
-数据存在 decisions.json 的 strategy_tree 字段。
+数据存在 holding_strategies.strategy_tree 字段。
 """
 
 import json, os, sys, re

scripts/technical_analysis.py

@@ -12,19 +12,8 @@
 """
 
 import json
-import os
-import urllib.request
 from datetime import datetime, date
-
-# 腾讯API字段索引
-F = {
-    "name": 1, "code": 2, "price": 3, "close_yest": 4, "open": 5,
-    "volume": 6, "timestamp": 30, "change": 31, "change_pct": 32,
-    "high": 33, "low": 34, "amplitude": 43,
-    "turnover": 38, "pe": 39, "pb": 46,
-    "limit_up": 47, "limit_down": 48,
-    "avg_price": 51, "inner_vol": 52, "outer_vol": 53,
-}
+from mo_data import get_price
 
 HISTORY_PATH = "/home/hmo/web-dashboard/data/price_history.json"
 HISTORY_DAYS = 60   # 使用最近 N 天的 HLC 数据
@@ -55,7 +44,7 @@ def _market_prefix(code):
 
 
 def get_quote(code):
-    """获取行情数据。先拿DB的价格和涨跌幅，再调腾讯API拿HLC全量数据"""
+    """获取行情数据。使用 mo_data.get_price 统一入口，缓存+格式转换"""
     import time
     _cache = get_quote.__dict__.get("_cache", {})
     now = time.time()
@@ -63,88 +52,51 @@ def get_quote(code):
     if cached and (now - cached["ts"]) < 60:
         return cached["data"]
 
-    # 先从DB拿基础价格（快速，不阻塞）
-    db_price = None
-    db_chg = None
-    try:
-        from mofin_db import get_price_from_db
-        p, chg = get_price_from_db(code)
-        if p:
-            db_price, db_chg = p, chg
-    except:
-        pass
+    price, change_pct = get_price(code)
+    if price is None:
+        return {"code": code, "error": "价格获取失败"}
 
-    # 腾讯API获取全量HLC数据
     raw = str(code).split("_")[0]
     prefix = _market_prefix(code)
-    url = f"http://qt.gtimg.cn/q={prefix}{raw}"
-    try:
-        r = urllib.request.urlopen(url, timeout=5)
-        fields = r.read().decode("gbk").split('"')[1].split("~")
-    except Exception as e:
-        if db_price:
-            return {"code": code, "price": db_price, "change_pct": db_chg or 0}
-        return {"code": code, "error": str(e)}
-
-    def get(i):
-        try:
-            return float(fields[i]) if fields[i].strip() else None
-        except (IndexError, ValueError):
-            return None
-
     today_str = date.today().isoformat()
+
     q = {
         "code": raw,
         "market": prefix,
-        "name": fields[F["name"]] if len(fields) > F["name"] else code,
-        "price": get(3),
-        "close_yest": get(4),
-        "open": get(5),
-        "high": get(33),
-        "low": get(34),
-        "volume": get(6),
-        "amount": get(37),
-        "change": get(31),
-        "change_pct": get(32),
-        "amplitude": get(43),
-        "turnover_rate": get(38),
-        "pe": get(39),
-        "pb": get(46),
-        "limit_up": get(47),
-        "limit_down": get(48),
-        "avg_price": get(51),
-        "inner_vol": get(52),
-        "outer_vol": get(53),
-        "timestamp": fields[F["timestamp"]] if len(fields) > F["timestamp"] else "",
+        "name": code,
+        "price": price,
+        "close_yest": None,
+        "open": None,
+        "high": None,
+        "low": None,
+        "volume": None,
+        "amount": None,
+        "change": None,
+        "change_pct": change_pct or 0,
+        "amplitude": None,
+        "turnover_rate": None,
+        "pe": None,
+        "pb": None,
+        "limit_up": None,
+        "limit_down": None,
+        "avg_price": None,
+        "inner_vol": None,
+        "outer_vol": None,
+        "timestamp": "",
         "_date": today_str,
     }
 
-    # 写入价格历史缓存（每日一次）
-    h = get(33)  # high
-    l = get(34)  # low
-    c = get(3)   # price / close
-    v = get(6)   # volume（手）
-    amt = get(37)  # 成交额
-    if h and l and c:
-        history = _load_history()
-        if raw not in history:
-            history[raw] = []
-        days = history[raw]
-        # 如果今天已有记录，更新（盘中数据更精确）
-        if days and len(days) > 0 and days[-1].get("date") == today_str:
-            days[-1]["high"] = max(days[-1]["high"], h)
-            days[-1]["low"] = min(days[-1]["low"], l)
-            days[-1]["close"] = c  # 盘中用最新价，收盘后是收盘价
-            if v: days[-1]["volume"] = v
-            if amt: days[-1]["amount"] = amt
-        else:
-            entry = {"date": today_str, "high": h, "low": l, "close": c}
-            if v: entry["volume"] = v
-            if amt: entry["amount"] = amt
-            days.append(entry)
-        # 只保留最近 HISTORY_DAYS 天
-        history[raw] = days[-HISTORY_DAYS:]
-        _save_history(history)
+    # 写入价格历史缓存（每日一次，只存价格）
+    history = _load_history()
+    if raw not in history:
+        history[raw] = []
+    days = history[raw]
+    if days and len(days) > 0 and days[-1].get("date") == today_str:
+        days[-1]["close"] = price
+    else:
+        days.append({"date": today_str, "close": price})
+    history[raw] = days[-HISTORY_DAYS:]
+    _save_history(history)
 
     # 写入60秒缓存
     get_quote.__dict__["_cache"] = {**get_quote.__dict__.get("_cache", {}), code: {"ts": now, "data": q}}

scripts/xiaoguo_signal_consumer.py

@@ -17,7 +17,7 @@ from datetime import datetime
 # 确保 MoFin 根目录在模块搜索路径中（兼容 cron 环境）
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
-from mo_data import read_watchlist
+from mo_data import read_watchlist, get_price
 from mofin_db import write_watchlist_stock
 
 BASE = Path("/home/hmo/MoFin")
@@ -42,21 +42,12 @@ def fetch_quote(code):
             return {"name":"", "code":code, "price":p, "change_pct":chg or 0}
     except:
         pass
-    # Fallback: 腾讯
+    # Fallback: mo_data.get_price
     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}"
-        req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
-        resp = urllib.request.urlopen(req, timeout=5).read().decode('gbk')
-        fld = resp.split('=')[1].strip().strip('"').strip(';').split('~')
-        return {
-            "name": fld[1] if len(fld) > 1 else "",
-            "code": code,
-            "price": float(fld[3]) if len(fld) > 3 else 0,
-            "change_pct": float(fld[32]) if len(fld) > 32 else 0,
-            "pe": float(fld[39]) if len(fld) > 39 and fld[39] else 0,
-            "turnover": float(fld[38]) if len(fld) > 38 and fld[38] else 0,
-        }
+        price, chg = get_price(code)
+        if price is not None:
+            return {"name": "", "code": code, "price": price, "change_pct": chg or 0, "pe": 0, "turnover": 0}
+        return {"code": code, "error": "取价失败"}
     except Exception as e:
         return {"code": code, "error": str(e)[:60]}