Ngày 28 tháng 4 năm 2026, tôi nhận được một tin nhắn từ anh Minh — một nhà giao dịch quyen tại TP.HCM. Câu chuyện của anh là bài học mà bất kỳ ai làm việc với dữ liệu Bybit đều nên đọc.

“Mình đã build xong backtest engine cho chiến lược iron condor trên quyen Bybit. Tất cả test offline đều hoàn hảo. Nhưng khi cần download 6 tháng dữ liệu OHLCV của 50 cặp quyen để chạy validation thực sự, server liên tục trả về ConnectionError: timeout. Mình đợi 3 tiếng, không có byte nào được ghi xuống disk.”

Anh Minh không phải người duy nhất gặp vấn đề này. Trong bài viết này, tôi sẽ phân tích sâu hai phương án phổ biến nhất để download dữ liệu quyen lịch sử Bybit: Tardis API và WebSocket Replay. Cuối bài, bạn sẽ có đủ thông tin để chọn giải pháp phù hợp với ngân sách và use case của mình.

Mục lục

Vấn đề thực tế: Tại sao download dữ liệu quyền chọn Bybit lại khó?

Bybit không cung cấp public REST endpoint để download dữ liệu quyền chọn lịch sử trực tiếp. Các nguồn chính thức và bán chính thức bao gồm:

Anh Minh gặp lỗi timeout vì Bybit rate-limit rất nghiêm ngặt. Theo tài liệu chính thức, IP không được phép gửi quá 10 requests/giây cho các endpoint historical. Với 50 cặp quyền chọn và 6 tháng dữ liệu (ước tính 180 ngày × 86400 giây / 60 = 259,200 candles 1-phút), việc download tuần tự sẽ mất hàng ngày và chắc chắn trigger rate limit.

Tardis API: Giải pháp SaaS cho dữ liệu market data

Tardis (tardis.dev) là dịch vụ cung cấp normalized historical market data từ hơn 30 sàn giao dịch, bao gồm Bybit. Họ thu thập dữ liệu bằng cách kết nối trực tiếp vào WebSocket feeds của exchange và lưu trữ dưới dạng có thể truy vấn.

Ưu điểm của Tardis

Nhược điểm của Tardis

Code mẫu Tardis API

# tardis_bybit_options.py

Yêu cầu: pip install tardis-sdk


from tardis_sdk import Tardis
import pandas as pd
from datetime import datetime, timedelta

class BybitOptionsDataFetcher:
    def __init__(self, api_key: str):
        self.client = Tardis(auth=api_key)
        self.exchange = "bybit"
    
    def fetch_ohlcv(
        self, 
        symbol: str, 
        start_date: datetime, 
        end_date: datetime,
        timeframe: str = "1m"
    ) -> pd.DataFrame:
        """
        Download OHLCV data cho cặp quyền chọn Bybit
        
        Args:
            symbol: VD "BTC-28MAR25-95000-C" (Bybit options format)
            start_date: Ngày bắt đầu
            end_date: Ngày kết thúc
            timeframe: "1m", "5m", "1h", "1d"
        
        Returns:
            DataFrame với columns: timestamp, open, high, low, close, volume
        """
        # Tardis sử dụng unified symbol format
        tardis_symbol = f"{symbol}"
        
        # Cấu hình replay job
        job = self.client.create_replay_job(
            exchange=self.exchange,
            symbols=[tardis_symbol],
            from_date=start_date.isoformat(),
            to_date=end_date.isoformat(),
            data_types=["ohlcv"] if timeframe != "1s" else ["trade"],
            timeframe=timeframe
        )
        
        # Chờ và lấy kết quả
        result = job.join()
        
        # Convert sang DataFrame
        records = []
        for candle in result:
            records.append({
                "timestamp": candle["timestamp"],
                "open": candle["open"],
                "high": candle["high"],
                "low": candle["low"],
                "close": candle["close"],
                "volume": candle["volume"]
            })
        
        return pd.DataFrame(records)


Sử dụng

fetcher = BybitOptionsDataFetcher(api_key="YOUR_TARDIS_API_KEY")


Download 1 tháng data cho 1 cặp quyền chọn

df = fetcher.fetch_ohlcv(
    symbol="BTC-28MAR25-95000-C",
    start_date=datetime(2026, 3, 1),
    end_date=datetime(2026, 4, 1),
    timeframe="5m"
)

print(f"Downloaded {len(df)} candles")
print(df.head())
# tardis_bulk_download.py

Download nhiều cặp quyền chọn cùng lúc với parallel requests


from tardis_sdk import Tardis
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import pandas as pd
import time

class BulkOptionsDownloader:
    def __init__(self, api_key: str, max_workers: int = 5):
        self.client = Tardis(auth=api_key)
        self.max_workers = max_workers
    
    def download_symbol(self, symbol: str, start: datetime, end: datetime) -> dict:
        """Download data cho 1 symbol - xử lý lỗi và retry"""
        retry_count = 3
        for attempt in range(retry_count):
            try:
                job = self.client.create_replay_job(
                    exchange="bybit",
                    symbols=[symbol],
                    from_date=start.isoformat(),
                    to_date=end.isoformat(),
                    data_types=["ohlcv"],
                    timeframe="5m"
                )
                result = job.join(timeout=600)  # 10 phút timeout
                
                records = [
                    {"symbol": symbol, **candle}
                    for candle in result
                ]
                return {"symbol": symbol, "data": records, "status": "success"}
            
            except Exception as e:
                print(f"Attempt {attempt+1} failed for {symbol}: {e}")
                if attempt < retry_count - 1:
                    time.sleep(2 ** attempt)  # Exponential backoff
                else:
                    return {"symbol": symbol, "data": [], "status": "failed", "error": str(e)}
    
    def download_all(self, symbols: list, start: datetime, end: datetime) -> pd.DataFrame:
        """Download tất cả symbols với parallel execution"""
        all_data = []
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {
                executor.submit(self.download_symbol, sym, start, end): sym
                for sym in symbols
            }
            
            for future in as_completed(futures):
                result = future.result()
                if result["status"] == "success":
                    all_data.extend(result["data"])
                    print(f"✓ {result['symbol']}: {len(result['data'])} records")
                else:
                    print(f"✗ {result['symbol']}: {result.get('error', 'Unknown error')}")
        
        return pd.DataFrame(all_data)


Sử dụng

downloader = BulkOptionsDownloader(
    api_key="YOUR_TARDIS_API_KEY",
    max_workers=3  # Tránh rate limit
)


Danh sách 10 cặp quyền chọn BTC phổ biến

symbols = [
    "BTC-28MAR25-95000-C", "BTC-28MAR25-95000-P",
    "BTC-28MAR25-100000-C", "BTC-28MAR25-100000-P",
    "BTC-25APR25-92000-C", "BTC-25APR25-92000-P",
    "BTC-25APR25-98000-C", "BTC-25APR25-98000-P",
    "BTC-30MAY25-97000-C", "BTC-30MAY25-97000-P"
]

df_all = downloader.download_all(
    symbols=symbols,
    start=datetime(2026, 2, 1),
    end=datetime(2026, 4, 1)
)

df_all.to_parquet("bybit_options_2m_2026.parquet")
print(f"\nTotal records: {len(df_all)}")

WebSocket Replay: Tự xây dựng pipeline

Phương pháp thứ hai là tự xây dựng hệ thống thu thập dữ liệu bằng cách kết nối trực tiếp vào Bybit WebSocket API. Bybit hỗ trợ historical WebSocket streams cho phép replay dữ liệu từ quá khứ (tính năng này được giới thiệu từ 2024).

Ưu điểm của WebSocket Replay

Nhược điểm của WebSocket Replay

Code mẫu WebSocket Replay

# bybit_websocket_replay.py

Yêu cầu: pip install websockets aiofiles pandas


import asyncio
import aiofiles
import json
import time
from datetime import datetime, timedelta
from typing import Optional
import pandas as pd

