AI 投资实战：从因子、模型到回测的完整代码指南

原创

用户12339161

发布于 2026-09-22 15:12:37

1720

摘要：本文从专业量化投资视角出发，系统讲解 AI 投资落地的完整链路：数据、因子、标签、滚动训练、组合构建、回测、绩效评估与实盘化风险控制。文中包含一套可运行的 Python 示例代码，适合想进入 AI 量化投资领域的开发者。

AI 投资不是预测涨跌，而是构建决策系统
专业 AI 投资系统架构
数据层：股票池与行情数据
因子层：从价格量到横截面因子
标签层：未来收益与防未来函数
模型层：滚动训练与时序验证
组合层：从预测分数到持仓权重
回测层：成本、换手与绩效
进阶：均值方差与风险平价
实盘化清单
完整代码
总结

一、AI 投资不是预测涨跌，而是构建决策系统

很多人对 AI 投资的误解是：训练一个模型，预测明天涨还是跌，然后满仓买入。

专业视角下，AI 投资的核心不是“猜涨跌”，而是构建一个可重复、可验证、可风控的决策系统：

数据 -> 因子 -> 模型 -> 信号 -> 组合 -> 执行 -> 风控 -> 归因 -> 迭代

AI 在其中的角色是：

从高维数据中提取非线性关系；
对多因子进行动态加权；
预测横截面收益排序；
辅助组合优化与风险控制。

但 AI 无法消除市场不确定性。真正决定长期结果的，往往是：

数据质量；
因子逻辑；
交易成本；
风险暴露；
执行纪律；
合规边界。

二、专业 AI 投资系统架构

一个完整的 AI 量化系统通常包含七层：

层级	职责	关键问题
数据层	行情、财报、另类数据	准确性、复权、缺失值
因子层	动量、价值、质量、波动	逻辑、中性化、IC
模型层	预测未来收益	过拟合、时序验证
组合层	选股与权重	风险预算、换手约束
执行层	下单与成本	滑点、冲击成本、容量
风控层	暴露与回撤	行业、市值、风格中性
评估层	绩效与归因	夏普、回撤、ICIR

本文用 Python 实现一个简化但完整的版本：

yfinance 数据 -> 横截面因子 -> LightGBM 滚动训练 -> TopN 组合 -> 回测绩效

三、数据层：股票池与行情数据

先安装依赖：

pip install pandas numpy scikit-learn lightgbm yfinance scipy matplotlib

下载示例股票池数据：

import numpy as np
import pandas as pd
import yfinance as yf

TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "META",
           "NVDA", "JPM", "XOM", "UNH", "V"]
START = "2015-01-01"
END = "2024-12-31"

raw = yf.download(
    TICKERS,
    start=START,
    end=END,
    auto_adjust=True,
    progress=False
)

close = raw["Close"].copy()
volume = raw["Volume"].copy()

close = close.dropna(how="all")
volume = volume.reindex(close.index).fillna(0)

print(close.tail())

专业注意点：

使用复权价格，避免分红送股导致虚假跳空；
避免幸存者偏差，股票池应包含退市股票；
避免前视偏差，财报数据必须按公告日对齐；
数据频率要与调仓频率匹配。

四、因子层：从价格量到横截面因子

因子是 AI 投资的原料。常见因子包括：

动量：过去 20 日、60 日收益；
反转：过去 1 日、5 日收益；
波动：20 日收益率标准差；
均线偏离：价格 / 均线 - 1；
量能：成交量 / 20 日均量；
超买超卖：RSI。

代码实现：

def build_features(close, volume):
    ret = close.pct_change()
    feat = {}

    feat["ret_1"] = ret
    feat["ret_5"] = close.pct_change(5)
    feat["ret_20"] = close.pct_change(20)

    feat["mom_20"] = close / close.shift(20) - 1
    feat["mom_60"] = close / close.shift(60) - 1

    feat["vol_20"] = ret.rolling(20).std()

    feat["ma_ratio_20"] = close / close.rolling(20).mean() - 1
    feat["ma_ratio_60"] = close / close.rolling(60).mean() - 1

    feat["volume_ratio_20"] = volume / volume.rolling(20).mean()

    delta = close.diff()
    gain = delta.clip(lower=0).rolling(14).mean()
    loss = (-delta.clip(upper=0)).rolling(14).mean()
    rs = gain / (loss + 1e-12)
    feat["rsi_14"] = 100 - 100 / (1 + rs)

    return feat

横截面标准化：

def cs_winsorize_zscore(x):
    lower = x.quantile(0.01)
    upper = x.quantile(0.99)
    x = x.clip(lower, upper)
    return (x - x.mean()) / (x.std() + 1e-12)

