บทนำ: ทำไมต้องใช้ Bybit Perpetual Futures API

สำหรับนักพัฒนาที่สร้างระบบเทรดอัตโนมัติหรือแอปพลิเคชันที่ต้องการข้อมูลราคาคริปโตแบบเรียลไทม์ Bybit Perpetual Futures API เป็นเครื่องมือที่ทรงพลังและใช้งานง่าย ในบทความนี้เราจะสอนวิธีดึงข้อมูลตลาดแบบเรียลไทม์ทั้งแบบ REST API และ WebSocket พร้อมตัวอย่างโค้ดที่พร้อมใช้งานจริง

เริ่มต้นใช้งาน Bybit API

1. สร้าง API Key

ขั้นตอนแรกคือการสร้าง API Key จาก Bybit โดยไปที่ Settings > API แล้วสร้าง Key ใหม่:

2. REST API: ดึงข้อมูล Market Ticker

สำหรับการดึงข้อมูลราคาปัจจุบันของสินทรัพย์ สามารถใช้ Public API ได้โดยไม่ต้องยืนยันตัวตน:

import requests
import json

def get_bybit_ticker(symbol="BTCUSDT"):
    """
    ดึงข้อมูล Ticker ของสินทรัพย์จาก Bybit Perpetual Futures API
    symbol: ชื่อคู่เทรด เช่น BTCUSDT, ETHUSDT, SOLUSDT
    """
    url = "https://api.bybit.com/v5/market/tickers"
    params = {
        "category": "linear",  # USDT Perpetual
        "symbol": symbol
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        data = response.json()
        
        if data["retCode"] == 0:
            ticker = data["result"]["list"][0]
            return {
                "symbol": ticker["symbol"],
                "price": float(ticker["lastPrice"]),
                "24h_change": f"{ticker["price24hPcnt"]}%",
                "24h_high": float(ticker["highPrice24h"]),
                "24h_low": float(ticker["lowPrice24h"]),
                "24h_volume": float(ticker["volume24h"]),
                "mark_price": float(ticker["markPrice"]),
                "funding_rate": f"{ticker["fundingRate"]}%",
                "next_funding_time": ticker["nextFundingTime"]
            }
        else:
            print(f"API Error: {data['retMsg']}")
            return None
            
    except requests.exceptions.Timeout:
        print("Request timeout - API server may be busy")
        return None
    except requests.exceptions.RequestException as e:
        print(f"Connection error: {e}")
        return None


ทดสอบการใช้งาน

if __name__ == "__main__":
    btc_data = get_bybit_ticker("BTCUSDT")
    if btc_data:
        print(f"สรุปข้อมูล {btc_data['symbol']}")
        print(f"ราคาล่าสุด: ${btc_data['price']:,.2f}")
        print(f"24h Change: {btc_data['24h_change']}")
        print(f"24h Volume: {btc_data['24h_volume']:,.2f} BTC")
        print(f"Funding Rate: {btc_data['funding_rate']}")

3. WebSocket: รับข้อมูลเรียลไทม์แบบ Streaming

สำหรับการรับข้อมูลแบบเรียลไทม์ (อัปเดตทุก 100ms) ควรใช้ WebSocket แทนการ Poll API ทุกวินาที:

import websocket
import json
import threading
import time

class BybitWebSocketClient:
    """Bybit WebSocket Client สำหรับรับข้อมูลเรียลไทม์"""
    
    def __init__(self, symbols=["BTCUSDT", "ETHUSDT"]):
        self.symbols = symbols
        self.ws = None
        self.last_price = {}
        self.is_running = False
        
    def on_message(self, ws, message):
        """จัดการเมื่อได้รับข้อความจาก WebSocket"""
        try:
            data = json.loads(message)
            
            # ตรวจสอบประเภทข้อความ
            if data.get("op") == "subscribe":
                print(f"✅ สมัครรับข้อมูลสำเร็จ: {data.get('req', '')}")
                return
                
            if data.get("topic"):
                topic = data["topic"]
                
                if topic.startswith("tickers."):
                    symbol = topic.split(".")[1]
                    ticker_data = data["data"]
                    
                    self.last_price[symbol] = {
                        "price": float(ticker_data["lastPrice"]),
                        "24h_change": float(ticker_data["price24hPcnt"]) * 100,
                        "volume": float(ticker_data["volume24h"]),
                        "timestamp": ticker_data.get("timestamp", int(time.time() * 1000))
                    }
                    
                    # แสดงผลข้อมูล
                    p = self.last_price[symbol]
                    print(f"[{symbol}] ราคา: ${p['price']:,.2f} | "
                          f"เปลี่ยนแปลง: {p['24h_change']:+.2f}% | "
                          f"Volume: {p['volume']:,.0f}")
                          
        except (json.JSONDecodeError, KeyError, ValueError) as e:
            print(f"Parse error: {e}")
            
    def on_error(self, ws, error):
        print(f"WebSocket Error: {error}")
        
    def on_close(self, ws, close_status_code, close_msg):
        print(f"WebSocket ปิดการเชื่อมต่อ ({close_status_code}): {close_msg}")
        if self.is_running:
            self.reconnect()
            
    def on_open(self, ws):
        """สมัครรับข้อมูลเมื่อเชื่อมต่อสำเร็จ"""
        for symbol in self.symbols:
            subscribe_msg = {
                "op": "subscribe",
                "args": [f"tickers.{symbol}"]
            }
            ws.send(json.dumps(subscribe_msg))
            print(f"กำลังสมัครรับข้อมูล: {symbol}")
            
    def connect(self):
        """เชื่อมต่อ WebSocket"""
        self.is_running = True
        self.ws = websocket.WebSocketApp(
            "wss://stream.bybit.com/v5/public/linear",
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        thread = threading.Thread(target=self.ws.run_forever)
        thread.daemon = True
        thread.start()
        print("WebSocket เริ่มเชื่อมต่อ...")
        
    def reconnect(self):
        """เชื่อมต่อใหม่เมื่อหลุดการเชื่อมต่อ"""
        print("กำลังเชื่อมต่อใหม่ใน 5 วินาที...")
        time.sleep(5)
        if self.is_running:
            self.connect()
            
    def stop(self):
        """หยุดการทำงาน"""
        self.is_running = False
        if self.ws:
            self.ws.close()
            
    def get_current_price(self, symbol):
        """ดึงราคาล่าสุดของสินทรัพย์"""
        return self.last_price.get(symbol, {}).get("price")


ทดสอบการใช้งาน

if __name__ == "__main__":
    client = BybitWebSocketClient(["BTCUSDT", "ETHUSDT", "SOLUSDT"])
    client.connect()
    
    # รัน 60 วินาที
    try:
        time.sleep(60)
    except KeyboardInterrupt:
        pass
    finally:
        client.stop()

ประยุกต์ใช้กับ AI: วิเคราะห์สัญญาณการซื้อขาย

เมื่อได้ข้อมูลจาก Bybit แล้ว สามารถส่งต่อไปยัง AI เพื่อวิเคราะห์สัญญาณการซื้อขายได้ ตัวอย่างนี้ใช้ HolySheep AI ซึ่งมีความเร็วตอบสนองต่ำกว่า 50ms และราคาประหยัดกว่า OpenAI ถึง 85%:

import requests
import json
from datetime import datetime

class TradingSignalAnalyzer:
    """ระบบวิเคราะห์สัญญาณการซื้อขายด้วย AI"""
    
    def __init__(self, api_key):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        
    def analyze_market(self, symbol, price, change_24h, volume, funding_rate):
        """
        วิเคราะห์ตลาดด้วย AI และสร้างสัญญาณการซื้อขาย
        ใช้ DeepSeek V3.2 ซึ่งมีราคาเพียง $0.42/MTok
        """
        
        prompt = f"""วิเคราะห์สัญญาณการซื้อขายสำหรับ {symbol}:

ข้อมูลตลาดปัจจุบัน:
- ราคาปัจจุบัน: ${price:,.2f}
- การเปลี่ยนแปลง 24 ชม.: {change_24h:+.2f}%
- ปริมาณการซื้อขาย 24 ชม.: {volume:,.2f}
- Funding Rate ปัจจุบัน: {funding_rate}%

กรุณาให้คำแนะนำ:
1. แนวโน้มตลาด (ขาขึ้น/ขาลง/ข้างเคียง)
2. ระดับความเสี่ยง (ต่ำ/กลาง/สูง)
3. สัญญาณ (ซื้อ/ขาย/รอ)
4. เหตุผลประกอบ

ตอบกลับเป็นภาษาไทยโดยย่อ"""

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "deepseek-chat",
            "messages": [
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 500
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=10
            )
            
            if response.status_code == 200:
                result = response.json()
                return result["choices"][0]["message"]["content"]
            else:
                return f"Error: {response.status_code} - {response.text}"
                
        except requests.exceptions.Timeout:
            return "AI API timeout - ลองใหม่อีกครั้ง"
        except requests.exceptions.RequestException as e:
            return f"Connection error: {e}"


ทดสอบการใช้งาน

if __name__ == "__main__":
    # ตั้งค่า API Key
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    
    analyzer = TradingSignalAnalyzer(API_KEY)
    
    # ดึงข้อมูลจาก Bybit
    from bybit_ticker import get_bybit_ticker
    btc_data = get_bybit_ticker("BTCUSDT")
    
    if btc_data:
        print(f"กำลังวิเคราะห์ {btc_data['symbol']}...")
        
        # วิเคราะห์ด้วย AI
        signal = analyzer.analyze_market(
            symbol=btc_data['symbol'],
            price=btc_data['price'],
            change_24h=float(btc_data['24h_change'].replace('%', '')),
            volume=btc_data['24h_volume'],
            funding_rate=btc_data['funding_rate']
        )
        
        print("\n📊 ผลการวิเคราะห์:")
        print(signal)

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

1. Error 10002 - Signature verification failed

สาเหตุ: API Signature ไม่ถูกต้อง มักเกิดจาก Timestamp ไม่ตรงกันหรือวิธีสร้าง Signature ผิด

# ❌ วิธีที่ผิด - ใช้ Timestamp ผิด
import time
timestamp = str(int(time.time() * 1000))  # หน่วยเป็น milliseconds


✅ วิธีที่ถูก - ใช้วิธีสร้าง Signature ตามเอกสาร Bybit

import hmac
import

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