{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.11","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":102335,"databundleVersionId":12518947,"sourceType":"competition"}],"dockerImageVersionId":31041,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:35.309960Z","iopub.execute_input":"2025-06-09T16:14:35.310170Z","iopub.status.idle":"2025-06-09T16:14:35.892448Z","shell.execute_reply.started":"2025-06-09T16:14:35.310145Z","shell.execute_reply":"2025-06-09T16:14:35.891688Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os, json, joblib, numpy as np, pandas as pd\nfrom pathlib import Path\nimport warnings \nwarnings.filterwarnings(\"ignore\")\n\n# 导入必要的库\nfrom sklearn.model_selection import train_test_split, StratifiedKFold, GridSearchCV\nfrom sklearn.preprocessing import StandardScaler, LabelEncoder\nfrom sklearn.utils.class_weight import compute_class_weight\nfrom sklearn.metrics import accuracy_score, f1_score, confusion_matrix, classification_report\nfrom sklearn.ensemble import RandomForestClassifier\nimport xgboost as xgb\nfrom sklearn.pipeline import Pipeline\n\nfrom tensorflow.keras.utils import Sequence, to_categorical, pad_sequences\nfrom tensorflow.keras.models import Model, load_model\nfrom tensorflow.keras.layers import (\n    Input, Conv1D, BatchNormalization, Activation, add, MaxPooling1D, Dropout,\n    Bidirectional, LSTM, GlobalAveragePooling1D, Dense, Multiply, Reshape,\n    Lambda, Concatenate\n)\nfrom tensorflow.keras.regularizers import l2\nfrom tensorflow.keras.optimizers import Adam\nfrom tensorflow.keras.callbacks import EarlyStopping\nfrom tensorflow.keras import backend as K\nimport tensorflow as tf\nimport polars as pl","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:35.894240Z","iopub.execute_input":"2025-06-09T16:14:35.894510Z","iopub.status.idle":"2025-06-09T16:14:52.285524Z","shell.execute_reply.started":"2025-06-09T16:14:35.894494Z","shell.execute_reply":"2025-06-09T16:14:52.284977Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 自定义评估指标\nclass CompetitionMetric:\n    def calculate_hierarchical_f1(self, true_df, pred_df):\n        \"\"\"计算层次化F1分数\"\"\"\n        # 这里简化实现，实际竞赛中可能有更复杂的层次结构\n        return f1_score(true_df['gesture'], pred_df['gesture'], average='macro')\n\n# 设置参数\nTRAIN = True                     \nRAW_DIR = Path(\"/kaggle/input/cmi-detect-behavior-with-sensor-data\")\nPRETRAINED_DIR = Path(\"/kaggle/input/pretrained-model\")  \nEXPORT_DIR = Path(\"./\")                                    \nBATCH_SIZE = 64\nPAD_PERCENTILE = 95\nLR_INIT = 1e-3\nWD = 3e-4\nMIXUP_ALPHA = 0.4\nEPOCHS = 160\nPATIENCE = 40\nN_OUTER_FOLDS = 5  # 外层交叉验证折数\nN_INNER_FOLDS = 3  # 内层交叉验证折数\n\nprint(\"▶ imports ready · tensorflow\", tf.__version__)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.286252Z","iopub.execute_input":"2025-06-09T16:14:52.286755Z","iopub.status.idle":"2025-06-09T16:14:52.292605Z","shell.execute_reply.started":"2025-06-09T16:14:52.286735Z","shell.execute_reply":"2025-06-09T16:14:52.291842Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Tensor操作函数\ndef time_sum(x):\n    return K.sum(x, axis=1)\n\ndef squeeze_last_axis(x):\n    return tf.squeeze(x, axis=-1)\n\ndef expand_last_axis(x):\n    return tf.expand_dims(x, axis=-1)\n\ndef se_block(x, reduction=8):\n    ch = x.shape[-1]\n    se = GlobalAveragePooling1D()(x)\n    se = Dense(ch // reduction, activation='relu')(se)\n    se = Dense(ch, activation='sigmoid')(se)\n    se = Reshape((1, ch))(se)\n    return Multiply()([x, se])\n\n# Residual CNN Block with SE\ndef residual_se_cnn_block(x, filters, kernel_size, pool_size=2, drop=0.3, wd=1e-4):\n    shortcut = x\n    for _ in range(2):\n        x = Conv1D(filters, kernel_size, padding='same', use_bias=False,\n                   kernel_regularizer=l2(wd))(x)\n        x = BatchNormalization()(x)\n        x = Activation('relu')(x)\n    x = se_block(x)\n    if shortcut.shape[-1] != filters:\n        shortcut = Conv1D(filters, 1, padding='same', use_bias=False,\n                          kernel_regularizer=l2(wd))(shortcut)\n        shortcut = BatchNormalization()(shortcut)\n    x = add([x, shortcut])\n    x = Activation('relu')(x)\n    x = MaxPooling1D(pool_size)(x)\n    x = Dropout(drop)(x)\n    return x\n\ndef attention_layer(inputs):\n    score = Dense(1, activation='tanh')(inputs)\n    score = Lambda(squeeze_last_axis)(score)\n    weights = Activation('softmax')(score)\n    weights = Lambda(expand_last_axis)(weights)\n    context = Multiply()([inputs, weights])\n    context = Lambda(time_sum)(context)\n    return context","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.293390Z","iopub.execute_input":"2025-06-09T16:14:52.294233Z","iopub.status.idle":"2025-06-09T16:14:52.318252Z","shell.execute_reply.started":"2025-06-09T16:14:52.294207Z","shell.execute_reply":"2025-06-09T16:14:52.317700Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Normalizes and cleans the time series sequence. \ndef preprocess_sequence(df_seq: pd.DataFrame, feature_cols: list[str], scaler: StandardScaler):\n    mat = df_seq[feature_cols].ffill().bfill().fillna(0).values\n    return scaler.transform(mat).astype('float32')\n\n# MixUp数据增强\nclass MixupGenerator(Sequence):\n    def __init__(self, X, y, batch_size, alpha=0.2):\n        self.X, self.y = X, y\n        self.batch = batch_size\n        self.alpha = alpha\n        self.indices = np.arange(len(X))\n        \n    def __len__(self):\n        return int(np.ceil(len(self.X) / self.batch))\n    \n    def __getitem__(self, i):\n        idx = self.indices[i*self.batch:(i+1)*self.batch]\n        Xb, yb = self.X[idx], self.y[idx]\n        lam = np.random.beta(self.alpha, self.alpha)\n        perm = np.random.permutation(len(Xb))\n        X_mix = lam * Xb + (1-lam) * Xb[perm]\n        y_mix = lam * yb + (1-lam) * yb[perm]\n        return X_mix, y_mix\n    \n    def on_epoch_end(self):\n        np.random.shuffle(self.indices)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.319850Z","iopub.execute_input":"2025-06-09T16:14:52.320045Z","iopub.status.idle":"2025-06-09T16:14:52.339046Z","shell.execute_reply.started":"2025-06-09T16:14:52.320030Z","shell.execute_reply":"2025-06-09T16:14:52.338406Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 构建双流模型\ndef build_two_branch_model(pad_len, imu_dim, tof_dim, n_classes, wd=1e-4):\n    inp = Input(shape=(pad_len, imu_dim+tof_dim))\n    imu = Lambda(lambda t: t[:, :, :imu_dim])(inp)\n    tof = Lambda(lambda t: t[:, :, imu_dim:])(inp)\n\n    # IMU deep branch\n    x1 = residual_se_cnn_block(imu, 64, 3, drop=0.3, wd=wd)\n    x1 = residual_se_cnn_block(x1, 128, 5, drop=0.3, wd=wd)\n\n    # TOF/Thermal lighter branch\n    x2 = Conv1D(64, 3, padding='same', use_bias=False, kernel_regularizer=l2(wd))(tof)\n    x2 = BatchNormalization()(x2); x2 = Activation('relu')(x2)\n    x2 = MaxPooling1D(2)(x2); x2 = Dropout(0.3)(x2)\n    x2 = Conv1D(128, 3, padding='same', use_bias=False, kernel_regularizer=l2(wd))(x2)\n    x2 = BatchNormalization()(x2); x2 = Activation('relu')(x2)\n    x2 = MaxPooling1D(2)(x2); x2 = Dropout(0.3)(x2)\n\n    merged = Concatenate()([x1, x2])\n\n    x = Bidirectional(LSTM(128, return_sequences=True, kernel_regularizer=l2(wd)))(merged)\n    x = Dropout(0.4)(x)\n    x = attention_layer(x)\n\n    for units, drop in [(256, 0.5), (128, 0.3)]:\n        x = Dense(units, use_bias=False, kernel_regularizer=l2(wd))(x)\n        x = BatchNormalization()(x); x = Activation('relu')(x)\n        x = Dropout(drop)(x)\n\n    out = Dense(n_classes, activation='softmax', kernel_regularizer=l2(wd))(x)\n    return Model(inp, out)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.339659Z","iopub.execute_input":"2025-06-09T16:14:52.339880Z","iopub.status.idle":"2025-06-09T16:14:52.354424Z","shell.execute_reply.started":"2025-06-09T16:14:52.339865Z","shell.execute_reply":"2025-06-09T16:14:52.353814Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 构建传统LSTM模型\ndef build_lstm_model(pad_len, feature_dim, n_classes, wd=1e-4):\n    inp = Input(shape=(pad_len, feature_dim))\n    \n    x = LSTM(128, return_sequences=True, kernel_regularizer=l2(wd))(inp)\n    x = Dropout(0.3)(x)\n    x = LSTM(128, kernel_regularizer=l2(wd))(x)\n    x = Dropout(0.3)(x)\n    \n    x = Dense(256, use_bias=False, kernel_regularizer=l2(wd))(x)\n    x = BatchNormalization()(x)\n    x = Activation('relu')(x)\n    x = Dropout(0.5)(x)\n    \n    out = Dense(n_classes, activation='softmax', kernel_regularizer=l2(wd))(x)\n    return Model(inp, out)\n\n# 特征提取函数 - 用于传统机器学习模型\ndef extract_features(X_sequences):\n    \"\"\"从序列数据中提取统计特征\"\"\"\n    features = []\n    for seq in X_sequences:\n        # 计算每个特征维度的统计量\n        mean = np.mean(seq, axis=0)\n        std = np.std(seq, axis=0)\n        min_val = np.min(seq, axis=0)\n        max_val = np.max(seq, axis=0)\n        median = np.median(seq, axis=0)\n        skewness = np.mean(((seq - mean) / std) ** 3, axis=0)\n        kurtosis = np.mean(((seq - mean) / std) ** 4, axis=0) - 3\n        \n        # 组合所有特征\n        combined = np.hstack([mean, std, min_val, max_val, median, skewness, kurtosis])\n        features.append(combined)\n    \n    return np.array(features)\n\n# 嵌套交叉验证评估模型\ndef nested_cross_validation(X, y, model_type, param_grid=None):\n    \"\"\"\n    执行嵌套交叉验证评估模型性能\n    \n    参数:\n    X: 特征数据\n    y: 标签数据\n    model_type: 模型类型 ('two_branch', 'lstm', 'xgb')\n    param_grid: 超参数网格\n    \n    返回:\n    每个外层折叠的评估结果\n    \"\"\"\n    outer_cv = StratifiedKFold(n_splits=N_OUTER_FOLDS, shuffle=True, random_state=42)\n    results = []\n    \n    for i, (train_idx, test_idx) in enumerate(outer_cv.split(X, y)):\n        print(f\"\\n=== 外层折叠 {i+1}/{N_OUTER_FOLDS} ===\")\n        \n        X_train, X_test = X[train_idx], X[test_idx]\n        y_train, y_test = y[train_idx], y[test_idx]\n        \n        # 内层交叉验证用于超参数优化\n        inner_cv = StratifiedKFold(n_splits=N_INNER_FOLDS, shuffle=True, random_state=42)\n        \n        if model_type == 'two_branch':\n            # 为简化示例，这里只使用固定参数\n            model = build_two_branch_model(\n                X_train.shape[1], \n                len(imu_cols), \n                len(tof_cols), \n                len(np.unique(y)),\n                wd=WD\n            )\n            steps = len(X_train)//BATCH_SIZE\n            lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n            model.compile(\n                optimizer=Adam(lr_sched),\n                loss='categorical_crossentropy',\n                metrics=['accuracy']\n            )\n            train_gen = MixupGenerator(X_train, to_categorical(y_train), BATCH_SIZE, MIXUP_ALPHA)\n            cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True)\n            model.fit(\n                train_gen, \n                epochs=EPOCHS, \n                validation_data=(X_test, to_categorical(y_test)),\n                callbacks=[cb],\n                verbose=0\n            )\n            y_pred = model.predict(X_test).argmax(1)\n            \n        elif model_type == 'lstm':\n            # 为简化示例，这里只使用固定参数\n            model = build_lstm_model(\n                X_train.shape[1], \n                X_train.shape[2], \n                len(np.unique(y)),\n                wd=WD\n            )\n            steps = len(X_train)//BATCH_SIZE\n            lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n            model.compile(\n                optimizer=Adam(lr_sched),\n                loss='categorical_crossentropy',\n                metrics=['accuracy']\n            )\n            train_gen = MixupGenerator(X_train, to_categorical(y_train), BATCH_SIZE, MIXUP_ALPHA)\n            cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True)\n            model.fit(\n                train_gen, \n                epochs=EPOCHS, \n                validation_data=(X_test, to_categorical(y_test)),\n                callbacks=[cb],\n                verbose=0\n            )\n            y_pred = model.predict(X_test).argmax(1)\n            \n        elif model_type == 'xgb':\n            # 提取统计特征用于XGBoost\n            X_train_features = extract_features(X_train)\n            X_test_features = extract_features(X_test)\n            \n            if param_grid:\n                # 执行网格搜索\n                search = GridSearchCV(\n                    estimator=xgb.XGBClassifier(objective='multi:softmax', use_label_encoder=False, eval_metric='merror'),\n                    param_grid=param_grid,\n                    cv=inner_cv,\n                    scoring='f1_macro',\n                    n_jobs=-1\n                )\n                search.fit(X_train_features, y_train)\n                model = search.best_estimator_\n                print(f\"最佳参数: {search.best_params_}\")\n            else:\n                model = xgb.XGBClassifier(\n                    objective='multi:softmax',\n                    n_estimators=100,\n                    max_depth=6,\n                    learning_rate=0.1,\n                    use_label_encoder=False,\n                    eval_metric='merror'\n                )\n                model.fit(X_train_features, y_train)\n                \n            y_pred = model.predict(X_test_features)\n        \n        # 评估模型\n        acc = accuracy_score(y_test, y_pred)\n        f1_macro = f1_score(y_test, y_pred, average='macro')\n        cm = confusion_matrix(y_test, y_pred)\n        \n        print(f\"准确率: {acc:.4f}\")\n        print(f\"Macro F1: {f1_macro:.4f}\")\n        print(f\"混淆矩阵:\\n{cm}\")\n        \n        results.append({\n            'accuracy': acc,\n            'f1_macro': f1_macro,\n            'confusion_matrix': cm,\n            'y_true': y_test,\n            'y_pred': y_pred\n        })\n    \n    return results\n\n    ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.355058Z","iopub.execute_input":"2025-06-09T16:14:52.355254Z","iopub.status.idle":"2025-06-09T16:14:52.378491Z","shell.execute_reply.started":"2025-06-09T16:14:52.355239Z","shell.execute_reply":"2025-06-09T16:14:52.377902Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def build_xgb_model(X_train, y_train, param_grid=None, inner_cv=None):\n    \"\"\"\n    封装 XGBoost 模型的训练过程，支持可选的网格搜索\n    返回训练好的模型\n    \"\"\"\n    # 提取统计特征\n    X_train_features = extract_features(X_train)\n    \n    if param_grid and inner_cv:\n        search = GridSearchCV(\n            estimator=xgb.XGBClassifier(objective='multi:softmax', use_label_encoder=False, eval_metric='merror'),\n            param_grid=param_grid,\n            cv=inner_cv,\n            scoring='f1_macro',\n            n_jobs=-1\n        )\n        search.fit(X_train_features, y_train)\n        model = search.best_estimator_\n        print(f\"XGB 最佳参数: {search.best_params_}\")\n    else:\n        model = xgb.XGBClassifier(\n            objective='multi:softmax',\n            n_estimators=200,\n            max_depth=10,\n            learning_rate=0.1,\n            use_label_encoder=False,\n            eval_metric='merror'\n        )\n        model.fit(X_train_features, y_train)\n    \n    return model\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.379395Z","iopub.execute_input":"2025-06-09T16:14:52.379637Z","iopub.status.idle":"2025-06-09T16:14:52.399571Z","shell.execute_reply.started":"2025-06-09T16:14:52.379613Z","shell.execute_reply":"2025-06-09T16:14:52.398879Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def nested_cross_validation(X, y, model_type, param_grid=None):\n    outer_cv = StratifiedKFold(n_splits=N_OUTER_FOLDS, shuffle=True, random_state=42)\n    results = []\n\n    for i, (train_idx, test_idx) in enumerate(outer_cv.split(X, y)):\n        print(f\"\\n=== 外层折叠 {i+1}/{N_OUTER_FOLDS} ===\")\n        X_train, X_test = X[train_idx], X[test_idx]\n        y_train, y_test = y[train_idx], y[test_idx]\n\n        inner_cv = StratifiedKFold(n_splits=N_INNER_FOLDS, shuffle=True, random_state=42)\n\n        if model_type == 'two_branch':\n            model = build_two_branch_model(X_train.shape[1], len(imu_cols), len(tof_cols), len(np.unique(y)), wd=WD)\n            steps = len(X_train)//BATCH_SIZE\n            lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n            model.compile(optimizer=Adam(lr_sched), loss='categorical_crossentropy', metrics=['accuracy'])\n            train_gen = MixupGenerator(X_train, to_categorical(y_train), BATCH_SIZE, MIXUP_ALPHA)\n            cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True)\n            model.fit(train_gen, epochs=EPOCHS, validation_data=(X_test, to_categorical(y_test)), callbacks=[cb], verbose=0)\n            y_pred = model.predict(X_test).argmax(1)\n\n        elif model_type == 'lstm':\n            model = build_lstm_model(X_train.shape[1], X_train.shape[2], len(np.unique(y)), wd=WD)\n            steps = len(X_train)//BATCH_SIZE\n            lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n            model.compile(optimizer=Adam(lr_sched), loss='categorical_crossentropy', metrics=['accuracy'])\n            train_gen = MixupGenerator(X_train, to_categorical(y_train), BATCH_SIZE, MIXUP_ALPHA)\n            cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True)\n            model.fit(train_gen, epochs=EPOCHS, validation_data=(X_test, to_categorical(y_test)), callbacks=[cb], verbose=0)\n            y_pred = model.predict(X_test).argmax(1)\n\n        elif model_type == 'xgb':\n            model = build_xgb_model(X_train, y_train, param_grid, inner_cv)\n            X_test_features = extract_features(X_test)\n            y_pred = model.predict(X_test_features)\n\n        # 评估\n        acc = accuracy_score(y_test, y_pred)\n        f1_macro = f1_score(y_test, y_pred, average='macro')\n        cm = confusion_matrix(y_test, y_pred)\n\n        print(f\"准确率: {acc:.4f}\")\n        print(f\"Macro F1: {f1_macro:.4f}\")\n\n        results.append({\n            'accuracy': acc,\n            'f1_macro': f1_macro,\n            'confusion_matrix': cm,\n            'y_true': y_test,\n            'y_pred': y_pred\n        })\n\n    return results\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.400229Z","iopub.execute_input":"2025-06-09T16:14:52.400381Z","iopub.status.idle":"2025-06-09T16:14:52.415305Z","shell.execute_reply.started":"2025-06-09T16:14:52.400369Z","shell.execute_reply":"2025-06-09T16:14:52.414809Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"if TRAIN:\n    print(\"▶ TRAIN MODE – loading dataset …\")\n    df = pd.read_csv(RAW_DIR / \"train.csv\")\n\n    le = LabelEncoder()\n    df['gesture_int'] = le.fit_transform(df['gesture'])\n    np.save(EXPORT_DIR / \"gesture_classes.npy\", le.classes_)\n\n    meta_cols = {'gesture', 'gesture_int', 'sequence_type', 'behavior', 'orientation', 'row_id', 'subject', 'phase', 'sequence_id', 'sequence_counter'}\n    feature_cols = [c for c in df.columns if c not in meta_cols]\n    imu_cols = [c for c in feature_cols if not (c.startswith('thm_') or c.startswith('tof_'))]\n    tof_cols = [c for c in feature_cols if c.startswith('thm_') or c.startswith('tof_')]\n    print(f\"  IMU {len(imu_cols)} | TOF/THM {len(tof_cols)} | total {len(feature_cols)} features\")\n\n    scaler = StandardScaler().fit(df[feature_cols].ffill().bfill().fillna(0).values)\n    joblib.dump(scaler, EXPORT_DIR / \"scaler.pkl\")\n\n    seq_gp = df.groupby('sequence_id')\n    X_list, y_list, lens = [], [], []\n    for seq_id, seq in seq_gp:\n        mat = preprocess_sequence(seq, feature_cols, scaler)\n        X_list.append(mat)\n        y_list.append(seq['gesture_int'].iloc[0])\n        lens.append(len(mat))\n\n    pad_len = int(np.percentile(lens, PAD_PERCENTILE))\n    np.save(EXPORT_DIR / \"sequence_maxlen.npy\", pad_len)\n    np.save(EXPORT_DIR / \"feature_cols.npy\", np.array(feature_cols))\n\n    X = pad_sequences(X_list, maxlen=pad_len, padding='post', truncating='post', dtype='float32')\n    y = np.array(y_list)\n\n    cw_vals = compute_class_weight('balanced', classes=np.arange(len(le.classes_)), y=y)\n    class_weight = dict(enumerate(cw_vals))\n\n    models_to_evaluate = {\n        'two_branch': '双流CNN+BiLSTM+注意力机制',\n        'lstm': '传统LSTM模型',\n        'xgb': 'XGBoost'\n    }\n\n    xgb_param_grid = {'max_depth': [3, 6, 10], 'learning_rate': [0.01, 0.1], 'n_estimators': [100, 200]}\n\n    all_results = {}\n    for model_type, model_name in models_to_evaluate.items():\n        print(f\"\\n===== 评估模型: {model_name} =====\")\n        results = nested_cross_validation(X, y, model_type, param_grid=xgb_param_grid if model_type == 'xgb' else None)\n        avg_acc = np.mean([r['accuracy'] for r in results])\n        avg_f1 = np.mean([r['f1_macro'] for r in results])\n        print(f\"{model_name} 平均准确率: {avg_acc:.4f}\")\n        print(f\"{model_name} 平均Macro F1: {avg_f1:.4f}\")\n        all_results[model_type] = {'results': results, 'avg_accuracy': avg_acc, 'avg_f1_macro': avg_f1}\n\n    print(\"\\n===== 模型性能比较 =====\")\n    for model_type, model_name in models_to_evaluate.items():\n        print(f\"{model_name}: 准确率={all_results[model_type]['avg_accuracy']:.4f}, Macro F1={all_results[model_type]['avg_f1_macro']:.4f}\")\n\n    best_model_type = max(all_results, key=lambda k: all_results[k]['avg_f1_macro'])\n    print(f\"\\n最佳模型: {models_to_evaluate[best_model_type]}\")\n\n    if best_model_type == 'two_branch':\n        final_model = build_two_branch_model(pad_len, len(imu_cols), len(tof_cols), len(le.classes_), wd=WD)\n        steps = len(X)//BATCH_SIZE\n        lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n        final_model.compile(optimizer=Adam(lr_sched), loss=tf.keras.losses.CategoricalCrossentropy(label_smoothing=0.1), metrics=['accuracy'])\n        train_gen = MixupGenerator(X, to_categorical(y), batch_size=BATCH_SIZE, alpha=MIXUP_ALPHA)\n        cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True, verbose=1)\n        final_model.fit(train_gen, epochs=EPOCHS, callbacks=[cb], verbose=1)\n        final_model.save(EXPORT_DIR / \"best_model.h5\")\n        \n    elif best_model_type == 'lstm':\n        final_model = build_lstm_model(pad_len, X.shape[2], len(le.classes_), wd=WD)\n        steps = len(X)//BATCH_SIZE\n        lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n        final_model.compile(optimizer=Adam(lr_sched), loss=tf.keras.losses.CategoricalCrossentropy(label_smoothing=0.1), metrics=['accuracy'])\n        train_gen = MixupGenerator(X, to_categorical(y), batch_size=BATCH_SIZE, alpha=MIXUP_ALPHA)\n        cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True, verbose=1)\n        final_model.fit(train_gen, epochs=EPOCHS, callbacks=[cb], verbose=1)\n        final_model.save(EXPORT_DIR / \"best_model.h5\")\n        \n    elif best_model_type == 'xgb':\n        final_model = build_xgb_model(X, y)  # 直接使用封装好的XGB函数\n        joblib.dump(final_model, EXPORT_DIR / \"best_model.pkl\")\n\n    print(\"✔ Training done – 最佳模型已保存\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-09T16:14:52.416111Z","iopub.execute_input":"2025-06-09T16:14:52.416345Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 主程序\nif TRAIN:\n    print(\"▶ TRAIN MODE – loading dataset …\")\n    df = pd.read_csv(RAW_DIR / \"train.csv\")\n\n    # 标签编码\n    le = LabelEncoder()\n    df['gesture_int'] = le.fit_transform(df['gesture'])\n    np.save(EXPORT_DIR / \"gesture_classes.npy\", le.classes_)\n\n    # 特征列表\n    meta_cols = {'gesture', 'gesture_int', 'sequence_type', 'behavior', 'orientation',\n                 'row_id', 'subject', 'phase', 'sequence_id', 'sequence_counter'}\n    feature_cols = [c for c in df.columns if c not in meta_cols]\n\n    imu_cols  = [c for c in feature_cols if not (c.startswith('thm_') or c.startswith('tof_'))]\n    tof_cols  = [c for c in feature_cols if c.startswith('thm_') or c.startswith('tof_')]\n    print(f\"  IMU {len(imu_cols)} | TOF/THM {len(tof_cols)} | total {len(feature_cols)} features\")\n\n    # 全局标准化器\n    scaler = StandardScaler().fit(df[feature_cols].ffill().bfill().fillna(0).values)\n    joblib.dump(scaler, EXPORT_DIR / \"scaler.pkl\")\n\n    # 构建序列\n    seq_gp = df.groupby('sequence_id')\n    X_list, y_list, lens = [], [], []\n    for seq_id, seq in seq_gp:\n        mat = preprocess_sequence(seq, feature_cols, scaler)\n        X_list.append(mat)\n        y_list.append(seq['gesture_int'].iloc[0])\n        lens.append(len(mat))\n    \n    pad_len = int(np.percentile(lens, PAD_PERCENTILE))\n    np.save(EXPORT_DIR / \"sequence_maxlen.npy\", pad_len)\n    np.save(EXPORT_DIR / \"feature_cols.npy\", np.array(feature_cols))\n\n    X = pad_sequences(X_list, maxlen=pad_len, padding='post', truncating='post', dtype='float32')\n    y = np.array(y_list)\n    \n    # 原始代码中的训练/验证分割\n    X_tr, X_val, y_tr, y_val = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)\n\n    # 类别权重\n    cw_vals = compute_class_weight('balanced', classes=np.arange(len(le.classes_)), y=y)\n    class_weight = dict(enumerate(cw_vals))\n\n    # 定义要评估的模型\n    models_to_evaluate = {\n        'two_branch': '双流CNN+BiLSTM+注意力机制',\n        'lstm': '传统LSTM模型',\n        'xgb': 'XGBoost'\n    }\n    \n    # XGBoost超参数网格\n    xgb_param_grid = {\n        'max_depth': [3, 6, 10],\n        'learning_rate': [0.01, 0.1],\n        'n_estimators': [100, 200]\n    }\n    \n    # 存储所有模型的评估结果\n    all_results = {}\n    \n    # 对每个模型进行嵌套交叉验证\n    for model_type, model_name in models_to_evaluate.items():\n        print(f\"\\n===== 评估模型: {model_name} =====\")\n        results = nested_cross_validation(X, y, model_type, \n                                         param_grid=xgb_param_grid if model_type == 'xgb' else None)\n        \n        # 计算平均性能\n        avg_acc = np.mean([r['accuracy'] for r in results])\n        avg_f1 = np.mean([r['f1_macro'] for r in results])\n        \n        print(f\"\\n{model_name} 平均准确率: {avg_acc:.4f}\")\n        print(f\"{model_name} 平均Macro F1: {avg_f1:.4f}\")\n        \n        all_results[model_type] = {\n            'results': results,\n            'avg_accuracy': avg_acc,\n            'avg_f1_macro': avg_f1\n        }\n    \n    # 比较模型性能\n    print(\"\\n===== 模型性能比较 =====\")\n    for model_type, model_name in models_to_evaluate.items():\n        print(f\"{model_name}: 准确率={all_results[model_type]['avg_accuracy']:.4f}, Macro F1={all_results[model_type]['avg_f1_macro']:.4f}\")\n    \n    # 选择最佳模型并使用全部数据训练\n    best_model_type = max(all_results, key=lambda k: all_results[k]['avg_f1_macro'])\n    print(f\"\\n最佳模型: {models_to_evaluate[best_model_type]}\")\n    \n    if best_model_type == 'two_branch':\n        final_model = build_two_branch_model(pad_len, len(imu_cols), len(tof_cols), len(le.classes_), wd=WD)\n        steps = len(X)//BATCH_SIZE\n        lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n        final_model.compile(\n            optimizer=Adam(lr_sched),\n            loss=tf.keras.losses.CategoricalCrossentropy(label_smoothing=0.1),\n            metrics=['accuracy']\n        )\n        train_gen = MixupGenerator(X, to_categorical(y), batch_size=BATCH_SIZE, alpha=MIXUP_ALPHA)\n        cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True, verbose=1)\n        final_model.fit(train_gen, epochs=EPOCHS, callbacks=[cb], verbose=1)\n        final_model.save(EXPORT_DIR / \"best_model.h5\")\n        \n    elif best_model_type == 'lstm':\n        final_model = build_lstm_model(pad_len, X.shape[2], len(le.classes_), wd=WD)\n        steps = len(X)//BATCH_SIZE\n        lr_sched = tf.keras.optimizers.schedules.CosineDecayRestarts(LR_INIT, first_decay_steps=5*steps)\n        final_model.compile(\n            optimizer=Adam(lr_sched),\n            loss=tf.keras.losses.CategoricalCrossentropy(label_smoothing=0.1),\n            metrics=['accuracy']\n        )\n        train_gen = MixupGenerator(X, to_categorical(y), batch_size=BATCH_SIZE, alpha=MIXUP_ALPHA)\n        cb = EarlyStopping(patience=PATIENCE, restore_best_weights=True, verbose=1)\n        final_model.fit(train_gen, epochs=EPOCHS, callbacks=[cb], verbose=1)\n        final_model.save(EXPORT_DIR / \"best_model.h5\")\n        \n    elif best_model_type == 'xgb':\n        # 提取特征并训练\n        X_features = extract_features(X)\n        final_model = xgb.XGBClassifier(\n            objective='multi:softmax',\n            n_estimators=200,\n            max_depth=10,\n            learning_rate=0.1,\n            use_label_encoder=False,\n            eval_metric='merror'\n        )\n        final_model.fit(X_features, y)\n        joblib.dump(final_model, EXPORT_DIR / \"best_model.pkl\")\n    \n    print(\"✔ Training done – 最佳模型已保存\")\n\nelse:\n    # 推理模式保持不变\n    print(\"▶ INFERENCE MODE – loading artefacts from\", PRETRAINED_DIR)\n    feature_cols   = np.load(PRETRAINED_DIR / \"feature_cols.npy\", allow_pickle=True).tolist()\n    pad_len        = int(np.load(PRETRAINED_DIR / \"sequence_maxlen.npy\"))\n    scaler         = joblib.load(PRETRAINED_DIR / \"scaler.pkl\")\n    gesture_classes = np.load(PRETRAINED_DIR / \"gesture_classes.npy\", allow_pickle=True)\n\n    imu_cols = [c for c in feature_cols if not (c.startswith('thm_') or c.startswith('tof_'))]\n    tof_cols = [c for c in feature_cols if c.startswith('thm_') or c.startswith('tof_')]\n\n    # 加载模型\n    try:\n        # 尝试加载深度学习模型\n        custom_objs = {\n            'time_sum': time_sum,\n            'squeeze_last_axis': squeeze_last_axis,\n            'expand_last_axis': expand_last_axis,\n            'se_block': se_block,\n            'residual_se_cnn_block': residual_se_cnn_block,\n            'attention_layer': attention_layer,\n        }\n        model = load_model(PRETRAINED_DIR / \"best_model.h5\",\n                           compile=False, custom_objects=custom_objs)\n        model_type = 'dl'\n    except:\n        # 尝试加载XGBoost模型\n        model = joblib.load(PRETRAINED_DIR / \"best_model.pkl\")\n        model_type = 'xgb'\n    \n    print(\"  model, scaler, pads loaded – ready for evaluation\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}