特征工程与选择:好特征值一千个数据点

特征工程与选择:好特征值一千个数据点

本节摘要:在经典机器学习里,数据的表示比算法的选择更重要。你拿了数据集,选了算法,训练,结果平庸,换成更花哨的算法还是平庸,花一周调超参数只有微小改善——然后有人把原始数据变成更好的特征,一个简单的逻辑回归就打败了你精心调过的梯度提升集成。这种事天天发生。一个用「面积」和「卧室数」的房价模型,无论学习器多高级,都会打败用「原始地址字符串」的模型;算法只能用你给它的东西。特征工程(Feature Engineering)把原始数据转成让模型更易发现规律的表示,特征选择(Feature Selection)丢掉只增噪声不增信号的冗余特征,两者合起来是经典 ML 里杠杆率最高的活动。本节将从零实现数值变换(标准化、min-max、对数、分箱)、类别编码(one-hot、标签、目标编码及其泄漏风险)、文本特征(TF-IDF)、缺失值填补,以及过滤式特征选择(方差阈值、相关性、互信息)。

学习目标

阅读完本节,你应当能够:

1. 从零实现数值变换(标准化、min-max 缩放、对数变换、分箱),说明各自的适用场景。
2. 为类别特征构建 one-hot、标签、目标编码,识别目标编码的数据泄漏风险。
3. 从零构造 TF-IDF 向量化器,解释它为何在文本分类上胜过原始词频。
4. 应用过滤式特征选择(方差阈值、相关性、互信息)降低维度。

一、问题与直觉

你有个数据集,选了算法,训练,结果平庸。换了更花哨的算法,还是平庸。花一周调超参数,边际改善。然后有人把原始数据变成更好的特征,一个简单的逻辑回归就打败了你调过的梯度提升集成。

这种事天天发生。在经典 ML 里,数据表示比算法选择更重要。一个用「面积」和「卧室数」的房价模型,无论学习器多高级,都会打败用「原始地址字符串」的模型。算法只能用你给它的东西。

特征工程把原始数据转成让模型更易发现规律的表示。特征选择丢掉只增噪声不增信号的特征。两者合起来是经典 ML 里杠杆率最高的活动。

特征流水线

数值特征

原始数字很少能直接喂模型。常见变换:

缩放:把特征放到同一范围,让基于距离的算法(K-Means、KNN、SVM)平等对待所有特征。min-max 缩放映射到 [0, 1],标准化(z 分数)映射到均值 0、标准差 1。

对数变换:压缩右偏分布(收入、人口、词频),把乘性关系变成加性。

分箱:把连续值变成类别。当特征与目标的关系是非线性但阶梯式时(如年龄段)有用。

多项式特征:造 x²、x³、x1*x2 项,让线性模型捕捉非线性关系,代价是特征变多。

类别特征

模型要数字,类别要编码。

one-hot 编码:每个类别一列二值。「color = 红/蓝/绿」变成 is_red、is_blue、is_green 三列。低基数特征好用,类别一多就爆炸。

标签编码:每个类别映射到一个整数:红=0、蓝=1、绿=2。引入虚假序(模型可能以为绿 > 蓝 > 红),只适合按单个值切分的树模型。

目标编码:每个类别用该类别的目标均值替换。强大但危险:数据泄漏风险高。只能在训练数据上算,再套用到测试数据。

文本特征

计数向量化:数每个词在文档里出现几次。「the cat sat on the mat」变成 {the: 2, cat: 1, sat: 1, on: 1, mat: 1}。

TF-IDF:词频-逆文档频率。按词在跨文档中的独特程度加权。常见词如「the」权重低,稀有且独特的词权重高。

TF(word, doc) = count(word in doc) / doc 总词数
IDF(word) = log(总文档数 / 含该词的文档数)
TF-IDF = TF * IDF
​

缺失值

真实数据有洞。策略:

