加密화폐 선물 Funding Rate套利는 무위험 수익처럼 보이지만, 실제로는 데이터 품질에 따라 수익률이 크게 달라집니다. 이번 튜토리얼에서는 HolySheep AI를 활용한 실시간 Funding Rate 모니터링부터 이상치 처리, 그리고夏普比率(Sharpe Ratio) 최적화까지 완전한 백테스팅 파이프라인을 구축하겠습니다.
실전 사례: 개인 트레이더의 Funding Rate Arbitrage
저는 2024년 초 주요 선물거래소 Binance, Bybit, OKX의 Funding Rate 스프레드를 활용하여 일 0.3%~0.8% 수익을 목표로套利 봇을 운영했습니다. 초기에는 단순한 Arbitrage 로직만 구현했지만, 이상치 데이터 처리 없이 백테스팅한 결과는 실제 거래와 40% 이상의 편차를 보였습니다.
구체적으로 문제가 되었던 부분:
- 极端市场波动时期的Funding Rate 급등락
- 거래소 API 지연으로 인한 스프레드 왜곡
- 유동성 부족 시점의 가격 이상치
이 튜토리얼은 HolySheep AI의 다중 거래소 실시간 데이터 통합 기능을 활용하여 데이터를 수집하고, Python 기반의 통계적 이상치 탐지 및 처리方法来夏普比率를 최적화하는整套流程을 다룹니다.
1. Funding Rate 데이터 수집 아키텍처
1.1 HolySheep AI 기반 실시간 데이터 파이프라인
HolySheep AI는 단일 API 키로 Binance, Bybit, OKX, HTX 등 주요 거래소의 데이터를 통합 수집할 수 있습니다. 이를 통해 크로스 거래소 Funding Rate 비교가 용이해집니다.
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
class FundingRateCollector:
"""HolySheep AI를 활용한 크로스 거래소 Funding Rate 수집"""
def __init__(self, api_key):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.exchanges = {
"binance": "https://api.binance.com",
"bybit": "https://api.bybit.com",
"okx": "https://www.okx.com"
}
def get_funding_rate(self, symbol, exchange="binance"):
"""특정 거래소의 현재 Funding Rate 조회"""
# HolySheep AI를 통한 통합 데이터 수집
endpoint = f"{self.base_url}/market/funding-rate"
params = {
"symbol": symbol,
"exchange": exchange,
"limit": 1
}
try:
response = requests.get(
endpoint,
headers=self.headers,
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
return {
"symbol": symbol,
"exchange": exchange,
"funding_rate": float(data.get("funding_rate", 0)),
"next_funding_time": data.get("next_funding_time"),
"timestamp": datetime.now().isoformat()
}
except requests.exceptions.RequestException as e:
print(f"[오류] {exchange} API 연결 실패: {e}")
return None
def get_cross_exchange_arbitrage_opportunity(self, symbol):
"""크로스 거래소 Funding Rate 비교를 통한 Arbitrage 기회 탐지"""
results = {}
for exchange in self.exchanges.keys():
rate_data = self.get_funding_rate(symbol, exchange)
if rate_data:
results[exchange] = rate_data
if len(results) >= 2:
rates = {k: v["funding_rate"] for k, v in results.items()}
max_exchange = max(rates, key=rates.get)
min_exchange = min(rates, key=rates.get)
spread = rates[max_exchange] - rates[min_exchange]
return {
"symbol": symbol,
"best_long": max_exchange,
"best_short": min_exchange,
"max_rate": rates[max_exchange],
"min_rate": rates[min_exchange],
"spread_bps": spread * 10000, # basis points 변환
"opportunity_score": self._calculate_opportunity_score(spread)
}
return None
def _calculate_opportunity_score(self, spread, risk_free_rate=0.05):
"""Funding Rate 스프레드 기반 기회 점수 계산 (연환산)"""
annual_spread = spread * 3 * 365 # 8시간 Funding 기준
# 단순화된 Sharpe Ratio 근사값
return annual_spread / risk_free_rate
사용 예시
collector = FundingRateCollector("YOUR_HOLYSHEEP_API_KEY")
opportunity = collector.get_cross_exchange_arbitrage_opportunity("BTCUSDT")
if opportunity:
print(f"[Arbitrage 기회 탐지]")
print(f"심볼: {opportunity['symbol']}")
print(f"Long 최적 거래소: {opportunity['best_long']} ({opportunity['max_rate']*100:.4f}%)")
print(f"Short 최적 거래소: {opportunity['best_short']} ({opportunity['min_rate']*100:.4f}%)")
print(f"스프레드: {opportunity['spread_bps']:.2f} bps")
print(f"기회 점수: {opportunity['opportunity_score']:.2f}")
1.2 히스토리컬 데이터 수집 및 저장
import sqlite3
import asyncio
from typing import List, Dict
from dataclasses import dataclass
import numpy as np
@dataclass
class HistoricalFundingData:
timestamp: datetime
symbol: str
exchange: str
funding_rate: float
mark_price: float
index_price: float
class HistoricalDataStore:
"""Funding Rate 히스토리컬 데이터를 SQLite에 저장 및 관리"""
def __init__(self, db_path="funding_rates.db"):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""데이터베이스 및 테이블 초기화"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS funding_rates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME NOT NULL,
symbol TEXT NOT NULL,
exchange TEXT NOT NULL,
funding_rate REAL NOT NULL,
mark_price REAL,
index_price REAL,
spread_bps REAL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE(timestamp, symbol, exchange)
)
""")
# 인덱스 생성 (쿼리 성능 최적화)
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_symbol_timestamp
ON funding_rates(symbol, timestamp)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_exchange_timestamp
ON funding_rates(exchange, timestamp)
""")
conn.commit()
conn.close()
def insert_batch(self, records: List[Dict]):
"""배치로 히스토리컬 데이터 삽입"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
data = [
(
r["timestamp"],
r["symbol"],
r["exchange"],
r["funding_rate"],
r.get("mark_price"),
r.get("index_price"),
r.get("spread_bps")
)
for r in records
]
cursor.executemany("""
INSERT OR REPLACE INTO funding_rates
(timestamp, symbol, exchange, funding_rate, mark_price, index_price, spread_bps)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", data)
conn.commit()
inserted_count = cursor.rowcount
conn.close()
print(f"[저장 완료] {inserted_count}건의 Funding Rate 데이터 삽입")
return inserted_count
def get_dataframe(self, symbol: str = None,
exchange: str = None,
start_date: datetime = None,
end_date: datetime = None) -> pd.DataFrame:
"""필터 조건에 따른 DataFrame 조회"""
conn = sqlite3.connect(self.db_path)
query = "SELECT * FROM funding_rates WHERE 1=1"
params = []
if symbol:
query += " AND symbol = ?"
params.append(symbol)
if exchange:
query += " AND exchange = ?"
params.append(exchange)
if start_date:
query += " AND timestamp >= ?"
params.append(start_date.isoformat())
if end_date:
query += " AND timestamp <= ?"
params.append(end_date.isoformat())
query += " ORDER BY timestamp ASC"
df = pd.read_sql_query(query, conn, params=params)
df["timestamp"] = pd.to_datetime(df["timestamp"])
conn.close()
return df
사용 예시
store = HistoricalDataStore("funding_rates.db")
최근 30일 Binance BTCUSDT Funding Rate 조회
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
df = store.get_dataframe(
symbol="BTCUSDT",
exchange="binance",
start_date=start_date,
end_date=end_date
)
print(f"[데이터 조회 완료] {len(df)}건 로드됨")
print(df.describe())
2. 데이터清洗 (Data Cleaning) Fundamentals
2.1 왜 Funding Rate 데이터清洗이 중요한가
Funding Rate Arbitrage 백테스팅에서 데이터 품질이 결과를 좌우하는 이유:
- 极端값 문제: 시장은剧烈波动 시 Funding Rate가 평소의 10~50배 이상 급등
- 거래소 버그: 가끔 잘못된 Funding Rate가 API로 전달됨
- 샘플링 편향: 유동성 부족 시간대의 왜곡된 가격
- 시계열 불연속: 거래소 운영 중단으로 인한 데이터 갭
저의 경험상, 적절한 이상치 처리 없이 백테스팅한 전략은 실제 수익률 대비 30~60% 과대추정되는 경향이 있습니다.
2.2 기본 데이터清洗 파이프라인
import scipy.stats as stats
from typing import Tuple, List
import warnings
warnings.filterwarnings('ignore')
class FundingRateDataCleaner:
"""Funding Rate 데이터 전용清洗 파이프라인"""
def __init__(self, z_score_threshold: float = 3.0,
iqr_multiplier: float = 1.5):
self.z_score_threshold = z_score_threshold
self.iqr_multiplier = iqr_multiplier
def clean_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
"""전체 데이터清洗 프로세스 실행"""
df_clean = df.copy()
# Step 1: 결측치 처리
df_clean = self._handle_missing_values(df_clean)
# Step 2: 시계열 연속성 검증
df_clean = self._validate_time_continuity(df_clean)
# Step 3: 물리적 범위 검증 (Funding Rate는 일반적으로 ±1% 이내)
df_clean = self._validate_physical_bounds(df_clean)
# Step 4: Z-Score 기반 이상치 제거
df_clean = self._remove_zscore_outliers(df_clean)
# Step 5: IQR 기반 이상치 제거
df_clean = self._remove_iqr_outliers(df_clean)
return df_clean
def _handle_missing_values(self, df: pd.DataFrame) -> pd.DataFrame:
"""결측치 처리: 선형 보간법 적용"""
initial_len = len(df)
# Funding Rate 결측치는 전후 값으로 선형 보간
if df["funding_rate"].isna().any():
df["funding_rate"] = df["funding_rate"].interpolate(method="linear")
# 시작/끝단의 결측치는 Forward/Backward fill
df["funding_rate"] = df["funding_rate"].fillna(method="ffill").fillna(method="bfill")
# 10% 이상 결측치가 있는 행은 제거
missing_ratio = df.isna().mean()
cols_to_drop = missing_ratio[missing_ratio > 0.1].index.tolist()
if cols_to_drop:
print(f"[경고] 결측치 10% 이상 컬럼 제거: {cols_to_drop}")
df = df.drop(columns=cols_to_drop)
removed = initial_len - len(df)
if removed > 0:
print(f"[결측치 처리] {removed}건 제거")
return df
def _validate_time_continuity(self, df: pd.DataFrame) -> pd.DataFrame:
"""8시간 간격 Funding Rate 시계열 연속성 검증"""
df = df.sort_values("timestamp").reset_index(drop=True)
df["time_diff"] = df["timestamp"].diff()
# 8시간 = 28800초, 허용 오차 ±30분
expected_interval = timedelta(hours=8)
tolerance = timedelta(minutes=30)
# 불연속 구간 식별
invalid_indices = []
for i in range(1, len(df)):
time_diff = df.loc[i, "time_diff"]
# 너무 짧은 간격 (중복 데이터)
if time_diff < timedelta(hours=7):
invalid_indices.append(i)
# 너무 긴 간격 (데이터 누락)
elif time_diff > (expected_interval + tolerance):
print(f"[경고] {df.loc[i, 'timestamp']}에서 데이터 갭 감지: {time_diff}")
if invalid_indices:
df = df.drop(invalid_indices).reset_index(drop=True)
print(f"[연속성 검증] {len(invalid_indices)}건 불연속 데이터 제거")
return df.drop(columns=["time_diff"])
def _validate_physical_bounds(self, df: pd.DataFrame) -> pd.DataFrame:
"""물리적 범위 검증: Funding Rate는 일반적으로 ±1% (연환산 ±1100%) 이내"""
min_rate = -0.01 # -1%
max_rate = 0.01 # +1%
before_count = len(df)
mask = (df["funding_rate"] >= min_rate) & (df["funding_rate"] <= max_rate)
outliers = df[~mask]
if len(outliers) > 0:
print(f"[물리적 범위] {len(outliers)}건 이상치 탐지 (범위: {min_rate*100}% ~ {max_rate*100}%)")
print(f" 이상치 범위: {outliers['funding_rate'].min()*100:.2f}% ~ {outliers['funding_rate'].max()*100:.2f}%")
df = df[mask].reset_index(drop=True)
return df
def _remove_zscore_outliers(self, df: pd.DataFrame,
column: str = "funding_rate") -> pd.DataFrame:
"""Z-Score 기반 이상치 탐지 및 제거"""
# 이동 창 기반 Z-Score 계산 (48시간 = 6개 Funding 주기)
window_size = 6
df["z_score"] = df[column].rolling(
window=window_size,
center=True,
min_periods=3
).apply(
lambda x: abs(stats.zscore(x)[len(x)//2]) if len(x) > 2 else 0
)
before_count = len(df)
df = df[df["z_score"] <= self.z_score_threshold].reset_index(drop=True)
removed = before_count - len(df)
if removed > 0:
print(f"[Z-Score] {removed}건 이상치 제거 (임계값: {self.z_score_threshold})")
return df.drop(columns=["z_score"])
def _remove_iqr_outliers(self, df: pd.DataFrame,
column: str = "funding_rate") -> pd.DataFrame:
"""IQR (Interquartile Range) 기반 이상치 탐지 및 제거"""
# 이동 창 기반 IQR 계산
window_size = 24 # 8일 (8시간 * 24 = 192시간)
def detect_outliers(series):
q1 = series.quantile(0.25)
q3 = series.quantile(0.75)
iqr = q3 - q1
lower_bound = q1 - self.iqr_multiplier * iqr
upper_bound = q3 + self.iqr_multiplier * iqr
return series.apply(
lambda x: x if lower_bound <= x <= upper_bound else None
)
df["iqr_cleaned"] = df[column].rolling(
window=window_size,
center=True,
min_periods=12
).apply(
lambda x: detect_outliers(pd.Series(x)).dropna().median()
)
# NaN을 중앙값으로 대체 (이상치만 치환)
mask = df[column].notna() & df["iqr_cleaned"].notna()
df.loc[mask, column] = df.loc[mask, "iqr_cleaned"]
return df.drop(columns=["iqr_cleaned"])
사용 예시
cleaner = FundingRateDataCleaner(z_score_threshold=3.0, iqr_multiplier=1.5)
df_clean = cleaner.clean_dataframe(df)
print(f"\n[清洗 결과]")
print(f"원본 데이터: {len(df)}건")
print(f"清洗 후: {len(df_clean)}건")
print(f"제거율: {(1 - len(df_clean)/len(df))*100:.2f}%")
3. 이상치 탐지 고급 기법
3.1 동적 임계값 기반 이상치 탐지
import ruptures as rpt # 변동점 탐지 라이브러리
from sklearn.ensemble import IsolationForest
from typing import Tuple
class AdvancedOutlierDetector:
"""변동점 탐지 + 머신러닝 기반 고급 이상치 탐지"""
def __init__(self, contamination: float = 0.01):
self.contamination = contamination
self.change_points = []
def detect_regime_changes(self, df: pd.DataFrame,
column: str = "funding_rate") -> List[dict]:
"""PELT 알고리즘 기반 구조 변화점(Regime Change) 탐지"""
signal = df[column].values
# PELT (Pruned Exact Linear Time) 변동점 탐지
#コスト함수: l2 (최소제곱)
algo = rpt.Pelt(model="rbf").fit(signal)
#惩罚 파라미터 (penalty): 높을수록 변동점 적게 탐지
self.change_points = algo.predict(pen=10)
results = []
for cp in self.change_points[:-1]: # 마지막은 항상 전체 길이
results.append({
"index": cp,
"timestamp": df.iloc[cp]["timestamp"] if cp < len(df) else None,
"funding_rate": df.iloc[cp][column] if cp < len(df) else None
})
print(f"[변동점 탐지] {len(results)}개 Regime Change 발견")
return results
def isolation_forest_outliers(self, df: pd.DataFrame,
features: List[str] = None) -> pd.DataFrame:
"""Isolation Forest 기반 다변량 이상치 탐지"""
if features is None:
features = ["funding_rate", "mark_price"]
# 결측치가 있는 행 제외
df_work = df.dropna(subset=features).copy()
if len(df_work) < 100:
print("[경고] 데이터 부족 - Isolation Forest 스킵")
return df
X = df_work[features].values
# Isolation Forest 모델 학습
iso_forest = IsolationForest(
contamination=self.contamination,
random_state=42,
n_estimators=100
)
# 이상치 예측 (-1: 이상치, 1: 정상)
df_work["anomaly_score"] = iso_forest.fit_predict(X)
df_work["anomaly_label"] = df_work["anomaly_score"].apply(
lambda x: "이상치" if x == -1 else "정상"
)
# 이상치 비율 출력
anomaly_ratio = (df_work["anomaly_score"] == -1).mean()
print(f"[Isolation Forest] 이상치 비율: {anomaly_ratio*100:.2f}%")
# 원본 DataFrame에 병합
df_result = df.merge(
df_work[["timestamp", "anomaly_label"]],
on="timestamp",
how="left"
)
df_result["anomaly_label"] = df_result["anomaly_label"].fillna("정상")
return df_result
def adaptive_threshold(self, df: pd.DataFrame,
column: str = "funding_rate",
lookback: int = 72) -> pd.DataFrame:
"""적응형 임계값: 최근 데이터 기반 동적 이상치 판단"""
df = df.sort_values("timestamp").reset_index(drop=True)
# 지수 이동평균 및 표준편차 기반 동적 임계값
df["ema"] = df[column].ewm(span=lookback, adjust=False).mean()
df["rolling_std"] = df[column].rolling(window=lookback, min_periods=24).std()
# 3시그마 규칙 적용
df["upper_bound"] = df["ema"] + 3 * df["rolling_std"]
df["lower_bound"] = df["ema"] - 3 * df["rolling_std"]
df["is_adaptive_outlier"] = (
(df[column] > df["upper_bound"]) |
(df[column] < df["lower_bound"])
)
outlier_count = df["is_adaptive_outlier"].sum()
print(f"[적응형 임계값] {outlier_count}건 이상치 탐지 ({outlier_count/len(df)*100:.2f}%)")
return df.drop(columns=["ema", "rolling_std", "upper_bound", "lower_bound"])
사용 예시
advanced_detector = AdvancedOutlierDetector(contamination=0.01)
변동점 탐지
change_points = advanced_detector.detect_regime_changes(df_clean)
Isolation Forest 이상치 탐지
df_with_anomalies = advanced_detector.isolation_forest_outliers(df_clean)
적응형 임계값 적용
df_final = advanced_detector.adaptive_threshold(df_with_anomalies)
print(f"\n[최종 데이터 상태]")
print(df_final.groupby("anomaly_label").size())
4. 夏普比率 (Sharpe Ratio) 계산 및 최적화
4.1 Funding Rate Arbitrage 특화 Sharpe Ratio 계산
import numpy as np
from scipy.optimize import minimize
class SharpeRatioOptimizer:
"""Funding Rate Arbitrage 특화 Sharpe Ratio 계산 및 최적화"""
def __init__(self, risk_free_rate: float = 0.0,
trading_fee: float = 0.0004, # 일반 선물 거래소 수수료
slippage: float = 0.0002): # 슬리피지 가정
self.risk_free_rate = risk_free_rate
self.trading_fee = trading_fee
self.slippage = slippage
def calculate_net_funding(self, funding_rates: np.array,
position_side: str = "long") -> np.array:
"""순 Funding Rate 계산 (수수료 및 슬리피지 차감)"""
if position_side == "long":
# Long 포지션: Funding Rate 수령 - 비용
net = funding_rates - self.trading_fee - self.slippage
else:
# Short 포지션: Funding Rate 지급 - 비용
net = -funding_rates - self.trading_fee - self.slippage
return net
def calculate_cumulative_returns(self, net_funding: np.array) -> np.array:
"""누적 수익률 계산 (복리)"""
# 1 + 수익률의 누적 곱
cum_returns = np.cumprod(1 + net_funding)
# 초기값 1로 정규화
cum_returns = cum_returns / cum_returns[0] if cum_returns[0] != 0 else cum_returns
return cum_returns
def sharpe_ratio(self, returns: np.array,
annualization_factor: int = 1095) -> float:
"""
Annualized Sharpe Ratio 계산
annualization_factor: 8시간 * 3일 * 365일 = 1095 ( Funding 3회/일 )
"""
if len(returns) < 2:
return 0.0
# 초과 수익률
excess_returns = returns - self.risk_free_rate / annualization_factor
# Sharpe Ratio = (평균 초과 수익률) / (표준편차) * √연간 거래 수
mean_return = np.mean(excess_returns)
std_return = np.std(excess_returns, ddof=1)
if std_return == 0:
return 0.0
sharpe = (mean_return / std_return) * np.sqrt(annualization_factor)
return sharpe
def max_drawdown(self, cumulative_returns: np.array) -> dict:
"""최대 낙폭 (Maximum Drawdown) 계산"""
running_max = np.maximum.accumulate(cumulative_returns)
drawdown = (cumulative_returns - running_max) / running_max
max_dd = np.min(drawdown)
max_dd_idx = np.argmin(drawdown)
# 복원 시간 계산
peak_idx = np.argmax(cumulative_returns[:max_dd_idx])
recovery_time = max_dd_idx - peak_idx if max_dd_idx > peak_idx else None
return {
"max_drawdown": max_dd,
"max_drawdown_date": max_dd_idx,
"peak_date": peak_idx,
"recovery_period": recovery_time
}
def sortino_ratio(self, returns: np.array,
target_return: float = 0,
annualization_factor: int = 1095) -> float:
"""Sortino Ratio: 하방 표준편차 기반 변동성 측정"""
excess_returns = returns - target_return
# 하방 편차 (음의 수익률만 고려)
downside_returns = returns[returns < target_return]
if len(downside_returns) == 0:
return np.inf
downside_std = np.std(downside_returns, ddof=1)
if downside_std == 0:
return 0.0
mean_return = np.mean(returns)
sortino = (mean_return / downside_std) * np.sqrt(annualization_factor)
return sortino
def optimize_weights(self, df: pd.DataFrame,
symbols: List[str],
risk_target: float = 0.02) -> dict:
"""
다중 심볼 Funding Rate Arbitrage 최적 포트폴리오 가중치 산출
Risk Parity 기반 최적화
"""
# 각 심볼의 Funding Rate 수익률 시계열
returns_dict = {}
for symbol in symbols:
if symbol in df.columns:
returns_dict[symbol] = df[symbol].values
symbols_available = [s for s in symbols if s in returns_dict]
if len(symbols_available) < 2:
print("[오류] 최적화 불가: 심볼 부족")
return None
# 수익률 행렬 생성
returns_matrix = np.column_stack([returns_dict[s] for s in symbols_available])
# 공분산 행렬
cov_matrix = np.cov(returns_matrix.T)
n_assets = len(symbols_available)
# 목적함수: Sharpe Ratio 최대화 (음수화하여 최소화 문제로 변환)
def neg_sharpe(weights):
portfolio_return = np.dot(returns_matrix, weights)
portfolio_vol = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
if portfolio_vol == 0:
return 0
sharpe = (np.mean(portfolio_return) / portfolio_vol) * np.sqrt(1095)
return -sharpe
# 제약조건
constraints = [
{"type": "eq", "fun": lambda x: np.sum(x) - 1}, # 가중치 합 = 1
]
# 범위: 각 자산 0~30%
bounds = tuple((0, 0.3) for _ in range(n_assets))
# 초기 가중치 (균등 분배)
initial_weights = np.array([1/n_assets] * n_assets)
# 최적화 실행
result = minimize(
neg_sharpe,
initial_weights,
method="SLSQP",
bounds=bounds,
constraints=constraints,
options={"maxiter": 1000}
)
optimal_weights = result.x
return {
"weights": {symbols_available[i]: optimal_weights[i]
for i in range(n_assets)},
"expected_sharpe": -result.fun,
"optimization_success": result.success
}
사용 예시
optimizer = SharpeRatioOptimizer(
risk_free_rate=0.05,
trading_fee=0.0004,
slippage=0.0002
)
Funding Rate 데이터가 있는 심볼들
symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT"]
각 심볼의 순 Funding Rate 계산
for symbol in symbols:
if symbol in df_final.columns:
net_funding = optimizer.calculate_net_funding(df_final[symbol].values)
df_final[f"{symbol}_net"] = net_funding
Sharpe Ratio 계산
for symbol in symbols:
net_col = f"{symbol}_net"
if net_col in df_final.columns:
sharpe = optimizer.sharpe_ratio(df_final[net_col].dropna().values)
mdd = optimizer.max_drawdown(
optimizer.calculate_cumulative_returns(df_final[net_col].dropna().values)
)
print(f"\n[{symbol}] 성과 지표")
print(f" Sharpe Ratio: {sharpe:.3f}")
print(f" Max Drawdown: {mdd['max_drawdown']*100:.2f}%")
5. 백테스팅 시스템 구축
5.1 완전한 백테스팅 프레임워크
from dataclasses import dataclass, field
from typing import Optional, List
from enum import Enum
import json
class PositionSide(Enum):
LONG = "long"
SHORT = "short"
FLAT = "flat"
@dataclass
class TradeResult:
entry_time: datetime
exit_time: datetime
side: PositionSide
entry_rate: float
exit_rate: float
pnl: float
fees: float
@dataclass
class BacktestConfig:
initial_capital: float = 10000.0
max_position_size: float = 0.95 # 최대 레버리지 1.0 (95% 사용)
min_spread_bps: float = 5.0 # 최소 Arbitrage 스프레드 (bps)
lookback_periods: int = 24 # 과거 데이터 참조 기간
outlier_threshold: float = 3.0 # Z-Score 임계값
@dataclass
class BacktestResult:
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
max_drawdown: float
sharpe_ratio: float
sortino_ratio: float
avg_trade_duration: float
trades: List[TradeResult] = field(default_factory=list)
class FundingRateArbitrageBacktester:
"""Funding Rate Arbitrage 전략 백테스터"""
def __init__(self, config: BacktestConfig):
self.config = config
self.data_cleaner = FundingRateDataCleaner(
z_score_threshold=config.outlier_threshold
)
self.optimizer = SharpeRatioOptimizer()
self.trades: List[TradeResult] = []
def run(self, df_long: pd.DataFrame,
df_short: pd.DataFrame) -> BacktestResult:
"""백테스트 실행"""
# 데이터清洗
df_long_clean = self.data_cleaner.clean_dataframe(df_long)
df_short_clean = self.data_cleaner.clean_dataframe(df_short)
# 정렬
df_long_clean = df_long_clean.sort_values("timestamp").reset_index(drop=True)
df_short_clean = df_short_clean.sort_values("timestamp").reset_index(drop=True)
# 병합 (timestamp 기준)
df_merged = pd.merge(
df_long_clean[["timestamp", "funding_rate"]],
df_short_clean[["timestamp", "funding_rate"]],
on="timestamp",
suffixes=("_long", "_short")
)
print(f"[백테스트 준비] {len(df_merged)}개 Funding 주기 분석")
# 스프레드 계산
df_merged["spread"] = df_merged["funding_rate_long"] + df_merged["funding_rate_short"]
df_merged["spread_bps"] = df_merged["spread"] * 10000
# Arbitrage 신호 생성
df_merged["signal"] = (df_merged["spread_bps"] >= self.config.min_spread_bps).astype(int)
# 신호 기반 트레이딩 시뮬레이션
capital = self.config.initial_capital
position = PositionSide.FLAT
for idx, row in df_merged.iterrows():
if row["signal"] == 1 and position == PositionSide.FLAT:
# Arbitrage 포지션 진입
position = PositionSide.LONG
entry_spread = row["spread"]
entry_time = row["timestamp"]
entry_long_rate = row["funding_rate_long"]
entry_short_rate = row["funding_rate_short"]
elif row["signal"] == 0 and position == PositionSide.LONG:
# Arbitrage 포지션 청산
exit_spread = row["spread"]
exit_time = row["timestamp"]
# PnL 계산 (순 Funding Rate 차감)
net_pnl = (exit_spread - entry_spread) - (self.optimizer.trading_fee * 2)
pnl_amount = capital * net_pnl * 2 # 양방향 포지션
# 수수료
fees = capital * self.optimizer.trading_fee * 2
trade = TradeResult(
entry_time=entry_time,
exit_time=exit_time,
side=position,
entry_rate=entry_spread,
exit_rate=exit_spread,
pnl=pnl_amount,
fees=fees
)
self.trades.append(trade)
capital += pnl_amount - fees