ในโลกของ DeFi และการซื้อขายสินทรัพย์ดิจิทัล การทำตลาด (Market Making) เป็นหัวใจสำคัญของสภาพคล่อง แต่การจัดการความเสี่ยงสินค้าคงคลัง (Inventory Risk) และการวิเคราะห์ PnL อย่างแม่นยำต้องอาศัยข้อมูล Order Book คุณภาพสูง บทความนี้จะพาคุณสร้างระบบวิเคราะห์ที่ขับเคลื่อนด้วยข้อมูล Tardis พร้อมโมเดลความเสี่ยงที่ใช้งานได้จริง
Tardis API: แหล่งข้อมูล Order Book ระดับโลก
Tardis เป็นผู้ให้บริการข้อมูลตลาดคริปโตที่ครอบคลุม Exchange หลายสิบแห่ง ให้ข้อมูล Trade, Order Book, และ OHLCV แบบ Real-time และ Historical โดยมีความหน่วง (Latency) ต่ำกว่า 100ms สำหรับ Feed สive
การตั้งค่า Tardis Client
# ติดตั้ง dependencies
pip install tardis-client asyncio aiohttp pandas numpy
import asyncio
from tardis_client import TardisClient, channels
import pandas as pd
from datetime import datetime, timedelta
class TardisOrderBookAnalyzer:
def __init__(self, api_key: str):
self.client = TardisClient(api_key=api_key)
self.order_book_state = {}
async def subscribe_orderbook(self, exchange: str, symbol: str):
"""สมัครสมาชิก Order Book Stream"""
return self.client.create_orderbook_stream(
exchange=exchange,
symbols=[symbol],
channels=[channels.ORDERBOOK_SNAPSHOT]
)
async def process_orderbook_data(self, message):
"""ประมวลผลข้อมูล Order Book"""
if message.type == 'snapshot':
self.order_book_state[message.exchange_symbol] = {
'bids': {float(p): float(q) for p, q in message.bids},
'asks': {float(p): float(q) for p, q in message.asks},
'timestamp': message.timestamp
}
def calculate_spread(self, symbol: str) -> dict:
"""คำนวณ Bid-Ask Spread และ Mid Price"""
if symbol not in self.order_book_state:
return None
state = self.order_book_state[symbol]
best_bid = max(state['bids'].keys())
best_ask = min(state['asks'].keys())
mid_price = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid_price
return {
'best_bid': best_bid,
'best_ask': best_ask,
'mid_price': mid_price,
'spread_bps': spread * 10000,
'timestamp': state['timestamp']
}
การใช้งาน
async def main():
analyzer = TardisOrderBookAnalyzer(api_key="YOUR_TARDIS_API_KEY")
async with analyzer.subscribe_orderbook("binance", "BTC-USDT") as stream:
async for message in stream:
await analyzer.process_orderbook_data(message)
spread_info = analyzer.calculate_spread("BTC-USDT")
if spread_info:
print(f"BTC Spread: {spread_info['spread_bps']:.2f} bps | Mid: ${spread_info['mid_price']:,.2f}")
asyncio.run(main())โมเดลความเสี่ยงสินค้าคงคลัง (Inventory Risk Model)
การทำตลาดที่มีประสิทธิภาพต้องจัดการ Inventory Risk อย่างเป็นระบบ โมเดลนี้ใช้การคำนวณ Position, Unrealized PnL, และ Risk Metrics แบบ Real-time
import numpy as np
from dataclasses import dataclass
from typing import Dict, List
import json
@dataclass
class Position:
asset: str
quantity: float
avg_entry_price: float
current_price: float
@dataclass
class MarketMakerState:
inventory: Dict[str, Position]
cash_balance: float
trades: List[dict]
class InventoryRiskModel:
def __init__(self, max_position_per_asset: float = 10.0,
risk_free_rate: float = 0.05):
self.state = MarketMakerState(
inventory={},
cash_balance=100000.0,
trades=[]
)
self.max_position = max_position_per_asset
self.risk_free_rate = risk_free_rate
self.price_history: Dict[str, List[float]] = {}
def update_price(self, symbol: str, price: float):
"""อัปเดตราคาล่าสุด"""
if symbol not in self.price_history:
self.price_history[symbol] = []
self.price_history[symbol].append(price)
# เก็บประวัติ 1000 ราคาล่าสุด
if len(self.price_history[symbol]) > 1000:
self.price_history[symbol] = self.price_history[symbol][-1000:]
def execute_trade(self, symbol: str, side: str, quantity: float,
price: float, fee_rate: float = 0.001):
"""บันทึกการซื้อขาย"""
asset = symbol.split('-')[0]
fee = quantity * price * fee_rate
trade = {
'symbol': symbol,
'side': side,
'quantity': quantity,
'price': price,
'fee': fee,
'timestamp': pd.Timestamp.now()
}
if asset not in self.state.inventory:
self.state.inventory[asset] = Position(
asset=asset,
quantity=0.0,
avg_entry_price=0.0,
current_price=price
)
pos = self.state.inventory[asset]
pos.current_price = price
if side == 'buy':
total_cost = quantity * price + fee
new_qty = pos.quantity + quantity
pos.avg_entry_price = (
(pos.quantity * pos.avg_entry_price + quantity * price) / new_qty
if new_qty > 0 else 0
)
pos.quantity = new_qty
self.state.cash_balance -= total_cost
else: # sell
total_proceeds = quantity * price - fee
pos.quantity -= quantity
self.state.cash_balance += total_proceeds
self.state.trades.append(trade)
self.update_price(symbol, price)
return trade
def calculate_unrealized_pnl(self, asset: str) -> float:
"""คำนวณ Unrealized PnL"""
if asset not in self.state.inventory:
return 0.0
pos = self.state.inventory[asset]
return (pos.current_price - pos.avg_entry_price) * pos.quantity
def calculate_var(self, asset: str, confidence: float = 0.95) -> float:
"""คำนวณ Value at Risk (VaR)"""
if asset not in self.price_history or len(self.price_history[asset]) < 30:
return 0.0
returns = np.diff(np.log(self.price_history[asset]))
var_quantile = 1 - confidence
return np.percentile(returns, var_quantile * 100) * len(self.price_history[asset])
def calculate_sharpe_ratio(self, asset: str, window: int = 100) -> float:
"""คำนวณ Sharpe Ratio จากประวัติราคา"""
if asset not in self.price_history or len(self.price_history[asset]) < window:
return 0.0
prices = np.array(self.price_history[asset][-window:])
returns = np.diff(np.log(prices))
excess_return = np.mean(returns) * 252 - self.risk_free_rate
volatility = np.std(returns) * np.sqrt(252)
return excess_return / volatility if volatility > 0 else 0.0
def get_risk_report(self) -> dict:
"""สร้างรายงานความเสี่ยง"""
total_pnl = 0.0
asset_metrics = []
for asset, pos in self.state.inventory.items():
unrealized = self.calculate_unrealized_pnl(asset)
var = self.calculate_var(asset)
sharpe = self.calculate_sharpe_ratio(asset)
total_pnl += unrealized
asset_metrics.append({
'asset': asset,
'quantity': pos.quantity,
'avg_price': pos.avg_entry_price,
'current_price': pos.current_price,
'unrealized_pnl': unrealized,
'var_95': var,
'sharpe_ratio': sharpe,
'position_value': pos.quantity * pos.current_price
})
return {
'total_unrealized_pnl': total_pnl,
'cash_balance': self.state.cash_balance,
'total_portfolio_value': self.state.cash_balance + total_pnl,
'asset_metrics': asset_metrics,
'total_trades': len(self.state.trades)
}
ทดสอบโมเดล
risk_model = InventoryRiskModel(max_position_per_asset=5.0)
risk_model.execute_trade('BTC-USDT', 'buy', 0.5, 45000)
risk_model.execute_trade('ETH-USDT', 'buy', 10.0, 2500)
risk_model.execute_trade('BTC-USDT', 'sell', 0.2, 47000)
for i in range(100):
risk_model.update_price('BTC-USDT', 45000 + np.random.randn() * 500)
risk_model.update_price('ETH-USDT', 2500 + np.random.randn() * 50)
report = risk_model.get_risk_report()
print(f"Total Unrealized PnL: ${report['total_unrealized_pnl']:,.2f}")
print(f"Portfolio Value: ${report['total_portfolio_value']:,.2f}")ระบบ PnL Attribution ด้วย AI
การวิเคราะห์ว่า PnL มาจากแหล่งใด (Spread, Inventory, Fees, Market Impact) เป็นสิ่งสำคัญ บริการ HolySheep AI ช่วยให้คุณสร้างโมเดล Attribution อัจฉริยะได้ง่ายขึ้น ด้วย API ที่รองรับ LLM หลายตัว เช่น GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash และ DeepSeek V3.2 ที่ราคาประหยัดมาก
import requests
import json
from datetime import datetime
HolySheep AI API Configuration
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # รับเครดิตฟรีเมื่อลงทะเบียน
class PnLAttributionAnalyzer:
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_pnl_breakdown(self, trades: list, market_data: dict) -> dict:
"""วิเคราะห์ PnL Breakdown ด้วย AI"""
prompt = f"""คุณเป็นนักวิเคราะห์ PnL สำหรับ Market Maker
วิเคราะห์ข้อมูลการซื้อขายต่อไปนี้และให้รายงาน PnL Attribution:
ข้อมูลการซื้อขาย:
{json.dumps(trades[:10], indent=2)}
ข้อมูลตลาด:
{json.dumps(market_data, indent=2)}
ให้วิเคราะห์:
1. Spread PnL (รายได้จาก Bid-Ask Spread)
2. Inventory PnL (กำไร/ขาดทุนจากสินค้าคงคลัง)
3. Fee PnL (ค่าธรรมเนียมที่จ่าย)
4. Market Impact PnL (ผลกระทบจากการเคลื่อนไหวของตลาด)
คืนค่าเป็น JSON ที่มีโครงสร้าง:
{{"spread_pnl": float, "inventory_pnl": float, "fee_pnl": float,
"market_impact_pnl": float, "total_pnl": float, "analysis": string}}
"""
payload = {
"model": "gpt-4.1", # $8/MTok - ราคาประหยัด
"messages": [
{"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้านการเงินเชิงปริมาณ"},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def generate_risk_report(self, risk_metrics: dict) -> str:
"""สร้างรายงานความเสี่ยงอัตโนมัติ"""
prompt = f"""สร้างรายงานความเสี่ยงสำหรับ Market Maker จาก Metrics ต่อไปนี้:
{json.dumps(risk_metrics, indent=2)}
รายงานควรประกอบด้วย:
1. สรุปภาพรวม (Executive Summary)
2. ความเสี่ยงที่สำคัญ (Key Risks)
3. คำแนะนำ (Recommendations)
4. ตัวชี้วัดสำคัญ (KPIs)
เขียนเป็นภาษาไทย กระชับ เข้าใจง่าย
"""
payload = {
"model": "claude-sonnet-4.5", # $15/MTok - เหมาะกับงานวิเคราะห์
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.5,
"max_tokens": 3000
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
raise Exception(f"Report generation failed: {response.status_code}")
ตัวอย่างการใช้งาน
analyzer = PnLAttributionAnalyzer(api_key=API_KEY)
sample_trades = [
{"symbol": "BTC-USDT", "side": "buy", "quantity": 0.5, "price": 45000, "fee": 22.5},
{"symbol": "BTC-USDT", "side": "sell", "quantity": 0.2, "price": 47000, "fee": 9.4},
{"symbol": "ETH-USDT", "side": "buy", "quantity": 10, "price": 2500, "fee": 25.0},
]
sample_market = {
"BTC_volatility": 0.02,
"ETH_volatility": 0.025,
"BTC_avg_spread": 0.0005,
"ETH_avg_spread": 0.0008,
"market_trend": "bullish"
}
try:
pnl_breakdown = analyzer.analyze_pnl_breakdown(sample_trades, sample_market)
print(f"Spread PnL: ${pnl_breakdown.get('spread_pnl', 0):,.2f}")
print(f"Inventory PnL: ${pnl_breakdown.get('inventory_pnl', 0):,.2f}")
print(f"Total PnL: ${pnl_breakdown.get('total_pnl', 0):,.2f}")
except Exception as e:
print(f"Error: {e}")ตารางเปรียบเทียบ AI Providers สำหรับ Financial Analysis
| โมเดล | ราคา ($/MTok) | Context Window | เหมาะกับงาน | ความเร็ว |
|---|---|---|---|---|
| GPT-4.1 | $8.00 | 128K | วิเคราะห์ PnL, Code Generation | รวดเร็ว |
| Claude Sonnet 4.5 | $15.00 | 200K | รายงานความเสี่ยง, Long-context | ปานกลาง |
| Gemini 2.5 Flash | $2.50 | 1M | งานทั่วไป, Batch Processing | เร็วมาก |
| DeepSeek V3.2 | $0.42 | 64K | งานที่ต้องการความคุ้มค่าสูง | เร็วมาก |
เหมาะกับใคร / ไม่เหมาะกับใคร
✅ เหมาะกับ
- Market Makers รายใหญ่ - ต้องการระบบวิเคราะห์ PnL แบบ Real-time ที่รองรับ Volume สูง
- Quant Funds - ต้องการโมเดลความเสี่ยงที่แม่นยำสำหรับการตัดสินใจลงทุน
- Exchange Operations - ต้องการ Monitor ความเสี่ยงของ Liquidity Providers
- DeFi Protocols - ต้องการเข้าใจ PnL ของ Vaults และ LP Positions
- Retail Traders ที่มีความรู้ - ต้องการเครื่องมือวิเคราะห์ระดับมืออาชีพ
❌ ไม่เหมาะกับ
- ผู้เริ่มต้น - ยังไม่เข้าใจพื้นฐาน Order Book และ Market Making
- งานที่ต้องการ Legal Compliance - ต้องใช้ระบบที่ผ่านการ Audit อย่างเป็นทางการ
- High-Frequency Trading - ต้องการ Latency ต่ำกว่า 1ms ซึ่งต้องใช้ Infrastructure เฉพาะทาง
ราคาและ ROI
การลงทุนในระบบ PnL Analysis คุ้มค่าหรือไม่? มาคำนวณกัน:
| รายการ | ต้นทุน/เดือน | หมายเหตุ |
|---|---|---|
| Tardis API (Historical) | $99 - $499 | ขึ้นอยู่กับ Data Coverage |
| HolySheep AI (DeepSeek V3.2) | ~$50 | 1M Tokens/เดือน @ $0.42/MTok |
| Compute Infrastructure | $100 - $500 | Cloud VPS หรือ Dedicated Server |
| รวมต้นทุนรายเดือน | $249 - $1,049 | แล้วแต่ Scale |
ROI Analysis: หากระบบช่วยลด Inventory Risk ได้เพียง 1-2% ของ AUM $1M ก็คุ้มค่าการลงทุนแล้ว สำหรับ Professional Market Makers ที่มี AUM สูงกว่า $10M ต้นทุนเหล่านี้ถือว่าน้อยมากเมื่อเทียบกับประโยชน์ที่ได้รับ
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. Rate Limit จาก Tardis API
# ❌ วิธีผิด: เรียก API บ่อยเกินไปโดยไม่มีการจัดการ
async def get_orderbook_urgent():
async with client.subscribe_orderbook("binance", "BTC-USDT") as stream:
async for message in stream:
# ประมวลผลทุก message
await process_immediately(message)
✅ วิธีถูก: ใช้ Rate Limiter และ Batching
from collections import deque
import time
class RateLimitedOrderBook:
def __init__(self, max_calls_per_second: int = 10):
self.max_calls = max_calls_per_second
self.call_times = deque(maxlen=max_calls_per_second)
async def get_orderbook(self, client, exchange: str, symbol: str):
now = time.time()
# รอจนกว่า Rate Limit จะพร้อม
while len(self.call_times) >= self.max_calls:
oldest = self.call_times[0]
wait_time = 1.0 - (now - oldest)
if wait_time > 0:
await asyncio.sleep(wait_time)
self.call_times.popleft()
self.call_times.append(time.time())
return await client.get_orderbook_snapshot(exchange, symbol)
async def batch_process(self, messages, batch_size: int = 100):
"""รวบรวม Messages เป็น Batch ก่อนประมวลผล"""
batch = []
for msg in messages:
batch.append(msg)
if len(batch) >= batch_size:
await self.process_batch(batch)
batch = []
if batch:
await self.process_batch(batch)2. Floating Point Precision ในการคำนวณ PnL
# ❌ วิธีผิด: ใช้ float ธรรมดาสำหรับเงิน
def calculate_pnl_bad(quantity: float, price: float):
return quantity * price # อาจผิดพลาดจาก Precision
✅ วิธีถูก: ใช้ Decimal สำหรับการเงิน
from decimal import Decimal, ROUND_HALF_UP, getcontext
getcontext().prec = 28 # กำหนด Precision
class PrecisePnL:
@staticmethod
def calculate_pnl(quantity: str, price: str, fee_rate: str = "0.001") -> dict:
qty = Decimal(quantity)
price = Decimal(price)
fee_r = Decimal(fee_rate)
gross = qty * price
fee = gross * fee_r
net = gross - fee
return {
'gross': float(gross.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)),
'fee': float(fee.quantize(Decimal('0.0001'), rounding=ROUND_HALF_UP)),
'net': float(net.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))
}
def aggregate_pnl(self, trades: list) -> dict:
total_gross = Decimal('0')
total_fee =