고성능 암호화폐 거래 전략 개발において、실시간 틱 데이터 분석은 핵심 요소입니다. 본 튜토리얼では、Tardis.dev의逐笔取引データを活用し、HolySheep AIのマルチモデル統合优势を活かしたプロダクション 수준의モメンタム戦略バックテストシステムを構築します。
개요: 왜 Tardis 데이터인가?
암호화폐 시장에서는 전통적인金融商品と異なり、24시간 연중무휴로 거래가 진행됩니다. Tardis.dev는 주요 거래소(Binance, Bybit, OKX 등)의 원시 틱 데이터를低レイテンシーで提供する 서비스입니다.
- tick 단위 거래 데이터 (가격, 수량, 시간)
- 순차적 주문一本书 데이터 (Level 2)
- 다양한 거래소 호환성
- Historical 데이터 최대 수년치 지원
아키텍처 설계
+-------------------+ +------------------+ +------------------+
| Tardis API |---->| Data Pipeline |---->| Signal Engine |
| (Tick/Market Data)| | (Rust/Python) | | (AI Inference) |
+-------------------+ +------------------+ +------------------+
|
+--------------------------------+
|
v
+-------------------+ +------------------+ +------------------+
| HolySheep AI |<----| LLM Analysis |---->| Backtest Engine |
| (Multi-Model) | | (Signal Valid.) | | (Vectorized) |
+-------------------+ +------------------+ +------------------+
본 시스템의 핵심 구성 요소:
- 데이터 파이프라인: Tardis WebSocket/ REST API에서 틱 데이터를 수집하여 시그널 엔진으로 전달
- 시그널 엔진: 모멘텀 지표를 계산하고 AI 모델로 시장 상황 분석
- LLM 분석 레이어: HolySheep AI의 DeepSeek V3.2를活用하여 시그널 품질 검증 및 개선
- 백테스트 엔진: 벡터라이즈된 방식으로 대규모Historical 데이터 테스트
환경 구축
# 필수 패키지 설치
pip install tardis-client pandas numpy scipy
pip install asyncpg redis aiohttp
pip install holy Sheep -ai # HolySheep SDK
프로젝트 구조
mkdir momentum-backtest/
cd momentum-backtest/
touch config.py data_pipeline.py signal_engine.py backtest.pyTardis 데이터 파이프라인 구현
# data_pipeline.py
import asyncio
import aiohttp
from tardis_client import TardisClient, Channel
from dataclasses import dataclass
from typing import List, Optional
import pandas as pd
from datetime import datetime
@dataclass
class TickData:
timestamp: int
price: float
volume: float
side: str # 'buy' or 'sell'
exchange: str
class TardisDataPipeline:
def __init__(self, api_key: str, exchanges: List[str]):
self.api_key = api_key
self.exchanges = exchanges
self.client = TardisClient(api_key=api_key)
self.tick_buffer = []
self.buffer_size = 1000
async def subscribe_realtime(
self,
symbols: List[str],
channels: List[str],
callback=None
):
"""실시간 틱 데이터 구독"""
for exchange in self.exchanges:
for symbol in symbols:
await self.client.subscribe(
exchange=exchange,
channels=[
Channel(
name=ch,
symbols=[symbol]
) for ch in channels
]
)
async for event in self.client.stream():
tick = self._parse_event(event)
if tick:
self.tick_buffer.append(tick)
# 버퍼가 차면 콜백 호출
if len(self.tick_buffer) >= self.buffer_size:
if callback:
await callback(self.tick_buffer)
self.tick_buffer = []
async def fetch_historical(
self,
exchange: str,
symbol: str,
start_time: datetime,
end_time: datetime
) -> pd.DataFrame:
"""Historical 데이터 조회"""
url = f"https://api.tardis.dev/v1/feeds/{exchange}:{symbol}"
params = {
"from": int(start_time.timestamp()),
"to": int(end_time.timestamp()),
"types": "trade"
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
data = await resp.json()
return self._convert_to_dataframe(data)
def _parse_event(self, event) -> Optional[TickData]:
if event.type == "trade":
return TickData(
timestamp=event.timestamp,
price=float(event.price),
volume=float(event.amount),
side=event.side,
exchange=event.exchange
)
return None
def _convert_to_dataframe(self, data: List[dict]) -> pd.DataFrame:
df = pd.DataFrame(data)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df = df.set_index('timestamp').sort_index()
return df
사용 예시
async def main():
pipeline = TardisDataPipeline(
api_key="YOUR_TARDIS_API_KEY",
exchanges=["binance", "bybit"]
)
# Historical 데이터로 백테스트
df = await pipeline.fetch_historical(
exchange="binance",
symbol="BTC-USDT",
start_time=datetime(2024, 1, 1),
end_time=datetime(2024, 6, 1)
)
print(f"데이터 포인트: {len(df):,}")
print(df.head())
if __name__ == "__main__":
asyncio.run(main())모멘텀 시그널 엔진
# signal_engine.py
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum
class SignalType(Enum):
STRONG_BUY = 1
BUY = 2
NEUTRAL = 3
SELL = 4
STRONG_SELL = 5
@dataclass
class MomentumSignal:
timestamp: pd.Timestamp
signal: SignalType
confidence: float
indicators: Dict[str, float]
raw_score: float
class MomentumSignalEngine:
def __init__(
self,
short_window: int = 20,
long_window: int = 50,
volume_window: int = 30,
volatility_threshold: float = 0.02
):
self.short_window = short_window
self.long_window = long_window
self.volume_window = volume_window
self.volatility_threshold = volatility_threshold
def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""기술적 지표 계산"""
df = df.copy()
# 이동평균선
df['sma_short'] = df['price'].rolling(self.short_window).mean()
df['sma_long'] = df['price'].rolling(self.long_window).mean()
# 모멘텀 (ROC)
df['momentum'] = df['price'].pct_change(periods=self.short_window)
# RSI
delta = df['price'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / (loss + 1e-10)
df['rsi'] = 100 - (100 / (1 + rs))
# 거래량 가중 모멘텀
df['volume_ma'] = df['volume'].rolling(self.volume_window).mean()
df['volume_ratio'] = df['volume'] / df['volume_ma']
# 변동성 (표준편차)
df['volatility'] = df['price'].rolling(self.long_window).std()
df['volatility_pct'] = df['volatility'] / df['price']
# MACD
exp1 = df['price'].ewm(span=12, adjust=False).mean()
exp2 = df['price'].ewm(span=26, adjust=False).mean()
df['macd'] = exp1 - exp2
df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
# Bollinger Bands
df['bb_mid'] = df['price'].rolling(20).mean()
df['bb_std'] = df['price'].rolling(20).std()
df['bb_upper'] = df['bb_mid'] + (df['bb_std'] * 2)
df['bb_lower'] = df['bb_mid'] - (df['bb_std'] * 2)
df['bb_position'] = (df['price'] - df['bb_lower']) / (
df['bb_upper'] - df['bb_lower'] + 1e-10
)
return df
def generate_signals(self, df: pd.DataFrame) -> List[MomentumSignal]:
"""모멘텀 시그널 생성"""
df = self.calculate_indicators(df)
signals = []
for idx in range(self.long_window, len(df)):
row = df.iloc[idx]
# 점수 계산
score = 0.0
indicators = {}
# 1. 이동평균 교차 (가중치: 0.3)
if row['sma_short'] > row['sma_long']:
score += 0.3
ma_signal = 1
else:
ma_signal = -1
indicators['ma_cross'] = ma_signal
# 2. 모멘텀 (가중치: 0.25)
momentum_score = np.clip(row['momentum'] * 10, -0.25, 0.25)
score += momentum_score
indicators['momentum'] = row['momentum']
# 3. RSI (가중치: 0.2)
if row['rsi'] < 30:
rsi_signal = 0.2
elif row['rsi'] > 70:
rsi_signal = -0.2
else:
rsi_signal = 0
score += rsi_signal
indicators['rsi'] = row['rsi']
# 4. 거래량 확인 (가중치: 0.15)
if row['volume_ratio'] > 1.5:
volume_signal = 0.15
elif row['volume_ratio'] < 0.5:
volume_signal = -0.15
else:
volume_signal = 0
score += volume_signal
indicators['volume_ratio'] = row['volume_ratio']
# 5. MACD (가중치: 0.1)
if row['macd'] > row['macd_signal']:
macd_signal = 0.1
else:
macd_signal = -0.1
score += macd_signal
indicators['macd'] = row['macd']
# 시그널 분류
if score > 0.6:
signal_type = SignalType.STRONG_BUY
confidence = min(abs(score) / 0.8, 1.0)
elif score > 0.2:
signal_type = SignalType.BUY
confidence = min(score / 0.4, 1.0)
elif score < -0.6:
signal_type = SignalType.STRONG_SELL
confidence = min(abs(score) / 0.8, 1.0)
elif score < -0.2:
signal_type = SignalType.SELL
confidence = min(abs(score) / 0.4, 1.0)
else:
signal_type = SignalType.NEUTRAL
confidence = 0.5
# 변동성 필터
if row['volatility_pct'] > self.volatility_threshold:
confidence *= 0.8
signals.append(MomentumSignal(
timestamp=row.name,
signal=signal_type,
confidence=confidence,
indicators=indicators,
raw_score=score
))
return signals
테스트
if __name__ == "__main__":
# 샘플 데이터 생성
dates = pd.date_range("2024-01-01", periods=200, freq="1h")
sample_data = pd.DataFrame({
"price": 50000 + np.cumsum(np.random.randn(200) * 100),
"volume": np.random.randint(100, 1000, 200)
}, index=dates)
engine = MomentumSignalEngine()
signals = engine.generate_signals(sample_data)
print(f"생성된 시그널 수: {len(signals)}")
for sig in signals[-5:]:
print(f"{sig.timestamp}: {sig.signal.name} (신뢰도: {sig.confidence:.2f})")HolySheep AI를 활용한 시그널 품질 개선
# signal_enhancer.py
import os
import json
from typing import List, Dict
from openai import AsyncOpenAI
from signal_engine import MomentumSignal, SignalType
HolySheep AI 설정
client = AsyncOpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class SignalEnhancer:
"""HolySheep AI를活用した 시그널 품질 개선"""
SYSTEM_PROMPT = """당신은 암호화폐 거래 전문가입니다.
주어진 기술적 지표 데이터를 분석하여 시그널의 신뢰도를 평가하고 개선建议你를 제공합니다.
응답 형식:
{
"validation": true/false,
"adjusted_confidence": 0.0-1.0,
"reasoning": "평가 근거",
"market_context": "시장 상황 설명"
}"""
def __init__(self, model: str = "deepseek/deepseek-chat-v3"):
self.model = model
self.rate_limit = 10 # RPM
self.request_count = 0
async def enhance_signal(self, signal: MomentumSignal) -> Dict:
"""개별 시그널 품질 개선"""
await self._check_rate_limit()
indicators_str = json.dumps(signal.indicators, indent=2)
response = await client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": f"""
기술적 지표 데이터:
- 시그널 타입: {signal.signal.name}
- 신뢰도: {signal.confidence}
- 원시 점수: {signal.raw_score}
- 지표: {indicators_str}
이 시그널의 품질을 평가하고 개선建议你를 제공해주세요.
"""}
],
temperature=0.3,
max_tokens=500
)
result = json.loads(response.choices[0].message.content)
self.request_count += 1
return {
"original_signal": signal,
"enhanced_confidence": result.get("adjusted_confidence", signal.confidence),
"validation": result.get("validation", True),
"reasoning": result.get("reasoning", ""),
"market_context": result.get("market_context", ""),
"cost": response.usage.total_tokens * 0.00042 / 1000 # DeepSeek V3.2 가격
}
async def batch_enhance(
self,
signals: List[MomentumSignal],
batch_size: int = 50
) -> List[Dict]:
"""배치 시그널 개선 (비용 최적화)"""
enhanced_signals = []
total_cost = 0
for i in range(0, len(signals), batch_size):
batch = signals[i:i+batch_size]
# 배치 요청으로 비용 절감
result = await self._batch_analysis(batch)
enhanced_signals.extend(result["signals"])
total_cost += result["cost"]
print(f"배치 {i//batch_size + 1}: {len(batch)}개 처리, 누적 비용: ${total_cost:.4f}")
return enhanced_signals
async def _batch_analysis(self, signals: List[MomentumSignal]) -> Dict:
"""배치 분석 (DeepSeek V3.2 활용)"""
signals_summary = []
for sig in signals:
signals_summary.append({
"idx": len(signals_summary),
"type": sig.signal.name,
"confidence": sig.confidence,
"score": sig.raw_score,
"momentum": sig.indicators.get("momentum", 0),
"rsi": sig.indicators.get("rsi", 50)
})
response = await client.chat.completions.create(
model="deepseek/deepseek-chat-v3",
messages=[
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": f"""
시그널 배치 분석:
{json.dumps(signals_summary, indent=2)}
각 시그널의 신뢰도를 재조정하고 전체 시장 동향도 분석해주세요.
"""}
],
temperature=0.2,
max_tokens=1000
)
# 파싱 로직 (실제 구현에서는 더严密한 파서 필요)
cost = response.usage.total_tokens * 0.42 / 1_000_000 # $0.42/MTok
return {
"signals": signals, # 실제 구현에서는 LLM 응답 기반 조정
"cost": cost
}
async def _check_rate_limit(self):
"""Rate limit 관리"""
if self.request_count >= self.rate_limit:
import asyncio
await asyncio.sleep(60 / self.rate_limit)
self.request_count = 0
사용 예시
async def main():
from signal_engine import MomentumSignalEngine, SignalType
import pandas as pd
import numpy as np
# 테스트 시그널 생성
signals = [
MomentumSignal(
timestamp=pd.Timestamp.now(),
signal=SignalType.BUY,
confidence=0.75,
indicators={"momentum": 0.05, "rsi": 45, "volume_ratio": 1.8},
raw_score=0.45
)
]
enhancer = SignalEnhancer()
result = await enhancer.enhance_signal(signals[0])
print(f"원시 신뢰도: {result['original_signal'].confidence}")
print(f"개선 신뢰도: {result['enhanced_confidence']}")
print(f"추론: {result['reasoning']}")
print(f"비용: ${result['cost']:.6f}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())벡터라이즈 백테스트 엔진
# backtest.py
import numpy as np
import pandas as pd
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from signal_engine import MomentumSignal, SignalType
@dataclass
class BacktestConfig:
initial_capital: float = 10000.0
commission_rate: float = 0.001 # 0.1%
slippage: float = 0.0005 # 0.05%
position_size: float = 1.0 # 풀 포지션
@dataclass
class Trade:
entry_time: pd.Timestamp
entry_price: float
exit_time: pd.Timestamp
exit_price: float
side: str # 'long' or 'short'
pnl: float
pnl_pct: float
holding_periods: int
class VectorizedBacktest:
"""NumPy 벡터라이즈를 활용한 고성능 백테스트"""
def __init__(self, config: BacktestConfig = None):
self.config = config or BacktestConfig()
self.trades: List[Trade] = []
self.equity_curve = []
def run(
self,
df: pd.DataFrame,
signals: List[MomentumSignal]
) -> Dict:
"""백테스트 실행"""
# 신호를 DataFrame으로 변환
signal_df = pd.DataFrame([
{"timestamp": s.timestamp, "signal": s.signal,
"confidence": s.confidence, "raw_score": s.raw_score}
for s in signals
])
# 가격 데이터와 머지
df = df.reset_index()
df = df.merge(signal_df, left_on='timestamp', right_on='timestamp', how='left')
df['signal'] = df['signal'].fillna(SignalType.NEUTRAL)
df['confidence'] = df['confidence'].fillna(0.5)
# 벡터라이즈 포지션 계산
positions = self._calculate_positions(df)
df['position'] = positions
# 수익률 계산
df['returns'] = df['price'].pct_change()
df['strategy_returns'] = df['position'].shift(1) * df['returns']
# 비용 적용
df['strategy_returns'] = df['strategy_returns'] - self.config.commission_rate
trades, df = self._extract_trades(df)
self.trades = trades
#Equity curve
equity = (1 + df['strategy_returns'].fillna(0)).cumprod() * self.config.initial_capital
self.equity_curve = equity
# 통계 계산
stats = self._calculate_statistics(df, trades, equity)
return stats
def _calculate_positions(self, df: pd.DataFrame) -> np.ndarray:
"""벡터라이즈 포지션 계산"""
n = len(df)
positions = np.zeros(n)
position = 0
confidence_threshold = 0.6
strong_threshold = 0.8
for i in range(n):
signal = df.iloc[i]['signal']
confidence = df.iloc[i]['confidence']
# 진입 로직
if position == 0:
if signal == SignalType.STRONG_BUY and confidence > strong_threshold:
position = 1
elif signal == SignalType.BUY and confidence > confidence_threshold:
position = 1
elif signal == SignalType.STRONG_SELL and confidence > strong_threshold:
position = -1
elif signal == SignalType.SELL and confidence > confidence_threshold:
position = -1
# 청산 로직
else:
if signal == SignalType.NEUTRAL:
position = 0
elif position == 1 and signal in [SignalType.SELL, SignalType.STRONG_SELL]:
position = 0
elif position == -1 and signal in [SignalType.BUY, SignalType.STRONG_BUY]:
position = 0
positions[i] = position
return positions
def _extract_trades(self, df: pd.DataFrame) -> Tuple[List[Trade], pd.DataFrame]:
"""거래 추출"""
trades = []
entry_idx = None
entry_price = None
entry_side = None
for i in range(len(df)):
if df.iloc[i]['position'] != 0 and entry_idx is None:
entry_idx = i
entry_price = df.iloc[i]['price']
entry_side = 'long' if df.iloc[i]['position'] > 0 else 'short'
elif df.iloc[i]['position'] == 0 and entry_idx is not None:
exit_price = df.iloc[i]['price']
if entry_side == 'long':
pnl = exit_price - entry_price
else:
pnl = entry_price - exit_price
pnl_pct = pnl / entry_price
trades.append(Trade(
entry_time=df.iloc[entry_idx]['timestamp'],
entry_price=entry_price,
exit_time=df.iloc[i]['timestamp'],
exit_price=exit_price,
side=entry_side,
pnl=pnl,
pnl_pct=pnl_pct,
holding_periods=i - entry_idx
))
entry_idx = None
entry_price = None
entry_side = None
return trades, df
def _calculate_statistics(
self,
df: pd.DataFrame,
trades: List[Trade],
equity: pd.Series
) -> Dict:
"""통계 계산"""
if not trades:
return {"error": "거래 없음"}
pnls = [t.pnl for t in trades]
returns = [t.pnl_pct for t in trades]
# 기본 통계
total_trades = len(trades)
winning_trades = sum(1 for p in pnls if p > 0)
losing_trades = total_trades - winning_trades
win_rate = winning_trades / total_trades if total_trades > 0 else 0
# 수익률 통계
avg_win = np.mean([p for p in pnls if p > 0]) if winning_trades > 0 else 0
avg_loss = np.mean([p for p in pnls if p < 0]) if losing_trades > 0 else 0
profit_factor = abs(sum(p for p in pnls if p > 0) / sum(p for p in pnls if p < 0)) if losing_trades > 0 else float('inf')
# 최대 낙폭 (MDD)
rolling_max = equity.cummax()
drawdown = (equity - rolling_max) / rolling_max
max_drawdown = drawdown.min()
# 샤프 비율
returns_series = df['strategy_returns'].dropna()
sharpe_ratio = returns_series.mean() / returns_series.std() * np.sqrt(365 * 24) if returns_series.std() > 0 else 0
# 총 수익
total_return = (equity.iloc[-1] - self.config.initial_capital) / self.config.initial_capital if len(equity) > 0 else 0
return {
"total_trades": total_trades,
"winning_trades": winning_trades,
"losing_trades": losing_trades,
"win_rate": win_rate,
"avg_win": avg_win,
"avg_loss": avg_loss,
"profit_factor": profit_factor,
"max_drawdown": max_drawdown,
"sharpe_ratio": sharpe_ratio,
"total_return": total_return,
"final_equity": equity.iloc[-1] if len(equity) > 0 else self.config.initial_capital,
"avg_holding_periods": np.mean([t.holding_periods for t in trades])
}
실행 예시
if __name__ == "__main__":
# 데이터 생성
dates = pd.date_range("2024-01-01", periods=500, freq="1h")
np.random.seed(42)
prices = 50000 + np.cumsum(np.random.randn(500) * 50)
volumes = np.random.randint(100, 1000, 500)
df = pd.DataFrame({
"timestamp": dates,
"price": prices,
"volume": volumes
}).set_index("timestamp")
# 시그널 생성
from signal_engine import MomentumSignalEngine
engine = MomentumSignalEngine()
signals = engine.generate_signals(df)
# 백테스트 실행
config = BacktestConfig(
initial_capital=10000,
commission_rate=0.001,
slippage=0.0005
)
backtest = VectorizedBacktest(config)
stats = backtest.run(df, signals)
print("=" * 50)
print("백테스트 결과")
print("=" * 50)
for key, value in stats.items():
if isinstance(value, float):
print(f"{key}: {value:.4f}")
else:
print(f"{key}: {value}")비용 최적화와 HolySheep AI 활용
본 시스템에서 HolySheep AI는 시그널 품질 개선에 활용됩니다. DeepSeek V3.2 모델의場合、비용이 매우 효율적입니다.
비용 비교표
| 항목 | 직접 API 사용 | HolySheep AI 게이트웨이 |
|---|---|---|
| DeepSeek V3.2 입력 | $0.27/MTok | $0.42/MTok |
| DeepSeek V3.2 출력 | $1.10/MTok | $1.10/MTok |
| 국제 신용카드 | 필요 | 불필요 |
| 다중 모델 통합 | 별도 계정 | 단일 키 |
| 한국 원화 결제 | 불가 | 지원 |
| 신규 가입 크레딧 | 없음 | 무료 크레딧 제공 |
시나리오별 비용 분석
매일 100건의 시그널을 AI 검증하는 경우:
- 월간 요청 수: 3,000회
- 평균 토큰: 입력 500 + 출력 200 = 700 토큰/요청
- 월간 총 토큰: 2,100,000 토큰
- HolySheep 비용: 약 $0.88/월 (입력만 계산)
자주 발생하는 오류 해결
1. Tardis API 연결 오류
# 오류: aiohttp.client_exceptions.ClientConnectorError
해결: 연결 설정 및 재시도 로직 구현
import asyncio
from aiohttp import ClientError, TCPConnector
class RobustTardisConnection:
def __init__(self, api_key: str, max_retries: int = 3):
self.api_key = api_key
self.max_retries = max_retries
async def connect_with_retry(self, url: str, params: dict):
connector = TCPConnector(
limit=100,
ttl_dns_cache=300,
keepalive_timeout=30
)
for attempt in range(self.max_retries):
try:
async with aiohttp.ClientSession(
connector=connector,
timeout=aiohttp.ClientTimeout(total=30)
) as session:
async with session.get(url, params=params) as resp:
if resp.status == 200:
return await resp.json()
elif resp.status == 429: # Rate limit
await asyncio.sleep(2 ** attempt)
continue
else:
raise Exception(f"HTTP {resp.status}")
except ClientError as e:
print(f"재시도 {attempt + 1}/{self.max_retries}: {e}")
await asyncio.sleep(2 ** attempt)
raise Exception("최대 재시도 횟수 초과")2. 시그널 지연 문제
# 오류: 시그널 생성 지연으로 인한 거래 시점 불일치
해결: 비동기 파이프라인 및 캐싱 적용
from functools import lru_cache
import hashlib
class CachedSignalEngine(MomentumSignalEngine):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._cache = {}
self._cache_size = 1000
def _get_cache_key(self, price: float, volume: float) -> str:
# 근사값으로 캐시 키 생성 (성능 최적화)
return hashlib.md5(
f"{price:.2f}{volume:.0f}".encode()
).hexdigest()[:12]
@lru_cache(maxsize=1000)
def _calculate_indicator_cached(self, price: float, volume: float) -> dict:
"""캐시된 지표 계산"""
return {"price": price, "volume": volume}
def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
# 이미 계산된 지표 재사용
if len(df) > self.long_window:
last_row = df.iloc[-1]
cache_key = self._get_cache_key(last_row['price'], last_row['volume'])
if cache_key in self._cache:
return self._cache[cache_key]
result = super().calculate_indicators(df)
# 캐시 관리
if len(self._cache) > self._cache_size:
# 가장 오래된 항목 제거
oldest_key = next(iter(self._cache))
del self._cache[oldest_key]
return result3. 백테스트 메모리 초과
# 오류: 대규모 데이터 백테스트 시 메모리 부족
해결: 청크 단위 처리 및 메모리 맵 활용
import numpy as np
import pandas as pd
from typing import Generator
class MemoryOptimizedBacktest(VectorizedBacktest):
def __init__(self, config: BacktestConfig = None, chunk_size: int = 10000):
super().__init__(config)
self.chunk_size = chunk_size
def run_chunked(
self,
file_path: str,
signals: List[MomentumSignal]
) -> Dict:
"""청크 단위 백테스트 실행"""
total_stats = None
# 신호를 청크 단위로 분할
signal_chunks = [
signals[i:i+self.chunk_size]
for i in range(0, len(signals), self.chunk_size)
]
for chunk_idx, signal_chunk in enumerate(signal_chunks):
# 메모리 맵으로 데이터 로드
df_chunk = pd.read_parquet(
file_path,
filters=[("chunk", "=", chunk_idx)]
)
# 청크별 백테스트
chunk_stats = self.run(df_chunk, signal_chunk)
# 결과 누적
if total_stats is None: