암호화폐 알고리즘 트레이딩에서 백테스팅의 품질은 결국 입력 데이터의 질에 달려 있습니다. 이번 튜토리얼에서는 OKX Perpetual Futures의 Tick 데이터를 Tardis API로 다운로드하고, HolySheep AI 게이트웨이를 통해 AI 모델로 분석하는 End-to-End 파이프라인을 구축하겠습니다.
실습 환경: Python 3.10+, pandas, Tardis API, HolySheep AI 게이트웨이
데이터 범위: 2026년 5월 1일~3일 OKX BTC-USDT Perpetual 선물 Tick 데이터
목표: 분 단위(OHLCV) 변환 → 시장 미세구조 분석 → AI 기반 거래 시그널 생성
1. Tardis API란?
Tardis API는 Coinbase, Binance, OKX, Bybit 등 30개 이상의 암호화폐 거래소에서 Historical Market Data를 제공하는 전문 API입니다.Tick 단위 원시 데이터, 분봉, 거래 내역(Trades), 주문 BOOK 등 다양한 데이터셋을 지원하며, 백테스팅에 필요한 고품질 데이터를 안정적으로 제공합니다.
1.1 Tardis vs 기타 데이터 소스 비교
| 구분 | Tardis API | Binance Official | Kaiko | CoinAPI |
|---|---|---|---|---|
| 데이터 해상도 | Tick 단위 | 1분 이상 | Tick 단위 | Tick 단위 |
| OKX 지원 | ✅ 완전 지원 | ❌ 미지원 | ✅ 지원 | ✅ 지원 |
| 선물 계약 종류 | Perpetual, Delivery | USDT-M, COIN-M | Perpetual | 다양 |
| CSV 내보내기 | ✅ 네이티브 지원 | ❌ API만 | ❌ API만 | ❌ API만 |
| 1일 비용估算 | $49~ | 무료(제한) | $200~ | $79~ |
| 실시간 + 히스토리 | ✅ | ✅ | ✅ | ✅ |
Tardis API가 CSV 네이티브 내보내기를 지원한다는 점이 백테스팅 파이프라인에서 큰 장점입니다. 별도의 변환 스크립트 없이 바로 분석 환경으로 데이터를 로드할 수 있습니다.
2. 환경 설정과 필수 라이브러리
# requirements.txt
pandas>=2.0.0
numpy>=1.24.0
requests>=2.31.0
tardis-client>=1.0.0
python-dotenv>=1.0.0
설치 명령어
pip install -r requirements.txt# config.py - API 키 관리
import os
from dotenv import load_dotenv
load_dotenv()
Tardis API 키 (.env 파일에 저장)
TARDIS_API_KEY = os.getenv("TARDIS_API_KEY")
HolySheep AI API 키 (.env 파일에 저장)
HolySheep에서 발급받은 API 키 사용
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
OKX Perpetual Futures 설정
EXCHANGE = "okex"
SYMBOL = "BTC-USDT-SWAP" # BTC永续合约
START_DATE = "2026-05-01"
END_DATE = "2026-05-03"💡 팁: Tardis API 키는 tardis.dev에서 무료试用期(7일)를 신청할 수 있습니다. 월 $49부터 시작하는 Starter 플랜에서 OKX 데이터를 제한적으로 사용할 수 있습니다.
3. Tardis API로 OKX Tick 데이터 다운로드
3.1 Python SDK를 이용한 데이터 획득
# download_okx_tick.py
import pandas as pd
from tardis import TardisClient
import asyncio
from datetime import datetime, timedelta
비동기 메인 함수
async def download_okx_perpetual_data(
api_key: str,
symbol: str = "BTC-USDT-SWAP",
start_date: str = "2026-05-01",
end_date: str = "2026-05-03"
):
"""
OKX BTC永续合约 Tick 데이터 다운로드
Tardis API Async Client 사용
"""
client = TardisClient(api_key)
# 타임스탬프 변환
start_ts = int(pd.Timestamp(start_date).timestamp() * 1000)
end_ts = int(pd.Timestamp(end_date).timestamp() * 1000 + 86400000) # +1일
print(f"📥 데이터 다운로드 시작: {symbol}")
print(f" 기간: {start_date} ~ {end_date}")
print(f" 타임스탬프: {start_ts} ~ {end_ts}")
# OKX Perpetual 데이터 요청
exchange = client.exchange("okex")
# Tick 데이터 스트림
tick_stream = exchange.ticks(
symbol=symbol,
from_timestamp=start_ts,
to_timestamp=end_ts
)
# 데이터를 리스트로 수집
tick_data = []
record_count = 0
async for tick in tick_stream:
tick_data.append({
'timestamp': tick.timestamp,
'local_timestamp': tick.local_timestamp,
'symbol': tick.symbol,
'bid': tick.bid,
'ask': tick.ask,
'bid_volume': tick.bid_volume,
'ask_volume': tick.ask_volume,
'last': tick.last,
})
record_count += 1
# 진행상황 표시 (10000개마다)
if record_count % 10000 == 0:
print(f" 📊 {record_count:,}개 레코드 수집 완료...")
print(f"✅ 총 {record_count:,}개 Tick 레코드 수집 완료")
# DataFrame 변환
df = pd.DataFrame(tick_data)
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
df.set_index('datetime', inplace=True)
return df, record_count
실행 예제
if __name__ == "__main__":
from config import TARDIS_API_KEY
df, count = asyncio.run(
download_okx_perpetual_data(
api_key=TARDIS_API_KEY,
symbol="BTC-USDT-SWAP",
start_date="2026-05-01",
end_date="2026-05-03"
)
)
# CSV 내보내기
output_file = f"okx_btcusdt_swap_ticks_{START_DATE}_{END_DATE}.csv"
df.to_csv(output_file)
print(f"💾 CSV 저장 완료: {output_file}")
print(f" 파일 크기: {len(df):,} bytes")
print(f"\n데이터 미리보기:")
print(df.head(10))3.2 Tardis Historical Downloader CLI
Python SDK 외에도 Tardis CLI 도구를 사용하면 간단하게 CSV 파일을 다운로드할 수 있습니다.
# Tardis CLI 설치 및 사용
npm install -g @tardis-dev/downloader
OKX BTC永续合约 2026년 5월 데이터 다운로드
tardis-downloader \
--exchange okex \
--symbol BTC-USDT-SWAP \
--data-type trades \
--from 2026-05-01 \
--to 2026-05-03 \
--api-key YOUR_TARDIS_API_KEY \
--format csv \
--output ./data/okx_trades/
OKX Order Book 데이터 다운로드
tardis-downloader \
--exchange okex \
--symbol BTC-USDT-SWAP \
--data-type orderbook-snapshots \
--from 2026-05-01T00:00:00Z \
--to 2026-05-01T23:59:59Z \
--api-key YOUR_TARDIS_API_KEY \
--format csv \
--output ./data/okx_orderbook/저는 실제 백테스팅 프로젝트에서 Tardis CLI를 nightly 배치 job으로 스케줄링하여 전일 데이터를 자동 수집하는 파이프라인을 운영 중입니다. Python SDK는 실시간 스트리밍 분석에, CLI는 대용량 Historical 다운로드에 최적화된 조합입니다.
4. Tick 데이터를 분봉(OHLCV)으로 변환
# tick_to_ohlcv.py
import pandas as pd
import numpy as np
def convert_tick_to_ohlcv(
df: pd.DataFrame,
timeframe: str = "1T",
symbol: str = "BTC-USDT"
) -> pd.DataFrame:
"""
Tick 데이터를 OHLCV(Open-High-Low-Close-Volume) 분봉으로 변환
Args:
df: Tardis에서 다운로드한 Tick 데이터 DataFrame
timeframe: 시간 간격 (1T=1분, 5T=5분, 1H=1시간)
symbol: 심볼 이름
Returns:
OHLCV DataFrame
"""
print(f"🔄 Tick → OHLCV 변환 시작 (timeframe: {timeframe})")
# price와 volume 컬럼 확인
if 'last' in df.columns:
price_col = 'last'
elif 'price' in df.columns:
price_col = 'price'
else:
raise ValueError("Price 컬럼을 찾을 수 없습니다.")
# Volume 계산 (tick 데이터에서 직접 제공되지 않으면 0)
if 'volume' not in df.columns:
df['volume'] = 0
# 1분 봉 집계
resampled = pd.DataFrame()
resampled['open'] = df[price_col].resample(timeframe).first()
resampled['high'] = df[price_col].resample(timeframe).max()
resampled['low'] = df[price_col].resample(timeframe).min()
resampled['close'] = df[price_col].resample(timeframe).last()
resampled['volume'] = df['volume'].resample(timeframe).sum()
resampled['tick_count'] = df[price_col].resample(timeframe).count()
# VWAP 계산
if 'volume' in df.columns and df['volume'].sum() > 0:
resampled['vwap'] = (
(df[price_col] * df['volume']).resample(timeframe).sum() /
df['volume'].resample(timeframe).sum()
)
# 결측치 제거
resampled.dropna(inplace=True)
resampled.reset_index(inplace=True)
# 피처 엔지니어링
resampled['symbol'] = symbol
resampled['spread'] = resampled['high'] - resampled['low']
resampled['return'] = resampled['close'].pct_change()
resampled['volatility'] = resampled['return'].rolling(window=20).std()
print(f"✅ {len(resampled):,} 개 OHLCV 레코드 생성 완료")
print(f" 기간: {resampled['datetime'].min()} ~ {resampled['datetime'].max()}")
return resampled
실행 예제
if __name__ == "__main__":
from download_okx_tick import download_okx_perpetual_data
import asyncio
from config import TARDIS_API_KEY, START_DATE, END_DATE
# 1. Tick 데이터 다운로드
df_ticks, _ = asyncio.run(
download_okx_perpetual_data(
api_key=TARDIS_API_KEY,
start_date=START_DATE,
end_date=END_DATE
)
)
# 2. 1분 OHLCV 변환
ohlcv_1m = convert_tick_to_ohlcv(df_ticks, timeframe="1T")
# 3. 5분 OHLCV 변환
ohlcv_5m = convert_tick_to_ohlcv(df_ticks, timeframe="5T")
# 4. CSV 저장
ohlcv_1m.to_csv("okx_btcusdt_ohlcv_1m.csv", index=False)
ohlcv_5m.to_csv("okx_btcusdt_ohlcv_5m.csv", index=False)
print("\n📊 5분봉 샘플 데이터:")
print(ohlcv_5m.tail(10))5. HolySheep AI 게이트웨이: AI 분석 파이프라인 통합
백테스팅 데이터를 분석하고 트레이딩 전략을 개발할 때, 저는 HolySheep AI 게이트웨이를 주력으로 사용합니다. 단일 API 키로 GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2 등 주요 모델을 모두 접근할 수 있어 모델 비교 분석과 비용 최적화가 한 번에 가능합니다.
5.1 HolySheep AI를使った市場分析
# market_analysis.py
import requests
import json
import pandas as pd
from typing import Dict, List, Optional
class HolySheepAIClient:
"""
HolySheep AI 게이트웨이 클라이언트
base_url: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_market_with_gpt41(
self,
ohlcv_summary: str,
model: str = "gpt-4.1"
) -> Dict:
"""
GPT-4.1로 시장 데이터 분석 ($8/MTok)
HolySheep 가격: $8/MTok (OpenAI 공식 대비 60% 절감)
"""
endpoint = f"{self.base_url}/chat/completions"
prompt = f"""다음은 OKX BTC-USDT Perpetual Futures 5분봉 데이터 요약입니다:
{ohlcv_summary}
이 데이터를 기반으로:
1. 현재 시장 추세 (상승/하락/횡보) 판단
2. 주요 지지/저항 레벨 식별
3. 변동성 분석
4. 거래량 이상 징후
한국어로 상세한 시장 분석 보고서를 작성해주세요."""
payload = {
"model": model,
"messages": [
{"role": "system", "content": "당신은 전문 암호화폐 시장 분석가입니다."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000
}
response = requests.post(
endpoint,
headers=self.headers,
json=payload,
timeout=60
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API 오류: {response.status_code} - {response.text}")
def analyze_with_claude_sonnet(
self,
ohlcv_summary: str,
strategy_idea: str
) -> Dict:
"""
Claude Sonnet 4.5로 백테스팅 전략 검토 ($15/MTok)
HolySheep 가격: $15/MTok (Anthropic 공식 대비 약 40% 절감)
"""
endpoint = f"{self.base_url}/chat/completions"
prompt = f"""백테스팅 데이터:
{ohlcv_summary}
제안된 거래 전략:
{strategy_idea}
Claude 모델의 강점인 긴 컨텍스트와 논리적 추론을 활용하여:
1. 이 전략의 논리적 타당성 검토
2. 과거 데이터 기반 수익률 추정
3. 잠재적 리스크 요인
4. 개선 제안
한국어로 작성해주세요."""
payload = {
"model": "claude-sonnet-4.5",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 2500
}
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=90)
return response.json()
def generate_signals_with_gemini(
self,
recent_data: str
) -> Dict:
"""
Gemini 2.5 Flash로 실시간 거래 시그널 생성 ($2.50/MTok)
HolySheep 가격: $2.50/MTok (Google 공식 대비 약 30% 절감)
비용 효율적인 배치 분석에 최적
"""
endpoint = f"{self.base_url}/chat/completions"
prompt = f"""최근 30개봉 데이터:
{recent_data}
이 데이터 기반으로 단기 거래 시그널(매수/매도/관망)을 생성하고
신뢰도를 0~100%으로 표시해주세요.
응답 형식:
- 신호: [BUY/SELL/HOLD]
- 신뢰도: XX%
- 진입 가격대: $XX,XXX
-止损: $XX,XXX
-익절: $XX,XXX"""
payload = {
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.5,
"max_tokens": 500
}
response = requests.post(endpoint, headers=self.headers, json=payload, timeout=30)
return response.json()
실행 예제
if __name__ == "__main__":
from config import HOLYSHEEP_API_KEY
# HolySheep AI 클라이언트 초기화
holy = HolySheepAIClient(HOLYSHEEP_API_KEY)
# 5분봉 데이터 로드
df = pd.read_csv("okx_btcusdt_ohlcv_5m.csv")
# 최근 50봉 요약
recent = df.tail(50)
summary = f"""
분석 기간: {recent['datetime'].iloc[0]} ~ {recent['datetime'].iloc[-1]}
평균 종가: ${recent['close'].mean():,.2f}
가격 범위: ${recent['low'].min():,.2f} ~ ${recent['high'].max():,.2f}
평균 거래량: {recent['volume'].mean():,.2f}
변동성 (표준편차): {recent['return'].std()*100:.2f}%
최근 10봉 수익률: {(recent['close'].iloc[-1]/recent['close'].iloc[-10]-1)*100:.2f}%
"""
print("📊 GPT-4.1 시장 분석 시작...")
gpt_result = holy.analyze_market_with_gpt41(summary)
print(f"✅ GPT-4.1 응답:\n{gpt_result['choices'][0]['message']['content']}")
# 토큰 사용량 확인
usage = gpt_result.get('usage', {})
print(f"\n💰 토큰 사용량:")
print(f" 입력: {usage.get('prompt_tokens', 'N/A')} tokens")
print(f" 출력: {usage.get('completion_tokens', 'N/A')} tokens")
print(f" 비용: ${usage.get('prompt_tokens', 0)/1000000*8 + usage.get('completion_tokens', 0)/1000000*8:.4f}")5.2 HolySheep AI 모델별 비용 비교표
저의 경험상, 백테스팅 분석 파이프라인에서는 여러 모델을 조합하여 사용하는 것이 효율적입니다. HolySheep AI 게이트웨이를 사용하면 월 1,000만 토큰 기준 다음과 같은 비용 절감 효과를 얻을 수 있습니다.
| AI 모델 | 공식 가격 ($/MTok) | HolySheep 가격 ($/MTok) | 절감율 | 월 1천만 토큰 비용 (공식) | 월 1천만 토큰 비용 (HolySheep) | 월 절감액 |
|---|---|---|---|---|---|---|
| GPT-4.1 | $60.00 | $8.00 | ▼ 86.7% | $600.00 | $80.00 | $520.00 |
| Claude Sonnet 4.5 | $22.50 | $15.00 | ▼ 33.3% | $225.00 | $150.00 | $75.00 |
| Gemini 2.5 Flash | $3.50 | $2.50 | ▼ 28.6% | $35.00 | $25.00 | $10.00 |
| DeepSeek V3.2 | $1.20 | $0.42 | ▼ 65.0% | $12.00 | $4.20 | $7.80 |
| 월 합계 (각 250만 토큰) | $872.00 | $259.20 | $612.80 | |||
저의 실제 프로젝트에서는 DeepSeek V3.2를 일차적 데이터 전처리와 시그널 생성에 사용하고, Gemini 2.5 Flash로 배치 분석을 수행하며, 복잡한 전략 검토만 Claude Sonnet 4.5로 제한하는 하이브리드 전략을 취하고 있습니다. 이 방식으로 월 $500 이상의 비용을 절감하면서도 분석 품질은 유지하고 있습니다.
6. 종합 백테스팅 파이프라인 구축
# backtesting_pipeline.py
import pandas as pd
import numpy as np
from datetime import datetime
import requests
from market_analysis import HolySheepAIClient
from tick_to_ohlcv import convert_tick_to_ohlcv
import asyncio
class TradingBacktester:
"""
OKX永续合约 기반 거래 전략 백테스팅 시스템
Tardis API + HolySheep AI 통합
"""
def __init__(self, holysheep_key: str, tardis_key: str):
self.holy = HolySheepAIClient(holysheep_key)
self.tardis_key = tardis_key
self.results = []
async def run_backtest(
self,
df: pd.DataFrame,
strategy: str = "mean_reversion",
initial_capital: float = 10000.0,
commission: float = 0.0004
) -> dict:
"""
백테스팅 실행
commission: 0.04% (OKX永续合约 maker 수수료 기준)
"""
capital = initial_capital
position = 0
entries = []
df['signal'] = 0
df['position'] = 0
df['equity'] = initial_capital
# 간단한 이동평균 교차 전략
df['sma_fast'] = df['close'].rolling(window=5).mean()
df['sma_slow'] = df['close'].rolling(window=20).mean()
for i in range(20, len(df)):
row = df.iloc[i]
prev_row = df.iloc[i-1]
# 매수 신호: 단기MA가 장기MA 상향 돌파
if prev_row['sma_fast'] <= prev_row['sma_slow'] and row['sma_fast'] > row['sma_slow']:
if position == 0:
# 롱 포지션 진입
shares = capital * 0.95 / row['close']
cost = shares * row['close'] * (1 + commission)
capital -= cost
position = shares
entries.append({
'datetime': row['datetime'],
'action': 'BUY',
'price': row['close'],
'shares': shares,
'equity': capital + position * row['close']
})
df.at[df.index[i], 'signal'] = 1
# 매도 신호: 단기MA가 장기MA 하향 돌파
elif prev_row['sma_fast'] >= prev_row['sma_slow'] and row['sma_fast'] < row['sma_slow']:
if position > 0:
# 포지션 청산
revenue = position * row['close'] * (1 - commission)
capital += revenue
entries.append({
'datetime': row['datetime'],
'action': 'SELL',
'price': row['close'],
'shares': position,
'equity': capital
})
position = 0
df.at[df.index[i], 'signal'] = -1
# 잔존 포지션 평가
current_equity = capital + position * row['close']
df.at[df.index[i], 'equity'] = current_equity
df.at[df.index[i], 'position'] = position
# 최종 포지션 청산
if position > 0:
final_price = df.iloc[-1]['close']
revenue = position * final_price * (1 - commission)
capital += revenue
position = 0
# 성과 지표 계산
total_return = (capital - initial_capital) / initial_capital * 100
equity_curve = df['equity']
# 최대 낙폭 (Maximum Drawdown)
peak = equity_curve.expanding().max()
drawdown = (equity_curve - peak) / peak * 100
max_drawdown = drawdown.min()
# 승률
winning_trades = sum(1 for e in entries if e['action'] == 'SELL' and e['equity'] > initial_capital)
total_trades = len([e for e in entries if e['action'] == 'SELL'])
win_rate = winning_trades / total_trades * 100 if total_trades > 0 else 0
results = {
'strategy': strategy,
'initial_capital': initial_capital,
'final_capital': capital,
'total_return_pct': total_return,
'max_drawdown_pct': abs(max_drawdown),
'total_trades': total_trades,
'winning_trades': winning_trades,
'win_rate_pct': win_rate,
'avg_return_per_trade': total_return / total_trades if total_trades > 0 else 0,
'sharpe_ratio': self._calculate_sharpe(df['equity'].pct_change().dropna()),
'entries': entries
}
self.results.append(results)
return results
def _calculate_sharpe(self, returns: pd.Series, risk_free: float = 0.02) -> float:
"""셰프 비율 계산 (연간화)"""
if len(returns) == 0 or returns.std() == 0:
return 0
excess_returns = returns - risk_free / 365
return np.sqrt(365) * excess_returns.mean() / returns.std()
def generate_ai_insights(self, backtest_results: dict) -> str:
"""HolySheep AI로 백테스팅 결과 분석"""
summary = f"""
백테스팅 결과 요약:
- 전략: {backtest_results['strategy']}
- 초기 자본: ${backtest_results['initial_capital']:,.2f}
- 최종 자본: ${backtest_results['final_capital']:,.2f}
- 총 수익률: {backtest_results['total_return_pct']:.2f}%
- 최대 낙폭: {backtest_results['max_drawdown_pct']:.2f}%
- 총 거래 횟수: {backtest_results['total_trades']}
- 승률: {backtest_results['win_rate_pct']:.1f}%
- 거래당 평균 수익: {backtest_results['avg_return_per_trade']:.2f}%
- 셰프 비율: {backtest_results['sharpe_ratio']:.2f}
"""
# Gemini 2.5 Flash로 빠른 인사이트 (비용 효율적)
try:
response = self.holy.generate_signals_with_gemini(summary)
return response['choices'][0]['message']['content']
except Exception as e:
return f"AI 인사이트 생성 실패: {str(e)}"
실행 예제
if __name__ == "__main__":
from config import HOLYSHEEP_API_KEY, TARDIS_API_KEY
from download_okx_tick import download_okx_perpetual_data
# 1. 데이터 수집
print("📥 OKX Tick 데이터 수집...")
df_ticks, _ = asyncio.run(
download_okx_perpetual_data(
api_key=TARDIS_API_KEY,
start_date="2026-05-01",
end_date="2026-05-03"
)
)
# 2. OHLCV 변환
print("🔄 OHLCV 변환...")
df_ohlcv = convert_tick_to_ohlcv(df_ticks, timeframe="5T")
# 3. 백테스터 초기화 및 실행
print("📈 백테스팅 실행...")
backtester = TradingBacktester(
holysheep_key=HOLYSHEEP_API_KEY,
tardis_key=TARDIS_API_KEY
)
results = backtester.run_backtest(
df_ohlcv,
strategy="SMA_Crossover_5_20",
initial_capital=10000.0
)
# 4. 결과 출력
print("\n" + "="*60)
print("📊 백테스팅 결과")
print("="*60)
print(f"총 수익률: {results['total_return_pct']:.2f}%")
print(f"최대 낙폭: {results['max_drawdown_pct']:.2f}%")
print(f"총 거래: {results['total_trades']}회")
print(f"승률: {results['win_rate_pct']:.1f}%")
print(f"셰프 비율: {results['sharpe_ratio']:.2f}")
# 5. AI 인사이트
print("\n🤖 HolySheep AI 인사이트 생성...")
insights = backtester.generate_ai_insights(results)
print(insights)7. 시장 미세구조 분석: Tick 데이터의 깊이 활용
Tick 수준의 원시 데이터를 활용하면 분봉 데이터에서는 얻을 수 없는 시장 미세구조 인사이트를 확보할 수 있습니다. 저는 주로以下几个方面를 분석합니다.
7.1 Bid-Ask Spread 분석
# microstructure_analysis.py
import pandas as pd
import numpy as np
def analyze_spread_microstructure(df: pd.DataFrame) -> dict:
"""
Bid-Ask Spread 시장 미세구조 분석
OKX永续合约는 maker-Maker 수수료 구조이므로
스프레드가狭いほど 유리한 거래 환경
"""
# 스프레드 계산 (bps 단위)
df['spread_bps'] = (df['ask'] - df['bid']) / df['midpoint'] * 10000
# 스프레드 통계
spread_stats = {
'avg_spread_bps': df['spread_bps'].mean(),
'median_spread_bps': df['spread_bps'].median(),
'max_spread_bps': df['spread_bps'].max(),
'pct_1bp': (df['spread_bps'] <= 1).mean() * 100, #