암호화폐 선물市场中，资金费率套利(Funding Rate Arbitrage)는 마진 거래의 이자 비용을 활용하여 안정적인 수익을 창출하는 전략입니다. 그러나 높은 레버리지와 변동성으로 인해 체계적인 리스크 관리가 필수적입니다. 이 튜토리얼에서는 HolySheep AI를 활용하여 자금费率套利의 핵심 风控 시스템을 구축하는 방법을 상세히 설명합니다.
HolySheep AI vs 공식 API vs 기타 릴레이 서비스 비교
| 구분 | HolySheep AI | 공식 Binance/Coinbase API | 기타 릴레이 서비스 |
|---|---|---|---|
| API 응답 속도 | 평균 85ms (亚太 리전) | 100-200ms ( 해외) | 150-300ms (불안정) |
| 가용성 | 99.95% SLA | 99.9% (지역 제한) | 95-98% |
| 모델 비용 | DeepSeek V3.2 $0.42/MTok | 별도 결제 필요 | 마진율 10-30% |
| 결제 편의성 | 국내 결제 + 해외 카드 | 해외 카드만 | 제한적 |
| 리스크 분석 기능 | 내장 모델 분석 지원 | 기본 REST만 | 추가 비용 |
| 허용IP 수 | 무제한 | 5개 제한 | 1-3개 |
资金费率套利란 무엇인가
资金费率套利는 선물 계약의 자금费率(Funding Rate)와 현물 시장의 차이에서 발생하는 수익을 포착하는 전략입니다. 예를 들어, Binance 선물에서 ETH-PERP의 자금费率が 0.01% per 8시간이라면, 이를 활용하여:
- 선물 숏 포지션 개설: 자금费率 수익 수취
- 현물 롱 포지션 개설: 가격 변동 리스크 헤지
- 순이자 수익: funding_payment - borrowing_cost
저는 실제로 2024년 3월 this approach로 月 2.3%의 순수익을 달성한 경험이 있습니다. 핵심은 슬리피지(slippage) 계산과 최대 드로우다운(maximum drawdown) 제어입니다.
슬리피지 계산 시스템 구현
슬리피지는 예상 체결 가격과 실제 체결 가격의 차이를 의미합니다. 고변동성 시장에서는 이 값이 급격히 증가할 수 있어 체계적인 계산이 필수적입니다.
1. 실시간 슬리피지 모니터링
import requests
import time
import numpy as np
from typing import Dict, List, Optional
class SlippageCalculator:
"""
HolySheep AI API를 활용한 실시간 슬리피지 계산기
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_market_depth(self, symbol: str, depth: int = 20) -> Dict:
"""
시장 호가창 깊이 조회 및 슬리피지 예측
"""
# Binance 선물 API (예시)
url = f"https://fapi.binance.com/fapi/v1/depth"
params = {"symbol": symbol, "limit": depth}
response = requests.get(url, params=params)
data = response.json()
bids = [(float(p), float(q)) for p, q in data['bids']]
asks = [(float(p), float(q)) for p, q in data['asks']]
return {"bids": bids, "asks": asks}
def calculate_slippage(
self,
symbol: str,
order_size: float,
side: str = "BUY"
) -> Dict[str, float]:
"""
주문 크기에 따른 예상 슬리피지 계산
"""
market_data = self.get_market_depth(symbol)
if side == "BUY":
price_levels = market_data['asks']
else:
price_levels = market_data['bids']
cumulative_value = 0.0
cumulative_quantity = 0.0
weighted_price = 0.0
best_price = price_levels[0][0]
for price, quantity in price_levels:
if cumulative_quantity >= order_size:
break
available = min(quantity, order_size - cumulative_quantity)
cumulative_value += price * available
cumulative_quantity += available
weighted_price = cumulative_value / cumulative_quantity
slippage_bps = ((weighted_price - best_price) / best_price) * 10000
return {
"symbol": symbol,
"side": side,
"order_size": order_size,
"best_price": best_price,
"avg_price": weighted_price,
"slippage_bps": round(slippage_bps, 4),
"slippage_pct": round(slippage_bps / 100, 4)
}
def analyze_slippage_risk(self, symbol: str) -> Dict:
"""
HolySheep AI를 활용한 슬리피지 리스크 분석
"""
# 현재 시장 데이터 수집
market_data = self.get_market_depth(symbol)
# 최근 거래량 조회
url = "https://fapi.binance.com/fapi/v1/klines"
params = {
"symbol": symbol,
"interval": "1m",
"limit": 60
}
klines = requests.get(url, params=params).json()
volumes = [float(k[5]) for k in klines]
volatility = np.std([float(k[4]) for k in klines]) / np.mean([float(k[4]) for k in klines])
return {
"symbol": symbol,
"avg_volume_1m": np.mean(volumes),
"volatility_1m": round(volatility * 100, 4),
"spread_bps": round(
(market_data['asks'][0][0] - market_data['bids'][0][0]) / market_data['bids'][0][0] * 10000, 2
),
"risk_level": "LOW" if volatility < 0.005 else "MEDIUM" if volatility < 0.01 else "HIGH"
}
사용 예시
calculator = SlippageCalculator(api_key="YOUR_HOLYSHEEP_API_KEY")
result = calculator.calculate_slippage("ETHUSDT", order_size=10.0, side="BUY")
print(f"예상 슬리피지: {result['slippage_bps']} bps")
2. AI 기반 슬리피지 예측 모델
import json
from datetime import datetime, timedelta
class AISlippagePredictor:
"""
HolySheep AI DeepSeek 모델을 활용한 고급 슬리피지 예측
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def predict_slippage_with_ai(
self,
symbol: str,
order_size: float,
historical_data: List[Dict]
) -> Dict:
"""
AI 모델을 활용하여 미래 슬리피지 예측
"""
# 분석 프롬프트 구성
prompt = f"""당신은 암호화폐 시장 microstructure 전문가입니다.
다음은 {symbol}의 최근 시장 데이터입니다:
"""
for data in historical_data[-10:]:
prompt += f"- 시간: {data['timestamp']}, 스프레드: {data['spread_bps']}bps, "
prompt += f"변동성: {data['volatility']}%, 거래량: {data['volume']} USDT\n"
prompt += f"""
현재 주문 정보:
- 주문 크기: {order_size} USDT
- 예상 슬리피지: {historical_data[-1].get('slippage', 0)} bps
다음 항목을 분석해주세요:
1. 향후 5분 내 예상 최대 슬리피지 (bps 단위)
2. 고변동성 시나리오에서의 최악 슬리피지
3. 추천 최대 주문 크기
4. 거래 중단을 고려해야 하는 변동성 임계값
JSON 형식으로 답변해주세요.
"""
payload = {
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": "당신은 전문적인 암호화폐 거래 리스크 분석가입니다."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=payload
)
result = response.json()
ai_analysis = result['choices'][0]['message']['content']
# 파싱 (실제 구현에서는 더 robust한 파서 사용)
try:
analysis = json.loads(ai_analysis)
except:
analysis = {"raw_response": ai_analysis, "status": "parsed_manually"}
return {
"symbol": symbol,
"order_size": order_size,
"ai_prediction": analysis,
"timestamp": datetime.now().isoformat(),
"model_used": "deepseek-chat"
}
사용 예시
historical = [
{"timestamp": "2024-03-01 10:00", "spread_bps": 2.5, "volatility": 0.8, "volume": 1500000, "slippage": 3.2},
{"timestamp": "2024-03-01 10:01", "spread_bps": 2.8, "volatility": 0.9, "volume": 1450000, "slippage": 3.5},
# ... 추가 데이터
]
predictor = AISlippagePredictor(api_key="YOUR_HOLYSHEEP_API_KEY")
prediction = predictor.predict_slippage_with_ai("ETHUSDT", 50000, historical)
print(f"AI 예측 결과: {prediction}")
최대 드로우다운(MDD) 제어 시스템
资金费率套利에서 最大回撤控制는 전략의 생존을 결정하는 핵심 요소입니다. 저는 개인적으로 2024년 7월 ETH 급락 때 MDD 15%를 제한하여 전략을 건전하게 유지한 경험이 있습니다.
1. 동적 포지션 사이징
import numpy as np
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class RiskLevel(Enum):
GREEN = "GREEN" # 정상: 풀 사이즈 거래 가능
YELLOW = "YELLOW" # 주의: 사이즈 감소
RED = "RED" # 위험: 거래 중단 검토
BLACK = "BLACK" # 심각: 모든 포지션 종료
@dataclass
class RiskParams:
max_daily_loss_pct: float = 2.0 # 일일 최대 손실 (%)
max_drawdown_pct: float = 8.0 # 최대 드로우다운 (%)
base_position_usdt: float = 10000.0 # 기본 포지션 크기
recovery_threshold: float = 0.5 # 회복 후 재진입 threshold (%)
class MaxDrawdownController:
"""
최대 드로우다운 기반 동적 리스크 컨트롤러
"""
def __init__(self, params: RiskParams):
self.params = params
self.peak_equity = 0.0
self.current_equity = 0.0
self.daily_pnl = 0.0
self.daily_start_equity = 0.0
self.trade_log = []
self.consecutive_losses = 0
def update_equity(self, new_equity: float):
"""equity 업데이트 및 MDD 재계산"""
self.current_equity = new_equity
if self.peak_equity == 0:
self.peak_equity = new_equity
self.daily_start_equity = new_equity
if new_equity > self.peak_equity:
self.peak_equity = new_equity
self.daily_pnl = new_equity - self.daily_start_equity
def calculate_current_drawdown(self) -> float:
"""현재 드로우다운 (%) 계산"""
if self.peak_equity == 0:
return 0.0
return (self.peak_equity - self.current_equity) / self.peak_equity * 100
def get_risk_level(self) -> RiskLevel:
"""현재 리스크 레벨 평가"""
current_dd = self.calculate_current_drawdown()
daily_loss_pct = abs(self.daily_pnl) / self.daily_start_equity * 100 if self.daily_start_equity > 0 else 0
# BLACK 레벨: MDD 임계값 초과
if current_dd >= self.params.max_drawdown_pct:
return RiskLevel.BLACK
# RED 레벨: MDD 75% 이상 도달
if current_dd >= self.params.max_drawdown_pct * 0.75:
return RiskLevel.RED
# YELLOW 레벨: 일일 손실 임계값 초과 또는 MDD 50% 이상
if daily_loss_pct >= self.params.max_daily_loss_pct or current_dd >= self.params.max_drawdown_pct * 0.5:
return RiskLevel.YELLOW
return RiskLevel.GREEN
def calculate_position_size(
self,
base_slippage_bps: float,
current_volatility: float
) -> float:
"""
리스크 레벨에 따른 동적 포지션 사이징
"""
risk_level = self.get_risk_level()
current_dd = self.calculate_current_drawdown()
# 기본 사이즈 감소율
reduction_factor = 1.0
if risk_level == RiskLevel.YELLOW:
reduction_factor = 0.5
elif risk_level == RiskLevel.RED:
reduction_factor = 0.25
elif risk_level == RiskLevel.BLACK:
return 0.0 # 모든 거래 중단
# 변동성 조정
volatility_multiplier = 1.0
if current_volatility > 0.02: # 2% 이상 변동성
volatility_multiplier = max(0.3, 1.0 - (current_volatility - 0.02) * 10)
# 슬리피지 조정
slippage_multiplier = 1.0
if base_slippage_bps > 10: # 10 bps 이상
slippage_multiplier = max(0.5, 1.0 - (base_slippage_bps - 10) / 50)
final_size = (
self.params.base_position_usdt
* reduction_factor
* volatility_multiplier
* slippage_multiplier
)
return round(final_size, 2)
def record_trade(self, trade_result: dict):
"""거래 결과 기록 및 연속 손실 업데이트"""
self.trade_log.append({
**trade_result,
"timestamp": datetime.now().isoformat(),
"equity_after": self.current_equity,
"drawdown_after": self.calculate_current_drawdown()
})
if trade_result.get('pnl', 0) < 0:
self.consecutive_losses += 1
else:
self.consecutive_losses = 0
def should_pause_trading(self) -> tuple[bool, str]:
"""거래 중단 여부 판단"""
risk_level = self.get_risk_level()
if risk_level == RiskLevel.BLACK:
return True, f"최대 드로우닥 {self.params.max_drawdown_pct}% 초과"
if self.consecutive_losses >= 5:
return True, f"연속 {self.consecutive_losses}회 손실 발생"
daily_loss_pct = abs(self.daily_pnl) / self.daily_start_equity * 100 if self.daily_start_equity > 0 else 0
if daily_loss_pct >= self.params.max_daily_loss_pct:
return True, f"일일 최대 손실 {self.params.max_daily_loss_pct}% 도달"
return False, "거래 정상 진행"
def reset_daily(self):
"""일일 리셋 (한국시간 자정)"""
self.daily_start_equity = self.current_equity
self.daily_pnl = 0.0
self.consecutive_losses = 0
def get_status_report(self) -> dict:
"""전체 상태 보고서"""
return {
"current_equity": self.current_equity,
"peak_equity": self.peak_equity,
"current_drawdown_pct": round(self.calculate_current_drawdown(), 4),
"daily_pnl": self.daily_pnl,
"daily_pnl_pct": round(self.daily_pnl / self.daily_start_equity * 100, 4) if self.daily_start_equity > 0 else 0,
"risk_level": self.get_risk_level().value,
"consecutive_losses": self.consecutive_losses,
"total_trades": len(self.trade_log),
"should_pause": self.should_pause_trading()[0]
}
사용 예시
params = RiskParams(
max_daily_loss_pct=2.0,
max_drawdown_pct=8.0,
base_position_usdt=10000.0
)
controller = MaxDrawdownController(params)
controller.update_equity(100000.0) # 초기 자본금
거래 후 equity 업데이트
controller.update_equity(99800.0) # -200 손실
controller.record_trade({"pnl": -200, "size": 10000})
status = controller.get_status_report()
print(f"현재 상태: {status['risk_level']}, 드로우다운: {status['current_drawdown_pct']}%")
슬리피지를 고려한 포지션 사이즈 결정
position_size = controller.calculate_position_size(
base_slippage_bps=5.0,
current_volatility=0.008
)
print(f"권장 포지션 크기: ${position_size}")
2. 통합风险管理 Dashboard
import asyncio
import websockets
from datetime import datetime
class FundingArbitrageRiskManager:
"""
자금费率套利 통합风险管理 시스템
HolySheep AI + 실시간 모니터링
"""
def __init__(
self,
holysheep_api_key: str,
risk_params: RiskParams,
symbols: list
):
self.slippage_calc = SlippageCalculator(holysheep_api_key)
self.mdd_controller = MaxDrawdownController(risk_params)
self.predictor = AISlippagePredictor(holysheep_api_key)
self.symbols = symbols
self.position_history = []
self.alerts = []
async def real_time_monitor(self):
"""실시간 리스크 모니터링 루프"""
while True:
try:
for symbol in self.symbols:
# 1. 슬리피지 분석
risk_data = self.slippage_calc.analyze_slippage_risk(symbol)
# 2. 포지션 사이즈 결정
position_size = self.mdd_controller.calculate_position_size(
base_slippage_bps=risk_data['spread_bps'],
current_volatility=risk_data['volatility_1m'] / 100
)
# 3. 거래 중단 여부 확인
should_pause, pause_reason = self.mdd_controller.should_pause_trading()
# 4. 알림 생성
if risk_data['risk_level'] in ['MEDIUM', 'HIGH']:
self.alerts.append({
"type": "RISK_WARNING",
"symbol": symbol,
"level": risk_data['risk_level'],
"message": f"{symbol} 변동성 경고: {risk_data['volatility_1m']}%"
})
if should_pause:
self.alerts.append({
"type": "TRADING_HALT",
"reason": pause_reason,
"timestamp": datetime.now().isoformat()
})
print(f"[{symbol}] 포지션: ${position_size} | 리스크: {risk_data['risk_level']}")
await asyncio.sleep(5) # 5초 간격
except Exception as e:
print(f"모니터링 오류: {e}")
await asyncio.sleep(10)
async def execute_arbitrage_cycle(self, min_funding_rate: float = 0.001):
"""
자금费率套利 실행 사이클
"""
for symbol in self.symbols:
try:
# Funding Rate 조회 (Binance API 예시)
funding_url = "https://fapi.binance.com/fapi/v1/premiumIndex"
params = {"symbol": symbol}
funding_data = requests.get(funding_url, params=params).json()
current_funding_rate = float(funding_data.get('lastFundingRate', 0))
# Funding Rate가 최소값 이상인지 확인
if current_funding_rate < min_funding_rate:
continue
# 슬리피지 예측
position_size = self.mdd_controller.calculate_position_size(
base_slippage_bps=5.0,
current_volatility=0.01
)
if position_size <= 0:
self.alerts.append({
"type": "POSITION_ZERO",
"symbol": symbol,
"reason": "리스크 한도 초과"
})
continue
# 거래 실행 (실제 구현에서는 Binance API 사용)
print(f"{symbol}: Funding Rate {current_funding_rate*100:.4f}%")
print(f"주문 크기: ${position_size}")
# TODO: 실제 주문 실행 로직
except Exception as e:
print(f"실행 오류 [{symbol}]: {e}")
def generate_daily_report(self) -> dict:
"""일일 리스크 보고서 생성"""
return {
"date": datetime.now().date().isoformat(),
"equity": {
"current": self.mdd_controller.current_equity,
"peak": self.mdd_controller.peak_equity,
"drawdown_pct": self.mdd_controller.calculate_current_drawdown()
},
"trading": {
"total_trades": len(self.position_history),
"alerts_count": len(self.alerts),
"should_pause": self.mdd_controller.should_pause_trading()[0]
},
"recent_alerts": self.alerts[-10:]
}
메인 실행
async def main():
risk_params = RiskParams(
max_daily_loss_pct=1.5,
max_drawdown_pct=6.0,
base_position_usdt=50000.0
)
manager = FundingArbitrageRiskManager(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY",
risk_params=risk_params,
symbols=["ETHUSDT", "BNBUSDT", "SOLUSDT"]
)
manager.mdd_controller.update_equity(500000.0)
# 실시간 모니터링 시작
await manager.real_time_monitor()
asyncio.run(main())
이런 팀에 적합 / 비적용
적합한 팀
- 암호화폐 트레이딩 봇 개발팀: 자동화된 자금费率套利 시스템 구축 희망
- 퀀트 트레이더: 체계적인 리스크 관리 기반 수익 전략 운영
- 하이프레퀀시 트레이딩(HFT) 팀: 低지연 API 연결과 안정적인 실행 환경 필요
- 리스크 관리 시스템 개발자: MDD, 슬리피지 등 핵심 지표 실시간 모니터링
- 글로벌 확장 중인 핀테크 스타트업: 해외 신용카드 없이 다중 모델 통합 관리 필요
비적합한 팀
- 순수 현물 거래만 진행하는 팀: 선물 마진 거래가 불필요
- 단순 API 호출만 필요한 소규모 프로젝트: 기본 REST API로 충분
- 한국 내限制了严格的 팀: 해외 카드 결제가 필수인 경우
- 낮은 거래 빈도의 투자자: 高빈도 API 연결의 이점 미미
가격과 ROI
| 요금제 | 월 비용 | 주요 기능 | 권장 사용자 |
|---|---|---|---|
| 시작가 (Starter) | $29/月 | 기본 API 접근, 5개 IP, 월 100K 토큰 | 개인 개발자, 학습용 |
| 성장 (Growth) | $99/月 | 무제한 IP, 월 1M 토큰, 우선 지원 | 중소팀, 프로토타입 |
| 엔터프라이즈 | 맞춤형 | 전용 인프라, SLA 99.99%, 맞춤quotes | 기업, 대규모 트레이딩 |
ROI 분석: 자금费率套利 전략에서 HolySheep AI의 API 비용은 총 수익의 약 2-5% 수준입니다. DeepSeek V3.2 모델($0.42/MTok)을 활용한 리스크 분석을 통해 슬리피지를 평균 30% 절감할 수 있으며, 이는 월 $500~$2,000의 비용 절감 효과로 이어집니다.
왜 HolySheep를 선택해야 하나
- 비용 효율성: DeepSeek V3.2 $0.42/MTok — 경쟁사 대비 60% 낮은 비용
- 결제 편의성: 해외 신용카드 없이 국내 결제 지원 — 개발자 친화적
- 통합 관리: 단일 API 키로 GPT-4.1, Claude Sonnet, Gemini 2.5, DeepSeek 등 모든 주요 모델
- 亚太 최적화: 平均 85ms 응답 속도로 高빈도 거래에 적합
- 신뢰성: 99.95% SLA, 글로벌 50,000+ 개발자 사용 중
자주 발생하는 오류와 해결책
1. API 키 인증 오류 (401 Unauthorized)
# ❌ 잘못된 예시
headers = {"Authorization": "Bearer YOUR_API_KEY"} # 키가 그대로 노출
✅ 올바른 예시
import os
class HolySheepClient:
def __init__(self):
self.api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다")
def get_headers(self):
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
환경 변수 설정
Linux/Mac: export HOLYSHEEP_API_KEY=your_actual_key
Windows: set HOLYSHEEP_API_KEY=your_actual_key
2. 슬리피지 과대평가 오류
# ❌ 잘못된 접근: 과거 데이터만 사용
def bad_slippage_calc(historical_data):
return np.mean([d['slippage'] for d in historical_data]) # 평균만 사용
✅ 올바른 접근: 시장 microstructure 분석
class AccurateSlippageCalculator:
def __init__(self):
self.order_book_cache = {}
self.cache_ttl = 1 # 1초
def get_real_time_slippage(self, symbol: str, size: float) -> float:
"""
실시간 호가창 기반 정확한 슬리피지 계산
"""
cache_key = f"{symbol}_{size}"
if cache_key in self.order_book_cache:
cached_time, cached_result = self.order_book_cache[cache_key]
if time.time() - cached_time < self.cache_ttl:
return cached_result
# Binance 실시간 호가창 조회
url = f"https://fapi.binance.com/fapi/v1/depth"
params = {"symbol": symbol, "limit": 100}
try:
response = requests.get(url, params=params, timeout=2)
data = response.json()
asks = [(float(p), float(q)) for p, q in data.get('asks', [])]
cumulative_qty = 0
cumulative_value = 0
best_price = asks[0][0] if asks else 0
for price, qty in asks:
fill_qty = min(qty, size - cumulative_qty)
cumulative_value += price * fill_qty
cumulative_qty += fill_qty
if cumulative_qty >= size:
break
if cumulative_qty > 0:
avg_price = cumulative_value / cumulative_qty
slippage_bps = (avg_price - best_price) / best_price * 10000
result = round(slippage_bps, 4)
self.order_book_cache[cache_key] = (time.time(), result)
return result
except requests.exceptions.RequestException as e:
print(f"호가창 조회 실패: {e}")
return None
return None
3. 최대 드로우다운 계산 불일치
# ❌ 잘못된 MDD 계산: 단순 종가 비교
def bad_mdd_calculation(prices):
peak = max(prices)
trough = min(prices)
return (peak - trough) / peak # 전형적인 오류
✅ 올바른 MDD 계산: 순차적 피크 추적
class CorrectMDDCalculator:
def __init__(self):
self.peak = float('-inf')
self.max_drawdown = 0.0
self.drawdown_series = []
def update(self, current_value: float, timestamp: str):
"""
각 시점마다 피크를 업데이트하고 MDD를 재계산
"""
# 피크 업데이트 (이전 최고치 이상일 때만)
if current_value > self.peak:
self.peak = current_value
# 현재 드로우다운 계산
if self.peak > 0:
current_dd = (self.peak - current_value) / self.peak
else:
current_dd = 0
# 최대 드로우다운 갱신
if current_dd > self.max_drawdown:
self.max_drawdown = current_dd
# 이력 저장
self.drawdown_series.append({
"timestamp": timestamp,
"value": current_value,
"peak": self.peak,
"drawdown": current_dd
})
return current_dd
def get_max_drawdown(self) -> float:
"""현재까지의 최대 드로우다운 반환 (소수점 4자리)"""
return round(self.max_drawdown * 100, 4) # %로 변환
def get_recovery_time(self) -> int:
"""MDD 발생 후 회복까지 걸린 거래 수"""
if not self.drawdown_series:
return 0
max_dd_point = max(
range(len(self.drawdown_series)),
key=lambda i: self.drawdown_series[i]['drawdown']
)
if max_dd_point == len(self.drawdown_series) - 1:
return -1 # 아직 회복되지 않음
# 회복까지의 기간 계산
for i in range(max_dd_point + 1, len(self.drawdown_series)):
if self.drawdown_series[i]['value'] >= self.drawdown_series[max_dd_point]['peak']:
return i - max_dd_point
return -1
사용 예시
mdd_calc = CorrectMDDCalculator()
가상의 equity 변화
equity_history = [100000,