feat: data-layering 第二批消费层读DB(divergence/staleness/accumulation/collect_eval/strategy_review/mo_provider/multi_timeframe/chip_factors)

deploy/profile-scripts/collect_evaluation_data.py

@@ -16,10 +16,8 @@ D6 资金面 — 成交额/换手率/量比
 """
 
 import json
-import urllib.request
 import os
 import sys
-import re
 from datetime import datetime
 from pathlib import Path
 
@@ -34,8 +32,6 @@ DATA_DIR = Path(__file__).parent.parent / "data"
 PROFILES_PATH = DATA_DIR / "stock_profiles.json"
 OUTPUT_PATH = DATA_DIR / "evaluation_input.json"
 
-UA = "Mozilla/5.0"
-
 
 def load_json(path, default=None):
     try:
@@ -52,52 +48,68 @@ def save_json(path, data):
 
 
 def fetch_tencent_data(symbols):
-    """批量拉行情。DB 优先，腾讯 API fallback"""
+    """批量拉行情。读 DB live_prices + stock_daily（2026-08-26 分层铁律：消费层不直连腾讯API）"""
     if not symbols:
         return {}
-    # DB 优先
+    import sqlite3
+    result = {}
     try:
-        from mofin_db import get_prices_batch_from_db
-        db = get_prices_batch_from_db(symbols)
-        if db:
-            return {code: {"name": "", "price": p, "prev_close": 0, "change_pct": chg or 0,
-                          "high": 0, "low": 0} for code, (p, chg) in db.items()}
-    except: pass
-    # Fallback: 腾讯
-    code_map = {}
-    query_symbols = []
-    for c in symbols:
-        sym = f"hk{c}" if len(c) == 5 else f"sh{c}" if c.startswith(("5", "6", "9")) else f"sz{c}"
-        query_symbols.append(sym)
-        code_map[sym] = c
-    url = f"http://qt.gtimg.cn/q={','.join(query_symbols)}"
-    try:
-        req = urllib.request.Request(url, headers={"User-Agent": UA})
-        resp = urllib.request.urlopen(req, timeout=15)
-        text = resp.read().decode("gbk")
+        conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=5)
+        syms = list(symbols)
+        ph = ",".join("?" * len(syms))
+        # 实时价：一次查
+        price_rows = conn.execute(
+            f"SELECT code, price, change_pct FROM live_prices WHERE code IN ({ph})", syms
+        ).fetchall()
+        prices = {r[0]: (r[1], r[2]) for r in price_rows}
+        # 名称
+        name_rows = conn.execute(
+            f"SELECT code, name FROM stocks WHERE code IN ({ph})", syms
+        ).fetchall()
+        names = {r[0]: r[1] or "" for r in name_rows}
+        # 最近日K（昨收/今开/高低/量）
+        sd = {}
+        for c in syms:
+            row = conn.execute(
+                "SELECT close, open, high, low, volume FROM stock_daily "
+                "WHERE code=? ORDER BY date DESC LIMIT 1", (c,)
+            ).fetchone()
+            if row:
+                sd[c] = row
+        conn.close()
+
+        for code in syms:
+            if code not in prices:
+                continue
+            price = prices[code][0]
+            change_pct = prices[code][1] or 0
+            if not price or price <= 0:
+                continue
+            row = sd.get(code)
+            row_c = row[0] if row else 0
+            row_o = row[1] if row else 0
+            row_h = row[2] if row else 0
+            row_l = row[3] if row else 0
+            row_v = row[4] if row else 0
+            result[code] = {
+                "name": names.get(code, ""),
+                "price": price,
+                "prev_close": row_c,
+                "open": row_o,
+                "change_pct": change_pct,
+                "high": row_h,
+                "low": row_l,
+                "volume": row_v,
+            }
     except Exception as e:
         print(f"行情拉取失败: {e}", file=sys.stderr)
         return {}
-    result = {}
-    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) < 35:
-            continue
-        sym = line.split("=", 1)[0].strip().lstrip("v_")
-        orig = code_map.get(sym)
-        if not orig:
-            continue
-        # 统一格式（A股和港股字段长度不同）
-        result[orig] = fields
     return result
 
 
 def fetch_indices():
-    """拉五大指数"""
+    """拉指数：读 DB stock_daily 最近收盘（2026-08-26 分层铁律：消费层不直连腾讯API）"""
+    import sqlite3
     index_codes = {
         "sh000001": "上证指数",
         "sz399001": "深证成指",
@@ -105,37 +117,33 @@ def fetch_indices():
         "hkHSI": "恒生指数",
         "hkHSTECH": "恒生科技",
     }
-    idx_map = {}
-    for c, n in index_codes.items():
-        sym = c  # 已经是完整符号
-        idx_map[sym] = n
-    url = f"http://qt.gtimg.cn/q={','.join(index_codes.keys())}"
+    result = {}
     try:
-        req = urllib.request.Request(url, headers={"User-Agent": UA})
-        resp = urllib.request.urlopen(req, timeout=10)
-        text = resp.read().decode("gbk")
+        conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=5)
+        for c, n in index_codes.items():
+            rows = conn.execute(
+                "SELECT date, close, open, high, low FROM stock_daily "
+                "WHERE code=? ORDER BY date DESC LIMIT 2", (c,)
+            ).fetchall()
+            if not rows:
+                continue  # 读不到则跳过（中性）
+            latest = rows[0]
+            prev = rows[1] if len(rows) > 1 else None
+            prev_close = prev[1] if prev else latest[1]
+            price = latest[1] or 0
+            change_pct = (price - prev_close) / prev_close * 100 if prev_close else 0
+            result[n] = {
+                "price": safe_float(price),
+                "prev_close": safe_float(prev_close if prev else None),
+                "change_pct": safe_float(change_pct),
+                "high": safe_float(latest[3] or price),
+                "low": safe_float(latest[4] or price),
+                "timestamp": latest[0],
+            }
+        conn.close()
     except Exception as e:
         print(f"指数拉取失败: {e}", file=sys.stderr)
         return {}
-    result = {}
-    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_")
-        name = idx_map.get(sym, sym)
-        result[name] = {
-            "price": safe_float(fields[3]),
-            "prev_close": safe_float(fields[4]),
-            "change_pct": safe_float(fields[32]),
-            "high": safe_float(fields[33]),
-            "low": safe_float(fields[34]),
-            "timestamp": fields[30] if len(fields) > 30 else "",
-        }
     return result