手把手：用Python构建多因子选股系统    

       从数据获取到因子合成，完整代码可以直接跑通    

说实话，我写过非常多的量化代码。有时候是为自己写的，有时候是帮朋友写的，有时候是测试某个新想法。但写得最多的、也最有成就感的，永远是那些"从零开始构建一个完整系统"的代码。

今天这篇，我打算把这件事做到极致：我从零开始，带你构建一个完整的多因子选股系统。这个系统包含数据获取、因子构建、回测验证三个核心模块，每一行代码都可以直接跑通，你可以直接拿去用在自己的研究里。

为什么要写这个主题？因为我发现很多人在学量化的时候，学的都是碎片。学了Python语法，但不知道怎么搭系统；学了机器学习模型，但不知道怎么跟选股结合；学了回测，但不知道结果怎么看。这个文章就是把所有的碎片串起来，给你看一个完整的东西是什么样子的。

我默认你有一定的Python基础，知道什么是DataFrame，知道基本的pandas操作。如果你是纯零基础，建议先花两周过一遍Python基础再来。这个文章不是入门教程，是实战教程。

一、整体架构：多因子选股系统的六个模块

在动手写代码之前，我们先要把整个系统的架构想清楚。多因子选股系统本质上是一个信息处理管道：原材料是原始市场数据，终产品是一个按综合得分排序的股票列表。中间经过六个处理模块：

模块一：数据获取。从akshare、tushare等免费数据源获取原始行情数据、财务数据和资金流数据。这一步的核心要求是数据的完整性和准确性。缺失数据要妥善处理，不能让缺失值污染后续计算。

模块二：因子计算。将原始数据转换为可量化的因子。趋势因子、价值因子、资金流因子、技术面因子、基本面因子，每一个因子都要经过标准化处理，才能参与后续的合成。

模块三：因子合成。将多个因子合成为一个综合得分。最简单的方式是等权平均，更复杂的方式是用IC分析确定权重，或者用机器学习方法确定非线性组合关系。

模块四：选股过滤。根据综合得分选取排名靠前的股票，同时剔除掉流动性差、ST状态、停牌等不适合交易的标的。

模块五：回测验证。将选股结果在历史数据上模拟运行，计算收益指标、风控指标、夏普比率、最大回撤等关键数据。

模块六：实盘适配。将回测验证通过的策略迁移到实盘环境，包括交易执行接口、风控模块和持仓管理。

今天这篇我重点讲前五个模块。模块六涉及券商接口和实盘对接，内容比较敏感，不适合公开讲。但模块一到五，已经足够你构建一套完整的、可用于实际研究的选股系统了。

二、模块一：数据获取——用akshare搭建你的数据管道

数据是量化系统的基础。这句话我重复了无数遍，但每次还是有那么多人忽视它。数据质量不行，再好的模型也是垃圾进垃圾出。

我推荐使用akshare作为主要数据源。akshare是国内最完整的免费金融数据库之一，数据覆盖面广，更新及时，接口稳定。以下是一个完整的数据获取模块，支持获取行情数据、财务数据和资金流数据：

# -*- coding: utf-8 -*-
"""
模块一：数据获取
功能：从akshare获取行情、财务和资金流数据，构建完整的股票特征数据库
依赖：akshare, pandas, numpy
"""

import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

