"""同步任务：全市场股票基础信息 + 股本快照。 数据流： 麦蕊智数 hslt/list（主）→ Baostock.query_stock_basic()（降级）→ 解析 → upsert stocks 可选：雪球 quote_detail 刷新股本（需要 XUEQIU_TOKEN） """ from __future__ import annotations import re import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any from app.core.config import settings from app.core.db import ops as db_ops from app.core.datasource.base import registry as ds_registry from app.core.sync.base import SyncTask from app.core.sync.registry import mark_sync_blocked from app.core.utils.logging import get_logger logger = get_logger("sync.stocks_basic") def _derive_listing_status(name: str) -> str: n = name.strip() if n.startswith("*ST") or n.startswith("ST"): return "st" return "normal" class SyncStocksBasic(SyncTask): dataset_id = "stock_basic" def _run(self, *, trigger_source: str = "manual", with_share: bool = True, **kwargs) -> dict[str, Any]: self._progress(message="选择股票基础信息数据源...") rows: list[dict] = [] source_name = "" # 1. 优先使用麦蕊智数 mairui = ds_registry.get("datasource_mairui") if mairui is not None: ok, msg = mairui.is_available() if ok: self._progress(message="使用麦蕊智数获取股票列表...") rows = mairui.fetch_stock_list() if rows: source_name = "麦蕊智数" self._progress(message=f"麦蕊智数返回 {len(rows)} 只股票") # 2. 无数据则报错 if not rows: return {"status": "error", "message": "麦蕊智数未返回股票数据"} self._progress( message=f"获取到 {len(rows)} 只 A 股，开始写入...", current=0, total=len(rows), ) new_cnt = upd_cnt = unchanged = 0 existing = {s["code"]: s for s in db_ops.fetch_all_stocks()} # 准备批量写入（性能：executemany 一次写 500 条，比单条快 10-20 倍） bulk_rows = [] for info in rows: code = info["code"] old = existing.get(code) if old is None: new_cnt += 1 elif (old.get("name") != info["name"] or old.get("exchange") != info["exchange"] or old.get("listing_status", "normal") != info["listing_status"]): upd_cnt += 1 else: unchanged += 1 bulk_rows.append({ "code": code, "name": info["name"], "exchange": info["exchange"], "listing_status": info["listing_status"], }) # 分块 executemany 写入 + 进度更新 CHUNK = 500 for i in range(0, len(bulk_rows), CHUNK): db_ops.upsert_stocks_bulk(bulk_rows[i: i + CHUNK]) done = min(i + CHUNK, len(bulk_rows)) if done % 500 == 0 or done == len(bulk_rows): self._progress( message=f"写入数据库 {done}/{len(bulk_rows)}...", current=done, total=len(bulk_rows), current_step=bulk_rows[done - 1]["code"], ) # 退市检测 delisted = 0 api_codes = {r["code"] for r in rows} for code, old in existing.items(): if code not in api_codes and old.get("listing_status") != "delisted": db_ops.upsert_stock( code=code, name=old.get("name", ""), exchange=old.get("exchange", ""), list_date=old.get("list_date", ""), listing_status="delisted", ) delisted += 1 msg = f"股票基础信息 {new_cnt}新 {upd_cnt}更 {unchanged}无变化 {delisted}退市" self._progress(message=msg, current=len(rows), total=len(rows)) # 股本快照（可选 + 需要 token） share_msg = "" if with_share and settings.xueqiu_token: self._progress(message="股票基础信息完成，开始刷新股本快照...") xq = ds_registry.get("datasource_xueqiu") if xq is not None: try: share_ok, share_skip, share_fail = self._refresh_shares( xq, [r["code"] for r in rows] ) share_msg = f"；股本 {share_ok}成 {share_skip}跳 {share_fail}败" except Exception as e: share_msg = f"；股本刷新异常: {e}" else: if with_share and not settings.xueqiu_token: share_msg = "；未配置 XUEQIU_TOKEN，跳过股本刷新" mark_sync_blocked(self.dataset_id, message=msg + share_msg) return { "status": "ok" if not share_msg.startswith("；未配置") else "blocked", "source": source_name, "message": f"[{source_name}] {msg}{share_msg}", "total": len(rows), "new": new_cnt, "updated": upd_cnt, "delisted": delisted, "unchanged": unchanged, } def _fetch_with_timeout(self, bs, *, timeout: int) -> list[dict]: result: list = [None] exc: list = [None] def _do(): try: result[0] = bs.fetch_stock_basic() except Exception as e: exc[0] = e t = threading.Thread(target=_do, daemon=True) t.start() t.join(timeout) if t.is_alive(): logger.warning(f"Baostock 拉取超时（{timeout}s）") return [] if exc[0]: raise exc[0] return result[0] or [] def _refresh_shares( self, xq, codes: list[str], max_workers: int = 10, ) -> tuple[int, int, int]: code6_list = [] for c in codes: c6 = re.sub(r"^(SH|SZ|BJ)", "", c) if c6.isdigit() and len(c6) == 6: code6_list.append(c6) code6_list = sorted(set(code6_list)) today = time.strftime("%Y-%m-%d") existing = {s["code"]: s for s in db_ops.fetch_all_stocks()} todo = [] for c6 in code6_list: old = existing.get(c6, {}) if str(old.get("share_updated_at", "") or "") == today and old.get("total_share", 0) > 0: continue todo.append(c6) if not todo: return 0, len(code6_list), 0 ok = skip = fail = 0 with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = {pool.submit(xq.fetch_share_snapshot, c6): c6 for c6 in todo} done = 0 for future in as_completed(futures): c6 = futures[future] done += 1 try: r = future.result() except Exception as e: fail += 1 logger.warning(f"[share {c6}] 异常: {e}") continue if r is None: fail += 1 else: # trade_date 必须是源数据返回的交易日(数据源层用 effective_market_date # 兜底为"最近已完成交易日"),绝不允许用 today 兜底 — 7月9日 早盘跑 # 时 today=7月9日,但股本快照实际对应 7月8日 收盘。 trade_date = r.get("trade_date") if not trade_date: fail += 1 logger.warning(f"[share {c6}] 源数据未返回 trade_date,跳过(避免用运行日兜底)") continue db_ops.update_stock_share_snapshot( code=self._to_hermes(c6), total_share=r["total_share"], float_share=r["float_share"], trade_date=trade_date, ) ok += 1 if done % 100 == 0: self._progress( message=f"刷新股本 {done}/{len(todo)}...", current=done, total=len(todo), current_step=c6, ) skip = len(code6_list) - len(todo) return ok, skip, fail @staticmethod def _to_hermes(code6: str) -> str: if code6.startswith(("5", "6", "9")): return f"SH{code6}" if code6.startswith(("4", "8")): return f"BJ{code6}" return f"SZ{code6}"