东方财富数据源

概述

东方财富是独有数据最全面的数据源（龙虎榜/解禁/两融/大宗/股东户数/分红/研报/ 新闻/财务指标/涨停池），但东财有严格的 IP 级风控，高频请求会触发 403 封禁。

K线/实时行情/市值/PE/PB 禁止以东财为主源（走腾讯或 TDX），东财仅作末位兜底且需记 warning；资金流向/行业排名/估值的主源分别为 TDX / 腾讯 getRank / 腾讯 quote。

实现文件

backend/app/services/data_sources/em_client.py — 限流入口（em_get）
backend/app/services/data_sources/em_paginated.py — 分页拉取 + push2 轮转
backend/app/services/market_data_service/providers_eastmoney.py — 市场概览侧取数（em_fetch_with_retry）

防封策略（参考 AkShare + a-stock-data）

1. 编号 push2 主机轮转

主机池统一定义在 em_paginated.py（单一来源原则），em_client.py 通过导入引用：

# em_paginated.py — 唯一定义源
PUSH2_HOSTS = [
    "push2.eastmoney.com",
    "17.push2.eastmoney.com",
    "29.push2.eastmoney.com",
    "40.push2.eastmoney.com",
    "79.push2.eastmoney.com",
    "91.push2.eastmoney.com",
]

def get_push2_url(path: str = "/api/qt/clist/get") -> str:
    """轮转获取带编号主机的 URL，分散 CDN 负载"""
    return f"https://{_next_push2_host()}{path}"

# em_client.py — 从 em_paginated 导入，不再重复定义
from .em_paginated import PUSH2_HOSTS, PUSH2_HIS_HOSTS

配套 HostFallbackManager：单主机失败自动切镜像主机，连续 HOST_FAILURE_THRESHOLD=2 次失败进入 30s 域名冷却；CDN 坏节点返回的网关 HTML（可能带 200 状态码）也会被判为无效响应并切换下一主机。

2. 请求限流（em_get）

# em_client.py 核心逻辑（EM_MIN_INTERVAL 从 core/config.py 读取）
def _em_get_impl(url, params=None, headers=None, timeout=15, total_budget=None):
    with _em_throttle_lock:   # 加锁：多线程批量下载时"等待+时间戳更新"必须原子
        wait = s.EM_MIN_INTERVAL - (time.time() - _em_last_call[0])
        if wait > 0:
            time.sleep(wait + random.uniform(s.EM_JITTER_MIN, s.EM_JITTER_MAX))
        _em_last_call[0] = time.time()
    ...

EM_MIN_INTERVAL = 0.3s（批量同步实测安全值，可经环境变量调回更保守值）
EM_JITTER_MIN / EM_JITTER_MAX = 0.1 / 0.5s
EM_SESSION 用 curl_cffi 模拟 Chrome TLS 指纹，大幅降低 403 封锁率
total_budget：多主机降级链总时间预算，超时立即抛出，避免串行重试拖垮请求方

3. 403 不重试 + 429 退避 + 熔断

# providers_eastmoney.py::em_fetch_with_retry
if resp.status_code == 403:
    # 东财 IP 级风控：不重试，直接触发熔断
    em_breaker.record_failure()
    em_record_host_failure(host)     # 风控信号：同步共享冷却
    raise RateLimitError("403 Forbidden (东财风控)", source="eastmoney", status_code=403)

if resp.status_code == 429:
    # 限流：较长退避后重试，重试耗尽抛出 RateLimitError
    last_err = RateLimitError("429 Too Many Requests", source="eastmoney", status_code=429)
    await asyncio.sleep(2 ** attempt * 1.5 + random.uniform(0.2, 0.8))
    continue

镜像池全部冷却时 em_all_cooling(host) 为真，直接抛 NetworkError 快速失败，不再白等 3 次退避。网络瞬态错误（超时/连接失败）自动包装为 NetworkError，上层按异常类型精确决策重试/降级（详见「结构化异常体系」页）。

4. 指数退避重试

# 仅对瞬态网络错误重试（超时/连接失败）
await asyncio.sleep(2 ** attempt * 0.5 + random.uniform(0.1, 0.4))

通用装饰器版本见 services/data_sources/retry_policy.py： with_retry(max_attempts=3, backoff_base=1.0, backoff_factor=2.0, max_backoff=30.0)， wait = min(backoff_base * backoff_factor**attempt, max_backoff)。

