343 lines
15 KiB
Python
343 lines
15 KiB
Python
"""ETF 全市场数据同步（列表走东财快照，K 线走 Tushare fund_daily）。
|
||
|
||
分工：
|
||
- 东财 push2 clist：全市场 ETF 名单 + 总市值/流通市值/换手率（quicksync 镜像上
|
||
fund_etf_basic 不存在，规模字段无替代源）-> etf_basic；
|
||
- Tushare fund_daily（quicksync 可用，与股票 daily 同源同控频）-> candles
|
||
不复权底座，单位换算与股票一致（vol 手->份 ×100、amount 千元->元 ×1000）。
|
||
|
||
同步策略（全串行，Tushare 按分钟限频，并发无意义）：
|
||
- 逐日模式：交易日历里尚无任何 ETF 日线的日期，一天一调用拿全市场基金日线
|
||
（过滤到 etf_basic 符号），日常增量通常只有当天 1 次调用；
|
||
- 逐只模式：无任何缓存的 ETF（新上市/历史缺口）按 ts_code 全量拉取，每次 1 调用；
|
||
full=true 时对所有 ETF 重拉（修数/回补 amount 用）。
|
||
|
||
- 进程内后台任务（与 screener.market_sync 同款模式），前端轮询 /api/etf/sync/status；
|
||
- 复权：quicksync 无 fund_adj_factor，ETF 暂无因子，qfq/hfq 切换时按无因子
|
||
原样返回（api._adjust_bars 的既有语义）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
|
||
from sqlalchemy import delete, func, select, text
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from .. import cache
|
||
from ..config import settings
|
||
from . import etf_provider
|
||
from .sync_utils import call_retry, get_pro_lazy, utcnow
|
||
|
||
# 进程内单例任务状态（uvicorn 单进程场景够用）
|
||
_state: dict = {
|
||
"running": False,
|
||
"step": None,
|
||
"total": 0, # 本次需处理的单元数（缺失交易日 + 需拉取的 ETF 只数）
|
||
"done": 0,
|
||
"error": None,
|
||
"started_at": None,
|
||
"finished_at": None,
|
||
}
|
||
_task: asyncio.Task | None = None
|
||
_lock = asyncio.Lock()
|
||
|
||
_BATCH = 3000 # upsert 分批行数（asyncpg 单语句参数上限 32766，10 列/行）
|
||
# fund_daily 返回全市场基金 ~2100 行，一天一批远小于上限
|
||
|
||
|
||
def _parse_d(s: str) -> datetime:
|
||
return datetime.strptime(str(s), "%Y%m%d")
|
||
|
||
|
||
async def _sync_spot(session: AsyncSession) -> int:
|
||
"""快照 -> etf_basic（upsert + 删除已退市），返回列表行数。"""
|
||
from ..models import EtfBasic
|
||
|
||
async with etf_provider.new_client() as client:
|
||
rows = await etf_provider.fetch_etf_spot(client)
|
||
now = utcnow()
|
||
stmt = pg_insert(EtfBasic).values([{**r, "updated_at": now} for r in rows])
|
||
stmt = stmt.on_conflict_do_update(
|
||
index_elements=["ts_code"],
|
||
set_={
|
||
"name": stmt.excluded.name,
|
||
"exchange": stmt.excluded.exchange,
|
||
"total_mv": stmt.excluded.total_mv,
|
||
"circ_mv": stmt.excluded.circ_mv,
|
||
"turnover_rate": stmt.excluded.turnover_rate,
|
||
"updated_at": stmt.excluded.updated_at,
|
||
},
|
||
)
|
||
await session.execute(stmt)
|
||
# 快照外的 ETF 已退市（东财列表不再返回）
|
||
await session.execute(delete(EtfBasic).where(EtfBasic.ts_code.not_in({r["ts_code"] for r in rows})))
|
||
await session.commit()
|
||
return len(rows)
|
||
|
||
|
||
def _fetch_day_sync(pro, d: str) -> list[dict]:
|
||
"""拉某交易日全市场场内基金日线（fund_daily；未生成的日期返回空）。"""
|
||
time.sleep(settings.screener_sync_interval)
|
||
df = call_retry(pro.fund_daily, trade_date=d)
|
||
if df is None or df.empty:
|
||
return []
|
||
rows = []
|
||
for _, r in df.iterrows():
|
||
amt = r.get("amount")
|
||
rows.append({
|
||
"ts": _parse_d(d),
|
||
"ts_code": r["ts_code"],
|
||
"open": float(r["open"]), "high": float(r["high"]),
|
||
"low": float(r["low"]), "close": float(r["close"]),
|
||
"vol": float(r["vol"]), # 手
|
||
"amount": (float(amt) if amt is not None and amt == amt else None), # 千元
|
||
})
|
||
return rows
|
||
|
||
|
||
def _fetch_symbol_sync(pro, ts_code: str, start: str | None, end: str | None) -> list[dict]:
|
||
"""按 ts_code 增量/全量拉单只 ETF 日线（start=None 即上市以来全量）。"""
|
||
time.sleep(settings.screener_sync_interval)
|
||
df = call_retry(pro.fund_daily, ts_code=ts_code, start_date=start, end_date=end)
|
||
if df is None or df.empty:
|
||
return []
|
||
df = df.sort_values("trade_date")
|
||
rows = []
|
||
for _, r in df.iterrows():
|
||
amt = r.get("amount")
|
||
rows.append({
|
||
"ts": _parse_d(r["trade_date"]),
|
||
"ts_code": ts_code,
|
||
"open": float(r["open"]), "high": float(r["high"]),
|
||
"low": float(r["low"]), "close": float(r["close"]),
|
||
"vol": float(r["vol"]),
|
||
"amount": (float(amt) if amt is not None and amt == amt else None),
|
||
})
|
||
return rows
|
||
|
||
|
||
def _upsert_candles_stmt(batch: list[dict]):
|
||
from ..models import Candle
|
||
|
||
stmt = pg_insert(Candle).values(batch)
|
||
return stmt.on_conflict_do_update(
|
||
index_elements=["symbol", "timeframe", "ts"],
|
||
set_={"open": stmt.excluded.open, "high": stmt.excluded.high,
|
||
"low": stmt.excluded.low, "close": stmt.excluded.close,
|
||
"volume": stmt.excluded.volume,
|
||
# 增量源缺额时保留库内旧值（与股票底座同语义）
|
||
"amount": func.coalesce(stmt.excluded.amount, Candle.amount)},
|
||
)
|
||
|
||
|
||
def _day_batch(rows: list[dict], sym_set: set[str], d_str: str) -> list[dict]:
|
||
"""某日 fund_daily 行 -> candles 批（过滤到 etf_basic 符号；手->份、千元->元）。"""
|
||
return [
|
||
{"symbol": r["ts_code"].split(".")[0], "timeframe": "1d",
|
||
"ts": _parse_d(d_str),
|
||
"open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"],
|
||
"volume": r["vol"] * 100.0,
|
||
"amount": (r["amount"] * 1000.0) if r["amount"] is not None else None,
|
||
"turnover": None}
|
||
for r in rows if r["ts_code"].split(".")[0] in sym_set
|
||
]
|
||
|
||
|
||
def _symbol_batch(symbol: str, rows: list[dict]) -> list[dict]:
|
||
return [
|
||
{"symbol": symbol, "timeframe": "1d", "ts": r["ts"],
|
||
"open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"],
|
||
"volume": r["vol"] * 100.0,
|
||
"amount": (r["amount"] * 1000.0) if r["amount"] is not None else None,
|
||
"turnover": None}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
async def _write_batches(session: AsyncSession, batch: list[dict]) -> None:
|
||
for i in range(0, len(batch), _BATCH):
|
||
await session.execute(_upsert_candles_stmt(batch[i : i + _BATCH]))
|
||
await session.commit()
|
||
|
||
|
||
async def _backfill_list_dates(session: AsyncSession, dates: dict[str, str]) -> None:
|
||
"""首根 K 线日回填 list_date（仅空缺处）：一条 UPDATE ... FROM (VALUES)。"""
|
||
pairs = list(dates.items())
|
||
for i in range(0, len(pairs), 1000):
|
||
chunk = pairs[i : i + 1000]
|
||
vals = ", ".join(f"(:s{j}, :d{j})" for j in range(len(chunk)))
|
||
params = {f"s{j}": s for j, (s, _) in enumerate(chunk)}
|
||
params.update({f"d{j}": d for j, (_, d) in enumerate(chunk)})
|
||
await session.execute(text(
|
||
f"UPDATE etf_basic e SET list_date = v.d FROM (VALUES {vals}) AS v(symbol, d) "
|
||
"WHERE e.symbol = v.symbol AND e.list_date IS NULL"
|
||
), params)
|
||
|
||
|
||
async def _run_sync(full: bool) -> None:
|
||
from ..db import async_session
|
||
from ..models import Candle, EtfBasic, TradeCalendar
|
||
|
||
try:
|
||
pro = await asyncio.to_thread(get_pro_lazy)
|
||
|
||
# 1) 快照 -> etf_basic
|
||
_state["step"] = "正在拉取 ETF 列表"
|
||
async with async_session() as session:
|
||
n_list = await _sync_spot(session)
|
||
etfs = (await session.execute(
|
||
select(EtfBasic.ts_code, EtfBasic.symbol, EtfBasic.exchange, EtfBasic.list_date)
|
||
)).all()
|
||
# 零缓存 ETF（新上市/历史缺口，需逐只全量拉）：
|
||
# NOT EXISTS 走 (symbol,timeframe,ts) 索引探测，1514 次 ms 级；
|
||
# 比对「全 ETF 符号 GROUP BY max(ts)」（扫数百万索引行）便宜得多
|
||
have_any = set((await session.execute(text(
|
||
"SELECT e.symbol FROM etf_basic e WHERE EXISTS (SELECT 1 FROM candles c "
|
||
"WHERE c.symbol = e.symbol AND c.timeframe = '1d')"
|
||
))).scalars())
|
||
# 已落库的 ETF 交易日（YYYYMMDD）：逐日模式的跳过依据。
|
||
# 只看近 40 天（索引范围扫）——更早的历史缺口由 full 全量重拉兜底，
|
||
# 全表 distinct 对千万行 candles 表要几十秒，不能每次同步都付
|
||
fresh = not have_any or full
|
||
have_dates: set[str] = set() if fresh else {
|
||
r.strftime("%Y%m%d") for r in (await session.execute(
|
||
select(func.distinct(func.date(Candle.ts))).where(
|
||
Candle.timeframe == "1d",
|
||
Candle.symbol.in_(select(EtfBasic.symbol)),
|
||
Candle.ts >= datetime.now() - timedelta(days=40),
|
||
)
|
||
)).scalars() if r is not None
|
||
}
|
||
cal = (await session.execute(
|
||
select(TradeCalendar.trade_date).order_by(TradeCalendar.trade_date.desc())
|
||
)).scalars().all()
|
||
|
||
sym_set = {e.symbol for e in etfs}
|
||
today = datetime.now().strftime("%Y%m%d")
|
||
since40 = (datetime.now() - timedelta(days=40)).strftime("%Y%m%d")
|
||
|
||
# 2) 任务编排：逐日模式补近窗缺失交易日（fresh 库改走逐只全量）；
|
||
# 逐只模式拉零缓存 ETF，full=true 全量重拉
|
||
todo_dates = [] if fresh else [
|
||
d for d in cal if since40 <= d <= today and d not in have_dates
|
||
]
|
||
per_symbol = list(etfs) if fresh else [
|
||
e for e in etfs if e.symbol not in have_any
|
||
]
|
||
_state["total"] = len(todo_dates) + len(per_symbol)
|
||
_state["done"] = 0
|
||
fails: list[str] = []
|
||
new_list_dates: dict[str, str] = {}
|
||
written_rows = 0 # 实际写入的 K 线行数（决定是否作废 candles 相关缓存）
|
||
|
||
# 3) 逐日模式：一天一调用（交易日历空时跳过——由逐只模式兜底）
|
||
for d in sorted(todo_dates):
|
||
_state["step"] = f"正在同步 {d} 日线（{_state['done'] + 1}/{_state['total']}）"
|
||
try:
|
||
rows = await asyncio.to_thread(_fetch_day_sync, pro, d)
|
||
if rows:
|
||
batch = _day_batch(rows, sym_set, d)
|
||
if batch:
|
||
written_rows += len(batch)
|
||
async with async_session() as session:
|
||
await _write_batches(session, batch)
|
||
except Exception as ex: # noqa: BLE001 —— 单日失败不拖垮整体
|
||
fails.append(f"{d}: {str(ex)[:80]}")
|
||
_state["done"] += 1
|
||
|
||
# 4) 逐只模式：零缓存 ETF 全量拉取（start=None 即上市以来；full 同理）
|
||
for ts_code, symbol, _exch, has_list_date in per_symbol:
|
||
_state["step"] = f"正在同步 ETF 日线 {symbol}（{_state['done'] + 1}/{_state['total']}）"
|
||
try:
|
||
rows = await asyncio.to_thread(_fetch_symbol_sync, pro, ts_code, None, None)
|
||
except Exception as ex: # noqa: BLE001
|
||
fails.append(f"{ts_code}: {str(ex)[:80]}")
|
||
_state["done"] += 1
|
||
continue
|
||
if rows:
|
||
# 全量拉取的首根 = 真实上市日
|
||
if not has_list_date:
|
||
new_list_dates[symbol] = rows[0]["ts"].strftime("%Y%m%d")
|
||
batch = _symbol_batch(symbol, rows)
|
||
written_rows += len(batch)
|
||
try:
|
||
async with async_session() as session:
|
||
await _write_batches(session, batch)
|
||
except Exception as ex: # noqa: BLE001
|
||
fails.append(f"{ts_code}: {str(ex)[:80]}")
|
||
_state["done"] += 1
|
||
|
||
# 5) list_date 回填（一次 SQL）
|
||
if new_list_dates:
|
||
async with async_session() as session:
|
||
await _backfill_list_dates(session, new_list_dates)
|
||
await session.commit()
|
||
|
||
# candles 已更新：作废旧 K 线预览缓存（只在真的写了行时——空跑不作废，
|
||
# 免得每次同步都触发一轮 >10s 的统计重聚合）；etf 版本号作废 ETF 列表缓存
|
||
if written_rows:
|
||
await cache.bump_version("candles")
|
||
await cache.bump_version("etf")
|
||
|
||
parts = [f"同步完成（{n_list} 只 ETF"]
|
||
if todo_dates:
|
||
parts.append(f"{len(todo_dates)} 个交易日")
|
||
if per_symbol:
|
||
parts.append(f"{len(per_symbol)} 只逐只补数")
|
||
_state["step"] = "，".join(parts) + ")" + (f"，{len(fails)} 项失败" if fails else "")
|
||
if fails:
|
||
_state["error"] = "部分失败：" + "；".join(fails[:3]) + ("…" if len(fails) > 3 else "")
|
||
except Exception as e: # noqa: BLE001
|
||
_state["error"] = f"同步失败：{str(e)[:300]}"
|
||
_state["step"] = "同步失败"
|
||
finally:
|
||
_state["running"] = False
|
||
_state["finished_at"] = datetime.now()
|
||
|
||
|
||
async def start_sync(full: bool = False) -> dict:
|
||
"""幂等启动后台同步；已在跑则直接返回当前状态。full=true 所有 ETF 全量重拉。"""
|
||
global _task
|
||
async with _lock:
|
||
if _state["running"] and _task and not _task.done():
|
||
return dict(_state)
|
||
_state.update({
|
||
"running": True, "step": "准备同步", "total": 0, "done": 0,
|
||
"error": None, "started_at": datetime.now(), "finished_at": None,
|
||
})
|
||
_task = asyncio.create_task(_run_sync(full))
|
||
return dict(_state)
|
||
|
||
|
||
async def get_status(session: AsyncSession) -> dict:
|
||
"""任务状态 + DB 实况。行情侧只做单符号 max(ts) 索引探测（candles 是千万行表，
|
||
全表聚合 >10s，绝不能落在轮询热路径上），取最早上市且已有日线的一只当「数据更新至」。"""
|
||
from ..models import Candle, EtfBasic
|
||
|
||
etfs = int(await session.scalar(select(func.count()).select_from(EtfBasic)) or 0)
|
||
# 探测样本：优先最早上市（历史最长最稳）且已有日线的一只（EXISTS 走索引，ms 级）
|
||
probe = await session.scalar(
|
||
text("""
|
||
SELECT e.symbol FROM etf_basic e
|
||
WHERE EXISTS (SELECT 1 FROM candles c
|
||
WHERE c.symbol = e.symbol AND c.timeframe = '1d')
|
||
ORDER BY e.list_date NULLS LAST, e.symbol LIMIT 1
|
||
""")
|
||
)
|
||
last = None
|
||
if probe:
|
||
last = await session.scalar(
|
||
select(func.max(Candle.ts)).where(Candle.symbol == probe, Candle.timeframe == "1d")
|
||
)
|
||
status = dict(_state)
|
||
status.update({
|
||
"stats": {"etfs": etfs},
|
||
"last_trade_date": last,
|
||
"last_synced_at": _state.get("finished_at") or _state.get("started_at"),
|
||
"ready": etfs > 0 and last is not None,
|
||
})
|
||
return status
|