class BybitOptionsReplayClient:
    """
    Bybit WebSocket Historical Replay Client
    Kết nối vào public Bybit WebSocket và replay dữ liệu lịch sử
    """
    
    WS_URL = "wss://stream.bybit.com/v5/public/linear"
    
    def __init__(self, output_dir: str = "./data"):
        self.output_dir = output_dir
        self.buffer = {}
        self.connection = None
    
    async def connect(self):
        """Thiết lập WebSocket connection"""
        import websockets
        self.connection = await websockets.connect(self.WS_URL)
        print(f"Connected to Bybit WebSocket: {self.WS_URL}")
    
    async def subscribe(self, symbols: list):
        """Subscribe vào topics cho các symbols cần replay"""
        # Bybit v5 format: {"op": "subscribe", "args": ["topic.symbol"]}
        for symbol in symbols:
            # Options ký hiệu: BTC-28MAR25-95000-C
            subscribe_msg = {
                "op": "subscribe",
                "args": [f"option.{symbol}"]
            }
            await self.connection.send(json.dumps(subscribe_msg))
            print(f"Subscribed to: option.{symbol}")
            await asyncio.sleep(0.1)  # Rate limit protection
    
    async def replay_historical(
        self, 
        symbol: str, 
        start_time: int,  # Timestamp in milliseconds
        end_time: int
    ):
        """
        Replay dữ liệu cho 1 symbol trong khoảng thời gian
        
        Bybit hỗ trợ endpoint để lấy historical kline:
        GET /v5/market/kline?category=option&symbol={symbol}&interval=1&start={start}&end={end}
        """
        import aiohttp
        
        async with aiohttp.ClientSession() as session:
            url = "https://api.bybit.com/v5/market/kline"
            params = {
                "category": "option",
                "symbol": symbol,
                "interval": "1",  # 1 minute
                "start": start_time,
                "end": end_time,
                "limit": 1000  # Max per request
            }
            
            all_candles = []
            current_start = start_time
            
            while current_start < end_time:
                params["start"] = current_start
                
                try:
                    async with session.get(url, params=params) as response:
                        if response.status == 200:
                            data = await response.json()
                            
                            if data["retCode"] == 0:
                                candles = data["result"]["list"]
                                if not candles:
                                    break
                                
                                all_candles.extend(candles)
                                # Cập nhật start time cho request tiếp theo
                                current_start = int(candles[-1]["start"]) + 60000
                                
                                print(f"{symbol}: {len(all_candles)} candles downloaded")
                            else:
                                print(f"API Error: {data['retMsg']}")
                                # Retry sau 1 giây
                                await asyncio.sleep(1)
                                continue
                        else:
                            # Rate limit hoặc lỗi khác
                            print(f"HTTP {response.status}")
                            await asyncio.sleep(5)
                
                except Exception as e:
                    print(f"Request failed: {e}")
                    await asyncio.sleep(2)
                
                # Tránh rate limit
                await asyncio.sleep(0.2)
        
        return all_candles
    
    async def process_and_save(self, symbol: str, candles: list):
        """Convert candles sang DataFrame và lưu"""
        if not candles:
            return
        
        records = []
        for candle in candles:
            records.append({
                "symbol": symbol,
                "timestamp": datetime.fromtimestamp(int(candle["start"]) / 1000),
                "open": float(candle["open"]),
                "high": float(candle["high"]),
                "low": float(candle["low"]),
                "close": float(candle["close"]),
                "volume": float(candle["volume"]),
                "turnover": float(candle.get("turnover", 0)),
                "open_interest": float(candle.get("openInterest", 0))
            })
        
        df = pd.DataFrame(records)
        
        # Save to parquet (efficient storage)
        filepath = f"{self.output_dir}/{symbol.replace('-', '_')}.parquet"
        df.to_parquet(filepath, index=False)
        print(f"Saved {len(df)} records to {filepath}")
        
        return df

async def main():
    client = BybitOptionsReplayClient(output_dir="./bybit_options_data")
    
    # Cấu hình thời gian replay: 1 tháng trước
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(days=30)).timestamp() * 1000)
    
    # Danh sách symbols cần replay
    symbols = [
        "BTC-28MAR25-95000-C",
        "BTC-28MAR25-100000-C",
        "BTC-25APR25-92000-P"
    ]
    
    for symbol in symbols:
        print(f"\n{'='*50}")
        print(f"Replaying: {symbol}")
        print(f"{'='*50}")
        
        candles = await client.replay_historical(
            symbol=symbol,
            start_time=start_time,
            end_time=end_time
        )
        
        await client.process_and_save(symbol, candles)
        
        # Nghỉ giữa các symbols để tránh rate limit
        await asyncio.sleep(5)

if __name__ == "__main__":
    asyncio.run(main())
# bybit_options_pipeline.py

Full pipeline với error handling, retry, và progress tracking


