import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler from typing import Tuple, Optional def preprocess_for_lstm( df: pd.DataFrame, target_col: str = "Close", sequence_length: int = 60, test_split: float = 0.2, ) -> dict: """ Preprocess stock data for LSTM training. Returns: dict with X_train, y_train, X_test, y_test, scaler, scaled_data, and split_index. """ # Extract target column data = df[[target_col]].values.astype(float) if len(data) < sequence_length + 10: raise ValueError( f"Not enough data. Need at least {sequence_length + 10} rows, got {len(data)}." ) # Fit MinMaxScaler on full data scaler = MinMaxScaler(feature_range=(0, 1)) scaled_data = scaler.fit_transform(data) # Build sequences X, y = [], [] for i in range(sequence_length, len(scaled_data)): X.append(scaled_data[i - sequence_length : i, 0]) y.append(scaled_data[i, 0]) X, y = np.array(X), np.array(y) X = X.reshape((X.shape[0], X.shape[1], 1)) # (samples, timesteps, features) # Train / test split (no shuffle – temporal order matters) split_idx = int(len(X) * (1 - test_split)) X_train, X_test = X[:split_idx], X[split_idx:] y_train, y_test = y[:split_idx], y[split_idx:] return { "X_train": X_train, "y_train": y_train, "X_test": X_test, "y_test": y_test, "scaler": scaler, "scaled_data": scaled_data, "split_index": split_idx + sequence_length, # index in original df "sequence_length": sequence_length, "raw_data": data, } def inverse_transform(scaler: MinMaxScaler, values: np.ndarray) -> np.ndarray: """Inverse-scale predictions back to original price range.""" values = np.array(values).reshape(-1, 1) return scaler.inverse_transform(values).flatten() def build_prediction_input( scaled_data: np.ndarray, sequence_length: int = 60, ) -> np.ndarray: """Build the last sequence for making future predictions.""" last_seq = scaled_data[-sequence_length:, 0] return last_seq.reshape(1, sequence_length, 1) def generate_future_dates(last_date: str, n_days: int = 30) -> list: """Generate business-day future dates starting from last_date.""" import pandas as pd start = pd.Timestamp(last_date) + pd.Timedelta(days=1) future_dates = pd.bdate_range(start=start, periods=n_days) return [d.strftime("%Y-%m-%d") for d in future_dates] def calculate_metrics(y_true: np.ndarray, y_pred: np.ndarray) -> dict: """Calculate RMSE, MAE, MAPE for model evaluation.""" y_true = np.array(y_true, dtype=float) y_pred = np.array(y_pred, dtype=float) rmse = float(np.sqrt(np.mean((y_true - y_pred) ** 2))) mae = float(np.mean(np.abs(y_true - y_pred))) mape = float(np.mean(np.abs((y_true - y_pred) / (y_true + 1e-8))) * 100) r2 = float(1 - np.sum((y_true - y_pred)**2) / (np.sum((y_true - np.mean(y_true))**2) + 1e-8)) return {"rmse": round(rmse, 4), "mae": round(mae, 4), "mape": round(mape, 4), "r2": round(r2, 4)}