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

ทำไมต้องพยากรณ์ความผันผวนของ Bitcoin

Bitcoin เป็นสินทรัพย์ที่มีความผันผวนสูงมาก โดยในช่วงตลาดขาขึ้น ความผันผวนรายวันอาจสูงถึง 5-10% และในช่วงวิกฤตอาจพุ่งเกิน 15% การพยากรณ์ความผันผวนช่วยให้นักเทรด:

การตั้งค่า Data Pipeline จาก Tardis

Tardis เป็นผู้ให้บริการข้อมูลตลาด Cryptocurrency ระดับ institutional ที่รวบรวม Order Book, Trade, Quote และ Funding Rate จาก Exchange ชั้นนำ ในโปรเจกต์นี้เราจะดึงข้อมูล OHLCV รายนาทีและ Funding Rate เพื่อสร้างฟีเจอร์สำหรับการพยากรณ์

import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class TardisDataFetcher:
    """
    คลาสสำหรับดึงข้อมูลจาก Tardis Exchange API
    รองรับ: Binance, Bybit, OKX, Deribit
    """
    
    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'
        })
    
    def fetch_ohlcv(self, exchange: str, symbol: str, 
                    start_date: str, end_date: str, 
                    timeframe: str = '1m') -> pd.DataFrame:
        """
        ดึงข้อมูล OHLCV จาก Exchange ที่ระบุ
        
        Args:
            exchange: ชื่อ exchange (binance, bybit, okx)
            symbol: คู่เทรด (BTCUSDT)
            start_date: วันที่เริ่มต้น (YYYY-MM-DD)
            end_date: วันที่สิ้นสุด (YYYY-MM-DD)
            timeframe: ระยะเวลา (1m, 5m, 1h, 1d)
        
        Returns:
            DataFrame ที่มี columns: timestamp, open, high, low, close, volume
        """
        # คำนวณช่วงเวลาที่ต้องดึง (Tardis จำกัด 30 วันต่อ request)
        start = datetime.strptime(start_date, '%Y-%m-%d')
        end = datetime.strptime(end_date, '%Y-%m-%d')
        
        all_data = []
        current_start = start
        
        while current_start < end:
            current_end = min(current_start + timedelta(days=28), end)
            
            # สร้าง request payload
            payload = {
                "exchange": exchange,
                "symbol": symbol,
                "from": current_start.isoformat(),
                "to": current_end.isoformat(),
                "format": "json",
                "columns": ["timestamp", "open", "high", "low", "close", "volume"],
                "timeframe": timeframe
            }
            
            # ดึงข้อมูลจาก Exchange symbols endpoint
            url = f"{self.BASE_URL}/exchange/{exchange}/symbols/{symbol}"
            response = self.session.post(url, json=payload, timeout=60)
            response.raise_for_status()
            
            data = response.json()
            if data:
                df_chunk = pd.DataFrame(data)
                df_chunk['timestamp'] = pd.to_datetime(df_chunk['timestamp'])
                all_data.append(df_chunk)
            
            current_start = current_end + timedelta(minutes=1)
        
        if not all_data:
            return pd.DataFrame()
        
        # รวมข้อมูลและเรียงลำดับ
        result = pd.concat(all_data, ignore_index=True)
        result = result.drop_duplicates(subset=['timestamp']).sort_values('timestamp')
        result = result.reset_index(drop=True)
        
        return result
    
    def fetch_funding_rate(self, exchange: str, symbol: str,
                          start_date: str, end_date: str) -> pd.DataFrame:
        """
        ดึงข้อมูล Funding Rate สำหรับคำนวณความผันผวนโดยนัย
        """
        payload = {
            "exchange": exchange,
            "symbol": symbol,
            "from": start_date,
            "to": end_date,
            "format": "json",
            "columns": ["timestamp", "funding_rate", "mark_price"]
        }
        
        url = f"{self.BASE_URL}/exchange/{exchange}/funding-rate"
        response = self.session.post(url, json=payload, timeout=60)
        response.raise_for_status()
        
        return pd.DataFrame(response.json())


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

fetcher = TardisDataFetcher(api_key="YOUR_TARDIS_API_KEY")


ดึงข้อมูล Bitcoin รายชั่วโมงย้อนหลัง 90 วัน

