การทดสอบย้อนกลับกลยุทธ์สัญญา OKX: กระบวนการทำความสะอาดข้อมูลประวัติและการคำนวณตัวชี้วัดอย่างครบวงจร

บทนำ

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

ข้อมูลเบื้องต้น

ก่อนเริ่มกระบวนการ คุณต้องเตรียมข้อมูล OHLCV (Open, High, Low, Close, Volume) จาก OKX API โดยข้อมูลดิบที่ได้มามักมีปัญหาหลายประการที่ต้องแก้ไขก่อนนำไปใช้คำนวณตัวชี้วัด

กระบวนการทำความสะอาดข้อมูลประวัติ

import pandas as pd
import numpy as np
from datetime import datetime

class OKXDataCleaner:
    def __init__(self, data: pd.DataFrame):
        self.data = data.copy()
    
    def basic_validation(self):
        # ตรวจสอบคอลัมน์ที่จำเป็น
        required_cols = ['timestamp', 'open', 'high', 'low', 'close', 'volume']
        missing_cols = [col for col in required_cols if col not in self.data.columns]
        
        if missing_cols:
            raise ValueError(f"คอลัมน์ที่ขาดหายไป: {missing_cols}")
        
        # ตรวจสอบค่าว่าง
        null_count = self.data.isnull().sum()
        print(f"ค่าว่างในแต่ละคอลัมน์:\n{null_count}")
        
        return self
    
    def remove_anomalies(self, price_change_threshold=0.5):
        """
        ลบข้อมูลที่ผิดปกติ เช่น:
        - ราคาปิดต่ำกว่าราคาต่ำสุด
        - ราคาเปิดสูงกว่าราคาสูงสุด
        - การเปลี่ยนแปลงราคาผิดปกติ
        """
        initial_len = len(self.data)
        
        # กรอง: ราคาปิดต้องอยู่ระหว่าง low และ high
        self.data = self.data[
            (self.data['close'] >= self.data['low']) & 
            (self.data['close'] <= self.data['high'])
        ]
        
        # กรอง: ราคาเปิดต้องอยู่ระหว่าง low และ high
        self.data = self.data[
            (self.data['open'] >= self.data['low']) & 
            (self.data['open'] <= self.data['high'])
        ]
        
        # ลบการเปลี่ยนแปลงราคาที่ผิดปกติ
        self.data['price_change'] = self.data['close'].pct_change()
        self.data = self.data[
            (self.data['price_change'].abs() < price_change_threshold) | 
            (self.data['price_change'].isnull())
        ]
        self.data = self.data.drop('price_change', axis=1)
        
        removed = initial_len - len(self.data)
        print(f"ลบข้อมูลผิดปกติ {removed} แถว ({removed/initial_len*100:.2f}%)")
        
        return self
    
    def handle_missing_values(self, method='ffill'):
        """
        จัดการค่าว่าง:
        - ffill: ใช้ค่าก่อนหน้า
        - bfill: ใช้ค่าถัดไป
        - interpolate: คำนวณค่าระหว่างกลาง
        """
        fill_count = self.data.isnull().sum().sum()
        
        if method == 'ffill':
            self.data = self.data.fillna(method='ffill')
        elif method == 'bfill':
            self.data = self.data.fillna(method='bfill')
        elif method == 'interpolate':
            self.data = self.data.interpolate(method='linear')
        
        print(f"เติมค่าว่าง {fill_count} ค่าด้วยวิธี {method}")
        
        return self
    
    def remove_duplicates(self):
        """ลบข้อมูลซ้ำโดยใช้ timestamp เป็น key"""
        initial_len = len(self.data)
        self.data = self.data.drop_duplicates(subset=['timestamp'], keep='first')
        
        removed = initial_len - len(self.data)
        print(f"ลบข้อมูลซ้ำ {removed} แถว")
        
        return self
    
    def sort_by_time(self):
        """เรียงข้อมูลตามเวลา"""
        self.data = self.data.sort_values('timestamp').reset_index(drop=True)
        return self
    
    def get_cleaned_data(self) -> pd.DataFrame:
        return self.data.copy()