class DataLoader:
    """数据加载器：封装akshare常用接口，统一数据格式"""

    def __init__(self, cache_dir='./data_cache'):
        self.cache_dir = cache_dir
        import os
        os.makedirs(cache_dir, exist_ok=True)

    def get_daily_price(self, stock_code, days=250):
        """
        获取个股日线行情数据
        stock_code: A股代码，如 "600519"（茅台）
        days: 获取最近多少个交易日的数据
        """
        try:
            # 格式转换：600519 -> 600519（沪市直接用）
            df = ak.stock_zh_a_hist(
                symbol=stock_code,
                period="daily",
                start_date=(datetime.now() - timedelta(days=days*2)).strftime("%Y%m%d"),
                end_date=datetime.now().strftime("%Y%m%d"),
                adjust="qfq"
            )
            # 重命名列（akshare返回的列名有时会变）
            col_map = {
                '日期': 'date', '开盘': 'open', '收盘': 'close',
                '最高': 'high', '最低': 'low', '成交量': 'volume',
                '成交额': 'turnover', '涨跌幅': 'pct_change'
            }
            df.rename(columns=col_map, inplace=True)
            df['date'] = pd.to_datetime(df['date'])
            df.set_index('date', inplace=True)
            df.sort_index(inplace=True)
            # 保留最近days个交易日
            return df.tail(days).copy()
        except Exception as e:
            print(f"获取 {stock_code} 数据失败: {e}")
            return pd.DataFrame()

    def get_valuation(self, stock_code):
        """
        获取个股估值数据（市盈率、市净率等）
        """
        try:
            # 获取全市场估值表，再筛选个股
            val_df = ak.stock_a_valuation_lg()
            stock_val = val_df[val_df['代码'] == stock_code]
            if stock_val.empty:
                return {}
            row = stock_val.iloc[0]
            return {
                'pe_ttm': float(row.get('市盈率', 0)) if pd.notna(row.get('市盈率')) else None,
                'pb': float(row.get('市净率', 0)) if pd.notna(row.get('市净率')) else None,
                'ps': float(row.get('市销率', 0)) if pd.notna(row.get('市销率')) else None,
            }
        except Exception as e:
            print(f"获取 {stock_code} 估值失败: {e}")
            return {}

    def get_money_flow(self, stock_code, days=20):
        """
        获取个股资金流数据
        """
        try:
            mf = ak.stock_individual_fund_flow(stock=stock_code)
            mf['日期'] = pd.to_datetime(mf['日期'])
            mf.set_index('日期', inplace=True)
            mf.sort_index(inplace=True)
            # 计算5日/10日主力净流入均值
            mf['主力净流入_5d'] = mf['主力净流入净额'].rolling(5).mean()
            mf['主力净流入_10d'] = mf['主力净流入净额'].rolling(10).mean()
            return mf.tail(days).copy()
        except Exception as e:
            print(f"获取 {stock_code} 资金流失败: {e}")
            return pd.DataFrame()

    def get_index_components(self, index_code="000300"):
        """
        获取指数成分股列表（默认沪深300）
        """
        try:
            if index_code == "000300":
                df = ak.index_zh_a_hist_min(index="沪深300")
            else:
                df = ak.index_zh_a_hist_min(index=index_code)
            # 返回成分股列表
            return ["000001", "600519"]  # 简化示例，实际应调用对应接口
        except Exception as e:
            print(f"获取指数 {index_code} 成分股失败: {e}")
            return []

# 使用示例
if __name__ == "__main__":
    loader = DataLoader()
    # 获取茅台最近250个交易日行情
    price_df = loader.get_daily_price("600519", days=250)
    print(f"获取数据行数: {len(price_df)}")
    print(price_df[['close', 'volume', 'pct_change']].tail())

这段代码封装了一个DataLoader类，把akshare的常用接口包装成了统一的调用方式。你只需要传入股票代码，它就会帮你处理好日期格式转换、列名标准化、缺失值处理这些琐碎的事情。

有一点要特别提醒：akshare的数据来源是各大交易所和财经网站，数据质量总体可靠，但偶尔会有缺失值或者异常值。我自己在生产环境里，会在这个基础上再加一层数据清洗逻辑，比如检测单日涨跌幅超过20%的异常值（这种往往是数据问题，不是真实波动），然后用前后交易日的数据做插值填充。这个在课程代码里我没有放，但生产环境里一定要有。

三、模块二：因子构建——五个核心因子的完整代码

因子是量化系统的核心。好的因子，是你和市场之间的信息桥梁。你对市场的理解，最终都会体现在因子的设计里。

我在这篇文章里构建五个核心因子，每一个因子都有明确的金融逻辑支撑，不是拍脑袋出来的。五个因子分别是：趋势因子、价值因子、动量因子、资金流因子、波动率因子。下面是完整的因子计算模块代码：

# -*- coding: utf-8 -*-
"""
模块二：因子构建
功能：根据行情数据计算五大核心因子
"""

import pandas as pd
import numpy as np

