efinance、akshare 常见报错解决方案

用户头像sh_****1449ws
2026-05-20 发布

efinance、akshare 被限流、封 IP、弹验证码怎么办?2026 年稳定获取 A 股数据的解决方案

用 efinance 或 akshare 跑全市场行情,跑着跑着就报错了?请求被拒绝、弹出验证码、IP 被拉黑?这不是你代码的问题,而是底层数据源的限制。本文分析问题根源,并给出一个可以长期稳定使用的替代方案。


一、你是不是遇到了这些问题?

如果你正在用 efinance 或 akshare 获取 A 股数据,以下场景大概率遇到过:

场景 1:批量获取 K 线中途报错

import efinance as ef

# 想拉全市场5000+只股票的K线
for code in stock_list:
    df = ef.stock.get_quote_history(code)  # 跑到几百只就报错了

报错信息类似:

requests.exceptions.ConnectionError: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))

或者直接返回空 DataFrame,没有任何报错但数据是空的。

场景 2:刷全市场实时行情被封

import akshare as ak

# 循环获取实时行情
while True:
    df = ak.stock_zh_a_spot_em()  # 几分钟后开始返回空数据或报错
    time.sleep(1)

跑了一会之后,要么返回空数据,要么直接被重定向到验证码页面。

场景 3:IP 被拉入"小黑屋"

频繁访问后,你的 IP 被东方财富标记了。即使停止程序,等一两个小时甚至一两天再试,依然无法正常获取数据。浏览器打开东方财富网页也会弹出验证码。

场景 4:加了代理池还是不行

有人尝试用代理 IP 池来绕过限流:

proxies = {"http": "//proxy:port"}
# 换了 IP 还是被检测到异常流量

效果不理想。东方财富的反爬策略不仅看 IP,还看请求频率模式、User-Agent、Cookie 等。而且维护代理池本身就是额外的成本和麻烦。


二、问题的根源是什么?

要理解这些问题为什么无法从根本上解决,需要看清底层逻辑。

efinance 和 akshare 的本质

它们都是 爬虫库,底层是从东方财富(eastmoney.com)的网页接口抓取数据:

你的代码 → efinance/akshare → HTTP 请求 → 东方财富网页接口 → 返回数据

这些网页接口本来是给浏览器用的,不是为程序化批量调用设计的。

为什么现在越来越严格?

  1. 量化参与者暴增:AI 降低了门槛,越来越多人用爬虫拉数据,东方财富带宽成本急剧上升
  2. 反爬策略升级:东方财富在 2025 年开始大幅加强限流,包括 IP 封禁、验证码验证、请求频率检测
  3. 没有 SLA 保障:网页接口随时可能修改、下线,没有任何稳定性承诺

你能做什么?

很多人尝试过的"解决办法":

方法 效果 问题
加 sleep 降低频率 有一定效果 全市场 5000 只股票要跑很久
换代理 IP 短期有效 成本高,维护麻烦,仍可能被检测
换 User-Agent / Cookie 几乎无效 东方财富检测维度很多
等一段时间再跑 临时有效 不解决根本问题
换其他爬虫库 无效 底层都是同一个数据源

这些都是治标不治本。根本问题在于:你在用非官方接口做批量数据获取,这条路本身就不可持续。


三、正确的解决思路

与其和反爬策略斗智斗勇,不如换一个从架构上就没有这个问题的数据源。

一个稳定的数据源需要满足:

  1. 不依赖爬虫:数据从正规渠道获取,不是爬网页
  2. 有明确的 API:为程序化调用设计,不是网页接口
  3. 支持批量请求:一次请求多只股票,不需要循环
  4. 不会被封 IP:合法使用不会触发任何限制
  5. 有免费层:学习和研究阶段不需要花钱

四、替代方案:TickFlow

TickFlow 是一个专为量化研究设计的数据服务,底层不依赖爬虫,提供标准化 API 和 Python SDK。

核心区别

爬虫方案:你的代码 → 爬虫库 → 东方财富网页 → 可能被封
TickFlow:你的代码 → TickFlow SDK → TickFlow API → 稳定返回数据

TickFlow 的数据不是从东方财富爬的,而是从正规数据渠道获取,经过清洗和标准化后通过 API 提供。不存在被封 IP、弹验证码的问题。


五、对比:爬虫方案 vs TickFlow

下面用具体场景做对比。

场景 1:获取单只股票日 K 线

efinance 写法

import efinance as ef

df = ef.stock.get_quote_history("600000")
print(df.tail())
# 正常情况下能用,但高频调用后可能返回空数据

TickFlow 写法

from tickflow import TickFlow

tf = TickFlow.free()  # 免费,无需注册

df = tf.klines.get("600000.SH", period="1d", count=10000, as_dataframe=True)
print(df.tail())
# 永远不会被封,单次最多获取 10000 根 K 线

