同一个选股任务，用 akshare、Tushare、AlphaFeed 各实现一遍——代码量差了多少？
很多对比文章只说"这个好那个差"，不给代码。
这篇文章不扯虚的。我挑了一个非常常见的量化选股任务：
从全市场 A 股中，找出今天成交额 > 5 亿、涨幅 > 3%、且最新价站上 20 日均线的股票。
然后分别用 akshare、Tushare Pro、AlphaFeed 实现，每一行代码都贴出来。你可以自己判断哪个用起来舒服。
任务拆解
要完成这个选股任务，实际上要做四步：
- 获取全市场实时行情 — 拿到每只票的涨幅和成交额
- 初筛 — 过滤出成交额 > 5 亿、涨幅 > 3% 的票
- 获取初筛结果的 K 线 — 拿到这些票最近 20 天的日 K 线
- 计算 MA20 并二次筛选 — 最新价是否站上 20 日均线
一步一步来对比。
akshare 的实现
第 1 步：获取全市场实时行情
akshare 没有一个函数能直接拿到全市场行情，需要用 ak.stock_zh_a_spot_em() 拉东方财富的全量数据：
import akshare as ak
import pandas as pd
import time
# 获取全市场实时行情
df_all = ak.stock_zh_a_spot_em()
# 问题 1：列名是中文，需要手动对应
# '最新价', '涨跌幅', '成交额' — 不同版本列名可能不同
# 问题 2：这个函数底层是爬东方财富，如果东财改版就会挂
第 2 步：初筛
df_all["涨跌幅"] = pd.to_numeric(df_all["涨跌幅"], errors="coerce")
df_all["成交额"] = pd.to_numeric(df_all["成交额"], errors="coerce")
df_all["最新价"] = pd.to_numeric(df_all["最新价"], errors="coerce")
candidates = df_all[
(df_all["成交额"] > 5e8) & # 成交额 > 5 亿
(df_all["涨跌幅"] > 3) # 涨幅 > 3%
].copy()
# 需要把代码转换成标准格式来拉 K 线
# akshare 的代码是纯数字如 '600519'，后续需要加前缀
symbols = candidates["代码"].tolist()
print(f"初筛 {len(symbols)} 只票")
第 3 步：获取 K 线并计算 MA20
results = []
for i, code in enumerate(symbols):
try:
df_k = ak.stock_zh_a_hist(
symbol=code,
period="daily",
start_date="20260601", # 需要手动算开始日期
end_date="20260817",
adjust="qfq"
)
if df_k is None or len(df_k) < 20:
continue
df_k["收盘"] = pd.to_numeric(df_k["收盘"], errors="coerce")
ma20 = df_k["收盘"].rolling(20).mean().iloc[-1]
last_price = df_k["收盘"].iloc[-1]
if last_price > ma20:
results.append({
"代码": code,
"最新价": last_price,
"MA20": round(ma20, 2)
})
except Exception as e:
print(f"[{code}] 失败: {e}")
continue
# 必须加 sleep，否则被封 IP
if i % 5 == 0 and i > 0:
time.sleep(1)
print(f"最终筛选出 {len(results)} 只")
akshare 小结
| 指标 | 情况 |
|---|---|
| 代码行数 | 约 40 行 |
| 循环次数 | 初筛出几十只票，每只一次 HTTP 请求 |
| 需要 sleep | ✅ 不加大概率被封 |
| 中文列名 | ✅ 需要记住或查文档 |
| 出错处理 | 手动 try/except |
| 预计耗时 | 初筛快（几秒），K 线循环慢（几十秒到几分钟） |
Tushare Pro 的实现
第 1 步：获取全市场实时行情
Tushare Pro 的日线行情通过 daily 接口获取，但实时行情需要较高积分：
import tushare as ts
import pandas as pd
import time
pro = ts.pro_api("你的token")
# 获取当日行情（需要积分 ≥ 120）
df_all = pro.daily(trade_date="20260817")
# 问题：这是收盘后的日行情，不是盘中实时数据
# 实时数据需要更高积分或用其他接口
第 2 步：初筛
candidates = df_all[
(df_all["amount"] > 500000) & # 成交额单位是千元
(df_all["pct_chg"] > 3) # 涨跌幅
].copy()
symbols = candidates["ts_code"].tolist()
print(f"初筛 {len(symbols)} 只票")
第 3 步：获取 K 线并计算 MA20
results = []
for i, code in enumerate(symbols):
try:
df_k = pro.daily(
ts_code=code,
start_date="20260701",
end_date="20260817"
)
if df_k is None or len(df_k) < 20:
continue
df_k = df_k.sort_values("trade_date") # Tushare 返回倒序，需要手动排序
# Tushare daily 接口返回的是不复权数据
# 前复权需要额外调用 adj_factor 接口
adj = pro.adj_factor(ts_code=code, start_date="20260701", end_date="20260817")
adj = adj.sort_values("trade_date")
df_k = df_k.merge(adj[["trade_date", "adj_factor"]], on="trade_date")
df_k["close_adj"] = df_k["close"] * df_k["adj_factor"] / df_k["adj_factor"].iloc[-1]
ma20 = df_k["close_adj"].rolling(20).mean().iloc[-1]
last_price = df_k["close_adj"].iloc[-1]
if last_price > ma20:
results.append({
"代码": code,
"最新价": round(last_price, 2),
"MA20": round(ma20, 2)
})
except Exception as e:
print(f"[{code}] 失败: {e}")
continue
# Tushare 有频率限制（免费用户 200 次/分钟）
if i % 10 == 0 and i > 0:
time.sleep(1)
print(f"最终筛选出 {len(results)} 只")
Tushare 小结
| 指标 | 情况 |
|---|---|
| 代码行数 | 约 45 行 |
| 循环次数 | 每只票 2 次请求（K 线 + 复权因子） |
| 需要 sleep | ✅ 超过频率限制会报错 |
| 复权处理 | 需要手动拉复权因子并计算 |
| 排序 | 返回倒序，需手动 sort |
| 预计耗时 | 几十秒到几分钟，取决于初筛数量 |
AlphaFeed 的实现
完整代码
from alphafeed import AlphaFeed
import pandas as pd
af = AlphaFeed()
# 第 1 步：一行拿全市场实时行情
all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
# 第 2 步：初筛
candidates = all_cn[
(all_cn["amount"] > 5e8) &
(all_cn["change_rate"] > 0.03)
].copy()
symbols = candidates["symbol"].tolist()
print(f"初筛 {len(symbols)} 只票")
# 第 3 步：批量拉 K 线 + MA20 筛选
dfs = af.klines.batch(symbols, period="1d", count=25,
adjust="forward", to_dataframe=True, show_progress=True)
results = []
for symbol, df_k in dfs.items():
if len(df_k) < 20:
continue
df_k["ma20"] = df_k["close"].rolling(20).mean()
last_row = df_k.iloc[-1]
if last_row["close"] > last_row["ma20"]:
results.append({
"代码": symbol,
"最新价": round(last_row["close"], 2),
"MA20": round(last_row["ma20"], 2)
})
print(f"最终筛选出 {len(results)} 只")
pd.DataFrame(results).to_csv("selected.csv", index=False)
就这些。
AlphaFeed 小结
| 指标 | 情况 |
|---|---|
| 代码行数 | 约 20 行 |
| 循环次数 | 0（全市场一次 + 批量 K 线一次） |
| 需要 sleep | ❌ SDK 内部自动控制并发和重试 |
| 复权处理 | 参数adjust="forward" 直接搞定 |
| 排序 | 不需要，返回就是时间正序 |
| 预计耗时 | 几秒 |
并排对比
| 对比维度 | akshare | Tushare Pro | AlphaFeed |
|---|---|---|---|
| 代码总行数 | ~40 行 | ~45 行 | ~20 行 |
| 全市场行情 | 有（爬虫） | 有（需积分） | 有（universes） |
| 批量 K 线 | ❌ 循环 | ❌ 循环 | ✅ batch |
| 复权 | 参数指定 | 手动拉因子计算 | 参数指定 |
| sleep | 必须 | 必须 | 不需要 |
| 出错处理 | 手动写 | 手动写 | SDK 内置 |
| 数据排序 | 已排序 | 倒序要翻转 | 已排序 |
| 列名 | 中文 | 英文 | 英文 |
| 总耗时 | 1-5 分钟 | 1-3 分钟 | 几秒 |
| 费用 | 免费 | Token 积分制 | 免费可完成 |
这个差距是怎么来的
不是说 akshare 或 Tushare 写得差——它们在各自的设计目标下做得很好。差距来自架构层面：
1. 有没有全市场一次查询
akshare 和 Tushare 都是"给我一个代码，我返回这个代码的数据"。AlphaFeed 多了一层抽象：universes，你可以按市场一次拿到所有标的的数据。
2. 有没有原生批量接口
akshare 和 Tushare 拉多只票的数据，就是循环调用。AlphaFeed 的 batch 是 SDK 层面的——内部自动分块、多线程并发、失败重试，你只需要传一个列表进去。
3. 复权是不是一等公民
AlphaFeed 的复权是 API 参数，支持 5 种模式。Tushare 的日线接口返回不复权数据，前复权需要额外拉复权因子自己算。akshare 好一点，参数可以指定，但爬虫稳定性是另一回事。
4. SDK 是否帮你处理了脏活
重试、并发控制、错误处理、数据格式统一……这些在 AlphaFeed SDK 里是内置的，其他数据源需要你自己写。
换一个任务也是一样
上面用的是"涨幅 + 成交额 + 均线"选股。你换成任何其他任务——放量突破、缩量回调、跨市场比较、盘口分析——同样的模式：
- akshare / Tushare：循环 + sleep + try/except + 手动处理
- AlphaFeed：universes + batch + 参数 → 结果
差距不是某一行代码快一点，而是整个工作流的效率不在一个级别。
你可以自己试
别听我说。用你现在的数据源，把上面那个任务跑一遍。记录一下：
- 你写了多少行代码
- 等了多久
- 中间出了几次错
- 你花了多少时间在"数据获取"而不是"策略逻辑"上
然后用 AlphaFeed 免费版跑一遍同样的任务，对比一下。
pip install alphafeed
注册后免费即可完成上面的全部代码。
- AlphaFeed 官网：https://alphafeed.org/
- Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart