Bắt Đầu Từ Một Câu Chuyện Thật

Tháng 3 năm ngoái, một nhà giao dịch tại TP.HCM — tạm gọi anh Minh — đã mất 3 tuần xây dựng chiến lược arbitrage trên sàn Binance Futures. Mọi thứ hoàn hảo trên giấy: drawdown 8%, Sharpe ratio 2.3, win rate 68%. Khi deploy lên live, tài khoản bốc hơi 40% trong 2 ngày. Nguyên nhân? Dữ liệu backtest có "lỗ hổng" — thiếu 12% giao dịch trong các khung giờ volatility cao, OHLCV bị làm tròn sai ở timeframe nhỏ, và quan trọng nhất: tick data không đồng bộ timestamp với real market. Anh Minh kể lại: "Tôi đã tiết kiệm chi phí bằng cách dùng dữ liệu miễn phí từ nhiều nguồn khác nhau. Sai lầm lớn nhất là nghĩ rằng dữ liệu free = dữ liệu đủ tốt." Bài viết này sẽ hướng dẫn bạn xây dựng pipeline làm sạch dữ liệu Tardis chuẩn production, giúp backtest sát thực tế hơn 95%.

Tardis.dev Là Gì Và Tại Sao Nên Dùng

Tardis.dev là dịch vụ cung cấp historical market data cho crypto với độ chính xác cao. Khác với các nguồn miễn phí, Tardis cung cấp: Tuy nhiên, dữ liệu thô từ Tardis cần qua nhiều bước transform trước khi đưa vào backtest engine. Đó là lý do cần pipeline làm sạch.

Kiến Trúc Pipeline Tổng Quan


┌─────────────────────────────────────────────────────────────────────┐
│                     PIPELINE KIẾN TRÚC                              │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌──────────┐    ┌──────────────┐    ┌─────────────┐    ┌─────────┐ │
│  │ TARDIS   │───▶│  DOWNLOAD    │───▶│   CLEAN     │───▶│  STORE  │ │
│  │  API     │    │   RAW DATA   │    │   TRANSFORM │    │   DB    │ │
│  └──────────┘    └──────────────┘    └─────────────┘    └─────────┘ │
│       │                                     │                        │
│       ▼                                     ▼                        │
│  ┌──────────┐                      ┌─────────────────┐               │
│  │ MACHINE  │                      │  VALIDATION &   │               │
│  │ LEARNING │◀─────────────────────│  QUALITY CHECK  │               │
│  │ FEATURES │                      └─────────────────┘               │
│  └──────────┘                                                         │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Cài Đặt Môi Trường Và Thư Viện

# Cài đặt môi trường Python cho data pipeline
python -m venv tardis_pipeline
source tardis_pipeline/bin/activate  # Linux/Mac

tardis_pipeline\Scripts\activate # Windows

Cài các thư viện cần thiết

pip install pandas numpy pyarrow parquet-tools pip install tardis-client asyncio aiohttp pip install sqlalchemy duckdb # Database cho large-scale data pip install pydantic redis # Validation và caching pip install holySheep # Integration với HolySheep AI cho feature engineering

Download Dữ Liệu Thô Từ Tardis

import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd

