第二阶段：AI Quant Model（Day21～Day35）

今日主题：Feature Scaling & Normalization

预计学习时间：60～90 分钟。

今天我们继续开发已有的 quant_ml工程。

首先回顾一下前面的进度。

今天要解决一个问题：

为什么 PE=10、PB=1.5、ROE=0.15、Momentum=0.08 不能简单地放在一起比较？

以及：

为什么同样一组因子，使用不同的标准化方法，可能得到完全不同的机器学习结果？

今天不仅要理解理论，还要完成一个可以长期复用的特征变换模块。

一、今日工程目标

我们已经有：

data/processed/ml_dataset_cleaned.parquet

假设包含以下数据：

以上数据仅为示例，实践仍然使用我们前几天通过 AkShare 获取的真实数据。

现在观察：

pe = 28.6
pb = 9.2
roe = 0.30
momentum_20 = 0.06

这些因子数值差异很大。

但数值大，不代表因子更重要。

例如：

• PE=28.6，不代表 PE 的重要性是 ROE 的 95 倍。
• ROE=0.30，不代表盈利能力很弱。
• Momentum=0.06，实际上表示过去 20 日上涨 6%。

它们的量纲、分布、经济含义都不同。

所以，我们需要研究特征标准化。

今日交付

新增：

src/feature_scaling.py
scripts/scale_ml_dataset.py
tests/test_feature_scaling.py

生成：

data/processed/ml_dataset_scaled.parquet
data/processed/scaling_report.csv

不过，今天还要建立一个重要的工程约定：

Day25 只实现“预测时点可得的截面标准化”，不使用全历史数据拟合标准化参数。

需要使用训练集拟合参数的标准化方法，我们会在 Day26 完成时间切分后再正式应用。

这样可以避免未来信息泄漏。

第一部分：为什么机器学习需要标准化？

二、先看一个线性回归模型

假设我们的模型为：

Ŷ = w1 * PE + w2 * PB + w3 * ROE + w4 * Momentum + b

其中：

• PE：市盈率。
• PB：市净率。
• ROE：净资产收益率。
• Momentum：动量。

假设：

PE = 30
PB = 5
ROE = 0.20
Momentum = 0.05

那么：

Ŷ = 30 * w1 + 5 * w2 + 0.20 * w3 + 0.05 * w4 + b

可以看到，不同特征的数值尺度相差很大。

这可能影响什么？

1. 梯度下降的收敛速度

你在前面学习 PyTorch 时已经知道：

θ_{t+1} = θ_t - η * ∇L(θ_t)

其中：

η

是学习率。

如果不同特征的尺度相差很大，那么不同参数方向上的损失函数曲率可能差异明显。

这会使梯度下降更难选择合适的学习率。

标准化通常有助于改善优化过程。

2. 正则化的效果

例如 Ridge Regression：

L = Σ (Y_i - Ŷ_i)^2 + λ * Σ w_j^2

这里的正则化项：

λ * Σ w_j^2

直接作用于参数大小。

如果不同特征的尺度不同，那么相同预测影响所需的参数大小也不同。

这会导致正则化对不同特征的惩罚不均衡。

因此，对于带正则化的线性模型，标准化通常很重要。

3. 神经网络训练

MLP 的第一层：

H = WX + b

如果输入特征尺度差异过大，可能影响激活值分布与训练稳定性。

因此，神经网络通常也需要合理的输入特征变换。

但注意：

并不是所有机器学习模型都必须标准化。

例如 LightGBM、CatBoost 等树模型通常对单调尺度变换不敏感。

这一点后面会详细讲解。

第二部分：Z-Score 标准化

三、什么是 Z-Score？

Z-Score 是最常见的标准化方法之一。

公式：

Z = (X - μ) / σ

其中：

• X：原始因子。
• μ：均值。
• σ：标准差。

假设某一天 5 只股票的 PE：

pe = [10, 15, 20, 25, 30]

均值：

μ = (10 + 15 + 20 + 25 + 30) / 5 = 20

总体标准差：

σ = sqrt( ((10-20)^2 + (15-20)^2 + ... + (30-20)^2) / 5 )

计算得到：