btc_data = fetcher.fetch_ohlcv(
    exchange='binance',
    symbol='BTCUSDT',
    start_date='2024-10-01',
    end_date='2025-01-01',
    timeframe='1h'
)

print(f"ดึงข้อมูลสำเร็จ: {len(btc_data)} rows")
print(btc_data.tail())

การสร้างฟีเจอร์สำหรับ Volatility Modeling

ความผันผวนที่แท้จริง (Realized Volatility) คำนวณจาก Log Return ยกกำลังสอง รวมกับฟีเจอร์เพิ่มเติมจาก Order Book และ Funding Rate

import pandas as pd
import numpy as np
from scipy import stats

class VolatilityFeatureEngineer:
    """
    สร้างฟีเจอร์สำหรับโมเดลพยากรณ์ความผันผวน
    """
    
    def __init__(self, data: pd.DataFrame):
        self.data = data.copy()
        self.data.set_index('timestamp', inplace=True)
    
    def compute_returns(self, price_col: str = 'close') -> pd.DataFrame:
        """คำนวณ Log Return และ Realized Variance"""
        self.data['log_return'] = np.log(self.data[price_col] / self.data[price_col].shift(1))
        self.data['realized_var'] = self.data['log_return'] ** 2
        return self.data
    
    def compute_realized_volatility(self, windows: list = [5, 15, 30, 60]) -> pd.DataFrame:
        """
        คำนวณ Realized Volatility ที่หลาย Timeframe
        
        Args:
            windows: รายการความยาว window (นาที/ชั่วโมง)
        """
        for window in windows:
            # Realized Variance = Sum of squared returns
            rv = self.data['realized_var'].rolling(window=window).sum()
            self.data[f'rv_{window}'] = np.sqrt(rv * 252 * 24)  # Annualized
            
            # Parkinson Volatility (ใช้ High-Low)
            hl = np.log(self.data['high'] / self.data['low'])
            parkinson = (hl ** 2) * (1 / (4 * np.log(2))) * window
            self.data[f'parkinson_{window}'] = np.sqrt(parkinson * 252 * 24)
            
            # Rogers-Satchell Volatility
            log_hc = np.log(self.data['high'] / self.data['close'])
            log_cl = np.log(self.data['close'] / self.data['low'])
            log_ho = np.log(self.data['high'] / self.data['open'])
            log_ol = np.log(self.data['open'] / self.data['low'])
            rs = (log_hc * log_cl + log_ho * log_ol).rolling(window=window).mean()
            self.data[f'rogers_satchell_{window}'] = np.sqrt(rs * 252 * 24)
        
        return self.data
    
    def compute_order_flow_features(self, order_book_data: pd.DataFrame) -> pd.DataFrame:
        """
        คำนวณ Order Flow Imbalance และ Depth Features
        """
        # Order Imbalance
        order_imbalance = (order_book_data['bid_volume'] - order_book_data['ask_volume']) / \
                         (order_book_data['bid_volume'] + order_book_data['ask_volume'])
        self.data['order_imbalance'] = order_imbalance.reindex(self.data.index, method='ffill')
        
        # Depth Imbalance
        depth_ratio = order_book_data['bid_depth'] / order_book_data['ask_depth']
        self.data['depth_ratio'] = depth_ratio.reindex(self.data.index, method='ffill')
        
        return self.data
    
    def add_temporal_features(self) -> pd.DataFrame:
        """เพิ่มฟีเจอร์ทางเวลา"""
        self.data['hour'] = self.data.index.hour
        self.data['day_of_week'] = self.data.index.dayofweek
        self.data['is_weekend'] = self.data['day_of_week'].isin([5, 6]).astype(int)
        
        # Cyclical encoding
        self.data['hour_sin'] = np.sin(2 * np.pi * self.data['hour'] / 24)
        self.data['hour_cos'] = np.cos(2 * np.pi * self.data['hour'] / 24)
        self.data['dow_sin'] = np.sin(2 * np.pi * self.data['day_of_week'] / 7)
        self.data['dow_cos'] = np.cos(2 * np.pi * self.data['day_of_week'] / 7)
        
        return self.data
    
    def add_lagged_features(self, target_col: str = 'rv_60', lags: list = [1, 2, 3, 6, 12, 24]) -> pd.DataFrame:
        """เพิ่ม Lagged Values ของความผันผวน"""
        for lag in lags:
            self.data[f'{target_col}_lag{lag}'] = self.data[target_col].shift(lag)
        
        # Rolling statistics
        for window in [6, 12, 24]:
            self.data[f'{target_col}_ma{window}'] = self.data[target_col].rolling(window).mean()
            self.data[f'{target_col}_std{window}'] = self.data[target_col].rolling(window).std()
        
        return self.data
    
    def prepare_dataset(self, target_col: str = 'rv_60', drop_na: bool = True) -> tuple:
        """
        เตรียมข้อมูลสำหรับ Training
        
        Returns:
            X: Features DataFrame
            y: Target Series
        """
        # กำหนด Target (Volatility ของ period ถัดไป)
        self.data['target'] = self.data[target_col].shift(-1)
        
        # เลือก Features
        feature_cols = [col for col in self.data.columns 
                       if col not in ['open', 'high', 'low', 'close', 'volume', 'target']]
        
        X = self.data[feature_cols]
        y = self.data['target']
        
        if drop_na:
            valid_idx = ~(X.isna().any(axis=1) | y.isna())
            X = X[valid_idx]
            y = y[valid_idx]
        
        return X, y


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

