บทนำ: ทำไมต้อง OKX API

ในโลกของการเทรดคริปโตเชิงปริมาณ (Quantitative Trading) ความเร็วและความแม่นยำคือทุกอย่าง การใช้ OKX API ช่วยให้นักพัฒนาสามารถสร้างระบบเทรดอัตโนมัติที่เชื่อมต่อกับตลาดสปอต สัญญาซื้อขายล่วงหน้า (Perpetual Futures) และออปชันได้อย่างไร้รอยต่อ บทความนี้จะพาคุณเจาะลึกการใช้งาน OKX Open API ตั้งแต่พื้นฐานจนถึงการสร้างระบบ Quantitative Trading ที่พร้อมใช้งานจริง สำหรับนักพัฒนาที่ต้องการเพิ่มความสามารถในการประมวลผลข้อมูลตลาดด้วย AI คุณสามารถใช้ HolySheep AI ซึ่งรองรับโมเดลภาษาขนาดใหญ่หลากหลาย เช่น GPT-4.1, Claude Sonnet 4.5 และ Gemini 2.5 Flash ในราคาที่ประหยัดถึง 85% พร้อมความเร็วตอบสนองต่ำกว่า 50 มิลลิวินาที

การตั้งค่า OKX API Key และ Environment

ก่อนเริ่มต้นเขียนโค้ด คุณต้องสร้าง API Key จากบัญชี OKX ของคุณก่อน ทำตามขั้นตอนดังนี้:
  1. เข้าสู่ระบบ OKX และไปที่ Settings > API
  2. คลิก "Create V5 API Key"
  3. ตั้งค่าสิทธิ์: Read Only, Trade, Transfer ตามความต้องการ
  4. บันทึก API Key, Secret Key และ Passphrase อย่างปลอดภัย
# ติดตั้ง library ที่จำเป็น
pip install okx-sdk requests hmac hashlib time


config.py - การตั้งค่า OKX API

import os
from datetime import datetime
from typing import Dict

class OKXConfig:
    # ข้อมูล API ของคุณ (เก็บเป็น Environment Variable)
    API_KEY = os.getenv('OKX_API_KEY', 'your_api_key_here')
    SECRET_KEY = os.getenv('OKX_SECRET_KEY', 'your_secret_key_here')
    PASSPHRASE = os.getenv('OKX_PASSPHRASE', 'your_passphrase_here')
    
    # Endpoint สำหรับ Production
    BASE_URL = 'https://www.okx.com'
    
    # สำหรับ Demo Trading ใช้ Endpoint นี้แทน
    # BASE_URL = 'https://www.okx.com/v3'

    @staticmethod
    def get_timestamp() -> str:
        return datetime.utcnow().isoformat() + 'Z'

print("OKX Configuration พร้อมใช้งาน")

ระบบ Authentication สำหรับ OKX API

OKX ใช้ระบบ HMAC SHA256 สำหรับการยืนยันตัวตนทุก Request การเซ็นอินที่ถูกต้องเป็นสิ่งจำเป็นมาก เพราะ Request ที่ไม่ผ่านการยืนยันจะถูกปฏิเสธทันที 
# okx_auth.py - ระบบ Authentication สำหรับ OKX
import hmac
import base64
import hashlib
import time
from typing import Dict, Tuple

class OKXAuthenticator:
    def __init__(self, api_key: str, secret_key: str, passphrase: str):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase

    def sign(self, message: str) -> str:
        """สร้าง HMAC SHA256 signature"""
        mac = hmac.new(
            self.secret_key.encode('utf-8'),
            message.encode('utf-8'),
            hashlib.sha256
        )
        return base64.b64encode(mac.digest()).decode('utf-8')

    def get_headers(self, method: str, path: str, body: str = '') -> Dict[str, str]:
        """สร้าง Headers สำหรับ Request"""
        timestamp = time.strftime('%Y-%m-%dT%H:%M:%S.000Z', time.gmtime())
        
        # สำหรับ GET request body จะว่างเปล่า
        message = timestamp + method + path + body
        
        signature = self.sign(message)
        
        return {
            'OK-ACCESS-KEY': self.api_key,
            'OK-ACCESS-SIGN': signature,
            'OK-ACCESS-TIMESTAMP': timestamp,
            'OK-ACCESS-PASSPHRASE': self.passphrase,
            'Content-Type': 'application/json'
        }


