⚠️ บทความนี้สอนวิธีดาวน์โหลดข้อมูลย้อนหลัง OKX Options ผ่าน Tardis API
สำหรับนักเทรดและนักพัฒนาที่ต้องการวิเคราะห์ตลาดออปชัน
สรุปคำตอบ (TL;DR)
- Tardis API ให้บริการข้อมูล OKX Options history ผ่าน endpoints
/options/chain รองรับข้อมูลราคา IV, Greeks และ OI
- รูปแบบข้อมูลเป็น JSON streaming ส่งผ่าน SSE (Server-Sent Events)
- ความหน่วงของข้อมูลย้อนหลังอยู่ที่ 5-15 นาที หลังเกิดเหตุการณ์
- ทางเลือกที่ประหยัดกว่า: HolySheep AI รองรับ real-time + historical data พร้อม rate ¥1=$1 ประหยัด 85%+
Tardis options_chain คืออะไร
Tardis Machine เป็นแพลตฟอร์มรวบรวมข้อมูลตลาด crypto แบบ low-latency ให้บริการข้อมูล OKX Options ผ่าน WebSocket และ REST API รูปแบบ options_chain จะคืนข้อมูล chain ทั้งหมดของ underlying asset ที่ระบุ
รายละเอียดข้อมูล Tardis OKX Options
| Field | Type | คำอธิบาย | ตัวอย่าง |
| instrument_id | string | รหัสสัญญา | BTC-USD-240630-95000-C |
| underlying | string | สินทรัพย์อ้างอิง | BTC-USD |
| strike | number | ราคา Strike | 95000 |
| expiry | string | วันหมดอายุ | 2024-06-30 |
| option_type | string | C (Call) หรือ P (Put) | C |
| last_price | number | ราคาล่าสุด (USD) | 0.0542 |
| bid_price | number | ราคา Bid | 0.0540 |
| ask_price | number | ราคา Ask | 0.0545 |
| mark_price | number | ราคา Mark (เฉลี่ย) | 0.0542 |
| iv_bid | number | IV ฝั่ง Bid (%) | 58.45 |
| iv_ask | number | IV ฝั่ง Ask (%) | 59.12 |
| iv_mark | number | IV Mark (%) | 58.78 |
| delta | number | Delta (-1 ถึง 1) | 0.4521 |
| gamma | number | Gamma | 0.0000234 |
| theta | number | Theta (ต่อวัน) | -0.001234 |
| vega | number | Vega | 0.000892 |
| open_interest | number | OI (สัญญา) | 12450 |
| volume | number | Volume 24h | 3420 |
| timestamp | number | Unix timestamp (ms) | 1719123456000 |
| date | string | ISO date string | 2024-06-23T08:30:56Z |
วิธีดาวน์โหลดข้อมูล OKX Options History
วิธีที่ 1: REST API (แนะนำสำหรับ Batch Download)
import requests
import json
from datetime import datetime, timedelta
class TardisOKXOptionsDownloader:
"""ดาวน์โหลดข้อมูล OKX Options history จาก Tardis API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
def get_options_chain(
self,
underlying: str = "BTC-USD",
start_date: str = "2024-06-01",
end_date: str = "2024-06-23",
exchange: str = "okx"
) -> list:
"""
ดึงข้อมูล options chain จาก Tardis
Args:
underlying: สินทรัพย์อ้างอิง (BTC-USD, ETH-USD)
start_date: วันที่เริ่มต้น (YYYY-MM-DD)
end_date: วันที่สิ้นสุด (YYYY-MM-DD)
exchange: exchange name
Returns:
list: รายการข้อมูล options chain
"""
url = f"{self.base_url}/charts/okx/options/chain"
params = {
"api_key": self.api_key,
"underlying": underlying,
"from": int(datetime.strptime(start_date, "%Y-%m-%d").timestamp()),
"to": int(datetime.strptime(end_date, "%Y-%m-%d").timestamp()),
"limit": 1000 # ต่อ request
}
headers = {
"Accept": "application/x-ndjson" # Newline-delimited JSON
}
all_data = []
offset = 0
while True:
params["offset"] = offset
response = requests.get(url, params=params, headers=headers)
response.raise_for_status()
# Parse NDJSON (newline-delimited JSON)
lines = response.text.strip().split('\n')
if not lines or lines == ['']:
break
batch = [json.loads(line) for line in lines]
all_data.extend(batch)
if len(batch) < params["limit"]:
break
offset += params["limit"]
print(f"ดาวน์โหลดไปแล้ว {len(all_data)} รายการ...")
return all_data
def filter_by_expiry(self, data: list, expiry: str) -> list:
"""กรองข้อมูลตามวันหมดอายุ"""
return [d for d in data if d.get("expiry") == expiry]
def calculate_iv_surface(self, data: list) -> dict:
"""คำนวณ IV surface จากข้อมูล chain"""
strikes = set()
expiries = set()
for d in data:
if "strike" in d:
strikes.add(d["strike"])
if "expiry" in d:
expiries.add(d["expiry"])
return {
"unique_strikes": len(strikes),
"unique_expiries": len(expiries),
"strike_range": [min(strikes), max(strikes)] if strikes else None
}
วิธีใช้งาน
downloader = TardisOKXOptionsDownloader(api_key="YOUR_TARDIS_API_KEY")
ดาวน์โหลดข้อมูล BTC Options 1 เดือน
data = downloader.get_options_chain(
underlying="BTC-USD",
start_date="2024-05-01",
end_date="2024-06-23"
)
print(f"ดาวน์โหลดสำเร็จ: {len(data)} รายการ")
วิเคราะห์ IV surface
surface = downloader.calculate_iv_surface(data)
print(f"Strikes: {surface['unique_strikes']}, Expiries: {surface['unique_expiries']}")
วิธีที่ 2: WebSocket Streaming (สำหรับ Real-time + Replay)
import asyncio
import json
import websockets
from datetime import datetime
class TardisOKXOptionsWebSocket:
"""เชื่อมต่อ WebSocket สำหรับ OKX Options ผ่าน Tardis"""
def __init__(self, api_key: str):
self.api_key = api_key
self.ws_url = "wss://api.tardis.dev/v1/stream"
async def subscribe_to_options(
self,
underlying: str = "BTC-USD",
replay_from: int = None # Unix timestamp สำหรับ replay
):
"""
เชื่อมต่อ WebSocket และรับข้อมูล OKX Options
Args:
underlying: สินทรัพย์อ้างอิง
replay_from: Unix timestamp สำหรับ replay historical data
"""
params = {
"api_key": self.api_key,
"exchange": "okx",
"channel": "options_chain",
"underlying": underlying
}
if replay_from:
params["from"] = replay_from
params["mode"] = "historical" # หรือ "live"
async with websockets.connect(f"{self.ws_url}?{self._encode_params(params)}") as ws:
print(f"เชื่อมต่อสำเร็จ: {underlying} Options Chain")
async for message in ws:
data = json.loads(message)
await self._process_message(data)
async def _process_message(self, data: dict):
"""ประมวลผลข้อมูลที่ได้รับ"""
msg_type = data.get("type", "")
if msg_type == "options_chain":
# ข้อมูล options chain
chain_data = data.get("data", {})
timestamp = datetime.fromtimestamp(
chain_data.get("timestamp", 0) / 1000
)
print(f"[{timestamp}] {chain_data.get('underlying')} | "
f"Strike: {chain_data.get('strike')} | "
f"IV: {chain_data.get('iv_mark', 0):.2f}% | "
f"Delta: {chain_data.get('delta', 0):.4f}")
elif msg_type == "heartbeat":
# ตรวจสอบ connection
print(f"Heartbeat: {data.get('timestamp')}")
elif msg_type == "error":
print(f"Error: {data.get('message')}")
@staticmethod
def _encode_params(params: dict) -> str:
"""เข้ารหัสพารามิเตอร์สำหรับ URL"""
return "&".join([f"{k}={v}" for k, v in params.items()])
async def main():
"""ตัวอย่างการใช้งาน WebSocket"""
client = TardisOKXOptionsWebSocket(api_key="YOUR_TARDIS_API_KEY")
# รับข้อมูล real-time BTC Options
await client.subscribe_to_options(underlying="BTC-USD")
# หรือ replay ข้อมูลย้อนหลัง (30 นาทีก่อน)
# from datetime import datetime, timedelta
# replay_time = int((datetime.now() - timedelta(minutes=30)).timestamp())
# await client.subscribe_to_options(underlying="BTC-USD", replay_from=replay_time)
รัน WebSocket client
asyncio.run(main())
วิธีที่ 3: Python SDK อย่างเป็นทางการ
# ติดตั้ง: pip install tardis-machine
from tardis import Tardis
from datetime import datetime, timedelta
สร้าง instance
tardis = Tardis(api_key="YOUR_TARDIS_API_KEY")
ดึงข้อมูล OKX Options Chain
options = tardis.options(
exchange="okx",
underlying="BTC-USD",
start_date=datetime(2024, 5, 1),
end_date=datetime(2024, 6, 23),
channels=["chain"]
)
Iterate ผ่านข้อมูล
for msg in options:
if msg.type == "options_chain":
print(f"Strike: {msg.data.strike} | "
f"IV: {msg.data.iv_mark:.2f}% | "
f"OI: {msg.data.open_interest:,}")
ดึงเฉพาะ IV Surface
iv_surface = tardis.options_surface(
exchange="okx",
underlying="BTC-USD",
date=datetime(2024, 6, 23)
)
print(f"Surface timestamp: {iv_surface.timestamp}")
print(f"Data points: {len(iv_surface.chain)}")
เปรียบเทียบราคาและบริการ: HolySheep vs Tardis vs คู่แข่ง
| เกณฑ์เปรียบเทียบ | HolySheep AI | Tardis Machine | CoinAPI | Twelve Data |
| ราคา (Historical Data) | ¥0.42/MTok | $50-500/เดือน | $79+/เดือน | $120+/เดือน |
| Real-time Feed | รวมในแพลน | แยกจ่าย | แยกจ่าย | แยกจ่าย |
| ความหน่วง (Latency) | <50ms | 100-200ms | 200-500ms | 100-300ms |
| OKX Options Data | ✅ มี | ✅ มี | ❌ ไม่มี | ❌ ไม่มี |
| Deribit Options | ✅ มี | ✅ มี | ❌ ไม่มี | ❌ ไม่มี |
| วิธีชำระเงิน | WeChat/Alipay/Crypto | บัตรเครดิต/PayPal | บัตรเครดิต | บัตรเครดิต |
| อัตราแลกเปลี่ยน | ¥1=$1 (85%+ ประหยัด) | ราคาดอลลาร์ | ราคาดอลลาร์ | ราคาดอลลาร์ |
| เครดิตฟรี | ✅ มีเมื่อลงทะเบียน | ❌ ไม่มี | Trial 7 วัน | Trial 7 วัน |
| ปริมาณฟรีต่อเดือน | หลายล้าน tokens | จำกัด | จำกัด | จำกัด |
| Webhook/WebSocket | ✅ มี | ✅ มี | ✅ มี | ✅ มี |
| Support | 24/7 WeChat | Email ธุรกิจ | Email | Email |
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
ข้อผิดพลาดที่ 1: HTTP 401 Unauthorized - Invalid API Key
# ❌ ข้อผิดพลาดที่พบบ่อย
Response: {"error": "Invalid API key"}
🔧 วิธีแก้ไข
1. ตรวจสอบว่า API key ถูกต้อง
TARDIS_API_KEY = "ts_live_xxxx" # ต้องขึ้นต้นด้วย ts_live_
2. ตรวจสอบสิทธิ์ของ API key
ไปที่ https://tardis.dev/settings/api-keys
ตรวจสอบว่าเปิด permissions สำหรับ okx options
3. หรือใช้ HolySheep แทน (ราคาถูกกว่า 85%)
import requests
def test_api_connection(api_key: str) -> bool:
"""ทดสอบการเชื่อมต่อ API"""
response = requests.get(
"https://api.tardis.dev/v1/account",
params={"api_key": api_key}
)
if response.status_code == 401:
print("❌ API Key ไม่ถูกต้อง หรือหมดอายุ")
print("💡 ลองสมัคร HolySheep แทน: https://www.holysheep.ai/register")
return False
return response.ok
ทดสอบ
is_valid = test_api_connection("ts_live_xxxx")
print(f"สถานะ: {'✅ ถูกต้อง' if is_valid else '❌ ไม่ถูกต้อง'}")
ข้อผิดพลาดที่ 2: Rate Limit Exceeded - Too Many Requests
# ❌ ข้อผิดพลาด
Response: {"error": "Rate limit exceeded", "retry_after": 60}
🔧 วิธีแก้ไข
import time
import requests
from ratelimit import limits, sleep_and_retry
class RateLimitedDownloader:
"""Downloader พร้อมระบบจำกัดความถี่"""
# Tardis: 60 requests/minute สำหรับ free tier
CALLS = 60
PERIOD = 60 # วินาที
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
@sleep_and_retry
@limits(calls=CALLS, period=PERIOD)
def _make_request(self, url: str, params: dict) -> requests.Response:
"""ส่ง request พร้อมจำกัดความถี่อัตโนมัติ"""
params["api_key"] = self.api_key
response = requests.get(url, params=params)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"⏳ รอ {retry_after} วินาที...")
time.sleep(retry_after)
return self._make_request(url, params) # ลองใหม่
response.raise_for_status()
return response
def batch_download(self, dates: list) -> list:
"""ดาวน์โหลดข้อมูลหลายวันโดยไม่โดน limit"""
all_data = []
for i, date in enumerate(dates):
print(f"📥 ดาวน์โหลดวันที่ {date} ({i+1}/{len(dates)})")
data = self._get_data_for_date(date)
all_data.extend(data)
# พักเล็กน้อยระหว่าง request
if i < len(dates) - 1:
time.sleep(1)
return all_data
หรือใช้ HolySheep ที่ไม่มี strict rate limit
https://www.holysheep.ai/register
ข้อผิดพลาดที่ 3: Empty Response - No Data for Date Range
# ❌ ข้อผิดพลาด
Response: [] หรือ ไม่มีข้อมูลเลย
🔧 วิธีแก้ไข
from datetime import datetime, timedelta
def debug_empty_response(
api_key: str,
underlying: str,
start_date: str,
end_date: str
):
"""วินิจฉัยปัญหา empty response"""
issues = []
# 1. ตรวจสอบรูปแบบวันที่
try:
start = datetime.strptime(start_date, "%Y-%m-%d")
end = datetime.strptime(end_date, "%Y-%m-%d")
if start > end:
issues.append("❌ วันที่เริ่มต้น > วันที่สิ้นสุด")
# Tardis เก็บข้อมูลย้อนหลังได้เท่าไหร่?
# Free tier: 7 วัน, Pro: 1 ปี
max_lookback = timedelta(days=7) # ปรับตาม plan
if (datetime.now() - start) > max_lookback:
issues.append(f"⚠️ ข้อมูลย้อนหลังเกิน {max_lookback.days} วัน (limit ของ plan)")
except ValueError:
issues.append("❌ รูปแบบวันที่ไม่ถูกต้อง ใช้ YYYY-MM-DD")
# 2. ตรวจสอบ underlying name
valid_underlying = ["BTC-USD", "ETH-USD"]
if underlying not in valid_underlying:
issues.append(f"❌ Underlying '{underlying}' ไม่ถูกต้อง")
issues.append(f" รองรับ: {valid_underlying}")
# 3. ตรวจสอบเวลาตลาด
# OKX Options ซื้อขาย 24/7 แต่ data update ทุก 5 นาที
now = datetime.now()
if now.hour < 2: # ก่อน 02:00 UTC
issues.append("⚠️ ข้อมูลอาจยังไม่อัพเดทเนื่องจากเป็นช่วง maintenance")
# แสดงผลปัญหาที่พบ
if issues:
print("🔍 การวินิจฉัยปัญหา:")
for issue in issues:
print(f" {issue}")
return False
else:
print("✅ ไม่พบปัญหา ลองตรวจสอบ API key อีกครั้ง")
print("💡 หรือสมัคร HolySheep ที่ https://www.holysheep.ai/register")
return True
ทดสอบ
debug_empty_response(
api_key="ts_live_xxxx",
underlying="BTC-USD",
start_date="2024-06-20",
end_date="2024-06-23"
)
ข้อผิดพลาดที่ 4: WebSocket Disconnection - Connection Closed Unexpectedly
# ❌ ข้อผิดพลาด
websockets.exceptions.ConnectionClosed: code=1006, reason=
🔧 วิธีแก้ไข
import asyncio
import websockets
import json
async def robust_websocket_client(api_key: str, underlying: str = "BTC-USD"):
"""WebSocket client พร้อมระบบ reconnect อัตโนมัติ"""
max_retries = 5
retry_delay = 2 # วินาที
for attempt in range(max_retries):
try:
ws_url = (f"wss://api.tardis.dev/v1/stream"
f"?api_key={api_key}"
f"&exchange=okx"
f"&channel=options_chain"
f"&underlying={underlying}")
async with websockets.connect(ws_url, ping_interval=20) as ws:
print(f"✅ เชื่อมต่อสำเร็จ (attempt {attempt + 1})")
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=30)
data = json.loads(message)
process_options_data(data)
except asyncio.TimeoutError:
# ส่ง ping เพื่อรักษา connection
await ws.ping()
print("📡 Heartbeat...")
except websockets.exceptions.ConnectionClosed as e:
print(f"❌ Connection closed: {e.code} - {e.reason}")
if attempt < max_retries - 1:
delay = retry_delay * (2 ** attempt) # Exponential backoff
print(f"🔄 Retry ครั้งที่ {attempt + 2} ใน {delay} วินาที...")
await asyncio.sleep(delay)
else:
print("❌ เลิกพยายามหลังจาก retry 5 ครั้ง")
print("💡 ลองใช้ HolySheep ที่มี connection ที่เสถียรกว่า")
break
except Exception as e:
print(f"❌ Unexpected error: {e}")
break
def process_options_data(data: dict):
"""ประมวลผลข้อมูล options"""
if data.get("type") == "options_chain":
d = data.get("data", {})
print(f"IV: {d.get('iv_mark', 0):.2f}% | "
f"Delta: {d.get('delta', 0):.4f}")
รัน client
asyncio.run(robust_websocket_client("YOUR_API_KEY"))
เหมาะกับใคร / ไม่เหมาะกับใคร
| กลุ่มผู้ใช้ | เหมาะกับ HolySheep | เหมาะกับ Tardis |
| นักเทรดรายบุคคล | ✅ ราคาถูก, มีเครดิตฟรี, รองรับ WeChat/Alipay | ❌ ราคาแพงเกินไปสำหรับรายบุคคล |
| Quant Fund / สถาบัน | ✅ Low latency <50ms, volume สูง | ✅ Enterprise support, SLA 99.9% |
| นักพัฒนา/รีเสิร์ช | ✅ ใช้งานง่าย, มี free tier | ✅ ข้อมู
แหล่งข้อมูลที่เกี่ยวข้องบทความที่เกี่ยวข้อง
🔥 ลอง HolySheep AIเกตเวย์ AI API โดยตรง รองรับ Claude, GPT-5, Gemini, DeepSeek — หนึ่งคีย์ ไม่ต้อง VPN 👉 สมัครฟรี →
|