"""QMT Bridge 数据源（本地行情桥 HTTP 接口）。 将 xtquant 行情 API 通过 HTTP 暴露的本地服务，提供日 K、5 分钟 K 线和指数日 K。 无需凭证（本地服务），超时短（局域网），不限速。 """ from __future__ import annotations import json import urllib.request from datetime import datetime from typing import Any, Optional import pandas as pd from app.core.config import settings from app.core.datasource.base import DataSource from app.core.datasource.utils import code6_to_mairui, normalize_5min from app.core.utils.logging import get_logger logger = get_logger("qmt_bridge") KLINE_FIELDS = "close,open,high,low,volume" KLINE_5MIN_FIELDS = "close,open,high,low,volume,amount" class QmtBridgeSource(DataSource): key = "datasource_qmt_bridge" name = "QMT Bridge（本地行情桥）" provides = ["kline_daily", "kline_5min", "index_daily"] requires_credential = False credential_key = "" def __init__(self) -> None: self._base_url = (settings.qmt_bridge_url or "http://127.0.0.1:8610").rstrip("/") # ── 健康检查 / 可用性 ───────────────────────────────── def is_available(self) -> tuple[bool, str]: try: resp = self._get("/health") if isinstance(resp, dict) and resp.get("status") == "ok": return True, "就绪" return False, f"bridge 返回异常: {resp}" except Exception as e: return False, f"bridge 不可达: {e}" def health_check(self) -> dict[str, Any]: try: resp = self._get("/health") if isinstance(resp, dict) and resp.get("status") == "ok": return {"success": True, "message": f"bridge 连接正常"} return {"success": False, "message": f"bridge 返回异常: {resp}"} except Exception as e: return {"success": False, "message": f"bridge 连接失败: {e}"} # ── K 线 fetch ───────────────────────────────────── def fetch_kline_daily(self, code6: str, start: str, end: str) -> pd.DataFrame: """日 K 线。 Args: code6: 6 位股票代码（如 "600519"） start: "YYYY-MM-DD" end: "YYYY-MM-DD" Returns: DataFrame 列: trade_date, open, high, low, close, volume """ symbol = code6_to_mairui(code6) params = { "code": symbol, "period": "1d", "start": start.replace("-", ""), "end": end.replace("-", ""), "count": "-1", "fields": KLINE_FIELDS, } data = self._get("/kline", params) records = self._parse_kline_data(data, is_daily=True) if not records: return pd.DataFrame() df = pd.DataFrame(records) df["trade_date"] = pd.to_datetime(df["trade_date"]) return self._normalize_kline(df) def fetch_kline_5min(self, code6: str, start: str, end: str) -> pd.DataFrame: """5 分钟 K 线。 Args: code6: 6 位股票代码（如 "600519"） start: "YYYY-MM-DD" end: "YYYY-MM-DD" Returns: DataFrame 列: bar_time, open, high, low, close, volume, amount, turnover_rate """ symbol = code6_to_mairui(code6) params = { "code": symbol, "period": "5m", "start": start.replace("-", ""), "end": end.replace("-", ""), "count": "-1", "fields": KLINE_5MIN_FIELDS, } data = self._get("/kline", params) records = self._parse_kline_data(data, is_daily=False) if not records: return pd.DataFrame() df = pd.DataFrame(records) # 用 normalize_5min 标准化 return normalize_5min(df) def fetch_index_daily(self, index_code: str, start: str, end: str) -> pd.DataFrame: """指数日 K 线。 Args: index_code: 带交易所后缀的指数代码（如 "000300.SH"） start: "YYYY-MM-DD" end: "YYYY-MM-DD" Returns: DataFrame 列: trade_date, open, high, low, close, volume """ params = { "code": index_code, "period": "1d", "start": start.replace("-", ""), "end": end.replace("-", ""), "count": "-1", "fields": KLINE_FIELDS, } data = self._get("/kline", params) records = self._parse_kline_data(data, is_daily=True) if not records: return pd.DataFrame() df = pd.DataFrame(records) df["trade_date"] = pd.to_datetime(df["trade_date"]) return self._normalize_kline(df) # ── 内部方法 ─────────────────────────────────────── def _get(self, path: str, params: dict[str, str] | None = None) -> Any: """GET 请求 bridge，返回解析后的 JSON。""" url = f"{self._base_url}{path}" if params: qs = "&".join(f"{k}={v}" for k, v in params.items() if v) url = f"{url}?{qs}" req = urllib.request.Request(url, headers={"Accept": "application/json"}) try: with urllib.request.urlopen(req, timeout=30) as resp: raw = resp.read().decode("utf-8") return json.loads(raw) except Exception as e: logger.warning(f"[qmt_bridge] GET {path} 失败: {e}") raise def _parse_kline_data( self, data: dict[str, Any], is_daily: bool ) -> list[dict[str, Any]]: """桥响应解析为统一记录列表。跳过全零记录（停牌/未交易）。""" records = [] raw_list = data.get("data") if isinstance(data, dict) else None if not isinstance(raw_list, list): return records for item in raw_list: try: t = str(item.get("time", "") or "") if not t: continue # 跳过全零记录（停牌/未交易） o = float(item.get("open") or 0) h = float(item.get("high") or 0) l = float(item.get("low") or 0) c = float(item.get("close") or 0) v = float(item.get("volume") or 0) if o <= 0 and h <= 0 and l <= 0 and c <= 0 and v <= 0: continue if is_daily: # 日 K: time = "YYYYMMDD" if len(t) >= 8: trade_date = f"{t[:4]}-{t[4:6]}-{t[6:8]}" else: continue records.append({ "trade_date": trade_date, "open": o, "high": h, "low": l, "close": c, "volume": v, }) else: # 5min K: time = "YYYYMMDDHHmmss" if len(t) >= 14: bar_time = f"{t[:4]}-{t[4:6]}-{t[6:8]} {t[8:10]}:{t[10:12]}:{t[12:14]}" elif len(t) >= 8: bar_time = f"{t[:4]}-{t[4:6]}-{t[6:8]} 00:00:00" else: continue amt = float(item.get("amount") or 0) records.append({ "bar_time": bar_time, "open": o, "high": h, "low": l, "close": c, "volume": v, "amount": amt, "turnover_rate": 0.0, # bridge 不提供 }) except (KeyError, ValueError, TypeError): continue return records @staticmethod def _normalize_kline(df: pd.DataFrame) -> pd.DataFrame: """日 K 标准化：类型转换 + OHLC 校验。""" if df.empty: return df for col in ["open", "high", "low", "close", "volume"]: if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.dropna(subset=["trade_date", "open", "high", "low", "close"]) df["volume"] = df["volume"].fillna(0) # OHLC 合理性检查 valid = ( (df["open"] > 0) & (df["high"] > 0) & (df["low"] > 0) & (df["close"] > 0) & (df["volume"] >= 0) & (df["high"] >= df[["open", "low", "close"]].max(axis=1) - 0.001) & (df["low"] <= df[["open", "high", "close"]].min(axis=1) + 0.001) ) df = df.loc[valid].sort_values("trade_date") return df.drop_duplicates(subset=["trade_date"], keep="last").reset_index(drop=True)