自动化特征工程工具实战指南：Featuretools、AutoFeat与tsfresh全面教程（2026版）

import featuretools as ft

# 创建EntitySet
es = ft.EntitySet(id="retail")

# 添加用户实体
es = es.add_dataframe(
    dataframe_name="users",
    dataframe=users_df,
    index="user_id",
    time_index="signup_date"
)

# 添加订单实体，关联到用户
es = es.add_dataframe(
    dataframe_name="orders",
    dataframe=orders_df,
    index="order_id",
    time_index="order_date"
)
es = es.add_relationship("users", "user_id", "orders", "user_id")

# 定义自定义primitives
agg_primitives = ["sum", "mean", "max", "min", "count", "std", "trend"]
trans_primitives = ["year", "month", "day", "diff"]

# 执行深度特征合成
feature_matrix, feature_defs = ft.dfs(
    entityset=es,
    target_dataframe_name="users",       # 为用户生成特征
    agg_primitives=agg_primitives,       # 聚合操作
    trans_primitives=trans_primitives,   # 变换操作
    max_depth=2,                         # 特征合成深度
    cutoff_time=cutoff_times             # 时序截止点，防止数据泄漏
)

cutoff_times = pd.DataFrame({
    'user_id': [1, 2, 3],
    'time': pd.to_datetime(['2026-01-15', '2026-01-15', '2026-01-15'])
})

# 所有特征只基于2026-01-15之前的数据计算
feature_matrix, _ = ft.dfs(entityset=es, target_dataframe_name="users",
                           cutoff_time=cutoff_times)

# 保存特征定义
ft.save_features(feature_defs, "./feature_definitions.json")

# 后续加载并应用到新数据
loaded_defs = ft.load_features("./feature_definitions.json")
new_features = ft.calculate_feature_matrix(loaded_defs, entityset=new_es)

from autofeat import AutoFeatRegressor

# 一步完成特征生成+选择+模型训练
model = AutoFeatRegressor(
    feateng_steps=2,      # 特征组合深度
    max_gb=16,            # 最大内存使用
    units=["log", "sqrt", "square", "exp"]  # 允许的数学变换
)

# 自动拟合：内部完成特征工程+特征选择+回归
y_pred = model.fit_transform(X_train, y_train)

# 查看生成的特征名称
print(model.new_features_fc_)

from tsfresh import extract_features, select_features
from tsfresh.utilities.dataframe_functions import impute

# 假设df格式：id, time, value, label
# id=实体标识, time=时间戳, value=观测值

# Step 1: 自动提取800+特征
extracted = extract_features(
    df,
    column_id="sensor_id",      # 实体列
    column_sort="timestamp",    # 时间列
    column_value="reading",     # 值列
    default_fc_parameters="comprehensive"  # 提取全部特征
)

# Step 2: 缺失值处理
extracted = impute(extracted)

# Step 3: FRESH算法筛选相关特征（需标签）
selected = select_features(extracted, y)

print(f"原始特征数: {extracted.shape[1]}")
print(f"筛选后特征数: {selected.shape[1]}")

from sklearn.pipeline import Pipeline
from sklearn.ensemble import RandomForestClassifier
from tsfresh.transformers import RelevantFeatureAugmenter

pipeline = Pipeline([
    ("augmenter", RelevantFeatureAugmenter(column_id="id", column_sort="time")),
    ("classifier", RandomForestClassifier())
])

pipeline.fit(X_train, y_train)
predictions = pipeline.predict(X_test)

import dask.dataframe as dd
from dask.diagnostics import ProgressBar

# 用Dask DataFrame替代Pandas DataFrame
dask_users = dd.read_parquet("users/*.parquet")
dask_orders = dd.read_parquet("orders/*.parquet")

es = ft.EntitySet(id="big_retail")
es = es.add_dataframe("users", dask_users, index="user_id")
es = es.add_dataframe("orders", dask_orders, index="order_id")

# DFS自动利用Dask分布式计算
feature_matrix = ft.dfs(entityset=es, ...)

extract_features(
    df,
    column_id="sensor_id",
    column_sort="timestamp",
    n_jobs=8,           # 使用8个并行进程
    chunksize=1000,     # 每个进程处理1000个时间序列
    distributor=None    # 或使用dask/ray distributor
)

import featuretools as ft
from sklearn.model_selection import train_test_split
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import roc_auc_score

# ========== 数据准备 ==========
X = transactions_df.drop('is_fraud', axis=1)
y = transactions_df['is_fraud']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# ========== 基线模型（原始特征） ==========
clf_base = GradientBoostingClassifier(n_estimators=100)
clf_base.fit(X_train, y_train)
auc_base = roc_auc_score(y_test, clf_base.predict_proba(X_test)[:, 1])
print(f"基线AUC: {auc_base:.4f}")  # 如 0.8234

# ========== Featuretools增强 ==========
es = ft.EntitySet(id="fraud")
es = es.add_dataframe("transactions", X_train, index="transaction_id")
es = es.add_dataframe("cards", cards_df, index="card_id")
es.add_relationship("cards", "card_id", "transactions", "card_id")

fm, features = ft.dfs(entityset=es, target_dataframe_name="transactions",
                      agg_primitives=["sum", "mean", "count", "std", "max", "min"],
                      trans_primitives=["day", "hour", "month"],
                      max_depth=2)

# 测试集特征用相同特征定义生成
fm_test = ft.calculate_feature_matrix(features, entityset=es_test)

clf_enhanced = GradientBoostingClassifier(n_estimators=100)
clf_enhanced.fit(fm, y_train)
auc_enhanced = roc_auc_score(y_test, clf_enhanced.predict_proba(fm_test)[:, 1])
print(f"增强AUC: {auc_enhanced:.4f}")  # 如 0.8912

print(f"特征数: {X_train.shape[1]} → {fm.shape[1]}")
print(f"AUC提升: +{(auc_enhanced - auc_base):.4f}")

from sklearn.base import BaseEstimator, TransformerMixin

class FeaturetoolsTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, entityset, target, primitives, max_depth=2):
        self.entityset = entityset
        self.target = target
        self.primitives = primitives
        self.max_depth = max_depth
        self.features = None
    
    def fit(self, X, y=None):
        _, self.features = ft.dfs(entityset=self.entityset, ...)
        return self
    
    def transform(self, X):
        return ft.calculate_feature_matrix(self.features, entityset=self.entityset)

# 集成到sklearn Pipeline
pipeline = Pipeline([
    ('feat_eng', FeaturetoolsTransformer(es, 'users', [...])),
    ('classifier', RandomForestClassifier())
])