Trong lĩnh vực trading crypto, việc dự đoán chính xác funding rate (phí funding) là một trong những thách thức phức tạp nhất. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống feature engineering hoàn chỉnh để dự đoán funding rate, từ việc thu thập dữ liệu, xử lý features, cho đến huấn luyện mô hình machine learning.

Kịch bản lỗi thực tế

Tôi vẫn nhớ rõ ngày đầu tiên triển khai hệ thống dự đoán funding rate cho một sàn futures phổ biến. Mọi thứ看似 hoàn hảo cho đến khi logs bắt đầu xuất hiện một loạt lỗi:

2024-11-15 03:24:15 | ERROR | FundingRateCollector
ConnectionError: HTTPSConnectionPool(host='api.binance.com', port=443): 
Max retries exceeded with url: /fapi/v1/premiumIndex (Caused by 
ConnectTimeoutError: <Connection(HTTPConnectionPool(host='api.binance.com', 
port=443): Connect timeout: 5 seconds)>)

2024-11-15 03:24:15 | WARNING | Rate limit exceeded. Retrying in 300 seconds...
2024-11-15 03:24:15 | CRITICAL | Data gap detected: 15 minutes of missing funding data

Không chỉ có timeout, mà hệ thống còn gặp phải vấn đề về rate limiting và xử lý dữ liệu thiếu liên tục. Sau 3 ngày debug liên tục, tôi đã tìm ra giải pháp toàn diện và chia sẻ trong bài viết này.

Tổng quan hệ thống

Hệ thống feature engineering cho funding rate prediction bao gồm 4 module chính:

1. Data Collector - Module thu thập dữ liệu

Việc thu thập dữ liệu funding rate từ nhiều sàn là bước nền tảng. Dưới đây là implementation hoàn chỉnh với xử lý lỗi và retry mechanism:

import requests
import time
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
from threading import Lock
import asyncio

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class FundingRateData:
    symbol: str
    funding_rate: float
    mark_price: float
    index_price: float
    next_funding_time: datetime
    timestamp: datetime

class FundingRateCollector:
    """
    Collector for funding rate data from multiple exchanges.
    Handles rate limiting, retries, and data validation.
    """
    
    def __init__(self, api_base_url: str, api_key: str):
        self.api_base_url = api_base_url
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-KEY': api_key,
            'Content-Type': 'application/json'
        })
        self.rate_limit_lock = Lock()
        self.last_request_time = {}
        self.min_request_interval = 0.1  # 100ms between requests
        
        # Retry configuration
        self.max_retries = 3
        self.retry_delay = 5  # seconds
        self.timeout = 10  # seconds
        
    def _rate_limit(self, endpoint: str):
        """Enforce rate limiting per endpoint"""
        with self.rate_limit_lock:
            current_time = time.time()
            if endpoint in self.last_request_time:
                elapsed = current_time - self.last_request_time[endpoint]
                if elapsed < self.min_request_interval:
                    time.sleep(self.min_request_interval - elapsed)
            self.last_request_time[endpoint] = time.time()
    
    def _make_request(self, url: str, params: Optional[Dict] = None) -> Dict:
        """Make HTTP request with retry logic"""
        for attempt in range(self.max_retries):
            try:
                self._rate_limit(url)
                response = self.session.get(
                    url, 
                    params=params, 
                    timeout=self.timeout
                )
                
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    logger.warning(f"Rate limit hit. Retrying in {self.retry_delay}s...")
                    time.sleep(self.retry_delay)
                elif response.status_code == 401:
                    raise ValueError("401 Unauthorized: Invalid API key")
                else:
                    raise requests.HTTPError(f"HTTP {response.status_code}: {response.text}")
                    
            except requests.exceptions.Timeout:
                logger.warning(f"Timeout on attempt {attempt + 1}/{self.max_retries}")
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay * (attempt + 1))
            except requests.exceptions.ConnectionError as e:
                logger.warning(f"ConnectionError: {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay * (attempt + 1))
                    
        raise ConnectionError(f"Failed after {self.max_retries} attempts")
    
    def get_funding_rate(self, symbol: str) -> Optional[FundingRateData]:
        """Fetch current funding rate for a symbol"""
        url = f"{self.api_base_url}/fapi/v1/premiumIndex"
        params = {'symbol': symbol}
        
        try:
            data = self._make_request(url, params)
            
            return FundingRateData(
                symbol=symbol,
                funding_rate=float(data['lastFundingRate']) * 100,  # Convert to percentage
                mark_price=float(data['markPrice']),
                index_price=float(data['indexPrice']),
                next_funding_time=datetime.fromtimestamp(
                    data['nextFundingTime'] / 1000
                ),
                timestamp=datetime.now()
            )
        except Exception as e:
            logger.error(f"Error fetching funding rate for {symbol}: {e}")
            return None
    
    def get_funding_history(self, symbol: str, days: int = 30) -> List[Dict]:
        """Fetch historical funding rates"""

Tài nguyên liên quan
Bài viết liên quan