engineer = VolatilityFeatureEngineer(btc_data)
engineer.compute_returns()
engineer.compute_realized_volatility(windows=[5, 15, 30, 60])
engineer.add_temporal_features()
engineer.add_lagged_features()

X, y = engineer.prepare_dataset()
print(f"Dataset shape: X={X.shape}, y={y.shape}")
print(f"Features: {list(X.columns)}")

วิธีที่ 1: GARCH Model — วิธีคลาสสิกทางสถิติ

GARCH (Generalized Autoregressive Conditional Heteroskedasticity) เป็นโมเดลทางสถิติที่ใช้กันอย่างแพร่หลายในการพยากรณ์ความผันผวน โดยอาศัยหลักการว่าความแปรปรวนในปัจจุบันขึ้นอยู่กับความแปรปรวนในอดีตและ squared returns

ทฤษฎีพื้นฐานของ GARCH

โมเดล GARCH(p,q) มีสมการดังนี้:

โดยที่ ω > 0, α_i ≥ 0, β_j ≥ 0 และ Σ(α_i + β_j) < 1 (เงื่อนไข Stationarity)

import numpy as np
import pandas as pd
from arch import arch_model
from scipy.optimize import minimize
import warnings
warnings.filterwarnings('ignore')

class GARCHVolatilityModel:
    """
    โมเดล GARCH สำหรับพยากรณ์ความผันผวนของ Bitcoin
    รองรับ: GARCH, EGARCH, GJR-GARCH, TGARCH
    """
    
    def __init__(self, model_type: str = 'GARCH'):
        self.model_type = model_type
        self.model = None
        self.fitted_model = None
        self.params = None
    
    def build_model(self, returns: pd.Series, p: int = 1, q: int = 1, 
                   vol_kwargs: dict = None):
        """
        สร้างโมเดล GARCH
        
        Args:
            returns: Series ของ log returns
            p: GARCH order (lag ของ squared residuals)
            q: ARCH order (lag ของ conditional variance)
            vol_kwargs: parameters เพิ่มเติม (dist='t', o=1 for GJR)
        """
        if vol_kwargs is None:
            vol_kwargs = {}
        
        # สร้างโมเดลตามประเภท
        if self.model_type == 'GARCH':
            self.model = arch_model(returns, vol='Garch', p=p, q=q, **vol_kwargs)
        elif self.model_type == 'EGARCH':
            self.model = arch_model(returns, vol='EGARCH', p=p, q=q, **vol_kwargs)
        elif self.model_type == 'GJR-GARCH':
            self.model = arch_model(returns, vol='GARCH', p=p, o=1, q=q, **vol_kwargs)
        elif self.model_type == 'TGARCH':
            self.model = arch_model(returns, vol='GARCH', p=p, o=1, q=q, **vol_kwargs)
    
    def fit(self, returns: pd.Series, p: int = 1, q: int = 1,
           vol_kwargs: dict = None, show_summary: bool = False):
        """
        Train โมเดล GARCH
        """
        self.build_model(returns, p, q, vol_kwargs)
        
        # Fit โมเดลด้วย Fixed Variance (ใช้ 252 สำหรับ annualize)
        self.fitted_model = self.model.fit(disp='off', options={'maxiter': 1000})
        
        if show_summary:
            print(self.fitted_model.summary())
        
        return self.fitted_model
    
    def forecast(self, horizon: int = 1) -> np.ndarray:
        """
        พยากรณ์ความผันผวนล่วงหน้า
        
        Args:
            horizon: จำนวน period ที่ต้องการพยากรณ์
        
        Returns:
            Array ของ forecasted volatility (annualized)
        """
        if self.fitted_model is None:
            raise ValueError("ต้อง fit โมเดลก่อน")
        
        # พยากรณ์ด้วย ARCH package
        forecast = self.fitted_model.forecast(horizon=horizon)
        
        # คืนค่า variance แล้ว sqrt เพื่อได้ volatility
        variance_forecast = forecast.variance.values[-1, :]
        volatility_forecast = np.sqrt(variance_forecast) * np.sqrt(252 * 24)  # Annualize
        
        return volatility_forecast
    
    def rolling_forecast(self, returns: pd.Series, window: int = 504, 
                        horizon: int = 1) -> pd.Series:
        """
        Rolling Window Forecast สำหรับ Backtesting
        
        Args:
            returns: ข้อมูล returns ทั้งหมด
            window: ขนาด window สำหรับ train
            horizon: horizon สำหรับ forecast
        
        Returns:
            Series ของ forecasted volatility
        """
        forecasts = []
        
        for i in range(window, len(returns)):
            # Train โมเดลด้วย window
            train_data = returns.iloc[i-window:i]
            
            try:
                model = arch_model(train_data * 100, vol='Garch', p=1, q=1)
                res = model.fit(disp='off', options={'maxiter': 500})
                forecast = res.forecast(horizon=horizon)
                pred_vol = np.sqrt(forecast.variance.values[-1, 0]) / 100
                forecasts.append(pred_vol)
            except:
                forecasts.append(np.nan)
        
        return pd.Series(forecasts, index=returns.index[window:])


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