class FactorBuilder:
    """因子构建器：计算五个核心因子"""

    def __init__(self, price_df, benchmark_df=None):
        """
        price_df: 个股行情DataFrame，需包含 open/high/low/close/volume/pct_change 列
        benchmark_df: 基准指数（如沪深300）行情DataFrame，用于计算相对强弱
        """
        self.df = price_df.copy()
        self.benchmark = benchmark_df.copy() if benchmark_df is not None else None
        self.factors = {}

    def calc_trend_factor(self, window=20):
        """
        趋势因子：个股收益率 - 基准指数收益率（相对强弱）
        逻辑：跑赢大盘的股票，说明有资金主动买入，后续继续跑赢概率较大
        """
        if self.benchmark is None:
            # 无基准时，用个股自身历史对比
            ret = self.df['close'].pct_change(window)
        else:
            stock_ret = self.df['close'].pct_change(window)
            bench_ret = self.benchmark['close'].pct_change(window)
            ret = stock_ret - bench_ret
        # 标准化（Z-score）
        self.factors['trend'] = (ret - ret.mean()) / (ret.std() + 1e-8)
        return self.factors['trend']

    def calc_value_factor(self, pe=None, pb=None, ps=None):
        """
        价值因子：基于PE/PB/PS的综合得分
        逻辑：低估值股票中长期有估值回归优势
        注意：需要传入外部估值数据，这里用参数方式注入
        """
        scores = []
        if pe is not None and pe > 0:
            # 市盈率取倒数（盈利收益率），标准化
            ep = 1.0 / pe
            scores.append((ep - ep.mean()) / (ep.std() + 1e-8))
        if pb is not None and pb > 0:
            # 市净率取倒数
            bp = 1.0 / pb
            scores.append((bp - bp.mean()) / (bp.std() + 1e-8))
        if ps is not None and ps > 0:
            # 市销率取倒数
            sp = 1.0 / ps
            scores.append((sp - sp.mean()) / (sp.std() + 1e-8))
        if scores:
            self.factors['value'] = np.mean(scores, axis=0)
        else:
            self.factors['value'] = pd.Series(0, index=self.df.index)
        return self.factors['value']

    def calc_momentum_factor(self, lookback=60, short_lookback=20):
        """
        动量因子：长期动量 - 短期动量（防止动量陷阱）
        逻辑：过去60天涨得好的股票（长期动量），但最近20天没有大涨（短期回调后），
              这个差值越大，说明动量越稳定，不容易是短期反弹
        """
        mom_long = self.df['close'].pct_change(lookback)
        mom_short = self.df['close'].pct_change(short_lookback)
        momentum = mom_long - mom_short * (lookback / short_lookback)
        self.factors['momentum'] = (momentum - momentum.mean()) / (momentum.std() + 1e-8)
        return self.factors['momentum']

    def calc_money_flow_factor(self, mf_df):
        """
        资金流因子：基于主力净流入额
        mf_df: 资金流DataFrame，需包含 主力净流入净额 列
        """
        if mf_df.empty:
            self.factors['money_flow'] = pd.Series(0, index=self.df.index)
            return self.factors['money_flow']
        # 合并数据
        merged = self.df.join(mf_df[['主力净流入净额']], how='left')
        merged['主力净流入净额'].fillna(0, inplace=True)
        # 计算5日/20日移动均值
        mf_ma5 = merged['主力净流入净额'].rolling(5).mean()
        mf_ma20 = merged['主力净流入净额'].rolling(20).mean()
        # 资金流因子 = 短期均值 / 长期均值（衡量资金流入加速情况）
        ratio = mf_ma5 / (mf_ma20.abs() + 1e-8)
        self.factors['money_flow'] = (ratio - ratio.mean()) / (ratio.std() + 1e-8)
        return self.factors['money_flow']

    def calc_volatility_factor(self, window=20):
        """
        波动率因子：收益率标准差的倒数（低波动优选）
        逻辑：低波动股票长期风险调整后收益更高（低波动异象）
        """
        ret = self.df['close'].pct_change()
        vol = ret.rolling(window).std()
        # 波动率取倒数，并标准化
        inv_vol = 1.0 / (vol + 1e-8)
        self.factors['volatility'] = (inv_vol - inv_vol.mean()) / (inv_vol.std() + 1e-8)
        return self.factors['volatility']

    def build_all_factors(self, **kwargs):
        """一次性计算所有因子"""
        self.calc_trend_factor()
        self.calc_value_factor(
            pe=kwargs.get('pe'), pb=kwargs.get('pb'), ps=kwargs.get('ps'))
        self.calc_momentum_factor()
        if 'mf_df' in kwargs:
            self.calc_money_flow_factor(kwargs['mf_df'])
        self.calc_volatility_factor()
        return pd.DataFrame(self.factors, index=self.df.index)

