特征工程实战：构建工业级销量预测特征体系——从时序特征到Embedding特征的全方位解析

本文是销量预测系列的深度技术文，聚焦机器学习中最关键也最耗时的环节——特征工程。在Kaggle竞赛和工业实践中有一个共识："数据和特征决定了机器学习的上限，而模型和算法只是逼近这个上限。"本文将从时序特征、统计特征、交叉特征到深度学习Embedding特征，手把手构建一套完整的工业级特征工程体系，包含完整的Python代码和实战经验总结。

一、特征工程的重要性再思考

为什么特征工程决定了模型上限？

在讨论具体的特征工程方法之前，我们先从数学角度理解特征工程的本质。

机器学习的本质是找到一个函数 ( f(x) ) 来拟合输入 ( x ) 和输出 ( y ) 之间的映射关系。模型（如LightGBM、神经网络）的能力在于拟合这个函数的形式，但输入特征 ( x ) 的表达能力才是决定模型能否学到真实规律的关键。

举例来说，如果我们要预测明天的销量，原始数据可能只有"昨天的销量"这一个字段。模型当然可以学习到 ( f(昨天的销量) = 明天的销量 ) 这个简单关系，但这忽略了：

• 周期性规律：周末/工作日、月初/月末、节假日
• 趋势变化：销量的长期增长/下降趋势
• 外部因素：天气、促销活动、竞争对手行为
• 多维度关联：不同商品品类之间的销售关联

特征工程的目标就是通过各种手段，将这些隐含的、复杂的规律显式化为模型可以理解的数值特征。

工业级特征工程的四大原则

经过大量项目实践，我总结了工业级特征工程的四大原则：

二、基础时序特征：从日期到时间的全面提取

日期时间特征是时序预测的基石

几乎所有的时序预测任务都离不开日期时间特征。一个简单的日期字段，可以拆解出丰富的时间信息。

1. 基础时间特征提取

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')

class TimeFeatureExtractor:
"""时间特征提取器"""

def __init__(self, date_column: str = 'date'):
self.date_column = date_column

def extract_basic_features(self, df: pd.DataFrame) –> pd.DataFrame:
"""提取基础时间特征"""
df = df.copy()

# 确保日期列是datetime类型
df[self.date_column] = pd.to_datetime(df[self.date_column])

# ===== 基础时间单元 =====
df['year'] = df[self.date_column].dt.year
df['quarter'] = df[self.date_column].dt.quarter
df['month'] = df[self.date_column].dt.month
df['week'] = df[self.date_column].dt.isocalendar().week.astype(int)
df['day'] = df[self.date_column].dt.day
df['day_of_year'] = df[self.date_column].dt.dayofyear
df['day_of_week'] = df[self.date_column].dt.dayofweek # 0=周一, 6=周日
df['week_of_year'] = df[self.date_column].dt.isocalendar().week.astype(int)

# ===== 周/月的位置特征 =====
df['week_of_month'] = (df['day'] – 1) // 7 + 1 # 周在月中的位置
df['is_month_start'] = df[self.date_column].dt.is_month_start.astype(int)
df['is_month_end'] = df[self.date_column].dt.is_month_end.astype(int)
df['is_quarter_start'] = df[self.date_column].dt.is_quarter_start.astype(int)
df['is_quarter_end'] = df[self.date_column].dt.is_quarter_end.astype(int)
df['is_year_start'] = df[self.date_column].dt.is_year_start.astype(int)
df['is_year_end'] = df[self.date_column].dt.is_year_end.astype(int)

# ===== 工作日/休息日特征 =====
df['is_weekend'] = (df['day_of_week'] >= 5).astype(int)
df['is_workday'] = (df['day_of_week'] < 5).astype(int)

# ===== 特殊日期标记 =====
# 节假日（需要结合具体国家/地区的节假日数据）
df['is_holiday'] = self._extract_holiday_features(df)

# 节假日前后几天（通常销量会有波动）
df['days_to_holiday'] = self._calculate_days_to_holiday(df)
df['days_from_holiday'] = self._calculate_days_from_holiday(df)

# ===== 月初/月中/月末分段 =====
df['month_period'] = pd.cut(
df['day'],
bins=[0, 10, 20, 31],
labels=[0, 1, 2]
).astype(int)

return df

def extract_cyclical_features(self, df: pd.DataFrame) –> pd.DataFrame:
"""提取周期特征（正弦/余弦编码）

为什么需要周期性编码？
例如 day_of_week=0（周一）和 day_of_week=6（周日）在数值上差6，
但实际上它们是"相邻"的。通过正弦/余弦编码可以让模型理解这种周期性。
"""
df = df.copy()