ตัวอย่างการใช้งาน
def clean_okx_data(raw_data: pd.DataFrame) -> pd.DataFrame:
    cleaner = OKXDataCleaner(raw_data)
    return (cleaner
            .basic_validation()
            .remove_duplicates()
            .remove_anomalies(price_change_threshold=0.5)
            .handle_missing_values(method='interpolate')
            .sort_by_time()
            .get_cleaned_data())

การคำนวณตัวชี้วัดประสิทธิภาพ

import pandas as pd
import numpy as np

class BacktestMetrics:
    def __init__(self, df: pd.DataFrame, initial_capital: float = 10000):
        """
        df: DataFrame ที่มีคอลัมน์ timestamp, close, volume
        initial_capital: เงินทุนเริ่มต้น (USD)
        """
        self.df = df.copy()
        self.initial_capital = initial_capital
        self.trades = []
        
    def calculate_returns(self):
        """คำนวณผลตอบแทนรายวัน"""
        self.df['daily_return'] = self.df['close'].pct_change()
        self.df['log_return'] = np.log(self.df['close'] / self.df['close'].shift(1))
        return self
    
    def calculate_cumulative_returns(self):
        """คำนวณผลตอบแทนสะสม"""
        self.df['cumulative_return'] = (1 + self.df['daily_return']).cumprod()
        self.df['equity_curve'] = self.initial_capital * self.df['cumulative_return']
        return self
    
    def calculate_sharpe_ratio(self, risk_free_rate: float = 0.02, periods_per_year: int = 365) -> float:
        """คำนวณ Sharpe Ratio"""
        excess_returns = self.df['daily_return'].dropna() - (risk_free_rate / periods_per_year)
        return np.sqrt(periods_per_year) * excess_returns.mean() / excess_returns.std()
    
    def calculate_sortino_ratio(self, risk_free_rate: float = 0.02, periods_per_year: int = 365) -> float:
        """คำนวณ Sortino Ratio (ใช้ downside deviation)"""
        excess_returns = self.df['daily_return'].dropna() - (risk_free_rate / periods_per_year)
        downside_returns = excess_returns[excess_returns < 0]
        
        if len(downside_returns) == 0:
            return np.inf
        
        downside_deviation = downside_returns.std()
        return np.sqrt(periods_per_year) * excess_returns.mean() / downside_deviation
    
    def calculate_max_drawdown(self) -> dict:
        """คำนวณ Maximum Drawdown"""
        equity = self.df['equity_curve']
        running_max = equity.expanding().max()
        drawdown = (equity - running_max) / running_max
        
        max_dd = drawdown.min()
        peak = equity[:drawdown.idxmin()].idxmax()
        trough = drawdown.idxmin()
        
        return {
            'max_drawdown': max_dd,
            'peak_date': peak,
            'trough_date': trough,
            'drawdown_series': drawdown
        }
    
    def calculate_win_rate(self, trades: list) -> float:
        """คำนวณ Win Rate จากรายการเทรด"""
        if not trades:
            return 0.0
        
        winning_trades = sum(1 for t in trades if t.get('pnl', 0) > 0)
        return winning_trades / len(trades)
    
    def calculate_profit_factor(self, trades: list) -> float:
        """คำนวณ Profit Factor"""
        if not trades:
            return 0.0
        
        gross_profit = sum(t.get('pnl', 0) for t in trades if t.get('pnl', 0) > 0)
        gross_loss = abs(sum(t.get('pnl', 0) for t in trades if t.get('pnl', 0) < 0))
        
        if gross_loss == 0:
            return np.inf if gross_profit > 0 else 0.0
        
        return gross_profit / gross_loss
    
    def calculate_calmar_ratio(self, periods_per_year: int = 365) -> float:
        """คำนวณ Calmar Ratio"""
        annual_return = (self.df['cumulative_return'].iloc[-1] - 1) * 100
        max_dd_info = self.calculate_max_drawdown()
        max_dd = abs(max_dd_info['max_drawdown'])
        
        if max_dd == 0:
            return np.inf
        
        return annual_return / max_dd
    
    def get_full_report(self, trades: list = None) -> dict:
        """สร้างรายงานผลตอบแทนแบบครบถ้วน"""
        self.calculate_returns()
        self.calculate_cumulative_returns()
        
        report = {
            'total_return': (self.df['cumulative_return'].iloc[-1] - 1) * 100,
            'total_trades': len(trades) if trades else 0,
            'sharpe_ratio': self.calculate_sharpe_ratio(),
            'sortino_ratio': self.calculate_sortino_ratio(),
            'calmar_ratio': self.calculate_calmar_ratio(),
            'max_drawdown': self.calculate_max_drawdown()['max_drawdown'] * 100,
            'win_rate': self.calculate_win_rate(trades) * 100 if trades else 0,
            'profit_factor': self.calculate_profit_factor(trades),
            'final_equity': self.df['equity_curve'].iloc[-1],
            'volatility': self.df['daily_return'].std() * np.sqrt(365) * 100
        }
        
        return report


ตัวอย่างการใช้งาน
def run_backtest_analysis(df: pd.DataFrame, trades: list, initial_capital: float = 10000):
    metrics = BacktestMetrics(df, initial_capital)
    report = metrics.get_full_report(trades)
    
    print("=" * 50)
    print("รายงานผลตอบแทนจากการทดสอบย้อนกลับ")
    print("=" * 50)
    print(f"ผลตอบแทนรวม: {report['total_return']:.2f}%")
    print(f"จำนวนเทรด: {report['total_trades']}")
    print(f"Sharpe Ratio: {report['sharpe_ratio']:.2f}")
    print(f"Sortino Ratio: {report['sortino_ratio']:.2f}")
    print(f"Calmar Ratio: {report['calmar_ratio']:.2f}")
    print(f"Max Drawdown: {report['max_drawdown']:.2f}%")
    print(f"Win Rate: {report['win_rate']:.2f}%")
    print(f"Profit Factor: {report['profit_factor']:.2f}")
    print(f"Volatility: {report['volatility']:.2f}%")
    print(f"เงินทุนสุดท้าย: ${report['final_equity']:,.2f}")
    
    return report

การประยุกต์ใช้กับ HolySheep AI

สำหรับทีมพัฒนาที่ต้องการสร้างระบบ Backtesting อัตโนมัติ สามารถใช้ HolySheep AI เพื่อประมวลผลข้อมูลและสร้างรายงานได้อย่างรวดเร็ว ด้วย Latency ต่ำกว่า 50ms และราคาที่ประหยัดกว่า 85% เมื่อเทียบกับบริการอื่น

เหมาะกับใคร / ไม่เหมาะกับใคร

เหมาะกับ	ไม่เหมาะกับ
นักเทรดที่ต้องการทดสอบกลยุทธ์ก่อนใช้งานจริง	ผู้ที่ไม่มีพื้นฐานการเขียนโปรแกรม Python
ทีม Quant ที่ต้องการประมวลผลข้อมูลจำนวนมาก	ผู้ที่ต้องการผลลัพธ์แบบ Real-time
นักพัฒนา Bot ที่ต้องการ API ที่เสถียร	ผู้ที่ต้องการ GUI สำเร็จรูปทั้งหมด
ผู้ที่ต้องการประหยัดค่าใช้จ่าย API	ผู้ที่ใช้งาน OpenAI หรือ Anthropic เป็นหลัก

ราคาและ ROI

โมเดล	ราคาต่อ MTok	เทียบกับ OpenAI
GPT-4.1	$8.00	-
Claude Sonnet 4.5	$15.00	-
Gemini 2.5 Flash	$2.50	-
DeepSeek V3.2	$0.42	ประหยัด 85%+

ทำไมต้องเลือก HolySheep

