File size: 3,890 Bytes
e845ced |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
import pandas as pd
import numpy as np
from pmdarima.arima import auto_arima
from sklearn.metrics import mean_absolute_error, mean_squared_error
import matplotlib.pyplot as plt
# --- Step 1: Load stock data ---
stock_prices = pd.read_csv("/work/GOOGL.csv", parse_dates=["Date"], index_col="Date")["Close"]
# --- Step 2: Compute Log Returns ---
log_returns = np.log(stock_prices / stock_prices.shift(1)).dropna()
# --- Step 3: Sliding Window Evaluation ---
def evaluate_window(log_returns, stock_prices, window_size, test_size=0.2):
train_size = int(len(log_returns) * (1 - test_size))
train, test = log_returns[:train_size], log_returns[train_size:]
predictions = []
price_predictions = []
last_train_price = stock_prices.iloc[train_size - 1]
price_predictions.append(last_train_price)
for t in range(len(test)):
# Define rolling window
start_idx = train_size + t - window_size
if start_idx < 0:
window_data = log_returns[:train_size + t]
else:
window_data = log_returns[start_idx:train_size + t]
# Fit ARIMA
model = auto_arima(
window_data.values,
seasonal=False,
stepwise=True,
suppress_warnings=True,
error_action="ignore"
)
# Forecast 1-step log return
forecast = model.predict(n_periods=1)[0]
predictions.append(forecast)
# Convert to price
price_predictions.append(price_predictions[-1] * np.exp(forecast))
# Drop the initial seed (last_train_price)
price_predictions = price_predictions[1:]
# --- Ensure same length ---
predictions = np.array(predictions)
test = test[:len(predictions)]
actual_prices = stock_prices.iloc[train_size:train_size + len(price_predictions)]
# --- Metrics in log-return space ---
mae_log = mean_absolute_error(test, predictions)
rmse_log = np.sqrt(mean_squared_error(test, predictions))
# --- Metrics in price space ---
mae_price = mean_absolute_error(actual_prices, price_predictions)
rmse_price = np.sqrt(mean_squared_error(actual_prices, price_predictions))
# --- Direction Accuracy ---
direction_accuracy = np.mean(
np.sign(np.diff(actual_prices.values)) == np.sign(np.diff(price_predictions))
)
return {
"MAE_Log": mae_log,
"RMSE_Log": rmse_log,
"MAE_Price": mae_price,
"RMSE_Price": rmse_price,
"Direction_Accuracy": direction_accuracy,
"Price_Predictions": price_predictions, # Store predictions
"Actual_Prices": actual_prices # Store actual prices
}
# --- Step 4: Test multiple window sizes ---
window_sizes = [30, 60, 90, 120, 180, 200, 250]
results = {}
for w in window_sizes:
print(f"Evaluating window size: {w}")
metrics = evaluate_window(log_returns, stock_prices, w)
results[w] = metrics
results_df = pd.DataFrame({k: {kk: vv for kk, vv in v.items() if kk not in ['Price_Predictions', 'Actual_Prices']} for k, v in results.items()}).T.sort_values("RMSE_Price")
print("\nSliding Window Evaluation Results:")
print(results_df)
best_rmse_window = results_df.index[0]
print(f"\n✅ Best window for RMSE (Price): {best_rmse_window} days")
# --- Step 5: Plot the forecast for the best window ---
best_metrics = results[best_rmse_window]
actual_prices = best_metrics['Actual_Prices']
price_predictions = best_metrics['Price_Predictions']
# Create the plot
plt.figure(figsize=(12, 6))
plt.plot(actual_prices.index, actual_prices, label='Actual Prices', color='blue')
plt.plot(actual_prices.index[:len(price_predictions)], price_predictions, label='Predicted Prices', color='orange', linestyle='--')
plt.title(f'ARIMA Forecast vs Actual Prices (Window Size: {best_rmse_window} days)')
plt.xlabel('Date')
plt.ylabel('Stock Price (USD)')
plt.legend()
plt.grid(True)
plt.show() |