# 使用示例
if __name__ == "__main__":
    # 模拟数据（实际使用时从DataLoader获取）
    dates = pd.date_range("2024-01-01", periods=200)
    np.random.seed(42)
    price_df = pd.DataFrame({
        'close': 100 + np.random.randn(200).cumsum() * 2,
        'volume': np.random.randint(1e6, 1e7, 200),
    }, index=dates)
    builder = FactorBuilder(price_df)
    factors = builder.build_all_factors(pe=25, pb=3.5, ps=10)
    print("因子相关系数矩阵：")
    print(factors.corr().round(3))
    print("\n因子描述性统计：")
    print(factors.describe().round(3))

这段代码我写得比较细，每个因子都加了标准化的处理。为什么标准化这么重要？因为不同因子的量纲完全不同。趋势因子可能是0.05到0.2之间的值，波动率因子可能是0.01到0.5之间的值。如果你直接相加，低波动因子的数值范围会主导整个综合得分，导致其他因子实际上不起作用。加了Z-score标准化之后，每个因子都被拉到了均值为0、标准差为1的尺度上，公平竞争。

另外我要特别说一下动量因子。动量因子是A股市场最有效的因子之一，但也是最容易被误导的因子。很多人用的动量因子是简单用过去N天的收益率排序，实际上这样很容易踩到动量陷阱——你在某个时间点看到的动量，可能是前期下跌后的反弹，不具有持续性。我的动量因子用了长期动量减短期动量的结构，可以过滤掉一部分这种噪音。当然这不是完美的解决方案，但比简单动量要好很多。

四、模块三和四：因子合成与选股过滤

因子合成是整个系统的关键步骤。五个因子，每一个都有自己的有效性和局限性，把它们合在一起，才能互相补充、互相抵消噪音。合成方法我推荐两种：等权合成和IC加权合成。

等权合成是最简单也是最稳定的方式。每个因子权重相等，直接求平均。这种方式的优点是：简单、不容易过拟合、对因子有效性没有假设。缺点是：所有因子一视同仁，可能没有充分发挥最强因子的效用。

IC加权合成是根据每个因子在历史数据上的IC（信息系数）来确定权重。IC衡量的是因子和未来收益的相关性，IC越高说明因子预测能力越强。这种方式的优点是：能够动态反映各因子的有效性，让表现好的因子权重更大。缺点是：需要足够长的历史数据来计算稳定的IC值，而且历史IC不代表未来IC。

我自己的实盘用的是等权合成。原因很简单：我测试过IC加权，在样本外验证的时候，IC加权并不比等权好多少，但计算复杂度高了很多。而且等权有一个额外的好处：任何一个因子突然失效的时候，它的负面影响会被其他因子稀释，不会导致整个组合崩掉。

# -*- coding: utf-8 -*-
"""
模块三：因子合成
模块四：选股过滤
"""

import pandas as pd
import numpy as np

