Upload ZetaNet-v2.py

ZetaNet-v2.py

@@ -0,0 +1,383 @@
+import pandas as pd
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.optim as optim
+import matplotlib.pyplot as plt
+from sklearn.model_selection import train_test_split
+from sklearn.preprocessing import StandardScaler
+from time import time
+import warnings
+warnings.filterwarnings('ignore')
+
+class ImprovedZetaNet(nn.Module):
+    def __init__(self, input_size=2, hidden_sizes=[128, 256, 128, 64], output_size=2, dropout_rate=0.1):
+        super(ImprovedZetaNet, self).__init__()
+        
+        # Construir camadas dinamicamente
+        layers = []
+        prev_size = input_size
+        
+        for hidden_size in hidden_sizes:
+            layers.extend([
+                nn.Linear(prev_size, hidden_size),
+                nn.BatchNorm1d(hidden_size),
+                nn.ReLU(),
+                nn.Dropout(dropout_rate)
+            ])
+            prev_size = hidden_size
+        
+        # Camada de saída sem ativação
+        layers.append(nn.Linear(prev_size, output_size))
+        
+        self.network = nn.Sequential(*layers)
+        
+        # Inicialização Xavier/Glorot
+        self._initialize_weights()
+    
+    def _initialize_weights(self):
+        for module in self.modules():
+            if isinstance(module, nn.Linear):
+                nn.init.xavier_normal_(module.weight)
+                if module.bias is not None:
+                    nn.init.constant_(module.bias, 0)
+    
+    def forward(self, x):
+        return self.network(x)
+
+class ZetaTrainer:
+    def __init__(self, model, device='cpu'):
+        self.model = model.to(device)
+        self.device = device
+        self.train_losses = []
+        self.val_losses = []
+        
+    def train_epoch(self, train_loader, optimizer, criterion):
+        self.model.train()
+        total_loss = 0
+        num_batches = 0
+        
+        for batch_x, batch_y in train_loader:
+            batch_x, batch_y = batch_x.to(self.device), batch_y.to(self.device)
+            
+            optimizer.zero_grad()
+            predictions = self.model(batch_x)
+            loss = criterion(predictions, batch_y)
+            loss.backward()
+            
+            # Gradient clipping para estabilidade
+            torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)
+            
+            optimizer.step()
+            
+            total_loss += loss.item()
+            num_batches += 1
+            
+        return total_loss / num_batches
+    
+    def validate(self, val_loader, criterion):
+        self.model.eval()
+        total_loss = 0
+        num_batches = 0
+        
+        with torch.no_grad():
+            for batch_x, batch_y in val_loader:
+                batch_x, batch_y = batch_x.to(self.device), batch_y.to(self.device)
+                predictions = self.model(batch_x)
+                loss = criterion(predictions, batch_y)
+                total_loss += loss.item()
+                num_batches += 1
+                
+        return total_loss / num_batches
+    
+    def train(self, train_loader, val_loader, epochs=200, learning_rate=0.001, patience=20):
+        # Usar Adam com weight decay
+        optimizer = optim.AdamW(self.model.parameters(), lr=learning_rate, weight_decay=1e-5)
+        
+        # Learning rate scheduler
+        scheduler = optim.lr_scheduler.ReduceLROnPlateau(
+            optimizer, mode='min', factor=0.5, patience=10, verbose=True
+        )
+        
+        criterion = nn.MSELoss()
+        best_val_loss = float('inf')
+        patience_counter = 0
+        
+        print(f"Iniciando treinamento por {epochs} épocas...")
+        print("-" * 60)
+        
+        for epoch in range(epochs):
+            # Treinar
+            train_loss = self.train_epoch(train_loader, optimizer, criterion)
+            
+            # Validar
+            val_loss = self.validate(val_loader, criterion)
+            
+            # Atualizar scheduler
+            scheduler.step(val_loss)
+            
+            # Salvar histórico
+            self.train_losses.append(train_loss)
+            self.val_losses.append(val_loss)
+            
+            # Early stopping
+            if val_loss < best_val_loss:
+                best_val_loss = val_loss
+                patience_counter = 0
+                # Salvar melhor modelo
+                torch.save(self.model.state_dict(), 'best_zetanet.pth')
+            else:
+                patience_counter += 1
+            
+            # Print progress
+            if (epoch + 1) % 20 == 0 or epoch == 0:
+                current_lr = optimizer.param_groups[0]['lr']
+                print(f"Época {epoch+1:3d}/{epochs} | "
+                      f"Train Loss: {train_loss:.6f} | "
+                      f"Val Loss: {val_loss:.6f} | "
+                      f"LR: {current_lr:.2e}")
+            
+            # Early stopping
+            if patience_counter >= patience:
+                print(f"\nEarly stopping na época {epoch+1}")
+                break
+        
+        # Carregar melhor modelo
+        self.model.load_state_dict(torch.load('best_zetanet.pth'))
+        print(f"\nTreinamento concluído! Melhor perda de validação: {best_val_loss:.6f}")
+
+def parse_complex_improved(value):
+    """Função melhorada para parsing de números complexos"""
+    if pd.isna(value):
+        return np.nan
+    
+    value = str(value).strip()
+    
+    # Remover parênteses
+    value = value.replace('(', '').replace(')', '')
+    
+    # Substituir vírgulas por pontos
+    value = value.replace(',', '.')
+    
+    # Casos especiais
+    if value == '' or value.lower() == 'nan':
+        return np.nan
+    
+    try:
+        # Se não tem 'j' ou 'i', adicionar 'j' no final
+        if 'j' not in value.lower() and 'i' not in value.lower():
+            if '+' in value or '-' in value[1:]:  # Tem parte real e imaginária
+                value += 'j'
+            else:  # Só parte real
+                return complex(float(value), 0)
+        
+        # Substituir 'i' por 'j'
+        value = value.replace('i', 'j')
+        
+        return complex(value)
+    except (ValueError, TypeError):
+        return np.nan
+
+def load_and_preprocess_data(filepath, test_size=0.2, random_state=42):
+    """Carrega e preprocessa os dados com melhor tratamento de erros"""
+    print("Carregando dados...")
+    
+    try:
+        data = pd.read_csv(filepath)
+        print(f"Dados carregados: {len(data)} amostras")
+    except FileNotFoundError:
+        print(f"Arquivo {filepath} não encontrado!")
+        return None
+    
+    # Limpar e converter dados complexos
+    print("Processando números complexos...")
+    data['s'] = data['s'].apply(parse_complex_improved)
+    data['zeta(s)'] = data['zeta(s)'].apply(parse_complex_improved)
+    
+    # Remover valores inválidos
+    initial_len = len(data)
+    data = data.dropna()
+    final_len = len(data)
+    
+    if final_len < initial_len:
+        print(f"Removidas {initial_len - final_len} amostras inválidas")
+    
+    if len(data) == 0:
+        print("Nenhum dado válido encontrado!")
+        return None
+    
+    # Separar partes real e imaginária
+    data['s_real'] = data['s'].apply(lambda x: x.real)
+    data['s_imag'] = data['s'].apply(lambda x: x.imag)
+    data['zeta_real'] = data['zeta(s)'].apply(lambda x: x.real)
+    data['zeta_imag'] = data['zeta(s)'].apply(lambda x: x.imag)
+    
+    # Preparar features e targets
+    X = data[['s_real', 's_imag']].values
+    y = data[['zeta_real', 'zeta_imag']].values
+    
+    # Split treino/validação
+    X_train, X_val, y_train, y_val = train_test_split(
+        X, y, test_size=test_size, random_state=random_state
+    )
+    
+    # Normalização robusta
+    scaler_X = StandardScaler()
+    scaler_y = StandardScaler()
+    
+    X_train_scaled = scaler_X.fit_transform(X_train)
+    X_val_scaled = scaler_X.transform(X_val)
+    y_train_scaled = scaler_y.fit_transform(y_train)
+    y_val_scaled = scaler_y.transform(y_val)
+    
+    # Converter para tensores
+    X_train_tensor = torch.FloatTensor(X_train_scaled)
+    X_val_tensor = torch.FloatTensor(X_val_scaled)
+    y_train_tensor = torch.FloatTensor(y_train_scaled)
+    y_val_tensor = torch.FloatTensor(y_val_scaled)
+    
+    print(f"Dados preprocessados:")
+    print(f"  Treino: {len(X_train_tensor)} amostras")
+    print(f"  Validação: {len(X_val_tensor)} amostras")
+    
+    return {
+        'train': (X_train_tensor, y_train_tensor),
+        'val': (X_val_tensor, y_val_tensor),
+        'scalers': (scaler_X, scaler_y),
+        'raw_data': data
+    }
+
+def create_data_loaders(data_dict, batch_size=64):
+    """Cria DataLoaders do PyTorch"""
+    train_dataset = torch.utils.data.TensorDataset(
+        data_dict['train'][0], data_dict['train'][1]
+    )
+    val_dataset = torch.utils.data.TensorDataset(
+        data_dict['val'][0], data_dict['val'][1]
+    )
+    
+    train_loader = torch.utils.data.DataLoader(
+        train_dataset, batch_size=batch_size, shuffle=True
+    )
+    val_loader = torch.utils.data.DataLoader(
+        val_dataset, batch_size=batch_size, shuffle=False
+    )
+    
+    return train_loader, val_loader
+
+def plot_results(trainer, data_dict, model):
+    """Plota resultados do treinamento e predições"""
+    fig, axes = plt.subplots(2, 2, figsize=(15, 12))
+    
+    # 1. Curvas de perda
+    axes[0,0].plot(trainer.train_losses, label='Treino', alpha=0.8)
+    axes[0,0].plot(trainer.val_losses, label='Validação', alpha=0.8)
+    axes[0,0].set_xlabel('Época')
+    axes[0,0].set_ylabel('MSE Loss')
+    axes[0,0].set_title('Curvas de Aprendizado')
+    axes[0,0].legend()
+    axes[0,0].grid(True, alpha=0.3)
+    axes[0,0].set_yscale('log')
+    
+    # 2. Predições vs Real (parte real)
+    model.eval()
+    with torch.no_grad():
+        X_val, y_val = data_dict['val']
+        predictions = model(X_val)
+        
+        # Denormalizar
+        scaler_y = data_dict['scalers'][1]
+        y_val_denorm = scaler_y.inverse_transform(y_val.numpy())
+        pred_denorm = scaler_y.inverse_transform(predictions.numpy())
+    
+    axes[0,1].scatter(y_val_denorm[:, 0], pred_denorm[:, 0], alpha=0.6, s=1)
+    axes[0,1].plot([y_val_denorm[:, 0].min(), y_val_denorm[:, 0].max()], 
+                   [y_val_denorm[:, 0].min(), y_val_denorm[:, 0].max()], 'r--')
+    axes[0,1].set_xlabel('ζ(s) Real - Valor Real')
+    axes[0,1].set_ylabel('ζ(s) Real - Predição')
+    axes[0,1].set_title('Parte Real: Predição vs Real')
+    axes[0,1].grid(True, alpha=0.3)
+    
+    # 3. Predições vs Real (parte imaginária)
+    axes[1,0].scatter(y_val_denorm[:, 1], pred_denorm[:, 1], alpha=0.6, s=1)
+    axes[1,0].plot([y_val_denorm[:, 1].min(), y_val_denorm[:, 1].max()], 
+                   [y_val_denorm[:, 1].min(), y_val_denorm[:, 1].max()], 'r--')
+    axes[1,0].set_xlabel('ζ(s) Imag - Valor Real')
+    axes[1,0].set_ylabel('ζ(s) Imag - Predição')
+    axes[1,0].set_title('Parte Imaginária: Predição vs Real')
+    axes[1,0].grid(True, alpha=0.3)
+    
+    # 4. Distribuição dos erros
+    errors_real = np.abs(y_val_denorm[:, 0] - pred_denorm[:, 0])
+    errors_imag = np.abs(y_val_denorm[:, 1] - pred_denorm[:, 1])
+    
+    axes[1,1].hist(errors_real, bins=50, alpha=0.7, label='Erro Parte Real')
+    axes[1,1].hist(errors_imag, bins=50, alpha=0.7, label='Erro Parte Imag')
+    axes[1,1].set_xlabel('Erro Absoluto')
+    axes[1,1].set_ylabel('Frequência')
+    axes[1,1].set_title('Distribuição dos Erros')
+    axes[1,1].legend()
+    axes[1,1].grid(True, alpha=0.3)
+    axes[1,1].set_yscale('log')
+    
+    plt.tight_layout()
+    plt.savefig('zetanet_results.png', dpi=300, bbox_inches='tight')
+    plt.show()
+    
+    # Estatísticas
+    print(f"\nEstatísticas de Erro:")
+    print(f"Erro médio (parte real): {errors_real.mean():.6f}")
+    print(f"Erro médio (parte imag): {errors_imag.mean():.6f}")
+    print(f"Erro máximo (parte real): {errors_real.max():.6f}")
+    print(f"Erro máximo (parte imag): {errors_imag.max():.6f}")
+
+def main():
+    start_time = time()
+    
+    # Configurações
+    FILEPATH = "/content/combined_zeta_data.csv"  # Ajuste o caminho
+    DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+    print(f"Usando dispositivo: {DEVICE}")
+    
+    # Carregar e preprocessar dados
+    data_dict = load_and_preprocess_data(FILEPATH)
+    if data_dict is None:
+        return
+    
+    # Criar data loaders
+    train_loader, val_loader = create_data_loaders(data_dict, batch_size=128)
+    
+    # Criar modelo melhorado
+    model = ImprovedZetaNet(
+        input_size=2,
+        hidden_sizes=[128, 256, 256, 128, 64],
+        output_size=2,
+        dropout_rate=0.1
+    )
+    
+    print(f"\nArquitetura do modelo:")
+    print(model)
+    
+    # Contar parâmetros
+    total_params = sum(p.numel() for p in model.parameters())
+    trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
+    print(f"\nParâmetros totais: {total_params:,}")
+    print(f"Parâmetros treináveis: {trainable_params:,}")
+    
+    # Treinar modelo
+    trainer = ZetaTrainer(model, DEVICE)
+    trainer.train(
+        train_loader, val_loader, 
+        epochs=300, 
+        learning_rate=0.001, 
+        patience=30
+    )
+    
+    # Plotar resultados
+    plot_results(trainer, data_dict, model)
+    
+    end_time = time()
+    print(f"\nTempo total de execução: {(end_time - start_time):.2f} segundos")
+
+if __name__ == "__main__":
+    main()