ในโลกของการเทรดเชิงปริมาณ ข้อมูลคือหัวใจสำคัญที่สุด ไม่ว่าจะเป็นการสร้างกลยุทธ์ ทดสอบย้อนกลับ (Backtest) หรือปรับแต่งพารามิเตอร์ ทุกอย่างล้วนต้องพึ่งพาข้อมูลที่ถูกต้องแม่นยำ โดยเฉพาะข้อมูล K-Line (ฉายาแท่งเทียน) จากกระดานเทรร์ที่นิยมอย่าง OKX ซึ่งเป็นแหล่งข้อมูลที่นักเทรดเชิงปริมาณทั่วโลกใช้กันอย่างแพร่หลาย

บทความนี้จะพาคุณไปทำความรู้จักกับวิธีการดึงข้อมูลประวัติศาสตร์จาก OKX API มาประมวลผลผ่านระบบ Backtest โดยใช้ Python เพื่อให้คุณสามารถทดสอบกลยุทธ์การเทรดของตัวเองได้อย่างมีประสิทธิภาพ เราจะมาเริ่มจากสถานการณ์ข้อผิดพลาดจริงที่เคยเกิดขึ้น และวิธีแก้ไขอย่างละเอียด

ปัญหาจริงที่ผมเจอ: ConnectionError และ Rate Limit

ตอนที่ผมเริ่มสร้างระบบ Backtest สำหรับกลยุทธ์ Mean Reversion บนคู่เทรร์ BTC/USDT ผมเจอปัญหาใหญ่หลวงตั้งแต่วันแรก

Traceback (most recent call last):
  File "fetch_okx_data.py", line 45, in <module>
    data = fetch_kline_data(symbol="BTC-USDT", timeframe="1h", limit=1000)
  File "fetch_okx_data.py", line 28, in fetch_kline_data
    response = requests.get(url, headers=headers, timeout=10)
  File "/usr/lib/python3.10/site-packages/requests/api.py", line 76, in get
    return session.request(method=GET, url=url, **kwargs)
  File "/usr/lib/python3.10/site-packages/requests/api.py", line 118, in session.request
    ConnectionError: HTTPSConnectionPool(host='www.okx.com', port=443): 
    Max retries exceeded with url: /api/v5/market/history-candles

หลังจากแก้ปัญหา ConnectionError ได้แล้ว ผมก็เจออุปสรรคต่อไปนั่นคือ Rate Limit

{"code": "60001", "msg": "Return to interface frequency, please retry later", "data": []}

ปัญหาเหล่านี้เป็นบทเรียนที่ทำให้ผมต้องศึกษาวิธีการใช้ OKX API อย่างถูกต้อง และหาทางออกที่เหมาะสมสำหรับระบบ Production

OKX API 101: พื้นฐานที่ต้องเข้าใจก่อนเชื่อมต่อ

ก่อนจะเข้าสู่โค้ดจริง คุณต้องเข้าใจโครงสร้างของ OKX API ก่อน OKX เป็นกระดานเทรร์ที่มี API ที่ค่อนข้างเสถียร แต่มีข้อจำกัดเรื่อง Rate Limit ที่ต้องคำนึงถึง

ประเภทของ API Endpoint ที่เกี่ยวกับ K-Line

โครงสร้าง Timeframe ที่ OKX รองรับ

การดึงข้อมูล K-Line ด้วย Python

มาเริ่มเขียนโค้ดกันเลย ผมจะสร้างฟังก์ชันที่สมบูรณ์สำหรับดึงข้อมูล K-Line จาก OKX

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

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OKXDataFetcher:
    """คลาสสำหรับดึงข้อมูล K-Line จาก OKX API"""
    
    BASE_URL = "https://www.okx.com"
    
    def __init__(self, max_retries=3, retry_delay=2):
        self.max_retries = max_retries
        self.retry_delay = retry_delay
    
    def get_history_candles(self, inst_id: str, bar: str, after: str = None, 
                           before: str = None, limit: int = 100) -> pd.DataFrame:
        """
        ดึงข้อมูล K-Line ประวัติศาสตร์จาก OKX
        
        Args:
            inst_id: รหัสคู่เทรร์ เช่น BTC-USDT
            bar: Timeframe เช่น 1m, 5m, 1H, 1D
            after: Timestamp สิ้นสุด (milliseconds)
            before: Timestamp เริ่มต้น (milliseconds)
            limit: จำนวนข้อมูลสูงสุด 100-1000
        
        Returns:
            DataFrame ที่มีข้อมูล OHLCV
        """
        endpoint = "/api/v5/market/history-candles"
        url = f"{self.BASE_URL}{endpoint}"
        
        params = {
            "instId": inst_id,
            "bar": bar,
            "limit": min(limit, 1000)  # สูงสุด 1000 รายการ
        }
        
        if after:
            params["after"] = after
        if before:
            params["before"] = before
        
        for attempt in range(self.max_retries):
            try:
                response = requests.get(url, params=params, timeout=30)
                
                if response.status_code == 429:
                    logger.warning(f"Rate limit hit, retrying in {self.retry_delay}s...")
                    time.sleep(self.retry_delay)
                    continue
                
                response.raise_for_status()
                data = response.json()
                
                if data.get("code") != "0":
                    error_msg = data.get("msg", "Unknown error")
                    raise ValueError(f"OKX API Error: {error_msg}")
                
                return self._parse_candles(data.get("data", []))
                
            except requests.exceptions.ConnectionError as e:
                logger.error(f"Connection error: {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay * (attempt + 1))
                else:
                    raise
        
        raise Exception("Max retries exceeded")
    
    def _parse_candles(self, raw_data: list) -> pd.DataFrame:
        """แปลงข้อมูลดิบจาก API เป็น DataFrame"""
        if not raw_data:
            return pd.DataFrame()
        
        columns = ["timestamp", "open", "high", "low", "close", "vol", "vol_ccy"]
        df = pd.DataFrame(raw_data, columns=columns)
        
        # แปลงประเภทข้อมูล
        numeric_cols = ["open", "high", "low", "close", "vol", "vol_ccy"]
        for col in numeric_cols:
            df[col] = pd.to_numeric(df[col], errors="coerce")
        
        # แปลง timestamp เป็น datetime
        df["timestamp"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms")
        df["datetime"] = df["timestamp"].dt.tz_localize("UTC").dt.tz_convert("Asia/Bangkok")
        
        # เรียงลำดับจากเก่าไปใหม่
        df = df.sort_values("timestamp").reset_index(drop=True)
        
        return df


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

if __name__ == "__main__":
    fetcher = OKXDataFetcher()
    
    # ดึงข้อมูล BTC/USDT รายชั่วโมง 1000 แท่งล่าสุด
    df = fetcher.get_history_candles(
        inst_id="BTC-USDT",
        bar="1H",
        limit=1000
    )
    
    print(f"ได้ข้อมูล {len(df)} แท่ง")
    print(df.tail())

ระบบดึงข้อมูลแบบ Batch สำหรับ Backtest ระยะยาว

สำหรับการทำ Backtest ที่ต้องการข้อมูลย้อนหลังหลายปี คุณต้องแบ่งการดึงข้อมูลเป็นช่วงๆ เพราะ OKX API จำกัดการดึงได้ครั้งละ 100-1000 รายการ

import requests
import pandas as pd
from datetime import datetime, timedelta
import time
from typing import Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OKXBatchFetcher:
    """คลาสสำหรับดึงข้อมูล K-Line เป็นชุดใหญ่สำหรับ Backtest"""
    
    BASE_URL = "https://www.okx.com"
    
    # ข้อจำกัดของ API
    MAX_LIMIT = 1000
    RATE_LIMIT_DELAY = 0.2  # วินาทีระหว่าง request
    
    def __init__(self):
        self.last_request_time = 0
    
    def _rate_limit_wait(self):
        """รอเพื่อหลีกเลี่ยง Rate Limit"""
        elapsed = time.time() - self.last_request_time
        if elapsed < self.RATE_LIMIT_DELAY:
            time.sleep(self.RATE_LIMIT_DELAY - elapsed)
        self.last_request_time = time.time()
    
    def fetch_date_range(self, inst_id: str, bar: str, 
                        start_date: datetime, end_date: datetime) -> pd.DataFrame:
        """
        ดึงข้อมูล K-Line ในช่วงวันที่กำหนด
        
        Args:
            inst_id: คู่เทรร์ เช่น BTC-USDT
            bar: Timeframe เช่น 1H, 1D
            start_date: วันเริ่มต้น
            end_date: วันสิ้นสุด
        
        Returns:
            DataFrame ที่มีข้อมูลทั้งหมด
        """
        all_data = []
        current_end = end_date
        
        # แปลง datetime เป็น milliseconds
        end_ms = int(end_date.timestamp() * 1000)
        start_ms = int(start_date.timestamp() * 1000)
        
        logger.info(f"กำลังดึงข้อมูล {inst_id} {bar} จาก {start_date} ถึง {end_date}")
        
        while True:
            self._rate_limit_wait()
            
            params = {
                "instId": inst_id,
                "bar": bar,
                "after": str(end_ms),
                "before": str(start_ms),
                "limit": self.MAX_LIMIT
            }
            
            try:
                response = requests.get(
                    f"{self.BASE_URL}/api/v5/market/history-candles",
                    params=params,
                    timeout=30
                )
                
                if response.status_code == 429:
                    logger.warning("Rate limit - รอ 5 วินาที...")
                    time.sleep(5)
                    continue
                
                data = response.json()
                
                if data.get("code") != "0":
                    logger.error(f"API Error: {data.get('msg')}")
                    break
                
                candles = data.get("data", [])
                
                if not candles:
                    logger.info("ไม่มีข้อมูลเพิ่มเติม - เสร็จสิ้นการดึงข้อมูล")
                    break
                
                all_data.extend(candles)
                
                # อัพเดท end timestamp สำหรับการดึงครั้งต่อไป
                # ข้อมูลเรียงจากใหม่ไปเก่า ดังนั้น timestamp สุดท้ายคือข้อมูลที่เก่าที่สุด
                oldest_ts = int(candles[-1][0])
                end_ms = oldest_ts - 1  # ขอข้อมูลก่อนหน้านี้
                
                logger.info(f"ดึงได้ {len(candles)} รายการ, ต่อจาก timestamp {oldest_ts}")
                
                # ถ้า timestamp ที่ได้เก่ากว่าวันเริ่มต้น ให้หยุด
                if oldest_ts <= start_ms:
                    break
                
            except Exception as e:
                logger.error(f"เกิดข้อผิดพลาด: {e}")
                time.sleep(2)
                continue
        
        # แปลงข้อมูลทั้งหมดเป็น DataFrame
        df = self._process_data(all_data)
        
        logger.info(f"รวมได้ข้อมูลทั้งหมด {len(df)} แท่ง")
        
        return df
    
    def _process_data(self, raw_data: list) -> pd.DataFrame:
        """ประมวลผลข้อมูลดิบเป็น DataFrame ที่พร้อมใช้งาน"""
        if not raw_data:
            return pd.DataFrame()
        
        df = pd.DataFrame(
            raw_data,
            columns=["timestamp", "open", "high", "low", "close", "vol", "vol_ccy"]
        )
        
        # แปลงประเภทข้อมูล
        for col in ["open", "high", "low", "close", "vol", "vol_ccy"]:
            df[col] = pd.to_numeric(df[col], errors="coerce")
        
        # แปลง timestamp
        df["datetime"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms")
        df["datetime"] = df["datetime"].dt.tz_localize("UTC").dt.tz_convert("Asia/Bangkok")
        
        # ลบข้อมูลซ้ำและเรียงลำดับ
        df = df.drop_duplicates(subset=["timestamp"]).sort_values("timestamp")
        
        return df.reset_index(drop=True)



ตัวอย่างการใช้งานสำหรับ Backtest 2 ปี

if __name__ == "__main__":
    fetcher = OKXBatchFetcher()
    
    # ดึงข้อมูล 2 ปีย้อนหลัง
    end_date = datetime.now()
    start_date = end_date - timedelta(days=730)  # 2 ปี
    
    df = fetcher.fetch_date_range(
        inst_id="BTC-USDT",
        bar="1H",
        start_date=start_date,
        end_date=end_date
    )
    
    # บันทึกเป็น CSV
    df.to_csv("btc_usdt_2y_1h.csv", index=False)
    print(f"ข้อมูลถูกบันทึกแล้ว: {len(df)} แท่ง")
    print(f"ช่วงเวลา: {df['datetime'].min()} ถึง {df['datetime'].max()}")

เชื่อมต่อกับระบบ Backtest

หลังจากได้ข้อมูลแล้ว ขั้นตอนต่อไปคือการนำไปใช้กับระบบ Backtest ผมจะแสดงตัวอย่างการสร้าง Backtest Engine แบบง่ายที่คุณสามารถนำไปต่อยอดได้

import pandas as pd
import numpy as np
from datetime import datetime
from typing import List, Tuple, Optional
from dataclasses import dataclass
from enum import Enum

class Signal(Enum):
    """สัญญาณการเทรด"""
    BUY = 1
    SELL = -1
    HOLD = 0

@dataclass
class Trade:
    """ข้อมูลการเทรด"""
    entry_time: datetime
    entry_price: float
    quantity: float
    exit_time: Optional[datetime] = None
    exit_price: Optional[float] = None
    pnl: Optional[float] = None
    pnl_pct: Optional[float] = None

class SimpleBacktester:
    """
    Backtest Engine แบบง่ายสำหรับทดสอบกลยุทธ์
    
    รองรับ:
    - กลยุทธ์ Moving Average Crossover
    - กลยุทธ์ RSI
    - กลยุทธ์ Mean Reversion
    """
    
    def __init__(self, initial_capital: float = 10000, fee_rate: float = 0.001):
        self.initial_capital = initial_capital
        self.fee_rate = fee_rate
        self.trades: List[Trade] = []
        self.current_position = 0
        self.cash = initial_capital
        self.equity_curve = []
    
    def load_data(self, df: pd.DataFrame):
        """โหลดข้อมูล OHLCV"""
        self.df = df.copy()
        self._calculate_indicators()
    
    def _calculate_indicators(self):
        """คำนวณ Indicators พื้นฐาน"""
        self.df["sma_fast"] = self.df["close"].rolling(10).mean()
        self.df["sma_slow"] = self.df["close"].rolling(50).mean()
        self.df["rsi"] = self._calculate_rsi(self.df["close"], 14)
        self.df["bb_middle"] = self.df["close"].rolling(20).mean()
        self.df["bb_std"] = self.df["close"].rolling(20).std()
        self.df["bb_upper"] = self.df["bb_middle"] + 2 * self.df["bb_std"]
        self.df["bb_lower"] = self.df["bb_middle"] - 2 * self.df["bb_std"]
    
    @staticmethod
    def _calculate_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
        """คำนวณ RSI"""
        delta = prices.diff()
        gain = (delta.where(delta > 0, 0)).rolling(period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(period).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))
    
    def ma_crossover_strategy(self, fast: int = 10, slow: int = 50):
        """กลยุทธ์ Moving Average Crossover"""
        self.df["signal"] = 0
        self.df.loc[self.df["sma_fast"] > self.df["sma_slow"], "signal"] = 1
        self.df.loc[self.df["sma_fast"] < self.df["sma_slow"], "signal"] = -1
        
        # เปลี่ยนสัญญาณเฉพาะจุดตัด
        self.df["signal"] = self.df["signal"].diff().fillna(0)
        self.df.loc[self.df["signal"] == 2, "signal"] = 1  # Golden Cross
        self.df.loc[self.df["signal"] == -2, "signal"] = -1  # Death Cross
    
    def rsi_strategy(self, oversold: int = 30, overbought: int = 70):
        """กลยุทธ์ RSI"""
        self.df["signal"] = 0
        self.df.loc[self.df["rsi"] < oversold, "signal"] = 1  # ซื้อเมื่อ oversold
        self.df.loc[self.df["rsi"] > overbought, "signal"] = -1  # ขายเมื่อ overbought
    
    def mean_reversion_strategy(self, bb_period: int = 20, bb_std: int = 2):
        """กลยุทธ์ Mean Reversion บน Bollinger Bands"""
        self.df["signal"] = 0
        self.df.loc[self.df["close"] < self.df["bb_lower"], "signal"] = 1  # ซื้อเมื่อราคาต่ำกว่า BB ล่าง
        self.df.loc[self.df["close"] > self.df["bb_upper"], "signal"] = -1  # ขายเมื่อราคาสูงกว่า BB บน
    
    def run(self) -> dict:
        """รัน Backtest"""
        self.trades = []
        self.current_position = 0
        self.cash = self.initial_capital
        
        for idx, row in self.df.iterrows():
            signal = row["signal"]
            price = row["close"]
            timestamp = row["datetime"]
            
            # ซื้อ
            if signal == 1 and self.current_position == 0:
                fee = price * self.fee_rate
                quantity = (self.cash - fee) / price
                self.current_position = quantity
                self.cash = 0
                self.trades.append(Trade(
                    entry_time=timestamp,
                    entry_price=price,
                    quantity=quantity
                ))
            
            # ขาย
            elif signal == -1 and self.current_position > 0:
                fee = self.current_position * price * self.fee_rate
                proceeds = self.current_position * price - fee
                self.cash = proceeds
                self.trades[-1].exit_time = timestamp
                self.trades[-1].exit_price = price
                self.trades[-1].pnl = proceeds - self.trades[-1].entry_price * self.trades[-1].quantity
                self.trades[-1].pnl_pct = (price / self.trades[-1].entry_price - 1) * 100
                self.current_position = 0
        
        return self._calculate_performance()
    
    def _calculate_performance(self) -> dict:
        """คำนวณผลตอบแทนและ Metrics"""
        completed_trades = [t for t in self.trades if t.exit_price is not None]
        
        if not completed_trades:
            return {"error": "ไม่มีการเทรดที่สมบูรณ์"}
        
        total_pnl = sum(t.pnl for t in completed_trades)
        win_trades = [t for t in completed_trades if t.pnl > 0]
        lose_trades = [t for t in completed_trades if t.pnl <= 0]
        
        return {
            "initial_capital": self.initial_capital,
            "final_capital": self.cash + self.current_position * self.df.iloc[-1]["close"],
            "total_pnl": total_pnl,
            "total_return_pct": (total_pnl / self.initial_capital) * 100,
            "total_trades": len(completed_trades),
            "win_rate": len(win_trades) / len(completed_trades) * 100,
            "avg_win": np.mean([t.pnl for t in win_trades]) if win_trades else 0,
            "avg_loss": np.mean([t.pnl for t in lose_trades]) if lose_trades else 0,
            "profit_factor": abs(sum(t.pnl for t in win_trades) / sum(t.pnl for t in lose_trades)) if lose_trades else float("inf"),
            "max_drawdown": self._calculate_max_drawdown()
        }
    
    def _calculate_max_drawdown(self) -> float:
        """คำนวณ Maximum Drawdown"""
        if not self.trades:
            return 0
        
        peak = self.initial_capital
        max_dd = 0
        
        for trade in self.trades:
            if trade.exit_price:
                equity = self.initial_capital + sum(t.pnl for t in self.trades if t.exit_time <= trade.exit_time)
                peak = max(peak, equity)
                dd = (peak - equity) / peak * 100
                max_dd = max(max_dd, dd)
        
        return max_dd



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

if __name__ == "__main__":
    # โหลดข้อมูลจากไฟล์ CSV (ที่ดึงมาจาก OKX)
    df = pd.read_csv("btc_usdt_2y_1h.csv", parse_dates=["datetime"])
    
    # สร้าง Backtester
    backtester = SimpleBacktester(initial_capital=10000, fee_rate=0.001)
    backtester.load_data(df)
    
    # เลือกกลยุทธ์
    backtester.ma_crossover_strategy(fast=10, slow=50)
    
    # รัน Backtest
    results = backtester.run()
    
    print("=" * 50)
    print("ผลลัพธ์ Backtest")
    print("=" * 50)
    for key, value in results.items():
        if isinstance(value, float):
            print(f"{key}: {value:.2f}")
        else:
            print(f"{key}: {value}")

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. ConnectionError: HTTPSConnectionPool Timeout

# ปัญหา: เชื่อมต่อไม่ได้เนื่องจาก Timeout


วิธีแก้ไข: เพิ่ม timeout และ retry logic


import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session_with_retry():
    """สร้าง session ที่มี retry mechanism"""
    session = requests.Session()
    
    # ตั้งค่า retry strategy
    retry_strategy = Retry(
        total=5,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "OPTIONS"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session


ใช้งาน

session = create_session_with_retry()
response = session.get(url, timeout=60)

2. 401 Unauthorized / API Error 50101

# ปัญหา: ไม่ได้รับ

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