SakurasanandClaude
1550291d6b
非交易时段（凌晨/早盘前/周末）访问题材接口时，旧逻辑把上个交易日 数据缓存 18 小时，导致开盘后缓存未过期仍读到旧数据。 - 非交易时段缓存 TTL 由固定 18h 改为"截止到下一次开盘时刻"： 早盘前→9:30 失效、午休→13:00 失效、收盘后/周末→下一交易日 9:30 失效 - 题材列表交易时段强制不读缓存，双保险实时拉取 - 热点穿透图/题材股票缓存同样收敛到开盘前失效 Co-Authored-By: Claude <noreply@anthropic.com>
433 lines
16 KiB
Python
433 lines
16 KiB
Python
"""东方财富题材数据服务：题材列表、题材详情、题材相关股票
|
||
|
||
逆向自 emrnweb.eastmoney.com/investment 的 H5 接口：
|
||
- 题材列表: POST https://emcfgdata.eastmoney.com/api/themeInvest/getThemeList
|
||
- 题材详情: GET https://emcfgdata.securities.eastmoney.com/api/themeInvest/getDetail/{themeCode}
|
||
- 相关股票: POST https://emcfgdata.eastmoney.com/api/themeInvest/getStockList
|
||
|
||
两个 POST 接口需要「移动端包装结构」：
|
||
{args:{...业务参数}, appKey, client, clientVersion, clientType, randomCode, timestamp}
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import random
|
||
import string
|
||
import time
|
||
from datetime import datetime, time as dtime, timedelta, timezone
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
|
||
from services.cache import get_cache, set_cache
|
||
|
||
# ---- 东方财富移动端配置中心域名 ----
|
||
|
||
_PZ_URL = "https://emcfgdata.eastmoney.com"
|
||
_PZ_CDN_URL = "https://emcfgdata.securities.eastmoney.com"
|
||
|
||
# appKey 与页面场景对应：题材列表索引页 / 题材详情页
|
||
_APP_KEY_INDEX = "rn-themeIndex"
|
||
_APP_KEY_DETAIL = "rn-themeDetail"
|
||
|
||
_HEADERS = {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
|
||
"Origin": "https://emrnweb.eastmoney.com",
|
||
"Referer": "https://emrnweb.eastmoney.com/",
|
||
"Accept": "application/json",
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
}
|
||
|
||
# ---- 交易时段感知缓存 ----
|
||
|
||
_CST = timezone(timedelta(hours=8)) # 北京时间
|
||
_TRADING_MORNING = (dtime(9, 30), dtime(11, 30))
|
||
_TRADING_AFTERNOON = (dtime(13, 0), dtime(15, 0))
|
||
|
||
|
||
def _is_trading_time() -> bool:
|
||
"""判断当前是否为 A 股交易时段（周一至周五 9:30-11:30 / 13:00-15:00）"""
|
||
now = datetime.now(_CST)
|
||
if now.weekday() >= 5:
|
||
return False
|
||
t = now.time()
|
||
return (_TRADING_MORNING[0] <= t <= _TRADING_MORNING[1]
|
||
or _TRADING_AFTERNOON[0] <= t <= _TRADING_AFTERNOON[1])
|
||
|
||
|
||
def _next_open_delta_seconds() -> int:
|
||
"""非交易时段写入的缓存距下次开盘的秒数。
|
||
|
||
缓存只允许存活到下一次开盘（早盘 9:30 / 午休后 13:00）前一刻，
|
||
保证交易日开盘后缓存必然过期并实时拉取，不会读到上个交易日写入的旧数据。
|
||
"""
|
||
now = datetime.now(_CST)
|
||
# 午休 11:30-13:00 → 截止今天 13:00
|
||
if _TRADING_MORNING[1] < now.time() < _TRADING_AFTERNOON[0]:
|
||
open_dt = now.replace(hour=13, minute=0, second=0, microsecond=0)
|
||
return max(0, int((open_dt - now).total_seconds()))
|
||
# 其余非交易时段（早盘前 / 收盘后 / 周末 / 节假日）→ 下一个工作日 9:30
|
||
for days in range(0, 8):
|
||
d = (now + timedelta(days=days)).date()
|
||
if d.weekday() >= 5: # 跳过周末
|
||
continue
|
||
open_dt = datetime(d.year, d.month, d.day, 9, 30, tzinfo=_CST)
|
||
if open_dt > now:
|
||
return max(0, int((open_dt - now).total_seconds()))
|
||
return 0
|
||
|
||
|
||
def _list_ttl_seconds() -> int:
|
||
"""题材列表缓存秒数：交易时段 0（不缓存、实时拉取）；非交易时段缓存到下次开盘前失效"""
|
||
return 0 if _is_trading_time() else _next_open_delta_seconds()
|
||
|
||
|
||
# ---- 请求封装 ----
|
||
|
||
def _build_payload(args: Optional[dict] = None, app_key: str = _APP_KEY_INDEX) -> dict:
|
||
"""构建东方财富移动端请求包装结构"""
|
||
return {
|
||
"args": args or {},
|
||
"appKey": app_key,
|
||
"client": "iOS",
|
||
"clientVersion": "8.3",
|
||
"clientType": "cfw",
|
||
"randomCode": "".join(random.choices(string.ascii_uppercase + string.ascii_lowercase + string.digits, k=16)),
|
||
"timestamp": int(time.time() * 1000),
|
||
}
|
||
|
||
|
||
async def _post(path: str, args: dict, app_key: str = _APP_KEY_INDEX) -> Optional[dict]:
|
||
"""POST 到配置中心接口，返回 data 层 JSON"""
|
||
payload = _build_payload(args, app_key)
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
resp = await client.post(_PZ_URL + path, json=payload, headers=_HEADERS)
|
||
if resp.status_code != 200:
|
||
return None
|
||
body = resp.json()
|
||
except Exception as e:
|
||
print(f"[themes] POST {path} 失败: {e}")
|
||
return None
|
||
if body.get("code") != 0:
|
||
print(f"[themes] POST {path} 返回错误: {body.get('message')}")
|
||
return None
|
||
return body.get("data")
|
||
|
||
|
||
async def _get_cdn(path: str, app_key: str = _APP_KEY_DETAIL) -> Optional[dict]:
|
||
"""GET 到配置中心 CDN 接口（题材详情），data 包装结构放 query 参数"""
|
||
payload = _build_payload({}, app_key)
|
||
params = {"data": json.dumps(payload, ensure_ascii=False)}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as client:
|
||
resp = await client.get(_PZ_CDN_URL + path, params=params, headers=_HEADERS)
|
||
if resp.status_code != 200:
|
||
return None
|
||
body = resp.json()
|
||
except Exception as e:
|
||
print(f"[themes] GET {path} 失败: {e}")
|
||
return None
|
||
if body.get("code") != 0:
|
||
print(f"[themes] GET {path} 返回错误: {body.get('message')}")
|
||
return None
|
||
return body.get("data")
|
||
|
||
|
||
# ---- 题材列表 ----
|
||
|
||
# sortField 映射（题材列表页）：1=涨幅(bf3) 3=强度(strengthValue) 4=热度排名(hotRank) 5=成交额(fex5)
|
||
_LIST_PAGE_SIZE = 500
|
||
|
||
|
||
async def fetch_theme_list(sort_field: int = 1, asc: bool = False) -> list[dict]:
|
||
"""获取全部题材列表（内部循环分页拉全，约 623 个，最多 2 页）
|
||
|
||
Args:
|
||
sort_field: 排序字段 1/3/4/5
|
||
asc: True=升序, False=降序
|
||
"""
|
||
cache_key = f"theme_list:{sort_field}:{asc}"
|
||
# 交易时段强制实时：跳过缓存读取，避免命中非交易时段写入的上个交易日旧数据
|
||
if not _is_trading_time():
|
||
cached = get_cache(cache_key)
|
||
if cached is not None:
|
||
return json.loads(cached)
|
||
|
||
sort = 1 if asc else -1
|
||
# hotRank 数值越小越热，"热度降序(最热在前)" 需反转为接口升序
|
||
if sort_field == 4:
|
||
sort = -sort
|
||
items: list[dict] = []
|
||
page = 1
|
||
total = None
|
||
|
||
for _ in range(5): # 安全上限
|
||
data = await _post(
|
||
"/api/themeInvest/getThemeList",
|
||
{"pageSize": _LIST_PAGE_SIZE, "pageNum": page, "sort": sort, "sortField": sort_field},
|
||
)
|
||
if not data:
|
||
break
|
||
if total is None:
|
||
total = data.get("total", 0)
|
||
page_items = data.get("list", [])
|
||
if not page_items:
|
||
break
|
||
items.extend(page_items)
|
||
if len(items) >= total:
|
||
break
|
||
page += 1
|
||
|
||
if items:
|
||
ttl_s = _list_ttl_seconds()
|
||
if ttl_s > 0:
|
||
set_cache(cache_key, json.dumps(items, ensure_ascii=False), ttl_seconds=ttl_s)
|
||
return items
|
||
|
||
|
||
# ---- 题材详情 ----
|
||
|
||
async def fetch_theme_detail(theme_code: str) -> Optional[dict]:
|
||
"""获取题材详情（简介 + 热点事件 + 相关新闻），缓存 1 小时"""
|
||
cache_key = f"theme_detail:{theme_code}"
|
||
cached = get_cache(cache_key)
|
||
if cached is not None:
|
||
return json.loads(cached)
|
||
|
||
data = await _get_cdn(f"/api/themeInvest/getDetail/{theme_code}", app_key=_APP_KEY_DETAIL)
|
||
if not data:
|
||
return None
|
||
|
||
set_cache(cache_key, json.dumps(data, ensure_ascii=False), ttl_hours=1)
|
||
return data
|
||
|
||
|
||
# ---- 题材相关股票 ----
|
||
|
||
_STOCK_PAGE_SIZE = 100
|
||
|
||
|
||
async def fetch_theme_stocks(theme_code: str) -> dict:
|
||
"""获取题材下全部相关股票（分页拉全），返回 {stockList, statistic, total}"""
|
||
cache_key = f"theme_stocks:{theme_code}"
|
||
cached = get_cache(cache_key)
|
||
if cached is not None:
|
||
return json.loads(cached)
|
||
|
||
stock_list: list[dict] = []
|
||
statistic = {}
|
||
total = 0
|
||
page = 1
|
||
|
||
for _ in range(20): # 安全上限
|
||
data = await _post(
|
||
"/api/themeInvest/getStockList",
|
||
{"themeCode": theme_code, "pageSize": _STOCK_PAGE_SIZE, "pageNum": page, "sort": -1, "sortField": "f3"},
|
||
app_key=_APP_KEY_DETAIL,
|
||
)
|
||
if not data:
|
||
break
|
||
if not statistic and data.get("statistic"):
|
||
statistic = data["statistic"]
|
||
page_items = data.get("stockList", [])
|
||
if not page_items:
|
||
break
|
||
stock_list.extend(page_items)
|
||
total = data.get("total", 0)
|
||
if len(stock_list) >= total:
|
||
break
|
||
page += 1
|
||
|
||
result = {"stockList": stock_list, "statistic": statistic, "total": total}
|
||
if stock_list:
|
||
ttl_s = _graph_ttl_seconds()
|
||
if ttl_s > 0:
|
||
set_cache(cache_key, json.dumps(result, ensure_ascii=False), ttl_seconds=ttl_s)
|
||
return result
|
||
|
||
|
||
# ---- 热点穿透：题材-股票 网状关系图数据 ----
|
||
|
||
# 合并采样：涨幅榜 Top N + 热度榜 Top N 去重合并，
|
||
# 避免单一榜单导致股票覆盖题材数被低估（如有研新材覆盖 10+ 题材，仅涨幅榜只能采到 2 个）。
|
||
|
||
# 盘中图聚合结果/题材股票子层的缓存秒数。交易时段数据波动快，用 60s 短缓存；
|
||
# 非交易时段缓存 18 小时（覆盖到下一交易日）。
|
||
_GRAPH_CACHE_SECONDS = 60
|
||
|
||
|
||
def _graph_ttl_seconds() -> int:
|
||
"""图聚合结果与题材股票子层的缓存秒数：盘中 60 秒；非盘中缓存到下次开盘前失效"""
|
||
return _GRAPH_CACHE_SECONDS if _is_trading_time() else _next_open_delta_seconds()
|
||
|
||
|
||
# 后台重建锁：cache_key -> asyncio.Lock，幂等去重，防止并发重复聚合
|
||
_REBUILD_LOCKS: dict[str, asyncio.Lock] = {}
|
||
|
||
|
||
def _set_graph_cache(cache_key: str, data: dict) -> None:
|
||
"""写入图聚合缓存（存 data + built_at + expires_at，epoch 秒）"""
|
||
ttl_s = _graph_ttl_seconds()
|
||
if ttl_s <= 0:
|
||
return
|
||
now = time.time()
|
||
entry = {"data": data, "built_at": now, "expires_at": now + ttl_s}
|
||
set_cache(cache_key, json.dumps(entry, ensure_ascii=False), ttl_seconds=ttl_s)
|
||
|
||
|
||
def _get_graph_cache(cache_key: str) -> tuple[Optional[dict], Optional[float]]:
|
||
"""读取图聚合缓存，返回 (data, expires_at)；无缓存/损坏返回 (None, None)"""
|
||
cached = get_cache(cache_key)
|
||
if cached is None:
|
||
return None, None
|
||
try:
|
||
entry = json.loads(cached)
|
||
return entry.get("data"), entry.get("expires_at")
|
||
except (json.JSONDecodeError, TypeError):
|
||
return None, None
|
||
|
||
|
||
async def _build_theme_graph(sort_field: int, top_n: int) -> dict:
|
||
"""构建完整图数据（全量统计，边由前端从 stocks[].themeCodes 重建）
|
||
|
||
Args:
|
||
sort_field: 题材排序 1=涨幅, 4=热度（当前榜在前，另一榜合并补充）
|
||
top_n: 每个榜单的题材数量（1-60）
|
||
|
||
Returns:
|
||
{
|
||
"themes": [{themeCode, themeName, stockCount}],
|
||
"stocks": [{securityCode, securityName, coverCount, f3, f2, f62, f100, themeCodes[]}],
|
||
"stats": {themeCount, stockCount, coreCount, maxCover}
|
||
}
|
||
"""
|
||
# 1. 合并采样涨幅榜 + 热度榜（当前榜优先在前，另一榜补充），按 themeCode 去重
|
||
primary_list = await fetch_theme_list(sort_field, asc=False)
|
||
other_field = 4 if sort_field == 1 else 1
|
||
other_list = await fetch_theme_list(other_field, asc=False)
|
||
merged: dict[str, dict] = {}
|
||
for t in primary_list[:top_n]:
|
||
merged.setdefault(t["themeCode"], t)
|
||
for t in other_list[:top_n]:
|
||
merged.setdefault(t["themeCode"], t)
|
||
themes = list(merged.values())
|
||
if not themes:
|
||
return {"themes": [], "stocks": [], "stats": {}}
|
||
|
||
# 2. 并发拉取每题材股票（限流保护）
|
||
sem = asyncio.Semaphore(5)
|
||
|
||
async def _fetch_with_limit(code: str):
|
||
async with sem:
|
||
return await fetch_theme_stocks(code)
|
||
|
||
results = await asyncio.gather(*[_fetch_with_limit(t["themeCode"]) for t in themes])
|
||
|
||
# 3. 构建 M:N 关系：统计每股覆盖的题材数
|
||
theme_map = {t["themeCode"]: t for t in themes}
|
||
stock_map: dict[str, dict] = {} # securityCode -> stock dict
|
||
|
||
for t, res in zip(themes, results):
|
||
stock_list = res.get("stockList", [])
|
||
theme_map[t["themeCode"]]["stockCount"] = len(stock_list)
|
||
for s in stock_list:
|
||
code = s["securityCode"]
|
||
if code not in stock_map:
|
||
stock_map[code] = {
|
||
"securityCode": code,
|
||
"securityName": s.get("securityName", ""),
|
||
"coverCount": 0,
|
||
"f3": s.get("f3"),
|
||
"f2": s.get("f2"),
|
||
"f62": s.get("f62"),
|
||
"f100": s.get("f100", ""),
|
||
"themeCodes": [],
|
||
}
|
||
stock_map[code]["coverCount"] += 1
|
||
stock_map[code]["themeCodes"].append(t["themeCode"])
|
||
|
||
# 4. 排序：覆盖题材数越多（穿透越强）排越前
|
||
stocks = sorted(stock_map.values(), key=lambda x: (-x["coverCount"], -(x.get("f3") or 0)))
|
||
|
||
stats = {
|
||
"themeCount": len(themes),
|
||
"stockCount": len(stocks),
|
||
"coreCount": sum(1 for s in stocks if s["coverCount"] >= 2),
|
||
"maxCover": max((s["coverCount"] for s in stocks), default=1),
|
||
}
|
||
|
||
return {
|
||
"themes": [
|
||
{"themeCode": t["themeCode"], "themeName": t["themeName"], "stockCount": t["stockCount"]}
|
||
for t in theme_map.values()
|
||
],
|
||
"stocks": stocks,
|
||
"stats": stats,
|
||
}
|
||
|
||
|
||
def _trim_graph_result(result: dict, limit: int) -> dict:
|
||
"""按 limit 裁剪 stocks（保留穿透度最高的 N 只），仅影响下发体积，不影响 coverCount 统计"""
|
||
return {
|
||
"themes": result.get("themes", []),
|
||
"stocks": result.get("stocks", [])[:limit],
|
||
"stats": result.get("stats", {}),
|
||
}
|
||
|
||
|
||
def _spawn_rebuild(cache_key: str, sort_field: int, top_n: int) -> None:
|
||
"""幂等触发后台重建：已有重建任务在跑则跳过"""
|
||
lock = _REBUILD_LOCKS.setdefault(cache_key, asyncio.Lock())
|
||
if lock.locked():
|
||
return
|
||
asyncio.create_task(_rebuild_task(cache_key, sort_field, top_n, lock))
|
||
|
||
|
||
async def _rebuild_task(cache_key: str, sort_field: int, top_n: int, lock: asyncio.Lock) -> None:
|
||
"""后台重建：拿锁后 double-check 缓存是否已被刷新，避免重复聚合"""
|
||
async with lock:
|
||
try:
|
||
data, expires_at = _get_graph_cache(cache_key)
|
||
if data is not None and expires_at and expires_at > time.time():
|
||
return # 已被其他任务刷新
|
||
data = await _build_theme_graph(sort_field, top_n)
|
||
_set_graph_cache(cache_key, data)
|
||
print(f"[themes] graph 后台重建完成: {cache_key}")
|
||
except Exception as e:
|
||
print(f"[themes] graph 后台重建失败: {cache_key} {e}")
|
||
|
||
|
||
async def fetch_theme_graph(sort_field: int = 1, top_n: int = 30, limit: int = 1000) -> dict:
|
||
"""获取热点穿透图数据（盘中 60s 缓存 + stale-while-revalidate）
|
||
|
||
缓存命中且未过期 → 直接返回；已过期 → 返回旧数据并后台异步重建（秒开）；
|
||
无缓存 → 同步构建（并发下加锁去重）。返回前按 limit 裁剪 stocks。
|
||
|
||
Args:
|
||
sort_field: 题材排序 1=涨幅, 4=热度
|
||
top_n: 每个榜单的题材数量（1-60）
|
||
limit: 下发 stocks 上限（穿透度最高的 N 只）
|
||
"""
|
||
cache_key = f"theme_graph:{sort_field}:{top_n}"
|
||
|
||
data, expires_at = _get_graph_cache(cache_key)
|
||
if data is not None:
|
||
# 有缓存：新鲜直接返回；过期返回旧数据并后台刷新
|
||
if not (expires_at and expires_at > time.time()):
|
||
_spawn_rebuild(cache_key, sort_field, top_n)
|
||
return _trim_graph_result(data, limit)
|
||
|
||
# 无缓存：同步构建（并发下加锁去重）
|
||
lock = _REBUILD_LOCKS.setdefault(cache_key, asyncio.Lock())
|
||
async with lock:
|
||
data, expires_at = _get_graph_cache(cache_key)
|
||
if data is not None:
|
||
# 等待锁期间已被其他请求写入
|
||
if not (expires_at and expires_at > time.time()):
|
||
_spawn_rebuild(cache_key, sort_field, top_n)
|
||
return _trim_graph_result(data, limit)
|
||
data = await _build_theme_graph(sort_field, top_n)
|
||
_set_graph_cache(cache_key, data)
|
||
return _trim_graph_result(data, limit)
|