Lần đầu tiên chạy bot arbitrage funding rate, tôi nhận được ConnectionError: timeout after 30s khi cố gắng lấy dữ liệu từ Binance WebSocket. Sau 3 ngày debug liên tục với log lỗi chồng chất, tôi nhận ra vấn đề không nằm ở code mà ở cách tôi xử lý connection pooling và retry logic. Bài viết này là tổng hợp 6 tháng nghiên cứu và thực chiến chiến lược arbitrage phí tài trợ (funding rate arbitrage) trên thị trường BTC perpetual futures, giúp bạn tránh những sai lầm tôi đã mắc phải.
Funding Rate Arbitrage Là Gì? Cơ Chế Sinh Lời Chi Tiết
Trước khi đi vào chi tiết kỹ thuật, chúng ta cần hiểu rõ cơ chế tạo ra lợi nhuận. Funding rate là khoản phí được trao đổi giữa người holding vị thế long và short mỗi 8 giờ trên các sàn giao dịch perpetual futures. Khi thị trường bullish, funding rate dương → người long trả phí cho người short. Ngược lại, khi bearish, funding rate âm → người short trả phí cho người long.
Chiến lược arbitrage cơ bản nhất: Long spot BTC + Short perpetual futures cùng giá trị. Khi funding rate dương, bạn nhận phí từ người holding long. Khi funding rate âm, vị thế short spot (nếu có) sẽ bù đắp. Mục tiêu là thu phí funding liên tục trong khi loại bỏ rủi ro biến động giá.
Kiến Trúc Hệ Thống Thu Thập Dữ Liệu
Để implement chiến lược này, bạn cần real-time data về funding rate từ nhiều sàn. Tôi sử dụng HolySheep AI để xử lý và phân tích dữ liệu với độ trễ dưới 50ms, giúp đưa ra quyết định nhanh hơn đối thủ cạnh tranh.
Thu Thập Dữ Liệu Funding Rate Từ Nhiều Sàn
#!/usr/bin/env python3
"""
BTC Funding Rate Arbitrage - Data Collector
Thu thập funding rate real-time từ multiple sàn giao dịch
"""
import asyncio
import aiohttp
import json
from datetime import datetime
from typing import Dict, List
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class FundingRateCollector:
"""Collector cho funding rate data từ multiple exchanges"""
def __init__(self, holysheep_api_key: str):
self.api_key = holysheep_api_key
self.base_url = "https://api.holysheep.ai/v1"
self.funding_data = {}
self.update_interval = 60 # seconds
async def fetch_binance_funding(self, session: aiohttp.ClientSession) -> Dict:
"""Lấy funding rate từ Binance API"""
try:
url = "https://fapi.binance.com/fapi/v1/premiumIndex"
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
if response.status == 200:
data = await response.json()
btc_data = [item for item in data if item['symbol'] == 'BTCUSDT'][0]
return {
'exchange': 'binance',
'symbol': 'BTCUSDT',
'funding_rate': float(btc_data['lastFundingRate']) * 100,
'next_funding_time': btc_data['nextFundingTime'],
'mark_price': float(btc_data['markPrice']),
'index_price': float(btc_data['indexPrice']),
'timestamp': datetime.utcnow().isoformat()
}
except asyncio.TimeoutError:
logger.error("Binance API timeout - kiểm tra network connection")
return None
except Exception as e:
logger.error(f"Binance fetch error: {e}")
return None
async def fetch_bybit_funding(self, session: aiohttp.ClientSession) -> Dict:
"""Lấy funding rate từ Bybit API"""
try:
url = "https://api.bybit.com/v5/market/funding/history"
params = {'category': 'linear', 'symbol': 'BTCUSDT', 'limit': 1}
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=10)) as response:
if response.status == 200:
data = await response.json()
if data['retCode'] == 0 and data['result']['list']:
funding_info = data['result']['list'][0]
return {
'exchange': 'bybit',
'symbol': 'BTCUSDT',
'funding_rate': float(funding_info['fundingRate']) * 100,
'funding_timestamp': funding_info['fundingRateTimestamp'],
'mark_price': float(funding_info.get('markPrice', 0)),
'timestamp': datetime.utcnow().isoformat()
}
except Exception as e:
logger.error(f"Bybit fetch error: {e}")
return None
async def fetch_okx_funding(self, session: aiohttp.ClientSession) -> Dict:
"""Lấy funding rate từ OKX API"""
try:
url = "https://www.okx.com/api/v5/market/ticker"
params = {'instId': 'BTC-USDT-SWAP'}
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=10)) as response:
if response.status == 200:
data = await response.json()
if data['code'] == '0':
ticker = data['data'][0]
# OKX funding rate cần fetch riêng
funding_url = "https://www.okx.com/api/v5/public/funding-rate"
funding_params = {'instId': 'BTC-USDT-SWAP'}
async with session.get(funding_url, params=funding_params) as funding_response:
funding_data = await funding_response.json()
if funding_data['code'] == '0':
return {
'exchange': 'okx',
'symbol': 'BTC-USDT-SWAP',
'funding_rate': float(funding_data['data'][0]['fundingRate']) * 100,
'next_funding_time': funding_data['data'][0]['nextFundingTime'],
'mark_price': float(ticker['last']),
'timestamp': datetime.utcnow().isoformat()
}
except Exception as e:
logger.error(f"OKX fetch error: {e}")
return None
async def analyze_arbitrage_opportunity(self, binance_data: Dict, bybit_data: Dict, okx_data: Dict) -> Dict:
"""Phân tích cơ hội arbitrage giữa các sàn"""
opportunities = []
all_rates = [binance_data, bybit_data, okx_data]
for i, rate1 in enumerate(all_rates):
if not rate1:
continue
for rate2 in all_rates[i+1:]:
if not rate2:
continue
rate_diff = abs(rate1['funding_rate'] - rate2['funding_rate'])
# Tính annualize funding rate
daily_rate = rate_diff * 3 # 3 funding periods/day
annual_rate = daily_rate * 365
opportunities.append({
'pair': f"{rate1['exchange']} vs {rate2['exchange']}",
'rate_1': rate1['funding_rate'],
'rate_2': rate2['funding_rate'],
'rate_diff': rate_diff,
'annualized_rate': annual_rate,
'is_profitable': annual_rate > 5, # >5% annual yield
'confidence': 'high' if annual_rate > 20 else 'medium' if annual_rate > 10 else 'low'
})
return {
'timestamp': datetime.utcnow().isoformat(),
'opportunities': opportunities,
'best_opportunity': max(opportunities, key=lambda x: x['annualized_rate']) if opportunities else None
}
async def save_to_holysheep(self, analysis_result: Dict):
"""Lưu kết quả phân tích vào HolySheep AI"""
headers = {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
async with aiohttp.ClientSession() as session:
url = f"{self.base_url}/arbitrage/analysis"
async with session.post(url, json=analysis_result, headers=headers) as response:
if response.status == 200:
logger.info("Analysis saved to HolySheep AI successfully")
return await response.json()
elif response.status == 401:
logger.error("401 Unauthorized - kiểm tra API key")
raise Exception("Invalid API key")
else:
logger.error(f"Failed to save: {response.status}")
return None
async def run_collector(self):
"""Main loop cho collector"""
connector = aiohttp.TCPConnector(limit=10, limit_per_host=5)
async with aiohttp.ClientSession(connector=connector) as session:
while True:
try:
# Fetch từ tất cả sàn song song
results = await asyncio.gather(
self.fetch_binance_funding(session),
self.fetch_bybit_funding(session),
self.fetch_okx_funding(session),
return_exceptions=True
)
binance_data, bybit_data, okx_data = results[0], results[1], results[2]
# Log raw data
logger.info(f"Binance: {binance_data}")
logger.info(f"Bybit: {bybit_data}")
logger.info(f"OKX: {okx_data}")
# Phân tích arbitrage
analysis = await self.analyze_arbitrage_opportunity(
binance_data, bybit_data, okx_data
)
# Lưu vào HolySheep AI
await self.save_to_holysheep(analysis)
# Log kết quả
if analysis['best_opportunity']:
best = analysis['best_opportunity']
logger.info(f"Best arbitrage: {best['pair']} - Diff: {best['rate_diff']:.4f}% - Annual: {best['annualized_rate']:.2f}%")
except Exception as e:
logger.error(f"Collector loop error: {e}")
await asyncio.sleep(self.update_interval)
if __name__ == "__main__":
collector = FundingRateCollector(holysheep_api_key="YOUR_HOLYSHEEP_API_KEY")
asyncio.run(collector.run_collector())
Chiến Lược Giao Dịch Tự Động
#!/usr/bin/env python3
"""
BTC Funding Rate Arbitrage - Trading Strategy
Chiến lược giao dịch tự động với risk management
"""
import asyncio
import aiohttp
import ccxt
from decimal import Decimal, ROUND_DOWN
from dataclasses import dataclass
from typing import Optional, Tuple
from datetime import datetime, timedelta
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class TradeConfig:
"""Configuration cho trading strategy"""
min_funding_diff: float = 0.01 # Minimum funding diff để vào lệnh (%)
min_profit_threshold: float = 0.005 # Minimum profit per cycle (%)
max_position_size: float = 1000 # USDT max position
stop_loss_pct: float = 0.5 # Stop loss 0.5%
holysheep_api_key: str = "YOUR_HOLYSHEEP_API_KEY"
holysheep_base_url: str = "https://api.holysheep.ai/v1"
class ArbitrageStrategy:
"""Chiến lược arbitrage funding rate tự động"""
def __init__(self, config: TradeConfig):
self.config = config
self.exchanges = {}
self.active_positions = {}
self.trade_history = []
self.initial_capital = 0
async def initialize_exchanges(self):
"""Khởi tạo kết nối các sàn giao dịch"""
# Binance - spot cho long
self.exchanges['binance_spot'] = ccxt.binance({
'enableRateLimit': True,
'options': {'defaultType': 'spot'}
})
# Binance - futures cho short
self.exchanges['binance_futures'] = ccxt.binance({
'enableRateLimit': True,
'options': {'defaultType': 'swap'}
})
# Bybit - perpetual
self.exchanges['bybit'] = ccxt.bybit({
'enableRateLimit': True,
'options': {'defaultMarketType': 'swap', 'defaultSubType': 'linear'}
})
# Load markets
for name, exchange in self.exchanges.items():
await asyncio.to_thread(exchange.load_markets)
logger.info(f"Initialized {len(self.exchanges)} exchanges")
def calculate_position_size(self, capital: float, funding_diff: float,
funding_periods_per_day: int = 3) -> Tuple[float, float]:
"""
Tính toán position size tối ưu dựa trên funding rate
Args:
capital: Vốn khả dụng
funding_diff: Chênh lệch funding rate (%)
funding_periods_per_day: Số lần funding mỗi ngày
Returns:
(position_size, expected_daily_profit)
"""
# Target: Risk 1% capital per day
risk_per_day = capital * 0.01
profit_per_period = funding_diff / 100 * capital
expected_profit_per_day = profit_per_period * funding_periods_per_day
# Nếu profit không đủ, không vào lệnh
if expected_profit_per_day < self.config.min_profit_threshold * capital:
return 0, 0
# Position size giới hạn
position_size = min(
self.config.max_position_size,
capital * 0.5 # Max 50% capital per side
)
return position_size, expected_profit_per_day
async def execute_arbitrage(self, exchange_long: str, exchange_short: str,
symbol: str, position_size: float,
binance_rate: float, bybit_rate: float) -> Optional[Dict]:
"""
Thực hiện arbitrage: Long spot + Short perpetual cùng giá trị
Args:
exchange_long: Sàn holding long position
exchange_short: Sàn holding short position
symbol: Trading pair
position_size: USDT amount
binance_rate: Funding rate của Binance
bybit_rate: Funding rate của Bybit
"""
try:
# Xác định hướng dựa trên funding rate
if binance_rate > bybit_rate:
# Long trên sàn có funding cao, short trên sàn có funding thấp
long_exchange = self.exchanges['binance_spot']
short_exchange = self.exchanges['bybit']
long_rate = binance_rate
short_rate = bybit_rate
else:
long_exchange = self.exchanges['bybit']
short_exchange = self.exchanges['binance_spot']
long_rate = bybit_rate
short_rate = binance_rate
funding_diff = (long_rate - short_rate) * 100 # Convert to percentage
# Lấy giá hiện tại
ticker = await asyncio.to_thread(
long_exchange.fetch_ticker, symbol
)
current_price = ticker['last']
# Tính số lượng BTC
btc_amount = position_size / current_price
btc_amount = float(Decimal(str(btc_amount)).quantize(
Decimal('0.00001'), rounding=ROUND_DOWN
))
# Entry orders
entry_time = datetime.utcnow()
# Long spot
long_order = await asyncio.to_thread(
long_exchange.create_market_buy_order,
symbol, btc_amount
)
# Short futures/perpetual
short_symbol = symbol.replace('BTC/', 'BTC/USDT:') if '/' in symbol else symbol
short_order = await asyncio.to_thread(
short_exchange.create_market_sell_order,
short_symbol, btc_amount
)
# Tính entry prices
entry_long_price = long_order['average'] if long_order['average'] else current_price
entry_short_price = short_order['average'] if short_order['average'] else current_price
# Lưu position
position = {
'id': f"arb_{entry_time.strftime('%Y%m%d_%H%M%S')}",
'symbol': symbol,
'long_exchange': long_exchange.id,
'short_exchange': short_exchange.id,
'btc_amount': btc_amount,
'entry_long_price': entry_long_price,
'entry_short_price': entry_short_price,
'entry_time': entry_time,
'funding_diff': funding_diff,
'position_size_usdt': position_size,
'status': 'open',
'target_exit_time': entry_time + timedelta(hours=8) # Next funding
}
self.active_positions[position['id']] = position
logger.info(f"Opened arbitrage position: {position['id']}")
logger.info(f" Long: {long_exchange.id} @ {entry_long_price}")
logger.info(f" Short: {short_exchange.id} @ {entry_short_price}")
logger.info(f" Funding diff: {funding_diff:.4f}%")
logger.info(f" Expected daily profit: {funding_diff * 3 * position_size / 100:.2f} USDT")
# Gửi notification qua HolySheep
await self.notify_holysheep(position)
return position
except ccxt.InsufficientFunds as e:
logger.error(f"Insufficient funds: {e}")
# Retry với position size nhỏ hơn
smaller_size = position_size * 0.8
return await self.execute_arbitrage(
exchange_long, exchange_short, symbol,
smaller_size, binance_rate, bybit_rate
)
except Exception as e:
logger.error(f"Execute arbitrage error: {e}")
return None
async def close_arbitrage(self, position_id: str) -> Optional[Dict]:
"""Đóng arbitrage position"""
position = self.active_positions.get(position_id)
if not position:
logger.warning(f"Position {position_id} not found")
return None
try:
# Get exchanges
long_exchange = self.exchanges.get(position['long_exchange'].replace('_spot', ''))
short_exchange = self.exchanges.get(position['short_exchange'].replace('_spot', ''))
if not long_exchange or not short_exchange:
logger.error("Exchange not found for position closure")
return None
btc_amount = position['btc_amount']
symbol = position['symbol']
# Đóng long (sell spot)
long_close = await asyncio.to_thread(
long_exchange.create_market_sell_order,
symbol, btc_amount
)
# Đóng short (buy futures)
short_symbol = symbol.replace('BTC/', 'BTC/USDT:') if '/' in symbol else symbol
short_close = await asyncio.to_thread(
short_exchange.create_market_buy_order,
short_symbol, btc_amount
)
# Tính PnL
exit_long_price = long_close['average'] if long_close['average'] else position['entry_long_price']
exit_short_price = short_close['average'] if short_close['average'] else position['entry_short_price']
# Funding profit (từ lúc vào lệnh đến giờ)
hours_held = (datetime.utcnow() - position['entry_time']).total_seconds() / 3600
funding_profit = position['funding_diff'] / 100 * position['position_size_usdt'] * (hours_held / 8)
# Price PnL (nên = 0 vì hedge)
price_pnl = (exit_short_price - position['entry_short_price']) * btc_amount
price_pnl -= (exit_long_price - position['entry_long_price']) * btc_amount
total_pnl = funding_profit + price_pnl
# Update position
position.update({
'status': 'closed',
'exit_time': datetime.utcnow(),
'exit_long_price': exit_long_price,
'exit_short_price': exit_short_price,
'funding_profit': funding_profit,
'price_pnl': price_pnl,
'total_pnl': total_pnl,
'hours_held': hours_held
})
# Remove from active positions
del self.active_positions[position_id]
# Add to history
self.trade_history.append(position)
logger.info(f"Closed position {position_id}: PnL = {total_pnl:.2f} USDT")
return position
except Exception as e:
logger.error(f"Close arbitrage error: {e}")
return None
async def notify_holysheep(self, position: Dict):
"""Gửi notification qua HolySheep AI webhook"""
headers = {
'Authorization': f'Bearer {self.config.holysheep_api_key}',
'Content-Type': 'application/json'
}
payload = {
'event': 'arbitrage_position_opened',
'position': position,
'timestamp': datetime.utcnow().isoformat()
}
async with aiohttp.ClientSession() as session:
url = f"{self.config.holysheep_base_url}/notifications"
try:
async with session.post(url, json=payload, headers=headers,
timeout=aiohttp.ClientTimeout(total=5)) as response:
if response.status == 200:
logger.info("Notification sent to HolySheep AI")
else:
logger.warning(f"Notification failed: {response.status}")
except aiohttp.ClientError as e:
logger.error(f"Holysheep notification error: {e}")
async def run_strategy(self):
"""Main strategy loop"""
await self.initialize_exchanges()
# Initialize capital
self.initial_capital = self.config.max_position_size * 2
logger.info(f"Starting strategy with capital: {self.initial_capital} USDT")
while True:
try:
# Fetch funding rates
binance = self.exchanges['binance_futures']
bybit = self.exchanges['bybit']
binance_ticker = await asyncio.to_thread(
binance.fetch_ticker, 'BTC/USDT'
)
# Tính funding diff (demo - thực tế cần fetch funding rate history)
binance_rate = 0.01 # Demo rate
bybit_rate = 0.008 # Demo rate
funding_diff = (binance_rate - bybit_rate) * 100
if funding_diff >= self.config.min_funding_diff:
# Tính position size
pos_size, expected_profit = self.calculate_position_size(
self.initial_capital, funding_diff
)
if pos_size > 100: # Minimum position
logger.info(f"Found opportunity: diff={funding_diff:.4f}%, pos_size={pos_size}")
await self.execute_arbitrage(
'binance', 'bybit', 'BTC/USDT',
pos_size, binance_rate, bybit_rate
)
# Check và đóng các position đã đến hạn
current_time = datetime.utcnow()
for pos_id, position in list(self.active_positions.items()):
if current_time >= position['target_exit_time']:
await self.close_arbitrage(pos_id)
# Calculate performance
if self.trade_history:
total_pnl = sum(p['total_pnl'] for p in self.trade_history)
logger.info(f"Total PnL: {total_pnl:.2f} USDT | Active: {len(self.active_positions)}")
except Exception as e:
logger.error(f"Strategy loop error: {e}")
await asyncio.sleep(30) # Check every 30 seconds
Usage
config = TradeConfig(
min_funding_diff=0.01,
min_profit_threshold=0.005,
max_position_size=1000
)
strategy = ArbitrageStrategy(config)
asyncio.run(strategy.run_strategy())
Phân Tích Dữ Liệu Và Backtesting Chiến Lược
Để validate chiến lược, tôi đã backtest 6 tháng dữ liệu funding rate từ Binance, Bybit và OKX. Kết quả cho thấy funding rate trung bình hàng ngày vào khoảng 0.01% - 0.03%, với volatility cao vào các đợt market dump hoặc pump. Điểm quan trọng nhất: chênh lệch funding rate giữa các sàn thường xuyên đạt 0.02% - 0.05%, tương đương 20-50% annualized nếu bạn capture được chênh lệch này.
#!/usr/bin/env python3
"""
Backtesting Engine cho Funding Rate Arbitrage Strategy
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
import json
class FundingArbitrageBacktest:
"""Backtesting engine với realistic fee simulation"""
def __init__(self, initial_capital: float = 10000):
self.initial_capital = initial_capital
self.capital = initial_capital
self.trades = []
self.fees = {
'binance_spot': {'maker': 0.001, 'taker': 0.001},
'binance_futures': {'maker': 0.0002, 'taker': 0.0004},
'bybit': {'maker': 0.0002, 'taker': 0.0005}
}
def calculate_fees(self, exchange: str, side: str, amount: float) -> float:
"""Tính phí giao dịch thực tế"""
fee_rate = self.fees[exchange]['taker']
return amount * fee_rate
def simulate_trade(self, entry_time: datetime,
binance_rate: float, bybit_rate: float,
holding_hours: int = 8) -> Dict:
"""Simulate một arbitrage trade"""
# Xác định hướng: Long sàn có funding cao, Short sàn có funding thấp
if binance_rate > bybit_rate:
long_exchange = 'binance_spot'
short_exchange = 'bybit'
funding_earned = binance_rate # Nhận từ long
funding_paid = bybit_rate # Trả cho short
else:
long_exchange = 'bybit'
short_exchange = 'binance_spot'
funding_earned = bybit_rate
funding_paid = binance_rate
funding_diff = funding_earned - funding_paid
# Position size = 50% capital
position_size = self.capital * 0.5
btc_amount = position_size / 45000 # Demo BTC price
# Tính phí vào/ra (2 legs × 2 sides = 4 fees)
fee_1 = self.calculate_fees(long_exchange, 'buy', position_size)
fee_2 = self.calculate_fees(short_exchange, 'sell', position_size)
fee_3 = self.calculate_fees(long_exchange, 'sell', position_size)
fee_4 = self.calculate_fees(short_exchange, 'buy', position_size)
total_fees = fee_1 + fee_2 + fee_3 + fee_4
# Funding profit
funding_profit = position_size * funding_diff * (holding_hours / 8)
# Net PnL (funding - fees)
net_pnl = funding_profit - total_fees
return {
'entry_time': entry_time,
'exit_time': entry_time + timedelta(hours=holding_hours),
'long_exchange': long_exchange,
'short_exchange': short_exchange,
'funding_diff': funding_diff * 100,
'position_size': position_size,
'btc_amount': btc_amount,
'fees': total_fees,
'funding_profit': funding_profit,
'net_pnl': net_pnl,
'roi': net_pnl / self.capital * 100
}
def run_backtest(self, historical_data: pd.DataFrame,
min_funding_diff: float = 0.01) -> Dict:
"""
Run backtest với dữ liệu lịch sử
Args:
historical_data: DataFrame với columns ['timestamp', 'binance_rate', 'bybit_rate']
min_funding_diff: Minimum funding diff để trigger trade
"""
self.capital = self.initial_capital
self.trades = []
for idx, row in historical_data.iterrows():
funding_diff = (row['binance_rate'] - row['bybit_rate']) * 100
if abs(funding_diff) >= min_funding_diff:
trade = self.simulate_trade(
entry_time=row['timestamp'],
binance_rate=row['binance_rate'],
bybit_rate=row['bybit_rate'],
holding_hours=8
)
self.trades.append(trade)
self.capital += trade['net_pnl']
return self.generate_report()
def generate_report(self) -> Dict:
"""Generate backtest report"""
if not self.trades:
return {'error': 'No trades generated'}
trades_df = pd.DataFrame(self.trades)
total_pnl = trades_df['net_pnl'].sum()
total_fees = trades_df['fees'].sum()
total_funding = trades_df['funding_profit'].sum()
win_rate = (trades_df['net_pnl'] > 0).mean() * 100
avg_pnl = trades_df['net_pnl'].mean()
max_drawdown = self.calculate_max_drawdown(trades_df)
return {
'initial_capital': self.initial_capital,
'final_capital': self.capital,
'total_pnl': total_pnl,
'total_return_pct': (self.capital / self.initial_capital - 1) * 100,
'total_trades': len(self.trades),
'win_rate': win_rate,
'avg_pnl_per_trade': avg_pnl,
'total_fees': total_fees,
'total_funding_received': total_funding,
'max_drawdown': max_drawdown,
'sharpe_ratio': self.calculate_sharpe(trades_df),
'trades': trades_df.to_dict('records')
}
def calculate_max_drawdown(self, trades_df: pd.DataFrame) -> float:
"""Tính maximum drawdown"""
cumulative = trades_df['net_pnl'].cumsum()
running_max = cumulative.cummax()
drawdown = cumulative - running_max
return abs(drawdown.min())
def calculate_sharpe(self, trades_df: pd.DataFrame, risk_free_rate: float = 0.02) -> float:
"""Tính Sharpe ratio"""
returns = trades_df['net_pnl'] / self.initial_capital
excess_returns = returns - risk_free_rate / 365
if excess_returns.std() == 0:
return 0