SakurasanandClaude
e28d1a2fd0
- 新增 /hot-map 热点穿透页：d3-force 力导向网状图展示题材-股票 M:N 关系 - 股票节点面积/颜色按覆盖题材数分级，穿透核心股高亮 - 图/核心股列表双视图切换，列表覆盖题材>2 公司降序展示 - 悬停信息卡、缩放平移、点击跳转股票/题材详情 - 数据采样改为涨幅+热度双榜合并（各 Top50 去重=82 题材），修复覆盖数低估 （有研新材覆盖数 2→10，核心股 66→1124 家） - 移动端适配：触屏 tap 选中底部浮层、标签缩放阈值防重叠、自动 fit、紧凑图例 - 主页/题材列表增加热点穿透入口 Co-Authored-By: Claude <noreply@anthropic.com>
324 lines
11 KiB
Python
324 lines
11 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 _dynamic_ttl() -> int:
|
||
"""盘中返回 2 分钟缓存 TTL，非交易时段 18 小时（覆盖到下一交易日）"""
|
||
return 0 if _is_trading_time() else 18
|
||
|
||
|
||
# ---- 请求封装 ----
|
||
|
||
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}"
|
||
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 = _dynamic_ttl()
|
||
if ttl > 0:
|
||
set_cache(cache_key, json.dumps(items, ensure_ascii=False), ttl_hours=ttl)
|
||
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 = _dynamic_ttl()
|
||
if ttl > 0:
|
||
set_cache(cache_key, json.dumps(result, ensure_ascii=False), ttl_hours=ttl)
|
||
return result
|
||
|
||
|
||
# ---- 热点穿透：题材-股票 网状关系图数据 ----
|
||
|
||
# 合并采样：涨幅榜 Top N + 热度榜 Top N 去重合并，
|
||
# 避免单一榜单导致股票覆盖题材数被低估（如有研新材覆盖 10+ 题材，仅涨幅榜只能采到 2 个）。
|
||
_GRAPH_MERGE_TOP = 50
|
||
|
||
|
||
async def fetch_theme_graph(sort_field: int = 1, top_n: int = 30) -> dict:
|
||
"""构建题材-股票网状关系图数据（热点穿透）
|
||
|
||
合并采样涨幅榜 + 热度榜 Top N 题材（去重），并发拉取每题材全部股票，
|
||
统计每股覆盖的题材数（M:N 关系），返回前端可直接渲染的图结构。
|
||
|
||
Args:
|
||
sort_field: 题材排序 1=涨幅, 4=热度（当前榜在前，另一榜合并补充）
|
||
top_n: 每个榜单的题材数量（1-60）
|
||
|
||
Returns:
|
||
{
|
||
"themes": [{themeCode, themeName, bf3, hotValue, strengthValue, stockCount}],
|
||
"stocks": [{securityCode, securityName, coverCount, f3, f2, f62, f100, themeCodes[]}],
|
||
"edges": [{themeCode, securityCode}],
|
||
"stats": {themeCount, stockCount, coreCount, maxCover}
|
||
}
|
||
"""
|
||
cache_key = f"theme_graph:{sort_field}:{top_n}"
|
||
cached = get_cache(cache_key)
|
||
if cached is not None:
|
||
return json.loads(cached)
|
||
|
||
# 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": [], "edges": [], "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. 构建图数据
|
||
theme_map = {t["themeCode"]: t for t in themes}
|
||
stock_map: dict[str, dict] = {} # securityCode -> stock dict
|
||
edges: list[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"]
|
||
edges.append({"themeCode": t["themeCode"], "securityCode": code})
|
||
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),
|
||
}
|
||
|
||
result = {
|
||
"themes": list(theme_map.values()),
|
||
"stocks": stocks,
|
||
"edges": edges,
|
||
"stats": stats,
|
||
}
|
||
|
||
ttl = _dynamic_ttl()
|
||
if ttl > 0:
|
||
set_cache(cache_key, json.dumps(result, ensure_ascii=False), ttl_hours=ttl)
|
||
return result
|