专业注意点：

因子必须做横截面去极值和标准化；
市值、行业中性化能减少风格暴露；
因子相关性过高会导致组合集中；
因子有效性用 IC、RankIC、ICIR、分组单调性评估。

RankIC 计算：

def calc_rank_ic(pred_df):
    ic = pred_df.groupby(level="date").apply(
        lambda x: x["pred"].corr(x["label"], method="spearman")
    )
    return ic.dropna()

五、标签层：未来收益与防未来函数

标签设计决定模型目标。常见标签：

未来 1 日收益；
未来 5 日收益；
未来 20 日收益；
横截面超额收益；
风险调整后收益。

示例：预测未来 5 日横截面超额收益。

# t+1 收盘买入，持有到 t+6 收盘
fwd_ret_5 = close.shift(-6) / close.shift(-1) - 1

# 横截面去均值，预测超额收益
label = fwd_ret_5.sub(fwd_ret_5.mean(axis=1), axis=0)

# 回测用次日收益：t+1 收盘买入，t+2 收盘卖出
exec_ret_1 = close.shift(-2) / close.shift(-1) - 1

组装面板数据：

def build_dataset(close, volume):
    feat = build_features(close, volume)

    fwd_ret_5 = close.shift(-6) / close.shift(-1) - 1
    label = fwd_ret_5.sub(fwd_ret_5.mean(axis=1), axis=0)
    feat["label"] = label

    exec_ret_1 = close.shift(-2) / close.shift(-1) - 1
    feat["exec_ret_1"] = exec_ret_1

    df = pd.concat({k: v.stack() for k, v in feat.items()}, axis=1)
    df.index.names = ["date", "ticker"]
    df = df.replace([np.inf, -np.inf], np.nan).dropna()
    return df

关键原则：

特征只能使用 t 时刻及之前的数据；
标签可以使用未来数据，但只能用于训练；
回测收益必须与真实可执行时间对齐；
不要用全样本标准化，必须按日横截面处理。

六、模型层：滚动训练与时序验证

金融数据不能随机划分训练集和测试集。必须使用时序滚动训练：

2015-2017 训练 -> 2018 测试
2016-2018 训练 -> 2019 测试
...

代码：

import lightgbm as lgb

def preprocess(df, feature_cols):
    for col in feature_cols:
        df[col] = df.groupby(level="date")[col].transform(cs_winsorize_zscore)
    return df


def walk_forward_train(df, feature_cols, label_col="label",
                       start_year=2018, end_year=2024, train_years=3):
    df = df.sort_index()
    dates = df.index.get_level_values("date")
    preds = []
    models = {}

    for test_year in range(start_year, end_year + 1):
        train_start = pd.Timestamp(f"{test_year - train_years}-01-01")
        train_end = pd.Timestamp(f"{test_year - 1}-12-31")
        test_start = pd.Timestamp(f"{test_year}-01-01")
        test_end = pd.Timestamp(f"{test_year}-12-31")

        train_mask = (dates >= train_start) & (dates <= train_end)
        test_mask = (dates >= test_start) & (dates <= test_end)

        train = df[train_mask]
        test = df[test_mask]

        if len(train) < 1000 or len(test) < 100:
            continue

        model = lgb.LGBMRegressor(
            n_estimators=300,
            learning_rate=0.03,
            num_leaves=31,
            subsample=0.8,
            colsample_bytree=0.8,
            random_state=42,
            n_jobs=-1,
        )

        model.fit(train[feature_cols], train[label_col])

        tmp = test.copy()
        tmp["pred"] = model.predict(test[feature_cols])
        preds.append(tmp)
        models[test_year] = model

    return pd.concat(preds).sort_index(), models

专业注意点：

使用 Purged Walk-Forward，训练集和测试集之间留 embargo；
避免重叠标签导致信息泄漏；
模型越复杂，越需要正则化和样本外验证；
不要只看准确率，要看 IC、IR、分组收益和换手。

七、组合层：从预测分数到持仓权重

模型输出的是预测分数，不是持仓。组合层负责：

选股：TopN、分位数、多空；
加权：等权、市值加权、风险平价；
约束：单票上限、行业中性、换手控制；
成本：佣金、印花税、滑点、冲击成本。

简化 TopN 等权回测：