σ ≈ 7.071

那么：

现在所有 PE 都变成了无量纲数值。

我们可以这样理解：

• Z=0：等于平均水平。
• Z=1：高于平均值一个标准差。
• Z=-1：低于平均值一个标准差。

注意：Z-Score 不会让原始数据自动服从正态分布。

它只是进行平移和缩放。

四、Python 实现 Z-Score

【教学实验】

这段代码只用于理解原理，不需要加入正式工程。

import numpy as np

pe = np.array(
    [10, 15, 20, 25, 30],
    dtype=float,
)

mean = pe.mean()

std = pe.std(ddof=0)

zscore = (pe - mean) / std

print("Mean:", mean)
print("Std:", std)
print("Z-Score:", zscore)

预期输出：

Mean: 20.0
Std: 7.0710678118654755

Z-Score:
[-1.41421356 -0.70710678  0.
  0.70710678  1.41421356]

这里使用：

ddof=0

表示计算总体标准差。

Pandas 的 Series.std()默认：

ddof=1

计算样本标准差。

两种方式都可以用于标准化，但整个工程应该保持一致。

今天我们统一使用 ddof=0。

第三部分：截面标准化与时间序列标准化

五、这是量化中最重要的区别之一

在普通机器学习中，我们经常使用：

from sklearn.preprocessing import StandardScaler

但在量化选股中，标准化还涉及一个特殊问题：

我们到底希望比较不同股票，还是比较同一只股票在不同时间的状态？

这是两种不同的研究目标。

六、截面标准化（Cross-Sectional Standardization）

假设 2025 年 3 月 3 日有 3000 只股票。

我们希望知道：

某只股票的 PE，相对于当天整个股票池，是高还是低？

那么就应该在同一个交易日计算：

μ_t = (1 / N_t) * Σ X_{i,t}

σ_t = sqrt( (1 / N_t) * Σ (X_{i,t} - μ_t)^2 )

然后：

Z_{i,t} = (X_{i,t} - μ_t) / σ_t

这里：

• i：股票。
• t：交易日。
• N_t：当天股票数量。

例如：

这些 Z-Score 是示意值。

如果某只股票：

pe_zscore = -1.2

说明：

它的 PE 比当天股票池平均水平低 1.2 个标准差。

对于 Top-K 横截面选股，这种标准化很有价值。

七、时间序列标准化（Time-Series Standardization）

现在换一个问题。

假设我们只研究贵州茅台。

它过去一段时间的 PE：

2025-01：25

2025-02：28

2025-03：30

2025-04：27

2025-05：26

我们想知道：

当前 PE 相对于它自身过去的估值水平，是高还是低？

这时就需要时间序列标准化。

例如：

Z_{i,t}^{TS} = (X_{i,t} - μ_{i,t}^{rolling}) / σ_{i,t}^{rolling}

其中：

μ_{i,t}^{rolling}

表示该股票过去一段时间的滚动均值。

例如：

df["pe_ts_zscore"] = (
    df.groupby("symbol")["pe"]
      .transform(
lambda x: (
              x - x.rolling(60).mean()
          ) / x.rolling(60).std()
      )
)

这里的 60 表示 60 个历史观测值。

两种标准化的区别

我们当前项目的核心任务是：

每天对股票池进行横截面预测和排名。

因此，今天正式工程优先实现：

Cross-Sectional Z-Score

时间序列标准化先作为扩展知识。

第四部分：Robust Scaling

八、为什么 Z-Score 仍然可能受到极端值影响？

因为 Z-Score 使用：

μ

和：

σ

而均值、标准差都可能受到极端值影响。

例如：

pe = [10, 12, 15, 18, 1000]

极端值 1000 会显著影响均值和标准差。

虽然我们 Day24 已经进行了 Winsorization，但金融因子往往仍然存在偏态分布。

因此，可以考虑另一种标准化方法：

Robust Scaling（稳健缩放）。

公式：

X' = (X - Median(X)) / IQR(X)

其中：

IQR = Q_0.75 - Q_0.25

也就是四分位距。

九、为什么叫 Robust？

Robust 的意思是稳健。

因为：

