ในโลกของ DeFi และการเทรดสินทรัพย์ดิจิทัล การสร้าง стратегия แบบ Delta Neutral เป็นหนึ่งในเทคนิคที่ได้รับความนิยมมากที่สุดสำหรับนักลงทุนที่ต้องการสร้างผลตอบแทนจาก funding rate โดยไม่รับความเสี่ยงจากความผันผวนของราคา ในบทความนี้ ผมจะพาคุณไปดูวิธีการดึงข้อมูล position จาก OKX contract API อย่างละเอียด พร้อมทั้ง архитектура การ optimize performance และตัวอย่างโค้ด production-ready ที่สามารถนำไปใช้งานได้จริง
ทำความรู้จัก OKX Contract API และ Position Data
OKX (เดิมชื่อ OKEx) เป็นหนึ่งใน exchange ชั้นนำของโลกที่มี volume การซื้อขายสูงมาก โดยเฉพาะในตลาด futures และ perpetual swaps API ของ OKX มีความ robust และครอบคลุมมาก โดยสามารถแบ่งออกเป็น 2 ประเภทหลักๆ คือ REST API สำหรับการดึงข้อมูลแบบ synchronous และ WebSocket API สำหรับการรับข้อมูลแบบ real-time
REST API Endpoint สำหรับ Position Data
# OKX REST API - Get Position Information
import requests
import hmac
import base64
import datetime
import json
class OKXPositionFetcher:
def __init__(self, api_key: str, api_secret: str, passphrase: str, use_sandbox: bool = False):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.base_url = "https://www.okx.com" if not use_sandbox else "https://www.okx.com/v3/"
def _get_sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""สร้าง HMAC signature สำหรับ authentication"""
message = timestamp + method + path + body
mac = hmac.new(
self.api_secret.encode('utf-8'),
message.encode('utf-8'),
digestmod='sha256'
)
return base64.b64encode(mac.digest()).decode('utf-8')
def _get_headers(self, method: str, path: str, body: str = "") -> dict:
"""สร้าง headers พร้อม authentication"""
timestamp = datetime.datetime.utcnow().isoformat() + 'Z'
sign = self._get_sign(timestamp, method, path, body)
return {
'OK-ACCESS-KEY': self.api_key,
'OK-ACCESS-SIGN': sign,
'OK-ACCESS-TIMESTAMP': timestamp,
'OK-ACCESS-PASSPHRASE': self.passphrase,
'Content-Type': 'application/json'
}
def get_positions(self, inst_type: str = "SWAP", uly: str = None) -> dict:
"""
ดึงข้อมูล position ทั้งหมด
inst_type: MARGIN, SWAP, FUTURES, OPTION
uly: underlying asset เช่น BTC-USDT
"""
path = "/api/v5/account/positions"
params = f"?instType={inst_type}"
if uly:
params += f"&uly={uly}"
headers = self._get_headers("GET", path + params)
response = requests.get(
self.base_url + path + params,
headers=headers
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
ตัวอย่างการใช้งาน
fetcher = OKXPositionFetcher(
api_key="YOUR_API_KEY",
api_secret="YOUR_API_SECRET",
passphrase="YOUR_PASSPHRASE"
)
ดึงข้อมูล perpetual swap positions
positions = fetcher.get_positions(inst_type="SWAP", uly="BTC-USDT")
print(f"พบ {len(positions.get('data', []))} positions")WebSocket API สำหรับ Real-time Position Updates
# OKX WebSocket API - Real-time Position Stream
import asyncio
import json
import websockets
import hmac
import base64
import datetime
from typing import Callable, Optional
class OKXWebSocketPosition:
def __init__(self, api_key: str, api_secret: str, passphrase: str):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.ws_url = "wss://ws.okx.com:8443/ws/v5/private"
async def _generate_auth_signature(self) -> dict:
"""สร้าง WebSocket authentication signature"""
timestamp = datetime.datetime.utcnow().isoformat() + 'Z'
message = timestamp + 'GET' + '/users/self/verify'
mac = hmac.new(
self.api_secret.encode('utf-8'),
message.encode('utf-8'),
digestmod='sha256'
)
sign = base64.b64encode(mac.digest()).decode('utf-8')
return {
'op': "login",
'args': [{
'apiKey': self.api_key,
'passphrase': self.passphrase,
'timestamp': timestamp,
'sign': sign
}]
}
async def subscribe_positions(self, callback: Callable[[dict], None]):
"""
Subscribe ไปยัง position updates แบบ real-time
callback: function ที่จะถูกเรียกเมื่อมี position update
"""
async with websockets.connect(self.ws_url) as ws:
# Authenticate
auth_msg = await self._generate_auth_signature()
await ws.send(json.dumps(auth_msg))
auth_response = await ws.recv()
if json.loads(auth_response).get('code') != '0':
raise Exception(f"Authentication failed: {auth_response}")
# Subscribe to position channel
subscribe_msg = {
'op': "subscribe",
'args': [{
'channel': "positions",
'instType': "SWAP",
'uly': "BTC-USDT"
}]
}
await ws.send(json.dumps(subscribe_msg))
# Listen for updates
async for message in ws:
data = json.loads(message)
if data.get('arg', {}).get('channel') == 'positions':
# มี position update
position_data = data.get('data', [])
for pos in position_data:
await callback(pos)
elif data.get('event') == 'subscribe':
print(f"✅ Subscribed to: {data.get('arg', {}).get('channel')}")
elif data.get('event') == 'error':
print(f"❌ Error: {data.get('msg')}")
ตัวอย่าง callback function
async def handle_position_update(position: dict):
"""ประมวลผล position update แต่ละครั้ง"""
pos_side = position.get('posSide', 'net')
inst_id = position.get('instId')
avail_qty = position.get('availPos', '0')
notional = position.get('notionalUsd', '0')
print(f"📊 {inst_id} | Side: {pos_side} | Available: {avail_qty} | Notional: ${notional}")
รัน WebSocket listener
async def main():
ws_client = OKXWebSocketPosition(
api_key="YOUR_API_KEY",
api_secret="YOUR_API_SECRET",
passphrase="YOUR_PASSPHRASE"
)
await ws_client.subscribe_positions(handle_position_update)
asyncio.run(main())
Delta Neutral Strategy: หลักการและการคำนวณ
ก่อนที่เราจะไปถึงการดึงข้อมูลและประมวลผล มาทำความเข้าใจพื้นฐานของ Delta Neutral Strategy กันก่อน
Delta คืออะไร?
Delta (Δ) คืออัตราส่วนที่บอกว่าราคาของ derivative จะเปลี่ยนแปลงอย่างไรเมื่อราคาของ underlying asset เปลี่ยนไป 1 หน่วย สำหรับ position ใน OKX:
- Long Position (Buy): Delta = +1 (ราคาขึ้น 1 USD → profit 1 USD)
- Short Position (Sell): Delta = -1 (ราคาขึ้น 1 USD → loss 1 USD)
- Spot Holdings: Delta = +1 ต่อ 1 unit
หลักการของ Delta Neutral
เราบอกว่า portfolio มีความเป็นกลาง (neutral) เมื่อ:
Delta_total = Delta_spot + Delta_futures = 0
หรือเขียนในรูปแบบ position size:
Size_spot × 1 + Size_futures × (-1) = 0
Size_spot = Size_futures
นั่นหมายความว่า:
- Long 1 BTC ใน spot market
- Short 1 BTC ใน futures/perpetual swap
= Delta = 0 (ไม่รับความเสี่ยงจากราคา)
เมื่อ portfolio อยู่ในสถานะ delta neutral แล้ว ผลกำไรหรือขาดทุนจะมาจาก funding rate เท่านั้น ซึ่งเป็นการจ่าย period ที่ผู้ที่มี position net long จ่ายให้ผู้ที่มี position net short (หรือกลับกัน) เพื่อให้ราคา perpetual swap อยู่ใกล้เคียงกับราคา spot
โค้ด Complete Delta Neutral Position Calculator
# Delta Neutral Position Calculator - Production Ready
import asyncio
import aiohttp
import json
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
class PositionSide(Enum):
LONG = "long"
SHORT = "short"
NET = "net"
@dataclass
class Position:
inst_id: str # Instrument ID เช่น BTC-USDT-SWAP
inst_type: str # SWAP, FUTURES, MARGIN
pos_side: str # long, short, net
pos: float # Position size (ใน base currency)
pos_currency: str # Base currency เช่น BTC, ETH
notional: float # USD notional value
avail_pos: float # Available position for close
margin: float # Margin used
leverage: int # Leverage level
upl: float # Unrealized PnL
upl_ratio: float # Upl ratio percentage
funding_rate: float # Current funding rate
next_funding_time: str # Next funding time
@dataclass
class DeltaNeutralState:
spot_holdings: float # จำนวน asset ที่ถือ spot
futures_position: float # จำนวน futures position
spot_delta: float # Spot delta (usually 1)
futures_delta: float # Futures delta (-1 for short, +1 for long)
total_delta: float # ผลรวม delta
delta_ratio: float # สัดส่วน delta (1 = perfect neutral)
notional_usd: float # Total notional value in USD
funding_rate: float # Current funding rate (annual)
estimated_daily_funding: float # ประมาณการ funding รายวัน
class DeltaNeutralCalculator:
"""Calculator สำหรับ Delta Neutral Strategy"""
def __init__(self, okx_fetcher: 'OKXPositionFetcher'):
self.fetcher = okx_fetcher
self.spot_positions: List[Position] = []
self.futures_positions: List[Position] = []
def _parse_position(self, data: dict) -> Position:
"""Parse raw API data เป็น Position object"""
return Position(
inst_id=data.get('instId', ''),
inst_type=data.get('instType', ''),
pos_side=data.get('posSide', 'net'),
pos=float(data.get('pos', 0)),
pos_currency=self._extract_currency(data.get('instId', '')),
notional=float(data.get('notionalUsd', 0)),
avail_pos=float(data.get('availPos', 0)),
margin=float(data.get('margin', 0)),
leverage=int(data.get('lever', 1)),
upl=float(data.get('upl', 0)),
upl_ratio=float(data.get('uplRatio', 0)),
funding_rate=float(data.get('fundingRate', 0)),
next_funding_time=data.get('nextFundingTime', '')
)
def _extract_currency(self, inst_id: str) -> str:
"""แยก base currency จาก instrument ID"""
# BTC-USDT-SWAP -> BTC
# BTC-USDT-211225 -> BTC
parts = inst_id.split('-')
return parts[0] if parts else ''
async def fetch_all_positions(self, uly: str = "BTC-USDT"):
"""ดึงข้อมูล position ทั้งหมดและคำนวณ delta"""
# Fetch spot positions (MARGIN)
try:
spot_data = self.fetcher.get_positions(inst_type="MARGIN", uly=uly)
self.spot_positions = [
self._parse_position(p)
for p in spot_data.get('data', [])
if float(p.get('pos', 0)) != 0
]
except Exception as e:
print(f"⚠️ ไม่สามารถดึง spot positions: {e}")
self.spot_positions = []
# Fetch futures/swap positions
try:
swap_data = self.fetcher.get_positions(inst_type="SWAP", uly=uly)
self.futures_positions = [
self._parse_position(p)
for p in swap_data.get('data', [])
if float(p.get('pos', 0)) != 0
]
except Exception as e:
print(f"⚠️ ไม่สามารถดึง swap positions: {e}")
self.futures_positions = []
def calculate_delta_neutral_state(self) -> DeltaNeutralState:
"""คำนวณสถานะ Delta Neutral ของ portfolio"""
# คำนวณ total spot holdings
spot_delta = 0
for pos in self.spot_positions:
if pos.pos_side in ['long', 'net']:
spot_delta += pos.pos
elif pos.pos_side == 'short':
spot_delta -= pos.pos
# คำนวณ total futures position
futures_delta = 0
avg_funding_rate = 0
total_notional = 0
for pos in self.futures_positions:
# Futures delta: short = -1, long = +1
if pos.pos_side == 'short':
futures_delta -= pos.pos # Short มี delta ติดลบ
avg_funding_rate += pos.funding_rate
elif pos.pos_side == 'long':
futures_delta += pos.pos # Long มี delta บวก
avg_funding_rate += pos.funding_rate
total_notional += abs(pos.notional)
# Normalize funding rate
if self.futures_positions:
avg_funding_rate /= len(self.futures_positions)
# คำนวณ total delta
total_delta = spot_delta + futures_delta
# Delta ratio: 1 = perfect neutral, >1 = long bias, <1 = short bias
if futures_delta != 0:
delta_ratio = abs(spot_delta / futures_delta)
else:
delta_ratio = float('inf') if spot_delta != 0 else 1.0
# ประมาณการ funding รายวัน (8 ชั่วโมงต่อครั้ง = 3 ครั้งต่อวัน)
estimated_daily_funding = total_notional * avg_funding_rate * 3
return DeltaNeutralState(
spot_holdings=spot_delta,
futures_position=futures_delta,
spot_delta=spot_delta,
futures_delta=futures_delta,
total_delta=total_delta,
delta_ratio=delta_ratio,
notional_usd=total_notional,
funding_rate=avg_funding_rate,
estimated_daily_funding=estimated_daily_funding
)
def calculate_rebalance_amount(self, target_delta: float = 0) -> Dict[str, float]:
"""
คำนวณจำนวนที่ต้อง rebalance เพื่อให้ได้ target delta
target_delta = 0 หมายถึง perfect delta neutral
"""
state = self.calculate_delta_neutral_state()
# หาความต่างจาก target
delta_diff = target_delta - state.total_delta
# คำนวณว่าต้อง trade ฝั่งไหน
if delta_diff > 0:
# ต้องเพิ่ม delta = ต้อง long spot หรือ close short futures
return {
'action': 'increase_delta',
'amount': abs(delta_diff),
'spot_to_add': abs(delta_diff),
'futures_to_close': 0,
'direction': 'long_spot_or_close_short'
}
else:
# ต้องลด delta = ต้อง sell spot หรือ open short futures
return {
'action': 'decrease_delta',
'amount': abs(delta_diff),
'spot_to_sell': abs(delta_diff),
'futures_to_open': 0,
'direction': 'short_futures_or_sell_spot'
}
def generate_rebalance_report(self) -> str:
"""สร้างรายงานสถานะ Delta Neutral"""
state = self.calculate_delta_neutral_state()
rebalance = self.calculate_rebalance_amount()
report = f"""
╔══════════════════════════════════════════════════════════════╗
║ DELTA NEUTRAL PORTFOLIO REPORT ║
╠══════════════════════════════════════════════════════════════╣
║ 📊 POSITION SUMMARY ║
║ ├─ Spot Holdings: {state.spot_delta:>15.6f} units ║
║ ├─ Futures Position: {state.futures_delta:>15.6f} units ║
║ ├─ Total Delta: {state.total_delta:>15.6f} ║
║ └─ Delta Ratio: {state.delta_ratio:>15.4f} ║
╠══════════════════════════════════════════════════════════════╣
║ 💰 FUNDING INFO ║
║ ├─ Funding Rate: {state.funding_rate * 100:>14.4f}% ║
║ ├─ Est. Daily: ${state.estimated_daily_funding:>14.2f} ║
║ └─ Total Notional: ${state.notional_usd:>14.2f} ║
╠══════════════════════════════════════════════════════════════╣
║ ⚖️ REBALANCE ACTION ║
║ └─ {rebalance.get('direction', 'N/A'):<55} ║
╚══════════════════════════════════════════════════════════════╝
"""
return report
ตัวอย่างการใช้งาน
async def main():
fetcher = OKXPositionFetcher(
api_key="YOUR_API_KEY",
api_secret="YOUR_API_SECRET",
passphrase="YOUR_PASSPHRASE"
)
calculator = DeltaNeutralCalculator(fetcher)
# ดึงข้อมูลและคำนวณ
await calculator.fetch_all_positions("BTC-USDT")
# แสดงรายงาน
print(calculator.generate_rebalance_report())
# คำนวณว่าต้อง rebalance เท่าไหร่
rebalance = calculator.calculate_rebalance_amount(target_delta=0)
print(f"ต้อง rebalance: {rebalance['amount']:.6f} units")
asyncio.run(main())
การเพิ่มประสิทธิภาพและ Production Considerations
Concurrent Position Monitoring หลาย Asset
# Production-grade Concurrent Position Monitor
import asyncio
import aiohttp
from typing import Dict, List, Tuple
from dataclasses import dataclass
import time
from collections import defaultdict
@dataclass
class AssetMonitor:
symbol: str # BTC-USDT, ETH-USDT, etc.
target_delta: float # Delta target (0 = neutral)
current_delta: float # Current portfolio delta
notional_usd: float # Position notional
funding_rate: float # Current funding rate
last_update: float # Timestamp of last update
status: str # OK, WARNING, CRITICAL
class ConcurrentPositionMonitor:
"""
Monitor positions หลาย assetพร้อมกันด้วย async/await
รองรับ rate limiting และ error handling
"""
def __init__(self, fetcher: 'OKXPositionFetcher', rate_limit: int = 10):
self.fetcher = fetcher
self.rate_limit = rate_limit # requests per second
self.monitors: Dict[str, AssetMonitor] = {}
self.last_request_time = 0
self.request_count = 0
self.lock = asyncio.Semaphore(rate_limit)
# Metrics
self.metrics = {
'total_requests': 0,
'failed_requests': 0,
'avg_latency_ms': 0,
'cache_hits': 0
}
async def _rate_limited_request(self, coro):
"""Rate limiting wrapper"""
async with self.lock:
# คำนวณ delay ถ้าจำเป็น
now = time.time()
elapsed = now - self.last_request_time
if elapsed < (1 / self.rate_limit):
await asyncio.sleep((1 / self.rate_limit) - elapsed)
self.last_request_time = time.time()
self.request_count += 1
result = await coro
self.metrics['total_requests'] += 1
return result
async def fetch_single_asset(self, symbol: str, inst_type: str = "SWAP") -> List[dict]:
"""ดึงข้อมูล position ของ asset เดียว"""
loop = asyncio.get_event_loop()
async def _fetch():
return await loop.run_in_executor(
None,
lambda: self.fetcher.get_positions(inst_type=inst_type, uly=symbol)
)
return await self._rate_limited_request(_fetch())
async def monitor_multiple_assets(self, symbols: List[str]) -> Dict[str, AssetMonitor]:
"""
Monitor positions หลาย asset พร้อมกัน
ใช้ asyncio.gather สำหรับ concurrent execution
"""
# สร้าง tasks สำหรับทุก symbol
tasks = []
for symbol in symbols:
task = self._fetch_and_process_symbol(symbol)
tasks.append(task)
# รัน tasks พร้อมกัน
results = await asyncio.gather(*tasks, return_exceptions=True)
# รวบรวมผลลัพธ์
monitors = {}
for symbol, result in zip(symbols, results):
if isinstance(result, Exception):
print(f"❌ Error fetching {symbol}: {result}")
monitors[symbol] = AssetMonitor(
symbol=symbol,
target_delta=0,
current_delta=0,
notional_usd=0,
funding_rate=0,
last_update=time.time(),
status='CRITICAL'
)
else:
monitors[symbol] = result
self.monitors = monitors
return monitors
async def _fetch_and_process_symbol(self, symbol: str) -> AssetMonitor:
"""ดึงข้อมูลและประมวลผล symbol เดียว"""
start_time = time.time()
try:
# ดึงข้อมูลทั้ง swap และ margin
swap_data = await self.fetch_single_asset(symbol, "SWAP")
margin_data = await self.fetch_single_asset(symbol, "MARGIN")
# คำนวณ delta
spot_delta = sum(float(p.get('pos', 0)) for p in margin_data.get('data', []))
futures_delta = sum(
-float(p.get('pos', 0)) if p.get('posSide') == 'short'
else float(p.get('pos', 0))
for p in swap_data.get('data', [])
)
current_delta = spot_delta + futures_delta
# คำนวณ notional
total_notional = sum(
abs(float(p.get('notionalUsd', 0)))
for p in swap_data.get('data', [])
)
# คำนวณ funding rate
funding_rates = [
float(p.get('fundingRate', 0))
for p in swap_data.get('data', [])
if float(p.get('pos', 0)) != 0
]
avg_funding = sum(funding_rates) / len(funding_rates) if funding_rates else 0
# กำหนด status
if abs(current_delta) < 0.01:
status = 'OK'
elif abs(current_delta) < 0.1:
status = 'WARNING'
else:
status = 'CRITICAL'
latency_ms = (time.time() - start_time) * 1000
self.metrics['avg_latency_ms'] = (
(self.metrics['avg_latency_ms'] * (self.metrics['total_requests'] - 1) + latency_ms)
/ self.metrics['total_requests']
)
return AssetMonitor(
symbol=symbol,
target_delta=0,
current_delta=current_delta,
notional_usd=total_notional,
funding_rate=avg_funding,
last_update=time.time(),
status=status
)
except Exception as e:
self.metrics['failed_requests'] += 1
raise
def generate_monitoring_report(self) -> str:
"""สร้างรายงาน monitoring ทั้งหมด"""
header = f"""
╔════════════════════════════════════════════════════════════════╗
║ DELTA NEUTRAL MONITORING REPORT ║
║ {time.strftime('%Y-%m-%d %H:%M:%S UTC'):<44} ║
╠════════════════════════════════════════════════════════════════╣
"""
rows = []
total_notional = 0
total_daily_funding = 0
for symbol, monitor in sorted(self.monitors.items()):
status_icon = {'OK': '✅', 'WARNING': '⚠️', 'CRITICAL': '🚨'}.get(monitor.status, '❓')
daily_funding = monitor.notional_usd * monitor.funding_rate * 3
total_notional += monitor.notional_usd
total_daily_funding += daily_funding
rows.append(
f"║ {status_icon} {symbol:<12} │ Delta: {monitor.current_delta:>+8.4f} "
f"│ ${monitor.notional_usd:>12,.2f} │ Funding: {monitor.funding_rate*100:>+7.4f}% ║"
)
footer = f"""
╠════════════════════════════════════════════════════════════════╣
║ 📈 PORTFOLIO TOTALS ║
║ ├─ Total Notional: ${total_notional:>15,.2f} ║
║ ├─ Est. Daily Funding: ${total_daily_funding:>15,.2f} ║
║ └─ Annual Projection: ${total_daily_funding * 365:>15,.2f} ║
╠════════════════════════════════════════════════════════════════╣