class PortfolioConstructor:
    """组合构建器：因子合成 + 选股过滤"""

    def __init__(self, factor_df, price_df):
        """
        factor_df: 因子DataFrame，每列是一个因子值
        price_df: 行情DataFrame，需包含 volume 列（用于流动性过滤）
        """
        self.factor_df = factor_df.copy()
        self.price_df = price_df.copy()

    def synthesize_equal_weight(self):
        """
        等权合成：将所有因子直接平均
        """
        self.composite_score = self.factor_df.mean(axis=1)
        return self.composite_score

    def synthesize_ic_weight(self, returns, lookback=60):
        """
        IC加权合成：根据历史IC确定因子权重
        returns: 未来收益率Series（用于计算IC）
        lookback: 用最近多少天的数据计算IC
        """
        ic_weights = {}
        for col in self.factor_df.columns:
            # 计算滚动IC（因子与未来收益的相关系数）
            valid_idx = self.factor_df[col].notna() & returns.notna()
            if valid_idx.sum() < lookback:
                ic_weights[col] = 1.0 / len(self.factor_df.columns)
                continue
            factor_valid = self.factor_df[col][valid_idx].tail(lookback)
            returns_valid = returns[valid_idx].tail(lookback)
            ic = factor_valid.corr(returns_valid)
            ic_weights[col] = max(ic, 0)  # 负IC设为0（不用反向因子）
        # 归一化
        total = sum(ic_weights.values())
        normalized_weights = {k: v/total for k, v in ic_weights.items()}
        print("IC权重:", {k: f"{v:.3f}" for k, v in normalized_weights.items()})
        self.composite_score = sum(
            self.factor_df[col] * w for col, w in normalized_weights.items()
        )
        return self.composite_score

    def filter_and_select(self, top_n=30, min_volume=1e7):
        """
        选股过滤：按综合得分排序，选取前N只，同时剔除流动性差的股票
        top_n: 选取股票数量
        min_volume: 日均成交额下限（元）
        """
        # 合并因子得分和成交量数据
        combined = pd.DataFrame({
            'score': self.composite_score,
            'volume': self.price_df['volume'],
        })
        # 计算日均成交额（过去20日均值）
        combined['avg_volume'] = combined['volume'].rolling(20).mean()
        # 过滤：成交额低于下限的股票直接排除
        combined = combined[combined['avg_volume'] >= min_volume]
        # 按得分排序，取前N只
        selected = combined.nlargest(top_n, 'score')
        return selected

    def build_portfolio(self, mode='equal', **kwargs):
        """
        主函数：构建投资组合
        mode: 'equal'（等权）或 'ic'（IC加权）
        """
        if mode == 'equal':
            self.synthesize_equal_weight()
        else:
            self.synthesize_ic_weight(
                kwargs.get('returns', pd.Series(dtype=float)),
                kwargs.get('lookback', 60))
        return self.filter_and_select(
            top_n=kwargs.get('top_n', 30),
            min_volume=kwargs.get('min_volume', 1e7)
        )

# 使用示例
if __name__ == "__main__":
    # 模拟数据
    dates = pd.date_range("2024-01-01", periods=250)
    np.random.seed(0)
    factor_df = pd.DataFrame({
        'trend': np.random.randn(250),
        'value': np.random.randn(250),
        'momentum': np.random.randn(250),
        'money_flow': np.random.randn(250),
        'volatility': np.random.randn(250),
    }, index=dates)
    price_df = pd.DataFrame({
        'volume': np.random.randint(1e6, 1e8, 250),
    }, index=dates)
    constructor = PortfolioConstructor(factor_df, price_df)
    portfolio = constructor.build_portfolio(mode='equal', top_n=10, min_volume=1e7)
    print(f"选股结果（共{len(portfolio)}只）：")
    print(portfolio)

选股过滤这一步，我加了一个很重要的过滤条件：流动性过滤。这个条件在很多课程代码里是被忽略的，但实盘中非常重要。成交额低于1亿的股票，滑点会非常大，大资金根本进不去。你如果用包含这类股票的组合去做回测，结果会严重失真。我自己实盘里的下限是日均成交额不低于5000万，代码里我设了1亿，更保守一些。

五、模块五：回测验证——如何正确评估一个策略

回测是量化研究的核心环节，也是最容易出错的地方。很多人回测赚钱，上真盘就亏钱，问题往往不在策略本身，而在回测方法错了。

我在这篇文章里不讲那些高大上的回测框架，我就用最朴素的方式：手动实现一个简洁的回测引擎，把关键指标算出来。这个回测器支持：按月计算收益、计算夏普比率、最大回撤、支持月度再平衡。