场景 2:批量获取全市场 K 线(最容易被封的场景)

efinance 写法(大概率中途被封):

import efinance as ef
import time

stock_list = ef.stock.get_realtime_quotes()["股票代码"].tolist()

data = {}
for i, code in enumerate(stock_list):
    try:
        df = ef.stock.get_quote_history(code)
        data[code] = df
    except Exception as e:
        print(f"第 {i} 只报错: {e}")
        time.sleep(10)  # 被封了就等一等,但越等越慢
    time.sleep(0.5)  # 加延迟降低频率

# 问题:5000只股票,每只0.5秒,至少要跑40多分钟
# 而且跑到中途很可能被封

TickFlow 写法(稳定且快):

from tickflow import TickFlow

tf = TickFlow.free()

# 获取全市场标的
universe = tf.universes.get("CN_Equity_A")
symbols = universe["symbols"]
print(f"共 {len(symbols)} 只 A 股")

# 批量获取,SDK 自动分批并发
dfs = tf.klines.batch(
    symbols,
    period="1d",
    count=200,
    as_dataframe=True,
    show_progress=True,
)

print(f"成功获取 {len(dfs)} 只股票的数据")
# 1-2 分钟完成全市场下载,不会被封

场景 3:实时行情(最刚需也最容易出问题的场景)

akshare 写法(容易被封):

import akshare as ak
import time

while True:
    try:
        df = ak.stock_zh_a_spot_em()  # 全市场行情
        print(f"获取到 {len(df)} 只股票")
    except Exception as e:
        print(f"报错了: {e}")
        time.sleep(60)  # 被封就等一分钟
    time.sleep(3)

# 问题:跑几轮就可能被东方财富限流

TickFlow 写法(稳定):

from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

# 一行获取全市场 A 股实时行情
quotes = tf.quotes.get(universes=["CN_Equity_A"], as_dataframe=True)
print(f"获取到 {len(quotes)} 只 A 股行情")

# 也可以只查几只
quotes = tf.quotes.get(symbols=["600519.SH", "000001.SZ"], as_dataframe=True)
print(quotes[["symbol", "last_price", "volume", "amount"]])

如果需要更低延迟的持续推送,可以用 WebSocket:

import datetime
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")
stream = tf.stream

@stream.on_quotes
def on_quotes(quotes):
    for q in quotes:
        ts = datetime.datetime.fromtimestamp(q["timestamp"] / 1000)
        ext = q.get("ext", {})
        name = ext.get("name", "")
        change_pct = ext.get("change_pct")
        change_str = f"{change_pct:+.2%}" if change_pct is not None else "N/A"
        print(f"[{ts:%H:%M:%S}] {q['symbol']} {name} {q['last_price']} {change_str}")

stream.subscribe("quotes", ["600519.SH", "000001.SZ", "600000.SH"])
stream.connect()

场景 4:分钟 K 线

akshare 写法

import akshare as ak

# akshare 的分钟K线数据有限,且同样面临限流问题
df = ak.stock_zh_a_hist_min_em(symbol="600000", period="5", adjust="qfq")

TickFlow 写法

from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

# 历史分钟 K 线
df = tf.klines.get("600000.SH", period="5m", count=1000, as_dataframe=True)
print(df.tail())

# 当日日内分钟 K 线
df_today = tf.klines.intraday("600000.SH", period="1m", as_dataframe=True)
print(df_today.tail())

# 批量获取多只股票的日内分钟 K
dfs = tf.klines.intraday_batch(
    ["600000.SH", "000001.SZ", "600519.SH"],
    as_dataframe=True,
    show_progress=True,
)

六、常见问题解答

Q: TickFlow 免费层能做什么?

免费层无需注册,直接使用:

  • 历史日 K 线(1d、1w、1M、1Q、1Y)
  • 标的信息查询
  • 标的池查询
from tickflow import TickFlow

tf = TickFlow.free()

# 这些全部免费
df = tf.klines.get("600000.SH", period="1d", count=10000, as_dataframe=True)
instruments = tf.instruments.batch(symbols=["600000.SH", "000001.SZ"])
universe = tf.universes.get("CN_Equity_A")

对于只做日线级别回测和研究的用户,免费层完全够用。

Q: 实时行情和分钟 K 需要付费吗?

需要注册获取 API key。在 tickflow.org 注册后在控制台生成。

Q: TickFlow 会不会也被封?

不会。TickFlow 是正规的 API 数据服务,你的请求是发给 TickFlow 服务器,不是发给东方财富。只要在合理的调用频率内使用,不会被封。SDK 内置了自动重试和频率控制。

Q: 代码迁移成本大吗?

很低。主要就是换一下数据获取的部分,后续的 pandas 分析逻辑完全不变:

# 之前(efinance)
import efinance as ef
df = ef.stock.get_quote_history("600000")

# 现在(TickFlow)
from tickflow import TickFlow
tf = TickFlow.free()
df = tf.klines.get("600000.SH", period="1d", count=10000, as_dataframe=True)

# 后续分析代码不需要改
df["ma5"] = df["close"].rolling(5).mean()
df["ma20"] = df["close"].rolling(20).mean()

返回的 DataFrame 包含 openhighlowclosevolumeamount 等标准字段,和 efinance/akshare 的数据结构类似。

Q: 支持哪些市场?

  • A 股(SH / SZ / BJ)
  • 国内期货(SHF / DCE / ZCE / CFX / INE / GFE)
  • 美股(US)
  • 港股(HK)

标的代码格式统一为 代码.市场后缀,例如 600000.SHAAPL.US00700.HK


七、迁移示例:从 efinance 迁移一个完整策略

假设你之前用 efinance 写了一个均线策略,迁移到 TickFlow 只需要改数据获取部分:

迁移前(efinance)

import efinance as ef
import pandas as pd
import time

codes = ["600000", "000001", "600519"]
data = {}

for code in codes:
    try:
        df = ef.stock.get_quote_history(code)
        data[code] = df
    except:
        print(f"{code} 获取失败")
    time.sleep(1)  # 不加延迟必被封

for code, df in data.items():
    df["ma5"] = df["收盘"].rolling(5).mean()
    df["ma20"] = df["收盘"].rolling(20).mean()
    latest = df.iloc[-1]
    if latest["ma5"] > latest["ma20"]:
        print(f"{code}: 均线多头")

问题:

  • 股票多了就被封
  • 列名是中文("收盘"而不是 "close"),不方便
  • 没有复权选项

迁移后(TickFlow)

from tickflow import TickFlow

tf = TickFlow.free()

symbols = ["600000.SH", "000001.SZ", "600519.SH"]

# 一行批量获取,不会被封
dfs = tf.klines.batch(symbols, period="1d", count=200, adjust="forward", as_dataframe=True)

for symbol, df in dfs.items():
    df["ma5"] = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    latest = df.iloc[-1]
    if latest["ma5"] > latest["ma20"]:
        print(f"{symbol}: 均线多头")

改动点:

  • ef.stock.get_quote_history(code)tf.klines.batch(symbols, ...)
  • 列名从中文变为英文标准字段
  • 不需要 time.sleep
  • 不需要 try/except 处理封禁
  • 自带复权支持

八、功能对比总结

功能 efinance akshare TickFlow
数据来源 爬东方财富网页 爬东方财富等网页 正规数据渠道 API
是否会被封 IP 频繁使用会被封 频繁使用会被封 不会
验证码问题
日 K 线 支持 支持 支持(免费)
分钟 K 线 有限 有限且不稳定 1m/5m/15m/30m/60m
实时行情 不稳定 不稳定 稳定(REST + WebSocket)
批量获取速度 慢(逐只请求+延迟) 快(5000只/1-2分钟)
复权方式 有限 依赖数据源 5种复权方式
免费层 完全免费 完全免费 日K免费,实时行情需注册
长期稳定性 差(接口随时变) 好(有 SLA)

九、安装和快速上手

pip install "tickflow[all]" --upgrade
from tickflow import TickFlow

# 免费服务,直接用,不注册不登录
tf = TickFlow.free()

# 日 K 线
df = tf.klines.get("600519.SH", period="1d", count=5000, as_dataframe=True)
print(df.tail())

# 批量
symbols = ["600000.SH", "000001.SZ", "600519.SH"]
dfs = tf.klines.batch(symbols, period="1d", count=1000, as_dataframe=True, show_progress=True)

# 标的信息
inst = tf.instruments.get("600519.SH")
print(f"{inst['symbol']}: {inst['name']}")

需要实时行情时,去 tickflow.org 注册获取 API key:

tf = TickFlow(api_key="your-api-key")

# 全市场实时行情
quotes = tf.quotes.get(universes=["CN_Equity_A"], as_dataframe=True)
print(f"共 {len(quotes)} 只 A 股")

十、总结

efinance 和 akshare 被封 IP、限流、弹验证码,本质原因是它们通过爬虫从东方财富网页抓数据,而东方财富在持续加强反爬。这个趋势只会越来越严格,不会放松。

与其花时间对抗反爬(加代理、加延迟、换 User-Agent),不如直接换一个从设计上就不存在这些问题的数据源。

TickFlow 的方案:

  • 免费层日 K 直接用,无需注册
  • 批量获取全市场 1-2 分钟,不会被封
  • 实时行情稳定,支持 WebSocket
  • SDK 设计统一,迁移成本低

数据获取应该是最简单的一步,而不是最头疼的一步。


相关链接


你的时间应该花在策略上,而不是跟反爬做斗争。

评论