2.3 特征工程：量化交易的“炼金术”¶

这一节讲什么？

从 OHLCV 原始数据到能喂给机器学习模型的特征矩阵，这一节是关键的"炼金术"。核心原则：用收益率而非价格（平稳性），用滚动标准化而非全局缩放（防止未来泄露）。

学习目标¶

• 从 OHLCV 原始数据构建有价值的预测特征
• 严谨性：掌握平稳性检验（ADF Test）处理非平稳价格序列
• 防御性：理解滚动标准化（Rolling Scaling）防止未来泄露
• 构建滞后特征与未来标签，避免 Look-ahead Bias

In [1]:

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import yfinance as yf
import seaborn as sns
from statsmodels.tsa.stattools import adfuller

# 下载数据
raw = yf.download('AAPL', start='2020-01-01', end='2024-01-01', progress=False)
df = pd.DataFrame({
    'Open': raw['Open'].squeeze(),
    'High': raw['High'].squeeze(),
    'Low': raw['Low'].squeeze(),
    'Close': raw['Close'].squeeze(),
    'Volume': raw['Volume'].squeeze(),
})
print(f'原始数据准备就绪: {df.shape}')

import numpy as np import pandas as pd import matplotlib.pyplot as plt import yfinance as yf import seaborn as sns from statsmodels.tsa.stattools import adfuller # 下载数据 raw = yf.download('AAPL', start='2020-01-01', end='2024-01-01', progress=False) df = pd.DataFrame({ 'Open': raw['Open'].squeeze(), 'High': raw['High'].squeeze(), 'Low': raw['Low'].squeeze(), 'Close': raw['Close'].squeeze(), 'Volume': raw['Volume'].squeeze(), }) print(f'原始数据准备就绪: {df.shape}')

原始数据准备就绪: (1006, 5)

In [2]:

import matplotlib.pyplot as plt

# 1. 设置系统自带的中文字体（这里使用黑体 SimHei）
plt.rcParams['font.sans-serif'] = ['SimHei']  # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei']

# 2. 解决更换字体后，负号（-）显示为方块的问题
plt.rcParams['axes.unicode_minus'] = False

import matplotlib.pyplot as plt # 1. 设置系统自带的中文字体（这里使用黑体 SimHei） plt.rcParams['font.sans-serif'] = ['SimHei'] # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei'] # 2. 解决更换字体后，负号（-）显示为方块的问题 plt.rcParams['axes.unicode_minus'] = False

1. 为什么“价格”不能直接做特征？ (平稳性原则)¶

QuantEcon 核心观点： 大多数量化模型要求输入数据是平稳的 (Stationary)。平稳意味着数据的均值和方差不会随着时间发生巨大漂移。

价格是不平稳的（今天 100 元，明年可能 200 元），直接用价格训练模型会导致“伪回归”。我们需要使用收益率或价格差分。

In [3]:

# ADF 检验：判别数据是否平稳 (p-value < 0.05 则平稳)
def check_stationarity(series, name):
    result = adfuller(series.dropna())
    print(f"{name} ADF p-value: {result[1]:.5f} {'(平稳 ✅)' if result[1] < 0.05 else '(不平稳 ❌)'}")

check_stationarity(df['Close'], "收盘价")
check_stationarity(df['Close'].pct_change(), "日收益率")

print("结论：在特征工程中，请尽量使用收益率或振幅，而不是原始价格。")

# ADF 检验：判别数据是否平稳 (p-value < 0.05 则平稳) def check_stationarity(series, name): result = adfuller(series.dropna()) print(f"{name} ADF p-value: {result[1]:.5f} {'(平稳 ✅)' if result[1] < 0.05 else '(不平稳 ❌)'}") check_stationarity(df['Close'], "收盘价") check_stationarity(df['Close'].pct_change(), "日收益率") print("结论：在特征工程中，请尽量使用收益率或振幅，而不是原始价格。")

收盘价 ADF p-value: 0.52759 (不平稳 ❌)
日收益率 ADF p-value: 0.00000 (平稳 ✅)
结论：在特征工程中，请尽量使用收益率或振幅，而不是原始价格。

2. 构造基础特征矩阵¶

In [4]:

# 收益率特征
df['ret_1d'] = df['Close'].pct_change()
df['ret_5d'] = df['Close'].pct_change(5)

# 波动与动量
df['vol_20d'] = df['ret_1d'].rolling(20).std()
df['ma_ratio'] = df['Close'] / df['Close'].rolling(20).mean() - 1

