feat(pipeline): VPS 价格源适配器——桥分钟bar+dbbardata日线（偏差日报三价链锚点） [vps]

2026-09-26 07:57:37 +08:00
parent ff07be83e4
commit 68ea844e8b
2 changed files with 152 additions and 0 deletions
@@ -0,0 +1,138 @@
+# -*- coding: utf-8 -*-
+"""VPS 价格源适配器（偏差日报三价链锚点；Mac 不可跑——桥/dbbardata 仅 VPS）。
+
+分钟 close：桥 xtdata shim（PYTHONPATH 前置 C:\\sanguo_bigqmt\\xtquant_bridge +
+env BIGQMT_ACCOUNT_ID/BIGQMT_REDIS_PASSWORD——口令运行时读 C:\\redis\\redis.conf，
+永不落盘/回显）。日线 OHLC/prev_close：quant_trading.db dbbardata（datareader
+同款列名）。symbol 规范化：日志/台账四种形态 → xtdata 的 XXXXXX.SH/SZ。
+"""
+from __future__ import annotations
+
+import os
+import re
+import sqlite3
+from datetime import datetime, timedelta
+
+_BRIDGE_PYROOT = os.environ.get(
+    "SANGUO_BRIDGE_PYROOT", r"C:\sanguo_bigqmt\xtquant_bridge")
+_ACCOUNT = os.environ.get("BIGQMT_ACCOUNT_ID", "66639661")
+_DAILY_DB = os.environ.get(
+    "SANGUO_DAILY_DB", r"C:\sanguo_vnpy_v2\data\quant_trading.db")
+
+
+def normalize_symbol(sym: str) -> str:
+    """600276.SH/600276/sh600276/600276.XSHG → 600276.SH（SZ 对称）。"""
+    s = sym.strip().upper()
+    m = re.match(r"^(?:SH|SZ)?(\d{6})(?:\.(?:SH|SZ|XSHG|XSHE))?$", s)
+    if not m:
+        return s
+    code = m.group(1)
+    exch = "SH" if s.startswith("SH") or s.endswith(("SH", "XSHG")) \
+        or code.startswith(("6", "9", "5")) else "SZ"
+    return f"{code}.{exch}"
+
+
+def _read_redis_password() -> str:
+    with open(r"C:\redis\redis.conf", "r", errors="replace") as f:
+        for line in f:
+            m = re.match(r"^requirepass\s+(\S+)", line)
+            if m:
+                return m.group(1).strip()
+    raise SystemExit("NO_REQUIREPASS_IN_CONF")
+
+
+def _ensure_bridge_env() -> None:
+    parts = [p for p in os.environ.get("PYTHONPATH", "").split(os.pathsep)
+             if p]
+    if _BRIDGE_PYROOT not in parts:
+        os.environ["PYTHONPATH"] = os.pathsep.join([_BRIDGE_PYROOT] + parts)
+    os.environ.setdefault("BIGQMT_ACCOUNT_ID", _ACCOUNT)
+    if "BIGQMT_REDIS_PASSWORD" not in os.environ:
+        os.environ["BIGQMT_REDIS_PASSWORD"] = _read_redis_password()
+
+
+def _minute_closes(symbols: list[str], day: str) -> dict[tuple[str, str], float]:
+    """桥拉一日分钟 close → {(symbol, 'HH:MM'): close}（当日锚点一次拉齐）。"""
+    _ensure_bridge_env()
+    from xtquant import xtdata  # noqa: WPS433（shim 需 env 先就位，延迟导入）
+    start = f"{day} 09:00:00"
+    end = f"{day} 15:30:00"
+    out: dict[tuple[str, str], float] = {}
+    codes = sorted({normalize_symbol(s) for s in symbols})
+    for i in range(0, len(codes), 50):
+        df = xtdata.get_market_data(
+            field_list=["close"], stock_list=codes[i:i + 50], period="1m",
+            start_time=start, end_time=end)
+        if df is None or len(df) == 0:
+            continue
+        closes = df["close"] if hasattr(df, "keys") and "close" in df else df
+        for code in codes[i:i + 50]:
+            try:
+                sub = closes[code]
+            except (KeyError, TypeError, IndexError):
+                continue
+            for idx, v in sub.items() if hasattr(sub, "items") else []:
+                minute = str(idx)[-8:-3] if len(str(idx)) >= 8 else str(idx)[-5:]
+                if v == v:  # NaN 剔除
+                    out[(code, minute)] = float(v)
+    return out
+
+
+def _daily_bars(symbol: str, day: str) -> dict | None:
+    """dbbardata 当日+前一日日线 → {open/high/low/close/prev_close}。
+
+    exchange 映射（infra 核修）：dbbardata 用 vt 约定 SSE/SZSE（datareader.py:158
+    同款），非 SH/SZ——照抄 split 结果会静默查空、锚点全 None。
+    """
+    code = normalize_symbol(symbol)
+    bare, exch = code.split(".")
+    exch = {"SH": "SSE", "SZ": "SZSE"}[exch]
+    prev = (datetime.strptime(day, "%Y-%m-%d") - timedelta(days=10)
+            ).strftime("%Y-%m-%d")
+    end_excl = (datetime.strptime(day, "%Y-%m-%d") + timedelta(days=1)
+                ).strftime("%Y-%m-%d")
+    with sqlite3.connect(_DAILY_DB) as conn:
+        rows = conn.execute(
+            "SELECT datetime, open_price, high_price, low_price, close_price "
+            "FROM dbbardata WHERE symbol=? AND exchange=? AND interval='d' "
+            "AND datetime>=? AND datetime<? ORDER BY datetime",
+            (bare, exch, prev, end_excl)).fetchall()
+    rows = [r for r in rows if str(r[0])[:10] <= day]
+    if not rows:
+        return None
+    cur = rows[-1]
+    return {"open": cur[1], "high": cur[2], "low": cur[3], "close": cur[4],
+            "prev_close": rows[-2][4] if len(rows) >= 2 else cur[1]}
+
+
+def make_price_fn():
+    """PriceFn：minute≠None 走桥分钟缓存；None 走 dbbardata 日 close。"""
+    cache: dict[str, dict[tuple[str, str], float]] = {}
+
+    def fn(symbol: str, day: str, minute: str | None):
+        code = normalize_symbol(symbol)
+        if minute is None:
+            bars = _daily_bars(code, day)
+            return bars and bars["close"]
+        if day not in cache:
+            cache[day] = _minute_closes([code], day)
+        return cache[day].get((code, minute))
+    return fn
+
+
+def make_ohlc_fn():
+    def fn(symbol: str, day: str):
+        return _daily_bars(normalize_symbol(symbol), day)
+    return fn
+
+
+def probe(day: str, symbol: str = "600276.SH") -> None:
+    """fail-loud 自检（infra 核修）：已知 symbol 查不到日线/分钟→直接 raise，
+    绝不静默空转（VPS 首跑/部署后冒烟必调；测试不依赖）。"""
+    bars = _daily_bars(symbol, day)
+    if not bars or not bars.get("close"):
+        raise RuntimeError(f"probe: dbbardata 无 {symbol}@{day} 日线（exchange 映射/库路径错？）")
+    mins = _minute_closes([symbol], day)
+    if not mins:
+        raise RuntimeError(f"probe: 桥分钟 close 空 {symbol}@{day}（shim env/形态错？）")
+    print(f"probe ok: close={bars['close']} minutes={len(mins)}")