• 中位数对极端值不敏感。
• 四分位距通常比标准差更不容易受到极端值影响。

例如：

values = [10, 12, 15, 18, 1000]

使用：

import pandas as pd

s = pd.Series(values)

median = s.median()

q1 = s.quantile(0.25)

q3 = s.quantile(0.75)

iqr = q3 - q1

print(median)
print(iqr)

结果：

15.0
6.0

因此：

RobustScore = (X - 15) / 6

不过，Robust Scaling 也不是万能的。

例如，如果大部分股票的某个因子都完全相同：

IQR = 0

就会出现除零问题。

因此正式代码需要处理零分母。

第五部分：Min-Max Scaling

十、什么是 Min-Max？

公式：

X' = (X - X_min) / (X_max - X_min)

它可以把训练范围内的数据映射到：

[0, 1]

例如：

pe = [10, 20, 30]

得到：

10 → 0

20 → 0.5

30 → 1

这种方法适合一些需要明确输入范围的场景。

但金融因子存在一个问题：

极端值会严重影响 Min-Max 的缩放效果。

例如：

pe = [10, 12, 15, 18, 1000]

由于最大值是 1000，其他股票可能全部挤在接近 0 的范围内。

因此，在我们的量化选股基线中，Min-Max 不作为首选。

第六部分：行业中性化

十一、为什么同样的 PE 不能直接跨行业比较？

假设：

如果直接按照 PE 从低到高排名：

银行股票全部排在前面

但这未必代表银行股票都更有投资价值。

因为不同行业具有不同的：

• 盈利模式
• 资本结构
• 成长性
• 风险水平
• 估值习惯

因此，我们可以考虑：

Industry Neutralization（行业中性化）。

十二、最简单的行业中性化

一种简单方法是：

行业内标准化。

例如：

Z_{i,t}^{industry} = (X_{i,t} - μ_{industry,t}) / σ_{industry,t}

这样：

银行股票与银行股票比较

科技股票与科技股票比较

例如：

注意：

行业内标准化只是行业中性化的一种简化方法。

在机构量化中，还可能使用截面回归残差进行行业和市值中性化：

X_{i,t} = α_t + Σ β_{k,t} * Industry_{i,k} + γ_t * log(MarketCap_{i,t}) + ε_{i,t}

然后使用：

ε_{i,t}

作为中性化后的因子。

这样可以同时控制行业和市值暴露。

但今天我们暂时不把行业中性化加入正式工程。

原因很简单：

目前我们还没有确认 Day21 的数据是否包含可靠的历史行业分类。

如果行业分类来自当前最新数据，直接回填历史日期，也可能引入历史分类偏差。

所以今天只学习原理，不为了实现而编造行业数据。

第七部分：不同模型是否都需要标准化？

十三、线性模型、神经网络和树模型的区别

这是今天非常重要的工程知识。

这里有一个值得理解的数学事实。

假设：

pe = [10, 20, 30]

标准化后：

pe_scaled = [-1.22, 0, 1.22]

对于决策树来说，原来的：

PE < 20

可以变成：

PE_scaled < 0

如果变换严格单调，那么数据顺序没有改变。

因此，树模型通常不需要通过标准化来改善分裂逻辑。

但对于 MLP：

H = WX + b

输入尺度会直接影响神经网络的数值计算。

所以标准化通常更加重要。

对我们的工程有什么意义？

未来我们会训练：

Linear Regression
LightGBM
CatBoost
MLP

我建议保留两套特征数据：

原始清洗特征

以及：

标准化特征

这样以后可以公平比较不同模型。

第八部分：正式工程实践

今天开始写代码。

我们继续遵守约定：

• 【正式工程代码】：必须加入工程。
• 【辅助工具】：用于调试和分析。
• 【教学实验】：不参与正式工程。

十四、项目目录

今天新增：

