วิธีการรวบรวมและประมวลผลข้อมูล OKX Candle สำหรับ Trading Bot

บทนำ: ทำไมต้องเรียนรู้ OKX Candle Data Aggregation

การดึงข้อมูล OHLCV (Open-High-Low-Close-Volume) จาก OKX คือหัวใจสำคัญของการสร้างระบบเทรดอัตโนมัติ ไม่ว่าจะเป็น Grid Trading, DCA Bot, หรือ AI Signal Generator บทความนี้จะสอนวิธีการดึงข้อมูล candle จาก OKX API, วิธีการ aggregate ข้อมูลหลาย timeframe, และเทคนิคการ optimize สำหรับ Bot ที่ทำงานเร็วและเสถียร สำหรับนักพัฒนาที่ต้องการสร้าง AI-powered trading analysis คุณสามารถใช้ HolySheep AI เพื่อวิเคราะห์ข้อมูล candle ที่รวบรวมได้ — รองรับ GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash และ DeepSeek V3.2 ในราคาที่ประหยัดกว่า 85% เมื่อเทียบกับ OpenAI โดยตรง

พื้นฐาน OKX Candle API

OKX ให้บริการ Candle REST API ผ่าน endpoint หลัก:

GET https://www.okx.com/api/v5/market/history-candles?instId=BTC-USDT&bar=1H&after=1704067200000&before=1704153600000

พารามิเตอร์สำคัญ:

• instId: คู่เทรด เช่น BTC-USDT, ETH-USDT-SWAP
• bar: Timeframe (1m, 5m, 15m, 1H, 4H, 1D, 1W)
• after/before: Unix timestamp มิลลิวินาที (เรียงจากใหม่ไปเก่า)
• limit: จำนวน candle สูงสุด 100 ต่อ request

ข้อมูล candle ที่ได้กลับมามี format:

{
  "code": "0",
  "data": [
    ["1704067200000", "42150.5", "42300.0", "42080.3", "42200.8", "1250.5", "52890123.5"],
    // [timestamp, open, high, low, close, volume, quoteVolume]
  ],
  "msg": ""
}

วิธีการที่ 1: Sequential Aggregation

วิธีที่ง่ายที่สุดคือดึงข้อมูลหลาย timeframe แล้วรวมเข้าด้วยกัน:

import requests
from datetime import datetime, timedelta

def fetch_okx_candles(inst_id: str, bar: str, start_ts: int, end_ts: int):
    """ดึงข้อมูล candle จาก OKX API แบบ sequential"""
    base_url = "https://www.okx.com/api/v5/market/history-candles"
    all_candles = []
    current_after = end_ts  # เริ่มจาก timestamp ล่าสุด
    
    while current_after > start_ts:
        params = {
            "instId": inst_id,
            "bar": bar,
            "after": current_after,
            "before": start_ts,
            "limit": 100
        }
        
        response = requests.get(base_url, params=params, timeout=10)
        data = response.json()
        
        if data["code"] != "0":
            raise Exception(f"OKX API Error: {data['msg']}")
        
        candles = data["data"]
        if not candles:
            break
            
        all_candles.extend(candles)
        # ปรับ timestamp เพื่อดึง batch ถัดไป
        current_after = int(candles[-1][0]) - 1
        
    return all_candles

ตัวอย่างการใช้งาน
start = int((datetime.now() - timedelta(days=30)).timestamp() * 1000)
end = int(datetime.now().timestamp() * 1000)

candles_1h = fetch_okx_candles("BTC-USDT", "1H", start, end)
print(f"ได้ข้อมูล {len(candles_1h)} candles")

วิธีการที่ 2: Parallel Aggregation พร้อม Rate Limiting

สำหรับ Bot ที่ต้องการความเร็ว ควรใช้ concurrent requests:

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

async def fetch_candles_async(session, inst_id: str, bar: str, after: str):
    """ดึงข้อมูล async พร้อม handle rate limit"""
    url = "https://www.okx.com/api/v5/market/history-candles"
    params = {"instId": inst_id, "bar": bar, "after": after, "limit": 100}
    
    async with session.get(url, params=params) as response:
        if response.status == 429:
            await asyncio.sleep(1)  # Wait on rate limit
            return await fetch_candles_async(session, inst_id, bar, after)
        
        data = await response.json()
        if data["code"] != "0":
            return []
        return data["data"]