主要 API 端点

端点	用途	参数
`/api/qt/ulist.np/get`	指数实时行情	secids, fields
`/api/qt/clist/get`	板块/个股列表	fs, fid, fields, pn, pz
`/api/qt/stock/get`	单股详情	secid, fields
`/api/qt/stock/kline/get`	K线历史	secid, klt, fqt, lmt
`push2his.eastmoney.com`	历史K线	同上

push2ex 涨停板行情中心（`limit_board.py`）

端点	用途	`sort`
`getTopicZTPool`	涨停池（指定日期）	`fbt:asc`
`getTopicZBPool`	炸板池	`fbt:asc`
`getTopicDTPool`	跌停池	`fund:asc`
`getYesterdayZTPool`	昨日涨停池（昨涨停今表现，算晋级率/赚钱效应）	必须 `zs:desc`

date 传 YYYYMMDD；非交易日会回落到「该日之前最近交易日」的池子。 limit_up_pool_unified(date) = 封板池 + 腾讯口径补充条目（与首页涨停数一致）。

push2ex 的 rc 是业务层状态码，HTTP 200 ≠ 业务成功

{"rc":102,"data":null} 这种响应 HTTP 状态是 200，但 data 为 null。若代码只写 (body.get("data") or {}).get("pool") or []，就会被静默吞成空池 —— 面板整块空掉，日志里一条线索都没有。_em_zt_api 现已显式校验：

rc = body.get("rc")
if rc not in (0, None):          # None 兼容缺字段的老响应
    raise RuntimeError(f"{endpoint} 业务错误 rc={rc} (date={date}, sort={sort})")

已知触发：getYesterdayZTPool 只接受 sort=zs:desc，传 fbt:asc 即返回 rc=102。调用方宜准备回落路径（参见 abnormal_moves._yesterday_pool()：主源为空时回落 limit_up_pool(上一交易日)，实测两路成分完全一致）。

腾讯口径代码集是「当日实时」的，不要用在历史日期上

limit_up_pool_unified 会用 tencent_limit_codes() 补充「东财未计封板」的个股，而该函数没有日期参数（取当日实时行情）。若查询历史日期还去合并，会把今日涨停股混进历史池 —— 静默串数据。故该函数已加守卫：仅当查询「今天」时才合并。

分页拉取（参考 AkShare fetch_paginated_data）

全市场数据需分页拉取，页间随机延迟避免风控：

async def fetch_paginated_data_async(url, params, max_pages=50, page_delay=(0.3, 0.8)):
    for page in range(1, max_pages + 1):
        params["pn"] = str(page)
        resp = await client.get(url, params=params)
        diff = resp.json()["data"]["diff"]
        all_rows.extend(diff)
        if page * len(diff) >= total:
            break
        await asyncio.sleep(random.uniform(*page_delay))  # 页间延迟

数据标准化

东财返回的数值字段可能为 "-"（无数据），必须通过 safe_float() 转换：

def safe_float(value, default=0.0) -> float:
    if value is None: return default
    if isinstance(value, (int, float)): return float(value)
    if isinstance(value, str):
        value = value.strip()
        if not value or value == "-" or value == "--": return default
        try: return float(value)
        except ValueError: return default
    return default

注意事项

东财风控要点

403 绝对不重试：重试只会加重封禁
请求间隔 ≥ 0.3 秒（EM_MIN_INTERVAL 默认值）+ 0.1~0.5 秒随机抖动；风控收紧时可用环境变量调大做保守模式
编号主机分散：避免单节点过载；单主机连续失败 2 次（HOST_FAILURE_THRESHOLD）冷却 30s
熔断器保护：连续 4 次失败（EM_BREAKER_FAILURE_THRESHOLD）后跳过 120 秒（EM_BREAKER_COOLDOWN_SEC）
不缓存空数据：东财被封时不写入缓存

概述​

实现文件​

防封策略（参考 AkShare + a-stock-data）​

1. 编号 push2 主机轮转​

2. 请求限流（em_get）​

3. 403 不重试 + 429 退避 + 熔断​

4. 指数退避重试​

主要 API 端点​

push2ex 涨停板行情中心（limit_board.py）​

分页拉取（参考 AkShare fetch_paginated_data）​

数据标准化​

注意事项​

相关文档​

概述