quant_ml/
│
├── data/
│   └── processed/
│       ├── ml_dataset.parquet
│       ├── ml_dataset_cleaned.parquet
│       ├── ml_dataset_scaled.parquet
│       └── scaling_report.csv
│
├── src/
│   ├── data_loader.py
│   ├── features.py
│   ├── financial_features.py
│   ├── labels.py
│   ├── data_cleaning.py
│   └── feature_scaling.py          # Day25 新增
│
├── scripts/
│   ├── build_ml_features.py
│   ├── build_ml_dataset.py
│   ├── clean_ml_dataset.py
│   └── scale_ml_dataset.py         # Day25 新增
│
└── tests/
    ├── test_labels.py
    ├── test_data_cleaning.py
    └── test_feature_scaling.py     # Day25 新增

第九部分：实践一——实现截面 Z-Score

十五、【正式工程代码】src/feature_scaling.py

首先实现截面标准化。

完整代码：

import numpy as np
import pandas as pd


defvalidate_scaling_columns(
    df: pd.DataFrame,
    feature_columns: list[str],
) -> None:
"""
    检查标准化输入。
    """

    required = [
"symbol",
"trade_date",
    ]

    missing = [
        col
for col in required + feature_columns
if col notin df.columns
    ]

if missing:
raise ValueError(
f"缺少字段：{missing}"
        )

ifnot feature_columns:
raise ValueError(
"feature_columns 不能为空"
        )

if len(feature_columns) != len(set(feature_columns)):
raise ValueError(
"feature_columns 存在重复字段"
        )

if df.duplicated(
        ["symbol", "trade_date"]
    ).any():
raise ValueError(
"存在重复的股票-日期记录"
        )

for col in feature_columns:

ifnot pd.api.types.is_numeric_dtype(
            df[col]
        ):
raise TypeError(
f"{col} 不是数值类型"
            )


defcross_sectional_zscore(
    df: pd.DataFrame,
    feature_columns: list[str],
    suffix: str = "_zscore",
) -> pd.DataFrame:
"""
    按交易日期计算截面 Z-Score。

    使用总体标准差 ddof=0。

    如果某日标准差为 0，
    则有效观测值标准化为 0。

    原始 NaN 仍保留 NaN。
    """

    validate_scaling_columns(
        df,
        feature_columns,
    )

    data = df.copy()

for col in feature_columns:

        grouped = data.groupby(
"trade_date"
        )[col]

        mean = grouped.transform("mean")

        std = grouped.transform(
lambda x: x.std(ddof=0)
        )

        result = (
            data[col] - mean
        ) / std.replace(0, np.nan)

# 对零方差的有效观测赋值 0
        zero_std = (
            std.eq(0)
            & data[col].notna()
        )

        result = result.mask(
            zero_std,
0.0,
        )

        data[f"{col}{suffix}"] = result

return data


defcross_sectional_robust_scale(
    df: pd.DataFrame,
    feature_columns: list[str],
    suffix: str = "_robust",
) -> pd.DataFrame:
"""
    按交易日期进行稳健缩放。

    公式：
    (X - Median) / IQR
    """

    validate_scaling_columns(
        df,
        feature_columns,
    )

    data = df.copy()

for col in feature_columns:

        grouped = data.groupby(
"trade_date"
        )[col]

        median = grouped.transform(
"median"
        )

        q1 = grouped.transform(
lambda x: x.quantile(0.25)
        )

        q3 = grouped.transform(
lambda x: x.quantile(0.75)
        )

        iqr = q3 - q1

        result = (
            data[col] - median
        ) / iqr.replace(0, np.nan)

        zero_iqr = (
            iqr.eq(0)
            & data[col].notna()
        )

        result = result.mask(
            zero_iqr,
0.0,
        )

        data[f"{col}{suffix}"] = result

return data


defbuild_scaling_report(
    df: pd.DataFrame,
    feature_columns: list[str],
    suffix: str = "_zscore",
) -> pd.DataFrame:
"""
    生成标准化质量报告。
    """

    records = []

for col in feature_columns:

        scaled_col = f"{col}{suffix}"

        series = df[scaled_col]

        records.append({
"feature": col,
"scaled_feature": scaled_col,
"count": int(series.count()),
"missing_count": int(
                series.isna().sum()
            ),
"mean": float(
                series.mean()
            ),
"std": float(
                series.std(ddof=0)
            ),
"min": float(
                series.min()
            ),
"max": float(
                series.max()
            ),
        })

return pd.DataFrame(records)

