用 Python 做多因子选股：从单一指标到因子打分体系

2601_96288572

278人浏览 · 2026-06-16 12:55:52

2601_96288572 · 2026-06-16 12:55:52 发布

前面几篇文章里，我们做过各种选股：按涨幅筛、按放量筛、按均线排列筛、按动量排名。这些筛选条件本质上都是因子——用来描述一只股票某个维度特征的量化指标。

但单因子选股有个问题：每个因子只能捕捉一个维度的信息。动量因子选出来的票可能流动性很差；换手率高的票可能都是在跌。你需要一种方法把多个因子组合起来，做出更稳健的判断。

这就是多因子选股的核心思路：给每只股票在多个维度上打分，然后按综合得分排序。

这篇文章用 AlphaFeed 的数据，从零搭一个轻量级多因子选股框架。不用任何第三方因子库，纯 Python + pandas 实现。

1. 什么是因子

用一句话说：因子就是你认为跟股票未来收益有关的任何量化指标。

因子类别	例子	逻辑
动量	过去 20 天收益率	涨得好的继续涨
波动率	过去 20 天收益率标准差	低波动可能意味着稳定
换手率	成交量 / 流通股本	高换手说明关注度高
量价背离	价格涨但成交量跌	可能是上涨动力不足
反转	过去 5 天收益率（取反）	短期跌多了可能反弹

因子研究的目标不是找到一个"万能因子"，而是找到若干个有一定预测力的因子，然后组合起来使用。

2. 获取全市场数据

多因子选股需要横截面数据——同一时刻所有股票的因子值。我们先拉全市场行情和历史 K 线：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

today_quotes = af.quotes.get(universes="CN_Stock", to_dataframe=True)

liquid = today_quotes[today_quotes["amount"] > 3e8].copy()
print(f"流动性过滤后: {len(liquid)} 只")

symbols = liquid["symbol"].tolist()

if len(symbols) > 500:
    symbols = symbols[:500]