# 周期的秒数
periods = {
'day_of_week': 7,
'day_of_month': 30,
'day_of_year': 365,
'month': 12,
'week_of_year': 52
}

for feature, period in periods.items():
if feature in df.columns:
# 正弦编码
df[f'{feature}_sin'] = np.sin(2 * np.pi * df[feature] / period)
# 余弦编码
df[f'{feature}_cos'] = np.cos(2 * np.pi * df[feature] / period)

return df

def _extract_holiday_features(self, df: pd.DataFrame) –> pd.Series:
"""提取节假日特征（以中国为例）"""
holidays = self._get_chinese_holidays(df[self.date_column].max().year)

is_holiday = df[self.date_column].isin(holidays).astype(int)

# 调休日（周末但需要上班）标记为-1
# 这个需要单独处理，在此简化
return is_holiday

def _get_chinese_holidays(self, year: int) –> list:
"""获取中国法定节假日（简化版，实际建议用holidays库）"""
# 春节（除夕到初七，通常是15天左右）
spring_festival_start = pd.Timestamp(f'{year}-01-21')
spring_festival = pd.date_range(spring_festival_start, periods=15)

# 清明节
qingming = [pd.Timestamp(f'{year}-04-0{x}') for x in range(3, 6)]

# 劳动节
labor_day = [pd.Timestamp(f'{year}-05-0{x}') for x in range(1, 4)]

# 国庆节
national_day = pd.date_range(f'{year}-10-01', periods=7)

all_holidays = list(spring_festival) + qingming + labor_day + list(national_day)
return all_holidays

def _calculate_days_to_holiday(self, df: pd.DataFrame) –> pd.Series:
"""计算距离下一个节假日的天数"""
# 这个实现比较复杂，实际项目中建议用专门的库
# 这里返回全0作为占位
return pd.Series(0, index=df.index)

def _calculate_days_from_holiday(self, df: pd.DataFrame) –> pd.Series:
"""计算距离上一个节假日的天数"""
return pd.Series(0, index=df.index)

2. 时钟时间特征（针对小时级数据）

def extract_hourly_features(self, df: pd.DataFrame) –> pd.DataFrame:
"""提取小时级时间特征"""
df = df.copy()

if 'hour' not in df.columns and self.date_column in df.columns:
df['hour'] = pd.to_datetime(df[self.date_column]).dt.hour

# 基础小时特征
df['hour_sin'] = np.sin(2 * np.pi * df['hour'] / 24)
df['hour_cos'] = np.cos(2 * np.pi * df['hour'] / 24)

# 时段划分（电商场景）
df['time_period'] = pd.cut(
df['hour'],
bins=[–1, 6, 9, 12, 14, 18, 21, 24],
labels=[0, 1, 2, 3, 4, 5, 6]
).astype(int)

# 特殊时段标记
df['is_morning_rush'] = ((df['hour'] >= 7) & (df['hour'] <= 9)).astype(int)
df['is_lunch_time'] = ((df['hour'] >= 11) & (df['hour'] <= 13)).astype(int)
df['is_evening_rush'] = ((df['hour'] >= 17) & (df['hour'] <= 19)).astype(int)
df['is_late_night'] = ((df['hour'] >= 22) | (df['hour'] <= 5)).astype(int)

return df

三、滞后特征与滑窗特征：时序预测的核心

滞后特征：最简单的时序特征

滞后特征是最直观的时序特征，它直接使用历史数据作为预测当前时刻的特征。

class LagFeatureGenerator:
"""滞后特征生成器"""

def __init__(self, group_columns: list = None):
"""
Args:
group_columns: 分组列，如['store', 'family']，表示按门店和品类分组
"""
self.group_columns = group_columns or []

def create_lag_features(self,
df: pd.DataFrame,
target_column: str,
lags: list = None,
sort_column: str = 'date') –> pd.DataFrame:
"""
创建滞后特征

Args:
df: 输入数据
target_column: 目标列名
lags: 滞后阶数列表，如[1, 2, 3, 7, 14, 21, 28]
sort_column: 排序列
"""
df = df.copy()

if lags is None:
lags = [1, 2, 3, 7, 14, 21, 28]

# 确保数据按时间和分组排序
if self.group_columns:
df = df.sort_values(self.group_columns + [sort_column])

