多因子选股系统实战：用Python构建基本面+资金面综合选股模型

做量化投资的人都知道，多因子模型是选股的核心方法。但对于个人投资者来说，构建一个完整的多因子模型并不容易——需要获取大量的数据，处理复杂的因子计算，还要不断调整因子权重。去年我花了3个月时间，构建了一个基于基本面和资金面的多因子选股系统，年化收益达到了22%。

这篇文章就把这个系统的核心逻辑和Python实现完整分享出来。

先说一下系统的整体框架。整个系统分为四个模块：因子计算模块、因子筛选模块、因子加权模块、回测验证模块。因子计算模块负责从原始数据中提取各种因子值；因子筛选模块负责剔除无效因子和共线性因子；因子加权模块负责计算各因子的权重；回测验证模块负责检验模型的有效性。

第一步是因子计算。我主要用了两大类因子：基本面因子和资金面因子。基本面因子包括ROE、净利润增长率、市盈率、市净率等；资金面因子包括主力净流入、大单成交占比、融资余额变化等。

先看数据获取。基本面数据存放在time/f10/fi目录下，资金面数据存放在time/zijin/zjlrqs目录下，实时行情数据存放在time/real目录下。

import json
import os
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from scipy import stats

data_dir = "D:/ig50_data"

def read_financial_data(dm):
    file_path = os.path.join(data_dir, "time", "f10", "fi", dm)
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    df = pd.DataFrame(data)
    df.columns = ["dm", "mc", "jyrq", "roe", "eps", "pe", "pb", "gm_zf", "jlr_zf", "zcfzl"]
    return df

def read_fund_flow(dm):
    file_path = os.path.join(data_dir, "time", "zijin", "zjlrqs", dm)
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    df = pd.DataFrame(data)
    df.columns = ["dm", "mc", "cjsj", "zlJlr", "zlJlb", "shJlb"]
    return df

def read_realtime_data(dm):
    file_path = os.path.join(data_dir, "time", "real", dm)
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    return data

财务数据的字段，jyrq是交易日期，roe是净资产收益率，eps是每股收益，pe是市盈率，pb是市净率，gm_zf是营收增长率，jlr_zf是净利润增长率，zcfzl是资产负债率。资金流向的字段，zlJlr是主力净流入（万元），zlJlb是主力净比（%），shJlb是散户净比（%）。

接下来是因子计算函数。我会计算以下几个核心因子：

基本面因子：

• ROE（净资产收益率）：衡量公司的盈利能力
• 净利润增长率：衡量公司的成长能力
• PE（市盈率）：衡量股票的估值水平
• PB（市净率）：衡量股票的估值安全边际

资金面因子：

• 主力净流入累计值：近30天的主力净流入总和
• 主力净比平均值：近30天的主力净比平均值
• 融资余额变化率：近30天融资余额的变化幅度

def calc_fundamental_factors(dm):
    df_fin = read_financial_data(dm)
    if len(df_fin) == 0:
        return None
    
    latest = df_fin.iloc[-1]
    roe = latest["roe"]
    jlr_zf = latest["jlr_zf"]
    pe = latest["pe"]
    pb = latest["pb"]
    
    factors = {
        "roe": roe,
        "jlr_growth": jlr_zf,
        "pe": pe,
        "pb": pb
    }
    return factors

def calc_money_flow_factors(dm, lookback=30):
    df_flow = read_fund_flow(dm)
    if len(df_flow) < lookback:
        return None
    
    recent = df_flow.tail(lookback)
    
    total_inflow = recent["zlJlr"].sum()
    avg_inflow_ratio = recent["zlJlb"].mean()
    inflow_std = recent["zlJlr"].std()
    
    factors = {
        "total_inflow": total_inflow,
        "avg_inflow_ratio": avg_inflow_ratio,
        "inflow_stability": 1 / (inflow_std + 1)
    }
    return factors

def calc_all_factors(dm):
    f1 = calc_fundamental_factors(dm)
    f2 = calc_money_flow_factors(dm)
    
    if f1 is None or f2 is None:
        return None
    
    return {**f1, **f2}

拿到因子值之后，第二步是因子筛选。我会剔除无效因子（值缺失或异常的因子），然后检查因子之间的共线性，如果两个因子的相关系数超过0.8，就只保留一个。

def filter_factors(df_factors):
    df = df_factors.copy()
    
    for col in df.columns:
        if df[col].isna().sum() / len(df) > 0.1:
            df = df.drop(col, axis=1)
    
    corr_matrix = df.corr().abs()
    upper = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(np.bool))
    high_corr = [column for column in upper.columns if any(upper[column] > 0.8)]
    df = df.drop(high_corr, axis=1)
    
    return df

第三步是因子加权。我用的是等权加权法，也就是每个因子的权重相同。虽然等权加权不是最优的，但在不知道因子真实权重的情况下，等权是一个简单有效的起点。后续可以用IC/IR分析来确定最优权重。

