功能工程和选择
一个好功能值一千个数据点.
Type: Build
Languages: Python
Prerequisites: Phase 1 (Statistics for ML, Linear Algebra), Phase 2 Lessons 1-7
Time: ~90 minutes
学习目标
- 实施数值转换 (标准化,最小最大规模化,日志转换,结) 并解释每一个转换的适当时
- 建立一个热点,标签和目标编码的类别特征,并确定目标编码中的数据泄漏风险
- 从零开始构建TF-IDF向量表,并解释为什么它超过文本分类的原始字数
- 应用基于过器的特征选择 (变异门,相关性,相互信息) 减少维度性
问题
你有数据集,你选择一个算法,你训练它.结果是中等的.你试着一个更精彩的算法.仍然是中等的.你花了一个星期调整超参数.边缘改进.
然后有人将原始数据转化为更好的功能, 一个简单的物流回归比你的调节的梯度增强组件更好.
在经典的ML中,数据的表示比算法的选择更重要.一个"平方录像"和"卧室数量"的房价模型将比一个"地址作为原始字符串"的模型更重要,无论学习者多么复杂.算法只能根据你给出的东西工作.
功能工程是将原始数据转化为模型更容易找到的表现.功能选择是抛弃没有添加信号的噪音特性的过程.它们一起是经典 ML 中最大的杆活动.
概念
功能管道
flowchart LR
A[Raw Data] --> B[Handle Missing Values]
B --> C[Numerical Transforms]
B --> D[Categorical Encoding]
B --> E[Text Features]
C --> F[Feature Interactions]
D --> F
E --> F
F --> G[Feature Selection]
G --> H[Model-Ready Data]数字特征
基本的数字很少是模型准备的.
Scaling:设置功能在相同的范围,以使距离基于的算法 (K-Means, KNN, SVM) 对待所有功能均等.最小最大扩展地图到 [0, 1].标准化 (z-score) 地图到 mean=0, std=1.
Log transform:压缩了右倾分布 (收入,人口,字数). 转化了乘法关系为加值关系.
Binning:转换连续值为类别. 当特征与目标之间的关系非线性,但步骤式 (例如年龄组) 时有用.
Polynomial features:允许线性模型以更多的特性来捕捉非线性关系.
类别特征
模型需要数字,类别需要编码.
One-hot encoding:创建一个对每个类别的二进制列. "色 = 红/蓝/绿"变成三个列: is_red, is_blue, is_green.适用于低cardinality功能,但在许多类别中爆炸.
Label encoding:绘制每个类别的整数:红=0,蓝=1,绿=2. 引入错误的排序 (模型可能认为绿 >蓝 >红). 仅适用于基于树的模型,分为单个值.
Target encoding:强大但危险:高数据泄漏风险.只需基于训练数据计算,应应用于测试数据.
文字的特征
Count vectorizer:计算每一个词在文档中出现在多少次. "猫坐在床上"变成 {the: 2, cat: 1, sat: 1, on: 1, mat: 1}.
TF-IDF:频率-反文档频率. 根据文档中的独特性,重量词. "the"等常见词变得较低. 罕见的,独特的词变得更重.
TF(word, doc) = count(word in doc) / total words in doc
IDF(word) = log(total docs / docs containing word)
TF-IDF = TF * IDF缺失的价值观
实际数据有漏洞.
- Drop rows:只有缺失数据是罕见的,随机的
- Mean/median imputation:简单,保持分布形状 (中位数更坚固到异常)
- Mode imputation:对于类别特征
- Indicator column:在计算之前添加一个二进制列"was_this_missing".数据缺失的事实本身可以是信息性的
- Forward/backward fill:时间序列数据
功能互动
有时,这种关系是结合中的.单独的"身高"和"体重"比"BMI =体重/身高^2"更少的预测性.
功能选择
没有相关的功能会增加噪音,增加训练时间,并可能导致过度适应.
Filter methods (pre-model):
- 相关性:消除高度相互相关的特征 (冗余)
- 互通信息:衡量知道特征的程度,减少了对目标的不确定性
- 变异门:删除几乎不变的功能
Wrapper methods (model-based):
- L1规律化 (Lasso):将无关性的特征权重达到完全零
- 复发性功能消除:训练,删除最不重要的功能,重复
Why selection matters:具有10个好功能的模型通常会超过具有10个好功能和90个噪音的模型.噪音的功能使模型有机会过度适应不通用的训练数据模式.
建立它
步骤1:从零开始进行数值转换
pythonimport 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:从零开始编码类别
pythondef 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:从零开始的文字功能
pythondef 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:从零开始错失的值归因
pythondef 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:从零开始选择功能
pythondef 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: 完整的管道和演示
pythonimport 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,这些转换是可组合的管道:
pythonfrom 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- 系统地从原始数据中进行工程的提示
运动
- 加入强大的规模化 (使用中位数和四分之一间范围而不是平均和标准偏差) 数字变化.将其与极端异常值的数据的标准规模化进行比较.
- 实现单独的目标编码:对于每个行,计算目标平均值,排除该行的目标值. 展示如何减少过度适应与天真的目标编码相比.
- 建立一个自动化特征选择管道,结合变异门,相关性过和相互信息排名.将其应用到住房数据集,并将模型性能 (使用简单的线性回归) 与所有特征和选定的特征进行比较.
关键词
| Term | What people say | What it actually means |
|---|---|---|
| Feature engineering | "Making new columns" | Transforming raw data into representations that expose patterns to the model |
| Standardization | "Making it normal" | Subtracting the mean and dividing by standard deviation so the feature has mean=0 and std=1 |
| One-hot encoding | "Making dummy variables" | Creating one binary column per category, where exactly one column is 1 for each row |
| Target encoding | "Using the answer to encode" | Replacing each category with the average target value for that category, with smoothing to prevent overfitting |
| TF-IDF | "Fancy word counts" | Term Frequency times Inverse Document Frequency: words weighted by how distinctive they are across the corpus |
| Imputation | "Filling in blanks" | Replacing missing values with estimated values (mean, median, mode, or model-predicted) |
| Feature selection | "Throwing out bad columns" | Removing features that add noise or redundancy, keeping only those with signal about the target |
| Mutual information | "How much one thing tells you about another" | A measure of the reduction in uncertainty about variable Y gained by observing variable X |
| Data leakage | "Accidentally cheating" | Using information during training that would not be available at prediction time, giving falsely optimistic results |
进一步阅读
- Feature Engineering and Selection (Max Kuhn & Kjell Johnson)- 免费的网上书,涵盖了整个功能工程领域
- scikit-learn Preprocessing Guide- 对于所有标准转换的实用参考
- Target Encoding Done Right (Micci-Barreca, 2001)- 目标编码的原始文件
This free lesson is part of the AI Engineering from Scratch curriculum. Read the full explanation, run the lesson code, and verify the result in the interactive reader or from the repository source.
Browse the complete course catalog or open this lesson on GitHub.