本實戰內容取自筆者參加的首屆中國心電智能大賽項目，初賽要求為設計一個自動識別心電圖波形算法。筆者使用Keras框架設計了基于Conv1D結構的模型，并且開源了代碼作為Baseline。內容包括數據預處理，模型搭建，網絡訓練，模型應用等，此Baseline采用最簡單的一維卷積達到了88%測試準確率。有多支隊伍在筆者基線代碼基礎上調優取得了優異成績，順利進入復賽。

數據介紹

下載完整的訓練集和測試集，共1000例常規心電圖，其中訓練集中包含600例，測試集中共400例。該數據是從多個公開數據集中獲取。參賽團隊需要利用有正常/異常兩類標簽的訓練集數據設計和實現算法，并在沒有標簽的測試集上做出預測。

該心電數據的采樣率為500 Hz。為了方便參賽團隊用不同編程語言都能讀取數據，所有心電數據的存儲格式為MAT格式。該文件中存儲了12個導聯的電壓信號。訓練數據對應的標簽存儲在txt文件中，其中0代表正常，1代表異常。

賽題分析

簡單分析一下，初賽的數據集共有1000個樣本，其中訓練集中包含600例，測試集中共400例。其中訓練集中包含600例是具有label的，可以用于我們訓練模型；測試集中共400例沒有標簽，需要我們使用訓練好的模型進行預測。

賽題就是一個二分類預測問題，解題思路應該包括以下內容

數據讀取與處理

網絡模型搭建

模型的訓練

模型應用與提交預測結果

實戰應用

經過對賽題的分析，我們把任務分成四個小任務，首先第一步是：

1.數據讀取與處理

該心電數據的采樣率為500 Hz。為了方便參賽團隊用不同編程語言都能讀取數據，所有心電數據的存儲格式為MAT格式。該文件中存儲了12個導聯的電壓信號。訓練數據對應的標簽存儲在txt文件中，其中0代表正常，1代表異常。

我們由上述描述可以得知，

我們的數據保存在MAT格式文件中（這決定了后面我們要如何讀取數據）

采樣率為500 Hz（這個信息并沒有怎么用到，大家可以簡單了解一下，就是1秒采集500個點，由后面我們得知每個數據都是5000個點，也就是10秒的心電圖片）

12個導聯的電壓信號（這個是指采用12種導聯方式，大家可以簡單理解為用12個體溫計量體溫，從而得到更加準確的信息，下圖為導聯方式簡單介紹，大家了解下即可。要注意的是，既然提供了12種導聯，我們應該全部都用到，雖然我們僅使用一種導聯方式也可以進行訓練與預測，但是經驗告訴我們，采取多個特征會取得更優效果）

數據處理函數定義：

import kerasfrom scipy.io import loadmatimport matplotlib.pyplot as pltimport globimport numpy as npimport pandas as pdimport mathimport osfrom keras.layers import *from keras.models import *from keras.objectives import *BASE_DIR = “preliminary/TRAIN/”#進行歸一化def normalize（v）： return （v - v.mean（axis=1）.reshape（（v.shape［0］，1））） / （v.max（axis=1）.reshape（（v.shape［0］，1）） + 2e-12）loadmat打開文件def get_feature（wav_file，Lens = 12，BASE_DIR=BASE_DIR）： mat = loadmat（BASE_DIR+wav_file） dat = mat［“data”］ feature = dat［0:12］ return（normalize（feature）.transopse（））#把標簽轉成oneHot形式def convert2oneHot（index，Lens）： hot = np.zeros（（Lens，）） hot［index］ = 1 return（hot）TXT_DIR = “preliminary/reference.txt”MANIFEST_DIR = “preliminary/reference.csv”

讀取一條數據進行顯示

if name__ == “__main”： dat1 = get_feature（“preliminary/TRAIN/TRAIN101.mat”） print（dat1.shape） #one data shape is （12， 5000） plt.plt（dat1［：，0］） plt.show（）

我們由上述信息可以看出每種導聯都是由5000個點組成的列表，12種導聯方式使每個樣本都是12*5000的矩陣，類似于一張分辨率為12x5000的照片。

我們需要處理的就是把每個讀取出來，歸一化一下，送入網絡進行訓練可以了。

標簽處理方式

def create_csv（TXT_DIR=TXT_DIR）： lists = pd.read_csv（TXT_DIR，sep=r“\t”，header=None） lists = lists.sample（frac=1） lists.to_csv（MANIFEST_DIR，index=None） print（“Finish save csv”）

