Trong bài viết này, tôi sẽ chia sẻ cách đội ngũ của tôi xây dựng pipeline tự động chuyển đổi dữ liệu trades từ Tardis.dev CSV sang định dạng sẵn sàng cho backtest, tất cả được xử lý thông qua HolySheep AI với độ trễ dưới 50ms và chi phí chỉ từ $0.42/MTok.
Bối Cảnh Và Tại Sao Cần Data Cleaning
Dữ liệu raw từ Tardis.dev khi xuất ra CSV thường gặp các vấn đề: timestamp không đồng nhất, missing values, duplicate trades, và format không tương thích với các framework backtest phổ biến như Backtrader, Zipline hay VectorBT. Năm 2025, khi đội ngũ của tôi chuyển từ việc sử dụng API chính thức của exchange sang relay dữ liệu, chúng tôi phải đối mặt với hàng triệu rows dữ liệu cần làm sạch mỗi ngày.
Kiến Trúc Pipeline Đề Xuất
#!/usr/bin/env python3
"""
Bybit BTCUSDT Data Cleaning Pipeline
Tardis.dev CSV -> Clean DataFrame -> Backtest Ready Format
Author: HolySheep AI Technical Team
"""
import pandas as pd
import numpy as np
from datetime import datetime, timezone
from typing import Optional, Dict, List
import json
import hashlib
import logging
Cấu hình logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class BybitTradeCleaner:
"""Pipeline làm sạch dữ liệu trades từ Tardis.dev"""
# Mapping columns từ Tardis.dev format
TARDIS_COLUMNS = {
'local_time': 'timestamp',
'exchange': 'exchange',
'pair': 'symbol',
'side': 'side',
'price': 'price',
'amount': 'amount',
'cost': 'cost',
'fee': 'fee',
'fee_currency': 'fee_currency',
'trade_id': 'trade_id',
'order_id': 'order_id'
}
# Symbols cần xử lý
SUPPORTED_SYMBOLS = ['BTCUSDT', 'ETHUSDT', 'SOLUSDT']
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key
self.cleaning_stats = {
'total_rows': 0,
'valid_rows': 0,
'dropped_duplicates': 0,
'missing_values': 0,
'invalid_prices': 0,
'processing_time_ms': 0
}
def load_tardis_csv(self, filepath: str) -> pd.DataFrame:
"""Đọc và validate file CSV từ Tardis.dev"""
start_time = datetime.now()
try:
# Đọc CSV với dtype specification để tối ưu memory
df = pd.read_csv(
filepath,
dtype={
'price': 'float64',
'amount': 'float64',
'cost': 'float64',
'fee': 'float64'
},
parse_dates=['local_time']
)
# Rename columns theo mapping
df = df.rename(columns=self.TARDIS_COLUMNS)
self.cleaning_stats['total_rows'] = len(df)
logger.info(f"Đã đọc {len(df)} rows từ {filepath}")
return df
except Exception as e:
logger.error(f"Lỗi khi đọc file: {e}")
raise
def clean_timestamp(self, df: pd.DataFrame) -> pd.DataFrame:
"""Chuẩn hóa timestamp về UTC milliseconds"""
df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)
df['timestamp_ms'] = df['timestamp'].astype('int64') // 10**6
df['date'] = df['timestamp'].dt.date
df['hour'] = df['timestamp'].dt.hour
return df
def remove_duplicates(self, df: pd.DataFrame) -> pd.DataFrame:
"""Loại bỏ duplicate trades dựa trên trade_id"""
initial_len = len(df)
df = df.drop_duplicates(subset=['trade_id'], keep='first')
dropped = initial_len - len(df)
self.cleaning_stats['dropped_duplicates'] = dropped
logger.info(f"Đã loại bỏ {dropped} duplicate trades")
return df
def validate_prices(self, df: pd.DataFrame) -> pd.DataFrame:
"""Validate giá: loại bỏ giá <= 0 hoặc outliers"""
initial_len = len(df)
# Loại bỏ giá không hợp lệ
df = df[df['price'] > 0]
# Loại bỏ outliers dựa trên IQR (Interquartile Range)
Q1 = df['price'].quantile(0.25)
Q3 = df['price'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 3 * IQR
upper_bound = Q3 + 3 * IQR
df = df[(df['price'] >= lower_bound) & (df['price'] <= upper_bound)]
invalid = initial_len - len(df)
self.cleaning_stats['invalid_prices'] = invalid
logger.info(f"Đã loại bỏ {invalid} giá không hợp lệ")
return df
def handle_missing_values(self, df: pd.DataFrame) -> pd.DataFrame:
"""Xử lý missing values"""
missing_before = df.isnull().sum().sum()
# Fill cost nếu missing (price * amount)
if 'cost' in df.columns:
df['cost'] = df['cost'].fillna(df['price'] * df['amount'])
# Drop rows với missing essential columns
essential_cols = ['timestamp', 'price', 'amount', 'side']
df = df.dropna(subset=essential_cols)
self.cleaning_stats['missing_values'] = missing_before
logger.info(f"Đã xử lý {missing_before} missing values")
return df
def add_computed_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""Thêm features tính toán cho backtest"""
# Volume USDT
df['volume_usdt'] = df['price'] * df['amount']
# Spread từ mid price (nếu có)
df['mid_price'] = df['price'] # Simplified, có thể mở rộng
# Trade hash để verify
df['trade_hash'] = df.apply(
lambda x: hashlib.sha256(
f"{x['trade_id']}{x['price']}{x['amount']}".encode()
).hexdigest()[:16],
axis=1
)
return df
def filter_symbol(self, df: pd.DataFrame, symbol: str) -> pd.DataFrame:
"""Filter chỉ lấy symbol cụ thể"""
if symbol not in self.SUPPORTED_SYMBOLS:
logger.warning(f"Symbol {symbol} không được hỗ trợ")
return df
original_len = len(df)
df = df[df['symbol'] == symbol]
logger.info(f"Filtered {original_len} -> {len(df)} rows cho {symbol}")
return df
def export_for_backtest(self, df: pd.DataFrame, format: str = 'parquet') -> str:
"""Export data sang format phù hợp cho backtest"""
# Chọn columns cần thiết
backtest_cols = [
'timestamp_ms', 'timestamp', 'date', 'hour',
'symbol', 'side', 'price', 'amount', 'cost',
'volume_usdt', 'trade_id', 'trade_hash'
]
df_export = df[[c for c in backtest_cols if c in df.columns]]
# Sort theo timestamp
df_export = df_export.sort_values('timestamp_ms')
# Generate output filename
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
output_path = f'cleaned_trades_{timestamp}.{format}'
if format == 'parquet':
df_export.to_parquet(output_path, index=False)
elif format == 'csv':
df_export.to_csv(output_path, index=False)
elif format == 'feather':
df_export.to_feather(output_path)
logger.info(f"Đã export {len(df_export)} rows sang {output_path}")
return output_path
def run_pipeline(self, input_path: str, symbol: str = 'BTCUSDT',
output_format: str = 'parquet') -> Dict:
"""Chạy toàn bộ pipeline"""
pipeline_start = datetime.now()
logger.info(f"Bắt đầu pipeline cho {symbol}")
# Load data
df = self.load_tardis_csv(input_path)
# Filter symbol
df = self.filter_symbol(df, symbol)
# Cleaning steps
df = self.clean_timestamp(df)
df = self.remove_duplicates(df)
df = self.validate_prices(df)
df = self.handle_missing_values(df)
df = self.add_computed_features(df)
# Export
output_path = self.export_for_backtest(df, output_format)
pipeline_end = datetime.now()
self.cleaning_stats['processing_time_ms'] = (
pipeline_end - pipeline_start
).total_seconds() * 1000
self.cleaning_stats['valid_rows'] = len(df)
self.cleaning_stats['success_rate'] = (
len(df) / self.cleaning_stats['total_rows'] * 100
if self.cleaning_stats['total_rows'] > 0 else 0
)
logger.info(f"Pipeline hoàn thành trong "
f"{self.cleaning_stats['processing_time_ms']:.2f}ms")
return {
'output_path': output_path,
'stats': self.cleaning_stats,
'data_preview': df.head(5).to_dict('records')
}
Sử dụng với HolySheep AI để enhanced cleaning
class HolySheepTradeEnricher:
"""Sử dụng HolySheep AI API để enrich data với insights"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
def analyze_trade_patterns(self, df: pd.DataFrame) -> Dict:
"""Gọi API để phân tích patterns trong trades"""
# Tóm tắt data thành prompt
summary = {
'total_trades': len(df),
'avg_price': df['price'].mean(),
'avg_volume': df['volume_usdt'].mean(),
'buy_ratio': (df['side'] == 'buy').mean(),
'hourly_distribution': df.groupby('hour').size().to_dict()
}
prompt = f"""Analyze these BTCUSDT trade statistics:
{json.dumps(summary, indent=2)}
Provide insights on:
1. Trading patterns and volatility
2. Optimal entry/exit times
3. Risk assessment
Respond in Vietnamese, structured JSON format."""
# Gọi HolySheep AI
response = self._call_holysheep(prompt)
return response
def _call_holysheep(self, prompt: str) -> Dict:
"""Internal method để gọi HolySheep AI API"""
import requests
headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
payload = {
'model': 'deepseek-v3.2', # $0.42/MTok - best cost efficiency
'messages': [
{'role': 'user', 'content': prompt}
],
'temperature': 0.3,
'max_tokens': 1000
}
try:
response = requests.post(
f'{self.BASE_URL}/chat/completions',
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"Lỗi khi gọi HolySheep API: {e}")
return {'error': str(e)}
if __name__ == '__main__':
# Ví dụ sử dụng
cleaner = BybitTradeCleaner()
result = cleaner.run_pipeline(
input_path='tardis_btcusdt_trades.csv',
symbol='BTCUSDT',
output_format='parquet'
)
print("\n=== Pipeline Results ===")
print(f"Output: {result['output_path']}")
print(f"Success Rate: {result['stats']['success_rate']:.2f}%")
print(f"Processing Time: {result['stats']['processing_time_ms']:.2f}ms")
print(f"Valid Rows: {result['stats']['valid_rows']:,}")
# Enrich với HolySheep AI (nếu có API key)
import os
api_key = os.getenv('HOLYSHEEP_API_KEY')
if api_key:
enricher = HolySheepTradeEnricher(api_key)
df = pd.read_parquet(result['output_path'])
insights = enricher.analyze_trade_patterns(df)
print("\n=== HolySheep AI Insights ===")
print(json.dumps(insights, indent=2, ensure_ascii=False))Tích Hợp Với Framework Backtest
#!/usr/bin/env python3
"""
Integration với VectorBT - High-performance Backtesting Library
"""
import vectorbt as vbt
import pandas as pd
import numpy as np
from datetime import datetime
def create_vectorbt_dataset(cleaned_parquet_path: str) -> pd.DataFrame:
"""Chuyển đổi cleaned data sang format của VectorBT"""
# Đọc cleaned data
df = pd.read_parquet(cleaned_parquet_path)
# Resample về 1-minute OHLCV
df.set_index('timestamp', inplace=True)
ohlcv = df.resample('1min').agg({
'price': ['first', 'max', 'min', 'last'],
'amount': 'sum',
'volume_usdt': 'sum'
})
# Flatten column names
ohlcv.columns = ['open', 'high', 'low', 'close', 'volume', 'volume_usdt']
ohlcv = ohlcv.dropna()
return ohlcv
def run_momentum_backtest(ohlcv: pd.DataFrame,
fast_period: int = 10,
slow_period: int = 50) -> dict:
"""Chạy backtest với SMA crossover strategy"""
# Calculate SMAs
fast_ma = vbt.MA.run(ohlcv['close'], fast_period)
slow_ma = vbt.MA.run(ohlcv['close'], slow_period)
# Generate signals
entries = fast_ma.ma_crossed_above(slow_ma)
exits = fast_ma.ma_crossed_below(slow_ma)
# Run backtest
pf = vbt.Portfolio.from_signals(
ohlcv['close'],
entries=entries,
exits=exits,
direction='long',
freq='1min',
init_cash=100000, # $100k initial
fees=0.001, # 0.1% trading fee
slippage=0.0005 # 0.05% slippage
)
# Calculate metrics
total_return = pf.total_return()
sharpe_ratio = pf.sharpe_ratio()
max_drawdown = pf.max_drawdown()
win_rate = pf.trades.win_rate()
avg_trade_duration = pf.trades.duration.mean()
return {
'total_return': f"{total_return * 100:.2f}%",
'sharpe_ratio': f"{sharpe_ratio:.2f}",
'max_drawdown': f"{max_drawdown * 100:.2f}%",
'win_rate': f"{win_rate * 100:.2f}%",
'avg_duration_minutes': f"{avg_trade_duration.total_seconds() / 60:.1f}",
'total_trades': len(pf.trades),
'portfolio_stats': pf.stats()
}
def run_rsi_backtest(ohlcv: pd.DataFrame,
rsi_period: int = 14,
oversold: float = 30,
overbought: float = 70) -> dict:
"""Chạy backtest với RSI mean reversion strategy"""
# Calculate RSI
rsi = vbt.RSI.run(ohlcv['close'], rsi_period)
# Generate signals
entries = rsi.rsi_below(oversold)
exits = rsi.rsi_above(overbought)
# Run backtest
pf = vbt.Portfolio.from_signals(
ohlcv['close'],
entries=entries,
exits=exits,
direction='long',
freq='1min',
init_cash=100000,
fees=0.001,
slippage=0.0005
)
return {
'total_return': f"{pf.total_return() * 100:.2f}%",
'sharpe_ratio': f"{pf.sharpe_ratio():.2f}",
'max_drawdown': f"{pf.max_drawdown() * 100:.2f}%",
'win_rate': f"{pf.trades.win_rate() * 100:.2f}%",
'total_trades': len(pf.trades)
}
if __name__ == '__main__':
# Load cleaned data
ohlcv = create_vectorbt_dataset('cleaned_trades_20260101_120000.parquet')
print(f"Dataset: {len(ohlcv)} bars")
print(f"Date range: {ohlcv.index[0]} to {ohlcv.index[-1]}")
# Run strategies
print("\n=== SMA Crossover Strategy ===")
sma_results = run_momentum_backtest(ohlcv)
for key, value in sma_results.items():
print(f"{key}: {value}")
print("\n=== RSI Mean Reversion Strategy ===")
rsi_results = run_rsi_backtest(ohlcv)
for key, value in rsi_results.items():
print(f"{key}: {value}")Tối Ưu Hóa Chi Phí Với HolySheep AI
Trong quá trình xây dựng pipeline, đội ngũ của tôi nhận ra rằng việc sử dụng các model AI đắt đỏ để phân tích patterns là không cần thiết. Với HolySheep AI, chúng tôi tiết kiệm được 85%+ chi phí API so với việc sử dụng OpenAI hay Anthropic trực tiếp.
#!/usr/bin/env python3
"""
Cost Optimization với HolySheep AI
So sánh chi phí giữa các providers cho trade analysis
"""
import time
import requests
from datetime import datetime
class CostComparison:
"""So sánh chi phí API giữa các providers"""
# Pricing per 1M tokens (2026)
PRICING = {
'GPT-4.1': {'input': 8.00, 'output': 8.00, 'provider': 'OpenAI'},
'Claude Sonnet 4.5': {'input': 15.00, 'output': 15.00, 'provider': 'Anthropic'},
'Gemini 2.5 Flash': {'input': 2.50, 'output': 10.00, 'provider': 'Google'},
'DeepSeek V3.2': {'input': 0.42, 'output': 1.10, 'provider': 'HolySheep'},
'GPT-4o-mini': {'input': 1.50, 'output': 6.00, 'provider': 'OpenAI'},
'Qwen-2.5-72B': {'input': 0.90, 'output': 0.90, 'provider': 'HolySheep'}
}
# Cấu hình HolySheep
HOLYSHEEP_CONFIG = {
'base_url': 'https://api.holysheep.ai/v1',
'supported_models': ['deepseek-v3.2', 'qwen-2.5-72b', 'gpt-4o-mini']
}
def __init__(self):
self.test_prompt = """Analyze BTCUSDT trading patterns from 1 month data:
- 50,000 trades
- Average price: $45,000
- Volatility: 3.2%
- Provide insights on optimal entry/exit timing"""
def calculate_monthly_cost(self, model: str, daily_requests: int,
avg_tokens_per_request: int) -> dict:
"""Tính chi phí hàng tháng"""
pricing = self.PRICING.get(model, {})
if not pricing:
return {'error': f'Model {model} not found'}
daily_tokens = daily_requests * avg_tokens_per_request
monthly_tokens = daily_tokens * 30
monthly_tokens_millions = monthly_tokens / 1_000_000
# Giả định 50% input, 50% output
input_cost = monthly_tokens_millions * pricing['input'] * 0.5
output_cost = monthly_tokens_millions * pricing['output'] * 0.5
total_cost = input_cost + output_cost
return {
'model': model,
'provider': pricing['provider'],
'daily_requests': daily_requests,
'tokens_per_request': avg_tokens_per_request,
'monthly_tokens_millions': round(monthly_tokens_millions, 2),
'monthly_cost_usd': round(total_cost, 2),
'cost_per_1m_tokens': round(
(pricing['input'] + pricing['output']) / 2, 2
)
}
def generate_comparison_table(self) -> str:
"""Tạo bảng so sánh chi phí"""
scenarios = [
# (daily_requests, tokens_per_request, use_case)
(100, 2000, 'Light analysis - 100 calls/day'),
(500, 2000, 'Medium analysis - 500 calls/day'),
(1000, 3000, 'Heavy analysis - 1000 calls/day'),
(5000, 2000, 'Production scale - 5000 calls/day')
]
results = []
for daily_req, tokens_req, use_case in scenarios:
row = {'use_case': use_case}
for model in ['DeepSeek V3.2', 'Qwen-2.5-72B', 'GPT-4o-mini',
'Gemini 2.5 Flash', 'GPT-4.1', 'Claude Sonnet 4.5']:
cost = self.calculate_monthly_cost(
model, daily_req, tokens_req
)
row[model] = f"${cost.get('monthly_cost_usd', 'N/A')}"
results.append(row)
return results
def call_holysheep_api(self, api_key: str, model: str = 'deepseek-v3.2',
timeout: int = 30) -> dict:
"""Test HolySheep API với đo thời gian"""
headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
payload = {
'model': model,
'messages': [
{'role': 'user', 'content': self.test_prompt}
],
'temperature': 0.3,
'max_tokens': 500
}
start_time = time.time()
try:
response = requests.post(
f'{self.HOLYSHEEP_CONFIG["base_url"]}/chat/completions',
headers=headers,
json=payload,
timeout=timeout
)
elapsed_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
return {
'status': 'success',
'latency_ms': round(elapsed_ms, 2),
'model': model,
'tokens_used': result.get('usage', {}).get('total_tokens', 0),
'response_preview': result['choices'][0]['message']['content'][:200]
}
else:
return {
'status': 'error',
'status_code': response.status_code,
'latency_ms': round(elapsed_ms, 2)
}
except requests.exceptions.Timeout:
return {
'status': 'timeout',
'latency_ms': round((time.time() - start_time) * 1000, 2)
}
except Exception as e:
return {
'status': 'error',
'error': str(e),
'latency_ms': round((time.time() - start_time) * 1000, 2)
}
def print_cost_table():
"""In bảng so sánh chi phí"""
comparison = CostComparison()
results = comparison.generate_comparison_table()
print("\n" + "="*120)
print("BẢNG SO SÁNH CHI PHÍ API HÀNG THÁNG (USD)")
print("="*120)
print(f"{'Use Case':<45} | {'DeepSeek':>10} | {'Qwen-72B':>10} | {'GPT-4o':>10} | {'Gemini':>10} | {'GPT-4.1':>10} | {'Claude':>10}")
print("-"*120)
for row in results:
print(f"{row['use_case']:<45} | {row['DeepSeek V3.2']:>10} | {row['Qwen-2.5-72B']:>10} | "
f"{row['GPT-4o-mini']:>10} | {row['Gemini 2.5 Flash']:>10} | "
f"{row['GPT-4.1']:>10} | {row['Claude Sonnet 4.5']:>10}")
print("-"*120)
print("\n💡 Kết luận: DeepSeek V3.2 qua HolySheep rẻ nhất với $0.42/MTok")
print(" Tiết kiệm 85%+ so với Claude Sonnet 4.5 ($15/MTok)")
if __name__ == '__main__':
print_cost_table()
# Test API thực tế (cần API key)
import os
api_key = os.getenv('HOLYSHEEP_API_KEY', 'YOUR_HOLYSHEEP_API_KEY')
if api_key and api_key != 'YOUR_HOLYSHEEP_API_KEY':
comparison = CostComparison()
result = comparison.call_holysheep_api(api_key, 'deepseek-v3.2')
print("\n=== HolySheep API Test ===")
print(f"Status: {result['status']}")
print(f"Latency: {result.get('latency_ms', 'N/A')}ms")
if result['status'] == 'success':
print(f"Tokens used: {result['tokens_used']}")
print(f"Preview: {result['response_preview']}...")
else:
print("\n⚠️ Để test API thực, set HOLYSHEEP_API_KEY environment variable")Bảng So Sánh Chi Phí Và Hiệu Suất
| Model | Provider | Giá Input ($/MTok) | Giá Output ($/MTok) | Độ Trễ Trung Bình | Phù Hợp Cho | Tiết Kiệm vs Claude |
|---|---|---|---|---|---|---|
| DeepSeek V3.2 | HolySheep AI | $0.42 | $1.10 | <50ms | Trade Analysis, Data Processing | 97% |
| Qwen-2.5-72B | HolySheep AI | $0.90 | $0.90 | <50ms | Complex Reasoning, Strategy Design | 94% |
| GPT-4o-mini | HolySheep AI | $1.50 | $6.00 | <100ms | Quick Tasks, Code Generation | 90% |
| Gemini 2.5 Flash | $2.50 | $10.00 | <80ms | Long Context Tasks | 83% | |
| GPT-4.1 | OpenAI | $8.00 | $8.00 | <200ms | Complex Analysis | 47% |
| Claude Sonnet 4.5 | Anthropic | $15.00 | $15.00 | <300ms | Long Documents, Deep Analysis | Baseline |
Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi "Missing Columns" Khi Load CSV
Mô tả: Tardis.dev export CSV có format columns khác với expected format.
# Vấn đề: CSV có header không đúng format
Error: KeyError: 'price' when loading Tardis CSV
Giải pháp 1: Dynamic column detection
def auto_detect_columns(df: pd.DataFrame) -> pd.DataFrame:
"""Tự động detect và rename columns"""
column_mappings = {
'local_time': ['timestamp', 'time', 'datetime', 'ts'],
'price': ['price', 'last_price', 'p'],
'amount': ['amount', 'quantity', 'size', 'qty', 'vol'],
'side': ['side', 'direction', 'type', 'action'],
'cost': ['cost', 'notional', 'value']
}
for target, aliases in column_mappings.items():
for alias in aliases:
if alias in df.columns:
df = df.rename(columns={alias: target})
break
return df
Giải pháp 2: Validate required columns
REQUIRED_COLUMNS = ['timestamp', 'price', 'amount', 'side']
def validate_csv_schema(df: pd.DataFrame) -> bool:
"""Validate CSV có đủ columns cần thiết"""
missing = [col for col in REQUIRED_COLUMNS if col not in df.columns]
if missing:
raise ValueError(f"Missing required columns: {missing}")
return True
Sử dụng:
df = auto_detect_columns(df)
validate_csv_schema(df)2. Lỗi Duplicate Trades Và Trade ID Conflicts
Mô tả: Cùng một trade xuất hiện nhiều lần trong dataset, gây sai lệch backtest.
# Vấn đề: Tardis.dev có thể export overlapping data
Dẫn đến duplicate trades và tính toán sai
Giải pháp: Multi-level deduplication
def smart_deduplicate(df: pd.DataFrame) -> pd.DataFrame:
"""Deduplicate thông minh với nhiều strategies"""
initial_count = len(df)
# Strategy 1: Dựa trên trade_id (nếu có)
if 'trade_id' in df.columns:
df = df.drop_duplicates(subset=['trade