# 创建滞后特征
for lag in lags:
lag_col = f'{target_column}_lag_{lag}'
if self.group_columns:
df[lag_col] = df.groupby(self.group_columns)[target_column].shift(lag)
else:
df[lag_col] = df[target_column].shift(lag)

return df

def create_rolling_features(self,
df: pd.DataFrame,
target_column: str,
windows: list = None,
agg_functions: list = None) –> pd.DataFrame:
"""
创建滑窗统计特征

Args:
df: 输入数据
target_column: 目标列名
windows: 窗口大小列表，如[7, 14, 28, 56]
agg_functions: 聚合函数列表
"""
df = df.copy()

if windows is None:
windows = [7, 14, 28, 56]
if agg_functions is None:
agg_functions = ['mean', 'std', 'min', 'max', 'median']

for window in windows:
# 分组滑窗（如果有分组列）
if self.group_columns:
grouped = df.groupby(self.group_columns)[target_column]
else:
grouped = df[target_column]

# 滚动统计
rolling = grouped.transform(
lambda x: x.shift(1).rolling(window=window, min_periods=1)
)

# 均值
if 'mean' in agg_functions:
df[f'{target_column}_rolling_mean_{window}'] = rolling.mean()

# 标准差
if 'std' in agg_functions:
df[f'{target_column}_rolling_std_{window}'] = rolling.std()

# 最小值
if 'min' in agg_functions:
df[f'{target_column}_rolling_min_{window}'] = rolling.min()

# 最大值
if 'max' in agg_functions:
df[f'{target_column}_rolling_max_{window}'] = rolling.max()

# 中位数
if 'median' in agg_functions:
df[f'{target_column}_rolling_median_{window}'] = rolling.median()

return df

def create_expanding_features(self,
df: pd.DataFrame,
target_column: str) –> pd.DataFrame:
"""
创建累积特征（从历史起点累积到当前）
"""
df = df.copy()

if self.group_columns:
grouped = df.groupby(self.group_columns)[target_column]
else:
grouped = df[target_column]

# 累积均值
df[f'{target_column}_expanding_mean'] = grouped.transform(
lambda x: x.shift(1).expanding().mean()
)

# 累积最大/最小
df[f'{target_column}_expanding_max'] = grouped.transform(
lambda x: x.shift(1).expanding().max()
)
df[f'{target_column}_expanding_min'] = grouped.transform(
lambda x: x.shift(1).expanding().min()
)

return df

def create_momentum_features(self,
df: pd.DataFrame,
target_column: str,
periods: list = None) –> pd.DataFrame:
"""
创建动量特征（变化率、增长率）
"""
df = df.copy()

if periods is None:
periods = [7, 14, 28]

for period in periods:
lag_col = f'{target_column}_lag_{period}'
if lag_col not in df.columns:
df = self.create_lag_features(df, target_column, [period])

# 变化量（绝对值）
df[f'{target_column}_diff_{period}'] = df[target_column] – df[lag_col]

# 变化率（相对值）
df[f'{target_column}_pct_change_{period}'] = (
(df[target_column] – df[lag_col]) / (df[lag_col] + 1)
)

# 同期增长率（周同比、月同比）
if period == 7:
df[f'{target_column}_wow_growth'] = df[f'{target_column}_pct_change_7']
elif period == 28:
df[f'{target_column}_mom_growth'] = df[f'{target_column}_pct_change_28']

return df

实战：完整的滞后与滑窗特征构建

def build_time_series_features(df: pd.DataFrame,
target_column: str = 'sales',
group_columns: list = ['store', 'family'],
prediction_horizon: int = 16) –> pd.DataFrame:
"""
构建完整的时序特征体系

Args:
df: 原始数据，必须包含date, store, family, sales列
target_column: 目标列
group_columns: 分组列
prediction_horizon: 预测范围，用于创建不同步长的滞后特征
"""
print(f"原始数据形状: {df.shape}")

# 1. 基础滞后特征
lag_generator = LagFeatureGenerator(group_columns=group_columns)

# 不同预测范围使用不同的滞后阶数
if prediction_horizon == 1:
# 预测次日：短期滞后更重要
lags = [1, 2, 3, 7, 14, 21]
elif prediction_horizon <= 7:
# 预测一周：中期滞后
lags = [7, 14, 21, 28, 35]
else:
# 预测多周：长期滞后
lags = [7, 14, 28, 56, 84]

df = lag_generator.create_lag_features(df, target_column, lags)

# 2. 滑窗统计特征
df = lag_generator.create_rolling_features(
df, target_column,
windows=[7, 14, 28, 56],
agg_functions=['mean', 'std', 'min', 'max', 'median']
)

