บทนำ

ในโลกของการเทรดคริปโต ความผันผวน (Volatility) คือหัวใจสำคัญที่นักเทรดทุกคนต้องเข้าใจ การทำนายความผันผวนของ Bitcoin อย่างแม่นยำสามารถสร้างความได้เปรียบในการตัดสินใจซื้อขายได้อย่างมาก บทความนี้จะพาคุณสร้างโมเดลทำนายความผันผวนโดยใช้ข้อมูลจาก Tardis ซึ่งเป็นแพลตฟอร์มรวบรวมข้อมูลตลาดคริปโตชั้นนำ และเปรียบเทียบประสิทธิภาพระหว่างวิธีการทางสถิติแบบดั้งเดิม (GARCH) กับ Machine Learning สำหรับวิศวกรที่ต้องการลงมือทำ ผมจะแสดงโค้ดที่ใช้งานได้จริงใน production โดยใช้ HolySheep AI สำหรับงาน inference และการประมวลผลข้อมูล

ทำไมต้องใช้ Tardis Data

Tardis เป็นผู้ให้บริการข้อมูลตลาดคริปโตที่มีคุณภาพสูง ครอบคลุมข้อมูล OHLCV (Open, High, Low, Close, Volume) จากหลาย exchange รวมถึง: คุณภาพของข้อมูลมีผลโดยตรงต่อความแม่นยำของโมเดล ผมทดสอบพบว่า Tardis ให้ข้อมูลที่มีความสอดคล้องกัน (consistency) สูงเมื่อเทียบกับแหล่งข้อมูลอื่น ทำให้สามารถสร้างโมเดลที่ robust ต่อ outliers ได้ดี

การตั้งค่า Environment และการดึงข้อมูล

เริ่มต้นด้วยการติดตั้ง dependencies และตั้งค่า API: 
# requirements.txt
pandas>=2.0.0
numpy>=1.24.0
scipy>=1.10.0
arch>=6.0.0
scikit-learn>=1.3.0
xgboost>=1.7.0
lightgbm>=4.0.0
requests>=2.28.0
ta>=0.10.0
ta-lib # ต้องติดตั้งจาก source สำหรับ M1/M2 Mac
joblib>=1.2.0
matplotlib>=3.7.0
seaborn>=0.12.0
โค้ดด้านล่างแสดงการดึงข้อมูล BTC perpetuals จาก Tardis API: 
import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Optional, Dict, List
import time
import json

