import os, sys, joblib import numpy as np import pandas as pd import MetaTrader5 as mt5 import torch import torch.nn as nn from sklearn.preprocessing import MinMaxScaler, RobustScaler # --- Utility Class --- sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../../common'))) from AIClass import EAUtils from AIBaseClass import * from AILoss import * # --- Config --- LOOKBACK = 3 FOLDS = [1, 2, 3] DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu") EPOCHS = 50 LR = 1e-5 L2_LAMBDA = 0.03 class TransferCrossTF: def __init__(self, basesymbol, targetsymbol, baseTF, targetTF, basemodelpath, targetmodelpath ): self.BaseSymbol = basesymbol self.TargetSymbol = targetsymbol self.BaseTimeframe = baseTF self.TargetTimeframe = targetTF self.BaseModelPath = basemodelpath self.TargetModelDir = targetmodelpath def FetchTargetData(self): util = EAUtils(self.TargetSymbol, timestep=LOOKBACK, feature=10) df = util.fetch_dataV2(self.TargetSymbol, mt5.TIMEFRAME_W1, 520) if df is None or df.empty or len(df) < LOOKBACK + 1: raise ValueError(f"[{self.TargetSymbol}] ❌ Insufficient data") df = df[['open', 'high', 'low', 'close', 'tick_volume']] df = EAUtils.add_features(df) df = util.add_time_feature(df) last_close = round(df['close'].iloc[-1], EAUtils.get_digit_from_symbol(self.TargetSymbol)) return df, last_close def ScaledData(self): df, last_close = self.FetchTargetData() # ===== Fit new scalers for target ===== scaler_price = MinMaxScaler(feature_range=(-1, 1)); scaler_price.fit(df[['open', 'high', 'low', 'close']].values) scaler_adx = MinMaxScaler(feature_range=(-1, 1)); scaler_adx.fit(df[['adx']].values) scaler_atr = MinMaxScaler(feature_range=(-1, 1)); scaler_atr.fit(df[['atr']].values) scaler_tick = RobustScaler(); scaler_tick.fit(df[['tick_volume']].values) scaler_target_high = MinMaxScaler(feature_range=(-1, 1)); scaler_target_high.fit(df[['high']].values) scaler_target_low = MinMaxScaler(feature_range=(-1, 1)); scaler_target_low.fit(df[['low']].values) return scaler_price, scaler_adx, scaler_atr, scaler_tick, scaler_target_high, scaler_target_low, df, last_close def SaveScaler(self, scalers): for fold in FOLDS: fold_dir = f"{self.TargetModelDir}scaler_fold{fold}/" os.makedirs(fold_dir, exist_ok=True) joblib.dump(scalers['price'], f"{fold_dir}scaler_price.pkl") joblib.dump(scalers['adx'], f"{fold_dir}scaler_adx.pkl") joblib.dump(scalers['atr'], f"{fold_dir}scaler_atr.pkl") joblib.dump(scalers['tick'], f"{fold_dir}scaler_tick.pkl") joblib.dump(scalers['target_high'], f"{fold_dir}scaler_target_high.pkl") joblib.dump(scalers['target_low'], f"{fold_dir}scaler_target_low.pkl") def FreezeLayer(self,model): # --- Freeze feature extractor layers --- for name, param in model.named_parameters(): # Freeze Conv if "conv1" in name or "conv_bn" in name: param.requires_grad = False # Freeze GRU elif "gru" in name: param.requires_grad = False # Freeze Positional Encoding elif "pos_encoder" in name: param.requires_grad = False # Freeze Transformer elif "transformer" in name: param.requires_grad = False else: # Keep fc_high, fc_low trainable param.requires_grad = True def PerformTransfer(self, fold, results): scaler_price, scaler_adx, scaler_atr, scaler_tick, scaler_target_high, scaler_target_low, df, last_close = self.ScaledData() scalers = { 'price': scaler_price, 'adx': scaler_adx, 'atr': scaler_atr, 'tick': scaler_tick, 'target_high': scaler_target_high, 'target_low': scaler_target_low } self.SaveScaler(scalers) # --- 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[-5:], dtype=torch.float32).unsqueeze(0).to(DEVICE) # --- Load base model --- base_model_path = f'{self.BaseModelPath}' checkpoint = torch.load(base_model_path,weights_only=False) model = ConvGRUTransformerHL(input_dim=10, seq_len=4, kernel_size=3).to(DEVICE) model_dict = model.state_dict() # โหลดเฉพาะ layer ที่ shape ตรงกับ model ใหม่ pretrained_dict = {k: v for k, v in checkpoint['model_state_dict'].items() if k in model_dict and v.size() == model_dict[k].size()} model_dict.update(pretrained_dict) model.load_state_dict(model_dict) self.FreezeLayer(model) # --- Prepare target --- target_high = scaler_target_high.transform(df['high'].values[-1].reshape(-1,1)) target_low = scaler_target_low.transform(df['low'].values[-1].reshape(-1,1)) target_seq = torch.tensor(np.concatenate([target_high, target_low], axis=1), dtype=torch.float32).unsqueeze(0).to(DEVICE) # --- Optimizer + Loss --- optimizer = torch.optim.Adam(filter(lambda p: p.requires_grad, model.parameters()), lr=LR, weight_decay=L2_LAMBDA) loss_fn_high = DynamicBetaSmoothL1Loss(init_beta=0.5, alpha=0.1, reduction='mean') loss_fn_low = DynamicBetaSmoothL1Loss(init_beta=0.5, alpha=0.1, reduction='mean') scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, 'min', patience=5) # --- Fine-tune --- model.train() for _ in range(EPOCHS): optimizer.zero_grad() pred_high, pred_low = model(input_seq) target_high_t = target_seq[:,:,0] target_low_t = target_seq[:,:,1] loss_high = loss_fn_high(pred_high.squeeze(-1), target_high_t) loss_low = loss_fn_low(pred_low.squeeze(-1), target_low_t) loss = loss_high + loss_low loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 0.5) optimizer.step() scheduler.step(loss.item()) # --- Predict after fine-tune --- #model.eval() # with torch.no_grad(): # pred = model(input_seq) # pred = torch.cat(pred, dim=-1).squeeze(0).cpu().numpy() # pred_high = scaler_target_high.inverse_transform(pred[:,0].reshape(-1,1))[0][0] # pred_low = scaler_target_low.inverse_transform(pred[:,1].reshape(-1,1))[0][0] # results.append({ # 'symbol': self.TargetSymbol, # 'fold': fold, # 'predicted_high': round(pred_high, EAUtils.get_digit_from_symbol(self.TargetSymbol)), # 'predicted_low': round(pred_low, EAUtils.get_digit_from_symbol(self.TargetSymbol)), # 'last_close': last_close # }) # --- Save fine-tuned model --- final_model_path = f"{self.TargetModelDir}{self.TargetSymbol}_transformer_fold{fold}.pth" torch.save(model.state_dict(), final_model_path) print(f"[{self.TargetSymbol}] ✅ Saved transfer model fold {fold} -> {final_model_path}") return results # --- Run Transfer --- basesymbol = ['DXYm','EURUSDm','GBPUSDm','USDJPYm','USDCADm','XAUUSDm'] targetsymbol = ['DXYm','EURUSDm','GBPUSDm','USDJPYm','USDCADm','XAUUSDm'] for base, target in zip(basesymbol, targetsymbol): print(f"ฺBegin Transfer {base} from time frame Month to → {target} time frame Week") results = [] for f in FOLDS: transfer = TransferCrossTF( base, target, mt5.TIMEFRAME_MN1, mt5.TIMEFRAME_W1, f"Model/Month/{base}/TRANSFORMER-PYTORCH-V03/{base}_transformer_fold{f}_03.pth", f'Model/Week/{target}/TRANSFORMER-PYTORCH-V02/' ) results = transfer.PerformTransfer(f, results) # Save results summary per symbol df_result = pd.DataFrame(results) out_path = f"Model/Week/{target}/transfer_results.csv" df_result.to_csv(out_path, index=False) print(f"\n✅ {target} Transfer Learning complete! -> saved {out_path}") print(df_result)