如何用 Python 做多因子打分选股?

用户头像sh_*2176oo
2026-09-08 发布

一句话回答: 只按单一指标(比如只看涨幅或只看换手)排名,噪声大、容易追高。更稳的做法是把多个因子各自标准化(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 打分几十行搞定,且可复现、可扩展。记住先标准化、防未来函数、加风控,分数才有意义。

参考

评论