这里有两个值得注意的设计

第一：不覆盖原始因子。

例如：

pe

标准化后生成：

pe_zscore

这样我们可以同时保留：

pe
pe_zscore

未来训练不同模型时，可以选择不同特征。

第二：不处理标签。

例如：

label_return_20d

不会进入标准化函数。

因为我们明确指定：

feature_columns

而不是自动处理所有数值列。

第十部分：为什么要处理标准差为 0 的情况？

十六、零方差问题

假设某一天只有 3 只股票：

pe = [10, 10, 10]

那么：

μ = 10
σ = 0

Z-Score：

Z = (10 - 10) / 0

无法正常计算。

这种情况在量化中可能发生。

例如：

• 某个因子当天全部为相同值。
• 数据源出现异常。
• 股票池规模太小。
• 某个因子被统一填充。

我们的处理方法是：

zero_std = (
    std.eq(0)
    & data[col].notna()
)

result = result.mask(
    zero_std,
0.0,
)

含义是：

如果当天有效因子的标准差为 0，那么所有有效观测值都赋值为 0。

因为这些股票在这个因子上没有相对差异。

但如果原始数据是 NaN，仍然保留 NaN。

这与把所有缺失值直接设置为 0 不同。

第十一部分：实践二——建立标准化入口

十七、【正式工程代码】scripts/scale_ml_dataset.py

创建：

scripts/scale_ml_dataset.py

完整代码：

from pathlib import Path

import pandas as pd

from src.feature_scaling import (
    cross_sectional_zscore,
    build_scaling_report,
)


DATA_DIR = Path("data/processed")

INPUT_PATH = (
    DATA_DIR / "ml_dataset_cleaned.parquet"
)

OUTPUT_PATH = (
    DATA_DIR / "ml_dataset_scaled.parquet"
)

REPORT_PATH = (
    DATA_DIR / "scaling_report.csv"
)


FEATURE_COLUMNS = [
"pe",
"pb",
"roe",
"momentum_5",
"momentum_20",
"volatility_20",
"amount_ratio_20",
]


defmain():

ifnot INPUT_PATH.exists():
raise FileNotFoundError(
f"找不到输入文件：{INPUT_PATH}"
        )

    df = pd.read_parquet(
        INPUT_PATH
    )

    print("\n输入数据：")
    print(df.shape)

    feature_columns = [
        col
for col in FEATURE_COLUMNS
if col in df.columns
    ]

    missing_features = [
        col
for col in FEATURE_COLUMNS
if col notin df.columns
    ]

if missing_features:
        print(
"\n不存在的候选因子：",
            missing_features,
        )

ifnot feature_columns:
raise ValueError(
"没有可标准化的特征"
        )

    print("\n标准化因子：")
    print(feature_columns)

    scaled = cross_sectional_zscore(
        df,
        feature_columns,
    )

    report = build_scaling_report(
        scaled,
        feature_columns,
    )

    OUTPUT_PATH.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    scaled.to_parquet(
        OUTPUT_PATH,
        index=False,
    )

    report.to_csv(
        REPORT_PATH,
        index=False,
        encoding="utf-8-sig",
    )

    print("\n标准化完成：")
    print(scaled.shape)

    print("\n标准化报告：")
    print(report.to_string(index=False))

    print("\n输出文件：")
    print(OUTPUT_PATH)
    print(REPORT_PATH)


if __name__ == "__main__":
    main()

十八、运行工程

在项目根目录执行：

python -m scripts.scale_ml_dataset

成功后生成：

data/processed/ml_dataset_scaled.parquet

以及：

data/processed/scaling_report.csv

预期数据结构

例如：

以上仅为结构示例。

正式输出会保留 Day24 的所有字段，并增加：

pe_zscore
pb_zscore
roe_zscore
momentum_5_zscore
momentum_20_zscore
volatility_20_zscore
amount_ratio_20_zscore

具体字段以实际存在的特征为准。

第十二部分：自动化测试

十九、【自动化测试】tests/test_feature_scaling.py

创建：

tests/test_feature_scaling.py

完整代码：

import numpy as np
import pandas as pd

