File size: 5,840 Bytes
a6fae72 | 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 126 127 128 | import os, sys,joblib
import numpy as np
import pandas as pd
import MetaTrader5 as mt5
import torch
from tqdm import tqdm
import sklearn.preprocessing._data # สำหรับ StandardScaler
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../../common')))
from AIClass import EAUtils
from AIBaseClass import * # ต้องมี TransformerSingleStep ในที่นี้
from ClassUtils import *
# --- Config ---
LOOKBACK = 3
FOLDS = [1, 2, 3]
symbollist = ['DXYm','EURUSDm','GBPUSDm','USDJPYm','USDCADm','XAUUSDm']
# --- Device ---
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
results = []
calibrators = {symbol: {fold: CalibratorSingleStep(method='SMA') for fold in FOLDS} for symbol in symbollist}
for symbol in tqdm(symbollist):
SAVE_DIR = f"Model/Month/{symbol}/TRANSFORMER-PYTORCH-V03/"
try:
util = EAUtils(modelname=symbol, timestep=LOOKBACK, feature=5)
df = util.fetch_dataV4(symbol, mt5.TIMEFRAME_MN1,120)
df = EAUtils.add_features(df)
df = df[['open','high','low','close','tick_volume','adx','atr','hour','weekday','month']].dropna()
if len(df) < LOOKBACK + 1:
print(f"[{symbol}] ❌ Insufficient data")
continue
fold_table = []
for fold in FOLDS:
try:
model_path = os.path.join(SAVE_DIR, f"{symbol}_transformer_fold{fold}_04.pth")
if not os.path.exists(model_path):
print(f"[{symbol} fold{fold}] ❌ Model file not found: {model_path}")
continue
# --- Load scalers ---
scaler_price = joblib.load(f"{SAVE_DIR}scaler_fold{fold}/scaler_price.pkl")
scaler_adx = joblib.load(f"{SAVE_DIR}scaler_fold{fold}/scaler_adx.pkl")
scaler_atr = joblib.load(f"{SAVE_DIR}scaler_fold{fold}/scaler_atr.pkl")
scaler_target_high = joblib.load(f"{SAVE_DIR}scaler_fold{fold}/scaler_target_high.pkl")
scaler_target_low = joblib.load(f"{SAVE_DIR}scaler_fold{fold}/scaler_target_low.pkl")
scaler_tick = joblib.load(f"{SAVE_DIR}scaler_fold{fold}/scaler_tick_volume.pkl")
# --- Prepare Input ---
scaled_price = scaler_price.transform(df[['open','high','low','close']].values)
scaled_adx = scaler_adx.transform(df[['adx']].values)
scaled_atr = scaler_atr.transform(df[['atr']].values)
scaled_tick = scaler_tick.transform(df[['tick_volume']].values)
time_data = df[['hour','weekday','month']].values
X_scaled = np.concatenate([scaled_price, scaled_adx, scaled_atr,scaled_tick,time_data], axis=1)
input_seq = torch.tensor(X_scaled[-LOOKBACK:], dtype=torch.float32).unsqueeze(0).to(device)
checkpoint = torch.load(model_path, map_location=device,weights_only=False)
# --- Load Model ---
model = ConvGRUTransformerHLV10(input_dim=10, seq_len=LOOKBACK, kernel_size=LOOKBACK).to(device)
model.load_state_dict(checkpoint['model_state_dict'])
model.eval()
# --- Predict ---
with torch.no_grad():
pred_high,pred_low = model(input_seq)
# detach & convert to numpy
pred_high_np = pred_high.detach().cpu().numpy().reshape(-1, 1) # shape = [B,1]
pred_low_np = pred_low.detach().cpu().numpy().reshape(-1, 1)
pred_real_high = scaler_target_high.inverse_transform(pred_high_np)[0][0]
pred_real_low = scaler_target_low.inverse_transform(pred_low_np)[0][0]
latest_atr = df['atr'].iloc[-1]
pred_real_high = pred_real_high + (latest_atr*0.1)
pred_real_low = pred_real_low - (latest_atr*0.1)
# --- Record Results ---
last_close = float(df['close'].iloc[-1])
digits = EAUtils.get_digit_from_symbol(symbol)
fold_table.append({
'Fold': fold,
'Last Close': round(last_close, digits),
'Predicted High': round(pred_real_high, digits),
'Predicted Low': round(pred_real_low, digits)
})
# Bias Calibration
cal = calibrators[symbol][fold]
cal.update_history(pred_real_high, pred_real_low, df['high'].iloc[-1], df['low'].iloc[-1])
cal.refit()
pred_high_calib, pred_low_calib = cal.calibrate(pred_real_high, pred_real_low)
cal.update_bias(pred_high_calib, pred_low_calib, df['high'].iloc[-1], df['low'].iloc[-1])
results.append({
'symbol': symbol,
'fold': fold,
'predicted_high': round(pred_high_calib, digits),
'predicted_low': round(pred_low_calib, digits),
'last_close': round(last_close, digits)
})
except Exception as e_fold:
print(f"[{symbol} fold{fold}] ❌ Error during prediction: {e_fold}")
# print(f"\n=== Prediction Table for {symbol} ===")
# print(pd.DataFrame(fold_table))
except Exception as e:
print(f"[{symbol}] ❌ Error: {e}")
# --- Save overall results ---
df_result = pd.DataFrame(results)
out_csv = "predicted_single_step_no_optuna.csv"
df_result.to_csv(out_csv, index=False)
print(f"\n✅ Prediction complete! Results saved to: {out_csv}")
print(df_result)
|