人工智能之核心基础 机器学习 第十五章 数据预处理

import pandas as pd
df = pd.read_csv('data.csv')
print(df.info())        # 数据类型、非空数
print(df.describe())    # 数值特征统计（均值、std、min、max等）
print(df['category'].value_counts())  # 类别分布

# 缺失值比例
print(df.isnull().mean() * 100)

# 异常值：用箱线图快速查看
import matplotlib.pyplot as plt
df.boxplot(column=['income', 'age'])
plt.show()

# 数值特征：直方图看是否偏斜
df['income'].hist(bins=30)
plt.title("收入分布")
plt.show()

# 类别特征：条形图
df['city'].value_counts().plot(kind='bar')
plt.show()

from sklearn.impute import SimpleImputer, KNNImputer

# 均值/中位数/众数填充
num_imputer = SimpleImputer(strategy='median')      # 数值用中位数（抗异常值）
cat_imputer = SimpleImputer(strategy='most_frequent')  # 类别用众数

# 插值填充（时间序列）
df['price'].interpolate(method='linear', inplace=True)

# KNN模型填充（利用相似样本）
knn_imputer = KNNImputer(n_neighbors=5)
X_filled = knn_imputer.fit_transform(X)

# IQR方法
Q1 = df['income'].quantile(0.25)
Q3 = df['income'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 标记异常值
df['is_outlier'] = (df['income'] < lower_bound) | (df['income'] > upper_bound)

from sklearn.preprocessing import LabelEncoder, OneHotEncoder
from category_encoders import TargetEncoder

# 标签编码（用于树模型）
le = LabelEncoder()
df['city_encoded'] = le.fit_transform(df['city'])

# 独热编码（用于线性模型）
ohe = OneHotEncoder(sparse_output=False)
city_ohe = ohe.fit_transform(df[['city']])

# 目标编码（需有标签y）
te = TargetEncoder()
df['city_target'] = te.fit_transform(df[['city']], y)

# 等宽分箱
df['age_bin'] = pd.cut(df['age'], bins=5, labels=False)

# 等频分箱（每箱样本数相同）
df['income_qbin'] = pd.qcut(df['income'], q=4, labels=False)

from sklearn.preprocessing import StandardScaler, MinMaxScaler

# 标准化（推荐默认）
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 归一化
minmax = MinMaxScaler()
X_minmax = minmax.fit_transform(X)

from sklearn.feature_selection import VarianceThreshold, SelectKBest, f_classif, RFE
from sklearn.linear_model import LogisticRegression

# 过滤式：低方差特征删除
vt = VarianceThreshold(threshold=0.1)
X_vt = vt.fit_transform(X)

# 过滤式：基于统计检验
skb = SelectKBest(score_func=f_classif, k=10)
X_skb = skb.fit_transform(X, y)

# 包裹式：递归特征消除
estimator = LogisticRegression()
rfe = RFE(estimator, n_features_to_select=10)
X_rfe = rfe.fit_transform(X, y)

# 嵌入式：L1正则化（Lasso）
lasso = LogisticRegression(penalty='l1', solver='liblinear')
lasso.fit(X_scaled, y)
selected = lasso.coef_ != 0

# 图像增强示例（用于对比学习）
from torchvision import transforms

augment = transforms.Compose([
    transforms.RandomResizedCrop(224, scale=(0.2, 1.0)),
    transforms.RandomHorizontalFlip(),
    transforms.ColorJitter(0.4, 0.4, 0.4, 0.1),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])

 # 错误做法：分别处理
  X_labeled_scaled = scaler.fit_transform(X_labeled)
  X_unlabeled_scaled = scaler.fit_transform(X_unlabeled)  # ❌ 数据泄露！
  
  # 正确做法：仅用标签数据拟合
  scaler = StandardScaler().fit(X_labeled)
  X_labeled_scaled = scaler.transform(X_labeled)
  X_unlabeled_scaled = scaler.transform(X_unlabeled)      # ✅

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.impute import SimpleImputer
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# ========================
# 1. 加载数据
# ========================
# 模拟数据：user_id, age, income, city, clicks, purchase (0/1)
np.random.seed(42)
n_labeled = 200    # 少量标签
n_unlabeled = 2000 # 大量无标签

# 生成有标签数据
df_labeled = pd.DataFrame({
    'age': np.random.randint(18, 70, n_labeled),
    'income': np.random.exponential(50000, n_labeled),
    'city': np.random.choice(['北京', '上海', '广州', '深圳', '其他'], n_labeled),
    'clicks': np.random.poisson(10, n_labeled),
    'purchase': np.random.binomial(1, 0.3, n_labeled)
})

# 生成无标签数据（同分布）
df_unlabeled = pd.DataFrame({
    'age': np.random.randint(18, 70, n_unlabeled),
    'income': np.random.exponential(50000, n_unlabeled),
    'city': np.random.choice(['北京', '上海', '广州', '深圳', '其他'], n_unlabeled),
    'clicks': np.random.poisson(10, n_unlabeled),
    'purchase': -1  # 无标签标记为-1
})

# 合并（保留标识）
df_all = pd.concat([df_labeled, df_unlabeled], ignore_index=True)
df_all['is_labeled'] = df_all['purchase'] != -1

# ========================
# 2. EDA
# ========================
print("缺失值比例:\n", df_all.isnull().mean())
print("\n收入分布偏度:", df_all['income'].skew())  # 通常>1，右偏

# ========================
# 3. 缺失值处理（本例无缺失，模拟一个）
# ========================
# 假设income有10%缺失
df_all.loc[np.random.choice(df_all.index, size=int(0.1*len(df_all)), replace=False), 'income'] = np.nan

num_imputer = SimpleImputer(strategy='median')
df_all['income'] = num_imputer.fit_transform(df_all[['income']])

# ========================
# 4. 异常值处理
# ========================
Q1 = df_all['income'].quantile(0.25)
Q3 = df_all['income'].quantile(0.75)
IQR = Q3 - Q1
upper_bound = Q3 + 1.5 * IQR
df_all['income'] = np.clip(df_all['income'], None, upper_bound)  # 修正上限

# ========================
# 5. 特征编码
# ========================
# 城市：目标编码（因有少量标签）
from category_encoders import TargetEncoder
te = TargetEncoder()
# 仅用有标签数据拟合
labeled_mask = df_all['is_labeled']
df_all['city_te'] = te.fit_transform(
    df_all.loc[labeled_mask, ['city']], 
    df_all.loc[labeled_mask, 'purchase']
)
# 应用到全部数据
df_all['city_te'] = te.transform(df_all[['city']])['city']

# ========================
# 6. 特征缩放
# ========================
features = ['age', 'income', 'clicks', 'city_te']
scaler = StandardScaler()
# 仅用有标签数据拟合（避免数据泄露）
scaler.fit(df_all.loc[labeled_mask, features])
df_all[features] = scaler.transform(df_all[features])

# ========================
# 7. 特征选择（仅用有标签数据）
# ========================
X_labeled = df_all.loc[labeled_mask, features]
y_labeled = df_all.loc[labeled_mask, 'purchase']

selector = SelectKBest(score_func=f_classif, k=3)
X_selected = selector.fit_transform(X_labeled, y_labeled)
selected_features = np.array(features)[selector.get_support()]

print("选中的特征:", selected_features)

# ========================
# 8. 半监督训练：伪标签法
# ========================
# 步骤1: 用少量标签训练初始模型
rf = RandomForestClassifier(n_estimators=100, random_state=42)
rf.fit(X_labeled[selected_features], y_labeled)

# 步骤2: 预测无标签数据，取高置信度
X_unlabeled = df_all.loc[~labeled_mask, selected_features]
proba = rf.predict_proba(X_unlabeled)
pseudo_labels = rf.predict(X_unlabeled)
high_conf = proba.max(axis=1) > 0.8

# 步骤3: 构建增强训练集
X_train = pd.concat([
    X_labeled[selected_features],
    X_unlabeled[high_conf]
])
y_train = pd.concat([
    y_labeled,
    pd.Series(pseudo_labels[high_conf], index=X_unlabeled[high_conf].index)
])

# 步骤4: 重新训练
rf_final = RandomForestClassifier(n_estimators=100, random_state=42)
rf_final.fit(X_train, y_train)

# ========================
# 9. 评估（假设我们有隐藏测试集）
# ========================
# 模拟测试集
X_test = pd.DataFrame({
    'age': np.random.randint(18, 70, 500),
    'income': np.random.exponential(50000, 500),
    'clicks': np.random.poisson(10, 500)
})
# 目标编码需用原te
X_test['city_te'] = te.transform(pd.DataFrame({'city': np.random.choice(['北京','上海','其他'], 500)}))['city']
X_test_scaled = scaler.transform(X_test[features])
X_test_final = pd.DataFrame(X_test_scaled, columns=features)[selected_features]

y_test = np.random.binomial(1, 0.3, 500)  # 简化

acc = accuracy_score(y_test, rf_final.predict(X_test_final))
print(f"\n半监督模型准确率: {acc:.2%}")