From daef609e8b6191b68e2d04e25fad4a98908a902f Mon Sep 17 00:00:00 2001 From: gao Date: Thu, 23 Jul 2026 23:39:59 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20QMT=20Bridge=20=E8=AE=BE=E4=B8=BA?= =?UTF-8?q?=E4=B8=BB=E6=95=B0=E6=8D=AE=E6=BA=90=EF=BC=8C=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E6=8C=87=E6=95=B0=E6=97=A5K=E6=94=AF=E6=8C=81=E5=8F=8A?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=BA=90=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - QmtBridgeSource: provides 增加 index_daily，新增 fetch_index_daily 方法 - task_kline_index: 优先级改为 qmt_bridge → mairui → sina 三级降级 - task_kline_5min: 优先用 qmt_bridge，不可用时降级 mairui - task_kline_daily: 优先级加入 qmt_bridge（首位） - config: 新增 qmt_bridge_url 配置项 - registry: qmt_bridge 注册信息同步更新 - docs: 新增 DATASETS_AND_SOURCES.md，完整说明数据集与数据源依赖关系 - AGENTS.md: 同步更新指数数据源描述 --- AGENTS.md | 2 +- app/core/config.py | 3 + app/core/datasource/registry.py | 12 ++ app/core/sync/registry.py | 2 +- app/sources/qmt_bridge.py | 235 ++++++++++++++++++++++++++++++++ app/tasks/task_kline_5min.py | 20 ++- app/tasks/task_kline_daily.py | 5 +- app/tasks/task_kline_index.py | 38 ++++-- docs/DATASETS_AND_SOURCES.md | 152 +++++++++++++++++++++ 9 files changed, 448 insertions(+), 21 deletions(-) create mode 100644 app/sources/qmt_bridge.py create mode 100644 docs/DATASETS_AND_SOURCES.md diff --git a/AGENTS.md b/AGENTS.md index c5a7b85..9c1d93a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,7 +42,7 @@ task_watch 在后台运行（PID 见 `logs/task_watch.log`），持续跟踪运 |---|---|---|---|---| | `stock_basic` | `market_data.stocks` | 原始源 | 麦蕊 `/hslt/list`（主）+ Baostock `query_stock_basic`（备） | 全市场 A 股代码/名称/交易所/上市状态；`stock_basic` 自身可附带雪球股本快照 | | `kline_daily` | `market_data.kline_stock` | 原始源 | 雪球 `kline`（主）→ 麦蕊 `hsstock/history/*/d/n` → 新浪 `getKLineData`（备） | 全市场日 K OHLCV；支持 per-stock fallback | -| `kline_index` | `market_data.kline_index` + `indices` | 原始源 | 麦蕊 `hsindex/history`（主）→ 新浪指数日 K（备） | 上证/深证/创业板/沪深300/中证500/中证1000 | +| `kline_index` | `market_data.kline_index` + `indices` | 原始源 | QMT Bridge（主）→ 麦蕊 `hsindex/history` → 新浪指数日 K（备） | 上证/深证/创业板/沪深300/中证500/中证1000 | | `kline_5min` | `market_data.kline_5min` | 原始源 | 麦蕊 `hsstock/history/*/5/n` | 全市场 5 分钟 K；历史深度 2023-06-14 起 | | `tick_trade` | `market_data.tick_trade` | 原始源 | 麦蕊 `hsrl/zbjy` | 当天逐笔成交；每日 21:00 后发布 | | `moneyflow` | `market_data.moneyflow` | 原始源 | 麦蕊 `hsstock/history/transaction` | 主力/大/中/小单净额；每日 21:30 后发布 | diff --git a/app/core/config.py b/app/core/config.py index 5031202..913f72a 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -62,6 +62,9 @@ class Settings(BaseSettings): pg_password: str = "market_sync" pg_db_name: str = "market_data" + # QMT Bridge + qmt_bridge_url: str = "http://127.0.0.1:8610" + # Trading calendar trading_holidays: str = "" diff --git a/app/core/datasource/registry.py b/app/core/datasource/registry.py index a49442e..8b88352 100644 --- a/app/core/datasource/registry.py +++ b/app/core/datasource/registry.py @@ -23,6 +23,16 @@ logger = get_logger("datasource_health") _DATASOURCE_SEED: list[tuple[str, dict, str]] = [ + ( + "datasource_qmt_bridge", + { + "name": "QMT Bridge（本地行情桥）", + "provides": ["kline_daily", "kline_5min", "index_daily"], + "requiresCredential": False, + "note": "本地 QMT HTTP 桥，日K + 5分钟K + 指数日K 主源", + }, + "日 K + 5 分钟 K + 指数日 K 主源（替代 mairui）", + ), ( "datasource_xinlang", { @@ -248,10 +258,12 @@ def build_default_registry() -> None: from app.sources.baostock import BaostockSource from app.sources.xueqiu import XueqiuSource from app.sources.mairui import MairuiSource + from app.sources.qmt_bridge import QmtBridgeSource registry.register(SinaSource()) registry.register(BaostockSource()) registry.register(MairuiSource()) + registry.register(QmtBridgeSource()) if settings.xueqiu_token: registry.register(XueqiuSource()) else: diff --git a/app/core/sync/registry.py b/app/core/sync/registry.py index 4804e20..a52bdfe 100644 --- a/app/core/sync/registry.py +++ b/app/core/sync/registry.py @@ -53,7 +53,7 @@ SYNC_DEFINITIONS: list[dict[str, Any]] = [ "storage_uri": "PG market_data.kline_index + indices", "storage_layer": "pg", "management_role": "原始源", - "source": "新浪指数日K", + "source": "QMT Bridge（主）→ 麦蕊 → 新浪 逐级 fallback", "sync_script": "app.tasks.task_kline_index:run", "dependency_ids": [], "sort_order": 30, diff --git a/app/sources/qmt_bridge.py b/app/sources/qmt_bridge.py new file mode 100644 index 0000000..b3bfb1e --- /dev/null +++ b/app/sources/qmt_bridge.py @@ -0,0 +1,235 @@ +"""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) diff --git a/app/tasks/task_kline_5min.py b/app/tasks/task_kline_5min.py index d645232..a6bdc08 100644 --- a/app/tasks/task_kline_5min.py +++ b/app/tasks/task_kline_5min.py @@ -77,13 +77,25 @@ class SyncKline5Min(SyncTask): max_workers: int = 5, **kwargs, ) -> dict[str, Any]: - primary = ds_registry.get("datasource_mairui") + # QMT Bridge 优先，不可用时降级到 mairui + primary = ds_registry.get("datasource_qmt_bridge") or ds_registry.get("datasource_mairui") if primary is None: - return {"status": "error", "message": "5min 数据源 mairui 未注册"} + return {"status": "error", "message": "5min 数据源 qmt_bridge/mairui 均未注册"} ok, reason = is_source_ready(primary.key) if not ok: - mark_sync_blocked(self.dataset_id, message=f"{primary.key} 未就绪: {reason}") - return {"status": "blocked", "message": f"{primary.key} 未就绪: {reason}"} + # 如果 bridge 不可用但 mairui 可用，尝试切换 + fallback = ds_registry.get("datasource_mairui") if primary.key == "datasource_qmt_bridge" else None + if fallback: + fb_ok, fb_reason = is_source_ready(fallback.key) + if fb_ok: + logger.warning(f"[5min] {primary.key} 不可用 ({reason}), 降级到 {fallback.key}") + primary = fallback + else: + mark_sync_blocked(self.dataset_id, message=f"{primary.key}/{fallback.key} 均不可用") + return {"status": "blocked", "message": f"{primary.key}/{fallback.key} 均不可用"} + else: + mark_sync_blocked(self.dataset_id, message=f"{primary.key} 未就绪: {reason}") + return {"status": "blocked", "message": f"{primary.key} 未就绪: {reason}"} if codes: stock_codes = [to_code6(c) for c in codes] diff --git a/app/tasks/task_kline_daily.py b/app/tasks/task_kline_daily.py index 3eec44a..f6a0f13 100644 --- a/app/tasks/task_kline_daily.py +++ b/app/tasks/task_kline_daily.py @@ -33,9 +33,8 @@ from app.core.utils.logging import get_logger logger = get_logger("sync.kline_daily") -# 源优先级：麦蕊 → 雪球 → 新浪 -# 2026-07-21 改: 麦蕊15:10已有数据,雪球要15:40后才齐,麦蕊为主源提速 -PRIMARY_PRIORITY = ["datasource_mairui", "datasource_xueqiu", "datasource_xinlang"] +# 源优先级：QMT Bridge（本地） → 麦蕊 → 雪球 → 新浪 +PRIMARY_PRIORITY = ["datasource_qmt_bridge", "datasource_mairui", "datasource_xueqiu", "datasource_xinlang"] DEFAULT_START = "2018-01-01" # 写库批量：攒够 BATCH_SIZE 行就 executemany 一次 + commit BATCH_SIZE = 2000 diff --git a/app/tasks/task_kline_index.py b/app/tasks/task_kline_index.py index 1adaa51..305e552 100644 --- a/app/tasks/task_kline_index.py +++ b/app/tasks/task_kline_index.py @@ -1,6 +1,6 @@ """同步任务：六大指数日 K 线。 -数据流：新浪指数日K接口 → 写 kline_index + indices 表。""" +数据流：QMT Bridge（主）→ 麦蕊 → 新浪 逐级 fallback → 写 kline_index + indices 表。""" from __future__ import annotations from datetime import datetime @@ -24,32 +24,46 @@ MAJOR_INDICES = { "000852": {"name": "中证1000", "market": "CN", "category": "broad_market"}, } +# 数据源优先级：qmt_bridge（免限速免凭证）→ mairui → sina +SOURCE_PRIORITY = [ + ("datasource_qmt_bridge", True, "qmt_bridge_index_daily"), + ("datasource_mairui", True, "mairui_index_daily"), + ("datasource_xinlang", False, "sina_index_daily"), +] + class SyncKlineIndex(SyncTask): dataset_id = "kline_index" def _run(self, *, trigger_source: str = "manual", **kwargs) -> dict[str, Any]: - # 优先 mairui（指数无单日空窗问题），sina 作为 fallback - primary = ds_registry.get("datasource_mairui") - use_mairui = primary is not None and is_source_ready(primary.key)[0] - if not use_mairui: - primary = ds_registry.get("datasource_xinlang") + # 按优先级选择第一个就绪的数据源 + primary = None + use_suffix = True + source_label = "" + for key, suffix, label in SOURCE_PRIORITY: + src = ds_registry.get(key) + if src is not None and is_source_ready(src.key)[0]: + primary = src + use_suffix = suffix + source_label = label + break + if primary is None: - return {"status": "error", "message": "指数数据源均不可用（mairui + 新浪）"} + return {"status": "error", "message": "指数数据源均不可用（qmt_bridge + mairui + 新浪）"} ok, reason = is_source_ready(primary.key) if not ok: mark_sync_blocked(self.dataset_id, message=f"{primary.key} 未就绪: {reason}") return {"status": "blocked", "message": f"{primary.key} 未就绪: {reason}"} - self._progress(message=f"开始同步 {len(MAJOR_INDICES)} 个指数...") + self._progress(message=f"开始同步 {len(MAJOR_INDICES)} 个指数，数据源: {source_label}...") total = len(MAJOR_INDICES) ok_cnt = fail_cnt = 0 results = {} for i, (code, info) in enumerate(MAJOR_INDICES.items(), 1): try: - # mairui 需带交易所后缀（如 000300.SH），sina 用纯6位 - if use_mairui: + # qmt_bridge / mairui 需带交易所后缀（如 000300.SH），sina 用纯6位 + if use_suffix: # 000xxx 上证 → .SH；399xxx 深证 → .SZ if code.startswith("399"): suffix = ".SZ" @@ -82,7 +96,7 @@ class SyncKlineIndex(SyncTask): db_ops.upsert_index( index_code=code, index_name=info["name"], market=info["market"], category=info["category"], - source="mairui_index_daily" if use_mairui else "sina_index_daily", enabled=True, + source=source_label, enabled=True, ) ok_cnt += 1 results[code] = { @@ -100,7 +114,7 @@ class SyncKlineIndex(SyncTask): current=i, total=total, current_step=code, ) - msg = f"六大指数 {ok_cnt}成 {fail_cnt}败" + msg = f"六大指数 {ok_cnt}成 {fail_cnt}败（{source_label}）" return { "status": "ok" if fail_cnt == 0 else "warning", "message": msg, diff --git a/docs/DATASETS_AND_SOURCES.md b/docs/DATASETS_AND_SOURCES.md new file mode 100644 index 0000000..9771421 --- /dev/null +++ b/docs/DATASETS_AND_SOURCES.md @@ -0,0 +1,152 @@ +# 数据集与数据源依赖说明 + +> 本文档列出项目同步的全部数据集、各数据集依赖的数据源及优先级， +> 以及数据集之间的上下游依赖关系。 + +--- + +## 一、数据集总览 + +项目目前定义 **12 个同步数据集**，按类型分为原始源（依赖外部 API）和衍生源（本地计算）。 + +### 1.1 原始源 + +| 数据集 ID | 目标表 | 说明 | 数据源与优先级 | +|---|---|---|---| +| `stock_basic` | `market_data.stocks` | 全市场 A 股代码/名称/交易所/上市状态 + 股本快照（雪球） | **Baostock** `query_stock_basic`（主）→ 麦蕊 `/hslt/list`（备） | +| `kline_daily` | `market_data.kline_stock` | 全市场日 K OHLCV | **QMT Bridge** `/kline?period=1d`（主）→ **麦蕊** `hsstock/history/*/d/n` → **雪球** `kline` → **新浪** `getKLineData`，per-stock 逐级 fallback | +| `kline_index` | `market_data.kline_index` + `indices` | 上证/深证/创业板/沪深300/中证500/中证1000 日线 | **QMT Bridge** `/kline?period=1d`（主）→ **麦蕊** `hsindex/history` → **新浪** 指数日 K | +| `kline_5min` | `market_data.kline_5min` | 全市场 5 分钟 K 线 | **QMT Bridge** `/kline?period=5m`（主）→ **麦蕊** `hsstock/history/*/5/n`（备） | +| `tick_trade` | `market_data.tick_trade` | 当天逐笔成交（每日 21:00 发布） | **麦蕊** `hsrl/zbjy`（唯一） | +| `moneyflow` | `market_data.moneyflow` | 个股资金流（每日 21:30 发布，21:35 触发） | **麦蕊** `hsstock/history/transaction`（唯一） | +| `longhubang` | `market_data.longhubang_daily` + `longhubang_seat` | 龙虎榜聚合层 + 席位层（每日 22:00 触发） | **akshare** `stock_lhb_detail_em` + `stock_lhb_stock_detail_em`（东方财富封装，无替代源） | +| `stock_node` | `market_data.node_categories` + `nodes` + `stock_node_map` | 股票-指数/行业/概念映射（每周六 11:30） | **麦蕊** `/hszg/{list,gg,zg}`（唯一） | +| `mairui_indicators` | `market_data.kline_stock_macd_daily` + `kdj_daily` + `boll_daily` | MACD/KDJ/BOLL 日频技术指标（每日 16:40 增量） | **麦蕊** `/hsstock/history/{macd,kdj,boll}/{symbol}/d/n`（唯一） | +| `share_snapshot` | `market_data.stocks.total_share/float_share` + `share` 表 | 全市场股本快照（雪球 `quote_detail`，需 `XUEQIU_TOKEN`） | **雪球** `quote_detail`（唯一） | + +### 1.2 衍生源（本地计算，无外部接口） + +| 数据集 ID | 目标表 | 说明 | 上游依赖 | +|---|---|---|---| +| `market_regime` | `market_data.market_regime_daily` | 涨跌家数 / advance_ratio / 恐慌标记 | **`kline_daily`**（本地 `kline_stock` 聚合） | +| `mairui_ma_daily` | `market_data.kline_stock_ma_daily` | MA5/10/20/60 均线 | **`kline_daily`**（本地 `kline_stock.close` 计算） | + +> 注：`mairui_ma_daily` 本可走麦蕊 `/hsdata` 端点，但免费 licence 返回"数据不存在"，故改为本地计算。 + +--- + +## 二、数据源总览 + +项目注册了 **6 个数据源**，各有 credential 要求和提供的能力。 + +| 数据源 Key | 名称 | 需凭证 | 提供能力 | 特性 | +|---|---|---|---|---| +| `datasource_qmt_bridge` | QMT Bridge（本地行情桥） | 否 | `kline_daily`, `kline_5min`, `index_daily` | 本地 xtquant HTTP 桥，免限速，低延迟，局域网 | +| `datasource_mairui` | 麦蕊智数（mairui.club） | 需 `MAIRUI_LICENCE` | `kline_daily`, `kline_5min`, `index_daily`, `stock_basic`, `moneyflow`, `tick_trade`, `stock_node`, `indicator_daily` | 数据最全，免费 licence 1 分钟 300 次 | +| `datasource_xinlang` | 新浪财经 | 否 | `kline_daily`, `index_daily` | 免费，RPS 3/s 限流 | +| `datasource_baostock` | Baostock | 否 | `kline_daily`, `stock_basic`, `industry` | 免费，基础信息 + 行业映射 | +| `datasource_xueqiu` | 雪球 | 需 `XUEQIU_TOKEN` | `kline_daily`, `share` | 股本快照 + K 线复核 | +| `datasource_akshare_lhb` | akshare 龙虎榜（东方财富） | 否 | `longhubang_daily`, `longhubang_seat` | 龙虎榜唯一源 | + +--- + +## 三、数据依赖关系 + +### 3.1 依赖拓扑图 + +``` +stock_basic (Baostock + 雪球) +├── kline_daily (QMT Bridge → 麦蕊 → 雪球 → 新浪) +│ ├── market_regime (衍生：本地 kline_stock 聚合) +│ └── mairui_ma_daily (衍生：本地 kline_stock.close 计算) +├── kline_5min (QMT Bridge → 麦蕊) +├── tick_trade (麦蕊) +├── moneyflow (麦蕊) +├── longhubang (akshare/东财) +├── stock_node (麦蕊) +└── share_snapshot (雪球) + +kline_index (QMT Bridge → 麦蕊 → 新浪) — 独立，无上游依赖 +``` + +### 3.2 依赖表 + +| 数据集 | 直接上游依赖 | 依赖说明 | +|---|---|---| +| `stock_basic` | 无 | 根任务，提供股票代码列表 | +| `kline_daily` | `stock_basic` | 依赖股票列表决定拉取范围 | +| `kline_index` | 无 | 独立运行，不依赖股票列表 | +| `kline_5min` | `stock_basic` | 依赖股票列表决定拉取范围 | +| `tick_trade` | `stock_basic` | 同上 | +| `moneyflow` | `stock_basic` | 同上 | +| `longhubang` | `stock_basic` | 同上 | +| `stock_node` | `stock_basic` | 同上 | +| `mairui_indicators` | `stock_basic` | 同上 | +| `share_snapshot` | `stock_basic` | 同上 | +| `market_regime` | `kline_daily` | 需 `kline_stock` 数据完整 | +| `mairui_ma_daily` | `kline_daily` | 需 `kline_stock.close` 数据完整 | + +--- + +## 四、默认调度时间（工作日） + +| 时间 | 数据集 | 备注 | +|---|---|---| +| 09:00 | `stock_basic` | 开盘前刷新 | +| 15:30 | `kline_index` | 指数日线 | +| 15:40 | `kline_daily` | 个股日线 | +| 16:00 | `kline_5min` | 5 分钟线 | +| 15:15 | `market_regime` | 市场情绪 | +| 16:30 | `mairui_ma_daily` | MA 均线 | +| 16:40 | `mairui_indicators` | MACD/KDJ/BOLL | +| 21:35 | `moneyflow` | 资金流（等 21:30 发布） | +| 22:00 | `longhubang` | 龙虎榜 | +| 周六 11:30 | `stock_node` | 节点映射 | +| 周六 11:30 | `share_snapshot` | 股本快照（周度） | + +--- + +## 五、数据回填级联规则 + +> 上游数据被回填或修复后，下游依赖它的衍生数据集必须手动/自动重跑， +> 否则会出现"上游新、下游旧"的不一致。 + +| 上游任务/表 | 触发条件 | 必须重跑的下游任务 | +|---|---|---| +| `kline_daily` / `kline_stock` | 回填历史 K 线、修复错误日线、补充漏掉的股票/日期 | `market_regime`、`mairui_ma_daily` | +| `stock_basic` / `stocks` | 新上市/退市、代码变更、上市状态修正 | `kline_daily`、`kline_5min`、`tick_trade`、`moneyflow`、`longhubang`、`stock_node`、`share_snapshot` | +| `kline_stock`（任意核心字段修复） | close/volume 等核心字段修正 | `market_regime`、`mairui_ma_daily` | + +**操作建议：** + +- 单次少量回填（如几只股票、几天）：用 CLI 参数指定 `codes` / `start` / `end`，然后按上表手动触发下游。 +- 大量回填（如全市场、多月/多年历史）：先跑上游，再按依赖链顺序跑下游；必要时禁用当日独立 timer，避免与 runall 并发。 +- 每日巡检 `bin/daily_sync_check.py` 已增加数据一致性检查：对比 `kline_stock` 与 `kline_stock_ma_daily`、`market_regime_daily` 的最新日期与缺失行数，发现缺口即告警。 + +--- + +## 六、数据源降级策略 + +各同步任务在代码中定义了具体的降级逻辑： + +| 任务 | 降级链 | 降级触发条件 | +|---|---|---| +| `kline_daily` | QMT Bridge → 麦蕊 → 雪球 → 新浪 | per-stock 逐级 fallback，雪球失败自动降级到麦蕊/新浪 | +| `kline_index` | QMT Bridge → 麦蕊 → 新浪 | 选第一个 `is_source_ready` 通过的源 | +| `kline_5min` | QMT Bridge → 麦蕊 | bridge 不可用时检查 mairui 是否就绪 | +| `tick_trade` | 仅麦蕊 | 外层 FETCH_HARD_TIMEOUT=120s 防止永久挂起 | +| `moneyflow` | 仅麦蕊 | 唯一源 | +| `longhubang` | 仅 akshare | 唯一源，无替代 | +| `stock_node` | 仅麦蕊 | 唯一源 | +| `stock_basic` | Baostock（主）→ 麦蕊（备） | Baostock 失败时切换 | +| `mairui_indicators` | 仅麦蕊 | 唯一源 | + +--- + +## 七、数据源健康检查 + +系统通过 `app/core/datasource/registry.py` 维护数据源健康状态： + +- **周期检查**：后台线程每 30 分钟对所有注册数据源执行一次 `health_check()`。 +- **`is_source_ready()`**：综合判断凭证是否配置 + `is_available()` 是否通过 + 最近健康检查是否成功，返回 `(bool, reason)`。 +- **`pick_source(capability)`**：遍历注册表，返回第一个 `is_source_ready()` 通过且提供该能力的数据源。