243 lines
8.4 KiB
Python
243 lines
8.4 KiB
Python
"""同花顺概念/行业板块（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]
|