async def aggregate_multiple_timeframes(inst_id: str, after: str):
    """รวบรวมข้อมูลหลาย timeframe พร้อมกัน"""
    timeframes = ["1m", "5m", "15m", "1H", "4H", "1D"]
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_candles_async(session, inst_id, tf, after)
            for tf in timeframes
        ]
        results = await asyncio.gather(*tasks)
        
    return dict(zip(timeframes, results))

ใช้งาน
results = asyncio.run(aggregate_multiple_timeframes("BTC-USDT", str(int(datetime.now().timestamp() * 1000))))
for tf, candles in results.items():
    print(f"{tf}: {len(candles)} candles")

วิธีการที่ 3: WebSocket Real-time Aggregation

สำหรับ Bot ที่ต้องการข้อมูล real-time:

import websocket
import json
import time

class OKXCandleAggregator:
    def __init__(self, inst_id: str, timeframes: list):
        self.inst_id = inst_id
        self.timeframes = timeframes
        self.candles = {tf: {} for tf in timeframes}
        self.ws = None
        
    def on_message(self, ws, message):
        data = json.loads(message)
        if data.get("arg", {}).get("channel") != "candles":
            return
            
        candle_data = data["data"][0]
        ts = int(candle_data[0])
        o, h, l, c, vol = candle_data[1:6]
        
        # อัพเดท candle ปัจจุบัน
        for tf in self.timeframes:
            tf_ts = self._get_timeframe_start(ts, tf)
            if tf_ts not in self.candles[tf]:
                self.candles[tf][tf_ts] = {
                    "open": float(o), "high": float(h),
                    "low": float(l), "close": float(c),
                    "volume": float(vol), "count": 1
                }
            else:
                self.candles[tf][tf_ts]["high"] = max(self.candles[tf][tf_ts]["high"], float(h))
                self.candles[tf][tf_ts]["low"] = min(self.candles[tf][tf_ts]["low"], float(l))
                self.candles[tf][tf_ts]["close"] = float(c)
                self.candles[tf][tf_ts]["volume"] += float(vol)
                
    def _get_timeframe_start(self, ts: int, tf: str) -> int:
        intervals = {"1m": 60000, "5m": 300000, "15m": 900000, "1H": 3600000}
        interval = intervals.get(tf, 3600000)
        return (ts // interval) * interval
        
    def start(self):
        ws_url = "wss://ws.okx.com:8443/ws/v5/public"
        self.ws = websocket.WebSocketApp(
            ws_url,
            on_message=self.on_message
        )
        
        for tf in self.timeframes:
            subscribe_msg = {
                "op": "subscribe",
                "args": [{"channel": "candles", "instId": self.inst_id, "bar": tf}]
            }
            self.ws.send(json.dumps(subscribe_msg))
            
        self.ws.run_forever(ping_interval=30)

ใช้งาน
aggregator = OKXCandleAggregator("BTC-USDT", ["1m", "5m", "15m"])
aggregator.start()

เชื่อมต่อ OKX Data กับ AI Analysis ผ่าน HolySheep

หลังจากรวบรวมข้อมูล candle แล้ว ขั้นตอนต่อไปคือการวิเคราะห์ด้วย AI เพื่อหา patterns หรือสร้างสัญญาณเทรด:

import requests
import json

def analyze_candles_with_ai(candles: list, api_key: str):
    """ส่งข้อมูล candle ให้ AI วิเคราะห์ผ่าน HolySheep"""
    
    # แปลงข้อมูล candle เป็น text summary
    latest_20 = candles[:20]
    summary = []
    for c in latest_20:
        ts = datetime.fromtimestamp(int(c[0])/1000).strftime("%Y-%m-%d %H:%M")
        summary.append(f"{ts} | O:{c[1]} H:{c[2]} L:{c[3]} C:{c[4]} V:{c[5]}")
    
    prompt = f"""Analyze these BTC-USDT candles and provide trading insights:
    
{chr(10).join(summary)}

Respond in Thai with:
1. Trend direction (up/down/sideways)
2. Key support/resistance levels
3. Volume analysis
4. Brief trading recommendation"""

    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-4.1",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3,
            "max_tokens": 500
        },
        timeout=30
    )
    
    return response.json()["choices"][0]["message"]["content"]

ใช้งาน — ราคาเพียง $8/MTok (GPT-4.1)
result = analyze_candles_with_ai(candles_1h, "YOUR_HOLYSHEEP_API_KEY")
print(result)

