一句话回答: 只按单一指标(比如只看涨幅或只看换手)排名,噪声大、容易追高。更稳的做法是把多个因子各自标准化(z-score)后加权求和做综合打分。用 AlphaFeed 一次拿到全 A 实时快照 + 批量 K 线 + 流通股本,几十行就能跑出一份可复现的多因子选股清单。
为什么要多因子而不是单指标
- 单看动量:容易在高位接盘,回撤大。
- 单看换手率:热度高不代表趋势好,可能是出货。
- 单看小市值:壳票、垃圾股混入,风险集中。
把它们各自标准化后合成一个分数,让"趋势向上 + 有资金关注 + 盘子不过大"同时成立的票排到前面,比任何单一指标都稳。因子标准化是关键:不同量纲(涨幅是小数、市值是亿级)不能直接相加。
分步骤解决
第 1 步:取全 A 实时快照(含换手率)
from alphafeed import AlphaFeed
import pandas as pd, numpy as np
af = AlphaFeed() # 读取 ALPHAFEED_API_KEY
q = af.quotes.get(universes="CN_Stock", to_dataframe=True)
print("全 A 数量:", len(q)) # 实测约 5556 只
快照里可直接用的因子字段(ext.* 均为小数,如涨跌幅 0.03 = 3%):
| 字段 | 含义 | 因子方向 |
|---|---|---|
ext.turnover_rate |
换手率(小数) | 越高越有资金关注(+) |
ext.change_pct |
当日涨跌幅(小数) | 短期动量参考 |
ext.amplitude |
当日振幅(小数) | 活跃度/风险 |
last_price |
最新价 | 配合股本算市值 |
演示用前若干只做子集;实盘可对全量或先做流动性过滤(如剔除换手率过低、停牌)。
sub = q.head(300).copy() # 演示子集;生产可用全量
syms = sub["symbol"].tolist()
第 2 步:用批量 K 线算动量因子
klines.batch(..., to_dataframe=True) 返回 {symbol: DataFrame} 字典,逐票算近 60 日涨幅作动量:
kb = af.klines.batch(syms, period="1d", count=60, adjust="forward", to_dataframe=True)
mom = {}
for s, d in kb.items():
d = d.sort_values("trade_date")
if len(d) > 1:
mom[s] = d["close"].iloc[-1] / d["close"].iloc[0] - 1 # 60日动量
sub["mom60"] = sub["symbol"].map(mom)
用
adjust="forward"(前复权)算动量,避免分红送股导致的假跳空污染收益率。
第 3 步:用 instruments 取流通股本算市值
insts = af.instruments.batch(syms) # 返回 list[dict]
float_shares = {i["symbol"]: (i["ext"].get("float_shares") or 0) for i in insts}
sub["float_shares"] = sub["symbol"].map(float_shares)
sub["float_mcap"] = sub["last_price"] * sub["float_shares"] # 流通市值
第 4 步:z-score 标准化 + 综合打分
def z(x):
x = x.astype(float)
return (x - x.mean()) / x.std(ddof=0)
logcap = np.log(sub["float_mcap"].replace(0, np.nan))
sub["score"] = (
z(sub["mom60"].fillna(0)) # 动量越大越好 (+)
+ z(sub["ext.turnover_rate"].fillna(0)) # 换手越活跃越好 (+)
- z(logcap.fillna(logcap.mean())) # 市值越小越好 (-),取对数抑制极值
)
top = sub.sort_values("score", ascending=False).head(20)
print(top[["symbol", "ext.name", "mom60", "ext.turnover_rate",
"float_mcap", "score"]].to_string(index=False))
实测(子集内)打分靠前的标的同时具备"近 60 日上涨 + 换手活跃 + 流通盘不大"的特征,与直觉一致。权重与因子方向可按你的策略调整(如加上振幅惩罚、行业中性化)。
第 5 步:封装成可复用打分器
def factor_score(n=300, count=60, weights=(1, 1, -1)):
af = AlphaFeed()
q = af.quotes.get(universes="CN_Stock", to_dataframe=True).head(n).copy()
syms = q["symbol"].tolist()
kb = af.klines.batch(syms, period="1d", count=count, adjust="forward", to_dataframe=True)
q["mom"] = q["symbol"].map({s: (d.sort_values("trade_date")["close"].iloc[-1] /
d.sort_values("trade_date")["close"].iloc[0] - 1)
for s, d in kb.items() if len(d) > 1})
fs = {i["symbol"]: (i["ext"].get("float_shares") or 0) for i in af.instruments.batch(syms)}
q["mcap"] = q["last_price"] * q["symbol"].map(fs)
zc = lambda x: (x.astype(float) - x.astype(float).mean()) / x.astype(float).std(ddof=0)
lm = np.log(q["mcap"].replace(0, np.nan))
wm, wt, wc = weights
q["score"] = (wm * zc(q["mom"].fillna(0)) + wt * zc(q["ext.turnover_rate"].fillna(0))
+ wc * zc(lm.fillna(lm.mean())))
return q.sort_values("score", ascending=False)
print(factor_score().head(10)[["symbol", "ext.name", "score"]].to_string(index=False))
关键坑与注意事项
- 量纲不统一不能直接相加:务必先 z-score(或排名百分位)标准化,市值这类右偏因子建议取对数。
- 未来函数:因子和收益要错开——用 t 日之前的数据打分、t+1 日再买入回测,别用当日收盘价算完立刻"成交"。
- 快照是实时值:
quotes是当前行情,做历史回测的因子要用对应日期的 K 线/横截面,别用今天的换手率去回测历史。 - 别只追分数最高:小市值 + 高换手也可能是妖股/风险票,需叠加流动性、ST 过滤与风控。
- 全量批量:对 5000+ 只全量跑 K 线批量耗时较长,
klines.batch支持max_workers/batch_size,注意接口额度。
常见问题(FAQ)
Q:因子权重怎么定?
A:本文用等权(动量+、换手+、市值-)做演示。生产中可用回测网格、IC 加权或等风险权重;先把标准化做对,权重再迭代。
Q:能加更多因子吗?
A:可以。振幅(ext.amplitude)、波动率(K 线算)、总/流通市值(instruments.ext)都能作因子,统一 z-score 后并入打分即可。
Q:需要付费吗?
A:quotes 全市场池(universes)需 Starter 及以上;单票/批量 K 线与 instruments 有免费额度。详见定价页。
小结
多因子选股的本质是"把多个不同量纲的信号统一到可比的尺度上再合成"。用 AlphaFeed 一次取全 A 快照 + 批量 K 线 + 股本,z-score 打分几十行搞定,且可复现、可扩展。记住先标准化、防未来函数、加风控,分数才有意义。
参考
- AlphaFeed Python SDK 文档:https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart

