"""同花顺概念/行业板块（ths_index 列表 + ths_daily 行情快照 + ths_member 成分）。 缓存分层（数据特性决定）： - 板块列表：一天不变 -> 直缓存（进程内 -> Redis 24h） - 行情快照：ths_daily 盘中即有当日（实测镜像），全市场一日 1877 行单次拿全 -> 整包 SWR（同 limit_board：盘中 5 分钟 / 盘后 4 小时，trade_date 回退定位） - 成分：每板块懒加载直缓存 24h；成分股行情 enrich（candles 最新+前收 LATERAL） 每次请求现算，不进缓存 type 代码：N 概念 / I 行业 / TH 主题 / S 特色 / R 地域 / BB 宽基 / ST 风格。 """ from __future__ import annotations import asyncio import time from datetime import date, datetime, timedelta from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from .. import cache from ..config import settings from .sync_utils import call_retry, f_clean, get_pro_lazy, s_clean _LIST_KEY = "ths_boards:list:v1" _DAILY_KEY = "ths_boards:daily:v1" _REDIS_TTL = 3600 _LIST_TTL = 86400 _INTRADAY_TTL = 300.0 _MAX_DATE_BACKTRACK = 5 class ThsBoardError(RuntimeError): """列表/行情全部拉不到 —— 接口层转 503。""" # ---------- 板块列表（直缓存：进程内 -> Redis 24h -> 拉取） ---------- _list_cache: list[dict] | None = None def _fetch_list_sync() -> list[dict]: time.sleep(settings.screener_sync_interval) df = call_retry(get_pro_lazy().ths_index, exchange="A") if df is None or df.empty: raise ThsBoardError("ths_index 板块列表为空") rows: list[dict] = [] for _, r in df.iterrows(): code = s_clean(r.get("ts_code")) if not code: continue rows.append({ "ts_code": code, "name": s_clean(r.get("name")), "type": s_clean(r.get("type")), "count": f_clean(r.get("count")), "list_date": s_clean(r.get("list_date")), }) return rows async def get_board_list() -> list[dict]: global _list_cache if _list_cache is not None: return _list_cache cached = await cache.cache_get(_LIST_KEY) if isinstance(cached, list) and cached: _list_cache = cached return cached rows = await asyncio.to_thread(_fetch_list_sync) _list_cache = rows await cache.cache_set(_LIST_KEY, rows, ttl=_LIST_TTL) return rows # ---------- 行情快照（整包 SWR，同 limit_board.py 三件套） ---------- _daily_state: dict = {"payload": None} _daily_refreshing = False _daily_refresh_error: str | None = None _bg_tasks: set[asyncio.Task] = set() def _fresh_ttl(is_trading_day: bool | None) -> float: """交易时段 5 分钟（镜像盘中即有当日快照）；其余 4 小时。""" if is_trading_day is None: is_trading_day = datetime.now().weekday() < 5 if is_trading_day: now = datetime.now() t = now.hour * 60 + now.minute if 9 * 60 + 15 <= t <= 15 * 60 + 30: return _INTRADAY_TTL return float(settings.market_eod_fresh_ttl) def _fetch_daily_sync() -> dict: pro = get_pro_lazy() for i in range(_MAX_DATE_BACKTRACK): d = (date.today() - timedelta(days=i)).strftime("%Y%m%d") time.sleep(settings.screener_sync_interval) df = call_retry(pro.ths_daily, trade_date=d) if df is None or df.empty: continue quotes: dict[str, dict] = {} for _, r in df.iterrows(): code = s_clean(r.get("ts_code")) if code: quotes[code] = { "close": f_clean(r.get("close")), "pct_change": f_clean(r.get("pct_change")), "vol": f_clean(r.get("vol")), "turnover_rate": f_clean(r.get("turnover_rate")), } return {"trade_date": f"{d[:4]}-{d[4:6]}-{d[6:]}", "quotes": quotes} raise ThsBoardError(f"近 {_MAX_DATE_BACKTRACK} 天均无 ths_daily 板块行情") async def _refresh_daily() -> dict: data = await asyncio.to_thread(_fetch_daily_sync) payload = {**data, "updated_at": datetime.now().isoformat(), "fetched_ts": time.time()} _daily_state["payload"] = payload await cache.cache_set(_DAILY_KEY, payload, ttl=_REDIS_TTL) return payload async def _refresh_daily_wrapped() -> None: global _daily_refresh_error, _daily_refreshing try: await _refresh_daily() _daily_refresh_error = None except Exception as e: # noqa: BLE001 _daily_refresh_error = f"板块行情后台刷新: {str(e)[:60]}" finally: _daily_refreshing = False def _spawn_daily_refresh() -> None: global _daily_refreshing if _daily_refreshing: return _daily_refreshing = True task = asyncio.create_task(_refresh_daily_wrapped()) _bg_tasks.add(task) task.add_done_callback(_bg_tasks.discard) async def _get_daily(is_trading_day: bool | None) -> dict: ttl = _fresh_ttl(is_trading_day) p = _daily_state["payload"] if p is not None and time.time() - p["fetched_ts"] < ttl: return p if p is None: cached = await cache.cache_get(_DAILY_KEY) if cached: p = cached _daily_state["payload"] = p if p is not None: _spawn_daily_refresh() return p return await _refresh_daily() async def fetch_boards(is_trading_day: bool | None) -> dict: """列表 + 当日行情合并（行情缺失的板块价格为 null）。""" boards, daily = await asyncio.gather(get_board_list(), _get_daily(is_trading_day)) quotes: dict = daily.get("quotes", {}) merged = [{**b, **quotes.get(b["ts_code"], {})} for b in boards] errors = [_daily_refresh_error] if _daily_refresh_error else [] return { "trade_date": daily.get("trade_date"), "updated_at": daily.get("updated_at"), "boards": merged, "errors": errors, } # ---------- 成分（每板块懒加载直缓存 24h + 行情 enrich 现算） ---------- _member_cache: dict[str, list[dict]] = {} def _fetch_members_sync(board_code: str) -> list[dict]: time.sleep(settings.screener_sync_interval) df = call_retry(get_pro_lazy().ths_member, ts_code=board_code) if df is None or df.empty: return [] rows: list[dict] = [] for _, r in df.iterrows(): code = s_clean(r.get("con_code")) if code: rows.append({"con_code": code, "con_name": s_clean(r.get("con_name"))}) rows.sort(key=lambda x: x["con_code"]) return rows async def get_raw_members(board_code: str) -> list[dict]: hit = _member_cache.get(board_code) if hit is not None: return hit key = f"ths_members:{board_code}" cached = await cache.cache_get(key) if isinstance(cached, list): _member_cache[board_code] = cached return cached rows = await asyncio.to_thread(_fetch_members_sync, board_code) _member_cache[board_code] = rows await cache.cache_set(key, rows, ttl=_LIST_TTL) return rows # 成分股行情：candles 最新价 + 前收算涨跌幅（北交所等无底座数据的为 NULL） _MEMBERS_ENRICH_SQL = text(""" SELECT m.code AS con_code, c.close AS close, CASE WHEN c.close IS NOT NULL AND prev.close IS NOT NULL AND prev.close <> 0 THEN round(((c.close / prev.close - 1) * 100)::numeric, 2) END AS pct_chg FROM (SELECT unnest(CAST(:codes AS text[])) AS code) m LEFT JOIN LATERAL ( SELECT close, ts FROM candles WHERE symbol = split_part(m.code, '.', 1) AND timeframe = '1d' ORDER BY ts DESC LIMIT 1 ) c ON true LEFT JOIN LATERAL ( SELECT close FROM candles WHERE symbol = split_part(m.code, '.', 1) AND timeframe = '1d' AND ts < c.ts ORDER BY ts DESC LIMIT 1 ) prev ON c.ts IS NOT NULL """) async def get_members(session: AsyncSession, board_code: str) -> list[dict]: """成分 + 现价/涨跌幅（行情不缓存，每请求现算）。""" members = await get_raw_members(board_code) if not members: return [] rows = (await session.execute(_MEMBERS_ENRICH_SQL, { "codes": [m["con_code"] for m in members], })).mappings().all() quote = {r["con_code"]: {"close": float(r["close"]) if r["close"] is not None else None, "pct_chg": float(r["pct_chg"]) if r["pct_chg"] is not None else None} for r in rows} return [{**m, **quote.get(m["con_code"], {"close": None, "pct_chg": None})} for m in members]