ทำไมต้องเลือก HolySheep สำหรับ Trading AI:

• ราคาถูกกว่า 85%: GPT-4.1 เพียง $8/MTok เทียบกับ OpenAI $60/MTok
• DeepSeek V3.2: $0.42/MTok สำหรับงานวิเคราะห์ที่ต้องการ volume สูง
• Latency ต่ำกว่า 50ms: เหมาะสำหรับ real-time trading signals
• รองรับ WeChat/Alipay: ซื้อเครดิตได้ง่ายสำหรับผู้ใช้ในไทย
• เครดิตฟรีเมื่อลงทะเบียน: ทดลองใช้งานก่อนตัดสินใจ

การ Optimize สำหรับ Production Trading Bot

1. Caching Strategy

import redis
import json
from functools import wraps

def cache_candles(ttl_seconds: int = 300):
    """Cache ข้อมูล candle ใน Redis"""
    r = redis.Redis(host='localhost', port=6379, db=0)
    
    def decorator(func):
        @wraps(func)
        def wrapper(inst_id, bar, *args, **kwargs):
            cache_key = f"okx:{inst_id}:{bar}"
            
            # ลองดึงจาก cache
            cached = r.get(cache_key)
            if cached:
                return json.loads(cached)
            
            # ถ้าไม่มี ดึงจาก API
            result = func(inst_id, bar, *args, **kwargs)
            r.setex(cache_key, ttl_seconds, json.dumps(result))
            
            return result
        return wrapper
    return decorator

@cache_candles(ttl_seconds=60)
def get_cached_candles(inst_id: str, bar: str):
    # ดึงจาก OKX API
    return fetch_okx_candles(inst_id, bar, start, end)

2. Database Schema สำหรับ Historical Data

CREATE TABLE o candles (
    id BIGSERIAL PRIMARY KEY,
    inst_id VARCHAR(20) NOT NULL,
    timeframe VARCHAR(5) NOT NULL,
    open_time TIMESTAMP NOT NULL,
    open_price DECIMAL(18,8),
    high_price DECIMAL(18,8),
    low_price DECIMAL(18,8),
    close_price DECIMAL(18,8),
    volume DECIMAL(18,8),
    quote_volume DECIMAL(18,8),
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(inst_id, timeframe, open_time)
);

CREATE INDEX idx_candles_lookup ON o candles(inst_id, timeframe, open_time DESC);
CREATE INDEX idx_candles_volume ON o candles(inst_id, timeframe, volume DESC);

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

กรณีที่ 1: Rate Limit 429 Error

อาการ: ได้รับ response 429 หลังจากดึงข้อมูลไปสักพัก สาเหตุ: OKX จำกัด requests ที่ 20 requests/second ต่อ IP วิธีแก้ไข:

import time
from collections import deque

class RateLimiter:
    """จำกัด request rate อัตโนมัติ"""
    def __init__(self, max_calls: int = 20, per_seconds: int = 1):
        self.max_calls = max_calls
        self.per_seconds = per_seconds
        self.calls = deque()
        
    def wait_if_needed(self):
        now = time.time()
        # ลบ requests เก่าที่หมดอายุ
        while self.calls and self.calls[0] < now - self.per_seconds:
            self.calls.popleft()
            
        if len(self.calls) >= self.max_calls:
            sleep_time = self.calls[0] + self.per_seconds - now
            time.sleep(max(0, sleep_time))
            self.calls.popleft()
            
        self.calls.append(now)

ใช้งาน
limiter = RateLimiter(max_calls=20, per_seconds=1)

def throttled_fetch(url, params):
    limiter.wait_if_needed()
    return requests.get(url, params=params, timeout=10)

กรณีที่ 2: Timestamp Offset Error

อาการ: ข้อมูล candle ที่ได้ไม่ตรงกับที่คาดหวัง เช่น candle 1 ชั่วโมงมี timeframe ไม่ถูกต้อง สาเหตุ: OKX API ใช้ timestamp มิลลิวินาที แต่บางครั้ง timezone ของ server ต่างกัน วิธีแก้ไข:

from datetime import datetime, timezone

