저는 최근 암호화폐 시장 데이터 분석 시스템 구축을 맡으면서 Bybit 선물 API의 逐笔成交(체크바이라 execução, 틱 단위 거래) 데이터를 실시간으로 수집하고 주문창(Order Book)을 재구성하는 과정을 깊이 탐구했습니다. 이 튜토리얼에서는 Bybit 공식 API 연동부터 HolySheep AI를 활용한 시장 분석까지, 전체 파이프라인을 단계별로 설명드리겠습니다.
Bybit 선물 API란?
Bybit는 전 세계 최대 선물 거래량 중 하나를 보유한 거래소로, 풍부한 시장 데이터 API를 무료로 제공합니다. 특히 逐笔成交(Tick-by-Tick Execution) 데이터는 밀리초 단위의 개별 거래를 기록하며, 고빈도 거래(HFT), 시장 미세구조 분석, 실시간 주문창 재구성에 필수적인 데이터입니다.
HolySheep vs Bybit 공식 API vs 타 릴레이 서비스 비교
| 비교 항목 | HolySheep AI | Bybit 공식 API | 일반 릴레이 서비스 |
|---|---|---|---|
| 주요 용도 | AI 모델 통합 게이트웨이 | 거래소 데이터/거래 | 단순 중계/프록시 |
| 逐笔成交 데이터 | ❌ 미지원 | ✅ WebSocket/REST 완전 지원 | ⚠️ 제한적 지원 |
| 주문창 재구성 | ❌ 미지원 | ✅ 실시간 WebSocket | ⚠️ 지연 가능 |
| AI 분석 기능 | ✅ GPT-4.1, Claude, Gemini 통합 | ❌ 미지원 | ❌ 미지원 |
| 거래 데이터 AI 분석 | ✅ HolySheep API로 시장 데이터 분석 가능 | ❌ 미지원 | ❌ 미지원 |
| 지연 시간 | 100-200ms (AI 분석) | 10-50ms (시장 데이터) | 200-500ms |
| 비용 | $0 (기본 무료 크레딧 포함) | 무료 (공식 API) | 월 $10-$100 |
| 로컬 결제 | ✅ 해외 신용카드 불필요 | N/A | ❌ 해외 카드 필요 |
이런 팀에 적합 / 비적합
✅ HolySheep AI가 적합한 팀
- 암호화폐 시장 데이터를 AI로 분석하고 싶은 개발자
- Bybit 거래 데이터를 기반으로 예측 모델을 구축하는 데이터 과학자
- 여러 AI 모델(GPT-4.1, Claude, Gemini)을 단일 API로 테스트하고 싶은 팀
- 해외 신용카드 없이 AI API 비용을 최적화したい 스타트업
- DeepSeek 같은 비용 효율적 모델로 거래 신호 분석을 원하는 퀀트 트레이더
❌ HolySheep AI가 적합하지 않은 팀
- Bybit API로 직접 거래(오더 실행)를 자동화하려는 분 — 이는 Bybit 공식 API로 구현해야 합니다
- 100ms 이하의 초저지연이 요구되는 고빈도 거래 시스템 — 전용 인프라가 필요합니다
- 순수히 Bybit 시장 데이터만 필요한 경우 — Bybit 공식 API가 가장 적합합니다
가격과 ROI
| 서비스 | 가격 | 적합한 사용량 | 월 비용估算 |
|---|---|---|---|
| HolySheep AI | DeepSeek V3.2: $0.42/MTok | 월 100만 토큰 | $420 (시장 데이터 AI 분석) |
| OpenAI 공식 | GPT-4o: $15/MTok | 월 100만 토큰 | $15,000 |
| Anthropic 공식 | Claude 3.5 Sonnet: $15/MTok | 월 100만 토큰 | $15,000 |
| Bybit 공식 API | 무료 | 무제한 시장 데이터 | $0 |
ROI 분석: HolySheep AI를 통해 Bybit 시장 데이터를 AI 분석하면, DeepSeek V3.2 사용 시 OpenAI 대비 97% 비용 절감이 가능합니다. 월 100만 토큰 사용 시 $420 vs $15,000의 차이가 발생합니다.
왜 HolySheep를 선택해야 하나
Bybit API로 수집한 逐笔成交 데이터와 주문창 데이터를 분석할 때, HolySheep AI는 다음과 같은 이점을 제공합니다:
- 단일 API 키로 다중 모델 통합: GPT-4.1, Claude 3.5 Sonnet, Gemini 2.5 Flash, DeepSeek V3.2를 하나의 API 키로切り替え 가능
- 비용 최적화: DeepSeek V3.2는 $0.42/MTok으로 시장 데이터 분석에 최적의 비용 효율성 제공
- 로컬 결제 지원: 해외 신용카드 없이 KRW, CNY 등 현지 결제 수단으로 구매 가능
- 무료 크레딧: 지금 가입하면 즉시 사용 가능한 무료 크레딧 제공
Bybit 선물 API 연동 준비
1. Bybit API 키 발급
Bybit Developer Portal(https://bybit-exchange.github.io/docs/)에서 API 키를 발급받습니다. 선물 거래를 위해서는 Futures 전용 API 키가 필요하며, 테스트넷을 먼저 사용하는 것을 권장합니다.
2. 필요한 Python 라이브러리 설치
# Bybit API 연동에 필요한 라이브러리 설치
pip install websocket-client requests asyncio aiohttp
#HolySheep AI API 연동 (시장 데이터 AI 분석용)
pip install openai
데이터 처리를 위한 추가 라이브러리
pip install pandas numpy逐笔成交 데이터 수집 구현
WebSocket을 통한 실시간 데이터 수신
import websocket
import json
import time
from datetime import datetime
class BybitTickCollector:
"""Bybit 선물，逐笔成交(Tick-by-Tick) 데이터 수집기"""
def __init__(self, symbol="BTCUSDT"):
self.symbol = symbol.upper()
self.ws_url = "wss://stream.bybit.com/v5/public/linear"
self.ticks = []
self.max_ticks = 10000 # 메모리 관리: 최대 10000개 tick 저장
def on_message(self, ws, message):
"""WebSocket 메시지 수신 처리"""
data = json.loads(message)
#逐笔成交(topic: execution) 메시지 파싱
if data.get("topic", "").startswith("execution."):
for tick in data.get("data", []):
parsed_tick = {
"timestamp": tick.get("T"), # 실행 시간
"symbol": tick.get("s"), # 심볼
"side": tick.get("S"), # Buy/Sell
"price": float(tick.get("p", 0)), # 실행 가격
"size": float(tick.get("v", 0)), # 실행 수량
"order_id": tick.get("o"), # 주문 ID
"is_maker": tick.get("m", False) # 메이커 여부
}
self.ticks.append(parsed_tick)
# 메모리 관리: 최대 저장 개수 초과 시 오래된 데이터 제거
if len(self.ticks) > self.max_ticks:
self.ticks = self.ticks[-self.max_ticks:]
# 실시간 출력 (디버깅용)
print(f"[{datetime.fromtimestamp(parsed_tick['timestamp']/1000)}] "
f"{parsed_tick['symbol']} {parsed_tick['side']} "
f"@{parsed_tick['price']} x {parsed_tick['size']}")
def on_error(self, ws, error):
print(f"WebSocket Error: {error}")
def on_close(self, ws, close_status_code, close_msg):
print(f"Connection closed: {close_status_code} - {close_msg}")
def on_open(self, ws):
"""WebSocket 연결 시 구독 요청"""
subscribe_msg = {
"op": "subscribe",
"args": [f"execution.{self.symbol}"] #逐笔成交 토픽 구독
}
ws.send(json.dumps(subscribe_msg))
print(f"Subscribed to execution.{self.symbol}")
def start(self):
"""WebSocket 연결 시작"""
ws = websocket.WebSocketApp(
self.ws_url,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
ws.run_forever(ping_interval=30)
사용 예제
if __name__ == "__main__":
collector = BybitTickCollector(symbol="BTCUSDT")
print("Starting Bybit Tick Data Collection...")
print("Press Ctrl+C to stop")
collector.start()주문창 재구성 구현
import heapq
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
import threading
@dataclass(order=True)
class Level:
"""주문창 레벨 (힙 정렬용)"""
price: float
size: float = field(compare=False)
timestamp: float = field(compare=False, default=0)
def __repr__(self):
return f"@{self.price} x {self.size}"
class OrderBookRebuilder:
"""
Bybit 주문창 재구성기
逐笔成交 데이터를 기반으로 주문창 상태를 실시간 업데이트
"""
def __init__(self, symbol: str = "BTCUSDT", depth: int = 20):
self.symbol = symbol
self.depth = depth
# 주문창 상태 (price -> size 매핑)
self.bids: Dict[float, float] = {} # 매수 주문 (가격 오름차순 정렬)
self.asks: Dict[float, float] = {} # 매도 주문 (가격 오름차순 정렬)
# 힙 구조 (빠른 최솟값/최댓값 접근용)
self.bid_heap: List[Level] = [] # 음수 가격으로 최대 힙 구현
self.ask_heap: List[Level] = [] # 최소 ힱ 구현
# 스레드 안전성
self.lock = threading.Lock()
# 통계
self.total_trades = 0
self.total_volume = 0.0
def process_execution(self, tick: dict):
"""逐笔成交 데이터로 주문창 업데이트"""
with self.lock:
price = tick['price']
size = tick['size']
side = tick['side']
is_maker = tick.get('is_maker', False)
# 시장가成交 또는 메이커 주문 처리
if is_maker:
# 메이커 주문: 주문창에 추가
if side == "Buy":
self.bids[price] = self.bids.get(price, 0) + size
heapq.heappush(self.bid_heap, Level(-price, size))
else:
self.asks[price] = self.asks.get(price, 0) + size
heapq.heappush(self.ask_heap, Level(price, size))
else:
# 테이커 주문: 주문창에서 차감
if side == "Buy" and price in self.asks:
# 매수 테이커 -> 매도 주문 충족
self.asks[price] = max(0, self.asks[price] - size)
if self.asks[price] <= 0:
del self.asks[price]
elif side == "Sell" and price in self.bids:
# 매도 테이커 -> 매수 주문 충족
self.bids[price] = max(0, self.bids[price] - size)
if self.bids[price] <= 0:
del self.bids[price]
# 통계 업데이트
self.total_trades += 1
self.total_volume += size
def get_snapshot(self) -> Dict:
"""현재 주문창 스냅샷 반환"""
with self.lock:
# 상위 depth개 가격 레벨 가져오기
bid_prices = sorted(self.bids.keys(), reverse=True)[:self.depth]
ask_prices = sorted(self.asks.keys())[:self.depth]
bid_levels = [
{"price": p, "size": self.bids[p], "total": sum(self.bids[p] for p in bid_prices[:i+1])}
for i, p in enumerate(bid_prices)
]
ask_levels = [
{"price": p, "size": self.asks[p], "total": sum(self.asks[p] for p in ask_prices[:i+1])}
for i, p in enumerate(ask_prices)
]
best_bid = bid_prices[0] if bid_prices else 0
best_ask = ask_prices[0] if ask_prices else 0
spread = best_ask - best_bid if best_bid and best_ask else 0
spread_pct = (spread / best_bid * 100) if best_bid else 0
return {
"symbol": self.symbol,
"timestamp": time.time(),
"bids": bid_levels,
"asks": ask_levels,
"best_bid": best_bid,
"best_ask": best_ask,
"spread": spread,
"spread_pct": round(spread_pct, 4),
"total_trades": self.total_trades,
"total_volume": self.total_volume
}
def __repr__(self):
snapshot = self.get_snapshot()
lines = [f"OrderBook({self.symbol})", f"Spread: {snapshot['spread']:.2f} ({snapshot['spread_pct']}%)"]
lines.append("Bids:" + " | ".join([f"{b['price']}:{b['size']}" for b in snapshot['bids'][:5]]))
lines.append("Asks:" + " | ".join([f"{a['price']}:{a['size']}" for a in snapshot['asks'][:5]]))
return "\n".join(lines)
주문창 재구성기와 Bybit 데이터 수집기 통합
def integrated_tick_processor():
"""逐笔成交 + 주문창 재구성 통합 프로세스"""
order_book = OrderBookRebuilder(symbol="BTCUSDT", depth=20)
class IntegratedCollector(BybitTickCollector):
def __init__(self, book_rebuilder):
super().__init__("BTCUSDT")
self.order_book = book_rebuilder
def on_message(self, ws, message):
data = json.loads(message)
if data.get("topic", "").startswith("execution."):
for tick in data.get("data", []):
parsed_tick = {
"timestamp": tick.get("T"),
"symbol": tick.get("s"),
"side": tick.get("S"),
"price": float(tick.get("p", 0)),
"size": float(tick.get("v", 0)),
"is_maker": tick.get("m", False)
}
# 주문창 업데이트
self.order_book.process_execution(parsed_tick)
return IntegratedCollector(order_book)
if __name__ == "__main__":
processor = integrated_tick_processor()
print("Starting Integrated Tick Processor with Order Book...")
processor.start()HolySheep AI로 시장 데이터 AI 분석
Bybit에서 수집한 逐笔成交 데이터와 주문창 스냅샷을 HolySheep AI로 전송하여 시장 상황을 분석하고 거래 신호를 생성하는 파이프라인을 구축해 보겠습니다.
HolySheep AI API 연동
import os
from openai import OpenAI
HolySheep AI API 키 설정
https://www.holysheep.ai/register에서 무료 크레딧과 함께 가입하세요
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
HolySheep AI API 클라이언트 초기화
client = OpenAI(
api_key=os.environ["HOLYSHEEP_API_KEY"],
base_url="https://api.holysheep.ai/v1" # HolySheep 게이트웨이 엔드포인트
)
def analyze_market_with_ai(order_book_snapshot: dict, recent_ticks: list) -> str:
"""
Bybit 주문창과逐笔成交 데이터를 HolySheep AI로 분석
Args:
order_book_snapshot: OrderBookRebuilder.get_snapshot() 결과
recent_ticks: 최근 50개 tick 데이터
Returns:
AI가 분석한 시장 상황 설명
"""
# 분석용 프롬프트 구성
analysis_prompt = f"""당신은 전문 암호화폐 트레이더입니다.
다음 Bybit 선물 시장 데이터를 분석하고 거래 신호를 제공해주세요.
현재 주문창 상태
- 심볼: {order_book_snapshot['symbol']}
- 최우선 매수호가: ${order_book_snapshot['best_bid']:,.2f}
- 최우선 매도호가: ${order_book_snapshot['best_ask']:,.2f}
- 스프레드: ${order_book_snapshot['spread']:.2f} ({order_book_snapshot['spread_pct']:.4f}%)
상위 매수 주문 (Bids)
{chr(10).join([f" - ${b['price']:,.2f}: {b['size']:.4f} BTC (누적: {b['total']:.4f})" for b in order_book_snapshot['bids'][:5]])}
상위 매도 주문 (Asks)
{chr(10).join([f" - ${a['price']:,.2f}: {a['size']:.4f} BTC (누적: {a['total']:.4f})" for a in order_book_snapshot['asks'][:5]])}
최근 거래 동향 (최근 10건)
{chr(10).join([f" - {'매수' if t['side']=='Buy' else '매도'} @ ${t['price']:,.2f} x {t['size']:.4f} (메이커: {'예' if t.get('is_maker') else '아니오'})" for t in recent_ticks[-10:]])}
분석 요청
1. 현재 시장 분위기 판단 (매수세 우세 / 매도세 우세 / 중립)
2. 주요 지지선 및 저항선 추정
3. 단기 거래 신호 (강력Buy / Buy / 중립 / Sell / 강력Sell)
4. 주요 위험 요소
5. 투자자 참고 사항
한국어로 명확하게 답변해주세요."""
# HolySheep AI API 호출 (DeepSeek V3.2 사용 - $0.42/MTok으로 비용 최적화)
response = client.chat.completions.create(
model="deepseek-chat", # HolySheep에서 DeepSeek V3.2 매핑
messages=[
{
"role": "system",
"content": "당신은 세계적 수준의 암호화폐 시장 분석 전문가입니다."
},
{
"role": "user",
"content": analysis_prompt
}
],
temperature=0.3, # 일관된 분석을 위해 낮은 온도
max_tokens=1000
)
return response.choices[0].message.content
HolySheep AI 모델 비교 테스트
def test_holy_sheep_models():
"""HolySheep AI에서 사용 가능한 주요 모델들의 응답 시간과 비용 비교"""
test_prompt = "BTC 현재 시장 상황을 3문장으로 요약해주세요."
models_to_test = [
("deepseek-chat", "DeepSeek V3.2", 0.42),
("gpt-4o-mini", "GPT-4o Mini", 0.75),
("gpt-4o", "GPT-4.1", 8.00),
]
results = []
for model_id, model_name, price_per_mtok in models_to_test:
try:
import time
start_time = time.time()
response = client.chat.completions.create(
model=model_id,
messages=[{"role": "user", "content": test_prompt}],
max_tokens=100
)
elapsed_ms = (time.time() - start_time) * 1000
output_tokens = response.usage.completion_tokens
cost = (output_tokens / 1_000_000) * price_per_mtok
results.append({
"model": model_name,
"latency_ms": round(elapsed_ms, 2),
"output_tokens": output_tokens,
"cost_usd": round(cost, 6),
"response": response.choices[0].message.content[:100] + "..."
})
print(f"✅ {model_name}: {elapsed_ms:.0f}ms | {output_tokens} tokens | ${cost:.6f}")
except Exception as e:
print(f"❌ {model_name}: 오류 - {str(e)}")
return results
사용 예제
if __name__ == "__main__":
print("Testing HolySheep AI Models...")
print("=" * 60)
test_holy_sheep_models()
print("\n" + "=" * 60)
print("AI Market Analysis Example")
print("=" * 60)
# 샘플 주문창 데이터
sample_order_book = {
"symbol": "BTCUSDT",
"best_bid": 67450.50,
"best_ask": 67452.00,
"spread": 1.50,
"spread_pct": 0.0022,
"bids": [
{"price": 67450.50, "size": 2.543, "total": 2.543},
{"price": 67448.00, "size": 1.892, "total": 4.435},
{"price": 67445.50, "size": 3.215, "total": 7.650},
],
"asks": [
{"price": 67452.00, "size": 1.234, "total": 1.234},
{"price": 67455.00, "size": 2.876, "total": 4.110},
{"price": 67458.50, "size": 1.543, "total": 5.653},
]
}
sample_ticks = [
{"side": "Buy", "price": 67451.00, "size": 0.543, "is_maker": False},
{"side": "Sell", "price": 67451.50, "size": 0.321, "is_maker": True},
{"side": "Buy", "price": 67450.00, "size": 1.200, "is_maker": False},
]
analysis = analyze_market_with_ai(sample_order_book, sample_ticks)
print(analysis)실시간 분석 파이프라인 통합
import time
import threading
from datetime import datetime
class RealTimeMarketAnalyzer:
"""
Bybit逐笔成交 + 주문창 + HolySheep AI 실시간 분석 파이프라인
"""
def __init__(self, symbol: str = "BTCUSDT", holy_sheep_api_key: str):
self.symbol = symbol
self.order_book = OrderBookRebuilder(symbol=symbol, depth=20)
self.tick_collector = BybitTickCollector(symbol=symbol)
self.ai_client = OpenAI(
api_key=holy_sheep_api_key,
base_url="https://api.holysheep.ai/v1"
)
# 분석 주기 설정
self.analysis_interval = 60 # 60초마다 AI 분석
self.last_analysis_time = 0
self.analysis_count = 0
self.running = False
def start(self):
"""분석 파이프라인 시작"""
self.running = True
self.tick_collector.start()
def process_tick(self, tick: dict):
"""단일 tick 처리: 주문창 업데이트 + 주기적 AI 분석"""
self.order_book.process_execution(tick)
# 주기적 AI 분석 실행
current_time = time.time()
if current_time - self.last_analysis_time >= self.analysis_interval:
self.run_analysis()
self.last_analysis_time = current_time
def run_analysis(self):
"""HolySheep AI로 시장 분석 실행"""
try:
snapshot = self.order_book.get_snapshot()
recent_ticks = self.tick_collector.ticks[-50:] # 최근 50개 tick
# HolySheep AI API 호출
response = self.ai_client.chat.completions.create(
model="deepseek-chat",
messages=[{
"role": "user",
"content": self._build_analysis_prompt(snapshot, recent_ticks)
}],
max_tokens=500
)
self.analysis_count += 1
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"\n{'='*60}")
print(f"[{timestamp}] 분석 #{self.analysis_count}")
print(f"{'='*60}")
print(f"스프레드: ${snapshot['spread']:.2f} ({snapshot['spread_pct']:.4f}%)")
print(f"누적 거래: {snapshot['total_trades']}건, {snapshot['total_volume']:.2f} BTC")
print("-" * 60)
print("AI 분석 결과:")
print(response.choices[0].message.content)
print("=" * 60 + "\n")
# ⚠️ 실제 거래 신호는 반드시 Bybit 공식 API로 구현
# HolySheep AI는 분석 용도로만 사용
except Exception as e:
print(f"AI 분석 오류: {e}")
def _build_analysis_prompt(self, snapshot: dict, ticks: list) -> str:
"""AI 분석용 프롬프트 구성"""
return f"""Bybit {snapshot['symbol']} 시장 분석 요청:
스프레드: ${snapshot['spread']:.2f} ({snapshot['spread_pct']:.4f}%)
매수호가: ${snapshot['best_bid']:,.2f} / 매도호가: ${snapshot['best_ask']:,.2f}
총 거래량: {snapshot['total_volume']:.4f} BTC ({snapshot['total_trades']}건)
최근 거래 동향 10건:
{chr(10).join([
f"{'매수' if t['side']=='Buy' else '매도'} @ ${t['price']:,.2f} x {t['size']:.4f}"
for t in ticks[-10:]
])}
한국어로 3문장 내외로 시장 상황을 요약해주세요."""
메인 실행
if __name__ == "__main__":
import os
HOLY_SHEEP_KEY = os.environ.get("HOLY_SHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
print("=" * 60)
print("Bybit + HolySheep AI 실시간 시장 분석 시스템")
print("=" * 60)
print("Bybit WebSocket: wss://stream.bybit.com/v5/public/linear")
print("AI 분석 주기: 60초마다")
print("사용 모델: DeepSeek V3.2 ($0.42/MTok)")
print("=" * 60)
analyzer = RealTimeMarketAnalyzer(
symbol="BTCUSDT",
holy_sheep_api_key=HOLY_SHEEP_KEY
)
print("\nStarting analyzer... (Press Ctrl+C to stop)")
analyzer.start()Bybit REST API를 통한 히스토리컬 데이터 수집
WebSocket 실시간 데이터와 함께 REST API를 사용하여 과거 逐笔成交 데이터를 수집하는 방법도 알아두면 유용합니다.
import requests
from datetime import datetime, timedelta
class BybitHistoricalFetcher:
"""Bybit REST API를 통한 히스토리컬逐笔成交 데이터 수집"""
BASE_URL = "https://api.bybit.com"
def __init__(self, testnet: bool = False):
if testnet:
self.BASE_URL = "https://api-testnet.bybit.com"
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"User-Agent": "BybitMarketAnalyzer/1.0"
})
def get_recent_executions(self, symbol: str = "BTCUSDT", limit: int = 100):
"""
최근逐笔成交(체결) 데이터 조회
API Docs: https://bybit-exchange.github.io/docs/v5/order/execution
"""
endpoint = "/v5/order/recent-execution"
params = {
"category": "linear", # 선물: linear, 옵션: option
"symbol": symbol,
"limit": min(limit, 1000) # 최대 1000개
}
try:
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("retCode") == 0:
return data.get("result", {}).get("list", [])
else:
print(f"API 오류: {data.get('retMsg')}")
return []
except requests.exceptions.RequestException as e:
print(f"요청 실패: {e}")
return []
def get_public_trading_history(self, symbol: str = "BTCUSDT",
start_time: int = None,
limit: int = 200):
"""
공용 거래 내역 조회 (인증 불필요)
API Docs: https://bybit-exchange.github.io/docs/v5/market/recent-tick
"""
endpoint = "/v5/market/recent-tick"
params = {
"category": "linear",
"symbol": symbol,
"limit": min(limit, 1000)
}
if start_time:
# milliseconds 타임스탬프
params["startTime"] = start_time
try:
response = self.session.get(
f"{self.BASE_URL}{endpoint}",
params=params,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("retCode") == 0:
ticks = data.get("result", {}).get("list", [])
# 데이터 정제
return [
{
"timestamp": int(t.get("T", 0)),
"datetime": datetime.fromtimestamp(int(t.get("T", 0))/1000).isoformat(),
"symbol": t.get("s"),
"price": float(t.get("p", 0)),
"size": float(t.get("v",