# 3. 累积特征
df = lag_generator.create_expanding_features(df, target_column)

# 4. 动量特征
df = lag_generator.create_momentum_features(df, target_column, [7, 14, 28])

# 5. 同期特征（上周同期、上月同期）
df = lag_generator.create_same_period_features(df, target_column)

print(f"时序特征后数据形状: {df.shape}")
return df

def create_same_period_features(self, df: pd.DataFrame, target_column: str) –> pd.DataFrame:
"""创建同期特征（上周同一时间、上月同一时间）"""
df = df.copy()

# 上周同期（7天前）
lag_7 = f'{target_column}_lag_7'
if lag_7 in df.columns:
# 上上周同期（14天前）
lag_14 = f'{target_column}_lag_14'
if lag_14 in df.columns:
df[f'{target_column}_2w_avg'] = (df[lag_7] + df[lag_14]) / 2

# 上月同期（28天前）
lag_28 = f'{target_column}_lag_28'
if lag_28 in df.columns:
df[f'{target_column}_4w_avg'] = (
df[lag_7] + df[lag_14] + df.get(f'{target_column}_lag_21', 0) + df[lag_28]
) / 4

return df

四、统计特征：分布与异常的全方位刻画

分组统计特征

分组统计是最常用的特征工程手段之一，它能够捕捉个体在群体中的相对位置。

class StatisticalFeatureGenerator:
"""统计特征生成器"""

def create_group_statistics(self,
df: pd.DataFrame,
group_columns: list,
target_column: str,
stats: list = None) –> pd.DataFrame:
"""
创建分组统计特征

Args:
df: 输入数据
group_columns: 分组列
target_column: 目标列
stats: 统计函数列表
"""
df = df.copy()

if stats is None:
stats = ['mean', 'std', 'min', 'max', 'median', 'sum', 'count']

# 确保分组列存在
for col in group_columns:
if col not in df.columns:
raise ValueError(f"分组列 {col} 不存在于数据中")

# 创建分组键
df['_group_key'] = df[group_columns].astype(str).agg('_'.join, axis=1)

# 计算分组统计
group_stats = df.groupby('_group_key')[target_column].agg(stats)
group_stats.columns = [f'{target_column}_group_{stat}' for stat in stats]
group_stats = group_stats.reset_index()

# 合并回原数据
df = df.merge(group_stats, on='_group_key', how='left')
df.drop('_group_key', axis=1, inplace=True)

# 计算个体在群体中的位置（归一化）
if 'mean' in stats and 'std' in stats:
mean_col = f'{target_column}_group_mean'
std_col = f'{target_column}_group_std'
df[f'{target_column}_zscore'] = (
(df[target_column] – df[mean_col]) / (df[std_col] + 1e-6)
)

return df

def create_rank_features(self,
df: pd.DataFrame,
group_columns: list,
value_column: str) –> pd.DataFrame:
"""创建排名特征"""
df = df.copy()

# 组内排名
df[f'{value_column}_rank_in_group'] = df.groupby(
group_columns
)[value_column].rank(method='dense', ascending=False)

# 组内百分比排名
df[f'{value_column}_pct_rank_in_group'] = df.groupby(
group_columns
)[value_column].rank(pct=True, method='average')

return df

def create_quantile_features(self,
df: pd.DataFrame,
column: str,
quantiles: list = None) –> pd.DataFrame:
"""创建分位数特征"""
df = df.copy()

if quantiles is None:
quantiles = [0.1, 0.25, 0.5, 0.75, 0.9, 0.95, 0.99]

q_values = df[column].quantile(quantiles)

for q in quantiles:
q_col = f'{column}_q{int(q*100)}'
df[q_col] = q_values[q]
df[f'{column}_above_q{int(q*100)}'] = (df[column] > q_values[q]).astype(int)

return df

时序异常检测特征

class AnomalyFeatureGenerator:
"""异常检测特征生成器"""

def create_anomaly_features(self,
df: pd.DataFrame,
target_column: str,
group_columns: list = None,
threshold_std: float = 3.0) –> pd.DataFrame:
"""
创建异常检测相关特征

Args:
df: 输入数据
target_column: 目标列
group_columns: 分组列
threshold_std: 异常判定阈值（标准差倍数）
"""
df = df.copy()

if group_columns:
grouped = df.groupby(group_columns)[target_column]
else:
grouped = df[target_column]

# Z-Score
mean_val = grouped.transform('mean')
std_val = grouped.transform('std')
df[f'{target_column}_zscore'] = (df[target_column] – mean_val) / (std_val + 1e-6)