class TardisDataDownloader:
    """
    Download historical K-line và tick data từ Tardis.dev
    Rate limit: 10 requests/second (free tier)
    """
    
    BASE_URL = "https://api.tardis.dev/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = None
        self.cache_dir = Path("./data/raw")
        self.cache_dir.mkdir(parents=True, exist_ok=True)
    
    async def fetch_klines(
        self,
        exchange: str,
        symbol: str,
        start_date: datetime,
        end_date: datetime,
        interval: str = "1m"
    ):
        """
        Download K-line data với chunk processing
        interval: '1m', '5m', '1h', '1d'
        """
        url = f"{self.BASE_URL}/historical/{exchange}/klines"
        
        # Tardis yêu cầu timestamp theo milliseconds
        params = {
            "symbol": symbol,
            "startTime": int(start_date.timestamp() * 1000),
            "endTime": int(end_date.timestamp() * 1000),
            "interval": interval
        }
        
        headers = {"Authorization": f"Bearer {self.api_key}"}
        
        if not self.session:
            self.session = aiohttp.ClientSession(headers=headers)
        
        try:
            async with self.session.get(url, params=params) as response:
                if response.status == 200:
                    data = await response.json()
                    return self._normalize_klines(data, exchange, symbol, interval)
                else:
                    raise Exception(f"API Error: {response.status}")
        except Exception as e:
            print(f"Lỗi fetch klines: {e}")
            return None
    
    def _normalize_klines(self, data, exchange, symbol, interval):
        """Chuẩn hóa format K-line về DataFrame thống nhất"""
        normalized = []
        for candle in data:
            normalized.append({
                "timestamp": pd.to_datetime(candle["timestamp"]),
                "open": float(candle["open"]),
                "high": float(candle["high"]),
                "low": float(candle["low"]),
                "close": float(candle["close"]),
                "volume": float(candle["volume"]),
                "quote_volume": float(candle.get("quoteVolume", 0)),
                "trades": int(candle.get("trades", 0)),
                "taker_buy_ratio": float(candle.get("takerBuyRatio", 0)),
                "exchange": exchange,
                "symbol": symbol,
                "interval": interval
            })
        return pd.DataFrame(normalized)
    
    async def fetch_trades(
        self,
        exchange: str,
        symbol: str,
        start_date: datetime,
        end_date: datetime
    ):
        """Download raw trade data (tick-by-tick)"""
        url = f"{self.BASE_URL}/historical/{exchange}/trades"
        
        params = {
            "symbol": symbol,
            "from": int(start_date.timestamp()),
            "to": int(end_date.timestamp()),
            "limit": 10000  # Max per request
        }
        
        headers = {"Authorization": f"Bearer {self.api_key}"}
        all_trades = []
        
        async with self.session.get(url, params=params) as response:
            if response.status == 200:
                data = await response.json()
                for trade in data:
                    all_trades.append({
                        "id": trade["id"],
                        "timestamp": pd.to_datetime(trade["timestamp"]),
                        "price": float(trade["price"]),
                        "amount": float(trade["amount"]),
                        "side": trade["side"],
                        "is_buyer_maker": trade.get("isBuyerMaker", False),
                        "exchange": exchange,
                        "symbol": symbol
                    })
        
        return pd.DataFrame(all_trades)

async def main():
    downloader = TardisDataDownloader(api_key="YOUR_TARDIS_API_KEY")
    
    # Download BTCUSDT 1h K-line từ Binance
    start = datetime(2024, 1, 1)
    end = datetime(2024, 12, 31)
    
    klines = await downloader.fetch_klines(
        exchange="binance",
        symbol="BTCUSDT",
        start_date=start,
        end_date=end,
        interval="1h"
    )
    
    # Download trades cùng period
    trades = await downloader.fetch_trades(
        exchange="binance",
        symbol="BTCUSDT",
        start_date=start,
        end_date=end
    )
    
    # Save raw data
    klines.to_parquet("./data/raw/btcusdt_klines_2024.parquet")
    trades.to_parquet("./data/raw/btcusdt_trades_2024.parquet")
    
    print(f"Downloaded: {len(klines)} klines, {len(trades)} trades")

asyncio.run(main())

Bước 1: Làm Sạch Dữ Liệu K-line

import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class DataQualityReport:
    """Báo cáo chất lượng dữ liệu sau cleaning"""
    total_rows: int
    rows_removed: int
    duplicates_found: int
    gaps_found: int
    outliers_found: int
    null_values: dict
    quality_score: float  # 0-100

