Implementação de Modelo Preditivo com PyTorch para COVID-19

Este guia detalha a implementação de um modelo de rede neural profunda para prever dados relacionados ao COVID-19, cobrindo desde a preparação dos dados até o treinamento e avaliação do modelo.

1. Configuração do Ambiente e Reprodutibilidade

Para garantir que os resultados sejam consistentes entre diferentes execuções, configuramos as sementes aleatórias para a biblioteca principal de computação numérica e para o framework de deep learning. As configurações abaixo forçam comportamentos determinísticos nas operações da GPU.

# Bibliotecas principais
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import numpy as np
import csv
import os
import matplotlib.pyplot as plt

# Configuração para reprodutibilidade
semente = 42069
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
np.random.seed(semente)
torch.manual_seed(semente)
if torch.cuda.is_available():
    torch.cuda.manual_seed_all(semente)

2. Funções Auxiliares

Funções utilitárias para verificar o dispositivo computacional disponível, visualizar o progresso do treinamento e analisra os resultados preditivos.

def obter_dispositivo():
    """Retorna 'cuda' se uma GPU estiver disponível, caso contrário 'cpu'."""
    return "cuda" if torch.cuda.is_available() else "cpu"

def grafico_curva_aprendizado(registro_loss, titulo=''):
    """Plota as curvas de loss de treino e validação ao longo das épocas."""
    total_passos = len(registro_loss['treino'])
    eixo_x1 = range(total_passos)
    passo_validacao = max(1, total_passos // len(registro_loss['validacao']))
    eixo_x2 = eixo_x1[::passo_validacao]

    plt.figure(figsize=(6, 4))
    plt.plot(eixo_x1, registro_loss['treino'], c='red', label='Treino')
    plt.plot(eixo_x2, registro_loss['validacao'], c='cyan', label='Validação')
    plt.ylim(0.0, 5.0)
    plt.xlabel('Passos de Treinamento')
    plt.ylabel('Erro Médio Quadrático (MSE)')
    plt.title(f'Curva de Aprendizado - {titulo}')
    plt.legend()
    plt.show()

def grafico_predicoes(conjunto_dados, modelo, dispositivo, limite=35):
    """Gera um gráfico de dispersão comparando predições com valores reais."""
    modelo.eval()
    predicoes, alvos = [], []
    with torch.no_grad():
        for x, y in conjunto_dados:
            x, y = x.to(dispositivo), y.to(dispositivo)
            pred = modelo(x)
            predicoes.append(pred.detach().cpu())
            alvos.append(y.detach().cpu())

    predicoes_np = torch.cat(predicoes, dim=0).numpy()
    alvos_np = torch.cat(alvos, dim=0).numpy()

    plt.figure(figsize=(5, 5))
    plt.scatter(alvos_np, predicoes_np, c='red', alpha=0.5)
    plt.plot([-0.2, limite], [-0.2, limite], c='blue')  # Linha de predição perfeita
    plt.xlim(-0.2, limite)
    plt.ylim(-0.2, limite)
    plt.xlabel('Valor Real')
    plt.ylabel('Valor Predito')
    plt.title('Valores Reais vs. Predições')
    plt.show()

3. Processamento dos Dados

Definição de uma classe de dataset personalizada para carregar, processar e servir os dados de COVID-19 em diferentes modos (treino, validação, teste).

class ConjuntoDadosCOVID(Dataset):
    def __init__(self, caminho_arquivo, modo='treino', somente_alvo=False):
        self.modo = modo
        with open(caminho_arquivo, 'r') as arquivo:
            dados_brutos = list(csv.reader(arquivo))
            matriz_dados = np.array(dados_brutos[1:])[:, 1:].astype(float)

        if not somente_alvo:
            indices_atributos = list(range(93))
        else:
            # Seleção de subconjunto de atributos
            indices_atributos = list(range(40)) + [57, 75]

        if modo == 'teste':
            self.dados = torch.FloatTensor(matriz_dados[:, indices_atributos])
        else:
            valores_alvo = matriz_dados[:, -1]
            dados_filtrados = matriz_dados[:, indices_atributos]

            if modo == 'treino':
                mascara = np.array([i % 10 != 0 for i in range(len(dados_filtrados))])
            elif modo == 'validacao':
                mascara = np.array([i % 10 == 0 for i in range(len(dados_filtrados))])

            self.dados = torch.FloatTensor(dados_filtrados[mascara])
            self.alvos = torch.FloatTensor(valores_alvo[mascara])

            # Normalização Z-score dos atributos numéricos (a partir da coluna 40)
            media = self.dados[:, 40:].mean(dim=0, keepdim=True)
            desvio_padrao = self.dados[:, 40:].std(dim=0, keepdim=True)
            self.dados[:, 40:] = (self.dados[:, 40:] - media) / (desvio_padrao + 1e-8)

    def __getitem__(self, idx):
        if self.modo in ['treino', 'validacao']:
            return self.dados[idx], self.alvos[idx]
        else:
            return self.dados[idx]

    def __len__(self):
        return len(self.dados)

def criar_dataloader(caminho_arquivo, modo, tamanho_lote, num_trabalhadores=0, somente_alvo=False):
    """Cria e configura um DataLoader para o conjunto de dados especificado."""
    dataset = ConjuntoDadosCOVID(caminho_arquivo, modo=modo, somente_alvo=somente_alvo)
    dataloader = DataLoader(
        dataset,
        batch_size=tamanho_lote,
        shuffle=(modo == 'treino'),
        drop_last=False,
        num_workers=num_trabalhadores,
        pin_memory=True
    )
    return dataloader

4. Definição do Modelo de Rede Neural

Arquitetura de uma rede neural totalmente conectada (fully-connected) para regressão. A estrutura é simples, composta por camadas lineares e uma função de ativação não-linear.

class RedeNeural(nn.Module):
    def __init__(self, dim_entrada):
        super(RedeNeural, self).__init__()
        self.camadas = nn.Sequential(
            nn.Linear(dim_entrada, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )
        self.funcao_perda = nn.MSELoss(reduction='mean')

    def forward(self, x):
        return self.camadas(x).squeeze(1)

    def calcular_perda(self, predicoes, alvos):
        return self.funcao_perda(predicoes, alvos)

5. Rotinas de Treinmaento, Validação e Teste

Funções que orquestram o ciclo completo de aprendizado: ajuste dos pesos do modelo com os dados de treino, avaliação periódica no conujnto de validação e geração das predições finais no conjunto de teste.

def treinar_modelo(dl_treino, dl_validacao, modelo, config, dispositivo):
    """Executa o loop principal de treinamento do modelo."""
    num_epocas = config['num_epocas']
    otimizador = getattr(torch.optim, config['otimizador'])(
        modelo.parameters(), **config['hiperparametros_otimizador']
    )

    melhor_perda_validacao = float('inf')
    historico_perdas = {'treino': [], 'validacao': []}
    epocas_sem_melhoria = 0
    epoca = 0

    while epoca < num_epocas:
        modelo.train()
        for lote_x, lote_y in dl_treino:
            otimizador.zero_grad()
            lote_x, lote_y = lote_x.to(dispositivo), lote_y.to(dispositivo)
            predicoes = modelo(lote_x)
            perda = modelo.calcular_perda(predicoes, lote_y)
            perda.backward()
            otimizador.step()
            historico_perdas['treino'].append(perda.item())

        perda_atual_val = validar_modelo(dl_validacao, modelo, dispositivo)
        historico_perdas['validacao'].append(perda_atual_val)

        if perda_atual_val < melhor_perda_validacao:
            melhor_perda_validacao = perda_atual_val
            torch.save(modelo.state_dict(), config['caminho_modelo'])
            epocas_sem_melhoria = 0
        else:
            epocas_sem_melhoria += 1

        epoca += 1
        if epocas_sem_melhoria >= config['paciencia_early_stop']:
            break

    return melhor_perda_validacao, historico_perdas

def validar_modelo(dl_validacao, modelo, dispositivo):
    """Calcula a loss média no conjunto de validação."""
    modelo.eval()
    perda_total = 0.0
    with torch.no_grad():
        for lote_x, lote_y in dl_validacao:
            lote_x, lote_y = lote_x.to(dispositivo), lote_y.to(dispositivo)
            predicoes = modelo(lote_x)
            perda_lote = modelo.calcular_perda(predicoes, lote_y)
            perda_total += perda_lote.item() * len(lote_x)
    return perda_total / len(dl_validacao.dataset)

def realizar_teste(dl_teste, modelo, dispositivo):
    """Gera as predições finais para o conjunto de teste."""
    modelo.eval()
    lista_predicoes = []
    with torch.no_grad():
        for lote_x in dl_teste:
            lote_x = lote_x.to(dispositivo)
            predicoes = modelo(lote_x)
            lista_predicoes.append(predicoes.cpu())
    return torch.cat(lista_predicoes, dim=0).numpy()

Considerações sobre Otimização

O desempenho de um modelo depende de fatores como a quantidade e qualidade dos dados, a relevância das features selecionadas e a capacidade da arquitetura da rede. Caso o desempenho inicial seja insatisfatório, estratégias como a engenharia de atributos (feature engineering), ajuste de hiperparâmetros (taxa de aprendizado, tamanho do lote, profundidade da rede) ou a aplicação de técnicas de regularização podem ser exploradas para melhorar a capacidade de generalização do modelo.

Tags: Pytorch Rede Neural Profunda (DNN) Pré-processamento de Dados COVID-19 regressão

Publicado em 6-6 00:09 por Thomas