# 是否为异常点
df[f'{target_column}_is_anomaly'] = (
np.abs(df[f'{target_column}_zscore']) > threshold_std
).astype(int)

# 异常程度
df[f'{target_column}_anomaly_score'] = np.abs(df[f'{target_column}_zscore'])

# 偏离均值百分比
df[f'{target_column}_deviation_pct'] = (
(df[target_column] – mean_val) / (mean_val + 1)
)

return df

def create_iqr_features(self,
df: pd.DataFrame,
column: str,
group_columns: list = None) –> pd.DataFrame:
"""基于IQR的异常特征"""
df = df.copy()

if group_columns:
grouped = df.groupby(group_columns)[column]
else:
grouped = df[column]

q1 = grouped.quantile(0.25)
q3 = grouped.quantile(0.75)
iqr = q3 – q1

df[f'{column}_q1'] = df[column].map(q1)
df[f'{column}_q3'] = df[column].map(q3)
df[f'{column}_iqr'] = df[column].map(iqr)

# IQR边界
lower_bound = q1 – 1.5 * iqr
upper_bound = q3 + 1.5 * iqr

df[f'{column}_lower_bound'] = df[column].map(lower_bound)
df[f'{column}_upper_bound'] = df[column].map(upper_bound)

# 是否在边界内
df[f'{column}_within_bounds'] = (
(df[column] >= df[f'{column}_lower_bound']) &
(df[column] <= df[f'{column}_upper_bound'])
).astype(int)

return df

五、类别特征编码：从One-Hot到Target Encoding

传统编码方法

from sklearn.preprocessing import LabelEncoder, OneHotEncoder
import category_encoders as ce

class CategoricalEncoder:
"""类别特征编码器"""

def __init__(self):
self.label_encoders = {}
self.target_encoders = {}

def label_encode(self,
df: pd.DataFrame,
columns: list) –> pd.DataFrame:
"""
Label Encoding（标签编码）
适用于树模型，不适用于线性模型
"""
df = df.copy()

for col in columns:
if col not in self.label_encoders:
self.label_encoders[col] = LabelEncoder()
df[f'{col}_label_encoded'] = self.label_encoders[col].fit_transform(
df[col].astype(str)
)
else:
# 处理未见过的类别
df[f'{col}_label_encoded'] = df[col].astype(str).map(
lambda x: self.label_encoders[col].transform([x])[0]
if x in self.label_encoders[col].classes_
else –1
)

return df

def target_encode(self,
df: pd.DataFrame,
columns: list,
target_column: str,
smoothing: float = 10.0) –> pd.DataFrame:
"""
Target Encoding（目标编码）

使用目标变量的均值对类别进行编码，配合平滑避免过拟合

smoothing参数：平滑系数，越大越保守
"""
df = df.copy()

global_mean = df[target_column].mean()

for col in columns:
# 计算每个类别的统计量
agg = df.groupby(col)[target_column].agg(['mean', 'count'])

# 平滑公式：smoothed_mean = (count * mean + smoothing * global_mean) / (count + smoothing)
smoothed_mean = (
(agg['count'] * agg['mean'] + smoothing * global_mean) /
(agg['count'] + smoothing)
)

df[f'{col}_target_encoded'] = df[col].map(smoothed_mean)

# 同时创建标准差编码
std_map = df.groupby(col)[target_column].std()
df[f'{col}_target_std'] = df[col].map(std_map).fillna(0)

return df

def leave_one–out_encode(self,
df: pd.DataFrame,
columns: list,
target_column: str,
n_folds: int = 5) –> pd.DataFrame:
"""
Leave-One-Out Target Encoding

避免直接使用目标编码导致的过拟合，使用K-Fold交叉验证
"""
df = df.copy()

for col in columns:
df[f'{col}_loo_encoded'] = 0.0

# K-Fold交叉验证
kfold = KFold(n_splits=n_folds, shuffle=True, random_state=42)

for train_idx, val_idx in kfold.split(df):
# 计算训练集的编码
train_data = df.iloc[train_idx]
target_mean = train_data.groupby(col)[target_column].mean()

# 应用到验证集
df.loc[df.index[val_idx], f'{col}_loo_encoded'] = (
df.iloc[val_idx][col].map(target_mean)
)

# 用全局均值填充缺失值
global_mean = df[target_column].mean()
df[f'{col}_loo_encoded'].fillna(global_mean, inplace=True)

return df

高级编码：WOE编码

