암호화폐 시장에서는 실시간 데이터 처리와 과거 데이터 기반 전략 검증이 성공적인 거래 시스템의 핵심입니다. Bybit 스팟 거래소의 Historical Data API를 활용하면 수년간의 시장 데이터를 효율적으로 수집하고, HolySheep AI 게이트웨이를 통해 AI 기반 거래 신호를 생성한 뒤 정밀한 백테스팅을 수행할 수 있습니다. 이 튜토리얼에서는 엔지니어 관점에서 프로덕션 수준의 백테스팅 파이프라인을 구축하는 방법을 상세히 다룹니다.
Bybit 스팟 Historical Data API 핵심 구조
Bybit은 스팟 거래소를 위한 comprehensive한 REST API를 제공합니다. Historical Data 엔드포인트는 캔들스틱, 거래 내역, 오더북 등 핵심 데이터를 지원하며,authenticated 요청과 공개 요청으로 구분됩니다. 저의 경험상高频 거래 전략 개발 시 데이터 수집层的 효율성이 전체 백테스팅 파이프라인의 병목이 되는 경우가 많아, 비동기 수집 아키텍처 설계가 중요합니다.
API 엔드포인트 종류와 Rate Limit
| 엔드포인트 카테고리 | 엔드포인트 경로 | Rate Limit | 데이터 주기 |
|---|---|---|---|
| 캔들스틱 | /v5/market/kline | 600 requests/10sec | 1min/3min/5min/15min/30min/1h/2h/4h/6h/12h/1d/1w/1M |
| 거래 내역 | /v5/market/recent-trade | 600 requests/10sec | Real-time streaming |
| 오더북 | /v5/market/orderbook | 600 requests/10sec | Level 50/200/500 |
| 인덱스 가격 | /v5/market/index-price-kline | 600 requests/10sec | 1min~1M |
| 펀딩 비율 | /v5/market/funding-band | 600 requests/10sec | Real-time |
카테고리별 Historical Data 접근
Bybit v5 API에서 카테고리(category) 파라미터는 데이터 유형을 구분합니다. 스팟 데이터의 경우 category="spot"을 사용하며, USDT Perpetual은 category="linear", USDC Perpetual은 category="inverse"를 지정합니다. 이 구분은 특히 크로스 자산 백테스팅 시 필수적으로 관리해야 하는 파라미터입니다.
import aiohttp
import asyncio
import time
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import pandas as pd
class BybitHistoricalDataCollector:
"""
Bybit 스팟 Historical Data API 비동기 수집기
프로덕션 레벨 병렬 수집 지원 및 Rate Limit 자동 관리
"""
BASE_URL = "https://api.bybit.com"
def __init__(self, api_key: str = None, api_secret: str = None):
self.api_key = api_key
self.api_secret = api_secret
self.rate_limiter = asyncio.Semaphore(10) # 동시 요청 제한
self.request_timestamps = [] # Rate Limit 추적용
async def _get_klines_async(
self,
session: aiohttp.ClientSession,
symbol: str,
interval: str = "1",
start_time: int = None,
end_time: int = None,
limit: int = 1000
) -> List[Dict]:
"""
캔들스틱 데이터 비동기 수집
interval: 1, 3, 5, 15, 30, 60, 120, 240, 360, 720, D, W, M
"""
endpoint = "/v5/market/kline"
params = {
"category": "spot",
"symbol": symbol,
"interval": interval,
"limit": limit
}
if start_time:
params["start"] = start_time
if end_time:
params["end"] = end_time
async with self.rate_limiter:
try:
async with session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
data = await response.json()
if data.get("retCode") == 0:
return data.get("result", {}).get("list", [])
else:
print(f"API Error: {data.get('retMsg')}")
return []
elif response.status == 429:
# Rate Limit 도달 시 재시도
await asyncio.sleep(2)
return await self._get_klines_async(
session, symbol, interval, start_time, end_time, limit
)
else:
print(f"HTTP Error: {response.status}")
return []
except Exception as e:
print(f"Request failed: {e}")
return []
async def collect_historical_data(
self,
symbol: str,
interval: str,
days_back: int = 365
) -> pd.DataFrame:
"""
지정된 기간의 히스토리컬 데이터 전체 수집
Auto-pagination 지원
"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days_back)).timestamp() * 1000)
all_candles = []
current_start = start_time
max_retries = 5
connector = aiohttp.TCPConnector(limit=20, limit_per_host=10)
async with aiohttp.ClientSession(connector=connector) as session:
while current_start < end_time:
retries = 0
candles = []
while retries < max_retries:
candles = await self._get_klines_async(
session, symbol, interval, current_start, end_time
)
if candles:
break
retries += 1
await asyncio.sleep(1 * retries) # 지수 백오프
if not candles:
break
all_candles.extend(candles)
# 다음 페이지: 마지막 캔들의 startTime 사용
current_start = int(candles[-1][0]) + 1
# Rate Limit 방지
await asyncio.sleep(0.2)
if len(all_candles) >= 100000: # 메모리 관리
break
# DataFrame 변환
df = pd.DataFrame(all_candles, columns=[
'start_time', 'open', 'high', 'low', 'close', 'volume', 'turnover'
])
if not df.empty:
df['start_time'] = pd.to_datetime(df['start_time'].astype(int), unit='ms')
for col in ['open', 'high', 'low', 'close', 'volume', 'turnover']:
df[col] = df[col].astype(float)
df = df.sort_values('start_time').drop_duplicates()
return df
사용 예시
async def main():
collector = BybitHistoricalDataCollector()
# BTC/USDT 1년치 1시간 데이터 수집
btc_data = await collector.collect_historical_data(
symbol="BTCUSDT",
interval="60",
days_back=365
)
print(f"수집된 데이터: {len(btc_data)} 건")
print(f"기간: {btc_data['start_time'].min()} ~ {btc_data['start_time'].max()}")
print(f"평균 지연: {len(btc_data) / 365 / 24:.1f} 캔들/일")
if __name__ == "__main__":
asyncio.run(main())AI 기반 거래 신호 생성: HolySheep AI 통합
히스토리컬 데이터를 수집했다면, 다음 단계는 이 데이터에서 거래 신호를 생성하는 것입니다. HolySheep AI 게이트웨이를 활용하면 단일 API 키로 GPT-4.1, Claude, DeepSeek 등 다양한 모델을 사용하여 기술적 분석, 패턴 인식, 감정 분석 기반의 거래 신호를 생성할 수 있습니다. 특히 HolySheep의 경우 해외 신용카드 없이 로컬 결제가 가능하여 국내 개발자들이 번거로운 해외 결제 없이 즉시 시작할 수 있습니다.
멀티 모델 신호 생성 파이프라인
import os
import json
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
class AISignalGenerator:
"""
HolySheep AI 게이트웨이 기반 멀티 모델 거래 신호 생성기
비용 최적화를 위한 모델 선택 로직 포함
"""
def __init__(self, api_key: str):
# HolySheep AI 게이트웨이 - 모든 모델 통합 엔드포인트
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
# 모델별 클라이언트 초기화
self.openai_client = AsyncOpenAI(
base_url=self.base_url,
api_key=api_key
)
self.anthropic_client = AsyncAnthropic(
base_url=self.base_url,
api_key=api_key
)
# 비용 최적화 매핑 (per million tokens)
self.model_costs = {
"gpt-4.1": 8.0, # GPT-4.1: $8/MTok
"claude-sonnet-4-5": 15.0, # Claude Sonnet 4.5: $15/MTok
"gemini-2.5-flash": 2.5, # Gemini 2.5 Flash: $2.50/MTok
"deepseek-v3.2": 0.42 # DeepSeek V3.2: $0.42/MTok
}
def _create_signal_prompt(self, market_data: dict) -> str:
"""거래 신호 생성을 위한 프롬프트 구성"""
recent_candles = market_data.get("recent_candles", [])
# 최근 20개 캔들 요약
price_summary = f"""
최근 시장 데이터:
- 현재가: ${market_data.get('current_price', 0):,.2f}
- 24h 변동률: {market_data.get('daily_change', 0):.2f}%
- 거래량: {market_data.get('volume', 0):,.0f}
- 최근 추세: {market_data.get('trend', '중립')}
최근 캔들(OHLCV):
{self._format_candles(recent_candles[-20:])}
"""
return price_summary
def _format_candles(self, candles: list) -> str:
"""캔들 데이터 포맷팅"""
lines = []
for c in candles:
lines.append(
f"시간: {c['start_time']}, "
f"시가: {c['open']:.2f}, 고가: {c['high']:.2f}, "
f"저가: {c['low']:.2f}, 종가: {c['close']:.2f}"
)
return "\n".join(lines)
async def generate_signal_deepseek(
self,
market_data: dict,
strategy_type: str = "momentum"
) -> dict:
"""
DeepSeek V3.2 활용 - 비용 최적화 (가장 저렴)
빠르고 효율적인 기술적 분석 수행
"""
prompt = self._create_signal_prompt(market_data)
system_prompt = f"""당신은 전문 암호화폐 트레이더입니다.
{strategy_type} 전략에 기반하여 다음 형식으로 응답하세요:
SIGNAL: [BUY/SELL/HOLD]
CONFIDENCE: [0-100]%
ENTRY_PRICE: [숫자 또는 null]
STOP_LOSS: [숫자 또는 null]
TAKE_PROFIT: [숫자 또는 null]
REASON: [간단한 이유]
"""
try:
response = await self.openai_client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=300
)
content = response.choices[0].message.content
return self._parse_signal_response(content, "deepseek-v3.2")
except Exception as e:
print(f"DeepSeek 신호 생성 실패: {e}")
return None
async def generate_signal_gpt4(
self,
market_data: dict,
strategy_type: str = "comprehensive"
) -> dict:
"""
GPT-4.1 활용 - 복잡한 시장 분석
다중 지표 종합 분석 및 리스크 평가
"""
prompt = self._create_signal_prompt(market_data)
system_prompt = f"""당신은 헤지펀드 수준 암호화폐 애널리스트입니다.
{strategy_type} 전략에 기반하여 포괄적인 시장 분석을 수행하세요:
응답 형식:
{{
"signal": "BUY/SELL/HOLD",
"confidence": 0-100,
"entry_price": 숫자 또는 null,
"stop_loss": 숫자 또는 null,
"take_profit": 숫자 또는 null,
"reason": "상세한 이유",
"risk_level": "LOW/MEDIUM/HIGH",
"support_levels": [숫자 배열],
"resistance_levels": [숫자 배열]
}}
"""
try:
response = await self.openai_client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
temperature=0.2,
max_tokens=500
)
content = response.choices[0].message.content
return self._parse_json_signal(content, "gpt-4.1")
except Exception as e:
print(f"GPT-4.1 신호 생성 실패: {e}")
return None
async def batch_generate_signals(
self,
market_data_list: list,
use_cheap_model: bool = True
) -> list:
"""
배치 신호 생성 - 비용 최적화
실시간성이 낮고 수량이 많은 경우 cheap 모델 사용
"""
results = []
# HolySheep 게이트웨이 활용 - 모델 라우팅
for data in market_data_list:
if use_cheap_model:
# DeepSeek로 99% 비용 절감
signal = await self.generate_signal_deepseek(data)
else:
# GPT-4.1로 고품질 분석
signal = await self.generate_signal_gpt4(data)
results.append(signal)
return results
def _parse_signal_response(self, content: str, model: str) -> dict:
"""신호 응답 파싱"""
# 응답 형식 파싱 로직
lines = content.strip().split('\n')
signal_data = {}
for line in lines:
if ':' in line:
key, value = line.split(':', 1)
signal_data[key.strip().lower()] = value.strip()
signal_data['model'] = model
signal_data['cost'] = self.model_costs.get(model, 0)
return signal_data
def _parse_json_signal(self, content: str, model: str) -> dict:
"""JSON 신호 응답 파싱"""
try:
# JSON 블록 추출
if '```json' in content:
content = content.split('``json')[1].split('``')[0]
elif '```' in content:
content = content.split('``')[1].split('``')[0]
signal_data = json.loads(content.strip())
signal_data['model'] = model
signal_data['cost'] = self.model_costs.get(model, 0)
return signal_data
except json.JSONDecodeError:
return self._parse_signal_response(content, model)
비용 계산 예시
def calculate_model_costs():
"""HolySheep AI 모델별 비용 비교"""
# 월간 100만 토큰 사용 시
monthly_tokens = 1_000_000
costs = {
"GPT-4.1": monthly_tokens * 8.0 / 1_000_000, # $8/MTok
"Claude Sonnet 4.5": monthly_tokens * 15.0 / 1_000_000, # $15/MTok
"Gemini 2.5 Flash": monthly_tokens * 2.5 / 1_000_000, # $2.50/MTok
"DeepSeek V3.2": monthly_tokens * 0.42 / 1_000_000, # $0.42/MTok
}
print("월간 100만 토큰 사용 시 비용:")
for model, cost in costs.items():
print(f" {model}: ${cost:.2f}")
# DeepSeek vs GPT-4.1 비용 절감율
savings = (costs["GPT-4.1"] - costs["DeepSeek V3.2"]) / costs["GPT-4.1"] * 100
print(f"\nDeepSeek 사용 시 GPT-4.1 대비 {savings:.1f}% 비용 절감")
if __name__ == "__main__":
# HolySheep API 키 설정
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
generator = AISignalGenerator(API_KEY)
calculate_model_costs()Quantitative 백테스팅 프레임워크 설계
데이터 수집과 AI 신호 생성이 준비되었다면, 이를 종합하는 백테스팅 엔진이 필요합니다. 프로덕션 수준의 백테스팅 프레임워크는 단순히 수익률 계산에 그치지 않고, 슬리피지, 수수료,流动性 제약, 하드포크·이벤트 기반 시장 변동 등 현실적 요소를 반영해야 합니다. 저는 과거 3년간 다양한量化 펀드에서 백테스팅 시스템을 구축하면서, 데이터 품질이 전략 성능의 70% 이상을 결정한다는 사실을 깨달았습니다.
벡터라이즈 백테스팅 vs 이벤트 기반 백테스팅
백테스팅 방식은 크게 벡터라이즈 방식과 이벤트 기반方式来 나뉩니다. 벡터라이즈 방식은 pandas 연산으로 전체 데이터셋을 한 번에 처리하여高速이지만 복잡한 전략에는 한계가 있습니다. 반면 이벤트 기반 방식은 각 캔들마다 신호를 검증하여 현실적인 시뮬레이션이 가능하지만 속도가 느립니다. 저는 초기 탐색 단계에서는 벡터라이즈 방식을, 최종 검증 단계에서는 이벤트 기반 방식을 사용하는 하이브리드 접근법을 권장합니다.
import numpy as np
import pandas as pd
from typing import List, Dict, Tuple, Callable
from dataclasses import dataclass, field
from enum import Enum
from collections import deque
import warnings
class PositionSide(Enum):
LONG = 1
SHORT = -1
FLAT = 0
@dataclass
class Trade:
"""거래 단위"""
entry_time: pd.Timestamp
exit_time: pd.Timestamp
side: PositionSide
entry_price: float
exit_price: float
size: float
pnl: float
pnl_pct: float
commission: float
slippage: float
@dataclass
class BacktestConfig:
"""백테스팅 설정"""
initial_capital: float = 10000.0
commission_rate: float = 0.001 # 0.1% taker fee
slippage_bps: float = 5 # 5 basis points
max_position_size: float = 1.0 # 최대 포지션 비율
leverage: float = 1.0
risk_free_rate: float = 0.04 # 연 4%
@dataclass
class BacktestResult:
"""백테스팅 결과"""
trades: List[Trade]
equity_curve: pd.Series
metrics: Dict = field(default_factory=dict)
class VectorizedBacktester:
"""
벡터라이즈 기반 고속 백테스팅 엔진
대량 데이터 테스트 및 파라미터 스위핑에 최적화
"""
def __init__(self, config: BacktestConfig = None):
self.config = config or BacktestConfig()
def run(
self,
df: pd.DataFrame,
signals: pd.Series,
price_col: str = 'close'
) -> BacktestResult:
"""
벡터라이즈 백테스트 실행
signals: 1=LONG, -1=SHORT, 0=FLAT
"""
df = df.copy()
df['signal'] = signals.shift(1).fillna(0) # 신호 발생 다음 캔들 실행
df['price'] = df[price_col]
# 수익률 계산
df['returns'] = df['price'].pct_change()
df['strategy_returns'] = df['signal'].shift(1) * df['returns']
# 슬리피지 적용
df['slippage_cost'] = np.abs(df['signal'].diff()) * self.config.slippage_bps / 10000
df['strategy_returns'] -= df['slippage_cost']
# 수수료 적용
df['commission_cost'] = np.abs(df['signal'].diff()) * self.config.commission_rate
df['strategy_returns'] -= df['commission_cost']
# 에쿼티 커브
df['equity'] = (1 + df['strategy_returns']).cumprod() * self.config.initial_capital
# 거래 목록 생성
trades = self._extract_trades(df, price_col)
# 지표 계산
metrics = self._calculate_metrics(df, trades)
return BacktestResult(
trades=trades,
equity_curve=df['equity'],
metrics=metrics
)
def _extract_trades(self, df: pd.DataFrame, price_col: str) -> List[Trade]:
"""벡터라이즈 트레이드 추출"""
trades = []
# 포지션 변경 지점 식별
position_changes = df[df['signal'].diff() != 0].index
if len(position_changes) == 0:
return trades
for i in range(len(position_changes)):
entry_idx = position_changes[i]
exit_idx = position_changes[i + 1] if i + 1 < len(position_changes) else df.index[-1]
entry_price = df.loc[entry_idx, price_col]
exit_price = df.loc[exit_idx, price_col]
side = PositionSide(df.loc[entry_idx, 'signal'])
# 슬리피지 적용
slippage = entry_price * self.config.slippage_bps / 10000
if side == PositionSide.LONG:
execution_price = entry_price * (1 + self.config.slippage_bps / 10000)
else:
execution_price = entry_price * (1 - self.config.slippage_bps / 10000)
pnl_pct = (exit_price - execution_price) / execution_price * side.value
pnl = self.config.initial_capital * pnl_pct
commission = self.config.initial_capital * self.config.commission_rate * 2
trades.append(Trade(
entry_time=entry_idx,
exit_time=exit_idx,
side=side,
entry_price=execution_price,
exit_price=exit_price,
size=self.config.initial_capital,
pnl=pnl,
pnl_pct=pnl_pct * 100,
commission=commission,
slippage=slippage * self.config.initial_capital
))
return trades
def _calculate_metrics(self, df: pd.DataFrame, trades: List[Trade]) -> Dict:
"""성과 지표 계산"""
if len(trades) == 0:
return {}
equity = df['equity']
returns = df['strategy_returns'].dropna()
total_return = (equity.iloc[-1] / equity.iloc[0] - 1) * 100
n_days = (df.index[-1] - df.index[0]).days
# 연환산 수익률
annual_return = ((1 + total_return / 100) ** (365 / n_days) - 1) * 100 if n_days > 0 else 0
# 일일 수익률 통계
daily_returns = returns.dropna()
sharpe_ratio = (
daily_returns.mean() / daily_returns.std() * np.sqrt(365)
if daily_returns.std() > 0 else 0
)
# 최대 낙폭
rolling_max = equity.cummax()
drawdown = (equity - rolling_max) / rolling_max * 100
max_drawdown = drawdown.min()
# 승률
winning_trades = [t for t in trades if t.pnl > 0]
win_rate = len(winning_trades) / len(trades) * 100 if trades else 0
# 평균 수익/손실
avg_win = np.mean([t.pnl for t in winning_trades]) if winning_trades else 0
losing_trades = [t for t in trades if t.pnl <= 0]
avg_loss = np.mean([t.pnl for t in losing_trades]) if losing_trades else 0
profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else 0
# Calmar 비율
calmar_ratio = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0
return {
'total_return': total_return,
'annual_return': annual_return,
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'win_rate': win_rate,
'total_trades': len(trades),
'profit_factor': profit_factor,
'calmar_ratio': calmar_ratio,
'avg_win': avg_win,
'avg_loss': avg_loss,
'avg_trade_pnl': np.mean([t.pnl for t in trades]) if trades else 0
}
class EventDrivenBacktester:
"""
이벤트 기반 백테스팅 엔진
신호 생성 함수, 포지션 관리, 리스크 관리 커스터마이징 지원
"""
def __init__(
self,
config: BacktestConfig,
signal_generator: Callable,
position_sizer: Callable = None,
risk_manager: Callable = None
):
self.config = config
self.signal_generator = signal_generator
self.position_sizer = position_sizer or self._default_position_sizer
self.risk_manager = risk_manager or self._default_risk_manager
self.cash = config.initial_capital
self.position = 0
self.position_side = PositionSide.FLAT
self.trades = []
self.equity_history = []
def _default_position_sizer(self, signal: int, equity: float, price: float) -> float:
"""기본 포지션 사이징"""
return equity * self.config.max_position_size / price
def _default_risk_manager(
self,
position: float,
price: float,
signal: int,
candle: pd.Series
) -> Tuple[float, float, float]:
"""
기본 리스크 관리
returns: (stop_loss, take_profit, position_size)
"""
risk_pct = 0.02 # 2% 리스크
atr = (candle['high'] - candle['low']) * 2 # 단순 ATR
stop_loss = price * (1 - risk_pct) if signal > 0 else price * (1 + risk_pct)
take_profit = price * (1 + risk_pct * 2) if signal > 0 else price * (1 - risk_pct * 2)
return stop_loss, take_profit, self.position
def run(self, df: pd.DataFrame, signals: pd.Series = None) -> BacktestResult:
"""이벤트 기반 백테스트 실행"""
self.cash = self.config.initial_capital
self.position = 0
self.position_side = PositionSide.FLAT
self.trades = []
self.equity_history = []
pending_order = None
for idx, row in df.iterrows():
current_price = row['close']
current_time = idx
# Equity 업데이트
equity = self.cash + self.position * current_price
self.equity_history.append((idx, equity))
# 신호 생성 (외부 함수 또는 시그널 시리즈)
if signals is not None:
signal = signals.loc[idx] if idx in signals.index else 0
else:
signal = self.signal_generator(row, self.position_side)
# Stop Loss / Take Profit 체크
if pending_order:
sl, tp = pending_order['sl'], pending_order['tp']
if self.position_side == PositionSide.LONG:
if current_price <= sl:
self._close_position(current_price, current_time, "Stop Loss")
pending_order = None
elif current_price >= tp:
self._close_position(current_price, current_time, "Take Profit")
pending_order = None
elif self.position_side == PositionSide.SHORT:
if current_price >= sl:
self._close_position(current_price, current_time, "Stop Loss")
pending_order = None
elif current_price <= tp:
self._close_position(current_price, current_time, "Take Profit")
pending_order = None
# 신호 기반 거래
if signal != 0 and self.position_side == PositionSide.FLAT:
# 포지션 오픈
sl, tp, position_size = self._default_risk_manager(
self.position, current_price, signal, row
)
# 슬리피지 적용
execution_price = current_price * (
1 + self.config.slippage_bps / 10000 if signal > 0
else 1 - self.config.slippage_bps / 10000
)
self.cash -= position_size * execution_price
self.cash -= position_size * execution_price * self.config.commission_rate
self.position = position_size * signal
self.position_side = PositionSide(signal)
pending_order = {'sl': sl, 'tp': tp}
elif signal == 0 and self.position_side != PositionSide.FLAT:
# 포지션 클로즈
execution_price = current_price * (
1 - self.config.slippage_bps / 10000 if self.position > 0
else 1 + self.config.slippage_bps / 10000
)
self._close_position(execution_price, current_time, "Signal Exit")
pending_order = None
# 잔여 포지션 정리
if self.position != 0:
final_price = df.iloc[-1]['close']
self._close_position(final_price, df.index[-1], "End of Backtest")
# 결과 구성
equity_curve = pd.Series(
[e[1] for e in self.equity_history],
index=[e[0] for e in self.equity_history]
)
metrics = VectorizedBacktester(self.config)._calculate_metrics(
pd.DataFrame({'equity': equity_curve}), self.trades
)
return BacktestResult(
trades=self.trades,
equity_curve=equity_curve,
metrics=metrics
)
def _close_position(self, price: float, time: pd.Timestamp, reason: str):
"""포지션 클로즈"""
pnl = self.position * (price - self.entry_price) if hasattr(self, 'entry_price') else 0
commission = abs(self.position) * price * self.config.commission_rate
self.trades.append(Trade(
entry_time=self.entry_time,
exit_time=time,
side=self.position_side,
entry_price=self.entry_price,
exit_price=price,
size=abs(self.position),
pnl=pnl - commission,
pnl_pct=pnl / (abs(self.position) * self.entry_price) * 100,
commission=commission,
slippage=0
))
self.cash += self.position * price - commission
self.position = 0
self.position_side = PositionSide.FLAT
사용 예시
def run_complete_backtest():
"""완전한 백테스트 파이프라인 실행"""
# 1. 데이터 수집 (이전 섹션의 collector 사용)
collector = BybitHistoricalDataCollector()
# 2. 백테스팅 설정
config = BacktestConfig(
initial_capital=10000,
commission_rate=0.001,
slippage_bps=5,
max_position_size=0.1, # 최대 10% 포지션
leverage=1.0
)
# 3. 벡터라이즈 백테스터 실행
vectorized_tester = VectorizedBacktester(config)
# 더미 데이터로 테스트
dates = pd.date_range('2023-01-01', '2024-01-01', freq='1h')
np.random.seed(42)
dummy_data = pd.DataFrame({
'start_time': dates,
'open': 30000 + np.cumsum(np.random.randn(len(dates)) * 100),
'high': 30000 + np.cumsum(np.random.randn(len(dates)) * 100) * 1.02,
'low': 30000 + np.cumsum(np.random.randn(len(dates)) * 100) * 0.98,
'close': 30000 + np.cumsum(np.random.randn(len(dates)) * 100),
'volume': np.random.randint(1000, 10000, len(dates))
}, index=dates)
dummy_data['high'] = dummy_data[['open', 'close', 'high']].max(axis=1)
dummy_data['low'] = dummy_data[['open', 'close', 'low']].min(axis=1)
# 단순 이동평균 크로스오버 신호
dummy_data['ma_fast'] = dummy_data['close'].rolling(20).mean()
dummy_data['ma_slow'] = dummy_data['close'].rolling(50).mean()
dummy_data['signal'] = np.where(
dummy_data