2026年泰迪杯数据挖掘挑战赛B题完整代码解析与新手避坑指南

1次阅读

共计 4050 个字符，预计需要花费 11 分钟才能阅读完成。

泰迪杯数据挖掘挑战赛 B 题通常涉及复杂的现实场景数据，2026 年赛题预计延续以下典型难点：

非结构化数据处理：原始数据常包含 JSON/ 文本日志等异构格式，需转换为结构化特征
时序特征提取：用户行为数据具有时间连续性，需滑动窗口统计与周期模式挖掘
高维稀疏特征：类别型变量经过 One-Hot 编码后易导致维度爆炸
评价指标特殊性：赛题可能采用 F1-score 等非标准指标，需自定义验证策略

缺失值检测与处理

# 检查各列缺失比例
missing_ratio = df.isnull().mean().sort_values(ascending=False)

# 对连续型变量用中位数填充，类别型用众数填充
num_cols = df.select_dtypes(include=np.number).columns
cat_cols = df.select_dtypes(exclude=np.number).columns

df[num_cols] = df[num_cols].fillna(df[num_cols].median())
df[cat_cols] = df[cat_cols].fillna(df[cat_cols].mode().iloc[0])

异常值修正

# 采用 IQR 方法检测异常值
Q1 = df['value'].quantile(0.25)
Q3 = df['value'].quantile(0.75)
IQR = Q3 - Q1
df = df[~((df['value'] < (Q1 - 1.5*IQR)) | 
          (df['value'] > (Q3 + 1.5*IQR)))]

数据类型转换

# 将时间字符串转为 datetime 类型
df['time'] = pd.to_datetime(df['time'], format='%Y-%m-%d %H:%M:%S')

# 对分类变量进行编码
from sklearn.preprocessing import LabelEncoder
le = LabelEncoder()
df['category'] = le.fit_transform(df['category'])

数据一致性检查

# 验证 ID 列是否唯一
assert df['id'].nunique() == len(df), "存在重复 ID"

# 检查时间顺序
assert df['time'].is_monotonic_increasing, "时间序列未排序"

存储优化

# 降低数值型内存占用
df[num_cols] = df[num_cols].apply(pd.to_numeric, downcast='integer')

# 使用 category 类型存储分类变量
df[cat_cols] = df[cat_cols].astype('category')

# 提取时间维度特征
df['hour'] = df['time'].dt.hour
df['day_of_week'] = df['time'].dt.dayofweek

def get_time_features(df):
    # 滑动窗口统计
    df['7d_avg'] = df['value'].rolling(window='7D').mean()

    # 周期特征（需根据业务调整周期）df['sin_hour'] = np.sin(2*np.pi*df['hour']/24)
    df['cos_hour'] = np.cos(2*np.pi*df['hour']/24)
    return df

# 交互特征（需结合业务逻辑）df['price_per_click'] = df['total_price'] / (df['click_count']+1)

# 分组统计特征
group_stats = df.groupby('user_id')['value'].agg(['mean','std'])
df = df.merge(group_stats, on='user_id', suffixes=('','_user'))

指标	XGBoost	LightGBM
训练速度	较慢	快（支持直方图算法）
内存占用	较高	较低
类别特征处理	需编码	原生支持
小数据集表现	优	可能过拟合
解释性	提供特征重要性	特征重要性更稳定

import warnings
warnings.filterwarnings('ignore')

# 数据加载与校验
try:
    raw_df = pd.read_csv('contest_b_data.csv', parse_dates=['timestamp'])
    assert set(raw_df.columns) >= {'id','target','feature1','feature2'}
except Exception as e:
    print(f"数据加载失败: {str(e)}")
    raise

# 特征工程管道
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer

preprocessor = ColumnTransformer(
    transformers=[('num', StandardScaler(), num_cols),
        ('cat', OneHotEncoder(handle_unknown='ignore'), cat_cols)
    ])

# 模型训练与验证
from sklearn.model_selection import TimeSeriesSplit
tscv = TimeSeriesSplit(n_splits=5)

model = Pipeline([('preprocess', preprocessor),
    ('model', LGBMClassifier(
        n_estimators=500,
        learning_rate=0.05,
        max_depth=7,
        random_state=42))
])

# 使用时序交叉验证
for train_idx, test_idx in tscv.split(X):
    X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]

    model.fit(X_train, y_train)
    preds = model.predict_proba(X_test)[:,1]

    # 使用赛题指定指标
    score = roc_auc_score(y_test, preds)
    print(f"Fold AUC: {score:.4f}")

使用 dtypes 参数指定列类型读取数据

dtype_dict = {'user_id':'int32', 'price':'float32'}
df = pd.read_csv('large_file.csv', dtype=dtype_dict)

分块处理大规模数据

chunk_iter = pd.read_csv('huge.csv', chunksize=100000)
results = []
for chunk in chunk_iter:
    processed = process_chunk(chunk)  # 自定义处理函数
    results.append(processed)
final_df = pd.concat(results)

时间先后校验

# 确保特征生成时间早于标签时间
assert (df['feature_time'] < df['label_time']).all()

避免使用未来信息

# 错误的滚动统计（包含未来数据）df['wrong_avg'] = df['value'].rolling(30, min_periods=1).mean()

# 正确的滞后统计
df['correct_avg'] = df['value'].shift(1).rolling(30).mean()

方法	适用场景	实现方式
随机欠采样	多数类远大于少数类	`RandomUnderSampler()`
SMOTE	特征空间线性可分	`SMOTE(k_neighbors=5)`
ADASYN	不同密度区域的少数类	`ADASYN(n_neighbors=5)`
类别权重	保持数据分布	`class_weight='balanced'`

数学表达（SMOTE 算法）：

对于每个少数类样本 $x_i$，从其 k 近邻中随机选择样本 $x_{zi}$，生成新样本：

$$x_{new} = x_i + \lambda (x_{zi} – x_i)$$

其中 $\lambda\in[0,1]$ 为随机数

常规分类问题

from sklearn.model_selection import StratifiedKFold
skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

时间序列问题

from sklearn.model_selection import TimeSeriesSplit
tscv = TimeSeriesSplit(n_splits=5)

自定义验证策略

# 按用户分组验证（避免同一用户出现在训练测试集）from sklearn.model_selection import GroupKFold
gkf = GroupKFold(n_splits=5)
for train_idx, test_idx in gkf.split(X, y, groups=user_ids):
    # 训练验证逻辑

访问以下链接获取简化版数据集和 Jupyter Notebook 模板：
示例数据下载

实践任务：
1. 尝试添加新的时序特征（如节假日标记）
2. 比较 XGBoost 与 LightGBM 的验证集表现差异
3. 实现早停机制防止过拟合

通过本次技术解析，参赛者应能掌握数据挖掘竞赛的完整处理流程，特别注意时序数据的特殊性和模型优化的核心技巧。建议在实际应用中持续监控数据分布变化，定期更新模型以适应数据漂移。

正文完