data_analysis_v0.0.py

# -*- coding: utf-8 -*-
"""
Improved LSTM-based Time Series Forecast for Next-Day Stock Price.
Handles 5000 records, uses feature scaling for both features and target,
and includes better training techniques to avoid straight-line predictions.
"""

import os
import re
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau

# ====================================
# CONFIG
# ====================================
INPUT_FILE = "Data.csv"      # Can be .csv or .xlsx
LOOKBACK = 60                # Number of historical days for LSTM
TRAIN_SPLIT = 0.70
VAL_SPLIT = 0.15
SEED = 42

tf.random.set_seed(SEED)
np.random.seed(SEED)

# ====================================
# HELPERS
# ====================================
def detect_file_type(file_path):
    ext = os.path.splitext(file_path)[-1].lower()
    if ext == ".csv":
        return "csv"
    elif ext in [".xlsx", ".xls"]:
        return "excel"
    else:
        raise ValueError("Unsupported file format. Use .csv or .xlsx")

def to_float_num(s):
    """Convert strings like '7,831.60' to float."""
    if pd.isna(s): return np.nan
    s = str(s).replace(",", "").replace("−", "-")
    try:
        return float(s)
    except:
        return np.nan

def parse_volume(v):
    """Convert Vol. to numeric shares (e.g., '134.56M' -> 134,560,000)."""
    if pd.isna(v): return np.nan
    s = str(v).replace(",", "").replace("−", "-")
    m = re.match(r"^\s*([+-]?\d*\.?\d+)\s*([KMBkmb])?\s*$", s)
    if m:
        num = float(m.group(1))
        suf = (m.group(2) or "").upper()
        mult = {"": 1, "K": 1e3, "M": 1e6, "B": 1e9}[suf]
        return num * mult
    try:
        return float(s)
    except:
        return np.nan

def parse_change_pct(x):
    """Convert '-0.22%' → -0.22."""
    if pd.isna(x): return np.nan
    s = str(x).replace("%", "").replace("−", "-").replace(",", "")
    try:
        return float(s)
    except:
        return np.nan

def make_sequences(features, targets, lookback):
    X, y = [], []
    for i in range(lookback, len(features)):
        X.append(features[i-lookback:i, :])
        y.append(targets[i])
    return np.array(X), np.array(y)

# ====================================
# LOAD DATA
# ====================================
file_type = detect_file_type(INPUT_FILE)

if file_type == "csv":
    df = pd.read_csv(INPUT_FILE)
else:
    df = pd.read_excel(INPUT_FILE, engine="openpyxl")

required_cols = ["Date", "Price", "Open", "High", "Low", "Vol.", "Change %"]
missing_cols = [c for c in required_cols if c not in df.columns]
if missing_cols:
    raise ValueError(f"Missing columns: {missing_cols}. Found: {df.columns.tolist()}")

# Clean data
df["Date"] = pd.to_datetime(df["Date"], errors="coerce", dayfirst=True)
for col in ["Price", "Open", "High", "Low"]:
    df[col] = df[col].apply(to_float_num)
df["Vol."] = df["Vol."].apply(parse_volume)
df["Change %"] = df["Change %"].apply(parse_change_pct)

df = df.sort_values("Date").dropna(subset=["Price", "Open", "High", "Low"])
df = df.reset_index(drop=True)

df["Vol."].fillna(method="ffill", inplace=True)
df["Change %"].fillna(0.0, inplace=True)

# ====================================
# FEATURE ENGINEERING
# ====================================
for w in [5, 10, 20, 50, 200]:
    df[f"SMA_{w}"] = df["Price"].rolling(w).mean()
    df[f"EMA_{w}"] = df["Price"].ewm(span=w, adjust=False).mean()

df["RET_1"] = df["Price"].pct_change()
df["HL_PCT"] = (df["High"] - df["Low"]) / df["Price"]
df["OC_PCT"] = (df["Price"] - df["Open"]) / df["Open"]
df["VOL_CHG"] = df["Vol."].pct_change()

df = df.fillna(method="ffill").fillna(method="bfill")

FEATURE_COLS = [
    "Price", "Open", "High", "Low", "Vol.", "Change %",
    "RET_1", "HL_PCT", "OC_PCT", "VOL_CHG"
] + [f"SMA_{w}" for w in [5,10,20,50,200]] + [f"EMA_{w}" for w in [5,10,20,50,200]]

# ====================================
# SCALING
# ====================================
feature_scaler = MinMaxScaler()
target_scaler = MinMaxScaler()

scaled_features = feature_scaler.fit_transform(df[FEATURE_COLS].values.reshape(-1, len(FEATURE_COLS)))
scaled_prices = target_scaler.fit_transform(df[["Price"]])

# ====================================
# TRAIN-TEST SPLIT
# ====================================
n = len(df)
train_end = int(n * TRAIN_SPLIT)
val_end = int(n * (TRAIN_SPLIT + VAL_SPLIT))

train_features = scaled_features[:train_end]
val_features = scaled_features[train_end:val_end]
test_features = scaled_features[val_end:]

train_prices = scaled_prices[:train_end]
val_prices = scaled_prices[train_end:val_end]
test_prices = scaled_prices[val_end:]

X_train, y_train = make_sequences(train_features, train_prices, LOOKBACK)
X_val, y_val = make_sequences(val_features, val_prices, LOOKBACK)
X_test, y_test = make_sequences(test_features, test_prices, LOOKBACK)

test_dates = df.iloc[val_end+LOOKBACK:]["Date"].values

# ====================================
# BUILD LSTM MODEL
# ====================================
model = Sequential([
    LSTM(128, return_sequences=True, input_shape=(LOOKBACK, len(FEATURE_COLS))),
    Dropout(0.3),
    LSTM(64, return_sequences=False),
    Dropout(0.3),
    Dense(32, activation="relu"),
    Dense(1)
])

optimizer = tf.keras.optimizers.Adam(learning_rate=0.001, clipnorm=1.0)
model.compile(optimizer=optimizer, loss="mse")

early_stop = EarlyStopping(monitor="val_loss", patience=15, restore_best_weights=True)
reduce_lr = ReduceLROnPlateau(monitor="val_loss", factor=0.5, patience=7, verbose=1)

history = model.fit(
    X_train, y_train,
    validation_data=(X_val, y_val),
    epochs=150,
    batch_size=64,
    callbacks=[early_stop, reduce_lr],
    verbose=1,
    shuffle=False
)

# ====================================
# PREDICT & INVERSE TRANSFORM
# ====================================
y_pred_scaled = model.predict(X_test).ravel()
y_pred = target_scaler.inverse_transform(y_pred_scaled.reshape(-1, 1)).ravel()
y_test_orig = target_scaler.inverse_transform(y_test.reshape(-1, 1)).ravel()

rmse = np.sqrt(mean_squared_error(y_test_orig, y_pred))
mae = mean_absolute_error(y_test_orig, y_pred)
mape = np.mean(np.abs((y_test_orig - y_pred) / np.maximum(1e-8, np.abs(y_test_orig)))) * 100
r2 = r2_score(y_test_orig, y_pred)

prev_actual = np.array([np.nan] + list(y_test_orig[:-1]))
actual_dir = np.sign(y_test_orig - prev_actual)
pred_dir = np.sign(y_pred - prev_actual)
mask = ~np.isnan(prev_actual)
directional_accuracy = (actual_dir[mask] == pred_dir[mask]).mean() * 100

print("\n=== Test Metrics ===")
print(f"RMSE: {rmse:.4f}")
print(f"MAE : {mae:.4f}")
print(f"MAPE: {mape:.2f}%")
print(f"R²  : {r2:.4f}")
print(f"Directional Accuracy: {directional_accuracy:.2f}%")

# ====================================
# PLOTS
# ====================================
plt.figure(figsize=(14, 6))
plt.plot(test_dates, y_test_orig, label="Actual Price")
plt.plot(test_dates, y_pred, label="Predicted Price")
plt.title("Actual vs Predicted Prices")
plt.xlabel("Date")
plt.ylabel("Price")
plt.legend()
plt.grid()
plt.tight_layout()
plt.show()

# Zoom in last 100 points
plt.figure(figsize=(14, 6))
plt.plot(test_dates[-100:], y_test_orig[-100:], label="Actual Price")
plt.plot(test_dates[-100:], y_pred[-100:], label="Predicted Price")
plt.title("Zoomed: Last 100 Predictions")
plt.xlabel("Date")
plt.ylabel("Price")
plt.legend()
plt.grid()
plt.tight_layout()
plt.show()

plt.figure(figsize=(8, 4))
plt.plot(history.history["loss"], label="Train Loss")
plt.plot(history.history["val_loss"], label="Val Loss")
plt.title("Training History (Loss)")
plt.xlabel("Epoch")
plt.ylabel("MSE Loss")
plt.legend()
plt.grid()
plt.tight_layout()
plt.show()

# ====================================
# SAVE OUTPUTS
# ====================================
os.makedirs("model_outputs", exist_ok=True)
model.save("model_outputs/lstm_nextday_price")
pd.DataFrame({
    "Date": pd.to_datetime(test_dates),
    "Actual": y_test_orig,
    "Predicted": y_pred
}).to_csv("model_outputs/test_predictions.csv", index=False)

print("\nSaved model to model_outputs/lstm_nextday_price")
print("Saved predictions to model_outputs/test_predictions.csv")