ZetaNet / ZetaNet-v2.py
JoseRFJunior's picture
Upload ZetaNet-v2.py
6a77297 verified
Raw History Blame Contribute Delete
13.7 kB
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()