"""东方财富题材数据服务：题材列表、题材详情、题材相关股票 逆向自 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） 缓存新鲜 → 直接返回； 非交易时段过期 → 返回旧数据并后台异步重建（秒开，非盘中行情无实时变化，旧值可接受）； 交易时段过期 / 无缓存 → 同步重建（加锁去重），绝不返回上个交易日的旧图。 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 and expires_at and expires_at > time.time(): # 缓存新鲜 → 直接返回 return _trim_graph_result(data, limit) # 非交易时段过期：行情无实时变化，先返回旧值（秒开），后台异步重建 if data is not None and not _is_trading_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 and expires_at and expires_at > time.time(): # 等待锁期间已被其他请求刷新 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)