File size: 4,448 Bytes
00397e8 2400b74 00397e8 2400b74 00397e8 2400b74 00397e8 2400b74 00397e8 | 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 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """
EUR/USD Direction Prediction — Inference
=========================================
Usage:
python predict.py
Requires: yfinance, lightgbm, xgboost, ta, scikit-learn, joblib, numpy, pandas
"""
import json
import numpy as np
import pandas as pd
import yfinance as yf
import ta
import joblib
# Optimal ensemble weights (grid-optimised over 198 walk-forward folds)
LGB_WEIGHT = 0.39
XGB_WEIGHT = 0.61
def compute_features(df):
"""Compute all features from OHLCV data."""
LOOKBACK_WINDOWS = [5, 10, 21, 63, 126, 252]
for w in LOOKBACK_WINDOWS:
df[f"log_return_{w}d"] = np.log(df["Close"] / df["Close"].shift(w))
df[f"momentum_{w}d"] = df["Close"] / df["Close"].shift(w) - 1
df["log_return_1d"] = np.log(df["Close"] / df["Close"].shift(1))
for w in [5, 10, 21, 63]:
df[f"volatility_{w}d"] = df["log_return_1d"].rolling(w).std()
for w in [5, 10, 20, 50, 100, 200]:
sma = df["Close"].rolling(w).mean()
df[f"price_to_sma_{w}"] = df["Close"] / sma - 1
df["ema_12"] = df["Close"].ewm(span=12).mean()
df["ema_26"] = df["Close"].ewm(span=26).mean()
df["ema_cross"] = df["ema_12"] / df["ema_26"] - 1
for w in [7, 14, 21]:
df[f"rsi_{w}"] = ta.momentum.RSIIndicator(df["Close"], window=w).rsi()
macd = ta.trend.MACD(df["Close"])
df["macd"] = macd.macd()
df["macd_signal"] = macd.macd_signal()
df["macd_diff"] = macd.macd_diff()
bb = ta.volatility.BollingerBands(df["Close"], window=20, window_dev=2)
df["bb_pband"] = bb.bollinger_pband()
df["bb_wband"] = bb.bollinger_wband()
atr = ta.volatility.AverageTrueRange(df["High"], df["Low"], df["Close"], window=14)
df["atr_14"] = atr.average_true_range()
df["atr_14_pct"] = df["atr_14"] / df["Close"]
stoch = ta.momentum.StochasticOscillator(df["High"], df["Low"], df["Close"])
df["stoch_k"] = stoch.stoch()
df["stoch_d"] = stoch.stoch_signal()
adx = ta.trend.ADXIndicator(df["High"], df["Low"], df["Close"], window=14)
df["adx"] = adx.adx()
df["adx_pos"] = adx.adx_pos()
df["adx_neg"] = adx.adx_neg()
df["williams_r"] = ta.momentum.WilliamsRIndicator(df["High"], df["Low"], df["Close"]).williams_r()
df["cci"] = ta.trend.CCIIndicator(df["High"], df["Low"], df["Close"]).cci()
df["obv"] = ta.volume.OnBalanceVolumeIndicator(df["Close"], df["Volume"]).on_balance_volume()
df["obv_pct"] = df["obv"].pct_change(5)
df["day_of_week"] = df.index.dayofweek
df["month"] = df.index.month
df["quarter"] = df.index.quarter
df["hl_range"] = (df["High"] - df["Low"]) / df["Close"]
df["oc_range"] = (df["Close"] - df["Open"]) / df["Close"]
for w in [10, 20, 50]:
rolling_high = df["High"].rolling(w).max()
rolling_low = df["Low"].rolling(w).min()
df[f"channel_pos_{w}"] = (df["Close"] - rolling_low) / (rolling_high - rolling_low + 1e-10)
return df
def predict():
# Load models
lgb_model = joblib.load("lgb_model.joblib")
xgb_model = joblib.load("xgb_model.joblib")
scaler = joblib.load("scaler.joblib")
feature_cols = json.load(open("feature_columns.json"))
# Fetch recent data (need 300 days for indicators to warm up)
df = yf.download("EURUSD=X", period="2y", progress=False)
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
df["Volume"] = df["Volume"].replace(0, 1)
df = compute_features(df)
df = df.dropna(subset=feature_cols)
# Predict latest
latest = df.iloc[[-1]]
X_latest = scaler.transform(latest[feature_cols].values)
lgb_prob = lgb_model.predict_proba(X_latest)[:, 1][0]
xgb_prob = xgb_model.predict_proba(X_latest)[:, 1][0]
ensemble_prob = LGB_WEIGHT * lgb_prob + XGB_WEIGHT * xgb_prob
direction = "UP" if ensemble_prob >= 0.5 else "DOWN"
confidence = max(ensemble_prob, 1 - ensemble_prob)
print(f"Date: {latest.index[0].date()}")
print(f"EUR/USD Close: {latest['Close'].values[0]:.5f}")
print(f"Prediction for next day: {direction}")
print(f"Confidence: {confidence:.1%}")
print(f" LightGBM prob(UP): {lgb_prob:.3f} (weight: {LGB_WEIGHT:.0%})")
print(f" XGBoost prob(UP): {xgb_prob:.3f} (weight: {XGB_WEIGHT:.0%})")
print(f" Ensemble prob(UP): {ensemble_prob:.3f}")
if __name__ == "__main__":
predict()
|