class TardisDataFetcher:
    """
    Fetcher สำหรับดึงข้อมูล OHLCV จาก Tardis Exchange API
    รองรับ Binance, Bybit, OKX perpetual futures
    """
    
    BASE_URL = "https://api.tardis.dev/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        })
        self.rate_limit_delay = 0.5  # รอระหว่าง request
    
    def get_exchanges(self) -> List[Dict]:
        """ดึงรายการ exchange ที่รองรับ"""
        response = self.session.get(f"{self.BASE_URL}/exchanges")
        response.raise_for_status()
        return response.json()
    
    def get_symbols(self, exchange: str) -> List[str]:
        """ดึง symbols ที่มีใน exchange"""
        cache_key = f"{exchange}_symbols"
        if hasattr(self, cache_key):
            return getattr(self, cache_key)
        
        url = f"{self.BASE_URL}/exchanges/{exchange}/symbols"
        response = self.session.get(url)
        response.raise_for_status()
        symbols = [s['symbol'] for s in response.json() if 'perpetual' in s.get('type', '').lower()]
        setattr(self, cache_key, symbols)
        return symbols
    
    def fetch_ohlcv(
        self,
        exchange: str,
        symbol: str,
        start_date: datetime,
        end_date: datetime,
        resolution: str = "1m",
        limit: int = 1000
    ) -> pd.DataFrame:
        """
        ดึงข้อมูล OHLCV จาก Tardis
        
        Args:
            exchange: ชื่อ exchange (เช่น 'binance', 'bybit', 'okx')
            symbol: ชื่อ symbol (เช่น 'BTC-USDT-PERP')
            start_date: วันที่เริ่มต้น
            end_date: วันที่สิ้นสุด
            resolution: ความละเอียด ('1m', '5m', '1h', '1d')
            limit: จำนวน candle สูงสุดต่อ request
        
        Returns:
            DataFrame พร้อม columns: timestamp, open, high, low, close, volume
        """
        url = f"{self.BASE_URL}/historical/candles"
        
        params = {
            'exchange': exchange,
            'symbol': symbol,
            'from': int(start_date.timestamp()),
            'to': int(end_date.timestamp()),
            'resolution': resolution,
            'limit': limit
        }
        
        all_candles = []
        current_from = int(start_date.timestamp())
        end_timestamp = int(end_date.timestamp())
        
        while current_from < end_timestamp:
            params['from'] = current_from
            
            try:
                response = self.session.get(url, params=params)
                response.raise_for_status()
                data = response.json()
                
                if not data or len(data) == 0:
                    break
                
                all_candles.extend(data)
                current_from = data[-1]['timestamp'] + 1
                
                time.sleep(self.rate_limit_delay)
                
            except requests.exceptions.RequestException as e:
                print(f"Error fetching data: {e}")
                if response.status_code == 429:
                    time.sleep(60)  # Rate limited, wait 1 min
                else:
                    break
        
        if not all_candles:
            return pd.DataFrame()
        
        df = pd.DataFrame(all_candles)
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
        df.set_index('timestamp', inplace=True)
        df = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
        df = df[~df.index.duplicated(keep='first')]
        
        return df.sort_index()
    
    def fetch_multiple_exchanges(
        self,
        symbol: str,
        start_date: datetime,
        end_date: datetime,
        resolution: str = "1m"
    ) -> Dict[str, pd.DataFrame]:
        """ดึงข้อมูลจากหลาย exchange และรวมกัน"""
        exchanges = ['binance', 'bybit', 'okx']
        result = {}
        
        for exchange in exchanges:
            try:
                print(f"Fetching {symbol} from {exchange}...")
                df = self.fetch_ohlcv(
                    exchange=exchange,
                    symbol=symbol,
                    start_date=start_date,
                    end_date=end_date,
                    resolution=resolution
                )
                if not df.empty:
                    result[exchange] = df
            except Exception as e:
                print(f"Failed to fetch from {exchange}: {e}")
        
        return result


ตัวอย่างการใช้งาน

if __name__ == "__main__":
    fetcher = TardisDataFetcher(api_key="YOUR_TARDIS_API_KEY")
    
    # ดึงข้อมูล 1 ปี
    end_date = datetime.now()
    start_date = end_date - timedelta(days=365)
    
    # ดึงจาก Binance
    btc_data = fetcher.fetch_ohlcv(
        exchange='binance',
        symbol='BTC-USDT-PERP',
        start_date=start_date,
        end_date=end_date,
        resolution='1h'
    )
    
    print(f"Fetched {len(btc_data)} candles")
    print(btc_data.tail())

Feature Engineering สำหรับ Volatility Prediction

การสร้าง features ที่เหมาะสมคือหัวใจของโมเดลทำนายความผันผวน ผมออกแบบ features ที่ครอบคลุมหลายมิติ: 
import pandas as pd
import numpy as np
from typing import List, Optional
from scipy import stats
import warnings
warnings.filterwarnings('ignore')

class VolatilityFeatureEngineer:
    """
    สร้าง features สำหรับ volatility prediction model
    ครอบคลุมทั้ง technical indicators, statistical features, และ market microstructure
    """
    
    def __init__(self, lookback_windows: List[int] = None):
        self.lookback_windows = lookback_windows or [1, 6, 12, 24, 72, 168]  # 1h-1w
        
    def calculate_returns(self, df: pd.DataFrame) -> pd.DataFrame:
        """คำนวณ returns หลายระดับ"""
        df = df.copy()
        df['returns'] = df['close'].pct_change()
        df['log_returns'] = np.log(df['close'] / df['close'].shift(1))
        
        for window in self.lookback_windows:
            df[f'returns_{window}h'] = df['close'].pct_change(window)
            df[f'log_returns_{window}h'] = np.log(
                df['close'] / df['close'].shift(window)
            )
        
        return df
    
    def calculate_volatility_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """คำนวณ features ที่เกี่ยวกับความผันผวน"""
        df = df.copy()
        
        for window in self.lookback_windows:
            # Realized Volatility (RV) - ใช้ squared returns
            df[f'rv_{window}h'] = np.sqrt(
                (df['returns'] ** 2).rolling(window).sum()
            )
            
            # Parkinson Volatility - ใช้ High-Low range
            df[f'parkinson_{window}h'] = np.sqrt(
                (1 / (4 * np.log(2))) * 
                ((np.log(df['high'] / df['low'])) ** 2).rolling(window).mean() * window
            )
            
            # Garman-Klass Volatility - รวม O-H-L-C
            df[f'garman_klass_{window}h'] = np.sqrt(
                (0.5 * (np.log(df['high'] / df['low'])) ** 2 - 
                 (2 * np.log(2) - 1) * (np.log(df['close'] / df['open'])) ** 2
                ).rolling(window).mean() * window
            )
            
            # Rogers-Satchell Volatility
            df[f'rogers_satchell_{window}h'] = np.sqrt(
                (np.log(df['high'] / df['close'])) * 
                (np.log(df['high'] / df['open'])) +
                (np.log(df['low'] / df['close'])) * 
                (np.log(df['low'] / df['open']))
            ).rolling(window).mean() * window
            
            # Extreme Value based - max drawdown volatility
            roll_max = df['close'].rolling(window).max()
            roll_min = df['close'].rolling(window).min()
            df[f'ewma_vol_{window}h'] = df['returns'].ewm(span=window).std()
            
        return df
    
    def calculate_momentum_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Momentum และ trend features"""
        df = df.copy()
        
        for window in self.lookback_windows:
            # Price momentum
            df[f'momentum_{window}h'] = df['close'] / df['close'].shift(window) - 1
            
            # RSI-like momentum
            delta = df['close'].diff()
            gain = delta.where(delta > 0, 0).rolling(window).mean()
            loss = (-delta.where(delta < 0, 0)).rolling(window).mean()
            rs = gain / (loss + 1e-10)
            df[f'rsi_{window}h'] = 100 - (100 / (1 + rs))
            
            # MACD
            exp1 = df['close'].ewm(span=12, adjust=False).mean()
            exp2 = df['close'].ewm(span=26, adjust=False).mean()
            macd = exp1 - exp2
            signal = macd.ewm(span=9, adjust=False).mean()
            df[f'macd_{window}h'] = macd - signal
            
            # Bollinger Bands position
            sma = df['close'].rolling(window).mean()
            std = df['close'].rolling(window).std()
            df[f'bb_position_{window}h'] = (df['close'] - sma) / (2 * std + 1e-10)
        
        return df
    
    def calculate_volume_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Volume-based features"""
        df = df.copy()
        
        for window in self.lookback_windows:
            # Volume momentum
            df[f'volume_ratio_{window}h'] = (
                df['volume'] / df['volume'].rolling(window).mean()
            )
            
            # VWAP proxy
            typical_price = (df['high'] + df['low'] + df['close']) / 3
            df[f'vwap_proxy_{window}h'] = (
                (typical_price * df['volume']).rolling(window).sum() / 
                df['volume'].rolling(window).sum()
            )
            
            # Price-Volume correlation
            df[f'pv_corr_{window}h'] = df['returns'].rolling(window).corr(df['volume'])
            
            # OBV-like indicator
            df['obv'] = (np.sign(df['returns']) * df['volume']).cumsum()
            df[f'obv_momentum_{window}h'] = df['obv'].diff(window)
        
        return df
    
    def calculate_statistical_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Statistical features - skewness, kurtosis, autocorrelation"""
        df = df.copy()
        
        for window in self.lookback_windows:
            # Return distribution features
            df[f'skewness_{window}h'] = df['returns'].rolling(window).apply(
                lambda x: stats.skew(x, nan_policy='omit'), raw=True
            )
            df[f'kurtosis_{window}h'] = df['returns'].rolling(window).apply(
                lambda x: stats.kurtosis(x, nan_policy='omit'), raw=True
            )
            
            # Autocorrelation of returns
            df[f' autocorr_{window}h'] = df['returns'].rolling(window).apply(
                lambda x: pd.Series(x).autocorr(lag=1), raw=True
            )
            
            # Volatility clustering indicator (squared returns autocorrelation)
            df[f'vol_clustering_{window}h'] = (df['returns']**2).rolling(window).apply(
                lambda x: pd.Series(x).autocorr(lag=1), raw=True
            )
            
            # Jarque-Bera like statistic
            returns = df['returns'].rolling(window)
            df[f'jarque_bera_stat_{window}h'] = (
                window / 6 * (
                    returns.skew()**2 + 0.25 * returns.kurtosis()**2
                )
            )
        
        return df
    
    def calculate_time_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Cyclical time features"""
        df = df.copy()
        
        # Hour of day (for hourly data)
        if df.index.freqstr and 'h' in str(df.index.freqstr):
            df['hour'] = df.index.hour
            df['hour_sin'] = np.sin(2 * np.pi * df['hour'] / 24)
            df['hour_cos'] = np.cos(2 * np.pi * df['hour'] / 24)
        
        # Day of week
        df['dayofweek'] = df.index.dayofweek
        df['dayofweek_sin'] = np.sin(2 * np.pi * df['dayofweek'] / 7)
        df['dayofweek_cos'] = np.cos(2 * np.pi * df['dayofweek'] / 7)
        
        # Is weekend
        df['is_weekend'] = (df.index.dayofweek >= 5).astype(int)
        
        # US market hours (BTC trades 24/7 but volume patterns differ)
        df['is_us_hours'] = ((df.index.hour >= 14) & (df.index.hour < 21)).astype(int)
        
        return df
    
    def create_all_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """สร้าง features ทั้งหมด"""
        df = df.copy()
        
        df = self.calculate_returns(df)
        df = self.calculate_volatility_features(df)
        df = self.calculate_momentum_features(df)
        df = self.calculate_volume_features(df)
        df = self.calculate_statistical_features(df)
        df = self.calculate_time_features(df)
        
        # Remove rows with NaN
        df = df.dropna()
        
        print(f"Total features created: {len(df.columns)}")
        print(f"Total samples: {len(df)}")
        
        return df



