Mở đầu: Tại sao tôi chuyển từ Hyperliquid native API sang HolySheep Tardis
Ba tháng trước, tôi đang xây dựng một chiến lược arbitrage bot cho Hyperliquid perpetual contracts. Hệ thống chạy ổn định trên môi trường testnet, nhưng khi lên production, tôi gặp ngay vấn đề: dữ liệu L2 orderbook snapshot từ native API có độ trễ 200-500ms, và quan trọng hơn, rate limit khiến backtesting historical data trở thành cơn ác mộng. Một lần backtest 30 ngày mất 6 tiếng chỉ vì phải chờ rate limit reset.
Sau khi thử nghiệm nhiều giải pháp, tôi tìm thấy
HolySheep Tardis Data API — dịch vụ cung cấp historical market data với latency dưới 50ms và chi phí chỉ bằng 15% so với các provider khác. Quá trình migration mất khoảng 2 ngày, nhưng hiệu quả mang lại vượt xa kỳ vọng: backtest 30 ngày giờ chỉ còn 23 phút, và tôi có thể access L2 snapshot data với độ chi tiết cao hơn.
Bài viết này sẽ hướng dẫn chi tiết cách tôi xây dựng pipeline từ đầu đến cuối.
Hyperliquid L2 Snapshot: Tại sao dữ liệu này quan trọng
L2 orderbook snapshot chứa toàn bộ bid/ask levels tại một thời điểm, không phải incremental updates. Với perpetual contracts như HYPE/USDC trên Hyperliquid, L2 snapshot cho phép:
- Tính toán realistic slippage cho large orders
- Phát hiện whale wall placements
- Backtest VWAP/TWAP strategies với độ chính xác cao
- Xây dựng market microstructure models
Native Hyperliquid API chỉ cung cấp websocket stream cho real-time data. Để lấy historical L2 snapshots, bạn cần ticker data từ blockchain events — điều này yêu cầu running một indexer node và xử lý hàng triệu events. HolySheep Tardis đơn giản hóa toàn bộ quy trình này.
Kiến trúc hệ thống
┌─────────────────────────────────────────────────────────────┐
│ Pipeline Architecture │
├─────────────────────────────────────────────────────────────┤
│ │
│ HolySheep Tardis API │
│ (https://api.holysheep.ai/v1/tardis) │
│ │ │
│ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ L2 Snapshot │───▶│ Normalize │───▶│ Backtest │ │
│ │ Fetcher │ │ Parser │ │ Engine │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ Local Cache │ │ Strategy │ │
│ │ (SQLite) │ │ Optimizer │ │
│ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
Cài đặt môi trường và dependencies
pip install requests pandas numpy pyarrow sqlalchemy aiohttp asyncio tqdm
# requirements.txt
requests>=2.31.0
pandas>=2.1.0
numpy>=1.26.0
pyarrow>=14.0.0
sqlalchemy>=2.0.0
aiohttp>=3.9.0
tqdm>=4.66.0
HolySheep Tardis API: Authentication và Rate Limits
HolySheep hỗ trợ API key authentication với format Bearer token. Điểm tôi đánh giá cao là free tier cho phép 10,000 requests/ngày — đủ để chạy multiple backtest cycles mà không cần thanh toán.
import requests
import os
class HolySheepTardisClient:
"""HolySheep Tardis Data API Client cho Hyperliquid L2 snapshots"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str = None):
self.api_key = api_key or os.environ.get("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("API key required. Get yours at https://www.holysheep.ai/register")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
def get_l2_snapshots(
self,
exchange: str = "hyperliquid",
market: str = "HYPE:USDC",
start_time: int, # Unix timestamp ms
end_time: int, # Unix timestamp ms
resolution: str = "1s" # 1s, 5s, 10s, 1m
) -> list:
"""
Fetch L2 orderbook snapshots từ HolySheep Tardis
Args:
exchange: Exchange identifier (hyperliquid)
market: Market pair (HYPE:USDC)
start_time: Start timestamp milliseconds
end_time: End timestamp milliseconds
resolution: Snapshot resolution
Returns:
List of L2 snapshot dictionaries
"""
endpoint = f"{self.BASE_URL}/tardis/historical"
payload = {
"exchange": exchange,
"market": market,
"type": "l2_snapshot",
"start_time": start_time,
"end_time": end_time,
"resolution": resolution,
"limit": 1000 # Max records per request
}
response = self.session.post(endpoint, json=payload)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate limit hit. Waiting {retry_after}s...")
import time
time.sleep(retry_after)
return self.get_l2_snapshots(exchange, market, start_time, end_time, resolution)
response.raise_for_status()
data = response.json()
return data.get("data", [])
def get_live_ticker(self, exchange: str = "hyperliquid", market: str = "HYPE:USDC") -> dict:
"""Fetch real-time ticker data"""
endpoint = f"{self.BASE_URL}/tardis/live"
params = {"exchange": exchange, "market": market}
response = self.session.get(endpoint, params=params)
response.raise_for_status()
return response.json()
Test connection
client = HolySheepTardisClient()
print("HolySheep Tardis API connected successfully")
Fetch và Cache L2 Snapshot Data
Để tối ưu chi phí và tránh repeated API calls, tôi xây dựng một caching layer với SQLite. Data được lưu theo market + date partition.
import sqlite3
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional
import time
class L2SnapshotCache:
"""SQLite-based cache cho L2 snapshots"""
def __init__(self, db_path: str = "l2_snapshots.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS l2_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
exchange TEXT,
market TEXT,
timestamp INTEGER,
asks TEXT, -- JSON string
bids TEXT, -- JSON string
UNIQUE(exchange, market, timestamp)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_lookup
ON l2_snapshots(exchange, market, timestamp)
""")
def save_snapshots(self, snapshots: list):
"""Batch save snapshots to cache"""
if not snapshots:
return
records = [
(
s["exchange"],
s["market"],
s["timestamp"],
str(s.get("asks", [])),
str(s.get("bids", []))
)
for s in snapshots
]
with sqlite3.connect(self.db_path) as conn:
conn.executemany("""
INSERT OR REPLACE INTO l2_snapshots
(exchange, market, timestamp, asks, bids)
VALUES (?, ?, ?, ?, ?)
""", records)
def get_cached_range(
self,
exchange: str,
market: str,
start: int,
end: int
) -> tuple[list, list]:
"""
Get cached timestamp range
Returns: (cached_timestamps, missing_ranges)
"""
with sqlite3.connect(self.db_path) as conn:
df = pd.read_sql("""
SELECT timestamp FROM l2_snapshots
WHERE exchange = ? AND market = ?
AND timestamp BETWEEN ? AND ?
ORDER BY timestamp
""", conn, params=[exchange, market, start, end])
if df.empty:
return [], [(start, end)]
cached = sorted(df["timestamp"].tolist())
# Find missing ranges
missing = []
last_end = start
for ts in cached:
if ts > last_end:
missing.append((last_end, ts - 1))
last_end = ts + 1000 # Assuming 1s resolution
if last_end < end:
missing.append((last_end, end))
return cached, missing
def load_as_dataframe(
self,
exchange: str,
market: str,
start: int,
end: int
) -> pd.DataFrame:
"""Load cached data as DataFrame"""
import json
with sqlite3.connect(self.db_path) as conn:
df = pd.read_sql("""
SELECT * FROM l2_snapshots
WHERE exchange = ? AND market = ?
AND timestamp BETWEEN ? AND ?
ORDER BY timestamp
""", conn, params=[exchange, market, start, end])
if not df.empty:
df["asks"] = df["asks"].apply(json.loads)
df["bids"] = df["bids"].apply(json.loads)
df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
return df
class HyperliquidL2Fetcher:
"""Main fetcher với smart caching"""
def __init__(self, api_key: str, cache: L2SnapshotCache = None):
self.client = HolySheepTardisClient(api_key)
self.cache = cache or L2SnapshotCache()
def fetch_range(
self,
exchange: str = "hyperliquid",
market: str = "HYPE:USDC",
start_time: datetime = None,
end_time: datetime = None,
resolution: str = "1s",
batch_size: int = 5000,
delay_between_batches: float = 0.5
) -> pd.DataFrame:
"""
Fetch L2 snapshots với automatic caching và progress bar
"""
start_ts = int(start_time.timestamp() * 1000)
end_ts = int(end_time.timestamp() * 1000)
# Check cache first
cached_ts, missing_ranges = self.cache.get_cached_range(
exchange, market, start_ts, end_ts
)
print(f"Cache status: {len(cached_ts)} snapshots found, "
f"{len(missing_ranges)} missing ranges")
# Fetch missing ranges
for start_missing, end_missing in missing_ranges:
print(f"Fetching range: {start_missing} - {end_missing}")
current = start_missing
while current < end_missing:
batch_end = min(current + batch_size * 1000, end_missing)
try:
snapshots = self.client.get_l2_snapshots(
exchange=exchange,
market=market,
start_time=current,
end_time=batch_end,
resolution=resolution
)
self.cache.save_snapshots(snapshots)
print(f" Fetched {len(snapshots)} snapshots")
current = batch_end + 1000
time.sleep(delay_between_batches) # Respect rate limits
except Exception as e:
print(f"Error fetching batch: {e}")
time.sleep(5) # Backoff on error
# Load combined data
return self.cache.load_as_dataframe(exchange, market, start_ts, end_ts)
Example usage
if __name__ == "__main__":
client = HolySheepTardisClient()
cache = L2SnapshotCache()
fetcher = HyperliquidL2Fetcher(client.api_key, cache)
# Fetch last 7 days of HYPE/USDC L2 snapshots
end_time = datetime.now()
start_time = end_time - timedelta(days=7)
df = fetcher.fetch_range(
market="HYPE:USDC",
start_time=start_time,
end_time=end_time,
resolution="1s"
)
print(f"\nTotal snapshots: {len(df)}")
print(df.head())
Parse và Phân tích L2 Orderbook
Sau khi có raw snapshots, bước tiếp theo là parse và extract các features quan trọng cho backtesting.
import json
import numpy as np
from dataclasses import dataclass
from typing import List, Tuple
@dataclass
class OrderBookLevel:
price: float
size: float
@dataclass
class L2Snapshot:
timestamp: int
exchange: str
market: str
asks: List[OrderBookLevel]
bids: List[OrderBookLevel]
@property
def best_bid(self) -> float:
return self.bids[0].price if self.bids else 0
@property
def best_ask(self) -> float:
return self.asks[0].price if self.asks else 0
@property
def mid_price(self) -> float:
return (self.best_bid + self.best_ask) / 2
@property
def spread(self) -> float:
return self.best_ask - self.best_bid
@property
def spread_bps(self) -> float:
return (self.spread / self.mid_price) * 10000 if self.mid_price else 0
def get_volume_at_level(self, levels: int = 5, side: str = "both") -> dict:
"""Calculate cumulative volume up to N levels"""
result = {}
if side in ("asks", "both"):
ask_vol = sum(a.size for a in self.asks[:levels])
result["ask_volume"] = ask_vol
if side in ("bids", "both"):
bid_vol = sum(b.size for b in self.bids[:levels])
result["bid_volume"] = bid_vol
if side == "both":
result["imbalance"] = (result["bid_volume"] - result["ask_volume"]) / \
(result["bid_volume"] + result["ask_volume"] + 1e-10)
return result
def estimate_slippage(self, order_size: float, side: str = "buy") -> dict:
"""
Estimate slippage for a market order
Returns: {avg_price, slippage_bps, fully_filled}
"""
levels = self.asks if side == "buy" else self.bids
remaining = order_size
total_cost = 0
filled_levels = 0
for level in levels:
fill_amount = min(remaining, level.size)
total_cost += fill_amount * level.price
remaining -= fill_amount
filled_levels += 1
if remaining <= 0:
break
avg_price = total_cost / (order_size - remaining)
execution_price = levels[0].price
slippage_bps = abs(avg_price - execution_price) / execution_price * 10000
return {
"avg_price": avg_price,
"slippage_bps": slippage_bps,
"fully_filled": remaining <= 0,
"filled_levels": filled_levels,
"remaining": remaining
}
def parse_raw_snapshots(df: pd.DataFrame) -> List[L2Snapshot]:
"""Parse DataFrame rows into L2Snapshot objects"""
snapshots = []
for _, row in df.iterrows():
asks = [
OrderBookLevel(price=float(a[0]), size=float(a[1]))
for a in row["asks"]
]
bids = [
OrderBookLevel(price=float(b[0]), size=float(b[1]))
for b in row["bids"]
]
snapshots.append(L2Snapshot(
timestamp=row["timestamp"],
exchange=row["exchange"],
market=row["market"],
asks=asks,
bids=bids
))
return snapshots
def compute_orderbook_features(df: pd.DataFrame) -> pd.DataFrame:
"""Compute orderbook features từ raw snapshots"""
snapshots = parse_raw_snapshots(df)
features = []
for snap in snapshots:
vol_5 = snap.get_volume_at_level(levels=5)
features.append({
"timestamp": snap.timestamp,
"best_bid": snap.best_bid,
"best_ask": snap.best_ask,
"mid_price": snap.mid_price,
"spread_bps": snap.spread_bps,
"bid_vol_5": vol_5.get("bid_volume", 0),
"ask_vol_5": vol_5.get("ask_volume", 0),
"imbalance_5": vol_5.get("imbalance", 0),
})
return pd.DataFrame(features)
Example: Calculate slippage for different order sizes
if __name__ == "__main__":
# Load sample data
cache = L2SnapshotCache("l2_snapshots.db")
df = cache.load_as_dataframe("hyperliquid", "HYPE:USDC",
start=int((datetime.now() - timedelta(hours=1)).timestamp() * 1000),
end=int(datetime.now().timestamp() * 1000))
if not df.empty:
snapshots = parse_raw_snapshots(df.head(100))
# Test slippage estimates
test_sizes = [1000, 5000, 10000, 50000]
print("\nSlippage Analysis for HYPE/USDC:")
print("-" * 70)
for size in test_sizes:
slippage_buy = snapshots[0].estimate_slippage(size, "buy")
slippage_sell = snapshots[0].estimate_slippage(size, "sell")
print(f"Order Size: {size:>6} | Buy Slippage: {slippage_buy['slippage_bps']:>6.2f} bps | "
f"Sell Slippage: {slippage_sell['slippage_bps']:>6.2f} bps")
Xây dựng Backtesting Engine
Bây giờ tôi sẽ xây dựng một simple backtesting engine sử dụng L2 snapshots để evaluate VWAP strategy.
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, List
import numpy as np
class OrderSide(Enum):
BUY = "buy"
SELL = "sell"
@dataclass
class Position:
entry_price: float = 0
size: float = 0
entry_time: int = 0
pnl: float = 0
@dataclass
class Trade:
timestamp: int
side: OrderSide
size: float
price: float
slippage_bps: float
fees: float
@dataclass
class BacktestResult:
trades: List[Trade] = field(default_factory=list)
total_pnl: float = 0
total_fees: float = 0
sharpe_ratio: float = 0
max_drawdown: float = 0
win_rate: float = 0
def summary(self) -> str:
return f"""
Backtest Results:
=================
Total Trades: {len(self.trades)}
Total PnL: ${self.total_pnl:.2f}
Total Fees: ${self.total_fees:.2f}
Net PnL: ${self.total_pnl - self.total_fees:.2f}
Sharpe Ratio: {self.sharpe_ratio:.2f}
Max Drawdown: {self.max_drawdown:.2f}%
Win Rate: {self.win_rate:.1%}
"""
class VWAPBacktester:
"""
VWAP strategy backtester sử dụng L2 snapshots
Entry signal: Price crosses above VWAP (long), below VWAP (short)
Exit signal: Mean reversion to VWAP or stop loss
"""
def __init__(
self,
snapshots: List[L2Snapshot],
fee_rate: float = 0.0004, # 4 bps taker fee
funding_rate: float = 0.0001, # Daily funding
initial_capital: float = 10000,
vwap_window: int = 300 # 5 minutes in 1s snapshots
):
self.snapshots = snapshots
self.fee_rate = fee_rate
self.funding_rate = funding_rate
self.initial_capital = initial_capital
self.vwap_window = vwap_window
self.capital = initial_capital
self.position: Optional[Position] = None
self.trades: List[Trade] = []
self.equity_curve: List[float] = []
def calculate_vwap(self, idx: int) -> float:
"""Calculate VWAP from past N snapshots"""
start_idx = max(0, idx - self.vwap_window)
window = self.snapshots[start_idx:idx+1]
total_pv = 0
total_vol = 0
for snap in window:
# Use mid price * volume as approximation
vol = sum(l.size for l in snap.bids[:3]) + sum(l.size for l in snap.asks[:3])
total_pv += snap.mid_price * vol
total_vol += vol
return total_pv / total_vol if total_vol > 0 else window[-1].mid_price
def execute_trade(self, snapshot: L2Snapshot, side: OrderSide, size: float):
"""Execute trade và record slippage"""
slippage_info = snapshot.estimate_slippage(size, side.value)
exec_price = slippage_info["avg_price"]
fee = size * exec_price * self.fee_rate
trade = Trade(
timestamp=snapshot.timestamp,
side=side,
size=size,
price=exec_price,
slippage_bps=slippage_info["slippage_bps"],
fees=fee
)
self.trades.append(trade)
self.total_fees += fee
return exec_price, fee
def run(self) -> BacktestResult:
"""Run backtest"""
self.capital = self.initial_capital
self.position = None
self.trades = []
self.equity_curve = [self.capital]
entry_price = 0
position_size = 0
for i in range(self.vwap_window, len(self.snapshots)):
snapshot = self.snapshots[i]
current_price = snapshot.mid_price
vwap = self.calculate_vwap(i)
# Entry signals
if self.position is None:
# Long signal: price crosses above VWAP by threshold
if current_price > vwap * 1.001: # 10 bps threshold
position_size = self.capital * 0.95 / current_price
exec_price, fee = self.execute_trade(snapshot, OrderSide.BUY, position_size)
self.position = Position(
entry_price=exec_price,
size=position_size,
entry_time=snapshot.timestamp
)
self.capital -= fee
# Exit signals
elif self.position is not None:
# Exit long: price crosses below VWAP
if current_price < vwap * 0.999:
exec_price, fee = self.execute_trade(snapshot, OrderSide.SELL, self.position.size)
pnl = (exec_price - self.position.entry_price) * self.position.size
self.capital += pnl - fee
self.trades[-1].fees += fee
self.position = None
# Stop loss: 2% from entry
elif current_price < self.position.entry_price * 0.98:
exec_price, fee = self.execute_trade(snapshot, OrderSide.SELL, self.position.size)
pnl = (exec_price - self.position.entry_price) * self.position.size
self.capital += pnl - fee
self.trades[-1].fees += fee
self.position = None
self.equity_curve.append(self.capital)
return self._compute_results()
def _compute_results(self) -> BacktestResult:
"""Compute performance metrics"""
returns = np.diff(self.equity_curve) / self.equity_curve[:-1]
returns = returns[~np.isnan(returns) & ~np.isinf(returns)]
sharpe = returns.mean() / returns.std() * np.sqrt(252 * 24 * 3600) if returns.std() > 0 else 0
# Max drawdown
peak = self.equity_curve[0]
max_dd = 0
for value in self.equity_curve:
if value > peak:
peak = value
dd = (peak - value) / peak * 100
max_dd = max(max_dd, dd)
# Win rate
pnl_list = []
for i in range(0, len(self.trades) - 1, 2):
if i + 1 < len(self.trades):
entry_trade = self.trades[i]
exit_trade = self.trades[i + 1]
pnl = (exit_trade.price - entry_trade.price) * entry_trade.size - \
entry_trade.fees - exit_trade.fees
pnl_list.append(pnl)
wins = sum(1 for p in pnl_list if p > 0)
win_rate = wins / len(pnl_list) if pnl_list else 0
return BacktestResult(
trades=self.trades,
total_pnl=sum(pnl_list),
total_fees=sum(t.fees for t in self.trades),
sharpe_ratio=sharpe,
max_drawdown=max_dd,
win_rate=win_rate
)
Run backtest example
if __name__ == "__main__":
# Load cached snapshots
cache = L2SnapshotCache("l2_snapshots.db")
df = cache.load_as_dataframe(
"hyperliquid", "HYPE:USDC",
start=int((datetime.now() - timedelta(days=3)).timestamp() * 1000),
end=int(datetime.now().timestamp() * 1000)
)
if len(df) > 100:
snapshots = parse_raw_snapshots(df)
print(f"Loaded {len(snapshots)} snapshots for backtesting")
backtester = VWAPBacktester(
snapshots=snapshots,
fee_rate=0.0004,
initial_capital=10000,
vwap_window=300
)
result = backtester.run()
print(result.summary())
else:
print("Insufficient data for backtesting")
Lỗi thường gặp và cách khắc phục
1. Lỗi 401 Unauthorized - Invalid API Key
# Error: {"error": "Invalid API key", "code": 401}
Nguyên nhân:
- API key không đúng hoặc đã hết hạn
- Key không có quyền truy cập Tardis endpoint
- Bearer token format sai
Khắc phục:
import os
def get_valid_api_key() -> str:
"""Validate và retrieve API key"""
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEY not found in environment. "
"Get your free key at: https://www.holysheep.ai/register"
)
if len(api_key) < 32:
raise ValueError("API key appears to be invalid (too short)")
return api_key
Verify key format
client = HolySheepTardisClient(get_valid_api_key())
print("API key validated successfully")
2. Lỗi 429 Rate Limit Exceeded
# Error: {"error": "Rate limit exceeded", "code": 429, "retry_after": 60}
Nguyên nhân:
- Request quá nhiều trong thời gian ngắn
- Free tier limit: 10,000 requests/ngày
- Không có exponential backoff
Khắc phục với smart retry:
import time
import random
from functools import wraps
def smart_retry(max_retries=5, base_delay=1):
"""Exponential backoff với jitter"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if "429" in str(e) or "rate limit" in str(e).lower():
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"Rate limited. Retry in {delay:.1f}s (attempt {attempt + 1})")
time.sleep(delay)
else:
raise
raise Exception(f"Max retries ({max_retries}) exceeded")
return wrapper
return decorator
@smart_retry(max_retries=5, base_delay=2)
def fetch_with_retry(client, **params):
"""Fetch với automatic retry"""
return client.get_l2_snapshots(**params)
Hoặc sử dụng caching để giảm API calls
class SmartCache:
"""Cache với timestamp-based invalidation"""
def __init__(self, ttl_seconds=3600):
self.cache = {}
self.ttl = ttl_seconds
def get(self, key):
if key in self.cache:
data, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
return data
del self.cache[key]
return None
def set(self, key, data):
self.cache[key] = (data, time.time())
3. Lỗi 422 Invalid Market Format
# Error: {"error": "Invalid market format", "code": 422}
Nguyên nhân:
- Market symbol không đúng format
- Hyperliquid yêu cầu format: BASE:QUOTE (ví dụ HYPE:USDC)
- Một số perpetual markets có suffix như HYPE:USDC-PERP
Khắc phục:
VALID_HYPERLIQUID_MARKETS = {
"HYPE:USDC": "HYPE/USDC Perpetual",
"BTC:USDC": "BTC/USDC Perpetual",
"ETH:USDC": "ETH/USDC Perpetual",
"SOL:USDC": "SOL/USDC Perpetual",
"ARBITRUM:USDC": "ARB/USDC Perpetual",
}
def normalize_market(market: str) -> str:
"""Normalize market symbol"""
market = market.upper().strip()
# Handle common variations
replacements = {
"/": ":",
"-PERP": ":USDC",
"USDT": "USDC",
"_USDC": ":USDC",
}
for old, new in replacements.items():
market = market.replace(old, new)
# Add :USDC if missing
if ":" not in market:
market = f"{market}:USDC"
if market not in VALID_HYPERLIQUID_MARKETS:
raise ValueError(
f"Market {market} not supported. "
f"Valid markets: {list(VALID_HYPERLIQUID_MARKETS.keys())}"
)
return market
Test
print(normalize_market("hype/usdt")) # "HYPE:USDC"
print(normalize_market("BTC-USDT-PERP")) # "BTC:USDC"
4. Lỗi Data Gaps - Missing Snapshots
# Problem: Backtest results không chính xác vì thiếu snapshots
Tài nguyên liên quan
Bài viết liên quan