ARTICLE · 1154227
AI + 量化开发工程师学习笔记 Day 21:机器学习在量化投资中的位置——从真实 A 股数据构建 Quant ML Dataset
第二阶段：AI Quant Model（Day21～Day35）
今天，我们正式进入第二阶段。
前面 20 天，我们从零实现了一个 MiniGPT，学习了 Transformer、PyTorch、模型训练、验证、优化器、Checkpoint 和模型推理。
从今天开始，我们要把这些知识应用到你一直在研究的量化投资领域。
而且这一阶段，我希望我们不只是学习几个机器学习算法，而是真正开发一个可以接入你现有量化工程的机器学习选股模块。
考虑到我们已经有一套比较完整的量化系统，涉及股票池、因子、回测、Top-K、交易成本等，我们不会从最基础的量化知识重新讲起，而是重点研究：
如何让机器学习模型利用历史金融数据，学习股票的收益规律，并参与股票排序和组合构建。
一、今天的学习目标
预计学习时间：90～120 分钟（实践可以分两次完成）。
今天我们不急着训练 LightGBM。
因为量化机器学习中有一件事比选择模型更重要：
构建一份真实、可靠、没有未来信息泄漏的训练数据集。
如果数据本身有问题，再复杂的神经网络也没有意义。
第一部分：理论基础
二、机器学习在量化系统中到底处于什么位置？
我们先看传统量化选股。
假设你已经构建了四个因子：
PE：市盈率 PB：市净率 ROE：净资产收益率 Momentum：动量
传统多因子策略可能采用：
Score_i = w1 * PE_i + w2 * PB_i + w3 * ROE_i + w4 * Momentum_i
这里有一个前提：不同因子已经经过适当的方向调整、异常值处理和标准化。
例如：
Score_i = -0.2 * Z(PE_i) - 0.2 * Z(PB_i) + 0.3 * Z(ROE_i) + 0.3 * Z(Momentum_i)
然后按照 Score 排序：
股票池
↓
计算因子
↓
因子标准化
↓
人工设置权重
↓
综合评分
↓
Top-K
↓
组合构建
↓
回测
这种方法你应该已经非常熟悉。
传统方法的核心问题
因子权重从哪里来？
例如：
weights = {
"pe": -0.2,
"pb": -0.2,
"roe": 0.3,
"momentum": 0.3,
}
为什么 ROE 是 0.3，而不是 0.5？
为什么 Momentum 是 0.3，而不是 0.1？
这些权重可以来自经验、统计检验或优化算法，但如果我们希望模型从历史数据中学习因子与未来收益之间的关系，就可以引入机器学习。
三、机器学习如何改造传统量化？
机器学习模型不要求我们预先确定所有因子的组合权重。
我们提供：
历史因子
+
历史未来收益标签
模型学习：
Feature → Label
例如：
注意：这是用于解释的虚构数据。
机器学习尝试学习：
r̂_{i,t+20} = f(X_{i,t}; θ)
其中：
i：股票t：预测日期X_{i,t}：该日期可获得的因子θ：模型参数r̂：模型预测的未来收益
例如训练完成后：
股票A → 预测收益 5.2%
股票B → 预测收益 -3.1%
股票C → 预测收益 7.8%
股票D → 预测收益 1.6%
股票E → 预测收益 -2.4%
然后：
按照预测收益排序
↓
选择 Top-K
↓
构建组合
↓
回测
这里必须强调：
预测收益不等于真实收益。
机器学习只是利用历史统计关系建立预测函数。金融市场存在噪声、非平稳性和交易成本，模型完全可能在样本外失效。
四、传统因子模型与 ML 模型的区别
注意一个容易混淆的地方：
传统多因子不等于没有机器学习。
例如，用线性回归学习因子权重，本身就是一种机器学习方法。
真正的区别在于建模方式、函数复杂度以及训练方法，而不是简单地把两者对立起来。
五、量化机器学习最重要的三个概念
我们在 MiniGPT 中学习过：
X = Token Context
Y = Next Token
现在换成量化：
X = 股票因子
Y = 未来收益率
1. Feature：特征
假设某只股票在 2025 年 3 月 31 日：
features = {
"pe": 15.2,
"pb": 1.8,
"roe": 0.16,
"momentum_20": 0.08,
"volatility_20": 0.025,
}
这就是一个特征向量。
数学表示：
X_{i,t} = [PE, PB, ROE, Momentum, Volatility]^T
但所有特征都必须是在预测时点已经可获得的数据。
例如，3 月 31 日尚未披露的年报 ROE，不能提前使用。
2. Label：标签
我们希望预测未来 20 个交易日收益：
Y_{i,t} = P_{i,t+20} / P_{i,t} - 1
例如：
2025-03-31 收盘价：10 元
20 个交易日后：11 元
则：
Y = 11 / 10 - 1 = 10%
这是监督学习的目标。
但这里还有一个交易执行细节：
如果使用当天收盘后才能计算出的特征，就不能假设自己还能以当天已经过去的收盘价成交。
因此正式回测时，需要根据实际信号生成时间、下单时间、成交价格和可交易性定义标签。
今天先建立数据结构，Day23 再系统讨论这个问题。
3. Sample：样本
一条样本通常是：
(Stock, Date, X, Y)
例如：
000001
2025-03-31
PE=15.2
PB=1.8
ROE=0.16
Momentum=0.08
FutureReturn20D=0.10
这是一条训练样本。
如果：
500 只股票
×
500 个交易日
理论上可以形成：
500 × 500 = 250000
条股票-日期记录。
但它们不是 25 万个相互独立的样本，因为不同股票受共同市场因素影响，同一股票相邻日期的 20 日收益标签也会大量重叠。
这一点在后面的时间序列验证中非常重要。
六、股票池应该先确定，还是先训练模型？
这是我们之前专门讨论过的问题。
我依然建议：
先定义股票池规则，再构建训练数据，最后训练和回测。
但股票池必须是动态、可历史重建的。
例如：
2024-01-02
符合条件的股票集合
2024-01-03
符合条件的股票集合
2024-01-04
符合条件的股票集合
不能拿 2026 年仍然上市的股票，直接作为 2020 年的全部历史股票池。
否则可能出现：
Survivorship Bias（幸存者偏差）。
举个例子：
假设 2020 年有 1000 只候选股票，到 2026 年有 100 只已经退市。
如果你只研究今天还活着的 900 只，就可能高估历史策略表现。
因此我们最终要实现：
universe = get_universe(
trade_date="2024-01-02"
)
而不是简单地：
universe = get_current_stocks()
然后拿它回测过去十年。
我们第二阶段暂定的股票池规则
未来可以采用：
这些是待验证的研究规则，不是已经证明最优的股票池配置。
另外，停牌和涨跌停限制应当在交易执行层再次检查，不能只在股票池层处理。
七、为什么今天不直接下载 5000 只股票？
因为我们首先需要验证：
数据源
↓
数据格式
↓
日期
↓
复权
↓
字段类型
↓
缺失值
↓
数据保存
全部正确。
今天采用一个工程上更合理的策略：
先用 3 只股票建立完整的数据处理链，再扩展到几百只、几千只。
测试股票：
这三只股票只是用于测试数据获取与处理流程，不构成投资建议，也不是最终的研究股票池。
第二部分：Day21 动手实践
今天我们建立独立项目 quant_ml，暂时不修改你现有的 quant_strategy工程。
八、macOS 开发环境
在终端执行：
mkdir -p ~/AI/projects/quant_ml
cd ~/AI/projects/quant_ml
python3 --version
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install akshare pandas numpy pyarrow matplotlib pytest
打开 VSCode：
code .
在 VSCode 中通过 Python: Select Interpreter选择：
quant_ml/.venv/bin/python
今天使用 AkShare 的 stock_zh_a_hist接口。官方文档支持按股票代码、起止日期、日频和复权方式获取 A 股历史行情。
九、项目目录
quant_ml/
├── .venv/
├── data/
│ ├── raw/
│ └── processed/
├── src/
│ ├── __init__.py
│ ├── data_loader.py
│ └── features.py
├── main.py
└── requirements.txt
先创建目录：
mkdir -p data/raw data/processed src
touch src/__init__.py
十、获取真实 A 股行情
创建 src/data_loader.py：
import time
import akshare as ak
import pandas as pd
COLUMN_MAP = {
"日期": "trade_date",
"股票代码": "symbol",
"开盘": "open",
"收盘": "close",
"最高": "high",
"最低": "low",
"成交量": "volume",
"成交额": "amount",
"换手率": "turnover",
}
deffetch_stock(
symbol: str,
start_date: str,
end_date: str,
) -> pd.DataFrame:
"""
下载单只股票历史日线。
本课程暂用后复权数据研究价格变化。
正式历史回测还需处理复权因子的时间可得性。
"""
# 腾讯接口（新版 AkShare）：成交量为股，换手率为小数。
# 输出转换为旧工程单位：volume 为手，turnover 为百分数。
market = "sh"if symbol.startswith("6") else"sz"
df = ak.stock_zh_a_hist_tx(
symbol=market + symbol,
start_date=start_date,
end_date=end_date,
adjust="hfq",
timeout=20,
)
required = {"date", "open", "close", "high", "low", "volume", "amount", "turnover"}
missing = required - set(df.columns)
if missing:
raise ValueError(f"腾讯接口字段不完整：{missing}；请更新 AkShare")
df = df.rename(columns={"date": "trade_date"}).copy()
df["symbol"] = symbol
df["volume"] = pd.to_numeric(df["volume"], errors="raise") / 100
df["turnover"] = pd.to_numeric(df["turnover"], errors="raise") * 100
df["source"] = "tencent"
df["adjust"] = "hfq"
if df.empty:
raise ValueError(f"{symbol} 没有返回行情数据")
df = df.rename(columns=COLUMN_MAP)
columns = list(COLUMN_MAP.values())
df = df[columns].copy()
df["trade_date"] = pd.to_datetime(
df["trade_date"]
)
df["symbol"] = df["symbol"].astype(str).str.zfill(6)
numeric_columns = [
"open", "high", "low", "close",
"volume", "amount", "turnover",
]
for col in numeric_columns:
df[col] = pd.to_numeric(
df[col], errors="coerce"
)
df = (
df.sort_values("trade_date")
.drop_duplicates(["symbol", "trade_date"])
.reset_index(drop=True)
)
return df
deffetch_stocks(
symbols: list[str],
start_date: str,
end_date: str,
) -> pd.DataFrame:
frames = []
for symbol in symbols:
print(f"正在下载：{symbol}")
try:
df = fetch_stock(
symbol, start_date, end_date
)
frames.append(df)
print(f"成功：{len(df)} 行")
except Exception as exc:
print(f"下载失败：{symbol}，原因：{exc}")
time.sleep(1)
ifnot frames:
raise RuntimeError("所有股票下载失败")
return pd.concat(
frames, ignore_index=True
).sort_values(
["symbol", "trade_date"]
).reset_index(drop=True)
这里有三个工程细节：
第一，股票代码必须使用字符串，否则 000001会丢失前导零。
第二，下载失败不能静默忽略。正式训练前，必须检查是否缺失股票及历史区间。
第三，后复权价格适合用于某些历史收益研究，但当前重新计算的复权价格不一定等于历史某时点实际可获得的价格。我们暂时把它作为教学数据，正式无未来泄漏回测还要进一步完善。
十一、构建第一批特征
创建 src/features.py：
import numpy as np
import pandas as pd
defbuild_features(
df: pd.DataFrame,
) -> pd.DataFrame:
"""
根据历史行情构建特征。
所有 rolling 操作都按股票分别计算，
避免不同股票的数据相互混合。
"""
df = df.sort_values(
["symbol", "trade_date"]
).copy()
groups = df.groupby(
"symbol", group_keys=False
)
# 日收益率
df["return_1d"] = groups["close"].pct_change(
fill_method=None
)
# 过去 5 个交易日收益率
df["momentum_5"] = groups["close"].pct_change(
periods=5,
fill_method=None,
)
# 过去 20 个交易日收益率
df["momentum_20"] = groups["close"].pct_change(
periods=20,
fill_method=None,
)
# 20 日波动率
df["volatility_20"] = (
df.groupby("symbol")["return_1d"]
.transform(
lambda s: s.rolling(
20, min_periods=20
).std()
)
)
# 过去 20 日平均成交额
df["amount_ma20"] = (
df.groupby("symbol")["amount"]
.transform(
lambda s: s.rolling(
20, min_periods=20
).mean()
)
)
# 过去 20 日成交额相对变化
df["amount_ratio_20"] = (
df["amount"] / df["amount_ma20"]
)
df = df.replace(
[np.inf, -np.inf], np.nan
)
return df
今天暂时只构建行情类特征。
PE、PB、ROE 属于估值与财务特征，我们在 Day22 引入。
注意：这些特征包含当天收盘价、成交额等信息，因此只能在当天行情确认之后使用，不能假设盘中提前知道最终结果。
十二、编写主程序
创建 main.py：
from pathlib import Path
from src.data_loader import fetch_stocks
from src.features import build_features
RAW_DIR = Path("data/raw")
PROCESSED_DIR = Path("data/processed")
RAW_DIR.mkdir(parents=True, exist_ok=True)
PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
defmain():
symbols = [
"000001",
"600519",
"000858",
]
# 固定历史区间，保证实验可重复
df = fetch_stocks(
symbols=symbols,
start_date="20240101",
end_date="20251231",
)
# 保存原始行情
df.to_csv(
RAW_DIR / "stock_daily.csv",
index=False,
encoding="utf-8-sig",
)
df.to_parquet(
RAW_DIR / "stock_daily.parquet",
index=False,
)
print("\n原始数据：")
print(df.head())
# 检查重复记录
duplicates = df.duplicated(
["symbol", "trade_date"]
).sum()
print("\n重复记录：", duplicates)
# 检查缺失值
print("\n缺失值：")
print(df.isna().sum())
# 检查价格
invalid_price = (
(df["close"] <= 0) |
(df["high"] < df["low"])
).sum()
print("\n异常价格记录：", invalid_price)
# 构建特征
features = build_features(df)
feature_columns = [
"return_1d",
"momentum_5",
"momentum_20",
"volatility_20",
"amount_ratio_20",
]
# 仅保留特征齐全的记录
dataset = features.dropna(
subset=feature_columns
).copy()
# 保存特征数据
dataset.to_csv(
PROCESSED_DIR / "features.csv",
index=False,
encoding="utf-8-sig",
)
dataset.to_parquet(
PROCESSED_DIR / "features.parquet",
index=False,
)
print("\n特征数据：")
print(
dataset[
["symbol", "trade_date"]
+ feature_columns
].tail(10)
)
print("\n样本数量：", len(dataset))
print("\n各股票样本数量：")
print(dataset.groupby("symbol").size())
print("\nDay21 数据构建完成！")
if __name__ == "__main__":
main()
运行：
python main.py
成功后应该生成：
data/
├── raw/
│ ├── stock_daily.csv
│ └── stock_daily.parquet
└── processed/
├── features.csv
└── features.parquet
实际行数、价格和因子值取决于下载结果，我这里不虚构运行数据。
这份数据目前还不是可以直接训练未来收益预测模型的完整 Dataset，因为它尚未包含 Label。
Day23 会补上未来 20 日收益标签。
第三部分：理解今天写的特征
十三、Momentum 是什么？
今天计算：
df["momentum_20"]
公式：
Momentum_20(t) = P_t / P_{t-20} - 1
假设：
20 个交易日前：10 元
今天：11 元
那么：
Momentum_20 = 10%
这是一个典型的趋势类特征。
但注意，20 日动量为正，不代表未来 20 日一定上涨。
机器学习要研究的，恰恰是这种历史特征与未来收益之间是否存在稳定的统计关系。
十四、Volatility 是什么？
我们使用过去 20 个日收益率的样本标准差：
Volatility_20(t) = Std(r_{t-19}, ..., r_t)
它衡量近期收益波动程度。
例如：
股票A：每天波动较小
股票B：每天波动较大
即使两只股票过去 20 日累计收益相同，波动率也可能不同。
模型可以学习波动率是否具有额外预测价值。
这里得到的是日收益率标准差，还没有年化。
若采用一年 252 个交易日的近似假设：
AnnualizedVolatility = Volatility_daily * sqrt(252)
第四部分：今日作业与答案
十五、理论作业
题1：为什么不能直接用今天的全部上市股票回测 2015 年？
参考答案：
因为今天的股票集合可能不包含 2015 年存在、后来退市的公司，会产生幸存者偏差。应该根据历史时点重建可投资股票池。
题2：为什么 PE、PB、ROE 不能简单按财报报告期使用？
参考答案：
因为报告期不等于披露日期。例如 2024 年年报可能在 2025 年才披露，模型不能在 2024 年末提前使用尚未公布的数据。
题3：机器学习模型是不是可以完全替代人工设计因子？
参考答案：
不一定。模型可以学习特征之间的复杂关系，但仍然需要合理的数据来源、特征设计、标签设计、验证方法和交易约束。
题4：为什么随机划分训练集和测试集可能导致量化回测失真？
参考答案：
随机划分可能使较晚日期的数据进入训练集，而较早日期进入测试集，破坏真实时间顺序。此外，相邻日期的特征和未来收益标签可能高度重叠，导致信息泄漏和过于乐观的评估。
十六、编程作业
题1：计算过去 10 日收益率
参考答案：
df["momentum_10"] = (
df.groupby("symbol")["close"]
.pct_change(
periods=10,
fill_method=None,
)
)
题2：计算过去 20 日平均成交额
参考答案：
df["amount_ma20"] = (
df.groupby("symbol")["amount"]
.transform(
lambda s: s.rolling(
window=20,
min_periods=20,
).mean()
)
)
题3：筛选某个交易日成交额最大的 10 只股票
参考答案：
trade_date = "2025-06-30"
selected = df[
df["trade_date"] == trade_date
].copy()
top10 = selected.nlargest(
10, "amount"
)
print(
top10[
["symbol", "trade_date", "amount"]
]
)
这里有一个值得注意的问题：我们今天只下载 3 只股票，因此不可能得到 10 只股票的结果。
这道题的真正目的，是让你理解以后如何对同一交易日的整个股票池进行截面筛选。
第五部分：今天的工程验收
Day21 结束时，你应该确认：
.venv | |
还有一个额外检查：三只股票的交易日期是否一致。如果不一致，要进一步区分停牌、上市时间差异和数据缺失，不能直接把所有缺失日期当作正常情况。
今天最重要的一句话
量化机器学习的起点不是选择 LightGBM、CatBoost 还是神经网络，而是建立一份符合真实时间顺序、具有明确股票池规则、能够追溯数据来源的高质量训练数据集。
下一课：Day22——特征工程与因子体系
我们将正式引入 PE、PB、ROE、Momentum 等因子，研究估值因子、质量因子与技术因子的区别，并重点解决一个真实量化工程问题：
如何确保每个历史交易日使用的财务数据，在那个交易日确实已经公布？
这是从教学型量化模型走向严谨的实盘研究系统必须解决的问题。
Day21 小结
今天我们正式进入第二阶段 AI Quant Model：
理解机器学习在量化系统中的位置：从人工权重转向 Feature → Label 掌握 Feature、Label、Sample 三个核心概念 认识 Survivorship Bias：股票池必须可历史重建 搭建独立 quant_ml工程，用 AkShare 获取真实 A 股行情构建第一批行情特征：Momentum、Volatility、成交额比率 完成数据质量检查，并保存 CSV / Parquet
Day21 最重要的一句话：
量化机器学习的起点不是选择模型，而是建立一份真实、可靠、无未来泄漏的训练数据集。
下一课：Day22 — 特征工程与因子体系：PE、PB、ROE 与 Point-in-Time 财务数据