안녕하세요, 저는 HolySheep AI의 기술 아키텍트입니다. 이번 튜토리얼에서는 Hyperliquid DEX의 온체인 주문 흐름 데이터를 분석하고, Tardis에서 제공하는 고품질 히스토리 데이터를 연동하여 정량 거래 백테스팅 파이프라인을 처음부터 구축하는 방법을 단계별로 안내하겠습니다. 초보자도 이해할 수 있도록 각 개념을 쉽게 설명하고, 실제 작동하는 코드와 구체적인 수치를 포함했습니다.
1. Hyperliquid DEX란 무엇인가
Hyperliquid는 퍼페추얼 스왑 거래를 제공하는 탈중앙화 거래소(DEX)로서, 자체 L1 블록체인을 운영합니다. 다른 DEX들과 달리 Hyperliquid는 CEX에 가까운 거래 속도와 낮은 가스비를 제공하는 것이 특징입니다. 핵심적으로 주목해야 할 점은 Hyperliquid가 모든 주문, 청산, 포지션 변경 등의 온체인 데이터를 공개한다는 것입니다. 이 데이터를 분석하면 시장 참여자들의 행동 패턴, 대형 포지션의 진입·출구 시점, 그리고 핍스퀘빙(피드 스퀘어) 압력 등을 파악할 수 있습니다.
주문 흐름 분석(Order Flow Analysis)은 전통적인 고频交易(HFT)에서 사용되는 기법으로, 누가 어떤 가격에 얼마나 많이 매수·매도하고 있는지를 추적하여 시장 방향성을 예측하는 방법론입니다. Hyperliquid에서는 다음 데이터를 활용할 수 있습니다:
- 주문 발주(Order Placement): 가격, 수량, 방향(매수/매도)
- 주문 취소(Order Cancellation): 취소 패턴 분석
- 체결(Trade Execution): 실제成交 가격과 수량
- 청산(Liquidation): 강제 청산 Events
- 펀딩비(Funding Rate): 순환 비용 지표
2. Tardis.historical이란
Tardis는 암호화폐 시장 데이터 제공자로, Hyperliquid를 포함한 주요 거래소의 Historical Market Data를 제공합니다. Tardis의 핵심 강점은 millisecond 단위의 정밀한 타임스탬프와 실제 체결 기반으로 정제된 데이터입니다. 다른 무료 데이터 소스와 비교했을 때 Tardis는 다음과 같은 차별점을 가집니다:
- 데이터 정제: 네트워크 문제로 인한 결측치와 이상치를 자동으로 처리
- 통합 포맷: 여러 거래소의 데이터를 동일한 스키마로 제공
- 저지연 스트리밍: Real-time 데이터도 millisecond 내외로 제공
- 카산드라 스토리지: 대용량 시계열 데이터를 효율적으로 질의
3. 왜 HolySheep AI가 필요한가
주문 흐름 분석을 수행하다 보면, 대량의 원시 데이터를 정제하고 패턴을 찾아내는 과정에서 AI 모델의 도움이 필수적입니다. HolySheep AI를 사용하면 단일 API 키로 다양한 AI 모델을 통합할 수 있어, 데이터 분석 파이프라인에서 모델 교체나 비교가 자유롭습니다. 예를 들어, 주문 흐름의 비정상 패턴을 감지할 때는 Claude Sonnet 4.5를, 실시간 신호 생성이 필요할 때는 Gemini 2.5 Flash를 사용하는 식입니다.
4. 환경 구성
시작하기 전에 필요한 도구들을 설치해야 합니다. 이 튜토리얼에서는 Python 3.10 이상을 사용하며, 필요한 패키지들을 순서대로 설치하겠습니다. 터미널에서 다음 명령어를 실행해주세요.
# 기본 의존성 설치
pip install pandas numpy requests asyncio aiohttp
Tardis API 연동을 위한 패키지
pip install tardis-dev
AI API 연동을 위한 OpenAI 호환 라이브러리
pip install openai httpx
데이터 시각화
pip install plotly kaleido
백테스팅 프레임워크
pip install backtrader vectorbt
HolySheep AI SDK (OpenAI 호환)
pip install holy-sheep-ai # 또는 httpx만으로 직접 구현5. Tardis에서 Hyperliquid 데이터 가져오기
Tardis는 REST API와 WebSocket 두 가지 방식으로 데이터를 제공합니다. 백테스팅을 위해서는 Historical 데이터를 CSV나 Parquet 형태로 다운로드하는 것이 효율적입니다. 먼저 Tardis 계정을 생성하고 API 키를 발급받아야 합니다.
5.1 Tardis API 키 발급
참고: Tardis 웹사이트(https://tardis.dev)에서 가입 후 대시보드에서 API 키를 확인할 수 있습니다. 무료 플랜으로도 제한된 범위의 데이터에 접근할 수 있지만, 본격적인 백테스팅에는 유료 플랜이 필요합니다. API 키는 "tardis-api-key-xxxxx" 형식으로 표시됩니다.
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
======= Tardis API 설정 =======
TARDIS_API_KEY = "your_tardis_api_key_here"
TARDIS_BASE_URL = "https://api.tardis.dev/v1"
Hyperliquid 심볼 목록 조회
def get_hyperliquid_symbols():
"""사용 가능한 Hyperliquid 거래 쌍 목록"""
url = f"{TARDIS_BASE_URL}/exchanges/hyperliquid/symbols"
headers = {"Authorization": f"Bearer {TARDIS_API_KEY}"}
response = requests.get(url, headers=headers)
if response.status_code == 200:
symbols = response.json()
print(f"총 {len(symbols)}개의 Hyperliquid 심볼을 찾았습니다:")
return symbols
else:
print(f"심볼 조회 실패: {response.status_code}")
print(response.text)
return []
테스트 실행
if __name__ == "__main__":
symbols = get_hyperliquid_symbols()
# 상위 10개 심볼 출력
for s in symbols[:10]:
print(f" - {s}")5.2 Historical 데이터 다운로드
백테스팅에 사용할 Historical 데이터를 다운로드하는 함수를 구현합니다. Tardis는 trades, quotes, orderbook 등의 데이터 타입을 제공하며, 이 튜토리얼에서는 체결 데이터(trades)와 주문장 데이터(orderbook)를 중점적으로 다룹니다.
import requests
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
import time
import os
TARDIS_API_KEY = "your_tardis_api_key_here"
TARDIS_BASE_URL = "https://api.tardis.dev/v1"
def download_trades(symbol: str, start_date: datetime, end_date: datetime, save_dir: str = "./data"):
"""
특정 심볼의 특정 기간 동안 체결 데이터를 다운로드합니다.
Args:
symbol: 거래 심볼 (예: "BTC-PERP")
start_date: 시작 일시
end_date: 종료 일시
save_dir: 저장 디렉토리
Returns:
저장된 파일 경로
"""
save_path = Path(save_dir)
save_path.mkdir(parents=True, exist_ok=True)
filename = f"{symbol.replace('-', '_')}_{start_date.strftime('%Y%m%d')}_{end_date.strftime('%Y%m%d')}.csv"
filepath = save_path / filename
# 이미 파일이 존재하면スキップ
if filepath.exists():
print(f"[INFO] {filename}이 이미 존재합니다. 다운로드をスキップ합니다.")
return str(filepath)
print(f"[INFO] {symbol} 데이터 다운로드 시작: {start_date} ~ {end_date}")
url = f"{TARDIS_BASE_URL}/export/console/hyperliquid/trades"
params = {
"symbol": symbol,
"from": int(start_date.timestamp()),
"to": int(end_date.timestamp()),
"format": "csv",
"datatype": "trades"
}
headers = {"Authorization": f"Bearer {TARDIS_API_KEY}"}
try:
response = requests.get(url, headers=headers, params=params, stream=True)
if response.status_code == 200:
with open(filepath, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
# 행 수 확인
df = pd.read_csv(filepath)
print(f"[SUCCESS] {filename} 저장 완료: {len(df):,}행, {filepath.stat().st_size / 1024 / 1024:.2f}MB")
return str(filepath)
else:
print(f"[ERROR] 다운로드 실패: {response.status_code}")
print(response.text)
return None
except Exception as e:
print(f"[ERROR] 예외 발생: {str(e)}")
return None
def batch_download_multi_symbols(symbols: list, start_date: datetime, end_date: datetime,
delay_seconds: float = 1.0, save_dir: str = "./data"):
"""
여러 심볼의 데이터를 순차적으로 다운로드합니다.
Rate Limit 방지를 위해 요청 사이에 지연 시간을 둡니다.
"""
results = {}
for i, symbol in enumerate(symbols):
print(f"\n[{i+1}/{len(symbols)}] 처리 중: {symbol}")
result = download_trades(symbol, start_date, end_date, save_dir)
results[symbol] = result
if i < len(symbols) - 1:
print(f" Rate Limit 방지를 위해 {delay_seconds}초 대기...")
time.sleep(delay_seconds)
# 결과 요약
success_count = sum(1 for v in results.values() if v is not None)
print(f"\n[SUMMARY] 완료: {success_count}/{len(symbols)} 성공")
return results
======= 메인 실행 코드 =======
if __name__ == "__main__":
# 2024년 11월 한 달간 주요 페어 데이터 다운로드
start = datetime(2024, 11, 1)
end = datetime(2024, 12, 1)
main_symbols = ["BTC-PERP", "ETH-PERP", "SOL-PERP"]
download_results = batch_download_multi_symbols(
symbols=main_symbols,
start_date=start,
end_date=end,
delay_seconds=2.0,
save_dir="./hyperliquid_data"
)
for symbol, path in download_results.items():
status = "✅" if path else "❌"
print(f" {status} {symbol}: {path or '실패'}")6. 주문 흐름 데이터 분석 파이프라인
다운로드한 체결 데이터를 주문 흐름 분석에 적합한 형태로 가공하는 파이프라인을 구축하겠습니다. 주문 흐름 분석의 핵심은 "누가"、"어떤 가격"、"얼마나" 사고팔았는지를 추적하는 것입니다.
import pandas as pd
import numpy as np
from pathlib import Path
from collections import defaultdict
class OrderFlowAnalyzer:
"""
Hyperliquid 주문 흐름 분석기
주요 지표:
- Delta: 매수/매도 수량 차이
- Cumulative Delta: 시간에 따른 Delta 누적
- Volume Weighted Average Price (VWAP)
- Trade Size Distribution
- Order Flow Imbalance (OFI)
"""
def __init__(self, data_dir: str = "./hyperliquid_data"):
self.data_dir = Path(data_dir)
self.trades = None
self.orderflow_metrics = {}
def load_trades(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
"""
Tardis에서 다운로드한 체결 데이터 로드
"""
filename = f"{symbol.replace('-', '_')}_{start_date.replace('-', '')}_{end_date.replace('-', '')}.csv"
filepath = self.data_dir / filename
if not filepath.exists():
raise FileNotFoundError(f"데이터 파일을 찾을 수 없습니다: {filepath}")
# Tardis CSV 포맷 로드
df = pd.read_csv(filepath)
# 타임스탬프 변환
if 'timestamp' in df.columns:
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
elif 'local_timestamp' in df.columns:
df['datetime'] = pd.to_datetime(df['local_timestamp'], unit='ms')
# UTC로 변환
df['datetime'] = df['datetime'].dt.tz_localize('UTC')
# Tardis 필드 매핑 (필드명 확인 후 조정 필요)
if 'side' not in df.columns:
# Tardis 필드명 확인 필요
print(f"사용 가능한 컬럼: {df.columns.tolist()}")
# 일반적인 필드명 시도
for side_col in ['direction', 'type', 'action']:
if side_col in df.columns:
df.rename(columns={side_col: 'side'}, inplace=True)
break
self.trades = df.sort_values('datetime').reset_index(drop=True)
print(f"[INFO] {len(self.trades):,}건의 체결 데이터 로드 완료")
print(f" 시간 범위: {self.trades['datetime'].min()} ~ {self.trades['datetime'].max()}")
return self.trades
def calculate_delta(self, bucket_seconds: int = 60) -> pd.DataFrame:
"""
시간 버킷별 Delta (순매수/순매도) 계산
Delta = 매수 체결 수량 - 매도 체결 수량
"""
if self.trades is None:
raise ValueError("먼저 load_trades()를 호출해주세요")
df = self.trades.copy()
# 시간 버킷 생성
df['bucket'] = df['datetime'].dt.floor(f'{bucket_seconds}s')
# Tardis 필드명 처리 (side가 없으면 price 기반 추정)
if 'side' not in df.columns:
# price와 previous_price 비교하여 방향 추정
if 'price' in df.columns:
df['side'] = np.where(df['price'] > df['price'].shift(1), 'buy', 'sell')
# 매수/매도 분리
buy_volume = df[df['side'] == 'buy'].groupby('bucket')['amount'].sum()
sell_volume = df[df['side'] == 'sell'].groupby('bucket')['amount'].sum()
# Delta 계산
delta_df = pd.DataFrame({
'buy_volume': buy_volume,
'sell_volume': sell_volume,
'delta': buy_volume - sell_volume,
'total_volume': buy_volume + sell_volume
})
delta_df['delta_cumulative'] = delta_df['delta'].cumsum()
delta_df['buy_ratio'] = delta_df['buy_volume'] / delta_df['total_volume']
self.orderflow_metrics['delta'] = delta_df
return delta_df
def calculate_vwap(self, bucket_seconds: int = 60) -> pd.DataFrame:
"""
VWAP (Volume Weighted Average Price) 계산
"""
if self.trades is None:
raise ValueError("먼저 load_trades()를 호출해주세요")
df = self.trades.copy()
df['bucket'] = df['datetime'].dt.floor(f'{bucket_seconds}s')
df['price_volume'] = df['price'] * df['amount']
vwap_df = df.groupby('bucket').agg({
'price_volume': 'sum',
'amount': 'sum',
'price': ['min', 'max', 'first', 'last']
})
vwap_df.columns = ['price_volume_sum', 'volume_sum', 'price_min', 'price_max', 'price_open', 'price_close']
vwap_df['vwap'] = vwap_df['price_volume_sum'] / vwap_df['volume_sum']
self.orderflow_metrics['vwap'] = vwap_df
return vwap_df
def detect_large_trades(self, threshold_percentile: float = 95) -> pd.DataFrame:
"""
이상하게 큰 거래 (블로그 크리에이터) 탐지
"""
if self.trades is None:
raise ValueError("먼저 load_trades()를 호출해주세요")
df = self.trades.copy()
# 상위 5% 거래를 "큰 거래"로 분류
threshold = df['amount'].quantile(threshold_percentile / 100)
large_trades = df[df['amount'] >= threshold].copy()
print(f"[INFO] 전체 거래 중 {threshold_percentile}% 이상 거래: {len(large_trades):,}건")
print(f" 임계값: {threshold:.4f} (거래량 단위)")
print(f" 전체 거래 대비 비중: {large_trades['amount'].sum() / df['amount'].sum() * 100:.2f}%")
return large_trades
def generate_orderflow_signal(self, delta_threshold: float = 0.6) -> pd.DataFrame:
"""
주문 흐름 기반 거래 시그널 생성
Args:
delta_threshold: 매수 비율 임계값 (0.6 = 60% 이상일 때 매수 신호)
Returns:
시그널이 포함된 DataFrame
"""
delta_df = self.orderflow_metrics.get('delta')
if delta_df is None:
self.calculate_delta()
delta_df = self.orderflow_metrics['delta']
signals = delta_df.copy()
# 시그널 생성 로직
signals['buy_signal'] = signals['buy_ratio'] > delta_threshold
signals['sell_signal'] = signals['buy_ratio'] < (1 - delta_threshold)
# 신호 강도 (범위: -1 ~ +1)
signals['signal_strength'] = (signals['buy_ratio'] - 0.5) * 2
# cumulative delta 방향 전환점
signals['delta_reversal'] = np.diff(signals['delta_cumulative'].values, prepend=0) > 0
return signals
======= 분석 실행 예시 =======
if __name__ == "__main__":
analyzer = OrderFlowAnalyzer(data_dir="./hyperliquid_data")
# 데이터 로드
trades_df = analyzer.load_trades(
symbol="BTC-PERP",
start_date="20241101",
end_date="20241201"
)
# Delta 분석
delta_df = analyzer.calculate_delta(bucket_seconds=60) # 1분봉
print("\n[Delta 분석 결과]")
print(delta_df.tail(10))
# VWAP 계산
vwap_df = analyzer.calculate_vwap(bucket_seconds=60)
print("\n[VWAP 분석 결과]")
print(vwap_df.tail(10))
# 대규모 거래 탐지
large_trades = analyzer.detect_large_trades(threshold_percentile=95)
print("\n[대규모 거래 탐지 결과]")
print(large_trades[['datetime', 'price', 'amount', 'side']].head(20))
# 시그널 생성
signals = analyzer.generate_orderflow_signal(delta_threshold=0.6)
print("\n[거래 시그널]")
print(signals[signals['buy_signal'] | signals['sell_signal']].tail(20))7. HolySheep AI를 활용한 주문 흐름 패턴 분석
이제 주문 흐름 데이터에서 발견한 패턴을 HolySheep AI를 사용하여 분석하고, 백테스팅 전략에 활용하는 방법을 알아보겠습니다. HolySheep AI는 다양한 AI 모델을 단일 API로 제공하므로, 상황에 맞게 모델을 선택할 수 있습니다.
7.1 HolySheep AI SDK 설정
import httpx
import json
from typing import Optional, List, Dict, Any
class HolySheepAIClient:
"""
HolySheep AI API 클라이언트
HolySheep는 OpenAI 호환 API를 제공하므로, base_url만 변경하면 기존 코드를 재사용할 수 있습니다.
HolySheep AI 주요 모델:
- GPT-4.1: $8.00/1M 토큰 (정밀 분석)
- Claude Sonnet 4.5: $15.00/1M 토큰 (복잡한 추론)
- Gemini 2.5 Flash: $2.50/1M 토큰 (빠른 처리)
- DeepSeek V3.2: $0.42/1M 토큰 (비용 효율적)
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1" # HolySheep 공식 엔드포인트
self.client = httpx.Client(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=120.0
)
def analyze_orderflow_pattern(self, orderflow_summary: Dict[str, Any],
model: str = "claude-sonnet-4.5") -> str:
"""
주문 흐름 분석 결과를 AI로 분석
Args:
orderflow_summary: 주문 흐름 요약 데이터
model: 사용할 AI 모델
Returns:
AI 분석 결과 텍스트
"""
prompt = f"""
당신은 암호화폐 주문 흐름 분석 전문가입니다. 다음 Hyperliquid DEX의 주문 흐름 데이터를 분석해주세요.
데이터 요약
{json.dumps(orderflow_summary, indent=2, ensure_ascii=False)}
분석 요청 사항
1. 이 기간 동안의 주요 시장 심리 파악
2. 대형 거래자들의 행동 패턴 분석
3. 예상される 단기 가격 방향성
4. 위험 요소 식별
한국어로 상세하게 분석해주세요.
"""
payload = {
"model": model,
"messages": [
{"role": "system", "content": "당신은 전문적인 암호화폐 시장 분석가입니다."},
{"role": "user", "content": prompt}
],
"max_tokens": 2000,
"temperature": 0.7
}
response = self.client.post("/chat/completions", json=payload)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
raise Exception(f"API 호출 실패: {response.status_code} - {response.text}")
def generate_trading_strategy(self, signals_df: pd.DataFrame,
market_context: str) -> Dict[str, Any]:
"""
주문 흐름 시그널 기반으로 거래 전략 생성
Args:
signals_df: 주문 흐름 시그널 DataFrame
market_context: 시장 맥락 정보
Returns:
생성된 거래 전략
"""
# 시그널 요약
buy_signals = signals_df['buy_signal'].sum() if 'buy_signal' in signals_df.columns else 0
sell_signals = signals_df['sell_signal'].sum() if 'sell_signal' in signals_df.columns else 0
summary = {
"period": f"{signals_df.index.min()} ~ {signals_df.index.max()}",
"total_buckets": len(signals_df),
"buy_signals": int(buy_signals),
"sell_signals": int(sell_signals),
"avg_delta": float(signals_df['delta'].mean()) if 'delta' in signals_df.columns else 0,
"avg_buy_ratio": float(signals_df['buy_ratio'].mean()) if 'buy_ratio' in signals_df.columns else 0.5
}
prompt = f"""
다음 주문 흐름 시그널 데이터를 바탕으로 Hyperliquid DEX용 거래 전략을 생성해주세요.
시장 맥락
{market_context}
시그널 요약
{json.dumps(summary, indent=2, ensure_ascii=False)}
요청 사항
1. 진입 조건 (Entry Conditions)
2. 손절절 수준 (Stop Loss)
3. 목표 수익률 (Take Profit)
4. 포지션 크기 결정 규칙
5.风险管理 가이드라인
응답은 반드시 다음 JSON 형식으로 제공해주세요:
{{
"strategy_name": "...",
"entry_conditions": [...],
"stop_loss_pct": 0.0,
"take_profit_pct": 0.0,
"position_sizing": "...",
"risk_management": [...]
}}
"""
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "당신은 전문적인 정량 거래 전략가입니다. 항상 유효한 JSON만 반환합니다."},
{"role": "user", "content": prompt}
],
"max_tokens": 3000,
"temperature": 0.3,
"response_format": {"type": "json_object"}
}
response = self.client.post("/chat/completions", json=payload)
if response.status_code == 200:
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
else:
raise Exception(f"API 호출 실패: {response.status_code} - {response.text}")
def close(self):
self.client.close()
======= HolySheep AI 사용 예시 =======
if __name__ == "__main__":
# HolySheep API 키 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # https://www.holysheep.ai/register 에서 발급
client = HolySheepAIClient(HOLYSHEEP_API_KEY)
# 분석기에서 데이터 가져오기
analyzer = OrderFlowAnalyzer(data_dir="./hyperliquid_data")
trades_df = analyzer.load_trades("BTC-PERP", "20241101", "20241201")
delta_df = analyzer.calculate_delta(bucket_seconds=300) # 5분봉
# 주문 흐름 요약 생성
orderflow_summary = {
"total_trades": len(trades_df),
"date_range": f"{trades_df['datetime'].min()} ~ {trades_df['datetime'].max()}",
"avg_delta_per_5min": float(delta_df['delta'].mean()),
"max_delta": float(delta_df['delta'].max()),
"min_delta": float(delta_df['delta'].min()),
"buy_dominance_ratio": float(delta_df['buy_ratio'].mean()),
"large_trade_count": len(analyzer.detect_large_trades(95))
}
print("[INFO] HolySheep AI로 주문 흐름 분석 시작...")
# AI 분석 요청
analysis_result = client.analyze_orderflow_pattern(
orderflow_summary=orderflow_summary,
model="claude-sonnet-4.5" # Claude Sonnet 4.5: 복잡한 추론에 적합
)
print("\n" + "="*60)
print("HolySheep AI 분석 결과")
print("="*60)
print(analysis_result)
client.close()8. 백테스팅 파이프라인 구축
이제 실제 백테스팅 프레임워크에 주문 흐름 시그널을 통합하는 파이프라인을 구축하겠습니다. 백트레이더(Backtrader)를 사용하여 과거 데이터에서 전략의 수익성을 검증합니다.
import backtrader as bt
import pandas as pd
import numpy as np
from datetime import datetime
from pathlib import Path
class OrderFlowStrategy(bt.Strategy):
"""
주문 흐름 기반 거래 전략
HolySheep AI에서 생성된 시그널을 백테스트에 활용
"""
params = (
('delta_threshold', 0.65), # 매수 신호 델타 임계값
('delta_threshold_short', 0.35), # 매도 신호 델타 임계값
('stop_loss_pct', 0.02), # 손절절 2%
('take_profit_pct', 0.04), # 목표 수익 4%
('position_size', 0.95), # 자본 대비 포지션 크기
('orderflow_data', None), # 주문 흐름 데이터 DataFrame
)
def __init__(self):
self.order = None
self.entry_price = None
# 주문 흐름 데이터 인덱싱
if self.params.orderflow_data is not None:
self.of_data = self.params.orderflow_data.copy()
self.of_data.index = pd.to_datetime(self.of_data.index)
def log(self, txt, dt=None):
"""로깅 함수"""
dt = dt or self.datas[0].datetime.datetime(0)
print(f'[LOG] {dt.isoformat()} - {txt}')
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'매수 체결: 가격 {order.executed.price:.4f}, 수량 {order.executed.size:.4f}')
elif order.issell():
self.log(f'매도 체결: 가격 {order.executed.price:.4f}, 수량 {order.executed.size:.4f}')
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('주문 실패/취소')
self.order = None
def next(self):
"""매 봉(clone)마다 실행되는 로직"""
if self.order:
return
# 현재 시간에 해당하는 주문 흐름 데이터 조회
current_dt = self.datas[0].datetime.datetime(0)
# 주문 흐름 신호 확인
of_signal = self.get_orderflow_signal(current_dt)
if of_signal is None:
return
buy_signal = of_signal.get('buy_signal', False)
sell_signal = of_signal.get('sell_signal', False)
signal_strength = of_signal.get('signal_strength', 0)
# 포지션 없으면 진입 검토
if not self.position:
if buy_signal:
self.execute_long_entry(signal_strength)
elif sell_signal:
self.execute_short_entry(signal_strength)
# 포지션 있으면 관리
else:
self.manage_position()
def get_orderflow_signal(self, current_dt):
"""현재 시간의 주문 흐름 신호 조회"""
if not hasattr(self, 'of_data') or self.of_data is None:
return None
try:
# 가장 가까운 시간의 데이터 찾기
mask = self.of_data.index <= pd.Timestamp(current_dt)
if not mask.any():
return None
current_row = self.of_data[mask].iloc[-1]
return {
'buy_signal': current_row.get('buy_signal', False),
'sell_signal': current_row.get('sell_signal', False),
'signal_strength': current_row.get('signal_strength', 0),
'buy_ratio': current_row.get('buy_ratio', 0.5),
'delta': current_row.get('delta', 0)
}
except Exception as e:
return None
def execute_long_entry(self, signal_strength):
"""롱 포지션 진입"""
size = (self.broker.getcash() * self.params.position_size) / self.data.close[0]
self.order = self.buy(size=size)
self.entry_price = self.data.close[0]
self.log(f'롱 진입: 신호 강도 {signal_strength:.3f}')
def execute_short_entry(self, signal_strength):
"""숏 포지션 진입"""
size = (self.broker.getcash() * self.params.position_size) / self.data.close[0]
self.order = self.sell(size=size)
self.entry_price = self.data.close[0]
self.log(f'숏 진입: 신호 강도 {signal_strength:.3f}')
def manage_position(self):
"""포지션 관리 (손절절/익절)"""
if not self.position:
return
current_price = self.data.close[0]
entry = self.entry_price
if self.position.size > 0: # 롱 포지션
pnl_pct = (current_price - entry) / entry
# 손절절
if pnl_pct <= -self.params.stop_loss_pct:
self.close()
self.log(f'손절절 실행: 손익 {pnl_pct*100:.2f}%')
# 익절
elif pnl_pct >= self.params.take_profit_pct:
self.close()
self.log(f'익절 실행: 손익 {pnl_pct*100:.2f}%')
else: # 숏 포지션
pnl_pct = (entry - current_price) / entry
# 손절절
if pnl_pct <= -self.params.stop_loss_pct:
self.close()
self.log(f'손절절 실행: 손익 {pnl_pct*100:.2f}%')
# 익절
elif pnl_pct >= self.params.take_profit_pct:
self.close()
self.log(f'익절 실행: 손익 {pnl_pct*100:.2f}%')
class OrderFlowData(bt.feeds.PandasData):
"""주문 흐름 데이터를 위한 커스텀 피드"""
params = (
('datetime', None),
('open', 'open'),
('high', 'high'),
('low', 'low'),
('close', 'close'),
('volume', 'volume'),
('openinterest', -1),
)
def run_backtest(trades_df: pd.DataFrame, orderflow_signals: pd.DataFrame,
initial_cash: float = 100000,
commission: float = 0.0004):
"""
백테스트 실행 함수
Args:
trades_df: 체결 데이터
orderflow_signals: 주문 흐름 시그널
initial_cash: 초기 자본
commission: 수수료율 (Hyperliquid 기준 약 0.04%)
"""
cerebro = bt.Cerebro()
# 주문 흐름 시그널 데이터프레임 준비
signal_df = orderflow_signals.copy()
signal_df.index = pd.to_datetime(signal_df.index)
# 가격 데이터 (OHLCV) 준비
# Tardis 데이터에서 OHLCV 구성
trades_copy = trades_df.copy()
trades_copy.set_index('datetime', inplace=True)
# 5분봉으로 리샘플링
price_data = trades_copy.resample('5min').agg({
'price': ['first', 'max', 'min