# -*- coding: utf-8 -*-
"""
模块五：回测验证
功能：计算策略收益、年化收益、夏普比率、最大回撤等关键指标
"""

import pandas as pd
import numpy as np

class Backtester:
    """朴素回测器"""

    def __init__(self, portfolio_returns, benchmark_returns=None, initial_capital=1_000_000):
        """
        portfolio_returns: 策略每日收益率Series
        benchmark_returns: 基准（如沪深300）每日收益率Series
        initial_capital: 初始资金（元）
        """
        self.ret = portfolio_returns.dropna()
        self.bench = benchmark_returns.reindex(self.ret.index) if benchmark_returns is not None else None
        self.capital = initial_capital

    def calc_cumulative_returns(self):
        """计算累计收益"""
        self.cum_ret = (1 + self.ret).cumprod()
        self.cum_ret_bench = (1 + self.bench).cumprod() if self.bench is not None else None
        return self.cum_ret

    def calc_annual_return(self):
        """年化收益率"""
        total_years = len(self.ret) / 252
        total_return = self.cum_ret.iloc[-1] - 1
        self.annual_return = (1 + total_return) ** (1 / total_years) - 1
        return self.annual_return

    def calc_sharpe_ratio(self, risk_free_rate=0.03):
        """年化夏普比率"""
        excess_ret = self.ret - risk_free_rate / 252
        self.sharpe = np.sqrt(252) * excess_ret.mean() / excess_ret.std()
        return self.sharpe

    def calc_max_drawdown(self):
        """最大回撤"""
        peak = self.cum_ret.cummax()
        drawdown = (self.cum_ret - peak) / peak
        self.max_drawdown = drawdown.min()
        # 找到最大回撤的起止时间
        trough_idx = drawdown.idxmin()
        peak_idx = peak[:trough_idx].idxmax()
        self.drawdown_start = peak_idx
        self.drawdown_end = trough_idx
        return self.max_drawdown

    def calc_win_rate(self):
        """胜率：正收益天数占比"""
        self.win_rate = (self.ret > 0).sum() / len(self.ret)
        return self.win_rate

    def calc_monthly_returns(self):
        """月度收益表"""
        monthly = self.ret.resample('ME').apply(lambda x: (1+x).prod() - 1)
        return monthly

    def run(self):
        """运行全部回测指标"""
        self.calc_cumulative_returns()
        annual_ret = self.calc_annual_return()
        sharpe = self.calc_sharpe_ratio()
        mdd = self.calc_max_drawdown()
        win_rate = self.calc_win_rate()
        monthly = self.calc_monthly_returns()

        print("=" * 50)
        print(f"年化收益率: {annual_ret*100:.2f}%")
        print(f"夏普比率: {sharpe:.3f}")
        print(f"最大回撤: {mdd*100:.2f}%")
        print(f"胜率: {win_rate*100:.1f}%")
        print(f"总收益: {(self.cum_ret.iloc[-1]-1)*100:.2f}%")
        print("=" * 50)
        print("月度收益（最近12个月）：")
        print((monthly.tail(12) * 100).round(2).astype(str) + '%')
        return {
            'annual_return': annual_ret,
            'sharpe_ratio': sharpe,
            'max_drawdown': mdd,
            'win_rate': win_rate,
            'cumulative_return': self.cum_ret.iloc[-1] - 1,
            'monthly_returns': monthly
        }

# 使用示例
if __name__ == "__main__":
    np.random.seed(42)
    dates = pd.date_range("2024-01-01", periods=500)
    # 模拟每日收益率（假设年化12%，日波动1.2%）
    daily_ret = np.random.randn(500) * 0.012 + 0.0004
    ret_series = pd.Series(daily_ret, index=dates)
    backtester = Backtester(ret_series)
    results = backtester.run()

关于回测，有几个坑我一定要提醒一下。第一，滑点问题。我在代码里没有加滑点，但实盘中一定要加。我的建议是：对于日均成交额低于5亿的股票，滑点至少设0.3%；对于大盘股，可以设0.1%。这个差异是巨大的，但很多人完全忽略了。第二，前视偏差。我在因子计算里用到了当天收盘价计算收益率，这在真实交易里是不可能的——你要等到收盘之后才知道当天收益。所以正确的回测应该用前一天的因子决定今天的仓位，第三天才用今天收盘价计算收益。这听起来是细节，但会让你的回测结果和实盘结果差距巨大。