• 删行:仅在缺失稀少且随机时
• 均值/中位数填补:简单,保分布形状(中位数更抗离群点)
• 众数填补:类别特征用
• 指示列:填补前加一列二值「这个是否缺失」。缺失这个事实本身可能就有信息
• 前后向填充:时间序列用

特征交互

有时关系藏在组合里。「身高」和「体重」单独不如「BMI = 体重 / 身高²」有预测力。特征交互让特征空间倍增,要用领域知识挑对的。

特征选择

特征不是越多越好。无关特征加噪声、增训练时间、引发过拟合。

过滤法(建模型前):

• 相关性:删彼此高度相关的特征(冗余)
• 互信息:度量知道一个特征能减少多少对目标的不确定
• 方差阈值:删几乎不变的特征

包装法(基于模型):

• L1 正则(Lasso):把无关特征权重驱动到恰好零
• 递归特征消除:训练、删最不重要的特征、重复

选择为何重要:一个 10 个好特征的模型,通常胜过一个 10 个好特征加 90 个噪声特征的模型。噪声特征给模型在训练数据上过拟合不泛化模式的机会。

二、从零实现

第 1 步:从零写数值变换

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 步:从零写类别编码

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
​

⚠️ 目标编码的泄漏陷阱:必须只用训练集的目标值算编码,再套到验证/测试集。若用了全量数据,目标信息就泄漏进特征里,验证分数虚高。smoothing 参数让样本少的类别向全局均值收缩,防过拟合。

第 3 步:从零写文本特征

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 步:从零写缺失值填补

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 步:从零写特征选择

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 步:完整流水线与演示

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
​

完整演示(跑通缺失值处理、数值变换、类别编码、文本特征、多项式特征、特征选择)见 code/feature_engineering.py。

三、框架对比

用 scikit-learn,这些变换可拼成可组合流水线:

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 —— 一个系统化从原始数据做特征工程的提示词。

从零变换函数集合(code/feature_engineering.py)可作为「无依赖」特征工具箱,在不能用 sklearn 的受限环境(如某些线上推理服务)里复用。

五、练习

1. 给数值变换加稳健缩放(用中位数和四分位距代替均值和标准差),在有极端离群点的数据上与标准缩放对比。
2. 实现留一目标编码:对每一行,算目标均值时排除该行自己的目标值。展示它如何比朴素目标编码降低过拟合。
3. 构建自动特征选择流水线,组合方差阈值、相关性过滤、互信息排名,套到房价数据上,用简单线性回归对比全特征与选后特征的模型表现。

本节要点回顾

1. 表示比算法重要:好特征让简单模型打败花哨算法,算法只能用你给的东西。
2. 数值变换按需选:标准化(均值 0、std 1)适合距离算法,min-max 映 [0,1],对数压右偏,分箱处理阶梯式非线性,多项式让线性模型学非线性。
3. 类别编码看基数:one-hot 低基数好用、标签编码只适合树(虚假序)、目标编码强大但高泄漏风险且必须只用训练集算。
4. TF-IDF 比词频强:词频乘逆文档频率,常见词权重低、稀有独特词权重高,文本分类首选。
5. 缺失值策略化:删行(稀少随机)、均值/中位数/众数填补、加缺失指示列(缺失本身是信号)、时间序列用前后向填充。
6. 交互藏组合里:BMI = 体重/身高² 比单看身高体重强,用领域知识挑对的。
7. 特征选择两路:过滤法(方差阈值、相关性、互信息)建模型前用,包装法(Lasso、递归消除)基于模型。
8. 少即是多:10 个好特征胜过 10 好 + 90 噪声,噪声给过拟合可乘之机。
9. 库版同数学加工程:sklearn 用 ColumnTransformer+Pipeline 拼可组合流水线,加稀疏、边界、持久化,但底层变换一致。

下一节,我们把镜头转向评估——准确率不够用,精确率、召回率、ROC、AUC、交叉验证如何告诉你模型真实的表现。