크립토 트레이딩 시스템을 구축하려면 고품질的历史 tick 데이터가 필수입니다. 이번 튜토리얼에서는 Tardis API로 OKX Perpetual Futures(OKX永续合约)의 실시간·과거 데이터를 수집하고, Parquet 포맷으로 변환해 백테스팅 시스템에 연결하는 End-to-End 파이프라인을 구축하겠습니다. HolySheep AI의 다중 모델 통합 게이트웨이를 통해 실시간 데이터 전처리부터 AI 기반 시그널 생성까지 한 번에 처리하는 실전 아키텍처를 공개합니다.
왜 Tardis API인가?
크립토 거래소 API는 Binance, Bybit, OKX 등 각 거래소마다 구조가 다르고 Rate Limit 정책도 상이합니다. Tardis API는 이 차이를 추상화하여 단일 인터페이스로 여러 거래소의 historical market data에 접근할 수 있게 해줍니다. 특히 OKX永续合约 같은 perpetual futures 데이터는 BitQuery나 자체 스크래핑 대비:
- 데이터 신뢰도: 거래소 공식 WebSocket 피드 기반
- 커버리지: 2019년부터现在的 모든 tick 데이터
- 포맷 일관성: unified JSON schema across exchanges
- 파티셔닝: 일별/심볼별 파일로 분할 제공
아키텍처 개요
┌─────────────────────────────────────────────────────────────────┐
│ OKX Perpetual Futures │
│ (永续合约 USDT-Market) │
└─────────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Tardis API │
│ • Historical Replays (과거 데이터) │
│ • Live Streaming (실시간 WebSocket) │
│ • Data Normalization │
└─────────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Python Data Pipeline │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Fetcher │───▶│ Transformer│───▶│ Parquet │ │
│ │ (asyncio) │ │ (pandas) │ │ Writer │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ HolySheep AI Gateway │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ GPT-4.1 │ │ Claude 3.5 │ │ DeepSeek V3 │ │
│ │ 시그널 생성 │ │ 리스크 분석 │ │ 가격 예측 │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 백테스팅 엔진 │
│ (Backtrader / VectorBT) │
└─────────────────────────────────────────────────────────────────┘
사전 준비
필수 환경
# Python 3.10+ 권장
python --version # Python 3.10.13
핵심 의존성 설치
pip install tardis-client aiohttp pandas pyarrow sqlalchemy
pip install backtrader vectorbt-pro # 백테스팅
HolySheep AI SDK
pip install openai anthropic
환경 변수 설정
export TARDIS_API_KEY="your_tardis_api_key_here"
export HOLYSHEEP_API_KEY="your_holysheep_api_key_here"Tardis API 키 발급
Tardis.dev에서 무료 계정 생성 후 API 키를 발급받습니다. 무료 플랜은 일 10,000개 메시지 제한이 있고, USDT-M futures 데이터만 접근 가능합니다. 상업용 백테스팅에는 Pro 플랜(월 $49~)을 권장합니다.
Tardis API 데이터 수집 파이프라인
1. OKX永续合约 Historical Data Fetcher
"""
OKX Perpetual Futures (OKX永续合约) Historical Tick Data Fetcher
Tardis API를 사용한 고성능 데이터 수집 파이프라인
"""
import asyncio
import json
import os
from datetime import datetime, timedelta
from typing import AsyncGenerator
import aiohttp
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from tardis_client import TardisClient, TradingPair
from tardis_client.models import BookL2UpdateEvent, TradeEvent
class OKXPerpetualDataFetcher:
"""
OKX USDT-M永续合约 Historical Data Fetcher
Tardis API의 unified schema를 활용하여
거래소별 상이한 데이터 포맷을 단일화
"""
BASE_URL = "https://api.tardis.dev/v1"
# OKX Perpetual Futures 마켓 식별자
OKX_PERPETUAL_EXCHANGE = "okx"
SUPPORTED_SYMBOLS = [
"BTC-USDT-SWAP",
"ETH-USDT-SWAP",
"SOL-USDT-SWAP",
"BNB-USDT-SWAP",
"XRP-USDT-SWAP"
]
def __init__(self, api_key: str):
self.api_key = api_key
self.client = None
async def fetch_historical_trades(
self,
symbol: str,
start_date: datetime,
end_date: datetime,
verbose: bool = True
) -> pd.DataFrame:
"""
OKX永续合约 과거 거래 데이터 수집
Args:
symbol: 거래 심볼 (e.g., "BTC-USDT-SWAP")
start_date: 데이터 시작일
end_date: 데이터 종료일
verbose: 진행 상황 출력 여부
Returns:
pd.DataFrame: 거래 데이터 (timestamp, price, size, side)
"""
url = f"{self.BASE_URL}/historical/trades"
params = {
"exchange": self.OKX_PERPETUAL_EXCHANGE,
"symbol": symbol,
"from_": int(start_date.timestamp()),
"to": int(end_date.timestamp()),
"limit": 50000 # 최대 페이지당 레코드 수
}
headers = {"Authorization": f"Bearer {self.api_key}"}
all_trades = []
offset = 0
async with aiohttp.ClientSession() as session:
while True:
params["offset"] = offset
async with session.get(
url,
params=params,
headers=headers,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status != 200:
error_text = await response.text()
raise RuntimeError(
f"Tardis API 오류: {response.status} - {error_text}"
)
data = await response.json()
if not data.get("data"):
break
trades = data["data"]
all_trades.extend(trades)
if verbose:
print(
f"[{symbol}] {len(trades)}건 수집됨 "
f"(총 {len(all_trades)}건, offset={offset})"
)
# 다음 페이지 확인
if not data.get("hasMore"):
break
offset += len(trades)
# Rate Limit 방지
await asyncio.sleep(0.1)
# DataFrame 변환
df = pd.DataFrame(all_trades)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
df = df.sort_values("timestamp").reset_index(drop=True)
return df
async def stream_realtime_trades(
self,
symbols: list[str],
duration_seconds: int = 3600
) -> AsyncGenerator[pd.DataFrame, None]:
"""
OKX永续合约 실시간 거래 스트리밍
WebSocket 기반으로 지연 시간 최소화
HolySheep AI와 연계하여 실시간 시그널 생성에 활용
Args:
symbols: 구독 심볼 목록
duration_seconds: 스트리밍 지속 시간
"""
# Tardis Realtime API 사용
realtime_url = (
f"{self.BASE_URL}/feeds"
f"?exchange={self.OKX_PERPETUAL_EXCHANGE}"
f"&symbols={','.join(symbols)}"
)
headers = {"Authorization": f"Bearer {self.api_key}"}
async with aiohttp.ClientSession() as session:
async with session.get(
realtime_url,
headers=headers,
timeout=aiohttp.ClientTimeout(total=duration_seconds + 10)
) as ws:
buffer = []
buffer_size = 100 # 배치 처리
start_time = asyncio.get_event_loop().time()
async for line in ws.content:
# Heartbeat check
elapsed = asyncio.get_event_loop().time() - start_time
if elapsed > duration_seconds:
break
try:
data = json.loads(line.decode())
if data.get("type") == "trade":
buffer.append({
"timestamp": pd.Timestamp.now(),
"symbol": data.get("symbol"),
"price": float(data.get("price")),
"size": float(data.get("amount")),
"side": data.get("side"),
"trade_id": data.get("id")
})
# 배치 처리
if len(buffer) >= buffer_size:
yield pd.DataFrame(buffer)
buffer = []
except json.JSONDecodeError:
continue
# 남은 데이터 flush
if buffer:
yield pd.DataFrame(buffer)
async def main():
"""데이터 수집 메인 실행"""
fetcher = OKXPerpetualDataFetcher(
api_key=os.environ["TARDIS_API_KEY"]
)
# BTC-USDT永续合约 2024년 1월 데이터 수집
start = datetime(2024, 1, 1)
end = datetime(2024, 1, 31)
btc_trades = await fetcher.fetch_historical_trades(
symbol="BTC-USDT-SWAP",
start_date=start,
end_date=end,
verbose=True
)
print(f"수집 완료: {len(btc_trades)}건")
print(f"시간 범위: {btc_trades['timestamp'].min()} ~ {btc_trades['timestamp'].max()}")
return btc_trades
if __name__ == "__main__":
df = asyncio.run(main())
# Parquet 저장
df.to_parquet(
"okx_btc_usdt_swap_2024_01.parquet",
engine="pyarrow",
compression="snappy"
)
print("Parquet 파일 저장 완료")2. Parquet 파이프라인: 데이터 변환과 최적화
"""
OKX永续合约 Tick 데이터 Parquet 변환 및 최적화 파이프라인
성능 최적화를 위한 컬럼 타입 지정 및 파티셔닝
"""
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
from datetime import datetime
from typing import Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OKXParquetPipeline:
"""
OKX永续合约 Tick 데이터를 Parquet로 변환하는 파이프라인
Parquet 포맷 선택 이유:
• Column-oriented: tick 데이터의 특정 컬럼만 읽기 가능
• 압축 효율:snappy로 약 70% 압축률
• 스키마 보존: 데이터 타입 자동 추론 문제 해결
• 분할 쿼리: 일별/심볼별 파티셔닝으로 대용량 데이터的高速 검색
"""
# Tardis API에서 받은 raw 데이터 스키마
RAW_SCHEMA = pa.schema([
("timestamp", pa.timestamp("ms")),
("symbol", pa.string()),
("price", pa.float64()),
("size", pa.float64()),
("side", pa.string()), # "buy" or "sell"
("trade_id", pa.string()),
("fee", pa.float64()),
("fee_currency", pa.string()),
])
# 최적화된 Parquet 스키마
OPTIMIZED_DTYPES = {
"timestamp": "datetime64[ms]",
"symbol": "category", # 반복 문자열 압축
"price": "float32", # 8바이트 → 4바이트
"size": "float32",
"side": "category",
"trade_id": "string",
"fee": "float32",
"fee_currency": "category",
}
def __init__(self, output_dir: str = "./data/parquet"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
def transform_to_ohlcv(
self,
df: pd.DataFrame,
timeframe: str = "1min"
) -> pd.DataFrame:
"""
Tick 데이터를 OHLCV 형태로 리샘플링
Args:
df: Tick 데이터 DataFrame
timeframe: 리샘플링 주기 ("1T", "5T", "15T", "1H", "4H", "1D")
Returns:
pd.DataFrame: OHLCV 데이터
"""
df = df.set_index("timestamp").sort_index()
ohlcv = pd.DataFrame()
ohlcv["open"] = df["price"].resample(timeframe).first()
ohlcv["high"] = df["price"].resample(timeframe).max()
ohlcv["low"] = df["price"].resample(timeframe).min()
ohlcv["close"] = df["price"].resample(timeframe).last()
ohlcv["volume"] = df["size"].resample(timeframe).sum()
ohlcv["trade_count"] = df["price"].resample(timeframe).count()
# VWAP 계산
df["dollar_volume"] = df["price"] * df["size"]
ohlcv["vwap"] = (
df["dollar_volume"].resample(timeframe).sum() /
df["size"].resample(timeframe).sum()
)
return ohlcv.dropna()
def write_partitioned_parquet(
self,
df: pd.DataFrame,
symbol: str,
year_month: str,
partition_cols: Optional[list] = None
) -> str:
"""
파티셔닝된 Parquet 파일 저장
저장 구조:
data/
├── symbol=btc_usdt_swap/
│ ├── dt=2024-01/
│ │ └── data.parquet
│ └── dt=2024-02/
│ └── data.parquet
└── symbol=eth_usdt_swap/
└── dt=2024-01/
└── data.parquet
Args:
df: 저장할 DataFrame
symbol: 심볼명
year_month: 파티션 키 (e.g., "2024-01")
partition_cols: 파티션 컬럼 리스트
Returns:
str: 저장된 파일 경로
"""
if partition_cols is None:
partition_cols = ["symbol", "dt"]
# 파티션 컬럼 추가
df = df.copy()
df["symbol"] = symbol.lower().replace("-", "_")
df["dt"] = year_month
# 저장 경로 생성
output_path = self.output_dir / f"symbol={df['symbol'].iloc[0]}" / f"dt={year_month}"
output_path.mkdir(parents=True, exist_ok=True)
file_path = output_path / "data.parquet"
# Parquet 작성 (최적화 옵션)
table = pa.Table.from_pandas(
df,
schema=self.RAW_SCHEMA,
preserve_index=False
)
# Parquet 메타데이터 최적화
parquet_args = {
"compression": "snappy", # 빠른 압축/해제
"use_dictionary": True, # 카테고리 컬럼 최적화
"write_statistics": True, # min/max 인덱싱
}
pq.write_table(
table,
str(file_path),
**parquet_args
)
file_size_mb = file_path.stat().st_size / 1024 / 1024
logger.info(
f"저장 완료: {file_path} "
f"(레코드: {len(df):,}, 크기: {file_size_mb:.2f}MB)"
)
return str(file_path)
def read_partitioned_data(
self,
symbol: Optional[str] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
columns: Optional[list] = None
) -> pd.DataFrame:
"""
파티셔닝된 Parquet 파일 읽기
핵심: 전체 스캔 대신 필요한 파티션만 읽기
Args:
symbol: 심볼 필터 (None이면 전체 심볼)
start_date: 시작일 필터
end_date: 종료일 필터
columns: 읽을 컬럼 리스트 (None이면 전체)
Returns:
pd.DataFrame: 필터링된 데이터
"""
# Parquet Dataset API로 프루닝
dataset = pq.ParquetDataset(
str(self.output_dir),
filters=None, # pandas에서 필터링
use_legacy_dataset=False
)
# 필요한 컬럼만 읽기 (I/O 최적화)
if columns:
table = dataset.read(columns=columns)
else:
table = dataset.read()
df = table.to_pandas()
# 필터 적용
if symbol:
df = df[df["symbol"] == symbol.lower().replace("-", "_")]
if start_date:
df = df[df["timestamp"] >= start_date]
if end_date:
df = df[df["timestamp"] <= end_date]
return df.sort_values("timestamp").reset_index(drop=True)
def get_data_stats(self) -> dict:
"""데이터셋 통계 정보 조회"""
dataset = pq.ParquetDataset(str(self.output_dir))
total_rows = 0
total_size_mb = 0
date_range = {"min": None, "max": None}
symbols = set()
for file_path in self.output_dir.rglob("*.parquet"):
table = pq.read_table(str(file_path))
total_rows += len(table)
total_size_mb += file_path.stat().st_size / 1024 / 1024
df = table.to_pandas()
if "symbol" in df.columns:
symbols.update(df["symbol"].unique())
if "timestamp" in df.columns:
ts_min = df["timestamp"].min()
ts_max = df["timestamp"].max()
if date_range["min"] is None or ts_min < date_range["min"]:
date_range["min"] = ts_min
if date_range["max"] is None or ts_max > date_range["max"]:
date_range["max"] = ts_max
return {
"total_rows": total_rows,
"total_size_mb": round(total_size_mb, 2),
"symbols": sorted(symbols),
"date_range": date_range
}
===== 실행 예제 =====
if __name__ == "__main__":
pipeline = OKXParquetPipeline(output_dir="./data/okx_perpetual")
# 기존 tick 데이터 로드 (Tardis Fetcher에서 생성)
tick_df = pd.read_parquet("okx_btc_usdt_swap_2024_01.parquet")
print(f"Tick 데이터 로드: {len(tick_df):,}건")
# OHLCV 변환
ohlcv_1h = pipeline.transform_to_ohlcv(tick_df, timeframe="1H")
print(f"1시간봉 OHLCV: {len(ohlcv_1h)}개")
# 파티셔닝된 Parquet 저장
pipeline.write_partitioned_parquet(
df=tick_df,
symbol="BTC-USDT-SWAP",
year_month="2024-01"
)
# 필터링된 읽기 테스트
filtered = pipeline.read_partitioned_data(
symbol="btc_usdt_swap",
start_date=datetime(2024, 1, 15),
end_date=datetime(2024, 1, 31)
)
print(f"필터링 결과: {len(filtered):,}건")
# 통계 확인
stats = pipeline.get_data_stats()
print(f"데이터셋 통계: {stats}")3. HolySheep AI 게이트웨이 연동: AI 시그널 생성
"""
HolySheep AI 게이트웨이를 통한 OKX永续合约 백테스팅 시그널 생성
단일 API 키로 다중 모델 활용: GPT-4.1, Claude, DeepSeek V3
base_url: https://api.holysheep.ai/v1 (절대 openai.com 사용 금지)
"""
import os
import json
from typing import Optional
from dataclasses import dataclass
from datetime import datetime
HolySheep AI SDK - 반드시 이 설정 사용
import openai
from openai import AsyncOpenAI
전역 클라이언트 초기화
HOLYSHEEP_CLIENT = AsyncOpenAI(
api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1" # ✅ HolySheep 게이트웨이
)
@dataclass
class TradingSignal:
"""AI 생성 거래 시그널"""
timestamp: datetime
symbol: str
direction: str # "long", "short", "neutral"
confidence: float # 0.0 ~ 1.0
entry_price: float
stop_loss: float
take_profit: float
reasoning: str
model_used: str
class HolySheepSignalGenerator:
"""
HolySheep AI 게이트웨이 기반 크립토 시그널 생성기
HolySheep 선택 이유:
1. 단일 API 키로 GPT-4.1, Claude Sonnet, DeepSeek V3 통합
2. 로컬 결제 지원 (해외 신용카드 불필요)
3. GPT-4.1: $8/MTok, DeepSeek V3: $0.42/MTok (업계 최저가)
4. 99.9% uptime SLA
"""
def __init__(self, api_key: str):
self.client = AsyncOpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
async def generate_rsi_divergence_signal(
self,
symbol: str,
ohlcv_data: list[dict],
rsi_value: float,
prev_rsi: float
) -> TradingSignal:
"""
RSI 다이버전스 기반 시그널 생성
GPT-4.1로 기술적 지표 분석 + 거래 신호 도출
"""
prompt = f"""OKX {symbol} Perpetual Futures 단타 시그널을 분석해주세요.
현재 RSI(14): {rsi_value:.2f}
이전 RSI: {prev_rsi:.2f}
최근 20개 봉 데이터:
{json.dumps(ohlcv_data[-20:], indent=2)}
분석 요청:
1. RSI 다이버전스 여부 판단 (bullish/bearish/none)
2. 매수/매도/관망 권고
3. 신뢰도 점수 (0.0~1.0)
4. 추천 진입가,止损,.takeprofit
5. 근거 설명 (한국어로)
"""
response = await self.client.chat.completions.create(
model="gpt-4.1", # HolySheep 게이트웨이 모델명
messages=[
{
"role": "system",
"content": "당신은 전문 크립토 트레이더입니다. JSON 형식으로 응답해주세요."
},
{
"role": "user",
"content": prompt
}
],
temperature=0.3, # 일관된 분석을 위해 낮은 temperature
max_tokens=800
)
result_text = response.choices[0].message.content
# JSON 파싱 (GPT 응답 형식화)
try:
# 마크다운 코드블록 제거
result_text = result_text.strip("`").strip("json\n")
signal_data = json.loads(result_text)
return TradingSignal(
timestamp=datetime.now(),
symbol=symbol,
direction=signal_data.get("direction", "neutral"),
confidence=signal_data.get("confidence", 0.5),
entry_price=signal_data.get("entry_price", 0),
stop_loss=signal_data.get("stop_loss", 0),
take_profit=signal_data.get("take_profit", 0),
reasoning=signal_data.get("reasoning", ""),
model_used="GPT-4.1 via HolySheep"
)
except json.JSONDecodeError:
# 파싱 실패 시 기본값 반환
return TradingSignal(
timestamp=datetime.now(),
symbol=symbol,
direction="neutral",
confidence=0.0,
entry_price=0,
stop_loss=0,
take_profit=0,
reasoning=f"AI 응답 파싱 실패: {result_text[:100]}",
model_used="GPT-4.1 via HolySheep"
)
async def analyze_risk_with_claude(
self,
trade_signal: TradingSignal,
portfolio_data: dict
) -> dict:
"""
Claude Sonnet 4.5로 리스크 분석
HolySheep AIなら同一キーでClaudeにもアクセス可能
DeepSeek 대비 고급 추론 capability 활용
"""
prompt = f"""다음 거래 시그널의 리스크를 분석해주세요.
시그널 정보:
- 심볼: {trade_signal.symbol}
- 방향: {trade_signal.direction}
- 진입가: ${trade_signal.entry_price:,.2f}
-止损: ${trade_signal.stop_loss:,.2f}
-利益確定: ${trade_signal.take_profit:,.2f}
- 신뢰도: {trade_signal.confidence:.0%}
포트폴리오 상태:
{json.dumps(portfolio_data, indent=2)}
분석 요청:
1. 리스크/리워드 비율 평가
2. 포트폴리오 적합성 (포지션 사이즈 권장)
3. 최대 손실 가능액 (Max Drawdown 예상)
4. 실행 추천 여부
"""
response = await self.client.chat.completions.create(
model="claude-sonnet-4-20250514", # HolySheep 게이트웨이 모델명
messages=[
{
"role": "system",
"content": "당신은 전문 리스크 관리 애널리스트입니다. JSON 형식으로 응답해주세요."
},
{
"role": "user",
"content": prompt
}
],
temperature=0.2,
max_tokens=600
)
result_text = response.choices[0].message.content
try:
result_text = result_text.strip("`").strip("json\n")
return json.loads(result_text)
except json.JSONDecodeError:
return {"error": "리스크 분석 파싱 실패", "raw_response": result_text}
async def batch_signal_with_deepseek(
self,
ohlcv_batch: list[dict]
) -> list[dict]:
"""
DeepSeek V3으로 배치 시그널 생성
HolySheep AI 가격:
- DeepSeek V3.2: $0.42/MTok (업계 최저가)
- 대량 처리 시 GPT-4 대비 95% 비용 절감
Args:
ohlcv_batch: OHLCV 데이터 배치 (최대 100개 심볼)
Returns:
list[dict]: 배치 시그널 결과
"""
symbols_data = "\n".join([
f"{i+1}. {item['symbol']}: O={item['open']:.2f} H={item['high']:.2f} "
f"L={item['low']:.2f} C={item['close']:.2f} V={item['volume']:.0f}"
for i, item in enumerate(ohlcv_batch)
])
prompt = f"""다음 암호화폐 가격 데이터를 분석하여 각 심볼별 단타 시그널을 생성해주세요.
데이터:
{symbols_data}
출력 형식 (JSON 배열):
[
{{"symbol": "BTC-USDT", "direction": "long|short|neutral", "confidence": 0.0~1.0, "reasoning": "..."}},
...
]
한국어로_reasoning을 작성해주세요."""
response = await self.client.chat.completions.create(
model="deepseek-chat-v3.2", # HolySheep 게이트웨이 모델명
messages=[
{
"role": "system",
"content": "당신은 다중 심볼 크립토 분석가입니다. 정확하게 JSON 배열로만 응답해주세요."
},
{
"role": "user",
"content": prompt
}
],
temperature=0.4,
max_tokens=2000
)
result_text = response.choices[0].message.content
try:
result_text = result_text.strip("`").strip("json\n")
return json.loads(result_text)
except json.JSONDecodeError:
return []
async def run_backtest_with_ai():
"""AI 시그널 기반 백테스트 실행 예제"""
from .parquet_pipeline import OKXParquetPipeline
# HolySheep AI 클라이언트 초기화
signal_gen = HolySheepSignalGenerator(
api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
)
# Parquet에서 데이터 로드
pipeline = OKXParquetPipeline("./data/okx_perpetual")
df = pipeline.read_partitioned_data(
symbol="btc_usdt_swap",
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 3, 31)
)
# 1시간봉 OHLCV 변환
ohlcv = pipeline.transform_to_ohlcv(df, timeframe="1H")
ohlcv_list = ohlcv.reset_index().to_dict("records")
# RSI 계산
delta = pd.Series(ohlcv["close"]).diff()
gain = delta.where(delta > 0, 0).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
signals = []
# 각 봉에서 AI 시그널 생성
for i in range(20, len(ohlcv_list)):
current_rsi = rsi.iloc[i]
prev_rsi = rsi.iloc[i-1]
# RSI 다이버전스 탐지
if abs(current_rsi - prev_rsi) > 5: # RSI 변화 임계값
signal = await signal_gen.generate_rsi_divergence_signal(
symbol="BTC-USDT-SWAP",
ohlcv_data=ohlcv_list[max(0, i-20):i+1],
rsi_value=current_rsi,
prev_rsi=prev_rsi
)
if signal.direction != "neutral":
signals.append(signal)
print(f"[{signal.timestamp}] {signal.direction.upper()} @ ${signal.entry_price:,.2f}")
print(f"\n총 {len(signals)}개 시그널 생성 완료")
return signals
if __name__ == "__main__":
import asyncio
signals = asyncio.run(run_backtest_with_ai())백테스팅 엔진 연동
"""
Backtrader 기반 AI 시그널 백테스팅
OKX永续合约 Parquet 데이터 + HolySheep AI 시그널 통합
"""
import backtrader as bt
import pandas as pd
from datetime import datetime
from typing import Optional
import asyncio
from .parquet_pipeline import OKXParquetPipeline
from .holy_sheep_gateway import HolySheepSignalGenerator, TradingSignal
class AISignalStrategy(bt.Strategy):
"""
HolySheep AI 시그널 기반 거래 전략
특징:
- HolySheep AI에서 실시간 시그널 수신
- RSI 다이버전스 + AI 판단 결합
- 파라미터 최적화 지원
"""
params = (
("rsi_period", 14),
("rsi_overbought", 70),
("rsi_oversold", 30),
("signal_generator", None), # HolySheepSignalGenerator 인스턴스
("symbols", ["BTC-USDT-SWAP"]),
("position_size", 0.95), # 계좌의 95% 사용
)
def __init__(self):
self.data_close = self.datas[0].close
self.data_open = self.datas[0].open
self.data_high = self.datas[0].high
self.data_low = self.datas[0].low
self.data_volume = self.datas[0].volume
# RSI 지표
self.rsi = bt.indicators.RSI(
self.data_close,
period=self.params.rsi_period
)
# 시그널 캐시
self.ai_signal: Optional[TradingSignal] = None
self.last_signal_time = None
def next(self):
# HolySheep AI에서 시그널 요청 (10봉마다)
if len(self) % 10 == 0 and self.params.signal_generator:
asyncio.create_task(self.request_ai_signal())
# 포지션 없음 +