Code/AIClass.py

# Anucha Maneeyotin

from datetime import datetime
import pandas as pd
#import tensorflow as tf
from tensorflow.keras.models import Model, Sequential,load_model
from keras.layers import *
from tensorflow.keras import regularizers
import numpy as np
from keras.optimizers import *
import pandas_ta as ta
from datetime import datetime, timedelta
from sklearn.preprocessing import MinMaxScaler
import joblib
import os
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

features = ['open','high', 'low', 'close']
class EAUtils:

    # create instant class
    #def __new__(cls,aaa):
    #    instance = super(EAUtils, cls).__new__(cls)
    #    return instance

    # create constructor
    def __init__(self, modelname,timestep,feature):
        self.modelname = modelname # can create property read modelname from parameter
        self.timestep = timestep
        self.feature = feature

    @staticmethod
    def MajorList():
        return ["DXYm","GBPUSDm","EURUSDm","NZDUSDm","AUDUSDm","USDCADm","USDJPYm","USDCHFm"]
    
    @staticmethod
    def MinorList():
        return ["EURJPYm","GBPJPYm","USDCHFm","XAUUSDm","EURAUDm"]
    
    @staticmethod
    def get_point_from_symbol(symbolname: str) -> float:
        point_map = {
            "EURUSDm": 0.00001,
            "GBPUSDm": 0.00001,
            "AUDUSDm": 0.00001,
            "NZDUSDm": 0.00001,
            "USDCADm": 0.00001,
            "USDCHFm": 0.00001,
            "USDJPYm": 0.001,
            "XAUUSDm": 0.001,
            "DXYm": 0.01,
        }
        return point_map.get(symbolname, 0.00001)
    
    @staticmethod
    def get_digit_from_symbol(symbolname: str) -> int:
        point_map = {
            "EURUSDm": 5,
            "GBPUSDm": 5,
            "AUDUSDm": 5,
            "NZDUSDm": 5,
            "USDCADm": 5,
            "USDCHFm": 5,
            "USDJPYm": 3,
            "XAUUSDm": 3,
            "DXYm": 2,
        }
        return point_map.get(symbolname, 5)

    def GenerateDayData(self):
       mt5.initialize()
       rates = mt5.copy_rates_from_pos(self.modelname, mt5.TIMEFRAME_D1, 0, 3650)
       df = pd.DataFrame(rates)
       df['time'] = pd.to_datetime(df['time'], unit='s')
       df.set_index('time', inplace=True)
       df.sort_index(ascending=True, inplace=True)
       df = df.filter(features)
       mt5.shutdown()
       return df

    def GenerateWeekData(self):
       mt5.initialize()
       rates = mt5.copy_rates_from_pos(self.modelname, mt5.TIMEFRAME_W1, 0, 520)
       df = pd.DataFrame(rates)
       df['time'] = pd.to_datetime(df['time'], unit='s')
       df.set_index('time', inplace=True)
       df.sort_index(ascending=True, inplace=True)
       df = df.filter(['open','high', 'low', 'close','tick_volume'])
       mt5.shutdown()
       return df

    def GenerateMonthData(self):
       #mt5.initialize(path='C:\Program Files\XM Global MT5\terminal64.exe',server='XMGlobal-MT5',login=4982544,password='Tungman300824')
       mt5.initialize()
       rates = mt5.copy_rates_from_pos(self.modelname, mt5.TIMEFRAME_MN1, 0, 120)
       df = pd.DataFrame(rates)
       df['time'] = pd.to_datetime(df['time'], unit='s')
       df.set_index('time', inplace=True)
       df.sort_index(ascending=True, inplace=True)
       df = df.filter(['open','high', 'low', 'close','tick_volume'])
       mt5.shutdown()
       return df

    def collect_dataset(self,df, history_size):
        """
        Collect dataset for the following regression problem:
        - input: history_size consecutive H1 bars;
        - output: close price for the next bar.

        :param df: D1 bars for a range of dates
        :param history_size: how many bars should be considered for making a prediction
        :return: features and labels
        """
        n = len(df)
        xs = []
        ys = []
        for i in range(n - history_size):
            w = df.iloc[i: i + history_size + 1]

            x = w[['open', 'high', 'low', 'close','tick_volume']].iloc[:-1].values
            #y = w.iloc[-1]['high','low']
            y = w[['high','low']].shift(-1)
            xs.append(x)
            ys.append(y)

        X = np.array(xs)
        y = np.array(ys)
        return X, y

    # Single step dataset preparation
    def singleStepSampler(self,df, window):
        xRes = []
        yRes = []
        for i in range(0, len(df) - window):
            res = []
            for j in range(0, window):
                r = []
                for col in df.columns:
                    r.append(df[col][i + j])
                    #r.append(df.iloc(i+j))
                res.append(r)
            xRes.append(res)
            yRes.append(df[['high','low']].iloc[i + window].to_numpy())
        return np.array(xRes), np.array(yRes)

    def create_sequences_unistep(self,data, n_steps):
        data_t = data.to_numpy()
        X = []
        y = []
   
        for i in range(len(data_t)-n_steps):
            row = [a for a in data_t[i:i+n_steps]]
            X.append(row)
   
            label = data_t[i+n_steps][0]
            y.append(label)
   
        return np.array(X), np.array(y)

    def split_sequence_multistep(self,data, n_steps_in, n_steps_out):
        # แปลง DataFrame -> ndarray
        if hasattr(data, "values"):
            data = data.values

        X, y = [], []
        for i in range(len(data) - n_steps_in - n_steps_out + 1):
            seq_x = data[i:i+n_steps_in, :]
            seq_y = data[i+n_steps_in:i+n_steps_in+n_steps_out, 1:3]  # high, low start at high but not include close
            X.append(seq_x)
            y.append(seq_y)
        return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32)

    #+-------------------------------------------------------------------+
    # 30082025 use this method in train loop                             |
    # get data from metatrader5 before tranin model                      |
    #+-------------------------------------------------------------------+
    def split_sequence_multistep_target(self,data, n_steps_in, n_steps_out):
        # แปลง DataFrame -> ndarray
        if hasattr(data, "values"):
            data = data.values

        X, y_high,y_low = [], [],[]
        for i in range(len(data) - n_steps_in - n_steps_out + 1):
            seq_x = data[i:i+n_steps_in, :]
            seq_y_high  = data[i+n_steps_in:i+n_steps_in+n_steps_out, 1:2] 
            seq_y_low   = data[i+n_steps_in:i+n_steps_in+n_steps_out, 2:3] 
            X.append(seq_x)
            y_high.append(seq_y_high)
            y_low.append(seq_y_low)
        return  np.array(X, dtype=np.float32), np.array(y_high, dtype=np.float32) , np.array(y_low, dtype=np.float32)

    #+-------------------------------------------------------------------+
    #| get data from metatrader5 before tranin model                     |
    #+-------------------------------------------------------------------+
    def fetch_data(self,symbol, timeframe='', num_bars=1200):
        if not mt5.initialize():
            print("initialize() failed")
            mt5.shutdown()

        df = mt5.copy_rates_from_pos(symbol, timeframe, 0, num_bars) 

        # Transform data to a DataFrame
        df = pd.DataFrame(df)
        df['time'] = pd.to_datetime(df['time'], unit='s')
        df.set_index('time', inplace=True)
        df.sort_index(ascending=True, inplace=True)
        df = df.filter(['open','high', 'low', 'close','tick_volume'])
        return df

    #+-------------------------------------------------------------------+
    #| get data from metatrader5 before tranin model and predict         |
    #+-------------------------------------------------------------------+
    def fetch_dataV2(self,symbol, timeframe='', num_bars=3650):
        if not mt5.initialize():
            print("initialize() failed")
            mt5.shutdown()

        df = mt5.copy_rates_from_pos(symbol, timeframe, 0, num_bars) 

        # Transform data to a DataFrame
        df = pd.DataFrame(df)
        df['time'] = pd.to_datetime(df['time'], unit='s')
        df.set_index('time', inplace=True)
        df.sort_index(ascending=True, inplace=True)
        df = df.filter(['open','high', 'low', 'close','tick_volume'])
        return df

    #+-------------------------------------------------------------------+
    #| get data from metatrader5 before tranin model and predict         |
    #+-------------------------------------------------------------------+
    def fetch_dataV3(self,symbol, timeframe='', start=700,end=3650):
        if not mt5.initialize():
            print("initialize() failed")
            mt5.shutdown()

        df = mt5.copy_rates_from_pos(symbol, timeframe, start, end) 

        # Transform data to a DataFrame
        df = pd.DataFrame(df)
        df['time'] = pd.to_datetime(df['time'], unit='s')
        df.set_index('time', inplace=True)
        df.sort_index(ascending=True, inplace=True)
        df = df.filter(['open','high', 'low', 'close','tick_volume'])
        return df
    
    #+-------------------------------------------------------------------+
    #| get data from metatrader5 before tranin model and predict         |
    #+-------------------------------------------------------------------+
    def fetch_dataV4(self, symbol, timeframe=mt5.TIMEFRAME_D1, num_bars=3650):
        if not mt5.initialize():
            print("initialize() failed")
            mt5.shutdown()

        # ดึงข้อมูลจาก MT5
        df = mt5.copy_rates_from_pos(symbol, timeframe, 0, num_bars) 

        # Transform data to a DataFrame
        df = pd.DataFrame(df)
        df['time'] = pd.to_datetime(df['time'], unit='s')

        # เลือก column ที่จำเป็น
        df = df.filter(['time','open','high','low','close','tick_volume'])

        # ✅ Sort by time ascending
        df = df.sort_values(by='time', ascending=True).reset_index(drop=True)

        # --- เพิ่ม time features ---
        df['hour']    = df['time'].dt.hour / 23.0
        df['weekday'] = df['time'].dt.weekday / 6.0
        df['month']   = (df['time'].dt.month - 1) / 11.0

        return df

    
    #+-------------------------------------------------------------------+
    #| get data from metatrader5 before tranin model                     |
    #+-------------------------------------------------------------------+
    def fetch_last5Years(self,symbol, timeframe='', num_bars=900):
        if not mt5.initialize():
            print("initialize() failed")
            mt5.shutdown()

        df = mt5.copy_rates_from_pos(symbol, timeframe, 900, num_bars) 

        # Transform data to a DataFrame
        df = pd.DataFrame(df)
        df['time'] = pd.to_datetime(df['time'], unit='s')
        df.set_index('time', inplace=True)
        df.sort_index(ascending=True, inplace=True)
        df = df.filter(['open','high', 'low', 'close','tick_volume'])
        return df

        #+-------------------------------------------------------------------+
    #| data = get data from fetch data                                   |
    #+-------------------------------------------------------------------+
    @staticmethod
    def add_features(data):

        data            = data.copy()

        adx             = data.ta.adx(length=5)
        data['adx']     = adx['ADX_5'] 

        atr             = data.ta.atr(length=5)
        data['atr']     = atr

        sar             = data.ta.psar(high=data['high'], low=data['low'], close=data['close'], af0=0.02, af=0.02, max_af=0.2)

        #print(sar.columns)

        data['psar']    = sar['PSARr_0.02_0.2']


        # data = EAUtils.add_seasonal_features(data)

        # # เพิ่ม rolling high/low/range
        # data['rolling_max_high_3'] = data['high'].rolling(window=3).max()
        # data['rolling_min_low_3'] = data['low'].rolling(window=3).min()
        # data['rolling_range_3'] = data['high'].rolling(window=3).max() - data['low'].rolling(window=3).min()

        # # Pivot point
        # data['pivot'] = (data['high'] + data['low'] + data['close']) / 3
        # data['support1'] = 2 * data['pivot'] - data['high']
        # data['resistance1'] = 2 * data['pivot'] - data['low']

        data = data.dropna()

        return data
    
    #+-------------------------------------------------------------------+
    #| add this feature after add indicator features                     |
    #+-------------------------------------------------------------------+
    @staticmethod
    def add_seasonal_features(df):

        df = df.copy()
        # สร้าง datetime object
        df['datetime'] = pd.to_datetime(df['time'], unit='s') if 'time' in df.columns else pd.to_datetime(df.index)

        # สร้าง seasonal feature
        df['hour'] = df['datetime'].dt.hour
        df['day_of_week'] = df['datetime'].dt.dayofweek     # 0=Monday
        df['day'] = df['datetime'].dt.day
        df['month'] = df['datetime'].dt.month
        df['week_of_year'] = df['datetime'].dt.isocalendar().week.astype(int)

        # one-hot encode (ถ้าจะใช้กับ DNN หรืออยากให้ model อ่านง่ายขึ้น)
        df = pd.get_dummies(df, columns=['day_of_week', 'month'])

        return df
    
    @staticmethod
    def add_time_feature(df):
        
        df = df.copy()
        
        df['hour'] = df.index.hour / 23.0
        df['weekday'] = df.index.weekday / 6.0
        df['month'] = (df.index.month - 1) / 11.0

        return df


    def add_candle_patterns_only(self,df: pd.DataFrame):
        df = df.copy()

        # รายชื่อ pattern สำคัญที่เน้นความแม่นและสื่อความหมายดี
        candle_patterns = [
            'doji', 'hammer', 'inverted_hammer', 'hanging_man',
            'shooting_star', 'engulfing', 'harami',
            'morning_star', 'evening_star',
            'piercing', 'dark_cloud_cover',
        ]

        for pattern in candle_patterns:
            try:
                df[f'cdl_{pattern}'] = ta.cdl_pattern(df, name=pattern)
            except Exception:
                df[f'cdl_{pattern}'] = 0  # fallback เผื่อ pattern ไม่ support

        return df
    #+-------------------------------------------------------------------+
    #| get data from metatrader5 for error model                         |
    #+-------------------------------------------------------------------+
    def fetch_history(self,symbol,timeframe,startdate,enddate):

        if not mt5.initialize():
            print("initialize() failed")
            mt5.shutdown()
        
        df = mt5.copy_ticks_range(symbol, startdate, enddate, mt5.COPY_TICKS_ALL)

        # Transform data to a DataFrame
        df = pd.DataFrame(df)
        df.sort_values(by='time', ascending=True, inplace=True)
        df = df.filter(['open','high', 'low', 'close'])
        return df
    
    #+-------------------------------------------------------------------+
    #| use this method after load data from metatrader5                  |
    #+-------------------------------------------------------------------+
    def create_sequences_singlestep3(self,data, time_step=10):

        data = EAUtils.add_features(data)  # Add feature ADX

        expected_cols = ['open', 'high', 'low', 'close', 'adx','atr']
        expected_cols += [col for col in data.columns if 'day_of_week_' in col or 'month_' in col]

        # จัด column ใหม่ให้ตรง
        data = data[expected_cols]

        X, y = [], []
        for i in range(len(data) - time_step):
            X.append(data.iloc[i:i+time_step].values)
            y.append(data.iloc[i+time_step][['high', 'low']].values)  # Predict high and low
        return np.array(X), np.array(y)
    

    def create_sequences_singlestep_futurewindow(self,data, timestep=5, future_window=7):
        """
        สร้างชุดข้อมูล X, y
        - X: ลำดับข้อมูล input (timestep แท่ง)
        - y: [max high, min low] ของช่วง future_window วันถัดไป
        """
        # เพิ่มฟีเจอร์
        data = EAUtils.add_features(data)  # เช่น adx, atr, day_of_week_xxx, month_xxx
        
        # กรองเฉพาะคอลัมน์ที่ต้องการ
        expected_cols = ['open', 'high', 'low', 'close', 'adx', 'atr']
        expected_cols += [col for col in data.columns if 'day_of_week_' in col or 'month_' in col]
        data = data[expected_cols]

        high = data['high'].values
        low = data['low'].values

        X, y = [], []

        for i in range(len(data) - timestep - future_window + 1):
            x_i = data.iloc[i:i + timestep].values  # shape = (timestep, num_features)
            future_high = high[i + timestep : i + timestep + future_window].max()
            future_low = low[i + timestep : i + timestep + future_window].min()
            X.append(x_i)
            y.append([future_high, future_low])

        return np.array(X), np.array(y)
    

    def create_sequences_singlestep_relative(self, data, timestep=5, future_window=1):

        # เตรียมดึงข้อมูลสำหรับสร้าง label
        high = data['high'].values
        low = data['low'].values
        close = data['close'].values

        X, y = [], []
        for i in range(len(data) - timestep - future_window + 1):
            x_i = data.iloc[i:i + timestep].values

            future_high = high[i + timestep : i + timestep + future_window].max()
            future_low = low[i + timestep : i + timestep + future_window].min()
            close_ref = close[i + timestep - 1]

            rel_high = (future_high / close_ref) - 1
            rel_low = (future_low / close_ref) - 1

            X.append(x_i)
            y.append([rel_high, rel_low])

        return np.array(X), np.array(y)
    
    def create_sequences_singlestep(self, x_data, y_data, timestep=5):
        """
        แยกสร้าง sequence สำหรับ single-step prediction
        - x_data : ndarray ที่รวม features ต่าง ๆ แล้ว (เช่น price + indicator)
        - y_data : ndarray ที่เป็น target เช่น high, low ที่ถูก scale แล้ว
        - timestep : จำนวนแท่งย้อนหลัง
        
        Return:
            X: shape (num_samples, timestep, num_features)
            y: shape (num_samples, num_targets)
        """
        X, y = [], []

        for i in range(len(x_data) - timestep):
            X.append(x_data[i:i + timestep])
            y.append(y_data[i + timestep])

        return np.array(X), np.array(y)


    def create_sequences_multistep_relative(self, data, timestep=5, forecast_horizon=5):

        # ดึง series สำหรับคำนวณ label
        high = data['high'].values
        low = data['low'].values
        close = data['close'].values

        X, y = [], []

        for i in range(len(data) - timestep - forecast_horizon + 1):
            x_i = data.iloc[i:i + timestep].values
            close_ref = close[i + timestep - 1]

            y_i = []
            for j in range(forecast_horizon):
                future_high = high[i + timestep + j]
                future_low = low[i + timestep + j]

                rel_high = (future_high / close_ref) - 1
                rel_low = (future_low / close_ref) - 1

                y_i.append([rel_high, rel_low])

            X.append(x_i)
            y.append(y_i)

        return np.array(X), np.array(y)

    def create_sequences_multistep_relative_log(self, data, timestep=30, forecast_horizon=5):
        # ดึง series สำหรับคำนวณ label
        high = data['high'].values
        low = data['low'].values
        close = data['close'].values

        X, y = [], []

        for i in range(len(data) - timestep - forecast_horizon + 1):
            x_i = data.iloc[i:i + timestep].values
            close_ref = close[i + timestep - 1]

            y_i = []
            for j in range(forecast_horizon):
                future_high = high[i + timestep + j]
                future_low = low[i + timestep + j]

                # ใช้ log-return แทน
                rel_high = np.log(future_high / close_ref)
                rel_low = np.log(future_low / close_ref)

                y_i.append([rel_high, rel_low])

            X.append(x_i)
            y.append(y_i)

        return np.array(X), np.array(y)
    

    def create_target_multistep_log_relative(self, data: pd.DataFrame, forecast_horizon: int = 5) -> pd.DataFrame:
        """
        สร้าง target multistep ที่เป็น log-relative ของ high/low เทียบกับ close ปัจจุบัน
        เช่น: log(high_t / close_t), log(low_t / close_t) สำหรับ t+1 ถึง t+n
        Return เป็น DataFrame ที่ flatten แล้ว (พร้อมใช้กับ Scaler)
        """
        log_targets = []
        col_names = []

        for i in range(forecast_horizon):
            col_names.append(f"log_rel_high_t{i+1}")
            col_names.append(f"log_rel_low_t{i+1}")

        for i in range(len(data) - forecast_horizon):
            window = data.iloc[i : i + forecast_horizon + 1]  # +1 สำหรับ close ref
            close_ref = window['close'].iloc[0]
            highs = window['high'].iloc[1:]
            lows = window['low'].iloc[1:]

            log_rel_highs = np.log(highs / close_ref)
            log_rel_lows = np.log(lows / close_ref)

            combined = np.empty((forecast_horizon * 2,))
            combined[0::2] = log_rel_highs  # ใส่ high ที่ตำแหน่งคู่
            combined[1::2] = log_rel_lows  # ใส่ low ที่ตำแหน่งคี่

            log_targets.append(combined)

        return pd.DataFrame(log_targets, columns=col_names)

    
    def create_sequences_multistep_from_scaled(
            self,
            X_scaled: np.ndarray,
            y_scaled: np.ndarray,
            timestep: int,
            forecast_horizon: int
        ):
        """
        สร้าง X, y ที่ใช้สำหรับ Multistep Prediction (แยก High กับ Low)

        Parameters:
            - X_scaled: (samples, features)
            - y_scaled: (samples, 2) -> column 0 = high, column 1 = low
            - timestep: จำนวนแท่งย้อนหลังใน input sequence
            - forecast_horizon: จำนวนวันที่ต้องการพยากรณ์ล่วงหน้า

        Returns:
            - X_seq: (num_samples, timestep, features)
            - y_high_seq: (num_samples, forecast_horizon)
            - y_low_seq: (num_samples, forecast_horizon)
        """
        X_seq = []
        y_high_seq = []
        y_low_seq = []

        for i in range(timestep, len(X_scaled) - forecast_horizon + 1):
            x_window = X_scaled[i - timestep:i]                           # (timestep, features)
            y_window = y_scaled[i:i + forecast_horizon]                   # (forecast_horizon, 2)

            X_seq.append(x_window)
            y_high_seq.append(y_window[:, 0])  # column 0 = high
            y_low_seq.append(y_window[:, 1])   # column 1 = low

        return np.array(X_seq), np.array(y_high_seq), np.array(y_low_seq)


    def create_sequences_singlestep_split_Y(self,x_data, y_data, timestep=5):
        X, y_high, y_low = [], [], []
        for i in range(len(x_data) - timestep):
            X.append(x_data[i:i + timestep])
            y_high.append(y_data[i + timestep][0])
            y_low.append(y_data[i + timestep][1])
        return np.array(X), np.array(y_high), np.array(y_low)


    import os, joblib

    @staticmethod
    def load_column_scalers(df, scaler_folder):
        scalers = {}
        for col in df.columns:
            scaler_path = os.path.join(scaler_folder, f"{col}.pkl")
            if os.path.exists(scaler_path):
                scalers[col] = joblib.load(scaler_path)
            else:
                print(f"⚠ Warning: Scaler not found for {col}, skipping scaling.")
                scalers[col] = None
        return scalers
    
    def load_column_scaler_types(self,df, scaler_folder):
        """
        Load scaler types per column from folder.
        Returns dict {column_name: scaler class or None}
        """
        scaler_types = {}
        for col in df.columns:
            scaler_path = os.path.join(scaler_folder, f"{col}.pkl")
            if os.path.exists(scaler_path):
                scaler_types[col] = joblib.load(scaler_path).__class__  # just class
            else:
                print(f"⚠ Warning: Scaler not found for {col}, skipping scaling.")
                scaler_types[col] = None
        return scaler_types

    def auto_weight_decay_range(self,scaler, target_feature='high'):
        """
        ให้ scaler ของ target (MinMaxScaler, StandardScaler, PowerTransformer)
        จะ return lower, upper ของ weight_decay แบบ log-scale
        """
        # ตรวจสอบชนิด scaler
        if hasattr(scaler, 'data_min_') and hasattr(scaler, 'data_max_'):
            # MinMaxScaler
            feature_min = scaler.data_min_[0]
            feature_max = scaler.data_max_[0]
        elif hasattr(scaler, 'scale_'):
            # StandardScaler / PowerTransformer
            feature_min = -scaler.scale_[0]*3   # 3 sigma
            feature_max = scaler.scale_[0]*3
        else:
            # fallback
            feature_min, feature_max = -1, 1

        # ขนาดของ target
        feature_range = feature_max - feature_min
        # map เป็น weight_decay range
        # scale ให้อยู่ใน 1e-7 ~ 1e-4
        lower = max(1e-7, feature_range * 1e-7)
        upper = max(lower*10, feature_range * 1e-5)

        return lower, upper

    def auto_lr_range(self,X_train, base_lr=1e-4):
        # X_train: numpy array shape (n_samples, timesteps, features)
        input_scale = np.std(X_train)
        # สมมติ input scale ใหญ่ -> lr ต้องเล็ก, scale เล็ก -> lr ใหญ่
        min_lr = base_lr / max(input_scale, 1.0)
        max_lr = base_lr * max(input_scale, 1.0)
        return min_lr, max_lr
    

    @staticmethod
    def dynamic_beta(y_true: torch.Tensor, y_pred: torch.Tensor, method='median', eps=1e-6):
        error = (y_true - y_pred).abs()
        if method == 'median':
            beta = error.median().item()
        elif method == 'std':
            beta = error.std().item()
        else:
            raise ValueError("method must be 'median' or 'std'")
        return max(beta, eps)

    @staticmethod
    def calc_weight_decay(lr, scale=0.1, min_wd=1e-8, max_wd=1e-3):
        """
        คำนวณ weight_decay จาก learning rate

        Parameters
        ----------
        lr : float
            learning rate ปัจจุบัน
        scale : float, optional
            สัดส่วนของ LR ที่จะใช้เป็น weight_decay, default 0.1
        min_wd : float, optional
            ค่าต่ำสุดของ weight_decay, default 1e-8
        max_wd : float, optional
            ค่าสูงสุดของ weight_decay, default 1e-3

        Returns
        -------
        float
            weight_decay ที่คำนวณแล้ว
        """
        wd = lr * scale
        wd = max(min_wd, min(max_wd, wd))  # clamp ให้อยู่ในช่วง min-max
        return wd
        
    @staticmethod
    def get_optimizer_groupsV01(model, base_lr=1e-7, head_lr=1e-4, bias_lr=2e-4, weight_decay=1e-5):
        backbone_params = []
        output_params   = []
        bias_params     = []

        for name, param in model.named_parameters():
            if not param.requires_grad:
                continue
            if "bias" in name:
                bias_params.append(param)
            elif ("transformer" in name) or ("input_fc" in name) or ("pos_encoder" in name):
                backbone_params.append(param)
            elif "output" in name:
                output_params.append(param)

        # --- Debug: print ขนาดแต่ละกลุ่ม ---
        total = sum(p.numel() for p in model.parameters() if p.requires_grad)
        print(f"Backbone: {sum(p.numel() for p in backbone_params)} params")
        print(f"Output:   {sum(p.numel() for p in output_params)} params")
        print(f"Bias:     {sum(p.numel() for p in bias_params)} params")
        print(f"Total:    {total} params")

        # --- Check no overlap ---
        seen = set()
        for group in [backbone_params, output_params, bias_params]:
            for p in group:
                assert id(p) not in seen, "Duplicate parameter detected!"
                seen.add(id(p))

        optimizer = torch.optim.Adam([
            {'params': backbone_params, 'lr': base_lr, 'weight_decay': weight_decay},
            {'params': output_params,   'lr': head_lr, 'weight_decay': weight_decay},
            {'params': bias_params,     'lr': bias_lr, 'weight_decay': 0.0},  # bias no decay
        ], weight_decay=weight_decay)

        return optimizer