"""个股参考数据（tushare 参考数据版块，详情页按需单查懒加载）。 10 个常规 kind 走同一泛型管道：拉取 -> 精确去重 -> 按 df dtype 清洗 -> 排序 -> 5 年窗切片 -> 单行 JSON 快照 upsert（stock_reference，行即缓存，rows_json NULL=墓碑）。 repurchase 特殊：tushare 不支持按 ts_code 过滤（实测入参被忽略，返回全市场默认页）， 改为全市场按月分块拉取（单块触顶 2000 行则窗口减半递归）后按股拆分入库； 全局状态记在 stock_sync_state('_MARKET_', 'repurchase')——从未回填时后台任务回填 近 24 个月（本次请求返回现有行，可能暂空），状态过期(>7 天)时内联补拉缺失月份。 单位沿用各接口原始口径（详见 _KINDS 注释与文档）：top10 hold_amount 股、pledge 万股、 block_trade vol 万股 / amount 万元、repurchase vol 股 / amount 元、share_float float_share 股。 """ from __future__ import annotations import asyncio import calendar import json import logging import time from dataclasses import dataclass from datetime import date from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from ..config import settings from ..db import async_session from ..models import StockReference from .sync_utils import call_retry, f_clean, fresh, get_pro_lazy, read_sync_state, s_clean, upsert_sync_state, utcnow log = logging.getLogger(__name__) _REFRESH_DAYS = 7 @dataclass(frozen=True) class _Kind: api: str fields: str # 期望列的文档化清单（实际不传 fields，见 _fetch_kind_sync 注释） windowed: bool # True: 传 start_date(=N年前0101)+end_date(=今天) 窗口参数 order: tuple[tuple[str, bool], ...] # 排序 (列, 是否倒序)；多列时靠稳定多趟排序 slice_col: str # 窗口切片依据列（未来日期 >= cutoff 自然保留） group: tuple[str, ...] | None = None # 聚合键：share_float 每股东一行，按解禁事件聚合才可展示 years: int = 5 # 窗口年数（moneyflow 这类日频近况数据取 1 年） refresh_days: int = 7 # 新鲜度门控（moneyflow 每日盘后更新，须 1 天） def _window_start() -> str: return f"{date.today().year - 5}0101" def _window_start_for(spec: _Kind) -> str: return f"{date.today().year - spec.years}0101" def _today() -> str: return date.today().strftime("%Y%m%d") _TOP10_FIELDS = ("ts_code,ann_date,end_date,holder_name,hold_amount,hold_ratio," "hold_float_ratio,hold_change,holder_type") _PLEDGE_STAT_FIELDS = "ts_code,end_date,pledge_count,unrest_pledge,rest_pledge,total_share,pledge_ratio" _PLEDGE_DETAIL_FIELDS = ("ts_code,ann_date,holder_name,pledge_amount,start_date,end_date," "is_release,release_date,holding_amount,pledged_amount," "p_total_ratio,h_total_ratio,pledgor,is_buyback") _REPURCHASE_FIELDS = "ts_code,ann_date,end_date,proc,exp_date,vol,amount,high_limit,low_limit" _FLOAT_FIELDS = "ts_code,ann_date,float_date,float_share,float_ratio,holder_name,share_type" _BLOCK_FIELDS = "ts_code,trade_date,price,vol,amount,buyer,seller" _HOLDERNUM_FIELDS = "ts_code,ann_date,end_date,holder_num" _HOLDERTRADE_FIELDS = ("ts_code,ann_date,holder_name,holder_type,in_de,change_vol," "change_ratio,after_share,after_ratio,avg_price,total_share") _SHOCK_FIELDS = "ts_code,trade_date,name,trade_market,reason,period" # 同花顺资金流向（万元）；2027-07-06 起源头停供 5 日主力净额与占比（net_d5_amount） _MF_FIELDS = ("ts_code,trade_date,name,pct_change,latest,net_amount,net_d5_amount," "buy_lg_amount,buy_lg_amount_rate,buy_md_amount,buy_md_amount_rate," "buy_sm_amount,buy_sm_amount_rate") _KINDS: dict[str, _Kind] = { # 前十大股东/流通股东：报告期窗；期内按持股数排序（tushare 返回顺序不稳定） "top10_holders": _Kind("top10_holders", _TOP10_FIELDS, True, (("end_date", True), ("hold_amount", True)), "end_date"), "top10_floatholders": _Kind("top10_floatholders", _TOP10_FIELDS, True, (("end_date", True), ("hold_amount", True)), "end_date"), # 质押统计：周频全量（接口无窗口参数），取近 5 年切片 "pledge_stat": _Kind("pledge_stat", _PLEDGE_STAT_FIELDS, False, (("end_date", True),), "end_date"), "pledge_detail": _Kind("pledge_detail", _PLEDGE_DETAIL_FIELDS, True, (("ann_date", True),), "ann_date"), "share_float": _Kind("share_float", _FLOAT_FIELDS, False, (("float_date", True),), "float_date", # 每个解禁股东一行（IPO 原始股东解禁可达数千行，接口 6000 行封顶会截断尾部）， # 按解禁事件聚合：股数/占比求和，股东名并成「N名股东」 group=("ann_date", "float_date", "share_type")), "block_trade": _Kind("block_trade", _BLOCK_FIELDS, True, (("trade_date", True),), "trade_date"), # 同花顺资金流向：日频盘后更新，窗口 1 年、新鲜度 1 天（其余 kind 默认 5 年/7 天） "moneyflow": _Kind("moneyflow_ths", _MF_FIELDS, True, (("trade_date", True),), "trade_date", years=1, refresh_days=1), "holdernumber": _Kind("stk_holdernumber", _HOLDERNUM_FIELDS, True, (("end_date", True), ("ann_date", True)), "end_date"), "holdertrade": _Kind("stk_holdertrade", _HOLDERTRADE_FIELDS, True, (("ann_date", True),), "ann_date"), "shock": _Kind("stk_shock", _SHOCK_FIELDS, True, (("trade_date", True),), "trade_date"), # quicksync 镜像实测无该接口数据：管道照常，墓碑优雅降级 "high_shock": _Kind("stk_high_shock", _SHOCK_FIELDS, True, (("trade_date", True),), "trade_date"), } #: 路由白名单（repurchase 走全市场特殊管道，不在 _KINDS 泛型表里） REFERENCE_KINDS = frozenset(_KINDS) | {"repurchase"} def _kv(v): """排序键归一：None < 字符串 < 数值，避免混合类型比较抛错。""" if v is None: return (0, 0.0) if isinstance(v, str): return (1, v) return (2, float(v)) def _clean_df(df, keep_ts_code: bool = False) -> list[dict]: """按 df dtype 清洗（数值列 f_clean、其余列 s_clean），默认去掉冗余 ts_code 列，精确去重。 数值列判定必须用 pandas.api.types.is_numeric_dtype：pandas 3 的字符串列 dtype 是 'str' 而非 'object'，按 `!= "object"` 判断会把日期列误当数值列（f_clean 变 float）。 """ import pandas as pd df = df.drop_duplicates() num_cols = {c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])} skip_ts = {"ts_code"} if not keep_ts_code else set() rows: list[dict] = [] for _, r in df.iterrows(): row = {c: (f_clean(r[c]) if c in num_cols else s_clean(r[c])) for c in df.columns if c not in skip_ts} if any(v is not None for v in row.values()): rows.append(row) return rows def _group_rows(rows: list[dict], spec: _Kind) -> list[dict]: """share_float 类聚合：数值列求和、holder_name 并成「N名股东」。""" agg: dict[tuple, dict] = {} order_keys: list[tuple] = [] for r in rows: key = tuple(r.get(c) for c in spec.group) # type: ignore[arg-type] if key not in agg: agg[key] = { **{c: r.get(c) for c in spec.group}, # type: ignore[misc] "_num": {}, "_holders": [], } order_keys.append(key) a = agg[key] for c, v in r.items(): if isinstance(v, (int, float)) and c not in spec.group: # type: ignore[operator] a["_num"][c] = a["_num"].get(c, 0.0) + v h = r.get("holder_name") if h: a["_holders"].append(h) out = [] for key in order_keys: a = agg[key] row = {c: a[c] for c in spec.group} # type: ignore[misc] row.update({c: round(v, 4) for c, v in a["_num"].items()}) hs = a["_holders"] row["holder_name"] = f"{hs[0]} 等{len(hs)}名股东" if len(hs) > 1 else (hs[0] if hs else None) row["holder_count"] = len(hs) out.append(row) return out def _slice_and_sort(rows: list[dict], spec: _Kind) -> list[dict]: if spec.group: rows = _group_rows(rows, spec) if spec.slice_col: cutoff = _window_start_for(spec) rows = [r for r in rows if r.get(spec.slice_col) is None or r[spec.slice_col] >= cutoff] # 多列排序：逆序逐列稳定排序（Python sort 稳定，后排的列为主键） for col, desc in reversed(spec.order): rows.sort(key=lambda r, c=col: _kv(r.get(c)), reverse=desc) return rows def _fetch_kind_sync(ts_code: str, kind: str) -> list[dict]: """同步拉单股单分类（需在 to_thread 里跑）；无数据返回空列表。 不传 fields：所用字段全为接口默认显示列，而 quicksync 镜像对部分接口 （实测 share_float）传 fields 时会忽略 ts_code 过滤、返回全市场数据—— 所以下面再加一道 ts_code 后过滤兜底。 """ spec = _KINDS[kind] time.sleep(settings.screener_sync_interval) params: dict = {"ts_code": ts_code} if spec.windowed: params["start_date"] = _window_start_for(spec) params["end_date"] = _today() df = call_retry(getattr(get_pro_lazy(), spec.api), **params) if df is None or df.empty: return [] if "ts_code" in df.columns: df = df[df["ts_code"] == ts_code] return _slice_and_sort(_clean_df(df), spec) # ---------- 行即缓存读写 ---------- def _parse(rows_json: str | None) -> list[dict]: if not rows_json: return [] try: return json.loads(rows_json) except (ValueError, TypeError): return [] async def _read_rows(session: AsyncSession, ts_code: str, kind: str) -> list[dict]: row = (await session.execute( select(StockReference).where( StockReference.ts_code == ts_code, StockReference.kind == kind ))).scalar_one_or_none() return _parse(row.rows_json) if row is not None else [] async def _upsert_row(session: AsyncSession, ts_code: str, kind: str, rows: list[dict]) -> None: """写单行快照（不 commit）；空列表写墓碑（rows_json NULL）。""" stmt = pg_insert(StockReference).values( ts_code=ts_code, kind=kind, rows_json=json.dumps(rows, ensure_ascii=False) if rows else None, updated_at=utcnow(), ) stmt = stmt.on_conflict_do_update( index_elements=["ts_code", "kind"], set_={"rows_json": stmt.excluded.rows_json, "updated_at": stmt.excluded.updated_at}, ) await session.execute(stmt) # per-(code,kind) 锁：同键首次并发 N 个请求只有 1 个打 tushare，其余等锁后双检命中 _key_locks: dict[tuple[str, str], asyncio.Lock] = {} _guard = asyncio.Lock() async def _get_generic(session: AsyncSession, ts_code: str, kind: str) -> list[dict]: """常规 kind 的读穿透：新鲜直返；否则锁内双检 -> to_thread 拉 -> upsert。""" spec = _KINDS[kind] row = (await session.execute( select(StockReference).where( StockReference.ts_code == ts_code, StockReference.kind == kind ))).scalar_one_or_none() if row is not None and fresh(row.updated_at, spec.refresh_days): return _parse(row.rows_json) # 释放请求会话连接（close 而非 rollback：rollback 会 expire 鉴权缓存的 User，见 finance.py 注释） await session.close() key = (ts_code, kind) async with _guard: lock = _key_locks.setdefault(key, asyncio.Lock()) async with lock: async with async_session() as s2: row = (await s2.execute( select(StockReference).where( StockReference.ts_code == ts_code, StockReference.kind == kind ))).scalar_one_or_none() if row is not None and fresh(row.updated_at, spec.refresh_days): return _parse(row.rows_json) old = _parse(row.rows_json) if row is not None else [] try: rows = await asyncio.to_thread(_fetch_kind_sync, ts_code, kind) except Exception: if old: # 降级：返旧行，不把「上游挂了」伪装成「无数据」 return old raise await _upsert_row(s2, ts_code, kind, rows) await s2.commit() return rows # ---------- repurchase：全市场按月分块管道 ---------- _MARKET_CODE = "_MARKET_" _BACKFILL_MONTHS = 24 _REPURCHASE_CHUNK_CAP = 2000 # 接口单次上限（触顶则窗口减半递归拆分） _repurchase_lock = asyncio.Lock() # 回填任务与内联增量共用，防止并发重复拉 _repurchase_task: asyncio.Task | None = None def _month_range(start_ym: tuple[int, int], end_ym: tuple[int, int]) -> list[tuple[int, int]]: """[(y, m), ...] 闭区间月列表。""" out = [] y, m = start_ym while (y, m) <= end_ym: out.append((y, m)) m += 1 if m > 12: y, m = y + 1, 1 return out def _fetch_repurchase_window(pro, start: str, end: str) -> list[dict]: """拉一个日期窗的全市场回购（同步，线程内跑）；触顶 2000 行则窗口减半递归。 不传 fields：镜像实测传 fields 时 repurchase 按日期窗也返回空，去掉后正常。 """ time.sleep(settings.screener_sync_interval) df = call_retry(pro.repurchase, start_date=start, end_date=end) if df is None or df.empty: return [] if len(df) >= _REPURCHASE_CHUNK_CAP and start < end: mid_ts = ( date(int(start[:4]), int(start[4:6]), int(start[6:8])) + (date(int(end[:4]), int(end[4:6]), int(end[6:8])) - date(int(start[:4]), int(start[4:6]), int(start[6:8]))) / 2 ) mid = mid_ts.strftime("%Y%m%d") if mid > start and mid < end: return _fetch_repurchase_window(pro, start, mid) + _fetch_repurchase_window(pro, mid, end) return _clean_df(df, keep_ts_code=True) # 保留 ts_code 作为按股分组键 def _fetch_repurchase_market(months: list[tuple[int, int]]) -> dict[str, list[dict]]: """按月分块拉全市场回购，按股分组返回（同步，线程内跑）。 注意 _clean_df 必须保留 ts_code（分组键），写入时再剥掉。 """ pro = get_pro_lazy() by_code: dict[str, list[dict]] = {} for y, m in months: last_day = calendar.monthrange(y, m)[1] for r in _fetch_repurchase_window(pro, f"{y}{m:02d}01", f"{y}{m:02d}{last_day}"): code = r.get("ts_code") if code: by_code.setdefault(code, []).append({k: v for k, v in r.items() if k != "ts_code"}) spec = _Kind("", _REPURCHASE_FIELDS, False, (("ann_date", True),), "ann_date") for code in by_code: by_code[code] = _slice_and_sort(by_code[code], spec) return by_code async def _write_repurchase_batch(session: AsyncSession, fetched: dict[str, list[dict]]) -> None: """与库内旧行合并去重后批量 upsert（不 commit；由调用方提交）。""" codes = list(fetched) if not codes: return old_rows = (await session.execute( select(StockReference.ts_code, StockReference.rows_json).where( StockReference.kind == "repurchase", StockReference.ts_code.in_(codes) ))).all() old_map = {ts: _parse(rj) for ts, rj in old_rows} def _dedupe_key(r: dict) -> str: return json.dumps(r, ensure_ascii=False, sort_keys=True) values = [] for code, rows in fetched.items(): merged = old_map.get(code, []) seen = {_dedupe_key(r) for r in merged} for r in rows: k = _dedupe_key(r) if k not in seen: seen.add(k) merged.append(r) merged.sort(key=lambda r: _kv(r.get("ann_date")), reverse=True) values.append({ "ts_code": code, "kind": "repurchase", "rows_json": json.dumps(merged, ensure_ascii=False), "updated_at": utcnow(), }) stmt = pg_insert(StockReference).values(values) stmt = stmt.on_conflict_do_update( index_elements=["ts_code", "kind"], set_={"rows_json": stmt.excluded.rows_json, "updated_at": stmt.excluded.updated_at}, ) await session.execute(stmt) async def _sync_repurchase_locked(only_current: bool) -> dict[str, list[dict]]: """在全局锁内拉取缺失月份并入库（调用方须已持有 _repurchase_lock）。""" async with async_session() as s: state = await read_sync_state(s, _MARKET_CODE, "repurchase") now = date.today() if state is None: months = _month_range((now.year - _BACKFILL_MONTHS // 12, now.month), (now.year, now.month)) else: last = state.last_synced_at # naive UTC；增量从其所在月起补 months = _month_range((last.year, last.month), (now.year, now.month)) if only_current and state is not None: # 内联增量拉近两个月（当前月 + 上月迟到公告），其余月份不会缺（首次已全量回填） months = months[-2:] fetched = await asyncio.to_thread(_fetch_repurchase_market, months) await _write_repurchase_batch(s, fetched) await upsert_sync_state(s, _MARKET_CODE, "repurchase", has_data=True) await s.commit() return fetched async def _repurchase_backfill() -> None: """后台全量回填（近 24 个月）；失败记录日志——下次触发重试。""" try: async with _repurchase_lock: await _sync_repurchase_locked(only_current=False) except Exception: # noqa: BLE001 后台任务无人接异常，至少留痕 log.warning("回购数据后台回填失败（下次触发重试）", exc_info=True) def _spawn_repurchase_backfill() -> None: global _repurchase_task if _repurchase_task is not None and not _repurchase_task.done(): return _repurchase_task = asyncio.create_task(_repurchase_backfill()) async def _get_repurchase(session: AsyncSession, ts_code: str) -> list[dict]: state = await read_sync_state(session, _MARKET_CODE, "repurchase") if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): return await _read_rows(session, ts_code, "repurchase") await session.close() # 释放请求会话连接（后面隔着网络调用） if state is None: # 从未回填：先返回现有行（可能空），后台任务补齐近 24 个月，下次打开即有 _spawn_repurchase_backfill() async with async_session() as s: return await _read_rows(s, ts_code, "repurchase") # 已回填但过期：内联补拉当前月（1 次调用，秒级） async with _repurchase_lock: async with async_session() as s: state = await read_sync_state(s, _MARKET_CODE, "repurchase") if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): return await _read_rows(s, ts_code, "repurchase") try: await _sync_repurchase_locked(only_current=True) except Exception: old = await _read_rows(s, ts_code, "repurchase") if old: return old # 降级返旧行 raise return await _read_rows(s, ts_code, "repurchase") async def get_reference(session: AsyncSession, ts_code: str, kind: str) -> list[dict]: """读穿透入口。返回空列表 = 确认无数据（墓碑）；抛异常 = 上游失败且无旧行可降级。""" if kind == "repurchase": return await _get_repurchase(session, ts_code) if kind not in _KINDS: raise ValueError(f"未知参考数据分类: {kind}") return await _get_generic(session, ts_code, kind)