def backtest_topn(pred_df, top_n=3, cost_bps=10.0):
    pred_df = pred_df.sort_index()
    dates = pred_df.index.get_level_values("date").unique().sort_values()

    daily_ret = []
    used_dates = []
    prev_w = pd.Series(dtype=float)

    for dt in dates:
        day = pred_df.xs(dt, level="date")
        if len(day) < top_n:
            continue

        top = day["pred"].nlargest(top_n)
        w = pd.Series(1.0 / len(top), index=top.index)

        all_tickers = w.index.union(prev_w.index)
        turnover = (
            w.reindex(all_tickers, fill_value=0)
            - prev_w.reindex(all_tickers, fill_value=0)
        ).abs().sum()

        port_ret = (w * day.loc[w.index, "exec_ret_1"]).sum()
        cost = turnover * cost_bps / 10000.0

        daily_ret.append(port_ret - cost)
        used_dates.append(dt)
        prev_w = w

    return pd.Series(daily_ret, index=used_dates, name="strategy")

专业注意点：

成本假设不能太低，A 股单边综合成本常按 10-30bp 估算；
高换手策略必须考虑冲击成本；
组合约束比模型精度更重要；
风险暴露要定期归因。

八、回测层：成本、换手与绩效

绩效指标：

def performance(ret, freq=252):
    ret = ret.dropna()
    if len(ret) == 0:
        return {}

    nav = (1 + ret).cumprod()

    total_return = nav.iloc[-1] - 1
    annual_return = nav.iloc[-1] ** (freq / len(ret)) - 1
    annual_vol = ret.std() * np.sqrt(freq)
    sharpe = (ret.mean() * freq) / (annual_vol + 1e-12)

    dd = nav / nav.cummax() - 1
    max_dd = dd.min()
    calmar = annual_return / abs(max_dd) if max_dd != 0 else np.nan
    win_rate = (ret > 0).mean()

    return {
        "total_return": total_return,
        "annual_return": annual_return,
        "annual_vol": annual_vol,
        "sharpe": sharpe,
        "max_drawdown": max_dd,
        "calmar": calmar,
        "win_rate": win_rate,
        "days": len(ret),
    }

专业评估不只看收益，还要看：

夏普、卡玛、最大回撤；
换手率、容量、成本敏感度；
分年度、分市场状态表现；
因子暴露与归因；
样本外稳定性。

九、进阶：均值方差与风险平价

组合优化是 AI 投资的重要环节。

最大夏普：

from scipy.optimize import minimize

def max_sharpe_weights(mu, cov, risk_free=0.0, max_weight=0.2):
    n = len(mu)

    def neg_sharpe(w):
        ret = w @ mu
        vol = np.sqrt(w @ cov @ w) + 1e-12
        return -(ret - risk_free) / vol

    cons = [{"type": "eq", "fun": lambda w: np.sum(w) - 1}]
    bounds = [(0, max_weight)] * n

    res = minimize(
        neg_sharpe,
        np.ones(n) / n,
        bounds=bounds,
        constraints=cons
    )
    return res.x if res.success else np.ones(n) / n

风险平价：

def risk_parity_weights(cov):
    n = cov.shape[0]

    def obj(w):
        port_var = w @ cov @ w
        mrc = w * (cov @ w)
        target = port_var / n
        return np.sum((mrc - target) ** 2)

    cons = [{"type": "eq", "fun": lambda w: np.sum(w) - 1}]
    bounds = [(0, 1)] * n

    res = minimize(
        obj,
        np.ones(n) / n,
        bounds=bounds,
        constraints=cons
    )
    return res.x if res.success else np.ones(n) / n

专业注意点：

均值方差对输入敏感，需做收缩估计；
风险平价更稳健，但需估计协方差矩阵；
实盘要加入换手惩罚、行业约束、因子约束；
Black-Litterman 可融合主观观点与市场均衡。

十、实盘化清单

从回测到实盘，至少检查：

数据是否按公告日对齐；
是否有幸存者偏差；
是否有前视偏差；
是否计入交易成本与滑点；
是否考虑涨跌停与停牌；
是否有容量限制；
是否有风控与熔断；
是否有模型监控与衰减预警；
是否有合规审查；
是否保留完整审计日志。

AI 投资不是“模型越复杂越好”，而是：

数据质量 > 因子逻辑 > 组合约束 > 模型精度 > 执行成本

十一、完整代码

以下脚本整合了本文核心流程，可直接保存为 ai_investment_demo.py：

# ai_investment_demo.py
# 仅用于研究，不构成投资建议
import numpy as np
import pandas as pd
import yfinance as yf
import lightgbm as lgb
from scipy.optimize import minimize

pd.set_option("display.max_columns", 50)
pd.set_option("display.width", 200)

TICKERS = ["AAPL", "MSFT", "GOOGL", "AMZN", "META",
           "NVDA", "JPM", "XOM", "UNH", "V"]
