Khoảng 3 giờ sáng, màn hình laptop của tôi đỏ lửa với hàng trăm dòng log lỗi. Tôi đã dành 6 tiếng xây dựng hệ thống backtest chiến lược giao dịch crypto bằng Python, và rồi ConnectionError: timeout after 30s xuất hiện ngay khi gọi API Binance.旗舰级系统还没上线就已经死在了摇篮里.
Bài viết này là tổng kết 2 năm kinh nghiệm thực chiến của tôi — từ những lỗi đau thương nhất đến hệ thống backtest hoàn chỉnh, và cách tích hợp AI để tối ưu chiến lược. Đặc biệt, tôi sẽ so sánh giải pháp HolySheep AI với các đối thủ quốc tế để bạn tiết kiệm 85%+ chi phí API.
Tại Sao K-line Data接入Là Bước Khó Nhất Trong量化系统?
Khi xây dựng AI量化策略回测系统, 80% thời gian không nằm ở thuật toán mà ở việc xử lý data. Binance cung cấp hàng triệu data point mỗi ngày, nhưng:
- Rate Limit khắc nghiệt: 1200 requests/phút cho kết nối IP đơn lẻ
- Dữ liệu thiếu sót: Gap trong lịch sử, missing candles khi market đóng cửa
- Định dạng phức tạp: Timestamp tính bằng milliseconds, OHLCV structure khác biệt
- Chi phí API: Nếu dùng các API chuẩn quốc tế, chi phí có thể lên đến $200-500/tháng
Kiến Trúc Hệ Thống Hoàn Chỉnh
Trước khi vào code, hãy xem kiến trúc tổng thể:
┌─────────────────────────────────────────────────────────────────┐
│ AI量化策略回测系统架构 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Binance │────▶│ Data │────▶│ Backtest │ │
│ │ K-line API │ │ Pipeline │ │ Engine │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Cache │ │ AI Strategy│ │
│ │ Layer │ │ Optimizer │ │
│ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ HolySheep │ │
│ │ AI API │ │
│ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘Code Thực Chiến: Data Fetcher Service
Đây là module core mà tôi đã viết lại 7 lần cho đến khi nó hoạt động hoàn hảo:
#!/usr/bin/env python3
"""
Binance K-line Data Fetcher - Phiên bản Production
Tác giả: HolySheep AI Team
Phiên bản: 2.1.0
"""
import requests
import time
import sqlite3
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BinanceKlineFetcher:
"""Fetch K-line data từ Binance với retry mechanism và cache"""
BASE_URL = "https://api.binance.com/api/v3"
RATE_LIMIT = 1200 # requests per minute
RETRY_ATTEMPTS = 3
RETRY_DELAY = 2 # seconds
def __init__(self, db_path: str = "kline_cache.db"):
self.db_path = db_path
self._init_database()
self.last_request_time = 0
self.request_count = 0
self.hour_window_start = time.time()
def _init_database(self):
"""Khởi tạo SQLite cache database"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS klines (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
interval TEXT NOT NULL,
open_time INTEGER NOT NULL,
open REAL,
high REAL,
low REAL,
close REAL,
volume REAL,
close_time INTEGER,
quote_volume REAL,
trades INTEGER,
UNIQUE(symbol, interval, open_time)
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_klines_lookup
ON klines(symbol, interval, open_time)
""")
conn.commit()
conn.close()
logger.info(f"Database initialized: {self.db_path}")
def _rate_limit_check(self):
"""Đảm bảo không vượt quá rate limit của Binance"""
current_time = time.time()
# Reset counter mỗi giờ
if current_time - self.hour_window_start >= 3600:
self.request_count = 0
self.hour_window_start = current_time
# Chờ nếu gần đạt limit
if self.request_count >= self.RATE_LIMIT * 0.9:
wait_time = 3600 - (current_time - self.hour_window_start)
if wait_time > 0:
logger.warning(f"Rate limit warning: waiting {wait_time:.1f}s")
time.sleep(min(wait_time, 60))
self.request_count += 1
def fetch_klines(
self,
symbol: str,
interval: str = "1h",
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 1000
) -> List[Dict]:
"""
Fetch K-line data từ Binance với automatic retry
Args:
symbol: Ví dụ 'BTCUSDT', 'ETHUSDT'
interval: '1m', '5m', '15m', '1h', '4h', '1d', '1w'
start_time: Timestamp in milliseconds
end_time: Timestamp in milliseconds
limit: Số lượng candles (max 1000)
Returns:
List chứa dictionary với OHLCV data
"""
endpoint = f"{self.BASE_URL}/klines"
params = {
"symbol": symbol.upper(),
"interval": interval,
"limit": limit
}
if start_time:
params["startTime"] = start_time
if end_time:
params["endTime"] = end_time
for attempt in range(self.RETRY_ATTEMPTS):
try:
self._rate_limit_check()
response = requests.get(endpoint, params=params, timeout=30)
if response.status_code == 429:
# Rate limit exceeded - exponential backoff
wait_time = 2 ** attempt * 10
logger.warning(f"429 Rate Limited, waiting {wait_time}s...")
time.sleep(wait_time)
continue
response.raise_for_status()
data = response.json()
candles = self._parse_klines(data)
self._cache_klines(symbol, interval, candles)
logger.info(f"Fetched {len(candles)} candles for {symbol}/{interval}")
return candles
except requests.exceptions.Timeout:
logger.warning(f"Timeout on attempt {attempt + 1}, retrying...")
time.sleep(self.RETRY_DELAY * (attempt + 1))
except requests.exceptions.ConnectionError as e:
logger.error(f"ConnectionError: {e}")
time.sleep(5 * (attempt + 1))
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise
raise ConnectionError(f"Failed to fetch klines after {self.RETRY_ATTEMPTS} attempts")
def _parse_klines(self, raw_data: List) -> List[Dict]:
"""Parse Binance API response thành structured format"""
candles = []
for k in raw_data:
candles.append({
"open_time": k[0],
"open": float(k[1]),
"high": float(k[2]),
"low": float(k[3]),
"close": float(k[4]),
"volume": float(k[5]),
"close_time": k[6],
"quote_volume": float(k[7]),
"trades": int(k[8]),
"taker_buy_volume": float(k[9]),
"taker_buy_quote_volume": float(k[10])
})
return candles
def _cache_klines(self, symbol: str, interval: str, candles: List[Dict]):
"""Lưu vào cache database để tránh fetch lại"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
for c in candles:
cursor.execute("""
INSERT OR REPLACE INTO klines
(symbol, interval, open_time, open, high, low, close,
volume, close_time, quote_volume, trades)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
symbol, interval, c["open_time"], c["open"], c["high"],
c["low"], c["close"], c["volume"], c["close_time"],
c["quote_volume"], c["trades"]
))
conn.commit()
conn.close()
def get_historical_data(
self,
symbol: str,
interval: str,
days: int = 365
) -> List[Dict]:
"""
Lấy dữ liệu lịch sử cho một khoảng thời gian dài
Tự động chunk request nếu vượt quá 1000 candles
"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
all_candles = []
current_start = start_time
while current_start < end_time:
candles = self.fetch_klines(
symbol=symbol,
interval=interval,
start_time=current_start,
end_time=end_time,
limit=1000
)
if not candles:
break
all_candles.extend(candles)
# Nếu trả về đủ 1000, có thể còn dữ liệu
if len(candles) == 1000:
current_start = candles[-1]["close_time"] + 1
# Tránh duplicate
all_candles = all_candles[:-1]
else:
break
# Delay nhỏ để tránh trigger rate limit
time.sleep(0.2)
logger.info(f"Total {len(all_candles)} candles retrieved for {symbol}")
return all_candles
Sử dụng
if __name__ == "__main__":
fetcher = BinanceKlineFetcher()
# Fetch dữ liệu 1 năm BTCUSDT khung 1 giờ
data = fetcher.get_historical_data(
symbol="BTCUSDT",
interval="1h",
days=365
)
print(f"Retrieved {len(data)} candles")
print(f"Date range: {datetime.fromtimestamp(data[0]['open_time']/1000)} to {datetime.fromtimestamp(data[-1]['open_time']/1000)}")AI量化策略回测Engine
Bây giờ tôi sẽ xây dựng engine backtest có thể test hàng nghìn chiến lược trong vài phút:
#!/usr/bin/env python3
"""
AI-Powered Backtest Engine - Tích hợp HolySheep AI
Sử dụng: Tạo chiến lược, test, và tối ưu với AI
"""
import json
import sqlite3
from dataclasses import dataclass, field
from typing import List, Dict, Callable, Optional
from datetime import datetime
import numpy as np
from scipy.optimize import differential_evolution
import requests
HolySheep AI Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng API key của bạn
@dataclass
class Trade:
"""Represents a single trade"""
entry_time: int
entry_price: float
exit_time: int
exit_price: float
size: float
side: str # 'long' or 'short'
pnl: float
pnl_pct: float
@dataclass
class BacktestResult:
"""Kết quả backtest"""
total_trades: int = 0
winning_trades: int = 0
losing_trades: int = 0
win_rate: float = 0.0
total_pnl: float = 0.0
total_pnl_pct: float = 0.0
max_drawdown: float = 0.0
sharpe_ratio: float = 0.0
trades: List[Trade] = field(default_factory=list)
def to_dict(self) -> Dict:
return {
"total_trades": self.total_trades,
"winning_trades": self.winning_trades,
"losing_trades": self.losing_trades,
"win_rate": f"{self.win_rate:.2%}",
"total_pnl": f"${self.total_pnl:.2f}",
"total_pnl_pct": f"{self.total_pnl_pct:.2f}%",
"max_drawdown": f"{self.max_drawdown:.2f}%",
"sharpe_ratio": f"{self.sharpe_ratio:.2f}"
}
class Strategy:
"""Base strategy class - implement các phương thức signal() và params()"""
def __init__(self, params: Dict):
self.params = params
def signal(self, candles: List[Dict], current_idx: int) -> Optional[str]:
"""
Trả về 'long', 'short', hoặc None
Override trong subclass
"""
raise NotImplementedError
@staticmethod
def get_param_bounds() -> List[tuple]:
"""Trả về bounds cho mỗi parameter để tối ưu"""
raise NotImplementedError
class MACrossStrategy(Strategy):
"""Moving Average Crossover Strategy"""
def signal(self, candles: List[Dict], current_idx: int) -> Optional[str]:
if current_idx < self.params["fast_period"]:
return None
fast_ma = self._calc_ma(candles, current_idx, self.params["fast_period"])
slow_ma = self._calc_ma(candles, current_idx, self.params["slow_period"])
prev_fast_ma = self._calc_ma(candles, current_idx - 1, self.params["fast_period"])
prev_slow_ma = self._calc_ma(candles, current_idx - 1, self.params["slow_period"])
# Golden Cross - Bullish
if prev_fast_ma <= prev_slow_ma and fast_ma > slow_ma:
return "long"
# Death Cross - Bearish
elif prev_fast_ma >= prev_slow_ma and fast_ma < slow_ma:
return "short"
return None
def _calc_ma(self, candles: List[Dict], idx: int, period: int) -> float:
prices = [c["close"] for c in candles[idx - period + 1:idx + 1]]
return sum(prices) / len(prices)
@staticmethod
def get_param_bounds() -> List[tuple]:
return [(5, 20), (20, 100)] # fast_period, slow_period
class RSIStrategy(Strategy):
"""RSI Mean Reversion Strategy"""
def signal(self, candles: List[Dict], current_idx: int) -> Optional[str]:
if current_idx < self.params["period"]:
return None
rsi = self._calc_rsi(candles, current_idx, self.params["period"])
oversold = self.params["oversold"]
overbought = self.params["overbought"]
if rsi < oversold:
return "long" # Mua khi quá bán
elif rsi > overbought:
return "short" # Bán khi quá mua
return None
def _calc_rsi(self, candles: List[Dict], idx: int, period: int) -> float:
prices = [c["close"] for c in candles[idx - period:idx]]
deltas = [prices[i] - prices[i-1] for i in range(1, len(prices))]
gains = [d if d > 0 else 0 for d in deltas]
losses = [-d if d < 0 else 0 for d in deltas]
avg_gain = sum(gains) / len(gains)
avg_loss = sum(losses) / len(losses)
if avg_loss == 0:
return 100
rs = avg_gain / avg_loss
return 100 - (100 / (1 + rs))
@staticmethod
def get_param_bounds() -> List[tuple]:
return [(7, 21), (20, 35), (65, 80)] # period, oversold, overbought
class BacktestEngine:
"""Engine thực thi backtest với position management"""
def __init__(
self,
initial_capital: float = 10000,
commission: float = 0.001, # 0.1%
slippage: float = 0.0005 # 0.05%
):
self.initial_capital = initial_capital
self.commission = commission
self.slippage = slippage
self.position = None # {'side': 'long', 'entry_price': x, 'size': y}
def run(self, candles: List[Dict], strategy: Strategy) -> BacktestResult:
"""Chạy backtest trên dữ liệu lịch sử"""
trades = []
equity_curve = [self.initial_capital]
current_capital = self.initial_capital
for i in range(len(candles)):
current_price = candles[i]["close"]
current_time = candles[i]["open_time"]
# Kiểm tra exit nếu đang có position
if self.position:
exit_signal = False
if self.position["side"] == "long":
# Stop loss và take profit
pnl_pct = (current_price - self.position["entry_price"]) / self.position["entry_price"]
if pnl_pct <= -self.position["stop_loss"]:
exit_signal = True
exit_price = current_price * (1 - self.slippage)
elif pnl_pct >= self.position["take_profit"]:
exit_signal = True
exit_price = current_price * (1 - self.slippage)
elif self.position["side"] == "short":
pnl_pct = (self.position["entry_price"] - current_price) / self.position["entry_price"]
if pnl_pct <= -self.position["stop_loss"]:
exit_signal = True
exit_price = current_price * (1 + self.slippage)
elif pnl_pct >= self.position["take_profit"]:
exit_signal = True
exit_price = current_price * (1 + self.slippage)
# Exit by signal
signal = strategy.signal(candles, i)
if signal and signal != self.position["side"]:
exit_signal = True
exit_price = current_price * (1 - self.slippage if signal == "long" else 1 + self.slippage)
if exit_signal:
pnl = self.position["size"] * (exit_price - self.position["entry_price"])
if self.position["side"] == "short":
pnl = -pnl
trade = Trade(
entry_time=self.position["entry_time"],
entry_price=self.position["entry_price"],
exit_time=current_time,
exit_price=exit_price,
size=self.position["size"],
side=self.position["side"],
pnl=pnl - (self.position["size"] * exit_price * self.commission * 2),
pnl_pct=pnl / current_capital * 100
)
trades.append(trade)
current_capital += trade.pnl
self.position = None
# Kiểm tra entry
if not self.position:
signal = strategy.signal(candles, i)
if signal:
entry_price = current_price * (1 + self.slippage if signal == "long" else 1 - self.slippage)
position_size = current_capital * 0.95 # 95% capital per trade
self.position = {
"side": signal,
"entry_price": entry_price,
"entry_time": current_time,
"size": position_size,
"stop_loss": strategy.params.get("stop_loss", 0.02),
"take_profit": strategy.params.get("take_profit", 0.04)
}
equity_curve.append(current_capital)
return self._calculate_metrics(trades, equity_curve)
def _calculate_metrics(self, trades: List[Trade], equity_curve: List[float]) -> BacktestResult:
"""Tính toán các metrics"""
if not trades:
return BacktestResult()
total_trades = len(trades)
winning_trades = len([t for t in trades if t.pnl > 0])
losing_trades = len([t for t in trades if t.pnl <= 0])
total_pnl = sum(t.pnl for t in trades)
total_pnl_pct = sum(t.pnl_pct for t in trades)
# Max Drawdown
peak = equity_curve[0]
max_dd = 0
for equity in equity_curve:
if equity > peak:
peak = equity
dd = (peak - equity) / peak * 100
if dd > max_dd:
max_dd = dd
# Sharpe Ratio (simplified)
returns = [equity_curve[i+1]/equity_curve[i] - 1 for i in range(len(equity_curve)-1)]
if np.std(returns) > 0:
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252)
else:
sharpe = 0
return BacktestResult(
total_trades=total_trades,
winning_trades=winning_trades,
losing_trades=losing_trades,
win_rate=winning_trades/total_trades if total_trades > 0 else 0,
total_pnl=total_pnl,
total_pnl_pct=total_pnl_pct,
max_drawdown=max_dd,
sharpe_ratio=sharpe,
trades=trades
)
class AIStrategyOptimizer:
"""
Sử dụng HolySheep AI để phân tích kết quả backtest
và đề xuất cải tiến chiến lược
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
def analyze_and_suggest(
self,
symbol: str,
strategy_name: str,
backtest_result: BacktestResult,
market_conditions: Dict
) -> str:
"""
Gửi kết quả backtest lên HolySheep AI để phân tích và đề xuất
"""
prompt = f"""Bạn là chuyên gia trading và quantitative analysis.
Hãy phân tích kết quả backtest cho chiến lược {strategy_name} trên {symbol}:
**Kết quả Backtest:**
- Tổng giao dịch: {backtest_result.total_trades}
- Win rate: {backtest_result.win_rate:.2%}
- Tổng P&L: ${backtest_result.total_pnl:.2f} ({backtest_result.total_pnl_pct:.2f}%)
- Max Drawdown: {backtest_result.max_drawdown:.2f}%
- Sharpe Ratio: {backtest_result.sharpe_ratio:.2f}
**Điều kiện thị trường:**
{json.dumps(market_conditions, indent=2)}
**Yêu cầu:**
1. Phân tích điểm mạnh và yếu của chiến lược
2. Đề xuất 3-5 cải tiến cụ thể
3. Đánh giá risk/reward ratio
4. Đề xuất thời điểm nên tạm dừng chiến lược này
Trả lời bằng tiếng Việt, format rõ ràng."""
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4o",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 2000
},
timeout=30
)
if response.status_code == 200:
result = response.json()
return result["choices"][0]["message"]["content"]
else:
return f"Lỗi API: {response.status_code} - {response.text}"
except requests.exceptions.Timeout:
return "Timeout khi gọi AI. Vui lòng thử lại."
except Exception as e:
return f"Lỗi: {str(e)}"
def optimize_strategy(
candles: List[Dict],
strategy_class: type,
param_bounds: List[tuple],
initial_capital: float = 10000
) -> Dict:
"""
Tối ưu hóa parameters sử dụng Differential Evolution
"""
engine = BacktestEngine(initial_capital=initial_capital)
def objective(params):
strategy = strategy_class(dict(zip(["fast_period", "slow_period"], map(int, params))))
result = engine.run(candles, strategy)
# Objective: minimize (negative Sharpe + drawdown penalty)
return -result.sharpe_ratio + result.max_drawdown * 0.1
result = differential_evolution(
objective,
param_bounds,
maxiter=50,
seed=42,
workers=1
)
return {
"optimal_params": result.x,
"best_sharpe": -result.fun
}
Sử dụng mẫu
if __name__ == "__main__":
# Khởi tạo fetcher để lấy dữ liệu
from binance_fetcher import BinanceKlineFetcher
fetcher = BinanceKlineFetcher()
candles = fetcher.get_historical_data("BTCUSDT", "1h", days=180)
# Chạy backtest với MA Cross strategy
strategy = MACrossStrategy({
"fast_period": 10,
"slow_period": 50,
"stop_loss": 0.02,
"take_profit": 0.04
})
engine = BacktestEngine(initial_capital=10000)
result = engine.run(candles, strategy)
print("=== BACKTEST RESULTS ===")
for key, value in result.to_dict().items():
print(f"{key}: {value}")
# Tối ưu với AI
ai_optimizer = AIStrategyOptimizer(HOLYSHEEP_API_KEY)
market_data = {
"trend": "sideways",
"volatility": "medium",
"volume": "increasing"
}
suggestion = ai_optimizer.analyze_and_suggest(
symbol="BTCUSDT",
strategy_name="MA Cross",
backtest_result=result,
market_conditions=market_data
)
print("\n=== AI SUGGESTIONS ===")
print(suggestion)Lỗi thường gặp và cách khắc phục
1. Lỗi 401 Unauthorized khi gọi API
Mô tả lỗi: Khi bạn nhận được response với status code 401 hoặc thông báo "Invalid API key", đây là vấn đề authentication phổ biến nhất.
# ❌ SAI - Key bị ẩn hoặc format sai
headers = {
"Authorization": "Bearer YOUR_API_KEY", # Không có khoảng trắng
"X-MBX-APIKEY": api_key
}
✅ ĐÚNG
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
Verify key format
import re
if not re.match(r'^[a-zA-Z0-9]{64}$', api_key):
print("API Key format không đúng!")
print("Key phải có 64 ký tự alphanumeric")2. Lỗi Connection Reset by Peer / ConnectionTimeout
Mô tả: Binance có rate limit rất nghiêm ngặt. Khi vượt quá, server sẽ reset connection hoặc timeout.
# ❌ KHÔNG NÊN - Gọi API liên tục không có delay
for i in range(10000):
candles = fetcher.fetch_klines(symbol, limit=1000)
# Sẽ bị 429 Rate Limit ngay!
✅ NÊN LÀM - Implement exponential backoff
import time
from functools import wraps
def retry_with_backoff(max_retries=5, initial_delay=1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except (ConnectionError, TimeoutError) as e:
if attempt == max_retries - 1:
raise
wait_time = delay * (2 ** attempt)
print(f"Attempt {attempt+1} failed, waiting {wait_time}s...")
time.sleep(wait_time)
return None
return wrapper
return decorator
@retry_with_backoff(max_retries=5, initial_delay=2)
def safe_fetch_klines(symbol, interval):
response = requests.get(endpoint, timeout=30)
response.raise_for_status()
return response.json()3. Lỗi Missing Data / Incomplete Candles
Mô tả: Dữ liệu K-line bị thiếu, đặc biệt khi fetch dữ liệu lịch sử dài. Điều này gây sai lệch trong backtest nghiêm trọng.
# ✅ Implement data validation và gap filling
def validate_and_fill_gaps(candles: List[Dict], interval: str) -> List[Dict]:
"""Validate và điền các gap trong dữ liệu"""
# Interval mapping sang milliseconds
interval_ms = {
"1m": 60000, "5m": 300000, "15m": 900000,
"1h": 3600000, "4h": 14400000, "1d": 86400000
}
interval_duration = interval_ms.get(interval, 3600000)
validated = []
for i in range(len(candles)):
expected_time = candles[i]["open_time"]
# Kiểm tra gap với candle trước
if i > 0:
prev_time = candles[i-1]["close_time"]
expected_gap = expected_time - prev_time
if expected_gap > interval_duration:
# Có gap - tạo dummy candles hoặc báo warning
print(f"WARNING: Gap detected from {prev_time} to {expected_time}")
print(f"Missing {expected_gap // interval_duration - 1} candles")
# Option 1: Skip gap (conservative)
# continue
# Option 2: Fill with forward fill (aggressive)
# for gap_idx in range(1, expected_gap // interval_duration):
# gap_time = prev_time + interval_duration * gap_idx
# validated.append({
# **candles[i-1],
# "open_time": gap_time,
# "close_time": gap_time + interval_duration,
# "is_gap_filled": True
# })
validated.append(candles[i])
return validated
Sử dụng
candles = fetcher.get_historical_data("BTCUSDT", "1h", days=365)
clean_candles = validate_and_fill_gaps(candles, "1h")
print(f"Original: {len(candles)}, After validation: {len(clean_candles)}")