import asyncio
import aiohttp
import aiofiles
import json
import hashlib
from datetime import datetime, timedelta
from pathlib import Path
from dataclasses import dataclass, field
from typing import List, Optional
import logging

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

@dataclass
class DownloadProgress:
    symbol: str
    total_requests: int = 0
    completed_requests: int = 0
    total_records: int = 0
    errors: List[str] = field(default_factory=list)
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None

class BybitOptionsPipeline:
    """
    Production-ready pipeline để download Bybit options historical data
    """
    
    BASE_URL = "https://api.bybit.com"
    RATE_LIMIT_DELAY = 0.2  # seconds between requests
    
    def __init__(self, output_dir: str):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.progress: dict[str, DownloadProgress] = {}
    
    async def download_with_retry(
        self,
        session: aiohttp.ClientSession,
        url: str,
        params: dict,
        max_retries: int = 3
    ) -> Optional[dict]:
        """Download với exponential backoff retry"""
        for attempt in range(max_retries):
            try:
                async with session.get(url, params=params) as response:
                    if response.status == 200:
                        data = await response.json()
                        
                        if data.get("retCode") == 0:
                            return data["result"]
                        elif data.get("retCode") == 10002:  # Rate limit
                            wait_time = 2 ** attempt
                            logger.warning(f"Rate limited, waiting {wait_time}s")
                            await asyncio.sleep(wait_time)
                        else:
                            logger.error(f"API error: {data.get('retMsg')}")
                            return None
                    elif response.status == 429:
                        await asyncio.sleep(5 * (attempt + 1))
                    else:
                        logger.error(f"HTTP {response.status}")
                        return None
                        
            except aiohttp.ClientError as e:
                logger.error(f"Request failed: {e}")
                await asyncio.sleep(2 ** attempt)
        
        return None
    
    async def download_symbol(
        self,
        session: aiohttp.ClientSession,
        symbol: str,
        start_time: int,
        end_time: int
    ) -> dict:
        """Download tất cả kline data cho 1 symbol"""
        
        progress = DownloadProgress(
            symbol=symbol,
            start_time=datetime.now()
        )
        self.progress[symbol] = progress
        
        url = f"{self.BASE_URL}/v5/market/kline"
        all_candles = []
        current_start = start_time
        
        while current_start < end_time:
            params = {
                "category": "option",
                "symbol": symbol,
                "interval": "1",
                "start": current_start,
                "end": min(current_start + 60000000, end_time),  # Max ~1 month per request
                "limit": 1000
            }
            
            progress.total_requests += 1
            
            result = await self.download_with_retry(session, url, params)
            
            if result and result.get("list"):
                candles = result["list"]
                all_candles.extend(candles)
                progress.completed_requests += 1
                progress.total_records += len(candles)
                
                # Cập nhật start time
                current_start = int(candles[-1]["start"]) + 60000
                
                logger.info(
                    f"{symbol}: {len(all_candles)} candles, "
                    f"progress: {progress.completed_requests}/{progress.total_requests}"
                )
            else:
                progress.errors.append(f"No data at start={current_start}")
                break
            
            await asyncio.sleep(self.RATE_LIMIT_DELAY)
        
        progress.end_time = datetime.now()
        return {
            "symbol": symbol,
            "candles": all_candles,
            "progress": progress
        }
    
    async def save_result(self, symbol: str, candles: list):
        """Lưu data xuống disk"""
        if not candles:
            return
        
        import pandas as pd
        
        records = [
            {
                "symbol": symbol,
                "start_time": c["start"],
                "open": float(c["open"]),
                "high": float(c["high"]),
                "low": float(c["low"]),
                "close": float(c["close"]),
                "volume": float(c["volume"]),
                "turnover": float(c.get("turnover", 0)),
            }
            for c in candles
        ]
        
        df = pd.DataFrame(records)
        filepath = self.output_dir / f"{symbol.replace('-', '_')}.parquet"
        df.to_parquet(filepath, index=False)
        
        logger.info(f"Saved {len(df)} records to {filepath}")
    
    async def run(self, symbols: List[str], days_back: int = 90):
        """Chạy pipeline cho tất cả symbols"""
        
        end_time = int(datetime.now().timestamp() * 1000)
        start_time = int((datetime.now() - timedelta(days=days_back)).timestamp() * 1000)
        
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.download_symbol(session, symbol, start_time, end_time)
                for symbol in symbols
            ]
            
            results = await asyncio.gather(*tasks)
        
        # Save all results
        for result in results:
            await self.save_result(result["symbol"], result["candles"])
        
        # Print summary
        self.print_summary()
    
    def print_summary(self):
        """In tổng kết quá trình download"""
        print("\n" + "="*60)
        print("DOWNLOAD SUMMARY")
        print("="*60)
        
        total_records = 0
        for symbol, progress in self.progress.items():
            duration = (progress.end_time - progress.start_time).total_seconds()
            print(f"\n{symbol}:")
            print(f"  Records: {progress.total_records}")
            print(f"  Requests: {progress.completed_requests}/{progress.total_requests}")
            print(f"  Duration: {duration:.1f}s")
            print(f"  Errors: {len(progress.errors)}")
            if progress.errors:
                print(f"  Error details: {progress.errors[:3]}")
            
            total_records += progress.total_records
        
        print(f"\nTotal records downloaded: {total_records}")
        print(f"Output directory: {self.output_dir}")
        print("="*60)


Chạy pipeline

async def main():
    symbols = [
        "BTC-28MAR25-95000-C",
        "BTC-28MAR25-100000-C",
        "BTC-28MAR25-90000-P",
        "ETH-28MAR25-3500-C",
        "ETH-28MAR25-3200-P"
    ]
    
    pipeline = BybitOptionsPipeline(output_dir="./bybit_options_historical")
    
    await pipeline.run(
        symbols=symbols,
        days_back=30  # 1 tháng data
    )

if __name__ == "__main__":
    asyncio.run(main())

So sánh chi tiết Tardis API vs WebSocket Replay

Tiêu chí Tardis API WebSocket Replay (Self-hosted)
Chi phí khởi điểm $29/tháng (Starter plan) ~$10-50/tháng (VPS/server)
Chi phí khi scale Tăng theo data volume, có thể lên $200+/tháng Server mạnh hơn, ~$50-150/tháng cho 1 triệu records
Thời gian triển khai 15-30 phút 2-5 ngày
Độ sâu data Depends on Tardis coverage Giới hạn bởi Bybit API (thường vài tháng)
Độ tin cậy 99.9% uptime SLA Phụ thuộc vào infrastructure của bạn
Latency data 15-60 phút trễ Near-realtime có thể
Format data Unified (chuẩn hóa across exchanges) Native Bybit format
Maintenance Không cần (managed service) Cần DevOps resource
Rate limit handling Tardis xử lý tự động Phải tự implement
Phù hợp cho Quick MVP, teams nhỏ, validation Production systems, volume lớn, custom needs

Giá và ROI: Con số cụ thể

Tardis API Pricing (2026)

Plan Giá/tháng API Credits Storage Phù hợp
Free $0 100,000 1GB Testing, personal projects
Starter $29 1,000,000 10GB Indie devs, small teams
Growth $99 5,000,000 100GB Medium trading teams
Pro $299 20,000,000 500GB Professional trading firms

Ước tính chi phí thực tế:

Self-hosted WebSocket Solution

Component Giá/tháng Ghi chú
VPS (4 vCPU, 8GB RAM) $20-40 Cho pipeline nhẹ
Storage (100GB NVMe SSD) $10-15 Parquet files, compressed
Data egress $0-10 Tùy provider
Monitoring/Alerting $5-10 Datadog, Grafana Cloud
Tổng $35-75/tháng Base infrastructure

Vì sao chọn HolySheep AI?

Sau khi so sánh hai phương án trên, bạn có thể thắc mắc: HolySheep AI có liên quan gì đến việc download dữ liệu quyền chọn?

Câu trả lời nằm ở use case tiếp theo: Sau khi có dữ liệu, bạn cần xử lý, phân tích, và đưa ra quyết định giao dịch. Đây là lúc HolySheep AI phát huy tác dụng.

HolySheep AI cho Crypto Trading

HolySheep AI là nền tảng API AI với chi phí cực thấp, lý tưởng cho các ứng dụng trading:

Với $1 bạn có thể xử lý hơn 2 triệu tokens bằng DeepSeek V3.2. So với việc

Tài nguyên liên quan

Bài viết liên quan