klines = af.klines.batch(
    symbols,
    period="1d",
    count=60,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

print(f"成功获取 K 线: {len(klines)} 只")

3. 计算单个因子：20 日动量

先从最简单的因子开始——过去 20 个交易日的涨跌幅：

import pandas as pd

def calc_momentum(klines: dict, period: int = 20) -> pd.DataFrame:
    records = []
    for sym, df in klines.items():
        df = df.sort_values("trade_date").reset_index(drop=True)
        if len(df) < period + 1:
            continue
        ret = df["close"].iloc[-1] / df["close"].iloc[-period] - 1
        records.append({"symbol": sym, "momentum": ret})
    return pd.DataFrame(records)

mom_df = calc_momentum(klines, period=20)
print(f"动量因子覆盖: {len(mom_df)} 只")
print(mom_df.sort_values("momentum", ascending=False).head(10))

4. 构建多个因子

我们再加几个维度的因子：

import numpy as np
import pandas as pd

def calc_all_factors(klines: dict) -> pd.DataFrame:
    records = []

    for sym, df in klines.items():
        df = df.sort_values("trade_date").reset_index(drop=True)
        if len(df) < 30:
            continue

        df["ret"] = df["close"].pct_change()

        momentum_20 = df["close"].iloc[-1] / df["close"].iloc[-20] - 1

        momentum_5 = df["close"].iloc[-1] / df["close"].iloc[-5] - 1

        volatility_20 = df["ret"].iloc[-20:].std() * np.sqrt(252)

        avg_amount_20 = df["amount"].iloc[-20:].mean()
        avg_amount_5 = df["amount"].iloc[-5:].mean()
        volume_change = avg_amount_5 / avg_amount_20 if avg_amount_20 > 0 else 1.0

        ma5 = df["close"].rolling(5).mean().iloc[-1]
        ma20 = df["close"].rolling(20).mean().iloc[-1]
        ma_ratio = ma5 / ma20 if ma20 > 0 else 1.0

        highest_20 = df["high"].iloc[-20:].max()
        price_position = df["close"].iloc[-1] / highest_20 if highest_20 > 0 else 1.0

        records.append({
            "symbol": sym,
            "name": df["name"].iloc[0] if "name" in df.columns else sym,
            "close": df["close"].iloc[-1],
            "momentum_20d": round(momentum_20, 4),
            "reversal_5d": round(-momentum_5, 4),
            "volatility_20d": round(volatility_20, 4),
            "volume_change": round(volume_change, 4),
            "ma_ratio": round(ma_ratio, 4),
            "price_position": round(price_position, 4),
        })

    return pd.DataFrame(records)

factors_df = calc_all_factors(klines)
print(f"因子表: {len(factors_df)} 只 × {len(factors_df.columns)} 列")
print(factors_df.head())

六个因子的含义：

因子名	计算方式	预期方向
`momentum_20d`	过去 20 天涨幅	越高越好（动量）
`reversal_5d`	过去 5 天涨幅取反	越高越好（短期跌多了反弹）
`volatility_20d`	20 日年化波动率	越低越好（低波动稳定）
`volume_change`	近 5 日成交额 / 近 20 日成交额	越高越好（量能放大）
`ma_ratio`	MA5 / MA20	> 1 偏强
`price_position`	当前价 / 20 日最高价	接近 1 说明在高位

5. 因子标准化

不同因子的数值范围差异很大——动量可能是 -0.2 到 0.5，波动率可能是 0.1 到 1.5。直接加权求和没有意义。

标准化的目的是把所有因子映射到同一个可比的尺度上。最常用的方法是横截面排序百分位：

def normalize_factors(df: pd.DataFrame, factor_cols: list) -> pd.DataFrame:
    """将因子值转换成横截面排序百分位（0~1）"""
    df = df.copy()
    for col in factor_cols:
        df[f"{col}_rank"] = df[col].rank(pct=True)
    return df

factor_cols = [
    "momentum_20d",
    "reversal_5d",
    "volatility_20d",
    "volume_change",
    "ma_ratio",
    "price_position",
]

factors_df = normalize_factors(factors_df, factor_cols)

rank_cols = [f"{c}_rank" for c in factor_cols]
print(factors_df[["symbol", "name"] + rank_cols].head(10))

排序百分位的好处是：不受异常值影响，不需要假设因子服从正态分布。排名第 1 的股票得分 1.0，排名最后的得分接近 0。

6. 因子方向对齐

有些因子是"越大越好"（动量），有些是"越小越好"（波动率）。在做加权求和之前，需要对齐方向：

factor_directions = {
    "momentum_20d": 1,
    "reversal_5d": 1,
    "volatility_20d": -1,
    "volume_change": 1,
    "ma_ratio": 1,
    "price_position": 1,
}

for col, direction in factor_directions.items():
    rank_col = f"{col}_rank"
    if direction == -1:
        factors_df[rank_col] = 1 - factors_df[rank_col]

现在所有 _rank 列都是"越高越好"。

7. 等权打分与综合排名

最简单的组合方式：所有因子等权平均。

rank_cols = [f"{c}_rank" for c in factor_cols]

factors_df["composite_score"] = factors_df[rank_cols].mean(axis=1)

factors_df = factors_df.sort_values("composite_score", ascending=False)

print("=== 多因子综合排名 Top 20 ===")
print(
    factors_df[["symbol", "name", "close", "composite_score"] + rank_cols]
    .head(20)
    .to_string(index=False)
)

print(f"\n=== 综合排名 Bottom 10 ===")
print(
    factors_df[["symbol", "name", "close", "composite_score"]]
    .tail(10)
    .to_string(index=False)
)

综合得分最高的 20 只就是你的候选池。你可以对比一下，和单因子选股（比如只看动量）选出来的结果有多大差异——通常多因子的结果更分散、更稳健。

8. 因子 IC 分析：这个因子有没有用

写了一堆因子，怎么知道哪些是真的有用的？最基础的检验方法是IC（Information Coefficient）——因子值和未来收益的相关性。

import numpy as np
import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

def calc_factor_ic(klines: dict, holding_days: int = 5) -> pd.DataFrame:
    """
    计算各因子与未来 N 天收益的 Rank IC
    """
    records = []

    for sym, df in klines.items():
        df = df.sort_values("trade_date").reset_index(drop=True)
        if len(df) < 40:
            continue

        df["ret"] = df["close"].pct_change()

        momentum_20 = df["close"].iloc[-1 - holding_days] / df["close"].iloc[-21 - holding_days] - 1
        volatility_20 = df["ret"].iloc[-21 - holding_days:-1 - holding_days].std()
        ma5 = df["close"].rolling(5).mean().iloc[-1 - holding_days]
        ma20 = df["close"].rolling(20).mean().iloc[-1 - holding_days]
        ma_ratio = ma5 / ma20 if ma20 > 0 else 1

        future_ret = df["close"].iloc[-1] / df["close"].iloc[-1 - holding_days] - 1

        records.append({
            "symbol": sym,
            "momentum_20d": momentum_20,
            "volatility_20d": volatility_20,
            "ma_ratio": ma_ratio,
            "future_ret": future_ret,
        })

    ic_df = pd.DataFrame(records).dropna()

    factor_names = ["momentum_20d", "volatility_20d", "ma_ratio"]
    ic_results = {}
    for f in factor_names:
        rank_ic = ic_df[f].corr(ic_df["future_ret"], method="spearman")
        ic_results[f] = round(rank_ic, 4)

    return ic_results

ic = calc_factor_ic(klines, holding_days=5)
print("=== Rank IC（因子与未来 5 天收益的秩相关）===")
for factor_name, ic_val in ic.items():
    strength = "强" if abs(ic_val) > 0.05 else ("中等" if abs(ic_val) > 0.03 else "弱")
    print(f"  {factor_name}: IC={ic_val:+.4f} ({strength})")

IC 的解读：

IC 绝对值	含义
> 0.05	有一定预测力，值得使用
0.03–0.05	有微弱信号，需要和其他因子配合
< 0.03	基本没有预测力

注意，IC 只是一个截面的度量。严谨的做法是计算IC 时间序列——每个月算一次 IC，然后看 IC 的均值和稳定性（IC 均值 > 0.03 且 IC_IR > 0.5 算不错了）。

9. 加权打分：给好因子更高权重

如果你通过 IC 分析发现动量因子的预测力最强、波动率因子较弱，可以按 IC 绝对值来分配权重：

import pandas as pd

ic_values = {
    "momentum_20d": 0.06,
    "reversal_5d": 0.03,
    "volatility_20d": -0.04,
    "volume_change": 0.02,
    "ma_ratio": 0.05,
    "price_position": 0.01,
}

total_abs_ic = sum(abs(v) for v in ic_values.values())
weights = {k: abs(v) / total_abs_ic for k, v in ic_values.items()}

print("=== 基于 IC 的因子权重 ===")
for name, w in weights.items():
    print(f"  {name}: {w:.2%}")

rank_cols = [f"{c}_rank" for c in ic_values.keys()]

factors_df["weighted_score"] = sum(
    factors_df[f"{name}_rank"] * w for name, w in weights.items()
)

factors_df = factors_df.sort_values("weighted_score", ascending=False)
print("\n=== IC 加权排名 Top 20 ===")
print(factors_df[["symbol", "name", "close", "weighted_score", "composite_score"]].head(20).to_string(index=False))

对比等权打分和 IC 加权打分的结果差异，你会发现加权后排名靠前的股票在预测力强的因子上得分更高。

10. 分层回测：因子真的有用吗

因子研究的终极检验是分层回测（Quintile Backtest）：把所有股票按因子值分成 5 组（或 10 组），看高分组是否比低分组的未来收益更高。

import pandas as pd
import numpy as np
from alphafeed import AlphaFeed

af = AlphaFeed()

symbols_list = factors_df["symbol"].tolist()[:300]

klines_long = af.klines.batch(
    symbols_list,
    period="1d",
    count=60,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

scored = factors_df[factors_df["symbol"].isin(klines_long.keys())].copy()

scored["quintile"] = pd.qcut(scored["composite_score"], q=5, labels=[1, 2, 3, 4, 5])

future_returns = {}
for sym, kdf in klines_long.items():
    kdf = kdf.sort_values("trade_date").reset_index(drop=True)
    if len(kdf) >= 10:
        ret_5d = kdf["close"].iloc[-1] / kdf["close"].iloc[-6] - 1
        future_returns[sym] = ret_5d

scored["future_5d_ret"] = scored["symbol"].map(future_returns)
scored = scored.dropna(subset=["future_5d_ret"])

quintile_perf = scored.groupby("quintile").agg(
    avg_return=("future_5d_ret", "mean"),
    median_return=("future_5d_ret", "median"),
    count=("symbol", "count"),
).round(4)

print("=== 分层回测结果（按综合因子分 5 组）===")
print("组 1 = 因子最差，组 5 = 因子最好")
print(quintile_perf)

spread = quintile_perf.loc[5, "avg_return"] - quintile_perf.loc[1, "avg_return"]
print(f"\n多空收益差（Q5 - Q1）: {spread:.4f} ({spread * 100:.2f}%)")

如果 Q5（因子最好的组）的收益显著高于 Q1（因子最差的组），说明因子有区分度。多空收益差就是因子的"超额收益来源"。

理想结果是各组收益单调递增：Q1 < Q2 < Q3 < Q4 < Q5。如果不是单调的，说明因子的线性预测能力有限。

11. 从研究到实盘：注意事项

问题	说明
数据延迟	因子用的是历史数据，实盘中你只能在收盘后计算完成后的第二天交易
调仓成本	多因子选股通常每月调一次仓，交易费用和冲击成本不可忽略
因子拥挤	如果很多人用相同的因子，超额收益会被套利掉
样本外检验	IC 和分层回测应该用"未来"数据验证，不能只在训练集上看
因子衰减	有些因子在过去有效，但最近两年已经失效了

多因子选股不是一个"配置好就不用管"的系统。它是一个持续迭代的研究流程：构建因子 → 检验 IC → 分层回测 → 组合优化 → 实盘观察 → 迭代因子。

12. 完整流程总结

全市场行情 (af.quotes, universes="CN_Stock")
    ↓
流动性过滤 (成交额 > 3 亿)
    ↓
批量拉取 K 线 (af.klines.batch)
    ↓
计算多个因子 (动量、波动率、换手、均线...)
    ↓
因子标准化 (横截面排序百分位)
    ↓
方向对齐 + 加权打分
    ↓
IC 检验 / 分层回测
    ↓
综合排名 → 候选池
    ↓
结合基本面/行业判断 → 最终决策

每一步都有代码，都可以独立修改。你可以随时增加新因子、调整权重、改变筛选条件，而不需要重写整个流程。