저는 3년 넘게 헤지펀드에서 퀀트 트레이딩 시스템을 개발해온 엔지니어입니다. 이번 글에서는 HolySheep AI를 활용하여 프로덕션 수준의 AI 트레이딩 전략 백테스터를 구축하는 방법을 상세히 다룹니다. HolySheep의 글로벌 AI API 게이트웨이 기능을 통해 단일 API 키로 여러 모델을 조합하고, 비용을 최적화하며, 지연 시간을 최소화하는 아키텍처를 설계하겠습니다.
아키텍처 개요
AI 트레이딩 백테스터의 핵심 요구사항은 다음과 같습니다:
- 대량 데이터 처리: 수년치 분단위 OHLCV 데이터 분석
- 복잡한 신호 생성: 다중 모델 협업으로 시장 심리, 패턴 인식 수행
- 빠른 이터레이션: 수백 개의 전략 변수를 동시에 테스트
- 비용 효율성: 수백만 건의 API 호출 비용 최적화
프로젝트 구조
ai-trading-backtester/
├── src/
│ ├── core/
│ │ ├── backtest_engine.py # 메인 백테스트 엔진
│ │ ├── portfolio.py # 포트폴리오 시뮬레이션
│ │ └── risk_manager.py # 리스크 관리
│ ├── models/
│ │ ├── signal_generator.py # AI 신호 생성기
│ │ ├── market_analyzer.py # 시장 분석기
│ │ └── sentiment_analyzer.py # 감정 분석기
│ ├── data/
│ │ ├── data_loader.py # 데이터 로딩
│ │ └── feature_engineering.py # 피처 엔지니어링
│ ├── api/
│ │ └── holysheep_client.py # HolySheep API 클라이언트
│ └── utils/
│ ├── cache.py # 응답 캐싱
│ └── rate_limiter.py # Rate Limiting
├── tests/
├── config.yaml
├── requirements.txt
└── main.py
HolySheep API 클라이언트 구현
가장 먼저 HolySheep AI API 클라이언트를 구현합니다. HolySheep는 단일 API 키로 GPT-4.1, Claude Sonnet, Gemini, DeepSeek 등 모든 주요 모델을 지원합니다. 이를 통해 모델별 강점을 활용하는 앙상블 전략을 구현할 수 있습니다.
# src/api/holysheep_client.py
import os
import json
import time
import hashlib
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
from openai import OpenAI
import anthropic
import httpx
@dataclass
class ModelConfig:
"""모델별 설정 및 가격 정보"""
model_id: str
provider: str # openai, anthropic, google
cost_per_mtok: float # 달러 단위
cost_per_ktok: float
max_tokens: int
avg_latency_ms: int # 예상 평균 지연 시간
class HolySheepAIClient:
"""
HolySheep AI 게이트웨이 클라이언트
단일 API 키로 여러 AI 모델 지원
"""
BASE_URL = "https://api.holysheep.ai/v1"
# 모델별 비용 및 성능 프로파일
MODELS = {
"gpt_4.1": ModelConfig(
model_id="gpt-4.1",
provider="openai",
cost_per_mtok=8.00,
cost_per_ktok=3.00,
max_tokens=128000,
avg_latency_ms=850
),
"claude_sonnet_4": ModelConfig(
model_id="claude-sonnet-4-5",
provider="anthropic",
cost_per_mtok=15.00,
cost_per_ktok=7.50,
max_tokens=200000,
avg_latency_ms=920
),
"gemini_2.5_flash": ModelConfig(
model_id="gemini-2.5-flash",
provider="google",
cost_per_mtok=2.50,
cost_per_ktok=0.50,
max_tokens=1000000,
avg_latency_ms=380
),
"deepseek_v3": ModelConfig(
model_id="deepseek-v3.2",
provider="openai-compatible",
cost_per_mtok=0.42,
cost_per_ktok=0.14,
max_tokens=64000,
avg_latency_ms=520
)
}
def __init__(self, api_key: str):
self.api_key = api_key
self.cache: Dict[str, Any] = {}
self.cache_hits = 0
self.cache_misses = 0
self.total_cost = 0.0
self.total_tokens = 0
self._setup_clients()
def _setup_clients(self):
"""각 제공자별 클라이언트 설정"""
# OpenAI 호환 클라이언트 (GPT, DeepSeek)
self.openai_client = OpenAI(
api_key=self.api_key,
base_url=self.BASE_URL,
timeout=60.0,
max_retries=3
)
# Anthropic 클라이언트
self.anthropic_client = anthropic.Anthropic(
api_key=self.api_key,
base_url=self.BASE_URL,
timeout=60.0
)
# Google AI용 HTTP 클라이언트
self.google_http = httpx.Client(
base_url=self.BASE_URL,
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=60.0
)
def _get_cache_key(self, model: str, messages: List[Dict]) -> str:
"""캐시 키 생성"""
content = json.dumps({"model": model, "messages": messages}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
def _get_from_cache(self, cache_key: str) -> Optional[Dict]:
"""캐시에서 응답 조회"""
if cache_key in self.cache:
self.cache_hits += 1
return self.cache[cache_key]
self.cache_misses += 1
return None
def _save_to_cache(self, cache_key: str, response: Dict, ttl: int = 3600):
"""응답 캐싱"""
self.cache[cache_key] = {
"response": response,
"timestamp": time.time(),
"ttl": ttl
}
def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 2048,
use_cache: bool = True,
cache_ttl: int = 3600
) -> Dict[str, Any]:
"""
범용 채팅 완료 API
Args:
model: 모델 ID (gpt_4.1, claude_sonnet_4, gemini_2.5_flash, deepseek_v3)
messages: 메시지 리스트
temperature: 창의성 레벨
max_tokens: 최대 토큰 수
use_cache: 캐시 사용 여부
cache_ttl: 캐시 TTL (초)
"""
if use_cache:
cache_key = self._get_cache_key(model, messages)
cached = self._get_from_cache(cache_key)
if cached:
return cached["response"]
model_config = self.MODELS.get(model)
if not model_config:
raise ValueError(f"Unknown model: {model}")
start_time = time.time()
try:
if model_config.provider in ["openai", "openai-compatible"]:
response = self.openai_client.chat.completions.create(
model=model_config.model_id,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
result = {
"content": response.choices[0].message.content,
"model": model,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
},
"latency_ms": int((time.time() - start_time) * 1000)
}
elif model_config.provider == "anthropic":
response = self.anthropic_client.messages.create(
model=model_config.model_id,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
result = {
"content": response.content[0].text,
"model": model,
"usage": {
"prompt_tokens": response.usage.input_tokens,
"completion_tokens": response.usage.output_tokens,
"total_tokens": response.usage.input_tokens + response.usage.output_tokens
},
"latency_ms": int((time.time() - start_time) * 1000)
}
else:
# Google Gemini 호환 API
response = self.google_http.post(
"/chat/completions",
json={
"model": model_config.model_id,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
response.raise_for_status()
data = response.json()
result = {
"content": data["choices"][0]["message"]["content"],
"model": model,
"usage": data.get("usage", {}),
"latency_ms": int((time.time() - start_time) * 1000)
}
# 비용 계산
total_tokens = result["usage"]["total_tokens"]
cost = (total_tokens / 1_000_000) * model_config.cost_per_mtok
result["cost"] = cost
self.total_cost += cost
self.total_tokens += total_tokens
if use_cache:
self._save_to_cache(cache_key, result, cache_ttl)
return result
except Exception as e:
print(f"API Error for {model}: {str(e)}")
raise
def batch_chat(
self,
requests: List[Dict[str, Any]],
model: str = "deepseek_v3",
max_concurrency: int = 10
) -> List[Dict[str, Any]]:
"""
배치 요청 처리 (비용 최적화용)
"""
import asyncio
from concurrent.futures import ThreadPoolExecutor
def single_request(req):
return self.chat_completion(
model=model,
messages=req["messages"],
temperature=req.get("temperature", 0.7),
max_tokens=req.get("max_tokens", 1024)
)
with ThreadPoolExecutor(max_workers=max_concurrency) as executor:
results = list(executor.map(single_request, requests))
return results
def get_stats(self) -> Dict[str, Any]:
"""사용 통계 반환"""
return {
"total_cost_usd": round(self.total_cost, 4),
"total_tokens": self.total_tokens,
"cache_hits": self.cache_hits,
"cache_misses": self.cache_misses,
"cache_hit_rate": round(
self.cache_hits / max(1, self.cache_hits + self.cache_misses) * 100, 2
)
}
def clear_cache(self):
"""캐시 초기화"""
self.cache.clear()
self.cache_hits = 0
self.cache_misses = 0
모듈 단위 테스트
if __name__ == "__main__":
client = HolySheepAIClient(api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"))
# DeepSeek V3.2로 간단한 테스트
response = client.chat_completion(
model="deepseek_v3",
messages=[
{"role": "system", "content": "당신은 금융 데이터 분석 전문가입니다."},
{"role": "user", "content": "BTC/USDT가 60,000달러 이상일 때의 시장 심리을 분석해주세요."}
],
max_tokens=500
)
print(f"Response: {response['content']}")
print(f"Latency: {response['latency_ms']}ms")
print(f"Cost: ${response['cost']:.4f}")
print(f"Stats: {client.get_stats()}")
AI 신호 생성 시스템
트레이딩 신호 생성은 다중 모델 협업으로 수행됩니다. 각 모델의 강점을 활용하여:
- DeepSeek V3.2: 구조화된 데이터 분석, 패턴 인식 (가장 저렴: $0.42/MTok)
- Gemini 2.5 Flash: 빠른 기술적 지표 계산 (저렴+고속: $2.50/MTok, ~380ms)
- Claude Sonnet 4: 고급 시장 심리 분석, 리스크 평가
- GPT-4.1: 종합 전략 결정 및 최종 신호 생성
# src/models/signal_generator.py
from dataclasses import dataclass
from typing import Optional, List, Dict
from enum import Enum
import pandas as pd
from datetime import datetime
class SignalType(Enum):
"""트레이딩 신호 타입"""
BUY = "BUY"
SELL = "SELL"
HOLD = "HOLD"
STRONG_BUY = "STRONG_BUY"
STRONG_SELL = "STRONG_SELL"
@dataclass
class TradingSignal:
"""트레이딩 신호"""
timestamp: datetime
symbol: str
signal_type: SignalType
confidence: float # 0.0 ~ 1.0
target_price: Optional[float]
stop_loss: Optional[float]
position_size: float # 0.0 ~ 1.0
reasoning: str
models_used: List[str]
api_cost: float
latency_ms: int
class AITradingSignalGenerator:
"""
AI 기반 트레이딩 신호 생성기
HolySheep의 다중 모델 협업으로 고품질 신호 생성
"""
def __init__(self, holysheep_client):
self.client = holysheep_client
self.usage_stats = {
"total_signals": 0,
"total_cost": 0.0,
"avg_confidence": 0.0,
"model_usage": {}
}
def generate_signal(
self,
symbol: str,
current_price: float,
ohlcv_data: pd.DataFrame,
market_context: str = ""
) -> TradingSignal:
"""
단일 심볼에 대한 트레이딩 신호 생성
3단계 파이프라인:
1. 기술적 분석 (Gemini 2.5 Flash)
2. 시장 심리 분석 (Claude Sonnet 4)
3. 최종 신호 결정 (GPT-4.1)
"""
start_time = datetime.now()
models_used = []
total_cost = 0.0
# ─────────────────────────────────────────────────────────
# Stage 1: 기술적 분석 (Gemini 2.5 Flash - 고속, 저렴)
# ─────────────────────────────────────────────────────────
tech_analysis = self._analyze_technical(
symbol, current_price, ohlcv_data
)
models_used.append("gemini_2.5_flash")
total_cost += tech_analysis.get("cost", 0)
# ─────────────────────────────────────────────────────────
# Stage 2: 시장 심리 분석 (Claude Sonnet 4 - 고급 분석)
# ─────────────────────────────────────────────────────────
sentiment_analysis = self._analyze_sentiment(
symbol, current_price, tech_analysis, market_context
)
models_used.append("claude_sonnet_4")
total_cost += sentiment_analysis.get("cost", 0)
# ─────────────────────────────────────────────────────────
# Stage 3: 최종 신호 결정 (GPT-4.1 - 최고 품질)
# ─────────────────────────────────────────────────────────
final_decision = self._make_final_decision(
symbol, current_price, tech_analysis, sentiment_analysis
)
models_used.append("gpt_4.1")
total_cost += final_decision.get("cost", 0)
# ─────────────────────────────────────────────────────────
# 신호 구성
# ─────────────────────────────────────────────────────────
signal = TradingSignal(
timestamp=start_time,
symbol=symbol,
signal_type=SignalType(final_decision["signal"]),
confidence=final_decision["confidence"],
target_price=final_decision.get("target_price"),
stop_loss=final_decision.get("stop_loss"),
position_size=final_decision["position_size"],
reasoning=final_decision["reasoning"],
models_used=models_used,
api_cost=total_cost,
latency_ms=int((datetime.now() - start_time).total_seconds() * 1000)
)
# 통계 업데이트
self._update_stats(signal)
return signal
def _analyze_technical(
self,
symbol: str,
price: float,
ohlcv: pd.DataFrame
) -> Dict:
"""기술적 분석 - Gemini 2.5 Flash 사용"""
# 피처 데이터 구성
recent_data = ohlcv.tail(30).to_dict('records')
prompt = f"""[기술적 분석 요청]
심볼: {symbol}
현재가: ${price:,.2f}
최근 30일 OHLCV 데이터:
{recent_data}
다음 항목을 분석해주세요:
1. RSI (Relative Strength Index) 해석
2. MACD 시그널 및 히스토그램
3. 이동평균선 골든크로스/데드크로스
4. 볼린저밴드 위치 및 폭
5. 거래량 추세
JSON 형식으로 결과를 반환:
{{"indicators": {{...}}, "summary": "...", "bias": "bullish|bearish|neutral"}}
"""
response = self.client.chat_completion(
model="gemini_2.5_flash",
messages=[
{"role": "system", "content": "당신은 전문 기술적 분석가입니다. JSON만 반환하세요."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=1500,
use_cache=True # 동일 데이터 캐싱
)
import json
try:
analysis = json.loads(response["content"])
analysis["cost"] = response["cost"]
analysis["latency_ms"] = response["latency_ms"]
return analysis
except:
return {"summary": response["content"], "bias": "neutral", "cost": response["cost"]}
def _analyze_sentiment(
self,
symbol: str,
price: float,
tech_analysis: Dict,
market_context: str
) -> Dict:
"""시장 심리 분석 - Claude Sonnet 4 사용"""
prompt = f"""[시장 심리 분석 요청]
심볼: {symbol}
현재가: ${price:,.2f}
기술적 분석 결과:
{tech_analysis.get('summary', 'N/A')}
Bias: {tech_analysis.get('bias', 'neutral')}
시장 맥락 정보:
{market_context or '일반 시장 상황'}
분석 요청사항:
1. 투자자 심리 상태 (공포/탐욕 지수 기반)
2. 기관 투자자动向
3. 뉴스/이벤트 영향도
4. 위험 요소 평가
JSON 형식으로 반환:
{{"sentiment_score": -100~100, "key_factors": [...], "risk_level": "low|medium|high", "outlook": "..."}}
"""
response = self.client.chat_completion(
model="claude_sonnet_4",
messages=[
{"role": "system", "content": "당신은 행동 금융 전문가입니다. 구조화된 JSON만 반환하세요."},
{"role": "user", "content": prompt}
],
temperature=0.4,
max_tokens=1200
)
import json
try:
analysis = json.loads(response["content"])
analysis["cost"] = response["cost"]
return analysis
except:
return {"sentiment_score": 0, "outlook": "중립", "risk_level": "medium", "cost": response["cost"]}
def _make_final_decision(
self,
symbol: str,
price: float,
tech_analysis: Dict,
sentiment_analysis: Dict
) -> Dict:
"""최종 트레이딩 신호 결정 - GPT-4.1 사용"""
prompt = f"""[트레이딩 신호 결정]
심볼: {symbol}
현재가: ${price:,.2f}
기술적 분석:
-Bias: {tech_analysis.get('bias', 'neutral')}
-요약: {tech_analysis.get('summary', 'N/A')}
심리 분석:
-점수: {sentiment_analysis.get('sentiment_score', 0)}/100
-리스크: {sentiment_analysis.get('risk_level', 'medium')}
-전망: {sentiment_analysis.get('outlook', '중립')}
상세 신호를 생성해주세요:
1. 신호: STRONG_BUY, BUY, HOLD, SELL, STRONG_SELL
2. 신뢰도: 0.0~1.0
3. 목표가: 현재가 대비 상승률 (매수 시)
4.止损: 현재가 대비 하락률 (매도 시)
5. 포지션 크기: 0.0~1.0
6. 이유: 상세한 설명
JSON 형식으로 반환:
{{"signal": "...", "confidence": 0.0~1.0, "target_price": null 또는 숫자, "stop_loss": null 또는 숫자, "position_size": 0.0~1.0, "reasoning": "..."}}
"""
response = self.client.chat_completion(
model="gpt_4.1",
messages=[
{"role": "system", "content": "당신은 전문 퀀트 트레이더입니다. 최고의 품질의 신호만 생성하세요. JSON만 반환하세요."},
{"role": "user", "content": prompt}
],
temperature=0.2,
max_tokens=800
)
import json
try:
decision = json.loads(response["content"])
decision["cost"] = response["cost"]
return decision
except:
return {
"signal": "HOLD",
"confidence": 0.5,
"position_size": 0.0,
"reasoning": "파싱 오류로 인한 중립 신호",
"cost": response["cost"]
}
def batch_generate_signals(
self,
symbols: List[str],
price_data: Dict[str, float],
ohlcv_data: Dict[str, pd.DataFrame],
market_context: str = ""
) -> List[TradingSignal]:
"""배치 신호 생성 (비용 최적화)"""
# DeepSeek V3.2로 비용 효율적인 일괄 분석
requests = []
for symbol in symbols:
prompt = f"{symbol} ({price_data[symbol]})의 간략 기술 분석"
requests.append({
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
})
# 배치 처리 (동시 요청으로 지연 시간 최소화)
responses = self.client.batch_chat(requests, model="deepseek_v3", max_concurrency=20)
signals = []
for symbol, response in zip(symbols, responses):
# 간단 분석 후 HOLD 신호 생성 (실제로는 위 파이프라인 사용)
signals.append(TradingSignal(
timestamp=datetime.now(),
symbol=symbol,
signal_type=SignalType.HOLD,
confidence=0.5,
target_price=None,
stop_loss=None,
position_size=0.0,
reasoning=response["content"][:200],
models_used=["deepseek_v3"],
api_cost=response["cost"],
latency_ms=response["latency_ms"]
))
return signals
def _update_stats(self, signal: TradingSignal):
"""사용 통계 업데이트"""
self.usage_stats["total_signals"] += 1
self.usage_stats["total_cost"] += signal.api_cost
prev_avg = self.usage_stats["avg_confidence"]
n = self.usage_stats["total_signals"]
self.usage_stats["avg_confidence"] = prev_avg + (signal.confidence - prev_avg) / n
for model in signal.models_used:
self.usage_stats["model_usage"][model] = \
self.usage_stats["model_usage"].get(model, 0) + 1
def get_usage_report(self) -> Dict:
"""사용량 리포트 반환"""
return {
**self.usage_stats,
"avg_cost_per_signal": self.usage_stats["total_cost"] / max(1, self.usage_stats["total_signals"])
}
테스트 코드
if __name__ == "__main__":
import os
# HolySheep API 초기화
client = HolySheepAIClient(api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"))
generator = AITradingSignalGenerator(client)
# 샘플 데이터 생성
import numpy as np
dates = pd.date_range(start='2024-01-01', periods=100, freq='D')
ohlcv = pd.DataFrame({
'open': 60000 + np.random.randn(100).cumsum() * 500,
'high': 60500 + np.random.randn(100).cumsum() * 500,
'low': 59500 + np.random.randn(100).cumsum() * 500,
'close': 60000 + np.random.randn(100).cumsum() * 500,
'volume': np.random.randint(1000000, 5000000, 100)
})
# 신호 생성 테스트
signal = generator.generate_signal(
symbol="BTC/USDT",
current_price=60234.56,
ohlcv_data=ohlcv,
market_context="BTC ETF 승인 기대감, 미联储 금리 동결 전망"
)
print(f"Signal: {signal.signal_type.value}")
print(f"Confidence: {signal.confidence:.2%}")
print(f"Position Size: {signal.position_size:.2%}")
print(f"API Cost: ${signal.api_cost:.4f}")
print(f"Latency: {signal.latency_ms}ms")
print(f"Models Used: {signal.models_used}")
print(f"\nReasoning:\n{signal.reasoning}")
print(f"\n{generator.get_usage_report()}")
백테스트 엔진 구현
# src/core/backtest_engine.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from datetime import datetime, timedelta
from enum import Enum
import pandas as pd
import numpy as np
class PositionStatus(Enum):
NO_POSITION = "NO_POSITION"
LONG = "LONG"
SHORT = "SHORT"
@dataclass
class Trade:
"""거래 내역"""
entry_time: datetime
exit_time: Optional[datetime]
entry_price: float
exit_price: Optional[float]
size: float
pnl: float
pnl_pct: float
signal: str
commission: float
@dataclass
class Portfolio:
"""포트폴리오 상태"""
initial_capital: float
current_capital: float
positions: Dict[str, float] = field(default_factory=dict)
trades: List[Trade] = field(default_factory=list)
equity_curve: List[float] = field(default_factory=list)
daily_returns: List[float] = field(default_factory=list)
def open_position(self, symbol: str, price: float, size: float, commission_rate: float = 0.001):
"""포지션 진입"""
cost = price * size
commission = cost * commission_rate
self.current_capital -= (cost + commission)
self.positions[symbol] = {
"size": size,
"entry_price": price,
"entry_time": datetime.now(),
"commission": commission
}
def close_position(self, symbol: str, exit_price: float, commission_rate: float = 0.001) -> Trade:
"""포지션 청산"""
if symbol not in self.positions:
raise ValueError(f"No position for {symbol}")
pos = self.positions[symbol]
size = pos["size"]
entry_price = pos["entry_price"]
entry_time = pos["entry_time"]
total_commission = pos["commission"] + (exit_price * size * commission_rate)
pnl = (exit_price - entry_price) * size - total_commission
pnl_pct = (exit_price / entry_price - 1) * 100
self.current_capital += (exit_price * size - total_commission)
trade = Trade(
entry_time=entry_time,
exit_time=datetime.now(),
entry_price=entry_price,
exit_price=exit_price,
size=size,
pnl=pnl,
pnl_pct=pnl_pct,
signal="AI_Signal",
commission=total_commission
)
self.trades.append(trade)
del self.positions[symbol]
return trade
class BacktestEngine:
"""
AI 트레이딩 전략 백테스트 엔진
HolySheep API를 활용한 AI 신호 기반 백테스팅
"""
def __init__(
self,
initial_capital: float = 100_000,
commission_rate: float = 0.001,
slippage: float = 0.0005
):
self.initial_capital = initial_capital
self.commission_rate = commission_rate
self.slippage = slippage
self.portfolio = Portfolio(
initial_capital=initial_capital,
current_capital=initial_capital
)
self.results: Optional[Dict] = None
def run(
self,
data: pd.DataFrame,
signal_generator: Callable,
rebalance_freq: str = "1D" # 일별 리밸런싱
) -> Dict:
"""
백테스트 실행
Args:
data: OHLCV 데이터 (DateTimeIndex 필요)
signal_generator: 신호 생성 함수
rebalance_freq: 리밸런싱 주기
"""
data = data.copy()
data = data.sort_index()
# 리밸런싱 일程 추출
rebalance_dates = data.resample(rebalance_freq).first().index
current_idx = 0
for rebal_date in rebalance_dates:
if rebal_date not in data.index:
continue
# 현재 시점 데이터
current_data = data.loc[:rebal_date].tail(30)
current_price = data.loc[rebal_date, 'close']
symbol = "UNKNOWN"
# 기존 포지션 청산 (스탑로스/테이크프로핃 적용)
self._check_exit_conditions(data.loc[rebal_date], symbol)
# AI 신호 생성
signal = signal_generator(
symbol=symbol,
current_price=current_price,
ohlcv_data=current_data,
market_context=f"Backtest Date: {rebal_date}"
)
# 신호에 따른 거래 실행
self._execute_signal(signal, current_price, rebal_date)
# 에쿼티 기록
portfolio_value = self._calculate_portfolio_value(current_price)
self.portfolio.equity_curve.append(portfolio_value)
current_idx += 1
return self._calculate_metrics()
def _check_exit_conditions(self, bar_data, symbol: str):
"""청산 조건 확인"""
if symbol not in self.portfolio.positions:
return
pos = self.portfolio.positions[symbol]
current_price = bar_data['close']
high_price = bar_data['high']
low_price = bar_data['low']
# 스탑로스 체크
stop_loss_pct = 0.02 # 2% 스탑로스
entry_price = pos['entry_price']
if current_price < entry_price * (1 - stop_loss_pct):
self.portfolio.close_position(symbol, entry_price * (1 - stop_loss_pct - self.slippage))
# 테이크프로핃 체크
take_profit_pct = 0.05 # 5% 테이크프로핃
if current_price > entry_price * (1 + take_profit_pct):
self.portfolio.close_position(symbol, entry_price * (1 + take_profit_pct + self.slippage))
def _execute_signal(self, signal, current_price: float, timestamp: datetime):
"""신호 실행"""
# 슬리피지 적용
exec_price = current_price * (1 + self.slippage)
if signal.signal_type.value in ["STRONG_BUY", "BUY"]:
position_size = self.portfolio.current_capital * signal.position_size / exec_price
self.portfolio.open_position(signal.symbol, exec_price, position_size)
elif signal.signal_type.value in ["STRONG_SELL", "SELL"]:
if signal.symbol in self.portfolio.positions:
self.portfolio.close_position(signal.symbol, exec_price * (1 - self.slippage))
def _calculate_portfolio_value(self, current_price: float) -> float:
"""현재 포트폴리오 가치 계산"""
position_value = 0
for symbol, pos in self.portfolio.positions.items():
position_value += pos['size'] * current_price
return self.portfolio.current_capital + position_value
def _calculate_metrics(self) -> Dict:
"""성과 지표 계산"""
equity = np.array(self.portfolio.equity_curve)
returns = np.diff(equity) / equity[:-1]
total_return = (equity[-1] / equity[0] - 1) * 100
annual_return = total_return / (len(equity) / 252) if len(equity) > 1 else 0
# 연평균 변동성 (annualized volatility)
volatility = np.std(returns) * np.sqrt(252) * 100 if len(returns) > 1 else 0
# 최대 낙폭 (Maximum Drawdown)
cumulative = equity / equity[0]
running_max = np.maximum.accumulate(cumulative)
drawdown = (cumulative - running_max) / running_max * 100
max_drawdown = np.min(drawdown)
# 셰프 지수 (Sharpe Ratio, 무위험 수익률 4% 가정)
risk_free_rate = 0.04
excess_return = (1 + annual_return/100) / (1 + risk_free_rate) - 1
sharpe_ratio = excess_return / (volatility/100) if volatility > 0 else 0
# 승률
winning_trades = [t for t in self.portfolio.trades if t.pnl > 0]
win_rate