Download ZetaNet-v2.py from JoseRFJunior/ZetaNet: direct link, hf CLI and curl.
- Browser
- Download file 13.7 kB
-
https://huggingface.co/JoseRFJunior/ZetaNet/resolve/main/ZetaNet-v2.py
- Command line
-
hf download hf://JoseRFJunior/ZetaNet/ZetaNet-v2.py
-
curl -L -o ZetaNet-v2.py https://huggingface.co/JoseRFJunior/ZetaNet/resolve/main/ZetaNet-v2.py
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() |