The Matrix of Machine Learning for Stock Prediction: Hype vs Reality

# -------------------------------------------------
# BEFORE: Naïve LSTM (the trap)
# -------------------------------------------------
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense

# Load data
df = pd.read_csv('AAPL_daily.csv')
close = df['Close'].values.reshape(-1, 1)

# Scale
scaler = MinMaxScaler()
close_scaled = scaler.fit_transform(close)

# Create sequences
def create_seq(data, lookback=60):
    X, y = [], []
    for i in range(len(data)-lookback):
        X.append(data[i:i+lookback])
        y.append(data[i+lookback])
    return np.array(X), np.array(y)

X, y = create_seq(close_scaled, lookback=60)

# Train/test split (random – uh oh!)
split = int(0.8 * len(X))
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]

# Model
model = Sequential([
    LSTM(50, activation='relu', input_shape=(X_train.shape[1],1)),
    Dense(1)
])
model.compile(optimizer='adam', loss='mse')
model.fit(X_train, y_train, epochs=20, batch_size=32, verbose=0)

# Predict next day price
pred_scaled = model.predict(X_test)
pred = scaler.inverse_transform(pred_scaled)
true = scaler.inverse_transform(y_test)

mse = np.mean((pred - true)**2)
print(f'MSE: {mse:.4f}')

# -------------------------------------------------
# AFTER: Probabilistic LSTM with walk‑forward validation
# -------------------------------------------------
import numpy as np
import pandas as pd
import tensorflow as tf
import tensorflow_probability as tfp
from sklearn.preprocessing import StandardScaler

tfd = tfp.distributions

# --- data prep -------------------------------------------------
df = pd.read_csv('AAPL_daily.csv')
returns = df['Close'].pct_change().dropna().values.reshape(-1,1)

scaler = StandardScaler()
returns_scaled = scaler.fit_transform(returns)

def make_windows(data, lookback=60):
    X, y = [], []
    for i in range(len(data)-lookback):
        X.append(data[i:i+lookback])
        y.append(data[i+lookback])
    return np.array(X), np.array(y)

X, y = make_windows(returns_scaled, lookback=60)

# --- walk‑forward (time‑series CV) -----------------------------
n_splits = 5
split_points = np.linspace(0.6, 0.9, n_splits)  # 60% → 90% train

sharpe_list = []

for train_frac in split_points:
    split_idx = int(len(X) * train_frac)
    X_tr, X_te = X[:split_idx], X[split_idx:]
    y_tr, y_te = y[:split_idx], y[split_idx:]

    # ---- model ------------------------------------------------
    model = tf.keras.Sequential([
        tf.keras.layers.LSTM(64, activation='tanh', input_shape=(X_tr.shape[1],1)),
        tf.keras.layers.Dense(2)   # we will output mu and sigma
    ])

    def nll(y_true, y_pred):
        mu, sigma = tf.split(y_pred, 2, axis=-1)
        sigma = tf.nn.softplus(sigma) + 1e-6   # ensure positivity
        dist = tfd.Normal(loc=mu, scale=sigma)
        return -tf.distribution.LogProb(dist, y_true)  # negative log‑likelihood

    model.compile(optimizer=tf.optimizers.Adam(0.001), loss=nll)
    model.fit(X_tr, y_tr, epochs=30, batch_size=32, verbose=0)

    # ---- predictive distribution -------------------------------
    mu_sigma = model.predict(X_te, verbose=0)
    mu, sigma_raw = np.split(mu_sigma, 2, axis=1)
    sigma = np.exp(sigma_raw)   # inverse softplus approx for simplicity
    preds = scaler.inverse_transform(mu)   # back to returns scale
    true_ret = scaler.inverse_transform(y_te)

    # ---- simple long‑only strategy based on predicted mean -----
    signal = (preds > 0).astype(int)   # go long if expected return positive
    strategy_ret = signal * true_ret
    # subtract a tiny transaction cost
    strategy_ret -= 0.0001

    # ---- performance metrics -----------------------------------
    mean_ret = np.mean(strategy_ret)
    vol_ret  = np.std(strategy_ret)
    sharpe   = mean_ret / vol_ret * np.sqrt(252) if vol_ret != 0 else 0
    sharpe_list.append(sharpe)

    print(f'Train {int(train_frac*100)}% → Sharpe: {sharpe:.2f}')

print(f'Average Sharpe over folds: {np.mean(sharpe_list):.2f}')