ทดสอบ Authentication

auth = OKXAuthenticator(
    api_key='demo_api_key',
    secret_key='demo_secret_key', 
    passphrase='demo_passphrase'
)
print("Authentication module พร้อมทำงาน")

การเทรดสปอต (Spot Trading) ผ่าน API

การเทรดสปอตเป็นจุดเริ่มต้นที่ดีสำหรับผู้ที่ต้องการเรียนรู้การใช้งาน OKX API เริ่มจากการดึงข้อมูลราคาตลาด ส่งคำสั่งซื้อขาย และจัดการ Portfolio 
# spot_trading.py - ระบบเทรดสปอต
import requests
import json
from okx_auth import OKXAuthenticator

class OKXSpotTrader:
    def __init__(self, authenticator: OKXAuthenticator, base_url: str):
        self.auth = authenticator
        self.base_url = base_url

    def get_ticker(self, inst_id: str = 'BTC-USDT') -> dict:
        """ดึงข้อมูล Ticker ของคู่เทรด"""
        path = f'/api/v5/market/ticker?instId={inst_id}'
        headers = self.auth.get_headers('GET', path)
        
        response = requests.get(self.base_url + path, headers=headers)
        return response.json()

    def get_orderbook(self, inst_id: str = 'BTC-USDT', sz: int = 10) -> dict:
        """ดึงข้อมูล Order Book"""
        path = f'/api/v5/market/books?instId={inst_id}&sz={sz}'
        headers = self.auth.get_headers('GET', path)
        
        response = requests.get(self.base_url + path, headers=headers)
        return response.json()

    def place_order(self, inst_id: str, side: str, ord_type: str, 
                    sz: str, px: str = '') -> dict:
        """
        วางคำสั่งซื้อขาย
        - inst_id: คู่เทรด เช่น BTC-USDT
        - side: buy หรือ sell
        - ord_type: market หรือ limit
        - sz: จำนวน
        - px: ราคา (สำหรับ limit order)
        """
        path = '/api/v5/trade/order'
        body = {
            'instId': inst_id,
            'tdMode': 'cash',  # สำหรับ Spot ใช้ cash
            'side': side,
            'ordType': ord_type,
            'sz': sz,
        }
        if px:
            body['px'] = px
            
        body_json = json.dumps(body)
        headers = self.auth.get_headers('POST', path, body_json)
        
        response = requests.post(
            self.base_url + path, 
            headers=headers, 
            data=body_json
        )
        return response.json()

    def get_account_balance(self) -> dict:
        """ดึงยอดคงเหลือในบัญชี"""
        path = '/api/v5/account/balance'
        headers = self.auth.get_headers('GET', path)
        
        response = requests.get(self.base_url + path, headers=headers)
        return response.json()


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


trader = OKXSpotTrader(auth, 'https://www.okx.com')


btc_ticker = trader.get_ticker('BTC-USDT')


print(f"BTC Price: ${btc_ticker['data'][0]['last']}")

ระบบเทรดสัญญาซื้อขายล่วงหน้า (Perpetual Futures)

สัญญาซื้อขายล่วงหน้า (Perpetual Swaps) เป็นผลิตภัณฑ์ที่ได้รับความนิยมมากในการเทรดเชิงปริมาณ เนื่องจากมีเลเวอเรจสูงและไม่มีวันหมดอายุ ในส่วนนี้เราจะสร้างระบบสำหรับเทรด Perpetual Futures พร้อมฟีเจอร์ Position Management และ Funding Rate Monitoring 
# perpetual_trading.py - ระบบเทรด Perpetual Futures
import requests
import json
from datetime import datetime

