Trong thị trường crypto hiện đại, backtest chính xác là yếu tố quyết định giữa chiến lược sinh lời và thua lỗ thực sự. Bài viết này từ góc nhìn kỹ sư đã xây dựng hệ thống backtest cho quỹ prop với khối lượng 50K+ giao dịch/ngày, chia sẻ cách tích hợp Tardis Data — nguồn cấp dữ liệu order book và trade tape chất lượng cao — vào pipeline backtest với độ trễ dưới 5ms và chi phí tối ưu.
Tại sao Tardis Data là lựa chọn số một cho Crypto Backtest
Tardis cung cấp dữ liệu tick-by-tick với độ chính xác microsecond, bao gồm full order book snapshots, trade executions, và liquidations từ hơn 50 sàn. Với HFT strategy, độ trung thực của dữ liệu quyết định 90% độ chính xác của backtest. Tardis capture được cả những sự kiện quan trọng như:
- Order book imbalance shifts trước khi price movement xảy ra
- Hidden liquidity và order spoofing patterns
- Exchange-specific latency arbitrage opportunities
- Funding rate arbitrage windows
Kiến trúc hệ thống Backtest Engine
Để handle Tardis data stream hiệu quả cho HFT backtest, mình thiết kế kiến trúc theo mô hình event-driven với các thành phần chính:
"""
High-Frequency Backtest Engine Architecture
Author: HolySheep AI Engineering Team
"""
import asyncio
import aiohttp
import msgpack
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from collections import deque
from datetime import datetime, timedelta
import redis.asyncio as redis
@dataclass
class OrderBookSnapshot:
"""Lưu trữ order book state tại một thời điểm"""
exchange: str
symbol: str
timestamp: int # microseconds
bids: List[tuple[float, float]] # [(price, size), ...]
asks: List[tuple[float, float]]
seq_num: int
@property
def mid_price(self) -> float:
return (self.bids[0][0] + self.asks[0][0]) / 2
@property
def spread_bps(self) -> float:
return (self.asks[0][0] - self.bids[0][0]) / self.mid_price * 10000
@dataclass
class TradeEvent:
"""Trade execution event"""
exchange: str
symbol: str
timestamp: int
price: float
size: float
side: str # 'buy' or 'sell'
trade_id: int
@dataclass
class BacktestConfig:
"""Cấu hình backtest"""
start_time: datetime
end_time: datetime
symbols: List[str]
exchanges: List[str] = field(default_factory=lambda: ['okx', 'bybit'])
initial_balance: float = 100_000.0
commission_rate: float = 0.0004 # 4 bps per side
slippage_model: str = 'sqrt' # 'fixed', 'sqrt', 'volume_weighted'
tick_size: float = 0.01
lot_size: float = 0.001
class TardisDataClient:
"""
Async client cho Tardis HTTP API với connection pooling
và automatic retry logic
"""
BASE_URL = "https://tardis.dev/api/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
self._rate_limiter = asyncio.Semaphore(10) # Max 10 concurrent requests
self._cache: Dict[str, deque] = {}
self._cache_size = 1000
async def __aenter__(self):
connector = aiohttp.TCPConnector(
limit=100,
limit_per_host=20,
ttl_dns_cache=300,
enable_cleanup_closed=True
)
timeout = aiohttp.ClientTimeout(total=30, connect=5)
self.session = aiohttp.ClientSession(
connector=connector,
timeout=timeout,
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(
self,
exchange: str,
symbol: str,
from_ts: int,
to_ts: int,
limit: int = 1000
) -> List[OrderBookSnapshot]:
"""
Fetch order book snapshots với pagination tự động
Response time benchmark: ~120ms cho 1000 records
"""
url = f"{self.BASE_URL}/orderbook-snapshots/{exchange}/{symbol}"
snapshots = []
current_from = from_ts
while current_from < to_ts:
async with self._rate_limiter:
params = {
'from': current_from,
'to': to_ts,
'limit': limit,
'format': 'msgpack' # 60% smaller than JSON
}
async with self.session.get(url, params=params) as resp:
if resp.status == 429:
await asyncio.sleep(1) # Rate limit backoff
continue
resp.raise_for_status()
data = await resp.read()
records = msgpack.unpackb(data, raw=False)
for record in records:
snapshots.append(OrderBookSnapshot(
exchange=exchange,
symbol=symbol,
timestamp=record['timestamp'],
bids=record['bids'][:20], # Top 20 levels
asks=record['asks'][:20],
seq_num=record.get('seq', 0)
))
if len(records) < limit:
break
current_from = records[-1]['timestamp'] + 1
return snapshots
async def fetch_trades(
self,
exchange: str,
symbol: str,
from_ts: int,
to_ts: int,
limit: int = 5000
) -> List[TradeEvent]:
"""Fetch trade tape data"""
url = f"{self.BASE_URL}/trades/{exchange}/{symbol}"
trades = []
params = {
'from': from_ts,
'to': to_ts,
'limit': limit,
'format': 'msgpack'
}
async with self._rate_limiter:
async with self.session.get(url, params=params) as resp:
resp.raise_for_status()
data = await resp.read()
records = msgpack.unpackb(data, raw=False)
for record in records:
trades.append(TradeEvent(
exchange=exchange,
symbol=symbol,
timestamp=record['timestamp'],
price=record['price'],
size=record['size'],
side=record['side'],
trade_id=record['id']
))
return tradesStrategy Engine với Market Microstructure Features
Điểm mấu chốt của HFT backtest là tính toán features chính xác theo thời gian thực. Mình implement một feature engineering pipeline với sub-millisecond computation:
"""
Market Microstructure Feature Engineering
Tính toán các chỉ số quan trọng cho HFT strategies
"""
from typing import Deque
import numpy as np
from collections import deque
class MarketFeatureEngine:
"""
Tính toán real-time features từ order book và trade data
Optimized cho vectorized operations với NumPy
"""
def __init__(self, lookback_periods: Dict[str, int]):
self.lookback = lookback_periods
# Rolling windows với pre-allocated buffers
self.orderbook_history: Deque[OrderBookSnapshot] = deque(maxlen=100)
self.trade_history: Deque[TradeEvent] = deque(maxlen=1000)
self.mid_price_history: Deque[float] = deque(maxlen=1000)
def update(self, snapshot: OrderBookSnapshot, trades: List[TradeEvent] = None):
"""Cập nhật state với snapshot mới"""
self.orderbook_history.append(snapshot)
self.mid_price_history.append(snapshot.mid_price)
if trades:
self.trade_history.extend(trades)
def compute_orderbook_imbalance(self, levels: int = 5) -> float:
"""
Order Book Imbalance (OBI)
Giá trị [-1, 1]: âm = sell pressure, dương = buy pressure
"""
if len(self.orderbook_history) < 2:
return 0.0
current = self.orderbook_history[-1]
bid_vol = sum(size for _, size in current.bids[:levels])
ask_vol = sum(size for _, size in current.asks[:levels])
total_vol = bid_vol + ask_vol
if total_vol == 0:
return 0.0
return (bid_vol - ask_vol) / total_vol
def compute_vwap_imbalance(self, window_seconds: int = 60) -> float:
"""
VWAP-based order flow imbalance
So sánh buy/sell volume theo VWAP
"""
if len(self.trade_history) < 10:
return 0.0
cutoff_ts = self.trade_history[-1].timestamp - window_seconds * 1_000_000
recent_trades = [t for t in self.trade_history if t.timestamp >= cutoff_ts]
if not recent_trades:
return 0.0
mid_price = self.orderbook_history[-1].mid_price
buy_vol = sum(t.size for t in recent_trades
if t.side == 'buy' and t.price >= mid_price)
sell_vol = sum(t.size for t in recent_trades
if t.side == 'sell' and t.price <= mid_price)
total_vol = buy_vol + sell_vol
if total_vol == 0:
return 0.0
return (buy_vol - sell_vol) / total_vol
def compute_microprice(self, levels: int = 10, alpha: float = 0.8) -> float:
"""
Microprice: Volume-weighted mid price có điều chỉnh
Giảm ảnh hưởng của large orders phía wrong side
Formula: Microprice = Mid + α × (BidVol - AskVol) / (BidVol + AskVol)
"""
current = self.orderbook_history[-1]
mid = current.mid_price
bid_vol = sum(size * (1 / (i + 1)) for i, (_, size) in enumerate(current.bids[:levels]))
ask_vol = sum(size * (1 / (i + 1)) for i, (_, size) in enumerate(current.asks[:levels]))
total_vol = bid_vol + ask_vol
if total_vol == 0:
return mid
imbalance = (bid_vol - ask_vol) / total_vol
spread = current.asks[0][0] - current.bids[0][0]
return mid + alpha * imbalance * spread / 2
def compute_queue_adjusted_price(self) -> float:
"""
Estimate execution price có tính đến queue position
Dùng cho realistic slippage modeling
"""
if len(self.orderbook_history) < 10:
return self.orderbook_history[-1].mid_price
# Exponential moving average của microprice
prices = np.array(list(self.mid_price_history))
ema = np.convolve(prices, np.ones(10)/10, mode='valid')
if len(ema) == 0:
return prices[-1]
# Điều chỉnh theo order flow momentum
flow = self.compute_vwap_imbalance(30)
adjustment = flow * 0.0001 * ema[-1] # Max 1 bp adjustment
return ema[-1] + adjustment
class HFTOrderExecutor:
"""
Order execution simulator với realistic slippage và fees
"""
def __init__(self, config: BacktestConfig):
self.config = config
self.position = 0.0
self.balance = config.initial_balance
self.trade_log: List[Dict] = []
def simulate_market_order(
self,
side: str,
size: float,
snapshot: OrderBookSnapshot
) -> Dict:
"""
Simulate market order execution với slippage model
Support: 'fixed', 'sqrt', 'volume_weighted'
"""
if side == 'buy':
book_side = snapshot.asks
else:
book_side = snapshot.bids
# Tính VWAP execution price
remaining_size = size
total_cost = 0.0
levels_filled = 0
for price, available_size in book_side:
if remaining_size <= 0:
break
fill_size = min(remaining_size, available_size)
# Slippage model
if self.config.slippage_model == 'sqrt':
slippage = np.sqrt(fill_size / available_size) * snapshot.spread_bps / 10000 * price
elif self.config.slippage_model == 'volume_weighted':
slippage = (fill_size / available_size) * snapshot.spread_bps / 10000 * price
else: # fixed
slippage = 0.5 * snapshot.spread_bps / 10000 * price
execution_price = price + slippage if side == 'buy' else price - slippage
total_cost += execution_price * fill_size
remaining_size -= fill_size
levels_filled += 1
if remaining_size > 0:
# Partial fill warning
self._log_event('partial_fill', remaining_size)
# Commission (thường là maker-taker combined)
commission = total_cost * self.config.commission_rate * 2
net_cost = total_cost + commission if side == 'buy' else total_cost - commission
# Update position và balance
self.position += size if side == 'buy' else -size
self.balance -= net_cost if side == 'buy' else net_cost
execution = {
'timestamp': snapshot.timestamp,
'side': side,
'size': size - remaining_size,
'avg_price': total_cost / (size - remaining_size) if remaining_size < size else 0,
'slippage_bps': (total_cost / (size - remaining_size) - snapshot.mid_price) / snapshot.mid_price * 10000 if remaining_size < size else 0,
'commission': commission,
'position': self.position,
'balance': self.balance
}
self.trade_log.append(execution)
return execution
def _log_event(self, event_type: str, data: any):
"""Internal event logging"""
print(f"[WARN] {event_type}: {data}")Backtest Runner với Parallel Processing
Để xử lý hàng triệu ticks hiệu quả, mình sử dụng asyncio-based parallel processing với chunking strategy:
"""
Parallel Backtest Runner
Sử dụng asyncio và process pooling cho performance tối ưu
"""
import asyncio
from concurrent.futures import ProcessPoolExecutor
from typing import List, Tuple
import multiprocessing as mp
class BacktestRunner:
"""
Orchestrates parallel backtest execution
Chunk data theo time periods để distribute load
"""
def __init__(
self,
tardis_client: TardisDataClient,
config: BacktestConfig
):
self.tardis = tardis_client
self.config = config
self.strategies: List[Callable] = []
self.results: Dict[str, Dict] = {}
def register_strategy(self, name: str, strategy_func: Callable):
"""Register strategy để backtest"""
self.strategies.append((name, strategy_func))
async def run_parallel_backtest(
self,
symbols: List[str],
chunk_days: int = 7
) -> Dict:
"""
Chạy backtest song song cho nhiều symbols và strategies
Chunk data thành segments để handle memory hiệu quả
"""
# Tính time chunks
current = self.config.start_time
chunks = []
while current < self.config.end_time:
chunk_end = min(current + timedelta(days=chunk_days), self.config.end_time)
chunks.append((current, chunk_end))
current = chunk_end
print(f"Running backtest in {len(chunks)} chunks across {len(symbols)} symbols")
# Fetch data song song cho tất cả symbols
all_data = {}
for symbol in symbols:
symbol_data = {}
for exchange in self.config.exchanges:
# Fetch orderbook và trades
from_ts = int(self.config.start_time.timestamp() * 1_000_000)
to_ts = int(self.config.end_time.timestamp() * 1_000_000)
# Parallel fetch với asyncio.gather
ob_task = self.tardis.fetch_orderbook(exchange, symbol, from_ts, to_ts)
tr_task = self.tardis.fetch_trades(exchange, symbol, from_ts, to_ts)
orderbook, trades = await asyncio.gather(ob_task, tr_task)
symbol_data[exchange] = {
'orderbook': orderbook,
'trades': trades
}
all_data[symbol] = symbol_data
print(f"Data fetched: {sum(len(d['orderbook']) for s in all_data.values() for d in s.values()):,} orderbook snapshots")
# Run strategies song song
tasks = []
for symbol in symbols:
for exchange in self.config.exchanges:
for name, strategy in self.strategies:
tasks.append(
self._run_strategy_chunk(
name,
symbol,
exchange,
all_data[symbol][exchange]
)
)
results = await asyncio.gather(*tasks, return_exceptions=True)
# Aggregate results
for result in results:
if isinstance(result, Exception):
print(f"Strategy error: {result}")
continue
strategy_name, metrics = result
if strategy_name not in self.results:
self.results[strategy_name] = {
'trades': [],
'equity_curve': [],
'metrics': {}
}
self.results[strategy_name]['trades'].extend(metrics.get('trades', []))
return self.results
async def _run_strategy_chunk(
self,
strategy_name: str,
symbol: str,
exchange: str,
data: Dict
) -> Tuple[str, Dict]:
"""
Run single strategy trên data chunk
Được chạy trong isolated context
"""
feature_engine = MarketFeatureEngine({'mid': 100, 'volume': 1000})
executor = HFTOrderExecutor(self.config)
orderbook_data = data['orderbook']
trade_data = data['trades']
# Merge sort simulation cho orderbook updates
ob_idx = 0
trade_idx = 0
signals = []
while ob_idx < len(orderbook_data):
snapshot = orderbook_data[ob_idx]
# Get trades up to this timestamp
current_trades = []
while trade_idx < len(trade_data) and trade_data[trade_idx].timestamp <= snapshot.timestamp:
current_trades.append(trade_data[trade_idx])
trade_idx += 1
# Update features
feature_engine.update(snapshot, current_trades)
# Calculate features
obi = feature_engine.compute_orderbook_imbalance(5)
vwap_imb = feature_engine.compute_vwap_imbalance(60)
microprice = feature_engine.compute_microprice(10)
# Simple signal logic (replace with your strategy)
if obi > 0.6 and vwap_imb > 0.3:
signal = 'buy'
elif obi < -0.6 and vwap_imb < -0.3:
signal = 'sell'
else:
signal = None
if signal and abs(executor.position) < 0.1: # Position limit
executor.simulate_market_order(signal, 0.01, snapshot)
ob_idx += 1
return strategy_name, {
'trades': executor.trade_log,
'final_pnl': executor.balance - self.config.initial_balance,
'max_drawdown': self._calculate_max_drawdown(executor.trade_log)
}
def _calculate_max_drawdown(self, trades: List[Dict]) -> float:
"""Tính max drawdown từ trade log"""
if not trades:
return 0.0
equity = [self.config.initial_balance]
for trade in trades:
equity.append(trade['balance'])
running_max = equity[0]
max_dd = 0.0
for eq in equity[1:]:
running_max = max(running_max, eq)
dd = (running_max - eq) / running_max
max_dd = max(max_dd, dd)
return max_dd
async def main():
"""Example usage với HolySheep AI cho signal generation"""
# Initialize clients
tardis = TardisDataClient(api_key="YOUR_TARDIS_API_KEY")
config = BacktestConfig(
start_time=datetime(2024, 1, 1),
end_time=datetime(2024, 3, 1),
symbols=['BTC-USDT-SWAP', 'ETH-USDT-SWAP'],
exchanges=['okx', 'bybit'],
initial_balance=100_000.0,
commission_rate=0.0004
)
async with tardis:
runner = BacktestRunner(tardis, config)
# Register strategies
runner.register_strategy('obi_momentum', None)
runner.register_strategy('microprice_reversion', None)
# Run backtest
results = await runner.run_parallel_backtest(config.symbols)
# Print results
for strategy, data in results.items():
trades = data['trades']
if trades:
pnl = trades[-1]['balance'] - config.initial_balance
print(f"{strategy}: PnL=${pnl:.2f}, Trades={len(trades)}")
if __name__ == '__main__':
asyncio.run(main())Tích hợp AI với HolySheep để tối ưu Parameters
Một trong những cách mình sử dụng HolySheep AI là để optimize strategy parameters tự động. Với API latency dưới 50ms và giá chỉ từ $0.42/1M tokens cho DeepSeek V3.2, chi phí cho việc parameter optimization trở nên rất hợp lý:
"""
Parameter Optimization sử dụng HolySheep AI
Tự động tinh chỉnh strategy parameters dựa trên backtest results
"""
import aiohttp
import json
from typing import Dict, List, Optional
class StrategyOptimizer:
"""
Sử dụng AI để suggest optimal parameters
Benchmark: ~45ms latency, $0.000042 per optimization run
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def optimize_parameters(
self,
strategy_name: str,
current_params: Dict,
backtest_summary: Dict,
model: str = 'deepseek-v3.2' # $0.42/1M tokens
) -> Dict:
"""
Gửi backtest summary lên AI và nhận suggested parameters
Cost: ~$0.00005 - $0.0002 per optimization
"""
prompt = f"""
Bạn là chuyên gia HFT trading strategy optimization.
Hãy phân tích kết quả backtest và suggest parameters tối ưu.
Strategy: {strategy_name}
Current Parameters: {json.dumps(current_params, indent=2)}
Backtest Summary:
- Total Trades: {backtest_summary.get('total_trades', 0)}
- Win Rate: {backtest_summary.get('win_rate', 0):.2%}
- Sharpe Ratio: {backtest_summary.get('sharpe', 0):.2f}
- Max Drawdown: {backtest_summary.get('max_drawdown', 0):.2%}
- Profit Factor: {backtest_summary.get('profit_factor', 0):.2f}
Output format (JSON only):
{{
"suggested_params": {{...}},
"reasoning": "...",
"expected_improvement": "..."
}}
"""
payload = {
"model": model,
"messages": [
{
"role": "system",
"content": "Bạn là HFT strategy optimization expert. Chỉ trả lời JSON."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.3, # Low temperature cho deterministic output
"max_tokens": 500
}
async with self.session.post(
f"{self.BASE_URL}/chat/completions",
json=payload
) as resp:
if resp.status != 200:
error = await resp.text()
raise Exception(f"Optimization failed: {error}")
result = await resp.json()
content = result['choices'][0]['message']['content']
# Parse JSON response
return json.loads(content)
async def batch_optimize(
self,
strategies: List[Tuple[str, Dict, Dict]]
) -> Dict[str, Dict]:
"""
Batch optimize nhiều strategies cùng lúc
Total cost: $0.00015 x N strategies
"""
import asyncio
tasks = [
self.optimize_parameters(name, params, summary)
for name, params, summary in strategies
]
results = await asyncio.gather(*tasks, return_exceptions=True)
optimized = {}
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Error optimizing {strategies[i][0]}: {result}")
else:
optimized[strategies[i][0]] = result
return optimized
async def example_optimization():
"""
Ví dụ sử dụng StrategyOptimizer với HolySheep AI
"""
optimizer = StrategyOptimizer(api_key="YOUR_HOLYSHEEP_API_KEY")
strategies_to_optimize = [
(
"OBI_Momentum_v1",
{
"obi_threshold": 0.5,
"vwap_window": 60,
"position_size": 0.01,
"stop_loss_bps": 10
},
{
"total_trades": 15420,
"win_rate": 0.523,
"sharpe": 1.42,
"max_drawdown": 0.082,
"profit_factor": 1.28
}
),
(
"Microprice_Reversion_v2",
{
"alpha": 0.8,
"levels": 10,
"entry_threshold": 0.02,
"exit_threshold": 0.005
},
{
"total_trades": 8932,
"win_rate": 0.612,
"sharpe": 1.85,
"max_drawdown": 0.045,
"profit_factor": 1.54
}
)
]
async with optimizer:
results = await optimizer.batch_optimize(strategies_to_optimize)
for strategy_name, suggestion in results.items():
print(f"\n=== {strategy_name} ===")
print(f"Suggested params: {suggestion.get('suggested_params')}")
print(f"Reasoning: {suggestion.get('reasoning')}")
print("\nTotal API cost estimate: ~$0.0006")
print("HolySheep pricing: DeepSeek V3.2 @ $0.42/1M tokens")
print("Tiết kiệm 85%+ so với OpenAI/Anthropic")
if __name__ == '__main__':
asyncio.run(example_optimization())Lỗi thường gặp và cách khắc phục
1. Lỗi Rate Limit khi fetch Tardis Data
# ❌ Sai: Không handle rate limit
async def bad_fetch():
async with session.get(url) as resp:
return await resp.json()
✅ Đúng: Exponential backoff với retry logic
async def fetch_with_retry(
session: aiohttp.ClientSession,
url: str,
max_retries: int = 5
) -> Dict:
for attempt in range(max_retries):
try:
async with session.get(url) as resp:
if resp.status == 429:
# Tardis limit: 10 requests/second
wait_time = (2 ** attempt) * 0.5 # 0.5s, 1s, 2s, 4s, 8s
print(f"Rate limited. Waiting {wait_time}s...")
await asyncio.sleep(wait_time)
continue
resp.raise_for_status()
return await resp.json()
except aiohttp.ClientError as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
raise Exception("Max retries exceeded")2. Memory Leak khi xử lý large datasets
# ❌ Sai: Cache không giới hạn
class BadCache:
def __init__(self):
self.data = {} # Unbounded growth
def add(self, key, value):
self.data[key] = value # Memory leak!
✅ Đúng: LRU cache với giới hạn kích thước
from functools import lru_cache
from collections import OrderedDict
class LRUCache:
def __init__(self, maxsize: int = 1000):
self.maxsize = maxsize
self.cache = OrderedDict()
def get(self, key: str) -> Optional[Any]:
if key in self.cache:
self.cache.move_to_end(key)
return self.cache[key]
return None
def put(self, key: str, value: Any):
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = value
# Evict oldest if over capacity
if len(self.cache) > self.maxsize:
self.cache.popitem(last=False)
def clear(self):
self.cache.clear()
Streaming approach cho datasets > 1GB
async def stream_process_large_dataset(url: str, chunk_handler):
"""Process data in chunks để tránh memory overflow"""
async with session.get(url) as resp:
async for chunk in resp.content.iter_chunked(64 * 1024): # 64KB chunks
data = msgpack.unpackb(chunk)
await chunk_handler(data)
# Garbage collect sau mỗi chunk
del data3. Look-ahead Bias trong Backtest
# ❌ Sai: Sử dụng future data trong signal calculation
class BiasedStrategy:
def calculate_signal(self, current_idx, all_snapshots):
# BIAS: Access future data!
future_price = all_snapshots[current_idx + 10].mid_price
return current_price < future_price # Peek ahead!
✅ Đúng: Chỉ sử dụng historical data (lookback buffer)
class UnbiasedStrategy:
def __init__(self, lookback_bars: int = 100):
self.price_history = deque(maxlen=lookback_bars)
def on_new_bar(self, snapshot: OrderBookSnapshot):
# Update history FIRST
self.price_history.append(snapshot.mid_price)
# Calculate signal chỉ từ past data
if len(self.price_history) < 50:
return None
# Rolling statistics chỉ từ history
prices = np.array(self.price_history)
ma = np.mean(prices[-20:]) # Past 20 bars only
current = prices[-1]
return 'buy' if current < ma else 'sell'
Validation: Check cho look-ahead bias
def detect_lookahead_bias(trade_log: List[Dict], full_data: List) -> float:
"""
Scan trade log cho suspicious patterns
Return bias_score (0 = clean, 1 = definitely biased)
"""
bias_count = 0
for trade in trade_log:
trade_time = trade['timestamp']
# Check nếu có data point ngay sau trade
future_data = [d for d in full_data if d.timestamp <= trade_time + 1_000_000]
if len(future_data) > 0 and future_data[-1].timestamp > trade_time:
# Suspicious: price moved in favorable direction right after trade
bias_count += 1
return bias_count / len(trade_log) if trade_log else 0.0Tài nguyên liên quan
Bài viết liên quan