ตัวอย่างการใช้งาน

if __name__ == "__main__":
    # สมมติว่ามีข้อมูลแล้ว
    # df = pd.read_parquet('btc_ohlcv.parquet')
    
    engineer = VolatilityFeatureEngineer()
    # df_features = engineer.create_all_features(df)
    
    print("Feature engineering module ready")

GARCH Model Implementation

GARCH (Generalized Autoregressive Conditional Heteroskedasticity) เป็นโมเดลทางสถิติที่ได้รับการยอมรับอย่างกว้างขวางในการ model ความผันผวน ความสามารถในการ capture "volatility clustering" ทำให้ GARCH เหมาะกับข้อมูล financial returns: 
import numpy as np
import pandas as pd
from arch import arch_model
from scipy.optimize import minimize
from typing import Tuple, Dict, Optional
import warnings
warnings.filterwarnings('ignore')

class GARCHVolatilityModel:
    """
    GARCH family models สำหรับ volatility prediction
    รองรับ: GARCH, EGARCH, GJR-GARCH, TGARCH
    """
    
    def __init__(self, model_type: str = 'GARCH'):
        """
        Args:
            model_type: 'GARCH', 'EGARCH', 'GJR-GARCH', 'TGARCH'
        """
        self.model_type = model_type.upper()
        self.model = None
        self.fitted_model = None
        self.params = None
        self.residuals = None
        self.conditional_volatility = None
        
    def fit(self, returns: pd.Series, p: int = 1, q: int = 1, 
            dist: str = 't') -> 'GARCHVolatilityModel':
        """
        Fit GARCH model
        
        Args:
            returns: Series of returns
            p: GARCH lag order
            q: ARCH lag order  
            dist: Distribution - 'normal', 't', 'skewt'
        """
        # Scale returns for numerical stability
        self.scale_factor = returns.std()
        scaled_returns = returns / self.scale_factor * 100
        
        # Build model
        vol_model = self.model_type.lower()
        
        self.model = arch_model(
            scaled_returns,
            mean='Constant',
            vol=vol_model,
            p=p,
            q=q,
            dist=dist
        )
        
        # Fit with constraints
        self.fitted_model = self.model.fit(
            disp='off',
            options={
                'maxiter': 1000,
                'ftol': 1e-8
            }
        )
        
        # Store results
        self.params = self.fitted_model.params
        self.residuals = self.fitted_model.resid
        self.conditional_volatility = self.fitted_model.conditional_volatility
        
        return self
    
    def predict_volatility(self, horizon: int = 1) -> Tuple[float, float]:
        """
        Predict future volatility
        
        Returns:
            (predicted_volatility, std_error)
        """
        forecast = self.fitted_model.forecast(horizon=horizon)
        mean_forecast = forecast.mean.iloc[-1].values[0]
        variance_forecast = forecast.variance.iloc[-1].values[0]
        
        # Scale back
        vol_scaled = np.sqrt(variance_forecast) / 100 * self.scale_factor
        std_scaled = np.sqrt(variance_forecast) / 100 * self.scale_factor
        
        return vol_scaled, std_scaled
    
    def rolling_forecast(self, returns: pd.Series, train_window: int, 
                        forecast_horizon: int) -> pd.DataFrame:
        """
        Rolling window forecast for backtesting
        
        Returns:
            DataFrame with actual vs predicted volatility
        """
        predictions = []
        actuals = []
        timestamps = []
        
        n = len(returns)
        
        for i in range(train_window, n - forecast_horizon + 1):
            train_data = returns.iloc[i - train_window:i]
            
            try:
                model = GARCHVolatilityModel(self.model_type)
                model.fit(train_data, p=1, q=1, dist='t')
                pred_vol, _ = model.predict_volatility(horizon=forecast_horizon)
                
                # Actual realized volatility
                actual_returns = returns.iloc[i:i + forecast_horizon]
                actual_vol = np.sqrt((actual_returns ** 2).sum())
                
                predictions.append(pred_vol)
                actuals.append(actual_vol)
                timestamps.append(returns.index[i])
                
            except Exception as e:
                continue
        
        return pd.DataFrame({
            'timestamp': timestamps,
            'predicted_vol': predictions,
            'actual_vol': actuals
        }).set_index('timestamp')
    
    def get_model_summary(self) -> Dict:
        """สรุปผล model"""
        if self.fitted_model is None:
            return {}
        
        return {
            'model_type': self.model_type,
            'params': self.params.to_dict(),
            'aic': self.fitted_model.aic,
            'bic': self.fitted_model.bic,
            'log_likelihood': self.fitted_model.loglikelihood,
            'converged': self.fitted_model.converged
        }


class MultiGARCHEnsemble:
    """Ensemble ของ GARCH variants"""
    
    def __init__(self):
        self.models = {
            'GARCH': GARCHVolatilityModel('GARCH'),
            'EGARCH': GARCHVolatilityModel('EGARCH'),
            'GJR-GARCH': GARCHVolatilityModel('GJR-GARCH'),
        }
        self.fitted_models = {}
        
    def fit_all(self, returns: pd.Series) -> Dict:
        """Fit ทุก model"""
        results = {}
        
        for name, model_template in self.models.items():
            try:
                model = GARCHVolatilityModel(name)
                model.fit(returns, p=1, q=1, dist='t')
                self.fitted_models[name] = model
                results[name] = model.get_model_summary()
            except Exception as e:
                print(f"Failed to fit {name}: {e}")
        
        return results
    
    def predict_ensemble(self, horizon: int = 1, 
                        weights: Optional[Dict[str, float]] = None) -> Tuple[float, float]:
        """
        Ensemble prediction โดยใช้ weighted average
        
        Args:
            horizon: forecast horizon
            weights: custom weights หรือ None สำหรับ equal weights
        
        Returns:
            (ensemble_volatility, uncertainty)
        """
        if not self.fitted_models:
            raise ValueError("No models fitted")
        
        if weights is None:
            weights = {name: 1.0 / len(self.fitted_models) 
                      for name in self.fitted_models}
        
        predictions = []
        uncertainties = []
        
        for name, model in self.fitted_models.items():
            pred, std = model.predict_volatility(horizon)
            predictions.append(pred)
            uncertainties.append(std)
        
        # Weighted average
        ensemble_vol = sum(
            w * p for w, p in zip(
                [weights.get(name, 0) for name in self.fitted_models],
                predictions
            )
        )
        
        # Average uncertainty
        ensemble_std = np.mean(uncertainties)
        
        return ensemble_vol, ensemble_std



