跳到正文

特征工程：为你的 Alpha 模型构建正确的输入

特征工程将原始市场数据转换为 alpha 模型的候选输入。好特征捕获经济上有意义的关系。坏特征捕获噪声。

第 5 篇 · 共 10 篇 约 2 分钟

在基于 ML 的量化金融中，特征工程是将原始市场数据（价格、成交量、基本面、另类数据）转换为预测模型的候选输入的过程。你的特征质量决定你模型表现的上限。

特征工程流水线

原始数据 → 预处理 → 特征生成 → 特征选择 → Alpha 模型

每个阶段都很重要。每个阶段的垃圾输入，垃圾输出。

特征类别

1. 基于价格的特征

从价格时间序列派生：

收益和对数收益：

import numpy as np
import pandas as pd

def compute_returns(prices):
    """简单收益：(P_t - P_{t-1}) / P_{t-1}"""
    return prices.pct_change()

def compute_log_returns(prices):
    """对数收益：ln(P_t / P_{t-1})"""
    return np.log(prices / prices.shift(1))

动量特征：

def momentum(prices, periods=[20, 60, 120, 252]):
    """
    多周期动量特征。
    返回每只股票的每个周期的动量。
    """
    mom = {}
    for p in periods:
        mom[f'mom_{p}d'] = prices / prices.shift(p) - 1
    return pd.DataFrame(mom)


def relative_momentum(prices, benchmark, periods=[60, 120, 252]):
    """
    相对动量：股票收益减去基准收益。
    """
    rel_mom = {}
    for p in periods:
        stock_ret = prices / prices.shift(p) - 1
        bench_ret = benchmark / benchmark.shift(p) - 1
        rel_mom[f'rel_mom_{p}d'] = stock_ret - bench_ret
    return pd.DataFrame(rel_mom)

波动率特征：

def rolling_volatility(returns, windows=[20, 60, 120]):
    """
    多窗口滚动波动率。
    """
    vol = {}
    for w in windows:
        vol[f'vol_{w}d'] = returns.rolling(w).std() * np.sqrt(252)
    return pd.DataFrame(vol)


def realized_volatility(prices, window=20):
    """
    已实现波动率：日内高频收益的方差估计。
    """
    intraday_returns = prices.pct_change()
    return intraday_returns.rolling(window).std() * np.sqrt(252)


def volatility_ratio(short_vol, long_vol):
    """
    波动率比率：短期波动率 / 长期波动率。
    高比率表示波动率正在收缩（趋势可能持续）。
    """
    return short_vol / long_vol

趋势强度特征：

def trend_strength(prices, lookback=60):
    """
    趋势强度：R-squared of linear time trend。

    接近 1 = 强趋势
    接近 0 = 区间震荡
    """
    log_prices = np.log(prices)

    x = np.arange(len(log_prices))
    x_mean = x.mean()

    ss_x = ((x - x_mean) ** 2).sum()
    ss_residuals = ((log_prices - log_prices.mean()) ** 2).sum()

    r_squared = 1 - ss_residuals / (ss_x + ss_residuals)
    return r_squared

2. 基本面特征

来自财务报表：

def fundamental_features(fundamentals):
    """
    从基本面数据计算标准价值和质量特征。
    """
    features = pd.DataFrame()

    # 估值特征
    features['pe_ratio'] = fundamentals['price'] / fundamentals['earnings_per_share']
    features['book_to_market'] = fundamentals['book_value'] / fundamentals['market_cap']
    features['ps_ratio'] = fundamentals['price'] / fundamentals['revenue_per_share']
    features['pcf_ratio'] = fundamentals['price'] / fundamentals['cash_flow_per_share']

    # 盈利能力特征
    features['roe'] = fundamentals['net_income'] / fundamentals['book_value']
    features['roa'] = fundamentals['net_income'] / fundamentals['total_assets']
    features['gross_margin'] = fundamentals['gross_profit'] / fundamentals['revenue']
    features['operating_margin'] = fundamentals['operating_income'] / fundamentals['revenue']

    # 财务杠杆特征
    features['debt_to_equity'] = fundamentals['total_debt'] / fundamentals['total_equity']
    features['debt_to_assets'] = fundamentals['total_debt'] / fundamentals['total_assets']
    features['interest_coverage'] = fundamentals['ebit'] / fundamentals['interest_expense']

    # 增长特征
    features['revenue_growth'] = fundamentals['revenue'].pct_change(4)  # YoY
    features['earnings_growth'] = fundamentals['earnings_per_share'].pct_change(4)
    features['book_value_growth'] = fundamentals['book_value'].pct_change(4)

    return features


def earnings_quality(fundamentals, prices):
    """
    盈利质量特征：应计项目。

    高应计项目（相对于现金流）可能表示盈利质量低。
    """
    cash_flow = fundamentals['operating_cash_flow']
    net_income = fundamentals['net_income']

    accruals = (net_income - cash_flow) / fundamentals['total_assets']
    return accruals

3. 另类数据特征

另类数据（替代传统财务和价格数据的数据）越来越重要：