START = "2015-01-01"
END = "2024-12-31"


def download_data(tickers, start, end):
    raw = yf.download(
        tickers,
        start=start,
        end=end,
        auto_adjust=True,
        progress=False
    )
    close = raw["Close"].copy()
    volume = raw["Volume"].copy()
    close = close.dropna(how="all")
    volume = volume.reindex(close.index).fillna(0)
    return close, volume


def build_features(close, volume):
    ret = close.pct_change()
    feat = {}

    feat["ret_1"] = ret
    feat["ret_5"] = close.pct_change(5)
    feat["ret_20"] = close.pct_change(20)

    feat["mom_20"] = close / close.shift(20) - 1
    feat["mom_60"] = close / close.shift(60) - 1

    feat["vol_20"] = ret.rolling(20).std()

    feat["ma_ratio_20"] = close / close.rolling(20).mean() - 1
    feat["ma_ratio_60"] = close / close.rolling(60).mean() - 1

    feat["volume_ratio_20"] = volume / volume.rolling(20).mean()

    delta = close.diff()
    gain = delta.clip(lower=0).rolling(14).mean()
    loss = (-delta.clip(upper=0)).rolling(14).mean()
    rs = gain / (loss + 1e-12)
    feat["rsi_14"] = 100 - 100 / (1 + rs)

    return feat


def build_dataset(close, volume):
    feat = build_features(close, volume)

    # 标签：t+1 收盘买入，持有到 t+6 收盘
    fwd_ret_5 = close.shift(-6) / close.shift(-1) - 1
    label = fwd_ret_5.sub(fwd_ret_5.mean(axis=1), axis=0)
    feat["label"] = label

    # 回测：t+1 收盘买入，t+2 收盘卖出
    exec_ret_1 = close.shift(-2) / close.shift(-1) - 1
    feat["exec_ret_1"] = exec_ret_1

    df = pd.concat({k: v.stack() for k, v in feat.items()}, axis=1)
    df.index.names = ["date", "ticker"]
    df = df.replace([np.inf, -np.inf], np.nan).dropna()
    return df


def cs_winsorize_zscore(x):
    lower = x.quantile(0.01)
    upper = x.quantile(0.99)
    x = x.clip(lower, upper)
    return (x - x.mean()) / (x.std() + 1e-12)


def preprocess(df, feature_cols):
    for col in feature_cols:
        df[col] = df.groupby(level="date")[col].transform(cs_winsorize_zscore)
    return df


def walk_forward_train(df, feature_cols, label_col="label",
                       start_year=2018, end_year=2024, train_years=3):
    df = df.sort_index()
    dates = df.index.get_level_values("date")
    preds = []
    models = {}

    for test_year in range(start_year, end_year + 1):
        train_start = pd.Timestamp(f"{test_year - train_years}-01-01")
        train_end = pd.Timestamp(f"{test_year - 1}-12-31")
        test_start = pd.Timestamp(f"{test_year}-01-01")
        test_end = pd.Timestamp(f"{test_year}-12-31")

        train_mask = (dates >= train_start) & (dates <= train_end)
        test_mask = (dates >= test_start) & (dates <= test_end)

        train = df[train_mask]
        test = df[test_mask]

        if len(train) < 1000 or len(test) < 100:
            continue

        model = lgb.LGBMRegressor(
            n_estimators=300,
            learning_rate=0.03,
            num_leaves=31,
            subsample=0.8,
            colsample_bytree=0.8,
            random_state=42,
            n_jobs=-1,
        )

        model.fit(train[feature_cols], train[label_col])

        tmp = test.copy()
        tmp["pred"] = model.predict(test[feature_cols])
        preds.append(tmp)
        models[test_year] = model

    if not preds:
        raise RuntimeError("没有生成任何预测，请检查数据区间和训练窗口")

    return pd.concat(preds).sort_index(), models


def calc_rank_ic(pred_df):
    ic = pred_df.groupby(level="date").apply(
        lambda x: x["pred"].corr(x["label"], method="spearman")
    )
    return ic.dropna()


def backtest_topn(pred_df, top_n=3, cost_bps=10.0):
    pred_df = pred_df.sort_index()
    dates = pred_df.index.get_level_values("date").unique().sort_values()

    daily_ret = []
    used_dates = []
    prev_w = pd.Series(dtype=float)

    for dt in dates:
        day = pred_df.xs(dt, level="date")
        if len(day) < top_n:
            continue

        top = day["pred"].nlargest(top_n)
        w = pd.Series(1.0 / len(top), index=top.index)

        all_tickers = w.index.union(prev_w.index)
        turnover = (
            w.reindex(all_tickers, fill_value=0)
            - prev_w.reindex(all_tickers, fill_value=0)
        ).abs().sum()

        port_ret = (w * day.loc[w.index, "exec_ret_1"]).sum()
        cost = turnover * cost_bps / 10000.0

        daily_ret.append(port_ret - cost)
        used_dates.append(dt)
        prev_w = w

    return pd.Series(daily_ret, index=used_dates, name="strategy")