class OKXPerpetualTrader:
    def __init__(self, authenticator, base_url: str):
        self.auth = authenticator
        self.base_url = base_url

    def get_funding_rate(self, inst_id: str = 'BTC-USDT-SWAP') -> dict:
        """ดึงข้อมูล Funding Rate ปัจจุบัน"""
        path = f'/api/v5/public/funding-rate?instId={inst_id}'
        headers = self.auth.get_headers('GET', path)
        
        response = requests.get(self.base_url + path, headers=headers)
        data = response.json()
        
        if data.get('data'):
            funding_info = data['data'][0]
            next_funding_time = int(funding_info.get('nextFundingTime', 0))
            return {
                'funding_rate': float(funding_info['fundingRate']),
                'next_funding_time': datetime.fromtimestamp(next_funding_time/1000),
                'inst_id': inst_id
            }
        return None

    def set_leverage(self, inst_id: str, lever: int = 10, 
                     mgn_mode: str = 'cross') -> dict:
        """
        ตั้งค่า Leverage
        - inst_id: คู่เทรด เช่น BTC-USDT-SWAP
        - lever: ระดับเลเวอเรจ เช่น 10x, 20x
        - mgn_mode: cross (Cross Margin) หรือ isolated (Isolated Margin)
        """
        path = '/api/v5/account/set-leverage'
        body = {
            'instId': inst_id,
            'lever': str(lever),
            'mgnMode': mgn_mode
        }
        body_json = json.dumps(body)
        headers = self.auth.get_headers('POST', path, body_json)
        
        response = requests.post(
            self.base_url + path, 
            headers=headers, 
            data=body_json
        )
        return response.json()

    def open_position(self, inst_id: str, side: str, sz: str, 
                      ord_type: str = 'market', px: str = '') -> dict:
        """
        เปิดสถานะ (Position)
        - side: long (ซื้อ) หรือ short (ขาย)
        """
        path = '/api/v5/trade/order'
        body = {
            'instId': inst_id,
            'tdMode': 'cross',  # Cross Margin สำหรับ Perpetual
            'side': side,
            'ordType': ord_type,
            'sz': sz,
            'posSide': side  # กำหนด side ของ position
        }
        if px:
            body['px'] = px
            
        body_json = json.dumps(body)
        headers = self.auth.get_headers('POST', path, body_json)
        
        response = requests.post(
            self.base_url + path, 
            headers=headers, 
            data=body_json
        )
        return response.json()

    def get_positions(self, inst_family: str = 'BTC') -> dict:
        """ดึงข้อมูล Positions ทั้งหมด"""
        path = f'/api/v5/account/positions?instFamily={inst_family}'
        headers = self.auth.get_headers('GET', path)
        
        response = requests.get(self.base_url + path, headers=headers)
        return response.json()


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


perpetual_trader = OKXPerpetualTrader(auth, 'https://www.okx.com')


funding_info = perpetual_trader.get_funding_rate('BTC-USDT-SWAP')


print(f"Funding Rate: {funding_info['funding_rate']:.4%}")

การเทรดออปชัน (Options Trading)

ออปชันเป็นผลิตภัณฑ์ที่ซับซ้อนกว่าสปอตและ Futures แต่ให้โอกาสในการทำกำไรในหลายรูปแบบ OKX มีออปชันหลากหลายสำหรับ BTC และ ETH 
# options_trading.py - ระบบเทรดออปชัน
import requests
import json
from datetime import datetime, timedelta

class OKXOptionsTrader:
    def __init__(self, authenticator, base_url: str):
        self.auth = authenticator
        self.base_url = base_url

    def get_option_instruments(self, underlying: str = 'BTC-USD', 
                               exp_date: str = None) -> dict:
        """
        ดึงรายการสัญญาออปชันที่มีทั้งหมด
        - underlying: สินทรัพย์อ้างอิง
        """
        path = f'/api/v5/public/instruments?instType=OPTION&underlyingInstId={underlying}'
        if exp_date:
            path += f'&exp={exp_date}'
            
        headers = self.auth.get_headers('GET', path)
        response = requests.get(self.base_url + path, headers=headers)
        return response.json()

    def get_option_ticker(self, inst_id: str) -> dict:
        """ดึงข้อมูล Ticker ของออปชัน"""
        path = f'/api/v5/market/ticker?instId={inst_id}'
        headers = self.auth.get_headers('GET', path)
        response = requests.get(self.base_url + path, headers=headers)
        return response.json()

    def get_index_ticker(self, inst_id: str = 'BTC-USD') -> dict:
        """ดึงราคา Index ของสินทรัพย์อ้างอิง"""
        path = f'/api/v5/market/ticker?instId={inst_id}'
        headers = self.auth.get_headers('GET', path)
        response = requests.get(self.base_url + path, headers=headers)
        return response.json()

    def buy_call_option(self, inst_id: str, sz: str, 
                        ord_type: str = 'market', px: str = '') -> dict:
        """ซื้อ Call Option"""
        path = '/api/v5/trade/order'
        body = {
            'instId': inst_id,
            'tdMode': 'cash',
            'side': 'buy',
            'ordType': ord_type,
            'sz': sz,
            'clOrdId': f'call_{int(datetime.now().timestamp())}'  # Client Order ID
        }
        if px:
            body['px'] = px
            
        body_json = json.dumps(body)
        headers = self.auth.get_headers('POST', path, body_json)
        
        response = requests.post(
            self.base_url + path, 
            headers=headers, 
            data=body_json
        )
        return response.json()

    def place_take_profit_order(self, inst_id: str, sz: str, 
                                 trigger_px: str, side: str = 'sell') -> dict:
        """วางคำสั่ง Take Profit แบบ-algo"""
        path = '/api/v5/trade/order-algo'
        body = {
            'instId': inst_id,
            'tdMode': 'cash',
            'side': side,
            'ordType': 'conditional',  # Trigger order
            'sz': sz,
            'triggerPx': trigger_px,  # ราคาที่จะ trigger
            'orderPx': '-1'  # Market price เมื่อ trigger
        }
        body_json = json.dumps(body)
        headers = self.auth.get_headers('POST', path, body_json)
        
        response = requests.post(
            self.base_url + path, 
            headers=headers, 
            data=body_json
        )
        return response.json()