• อัตราแลกเปลี่ยนพิเศษ ¥1 = $1 ประหยัดค่าใช้จ่ายสูงสุด 85%
• รองรับ WeChat และ Alipay สำหรับชำระเงิน
• Latency ต่ำกว่า 50ms เหมาะสำหรับงานที่ต้องการความเร็ว
• รับเครดิตฟรีเมื่อลงทะเบียน
• API ที่เสถียรสำหรับการประมวลผลข้อมูล Backtesting

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

1. ข้อผิดพลาด: ข้อมูลซ้ำทำให้ผลตอบแทนผิดปกติ

# ❌ วิธีผิด: ไม่ลบข้อมูลซ้ำ
df_with_duplicates = raw_data

✅ วิธีถูก: ลบข้อมูลซ้ำก่อนประมวลผล
df_cleaned = raw_data.drop_duplicates(subset=['timestamp'], keep='first')

2. ข้อผิดพลาด: คำนวณ Sharpe Ratio โดยไม่ลบค่าว่าง

# ❌ วิธีผิด: ใช้ข้อมูลที่มี NaN
sharpe = np.sqrt(365) * returns.mean() / returns.std()

✅ วิธีถูก: Drop NaN ก่อนคำนวณ
sharpe = np.sqrt(365) * returns.dropna().mean() / returns.dropna().std()

3. ข้อผิดพลาด: ใช้ Initial Capital ผิดในการคำนวณ Equity

# ❌ วิธีผิด: คำนวณ cumulative return โดยไม่รวม initial_capital
equity_wrong = initial_capital * (1 + returns).cumprod()

✅ วิธีถูก: ใช้ cumulative return ที่ถูกต้อง
df['cumulative_return'] = (1 + returns).cumprod()
df['equity_curve'] = initial_capital * df['cumulative_return']

4. ข้อผิดพลาด: Max Drawdown คำนวณจากราคาแทนที่จะเป็น Equity

# ❌ วิธีผิด: ใช้ close price แทน equity
max_dd_wrong = (close_price / close_price.expanding().max() - 1).min()

✅ วิธีถูก: ใช้ equity curve ที่รวมเงินทุนแล้ว
equity = initial_capital * cumulative_return
running_max = equity.expanding().max()
drawdown = (equity - running_max) / running_max
max_dd = drawdown.min()

ตัวอย่างการใช้งานแบบครบวงจร

# Main Script - การทดสอบย้อนกลับกลยุทธ์ OKX สัญญา Perpetual

import pandas as pd
import numpy as np
from okx_api import get_candlesticks  # สมมติว่ามีฟังก์ชันนี้

def main():
    # 1. ดึงข้อมูล
    raw_data = get_candlesticks(inst_id="BTC-USDT-SWAP", bar="1H", limit=1000)
    
    # 2. ทำความสะอาดข้อมูล
    cleaned_data = clean_okx_data(raw_data)
    
    # 3. กำหนดกลยุทธ์ (ตัวอย่าง: Moving Average Crossover)
    def ma_cross_strategy(df):
        df['ma_short'] = df['close'].rolling(20).mean()
        df['ma_long'] = df['close'].rolling(50).mean()
        df['signal'] = np.where(df['ma_short'] > df['ma_long'], 1, -1)
        return df
    
    data_with_signal = ma_cross_strategy(cleaned_data)
    
    # 4. จำลองการเทรด
    trades = simulate_trades(data_with_signal)
    
    # 5. คำนวณตัวชี้วัด
    report = run_backtest_analysis(data_with_signal, trades, initial_capital=10000)
    
    print("\n✅ การทดสอบย้อนกลับเสร็จสมบูรณ์")

if __name__ == "__main__":
    main()

สรุป

การทำความสะอาดข้อมูลประวัติและการคำนวณตัวชี้วัดอย่างถูกต้องเป็นพื้นฐานสำคัญสำหรับการทดสอบย้อนกลับที่เชื่อถือได้ การใช้ HolySheep AI ช่วยให้การประมวลผลข้อมูลจำนวนมากเป็นไปอย่างรวดเร็วและประหยัดค่าใช้จ่าย

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน