特征工程 (Feature Engineering) 与特征选择 (Feature Selection)
一个好特征胜过一千个数据点。
类型： 构建 语言： Python 前置知识： 第 1 阶段（机器学习统计、线性代数），第 2 阶段第 1-7 课 时间： ~90 分钟
学习目标
- 实现数值变换 (numerical transforms)（标准化 (standardization)、最小-最大缩放 (min-max scaling)、对数变换 (log transform)、分箱 (binning)），并说明各自适用的场景
- 构建分类特征 (categorical features) 的独热编码 (one-hot encoding)、标签编码 (label encoding) 和目标编码 (target encoding)，并识别目标编码中的数据泄漏 (data leakage) 风险
- 从零构建一个 TF-IDF 向量化器 (vectorizer)，并解释它为什么比原始词频计数更适合文本分类
- 应用基于过滤器的特征选择（方差阈值、相关性、互信息 (mutual information)）来降低维度
问题
你有一个数据集。你选了一个算法。你训练它。结果平平。你换一个更花哨的算法。还是平平。你花一周调超参数。提升依旧有限。
然后有人把原始数据变成了更好的特征，一个简单的逻辑回归就打败了你精调过的梯度提升集成模型。
这种事一直都在发生。在经典机器学习里，数据的表示方式往往比算法的选择更重要。一个拥有“面积”和“卧室数量”这类特征的房价模型，不管学习器多普通，都会胜过一个把“地址原样字符串”直接喂进去的模型。算法只能处理你提供给它的表示。
特征工程是把原始数据转换成更容易让模型发现模式的表示形式的过程。特征选择是把那些只会增加噪声、不会增加信号的特征丢掉的过程。两者结合起来，是经典机器学习中杠杆效应最高的工作。
概念
特征流水线
mermaid
flowchart LR
A[原始数据] --> B[处理缺失值]
B --> C[数值变换]
B --> D[类别编码]
B --> E[文本特征]
C --> F[特征交互]
D --> F
E --> F
F --> G[特征选择]
G --> H[可供模型使用的数据]数值特征
原始数字很少能直接用于建模。常见变换有：
缩放： 把各个特征放到相近的范围里，这样基于距离的算法（K-Means、KNN、SVM）才会平等对待所有特征。最小-最大缩放会把值映射到 [0, 1]。标准化（z-score）会把特征变成 mean=0、std=1。
对数变换： 压缩右偏分布（收入、人口、词频计数）。它还能把乘法关系变成加法关系。
分箱： 把连续值转换成类别。当特征与目标之间的关系不是线性的，而是分段台阶式的（例如年龄段）时，这很有用。
多项式特征 (polynomial features)： 创建 x^2、x^3、x1*x2 这样的项。它让线性模型也能捕捉非线性关系，但代价是特征数量会增加。
类别特征
模型需要数字。类别需要编码。
独热编码： 为每个类别创建一个二元列。“color = red/blue/green”会变成三列：is_red、is_blue、is_green。它很适合低基数特征，但类别一多就会迅速膨胀。
标签编码： 把每个类别映射成一个整数：red=0、blue=1、green=2。这会引入虚假的顺序关系（模型可能会以为 green > blue > red）。它只适用于会按单个取值进行切分的树模型。
目标编码： 用该类别对应目标变量的平均值替换该类别。它很强大，但也很危险：数据泄漏风险很高。必须只在训练数据上计算，并应用到测试数据上。
文本特征
词频向量化器 (count vectorizer)： 统计每个单词在文档里出现了多少次。“the cat sat on the mat”会变成 {the: 2, cat: 1, sat: 1, on: 1, mat: 1}。
TF-IDF： Term Frequency-Inverse Document Frequency。它会根据词在不同文档中的独特程度来赋权。像 “the” 这样的常见词权重很低。罕见且有辨识度的词权重很高。
TF(word, doc) = count(word in doc) / total words in doc
IDF(word) = log(total docs / docs containing word)
TF-IDF = TF * IDF缺失值
真实数据总是有空洞。常见策略：
- 删除行： 只在缺失很少而且是随机缺失时使用
- 均值/中位数插补： 简单有效，能保留分布形状（中位数对离群点更稳健）
- 众数插补： 适用于类别特征
- 指示列： 在插补前加一个二元列 “was_this_missing”。“缺失”这件事本身也可能包含信息
- 前向/后向填充： 适用于时间序列数据
特征交互
有时真正的关系藏在组合里。单独的“身高”和“体重”不如“BMI = weight / height^2”更有预测力。特征交互会扩大特征空间，因此要用领域知识来挑选真正有意义的组合。
特征选择
特征越多并不总是越好。无关特征会增加噪声、延长训练时间，还可能导致过拟合。
过滤式方法（建模前）：
- 相关性：移除彼此高度相关的特征（冗余）
- 互信息：衡量知道某个特征后，目标变量的不确定性减少了多少
- 方差阈值：移除几乎不变化的特征
包装式方法（基于模型）：
- L1 正则化（Lasso）：把无关特征的权重直接压到 0
- 递归特征消除：训练、删除最不重要的特征、重复
为什么选择很重要： 一个拥有 10 个好特征的模型，通常会胜过一个拥有同样 10 个好特征外加 90 个噪声特征的模型。噪声特征会给模型提供更多机会，让它在训练数据上过拟合那些无法泛化的模式。
动手实现
第 1 步：从零实现数值变换
python
import math
def min_max_scale(values):
min_val = min(values)
max_val = max(values)
if max_val == min_val:
return [0.0] * len(values)
return [(v - min_val) / (max_val - min_val) for v in values]
def standardize(values):
n = len(values)
mean = sum(values) / n
variance = sum((v - mean) ** 2 for v in values) / n
std = math.sqrt(variance) if variance > 0 else 1.0
return [(v - mean) / std for v in values]
def log_transform(values):
return [math.log(v + 1) for v in values]
def bin_values(values, n_bins=5):
min_val = min(values)
max_val = max(values)
bin_width = (max_val - min_val) / n_bins
if bin_width == 0:
return [0] * len(values)
result = []
for v in values:
bin_idx = int((v - min_val) / bin_width)
bin_idx = min(bin_idx, n_bins - 1)
result.append(bin_idx)
return result
def polynomial_features(row, degree=2):
n = len(row)
result = list(row)
if degree >= 2:
for i in range(n):
result.append(row[i] ** 2)
for i in range(n):
for j in range(i + 1, n):
result.append(row[i] * row[j])
return result第 2 步：从零实现类别编码
python
def one_hot_encode(values):
categories = sorted(set(values))
cat_to_idx = {cat: i for i, cat in enumerate(categories)}
n_cats = len(categories)
encoded = []
for v in values:
row = [0] * n_cats
row[cat_to_idx[v]] = 1
encoded.append(row)
return encoded, categories
def label_encode(values):
categories = sorted(set(values))
cat_to_int = {cat: i for i, cat in enumerate(categories)}
return [cat_to_int[v] for v in values], cat_to_int
def target_encode(feature_values, target_values, smoothing=10):
global_mean = sum(target_values) / len(target_values)
category_stats = {}
for feat, target in zip(feature_values, target_values):
if feat not in category_stats:
category_stats[feat] = {"sum": 0.0, "count": 0}
category_stats[feat]["sum"] += target
category_stats[feat]["count"] += 1
encoding = {}
for cat, stats in category_stats.items():
cat_mean = stats["sum"] / stats["count"]
weight = stats["count"] / (stats["count"] + smoothing)
encoding[cat] = weight * cat_mean + (1 - weight) * global_mean
return [encoding[v] for v in feature_values], encoding第 3 步：从零实现文本特征
python
def count_vectorize(documents):
vocab = {}
idx = 0
for doc in documents:
for word in doc.lower().split():
if word not in vocab:
vocab[word] = idx
idx += 1
vectors = []
for doc in documents:
vec = [0] * len(vocab)
for word in doc.lower().split():
vec[vocab[word]] += 1
vectors.append(vec)
return vectors, vocab
def tfidf(documents):
n_docs = len(documents)
vocab = {}
idx = 0
for doc in documents:
for word in doc.lower().split():
if word not in vocab:
vocab[word] = idx
idx += 1
doc_freq = {}
for doc in documents:
seen = set()
for word in doc.lower().split():
if word not in seen:
doc_freq[word] = doc_freq.get(word, 0) + 1
seen.add(word)
vectors = []
for doc in documents:
words = doc.lower().split()
word_count = len(words)
tf_map = {}
for word in words:
tf_map[word] = tf_map.get(word, 0) + 1
vec = [0.0] * len(vocab)
for word, count in tf_map.items():
tf = count / word_count
idf = math.log(n_docs / doc_freq[word])
vec[vocab[word]] = tf * idf
vectors.append(vec)
return vectors, vocab第 4 步：从零实现缺失值插补
python
def impute_mean(values):
present = [v for v in values if v is not None]
if not present:
return [0.0] * len(values), 0.0
mean = sum(present) / len(present)
return [v if v is not None else mean for v in values], mean
def impute_median(values):
present = sorted(v for v in values if v is not None)
if not present:
return [0.0] * len(values), 0.0
n = len(present)
if n % 2 == 0:
median = (present[n // 2 - 1] + present[n // 2]) / 2
else:
median = present[n // 2]
return [v if v is not None else median for v in values], median
def impute_mode(values):
present = [v for v in values if v is not None]
if not present:
return values, None
counts = {}
for v in present:
counts[v] = counts.get(v, 0) + 1
mode = max(counts, key=counts.get)
return [v if v is not None else mode for v in values], mode
def add_missing_indicator(values):
return [0 if v is not None else 1 for v in values]第 5 步：从零实现特征选择
python
def correlation(x, y):
n = len(x)
mean_x = sum(x) / n
mean_y = sum(y) / n
cov = sum((xi - mean_x) * (yi - mean_y) for xi, yi in zip(x, y)) / n
std_x = math.sqrt(sum((xi - mean_x) ** 2 for xi in x) / n)
std_y = math.sqrt(sum((yi - mean_y) ** 2 for yi in y) / n)
if std_x == 0 or std_y == 0:
return 0.0
return cov / (std_x * std_y)
def mutual_information(feature, target, n_bins=10):
feat_min = min(feature)
feat_max = max(feature)
bin_width = (feat_max - feat_min) / n_bins if feat_max != feat_min else 1.0
feat_binned = [
min(int((f - feat_min) / bin_width), n_bins - 1) for f in feature
]
n = len(feature)
target_classes = sorted(set(target))
feat_bins = sorted(set(feat_binned))
p_feat = {}
for b in feat_bins:
p_feat[b] = feat_binned.count(b) / n
p_target = {}
for t in target_classes:
p_target[t] = target.count(t) / n
mi = 0.0
for b in feat_bins:
for t in target_classes:
joint_count = sum(
1 for fb, tv in zip(feat_binned, target) if fb == b and tv == t
)
p_joint = joint_count / n
if p_joint > 0:
mi += p_joint * math.log(p_joint / (p_feat[b] * p_target[t]))
return mi
def variance_threshold(features, threshold=0.01):
n_features = len(features[0])
n_samples = len(features)
selected = []
for j in range(n_features):
col = [features[i][j] for i in range(n_samples)]
mean = sum(col) / n_samples
var = sum((v - mean) ** 2 for v in col) / n_samples
if var >= threshold:
selected.append(j)
return selected
def remove_correlated(features, threshold=0.9):
n_features = len(features[0])
n_samples = len(features)
to_remove = set()
for i in range(n_features):
if i in to_remove:
continue
col_i = [features[r][i] for r in range(n_samples)]
for j in range(i + 1, n_features):
if j in to_remove:
continue
col_j = [features[r][j] for r in range(n_samples)]
corr = abs(correlation(col_i, col_j))
if corr >= threshold:
to_remove.add(j)
return [i for i in range(n_features) if i not in to_remove]第 6 步：完整流水线与演示
python
import random
def make_housing_data(n=200, seed=42):
random.seed(seed)
data = []
for _ in range(n):
sqft = random.uniform(500, 5000)
bedrooms = random.choice([1, 2, 3, 4, 5])
age = random.uniform(0, 50)
neighborhood = random.choice(["downtown", "suburbs", "rural"])
has_pool = random.choice([True, False])
sqft_with_missing = sqft if random.random() > 0.05 else None
age_with_missing = age if random.random() > 0.08 else None
price = (
50 * sqft
+ 20000 * bedrooms
- 1000 * age
+ (50000 if neighborhood == "downtown" else 10000 if neighborhood == "suburbs" else 0)
+ (15000 if has_pool else 0)
+ random.gauss(0, 20000)
)
data.append({
"sqft": sqft_with_missing,
"bedrooms": bedrooms,
"age": age_with_missing,
"neighborhood": neighborhood,
"has_pool": has_pool,
"price": price,
})
return data
if __name__ == "__main__":
data = make_housing_data(200)
print("=== Raw Data Sample ===")
for row in data[:3]:
print(f" {row}")
sqft_raw = [d["sqft"] for d in data]
age_raw = [d["age"] for d in data]
prices = [d["price"] for d in data]
print("\n=== Missing Value Handling ===")
sqft_missing = sum(1 for v in sqft_raw if v is None)
age_missing = sum(1 for v in age_raw if v is None)
print(f" sqft missing: {sqft_missing}/{len(sqft_raw)}")
print(f" age missing: {age_missing}/{len(age_raw)}")
sqft_indicator = add_missing_indicator(sqft_raw)
age_indicator = add_missing_indicator(age_raw)
sqft_imputed, sqft_fill = impute_median(sqft_raw)
age_imputed, age_fill = impute_mean(age_raw)
print(f" sqft filled with median: {sqft_fill:.0f}")
print(f" age filled with mean: {age_fill:.1f}")
print("\n=== Numerical Transforms ===")
sqft_scaled = standardize(sqft_imputed)
age_scaled = min_max_scale(age_imputed)
sqft_log = log_transform(sqft_imputed)
age_binned = bin_values(age_imputed, n_bins=5)
print(f" sqft standardized: mean={sum(sqft_scaled)/len(sqft_scaled):.4f}, std={math.sqrt(sum(v**2 for v in sqft_scaled)/len(sqft_scaled)):.4f}")
print(f" age min-max: [{min(age_scaled):.2f}, {max(age_scaled):.2f}]")
print(f" age bins: {sorted(set(age_binned))}")
print("\n=== Categorical Encoding ===")
neighborhoods = [d["neighborhood"] for d in data]
ohe, ohe_cats = one_hot_encode(neighborhoods)
print(f" One-hot categories: {ohe_cats}")
print(f" Sample encoding: {neighborhoods[0]} -> {ohe[0]}")
le, le_map = label_encode(neighborhoods)
print(f" Label encoding map: {le_map}")
te, te_map = target_encode(neighborhoods, prices, smoothing=10)
print(f" Target encoding: {({k: round(v) for k, v in te_map.items()})}")
print("\n=== Text Features ===")
descriptions = [
"large modern house with pool",
"small cozy cottage near downtown",
"spacious family home with large yard",
"modern apartment downtown with view",
"rustic cabin in rural area",
]
cv, cv_vocab = count_vectorize(descriptions)
print(f" Vocabulary size: {len(cv_vocab)}")
print(f" Doc 0 non-zero features: {sum(1 for v in cv[0] if v > 0)}")
tf, tf_vocab = tfidf(descriptions)
print(f" TF-IDF vocabulary size: {len(tf_vocab)}")
top_words = sorted(tf_vocab.keys(), key=lambda w: tf[0][tf_vocab[w]], reverse=True)[:3]
print(f" Doc 0 top TF-IDF words: {top_words}")
print("\n=== Polynomial Features ===")
sample_row = [sqft_scaled[0], age_scaled[0]]
poly = polynomial_features(sample_row, degree=2)
print(f" Input: {[round(v, 4) for v in sample_row]}")
print(f" Polynomial: {[round(v, 4) for v in poly]}")
print(f" Features: [x1, x2, x1^2, x2^2, x1*x2]")
print("\n=== Feature Selection ===")
feature_matrix = [
[sqft_scaled[i], age_scaled[i], float(sqft_indicator[i]), float(age_indicator[i])]
+ ohe[i]
for i in range(len(data))
]
print(f" Total features: {len(feature_matrix[0])}")
surviving_var = variance_threshold(feature_matrix, threshold=0.01)
print(f" After variance threshold (0.01): {len(surviving_var)} features kept")
surviving_corr = remove_correlated(feature_matrix, threshold=0.9)
print(f" After correlation filter (0.9): {len(surviving_corr)} features kept")
binary_prices = [1 if p > sum(prices) / len(prices) else 0 for p in prices]
print("\n Mutual information with target:")
feature_names = ["sqft", "age", "sqft_missing", "age_missing"] + [f"neigh_{c}" for c in ohe_cats]
for j in range(len(feature_matrix[0])):
col = [feature_matrix[i][j] for i in range(len(feature_matrix))]
mi = mutual_information(col, binary_prices, n_bins=10)
print(f" {feature_names[j]}: MI={mi:.4f}")
print("\n Correlation with price:")
for j in range(len(feature_matrix[0])):
col = [feature_matrix[i][j] for i in range(len(feature_matrix))]
corr = correlation(col, prices)
print(f" {feature_names[j]}: r={corr:.4f}")实际使用
在 scikit-learn 中，这些变换可以组合成流水线：
python
from sklearn.preprocessing import StandardScaler, OneHotEncoder, PolynomialFeatures
from sklearn.impute import SimpleImputer
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.feature_selection import mutual_info_classif, VarianceThreshold
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
numeric_pipe = Pipeline([
("imputer", SimpleImputer(strategy="median")),
("scaler", StandardScaler()),
])
categorical_pipe = Pipeline([
("encoder", OneHotEncoder(sparse_output=False)),
])
preprocessor = ColumnTransformer([
("num", numeric_pipe, ["sqft", "age"]),
("cat", categorical_pipe, ["neighborhood"]),
])这些从零实现的版本准确展示了每个变换内部实际发生的事。库版本补上了边界情况处理、稀疏矩阵支持以及流水线组合能力，但数学本质是一样的。
交付成果
本课会产出：
outputs/prompt-feature-engineer.md- 一个用于系统化地从原始数据中进行特征工程的提示词
练习
- 在数值变换中加入稳健缩放（使用中位数和四分位距，而不是均值和标准差）。把它与标准缩放在存在极端离群点的数据上进行比较。
- 实现留一法目标编码：对每一行，计算去掉该行自身目标值后的目标均值。展示它相较于朴素目标编码如何降低过拟合。
- 构建一个自动化特征选择流水线，把方差阈值、相关性过滤和互信息排序结合起来。将它应用到房屋数据集上，并比较“全部特征”和“筛选后特征”下的模型表现（使用一个简单的线性回归）。
关键术语
| 术语 | 人们常说 | 实际含义 |
|---|---|---|
| 特征工程 | “新建几列” | 把原始数据转换为能让模型暴露出模式的表示形式 |
| 标准化 | “把它变正常” | 减去均值再除以标准差，使特征满足 mean=0、std=1 |
| 独热编码 | “做哑变量” | 为每个类别创建一个二元列，并且每一行恰好只有一列为 1 |
| 目标编码 | “用答案来编码” | 用该类别的平均目标值替换它，并通过平滑降低过拟合 |
| TF-IDF | “高级版词频计数” | 词频乘以逆文档频率：词语会按其在整个语料中的辨识度加权 |
| 插补 | “把空白补上” | 用估计值（均值、中位数、众数或模型预测值）替换缺失值 |
| 特征选择 | “把差列删掉” | 去除增加噪声或冗余的特征，只保留对目标有信号的特征 |
| 互信息 | “一个东西能告诉你另一个多少信息” | 观察变量 X 后，对变量 Y 不确定性减少程度的度量 |
| 数据泄漏 | “不小心作弊了” | 在训练时使用了预测时本不该可用的信息，导致结果看起来过于乐观 |
延伸阅读
- Feature Engineering and Selection (Max Kuhn & Kjell Johnson) - 覆盖特征工程全景的免费在线书籍
- scikit-learn Preprocessing Guide - 各类标准变换的实用参考
- Target Encoding Done Right (Micci-Barreca, 2001) - 关于带平滑目标编码的经典论文