garch = GARCHVolatilityModel(model_type='GARCH')


Fit โมเดล

returns = engineer.data['log_return'].dropna() * 100  # Scale for numerical stability
garch.fit(returns.iloc[-5000:], p=1, q=1, show_summary=True)


พยากรณ์ 1 ชั่วโมงข้างหน้า

next_hour_vol = garch.forecast(horizon=1)
print(f"ความผันผวนที่พยากรณ์ (1 ชั่วโมง): {next_hour_vol[0]*100:.4f}%")


Rolling forecast สำหรับทั้ง dataset

garch_forecasts = garch.rolling_forecast(returns, window=1000, horizon=1)
print(f"Rolling forecast shape: {garch_forecasts.shape}")

วิธีที่ 2: Machine Learning — โมเดล Deep Learning สำหรับ Volatility

วิธี ML ใช้ประโยชน์จากฟีเจอร์หลายมิติและความสัมพันธ์แบบไม่เชิงเส้นที่ GARCH อาจตรวจไม่พบ ในที่นี้เราจะใช้ LightGBM ซึ่งเร็วและแม่นยำ และเปรียบเทียบกับ LSTM สำหรับ Time-Series

import lightgbm as lgb
import numpy as np
import pandas as pd
from sklearn.model_selection import TimeSeriesSplit
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

