如何实现基于堆叠链式分类器的人类活动识别多阶段分类？

问题背景 #

现有人类活动识别数据集，包含mean_speed、max_speed、max_acc、mean_acc四个特征，以及activity标签（分为driving、walking、riding、motor-bike四类）。需要构建链式分类器：

1. 基分类器（Random Forest）先预测样本属于motorised（driving、motor-bike）或non-motorised（walking、riding），输出概率值
2. 两个元分类器：
   • 针对non-motorised子集，用Decision Tree区分walking和riding
   • 针对motorised子集，用SVC区分driving和motor-bike
3. 元分类器的输入为原始4个特征 + 基分类器输出的概率

数据集准备 #

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.tree import DecisionTreeClassifier
from sklearn.svm import SVC
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# 构建数据集
df = pd.DataFrame(
    { 
        'mean_speed': [40.01, 3.1, 2.88, 20.89, 5.82, 40.01, 33.1, 40.88, 20.89, 5.82, 40.018, 23.1], 
        'max_speed': [70.11, 6.71, 7.08, 39.63, 6.68, 70.11, 65.71, 71.08, 39.63, 13.68, 70.11, 35.71],
        'max_acc': [17.63, 2.93, 3.32, 15.57, 0.94, 17.63, 12.93, 3.32, 15.57, 0.94, 17.63, 12.93], 
        'mean_acc': [5.15, 1.97, 0.59, 5.11, 0.19, 5.15, 2.97, 0.59, 5.11, 0.19, 5.15, 2.97],
        'activity': ['driving', 'walking', 'walking', 'riding', 'walking', 'driving', 'motor-bike',
                    'motor-bike', 'riding', 'riding', 'motor-bike', 'riding']
    }
)

# 添加类型标签
class_mapping = {'driving':'motorised', 'motor-bike':'motorised', 'walking':'non-motorised', 'riding':'non-motorised'}
df['type'] = df['activity'].map(class_mapping)

步骤1：训练基分类器（Random Forest） #

基分类器用于预测样本的type（motorised/non-motorised），并输出概率值作为后续元分类器的特征之一。

# 分离基分类器的特征和标签
X_base = df[['mean_speed', 'max_speed', 'max_acc', 'mean_acc']]
y_base = df['type']

# 划分训练集和测试集
X_base_train, X_base_test, y_base_train, y_base_test = train_test_split(X_base, y_base, test_size=0.2, random_state=42)

# 训练Random Forest基分类器
base_clf = RandomForestClassifier(random_state=42)
base_clf.fit(X_base_train, y_base_train)

# 验证基分类器效果
y_base_pred = base_clf.predict(X_base_test)
print(f"基分类器准确率: {accuracy_score(y_base_test, y_base_pred):.2f}")

步骤2：生成概率特征并拆分数据集 #

为元分类器准备输入特征：原始4个特征 + 基分类器输出的motorised类概率（二者互补，选其一即可）。

# 为整个数据集生成基分类器的概率预测（取motorised类的概率）
df['motorised_prob'] = base_clf.predict_proba(X_base)[:, base_clf.classes_ == 'motorised'][0]

# 拆分motorised和non-motorised子集
motorised_df = df[df['type'] == 'motorised'].copy()
non_motorised_df = df[df['type'] == 'non-motorised'].copy()

步骤3：训练元分类器 #

3.1 针对non-motorised的Decision Tree（区分walking/riding） #

# 特征：原始4特征 + motorised_prob；标签：activity
X_non_motor = non_motorised_df[['mean_speed', 'max_speed', 'max_acc', 'mean_acc', 'motorised_prob']]
y_non_motor = non_motorised_df['activity']

# 划分训练测试集
X_non_train, X_non_test, y_non_train, y_non_test = train_test_split(X_non_motor, y_non_motor, test_size=0.2, random_state=42)

# 训练Decision Tree
dt_clf = DecisionTreeClassifier(random_state=42)
dt_clf.fit(X_non_train, y_non_train)

# 验证效果
y_non_pred = dt_clf.predict(X_non_test)
print(f"Non-motorised元分类器准确率: {accuracy_score(y_non_test, y_non_pred):.2f}")

3.2 针对motorised的SVC（区分driving/motor-bike） #

# 特征：原始4特征 + motorised_prob；标签：activity
X_motor = motorised_df[['mean_speed', 'max_speed', 'max_acc', 'mean_acc', 'motorised_prob']]
y_motor = motorised_df['activity']

# 划分训练测试集
X_motor_train, X_motor_test, y_motor_train, y_motor_test = train_test_split(X_motor, y_motor, test_size=0.2, random_state=42)

# 训练SVC（需设置probability=True以支持概率输出）
svc_clf = SVC(probability=True, random_state=42)
svc_clf.fit(X_motor_train, y_motor_train)

# 验证效果
y_motor_pred = svc_clf.predict(X_motor_test)
print(f"Motorised元分类器准确率: {accuracy_score(y_motor_test, y_motor_pred):.2f}")

步骤4：完整预测流程 #

封装成函数，实现从输入特征到最终activity预测的完整链式流程：

def predict_activity(features):
    # 输入需匹配特征列格式
    features_df = pd.DataFrame([features], columns=['mean_speed', 'max_speed', 'max_acc', 'mean_acc'])
    
    # 第一步：基分类器预测类型和概率
    type_pred = base_clf.predict(features_df)[0]
    motorised_prob = base_clf.predict_proba(features_df)[:, base_clf.classes_ == 'motorised'][0][0]
    
    # 添加概率特征到输入
    features_df['motorised_prob'] = motorised_prob
    
    # 第二步：根据类型选择元分类器预测最终activity
    if type_pred == 'non-motorised':
        return dt_clf.predict(features_df)[0]
    else:
        return svc_clf.predict(features_df)[0]

# 测试示例
test_sample = [40.01, 70.11, 17.63, 5.15]  # driving样本
print(f"测试样本预测结果: {predict_activity(test_sample)}")

test_sample2 = [3.1, 6.71, 2.93, 1.97]  # walking样本
print(f"测试样本2预测结果: {predict_activity(test_sample2)}")

内容的提问来源于stack exchange，提问作者arilwan