class WOEEncoder:
"""
Weight of Evidence 编码
常用于风控和信用评分场景，也可用于销量预测
"""

def fit_transform(self,
df: pd.DataFrame,
columns: list,
target_column: str) –> pd.DataFrame:
"""WOE编码"""
df = df.copy()

for col in columns:
# 计算每个类别的WOE
woe_map = {}
for category in df[col].unique():
category_data = df[df[col] == category]

# 好样本（target=1）和坏样本（target=0）的分布
events = category_data[target_column].sum() # 销量>0
non_events = len(category_data) – events # 销量=0

# 全局分布
total_events = df[target_column].sum()
total_non_events = len(df) – total_events

# 避免除零
if events / (total_events + 1e-6) == 0:
woe = –10
else:
woe = np.log(
(events / (total_events + 1e-6)) /
(non_events / (total_non_events + 1e-6) + 1e-6)
)

woe_map[category] = woe

df[f'{col}_woe'] = df[col].map(woe_map)

# IV（Information Value）计算
iv = self._calculate_iv(df, col, target_column, woe_map)
df[f'{col}_iv'] = iv

return df

def _calculate_iv(self, df, col, target, woe_map):
"""计算IV值"""
iv_total = 0

for category in df[col].unique():
category_data = df[df[col] == category]

events = category_data[target].sum()
non_events = len(category_data) – events
total_events = df[target].sum()
total_non_events = len(df) – total_events

event_dist = events / (total_events + 1e-6)
non_event_dist = non_events / (total_non_events + 1e-6)

woe = woe_map[category]
iv = (event_dist – non_event_dist) * woe
iv_total += iv

return iv_total

六、深度学习特征：Embedding与特征交叉

类别特征Embedding

import torch
import torch.nn as nn

class EmbeddingFeatureExtractor:
"""基于深度学习的Embedding特征提取"""

def __init__(self, embedding_dims: dict):
"""
Args:
embedding_dims: 字典，key为类别特征名，value为该特征的维度（唯一值数量）
"""
self.embedding_dims = embedding_dims
self.models = {}
self.embeddings = {}

