Kịch bản lỗi thực tế: ConnectionError: HTTPSConnectionPool(host='ws-api.kaiko.com', port=443): Max retries exceeded — Đây là lỗi mà tôi đã gặp phải vào tháng 3 năm 2024 khi đang chạy backtest chiến lược arbitrage trên 47 cặp giao dịch spot. Request timeout sau 30 giây, toàn bộ pipeline bị treo, và 3 tiếng dữ liệu lịch sử bị mất do không có error handling đúng cách. Bài viết này sẽ giúp bạn tránh những sai lầm tương tự.
Mục lục
- Kaiko Data API là gì và tại sao quan trọng với risk management
- Kiến trúc hệ thống backtesting tích hợp
- Hướng dẫn cài đặt và authentication
- Code mẫu Python — chi tiết từng bước
- Chiến lược risk management với dữ liệu Kaiko
- Lỗi thường gặp và cách khắc phục
- So sánh với giải pháp thay thế
- Kết luận và khuyến nghị
Kaiko Data API là gì
Kaiko là nhà cung cấp dữ liệu tiền mã hóa institutional-grade, cung cấp:
- Order book data với độ sâu 50 cấp
- Trade data với latency dưới 100ms
- OHLCV data từ 200+ sàn giao dịch
- Reference rate cho valuation
- Index data cho futures và perpetual
Với risk management, dữ liệu Kaiko cho phép:
- Tính toán VaR (Value at Risk) chính xác
- Stress testing với dữ liệu thực tế
- Backtesting chiến lược với slippage thực
- Cross-exchange arbitrage detection
Kiến trúc hệ thống Backtesting
┌─────────────────────────────────────────────────────────────────┐
│ BACKTESTING ARCHITECTURE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Kaiko API │────▶│ Data Layer │────▶│ Engine │ │
│ │ (Raw Data) │ │ (Cleaned) │ │ (Backtest) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Rate Limiter │ │ PostgreSQL │ │ Risk Engine │ │
│ │ + Retry │ │ (Storage) │ │ (VaR/Slip) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Report/ │ │
│ │ Dashboard │ │
│ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Cài đặt và Authentication
Yêu cầu ban đầu
# Python 3.10+
pip install kaiko-python pandas numpy psycopg2-binary aiohttp
pip install backtesting matplotlib scipy
pip install python-dotenv redis # Cho caching
Cấu hình API Key
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
Kaiko Configuration
KAIKO_API_KEY = os.getenv('KAIKO_API_KEY')
KAIKO_BASE_URL = 'https://ws-api.kaiko.com'
KAIKO_REST_URL = 'https://market-data.kaiko.io'
Rate Limiting
MAX_REQUESTS_PER_SECOND = 10
REQUEST_TIMEOUT = 30
Database Configuration
DB_CONFIG = {
'host': 'localhost',
'port': 5432,
'database': 'crypto_backtest',
'user': os.getenv('DB_USER'),
'password': os.getenv('DB_PASSWORD')
}
Risk Management Parameters
MAX_POSITION_SIZE = 0.02 # 2% max position
MAX_DRAWDOWN = 0.15 # 15% max drawdown
CONFIDENCE_LEVEL = 0.95 # VaR confidence level
Code mẫu Python — Chi tiết từng bước
Bước 1: Data Fetcher với Error Handling
# data_fetcher.py
import asyncio
import aiohttp
import time
from typing import Optional, Dict, List
from datetime import datetime, timedelta
import pandas as pd
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class KaikoDataFetcher:
"""Fetcher dữ liệu từ Kaiko API với retry logic và rate limiting"""
def __init__(self, api_key: str, base_url: str = 'https://market-data.kaiko.io'):
self.api_key = api_key
self.base_url = base_url
self.session: Optional[aiohttp.ClientSession] = None
self.last_request_time = 0
self.min_request_interval = 0.1 # 10 requests/second
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
'X-API-Key': self.api_key,
'Accept': 'application/json'
},
timeout=aiohttp.ClientTimeout(total=30)
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def _rate_limit(self):
"""Đảm bảo không vượt quá rate limit"""
current_time = time.time()
time_since_last = current_time - self.last_request_time
if time_since_last < self.min_request_interval:
await asyncio.sleep(self.min_request_interval - time_since_last)
self.last_request_time = time.time()
async def _request_with_retry(
self,
url: str,
max_retries: int = 3,
backoff_factor: float = 1.5
) -> Optional[Dict]:
"""Request với exponential backoff retry"""
for attempt in range(max_retries):
try:
await self._rate_limit()
async with self.session.get(url) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
# Rate limited - đợi lâu hơn
wait_time = backoff_factor ** attempt * 2
logger.warning(f"Rate limited, waiting {wait_time}s")
await asyncio.sleep(wait_time)
elif response.status == 401:
logger.error("Invalid API key!")
raise PermissionError("Invalid Kaiko API key")
elif response.status == 404:
logger.warning(f"Data not found: {url}")
return None
else:
logger.error(f"HTTP {response.status}: {await response.text()}")
except aiohttp.ClientError as e:
logger.warning(f"Attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
await asyncio.sleep(backoff_factor ** attempt)
else:
logger.error(f"All retries exhausted for {url}")
raise
return None
async def get_ohlcv(
self,
instrument: str,
interval: str = '1m',
start_time: datetime = None,
end_time: datetime = None
) -> pd.DataFrame:
"""
Lấy OHLCV data cho một instrument
Args:
instrument: VD 'btc-usd-spot' hoặc 'eth-usd-perpetual'
interval: '1m', '5m', '1h', '1d'
start_time: Thời gian bắt đầu
end_time: Thời gian kết thúc
"""
if not start_time:
start_time = datetime.now() - timedelta(days=7)
if not end_time:
end_time = datetime.now()
url = (
f"{self.base_url}/v2/data/trades.v1/"
f"spot/{instrument}/ohlcv/{interval}"
f"?start_time={start_time.isoformat()}"
f"&end_time={end_time.isoformat()}"
)
data = await self._request_with_retry(url)
if not data or 'data' not in data:
return pd.DataFrame()
df = pd.DataFrame(data['data'])
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.sort_values('timestamp')
return df
async def get_orderbook(
self,
instrument: str,
depth: int = 50
) -> Dict:
"""Lấy order book data với độ sâu chỉ định"""
url = (
f"{self.base_url}/v2/data/orderbooks.v1/"
f"spot/{instrument}/latest"
f"?depth={depth}"
)
return await self._request_with_retry(url)
async def get_all_instruments(self) -> List[Dict]:
"""Lấy danh sách tất cả instruments"""
url = f"{self.base_url}/v2/data/instruments.v1"
data = await self._request_with_retry(url)
return data.get('data', []) if data else []
Sử dụng
async def main():
async with KaikoDataFetcher(api_key='your-kaiko-key') as fetcher:
# Lấy 1 ngày dữ liệu BTC
btc_data = await fetcher.get_ohlcv(
instrument='btc-usd-spot',
interval='1h',
start_time=datetime.now() - timedelta(days=1)
)
print(f"Fetched {len(btc_data)} candles")
# Lấy order book
ob = await fetcher.get_orderbook('eth-usd-spot')
print(f"Best bid: {ob['data']['bids'][0]}")
asyncio.run(main())
Bước 2: Risk Management Engine
# risk_engine.py
import pandas as pd
import numpy as np
from typing import Dict, Tuple, Optional
from dataclasses import dataclass
from scipy import stats
@dataclass
class Position:
symbol: str
size: float
entry_price: float
current_price: float
@property
def pnl(self) -> float:
return (self.current_price - self.entry_price) * self.size
@property
def pnl_percent(self) -> float:
return (self.current_price / self.entry_price - 1) * 100
class RiskManager:
"""Engine quản lý rủi ro cho backtesting"""
def __init__(
self,
max_position_size: float = 0.02,
max_portfolio_exposure: float = 0.10,
max_drawdown: float = 0.15,
confidence_level: float = 0.95
):
self.max_position_size = max_position_size
self.max_portfolio_exposure = max_portfolio_exposure
self.max_drawdown = max_drawdown
self.confidence_level = confidence_level
self.equity_curve = []
self.peak_equity = float('inf')
self.current_drawdown = 0
def calculate_var_historical(
self,
returns: pd.Series,
portfolio_value: float
) -> float:
"""
Tính Value at Risk sử dụng phương pháp historical simulation
VaR = portfolio_value * percentile(returns, 1 - confidence)
"""
if len(returns) < 30:
return portfolio_value * 0.05 # Default 5% nếu không đủ data
var_percentile = (1 - self.confidence_level) * 100
var = np.percentile(returns, var_percentile)
return abs(portfolio_value * var)
def calculate_cvar(
self,
returns: pd.Series,
portfolio_value: float
) -> float:
"""
Conditional VaR (Expected Shortfall) - Trung bình các lỗ vượt VaR
CVaR cung cấp ước tính rủi ro tốt hơn VaR
"""
var_percentile = (1 - self.confidence_level) * 100
var = np.percentile(returns, var_percentile)
tail_returns = returns[returns <= var]
if len(tail_returns) == 0:
return self.calculate_var_historical(returns, portfolio_value)
return abs(portfolio_value * tail_returns.mean())
def calculate_sharpe_ratio(
self,
returns: pd.Series,
risk_free_rate: float = 0.0
) -> float:
"""Tính Sharpe Ratio"""
if returns.std() == 0:
return 0.0
excess_returns = returns - risk_free_rate / 252
return np.sqrt(252) * excess_returns.mean() / returns.std()
def calculate_max_drawdown(self, equity_curve: list) -> Tuple[float, int, int]:
"""
Tính maximum drawdown và thời gian phục hồi
Returns: (max_dd, peak_index, trough_index)
"""
if not equity_curve:
return 0.0, 0, 0
equity = pd.Series(equity_curve)
running_max = equity.expanding().max()
drawdown = (equity - running_max) / running_max
max_dd = drawdown.min()
trough_idx = drawdown.idxmin()
# Tìm peak trước trough
peak_idx = equity[:trough_idx].idxmax()
return abs(max_dd), int(peak_idx), int(trough_idx)
def check_position_limits(
self,
proposed_size: float,
current_exposure: float,
portfolio_value: float
) -> Tuple[bool, str]:
"""Kiểm tra giới hạn position size"""
position_value = proposed_size * portfolio_value
if proposed_size > self.max_position_size:
return False, f"Position size {proposed_size:.2%} exceeds max {self.max_position_size:.2%}"
if current_exposure + proposed_size > self.max_portfolio_exposure:
return False, f"Total exposure would exceed max {self.max_portfolio_exposure:.2%}"
if self.current_drawdown > self.max_drawdown:
return False, f"Drawdown {self.current_drawdown:.2%} exceeds limit {self.max_drawdown:.2%}"
return True, "OK"
def update_equity(self, new_equity: float):
"""Cập nhật equity curve và tính drawdown"""
self.equity_curve.append(new_equity)
if new_equity < self.peak_equity:
self.current_drawdown = (self.peak_equity - new_equity) / self.peak_equity
else:
self.peak_equity = new_equity
self.current_drawdown = 0
def get_risk_report(self, returns: pd.Series, portfolio_value: float) -> Dict:
"""Generate báo cáo rủi ro đầy đủ"""
return {
'VaR (95%)': self.calculate_var_historical(returns, portfolio_value),
'CVaR (95%)': self.calculate_cvar(returns, portfolio_value),
'Sharpe Ratio': self.calculate_sharpe_ratio(returns),
'Max Drawdown': self.calculate_max_drawdown(self.equity_curve)[0],
'Current Drawdown': self.current_drawdown,
'Volatility (annualized)': returns.std() * np.sqrt(252),
'Sortino Ratio': self._calculate_sortino(returns),
'Win Rate': (returns > 0).mean(),
'Avg Win': returns[returns > 0].mean() if (returns > 0).any() else 0,
'Avg Loss': returns[returns < 0].mean() if (returns < 0).any() else 0
}
def _calculate_sortino(self, returns: pd.Series) -> float:
"""Tính Sortino Ratio - chỉ tính downside deviation"""
downside_returns = returns[returns < 0]
if len(downside_returns) == 0 or downside_returns.std() == 0:
return 0.0
return np.sqrt(252) * returns.mean() / downside_returns.std()
Ví dụ sử dụng
if __name__ == '__main__':
# Tạo dummy returns data
np.random.seed(42)
returns = pd.Series(np.random.randn(252) * 0.02 + 0.0005)
risk_manager = RiskManager(
max_position_size=0.02,
max_drawdown=0.15,
confidence_level=0.95
)
# Simulate equity curve
equity = 100000
for ret in returns:
equity *= (1 + ret)
risk_manager.update_equity(equity)
# Generate report
report = risk_manager.get_risk_report(returns, 100000)
print("=" * 50)
print("RISK MANAGEMENT REPORT")
print("=" * 50)
for key, value in report.items():
if isinstance(value, float):
print(f"{key}: {value:,.2f}")
else:
print(f"{key}: {value:.2%}")
Bước 3: Backtesting Engine hoàn chỉnh
# backtester.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from risk_engine import RiskManager
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class Trade:
timestamp: datetime
symbol: str
side: str # 'buy' or 'sell'
price: float
size: float
fee: float = 0.0
@dataclass
class BacktestResult:
initial_capital: float
final_capital: float
total_trades: int
winning_trades: int
losing_trades: int
max_drawdown: float
sharpe_ratio: float
sortino_ratio: float
returns: pd.Series = field(default_factory=pd.Series)
trades: List[Trade] = field(default_factory=list)
equity_curve: List[float] = field(default_factory=list)
class CryptoBacktester:
"""
Backtesting engine cho chiến lược crypto
Tích hợp với Kaiko data và Risk Manager
"""
def __init__(
self,
initial_capital: float = 100000,
fee_rate: float = 0.001,
slippage: float = 0.0005,
risk_manager: Optional[RiskManager] = None
):
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.fee_rate = fee_rate
self.slippage = slippage
self.risk_manager = risk_manager or RiskManager()
self.positions: Dict[str, float] = {}
self.trades: List[Trade] = []
self.equity_curve: List[float] = []
self.returns: List[float] = []
self.data: Dict[str, pd.DataFrame] = {}
def load_data(self, symbol: str, data: pd.DataFrame):
"""Load dữ liệu cho backtesting"""
self.data[symbol] = data.sort_values('timestamp').reset_index(drop=True)
logger.info(f"Loaded {len(data)} rows for {symbol}")
def _get_slippage_price(self, price: float, side: str) -> float:
"""Tính giá sau slippage"""
if side == 'buy':
return price * (1 + self.slippage)
else:
return price * (1 - self.slippage)
def _execute_trade(
self,
symbol: str,
side: str,
price: float,
size: float,
timestamp: datetime
):
"""Thực thi trade với fee và slippage"""
exec_price = self._get_slippage_price(price, side)
fee = exec_price * size * self.fee_rate
total_cost = exec_price * size + fee if side == 'buy' else exec_price * size - fee
# Kiểm tra margin
if side == 'buy' and total_cost > self.current_capital:
logger.warning(f"Insufficient capital for {symbol} {side}")
return False
# Cập nhật position
if side == 'buy':
self.positions[symbol] = self.positions.get(symbol, 0) + size
self.current_capital -= total_cost
else:
self.positions[symbol] = self.positions.get(symbol, 0) - size
self.current_capital += total_cost
# Ghi nhận trade
self.trades.append(Trade(
timestamp=timestamp,
symbol=symbol,
side=side,
price=exec_price,
size=size,
fee=fee
))
logger.debug(f"{timestamp}: {side.upper()} {size} {symbol} @ {exec_price:.2f}")
return True
def run(
self,
strategy_func: Callable,
symbols: List[str],
start_date: datetime,
end_date: datetime
) -> BacktestResult:
"""
Chạy backtest với strategy function
strategy_func(data_dict, current_time, positions) -> List of signals
"""
logger.info(f"Starting backtest: {start_date} to {end_date}")
# Reset state
self.current_capital = self.initial_capital
self.positions = {}
self.trades = []
self.equity_curve = []
self.returns = []
# Merge all data timestamps
all_timestamps = set()
for symbol in symbols:
if symbol in self.data:
mask = (
(self.data[symbol]['timestamp'] >= start_date) &
(self.data[symbol]['timestamp'] <= end_date)
)
all_timestamps.update(
self.data[symbol][mask]['timestamp'].tolist()
)
sorted_timestamps = sorted(all_timestamps)
for i, ts in enumerate(sorted_timestamps):
# Lấy data hiện tại cho mỗi symbol
current_data = {}
for symbol in symbols:
if symbol in self.data:
df = self.data[symbol]
current_row = df[df['timestamp'] == ts]
if not current_row.empty:
current_data[symbol] = current_row.iloc[0]
if not current_data:
continue
# Tính current equity
position_value = 0
for symbol, size in self.positions.items():
if symbol in current_data:
position_value += size * current_data[symbol]['close']
total_equity = self.current_capital + position_value
self.equity_curve.append(total_equity)
# Update risk manager
self.risk_manager.update_equity(total_equity)
if i > 0:
daily_return = (total_equity / self.equity_curve[-2]) - 1
self.returns.append(daily_return)
# Gọi strategy để lấy signals
signals = strategy_func(
data=current_data,
timestamp=ts,
positions=self.positions,
capital=self.current_capital,
equity=total_equity
)
# Execute signals
if signals:
for signal in signals:
self._execute_trade(
symbol=signal['symbol'],
side=signal['side'],
price=current_data[signal['symbol']]['close'],
size=signal['size'],
timestamp=ts
)
# Tính metrics
returns_series = pd.Series(self.returns)
winning_trades = [t for t in self.trades if t.side == 'sell' and t.size > 0]
losing_trades = [t for t in self.trades if t.side == 'sell' and t.size < 0]
max_dd, _, _ = self.risk_manager.calculate_max_drawdown(self.equity_curve)
return BacktestResult(
initial_capital=self.initial_capital,
final_capital=self.equity_curve[-1] if self.equity_curve else self.initial_capital,
total_trades=len(self.trades),
winning_trades=len(winning_trades),
losing_trades=len(losing_trades),
max_drawdown=max_dd,
sharpe_ratio=self.risk_manager.calculate_sharpe_ratio(returns_series),
sortino_ratio=self.risk_manager._calculate_sortino(returns_series),
returns=returns_series,
trades=self.trades,
equity_curve=self.equity_curve
)
Ví dụ strategy đơn giản
def simple_momentum_strategy(data, timestamp, positions, capital, equity, **kwargs):
"""Chiến lược momentum đơn giản"""
signals = []
for symbol, row in data.items():
# Buy signal: SMA crossover
if row.get('close', 0) > row.get('sma_20', 0):
# Kiểm tra position hiện tại
if positions.get(symbol, 0) == 0:
signals.append({
'symbol': symbol,
'side': 'buy',
'size': 0.01 # 1% capital
})
# Sell signal: SMA crossover down
elif row.get('close', 0) < row.get('sma_20', 0):
if positions.get(symbol, 0) > 0:
signals.append({
'symbol': symbol,
'side': 'sell',
'size': positions[symbol]
})
return signals
Lỗi thường gặp và cách khắc phục
Lỗi 1: 401 Unauthorized - Invalid API Key
# ❌ SAI - Key không đúng format hoặc hết hạn
KAIKO_API_KEY = "invalid-key-format"
✅ ĐÚNG - Kiểm tra format và validate key
def validate_kaiko_key(api_key: str) -> bool:
"""Validate Kaiko API key format"""
if not api_key:
return False
# Kaiko key thường có format: sk_live_xxxxx hoặc sk_test_xxxxx
if not api_key.startswith(('sk_live_', 'sk_test_')):
logger.error("Invalid key format. Expected sk_live_ or sk_test_")
return False
# Test bằng cách call một endpoint đơn giản
import requests
response = requests.get(
'https://market-data.kaiko.io/v2/data/instruments.v1',
headers={'X-API-Key': api_key}
)
if response.status_code == 401:
logger.error("API key is invalid or expired")
return False
return True
Sử dụng
if not validate_kaiko_key(os.getenv('KAIKO_API_KEY')):
raise ValueError("Please provide a valid Kaiko API key")
Lỗi 2: Connection Timeout - Rate Limiting
# ❌ SAI - Không có rate limiting, gửi quá nhiều request
async def bad_fetch():
tasks = []
for symbol in symbols: # 200 symbols
tasks.append(fetch_ohlcv(symbol)) # Gửi tất cả cùng lúc!
return await asyncio.gather(*tasks)
✅ ĐÚNG - Implement semaphore và token bucket
import asyncio
import time
from collections import deque
class TokenBucket:
"""Token bucket algorithm cho rate limiting"""
def __init__(self, rate: float, capacity: int):
self.rate = rate # tokens per second
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
async def acquire(self, tokens: int = 1):
while True:
now = time.time()
elapsed = now - self.last_update
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.rate
)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return
wait_time = (tokens - self.tokens) / self.rate
await asyncio.sleep(wait_time)
class RateLimitedFetcher:
def __init__(self, rate: float = 10, burst: int = 20):
self.bucket = TokenBucket(rate, burst)
self.semaphore = asyncio.Semaphore(5) # Max 5 concurrent
async def fetch(self, url: str, session: aiohttp.ClientSession):
async with self.semaphore: # Giới hạn concurrent
await self.bucket.acquire() # Giới hạn rate
async with session.get(url) as response:
return await response.json()
Sử dụng
async def good_fetch(symbols: List[str]):
fetcher = RateLimitedFetcher(rate=10, burst=20)
async with aiohttp.ClientSession() as session:
tasks = [fetcher.fetch(url, session) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Xử lý errors riêng
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(f"Request {i} failed: {result}")
return [r for r in results if not isinstance(r, Exception)]
Lỗi 3: Data Quality - Missing Candles
# ❌ SAI - Giả định data luôn đầy đủ
df = pd.DataFrame(data['data'])
Không kiểm tra gaps -> phân tích sai
✅ ĐÚNG - Phát hiện và xử lý missing data
def validate_and_fill_ohlcv(
df: pd.DataFrame,
interval: str = '1h',
max_gap_hours: int = 4
) -> pd.DataFrame:
"""
Validate OHLCV data và fill missing candles
"""
df = df.copy()
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.sort_values('timestamp')
# Tạo complete date range
if len(df) > 0:
expected_interval = pd.Timedelta(interval)
full_range = pd.date_range(
start=df['timestamp'].min(),
end=df['timestamp'].max(),
freq=expected_interval
)
# Tìm missing timestamps
missing = full_range.difference(df['timestamp'])
if len(missing) > 0:
logger.warning(f"Found {len(missing)} missing candles")
# Fill method: forward fill hoặc interpolation
df = df.set_index('timestamp')
# Resample và forward fill
df = df.reindex(full_range)
df = df.ffill()
# Với OHLCV, có thể dùng interpolation
numeric_cols = ['open', 'high', 'low', 'close', 'volume']
for col in numeric_cols:
if col in df.columns:
df[col] = df[col].interpolate(method='linear')
df = df.reset_index().rename(columns={'index': 'timestamp'})
# Đánh dấu filled data
df['is_filled'] = df['timestamp'].isin(missing)
# Validate columns
required = ['open', 'high', 'low', 'close', 'volume']
for col in required:
if col not in df.columns:
raise ValueError(f"Missing required column: {col}")
# Check cho NaN
nan_count = df[col].isna().sum()
if nan_count > 0:
logger.warning(f"Column {col} has {nan_count} NaN values")
return df
Kiểm tra data quality trước backtest
def check_data_quality(df: pd.DataFrame, symbol: str) -> Dict:
"""Kiểm tra chất lượng data trước khi backtest"""
issues = []
# Check 1: Tỷ lệ NaN
nan_ratio = df.isna().mean()
if nan_ratio.max() > 0.05:
issues.append(f"High NaN ratio: {nan_ratio.max():.2%}")
# Check 2: Outliers trong returns
if 'close' in df.columns:
returns = df['close'].pct_change()