from src.feature_scaling import (
    cross_sectional_zscore,
    cross_sectional_robust_scale,
)


deftest_zscore():

    df = pd.DataFrame({
"symbol": [
"000001",
"000002",
"000003",
        ],
"trade_date": [
"2025-03-03",
"2025-03-03",
"2025-03-03",
        ],
"pe": [
10.0,
20.0,
30.0,
        ],
    })

    result = cross_sectional_zscore(
        df,
        feature_columns=["pe"],
    )

    values = result[
"pe_zscore"
    ].to_numpy()

assert np.isclose(
        values.mean(),
0.0,
    )

assert np.isclose(
        values.std(ddof=0),
1.0,
    )


deftest_zero_std():

    df = pd.DataFrame({
"symbol": [
"000001",
"000002",
"000003",
        ],
"trade_date": [
"2025-03-03",
"2025-03-03",
"2025-03-03",
        ],
"pe": [
10.0,
10.0,
10.0,
        ],
    })

    result = cross_sectional_zscore(
        df,
        feature_columns=["pe"],
    )

assert (
        result["pe_zscore"] == 0
    ).all()


deftest_robust_scale():

    df = pd.DataFrame({
"symbol": [
"000001",
"000002",
"000003",
"000004",
"000005",
        ],
"trade_date": [
"2025-03-03"
        ] * 5,
"pe": [
10.0,
12.0,
15.0,
18.0,
1000.0,
        ],
    })

    result = cross_sectional_robust_scale(
        df,
        feature_columns=["pe"],
    )

assert np.isclose(
        result["pe_robust"].median(),
0.0,
    )


deftest_label_unchanged():

    df = pd.DataFrame({
"symbol": [
"000001",
"000002",
"000003",
        ],
"trade_date": [
"2025-03-03"
        ] * 3,
"pe": [
10.0,
20.0,
30.0,
        ],
"label_return_20d": [
0.10,
-0.05,
            np.nan,
        ],
    })

    result = cross_sectional_zscore(
        df,
        feature_columns=["pe"],
    )

    pd.testing.assert_series_equal(
        result["label_return_20d"],
        df["label_return_20d"],
    )

运行：

pytest -q

如果 Day23、Day24 的测试都保留在工程中，那么 pytest 会自动运行所有测试。

第十三部分：一个容易忽略的问题

二十、标准化后的因子，是否可以直接用于模型训练？

我们已经得到：

ml_dataset_scaled.parquet

但我不建议今天直接训练模型。

原因是：

我们还没有完成训练集、验证集、测试集的时间划分。

你可能会问：

既然今天已经做了标准化，为什么还要担心数据泄漏？

因为不同标准化方法使用的数据范围不同。

例如：

方法 A：全历史标准化

mean = df["pe"].mean()

std = df["pe"].std()

df["pe_scaled"] = (
    df["pe"] - mean
) / std

这种方法可能使用 2025 年的 PE 数据计算 2023 年的标准化结果。

对于历史样本外评价，这不符合时间要求。

方法 B：同日截面标准化

df.groupby("trade_date")["pe"]

只使用对应交易日的数据。

如果当日股票池和因子值都在预测时点可获得，那么这种处理可以符合历史预测的信息约束。

但还有一个细节：

我们今天的截面标准化结果依赖当日参与计算的股票集合。

如果未来回测只允许交易某个历史股票池，那么计算截面均值和标准差时，也应该遵守相应的股票池规则。

因此，今天生成的文件是：

完成截面标准化的候选机器学习数据集。

Day26 会进一步建立训练集、验证集和测试集，并检查数据泄漏边界。

第十四部分：教学实验

下面的实验不需要放进正式工程。

你可以直接在 VSCode 中运行。

二十一、实验一：比较 Z-Score 和 Robust Scaling

【教学实验】

import pandas as pd

from src.feature_scaling import (
    cross_sectional_zscore,
    cross_sectional_robust_scale,
)


df = pd.DataFrame({
"symbol": [
"000001",
"000002",
"000003",
"000004",
"000005",
    ],
"trade_date": [
"2025-03-03"
    ] * 5,
"pe": [
10,
12,
15,
18,
1000,
    ],
})

