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

使用push2 数据接口，优化代码

Browse Source
This commit is contained in:
Sakurasan
2026-07-06 22:27:45 +08:00
parent 83c88ee945
commit 6390545e61
7 changed files with 750 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/eastmoney.py
+319 -109
View File
@@ -1,6 +1,7 @@
"""东方财富 API 客户端（资金流向数据）"""
import asyncio
import time
import httpx
import json
import os
@@ -10,7 +11,8 @@ from typing import Optional, List
from services.cache import get_cache, set_cache
# ---- API Key 轮询 ----
# ---- API Key 轮询（MX 备选源用）----
_api_keys: List[str] = []
_key_index: int = -1
@@ -61,18 +63,6 @@ def _rotate_key():
set_cache("_mx_key_index", str(_key_index), ttl_hours=24)
# ---- 工具函数 ----
def get_eastmoney_market(code: str) -> str:
"""获取东方财富格式的市场标识"""
if code.startswith("688"):
return "6"
if code.startswith("60"):
return "1"
return "0"
def parse_amount(text) -> float:
"""解析金额文本（支持 1.432亿元, -9915万元, 0元）"""
if isinstance(text, (int, float)):
@@ -96,7 +86,7 @@ def parse_amount(text) -> float:
return sign * value
# ---- MX API ----
# ---- MX API（备选源）----
async def _call_mx_api(api_key: str, name: str, days: int) -> Optional[dict]:
@@ -176,12 +166,10 @@ async def fetch_mx_api(name: str, days: int) -> Optional[List[dict]]:
"""调用东方财富妙想MX API获取资金流向（缓存6小时，多key轮询）"""
cache_key = f"mx_fund_flow:{name}:{days}"
# 缓存命中
cached = get_cache(cache_key)
if cached is not None:
return json.loads(cached)
# 多 key 轮询：按序尝试，遇到 113(超限) 自动切下一个 key
if not _ensure_keys():
return None
@@ -207,7 +195,6 @@ async def fetch_mx_api(name: str, days: int) -> Optional[List[dict]]:
_rotate_key()
continue
# 成功
data_list = _parse_mx_response(result)
if data_list:
set_cache(cache_key, json.dumps(data_list, ensure_ascii=False))
@@ -217,36 +204,196 @@ async def fetch_mx_api(name: str, days: int) -> Optional[List[dict]]:
return None
def get_eastmoney_market(code: str) -> str:
"""获取东方财富格式的市场标识"""
if code.startswith("688"):
return "6"
if code.startswith("60"):
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_TYPE_MAP = {
"industry": "m:90+t:2", # 行业板块
"concept": "m:90+t:3", # 概念板块
SECTOR_MEDIA_MAP = {
"industry": "m:90+s:4",
"concept": "m:90+t:3",
}
# 板块列表字段：f12=代码, f14=名称, f3=涨跌幅%, f62=主力净流入, f184=主力净流入占比
# f66=超大单净流入, f69=超大单净流入占比, f70=成交额, f78=小单净流入
SECTOR_FIELDS = "f12,f14,f2,f3,f4,f62,f184,f66,f69,f70,f78"
# 东方财富 UT 令牌管理
_em_ut: str = "8dec03ba335b81bf4ebdf7b29ec27d15"
_em_ut_lock = asyncio.Lock()
# ---- 板块数据（通过 akshare 调用同花顺数据源）----
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",
}
import akshare as ak
import pandas as pd
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
# 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="chrome120",
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",
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9",
},
timeout=10,
)
return _sector_session
# 内存缓存
_sector_cache: dict[str, tuple[list[dict], float]] = {}
_SECTOR_CACHE_TTL = 60
async def fetch_sector_list(sector_type: str) -> list[dict]:
"""
获取板块资金流向数据
sector_type: "industry" 或 "concept"
返回按主力净流入降序排列的板块列表
编码优先东方财富（BKxxxx），降级同花顺（6位数字）
"""
now = time.time()
if sector_type in _sector_cache:
data, ts = _sector_cache[sector_type]
if now - ts < _SECTOR_CACHE_TTL:
return data
data = await _fetch_push2(sector_type)
if data:
_sector_cache[sector_type] = (data, now)
return data
data = await _fetch_akshare(sector_type)
if data:
_sector_cache[sector_type] = (data, now)
return 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()
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}"
)
for attempt in range(2):
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
# 重建会话（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()
def _get_data():
# 1) 板块编码映射：东方财富 BK 编码 → 降级同花顺编码
def _get():
code_map = {}
try:
if sector_type == "industry":
@@ -268,7 +415,6 @@ async def fetch_sector_list(sector_type: str) -> list[dict]:
except Exception:
pass
# 2) 资金流向（10jqka 同花顺数据源）
if sector_type == "industry":
df = ak.stock_fund_flow_industry()
else:
@@ -277,92 +423,39 @@ async def fetch_sector_list(sector_type: str) -> list[dict]:
return code_map, df
try:
code_map, df = await loop.run_in_executor(None, _get_data)
code_map, df = await loop.run_in_executor(None, _get)
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": _safe_float(row.get("行业指数")),
"changePercent": _safe_float(row.get("行业-涨跌幅")),
"level": float(row.get("行业指数") or 0),
"changePercent": float(row.get("行业-涨跌幅") or 0),
"changeAmount": None,
"mainNetInflow": _safe_float(row.get("净额", 0)) * 100000000, # 亿→元
"mainNetInflow": float(row.get("净额", 0) or 0) * 100000000,
"mainNetInflowPercent": None,
"superLargeInflow": None,
"superLargeInflowPercent": None,
"turnover": _safe_float(row.get("流入资金", 0)) * 100000000 + _safe_float(row.get("流出资金", 0)) * 100000000,
"smallNetInflow": 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] 获取{sector_type}板块失败: {e}")
print(f"[eastmoney] akshare 获取{sector_type}板块失败: {e}")
return []
def _safe_float(val) -> float:
if val is None:
return 0.0
try:
return float(val)
except (ValueError, TypeError):
return 0.0
async def fetch_sector_list_direct(sector_type: str) -> list[dict]:
"""
直接从东方财富 push2 API 获取板块列表（备用，当 akshare 不可用时）
"""
fs = SECTOR_TYPE_MAP.get(sector_type)
if not fs:
return []
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"
)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "https://quote.eastmoney.com/",
}
async with httpx.AsyncClient() as client:
try:
resp = await client.get(url, headers=headers, timeout=10)
if resp.status_code != 200:
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": item.get("f4"),
"mainNetInflow": item.get("f62", 0),
"mainNetInflowPercent": item.get("f184", 0),
"superLargeInflow": item.get("f66", 0),
"superLargeInflowPercent": item.get("f69", 0),
"turnover": item.get("f70", 0),
"smallNetInflow": item.get("f78", 0),
})
return items
except Exception as e:
print(f"[eastmoney] 获取{sector_type}板块失败: {e}")
return []
# ---- 公司概况 ----
_F10_MARKET_MAP = {"6": "SH", "0": "SZ", "3": "SZ"}
@@ -733,14 +826,88 @@ async def fetch_business_segments(code: str, by_type: str = "product", years: in
return result
async def fetch_push2his(code: str, days: int) -> Optional[list]:
"""回退到东方财富 push2his 接口（自动重试一次）"""
async def fetch_kline_history(code: str, days: int = 90) -> Optional[list[dict]]:
"""从 push2his kline/get 获取日K线（curl_cffi 模拟浏览器 TLS 指纹）
可能受 IP 限流影响，调用方应有降级。
返回: [{date, open, close, high, low, volume(手), turnover, changePercent, amplitude, turnoverRate}]
"""
market = get_eastmoney_market(code)
secid = f"{market}.{code}"
url = (
f"https://push2his.eastmoney.com/api/qt/stock/kline/get"
f"?secid={secid}"
f"&ut=b2884a393a59ad64002292a3e90d46a5"
f"&fields1=f1,f2,f3,f4,f5,f6"
f"&fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61"
f"&klt=101&fqt=1&end=20500101&lmt={days}"
)
from curl_cffi.requests import AsyncSession
async with AsyncSession(impersonate="chrome120") as session:
for attempt in range(2):
try:
resp = await session.get(url, headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "https://quote.eastmoney.com/",
}, timeout=10)
if resp.status_code != 200:
if attempt == 0:
continue
return None
result = resp.json()
if result.get("rc") != 0:
if attempt == 0:
continue
return None
klines = result.get("data", {}).get("klines", [])
if not klines:
if attempt == 0:
continue
return None
parsed = []
for line in klines:
fields = line.split(",")
if len(fields) < 11:
continue
def _f(i): return float(fields[i]) if fields[i] else 0
parsed.append({
"date": fields[0],
"open": _f(1),
"close": _f(2),
"high": _f(3),
"low": _f(4),
"volume": int(_f(5)), # 手
"turnover": _f(6),
"amplitude": _f(7),
"changePercent": _f(8),
"change": _f(9),
"turnoverRate": _f(10),
})
parsed.reverse()
return parsed
except Exception as e:
if attempt == 0:
continue
return None
return None
async def fetch_fund_flow_daykline(code: str, days: int) -> Optional[list[dict]]:
"""从 push2his daykline/get 获取资金流向日线数据（结构化返回，替代 MX API）
返回字段：date, mainNetInflow, superLargeInflow/Outflow, largeInflow/Outflow,
mediumInflow/Outflow, smallInflow/Outflow, 各占比%, closePrice, changePercent,
mainForceNet, retailNet
"""
market = get_eastmoney_market(code)
secid = f"{market}.{code}"
url = (
f"https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get"
f"?lmt={days}&fields1=f1,f2,f3,f7"
f"&fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65,f66,f67,f68,f69"
f"&fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63"
f"&ut=b2884a393a59ad64002292a3e90d46a5&secid={secid}"
)
headers = {
@@ -755,19 +922,62 @@ async def fetch_push2his(code: str, days: int) -> Optional[list]:
if resp.status_code != 200:
if attempt == 0:
continue
print(f"[eastmoney] push2his HTTP {resp.status_code}")
return None
result = resp.json()
klines = result.get("data", {}).get("klines", [])
if not klines:
if attempt == 0:
continue
print(f"[eastmoney] push2his no klines in response")
return None
return klines
parsed = []
for line in klines:
fields = line.split(",")
if len(fields) < 13:
continue
def _f(i): return float(fields[i]) if fields[i] else 0
# 字段索引: 0=f51(date), 1=f52(主力), 2=f53(小单), 3=f54(中单),
# 4=f55(大单), 5=f56(超大单), 6=f57(主力%), 7=f58(小单%),
# 8=f59(中单%), 9=f60(大单%), 10=f61(超大单%), 11=f62(收盘价), 12=f63(涨跌幅)
date = fields[0]
main_net = _f(1)
small_net = _f(2)
medium_net = _f(3)
large_net = _f(4)
super_large_net = _f(5)
parsed.append({
"date": date,
"mainNetInflow": main_net,
"superLargeInflow": max(0, super_large_net),
"superLargeOutflow": abs(min(0, super_large_net)),
"largeInflow": max(0, large_net),
"largeOutflow": abs(min(0, large_net)),
"mediumInflow": max(0, medium_net),
"mediumOutflow": abs(min(0, medium_net)),
"smallInflow": max(0, small_net),
"smallOutflow": abs(min(0, small_net)),
"mainNetInflowPercent": _f(6),
"superLargeInflowPercent": _f(10),
"superLargeOutflowPercent": 0,
"largeInflowPercent": _f(9),
"largeOutflowPercent": 0,
"mediumInflowPercent": _f(8),
"mediumOutflowPercent": 0,
"smallInflowPercent": _f(7),
"smallOutflowPercent": 0,
"turnover": 0, # daykline 不含成交额，由调用方从 Tencent K 线合并
"mainForceNet": main_net,
"retailNet": small_net + medium_net,
"changePercent": _f(12),
"closePrice": _f(11),
})
return parsed
except Exception as e:
if attempt == 0:
continue
print(f"[eastmoney] push2his error: {e}")
print(f"[eastmoney] fetch_fund_flow_daykline error: {e}")
return None
return None
backend/services/tencent.py
+21 -4
View File
@@ -192,10 +192,13 @@ async def fetch_quote(code: str) -> Optional[dict]:
async def fetch_history(code: str, days: int = 90) -> List[dict]:
"""获取历史K线（前复权日K），主数据源"""
"""获取历史K线（前复权日K），含涨跌幅
多请求1天以计算第一条的涨跌幅，最终只返回 days 条。
"""
market = get_market_prefix(code)
stock_code = f"{market}{code}"
url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={stock_code},day,,,{days},qfq"
url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={stock_code},day,,,{days + 1},qfq"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "https://stockapp.finance.qq.com/",
@@ -212,17 +215,31 @@ async def fetch_history(code: str, days: int = 90) -> List[dict]:
if not isinstance(raw, list):
return []
result = []
prev_close = 0
for record in raw:
if not isinstance(record, list) or len(record) < 6:
continue
close = float(record[2]) if record[2] else 0
change_pct = 0
if prev_close > 0:
change_pct = (close - prev_close) / prev_close * 100
result.append({
"date": record[0],
"open": float(record[1]) if record[1] else 0,
"close": float(record[2]) if record[2] else 0,
"close": close,
"high": float(record[3]) if record[3] else 0,
"low": float(record[4]) if record[4] else 0,
"volume": int(record[5]) if record[5] else 0,
"volume": int(float(record[5])) if record[5] else 0,
"changePercent": round(change_pct, 2),
})
prev_close = close
# 丢弃最旧1条（它的 prev_close=0 导致涨跌幅为0），只保留最新 days 条
if len(result) > 1:
result = result[1:]
if len(result) > days:
result = result[-days:]
return result
except Exception:
return []