ตัวอย่างการใช้งาน

if __name__ == "__main__":
    # สมมติว่ามี returns series แล้ว
    # returns = df['returns']
    
    # Fit single model
    # garch = GARCHVolatilityModel('GARCH')
    # garch.fit(returns)
    # print(garch.get_model_summary())
    
    # Ensemble prediction
    # ensemble = MultiGARCHEnsemble()
    # ensemble.fit_all(returns)
    # pred_vol, uncertainty = ensemble.predict_ensemble(horizon=24)
    
    print("GARCH models ready")

Machine Learning Model Implementation

สำหรับ ML approach ผมใช้ XGBoost และ LightGBM ซึ่งเป็น gradient boosting ที่มีประสิทธิภาพสูงสำหรับ tabular data: 
import numpy as np
import pandas as pd
import lightgbm as lgb
import xgboost as xgb
from sklearn.model_selection import TimeSeriesSplit
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
from typing import Tuple, Dict, Optional, List
import joblib
import warnings
warnings.filterwarnings('ignore')

class VolatilityMLModel:
    """
    Machine Learning model สำหรับ volatility prediction
    รองรับ LightGBM และ XGBoost
    """
    
    def __init__(self, model_type: str = 'lightgbm'):
        self.model_type = model_type.lower()
        self.model = None
        self.feature_columns = None
        self.scaler = StandardScaler()
        
        # Default hyperparameters
        if model_type == 'lightgbm':
            self.params = {
                'objective': 'regression',
                'metric': 'rmse',
                'boosting_type': 'gbdt',
                'learning_rate': 0.05,
                'num_leaves': 31,
                'max_depth': 6,
                'min_child_samples': 20,
                'feature_fraction': 0.8,
                'bagging_fraction': 0.8,
                'bagging_freq': 5,
                'reg_alpha': 0.1,
                'reg_lambda': 0.1,
                'verbose': -1,
                'n_jobs': -1
            }
        else:  # xgboost
            self.params = {
                'objective': 'reg:squarederror',
                'eval_metric': 'rmse',
                'learning_rate': 0.05,
                'max_depth': 6,
                'min_child_weight': 5,
                'subsample': 0.8,
                'colsample_bytree': 0.8,
                'reg_alpha': 0.1,
                'reg_lambda': 0.1,
                'n_jobs': -1,
                'verbosity': 0
            }
    
    def prepare_data(
        self,
        df: pd.DataFrame,
        target_col: str = 'rv_24h',
        feature_cols: Optional[List[str]] = None,
        test_size: float = 0.2
    ) -> Tuple:
        """Prepare train/test split"""
        
        if feature_cols is None:
            # Auto-select features (exclude target and non-features)
            exclude_cols = [target_col, 'close', 'high', 'low', 'open', 
                          'volume', 'returns', 'log_returns', 'obv']
            feature_cols = [c for c in df.columns if c not in exclude_cols]
        
        self.feature_columns = feature_cols
        
        X = df[feature_cols].values
        y = df[target_col].values
        
        # Time series split
        split_idx = int(len(X) * (1 - test_size))
        
        X_train, X_test = X[:split_idx], X[split_idx:]
        y_train, y_test = y[:split_idx], y[split_idx:]
        
        # Scale features
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_test_scaled = self.scaler.transform(X_test)
        
        return X_train_scaled, X_test_scaled, y_train, y_test, feature_cols
    
    def train(
        self,
        X_train: np.ndarray,
        y_train: np.ndarray,
        X_val: Optional[Tuple] = None,
        n_estimators: int = 1000,
        early_stopping_rounds: int = 50
    ) -> 'VolatilityMLModel':
        """Train model with early stopping"""
        
        if self.model_type == 'lightgbm':
            train_data = lgb.Dataset(X_train, label=y_train)
            
            valid_sets = [train_data]

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง