| import os |
| import sys |
| import torch |
| import json |
| import numpy as np |
| import pandas as pd |
| from torch.utils.data import DataLoader |
| from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score |
| import matplotlib.pyplot as plt |
|
|
| sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) |
|
|
| from src.model import QuarterlyStockDataset, LSTMAutoEncoder |
|
|
| if __name__ == "__main__": |
| base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| processed_data_path = os.path.join(base_dir, 'data', 'processed', 'stock_data.parquet') |
| model_path = os.path.join(base_dir, 'models', 'lstm_autoencoder.pth') |
| metrics_path = os.path.join(base_dir, 'resources', 'metrics.json') |
|
|
| device = 'cuda' if torch.cuda.is_available() else 'cpu' |
| df = pd.read_parquet(processed_data_path) |
|
|
| test_df = df[df['Date'] >= '2024-01-01'] |
|
|
| sequence_length = 90 |
| test_dataset = QuarterlyStockDataset(test_df, sequence_length=sequence_length) |
| test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False) |
|
|
| best_params = { |
| 'hidden_size': 64, |
| 'latent_dim': 32, |
| 'num_layers': 2, |
| 'embed_dim': 16 |
| } |
|
|
| num_tickers = df['Ticker_Encoded'].nunique() |
|
|
| model = LSTMAutoEncoder( |
| input_dim=5, |
| num_tickers=num_tickers, |
| embed_dim=best_params['embed_dim'], |
| hidden_size=best_params['hidden_size'], |
| latent_dim=best_params['latent_dim'], |
| num_layers=best_params['num_layers'] |
| ).to(device) |
|
|
| model.load_state_dict(torch.load(model_path, map_location=device)) |
| model.eval() |
|
|
| all_actual, all_recon = [], [] |
|
|
| with torch.no_grad(): |
| for batch_x, batch_ticker in test_loader: |
| batch_x, batch_ticker = batch_x.to(device), batch_ticker.to(device) |
| recon = model(batch_x, batch_ticker) |
| all_actual.append(batch_x.cpu().numpy()) |
| all_recon.append(recon.cpu().numpy()) |
| X_test = np.concatenate(all_actual, axis=0) |
| X_recon = np.concatenate(all_recon, axis=0) |
| X_test_flat = X_test.reshape(-1, X_test.shape[-1]) |
| X_recon_flat = X_recon.reshape(-1, X_recon.shape[-1]) |
|
|
| mae = mean_absolute_error(X_test_flat, X_recon_flat) |
| rmse = np.sqrt(mean_squared_error(X_test_flat, X_recon_flat)) |
| r2 = r2_score(X_test_flat, X_recon_flat) |
|
|
| metrics = { |
| "MAE": float(mae), |
| "RMSE": float(rmse), |
| "R2": float(r2) |
| } |
|
|
| with open(metrics_path, 'w') as f: |
| json.dump(metrics, f, indent=4) |
| print(f"✅ Test metrics saved at: {metrics_path}") |
|
|