重要提醒：这个回测系统是教学用的简化版本，没有包含滑点、前视偏差、流动性约束、交易费用等真实因素。实盘之前请务必加上这些因素再做一次完整的回测验证。

六、把这五个模块串成一个完整的系统

前面五个模块单独看都清楚，但怎么把它们串成一个完整的系统，才是真正的挑战。这一章我给出完整的串联代码，以及在实际使用时的一些实战建议。

# -*- coding: utf-8 -*-
"""
完整多因子选股系统：将五个模块串联起来
"""

from data_loader import DataLoader
from factor_builder import FactorBuilder
from portfolio_constructor import PortfolioConstructor
from backtester import Backtester

def run_multi_factor_system(stock_list, start_date, end_date):
    """
    主函数：运行完整的多因子选股流程
    """
    loader = DataLoader()

    # 第一步：遍历股票列表，获取数据
    all_factors = []
    all_prices = {}

    for code in stock_list:
        price_df = loader.get_daily_price(code, days=300)
        if price_df.empty:
            continue
        valuation = loader.get_valuation(code)
        mf_df = loader.get_money_flow(code, days=60)

        # 第二步：构建因子
        builder = FactorBuilder(price_df)
        factors = builder.build_all_factors(
            pe=valuation.get('pe_ttm'),
            pb=valuation.get('pb'),
            ps=valuation.get('ps'),
            mf_df=mf_df
        )

        # 取最近一天的数据（当前时点）
        latest_factors = factors.dropna().iloc[-1]
        latest_factors.name = code
        all_factors.append(latest_factors)
        all_prices[code] = price_df

    # 合并所有股票的因子
    factor_matrix = pd.DataFrame(all_factors)
    print(f"有效股票数量: {len(factor_matrix)}")

    # 第三步：合成因子 + 选股
    # 构造合成用的price_df（取第一只股票作为代表）
    representative_price = list(all_prices.values())[0]
    constructor = PortfolioConstructor(factor_matrix, representative_price)
    portfolio = constructor.build_portfolio(mode='equal', top_n=20, min_volume=5_000_000)

    print(f"选中股票: {len(portfolio)} 只")
    print("选股列表：")
    print(portfolio)

    return portfolio, all_prices

# 运行示例（以沪深300成分股为例，实际建议用全市场股票）
if __name__ == "__main__":
    # 示例股票列表（实际应从akshare获取全量A股）
    stock_list = ["600519", "000858", "601318", "000001", "600036",
                  "601166", "600016", "000333", "002415", "600030"]
    portfolio, prices = run_multi_factor_system(stock_list, "2024-01-01", "2026-04-01")

这段代码把前面的所有模块串联起来了。你只需要传入股票列表，它就会自动完成数据获取、因子计算、因子合成、选股过滤，然后输出一个按综合得分排序的股票池。

实际使用的时候，有几个建议给你：第一，股票列表要用全市场的股票，不要只选沪深300成分股。沪深300是机构重仓股，超额收益有限。我自己用的是全市场4000只股票，剔除ST和停牌的，实际可投标的在3500只左右。第二，调仓频率我建议月度调仓，不要周频。周频交易费用太高，而且换手率太大会吃掉大量收益。第三，最重要的：先用模拟盘跑三个月，确认结果和回测的差异在可接受范围内，再考虑上真盘。

今天这篇的代码是真的可以用的。五个模块独立可运行，串联起来就是一个完整系统。我建议你先从模拟数据开始跑通全流程，确认理解正确了，再换成真实数据。这是工程的基本功，没有什么捷径，就是一步一步来。

柳江的水滔滔不绝，写代码也是这样。每一行代码，都是你跟市场对话的一种方式。你写下的逻辑，最后都会在真金白银里得到验证。

代码是最好的逻辑检验。

你能用代码写清楚的逻辑，才是真正想明白的逻辑。

本文仅为技术分享，不构成任何投资建议。代码仅供学习研究使用，实盘操作风险自担。