def performance(ret, freq=252):
    ret = ret.dropna()
    if len(ret) == 0:
        return {}

    nav = (1 + ret).cumprod()
    total_return = nav.iloc[-1] - 1
    annual_return = nav.iloc[-1] ** (freq / len(ret)) - 1
    annual_vol = ret.std() * np.sqrt(freq)
    sharpe = (ret.mean() * freq) / (annual_vol + 1e-12)

    dd = nav / nav.cummax() - 1
    max_dd = dd.min()
    calmar = annual_return / abs(max_dd) if max_dd != 0 else np.nan
    win_rate = (ret > 0).mean()

    return {
        "total_return": total_return,
        "annual_return": annual_return,
        "annual_vol": annual_vol,
        "sharpe": sharpe,
        "max_drawdown": max_dd,
        "calmar": calmar,
        "win_rate": win_rate,
        "days": len(ret),
    }


def max_sharpe_weights(mu, cov, risk_free=0.0, max_weight=0.2):
    n = len(mu)

    def neg_sharpe(w):
        ret = w @ mu
        vol = np.sqrt(w @ cov @ w) + 1e-12
        return -(ret - risk_free) / vol

    cons = [{"type": "eq", "fun": lambda w: np.sum(w) - 1}]
    bounds = [(0, max_weight)] * n

    res = minimize(
        neg_sharpe,
        np.ones(n) / n,
        bounds=bounds,
        constraints=cons
    )
    return res.x if res.success else np.ones(n) / n


def risk_parity_weights(cov):
    n = cov.shape[0]

    def obj(w):
        port_var = w @ cov @ w
        mrc = w * (cov @ w)
        target = port_var / n
        return np.sum((mrc - target) ** 2)

    cons = [{"type": "eq", "fun": lambda w: np.sum(w) - 1}]
    bounds = [(0, 1)] * n

    res = minimize(
        obj,
        np.ones(n) / n,
        bounds=bounds,
        constraints=cons
    )
    return res.x if res.success else np.ones(n) / n


if __name__ == "__main__":
    close, volume = download_data(TICKERS, START, END)
    df = build_dataset(close, volume)

    feature_cols = [
        "ret_1", "ret_5", "ret_20",
        "mom_20", "mom_60",
        "vol_20",
        "ma_ratio_20", "ma_ratio_60",
        "volume_ratio_20", "rsi_14",
    ]

    df = preprocess(df, feature_cols)
    pred_df, models = walk_forward_train(df, feature_cols)

    ic = calc_rank_ic(pred_df)
    print("RankIC 均值：", ic.mean())
    print("RankIC IR：", ic.mean() / (ic.std() + 1e-12))

    ret = backtest_topn(pred_df, top_n=3, cost_bps=10.0)
    perf = performance(ret)

    print("回测绩效：")
    for k, v in perf.items():
        if isinstance(v, float):
            print(f"{k}: {v:.4f}")
        else:
            print(f"{k}: {v}")

    last_date = pred_df.index.get_level_values("date").max()
    last_day = pred_df.xs(last_date, level="date")
    top = last_day["pred"].nlargest(5)
    tickers = top.index.tolist()

    recent_ret = close[tickers].pct_change().dropna().tail(120)
    cov = recent_ret.cov().values * 252
    mu = top.values

    w_ms = max_sharpe_weights(mu, cov)
    w_rp = risk_parity_weights(cov)

    print("最大夏普权重：", dict(zip(tickers, np.round(w_ms, 4))))
    print("风险平价权重：", dict(zip(tickers, np.round(w_rp, 4))))

十二、总结

AI 投资的专业落地，不是“模型预测涨跌”，而是：

数据 -> 因子 -> 标签 -> 模型 -> 组合 -> 回测 -> 风控 -> 实盘

记住三个核心原则：

防未来函数：特征、标签、回测时间必须严格对齐；
控制成本：换手、滑点、冲击成本会吃掉大部分 alpha；
风控优先：组合约束和风险暴露比模型精度更重要。

代码只是起点。真正决定 AI 投资成败的，是数据质量、工程能力、风控体系和合规边界。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

媒体 AI