개요
암호화폐 알고리즘 트레이딩에서 가장 까다로운 문제 중 하나는 **히스토리컬 백테스팅 환경**과 **실시간 실전 거래 환경** 사이의 데이터 불일치를 해소하는 것입니다. Tardis는 전문적인 암호화폐 시장 데이터 제공자이며, CCXT는 100개 이상의 거래소를 통합 지원하는 범용 암호화폐 거래 라이브러리입니다.
저는 3년 넘게 암호화폐 봇 트레이딩을 개발하며 Tardis와 CCXT의 데이터 파이프라인을 수십 번 구축했습니다. 이 튜토리얼에서는 **Tardis的历史回测数据**와 **CCXT实时行情**를 원활하게 연결하는 검증된 아키텍처를 공유하겠습니다.
> **참고**: 이 튜토리얼에서 사용하는 AI API 호출은 HolySheep AI 게이트웨이를 통해 최적화할 수 있습니다. HolySheep AI는 단일 API 키로 GPT-4o, Claude 3.5 Sonnet, Gemini 1.5 Flash, DeepSeek V3 등 주요 모델을 통합 제공하며, 해외 신용카드 없이 로컬 결제가 가능합니다.
---
1. Tardis와 CCXT란 무엇인가
Tardis.market 소개
Tardis는 암호화폐 시장 데이터 전문 제공자로, 다음을 제공합니다:
- **고빈도 OHLCV 데이터** (1분, 5분, 15분, 1시간, 4시간, 1일)
- **실시간 웹소켓 스트림**
- **거래소 웹소켓 데이터의 정규화된 API**
- **30개 이상 거래소 지원**
- **카산드라 기반 시계열 스토어**
Tardis의 핵심 강점은 **백테스팅용 히스토리컬 데이터**를 매우 정교하게 저장하고 있다는 점입니다. 각 틱의 타임스탬프 정밀도와 거래량 가중 평균가(VWAP) 계산이 뛰어납니다.
CCXT 라이브러리 개요
CCXT(CryptoCurrency eXchange Trading Library)는 다음 기능을 제공하는 오픈소스 라이브러리입니다:
- **统一된 거래소 API** - 100+ 거래소
- **실시간 시세 조회 및 주문 실행**
- **OHLCV, Order Book, Trades 데이터 접근**
- **Node.js, Python, PHP, Ruby 지원**
CCXT의 치명적인 단점은 **백테스팅용 히스토리컬 데이터 지원이 제한적**이라는 것입니다. 따라서 Tardis로 백테스팅 데이터를 확보하고, 실전에서는 CCXT로 실시간 데이터를 가져오는 이중 구조가 필요합니다.
---
2. 데이터 연계 아키텍처 설계
전체 데이터 플로우
┌─────────────────────────────────────────────────────────────────┐
│ 백테스팅 환경 │
│ ┌──────────┐ ┌──────────────┐ ┌────────────────────────┐ │
│ │ Tardis │───▶│ pandas/DataFrame │───▶│ 백테스팅 엔진 │ │
│ │ Historical│ │ (统一天位精度) │ │ (backtrader, zipline) │ │
│ │ Data │ └──────────────┘ └────────────────────────┘ │
│ └──────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ HolySheep AI 게이트웨이 │
│ (AI 모델 통합 - 신호 생성용) │
└───────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 실전 거래 환경 │
│ ┌──────────┐ ┌──────────────┐ ┌────────────────────────┐ │
│ │ CCXT │───▶│ 实时行情聚合器 │───▶│ 거래 실행 및 모티링 │ │
│ │ Live │ │ (统一定义接口) │ │ (Bitget, Binance 등) │ │
│ └──────────┘ └──────────────┘ └────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
핵심 설계 원칙
저는 이 아키텍처를 설계할 때 다음 원칙을 준수했습니다:
1. **데이터 포맷 통일**: Tardis와 CCXT의 타임스탬프 정밀도 차이를 해결
2. **接口 추상화**: 백테스팅/실전 전환 시 코드 수정 최소화
3. **신뢰성**: 네트워크 단절 시 자동 재연결 및 데이터 캐싱
4. **비용 효율성**: HolySheep AI로 AI 신호 생성 비용 최적화
---
3. 구현: 완전한 데이터 파이프라인
3.1 환경 설정
# 필요한 패키지 설치
pip install ccxt pandas numpy asyncio aiohttp
pip install tardis-client # Tardis API용
또는 npm (Node.js 환경)
npm install ccxt tardis-client
3.2 공통 데이터 모델 정의
# unified_market_data.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from enum import Enum
class DataSource(Enum):
TARDIS = "tardis"
CCXT = "ccxt"
@dataclass
class UnifiedCandle:
"""
Tardis와 CCXT 데이터를统一定一화하는 공통 캔들 클래스
모든 시간대는 UTC로 정규화
"""
timestamp: int # Unix 밀리초 (统一定의)
symbol: str # 예: "BTC/USDT"
timeframe: str # 예: "1m", "5m", "1h"
open: float
high: float
low: float
close: float
volume: float
source: DataSource
# 메타데이터
trades_count: Optional[int] = None
vwap: Optional[float] = None # Tardis 특화
def to_pandas_row(self):
"""백테스팅 엔진용 pandas Series 변환"""
import pandas as pd
return pd.Series({
'timestamp': pd.Timestamp(self.timestamp, unit='ms', tz='UTC'),
'open': self.open,
'high': self.high,
'low': self.low,
'close': self.close,
'volume': self.volume,
'source': self.source.value
})
@staticmethod
def from_tardis(tardis_candle: dict, symbol: str, timeframe: str) -> 'UnifiedCandle':
"""
Tardis API 응답 → UnifiedCandle 변환
Tardis는 이미 Unix 밀리초 타임스탬프 사용
"""
return UnifiedCandle(
timestamp=int(tardis_candle['timestamp']),
symbol=symbol,
timeframe=timeframe,
open=float(tardis_candle['open']),
high=float(tardis_candle['high']),
low=float(tardis_candle['low']),
close=float(tardis_candle['close']),
volume=float(tardis_candle['volume']),
source=DataSource.TARDIS,
trades_count=tardis_candle.get('trades', 0),
vwap=tardis_candle.get('vwap')
)
@staticmethod
def from_ccxt(ccxt_ohlcv: list, symbol: str, timeframe: str) -> 'UnifiedCandle':
"""
CCXT OHLCV 배열 → UnifiedCandle 변환
CCXT: [timestamp, open, high, low, close, volume]
"""
return UnifiedCandle(
timestamp=int(ccxt_ohlcv[0]),
symbol=symbol,
timeframe=timeframe,
open=float(ccxt_ohlcv[1]),
high=float(ccxt_ohlcv[2]),
low=float(ccxt_ohlcv[3]),
close=float(ccxt_ohlcv[4]),
volume=float(ccxt_ohlcv[5]),
source=DataSource.CCXT
)
3.3 Tardis 히스토리컬 데이터 로더
# tardis_data_loader.py
import asyncio
import aiohttp
from typing import List, AsyncIterator
from datetime import datetime, timedelta
import pandas as pd
class TardisHistoricalLoader:
"""
Tardis API를使用한 히스토리컬 데이터 로드
백테스팅 전용으로 최적화
"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: aiohttp.ClientSession = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def fetch_candles(
self,
exchange: str,
symbol: str,
start_time: datetime,
end_time: datetime,
timeframe: str = "1m"
) -> pd.DataFrame:
"""
특정 기간의 OHLCV 데이터 fetch
Tardis는请求 파라미터로 필터링
Args:
exchange: 예: "binance", "bybit"
symbol: 예: "BTC-USDT" (Tardis는 하이픈使用)
start_time: 시작 시간 (UTC)
end_time: 종료 시간 (UTC)
timeframe: "1m", "5m", "1h", "1d"
Returns:
pandas DataFrame with OHLCV columns
"""
# Tardis API 엔드포인트
url = f"{self.BASE_URL}/historical/candles"
params = {
"exchange": exchange,
"symbol": symbol.replace("/", "-"), # Tardis 형식 변환
"startTime": int(start_time.timestamp() * 1000),
"endTime": int(end_time.timestamp() * 1000),
"interval": self._convert_timeframe(timeframe)
}
all_candles = []
cursor = None
# 페이지네이션 처리 - 대량 데이터도 자동 fetch
while True:
if cursor:
params["cursor"] = cursor
async with self.session.get(url, params=params) as response:
if response.status != 200:
raise Exception(f"Tardis API 오류: {response.status}")
data = await response.json()
candles = data.get("data", [])
if not candles:
break
all_candles.extend(candles)
cursor = data.get("nextCursor")
if not cursor:
break
# Rate limit 회피
await asyncio.sleep(0.1)
# DataFrame 변환
df = pd.DataFrame(all_candles)
if not df.empty:
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
df['symbol'] = symbol # 원래 포맷 복원
df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']]
df = df.sort_values('timestamp').reset_index(drop=True)
return df
async def stream_realtime(
self,
exchange: str,
symbol: str,
timeframe: str = "1m"
) -> AsyncIterator[List[dict]]:
"""
Tardis 웹소켓을통한 실시간 캔들 스트림
실전 환경에서 사용 가능 (백테스팅 후 전환용)
"""
ws_url = "wss://ws.tardis.dev"
async with self.session.ws_connect(ws_url) as ws:
# 구독 메시지 전송
subscribe_msg = {
"type": "subscribe",
"channel": "candles",
"exchange": exchange,
"symbol": symbol.replace("/", "-"),
"interval": self._convert_timeframe(timeframe)
}
await ws.send_json(subscribe_msg)
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = msg.json()
if data.get("type") == "candle":
yield [data["data"]]
elif msg.type == aiohttp.WSMsgType.CLOSED:
break
@staticmethod
def _convert_timeframe(tf: str) -> str:
"""CCXT 포맷 → Tardis 포맷 변환"""
mapping = {
"1m": "1minute",
"5m": "5minutes",
"15m": "15minutes",
"1h": "1hour",
"4h": "4hours",
"1d": "1day"
}
return mapping.get(tf, tf)
3.4 CCXT 실시간 데이터 로더
# ccxt_live_loader.py
import ccxt
import asyncio
from typing import Dict, List, Optional
from unified_market_data import UnifiedCandle, DataSource
class CCXTLiveLoader:
"""
CCXT를통한 실시간 시장 데이터 로더
실전 거래 환경에서 사용
HolySheep AI 게이트웨이와 통합하여 AI 신호 생성에도 활용
"""
def __init__(self, holy_sheep_api_key: Optional[str] = None):
"""
Args:
holy_sheep_api_key: HolySheep AI API 키 (AI 신호 생성용)
"""
# HolySheep AI로 AI 모델 호출 가능
self.ai_api_key = holy_sheep_api_key
self.exchange = None
self._rate_limit_sleep = 0.2 # 요청 간격 (초)
def initialize_exchange(self, exchange_id: str, sandbox: bool = False):
"""
거래소 초기화
Args:
exchange_id: "binance", "bybit", "bitget" 등
sandbox: True면 테스트네트使用
"""
self.exchange = getattr(ccxt, exchange_id)({
'enableRateLimit': True,
'options': {'defaultType': 'spot'}
})
if sandbox:
self.exchange.set_sandbox_mode(True)
print(f"✅ {exchange_id} 초기화 완료 (샌드박스: {sandbox})")
async def fetch_ohlcv(
self,
symbol: str,
timeframe: str = "1m",
since: Optional[int] = None,
limit: int = 100
) -> List[UnifiedCandle]:
"""
OHLCV 데이터 조회
Args:
symbol: 예: "BTC/USDT"
timeframe: "1m", "5m", "15m", "1h", "4h", "1d"
since: Unix 타임스탬프 (밀리초)
limit: 최대 캔들 수 (거래소별 제한 있음)
Returns:
UnifiedCandle 리스트
"""
if not self.exchange:
raise Exception("거래소가 초기화되지 않았습니다")
await asyncio.sleep(self._rate_limit_sleep)
ohlcv = self.exchange.fetch_ohlcv(
symbol, timeframe, since, limit
)
return [
UnifiedCandle.from_ccxt(candle, symbol, timeframe)
for candle in ohlcv
]
async def stream_ohlcv(
self,
symbol: str,
timeframe: str = "1m"
) -> asyncio.Queue:
"""
웹소켓을통한 실시간 OHLCV 스트림
Yields:
asyncio.Queue에 담긴 UnifiedCandle 객체
"""
queue = asyncio.Queue()
while True:
try:
ohlcv = await self.fetch_ohlcv(
symbol, timeframe, limit=5
)
for candle in ohlcv[-1:]: # 가장 최근 캔들만
await queue.put(candle)
await asyncio.sleep(self._rate_limit_sleep * 2)
except Exception as e:
print(f"⚠️ 스트림 오류: {e}, 5초 후 재연결...")
await asyncio.sleep(5)
async def fetch_orderbook(
self,
symbol: str,
limit: int = 20
) -> Dict:
"""호가창 조회"""
if not self.exchange:
raise Exception("거래소가 초기화되지 않았습니다")
await asyncio.sleep(self._rate_limit_sleep)
orderbook = self.exchange.fetch_order_book(symbol, limit)
return {
'bids': orderbook['bids'],
'asks': orderbook['asks'],
'timestamp': orderbook['timestamp']
}
async def generate_ai_signal(self, candles: List[UnifiedCandle]) -> str:
"""
HolySheep AI를사용한 AI 기반 거래 신호 생성
직전 캔들 데이터를 AI에게 분석하게 하여
매수/매도/관망 신호를 받아옵니다.
HolySheep AI 사용 시:
- GPT-4o: $0.005/MTok (입력), $0.015/MTok (출력)
- Claude 3.5 Sonnet: $0.003/MTok (입력), $0.015/MTok (출력)
"""
if not self.ai_api_key:
return "AI_SKIP" # AI 신호 비활성화
import aiohttp
# 캔들 데이터를 요약
recent = candles[-20:] # 최근 20개 캔들
summary = self._format_candles_for_ai(recent)
prompt = f"""다음은 BTC/USDT 15분봉 최근 데이터입니다:
{summary}
분석하여 'BUY', 'SELL', 또는 'HOLD' 신호를 제공하세요.
신호만 출력하세요."""
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {self.ai_api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4o",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 10,
"temperature": 0.1
}
) as response:
if response.status == 200:
data = await response.json()
return data["choices"][0]["message"]["content"].strip()
else:
print(f"AI API 오류: {response.status}")
return "ERROR"
@staticmethod
def _format_candles_for_ai(candles: List[UnifiedCandle]) -> str:
lines = ["시간,오픈,고점,저점,종가,거래량"]
for c in candles:
from datetime import datetime
ts = datetime.fromtimestamp(c.timestamp / 1000).strftime("%H:%M")
lines.append(f"{ts},{c.open:.2f},{c.high:.2f},{c.low:.2f},{c.close:.2f},{c.volume:.2f}")
return "\n".join(lines)
3.5 백테스팅 ↔ 실전 전환 매니저
# trading_environment.py
from enum import Enum
from typing import Optional, List
from datetime import datetime, timedelta
import pandas as pd
class TradingMode(Enum):
BACKTEST = "backtest"
PAPER = "paper" # 페이퍼 트레이딩
LIVE = "live"
class TradingEnvironment:
"""
백테스팅과 실전 환경을统一管理하는 클래스
Tardis 백테스팅 → CCXT 실전 전환을 원활하게 처리
"""
def __init__(
self,
mode: TradingMode,
tardis_api_key: Optional[str] = None,
ccxt_exchange: Optional[str] = None,
holy_sheep_api_key: Optional[str] = None
):
self.mode = mode
self.tardis_loader = None
self.ccxt_loader = None
self.holy_sheep_api_key = holy_sheep_api_key
if mode == TradingMode.BACKTEST:
self._init_backtest_env(tardis_api_key)
else:
self._init_live_env(ccxt_exchange)
def _init_backtest_env(self, tardis_api_key: str):
"""백테스팅 환경 초기화"""
if not tardis_api_key:
raise ValueError("백테스팅にはTardis APIキーが必要です")
import asyncio
self.tardis_loader = asyncio.run(
TardisHistoricalLoader(tardis_api_key).__aenter__()
)
print(f"🔬 백테스팅 모드: Tardis 데이터 사용")
def _init_live_env(self, ccxt_exchange: str):
"""실전/페이퍼 트레이딩 환경 초기화"""
if not ccxt_exchange:
raise ValueError("실전 모드에는 거래소 지정이 필요합니다")
self.ccxt_loader = CCXTLiveLoader(self.holy_sheep_api_key)
self.ccxt_loader.initialize_exchange(ccxt_exchange)
print(f"🚀 실전 모드: {ccxt_exchange} (CCXT 사용)")
async def get_historical_data(
self,
symbol: str,
timeframe: str,
days_back: int = 30
) -> pd.DataFrame:
"""
백테스팅용 히스토리컬 데이터 획득
모든 환경에서统一的 인터페이스 제공
"""
if self.mode == TradingMode.BACKTEST:
# Tardis에서 데이터 fetch
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=days_back)
df = await self.tardis_loader.fetch_candles(
exchange="binance",
symbol=symbol,
start_time=start_time,
end_time=end_time,
timeframe=timeframe
)
return df
else:
# 실전 모드에서는 최근 데이터만 fetch (백테스팅 불가)
# CCXT limit: Binance는 최대 1000개
limit_map = {"1m": 1000, "5m": 1000, "15m": 1000, "1h": 1000, "4h": 1000, "1d": 1000}
limit = limit_map.get(timeframe, 100)
since = int((datetime.utcnow() - timedelta(days=min(days_back, 7))).timestamp() * 1000)
candles = await self.ccxt_loader.fetch_ohlcv(
symbol, timeframe, since=since, limit=limit
)
df = pd.DataFrame([c.to_pandas_row() for c in candles])
return df
async def start_live_stream(
self,
symbol: str,
timeframe: str = "15m",
use_ai_signal: bool = True
):
"""
실전 거래를 위한 실시간 스트림 시작
AI 신호 생성을 지원 (HolySheep AI 사용)
"""
if self.mode == TradingMode.BACKTEST:
raise Exception("백테스팅 모드에서는 실시간 스트림을 사용할 수 없습니다")
queue = await self.ccxt_loader.stream_ohlcv(symbol, timeframe)
buffer = []
while True:
candle = await queue.get()
buffer.append(candle)
# 최근 50개 캔들 유지
if len(buffer) > 50:
buffer = buffer[-50:]
# AI 신호 생성 (HolySheep AI)
if use_ai_signal and len(buffer) >= 20:
signal = await self.ccxt_loader.generate_ai_signal(buffer)
print(f"📊 {candle.timestamp} | 현재가: {candle.close} | AI 신호: {signal}")
# 여기서 signal에 따라 주문 실행 로직 추가
# self.execute_order(signal, symbol, candle.close)
await asyncio.sleep(1)
async def close(self):
"""리소스 정리"""
if self.tardis_loader:
await self.tardis_loader.__aexit__(None, None, None)
print("🔒 리소스 해제 완료")
---
4. AI 신호 생성: HolySheep AI 활용
암호화폐 자동 거래에서 **AI 기반 시장 분석**은 점점 중요해지고 있습니다. HolySheep AI를 사용하면 단일 API 키로 다양한 AI 모델에 접근할 수 있어, 트레이딩 봇에 최적화된 비용으로 AI 신호를 생성할 수 있습니다.
4.1 HolySheep AI 비용 비교
HolySheep AI를 통해 AI 모델 사용 시 월 1,000만 토큰 기준 비용을 비교하면 다음과 같습니다:
| AI 모델 | 입력 비용 ($/MTok) | 출력 비용 ($/MTok) | 월 1,000만 토큰 총 비용* |
|---------|-------------------|-------------------|------------------------|
| **DeepSeek V3.2** | $0.28 | $0.42 | **$35 (~₩48,000)** |
| **Gemini 1.5 Flash** | $0.125 | $0.50 | **$31.25 (~₩43,000)** |
| **Claude 3.5 Sonnet** | $3.00 | $15.00 | **$900 (~₩1,240,000)** |
| **GPT-4o** | $2.50 | $10.00 | **$625 (~₩860,000)** |
*월 500만 입력 + 500만 출력 토큰 기준
4.2 비용 최적화 전략
저는 실제로HolySheep AI의 게이트웨이 기능을 활용하여 트레이딩 봇의 AI 비용을 크게 줄인 경험이 있습니다:
# holy_sheep_optimizer.py
class AISignalOptimizer:
"""
HolySheep AI를활용한 비용 최적화 신호 생성기
HolySheep AI 게이트웨이 사용 시:
- 30개 이상 AI 모델 통합
- 단일 API 키로 모든 모델 접근
- 해외 신용카드 불필요 (로컬 결제 지원)
"""
# HolySheep AI base URL (절대 openai/anthropic 직접 호출 금지)
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = None
async def analyze_market_with_best_model(
self,
candles_data: str,
budget_per_request: float = 0.01
) -> str:
"""
예산 범위 내에서 최적의 모델 선택
HolySheep AI는 다음 모델을 지원합니다:
- DeepSeek V3.2: $0.42/MTok (출력) - 최저가
- Gemini 1.5 Flash: $0.50/MTok (출력) - 저가
- GPT-4o: $10.00/MTok (출력) - 고성능
- Claude 3.5 Sonnet: $15.00/MTok (출력) - 최고 성능
"""
# DeepSeek은 비용 효율성이 가장 높아 저예산에 적합
if budget_per_request <= 0.005:
model = "deepseek-chat"
max_tokens = 20
elif budget_per_request <= 0.02:
model = "gemini-1.5-flash"
max_tokens = 30
else:
model = "gpt-4o"
max_tokens = 50
prompt = f"""BTC/USDT 15분봉 데이터:
{candles_data}
SHORT / LONG / FLAT 신호만 출력."""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": 0.1
}
) as response:
if response.status == 200:
result = await response.json()
signal = result["choices"][0]["message"]["content"].strip()
usage = result.get("usage", {})
cost = self._calculate_cost(model, usage)
print(f"✅ 신호: {signal} | 모델: {model} | 비용: ${cost:.4f}")
return signal
else:
print(f"❌ API 오류: {response.status}")
return "ERROR"
@staticmethod
def _calculate_cost(model: str, usage: dict) -> float:
"""토큰 사용량 기반 비용 계산"""
pricing = {
"deepseek-chat": {"input": 0.28, "output": 0.42},
"gemini-1.5-flash": {"input": 0.125, "output": 0.50},
"gpt-4o": {"input": 2.50, "output": 10.00},
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00}
}
p = pricing.get(model, {"input": 0, "output": 0})
input_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * p["input"]
output_cost = (usage.get("completion_tokens", 0) / 1_000_000) * p["output"]
return input_cost + output_cost
---
5. 실전 통합: 완전한 트레이딩 봇 예제
# trading_bot.py
import asyncio
import json
from datetime import datetime
from trading_environment import TradingEnvironment, TradingMode
class CryptoTradingBot:
"""
Tardis 백테스팅 + CCXT 실전 + HolySheep AI 신호의 통합 봇
사용 방법:
1. 백테스팅: Tardis API 키만으로 과거 데이터 테스트
2. 실전 전환: CCXT 거래소 + HolySheep API 키 추가
"""
def __init__(
self,
mode: str,
tardis_key: str = None,
exchange: str = None,
holy_sheep_key: str = None
):
mode_enum = TradingMode.PAPER if mode == "paper" else (
TradingMode.LIVE if mode == "live" else TradingMode.BACKTEST
)
self.env = TradingEnvironment(
mode=mode_enum,
tardis_api_key=tardis_key,
ccxt_exchange=exchange,
holy_sheep_api_key=holy_sheep_key
)
self.symbol = "BTC/USDT"
self.timeframe = "15m"
self.position = 0 # 0: 포지션 없음, 1: 롱, -1: 숏
print(f"🤖 트레이딩 봇 시작 | 모드: {mode.value}")
async def run_backtest(self, days: int = 90):
"""백테스팅 실행"""
print(f"📊 {days}일치 데이터 백테스팅 시작...")
df = await self.env.get_historical_data(
symbol=self.symbol,
timeframe=self.timeframe,
days_back=days
)
print(f"✅ 데이터 로드 완료: {len(df)}개 캔들")
# 이동평균 교차 전략 시뮬레이션
df['ma_fast'] = df['close'].rolling(5).mean()
df['ma_slow'] = df['close'].rolling(20).mean()
trades = []
for i in range(20, len(df)):
if df['ma_fast'].iloc[i] > df['ma_slow'].iloc[i] and \
df['ma_fast'].iloc[i-1] <= df['ma_slow'].iloc[i-1]:
if self.position == 0:
self.position = 1
trades.append({
'time': df['timestamp'].iloc[i],
'type': 'BUY',
'price': df['close'].iloc[i]
})
elif df['ma_fast'].iloc[i] < df['ma_slow'].iloc[i] and \
df['ma_fast'].iloc[i-1] >= df['ma_slow'].iloc[i-1]:
if self.position == 1:
self.position = 0
trades.append({
'time': df['timestamp'].iloc[i],
'type': 'SELL',
'price': df['close'].iloc[i]
})
# 백테스팅 결과 출력
if len(trades) >= 2:
initial = 10000
for t in trades:
if t['type'] == 'BUY':
entry_price = t['price']
else:
pnl_pct = (t['price'] - entry_price) / entry_price * 100
initial *= (1 + pnl_pct / 100)
total_return = (initial - 10000) / 10000 * 100
print(f"📈 백테스팅 결과:")
print(f" 총 거래 횟수: {len(trades)}")
print(f" 최종 잔고: ${initial:.2f}")
print(f" 총 수익률: {total_return:.2f}%")
return trades
async def run_live_trading(self, duration_minutes: int = 60):
"""실시간 거래 실행"""
print(f"🚀 실시간 거래 시작 (최대 {duration_minutes}분)")
start_time = datetime.utcnow()
try:
await self.env.start_live_stream(
symbol=self.symbol,
timeframe=self.timeframe,
use_ai_signal=True
)
except KeyboardInterrupt:
print("\n⏹️ 사용자 중단")
finally:
await self.env.close()
실행 예제
if __name__ == "__main__":
# 백테스팅 예시
bot = CryptoTradingBot(
mode="backtest",
tardis_key="YOUR_TARDIS_API_KEY"
)
asyncio.run(bot.run_backtest(days=30))
# 실전 거래 예시 (주석 해제 후 사용)
# bot = CryptoTradingBot(
# mode="live",
# exchange="binance",
# holy_sheep_key="YOUR_HOLYSHEEP_API_KEY"
# )
# asyncio.run(bot.run_live_trading(duration_minutes=60))
---
자주 발생하는 오류와 해결책
오류 1: Tardis API 응답 형식 불일치
**문제