"""个股参考数据（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 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 _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 后台任务无人接异常 pass 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)