ในโลกของสกุลเงินดิจิทัล กลยุทธ์ Funding Rate Arbitrage เป็นหนึ่งในวิธีการที่ได้รับความนิยมในหมู่นักเทรดระดับมืออาชีพ บทความนี้จะพาคุณเรียนรู้การดึงข้อมูลประวัติ Funding Rate จาก Bybit API และการสร้างระบบ Backtesting เพื่อทดสอบประสิทธิภาพของกลยุทธ์

Funding Rate คืออะไรและทำงานอย่างไร

Funding Rate เป็นกลไกที่ใช้ในสัญญา Perpetual Futures เพื่อให้ราคาของสัญญาใกล้เคียงกับราคา Spot ของสินทรัพย์อ้างอิง โดยมีการชำระเงินทุก 8 ชั่วโมง:

การรับข้อมูล Funding Rate จาก Bybit API

Bybit มี Public API ที่สามารถใช้ดึงข้อมูล Funding Rate ได้ฟรีโดยไม่ต้องยืนยันตัวตน

import requests
import pandas as pd
from datetime import datetime, timedelta
import time

class BybitFundingRateCollector:
    """คลาสสำหรับรวบรวมข้อมูล Funding Rate จาก Bybit API"""
    
    BASE_URL = "https://api.bybit.com"
    
    def __init__(self, symbols=None):
        """
        กำหนด symbols ที่ต้องการเก็บข้อมูล
        ค่าเริ่มต้น: symbols ยอดนิยมในตลาด Perpetual Futures
        """
        self.symbols = symbols or [
            'BTCUSDT', 'ETHUSDT', 'SOLUSDT', 'BNBUSDT', 
            'XRPUSDT', 'ADAUSDT', 'DOGEUSDT', 'AVAXUSDT'
        ]
    
    def get_funding_rate(self, symbol, start_time=None, end_time=None, limit=200):
        """
        ดึงข้อมูล Funding Rate History ของ symbol ที่กำหนด
        
        Parameters:
        - symbol: ชื่อคู่เทรด เช่น 'BTCUSDT'
        - start_time: timestamp เริ่มต้น (milliseconds)
        - end_time: timestamp สิ้นสุด (milliseconds)
        - limit: จำนวนข้อมูลสูงสุด (max 1000)
        
        Returns:
        - DataFrame ที่มี columns: symbol, fundingRate, fundingRateTimestamp, settlementTime
        """
        endpoint = "/public/v3/perpetual/usdt-perpetual/public/contract/instrument-info"
        params = {
            'category': 'linear',
            'symbol': symbol,
            'limit': limit
        }
        
        try:
            response = requests.get(
                f"{self.BASE_URL}{endpoint}", 
                params=params, 
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            
            if data['retCode'] == 0:
                return self._parse_funding_data(data['result']['list'])
            else:
                print(f"Error for {symbol}: {data['retMsg']}")
                return pd.DataFrame()
                
        except requests.exceptions.RequestException as e:
            print(f"Request failed for {symbol}: {e}")
            return pd.DataFrame()
    
    def get_funding_rate_for_period(self, symbol, days=90):
        """
        ดึงข้อมูล Funding Rate ย้อนหลังตามจำนวนวันที่กำหนด
        
        Parameters:
        - symbol: ชื่อคู่เทรด
        - days: จำนวนวันย้อนหลัง
        
        Returns:
        - DataFrame พร้อมข้อมูล Funding Rate
        """
        end_time = int(time.time() * 1000)
        start_time = int((time.time() - days * 24 * 3600) * 1000)
        
        all_data = []
        current_start = start_time
        
        while current_start < end_time:
            df = self._fetch_funding_chunk(symbol, current_start, end_time)
            if df.empty:
                break
            
            all_data.append(df)
            
            # ดึงข้อมูลช่วงถัดไป
            last_timestamp = df['fundingRateTimestamp'].min()
            current_start = last_timestamp - 1
            
            # หยุดชั่วคราวเพื่อหลีกเลี่ยง rate limit
            time.sleep(0.2)
        
        if all_data:
            combined_df = pd.concat(all_data, ignore_index=True)
            combined_df = combined_df.drop_duplicates()
            combined_df = combined_df.sort_values('fundingRateTimestamp')
            return combined_df
        
        return pd.DataFrame()
    
    def _fetch_funding_chunk(self, symbol, start_time, end_time, limit=200):
        """ดึงข้อมูล Funding Rate เป็นช่วงๆ"""
        endpoint = "/public/v3/perpetual/usdt-perpetual/public/contract/instrument-info"
        params = {
            'category': 'linear',
            'symbol': symbol,
            'limit': limit
        }
        
        try:
            response = requests.get(
                f"{self.BASE_URL}{endpoint}", 
                params=params, 
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            
            if data['retCode'] == 0:
                return self._parse_funding_data(data['result']['list'])
            
        except Exception as e:
            print(f"Chunk fetch error: {e}")
        
        return pd.DataFrame()
    
    def _parse_funding_data(self, raw_data):
        """แปลงข้อมูลดิบเป็น DataFrame"""
        parsed = []
        for item in raw_data:
            parsed.append({
                'symbol': item.get('symbol'),
                'fundingRate': float(item.get('fundingRate', 0)),
                'fundingRateTimestamp': int(item.get('fundingRateE4', 0)) // 10000,
                'markPrice': float(item.get('markPrice', 0)),
                'indexPrice': float(item.get('indexPrice', 0)),
                'nextFundingTime': item.get('nextFundingTime'),
                'settleTime': item.get('settleTime')
            })
        
        return pd.DataFrame(parsed)
    
    def collect_all_symbols(self, days=90):
        """
        ดึงข้อมูล Funding Rate สำหรับทุก symbol ที่กำหนด
        
        Returns:
        - Dictionary ของ DataFrames โดยใช้ symbol เป็น key
        """
        results = {}
        
        for symbol in self.symbols:
            print(f"กำลังดึงข้อมูล {symbol}...")
            df = self.get_funding_rate_for_period(symbol, days)
            
            if not df.empty:
                results[symbol] = df
                print(f"✓ ได้ข้อมูล {symbol} {len(df)} รายการ")
            else:
                print(f"✗ ไม่พบข้อมูล {symbol}")
            
            time.sleep(0.3)  # หลีกเลี่ยง rate limit
        
        return results


การใช้งาน

if __name__ == "__main__":
    collector = BybitFundingRateCollector(symbols=['BTCUSDT', 'ETHUSDT', 'SOLUSDT'])
    
    # ดึงข้อมูลย้อนหลัง 90 วัน
    funding_data = collector.collect_all_symbols(days=90)
    
    # แสดงผลสถิติ
    for symbol, df in funding_data.items():
        print(f"\n{symbol} Funding Rate Statistics:")
        print(df['fundingRate'].describe())

การสร้างระบบ Backtesting สำหรับ Funding Rate Arbitrage

หลังจากได้ข้อมูล Funding Rate แล้ว ขั้นตอนถัดไปคือการสร้างระบบ Backtesting เพื่อทดสอบว่ากลยุทธ์มีประสิทธิภาพจริงหรือไม่

import pandas as pd
import numpy as np
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

class FundingRateArbitrageBacktester:
    """
    ระบบ Backtesting สำหรับกลยุทธ์ Funding Rate Arbitrage
    
    กลยุทธ์: 
    1. เข้า Short สัญญา Perpetual เมื่อ Funding Rate > threshold
    2. รับ Funding Payment ทุก 8 ชั่วโมง
    3. ปิดสถานะเมื่อ Funding Rate < exit_threshold หรือถึงเป้าหมายกำไร
    """
    
    def __init__(self, 
                 initial_capital=10000,
                 funding_threshold=0.001,  # 0.1% ต่อ 8 ชั่วโมง
                 exit_threshold=0.0001,     # 0.01% ต่อ 8 ชั่วโมง
                 leverage=3,
                 maker_fee=0.0002,
                 taker_fee=0.0006,
                 max_position_per_trade=0.3):  # ใช้ทุนไม่เกิน 30% ต่อสถานะ
        
        self.initial_capital = initial_capital
        self.funding_threshold = funding_threshold
        self.exit_threshold = exit_threshold
        self.leverage = leverage
        self.maker_fee = maker_fee
        self.taker_fee = taker_fee
        self.max_position_per_trade = max_position_per_trade
        
        self.trades = []
        self.equity_curve = []
        self.current_capital = initial_capital
    
    def run_backtest(self, funding_rate_df, price_df=None, symbol='UNKNOWN'):
        """
        รัน Backtesting บนข้อมูล Funding Rate
        
        Parameters:
        - funding_rate_df: DataFrame ที่มี columns [fundingRate, fundingRateTimestamp]
        - price_df: DataFrame ที่มี columns [open, high, low, close, timestamp]
        - symbol: ชื่อคู่เทรดสำหรับบันทึก
        
        Returns:
        - Dictionary ที่มีผลลัพธ์การ backtest
        """
        if funding_rate_df.empty:
            return self._empty_result()
        
        # คำนวณ Funding Rate รายวัน (8 ชั่วโมง = 3 ครั้งต่อวัน)
        funding_rate_df = funding_rate_df.copy()
        funding_rate_df['daily_funding_rate'] = funding_rate_df['fundingRate'] * 3
        funding_rate_df['date'] = pd.to_datetime(
            funding_rate_df['fundingRateTimestamp'], unit='s'
        )
        
        # กรองข้อมูลที่มีค่า funding rate เกิน threshold
        signals = funding_rate_df[
            funding_rate_df['fundingRate'].abs() >= self.funding_threshold
        ].copy()
        
        if signals.empty:
            return self._empty_result()
        
        # จำลองการเทรด
        for idx, row in signals.iterrows():
            trade_result = self._simulate_trade(row, symbol, price_df)
            
            if trade_result:
                self.trades.append(trade_result)
                
                # อัปเดต Capital
                pnl = trade_result['pnl']
                self.current_capital += pnl
                
                # บันทึก Equity Curve
                self.equity_curve.append({
                    'timestamp': row['fundingRateTimestamp'],
                    'equity': self.current_capital,
                    'trade_id': len(self.trades)
                })
        
        return self._calculate_performance()
    
    def _simulate_trade(self, signal_row, symbol, price_df=None):
        """
        จำลองการเทรดครั้งเดียว
        
        กลยุทธ์: Short Perpetual + Long Spot
        รับ Funding = position_size * funding_rate (ทุก 8 ชม.)
        จ่ายค่า Fee เมื่อเปิดและปิดสถานะ
        """
        entry_funding = signal_row['fundingRate']
        entry_time = signal_row['fundingRateTimestamp']
        
        # คำนวณขนาดสถานะ
        position_value = self.current_capital * self.max_position_per_trade
        position_size = position_value * self.leverage
        
        # ค่า Funding ที่คาดว่าจะได้รับต่อ 8 ชั่วโมง
        funding_per_period = position_size * entry_funding
        
        # ตรวจสอบว่าเป็น Short (รับ funding เมื่อ funding rate > 0)
        is_short_position = entry_funding > 0
        
        # คำนวณระยะเวลาถือสถานะ (จนกว่า funding rate จะลดลงต่ำกว่า exit threshold)
        holding_periods = 1  # อย่างน้อย 1 period
        
        # จำลองว่า holding ไปจนถึง signal ถัดไปที่ตรงเงื่อนไขออก
        # (ใน backtest จริง ต้องมี price data ประกอบ)
        
        # รายได้จาก Funding
        total_funding = funding_per_period * holding_periods
        
        # ค่าใช้จ่าย: Fees
        entry_fee = position_value * self.taker_fee
        exit_fee = position_value * self.taker_fee
        total_fees = entry_fee + exit_fee
        
        # คำนวณ PnL
        # สมมติว่าไม่มี position เปลี่ยนแปลงจากราคา (hedge กับ spot)
        # กำไรมาจาก Funding ลบค่า Fee
        pnl = total_funding - total_fees
        
        # คำนวณ ROI
        roi = (pnl / position_value) * 100
        
        return {
            'symbol': symbol,
            'entry_time': entry_time,
            'entry_funding_rate': entry_funding,
            'position_size': position_size,
            'holding_periods': holding_periods,
            'funding_earned': total_funding,
            'fees': total_fees,
            'pnl': pnl,
            'roi_percent': roi,
            'is_short': is_short_position
        }
    
    def _calculate_performance(self):
        """คำนวณผลตอบแทนและสถิติ"""
        if not self.trades:
            return self._empty_result()
        
        df_trades = pd.DataFrame(self.trades)
        
        total_pnl = df_trades['pnl'].sum()
        win_rate = (df_trades['pnl'] > 0).mean() * 100
        avg_pnl = df_trades['pnl'].mean()
        
        # คำนวณ Max Drawdown
        equity_series = pd.DataFrame(self.equity_curve)
        if not equity_series.empty:
            equity_series['cummax'] = equity_series['equity'].cummax()
            equity_series['drawdown'] = (equity_series['equity'] - equity_series['cummax']) / equity_series['cummax']
            max_drawdown = equity_series['drawdown'].min() * 100
        else:
            max_drawdown = 0
        
        # Sharpe Ratio (ประมาณ)
        if len(df_trades) > 1:
            returns = df_trades['pnl'] / self.initial_capital
            sharpe_ratio = returns.mean() / returns.std() * np.sqrt(365) if returns.std() > 0 else 0
        else:
            sharpe_ratio = 0
        
        return {
            'total_trades': len(df_trades),
            'total_pnl': total_pnl,
            'final_capital': self.current_capital,
            'total_return_percent': (self.current_capital / self.initial_capital - 1) * 100,
            'win_rate': win_rate,
            'avg_pnl_per_trade': avg_pnl,
            'max_drawdown_percent': max_drawdown,
            'sharpe_ratio': sharpe_ratio,
            'trades': df_trades
        }
    
    def _empty_result(self):
        return {
            'total_trades': 0,
            'total_pnl': 0,
            'final_capital': self.current_capital,
            'total_return_percent': 0,
            'win_rate': 0,
            'avg_pnl_per_trade': 0,
            'max_drawdown_percent': 0,
            'sharpe_ratio': 0,
            'trades': pd.DataFrame()
        }


การใช้งาน

if __name__ == "__main__":
    # สร้างข้อมูล Funding Rate ตัวอย่าง (ในการใช้งานจริง ดึงจาก API)
    sample_data = {
        'fundingRate': [0.001, 0.0015, 0.0012, 0.0008, 0.0005, 0.0018, 0.0011],
        'fundingRateTimestamp': [
            1704067200, 1704096000, 1704124800, 1704153600, 
            1704182400, 1704211200, 1704240000
        ]
    }
    df_funding = pd.DataFrame(sample_data)
    
    # รัน Backtest
    backtester = FundingRateArbitrageBacktester(
        initial_capital=10000,
        funding_threshold=0.0005,
        leverage=3
    )
    
    result = backtester.run_backtest(df_funding, symbol='BTCUSDT')
    
    print("=" * 50)
    print("BACKTEST RESULTS - Funding Rate Arbitrage")
    print("=" * 50)
    print(f"จำนวนการเทรด: {result['total_trades']}")
    print(f"กำไร/ขาดทุนรวม: ${result['total_pnl']:.2f}")
    print(f"ผลตอบแทนรวม: {result['total_return_percent']:.2f}%")
    print(f"อัตราชนะ: {result['win_rate']:.1f}%")
    print(f"Max Drawdown: {result['max_drawdown_percent']:.2f}%")
    print(f"Sharpe Ratio: {result['sharpe_ratio']:.2f}")

การวิเคราะห์ผลลัพธ์และปรับแต่งกลยุทธ์

หลังจากรัน Backtest แล้ว สิ่งสำคัญคือการวิเคราะห์ผลลัพธ์และปรับแต่งพารามิเตอร์ให้เหมาะสมกับสภาวะตลาด

การใช้ AI วิเคราะห์ข้อมูล Funding Rate อัตโนมัติ

ในการวิเคราะห์ข้อมูล Funding Rate จำนวนมาก การใช้ AI ช่วยวิเคราะห์จะช่วยประหยัดเวลาและค้นพบ patterns ที่ซ่อนอยู่ได้ดียิ่งขึ้น สมัครที่นี่ เพื่อทดลองใช้ HolySheep AI ซึ่งมี API ราคาถูกและเร็ว:

import requests
import json

class FundingRateAnalyzer:
    """ใช้ AI วิเคราะห์ข้อมูล Funding Rate อัตโนมัติ"""
    
    # ใช้ HolySheep AI API - ราคาถูกกว่า 85%+ เมื่อเทียบกับ OpenAI
    HOLYSHEEP_API_URL = "https://api.holysheep.ai/v1/chat/completions"
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # แทนที่ด้วย API key จริงของคุณ
    
    def __init__(self):
        self.headers = {
            "Authorization": f"Bearer {self.API_KEY}",
            "Content-Type": "application/json"
        }
    
    def analyze_funding_opportunity(self, symbol, current_funding_rate, 
                                    historical_avg, volatility, market_sentiment):
        """
        ใช้ AI วิเคราะห์โอกาส Funding Rate Arbitrage
        
        Returns:
        - คำแนะนำการเทรดพร้อมความมั่นใจ
        """
        
        prompt = f"""คุณเป็นผู้เชี่ยวชาญด้าน Crypto Trading

วิเคราะห์โอกาส Funding Rate Arbitrage สำหรับ {symbol}:

ข้อมูลปัจจุบัน:
- Funding Rate ปัจจุบัน: {current_funding_rate:.4f} ({current_funding_rate*100:.2f}% ต่อ 8 ชม.)
- ค่าเฉลี่ย historical: {historical_avg:.4f}
- Volatility: {volatility:.4f}
- Market Sentiment: {market_sentiment}

โปรดให้:
1. ความมั่นใจในการเข้าเทรด (0-100%)
2. Risk/Reward Ratio ที่คาดว่าจะได้รับ
3. ข้อควรระวังในการเทรด
4. ขนาดสถานะที่แนะนำ (% ของ Capital)

ตอบเป็น JSON format ดังนี้:
{{"confidence": number, "risk_reward": number, "warnings": string, "recommended_size": number}}"""
        
        payload = {
            "model": "gpt-4.1",  # หรือ deepseek-v3.2 สำหรับงานวิเคราะห์
            "messages": [
                {"role": "system", "content": "คุณเป็นที่ปรึกษาการลงทุนคริปโตที่มีประสบการณ์"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 500
        }
        
        try:
            response = requests.post(
                self.HOLYSHEEP_API_URL,
                headers=self.headers,
                json=payload,
                timeout=30
            )
            
            if response.status_code == 200:
                result = response.json()
                content = result['choices'][0]['message']['content']
                
                # แปลง response เป็น dict
                # ควรใช้ json.loads() ในการ parse
                return json.loads(content)
            else:
                print(f"API Error: {response.status_code}")
                return None
                
        except Exception as e:
            print(f"Request failed: {e}")
            return None
    
    def generate_backtest_report(self, backtest_results):
        """
        ใช้ AI สร้างรายงานการ Backtest อย่างละเอียด
        """
        
        prompt = f"""สร้างรายงานวิเคราะห์ Backtest สำหรับกลยุทธ์ Funding Rate Arbitrage:

ผลลัพธ์:
- จำนวนการเทรด: {backtest_results.get('total_trades', 0)}
- กำไร/ขาดทุนรวม: ${backtest_results.get('total_pnl', 0):.

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