mirrors2/auv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 删除板块资金流向，题材板块功能更全

Browse Source 
- 删除前端 /sectors 页面、首页入口按钮、stock-api 板块类型与 fetchSectors
- 删除后端 /api/sectors 路由，main.py 移除注册
- eastmoney.py 移除板块数据段(_fetch_push2/_fetch_akshare/UT令牌管理)，
  清理重复 import 与无用 datetime 子导入
- mootdx.py 移除板块降级方案(fetch_sector_list)
- routeTree.gen.ts 由 build 自动重新生成，移除 sectors 路由

题材热点/热点穿透/核心股已覆盖板块能力，功能更全，板块资金流向不再需要

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Sakurasan
2026-08-11 18:50:54 +08:00
co-authored by Claude
parent b0dbeef3fd
commit 90569918a3
8 changed files with 2 additions and 864 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/eastmoney.py
+1 -325
View File
@@ -6,13 +6,11 @@ import httpx
import json
import os
import re
from datetime import datetime, time as dtime, timedelta, timezone
from datetime import datetime
from typing import Optional, List
from services.cache import get_cache, set_cache
from services.cache import get_cache, set_cache
# ---- API Key 轮询（MX 备选源用）----
@@ -283,328 +281,6 @@ def get_eastmoney_market(code: str) -> str:
return "1"
return "0"
# ---- 板块数据 ----
# 从东方财富 bkzj/list.js 逆向的字段映射
# f62=主力净流入, f184=主力净流入占比
# f66=超大单净流入, f69=超大单净流入占比
# f72=大单净流入, f75=大单净流入占比
# f78=中单净流入, f81=中单净流入占比
# f84=小单净流入, f87=小单净流入占比
# f70=成交额
SECTOR_FIELDS = "f12,f14,f2,f3,f62,f184,f66,f69,f72,f75,f78,f81,f84,f87,f70"
# 东方财富板块类型映射
SECTOR_MEDIA_MAP = {
"industry": "m:90+s:4",
"concept": "m:90+t:3",
}
# 东方财富 UT 令牌管理
_em_ut: str = "8dec03ba335b81bf4ebdf7b29ec27d15"
_em_ut_lock = asyncio.Lock()
async def _refresh_em_ut() -> str:
"""
从东方财富前端 JS 中提取最新的 ut 令牌。
按优先级尝试：
1. bkzj/list.js（板块页专用）
2. common/emdataview.js（通用数据组件）
"""
urls = [
"https://data.eastmoney.com/newstatic/js/bkzj/list.js",
"https://data.eastmoney.com/newstatic/js/common/emdataview.js",
]
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://data.eastmoney.com/bkzj/hy.html",
}
async with httpx.AsyncClient() as client:
for url in urls:
try:
resp = await client.get(url, headers=headers, timeout=10)
if resp.status_code != 200:
continue
# 匹配 ut: 'xxxx' 或 ut:'xxxx' 或 ut: "xxxx"
m = re.search(r"""ut['"]?\s*:\s*['"]([a-f0-9]{32})['"]""", resp.text)
if m:
token = m.group(1)
print(f"[eastmoney] 已刷新 UT 令牌: {token[:8]}...")
return token
except Exception as e:
print(f"[eastmoney] 获取 UT 失败({url}): {e}")
return _em_ut # 保底返回当前值
async def get_em_ut(force_refresh: bool = False) -> str:
"""获取当前 UT，必要时刷新"""
global _em_ut
if force_refresh:
async with _em_ut_lock:
_em_ut = await _refresh_em_ut()
return _em_ut
# ---- 板块数据（市场时间感知缓存）----
_CST = timezone(timedelta(hours=8)) # 北京时间
_TRADING_MORNING = (dtime(9, 30), dtime(11, 30))
_TRADING_AFTERNOON = (dtime(13, 0), dtime(15, 0))
def _cst_now() -> datetime:
return datetime.now(_CST)
def _is_trading_time() -> bool:
"""判断当前是否为 A 股交易时段（周一至周五 9:30-11:30 / 13:00-15:00）"""
now = _cst_now()
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 _sector_ttl_hours() -> int:
"""根据是否在交易时段返回缓存 TTL
- 交易时段: 2 分钟（数据持续变化）
- 非交易时段: 18 小时（覆盖到下一个交易日）
"""
return 0 if _is_trading_time() else 18
# curl_cffi 模拟 Chrome TLS 指纹
from curl_cffi.requests import AsyncSession
_sector_session: Optional[AsyncSession] = None
def _get_sector_session() -> AsyncSession:
global _sector_session
if _sector_session is None:
_sector_session = AsyncSession(
impersonate="chrome131",
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",
"Referer": "https://data.eastmoney.com/bkzj/hy.html",
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9",
},
timeout=5,
)
return _sector_session
# 内存缓存（加速交易时段频繁请求）
_sector_cache: dict[str, tuple[list[dict], float]] = {}
_SECTOR_MEM_TTL = 60
async def fetch_sector_list(sector_type: str) -> list[dict]:
# 1. 内存缓存
now = time.time()
if sector_type in _sector_cache:
data, ts = _sector_cache[sector_type]
if now - ts < _SECTOR_MEM_TTL:
return data
cache_key = f"sector_list:{sector_type}"
# 2. 非交易时段：走磁盘持久缓存
if not _is_trading_time():
cached = get_cache(cache_key)
if cached is not None:
data = json.loads(cached)
_sector_cache[sector_type] = (data, now)
return data
# 3. 并发请求 push2 和 akshare，优先使用 push2
push2_task = asyncio.create_task(_fetch_push2(sector_type))
akshare_task = asyncio.create_task(_fetch_akshare(sector_type))
push2_data = await push2_task
if push2_data:
akshare_task.cancel()
try:
await akshare_task
except asyncio.CancelledError:
pass
_sector_cache[sector_type] = (push2_data, now)
ttl = _sector_ttl_hours()
if ttl > 0:
set_cache(cache_key, json.dumps(push2_data, ensure_ascii=False), ttl_hours=ttl)
return push2_data
# push2 失败，用 akshare（不写磁盘缓存）
akshare_data = await akshare_task
if akshare_data:
_sector_cache[sector_type] = (akshare_data, now)
return akshare_data
async def _fetch_push2(sector_type: str) -> list[dict]:
"""东方财富 push2 API（curl_cffi 模拟浏览器 TLS 指纹）"""
fs = SECTOR_MEDIA_MAP.get(sector_type)
if not fs:
return []
session = _get_sector_session()
ut = await get_em_ut()
for attempt in range(2):
url = (
f"https://push2.eastmoney.com/api/qt/clist/get"
f"?fs={fs}&fields={SECTOR_FIELDS}"
f"&fid=f62&po=1&pz=500&pn=1&np=1&fltt=2"
f"&invt=2&ut={ut}"
)
try:
resp = await session.get(url)
if resp.status_code != 200:
if attempt == 0:
await asyncio.sleep(1)
continue
return []
result = resp.json()
if result.get("rc") != 0:
return []
diff = result.get("data", {}).get("diff", [])
items = []
for item in diff:
items.append({
"code": item.get("f12", ""),
"name": item.get("f14", ""),
"level": item.get("f2"),
"changePercent": item.get("f3"),
"changeAmount": None,
"mainNetInflow": item.get("f62", 0) or 0,
"mainNetInflowPercent": item.get("f184", 0),
"superLargeInflow": item.get("f66", 0) or 0,
"superLargeInflowPercent": item.get("f69", 0),
"largeInflow": item.get("f72", 0) or 0,
"largeInflowPercent": item.get("f75", 0),
"mediumInflow": item.get("f78", 0) or 0,
"mediumInflowPercent": item.get("f81", 0),
"smallInflow": item.get("f84", 0) or 0,
"smallInflowPercent": item.get("f87", 0),
"turnover": item.get("f70", 0) or 0,
})
return items
except Exception as e:
err = str(e)
print(f"[eastmoney] push2 获取{sector_type}板块失败(attempt {attempt+1}): {err[:80]}")
# UT 可能过期，尝试刷新
if "disconnect" in err.lower() or "refused" in err.lower() or attempt == 1:
await get_em_ut(force_refresh=True)
ut = _em_ut
# 先尝试更新现有会话的 headers
try:
session.headers.update({"Referer": "https://data.eastmoney.com/bkzj/hy.html"})
except Exception:
pass
# 重建会话（TLS 指纹可能会被缓存）
global _sector_session
_sector_session = None
session = _get_sector_session()
if attempt == 0:
await asyncio.sleep(1)
return []
import akshare as ak
async def _fetch_akshare(sector_type: str) -> list[dict]:
"""akshare 降级方案（东方财富数据源）"""
loop = asyncio.get_event_loop()
code_map_key = f"board_codes:{sector_type}"
def _build_code_map():
"""获取板块代码映射（HTTP 较慢，结果单独缓存 24h）"""
code_map = {}
try:
if sector_type == "industry":
code_df = ak.stock_board_industry_name_em()
else:
code_df = ak.stock_board_concept_name_em()
if code_df is not None and not code_df.empty:
for _, r in code_df.iterrows():
code_map[str(r.get("f14", ""))] = str(r.get("f12", ""))
except Exception:
try:
if sector_type == "industry":
code_df = ak.stock_board_industry_name_ths()
else:
code_df = ak.stock_board_concept_name_ths()
if code_df is not None and not code_df.empty:
for _, r in code_df.iterrows():
code_map[str(r.get("name", ""))] = str(r.get("code", ""))
except Exception:
pass
return code_map
def _get_fund_flow():
if sector_type == "industry":
return ak.stock_fund_flow_industry()
else:
return ak.stock_fund_flow_concept()
# 1. 尝试从缓存读取 code_map
code_map = {}
cached_map = get_cache(code_map_key)
if cached_map is not None:
code_map = json.loads(cached_map)
try:
if code_map:
# 已有缓存，只需获取资金流
df = await loop.run_in_executor(None, _get_fund_flow)
else:
# 首次：code_map + 资金流并发获取
map_data, df = await asyncio.gather(
loop.run_in_executor(None, _build_code_map),
loop.run_in_executor(None, _get_fund_flow),
)
if map_data:
code_map = map_data
set_cache(code_map_key, json.dumps(code_map, ensure_ascii=False), ttl_hours=24)
if df is None or df.empty:
return []
df = df.sort_values("净额", ascending=False)
items = []
for _, row in df.iterrows():
name = str(row.get("行业", "")).strip()
inflow = float(row.get("流入资金", 0) or 0) * 100000000
outflow = float(row.get("流出资金", 0) or 0) * 100000000
items.append({
"code": code_map.get(name, ""),
"name": name,
"level": float(row.get("行业指数") or 0),
"changePercent": float(row.get("行业-涨跌幅") or 0),
"changeAmount": None,
"mainNetInflow": float(row.get("净额", 0) or 0) * 100000000,
"mainNetInflowPercent": None,
"superLargeInflow": None,
"superLargeInflowPercent": None,
"largeInflow": None,
"largeInflowPercent": None,
"mediumInflow": None,
"mediumInflowPercent": None,
"smallInflow": None,
"smallInflowPercent": None,
"turnover": inflow + outflow,
})
return items
except Exception as e:
print(f"[eastmoney] akshare 获取{sector_type}板块失败: {e}")
return []
# ---- 公司概况 ----
_F10_MARKET_MAP = {"6": "SH", "0": "SZ", "3": "SZ"}
backend/services/mootdx.py
-58
View File
@@ -3,7 +3,6 @@
通过 TCP 协议直连通达信行情服务器，不走 HTTP，不会被限流。
主要用途：
- K线数据：主数据源（稳定可靠）
- 板块数据：东方财富 push2 的降级方案
"""
import asyncio
@@ -69,60 +68,3 @@ async def fetch_kline_history(code: str, days: int = 90) -> Optional[List[dict]]
except Exception as e:
print(f"[mootdx] fetch_kline error: {e}")
return None
# ---- 板块数据（东方财富降级方案）----
def _sync_fetch_sectors(sector_type: str) -> Optional[list]:
from mootdx.consts import MARKET_SH, MARKET_SZ
client = _create_client()
# block() 返回 DataFrame，列：code, name 等
# 按板块类型过滤
block_df = client.block()
if block_df is None or block_df.empty:
return None
items = []
for _, row in block_df.iterrows():
name = str(row.get("name", "") or row.get("blockname", ""))
code = str(row.get("code", "") or row.get("blockcode", ""))
if not code or not name:
continue
items.append(
{
"code": code,
"name": name,
"level": None,
"changePercent": None,
"changeAmount": None,
"mainNetInflow": 0,
"mainNetInflowPercent": None,
"superLargeInflow": None,
"superLargeInflowPercent": None,
"largeInflow": None,
"largeInflowPercent": None,
"mediumInflow": None,
"mediumInflowPercent": None,
"smallInflow": None,
"smallInflowPercent": None,
"turnover": 0,
}
)
return items if items else None
async def fetch_sector_list(sector_type: str) -> Optional[List[dict]]:
"""获取板块列表（东方财富的降级方案，仅含代码和名称）"""
try:
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, _sync_fetch_sectors, sector_type)
except ImportError:
return None
except Exception as e:
print(f"[mootdx] fetch_sectors error: {e}")
return None