zscore_df = cross_sectional_zscore(
    df,
    ["pe"],
)

robust_df = cross_sectional_robust_scale(
    df,
    ["pe"],
)

result = pd.DataFrame({
"pe": df["pe"],
"zscore": zscore_df["pe_zscore"],
"robust": robust_df["pe_robust"],
})

print(result)

观察：

• 极端值 1000 对 Z-Score 的影响。
• Robust Scaling 是否更好地保留普通股票之间的差异。
• 为什么 Robust Scaling 不会自动消除极端值。

二十二、实验二：观察真实因子标准化后的分布

【教学实验】

import pandas as pd
import matplotlib.pyplot as plt


df = pd.read_parquet(
"data/processed/ml_dataset_scaled.parquet"
)

plt.figure(figsize=(10, 6))

plt.hist(
    df["pe_zscore"].dropna(),
    bins=80,
)

plt.title(
"PE Cross-Sectional Z-Score"
)

plt.xlabel("Z-Score")

plt.ylabel("Frequency")

plt.grid(alpha=0.3)

plt.show()

观察：

1. 分布是否以 0 附近为中心？
2. 是否仍然存在长尾？
3. 为什么 Z-Score 不能保证正态分布？

二十三、实验三：检查每个交易日的标准化结果

【教学实验】

import pandas as pd


df = pd.read_parquet(
"data/processed/ml_dataset_scaled.parquet"
)

daily_stats = (
    df.groupby("trade_date")[
"pe_zscore"
    ]
    .agg([
"mean",
"std",
"count",
    ])
)

print(
    daily_stats.head(10)
)

注意：

这里 Pandas 的：

std()

默认使用：

ddof=1

而我们的正式标准化使用：

ddof=0

所以即使某一天的标准化结果采用总体标准差 1，实验中输出的样本标准差也可能略大于 1。

如果希望完全一致，可以使用：

daily_std = (
    df.groupby("trade_date")[
"pe_zscore"
    ]
    .std(ddof=0)
)

第十五部分：今日练习题

二十四、理论题

题1：

为什么 Z-Score 标准化通常有利于 MLP 训练，但 LightGBM 往往不需要？

题2：

截面标准化与时间序列标准化的核心区别是什么？

题3：

为什么不能直接使用整个历史数据集的均值和标准差进行历史回测标准化？

题4：

为什么 Z-Score 不会让因子自动服从正态分布？

题5：

行业内标准化是否等于完整的行业中性化？为什么？

二十五、编程题

题目1：实现截面 Min-Max Scaling

要求：

实现函数：

cross_sectional_minmax(
    df,
    feature_columns,
)

按交易日期，将因子缩放到：

[0, 1]

如果某一天：

X_max = X_min

则所有有效值赋为 0。

缺失值保持 NaN。

题目2：比较不同标准化方法

对真实数据中的：

pe

分别执行：

Z-Score
Robust Scaling
Min-Max

比较结果的：

• 均值
• 标准差
• 最小值
• 最大值

题目3：验证截面 Z-Score

要求：

按交易日计算：

pe_zscore

的均值。

统计：

绝对均值 < 1e-8

的交易日数量。

同时检查是否存在零方差截面。

第十六部分：练习参考答案

二十六、理论题答案

第1题：

MLP 的输入尺度会影响线性层、激活值及梯度优化；LightGBM 主要依赖特征的排序和分裂阈值，通常不需要标准化。

第2题：

截面标准化比较同一天不同股票；时间序列标准化比较同一股票在不同日期的状态。

第3题：

全历史统计量可能包含预测时点之后的数据，从而造成前视偏差或样本外评价中的信息泄漏。

第4题：

Z-Score 只是线性变换，不改变原始分布的基本形状。

第5题：

不完全等于。行业内标准化是一种简化的行业调整方法，但无法自动消除所有行业暴露，也不控制市值等其他风险因子。

二十七、编程题答案

题目1

import numpy as np
import pandas as pd


defcross_sectional_minmax(
    df: pd.DataFrame,
    feature_columns: list[str],
) -> pd.DataFrame:

    data = df.copy()

