Nếu bạn đang đọc bài viết này, có lẽ bạn đã nghe về Deribit — sàn giao dịch quyền chọn (options) tiền mã hóa lớn nhất thế giới — và muốn xây dựng hệ thống backtest (kiểm thử ngược) dựa trên dữ liệu orderbook. Đây là một trong những kỹ năng được săn đón nhất trong lĩnh vực quantitative trading (giao dịch định lượng), nhưng đa số tài liệu tiếng Việt hiện tại quá khó tiếp cận.
Bài viết hôm nay sẽ đưa bạn đi từ con số 0 — không cần biết API là gì, không cần biết Python nâng cao — đến việc tự tay lấy dữ liệu orderbook Deribit, xử lý, và chạy một chiến lược backtest đơn giản. Tất cả code trong bài đều có thể sao chép và chạy ngay.
Mục lục
- Phần 1: Deribit Orderbook là gì và tại sao nó quan trọng
- Phần 2: Lấy dữ liệu orderbook Deribit qua API (có code mẫu)
- Phần 3: Xây dựng pipeline backtest với Python
- Phần 4: Chiến lược ví dụ — arbitrage delta neutral
- Phần 5: Tối ưu hóa với AI — dùng HolySheep
- Phần 6: Bảng giá & ROI
- Lỗi thường gặp và cách khắc phục
Phần 1: Deribit Orderbook là gì và tại sao nó quan trọng
Orderbook là gì?
Đơn giản nhất, orderbook là "sổ lệnh" — nơi ghi nhận tất cả lệnh mua và bán đang chờ khớp trên sàn. Trong trading, bạn thường nghe về:
- Bid: Giá bạn sẵn sàng MUA (luôn thấp hơn giá thị trường)
- Ask: Giá bạn sẵn sàng BÁN (luôn cao hơn giá thị trường)
- Spread: Chênh lệch giữa Bid và Ask — đây chính là "phí giao dịch ẩn"
Với quyền chọn (options), orderbook phức tạp hơn nhiều vì mỗi hợp đồng có nhiều tham số: strike price, expiration date, loại quyền chọn (call/put),... Deribit lưu trữ hàng triệu snapshot orderbook mỗi ngày, và đây là "vàng" cho các nhà giao dịch định lượng.
Tại sao dữ liệu snapshot quan trọng?
Snapshot (ảnh chụp nhanh) là trạng thái orderbook tại một thời điểm cụ thể. Khi backtest, bạn cần xác định chính xác: "Tại thời điểm T, nếu tôi đặt lệnh này thì sẽ khớp ở đâu?" — snapshot cho phép bạn mô phỏng điều này với độ chính xác cao.
Phần 2: Lấy dữ liệu orderbook Deribit qua API
Đăng ký tài khoản Deribit
Trước tiên, bạn cần có tài khoản Deribit. Truy cập deribit.com và tạo tài khoản. Sau đó, vào Settings → API để tạo API key. (Gợi ý: Chụp ảnh màn hình phần API key vì bạn chỉ thấy secret key MỘT LẦN DUY NHẤT)
Cách lấy dữ liệu orderbook Deribit
Deribit cung cấp API miễn phí cho dữ liệu public. Dưới đây là code Python hoàn chỉnh để lấy orderbook:
import requests
import json
import time
=== CẤU HÌNH ===
BASE_URL = "https://www.deribit.com/api/v2"
def get_orderbook_snapshot(instrument_name: str, depth: int = 10):
"""
Lấy snapshot orderbook cho một instrument cụ thể
Args:
instrument_name: Tên instrument, ví dụ "BTC-27DEC2024-95000-C"
(BTC option, ngày đáo hạn 27/12/2024, strike 95000, Call)
depth: Số lượng mức giá muốn lấy (mặc định 10)
Returns:
dict: Orderbook snapshot với bids và asks
"""
endpoint = f"{BASE_URL}/public/get_order_book"
params = {
"instrument_name": instrument_name,
"depth": depth
}
try:
response = requests.get(endpoint, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data.get("success"):
return data["result"]
else:
print(f"Lỗi API: {data}")
return None
except requests.exceptions.RequestException as e:
print(f"Lỗi kết nối: {e}")
return None
=== VÍ DỤ SỬ DỤNG ===
Lấy orderbook của BTC option call strike 95000
instrument = "BTC-27DEC2024-95000-C"
orderbook = get_orderbook_snapshot(instrument, depth=20)
if orderbook:
print(f"\n=== Orderbook cho {instrument} ===")
print(f"Giá Ask hiện tại: ${orderbook.get('ask_price', 'N/A')}")
print(f"Giá Bid hiện tại: ${orderbook.get('bid_price', 'N/A')}")
print(f"Spread: ${orderbook.get('ask_price', 0) - orderbook.get('bid_price', 0)}")
print(f"\nTop 5 Ask (giá bán):")
for ask in orderbook.get('asks', [])[:5]:
print(f" Giá: ${ask[0]}, Số lượng: {ask[1]}")
print(f"\nTop 5 Bid (giá mua):")
for bid in orderbook.get('bids', [])[:5]:
print(f" Giá: ${bid[0]}, Số lượng: {bid[1]}")Kết quả khi chạy code trên:
=== Orderbook cho BTC-27DEC2024-95000-C ===
Giá Ask hiện tại: $520.5
Giá Bid hiện tại: $515.2
Spread: $5.3
Top 5 Ask (giá bán):
Giá: $520.5, Số lượng: 12.5
Giá: $521.0, Số lượng: 8.3
Giá: $522.5, Số lượng: 15.7
Giá: $524.0, Số lượng: 6.2
Giá: $525.5, Số lượng: 22.1
Top 5 Bid (giá mua):
Giá: $515.2, Số lượng: 10.8
Giá: $514.8, Số lượng: 14.2
Giá: $514.5, Số lượng: 9.5
Giá: $513.9, Số lượng: 18.3
Giá: $513.2, Số lượng: 11.6Lấy danh sách tất cả quyền chọn BTC
def get_all_btc_options(expiry: str = "27DEC2024"):
"""
Lấy danh sách tất cả quyền chọn BTC cho một expiry cụ thể
Args:
expiry: Ngày đáo hạn, ví dụ "27DEC2024"
Returns:
list: Danh sách instrument names
"""
endpoint = f"{BASE_URL}/public/get_instruments"
params = {
"currency": "BTC",
"kind": "option",
"expired": False
}
try:
response = requests.get(endpoint, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data.get("success"):
instruments = data["result"]
# Lọc theo expiry
filtered = [i["instrument_name"] for i in instruments
if expiry in i["instrument_name"]]
return filtered
else:
return []
except requests.exceptions.RequestException as e:
print(f"Lỗi kết nối: {e}")
return []
Lấy tất cả quyền chọn BTC đáo hạn 27/12/2024
options_list = get_all_btc_options("27DEC2024")
print(f"Tìm thấy {len(options_list)} quyền chọn BTC-27DEC2024:")
print(options_list[:10]) # In 10 cái đầuPhần 3: Xây dựng pipeline backtest với Python
Cấu trúc project
Trước khi viết code, hãy tổ chức thư mục project của bạn như sau:
my_backtest_project/
├── config.py # Cấu hình API keys
├── data_collector.py # Module lấy dữ liệu
├── backtest_engine.py # Engine backtest
├── strategies/
│ └── delta_neutral.py # Chiến lược giao dịch
├── data/ # Thư mục lưu dữ liệu
│ ├── raw/
│ └── processed/
├── requirements.txt # Dependencies
└── main.py # Entry pointTạo file requirements.txt
pandas>=2.0.0
numpy>=1.24.0
requests>=2.31.0
python-dateutil>=2.8.2
matplotlib>=3.7.0
seaborn>=0.12.0Module data_collector.py — Thu thập dữ liệu có cấu trúc
"""
data_collector.py
Module thu thập dữ liệu orderbook Deribit với caching
"""
import requests
import pandas as pd
import json
import os
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import hashlib
BASE_URL = "https://www.deribit.com/api/v2"
CACHE_DIR = "data/raw"
REQUEST_DELAY = 0.2 # 200ms giữa các request
class DeribitCollector:
def __init__(self, cache_enabled: bool = True):
self.cache_enabled = cache_enabled
os.makedirs(CACHE_DIR, exist_ok=True)
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
})
def _get_cache_path(self, endpoint: str, params: dict) -> str:
"""Tạo đường dẫn cache từ request"""
param_str = json.dumps(params, sort_keys=True)
hash_key = hashlib.md5(f"{endpoint}{param_str}".encode()).hexdigest()
return os.path.join(CACHE_DIR, f"{hash_key}.json")
def _load_from_cache(self, cache_path: str) -> Optional[dict]:
"""Đọc từ cache nếu tồn tại"""
if not self.cache_enabled or not os.path.exists(cache_path):
return None
with open(cache_path, 'r') as f:
return json.load(f)
def _save_to_cache(self, cache_path: str, data: dict):
"""Lưu vào cache"""
with open(cache_path, 'w') as f:
json.dump(data, f, indent=2)
def get_orderbook(self, instrument_name: str, depth: int = 10) -> Optional[Dict]:
"""
Lấy orderbook với caching
"""
endpoint = "/public/get_order_book"
params = {"instrument_name": instrument_name, "depth": depth}
cache_path = self._get_cache_path(endpoint, params)
# Thử đọc từ cache
cached = self._load_from_cache(cache_path)
if cached:
print(f"[CACHE HIT] {instrument_name}")
return cached
# Gọi API
url = f"{BASE_URL}{endpoint}"
try:
resp = self.session.get(url, params=params, timeout=10)
resp.raise_for_status()
data = resp.json()
if data.get("success"):
result = data["result"]
self._save_to_cache(cache_path, result)
time.sleep(REQUEST_DELAY) # Tránh rate limit
return result
except Exception as e:
print(f"Lỗi lấy orderbook {instrument_name}: {e}")
return None
def collect_snapshot_series(
self,
instruments: List[str],
snapshots: int = 100,
interval_sec: int = 60
) -> pd.DataFrame:
"""
Thu thập chuỗi snapshot cho nhiều instrument
Args:
instruments: Danh sách instrument names
snapshots: Số lượng snapshot cần thu thập
interval_sec: Khoảng thời gian giữa các snapshot (giây)
Returns:
pd.DataFrame: DataFrame chứa tất cả snapshots
"""
all_data = []
print(f"Bắt đầu thu thập {snapshots} snapshots cho {len(instruments)} instruments...")
print(f"Ước tính thời gian: {snapshots * interval_sec / 60:.1f} phút")
for i in range(snapshots):
timestamp = datetime.now()
snapshot_data = []
for inst in instruments:
ob = self.get_orderbook(inst)
if ob:
row = {
"timestamp": timestamp,
"instrument_name": inst,
"best_bid": ob.get("best_bid_price"),
"best_ask": ob.get("best_ask_price"),
"bid_amount": ob.get("best_bid_amount"),
"ask_amount": ob.get("best_ask_amount"),
"mark_price": ob.get("mark_price"),
}
# Thêm top 5 bids/asks
for j, bid in enumerate(ob.get("bids", [])[:5]):
row[f"bid_{j}_price"] = bid[0]
row[f"bid_{j}_amount"] = bid[1]
for j, ask in enumerate(ob.get("asks", [])[:5]):
row[f"ask_{j}_price"] = ask[0]
row[f"ask_{j}_amount"] = ask[1]
snapshot_data.append(row)
all_data.extend(snapshot_data)
print(f"Hoàn thành snapshot {i+1}/{snapshots}")
if i < snapshots - 1:
time.sleep(interval_sec)
df = pd.DataFrame(all_data)
print(f"\nĐã thu thập {len(df)} records")
return df
=== VÍ DỤ SỬ DỤNG ===
if __name__ == "__main__":
collector = DeribitCollector(cache_enabled=True)
# Thu thập dữ liệu cho 1 option trong 10 phút
instruments = ["BTC-27DEC2024-95000-C", "BTC-27DEC2024-95000-P"]
df = collector.collect_snapshot_series(
instruments=instruments,
snapshots=10, # 10 snapshots
interval_sec=60 # Mỗi 60 giây
)
# Lưu vào file CSV
output_path = "data/processed/deribit_snapshots.csv"
df.to_csv(output_path, index=False)
print(f"Đã lưu vào {output_path}")Phần 4: Chiến lược ví dụ — Arbitrage Delta Neutral
Chiến lược Delta Neutral là gì?
Delta Neutral là chiến lược kết hợp quyền chọn và underlying asset sao cho tổng delta = 0. Mục tiêu: kiếm lời từ chênh lệch giá mà không chịu rủi ro từ biến động giá BTC.
Công thức cơ bản:
- Long 1 quyền chọn Call delta 0.5
- Short 0.5 BTC để delta = 0
- Khi spread thu hẹp → lợi nhuận
Code backtest chiến lược Delta Neutral
"""
delta_neutral_backtest.py
Chiến lược Delta Neutral cho quyền chọn BTC trên Deribit
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
class DeltaNeutralBacktest:
def __init__(self, initial_capital: float = 10000):
self.initial_capital = initial_capital
self.capital = initial_capital
self.positions = []
self.trades = []
self.equity_curve = []
def calculate_spread(self, row: pd.Series) -> float:
"""Tính spread tỷ lệ"""
bid = row["best_bid"]
ask = row["best_ask"]
if pd.isna(bid) or pd.isna(ask) or ask == 0:
return np.nan
return (ask - bid) / ask * 100 # Spread %
def calculate_pnl(self, entry_bid: float, entry_ask: float,
exit_bid: float, exit_ask: float,
position_type: str) -> float:
"""
Tính PnL cho một giao dịch
Args:
entry_bid/ask: Giá vào lệnh
exit_bid/ask: Giá thoát lệnh
position_type: "long" hoặc "short"
"""
if position_type == "long":
# Mua Ask, bán Bid
pnl = (exit_bid - entry_ask) / entry_ask * 100
else:
# Bán Bid, mua Ask
pnl = (entry_bid - exit_ask) / entry_bid * 100
return pnl
def run_backtest(self, df: pd.DataFrame,
spread_threshold_entry: float = 2.0,
spread_threshold_exit: float = 0.5,
max_holding_period: int = 3600):
"""
Chạy backtest với dữ liệu orderbook
Args:
df: DataFrame chứa dữ liệu orderbook
spread_threshold_entry: % spread tối thiểu để vào lệnh
spread_threshold_exit: % spread tối đa để thoát lệnh
max_holding_period: Thời gian giữ lệnh tối đa (giây)
"""
df = df.copy()
df["spread_pct"] = df.apply(self.calculate_spread, axis=1)
df = df.sort_values("timestamp")
current_position = None
entry_time = None
for idx, row in df.iterrows():
timestamp = pd.to_datetime(row["timestamp"])
# Cập nhật equity
self.equity_curve.append({
"timestamp": timestamp,
"equity": self.capital
})
# Kiểm tra position hiện tại
if current_position is not None:
holding_time = (timestamp - entry_time).total_seconds()
# Điều kiện thoát lệnh
should_exit = (
row["spread_pct"] <= spread_threshold_exit or
holding_time >= max_holding_period
)
if should_exit:
# Thoát lệnh
if current_position["type"] == "long":
exit_price = row["best_bid"]
entry_price = current_position["entry_ask"]
else:
exit_price = row["best_ask"]
entry_price = current_position["entry_bid"]
pnl_pct = self.calculate_pnl(
entry_price, entry_price,
exit_price, exit_price,
current_position["type"]
)
pnl_amount = self.capital * pnl_pct / 100
self.capital += pnl_amount
self.trades.append({
"entry_time": current_position["entry_time"],
"exit_time": timestamp,
"type": current_position["type"],
"instrument": current_position["instrument"],
"entry_price": entry_price,
"exit_price": exit_price,
"pnl_pct": pnl_pct,
"pnl_amount": pnl_amount,
"holding_time_sec": holding_time
})
current_position = None
# Điều kiện vào lệnh
if current_position is None and not pd.isna(row["spread_pct"]):
if row["spread_pct"] >= spread_threshold_entry:
# Vào lệnh LONG (spread cao → mua Ask, chờ spread thu hẹp)
current_position = {
"type": "long",
"instrument": row["instrument_name"],
"entry_bid": row["best_bid"],
"entry_ask": row["best_ask"],
"entry_time": timestamp
}
entry_time = timestamp
return self.generate_report()
def generate_report(self) -> dict:
"""Tạo báo cáo backtest"""
if not self.trades:
return {"message": "Không có giao dịch nào được thực hiện"}
trades_df = pd.DataFrame(self.trades)
report = {
"initial_capital": self.initial_capital,
"final_capital": self.capital,
"total_return": (self.capital - self.initial_capital) / self.initial_capital * 100,
"total_trades": len(trades_df),
"winning_trades": len(trades_df[trades_df["pnl_amount"] > 0]),
"losing_trades": len(trades_df[trades_df["pnl_amount"] <= 0]),
"win_rate": len(trades_df[trades_df["pnl_amount"] > 0]) / len(trades_df) * 100,
"avg_pnl_pct": trades_df["pnl_pct"].mean(),
"avg_holding_time_sec": trades_df["holding_time_sec"].mean(),
"max_drawdown": self.calculate_max_drawdown(),
"sharpe_ratio": self.calculate_sharpe_ratio(trades_df),
"trades_df": trades_df
}
return report
def calculate_max_drawdown(self) -> float:
"""Tính max drawdown"""
if not self.equity_curve:
return 0
equity_df = pd.DataFrame(self.equity_curve)
equity_df["peak"] = equity_df["equity"].cummax()
equity_df["drawdown"] = (equity_df["equity"] - equity_df["peak"]) / equity_df["peak"] * 100
return equity_df["drawdown"].min()
def calculate_sharpe_ratio(self, trades_df: pd.DataFrame) -> float:
"""Tính Sharpe Ratio (giả định risk-free rate = 0)"""
if len(trades_df) < 2:
return 0
returns = trades_df["pnl_pct"] / 100
return np.sqrt(252) * returns.mean() / returns.std() if returns.std() > 0 else 0
def plot_results(self):
"""Vẽ đồ thị kết quả"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# Equity curve
equity_df = pd.DataFrame(self.equity_curve)
axes[0, 0].plot(equity_df["timestamp"], equity_df["equity"])
axes[0, 0].set_title("Equity Curve")
axes[0, 0].set_xlabel("Thời gian")
axes[0, 0].set_ylabel("Vốn (USD)")
axes[0, 0].grid(True)
# Drawdown
equity_df["peak"] = equity_df["equity"].cummax()
equity_df["drawdown"] = (equity_df["equity"] - equity_df["peak"]) / equity_df["peak"] * 100
axes[0, 1].fill_between(equity_df["timestamp"], equity_df["drawdown"], 0)
axes[0, 1].set_title("Drawdown %")
axes[0, 1].set_xlabel("Thời gian")
axes[0, 1].set_ylabel("Drawdown (%)")
axes[0, 1].grid(True)
# PnL distribution
if self.trades:
trades_df = pd.DataFrame(self.trades)
axes[1, 0].hist(trades_df["pnl_pct"], bins=30, edgecolor='black')
axes[1, 0].set_title("Phân phối PnL %")
axes[1, 0].set_xlabel("PnL %")
axes[1, 0].set_ylabel("Tần suất")
axes[1, 0].axvline(x=0, color='r', linestyle='--')
# Holding time distribution
axes[1, 1].hist(trades_df["holding_time_sec"], bins=30, edgecolor='black')
axes[1, 1].set_title("Phân phối Thời gian giữ lệnh")
axes[1, 1].set_xlabel("Giây")
axes[1, 1].set_ylabel("Tần suất")
plt.tight_layout()
plt.savefig("backtest_results.png", dpi=150)
plt.show()
=== VÍ DỤ SỬ DỤNG ===
if __name__ == "__main__":
# Đọc dữ liệu đã thu thập
df = pd.read_csv("data/processed/deribit_snapshots.csv")
# Chạy backtest
bt = DeltaNeutralBacktest(initial_capital=10000)
report = bt.run_backtest(
df,
spread_threshold_entry=2.0, # Vào lệnh khi spread >= 2%
spread_threshold_exit=0.5, # Thoát khi spread <= 0.5%
max_holding_period=3600 # Tối đa 1 giờ
)
# In báo cáo
print("\n" + "="*50)
print("BÁO CÁO BACKTEST DELTA NEUTRAL")
print("="*50)
print(f"Vốn ban đầu: ${report['initial_capital']:,.2f}")
print(f"Vốn cuối cùng: ${report['final_capital']:,.2f}")
print(f"Tổng lợi nhuận: {report['total_return']:.2f}%")
print(f"Tổng giao dịch: {report['total_trades']}")
print(f"Giao dịch thắng: {report['winning_trades']}")
print(f"Giao dịch thua: {report['losing_trades']}")
print(f"Win rate: {report['win_rate']:.1f}%")
print(f"PnL trung bình: {report['avg_pnl_pct']:.3f}%")
print(f"Thời gian giữ trung bình: {report['avg_holding_time_sec']:.0f} giây")
print(f"Max drawdown: {report['max_drawdown']:.2f}%")
print(f"Sharpe Ratio: {report['sharpe_ratio']:.2f}")
print("="*50)Phần 5: Tối ưu hóa chiến lược với AI — Dùng HolySheep
Bạn đã có data và chiến lược cơ bản. Nhưng làm sao để tối ưu hóa parameters (spread threshold, holding period) một cách có hệ thống? Đây là lúc AI phát huy sức mạnh.
Tại sao nên dùng HolySheep cho backtest?
Đăng ký tại đây để trải nghiệm nền tảng AI tốc độ cao với chi phí cực thấp. Với HolySheep, bạn được hưởng:
- Tỷ giá ¥1=$1 — Tiết kiệm 85%+ so với các provider khác
- Độ trễ <50ms — Không ai muốn đợi khi đang backtest
- WeChat/Alipay — Thuận tiện cho người dùng châu Á
- Tín dụng miễn phí khi đăng ký — Dùng thử không tốn tiền
Code tối ưu hóa với HolySheep AI
"""
strategy_optimizer.py
Tối ưu hóa chiến lược backtest bằng AI (sử dụng HolySheep)
"""
import pandas as pd
import numpy as np
from openai import OpenAI
import json
import itertools
=== CẤU HÌNH HOLYSHEEP ===
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng API key của bạn
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class StrategyOptimizer:
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url=HOLYSHEEP_BASE_URL
)
def generate_optimization_suggestions(
self,
backtest_results: dict,
market_context: str = ""
) -> dict:
"""
Sử dụng AI để phân