"""个股财务数据（fina_indicator 财务指标 + 三大报表关键值，详情页按需单查懒加载）。 四源按报告期合并进 stock_financial 宽表（一行 = 一个报告期）： fina_indicator（doc 79，无 report_type 概念） + income / balancesheet / cashflow（report_type=1 合并报表） 近五年窗口（start_date = 当年-5年 的 0101）；季更数据 7 天新鲜度门控。 每源独立容错：部分接口失败不拖垮整体，缺失列靠 upsert 列级 coalesce 保旧值， 7 天后下次过期刷新自动重试失败源。四源全失败且无旧行 -> 抛异常（上游故障）。 stock_sync_state(kind='finance') 做新鲜度与墓碑（无数据股票不重复穿透控频调用）。 """ from __future__ import annotations import asyncio import time from datetime import date from sqlalchemy import func, 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 StockFinancial from .sync_utils import call_retry, f_clean, fresh, get_pro_lazy, read_sync_state, s_clean, upsert_sync_state, utcnow _REFRESH_DAYS = 7 # 四源显式字段（财务接口列多，必须显式传 fields；金额单位元、比率百分数沿用 tushare 原始） _FI_FIELDS = ("ts_code,ann_date,end_date,eps,bps,ocfps,roe,roe_dt,grossprofit_margin," "netprofit_margin,debt_to_assets,or_yoy,netprofit_yoy,dt_netprofit_yoy," "profit_dedt,rd_exp") _INC_FIELDS = "ts_code,ann_date,end_date,total_revenue,operate_profit,n_income_attr_p" _BS_FIELDS = "ts_code,ann_date,end_date,total_assets,total_hldr_eqy_exc_min_int" _CF_FIELDS = "ts_code,ann_date,end_date,n_cashflow_act" # tushare 字段 -> 模型列名（仅 balancesheet 归母权益超长名改短） _COL_MAP = {"total_hldr_eqy_exc_min_int": "total_hldr_eqy"} _SOURCES = ( ("fina_indicator", _FI_FIELDS, {}), ("income", _INC_FIELDS, {"report_type": "1"}), ("balancesheet", _BS_FIELDS, {"report_type": "1"}), ("cashflow", _CF_FIELDS, {"report_type": "1"}), ) _COLS = ( "ts_code", "end_date", "ann_date", "eps", "bps", "ocfps", "roe", "roe_dt", "grossprofit_margin", "netprofit_margin", "debt_to_assets", "or_yoy", "netprofit_yoy", "dt_netprofit_yoy", "profit_dedt", "rd_exp", "total_revenue", "operate_profit", "n_income_attr_p", "total_assets", "total_hldr_eqy", "n_cashflow_act", ) _VALUE_COLS = _COLS[3:] # ann_date 之外的其余业务列（批量 upsert 需统一键集） def fetch_finance_sync(ts_code: str) -> dict[str, dict]: """拉近五年四源财务数据并按报告期合并（同步网络 IO，需在 to_thread 里跑）。 返回 {end_date: 行dict}；四源全部失败且无任何数据时抛异常。 """ pro = get_pro_lazy() start = f"{date.today().year - 5}0101" # 含整五个年度的年报 merged: dict[str, dict] = {} failed = 0 for api, fields, kwargs in _SOURCES: time.sleep(settings.screener_sync_interval) try: df = call_retry(getattr(pro, api), ts_code=ts_code, start_date=start, fields=fields, **kwargs) except Exception: # noqa: BLE001 单源失败不拖垮整体（缺列保旧值，7 天后自动重试） failed += 1 continue if df is None or df.empty: continue # 同一报告期 tushare 可能返回多条（调整前/后副本），按公告日升序取最后一条 df = df.sort_values("ann_date", na_position="first").drop_duplicates("end_date", keep="last") for _, r in df.iterrows(): end = s_clean(r.get("end_date")) if not end: continue row = merged.setdefault(end, {"ts_code": ts_code, "end_date": end, "ann_date": None}) ann = s_clean(r.get("ann_date")) if ann: row["ann_date"] = ann for col in fields.split(","): if col in ("ts_code", "ann_date", "end_date"): continue v = f_clean(r.get(col)) if v is not None: row[_COL_MAP.get(col, col)] = v if failed == len(_SOURCES) and not merged: raise RuntimeError(f"tushare 财务四源均失败: {ts_code}") return merged def _row_dict(row: StockFinancial) -> dict: return {c: getattr(row, c) for c in _COLS} async def _read_rows(session: AsyncSession, ts_code: str) -> list[dict]: rs = (await session.execute( select(StockFinancial).where(StockFinancial.ts_code == ts_code) .order_by(StockFinancial.end_date.desc()))).scalars().all() return [_row_dict(r) for r in rs] async def _upsert_rows(session: AsyncSession, rows: list[dict]) -> None: """批量 upsert（不 commit）；列级 coalesce 保旧值——失败源的列、tushare 改为空值的列都保留库内旧值。""" now = utcnow() full = [] for r in rows: row = {c: None for c in _VALUE_COLS} # 统一键集：批量 insert 要求各 dict 同构 row.update(r) row["updated_at"] = now full.append(row) stmt = pg_insert(StockFinancial).values(full) set_ = {c: func.coalesce(stmt.excluded[c], getattr(StockFinancial, c)) for c in ("ann_date", *_VALUE_COLS)} set_["updated_at"] = stmt.excluded.updated_at stmt = stmt.on_conflict_do_update(index_elements=["ts_code", "end_date"], set_=set_) await session.execute(stmt) # per-code 锁：同 code 首次并发 N 个请求只有 1 个打 tushare，其余等锁后双检命中 # （uvicorn 单进程场景够用；多 worker 最坏情况是重复拉一次 + ON CONFLICT 幂等，无害） _code_locks: dict[str, asyncio.Lock] = {} _guard = asyncio.Lock() async def get_finance(session: AsyncSession, ts_code: str) -> list[dict] | None: """读穿透：同步状态新鲜直返；否则 per-code 锁 -> 新会话双检 -> to_thread 拉四源 -> upsert。 返回 None = 确认无财务数据（墓碑已落库，多为新股/退市老股）； 抛异常 = 四源均失败且无旧行可降级。 """ state = await read_sync_state(session, ts_code, "finance") if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): if not state.has_data: return None return await _read_rows(session, ts_code) # 释放请求会话持有的连接：后面可能隔着数秒的 tushare 调用，别长占连接池。 # 用 close() 而非 rollback()：rollback 会把会话身份映射里的实例全部 expire—— # 包括 require_user 刚塞进 auth._session_cache 的 User（expire_on_commit=False # 只保 commit，不保 rollback），下个请求命中鉴权缓存即 DetachedInstanceError 500。 # close() 同样归还连接，且已加载属性保持可访问（脱管安全，与鉴权缓存约定一致）。 await session.close() async with _guard: lock = _code_locks.setdefault(ts_code, asyncio.Lock()) async with lock: async with async_session() as s2: # 锁内重读 + 写入走新会话 state = await read_sync_state(s2, ts_code, "finance") if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): if not state.has_data: return None return await _read_rows(s2, ts_code) old = await _read_rows(s2, ts_code) try: merged = await asyncio.to_thread(fetch_finance_sync, ts_code) except Exception: # 降级：库内有旧行（哪怕超 7 天）照常返回，不把「上游挂了」伪装成「无数据」 if old: return old raise if merged: await _upsert_rows(s2, list(merged.values())) await upsert_sync_state(s2, ts_code, "finance", has_data=bool(merged) or bool(old)) await s2.commit() return await _read_rows(s2, ts_code)