gao
daef609e8b
- 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: 同步更新指数数据源描述
124 lines
5.1 KiB
Python
124 lines
5.1 KiB
Python
"""同步任务：六大指数日 K 线。
|
||
|
||
数据流：QMT Bridge（主）→ 麦蕊 → 新浪 逐级 fallback → 写 kline_index + indices 表。"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from app.core.db import ops as db_ops
|
||
from app.core.datasource.base import registry as ds_registry
|
||
from app.core.datasource.registry import is_source_ready
|
||
from app.core.sync.base import SyncTask, _is_market_closed
|
||
from app.core.sync.registry import mark_sync_blocked
|
||
from app.core.utils.logging import get_logger
|
||
|
||
logger = get_logger("sync.kline_index")
|
||
|
||
MAJOR_INDICES = {
|
||
"000001": {"name": "上证指数", "market": "CN", "category": "broad_market"},
|
||
"399001": {"name": "深证成指", "market": "CN", "category": "broad_market"},
|
||
"399006": {"name": "创业板指", "market": "CN", "category": "broad_market"},
|
||
"000300": {"name": "沪深300", "market": "CN", "category": "broad_market"},
|
||
"000905": {"name": "中证500", "market": "CN", "category": "broad_market"},
|
||
"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]:
|
||
# 按优先级选择第一个就绪的数据源
|
||
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": "指数数据源均不可用（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)} 个指数，数据源: {source_label}...")
|
||
total = len(MAJOR_INDICES)
|
||
ok_cnt = fail_cnt = 0
|
||
results = {}
|
||
|
||
for i, (code, info) in enumerate(MAJOR_INDICES.items(), 1):
|
||
try:
|
||
# qmt_bridge / mairui 需带交易所后缀（如 000300.SH），sina 用纯6位
|
||
if use_suffix:
|
||
# 000xxx 上证 → .SH；399xxx 深证 → .SZ
|
||
if code.startswith("399"):
|
||
suffix = ".SZ"
|
||
else:
|
||
suffix = ".SH"
|
||
fetch_code = f"{code}{suffix}"
|
||
else:
|
||
fetch_code = code
|
||
df = primary.fetch_index_daily(fetch_code, start="1990-01-01", end="2099-12-31")
|
||
if df is None or df.empty:
|
||
raise RuntimeError("返回空数据")
|
||
# 盘中未收盘则去掉当天
|
||
if not _is_market_closed():
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
df = df[df["trade_date"].dt.strftime("%Y-%m-%d") < today]
|
||
|
||
rows = [
|
||
{
|
||
"index_code": code,
|
||
"trade_date": r["trade_date"].strftime("%Y-%m-%d"),
|
||
"open": float(r["open"]),
|
||
"high": float(r["high"]),
|
||
"low": float(r["low"]),
|
||
"close": float(r["close"]),
|
||
"volume": float(r["volume"]),
|
||
}
|
||
for _, r in df.iterrows()
|
||
]
|
||
db_ops.upsert_kline_index(rows)
|
||
db_ops.upsert_index(
|
||
index_code=code, index_name=info["name"],
|
||
market=info["market"], category=info["category"],
|
||
source=source_label, enabled=True,
|
||
)
|
||
ok_cnt += 1
|
||
results[code] = {
|
||
"rows": len(rows),
|
||
"date_min": rows[0]["trade_date"] if rows else "",
|
||
"date_max": rows[-1]["trade_date"] if rows else "",
|
||
}
|
||
except Exception as e:
|
||
fail_cnt += 1
|
||
results[code] = {"error": str(e)}
|
||
logger.warning(f"[index {code}] 失败: {e}")
|
||
|
||
self._progress(
|
||
message=f"指数 {i}/{total} 完成 ({ok_cnt}成 {fail_cnt}败)",
|
||
current=i, total=total, current_step=code,
|
||
)
|
||
|
||
msg = f"六大指数 {ok_cnt}成 {fail_cnt}败（{source_label}）"
|
||
return {
|
||
"status": "ok" if fail_cnt == 0 else "warning",
|
||
"message": msg,
|
||
"ok": ok_cnt, "fail": fail_cnt,
|
||
"details": results,
|
||
}
|