我這里是采用從reference.txt讀取，然后打亂保存到reference.csv中，注意一定要進行數據打亂操作，不然訓練效果很差。因為原始數據前面便簽全部是1，后面全部是0

數據迭代方式

Batch_size = 20def xs_gen（path=MANIFEST_DIR，batch_size = Batch_size，train=True）：img_list = pd.read_csv（path）if train ： img_list = np.array（img_list）［：500］ print（“Found %s train items.”%len（img_list）） print（“list 1 is”，img_list［0］） steps = math.ceil（len（img_list） / batch_size） # 確定每輪有多少個batchelse： img_list = np.array（img_list）［500：］ print（“Found %s test items.”%len（img_list）） print（“list 1 is”，img_list［0］） steps = math.ceil（len（img_list） / batch_size） # 確定每輪有多少個batchwhile True： for i in range（steps）： batch_list = img_list［i * batch_size ： i * batch_size + batch_size］ np.random.shuffle（batch_list） batch_x = np.array（［get_feature（file） for file in batch_list［：，0］］） batch_y = np.array（［convert2oneHot（label，2） for label in batch_list［：，1］］） yield batch_x， batch_y

數據讀取的方式我采用的是生成器的方式，這樣可以按batch讀取，加快訓練速度，大家也可以采用一下全部讀取，看個人的習慣了

2.網絡模型搭建

數據我們處理好了，后面就是模型的搭建了，我使用keras搭建的，操作簡單便捷，tf，pytorch，sklearn大家可以按照自己喜好來。

網絡模型可以選擇CNN，RNN，Attention結構，或者多模型的融合，拋磚引玉，此Baseline采用的一維CNN方式，一維CNN學習地址

模型搭建

TIME_PERIODS = 5000num_sensors = 12def build_model（input_shape=（TIME_PERIODS，num_sensors），num_classes=2）： model = Sequential（） #model.add（Reshape（（TIME_PERIODS， num_sensors）， input_shape=input_shape）） model.add（Conv1D（16， 16，strides=2， activation=‘relu’，input_shape=input_shape）） model.add（Conv1D（16， 16，strides=2， activation=‘relu’，padding=“same”）） model.add（MaxPooling1D（2）） model.add（Conv1D（64， 8，strides=2， activation=‘relu’，padding=“same”）） model.add（Conv1D（64， 8，strides=2， activation=‘relu’，padding=“same”）） model.add（MaxPooling1D（2）） model.add（Conv1D（128， 4，strides=2， activation=‘relu’，padding=“same”）） model.add（Conv1D（128， 4，strides=2， activation=‘relu’，padding=“same”）） model.add（MaxPooling1D（2）） model.add（Conv1D（256， 2，strides=1， activation=‘relu’，padding=“same”）） model.add（Conv1D（256， 2，strides=1， activation=‘relu’，padding=“same”）） model.add（MaxPooling1D（2）） model.add（GlobalAveragePooling1D（）） model.add（Dropout（0.3）） model.add（Dense（num_classes， activation=‘softmax’）） return（model）

用model.summary（）輸出的網絡模型為

訓練參數比較少，大家可以根據自己想法更改。

3.網絡模型訓練

模型訓練

if name__ == “__main”： “”“dat1 = get_feature（”TRAIN101.mat“） print（”one data shape is“，dat1.shape） #one data shape is （12， 5000） plt.plot（dat1［0］） plt.show（）”“” if （os.path.exists（MANIFEST_DIR）==False）： create_csv（） train_iter = xs_gen（train=True） test_iter = xs_gen（train=False） model = build_model（） print（model.summary（）） ckpt = keras.callbacks.ModelCheckpoint（ filepath=‘best_model.{epoch:02d}-{val_acc：.2f}.h5’， monitor=‘val_acc’， save_best_only=True，verbose=1） model.compile（loss=‘categorical_crossentropy’， optimizer=‘adam’， metrics=［‘accuracy’］） model.fit_generator（ generator=train_iter， steps_per_epoch=500//Batch_size， epochs=20， initial_epoch=0， validation_data = test_iter， nb_val_samples = 100//Batch_size， callbacks=［ckpt］， ）

訓練過程輸出（最優結果：loss： 0.0565 - acc： 0.9820 - val_loss： 0.8307 - val_acc： 0.8800）

Epoch 10/2025/25 ［==============================］ - 1s 37ms/step - loss： 0.2329 - acc： 0.9040 - val_loss： 0.4041 - val_acc： 0.8700Epoch 00010： val_acc improved from 0.85000 to 0.87000， saving model to best_model.10-0.87.h5Epoch 11/2025/25 ［==============================］ - 1s 38ms/step - loss： 0.1633 - acc： 0.9380 - val_loss： 0.5277 - val_acc： 0.8300Epoch 00011： val_acc did not improve from 0.87000Epoch 12/2025/25 ［==============================］ - 1s 40ms/step - loss： 0.1394 - acc： 0.9500 - val_loss： 0.4916 - val_acc： 0.7400Epoch 00012： val_acc did not improve from 0.87000Epoch 13/2025/25 ［==============================］ - 1s 38ms/step - loss： 0.1746 - acc： 0.9220 - val_loss： 0.5208 - val_acc： 0.8100Epoch 00013： val_acc did not improve from 0.87000Epoch 14/2025/25 ［==============================］ - 1s 38ms/step - loss： 0.1009 - acc： 0.9720 - val_loss： 0.5513 - val_acc： 0.8000Epoch 00014： val_acc did not improve from 0.87000Epoch 15/2025/25 ［==============================］ - 1s 38ms/step - loss： 0.0565 - acc： 0.9820 - val_loss： 0.8307 - val_acc： 0.8800Epoch 00015： val_acc improved from 0.87000 to 0.88000， saving model to best_model.15-0.88.h5Epoch 16/2025/25 ［==============================］ - 1s 38ms/step - loss： 0.0261 - acc： 0.9920 - val_loss： 0.6443 - val_acc： 0.8400Epoch 00016： val_acc did not improve from 0.88000Epoch 17/2025/25 ［==============================］ - 1s 38ms/step - loss： 0.0178 - acc： 0.9960 - val_loss： 0.7773 - val_acc： 0.8700Epoch 00017： val_acc did not improve from 0.88000Epoch 18/2025/25 ［==============================］ - 1s 38ms/step - loss： 0.0082 - acc： 0.9980 - val_loss： 0.8875 - val_acc： 0.8600Epoch 00018： val_acc did not improve from 0.88000Epoch 19/2025/25 ［==============================］ - 1s 37ms/step - loss： 0.0045 - acc： 1.0000 - val_loss： 1.0057 - val_acc： 0.8600Epoch 00019： val_acc did not improve from 0.88000Epoch 20/2025/25 ［==============================］ - 1s 37ms/step - loss： 0.0012 - acc： 1.0000 - val_loss： 1.1088 - val_acc： 0.8600Epoch 00020： val_acc did not improve from 0.88000

4.模型應用預測結果

預測數據

if name__ == “__main”： “”“dat1 = get_feature（”TRAIN101.mat“） print（”one data shape is“，dat1.shape） #one data shape is （12， 5000） plt.plot（dat1［0］） plt.show（）”“” “”“if （os.path.exists（MANIFEST_DIR）==False）： create_csv（） train_iter = xs_gen（train=True） test_iter = xs_gen（train=False） model = build_model（） print（model.summary（）） ckpt = keras.callbacks.ModelCheckpoint（ filepath=‘best_model.{epoch:02d}-{val_acc：.2f}.h5’， monitor=‘val_acc’， save_best_only=True，verbose=1） model.compile（loss=‘categorical_crossentropy’， optimizer=‘adam’， metrics=［‘accuracy’］） model.fit_generator（ generator=train_iter， steps_per_epoch=500//Batch_size， epochs=20， initial_epoch=0， validation_data = test_iter， nb_val_samples = 100//Batch_size， callbacks=［ckpt］， ）”“” PRE_DIR = “sample_codes/answers.txt” model = load_model（“best_model.15-0.88.h5”） pre_lists = pd.read_csv（PRE_DIR，sep=r“ ”，header=None） print（pre_lists.head（）） pre_datas = np.array（［get_feature（item，BASE_DIR=“preliminary/TEST/”） for item in pre_lists［0］］） pre_result = model.predict_classes（pre_datas）#0-1概率預測 print（pre_result.shape） pre_lists［1］ = pre_result pre_lists.to_csv（“sample_codes/answers1.txt”，index=None，header=None） print（“predict finish”）

下面是前十條預測結果：

TEST394，0TEST313，1TEST484，0TEST288，0TEST261，1TEST310，0TEST286，1TEST367，1TEST149，1TEST160，1

展望

此Baseline采用最簡單的一維卷積達到了88%測試準確率（可能會因為隨機初始化值上下波動），大家也可以多嘗試GRU，Attention，和Resnet等結果，測試準確率會突破90+。