ตัวอย่าง: ดึงรายการออปชัน BTC ที่หมดอายุใน 7 วัน


options_trader = OKXOptionsTrader(auth, 'https://www.okx.com')


exp_date = (datetime.now() + timedelta(days=7)).strftime('%Y%m%d')


options = options_trader.get_option_instruments('BTC-USD', exp_date)

สร้างระบบ Quantitative Trading สมบูรณ์

ตอนนี้เรามาสร้างระบบ Quantitative Trading ที่ครบวงจรโดยรวมทุกส่วนเข้าด้วยกัน ระบบนี้จะมี Event-driven Architecture และสามารถประมวลผลสัญญาณตลาดแบบ Real-time 
# quantitative_trading_system.py - ระบบ Quantitative Trading สมบูรณ์
import asyncio
import json
import logging
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('QuantTradingSystem')

class SignalType(Enum):
    MACD_CROSSOVER = "macd_crossover"
    RSI_OVERSOLD = "rsi_oversold"
    RSI_OVERBOUGHT = "rsi_overbought"
    BOLLINGER_BREAK = "bollinger_break"
    VOLUME_SPIKE = "volume_spike"

@dataclass
class TradingSignal:
    symbol: str
    signal_type: SignalType
    strength: float  # 0.0 - 1.0
    price: float
    timestamp: datetime
    indicators: Dict

@dataclass
class Position:
    inst_id: str
    side: str
    size: float
    entry_price: float
    current_pnl: float = 0.0
    stop_loss: float = None
    take_profit: float = None