def fit(self, df: pd.DataFrame, target_column: str):
"""
训练Embedding模型
使用自编码器或者简单的Embedding层来学习稠密表示
"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

for col, n_unique in self.embedding_dims.items():
# Embedding维度选择经验公式：min(50, (n_unique + 1) // 2)
embedding_dim = min(50, (n_unique + 1) // 2)

# 构建模型
model = nn.Sequential(
nn.Embedding(n_unique, embedding_dim),
nn.Flatten()
).to(device)

# 简化：直接训练一个从类别到目标的Embedding
self._train_category_embedding(
model, df[col].values, df[target_column].values, device
)

self.models[col] = model
self.embedding_dims[col] = embedding_dim

def _train_category_embedding(self, model, X, y, device, epochs=50):
"""训练简单的Embedding映射"""
model.train()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
criterion = nn.MSELoss()

X_tensor = torch.LongTensor(X).to(device)
y_tensor = torch.FloatTensor(y).to(device)

for epoch in range(epochs):
optimizer.zero_grad()
pred = model(X_tensor)
loss = criterion(pred, y_tensor)
loss.backward()
optimizer.step()

def transform(self, df: pd.DataFrame) –> pd.DataFrame:
"""提取Embedding特征"""
df = df.copy()
device = next(iter(self.models.values())).device

for col, model in self.models.items():
model.eval()
with torch.no_grad():
X_tensor = torch.LongTensor(df[col].values).to(device)
embeddings = model(X_tensor).cpu().numpy()

# 创建多列Embedding特征
for i in range(self.embedding_dims[col]):
df[f'{col}_emb_{i}'] = embeddings[:, i]

return df

TabNet中的特征选择

"""
TabNet是一种专门为表格数据设计的深度学习架构，
它内置了特征选择机制，可以自动学习哪些特征更重要。
"""

from pytorch_tabnet.tab_model import TabNetRegressor
import numpy as np

def extract_tabnet_feature_importance(X_train, y_train, feature_names):
"""
使用TabNet提取特征重要性
"""
# 准备数据
X_train = X_train.values if hasattr(X_train, 'values') else X_train
y_train = y_train.values if hasattr(y_train, 'values') else y_train

# 训练TabNet
model = TabNetRegressor(
n_d=32, # 决策预测层宽度
n_a=32, # 注意力层宽度
n_steps=5, # 决策步骤数
gamma=1.5,
n_independent=2,
n_shared=2,
lambda_sparse=1e-4,
seed=42,
optimizer_fn=torch.optim.Adam,
optimizer_params=dict(lr=2e-2),
scheduler_params=dict(step_size=10, gamma=0.9),
scheduler_fn=torch.optim.lr_scheduler.StepLR,
mask_type='sparsemax'
)

model.fit(
X_train, y_train.reshape(–1, 1),
max_epochs=100,
patience=10,
batch_size=256,
virtual_batch_size=128,
num_workers=0
)

# 获取特征重要性
feature_importance = model.feature_importances_

# 排序
importance_df = pd.DataFrame({
'feature': feature_names,
'importance': feature_importance
}).sort_values('importance', ascending=False)

return importance_df

七、特征选择：剔除冗余，保留精华

基于相关性的特征选择

class FeatureSelector:
"""特征选择器"""

def remove_highly_correlated(self,
df: pd.DataFrame,
threshold: float = 0.95,
target_column: str = None) –> pd.DataFrame:
"""
移除高度相关的特征
"""
df = df.copy()

# 计算相关矩阵
numeric_cols = df.select_dtypes(include=[np.number]).columns
corr_matrix = df[numeric_cols].corr().abs()

# 找到高度相关的特征对
upper = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(bool))
to_drop = [
column for column in upper.columns
if any(upper[column] > threshold)
]

if target_column and target_column in to_drop:
to_drop.remove(target_column)

print(f"移除 {len(to_drop)} 个高度相关的特征: {to_drop[:10]}…")
df.drop(to_drop, axis=1, inplace=True)

return df

def remove_low_variance(self,
df: pd.DataFrame,
threshold: float = 0.01) –> pd.DataFrame:
"""移除低方差特征"""
df = df.copy()

numeric_cols = df.select_dtypes(include=[np.number]).columns
variances = df[numeric_cols].var()
low_var_cols = variances[variances < threshold].index.tolist()

if low_var_cols:
print(f"移除 {len(low_var_cols)} 个低方差特征")
df.drop(low_var_cols, axis=1, inplace=True)

return df

def select_by_importance(self,
df: pd.DataFrame,
target_column: str,
model=None,
n_features: int = 50) –> pd.DataFrame:
"""
基于模型重要性选择特征
"""
df = df.copy()

if model is None:
from sklearn.ensemble import GradientBoostingRegressor
model = GradientBoostingRegressor(n_estimators=100, random_state=42)

X = df.drop(columns=[target_column])
y = df[target_column]

model.fit(X, y)

# 获取特征重要性
importance = pd.DataFrame({
'feature': X.columns,
'importance': model.feature_importances_
}).sort_values('importance', ascending=False)

# 选择Top N特征
selected_features = importance.head(n_features)['feature'].tolist()
selected_features.append(target_column)

print(f"选择了 {len(selected_features)–1} 个特征")

return df[selected_features]

八、完整的特征工程Pipeline

class FeatureEngineeringPipeline:
"""完整的特征工程流水线"""

def __init__(self, config: dict = None):
self.config = config or self._default_config()
self.time_extractor = TimeFeatureExtractor()
self.lag_generator = LagFeatureGenerator()
self.stat_generator = StatisticalFeatureGenerator()
self.cat_encoder = CategoricalEncoder()
self.feature_selector = FeatureSelector()

def _default_config(self):
return {
'group_columns': ['store', 'family'],
'target_column': 'sales',
'date_column': 'date',
'lag_periods': [1, 2, 3, 7, 14, 21, 28],
'rolling_windows': [7, 14, 28],
'correlation_threshold': 0.95,
'n_features_final': 100
}

def fit_transform(self, df: pd.DataFrame) –> pd.DataFrame:
"""完整的特征工程流程"""
print(f"开始特征工程，原始数据形状: {df.shape}")

# 1. 时间特征
df = self.time_extractor.extract_basic_features(df)
df = self.time_extractor.extract_cyclical_features(df)
print(f"时间特征后: {df.shape}")

# 2. 滞后与滑窗特征
df = self.lag_generator.create_lag_features(
df, self.config['target_column'], self.config['lag_periods']
)
df = self.lag_generator.create_rolling_features(
df, self.config['target_column'], self.config['rolling_windows']
)
df = self.lag_generator.create_momentum_features(
df, self.config['target_column'], [7, 14, 28]
)
print(f"滞后与滑窗特征后: {df.shape}")

# 3. 统计特征
df = self.stat_generator.create_group_statistics(
df, self.config['group_columns'], self.config['target_column']
)
print(f"统计特征后: {df.shape}")

# 4. 异常检测特征
anomaly_gen = AnomalyFeatureGenerator()
df = anomaly_gen.create_anomaly_features(
df, self.config['target_column'], self.config['group_columns']
)
print(f"异常检测特征后: {df.shape}")

# 5. 类别特征编码
cat_cols = ['store', 'family', 'city', 'state', 'type']
existing_cat_cols = [c for c in cat_cols if c in df.columns]
if existing_cat_cols:
df = self.cat_encoder.target_encode(
df, existing_cat_cols, self.config['target_column']
)
print(f"类别编码后: {df.shape}")

# 6. 特征选择
df = self.feature_selector.remove_highly_correlated(
df, threshold=self.config['correlation_threshold'],
target_column=self.config['target_column']
)

# 7. 移除无穷值和缺失值过多的列
df = self._clean_features(df)

print(f"最终特征数量: {df.shape[1]}")
return df

def _clean_features(self, df: pd.DataFrame) –> pd.DataFrame:
"""清理特征"""
# 替换无穷值
df = df.replace([np.inf, –np.inf], np.nan)

# 移除缺失率过高的列
missing_rate = df.isnull().mean()
high_missing_cols = missing_rate[missing_rate > 0.5].index.tolist()
if high_missing_cols:
print(f"移除 {len(high_missing_cols)} 个缺失率过高的列")
df.drop(high_missing_cols, axis=1, inplace=True)

return df

九、特征工程最佳实践与避坑指南

训练/推理一致性

这是最常见也最致命的bug——训练和推理时特征计算不一致。

class ConsistentFeatureCalculator:
"""
保证训练和推理时特征计算一致性的基类
实际项目中建议用Feast、Tecton等Feature Store
"""

def __init__(self):
self.lag_offsets = None
self.rolling_windows = None
self.category_means = None
self.fitted = False

def fit(self, df: pd.DataFrame, target_column: str, group_columns: list):
"""训练时保存所有需要复用的统计量"""
self.group_columns = group_columns
self.target_column = target_column

# 保存分组统计
self.category_means = df.groupby(group_columns)[target_column].mean().to_dict()

# 保存分位数
self.quantiles = df[target_column].quantile([0.25, 0.5, 0.75]).to_dict()

self.fitted = True
print("特征计算器已拟合")

def transform(self, df: pd.DataFrame) –> pd.DataFrame:
"""推理时使用训练时保存的统计量"""
assert self.fitted, "必须先调用fit方法"

df = df.copy()

# 使用训练时保存的均值（而不是推理时计算）
df['group_mean'] = df.set_index(self.group_columns).index.map(
lambda x: self.category_means.get(tuple(x) if isinstance(x, list) else x, 0)
)

# 异常检测使用训练时的分位数边界
q1, q3 = self.quantiles[0.25], self.quantiles[0.75]
iqr = q3 – q1
df['is_outlier'] = (
(df[self.target_column] < q1 – 1.5 * iqr) |
(df[self.target_column] > q3 + 1.5 * iqr)
).astype(int)

return df

性能优化建议

十、特征工程清单

最后，给大家一个完整的特征工程检查清单：

□ 时序特征
□ 基础时间单元（年/季/月/周/日）
□ 周期性编码（正弦/余弦）
□ 工作日/周末标记
□ 节假日标记
□ 节假日前后天数

□ 滞后特征
□ T-1, T-2, T-3 短期滞后
□ T-7, T-14, T-21 中期滞后
□ T-28, T-56, T-84 长期滞后

□ 滑窗统计
□ 7天、14天、28天窗口均值
□ 7天、14天、28天窗口标准差
□ 7天、14天、28天窗口最值
□ 7天、14天、28天窗口中位数

□ 动量特征
□ 同比变化量
□ 环比变化量
□ 增长率/下降率

□ 同期特征
□ 上周同期
□ 上月同期
□ 去年同期

□ 类别特征
□ Label Encoding
□ Target Encoding（平滑）
□ 交叉特征

□ 异常特征
□ Z-Score
□ IQR边界
□ 是否为异常点

□ 特征选择
□ 移除高相关特征（>0.95）
□ 移除低方差特征
□ 基于重要性的选择

□ 一致性检查
□ 训练/推理特征计算一致
□ 无未来信息泄露
□ 缺失值处理合理

写在最后

特征工程是机器学习中最能体现"领域知识"和"工程经验"的环节。希望本文提供的完整特征体系和代码模板，能帮助你构建更强大的预测模型。

如果觉得有帮助，欢迎一键三连！有更多问题欢迎评论区交流~