def normalize_timestamp(ts) -> int:
    """แปลง timestamp ให้เป็น milliseconds UTC"""
    if isinstance(ts, str):
        ts = int(ts)
    
    # ถ้าเป็น seconds ให้คูณ 1000
    if ts < 1_000_000_000_000:
        ts = ts * 1000
        
    # แปลงเป็น UTC datetime
    dt = datetime.fromtimestamp(ts / 1000, tz=timezone.utc)
    
    # สร้าง timestamp ใหม่ที่ align กับ timeframe
    return int(dt.timestamp() * 1000)

def align_to_timeframe(ts: int, timeframe: str) -> int:
    """จัด timestamp ให้ตรงกับขอบเขต timeframe"""
    intervals = {
        "1m": 60_000,
        "5m": 300_000,
        "15m": 900_000,
        "1H": 3_600_000,
        "4H": 14_400_000,
        "1D": 86_400_000
    }
    interval = intervals.get(timeframe, 3_600_000)
    return (ts // interval) * interval

ทดสอบ
ts = "1704067200"  # string seconds
normalized = normalize_timestamp(ts)
aligned = align_to_timeframe(normalized, "1H")
print(f"Normalized: {datetime.fromtimestamp(normalized/1000, tz=timezone.utc)}")
print(f"Aligned to 1H: {datetime.fromtimestamp(aligned/1000, tz=timezone.utc)}")

กรณีที่ 3: Missing Candles / Data Gaps

อาการ: ข้อมูลที่ได้มีช่วงหายไปบางช่วง ไม่ต่อเนื่อง สาเหตุ: OKX อาจไม่มีข้อมูลบางช่วง (maintenance) หรือดึงข้อมูลผิด direction วิธีแก้ไข:

from typing import List, Optional

def fill_candle_gaps(candles: List[List], timeframe: str) -> List[List]:
    """เติมช่องว่างในข้อมูล candle"""
    intervals = {
        "1m": 60_000,
        "5m": 300_000,
        "15m": 900_000,
        "1H": 3_600_000,
        "4H": 14_400_000,
        "1D": 86_400_000
    }
    interval = intervals.get(timeframe, 3_600_000)
    
    if len(candles) < 2:
        return candles
        
    # Sort ตาม timestamp
    sorted_candles = sorted(candles, key=lambda x: int(x[0]))
    
    filled = []
    for i in range(len(sorted_candles) - 1):
        filled.append(sorted_candles[i])
        
        curr_ts = int(sorted_candles[i][0])
        next_ts = int(sorted_candles[i + 1][0])
        expected_diff = next_ts - curr_ts
        
        # ถ้าขาด interval มากกว่า 1 timeframe
        if expected_diff > interval:
            missing_count = expected_diff // interval - 1
            for j in range(int(missing_count)):
                gap_ts = curr_ts + interval * (j + 1)
                # ใช้ close ก่อนหน้าเป็น placeholder
                placeholder = [
                    str(gap_ts),
                    sorted_candles[i][4],  # use prev close as open
                    sorted_candles[i][4],  # high = prev close
                    sorted_candles[i][4],  # low = prev close
                    sorted_candles[i][4],  # close = prev close
                    "0",                   # zero volume
                    "0"
                ]
                filled.append(placeholder)
                
    filled.append(sorted_candles[-1])
    return filled

ตรวจสอบก่อนใช้งาน
original = fetch_okx_candles("BTC-USDT", "1H", start, end)
complete = fill_candle_gaps(original, "1H")
print(f"ก่อนเติม: {len(original)} | หลังเติม: {len(complete)}")

สรุป

การรวบรวมข้อมูล OKX candle อย่างมีประสิทธิภาพต้องอาศัย:

• การเลือกใช้ API ที่เหมาะสม: REST สำหรับ historical, WebSocket สำหรับ real-time
• การจัดการ Rate Limit: Implement throttling เพื่อไม่ให้โดน block
• Caching และ Database: เก็บข้อมูลที่ได้มาใช้ซ้ำ ลด API calls
• Error Handling ที่ดี: จัดการ edge cases ต่างๆ อย่างครบถ้วน

สำหรับนักพัฒนาที่ต้องการเชื่อมต่อข้อมูล candle กับ AI เพื่อสร้าง trading signals หรือ market analysis — HolySheep AI คือทางเลือกที่คุ้มค่าที่สุด ด้วยราคาเริ่มต้นเพียง $0.42/MTok (DeepSeek V3.2) และ latency ต่ำกว่า 50ms เหมาะสำหรับการทำ real-time analysis 👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน