Code/Predict-Month.py

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)