class KLineCleaner:
    """
    Pipeline làm sạch K-line data:
    1. Remove duplicates
    2. Fill gaps (interpolation)
    3. Validate OHLC logic
    4. Detect outliers
    5. Normalize timestamps
    """
    
    def __init__(self, df: pd.DataFrame, interval: str = "1h"):
        self.df = df.copy()
        self.interval = interval
        self.interval_seconds = self._interval_to_seconds(interval)
        self.report = None
    
    def _interval_to_seconds(self, interval: str) -> int:
        mapping = {
            "1m": 60, "5m": 300, "15m": 900,
            "1h": 3600, "4h": 14400, "1d": 86400
        }
        return mapping.get(interval, 3600)
    
    def clean(self) -> pd.DataFrame:
        """Main cleaning pipeline"""
        print("Bắt đầu làm sạch K-line data...")
        
        # Step 1: Sort và reset index
        self.df = self.df.sort_values("timestamp").reset_index(drop=True)
        
        # Step 2: Remove duplicates
        self._remove_duplicates()
        
        # Step 3: Validate và fix OHLC logic
        self._validate_ohlc()
        
        # Step 4: Fill gaps
        self._fill_gaps()
        
        # Step 5: Detect outliers
        self._detect_outliers()
        
        # Step 6: Normalize timestamps
        self._normalize_timestamps()
        
        # Generate report
        self._generate_report()
        
        return self.df
    
    def _remove_duplicates(self):
        """Loại bỏ duplicate rows dựa trên timestamp"""
        before = len(self.df)
        self.df = self.df.drop_duplicates(subset=["timestamp"], keep="first")
        after = len(self.df)
        print(f"  - Removed {before - after} duplicates")
    
    def _validate_ohlc(self):
        """
        Validate OHLC logic:
        - High >= Open, Close, Low
        - Low <= Open, Close, High
        - Giá > 0
        """
        invalid_count = 0
        
        # Check High >= all other prices
        mask_high = (
            (self.df["high"] < self.df["open"]) |
            (self.df["high"] < self.df["close"]) |
            (self.df["high"] < self.df["low"])
        )
        
        # Check Low <= all other prices
        mask_low = (
            (self.df["low"] > self.df["open"]) |
            (self.df["low"] > self.df["close"]) |
            (self.df["low"] > self.df["high"])
        )
        
        # Check positive prices
        mask_price = (self.df["open"] <= 0) | (self.df["close"] <= 0)
        
        invalid_mask = mask_high | mask_low | mask_price
        invalid_count = invalid_mask.sum()
        
        if invalid_count > 0:
            print(f"  - Found {invalid_count} rows with invalid OHLC, fixing...")
            # Fix bằng cách recalculate High/Low
            self.df.loc[mask_high, "high"] = self.df.loc[mask_high, 
                ["open", "close", "high"]].max(axis=1)
            self.df.loc[mask_low, "low"] = self.df.loc[mask_low, 
                ["open", "close", "low"]].min(axis=1)
            self.df.loc[mask_price, ["open", "high", "low", "close"]] = np.nan
    
    def _fill_gaps(self):
        """
        Phát hiện và fill gaps trong time series
        Sử dụng forward fill cho giá, reset volume về 0
        """
        self.df["timestamp"] = pd.to_datetime(self.df["timestamp"])
        self.df = self.df.set_index("timestamp")
        
        # Tạo complete time range
        expected_range = pd.date_range(
            start=self.df.index.min(),
            end=self.df.index.max(),
            freq=f"{self.interval_seconds}s"
        )
        
        # Find gaps
        existing_times = set(self.df.index)
        expected_times = set(expected_range)
        gaps = expected_times - existing_times
        
        if len(gaps) > 0:
            print(f"  - Found {len(gaps)} gaps, filling...")
            
            # Reindex với complete range
            self.df = self.df.reindex(expected_range)
            self.df.index.name = "timestamp"
            
            # Forward fill prices
            price_cols = ["open", "high", "low", "close"]
            self.df[price_cols] = self.df[price_cols].ffill()
            
            # Fill volume = 0 cho gap periods
            self.df["volume"] = self.df["volume"].fillna(0)
            self.df["trades"] = self.df["trades"].fillna(0)
            
            # Mark gap rows
            self.df["is_gap_filled"] = self.df["volume"] == 0
        
        self.df = self.df.reset_index()
    
    def _detect_outliers(self, n_std: float = 5.0):
        """
        Detect outliers dựa trên price change %
        Mặc định: outlier = price change > 5 std deviations
        """
        self.df["price_change_pct"] = self.df["close"].pct_change() * 100
        
        mean_change = self.df["price_change_pct"].mean()
        std_change = self.df["price_change_pct"].std()
        
        outlier_mask = (
            abs(self.df["price_change_pct"] - mean_change) > n_std * std_change
        )
        
        outliers = outlier_mask.sum()
        if outliers > 0:
            print(f"  - Detected {outliers} outliers (>{n_std} std)")
            # Có thể: remove, cap, hoặc flag outliers
            # Ở đây ta flag để user tự quyết định
            self.df["is_outlier"] = outlier_mask
    
    def _normalize_timestamps(self):
        """Đảm bảo timestamp được normalize về UTC"""
        self.df["timestamp"] = pd.to_datetime(
            self.df["timestamp"], utc=True
        ).dt.tz_convert(None)  # Convert về naive datetime
    
    def _generate_report(self):
        """Tạo báo cáo chất lượng dữ liệu"""
        total = len(self.df)
        nulls = self.df.isnull().sum().to_dict()
        
        quality_score = 100.0
        if nulls.get("close", 0) > 0:
            quality_score -= (nulls["close"] / total) * 30
        
        self.report = DataQualityReport(
            total_rows=total,
            rows_removed=0,  # Đã track ở các bước trên
            duplicates_found=0,
            gaps_found=0,
            outliers_found=self.df.get("is_outlier", pd.Series([False])).sum(),
            null_values=nulls,
            quality_score=quality_score
        )

