ในโลกของการเทรดคริปโตและตลาดการเงิน การพัฒนากลยุทธ์做市商 (Market Maker) ที่ทำกำไรได้อย่างยั่งยืนต้องอาศัยข้อมูลที่แม่นยำและการวิเคราะห์เชิงลึก บทความนี้จะพาคุณสำรวจวิธีการใช้ Tardis Order Book Data เพื่อทำ Backtesting และ Optimize กลยุทธ์做市商อย่างเป็นระบบ โดยใช้ LLM API จาก HolySheep AI เพื่อวิเคราะห์ข้อมูลและสร้างรายงานอัตโนมัติ
Tardis Order Book คืออะไร
Tardis เป็นแพลตฟอร์มที่รวบรวมข้อมูล Order Book คุณภาพสูงจากหลาย Exchange รวมถึง Binance, Bybit, OKX และอื่นๆ ข้อมูลที่ได้รับประกอบด้วย:
- Level 2 Order Book Data (ทุกระดับราคา)
- Trade Data พร้อม Timestamp แม่นยำถึง Microsecond
- Funding Rate และ Premium Index
- Liquidations และ Open Interest
- Full Depth Snapshots ทุก 100ms
ทำไมต้องใช้ LLM ในการวิเคราะห์做市商策略
การวิเคราะห์กลยุทธ์做市商แบบดั้งเดิมต้องใช้สูตรคณิตศาสตร์ซับซ้อนและการเขียนโค้ดหลายพันบรรทัด LLM สามารถช่วยให้การวิเคราะห์เร็วขึ้นหลายเท่า โดยสามารถ:
- วิเคราะห์รูปแบบ Order Flow อัตโนมัติ
- ระบุ Arbitrage Opportunity ระหว่าง Exchange
- สร้างรายงาน Performance ด้วย Natural Language
- แนะนำการ Optimize Parameter อย่างชาญฉลาด
การตั้งค่า Environment และ Dependencies
# สร้าง Virtual Environment
python -m venv market_maker_env
source market_maker_env/bin/activate # Linux/Mac
market_maker_env\Scripts\activate # Windows
ติดตั้ง Dependencies
pip install pandas numpy scipy requests
pip install tardis-client # Official Tardis API Client
pip install asyncio aiohttp # Async support
สร้างไฟล์ config.py
cat > config.py << 'EOF'
import os
HolySheep AI Configuration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Tardis Configuration
TARDIS_API_KEY = "YOUR_TARDIS_API_KEY"
TARDIS_EXCHANGE = "binance"
TARDIS_SYMBOL = "BTC-USDT-PERP"
Backtest Configuration
INITIAL_CAPITAL = 100_000 # USDT
SPREAD_BPS = 10 # Base spread in basis points
ORDER_SIZE_PCT = 0.01 # 1% of capital per side
REBALANCE_THRESHOLD = 0.02 # 2% price movement trigger
EOF
echo "Environment setup complete!"ระบบรวบรวมข้อมูล Tardis Order Book
import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import pandas as pd
from dataclasses import dataclass
@dataclass
class OrderBookSnapshot:
exchange_timestamp: int
local_timestamp: int
bids: List[tuple[float, float]] # (price, quantity)
asks: List[tuple[float, float]] # (price, quantity)
@dataclass
class Trade:
timestamp: int
price: float
quantity: float
side: str # 'buy' or 'sell'
class TardisDataCollector:
"""ระบบรวบรวมข้อมูล Order Book และ Trade จาก Tardis API"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def fetch_orderbook_snapshots(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> List[OrderBookSnapshot]:
"""ดึงข้อมูล Order Book Snapshots"""
url = f"{self.base_url}/filtered/orderbook_snapshots"
params = {
"exchange": exchange,
"symbol": symbol,
"from": int(start_date.timestamp() * 1000),
"to": int(end_date.timestamp() * 1000),
"limit": 1000
}
snapshots = []
async with self.session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
for item in data:
snapshots.append(OrderBookSnapshot(
exchange_timestamp=item["exchangeTimestamp"],
local_timestamp=item["localTimestamp"],
bids=[[b["price"], b["quantity"]] for b in item.get("bids", [])],
asks=[[a["price"], a["quantity"]] for a in item.get("asks", [])]
))
return snapshots
async def fetch_trades(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> List[Trade]:
"""ดึงข้อมูล Trade History"""
url = f"{self.base_url}/filtered/trades"
params = {
"exchange": exchange,
"symbol": symbol,
"from": int(start_date.timestamp() * 1000),
"to": int(end_date.timestamp() * 1000),
"limit": 5000
}
trades = []
async with self.session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
for item in data:
trades.append(Trade(
timestamp=item["timestamp"],
price=float(item["price"]),
quantity=float(item["quantity"]),
side=item["side"]
))
return trades
def calculate_orderbook_metrics(self, snapshot: OrderBookSnapshot) -> Dict:
"""คำนวณ Metrics จาก Order Book Snapshot"""
best_bid = snapshot.bids[0][0] if snapshot.bids else 0
best_ask = snapshot.asks[0][0] if snapshot.asks else 0
mid_price = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid_price * 10000 # in bps
bid_volume = sum(q for _, q in snapshot.bids[:10])
ask_volume = sum(q for _, q in snapshot.asks[:10])
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
return {
"mid_price": mid_price,
"spread_bps": spread,
"bid_volume_10": bid_volume,
"ask_volume_10": ask_volume,
"imbalance": imbalance,
"timestamp": snapshot.exchange_timestamp
}
ตัวอย่างการใช้งาน
async def main():
async with TardisDataCollector("YOUR_TARDIS_API_KEY") as collector:
start = datetime(2024, 1, 15, 0, 0, 0)
end = datetime(2024, 1, 15, 1, 0, 0) # 1 hour
print("กำลังดึงข้อมูล Order Book...")
orderbooks = await collector.fetch_orderbook_snapshots(
"binance-futures", "BTC-USDT-PERP", start, end
)
print("กำลังดึงข้อมูล Trade...")
trades = await collector.fetch_trades(
"binance-futures", "BTC-USDT-PERP", start, end
)
# คำนวณ Metrics
metrics_df = pd.DataFrame([
collector.calculate_orderbook_metrics(ob) for ob in orderbooks
])
print(f"ดึงข้อมูลสำเร็จ: {len(orderbooks)} snapshots, {len(trades)} trades")
print(metrics_df.describe())
return orderbooks, trades
asyncio.run(main())
Backtesting Engine สำหรับ做市商策略
import numpy as np
from typing import Dict, List, Tuple
from dataclasses import dataclass, field
from enum import Enum
class OrderSide(Enum):
BID = "bid"
ASK = "ask"
@dataclass
class Order:
order_id: str
side: OrderSide
price: float
quantity: float
timestamp: int
filled: bool = False
fill_price: float = 0.0
fill_timestamp: int = 0
@dataclass
class Position:
quantity: float # Positive = long, Negative = short
avg_entry: float
unrealized_pnl: float = 0.0
@dataclass
class BacktestResult:
total_pnl: float
total_trades: int
buy_trades: int
sell_trades: int
avg_spread_captured: float
max_drawdown: float
sharpe_ratio: float
win_rate: float
avg_trade_pnl: float
orders_placed: int
orders_filled: int
class MarketMakerBacktester:
"""
Backtesting Engine สำหรับกลยุทธ์做市商
รองรับการทำ Backtest ด้วยข้อมูล Order Book จริงจาก Tardis
"""
def __init__(
self,
initial_capital: float,
spread_bps: float = 10,
order_size_pct: float = 0.01,
rebalance_threshold: float = 0.02
):
self.initial_capital = initial_capital
self.spread_bps = spread_bps
self.order_size_pct = order_size_pct
self.rebalance_threshold = rebalance_threshold
# State
self.cash = initial_capital
self.position = Position(quantity=0, avg_entry=0)
self.orders: List[Order] = []
self.trade_history: List[Dict] = []
self.equity_curve: List[float] = []
self.pending_orders: Dict[str, Order] = {}
# Metrics
self.metrics = {
"filled_bids": 0,
"filled_asks": 0,
"spread_sum": 0,
"pnl_list": []
}
def _generate_order_id(self) -> str:
return f"MM_{len(self.trade_history)}_{np.random.randint(1000, 9999)}"
def _calculate_spread(self, mid_price: float) -> Tuple[float, float]:
"""คำนวณราคา Bid และ Ask จาก Mid Price"""
half_spread = mid_price * (self.spread_bps / 10000) / 2
bid_price = mid_price - half_spread
ask_price = mid_price + half_spread
return bid_price, ask_price
def _calculate_order_size(self, mid_price: float) -> float:
"""คำนวณขนาด Order ตาม % ของ Portfolio"""
portfolio_value = self.cash + abs(self.position.quantity * mid_price)
return (portfolio_value * self.order_size_pct) / mid_price
def _check_order_fill(
self,
order: Order,
trades: List[Trade],
current_time: int
) -> bool:
"""ตรวจสอบว่า Order ถูก Fill หรือไม่"""
# Filter trades ในช่วงเวลาที่ Order ถูกตั้ง
relevant_trades = [
t for t in trades
if order.timestamp <= t.timestamp <= current_time
]
for trade in relevant_trades:
if order.side == OrderSide.BID and trade.price <= order.price and trade.side == "buy":
order.filled = True
order.fill_price = trade.price
order.fill_timestamp = trade.timestamp
return True
elif order.side == OrderSide.ASK and trade.price >= order.price and trade.side == "sell":
order.filled = True
order.fill_price = trade.price
order.fill_timestamp = trade.timestamp
return True
return False
def place_orders(self, mid_price: float, timestamp: int) -> Tuple[Order, Order]:
"""วาง Bid และ Ask Orders"""
bid_price, ask_price = self._calculate_spread(mid_price)
order_size = self._calculate_order_size(mid_price)
bid_order = Order(
order_id=self._generate_order_id(),
side=OrderSide.BID,
price=bid_price,
quantity=order_size,
timestamp=timestamp
)
ask_order = Order(
order_id=self._generate_order_id(),
side=OrderSide.ASK,
price=ask_price,
quantity=order_size,
timestamp=timestamp
)
self.pending_orders[bid_order.order_id] = bid_order
self.pending_orders[ask_order.order_id] = ask_order
return bid_order, ask_order
def process_fill(self, order: Order, mid_price: float):
"""ประมวลผล Order ที่ถูก Fill"""
if order.side == OrderSide.BID:
# ซื้อ = Long Position
self.metrics["filled_bids"] += 1
new_quantity = self.position.quantity + order.quantity
if self.position.quantity == 0:
self.position.avg_entry = order.fill_price
else:
total_cost = (self.position.quantity * self.position.avg_entry) + \
(order.quantity * order.fill_price)
self.position.avg_entry = total_cost / new_quantity
self.position.quantity = new_quantity
self.cash -= order.fill_price * order.quantity
else: # ASK
# ขาย = Short Position หรือปิด Long
self.metrics["filled_asks"] += 1
if self.position.quantity > 0:
# ปิด Long Position
pnl = (order.fill_price - self.position.avg_entry) * \
min(order.quantity, self.position.quantity)
self.metrics["pnl_list"].append(pnl)
self.position.quantity -= order.quantity
if self.position.quantity < 0:
self.position.quantity = 0
else:
# เปิด Short Position
self.position.quantity -= order.quantity
self.cash += order.fill_price * order.quantity
# บันทึก Trade
self.trade_history.append({
"timestamp": order.fill_timestamp,
"side": order.side.value,
"price": order.fill_price,
"quantity": order.quantity,
"mid_price": mid_price,
"spread_captured": abs(order.fill_price - mid_price) / mid_price * 10000
})
self.metrics["spread_sum"] += abs(order.fill_price - mid_price) / mid_price * 10000
del self.pending_orders[order.order_id]
def update_equity(self, mid_price: float):
"""อัปเดต Equity Curve"""
position_value = self.position.quantity * mid_price
equity = self.cash + position_value
self.equity_curve.append(equity)
# คำนวณ Unrealized PnL
if self.position.quantity > 0:
self.position.unrealized_pnl = \
(mid_price - self.position.avg_entry) * self.position.quantity
else:
self.position.unrealized_pnl = \
(self.position.avg_entry - mid_price) * abs(self.position.quantity)
def run_backtest(
self,
orderbook_snapshots: List[OrderBookSnapshot],
trades: List[Trade]
) -> BacktestResult:
"""
รัน Backtest ด้วยข้อมูล Order Book และ Trade
Args:
orderbook_snapshots: ข้อมูล Order Book จาก Tardis
trades: ข้อมูล Trade History จาก Tardis
Returns:
BacktestResult: ผลลัพธ์การ Backtest
"""
print(f"เริ่ม Backtest: {len(orderbook_snapshots)} snapshots")
# Sort ข้อมูลตาม timestamp
snapshots_sorted = sorted(orderbook_snapshots, key=lambda x: x.exchange_timestamp)
for i, snapshot in enumerate(snapshots_sorted):
mid_price = (snapshot.bids[0][0] + snapshot.asks[0][0]) / 2
current_time = snapshot.exchange_timestamp
# ตรวจสอบ Order Fills
orders_to_remove = []
for order_id, order in self.pending_orders.items():
if self._check_order_fill(order, trades, current_time):
self.process_fill(order, mid_price)
# Rebalance: วาง Orders ใหม่ทุก N snapshots
if i % 10 == 0:
self.place_orders(mid_price, current_time)
# อัปเดต Equity
self.update_equity(mid_price)
if (i + 1) % 1000 == 0:
print(f"ประมวลผล {i + 1}/{len(snapshots_sorted)} snapshots...")
return self._calculate_results()
def _calculate_results(self) -> BacktestResult:
"""คำนวณผลลัพธ์ Backtest"""
equity = np.array(self.equity_curve)
returns = np.diff(equity) / equity[:-1]
# Max Drawdown
cummax = np.maximum.accumulate(equity)
drawdowns = (equity - cummax) / cummax
max_dd = abs(np.min(drawdowns))
# Sharpe Ratio
if len(returns) > 0 and np.std(returns) > 0:
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(365 * 24)
else:
sharpe = 0
total_trades = self.metrics["filled_bids"] + self.metrics["filled_asks"]
avg_trade_pnl = np.mean(self.metrics["pnl_list"]) if self.metrics["pnl_list"] else 0
return BacktestResult(
total_pnl=self.equity_curve[-1] - self.initial_capital,
total_trades=total_trades,
buy_trades=self.metrics["filled_bids"],
sell_trades=self.metrics["filled_asks"],
avg_spread_captured=self.metrics["spread_sum"] / total_trades if total_trades > 0 else 0,
max_drawdown=max_dd * 100,
sharpe_ratio=sharpe,
win_rate=len([p for p in self.metrics["pnl_list"] if p > 0]) / len(self.metrics["pnl_list"]) if self.metrics["pnl_list"] else 0,
avg_trade_pnl=avg_trade_pnl,
orders_placed=len(self.trade_history) * 2,
orders_filled=total_trades
)
ตัวอย่างการใช้งาน
def run_sample_backtest():
# สร้างข้อมูล Dummy (แทนที่ด้วยข้อมูลจริงจาก Tardis)
np.random.seed(42)
snapshots = []
trades = []
base_price = 50000
for i in range(10000):
ts = 1704067200000 + i * 100 # 100ms intervals
spread = np.random.uniform(5, 15) # bps
mid = base_price + np.random.normal(0, 50)
snapshots.append(OrderBookSnapshot(
exchange_timestamp=ts,
local_timestamp=ts,
bids=[[mid - spread/2 * mid/10000, np.random.uniform(0.5, 2)] for _ in range(10)],
asks=[[mid + spread/2 * mid/10000, np.random.uniform(0.5, 2)] for _ in range(10)]
))
if np.random.random() < 0.3:
trades.append(Trade(
timestamp=ts,
price=mid + np.random.choice([-1, 1]) * spread/2 * mid/10000,
quantity=np.random.uniform(0.1, 1),
side=np.random.choice(["buy", "sell"])
))
# Run Backtest
backtester = MarketMakerBacktester(
initial_capital=100_000,
spread_bps=10,
order_size_pct=0.01
)
result = backtester.run_backtest(snapshots, trades)
print("\n" + "="*50)
print("BACKTEST RESULTS")
print("="*50)
print(f"Total PnL: ${result.total_pnl:,.2f}")
print(f"Total Trades: {result.total_trades}")
print(f"Win Rate: {result.win_rate:.2%}")
print(f"Avg Spread Captured: {result.avg_spread_captured:.2f} bps")
print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
print(f"Max Drawdown: {result.max_drawdown:.2f}%")
return result
result = run_sample_backtest()
การใช้ LLM วิเคราะห์ผลลัพธ์ด้วย HolySheep AI
import requests
import json
from typing import Dict, List, Optional
import pandas as pd
class HolySheepAnalyzer:
"""
ใช้ HolySheep AI สำหรับวิเคราะห์ผลลัพธ์做市商策略
ราคาถูกกว่า Official API ถึง 85%+ (¥1=$1)
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "gpt-4.1" # $8/MTok - ราคาประหยัดมาก
def _make_request(self, messages: List[Dict], temperature: float = 0.7) -> str:
"""ส่ง Request ไปยัง HolySheep API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": 2000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def analyze_backtest_results(self, result: "BacktestResult") -> Dict:
"""วิเคราะห์ผลลัพธ์ Backtest ด้วย LLM"""
prompt = f"""คุณเป็นผู้เชี่ยวชาญด้านการวิเคราะห์做市商策略 (Market Making Strategy)
วิเคราะห์ผลลัพธ์ Backtest ต่อไปนี้และให้คำแนะนำในการ Optimize:
**ผลลัพธ์ Backtest:**
- Total PnL: ${result.total_pnl:,.2f}
- Total Trades: {result.total_trades}
- Buy Trades: {result.buy_trades}
- Sell Trades: {result.sell_trades}
- Win Rate: {result.win_rate:.2%}
- Avg Spread Captured: {result.avg_spread_captured:.2f} bps
- Sharpe Ratio: {result.sharpe_ratio:.2f}
- Max Drawdown: {result.max_drawdown:.2f}%
- Avg Trade PnL: ${result.avg_trade_pnl:.4f}
กรุณาวิเคราะห์และให้:
1. การประเมินโดยรวม (ดีมาก/ดี/ปานกลาง/ต้องปรับปรุง)
2. จุดแข็งของกลยุทธ์
3. จุดอ่อนและความเสี่ยง
4. คำแนะนำในการ Optimize (Parameter ที่ควรปรับ)
5. ข้อเสนอแนะเชิงกลยุทธ์
ตอบกลับเป็นภาษาไทย"""
messages = [
{"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้านการเงินและการเทรด"},
{"role": "user", "content": prompt}
]
analysis = self._make_request(messages)
return {"analysis": analysis, "model_used": self.model}
def generate_optimization_report(
self,
backtest_results: List["BacktestResult"],
parameter_combinations: List[Dict]
) -> str:
"""สร้างรายงาน Optimization จากหลาย Backtest Results"""
results_text = []
for i, (result, params) in enumerate(zip(backtest_results, parameter_combinations)):
results_text.append(f"""
**Configuration {i+1}:**
- Spread: {params.get('spread_bps', 'N/A')} bps
- Order Size: {params.get('order_size_pct', 'N/A')*100}%
- Rebalance Threshold: {params.get('rebalance_threshold', 'N/A')*100}%
ผลลัพธ์:
- PnL: ${result.total_pnl:,.2f}
- Sharpe: {result.sharpe_ratio:.2f}
- Max DD: {result.max_drawdown:.2f}%
""")
prompt = f"""คุณเป็นผู้เชี่ยวชาญด้านการ Optimize做市商策略
เปรียบเทียบและวิเคราะห์ผลลัพธ์ Backtest หลายรอบต่อไปนี้:
{''.join(results_text)}
ให้รายงานที่ประกอบด้วย:
1. ตารางเปรียบเทียบผลลัพธ์ทั้งหมด
2. Configuration ที่ดีที่สุดและเหตุผล
3. การวิเคราะห์ Correlation ระหว่าง Parameters กับผลลัพธ์
4. คำแนะนำ Final Configuration พร้อม Confidence Level
5. คำเตือนความเสี่ยงที่ควรระวัง
ตอบกลับเป็นภาษาไทย"""
messages = [
{"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้านการเงินเชิงปริมาณ"},
{"role": "user", "content": prompt}
]
return self._make_request(messages)
def analyze_order_flow(self, trades: List[Trade]) -> Dict:
"""วิเคราะห์ Order Flow Patterns"""
trades_df = pd.DataFrame([
{"timestamp": t.timestamp, "price": t.price,
"quantity": t.quantity, "side": t.side}
for t in trades
])
if len(trades_df) == 0:
return {"error": "No trades to analyze"}
# Calculate basic stats
buy_volume = trades_df[trades_df["side"] == "buy"]["quantity"].sum()
sell_volume = trades_df[trades_df["side"] == "sell"]["quantity"].sum()
buy_count = len(trades_df[trades_df["side"] == "buy"])
sell_count = len(trades_df[trades_df["side"] == "sell"])
prompt = f"""วิเคราะห์ Order Flow จากข้อมูลต่อไปนี้:
**Order Flow Statistics:**
- จำนวน Buy Orders: {buy_count}
- จำนวน Sell Orders: {sell_count}
- Buy Volume: {buy_volume:.4f}
- Sell Volume: {sell_volume:.4f}
- Buy/Sell Ratio: {buy_count/sell_count if sell_count > 0 else 'N/A':.2f}
- Volume Imbalance: {(buy_volume - sell_volume)/(buy_volume + sell_volume) if (buy_volume + sell_volume) > 0 else 0:.2%}
- ราคาเฉลี่ย: ${trades_df['price'].mean():,.2f}
- ความผันผวน: ${trades_df['price'].std():,.2f}
ให้การวิเคราะห์:
1. Order Flow Bias (Bullish/Bearish/Neutral)
2. รูปแบบการซื้อขายที่น่าสนใจ
3. ความเสี่ยงที่อาจเกิดขึ้น
4. คำแนะนำการปรับกลยุทธ์ตาม Order Flow
ตอบกลับเป็นภาษาไทย"""
messages = [
{"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้าน Market Microstructure"},
{"role": "user", "content": prompt}
]
analysis = self._make_request(messages)