class QuantitativeTradingSystem:
    def __init__(self, authenticator, base_url: str):
        self.spot_trader = OKXSpotTrader(authenticator, base_url)
        self.perpetual_trader = OKXPerpetualTrader(authenticator, base_url)
        self.options_trader = OKXOptionsTrader(authenticator, base_url)
        self.positions: List[Position] = []
        self.signal_history: List[TradingSignal] = []
        
    def calculate_rsi(self, prices: List[float], period: int = 14) -> float:
        """คำนวณ RSI Indicator"""
        if len(prices) < period + 1:
            return 50.0
            
        deltas = [prices[i] - prices[i-1] for i in range(1, len(prices))]
        gains = [d if d > 0 else 0 for d in deltas[-period:]]
        losses = [-d if d < 0 else 0 for d in deltas[-period:]]
        
        avg_gain = sum(gains) / period
        avg_loss = sum(losses) / period
        
        if avg_loss == 0:
            return 100.0
        rs = avg_gain / avg_loss
        return 100 - (100 / (1 + rs))

    def calculate_macd(self, prices: List[float]) -> Dict:
        """คำนวณ MACD Indicator"""
        if len(prices) < 26:
            return {'macd': 0, 'signal': 0, 'histogram': 0}
            
        # EMA calculations
        ema12 = self._ema(prices, 12)
        ema26 = self._ema(prices, 26)
        macd_line = ema12 - ema26
        signal_line = self._ema([macd_line] * 9, 9)  # Simplified
        histogram = macd_line - signal_line
        
        return {'macd': macd_line, 'signal': signal_line, 'histogram': histogram}

    def _ema(self, prices: List[float], period: int) -> float:
        """คำนวณ Exponential Moving Average"""
        if len(prices) < period:
            return sum(prices) / len(prices)
        multiplier = 2 / (period + 1)
        ema = sum(prices[:period]) / period
        for price in prices[period:]:
            ema = (price - ema) * multiplier + ema
        return ema

    async def analyze_market(self, symbol: str = 'BTC-USDT') -> TradingSignal:
        """วิเคราะห์ตลาดและสร้างสัญญาณ"""
        # ดึงข้อมูลราคา
        ticker = self.spot_trader.get_ticker(symbol)
        
        if not ticker.get('data'):
            logger.warning(f"ไม่สามารถดึงข้อมูล {symbol}")
            return None
            
        current_price = float(ticker['data'][0]['last'])
        
        # ดึง OHLCV data (ใช้ candles endpoint)
        # สร้าง fake price history สำหรับ demo
        price_history = [current_price * (1 + 0.01 * (i % 5 - 2)) 
                        for i in range(30)]
        
        # คำนวณ Indicators
        rsi = self.calculate_rsi(price_history)
        macd = self.calculate_macd(price_history)
        
        # ตรวจจับสัญญาณ
        signal_type = None
        strength = 0.5
        
        if rsi < 30:
            signal_type = SignalType.RSI_OVERSOLD
            strength = (30 - rsi) / 30
        elif rsi > 70:
            signal_type = SignalType.RSI_OVERBOUGHT
            strength = (rsi - 70) / 30
        
        if macd['histogram'] > 0 and signal_type is None:
            signal_type = SignalType.MACD_CROSSOVER
            strength = min(abs(macd['histogram']) / current_price * 100, 1.0)
        
        signal = TradingSignal(
            symbol=symbol,
            signal_type=signal_type or SignalType.MACD_CROSSOVER,
            strength=strength,
            price=current_price,
            timestamp=datetime.now(),
            indicators={'rsi': rsi, 'macd': macd}
        )
        
        self.signal_history.append(signal)
        return signal

    async def execute_strategy(self, signal: TradingSignal, 
                               position_size: float = 0.01):
        """ดำเนินการตามสัญญาณ"""
        if signal.strength < 0.6:  # ความแรงขั้นต่ำ
            logger.info(f"สัญญาณไม่แรงพอ: {signal.strength}")
            return
            
        if signal.signal_type == SignalType.RSI_OVERSOLD:
            logger.info(f"🟢 สัญญาณซื้อ RSI Oversold ที่ {signal.price}")
            result = self.spot_trader.place_order(
                inst_id=signal.symbol,
                side='buy',
                ord_type='market',
                sz=str(position_size)
            )
            logger.info(f"วางคำสั่งซื้อ: {result}")
            
        elif signal.signal_type == SignalType.RSI_OVERBOUGHT:
            logger.info(f"🔴 สัญญาณขาย RSI Overbought ที่ {signal.price}")

    async def run_trading_loop(self, symbol: str = 'BTC-USDT', 
                               interval: int = 60):
        """รันระบบเทรดแบบ Loop"""
        logger.info(f"เริ่มระบบ Quantitative Trading สำหรับ {symbol}")
        
        while True:
            try:
                signal = await self.analyze_market(symbol)
                if signal:
                    await self.execute_strategy(signal)
                    
                await asyncio.sleep(interval)
                
            except Exception as e:
                logger.error(f"เกิดข้อผิดพลาด: {e}")
                await asyncio.sleep(5)


การใช้งาน


system = QuantitativeTradingSystem(auth, 'https://www.okx.com')


asyncio.run(system.run_trading_loop('BTC-USDT', interval=60))

เปรียบเทียบผลิตภัณฑ์ OKX สำหรับ Quantitative Trading

ผลิตภัณฑ์ Leverage ความเสี่ยง เหมาะกับ API Complexity ค่าธรรมเนียม (Maker/Taker)
สปอต (Spot) 1x เท่านั้น

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

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

🔥 ลอง HolySheep AI

เกตเวย์ AI API โดยตรง รองรับ Claude, GPT-5, Gemini, DeepSeek — หนึ่งคีย์ ไม่ต้อง VPN

👉 สมัครฟรี →

© 2026 HolySheep AI · บทช่วยสอนเพิ่มเติม