Sử dụng cleaner

df_klines = pd.read_parquet("./data/raw/btcusdt_klines_2024.parquet") cleaner = KLineCleaner(df_klines, interval="1h") df_clean = cleaner.clean() print(f"\nQuality Report:") print(f" - Total rows: {cleaner.report.total_rows}") print(f" - Quality Score: {cleaner.report.quality_score:.1f}%") print(f" - Null values: {cleaner.report.null_values}")

Save cleaned data

df_clean.to_parquet("./data/cleaned/btcusdt_klines_2024_clean.parquet")

Bước 2: Xử Lý Tick Data (Raw Trades)

import pandas as pd
import numpy as np
from typing import Tuple

class TickDataProcessor:
    """
    Xử lý tick-by-tick trade data:
    1. Sync timestamps với exchange time
    2. Aggregate thành timeframe nhỏ hơn (tick -> 1s/1m bars)
    3. Calculate buy/sell pressure
    4. Detect wash trades và spoofing patterns
    """
    
    def __init__(self, df: pd.DataFrame):
        self.df = df.copy()
    
    def process(self) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """
        Main processing pipeline
        Returns: (cleaned_trades, aggregated_bars)
        """
        print("Processing tick data...")
        
        # Step 1: Clean raw trades
        cleaned = self._clean_trades()
        
        # Step 2: Detect suspicious trades
        cleaned = self._detect_suspicious_trades(cleaned)
        
        # Step 3: Calculate features
        cleaned = self._add_tick_features(cleaned)
        
        # Step 4: Aggregate thành 1-second bars
        bars_1s = self._aggregate_to_seconds(cleaned, seconds=1)
        bars_1m = self._aggregate_to_minutes(cleaned, minutes=1)
        
        return cleaned, bars_1s, bars_1m
    
    def _clean_trades(self) -> pd.DataFrame:
        """Clean raw tick data"""
        df = self.df.copy()
        
        # Convert timestamp
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        
        # Sort by timestamp
        df = df.sort_values("timestamp").reset_index(drop=True)
        
        # Remove duplicates
        before = len(df)
        df = df.drop_duplicates(subset=["timestamp", "id"], keep="first")
        print(f"  - Removed {before - len(df)} duplicate trades")
        
        # Validate prices
        df = df[df["price"] > 0]
        df = df[df["amount"] > 0]
        
        # Standardize side
        df["side"] = df["side"].str.lower()
        
        return df
    
    def _detect_suspicious_trades(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Detect potential wash trades:
        - Same address buy/sell in same second
        - Round number prices
        - Tiny amounts (dust trades)
        """
        df = df.copy()
        df["is_wash_trade"] = False
        
        # Detect dust trades (< $1 notional)
        df.loc[df["price"] * df["amount"] < 1, "is_wash_trade"] = True
        
        # Detect round number trades (price ending in .00 or .000)
        round_mask = (
            (df["price"] % 1 == 0) |
            (df["price"] % 0.001 < 0.0001)
        )
        df.loc[round_mask, "is_wash_trade"] = True
        
        wash_count = df["is_wash_trade"].sum()
        if wash_count > 0:
            print(f"  - Flagged {wash_count} potential wash trades")
        
        return df
    
    def _add_tick_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Thêm features hữu ích cho analysis"""
        
        # Time features
        df["hour"] = df["timestamp"].dt.hour
        df["minute"] = df["timestamp"].dt.minute
        df["second"] = df["timestamp"].dt.second
        df["day_of_week"] = df["timestamp"].dt.dayofweek
        
        # Notional value
        df["notional"] = df["price"] * df["amount"]
        
        # Tick direction (up/down from last price)
        df["price_change"] = df["price"].diff()
        df["tick_direction"] = np.sign(df["price_change"]).fillna(0)
        
        # Cumulative volume
        df["cum_volume"] = df["amount"].cumsum()
        
        # Trade intensity (trades per second in rolling window)
        df = df.set_index("timestamp")
        df["trades_per_second"] = (
            df["id"].rolling("1s").count()
        )
        df = df.reset_index()
        
        return df
    
    def _aggregate_to_seconds(self, df: pd.DataFrame, seconds: int = 1) -> pd.DataFrame:
        """Aggregate tick data thành N-second bars"""
        
        df = df.set_index("timestamp")
        freq = f"{seconds}s"
        
        bars = pd.DataFrame()
        bars["open"] = df["price"].resample(freq).first()
        bars["high"] = df["price"].resample(freq).max()
        bars["low"] = df["price"].resample(freq).min()
        bars["close"] = df["price"].resample(freq).last()
        
        bars["volume"] = df["amount"].resample(freq).sum()
        bars["trades"] = df["id"].resample(freq).count()
        
        # Buy volume vs sell volume
        buy_mask = df["side"] == "buy"
        bars["buy_volume"] = df.loc[buy_mask, "amount"].resample(freq).sum()
        bars["sell_volume"] = df.loc[~buy_mask, "amount"].resample(freq).sum()
        bars["buy_ratio"] = bars["buy_volume"] / bars["volume"].replace(0, np.nan)
        
        # VWAP
        bars["vwap"] = (
            (df["price"] * df["amount"]).resample(freq).sum() /
            df["amount"].resample(freq).sum()
        )
        
        bars = bars.dropna(how="all")
        bars = bars.reset_index()
        
        return bars
    
    def _aggregate_to_minutes(self, df: pd.DataFrame, minutes: int = 1) -> pd.DataFrame:
        """Aggregate tick data thành N-minute bars (similar logic)"""
        return self._aggregate_to_seconds(df, seconds=minutes * 60)

Sử dụng processor

df_trades = pd.read_parquet("./data/raw/btcusdt_trades_2024.parquet") processor = TickDataProcessor(df_trades) cleaned_trades, bars_1s, bars_1m = processor.process()

Save outputs

cleaned_trades.to_parquet("./data/processed/btcusdt_trades_clean.parquet") bars_1s.to_parquet("./data/processed/btcusdt_bars_1s.parquet") bars_1m.to_parquet("./data/processed/btcusdt_bars_1m.parquet") print(f"\nProcessed outputs:") print(f" - Cleaned trades: {len(cleaned_trades):,} rows") print(f" - 1-second bars: {len(bars_1s):,} rows") print(f" - 1-minute bars: {len(bars_1m):,} rows")

Bước 3: Merge K-line Và Tick Data

import pandas as pd
import numpy as np
from datetime import datetime

class DataMerger:
    """
    Merge K-line với aggregated tick data để tạo enriched dataset
    Bổ sung thông tin: buy/sell pressure, trade intensity, VWAP
    """
    
    def __init__(self, klines: pd.DataFrame, tick_bars: pd.DataFrame):
        self.klines = klines.copy()
        self.tick_bars = tick_bars.copy()
    
    def merge(self) -> pd.DataFrame:
        """Merge K-line với tick features"""
        
        # Ensure same timezone và format
        self.klines["timestamp"] = pd.to_datetime(self.klines["timestamp"])
        self.tick_bars["timestamp"] = pd.to_datetime(self.tick_bars["timestamp"])
        
        # Round tick bars timestamp về minute boundary
        self.tick_bars["timestamp"] = self.tick_bars["timestamp"].dt.floor("1min")
        
        # Aggregate tick data theo minute
        tick_features = self.tick_bars.groupby("timestamp").agg({
            "trades": "sum",
            "buy_volume": "sum",
            "sell_volume": "sum",
            "volume": "sum",
            "vwap": "mean",
            "trades_per_second": "mean"
        }).reset_index()
        
        # Rename columns
        tick_features.columns = [
            "timestamp",
            "tick_trades",
            "tick_buy_volume",
            "tick_sell_volume",
            "tick_total_volume",
            "tick_vwap",
            "avg_trade_intensity"
        ]
        
        # Calculate derived features
        tick_features["tick_buy_ratio"] = (
            tick_features["tick_buy_volume"] / 
            tick_features["tick_total_volume"].replace(0, np.nan)
        )
        
        tick_features["tick_sell_ratio"] = (
            tick_features["tick_sell_volume"] / 
            tick_total_volume.replace(0, np.nan)
        )
        
        # Merge với klines
        merged = self.klines.merge(
            tick_features,
            on="timestamp",
            how="left"
        )
        
        # Fill NaN cho periods không có tick data
        fill_values = {
            "tick_trades": 0,
            "tick_buy_volume": 0,
            "tick_sell_volume": 0,
            "tick_total_volume": 0,
            "avg_trade_intensity": 0
        }
        merged = merged.fillna(fill_values)
        
        # Validate merge quality
        merged["merge_quality"] = (
            merged["tick_trades"] / merged["trades"].replace(0, np.nan)
        ).clip(0, 1)
        
        print(f"Merge quality stats:")
        print(f"  - Perfect matches (>95%): {(merged['merge_quality'] > 0.95).sum()}")
        print(f"  - Partial matches (50-95%): {((merged['merge_quality'] > 0.5) & (merged['merge_quality'] <= 0.95)).sum()}")
        print(f"  - No matches (<50%): {(merged['merge_quality'] <= 0.5).sum()}")
        
        return merged

Merge data

df_klines = pd.read_parquet("./data/cleaned/btcusdt_klines_2024_clean.parquet") df_bars_1m = pd.read_parquet("./data/processed/btcusdt_bars_1m.parquet") merger = DataMerger(df_klines, df_bars_1m) df_enriched = merger.merge()

Save enriched dataset

df_enriched.to_parquet("./data/final/btcusdt_enriched_2024.parquet") print(f"\nEnriched dataset shape: {df_enriched.shape}") print(df_enriched.head())

Validation Pipeline — Kiểm Tra Chất Lượng Cuối Cùng

import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class ValidationResult:
    passed: bool
    checks: Dict[str, bool]
    warnings: List[str]
    errors: List[str]

class DataValidator:
    """
    Final validation trước khi đưa vào backtest
    Đảm bảo data đạt chuẩn quality
    """
    
    def __init__(self, df: pd.DataFrame):
        self.df = df
    
    def validate(self) -> ValidationResult:
        checks = {}
        warnings = []
        errors = []
        
        # 1. Check timestamp continuity
        checks["timestamp_continuity"] = self._check_timestamp_continuity()
        if not checks["timestamp_continuity"]:
            errors.append("Timestamp gaps detected")
        
        # 2. Check OHLC validity
        checks["ohlc_valid"] = self._check_ohlc()
        if not checks["ohlc_valid"]:
            errors.append("Invalid OHLC values")
        
        # 3. Check price sanity
        checks["price_sanity"] = self._check_price_sanity()
        if not checks["price_sanity"]:
            warnings.append("Price outliers detected")
        
        # 4. Check volume sanity
        checks["volume_sanity"] = self._check_volume()
        if not checks["volume_sanity"]:
            warnings.append("Zero or negative volume detected")
        
        # 5. Check for look-ahead bias
        checks["no_lookahead"] = self._check_no_lookahead()
        if not checks["no_lookahead"]:
            errors.append("Potential look-ahead bias detected")
        
        # 6. Check required columns
        required_cols = ["timestamp", "open", "high", "low", "close", "volume"]
        checks["required_columns"] = all(col in self.df.columns for col in required_cols)
        if not checks["required_columns"]:
            errors.append("Missing required columns")
        
        passed = len(errors) == 0
        
        return ValidationResult(
            passed=passed,
            checks=checks,
            warnings=warnings,
            errors=errors
        )
    
    def _check_timestamp_continuity(self) -> bool:
        """Kiểm tra không có gaps trong timestamp"""
        if "timestamp" not in self.df.columns:
            return False
        
        expected_diff = self.df["timestamp"].diff().mode()[0]
        gaps = self.df["timestamp"].diff() != expected_diff
        return gaps.sum() == 0
    
    def _check_ohlc(self) -> bool:
        """OHLC phải thỏa mãn: high >= max(open,close,low), low <= min(open,close,high)"""
        valid = (
            (self.df["high"] >= self.df[["open", "close", "low"]].max(axis=1)) &
            (self.df["low"] <= self.df[["open", "close", "high"]].min(axis=1)) &
            (self.df["open"] > 0) &
            (self.df["close"] > 0)
        )
        return valid.all()
    
    def _check_price_sanity(self) -> bool:
        """Giá không được thay đổi quá 50% trong 1 candle"""
        price_change = abs(self.df["close"].pct_change())
        return (price_change < 0.5).all()
    
    def _check_volume(self) -> bool:
        """Volume phải >= 0"""
        return (self.df["volume"] >= 0).all()
    
    def _check_no_lookahead(self) -> bool:
        """Đảm bảo không có future information leak"""
        # Với