class MLVolatilityModel:
    """
    Machine Learning Models สำหรับ Volatility Prediction
    รองรับ: LightGBM, XGBoost, Random Forest
    """
    
    def __init__(self, model_type: str = 'lightgbm'):
        self.model_type = model_type
        self.model = None
        self.scaler = StandardScaler()
        self.feature_importance = None
    
    def _get_base_model(self, params: dict = None):
        """สร้างโมเดลตามประเภท"""
        if params is None:
            params = self._default_params()
        
        if self.model_type == 'lightgbm':
            return lgb.LGBMRegressor(**params)
        elif self.model_type == 'xgboost':
            from xgboost import XGBRegressor
            return XGBRegressor(**params)
        elif self.model_type == 'rf':
            from sklearn.ensemble import RandomForestRegressor
            return RandomForestRegressor(**params)
    
    def _default_params(self) -> dict:
        """Default hyperparameters สำหรับ LightGBM"""
        return {
            'objective': 'regression',
            'metric': 'rmse',
            'boosting_type': 'gbdt',
            'num_leaves': 31,
            'learning_rate': 0.05,
            'feature_fraction': 0.8,
            'bagging_fraction': 0.8,
            'bagging_freq': 5,
            'min_child_samples': 20,
            'n_estimators': 500,
            'verbose': -1,
            'random_state': 42,
            'n_jobs': -1
        }
    
    def train(self, X: pd.DataFrame, y: pd.Series, 
              val_size: float = 0.2, params: dict = None):
        """
        Train โมเดล ML
        
        Args:
            X: Features DataFrame
            y: Target Series
            val_size: สัดส่วน validation set
            params: Hyperparameters
        """
        # แบ่งข้อมูล (Time-series split - ไม่สuffle)
        split_idx = int(len(X) * (1 - val_size))
        X_train, X_val = X.iloc[:split_idx], X.iloc[split_idx:]
        y_train, y_val = y.iloc[:split_idx], y.iloc[split_idx:]
        
        # Scale features
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_val_scaled = self.scaler.transform(X_val)
        
        # Train โมเดล
        self.model = self._get_base_model(params)
        
        if self.model_type == 'lightgbm':
            self.model.fit(
                X_train_scaled, y_train,
                eval_set=[(X_val_scaled, y_val)],
                callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)]
            )
        else:
            self.model.fit(X_train_scaled, y_train)
        
        # บันทึก feature importance
        if hasattr(self.model, 'feature_importances_'):
            self.feature_importance = pd.DataFrame({
                'feature': X.columns,
                'importance': self.model.feature_importances_
            }).sort_values('importance', ascending=False)
        
        return self
    
    def predict(self, X: pd.DataFrame) -> np.ndarray:
        """ทำนายความผันผวน"""
        X_scaled = self.scaler.transform(X)
        return self.model.predict(X_scaled)
    
    def rolling_train_predict(self, X: pd.DataFrame, y: pd.Series,
                             window: int = 1000, step: int = 24) -> pd.Series:
        """
        Rolling Window Training & Prediction สำหรับ Backtesting
        
        ใช้เมื่อต้องการโมเดลที่ปรับตัวตามสภาวะตลาด
        """
        predictions = []
        indices = []
        
        for i in range(window, len(X), step):
            # Train window
            train_end = i
            train_start = max(0, i - window)
            
            X_train = X.iloc[train_start:train_end]
            y_train = y.iloc[train_start:train_end]
            
            # Scale
            scaler = StandardScaler()
            X_train_scaled = scaler.fit_transform(X_train)
            
            # Train
            model = self._get_base_model()
            if self.model_type == 'lightgbm':
                model.fit(X_train_scaled, y_train, 
                         callbacks=[lgb.early_stopping(30, verbose=False)])
            else:
                model.fit(X_train_scaled, y_train)
            
            # Predict
            if i < len(X):
                X_pred = scaler.transform(X.iloc[i:i+1])
                pred = model.predict(X_pred)[0]
                predictions.append(pred)
                indices.append(X.index[i])
        
        return pd.Series(predictions, index=indices)
    
    def evaluate(self, y_true: pd.Series, y_pred: pd.Series) -> dict:
        """ประเมินประสิทธิภาพโมเดล"""
        return {
            'rmse': np.sqrt(mean_squared_error(y_true, y_pred)),
            'mae': mean_absolute_error(y_true, y_pred),
            'r2': r2_score(y_true, y_pred),
            'mape': np.mean(np.abs((y_true - y_pred) / y_true)) * 100
        }


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

ml_model = MLVolatilityModel(model_type='lightgbm')


Prepare dataset

X, y = engineer.prepare_dataset()
ml_model.train(X, y, val_size=0.2)


Predict

ml_predictions = ml_model.predict(X)


Feature importance

print("Top 10 Important Features:")
print(ml_model.feature_importance.head(10))


Rolling prediction

rolling_preds = ml_model.rolling_train_predict(X, y, window=2000, step=24)
print(f"Rolling predictions: {len(rolling_preds)} values")

การทำ Backtest และเปรียบเทียบประสิทธิภาพ

การทดสอบประสิทธิภาพโมเดลทั้งสองวิ

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

บทความที่เกี่ยวข้อง