基于CNN和BiGRU特征融合的八分类模型实现

本文使用CNN和BiGRU模型并行作为特征提取器，并将两个提取器的结果进行融合，最终使用全连接层进行八分类。代码使用PyTorch实现，并包含数据集的加载、预处理、模型训练和测试等步骤。

数据集

在一txt文件夹中有如上格式的数据集，每条样本前23位为特征值，最后一位为分类标签，标签共有8个类别。

7,7,183,233,10,10,3,10,3,10,0,25,21,42,194,0,0,2,1,0,0,32,1.23,4
7,7,183,233,10,10,3,10,3,10,0,25,21,0,0,2,78,2,1,0,0,86.6685638427734,1.25,4
7,7,183,233,10,10,3,10,3,10,0,25,21,90,80,20,10,2,1,0,0,86.64013671875,1.30,0
7,7,183,233,10,10,3,10,3,10,0,25,21,90,80,20,10,2,1,0,0,86.4980087280273,1.10,0
7,0,183,0,9,0,3,10,3,0,0,25,123,90,80,20,10,0,1,0,1,0,1.00,7
7,0,183,0,9,0,3,10,3,0,0,25,123,90,80,20,10,0,1,0,1,0,1.00,7

模型设计

CNN模型
- 输入数据需要转化为图片格式
- 使用两层卷积层和一层全连接层
BiGRU模型
- 输入数据为23位特征值
- 使用双向GRU层
特征融合
- 将CNN和BiGRU提取的特征向量进行拼接
全连接层
- 对融合后的特征进行八分类

代码实现

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import DataLoader, Dataset

import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
from PIL import Image

# Define the CNN model
class CNN(nn.Module):
    def __init__(self):
        super(CNN, self).__init__()
        self.conv1 = nn.Conv2d(1, 16, kernel_size=3, stride=1, padding=1)
        self.conv2 = nn.Conv2d(16, 32, kernel_size=3, stride=1, padding=1)
        self.fc1 = nn.Linear(32 * 7 * 7, 128)
        
    def forward(self, x):
        x = F.relu(self.conv1(x))
        x = F.max_pool2d(x, 2)
        x = F.relu(self.conv2(x))
        x = F.max_pool2d(x, 2)
        x = x.view(x.size(0), -1)
        x = F.relu(self.fc1(x))
        return x

# Define the BiGRU model
class BiGRU(nn.Module):
    def __init__(self):
        super(BiGRU, self).__init__()
        self.gru = nn.GRU(input_size=23, hidden_size=64, num_layers=2, bidirectional=True, batch_first=True)
        
    def forward(self, x):
        output, _ = self.gru(x)
        return output[:, -1, :]

# Define the fusion model
class FusionModel(nn.Module):
    def __init__(self):
        super(FusionModel, self).__init__()
        self.cnn = CNN()
        self.bigru = BiGRU()
        self.fc = nn.Linear(192, 8)
        
    def forward(self, x1, x2):
        x1 = self.cnn(x1)
        x2 = self.bigru(x2)
        x = torch.cat((x1, x2), dim=1)
        x = self.fc(x)
        return x

# Define the custom dataset
class CustomDataset(Dataset):
    def __init__(self, data, labels):
        self.data = data
        self.labels = labels
        
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        x = self.data[idx]
        y = self.labels[idx]
        return x, y

# Load the dataset
data = pd.read_csv('data.txt', header=None)
features = data.iloc[:, :-1].values
labels = data.iloc[:, -1].values

# Normalize the features
scaler = MinMaxScaler()
features = scaler.fit_transform(features)

# Split the dataset into train, validation, and test sets
X_train, X_test, y_train, y_test = train_test_split(features, labels, test_size=0.2, stratify=labels, random_state=42)
X_train, X_val, y_train, y_val = train_test_split(X_train, y_train, test_size=0.2, stratify=y_train, random_state=42)

# Convert the data to tensors
X_train = torch.Tensor(X_train)
X_val = torch.Tensor(X_val)
X_test = torch.Tensor(X_test)
y_train = torch.LongTensor(y_train)
y_val = torch.LongTensor(y_val)
y_test = torch.LongTensor(y_test)

# Create the dataloaders
train_dataset = CustomDataset(X_train, y_train)
val_dataset = CustomDataset(X_val, y_val)
test_dataset = CustomDataset(X_test, y_test)

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)
test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

# Initialize the fusion model
model = FusionModel()

# Define the loss function and optimizer
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# Train the model
num_epochs = 10
best_val_loss = float('inf')
for epoch in range(num_epochs):
    # Training
    model.train()
    train_loss = 0.0
    for inputs, labels in train_loader:
        optimizer.zero_grad()
        outputs = model(inputs.unsqueeze(1), inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        train_loss += loss.item() * inputs.size(0)
    train_loss /= len(train_loader.dataset)
    
    # Validation
    model.eval()
    val_loss = 0.0
    with torch.no_grad():
        for inputs, labels in val_loader:
            outputs = model(inputs.unsqueeze(1), inputs)
            loss = criterion(outputs, labels)
            val_loss += loss.item() * inputs.size(0)
        val_loss /= len(val_loader.dataset)
    
    # Save the best model
    if val_loss < best_val_loss:
        best_val_loss = val_loss
        torch.save(model.state_dict(), 'best_model.pt')
    
    print(f'Epoch {epoch+1}/{num_epochs}, Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}')

# Load the best model
model.load_state_dict(torch.load('best_model.pt'))

# Test the model
model.eval()
test_loss = 0.0
correct = 0
total = 0
with torch.no_grad():
    for inputs, labels in test_loader:
        outputs = model(inputs.unsqueeze(1), inputs)
        loss = criterion(outputs, labels)
        test_loss += loss.item() * inputs.size(0)
        _, predicted = torch.max(outputs.data, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()
test_loss /= len(test_loader.dataset)
accuracy = 100 * correct / total
print(f'Test Loss: {test_loss:.4f}, Accuracy: {accuracy:.2f}%')

注意

代码中使用了MinMaxScaler对特征进行归一化处理
将特征转换为图片格式的具体实现需要根据实际情况进行调整
代码中使用了torch.save和torch.load对模型进行保存和加载
训练过程中可以使用TensorBoard等工具进行可视化
可以根据实际情况调整模型参数，例如学习率、batch size等