저는 최근 암호화폐 시장에서의 AI 기반 양적 투자 전략 개발에 매진하고 있는 개발자입니다. 이번 글에서는 Tardis의 분 단위-historical 데이터와 HolySheep AI의 GPT-4.1 모델을 활용하여 분-minute 백테스팅 프레임워크를 구축하는 방법을 상세히 설명드리겠습니다. 이 튜토리얼을 마치시면, 자신만의 실시간 거래 시뮬레이션 시스템을 구현할 수 있게 됩니다.
백테스팅 프레임워크란 무엇인가?
백테스팅은 거래 전략을 과거 데이터에 적용하여 수익率和 위험도를 평가하는プロセス입니다. 분-minute 레벨의 백테스팅은 고빈도 거래(HFT) 및 스캘핑 전략에 필수적이며, 일반적인 일별 데이터 기반 백테스팅보다 훨씬 정확한 결과를 제공합니다. Tardis는 암호화폐 시장 최고의 historical 데이터 공급자 중 하나로, 주요 거래소(Kraken, Binance, Coinbase 등)의 분-minute 데이터를 제공합니다.
필수 도구 및 사전 준비
필요한 계정 및 API 키
- Tardis API: tardis.dev에서 가입 및 API 키 발급
- HolySheep AI: 지금 가입하여 AI 모델 접근 권한 확보
- Python 3.9+: 로컬 개발 환경
Tardis API 데이터 가져오기
Tardis는 재구성 가능한 시계열 데이터베이스(Replay API)를 제공하여 과거 데이터를 스트리밍 방식으로 가져올 수 있습니다. 이를 통해 분-minute 단위의 틱 데이터를 실시간처럼 처리할 수 있습니다.
# tardis_client.py
import httpx
import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json
class TardisHistoricalClient:
"""Tardis Replay API를 활용한 히스토리컬 데이터 클라이언트"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={"Authorization": f"Bearer {api_key}"},
timeout=60.0
)
async def get_replay_data(
self,
exchange: str,
symbol: str,
from_time: datetime,
to_time: datetime,
channels: List[str] = ["trade"]
) -> List[Dict]:
"""
특정 거래소, 심볼, 시간 범위의 데이터를 가져옵니다.
Args:
exchange: 거래소 이름 (예: 'binance', 'kraken', 'coinbase')
symbol: 거래쌍 (예: 'BTCUSD', 'ETHUSD')
from_time: 시작 시간
to_time: 종료 시간
channels: 구독 채널 (trade, book-iterated, ticker 등)
"""
params = {
"exchange": exchange,
"symbol": symbol,
"from": int(from_time.timestamp() * 1000),
"to": int(to_time.timestamp() * 1000),
"channels": json.dumps(channels)
}
response = await self.client.get("/replay", params=params)
response.raise_for_status()
return response.json()
async def get_minute_candles(
self,
exchange: str,
symbol: str,
from_time: datetime,
to_time: datetime
) -> List[Dict]:
"""분-minute 봉(캔들스틱) 데이터 생성"""
raw_trades = await self.get_replay_data(exchange, symbol, from_time, to_time)
candles = {}
for trade in raw_trades:
if trade.get("type") == "trade":
timestamp = trade["timestamp"]
minute_key = timestamp // 60000 * 60000
if minute_key not in candles:
candles[minute_key] = {
"timestamp": minute_key,
"open": trade["price"],
"high": trade["price"],
"low": trade["price"],
"close": trade["price"],
"volume": 0,
"count": 0
}
candles[minute_key]["high"] = max(candles[minute_key]["high"], trade["price"])
candles[minute_key]["low"] = min(candles[minute_key]["low"], trade["price"])
candles[minute_key]["close"] = trade["price"]
candles[minute_key]["volume"] += trade.get("amount", 0)
candles[minute_key]["count"] += 1
return sorted(candles.values(), key=lambda x: x["timestamp"])
async def close(self):
await self.client.aclose()
사용 예제
async def main():
client = TardisHistoricalClient(api_key="YOUR_TARDIS_API_KEY")
# BTC/USD 2024년 1월 1일 하루치 분-minute 데이터
from_time = datetime(2024, 1, 1, 0, 0, 0)
to_time = datetime(2024, 1, 1, 23, 59, 59)
candles = await client.get_minute_candles(
exchange="binance",
symbol="BTCUSDT",
from_time=from_time,
to_time=to_time
)
print(f"가져온 분-minute 봉 개수: {len(candles)}")
print(f"첫 번째 봉: {candles[0]}")
print(f"마지막 봉: {candles[-1]}")
await client.close()
if __name__ == "__main__":
asyncio.run(main())AI 기반 거래 전략 개발
이제 HolySheep AI를 활용하여 시장 패턴을 분석하고 매매 신호를 생성하는 AI 전략 모듈을 만들겠습니다. HolySheep는 지금 가입하면 GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash 등 다양한 모델을 단일 API 키로 사용할 수 있습니다.
# ai_strategy.py
import httpx
import asyncio
import json
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime
@dataclass
class TradingSignal:
"""매매 신호 데이터 클래스"""
timestamp: int
action: str # 'buy', 'sell', 'hold'
confidence: float
reasoning: str
entry_price: Optional[float] = None
stop_loss: Optional[float] = None
take_profit: Optional[float] = None
class AIStrategyEngine:
"""HolySheep AI를 활용한 양적 거래 전략 엔진"""
def __init__(self, holysheep_api_key: str):
self.api_key = holysheep_api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "gpt-4.1"
async def analyze_market_and_generate_signal(
self,
candles: List[Dict],
lookback_minutes: int = 60
) -> TradingSignal:
"""
HolySheep AI를 활용하여 시장 분석 및 매매 신호 생성
Args:
candles: 최근 N분간의 캔들스틱 데이터
lookback_minutes: 분석할 과거 기간 (분)
"""
# 최근 N분 데이터만 사용
recent_candles = candles[-lookback_minutes:] if len(candles) >= lookback_minutes else candles
# 프롬프트 구성
market_data_summary = self._format_candles_for_prompt(recent_candles)
prompt = f"""당신은 전문 암호화폐 트레이더입니다. 아래 BTC/USDT 마켓 데이터를 분석하고 매매 신호를 생성해주세요.
최근 {len(recent_candles)}개 분봉 데이터:
{market_data_summary}
응답 형식 (JSON만 출력):
{{
"action": "buy" 또는 "sell" 또는 "hold",
"confidence": 0.0부터 1.0 사이의 신뢰도,
"reasoning": "신호 생성 근거 (한국어)",
"entry_price": null 또는 진입 희망 가격,
"stop_loss": null 또는止损 가격 (action이 buy/sell일 때만),
"take_profit": null 또는利確 가격 (action이 buy/sell일 때만)
}}
규칙:
- 신호 신뢰도가 0.6 이상일 때만 buy/sell 신호 생성
- stop_loss는 진입가 대비 1-3% 이하로 설정
- take_profit은 진입가 대비 3-10%로 설정
- market volatility를 반드시 고려"""
# HolySheep AI API 호출
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": "당신은 전문 금융 AI 어시스턴트입니다."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
)
if response.status_code != 200:
raise Exception(f"AI API 오류: {response.status_code} - {response.text}")
result = response.json()
ai_response = result["choices"][0]["message"]["content"]
# JSON 파싱
signal_data = json.loads(ai_response)
return TradingSignal(
timestamp=recent_candles[-1]["timestamp"],
action=signal_data["action"],
confidence=signal_data["confidence"],
reasoning=signal_data["reasoning"],
entry_price=signal_data.get("entry_price"),
stop_loss=signal_data.get("stop_loss"),
take_profit=signal_data.get("take_profit")
)
def _format_candles_for_prompt(self, candles: List[Dict]) -> str:
"""캔들 데이터를 프롬프트용 문자열로 변환"""
lines = []
for i, c in enumerate(candles[-10:]): # 최근 10개만 포함
dt = datetime.fromtimestamp(c["timestamp"] / 1000)
lines.append(
f"{dt.strftime('%H:%M')} | O:{c['open']:.2f} H:{c['high']:.2f} "
f"L:{c['low']:.2f} C:{c['close']:.2f} V:{c['volume']:.2f}"
)
return "\n".join(lines)
async def batch_analyze(
self,
candles: List[Dict],
interval_minutes: int = 5
) -> List[TradingSignal]:
"""
여러 시점에 대한 일괄 분석 수행
Args:
candles: 전체 캔들 데이터
interval_minutes: 분석 간격 (분)
"""
signals = []
for i in range(interval_minutes, len(candles), interval_minutes):
lookback = min(60, i)
subset = candles[:i]
try:
signal = await self.analyze_market_and_generate_signal(
subset,
lookback_minutes=lookback
)
signals.append(signal)
print(f"[{datetime.now()}] 신호 생성: {signal.action} (신뢰도: {signal.confidence:.2f})")
# Rate limit 방지
await asyncio.sleep(0.5)
except Exception as e:
print(f"분석 오류: {e}")
continue
return signals
사용 예제
async def test_ai_strategy():
engine = AIStrategyEngine(holysheep_api_key="YOUR_HOLYSHEEP_API_KEY")
# 테스트용 샘플 데이터 (실제로는 Tardis에서 가져옴)
sample_candles = [
{
"timestamp": int((datetime(2024, 1, 1, 0, i)).timestamp() * 1000),
"open": 42000 + i * 10,
"high": 42100 + i * 10,
"low": 41900 + i * 10,
"close": 42050 + i * 10,
"volume": 1000 + i * 50
}
for i in range(60)
]
signal = await engine.analyze_market_and_generate_signal(sample_candles)
print(f"\n생성된 신호:")
print(f" 액션: {signal.action}")
print(f" 신뢰도: {signal.confidence:.2f}")
print(f" 근거: {signal.reasoning}")
if signal.stop_loss:
print(f" 止损: ${signal.stop_loss:.2f}")
if signal.take_profit:
print(f"利確: ${signal.take_profit:.2f}")
if __name__ == "__main__":
asyncio.run(test_ai_strategy())분-minute 백테스팅 엔진 구현
이제 Tardis에서 가져온 historical 데이터와 HolySheep AI의 신호를 결합하여 완전한 백테스팅 엔진을 구축하겠습니다.
# backtest_engine.py
import asyncio
import json
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum
import statistics
class PositionSide(Enum):
LONG = "long"
SHORT = "short"
FLAT = "flat"
@dataclass
class Position:
"""보유 포지션"""
side: PositionSide
entry_price: float
quantity: float
entry_time: int
stop_loss: Optional[float] = None
take_profit: Optional[float] = None
@dataclass
class Trade:
"""거래 기록"""
timestamp: int
action: str
price: float
quantity: float
pnl: float = 0.0
reasoning: str = ""
@dataclass
class BacktestResult:
"""백테스트 결과"""
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
max_drawdown: float
sharpe_ratio: float
trades: List[Trade] = field(default_factory=list)
def to_dict(self) -> Dict:
return {
"total_trades": self.total_trades,
"winning_trades": self.winning_trades,
"losing_trades": self.losing_trades,
"win_rate": f"{self.win_rate:.2%}",
"total_pnl": f"${self.total_pnl:.2f}",
"max_drawdown": f"{self.max_drawdown:.2%}",
"sharpe_ratio": f"{self.sharpe_ratio:.2f}"
}
class MinuteBacktestEngine:
"""
분-minute 단위 백테스팅 엔진
Tardis historical 데이터 + HolySheep AI 신호 기반
"""
def __init__(
self,
initial_capital: float = 10000.0,
position_size_pct: float = 0.1,
commission_rate: float = 0.001,
slippage_pct: float = 0.0005
):
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.position_size_pct = position_size_pct
self.commission_rate = commission_rate
self.slippage_pct = slippage_pct
self.position: Optional[Position] = None
self.trades: List[Trade] = []
self.equity_curve: List[float] = []
# 신호 생성 간격 (분)
self.signal_interval = 5
def _apply_commission_and_slippage(
self,
price: float,
action: str
) -> float:
"""수수료 및 슬리피지 적용"""
multiplier = 1 + self.slippage_pct
if action == "buy":
multiplier += self.commission_rate
else: # sell
multiplier -= self.commission_rate
return price * multiplier
def _check_stop_loss_take_profit(
self,
current_price: float
) -> Tuple[Optional[str], float]:
"""
止损 및利確 체크
Returns:
(action, exit_price) 또는 (None, 0)
"""
if not self.position or self.position.side == PositionSide.FLAT:
return None, 0
if self.position.side == PositionSide.LONG:
# 상승 추세에서 buy 포지션의 경우
if self.position.stop_loss and current_price <= self.position.stop_loss:
return "stop_loss", self.position.stop_loss
if self.position.take_profit and current_price >= self.position.take_profit:
return "take_profit", self.position.take_profit
elif self.position.side == PositionSide.SHORT:
if self.position.stop_loss and current_price >= self.position.stop_loss:
return "stop_loss", self.position.stop_loss
if self.position.take_profit and current_price <= self.position.take_profit:
return "take_profit", self.position.take_profit
return None, 0
def execute_trade(
self,
timestamp: int,
action: str,
price: float,
confidence: float,
reasoning: str,
stop_loss: Optional[float] = None,
take_profit: Optional[float] = None
) -> Optional[Trade]:
"""거래 실행"""
trade = None
# 신뢰도 threshold
if confidence < 0.6:
return None
if action == "buy" and not self.position:
# 롱 포지션 진입
exit_price = self._apply_commission_and_slippage(price, "buy")
quantity = (self.current_capital * self.position_size_pct) / exit_price
self.position = Position(
side=PositionSide.LONG,
entry_price=exit_price,
quantity=quantity,
entry_time=timestamp,
stop_loss=stop_loss,
take_profit=take_profit
)
self.current_capital -= (exit_price * quantity)
trade = Trade(
timestamp=timestamp,
action="buy",
price=exit_price,
quantity=quantity,
reasoning=reasoning
)
elif action == "sell" and self.position:
# 포지션 청산
exit_price = self._apply_commission_and_slippage(price, "sell")
pnl = (exit_price - self.position.entry_price) * self.position.quantity
if self.position.side == PositionSide.SHORT:
pnl = -pnl # 숏 포지션의 경우 반대
self.current_capital += (exit_price * self.position.quantity)
self.current_capital += pnl
trade = Trade(
timestamp=timestamp,
action="sell",
price=exit_price,
quantity=self.position.quantity,
pnl=pnl,
reasoning=reasoning
)
self.position = None
return trade
def run_backtest(
self,
candles: List[Dict],
signals: List[Dict]
) -> BacktestResult:
"""
백테스트 실행
Args:
candles: 분-minute 봉 데이터
signals: AI 신호 리스트 [(timestamp, action, confidence, ...), ...]
"""
# 신호를 timestamp로 인덱싱
signal_map = {s["timestamp"]: s for s in signals}
equity = self.initial_capital
peak_equity = equity
max_drawdown = 0.0
returns = []
for candle in candles:
timestamp = candle["timestamp"]
current_price = candle["close"]
#Equity 업데이트
if self.position:
if self.position.side == PositionSide.LONG:
position_value = self.position.quantity * current_price
unrealized_pnl = position_value - (self.position.entry_price * self.position.quantity)
equity = self.current_capital + position_value + unrealized_pnl - position_value
equity = self.current_capital + (current_price - self.position.entry_price) * self.position.quantity
else:
equity = self.current_capital - (current_price - self.position.entry_price) * self.position.quantity
else:
equity = self.current_capital
# Drawdown 계산
peak_equity = max(peak_equity, equity)
drawdown = (peak_equity - equity) / peak_equity if peak_equity > 0 else 0
max_drawdown = max(max_drawdown, drawdown)
self.equity_curve.append(equity)
#止损/利확 체크
action, exit_price = self._check_stop_loss_take_profit(current_price)
if action:
trade = self.execute_trade(
timestamp=timestamp,
action="sell",
price=exit_price,
confidence=1.0,
reasoning=f"{action} triggered"
)
if trade:
self.trades.append(trade)
if trade.pnl > 0:
returns.append(trade.pnl / self.current_capital)
else:
returns.append(trade.pnl / self.current_capital)
# AI 신호 체크
if timestamp in signal_map:
sig = signal_map[timestamp]
trade = self.execute_trade(
timestamp=timestamp,
action=sig["action"],
price=current_price,
confidence=sig["confidence"],
reasoning=sig.get("reasoning", ""),
stop_loss=sig.get("stop_loss"),
take_profit=sig.get("take_profit")
)
if trade:
self.trades.append(trade)
if trade.pnl > 0:
returns.append(trade.pnl / self.current_capital)
else:
returns.append(trade.pnl / self.current_capital)
# 결과 계산
winning_trades = [t for t in self.trades if t.pnl > 0]
losing_trades = [t for t in self.trades if t.pnl <= 0]
avg_return = statistics.mean(returns) if returns else 0
std_return = statistics.stdev(returns) if len(returns) > 1 else 1
sharpe_ratio = (avg_return / std_return) * (252 * 24 * 60) ** 0.5 if std_return > 0 else 0
total_pnl = self.current_capital - self.initial_capital
return BacktestResult(
total_trades=len(self.trades),
winning_trades=len(winning_trades),
losing_trades=len(losing_trades),
win_rate=len(winning_trades) / len(self.trades) if self.trades else 0,
total_pnl=total_pnl,
max_drawdown=max_drawdown,
sharpe_ratio=sharpe_ratio,
trades=self.trades
)
def save_results(self, filename: str = "backtest_results.json"):
"""결과를 JSON 파일로 저장"""
results = {
"backtest_config": {
"initial_capital": self.initial_capital,
"position_size_pct": self.position_size_pct,
"commission_rate": self.commission_rate,
"signal_interval": self.signal_interval
},
"final_capital": self.current_capital,
"result": BacktestResult(
total_trades=len(self.trades),
winning_trades=len([t for t in self.trades if t.pnl > 0]),
losing_trades=len([t for t in self.trades if t.pnl <= 0]),
win_rate=len([t for t in self.trades if t.pnl > 0]) / len(self.trades) if self.trades else 0,
total_pnl=self.current_capital - self.initial_capital,
max_drawdown=0,
sharpe_ratio=0
).to_dict(),
"equity_curve": self.equity_curve,
"trades": [
{
"timestamp": t.timestamp,
"action": t.action,
"price": t.price,
"quantity": t.quantity,
"pnl": t.pnl,
"reasoning": t.reasoning
}
for t in self.trades
]
}
with open(filename, "w", encoding="utf-8") as f:
json.dump(results, f, indent=2, ensure_ascii=False)
print(f"결과 저장 완료: {filename}")
메인 실행 예제
async def run_full_backtest():
from tardis_client import TardisHistoricalClient
from ai_strategy import AIStrategyEngine
# API 키 설정
tardis_client = TardisHistoricalClient(api_key="YOUR_TARDIS_API_KEY")
ai_engine = AIStrategyEngine(holysheep_api_key="YOUR_HOLYSHEEP_API_KEY")
# 2024년 1월 1일~7일 데이터 (1주일 백테스트)
from_time = datetime(2024, 1, 1, 0, 0, 0)
to_time = datetime(2024, 1, 7, 23, 59, 59)
print("Tardis에서 데이터 가져오는 중...")
candles = await tardis_client.get_minute_candles(
exchange="binance",
symbol="BTCUSDT",
from_time=from_time,
to_time=to_time
)
print(f"가져온 데이터: {len(candles)}개 분봉")
await tardis_client.close()
# AI 신호 생성 (5분 간격)
print("HolySheep AI로 신호 생성 중...")
signals_raw = await ai_engine.batch_analyze(
candles=candles,
interval_minutes=5
)
signals = [
{
"timestamp": s.timestamp,
"action": s.action,
"confidence": s.confidence,
"reasoning": s.reasoning,
"stop_loss": s.stop_loss,
"take_profit": s.take_profit
}
for s in signals_raw
]
print(f"생성된 신호: {len(signals)}개")
# 백테스트 실행
print("백테스트 실행 중...")
engine = MinuteBacktestEngine(
initial_capital=10000.0,
position_size_pct=0.1,
commission_rate=0.001
)
result = engine.run_backtest(candles, signals)
# 결과 출력
print("\n" + "="*50)
print("백테스트 결과")
print("="*50)
print(f"총 거래 횟수: {result.total_trades}")
print(f"승리 거래: {result.winning_trades}")
print(f"패배 거래: {result.losing_trades}")
print(f"승률: {result.win_rate:.2%}")
print(f"총 손익: ${result.total_pnl:.2f}")
print(f"최대 드로우다운: {result.max_drawdown:.2%}")
print(f"샤프 비율: {result.sharpe_ratio:.2f}")
# 결과 저장
engine.save_results("btc_backtest_2024_01.json")
await ai_engine.client.close()
if __name__ == "__main__":
asyncio.run(run_full_backtest())실시간 스트리밍 백테스트
historical 데이터 백테스트뿐만 아니라, Tardis Replay API를 활용하여 과거 데이터를 실시간 스트림처럼 처리하는 실시간 백테스트도 가능합니다.
# realtime_backtest.py
import asyncio
import httpx
import json
from datetime import datetime, timedelta
from typing import AsyncGenerator, Dict, Optional
import queue
import threading
class TardisReplayStreamer:
"""
Tardis Replay API를 활용한 실시간 스트리밍 백테스트
과거 데이터를 실시간처럼 처리하여 전략 검증
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
self.ws_url = "wss://api.tardis.dev/v1/replay/connect"
async def stream_replay(
self,
exchange: str,
symbol: str,
from_time: datetime,
to_time: datetime,
channels: list = None
) -> AsyncGenerator[Dict, None]:
"""
Replay 스트림에서 메시지를 비동기 제너레이터로Yield
Args:
exchange: 거래소 (binance, kraken, coinbase, etc.)
symbol: 거래쌍 (BTCUSDT, ETHUSD, etc.)
from_time: 시작 시간
to_time: 종료 시간
channels: 구독 채널 리스트
"""
if channels is None:
channels = ["trade"]
params = {
"exchange": exchange,
"symbol": symbol,
"from": int(from_time.timestamp() * 1000),
"to": int(to_time.timestamp() * 1000),
"channels": json.dumps(channels),
"compress": "lz4",
"normalize": "true"
}
# WebSocket 연결 수립
async with httpx.AsyncClient() as client:
async with client.ws_connect(
self.ws_url,
params=params,
headers={"Authorization": f"Bearer {self.api_key}"}
) as ws:
print(f"Replay 스트림 연결됨: {exchange} {symbol}")
async for message in ws:
if message.type == httpx.WSMsgType.TEXT:
data = json.loads(message.text)
# 하트비트 메시지 건너뛰기
if data.get("type") == "heartbeat":
continue
yield data
elif message.type == httpx.WSMsgType.ERROR:
print(f"WebSocket 오류: {message.data}")
break
async def backtest_with_stream(
self,
exchange: str,
symbol: str,
from_time: datetime,
to_time: datetime,
strategy_callback,
candle_interval: int = 60
):
"""
스트림 기반 실시간 백테스트
Args:
exchange: 거래소
symbol: 거래쌍
from_time: 시작 시간
to_time: 종료 시간
strategy_callback: 각 캔들 완료 시 호출될 전략 함수
candle_interval: 캔들 생성 간격 (초)
"""
current_candle = {
"timestamp": None,
"open": None,
"high": None,
"low": None,
"close": None,
"volume": 0,
"trades": 0
}
async for message in self.stream_replay(
exchange, symbol, from_time, to_time
):
# trade 메시지 처리
if message.get("type") == "trade" or message.get("channel") == "trade":
trade_data = message.get("data", {})
ts = trade_data.get("timestamp", 0)
# 새 캔들 시작
candle_ts = (ts // (candle_interval * 1000)) * (candle_interval * 1000)
if current_candle["timestamp"] is None:
current_candle["timestamp"] = candle_ts
current_candle["open"] = trade_data["price"]
current_candle["high"] = trade_data["price"]
current_candle["low"] = trade_data["price"]
current_candle["close"] = trade_data["price"]
#同一 캔들 내 데이터 업데이트
if candle_ts == current_candle["timestamp"]:
current_candle["high"] = max(
current_candle["high"],
trade_data["price"]
)
current_candle["low"] = min(
current_candle["low"],
trade_data["price"]
)
current_candle["close"] = trade_data["price"]
current_candle["volume"] += trade_data.get("amount", 0)
current_candle["trades"] += 1
# 새 캔들 시작 시 이전 캔들 처리
else:
# 전략 콜백 실행
await strategy_callback(current_candle.copy())
# 새 캔들 초기화
current_candle = {
"timestamp": candle_ts,
"open": trade_data["price"],
"high": trade_data["price"],
"low": trade_data["price"],
"close": trade_data["price"],
"volume": trade_data.get("amount", 0),
"trades": 1
}
사용 예제
async def simple_moving_average_strategy(candle: Dict):
"""단순 이동 평균 교차 전략 예제"""
# 실제 구현에서는.historical 데이터와 결합하여
# 이동평균線を計算하고 크로스오버 감지
print(f"[{datetime.fromtimestamp(candle['timestamp']/1000)}] "
f"캔들 완료: O={candle['open']:.2f} H={candle['high']:.2f} "
f"L={candle['low']:.2f} C={candle['close']:.2f} V={candle['volume']:.2f}")
async def main():
streamer = TardisReplayStreamer(api_key="YOUR_TARDIS_API_KEY")
# 2024년 1월 1일 0시~1시만 테스트 (1시간)
from_time = datetime(2024, 1, 1, 0, 0, 0)
to_time = datetime(2024, 1, 1, 1, 0, 0)
print("실시간 백테스트 시작...")
await streamer.backtest_with_stream(
exchange="binance",
symbol="BTCUSDT",
from_time=from_time,
to_time=to_time,
strategy_callback=simple_moving_average_strategy,
candle_interval=60 # 1분봉
)
if __name__ == "__main__":
asyncio.run(main())HolySheep AI vs 경쟁 서비스 비교
| 서비스 | GPT-4.1 ($/MTok) | Claude Sonnet 4.5 ($/MTok) | Gemini 2.5 Flash ($/MTok) | DeepSeek V3.2 ($/
관련 리소스관련 문서 |
|---|