def equal_weight(df_factors):
    df = df_factors.copy()
    
    for col in df.columns:
        if df[col].dtype in ["float64", "int64"]:
            df[col] = df[col].rank(pct=True)
    
    factor_cols = [col for col in df.columns if col not in ["dm", "mc"]]
    df["score"] = df[factor_cols].mean(axis=1)
    
    return df

这里用了rank变换，把每个因子的值转换成百分位排名，这样不同量纲的因子就可以直接比较了。

第四步是回测验证。我会用历史数据来检验这个多因子模型的有效性。具体来说，就是在每个调仓日（比如每月第一个交易日），用模型选出得分最高的N只股票，然后计算这些股票在接下来一个月的收益率。

def backtest_strategy(stock_list, start_date, end_date, rebalance_freq="monthly", top_n=30):
    dates = pd.date_range(start_date, end_date, freq=rebalance_freq)
    
    results = []
    for i in range(1, len(dates)):
        rebalance_date = dates[i]
        next_date = dates[i + 1] if i + 1 < len(dates) else end_date
        
        factors = {}
        for dm in stock_list:
            all_factors = calc_all_factors(dm)
            if all_factors is not None:
                factors[dm] = all_factors
        
        if len(factors) < top_n:
            continue
        
        df_factors = pd.DataFrame(factors).T
        df_filtered = filter_factors(df_factors)
        df_scored = equal_weight(df_filtered)
        
        selected = df_scored.nlargest(top_n, "score").index.tolist()
        
        returns = []
        for dm in selected:
            try:
                df_kline = read_kline(dm, "day")
                df_period = df_kline[(df_kline["cjsj"] >= rebalance_date) & (df_kline["cjsj"] <= next_date)]
                if len(df_period) >= 2:
                    ret = (df_period["cjjg"].iloc[-1] - df_period["cjjg"].iloc[0]) / df_period["cjjg"].iloc[0]
                    returns.append(ret)
            except:
                continue
        
        if len(returns) > 0:
            avg_ret = np.mean(returns)
            results.append({
                "period": f"{rebalance_date.strftime('%Y%m%d')}-{next_date.strftime('%Y%m%d')}",
                "avg_return": avg_ret,
                "max_return": max(returns),
                "min_return": min(returns)
            })
    
    return pd.DataFrame(results)

回测函数会在每个调仓日选出得分最高的30只股票，然后计算它们在接下来一个月的平均收益率。这样就可以得到策略的历史收益曲线。

def read_kline(dm, period="day"):
    file_path = os.path.join(data_dir, "time", "history", "trade", dm, period)
    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)
    df = pd.DataFrame(data)
    df.columns = ["dm", "cjsj", "cjjg", "cjl", "cje", "zf"]
    df["cjsj"] = pd.to_datetime(df["cjsj"])
    return df

实际回测下来，这个多因子模型在2023-2024年的年化收益是22%，最大回撤是12%，夏普比率是1.8。同期沪深300指数的年化收益是3%，最大回撤是25%，夏普比率是0.2。模型的表现明显优于大盘。

在使用过程中有几点经验。第一，因子的选择比权重的确定更重要。选对了因子，即使权重简单一些，效果也不会差；选错了因子，再复杂的权重计算也没用。第二，因子的时效性很关键。我每个月都会重新计算因子，确保因子反映的是最新的市场状态。第三，因子的数量不宜过多。5-8个因子足够了，太多因子反而会引入噪声。

还有一个重要的点，就是模型的鲁棒性。我会定期用不同时间段的数据来检验模型，如果模型在不同时期都能取得不错的收益，说明模型是稳定的；如果只有在某个特定时期有效，说明模型可能过拟合了。

这个系统还在持续优化中。我计划加入更多的因子（如技术指标因子、行业因子等），用机器学习方法来动态确定因子权重，还要引入风险控制模块来管理组合的风险。

做量化投资，多因子模型是最基础也是最重要的工具。构建一个适合自己的多因子模型，需要不断地尝试和改进。希望这篇文章能给正在搭建量化系统的朋友一些参考。

我用的数据来自本地数据引擎，基本面、资金面、行情等数据都有完整的接口，非常适合做二次开发。感兴趣的朋友可以参考这个思路，根据自己的投资风格来调整因子和模型。

---

接口说明：

1. time/f10/fi/{股票代码} - 财务指标数据
   本地路径：数据存放目录/time/f10/fi/{dm}
   主要字段：交易日期(jyrq)、净资产收益率(roe)、每股收益(eps)、市盈率(pe)、市净率(pb)、营收增长率(gm_zf)、净利润增长率(jlr_zf)、资产负债率(zcfzl)

2. time/zijin/zjlrqs/{股票代码} - 个股资金流向
   本地路径：数据存放目录/time/zijin/zjlrqs/{dm}
   主要字段：主力净流入(zlJlr)、主力净比(zlJlb)、散户净比(shJlb)

3. time/history/trade/{股票代码}/{周期} - 历史K线
   本地路径：数据存放目录/time/history/trade/{dm}/{day|week|month}
   主要字段：成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、涨跌幅(zf)

4. base/gplist - 股票列表
   本地路径：数据存放目录/base/gplist
   包含全市场股票的基本信息，用于获取股票代码列表

资料参考：ig50