最近微软、亚马逊等大型科技股的短期快速上涨成了不少开发者群里讨论的热点。大家关注的不仅是行情本身还有一个更偏技术的问题AI 是不是已经在参与“利润轮动”交易所谓利润轮动简单理解就是资金在盈利预期改善的行业或个股之间切换推动板块轮流走强。当这种切换速度越来越快、数据维度越来越多时传统的人工盯盘和单一均线策略已经很难跟上节奏AI 量化分析的价值就体现出来了。作为技术博主我不打算分析短线买卖点也不做市场行情预测而是想从一个更工程化的角度切入如果我们要从零搭建一套“AI 利润轮动分析系统”用来复盘微软、亚马逊这类个股相对大盘的强弱变化整个流程到底应该怎么做数据从哪里来特征怎么构造模型怎么训练回测怎么避免“未来函数”这些问题本文会完整拆解。文章适合有 Python 基础、对量化分析或 AI 应用开发感兴趣的读者。看完后你将掌握一套可复现的股票轮动分析框架包含数据采集、特征工程、模型训练、回测评估和常见坑点排查。需要提前说明的是本文所有代码和数据仅用于技术学习与研究不构成任何投资建议。1. 背景与核心概念从“利润轮动”到 AI 量化分析1.1 什么是利润轮动“利润轮动”并不是一个严格的金融学术术语但在实战投资圈里很常见。它描述的是这样一种现象市场资金会根据不同行业或公司的盈利变化趋势周期性调整持仓方向。举个例子当云服务、AI 基础设施相关业务出现景气度回升时市场可能更愿意给微软、亚马逊这类科技公司更高的估值而一旦半导体、新能源等板块的盈利预期更强资金又会切换到那些方向。这种切换并不是同时发生的而是呈现出明显的先后顺序因而形成“轮动”。传统的利润轮动分析通常依赖宏观经济指标、行业景气度调查、财报数据、分析师预期修正等信息。这些信息有很明显的缺点更新频率低、数据分散、维度多人工很难同时跟踪几十个行业的盈利变化。1.2 AI 在轮动分析中的角色AI 模型擅长处理高维、非线性的数据关系这也是它在利润轮动分析中越来越被关注的原因。我们可以把 AI 的任务定义成一个有监督的分类问题在某个时间点给定一只股票过去一段时间的技术面、资金面、盈利质量等特征判断未来一段时间内它跑赢基准指数比如标普 500的概率有多大。这里要注意AI 并不是“预测明天涨还是跌”而是输出一个相对强弱概率。这个概率可以辅助你理解当前市场是否正在向科技股倾斜或者某只股票相对大盘是否已经出现过热迹象。1.3 本文的技术范围为了把问题聚焦在工程实现上本文选用了公开可获取的日线行情数据以微软MSFT和亚马逊AMZN为例构建一个完整的 AI 轮动分析系统。整个项目会覆盖行情数据采集与清洗技术指标、相对强度、盈利代理特征构造随机森林 / LightGBM 分类模型训练滚动时间序列预测简单回测与结果解读常见问题与工程化最佳实践。下面我们从系统设计开始。2. 一套 AI 轮动分析系统应该怎么设计2.1 总体流程无论模型多复杂一套完整的量化分析系统都可以拆成五个环节数据采集获取个股和基准指数的历史行情、成交数据必要时加入财务数据。特征工程把原始行情转换成模型可以理解的特征例如均线、动量、RSI、MACD、相对强度等。模型训练用历史数据训练一个分类器输出“未来某段时间跑赢基准”的概率。信号生成根据模型概率值生成持仓信号。回测评估用历史数据模拟策略表现对比基准收益。这个流程并不难理解难的是每一步都做得严谨。很多刚接触量化的开发者往往把大量精力花在调模型参数上却忽略了数据清洗和回测逻辑中的前视偏差最终得到一份“看起来很赚钱、实盘却完全失效”的无效回测。2.2 技术选型考虑到文章的可读性和复现成本我选择了 Python 生态中比较成熟的库环节推荐工具说明数据采集yfinance、AkShareyfinance 获取美股很方便AkShare 更适合国内市场数据处理pandas、numpy表格数据清洗、特征计算的核心工具机器学习scikit-learn、LightGBM随机森林、梯度提升树在表格数据上表现稳定回测自定义循环 / backtrader本文为演示清晰采用自定义简洁回测可视化matplotlib、plotly用于画出净值曲线和特征重要性如果你所在网络环境访问 Yahoo 接口不稳定可以考虑替换为本地数据文件或使用其他合规的数据服务。本文的代码结构并不依赖特定数据源只需保证 DataFrame 中包含 open、high、low、close、volume 等标准字段。2.3 项目目录设计建议在实际开发中采用下面这种目录结构ai-rotation/ ├── data/ │ ├── raw/ # 原始行情数据 │ └── processed/ # 清洗后的数据 ├── src/ │ ├── config.py # 全局配置 │ ├── data.py # 数据下载与清洗 │ ├── features.py # 特征工程 │ ├── model.py # 模型训练与预测 │ ├── backtest.py # 回测逻辑 │ └── main.py # 主流程入口 ├── notebooks/ # 探索性分析 ├── requirements.txt └── README.md这样做的好处是数据、特征、模型、回测互相解耦后续你想替换数据源、增加特征或改模型都不需要重写整个项目。3. 环境准备与依赖安装3.1 运行环境本项目的代码基于 Python 3.9 到 3.11 编写建议使用虚拟环境管理依赖避免和系统 Python 环境冲突。操作系统方面Windows、macOS、Linux 均可下面命令以 Windows 和 macOS 通用写法为例。创建虚拟环境python -m venv venv source venv/bin/activate # macOS / Linux # venv\Scripts\activate # Windows3.2 安装依赖在项目根目录创建requirements.txtyfinance0.2.40 pandas2.0.0 numpy1.24.0 scikit-learn1.2.0 lightgbm4.0.0 matplotlib3.7.0安装依赖pip install -r requirements.txt需要注意yfinance 属于第三方非官方数据接口它封装了公开网络请求接口结构可能会随上游页面变化而变化。如果你的 yfinance 版本比较新部分 API 行为可能和旧版本略有差异例如列名返回 MultiIndex本文后续会专门处理。3.3 全局配置文件在src/config.py中集中管理参数# src/config.py TICKERS [MSFT, AMZN] BENCHMARK SPY START_DATE 2018-01-01 END_DATE 2025-01-01 HORIZON 20 # 未来 20 个交易日 TRAIN_SIZE 504 # 滚动训练窗口约 2 年 TEST_SIZE 63 # 滚动验证窗口约 3 个月 PROB_THRESHOLD 0.6 # 信号触发阈值这里的HORIZON决定了模型预测的是未来 20 个交易日相对基准的胜率属于中期轮动信号而不是日内高频信号。4. 第一步获取微软与亚马逊的行情数据4.1 下载数据我们编写一个data.py负责下载并清洗行情数据。为了避免每次运行都重新下载可以把数据缓存到本地 CSV 文件。# src/data.py import os import pandas as pd import yfinance as yf from config import TICKERS, BENCHMARK, START_DATE, END_DATE def download_data(ticker, startSTART_DATE, endEND_DATE, save_pathNone): 下载单只股票或指数的日线数据。 如果 save_path 存在则优先从本地读取。 if save_path and os.path.exists(save_path): return pd.read_csv(save_path, index_col0, parse_datesTrue) df yf.download(ticker, startstart, endend, auto_adjustTrue, progressFalse) # 多股票下载时列名可能是 MultiIndex这里统一处理 if isinstance(df.columns, pd.MultiIndex): df.columns df.columns.get_level_values(0) df.columns [col.lower() for col in df.columns] df df[~df.index.duplicated(keeplast)] df df.sort_index() if save_path: os.makedirs(os.path.dirname(save_path), exist_okTrue) df.to_csv(save_path) return df def load_all_data(): 加载个股和基准指数数据 os.makedirs(data/raw, exist_okTrue) data {} for ticker in TICKERS [BENCHMARK]: data[ticker] download_data( ticker, save_pathfdata/raw/{ticker}.csv ) return data if __name__ __main__: raw_data load_all_data() for name, frame in raw_data.items(): print(name, frame.shape) print(frame.tail(3))运行后可以看到类似下面的输出MSFT (1762, 6) AMZN (1762, 6) SPY (1762, 6)每一行代表一个交易日的open、high、low、close、volume等字段。auto_adjustTrue表示自动复权这样计算收益时不会被除权除息影响。4.2 数据清洗的必要性股价数据经常出现缺失值、重复索引和停牌导致的空行。上面的代码已经做了三件事用duplicated删除重复索引用sort_index保证时间顺序用to_csv缓存原始数据方便离线调试。如果某些日期没有交易你可能需要在后续特征计算时统一用ffill填充但要注意前向填充后的数据不能再用于直接计算收益率否则会引入偏差。5. 第二步构造轮动分析特征5.1 技术面特征技术面特征是最容易构造的一类特征。它不包含基本面信息但能刻画股价的动量、趋势和波动状态。# src/features.py import pandas as pd import numpy as np def add_technical_features(df: pd.DataFrame) - pd.DataFrame: 添加常用技术指标特征。 data df.copy() # 多周期均线 for window in [5, 10, 20, 60]: data[fma_{window}] data[close].rolling(window).mean() data[fret_{window}] data[close].pct_change(window) # RSI 14 delta data[close].diff() gain delta.clip(lower0).rolling(14).mean() loss (-delta.clip(upper0)).rolling(14).mean() rs gain / (loss 1e-10) data[rsi_14] 100 - 100 / (1 rs) # MACD ema12 data[close].ewm(span12, adjustFalse).mean() ema26 data[close].ewm(span26, adjustFalse).mean() data[macd] ema12 - ema26 data[macd_signal] data[macd].ewm(span9, adjustFalse).mean() data[macd_diff] data[macd] - data[macd_signal] return data这里每个特征都有实际意义ma_20代表短期趋势ret_20代表过去一个月的动量rsi_14用于观察短期超买或超卖macd_diff反映动能是否加速。需要注意的是rolling在窗口初期会产生大量NaN后文会统一按dropna处理。5.2 相对强度特征轮动研究的核心不是单看一只股票涨了多少而是看它相对大盘强多少。我们构造一个相对强度指标def add_relative_strength(stock_df: pd.DataFrame, bench_df: pd.DataFrame, window20) - pd.DataFrame: 计算个股相对基准指数的超额收益率。 stock_ret stock_df[close].pct_change(window) bench_ret bench_df[close].pct_change(window) rs stock_ret - bench_ret stock_df stock_df.copy() stock_df[frs_{window}] rs return stock_df相对强度之所以重要是因为它直接刻画了“轮动”这个词的含义。如果微软过去 20 天跑赢标普 500 指数说明资金正在向这只股票倾斜反之则说明它在当前阶段相对弱势。5.3 盈利基本面特征严格意义上的“利润轮动”应该把公司的盈利质量考虑进来。由于日频行情数据里不直接包含财务数据这里提供一个通用处理思路把季度财务指标重采样到日频。假设你已经手动整理或通过数据接口拿到了某公司的季度盈利数据例如销售毛利率、净资产收益率ROE、净利润同比增速financials pd.DataFrame({ date: [2024-03-31, 2024-06-30, 2024-09-30, 2024-12-31], gross_margin: [0.68, 0.69, 0.70, 0.69], roe: [0.25, 0.27, 0.28, 0.26], net_profit_yoy: [0.12, 0.15, 0.18, 0.20] })我们可以用交易日序列做前向填充def resample_financials(fin_df: pd.DataFrame, trade_dates: pd.DatetimeIndex) - pd.DataFrame: fin_df fin_df.copy() fin_df[date] pd.to_datetime(fin_df[date]) fin_df fin_df.set_index(date) result fin_df.reindex(trade_dates, methodffill) return result这样做的好处是每个交易日都能拿到最近一期的盈利数据模型可以把盈利变化与价格动量放在同一时间维度上分析。但要注意财报数据存在延迟披露实际使用中需要额外处理“财报发布日期”避免用到未来数据。5.4 合并特征在features.py中提供一个总入口def build_features(stock_df: pd.DataFrame, bench_df: pd.DataFrame) - pd.DataFrame: df add_technical_features(stock_df) df add_relative_strength(df, bench_df, window20) df add_relative_strength(df, bench_df, window60) df df.dropna() return df到这里特征工程部分就完成了。特征并不是越多越好尤其是财务数据量不足时增加特征反而容易过拟合。后续在模型环节中可以通过特征重要性来筛选有效变量。6. 第三步构建 AI 预测模型6.1 定义预测目标我们采用的标签是未来HORIZON个交易日内个股收益率是否跑赢基准指数。def create_label(stock_df: pd.DataFrame, bench_df: pd.DataFrame, horizon20) - pd.Series: stock_future stock_df[close].shift(-horizon) / stock_df[close] - 1 bench_future bench_df[close].shift(-horizon) / bench_df[close] - 1 label (stock_future bench_future).astype(int) return label这里有一点特别容易出错如果用pct_change(horizon)计算出来的是历史收益而我们需要的是未来收益所以要先shift(-horizon)。如果 shift 方向弄反模型就会“偷看未来”导致回测结果严重失真。6.2 构建训练数据def create_dataset(stock_df, bench_df, horizon20): df build_features(stock_df, bench_df) label create_label(stock_df.loc[df.index], bench_df.loc[df.index], horizonhorizon) df[label] label df df.dropna(subset[label]) return df经过处理后df中既有特征列又有label列。标签为 1表示未来 20 天该股票跑赢基准标签为 0表示跑输基准。6.3 使用时间序列交叉验证股票数据不能像普通分类问题那样随机打乱因为时间顺序本身包含信息。这里采用滚动训练窗口# src/model.py import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier def rolling_predict(df: pd.DataFrame, feature_cols, train_size504, test_size63): 滚动训练与预测。 train_size: 训练集长度交易日 test_size: 验证集长度交易日 X df[feature_cols].values y df[label].values probs np.full(len(df), np.nan) model RandomForestClassifier(n_estimators150, random_state42, n_jobs-1) start 0 while start train_size test_size len(df): train_idx slice(start, start train_size) test_idx slice(start train_size, start train_size test_size) model.fit(X[train_idx], y[train_idx]) prob model.predict_proba(X[test_idx])[:, 1] probs[test_idx] prob start test_size df[pred_prob] probs df[signal] (df[pred_prob] 0.6).astype(int) return df这里通过循环实现了“传统训练集 后续验证集”的滚动预测。在验证集上模型没有见过未来数据因此回测结果相对可信。6.4 模型选型说明本文选择随机森林主要是因为它对异常值不敏感调参简单并且能输出概率。实际项目中你也可以换成 LightGBM。LightGBM 训练更快但参数更多需要更小心的交叉验证。# LightGBM 示例 import lightgbm as lgb model lgb.LGBMClassifier( n_estimators200, learning_rate0.05, num_leaves31, random_state42 )如果换成 LightGBM特征重要性分析和概率输出仍然适用只要把model.predict_proba保持一致即可。7. 第四步回测与结果解读7.1 简单回测逻辑回测的目的是评估“按照模型信号交易”是否能跑赢一直持有基准指数的策略。# src/backtest.py import pandas as pd def run_backtest(df: pd.DataFrame, stock_df: pd.DataFrame, bench_df: pd.DataFrame): 策略逻辑前一天收盘后产生 signal 第二天按 signal 持有股票或持有基准。 stock_ret stock_df[close].pct_change() bench_ret bench_df[close].pct_change() # 避免使用当天信号shfit 一天模拟次日执行 signal df[signal].shift(1).fillna(0) # 对齐索引 result pd.DataFrame({ stock_ret: stock_ret, bench_ret: bench_ret, signal: signal }).dropna() result[strategy_ret] result[signal] * result[stock_ret] (1 - result[signal]) * result[bench_ret] result[strategy_cum] (1 result[strategy_ret]).cumprod() result[bench_cum] (1 result[bench_ret]).cumprod() return result这段代码的核心逻辑是当模型认为个股未来跑赢基准时持有该股票否则持有基准指数。shift(1)是为了把信号向后移动一天防止用到当天的收盘信息。7.2 回测结果展示在main.py中整合所有步骤# src/main.py from data import load_all_data from features import build_features from model import create_dataset, rolling_predict from backtest import run_backtest def main(): data load_all_data() # 统一对齐索引 common_index data[MSFT].index.intersection(data[SPY].index) for ticker in [MSFT, AMZN]: stock_df data[ticker].loc[common_index] bench_df data[SPY].loc[common_index] dataset create_dataset(stock_df, bench_df, horizon20) feature_cols [col for col in dataset.columns if col not in [label]] result_df rolling_predict(dataset, feature_cols) bt run_backtest(result_df, stock_df, bench_df) print(f {ticker} 回测结果 ) print(f策略累计收益: {bt[strategy_cum].iloc[-1]:.2%}) print(f基准累计收益: {bt[bench_cum].iloc[-1]:.2%}) if __name__ __main__: main()运行后你会得到类似下面的输出 MSFT 回测结果 策略累计收益: 132.45% 基准累计收益: 85.20% AMZN 回测结果 策略累计收益: 118.30% 基准累计收益: 85.20%注意这只是一个演示结果。由于数据区间、特征和模型参数不同实际收益会有所变化甚至可能跑输基准。跑赢基准并不是必然的尤其是利润轮动策略在震荡市中很容易出现频繁错误信号。7.3 回测结果怎么看回测结果不能只看最终收益还需要关注回撤幅度最大回撤是多少是否能接受换手率信号频繁变化会产生高交易成本。样本外表现滚动预测中真正样本外的部分占比多少鲁棒性换个时间段结果是否仍然稳定在真实的工程项目中我建议把策略收益曲线、基准收益曲线、持仓信号画到一张图上直观观察哪些时间段模型跑赢、哪些时间段模型被反复打脸。8. 常见问题与排查思路在实际动手过程中你最可能遇到下面这些问题。问题现象常见原因解决思路yfinance 下载数据为空网络受限、代码或公司代码错误检查网络环境换用其他公开数据源多股票下载后列名是 MultiIndexyfinance 新版默认行为用get_level_values(0)展平列名特征列大量 NaNrolling 窗口初期数据不足用dropna()或限制起始日期标签列全部为 0 或 1股票长期跑赢或跑输基准更换基准或缩短预测周期回测收益异常高可能使用了未来函数重点检查shift(-horizon)和信号执行顺序模型准确率不到 55%市场本身包含大量随机性调整特征、模型、阈值并接受低准确率下面重点解释两个典型问题。8.1 未来函数问题未来函数是量化回测中最致命的错误。常见原因是构造标签时用了未来数据却没有把未来数据从特征中排除。比如你在计算label时用了未来 20 天的收益但特征中也包含未来 20 天的均线那模型实际上已经“看到了答案”。解决办法是在同一时间轴下严格对齐并习惯性地在特征计算完成后做一次dropna同时用shift调整信号执行时间。8.2 类别不平衡问题当某只股票长期跑赢基准时标签为 1 的样本会远多于标签为 0 的样本模型会倾向于预测 1。应对办法包括调整HORIZON预测周期使用类别权重class_weightbalanced改用连续收益回归任务再设置阈值。不过要注意类别不平衡本身并不一定是坏事。如果你的策略就是做多强势股模型偏向于预测 1 反而符合逻辑。关键是回测时要区分真实信号能力还是简单偏向性。9. 最佳实践与工程建议9.1 数据版本化管理原始数据是量化系统的基石。建议每日或每周把下载到的数据保存为带日期的文件例如data/raw/msft_20250101.csv data/raw/msft_20250108.csv这样当模型结果异常时你可以随时回溯到某一天的数据而不是每次重新下载结果还和上次不一致。9.2 配置与代码分离不要把日期、股票代码、阈值等参数写死在代码里。上面的config.py就是最小化实践。更复杂的项目可以使用.yaml或.env文件配合配置管理工具进行多环境隔离。9.3 防止过拟合AI 模型在金融数据上非常容易过拟合因为金融数据信噪比极低。建议遵循以下原则只用少量、可解释的特征不要堆砌几百个衍生指标用滚动时间序列验证而不是随机 K 折每次改动策略后只回测一次避免“调参调到历史最优”记录模型的参数和特征组合方便后续审计。9.4 回测与实盘差异回测表现好不代表实盘能赚钱。常见偏差包括交易成本、滑点、涨跌停限制、流动性不足等。如果要向实盘靠近至少要在回测中扣除双边交易手续费和滑点。9.5 安全与合规如果未来你想把策略接入实盘交易接口必须注意交易接口的密钥和令牌不能出现在代码仓库中使用最小权限原则交易权限与查询权限分离在测试环境完成充分验证后再考虑实盘量化交易需遵守所在地区的法律法规。这部分不属于纯技术问题但工程化项目里一旦忽略后果会非常严重。10. 总结与下一步学习方向本文以微软、亚马逊为案例围绕“AI 利润轮动分析”这个主题完整搭建了一个量化研究小系统。我们从数据下载开始完成了行情清洗、技术特征与相对强度特征构造、随机森林模型训练、滚动预测和简单回测最终得到了一条可以对比基准的策略收益曲线。如果你完整运行了代码你应该已经掌握如何用 yfinance 下载股票和基准指数数据如何构造动量、RSI、MACD、相对强度等特征如何设计“未来 20 天是否跑赢基准”的分类任务如何在时间序列上做滚动训练避免未来函数如何用简单回测框架验证模型信号。下一步你可以从三个方向继续深入。第一引入财务数据把“利润轮动”中的盈利因子真正落地例如净利润同比、毛利率变化和 ROE这是和纯技术面策略差异最大的地方。第二引入新闻情感分析和 AI Agent 自动抓取市场热点让模型感知消息面变化。第三把回测框架升级为支持交易成本、滑点和组合管理的专业回测工具。最后想说的是AI 交易并不是什么“稳赚公式”。它能帮我们处理大量数据、发现潜在规律但市场永远存在不确定性。对开发者来说把数据流程做规范、把回测逻辑做严谨比追求一次惊艳的收益曲线更重要。如果本文对你有帮助可以收藏备用也欢迎在动手实践后回来交流你的回测结果。 从零搭建AI利润轮动分析系统：数据、特征、模型与回测
最近微软、亚马逊等大型科技股的短期快速上涨成了不少开发者群里讨论的热点。大家关注的不仅是行情本身还有一个更偏技术的问题AI 是不是已经在参与“利润轮动”交易所谓利润轮动简单理解就是资金在盈利预期改善的行业或个股之间切换推动板块轮流走强。当这种切换速度越来越快、数据维度越来越多时传统的人工盯盘和单一均线策略已经很难跟上节奏AI 量化分析的价值就体现出来了。作为技术博主我不打算分析短线买卖点也不做市场行情预测而是想从一个更工程化的角度切入如果我们要从零搭建一套“AI 利润轮动分析系统”用来复盘微软、亚马逊这类个股相对大盘的强弱变化整个流程到底应该怎么做数据从哪里来特征怎么构造模型怎么训练回测怎么避免“未来函数”这些问题本文会完整拆解。文章适合有 Python 基础、对量化分析或 AI 应用开发感兴趣的读者。看完后你将掌握一套可复现的股票轮动分析框架包含数据采集、特征工程、模型训练、回测评估和常见坑点排查。需要提前说明的是本文所有代码和数据仅用于技术学习与研究不构成任何投资建议。1. 背景与核心概念从“利润轮动”到 AI 量化分析1.1 什么是利润轮动“利润轮动”并不是一个严格的金融学术术语但在实战投资圈里很常见。它描述的是这样一种现象市场资金会根据不同行业或公司的盈利变化趋势周期性调整持仓方向。举个例子当云服务、AI 基础设施相关业务出现景气度回升时市场可能更愿意给微软、亚马逊这类科技公司更高的估值而一旦半导体、新能源等板块的盈利预期更强资金又会切换到那些方向。这种切换并不是同时发生的而是呈现出明显的先后顺序因而形成“轮动”。传统的利润轮动分析通常依赖宏观经济指标、行业景气度调查、财报数据、分析师预期修正等信息。这些信息有很明显的缺点更新频率低、数据分散、维度多人工很难同时跟踪几十个行业的盈利变化。1.2 AI 在轮动分析中的角色AI 模型擅长处理高维、非线性的数据关系这也是它在利润轮动分析中越来越被关注的原因。我们可以把 AI 的任务定义成一个有监督的分类问题在某个时间点给定一只股票过去一段时间的技术面、资金面、盈利质量等特征判断未来一段时间内它跑赢基准指数比如标普 500的概率有多大。这里要注意AI 并不是“预测明天涨还是跌”而是输出一个相对强弱概率。这个概率可以辅助你理解当前市场是否正在向科技股倾斜或者某只股票相对大盘是否已经出现过热迹象。1.3 本文的技术范围为了把问题聚焦在工程实现上本文选用了公开可获取的日线行情数据以微软MSFT和亚马逊AMZN为例构建一个完整的 AI 轮动分析系统。整个项目会覆盖行情数据采集与清洗技术指标、相对强度、盈利代理特征构造随机森林 / LightGBM 分类模型训练滚动时间序列预测简单回测与结果解读常见问题与工程化最佳实践。下面我们从系统设计开始。2. 一套 AI 轮动分析系统应该怎么设计2.1 总体流程无论模型多复杂一套完整的量化分析系统都可以拆成五个环节数据采集获取个股和基准指数的历史行情、成交数据必要时加入财务数据。特征工程把原始行情转换成模型可以理解的特征例如均线、动量、RSI、MACD、相对强度等。模型训练用历史数据训练一个分类器输出“未来某段时间跑赢基准”的概率。信号生成根据模型概率值生成持仓信号。回测评估用历史数据模拟策略表现对比基准收益。这个流程并不难理解难的是每一步都做得严谨。很多刚接触量化的开发者往往把大量精力花在调模型参数上却忽略了数据清洗和回测逻辑中的前视偏差最终得到一份“看起来很赚钱、实盘却完全失效”的无效回测。2.2 技术选型考虑到文章的可读性和复现成本我选择了 Python 生态中比较成熟的库环节推荐工具说明数据采集yfinance、AkShareyfinance 获取美股很方便AkShare 更适合国内市场数据处理pandas、numpy表格数据清洗、特征计算的核心工具机器学习scikit-learn、LightGBM随机森林、梯度提升树在表格数据上表现稳定回测自定义循环 / backtrader本文为演示清晰采用自定义简洁回测可视化matplotlib、plotly用于画出净值曲线和特征重要性如果你所在网络环境访问 Yahoo 接口不稳定可以考虑替换为本地数据文件或使用其他合规的数据服务。本文的代码结构并不依赖特定数据源只需保证 DataFrame 中包含 open、high、low、close、volume 等标准字段。2.3 项目目录设计建议在实际开发中采用下面这种目录结构ai-rotation/ ├── data/ │ ├── raw/ # 原始行情数据 │ └── processed/ # 清洗后的数据 ├── src/ │ ├── config.py # 全局配置 │ ├── data.py # 数据下载与清洗 │ ├── features.py # 特征工程 │ ├── model.py # 模型训练与预测 │ ├── backtest.py # 回测逻辑 │ └── main.py # 主流程入口 ├── notebooks/ # 探索性分析 ├── requirements.txt └── README.md这样做的好处是数据、特征、模型、回测互相解耦后续你想替换数据源、增加特征或改模型都不需要重写整个项目。3. 环境准备与依赖安装3.1 运行环境本项目的代码基于 Python 3.9 到 3.11 编写建议使用虚拟环境管理依赖避免和系统 Python 环境冲突。操作系统方面Windows、macOS、Linux 均可下面命令以 Windows 和 macOS 通用写法为例。创建虚拟环境python -m venv venv source venv/bin/activate # macOS / Linux # venv\Scripts\activate # Windows3.2 安装依赖在项目根目录创建requirements.txtyfinance0.2.40 pandas2.0.0 numpy1.24.0 scikit-learn1.2.0 lightgbm4.0.0 matplotlib3.7.0安装依赖pip install -r requirements.txt需要注意yfinance 属于第三方非官方数据接口它封装了公开网络请求接口结构可能会随上游页面变化而变化。如果你的 yfinance 版本比较新部分 API 行为可能和旧版本略有差异例如列名返回 MultiIndex本文后续会专门处理。3.3 全局配置文件在src/config.py中集中管理参数# src/config.py TICKERS [MSFT, AMZN] BENCHMARK SPY START_DATE 2018-01-01 END_DATE 2025-01-01 HORIZON 20 # 未来 20 个交易日 TRAIN_SIZE 504 # 滚动训练窗口约 2 年 TEST_SIZE 63 # 滚动验证窗口约 3 个月 PROB_THRESHOLD 0.6 # 信号触发阈值这里的HORIZON决定了模型预测的是未来 20 个交易日相对基准的胜率属于中期轮动信号而不是日内高频信号。4. 第一步获取微软与亚马逊的行情数据4.1 下载数据我们编写一个data.py负责下载并清洗行情数据。为了避免每次运行都重新下载可以把数据缓存到本地 CSV 文件。# src/data.py import os import pandas as pd import yfinance as yf from config import TICKERS, BENCHMARK, START_DATE, END_DATE def download_data(ticker, startSTART_DATE, endEND_DATE, save_pathNone): 下载单只股票或指数的日线数据。 如果 save_path 存在则优先从本地读取。 if save_path and os.path.exists(save_path): return pd.read_csv(save_path, index_col0, parse_datesTrue) df yf.download(ticker, startstart, endend, auto_adjustTrue, progressFalse) # 多股票下载时列名可能是 MultiIndex这里统一处理 if isinstance(df.columns, pd.MultiIndex): df.columns df.columns.get_level_values(0) df.columns [col.lower() for col in df.columns] df df[~df.index.duplicated(keeplast)] df df.sort_index() if save_path: os.makedirs(os.path.dirname(save_path), exist_okTrue) df.to_csv(save_path) return df def load_all_data(): 加载个股和基准指数数据 os.makedirs(data/raw, exist_okTrue) data {} for ticker in TICKERS [BENCHMARK]: data[ticker] download_data( ticker, save_pathfdata/raw/{ticker}.csv ) return data if __name__ __main__: raw_data load_all_data() for name, frame in raw_data.items(): print(name, frame.shape) print(frame.tail(3))运行后可以看到类似下面的输出MSFT (1762, 6) AMZN (1762, 6) SPY (1762, 6)每一行代表一个交易日的open、high、low、close、volume等字段。auto_adjustTrue表示自动复权这样计算收益时不会被除权除息影响。4.2 数据清洗的必要性股价数据经常出现缺失值、重复索引和停牌导致的空行。上面的代码已经做了三件事用duplicated删除重复索引用sort_index保证时间顺序用to_csv缓存原始数据方便离线调试。如果某些日期没有交易你可能需要在后续特征计算时统一用ffill填充但要注意前向填充后的数据不能再用于直接计算收益率否则会引入偏差。5. 第二步构造轮动分析特征5.1 技术面特征技术面特征是最容易构造的一类特征。它不包含基本面信息但能刻画股价的动量、趋势和波动状态。# src/features.py import pandas as pd import numpy as np def add_technical_features(df: pd.DataFrame) - pd.DataFrame: 添加常用技术指标特征。 data df.copy() # 多周期均线 for window in [5, 10, 20, 60]: data[fma_{window}] data[close].rolling(window).mean() data[fret_{window}] data[close].pct_change(window) # RSI 14 delta data[close].diff() gain delta.clip(lower0).rolling(14).mean() loss (-delta.clip(upper0)).rolling(14).mean() rs gain / (loss 1e-10) data[rsi_14] 100 - 100 / (1 rs) # MACD ema12 data[close].ewm(span12, adjustFalse).mean() ema26 data[close].ewm(span26, adjustFalse).mean() data[macd] ema12 - ema26 data[macd_signal] data[macd].ewm(span9, adjustFalse).mean() data[macd_diff] data[macd] - data[macd_signal] return data这里每个特征都有实际意义ma_20代表短期趋势ret_20代表过去一个月的动量rsi_14用于观察短期超买或超卖macd_diff反映动能是否加速。需要注意的是rolling在窗口初期会产生大量NaN后文会统一按dropna处理。5.2 相对强度特征轮动研究的核心不是单看一只股票涨了多少而是看它相对大盘强多少。我们构造一个相对强度指标def add_relative_strength(stock_df: pd.DataFrame, bench_df: pd.DataFrame, window20) - pd.DataFrame: 计算个股相对基准指数的超额收益率。 stock_ret stock_df[close].pct_change(window) bench_ret bench_df[close].pct_change(window) rs stock_ret - bench_ret stock_df stock_df.copy() stock_df[frs_{window}] rs return stock_df相对强度之所以重要是因为它直接刻画了“轮动”这个词的含义。如果微软过去 20 天跑赢标普 500 指数说明资金正在向这只股票倾斜反之则说明它在当前阶段相对弱势。5.3 盈利基本面特征严格意义上的“利润轮动”应该把公司的盈利质量考虑进来。由于日频行情数据里不直接包含财务数据这里提供一个通用处理思路把季度财务指标重采样到日频。假设你已经手动整理或通过数据接口拿到了某公司的季度盈利数据例如销售毛利率、净资产收益率ROE、净利润同比增速financials pd.DataFrame({ date: [2024-03-31, 2024-06-30, 2024-09-30, 2024-12-31], gross_margin: [0.68, 0.69, 0.70, 0.69], roe: [0.25, 0.27, 0.28, 0.26], net_profit_yoy: [0.12, 0.15, 0.18, 0.20] })我们可以用交易日序列做前向填充def resample_financials(fin_df: pd.DataFrame, trade_dates: pd.DatetimeIndex) - pd.DataFrame: fin_df fin_df.copy() fin_df[date] pd.to_datetime(fin_df[date]) fin_df fin_df.set_index(date) result fin_df.reindex(trade_dates, methodffill) return result这样做的好处是每个交易日都能拿到最近一期的盈利数据模型可以把盈利变化与价格动量放在同一时间维度上分析。但要注意财报数据存在延迟披露实际使用中需要额外处理“财报发布日期”避免用到未来数据。5.4 合并特征在features.py中提供一个总入口def build_features(stock_df: pd.DataFrame, bench_df: pd.DataFrame) - pd.DataFrame: df add_technical_features(stock_df) df add_relative_strength(df, bench_df, window20) df add_relative_strength(df, bench_df, window60) df df.dropna() return df到这里特征工程部分就完成了。特征并不是越多越好尤其是财务数据量不足时增加特征反而容易过拟合。后续在模型环节中可以通过特征重要性来筛选有效变量。6. 第三步构建 AI 预测模型6.1 定义预测目标我们采用的标签是未来HORIZON个交易日内个股收益率是否跑赢基准指数。def create_label(stock_df: pd.DataFrame, bench_df: pd.DataFrame, horizon20) - pd.Series: stock_future stock_df[close].shift(-horizon) / stock_df[close] - 1 bench_future bench_df[close].shift(-horizon) / bench_df[close] - 1 label (stock_future bench_future).astype(int) return label这里有一点特别容易出错如果用pct_change(horizon)计算出来的是历史收益而我们需要的是未来收益所以要先shift(-horizon)。如果 shift 方向弄反模型就会“偷看未来”导致回测结果严重失真。6.2 构建训练数据def create_dataset(stock_df, bench_df, horizon20): df build_features(stock_df, bench_df) label create_label(stock_df.loc[df.index], bench_df.loc[df.index], horizonhorizon) df[label] label df df.dropna(subset[label]) return df经过处理后df中既有特征列又有label列。标签为 1表示未来 20 天该股票跑赢基准标签为 0表示跑输基准。6.3 使用时间序列交叉验证股票数据不能像普通分类问题那样随机打乱因为时间顺序本身包含信息。这里采用滚动训练窗口# src/model.py import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier def rolling_predict(df: pd.DataFrame, feature_cols, train_size504, test_size63): 滚动训练与预测。 train_size: 训练集长度交易日 test_size: 验证集长度交易日 X df[feature_cols].values y df[label].values probs np.full(len(df), np.nan) model RandomForestClassifier(n_estimators150, random_state42, n_jobs-1) start 0 while start train_size test_size len(df): train_idx slice(start, start train_size) test_idx slice(start train_size, start train_size test_size) model.fit(X[train_idx], y[train_idx]) prob model.predict_proba(X[test_idx])[:, 1] probs[test_idx] prob start test_size df[pred_prob] probs df[signal] (df[pred_prob] 0.6).astype(int) return df这里通过循环实现了“传统训练集 后续验证集”的滚动预测。在验证集上模型没有见过未来数据因此回测结果相对可信。6.4 模型选型说明本文选择随机森林主要是因为它对异常值不敏感调参简单并且能输出概率。实际项目中你也可以换成 LightGBM。LightGBM 训练更快但参数更多需要更小心的交叉验证。# LightGBM 示例 import lightgbm as lgb model lgb.LGBMClassifier( n_estimators200, learning_rate0.05, num_leaves31, random_state42 )如果换成 LightGBM特征重要性分析和概率输出仍然适用只要把model.predict_proba保持一致即可。7. 第四步回测与结果解读7.1 简单回测逻辑回测的目的是评估“按照模型信号交易”是否能跑赢一直持有基准指数的策略。# src/backtest.py import pandas as pd def run_backtest(df: pd.DataFrame, stock_df: pd.DataFrame, bench_df: pd.DataFrame): 策略逻辑前一天收盘后产生 signal 第二天按 signal 持有股票或持有基准。 stock_ret stock_df[close].pct_change() bench_ret bench_df[close].pct_change() # 避免使用当天信号shfit 一天模拟次日执行 signal df[signal].shift(1).fillna(0) # 对齐索引 result pd.DataFrame({ stock_ret: stock_ret, bench_ret: bench_ret, signal: signal }).dropna() result[strategy_ret] result[signal] * result[stock_ret] (1 - result[signal]) * result[bench_ret] result[strategy_cum] (1 result[strategy_ret]).cumprod() result[bench_cum] (1 result[bench_ret]).cumprod() return result这段代码的核心逻辑是当模型认为个股未来跑赢基准时持有该股票否则持有基准指数。shift(1)是为了把信号向后移动一天防止用到当天的收盘信息。7.2 回测结果展示在main.py中整合所有步骤# src/main.py from data import load_all_data from features import build_features from model import create_dataset, rolling_predict from backtest import run_backtest def main(): data load_all_data() # 统一对齐索引 common_index data[MSFT].index.intersection(data[SPY].index) for ticker in [MSFT, AMZN]: stock_df data[ticker].loc[common_index] bench_df data[SPY].loc[common_index] dataset create_dataset(stock_df, bench_df, horizon20) feature_cols [col for col in dataset.columns if col not in [label]] result_df rolling_predict(dataset, feature_cols) bt run_backtest(result_df, stock_df, bench_df) print(f {ticker} 回测结果 ) print(f策略累计收益: {bt[strategy_cum].iloc[-1]:.2%}) print(f基准累计收益: {bt[bench_cum].iloc[-1]:.2%}) if __name__ __main__: main()运行后你会得到类似下面的输出 MSFT 回测结果 策略累计收益: 132.45% 基准累计收益: 85.20% AMZN 回测结果 策略累计收益: 118.30% 基准累计收益: 85.20%注意这只是一个演示结果。由于数据区间、特征和模型参数不同实际收益会有所变化甚至可能跑输基准。跑赢基准并不是必然的尤其是利润轮动策略在震荡市中很容易出现频繁错误信号。7.3 回测结果怎么看回测结果不能只看最终收益还需要关注回撤幅度最大回撤是多少是否能接受换手率信号频繁变化会产生高交易成本。样本外表现滚动预测中真正样本外的部分占比多少鲁棒性换个时间段结果是否仍然稳定在真实的工程项目中我建议把策略收益曲线、基准收益曲线、持仓信号画到一张图上直观观察哪些时间段模型跑赢、哪些时间段模型被反复打脸。8. 常见问题与排查思路在实际动手过程中你最可能遇到下面这些问题。问题现象常见原因解决思路yfinance 下载数据为空网络受限、代码或公司代码错误检查网络环境换用其他公开数据源多股票下载后列名是 MultiIndexyfinance 新版默认行为用get_level_values(0)展平列名特征列大量 NaNrolling 窗口初期数据不足用dropna()或限制起始日期标签列全部为 0 或 1股票长期跑赢或跑输基准更换基准或缩短预测周期回测收益异常高可能使用了未来函数重点检查shift(-horizon)和信号执行顺序模型准确率不到 55%市场本身包含大量随机性调整特征、模型、阈值并接受低准确率下面重点解释两个典型问题。8.1 未来函数问题未来函数是量化回测中最致命的错误。常见原因是构造标签时用了未来数据却没有把未来数据从特征中排除。比如你在计算label时用了未来 20 天的收益但特征中也包含未来 20 天的均线那模型实际上已经“看到了答案”。解决办法是在同一时间轴下严格对齐并习惯性地在特征计算完成后做一次dropna同时用shift调整信号执行时间。8.2 类别不平衡问题当某只股票长期跑赢基准时标签为 1 的样本会远多于标签为 0 的样本模型会倾向于预测 1。应对办法包括调整HORIZON预测周期使用类别权重class_weightbalanced改用连续收益回归任务再设置阈值。不过要注意类别不平衡本身并不一定是坏事。如果你的策略就是做多强势股模型偏向于预测 1 反而符合逻辑。关键是回测时要区分真实信号能力还是简单偏向性。9. 最佳实践与工程建议9.1 数据版本化管理原始数据是量化系统的基石。建议每日或每周把下载到的数据保存为带日期的文件例如data/raw/msft_20250101.csv data/raw/msft_20250108.csv这样当模型结果异常时你可以随时回溯到某一天的数据而不是每次重新下载结果还和上次不一致。9.2 配置与代码分离不要把日期、股票代码、阈值等参数写死在代码里。上面的config.py就是最小化实践。更复杂的项目可以使用.yaml或.env文件配合配置管理工具进行多环境隔离。9.3 防止过拟合AI 模型在金融数据上非常容易过拟合因为金融数据信噪比极低。建议遵循以下原则只用少量、可解释的特征不要堆砌几百个衍生指标用滚动时间序列验证而不是随机 K 折每次改动策略后只回测一次避免“调参调到历史最优”记录模型的参数和特征组合方便后续审计。9.4 回测与实盘差异回测表现好不代表实盘能赚钱。常见偏差包括交易成本、滑点、涨跌停限制、流动性不足等。如果要向实盘靠近至少要在回测中扣除双边交易手续费和滑点。9.5 安全与合规如果未来你想把策略接入实盘交易接口必须注意交易接口的密钥和令牌不能出现在代码仓库中使用最小权限原则交易权限与查询权限分离在测试环境完成充分验证后再考虑实盘量化交易需遵守所在地区的法律法规。这部分不属于纯技术问题但工程化项目里一旦忽略后果会非常严重。10. 总结与下一步学习方向本文以微软、亚马逊为案例围绕“AI 利润轮动分析”这个主题完整搭建了一个量化研究小系统。我们从数据下载开始完成了行情清洗、技术特征与相对强度特征构造、随机森林模型训练、滚动预测和简单回测最终得到了一条可以对比基准的策略收益曲线。如果你完整运行了代码你应该已经掌握如何用 yfinance 下载股票和基准指数数据如何构造动量、RSI、MACD、相对强度等特征如何设计“未来 20 天是否跑赢基准”的分类任务如何在时间序列上做滚动训练避免未来函数如何用简单回测框架验证模型信号。下一步你可以从三个方向继续深入。第一引入财务数据把“利润轮动”中的盈利因子真正落地例如净利润同比、毛利率变化和 ROE这是和纯技术面策略差异最大的地方。第二引入新闻情感分析和 AI Agent 自动抓取市场热点让模型感知消息面变化。第三把回测框架升级为支持交易成本、滑点和组合管理的专业回测工具。最后想说的是AI 交易并不是什么“稳赚公式”。它能帮我们处理大量数据、发现潜在规律但市场永远存在不确定性。对开发者来说把数据流程做规范、把回测逻辑做严谨比追求一次惊艳的收益曲线更重要。如果本文对你有帮助可以收藏备用也欢迎在动手实践后回来交流你的回测结果。