def alternative_features(web_data, sentiment_model=None):
    """
    另类数据特征：网络流量、情感、搜索量。

    Parameters
    ----------
    web_data : DataFrame
        包含 page_views, search_volume 等列
    sentiment_model : 可选，NLP 模型
    """
    features = pd.DataFrame()

    # 网络流量特征
    features['page_views_z'] = (
        web_data['page_views'] - web_data['page_views'].mean()
    ) / web_data['page_views'].std()
    features['search_volume_z'] = (
        web_data['search_index'] - web_data['search_index'].mean()
    ) / web_data['search_index'].std()
    features['engagement_rate'] = (
        web_data['unique_visitors'] / web_data['page_views']
    )

    # 情感特征（如果有 NLP 模型）
    if sentiment_model is not None:
        features['news_sentiment'] = sentiment_model.predict(
            web_data['news_headlines']
        )
        features['social_sentiment'] = sentiment_model.predict(
            web_data['social_posts']
        )

    return features

关键原则

原则 1：经济直觉第一

在跑 ML 之前问：为什么这个特征会预测收益？

具有经济直觉的特征比通过暴力搜索发现的特征更可能稳健。特征应反映使经济上说得通的市场无效性。

原则 2：平稳性

收益是平稳的。价格是非平稳的（它们有趋势）。大多数 ML 模型假设平稳输入。

始终使用收益，而不是价格 作为特征的基础。或者差分价格序列使其平稳。

def make_stationary(prices):
    """
    使价格序列平稳：通过差分。
    """
    return prices.pct_change()


def detrend(prices):
    """
    去趋势：从价格中移除线性时间趋势。
    """
    log_prices = np.log(prices)
    x = np.arange(len(log_prices))

    # 线性拟合
    coeffs = np.polyfit(x, log_prices, 1)
    trend = np.polyval(coeffs, x)

    return log_prices - trend

原则 3：归一化

不同特征有不同尺度。P/E 比可能范围是 5 到 50。动量收益范围是 -0.5 到 +0.5。归一化后再喂给 ML 模型：

from sklearn.preprocessing import StandardScaler

def normalize_features(features, scaler=None):
    """
    使用 StandardScaler 归一化特征。
    """
    if scaler is None:
        scaler = StandardScaler()

    normalized = pd.DataFrame(
        scaler.fit_transform(features),
        index=features.index,
        columns=features.columns
    )
    return normalized, scaler


def rank_normalize_features(features):
    """
    按排名归一化（对异常值更稳健）。
    """
    ranked = features.rank(pct=True)
    return ranked

原则 4：避免数据泄漏

你的特征在时间 t 必须只使用在 t 时可用的信息。常见泄漏来源：

def compute_feature_without_leakage(fundamentals, prices, fiscal_year_end_month):
    """
    计算基本面特征时不使用前视数据。

    使用最近一期已公布的基本面数据。
    """
    # 正确做法：使用最近一期已公布的数据
    # fiscal_year_end_month 告诉你公司财政年度何时结束
    pass

# 错误做法：
# future_earnings = fundamentals['earnings'].shift(-1)  # 前视！

# 正确做法：
# 使用当前已知的基本面数据
current_fundamentals = fundamentals

特征选择

1. 单变量 IC 筛选。 计算每个特征与标签之间的 IC。保留 IC 最高的特征。快但忽略特征交互。

def univariate_ic_screen(features, labels, top_k=50):
    """
    按 IC 筛选前 K 个特征。
    """
    ic_scores = {}
    for col in features.columns:
        valid = ~(features[col].isna() | labels.isna())
        if valid.sum() > 30:
            ic, _ = stats.spearmanr(
                features.loc[valid, col],
                labels[valid]
            )
            ic_scores[col] = ic

    sorted_features = sorted(ic_scores.items(), key=lambda x: abs(x[1]), reverse=True)
    return [f[0] for f in sorted_features[:top_k]]

2. 递归特征消除。 训练模型，移除最不重要的特征，重复。更准确但更慢。

3. L1 正则化（Lasso）。 带 L1 惩罚的 ML 模型自动将无关特征的权重设为零。

from sklearn.linear_model import LassoCV

def lasso_selection(X, y, alphas=None):
    """
    使用 Lasso 进行特征选择。
    """
    if alphas is None:
        alphas = np.logspace(-5, 1, 50)

    model = LassoCV(cv=5, alphas=alphas, random_state=42)
    model.fit(X, y)

    selected = X.columns[model.coef_ != 0].tolist()
    return selected, model

常见错误

数据挖掘噪声。 500 个特征和 5 年日数据，你会找到在样本内纯粹偶然预测的特征。始终使用样本外测试或保留集。

对小盘股过拟合。 许多特征只是小盘股的代理（大盘股行为与小盘股不同）。包括市值作为控制变量，或对大盘和小盘分别测试特征。

前视偏差在基本面数据中。 报告的基本面经常被修订。使用「报告」数据而不调整修订会引入前视偏差。使用时间点数据或等待数据被确认。

特征相关性与多重共线性。 高相关特征可能给 ML 模型问题。检查特征相关性矩阵，移除高度相关的特征。

参考文献

  • López de Prado, Advances in Financial Machine Learning, 第 3 章. Wiley, 2018. ISBN: 978-1-119-48208-6.
  • Ang, Asset Management, 第 9 章. Oxford University Press, 2014. ISBN: 978-0-19-995932-3.
  • Aronson, Evidence-Based Technical Analysis, 第 8 章. Wiley, 2006. ISBN: 978-0-471-75671-9.