ในโลกของการเทรดคริปโตที่มีการแข่งขันสูง การทำ Backtesting ด้วยข้อมูลประวัติศาสตร์เป็นกุญแจสำคัญที่ทำให้นักเทรดพัฒนากลยุทธ์ที่ทำกำไรได้จริง บทความนี้จะพาคุณเรียนรู้วิธีดึงข้อมูล Bybit Perpetual Futures มาประมวลผลด้วย Python Pandas และนำไปทำ Backtesting เพื่อทดสอบประสิทธิภาพของกลยุทธ์การเทรดของคุณ

กรณีศึกษา: ทีมสตาร์ทอัพ AI ในกรุงเทพฯ

บริบทธุรกิจ

ทีมสตาร์ทอัพ AI แห่งหนึ่งในกรุงเทพฯ กำลังพัฒนาระบบเทรดอัตโนมัติสำหรับ Bybit Perpetual Futures โดยใช้ Machine Learning เพื่อวิเคราะห์แนวโน้มราคาและสร้างสัญญาณซื้อขาย ทีมมีกลยุทธ์การเทรดที่พัฒนาขึ้นแล้ว 8 กลยุทธ์ และต้องการทำ Backtesting กับข้อมูลย้อนหลัง 2 ปี เพื่อทดสอบว่ากลยุทธ์ใดให้ผลตอบแทนดีที่สุด

จุดเจ็บปวด

ทีมเผชิญปัญหาสำคัญหลายประการ:

การย้ายมาใช้ HolySheep AI

หลังจากทดลองใช้หลายบริการ ทีมตัดสินใจย้ายมาใช้ HolySheep AI เนื่องจากเหตุผลหลักคือ อัตรา ¥1=$1 ซึ่งประหยัดได้มากกว่า 85% เมื่อเทียบกับบริการอื่น และรองรับการชำระเงินผ่าน WeChat/Alipay ทำให้การชำระเงินสะดวกสำหรับทีมในเอเชีย

ขั้นตอนการย้ายระบบ

1. การเปลี่ยน Base URL

ทีมเปลี่ยนจาก Base URL เดิมมาใช้ https://api.holysheep.ai/v1 ซึ่งให้ Performance ที่เหนือกว่าพร้อม Latency ต่ำกว่า 50ms

# โค้ดเดิม (ก่อนย้าย)
BASE_URL = "https://api.bybit.com/v5"


โค้ดใหม่ (หลังย้ายมาใช้ HolySheep)

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

2. การหมุนคีย์และการจัดการ API Keys

ทีมตั้งค่า API Key Rotation อัตโนมัติเพื่อป้องกันปัญหา Rate Limiting และเพิ่มความปลอดภัย

import os
from datetime import datetime, timedelta

class HolySheepAPI:
    def __init__(self, api_keys: list):
        self.api_keys = api_keys
        self.current_key_index = 0
        self.key_usage_count = {key: 0 for key in api_keys}
        self.reset_date = datetime.now() + timedelta(days=1)
    
    def get_current_key(self):
        # หมุนคีย์ทุก 1000 คำขอเพื่อหลีกเลี่ยง Rate Limit
        if self.key_usage_count[self.current_key()] >= 1000:
            self.rotate_key()
        return self.current_key()
    
    def current_key(self):
        return self.api_keys[self.current_key_index]
    
    def rotate_key(self):
        self.current_key_index = (self.current_key_index + 1) % len(self.api_keys)
        print(f"หมุนคีย์ไปยัง: {self.current_key()[:8]}... ที่ {datetime.now()}")
    
    def call_api(self, endpoint: str, params: dict = None):
        import requests
        headers = {"Authorization": f"Bearer {self.get_current_key()}"}
        response = requests.get(
            f"https://api.holysheep.ai/v1/{endpoint}",
            headers=headers,
            params=params
        )
        self.key_usage_count[self.current_key()] += 1
        return response.json()


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

api_keys = [
    "YOUR_HOLYSHEEP_API_KEY_1",
    "YOUR_HOLYSHEEP_API_KEY_2", 
    "YOUR_HOLYSHEEP_API_KEY_3"
]
api = HolySheepAPI(api_keys)

3. Canary Deployment