for col in feature_columns:

        grouped = data.groupby(
"trade_date"
        )[col]

        min_value = grouped.transform("min")

        max_value = grouped.transform("max")

        denominator = (
            max_value - min_value
        )

        result = (
            data[col] - min_value
        ) / denominator.replace(0, np.nan)

        zero_range = (
            denominator.eq(0)
            & data[col].notna()
        )

        result = result.mask(
            zero_range,
0.0,
        )

        data[f"{col}_minmax"] = result

return data

题目2

import pandas as pd

from src.feature_scaling import (
    cross_sectional_zscore,
    cross_sectional_robust_scale,
)


df = pd.read_parquet(
"data/processed/ml_dataset_cleaned.parquet"
)

zscore_df = cross_sectional_zscore(
    df,
    ["pe"],
)

robust_df = cross_sectional_robust_scale(
    df,
    ["pe"],
)

result = pd.DataFrame({
"zscore": zscore_df["pe_zscore"],
"robust": robust_df["pe_robust"],
})

print(
    result.describe()
)

Min-Max 的结果可以通过题目 1 中的函数补充。

题目3

import pandas as pd


df = pd.read_parquet(
"data/processed/ml_dataset_scaled.parquet"
)

daily_mean = (
    df.groupby("trade_date")[
"pe_zscore"
    ]
    .mean()
)

valid_days = daily_mean.notna()

passed = (
    daily_mean[valid_days]
    .abs() < 1e-8
)

print(
"有效交易日数量：",
    valid_days.sum()
)

print(
"均值检查通过数量：",
    passed.sum()
)

对于全体有效观测相同的截面，我们的函数会把标准化结果设置为 0，因此均值检查也可以通过。

第十七部分：Day25 工程验收

今天完成后，检查：

十八、Day25 总结

今天我们学习了四个重要概念：

1. Z-Score Standardization

Z = (X - μ) / σ

用于消除均值和尺度差异。

2. Robust Scaling

X' = (X - Median) / IQR

用于降低极端值对缩放过程的影响。

3. Cross-Sectional Standardization

Z_{i,t} = (X_{i,t} - μ_t) / σ_t

用于比较同一天不同股票的相对因子水平。

4. Industry Neutralization

通过行业内比较或截面回归，降低因子中的行业暴露。

我们还正式实现了：

src/feature_scaling.py

以及：

scripts/scale_ml_dataset.py

现在整个工程的数据流已经变成：

Day21：真实 A 股数据
        ↓
Day22：因子工程
        ↓
Day23：标签工程
        ↓
Day24：数据清洗
        ↓
Day25：特征标准化
        ↓
ml_dataset_scaled.parquet
        ↓
Day26：时间序列切分
        ↓
Day27：第一个机器学习模型

今天最重要的一句话

标准化的目的不是让所有因子变得一样，而是让模型在合理的数值尺度下学习不同因子的有效信息。

尤其对于我们未来要训练的 MLP，特征标准化非常重要。

但对于 LightGBM 和 CatBoost，标准化未必带来明显收益。

所以未来我们会通过实验来比较：

原始因子、清洗因子、标准化因子，究竟哪一种能够产生更好的样本外预测效果。

下一课 Day26：时间序列数据集划分——Train、Validation、Test、Purging 与 Embargo。

这一课将是我们进入正式模型训练之前，最重要的一道防线：确保模型没有在训练过程中提前看到未来。

Day25 小结

今天我们完成了量化特征标准化：

1. 理解不同因子量纲不同，数值大小不等于重要性
2. 掌握 Z-Score、Robust Scaling、Min-Max 的区别
3. 区分截面标准化与时间序列标准化，正式工程采用截面 Z-Score
4. 了解行业中性化原理，但暂不编造行业数据
5. 理解线性模型、MLP 通常需要标准化，树模型通常不需要
6. 实现 feature_scaling.py、标准化入口与测试
7. 输出 ml_dataset_scaled.parquet，且不使用全历史统计量

Day25 最重要的一句话：

标准化的目的不是让所有因子变得一样，而是让模型在合理的数值尺度下学习不同因子的有效信息。

下一课：Day26 — 时间序列数据集划分：Train、Validation、Test、Purging 与 Embargo