gao
05635b76b9
状态： - 9 个 sync task（stock_basic / kline_daily / kline_index / kline_5min / moneyflow / industry_sector / sector_features / share_snapshot / market_regime） - 数据源：baostock + mairui + 雪球(pysnowball) + 新浪（4 个） - 项目级约束：永远不用 akshare（已落实） - kline_5min 改用 DB 快照统一全量/增量逻辑 - 零后端 Chrome 扩展 xueqiu_sync（独立项目）
110 lines
4.6 KiB
Python
110 lines
4.6 KiB
Python
"""同步任务：六大指数日 K 线。
|
||
|
||
数据流：新浪指数日K接口 → 写 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"},
|
||
}
|
||
|
||
|
||
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")
|
||
if primary is None:
|
||
return {"status": "error", "message": "指数数据源均不可用（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)} 个指数...")
|
||
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:
|
||
# 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="mairui_index_daily" if use_mairui else "sina_index_daily", 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}败"
|
||
return {
|
||
"status": "ok" if fail_cnt == 0 else "warning",
|
||
"message": msg,
|
||
"ok": ok_cnt, "fail": fail_cnt,
|
||
"details": results,
|
||
}
|