ทีมใช้กลยุทธ์ Canary Deployment โดยเริ่มจากการย้าย 10% ของ Request ไปใช้ HolySheep ก่อน จากนั้นค่อยๆ เพิ่มสัดส่วนจนถึง 100%

import random

class CanaryDeploy:
    def __init__(self, canary_percentage: float = 0.1):
        self.canary_percentage = canary_percentage
        self.stats = {"holysheep": 0, "bybit": 0}
    
    def route_request(self) -> str:
        if random.random() < self.canary_percentage:
            self.stats["holysheep"] += 1
            return "holysheep"
        else:
            self.stats["bybit"] += 1
            return "bybit"
    
    def get_stats(self) -> dict:
        total = sum(self.stats.values())
        return {
            "holysheep_pct": self.stats["holysheep"] / total * 100,
            "bybit_pct": self.stats["bybit"] / total * 100,
            "total_requests": total
        }
    
    def increase_canary(self, increment: float = 0.1):
        self.canary_percentage = min(1.0, self.canary_percentage + increment)
        print(f"เพิ่ม Canary Traffic เป็น {self.canary_percentage * 100}%")


เริ่มต้นด้วย 10% ไป HolySheep

deployer = CanaryDeploy(canary_percentage=0.1)
print(deployer.get_stats())  # {'holysheep_pct': 10.0, 'bybit_pct': 90.0, ...}

ตัวชี้วัด 30 วันหลังจากย้าย

ตัวชี้วัด ก่อนย้าย (Bybit Direct) หลังย้าย (HolySheep) การปรับปรุง
API Response Time 420ms 180ms ↓ 57% (เร็วขึ้น 240ms)
ค่าใช้จ่ายรายเดือน $4,200 $680 ↓ 84% (ประหยัด $3,520)
เวลาประมวลผล Backtesting 8.5 ชั่วโมง 3.2 ชั่วโมง ↓ 62% (เร็วขึ้น 5.3 ชม.)
ความสำเร็จของ Request 96.2% 99.8% ↑ 3.6%

การดึงข้อมูล Bybit Perpetual Futures ด้วย Python

การติดตั้งและตั้งค่าเบื้องต้น

# ติดตั้งไลบรารีที่จำเป็น
pip install pandas numpy requests python-binance pandas-ta


นำเข้าไลบรารี

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


ตั้งค่า Timezone สำหรับประเทศไทย

THB_TZ = pytz.timezone('Asia/Bangkok')


ตั้งค่า HolySheep API

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

def fetch_bybit_data(symbol: str, interval: str, start_time: int, end_time: int):
    """
    ดึงข้อมูล OHLCV จาก Bybit ผ่าน HolySheep API
    สำหรับ Backtesting
    
    Parameters:
    - symbol: คู่เทรด เช่น BTCUSDT
    - interval: ช่วงเวลา เช่น 1m, 5m, 15m, 1h, 4h, 1d
    - start_time: timestamp เริ่มต้น (มิลลิวินาที)
    - end_time: timestamp สิ้นสุด (มิลลิวินาที)
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    params = {
        "category": "perpetual",
        "symbol": symbol,
        "interval": interval,
        "start": start_time,
        "end": end_time,
        "limit": 1000  # สูงสุด 1000 records ต่อครั้ง
    }
    
    try:
        response = requests.get(
            f"{BASE_URL}/market/kline",
            headers=headers,
            params=params,
            timeout=30
        )
        
        if response.status_code == 200:
            data = response.json()
            if data.get("retCode") == 0:
                return data.get("result", {}).get("list", [])
            else:
                print(f"เกิดข้อผิดพลาด: {data.get('retMsg')}")
                return None
        else:
            print(f"HTTP Error: {response.status_code}")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"Request Exception: {e}")
        return None

print("เตรียมพร้อมสำหรับการดึงข้อมูล Bybit Perpetual Futures")

การประมวลผลข้อมูลด้วย Pandas

def fetch_and_process_bybit_data(symbol: str, interval: str, days: int = 30):
    """
    ดึงและประมวลผลข้อมูล Bybit สำหรับ Backtesting
    
    Parameters:
    - symbol: คู่เทรด เช่น BTCUSDT, ETHUSDT
    - interval: ช่วงเวลา (1m, 5m, 15m, 1h, 4h, 1d)
    - days: จำนวนวันย้อนหลัง
    """
    # คำนวณ timestamp
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
    
    # ดึงข้อมูล
    raw_data = fetch_bybit_data(symbol, interval, start_time, end_time)
    
    if not raw_data:
        return None
    
    # สร้าง DataFrame
    df = pd.DataFrame(raw_data, columns=[
        'timestamp', 'open', 'high', 'low', 'close', 'volume', 'turnover'
    ])
    
    # แปลงประเภทข้อมูล
    numeric_cols = ['open', 'high', 'low', 'close', 'volume', 'turnover']
    for col in numeric_cols:
        df[col] = pd.to_numeric(df[col], errors='coerce')
    
    # แปลง timestamp เป็น datetime
    df['datetime'] = pd.to_datetime(df['timestamp'].astype(int), unit='ms')
    
    # แปลงเป็น timezone ประเทศไทย
    df['datetime_thb'] = df['datetime'].dt.tz_localize('UTC').dt.tz_convert('Asia/Bangkok')
    
    # เรียงข้อมูลจากเก่าไปใหม่
    df = df.sort_values('datetime').reset_index(drop=True)
    
    # เพิ่มคอลัมน์ที่มีประโยชน์สำหรับ Backtesting
    df['returns'] = df['close'].pct_change()  # อัตราผลตอบแทนราย period
    df['log_returns'] = np.log(df['close'] / df['close'].shift(1))  # Log returns
    
    # คำนวณ Moving Averages
    df['sma_20'] = df['close'].rolling(window=20).mean()
    df['sma_50'] = df['close'].rolling(window=50).mean()
    df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
    df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean()
    
    # คำนวณ RSI
    delta = df['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    df['rsi'] = 100 - (100 / (1 + rs))
    
    # คำนวณ Bollinger Bands
    df['bb_middle'] = df['close'].rolling(window=20).mean()
    df['bb_std'] = df['close'].rolling(window=20).std()
    df['bb_upper'] = df['bb_middle'] + (df['bb_std'] * 2)
    df['bb_lower'] = df['bb_middle'] - (df['bb_std'] * 2)
    
    # คำนวณ Volume Profile
    df['volume_sma_20'] = df['volume'].rolling(window=20).mean()
    
    return df


ตัวอย่างการดึงข้อมูล BTCUSDT 90 วันย้อนหลัง

df_btc = fetch_and_process_bybit_data("BTCUSDT", "1h", days=90)
print(f"ดึงข้อมูลสำเร็จ: {len(df_btc)} records")
print(df_btc.head())

การสร้างระบบ Backtesting

def backtest_strategy(df: pd.DataFrame, initial_capital: float = 10000, 
                      strategy_type: str = "sma_crossover"):
    """
    ระบบ Backtesting สำหรับทดสอบกลยุทธ์
    
    Parameters:
    - df: DataFrame ที่มีข้อมูล OHLCV และ indicators
    - initial_capital: เงินทุนเริ่มต้น (USD)
    - strategy_type: ประเภทกลยุทธ์ ("sma_crossover", "rsi", "bollinger")
    """
    df = df.copy()
    df['position'] = 0  # 0 = ถือเงินสด, 1 = ถือสินทรัพย์
    df['trade_signal'] = 0  # 1 = ซื้อ, -1 = ขาย, 0 = ถือ
    
    if strategy_type == "sma_crossover":
        # กลยุทธ์ SMA Crossover
        df['trade_signal'] = np.where(
            df['sma_20'] > df['sma_50'], 1, -1
        )
        
        # ส่งสัญญาณซื้อ/ขายเมื่อเกิดการตัดกัน
        df['trade_signal'] = df['trade_signal'].diff().fillna(0)
        df.loc[df['trade_signal'] == 2, 'trade_signal'] = 1   # ซื้อ
        df.loc[df['trade_signal'] == -2, 'trade_signal'] = -1  # ขาย
        
    elif strategy_type == "rsi":
        # กลยุทธ์ RSI Oversold/Overbought
        df.loc[df['rsi'] < 30, 'trade_signal'] = 1   # ซื้อเมื่อ RSI < 30
        df.loc[df['rsi'] > 70, 'trade_signal'] = -1  # ขายเมื่อ RSI > 70
        
    elif strategy_type == "bollinger":
        # กลยุทธ์ Bollinger Bands
        df.loc[df['close'] < df['bb_lower'], 'trade_signal'] = 1   # ซื้อเมื่อราคาต่ำกว่า Lower Band
        df.loc[df['close'] > df['bb_upper'], 'trade_signal'] = -1  # ขายเมื่อราคาสูงกว่า Upper Band
    
    # จำลองการเทรด
    capital = initial_capital
    position = 0  # จำนวนสินทรัพย์ที่ถือ
    position_size = 0.95  # ใช้ 95% ของทุนต่อครั้ง
    
    trades = []
    portfolio_values = []
    
    for i, row in df.iterrows():
        signal = row['trade_signal']
        price = row['close']
        
        if signal == 1 and position == 0:  # สัญญาณซื้อ
            buy_amount = capital * position_size
            position = buy_amount / price
            capital -= buy_amount
            trades.append({
                'datetime': row['datetime_thb'],
                'type': 'BUY',
                'price': price,
                'amount': position,
                'capital': capital
            })
            
        elif signal == -1 and position > 0:  # สัญญาณขาย
            sell_value = position * price
            capital += sell_value
            trades.append({
                'datetime': row['datetime_thb'],
                'type': 'SELL',
                'price': price,
                'amount': position,
                'capital': capital
            })
            position = 0
        
        # คำนวณมูลค่าพอร์ตรวม
        total_value = capital + (position * price)
        portfolio_values.append(total_value)
    
    df['portfolio_value'] = portfolio_values
    
    # คำนวณผลตอบแทน
    total_return = (portfolio_values[-1] - initial_capital) / initial_capital * 100
    
    # คำนวณ Max Drawdown
    df['cummax'] = df['portfolio_value'].cummax()
    df['drawdown'] = (df['portfolio_value'] - df['cummax']) / df['cummax'] * 100
    max_drawdown = df['drawdown'].min()
    
    # คำนวณ Sharpe Ratio
    daily_returns = df['portfolio_value'].pct_change().dropna()
    sharpe_ratio = (daily_returns.mean() / daily_returns.std()) * np.sqrt(252) if daily_returns.std() > 0 else 0
    
    return {
        'total_return': total_return,
        'max_drawdown': max_drawdown,
        'sharpe_ratio': sharpe_ratio,
        'total_trades': len(trades),
        'final_capital': portfolio_values[-1],
        'trades': pd.DataFrame(trades),
        'df': df
    }


ทดสอบกลยุทธ์

result = backtest_strategy(df_btc, initial_capital=10000, strategy_type="sma_crossover")
print(f"ผลตอบแทนรวม: {result['total_return']:.2f}%")
print(f"Max Drawdown: {result['max_drawdown']:.2f}%")
print(f"Sharpe Ratio: {result['sharpe_ratio']:.2f}")
print(f"จำนวนการเทรด: {result['total_trades']}")

การเปรียบเทียบกลยุทธ์หลายแบบ

def compare_strategies(df: pd.DataFrame, initial_capital: float = 10000):
    """
    เปรียบเทียบผลตอบแทนของกลยุทธ์หลายแบบ
    """
    strategies = ['sma_crossover', 'rsi', 'bollinger']
    results = []
    
    for strategy in strategies:
        result = backtest_strategy(df, initial_capital, strategy)
        results.append({
            'Strategy': strategy,
            'Total Return (%)': round(result['total_return'], 2),
            'Max Drawdown (%)': round(result['max_drawdown'], 2),
            'Sharpe Ratio': round(result['sharpe_ratio'], 2),
            'Total Trades': result['total_trades'],
            'Final Capital ($)': round(result['final_capital'], 2)
        })
    
    return pd.DataFrame(results)


เปรียบเทียบกลยุทธ์

comparison = compare_strategies(df_btc, initial_capital=10000)
print(comparison.to_string(index=False))


หากลยุทธ์ที่ดีที่สุด

best_strategy = comparison.loc[comparison['Total Return (%)'].idxmax()]
print(f"\nกลยุทธ์ที่ให้ผลตอบแทนดีที่สุด: {best_strategy['Strategy']}")
print(f"ผลตอบแทน: {best_strategy['Total Return (%)']}%")

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

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