# 价格形态特征
df['upper_shadow'] = (df['High'] - np.maximum(df['Close'], df['Open'])) / df['Close']
df['lower_shadow'] = (np.minimum(df['Close'], df['Open']) - df['Low']) / df['Close']

print("特征构建完成")

# 收益率特征 df['ret_1d'] = df['Close'].pct_change() df['ret_5d'] = df['Close'].pct_change(5) # 波动与动量 df['vol_20d'] = df['ret_1d'].rolling(20).std() df['ma_ratio'] = df['Close'] / df['Close'].rolling(20).mean() - 1 # 价格形态特征 df['upper_shadow'] = (df['High'] - np.maximum(df['Close'], df['Open'])) / df['Close'] df['lower_shadow'] = (np.minimum(df['Close'], df['Open']) - df['Low']) / df['Close'] print("特征构建完成")

特征构建完成

3. 防御性特征工程：滚动标准化¶

陷阱：全局标准化¶

很多教程教你用 StandardScaler().fit(X) 处理整个数据集。这会产生 Look-ahead Bias：在计算第 100 天的特征时，你实际上参考了第 500 天的数据（因为模型计算了全局均值）。

对策：滚动标准化 (Rolling Scaling)¶

只使用过去 N 天的数据来计算当天的均值和方差进行缩放。

In [5]:

def rolling_scale(series, window=60):
    mean = series.rolling(window).mean()
    std = series.rolling(window).std()
    return (series - mean) / std

df['ret_scaled'] = rolling_scale(df['ret_1d'])

plt.figure(figsize=(12, 4))
plt.plot(df['ret_1d'], label='原始收益率', alpha=0.5)
plt.plot(df['ret_scaled'], label='滚动标准化收益率', alpha=0.8)
plt.title("滚动标准化：确保每个时间点的数据仅包含该点之前的统计信息")
plt.legend()
plt.show()

def rolling_scale(series, window=60): mean = series.rolling(window).mean() std = series.rolling(window).std() return (series - mean) / std df['ret_scaled'] = rolling_scale(df['ret_1d']) plt.figure(figsize=(12, 4)) plt.plot(df['ret_1d'], label='原始收益率', alpha=0.5) plt.plot(df['ret_scaled'], label='滚动标准化收益率', alpha=0.8) plt.title("滚动标准化：确保每个时间点的数据仅包含该点之前的统计信息") plt.legend() plt.show()

4. 滞后特征与未来标签 (Target Engineering)¶

在预测任务中：

• 特征 ($X$)：必须是过去的数据（如昨日收益率 shift(1)）。
• 标签 ($y$)：是未来的数据（如明日收益率 shift(-1)）。

In [6]:

# 特征：滞后一天 (当日收盘时，我们知道昨天的收盘情况)
df['feat_ret_lag1'] = df['ret_1d'].shift(1)
df['feat_vol_lag1'] = df['vol_20d'].shift(1)

# 标签：明天的收益率
df['target_y'] = df['ret_1d'].shift(-1)

feature_cols = ['feat_ret_lag1', 'feat_vol_lag1', 'ma_ratio']
final_df = df[feature_cols + ['target_y']].dropna()

print("特征与标签对齐完成。可以喂给机器学习模型了。")
final_df.tail(3)

# 特征：滞后一天 (当日收盘时，我们知道昨天的收盘情况) df['feat_ret_lag1'] = df['ret_1d'].shift(1) df['feat_vol_lag1'] = df['vol_20d'].shift(1) # 标签：明天的收益率 df['target_y'] = df['ret_1d'].shift(-1) feature_cols = ['feat_ret_lag1', 'feat_vol_lag1', 'ma_ratio'] final_df = df[feature_cols + ['target_y']].dropna() print("特征与标签对齐完成。可以喂给机器学习模型了。") final_df.tail(3)

特征与标签对齐完成。可以喂给机器学习模型了。

Out[6]:

	feat_ret_lag1	feat_vol_lag1	ma_ratio	target_y
Date
2023-12-26	-0.005547	0.009071	-0.004032	0.000518
2023-12-27	-0.002841	0.009103	-0.004222	0.002226
2023-12-28	0.000518	0.009086	-0.003087	-0.005424

🎯 练习¶

1. 尝试对成交量进行 ADF 检验，看看成交量序列是否平稳？如果不平稳，该如何转换？
2. 构造一个“动量交互特征”：ma_ratio * ret_scaled，观察它在市场大幅波动时的表现。
3. 使用 df['Close'].diff() 计算价格一阶差分，并对结果进行 ADF 检验。

---

下一模块 → ../03_indicators/01_trend_indicators.ipynb（我们将把这些处理好的特征转化为具体的交易信号）

In [ ]: