저는 3년 넘게 암호화폐 자동매매 시스템을 개발하며 Bybit API를 하루에 수백만 건씩 호출해온 경험이 있습니다. 이 튜토리얼에서는 Bybit의 과거 체결(Trade) 데이터를 Python으로 안정적으로 가져오는 방법을 초보자도 이해할 수 있도록 단계별로 설명드리겠습니다.
Bybit API란 무엇인가
Bybit는 세계 3위권 암호화폐 선물거래소로, 투자자와 트레이더에게 풍부한 시장 데이터를 API 형태로 제공합니다. Bybit Public API를 이용하면 본인 거래 권한 없이도 시세, 호가창, 체결 내역 등 공개 데이터에 접근할 수 있습니다.
Bybit Public API가 제공하는 주요 데이터:
- 실시간 시세(Ticker)
- 호가창(Orderbook)
- 과거 체결 내역(Trade History)
- K线蜡烛图数据
- 거래소 공지
시작하기 전 준비물
아래 환경만 준비되면 됩니다. 별도 가입이나 비용이 들지 않습니다.
- Python 3.8 이상 — 아직 설치 안 했다면 python.org에서 다운로드
- pip — Python 패키지 설치 도구(보통 Python 설치 시 함께 포함)
- 인터넷 연결 — API 호출 시 필요
- IDE 또는 메모장 — 코딩을 작성할 편집기
1단계: 개발 환경 설정
먼저 프로젝트 폴더를 만들고 필요한 라이브러리를 설치하겠습니다.
# 터미널(명령 프롬프트)에서 실행하세요
프로젝트 폴더 생성
mkdir bybit_trade_tutorial
cd bybit_trade_tutorial
Python 가상환경 생성(권장)
python -m venv venv
Windows의 경우:
venv\Scripts\activate
macOS/Linux의 경우:
source venv/bin/activate
필수 라이브러리 설치
pip install requests pandas
💡 팁: 가상환경을 사용하면 프로젝트마다 라이브러리 버전 충돌을 방지할 수 있어 매우 중요합니다.
2단계: Bybit 과거 체결 데이터 가져오기
Bybit Public Trade API를 호출하여 과거 체결 데이터를 가져오는 기본 코드를 작성해보겠습니다.
import requests
import time
import json
def get_bybit_trade_history(symbol="BTCUSDT", limit=10):
"""
Bybit 과거 체결 데이터 가져오기
매개변수:
symbol: 거래 쌍 (기본값: BTCUSDT)
limit: 가져올 데이터 수 (최대 1000개)
반환:
체결 데이터 리스트
"""
# Bybit Public API 엔드포인트
url = "https://api.bybit.com/v5/market/recent-trade"
# 요청 파라미터
params = {
"category": "spot", # spot(현물), linear(선물), inverse(역계약)
"symbol": symbol, # 거래 쌍
"limit": limit # 데이터 개수 (1~1000)
}
try:
# API 호출
response = requests.get(url, params=params)
response.raise_for_status() # HTTP 에러 시 예외 발생
# JSON 데이터 파싱
data = response.json()
# 성공 여부 확인
if data["retCode"] == 0:
return data["result"]["list"]
else:
print(f"API 오류: {data['retMsg']}")
return None
except requests.exceptions.RequestException as e:
print(f"네트워크 오류: {e}")
return None
실제로 테스트해보기
if __name__ == "__main__":
print("=== Bybit BTCUSDT 최근 체결 데이터 ===\n")
trades = get_bybit_trade_history(symbol="BTCUSDT", limit=5)
if trades:
for i, trade in enumerate(trades, 1):
print(f"{i}. 체결 시간: {trade['tradeTime']}")
print(f" 가격: {trade['price']} USDT")
print(f" 수량: {trade['size']}")
print(f" 매도/매수: {'매도' if trade['side'] == 'Sell' else '매수'}")
print("-" * 40)
위 코드를 bybit_basic.py로 저장하고 실행하면 다음과 같은 결과가 나옵니다.
python bybit_basic.py
=== Bybit BTCUSDT 최근 체결 데이터 ===
1. 체결 시간: 1706691000000
가격: 42850.25 USDT
수량: 0.1532
매도/매수: 매수
----------------------------------------
2. 체결 시간: 1706690998000
가격: 42850.10 USDT
수량: 0.0521
매도/매수: 매도
----------------------------------------
(생략)
📋 반환되는 주요 필드 설명:
| 필드명 | 설명 | 예시값 |
|---|---|---|
| tradeTime | 체결 시각(밀리초 타임스탬프) | 1706691000000 |
| price | 체결 가격 | 42850.25 |
| size | 체결 수량 | 0.1532 |
| side | 매도(Sell) 또는 매수(Buy) | Buy |
| tradeId | 고유 체결 ID | 1234567890 |
| isBuyerMaker | 매수자 창작가 여부 | true/false |
3단계: 과거 특정 기간 데이터 가져오기
최근 체결이 아니라 특정 기간의 데이터를 가져오고 싶을 때가 있습니다. Bybit의 Unified Account API를 사용하면 됩니다.
import requests
import pandas as pd
from datetime import datetime, timedelta
def get_historical_trades(symbol="BTCUSDT", start_time=None, limit=100):
"""
Bybit 특정 기간 과거 체결 데이터 가져오기
매개변수:
symbol: 거래 쌍
start_time: 시작 시간 (밀리초 타임스탬프 또는 datetime)
limit: 한 번에 가져올 데이터 수 (최대 1000)
반환:
pandas DataFrame
"""
url = "https://api.bybit.com/v5/market/history-trade"
# start_time이 datetime 객체면 밀리초로 변환
if isinstance(start_time, datetime):
start_time = int(start_time.timestamp() * 1000)
# 기본값: 1시간 전
if start_time is None:
start_time = int((datetime.now() - timedelta(hours=1)).timestamp() * 1000)
params = {
"category": "spot",
"symbol": symbol,
"limit": min(limit, 1000), # 최대 1000개
"startTime": start_time
}
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data["retCode"] == 0:
trades = data["result"]["list"]
# DataFrame으로 변환
df = pd.DataFrame(trades)
# 타임스탬프를 읽기 쉬운 형태로 변환
df['tradeTime_dt'] = pd.to_datetime(df['tradeTime'].astype(int), unit='ms')
# 숫자형으로 변환
df['price'] = df['price'].astype(float)
df['size'] = df['size'].astype(float)
# 시간순 정렬 (오래된 것부터)
df = df.sort_values('tradeTime').reset_index(drop=True)
return df
else:
print(f"오류: {data['retMsg']}")
return None
except Exception as e:
print(f"예상치 못한 오류: {e}")
return None
사용 예시
if __name__ == "__main__":
print("=== 1시간 전부터 현재까지 BTCUSDT 체결 데이터 ===\n")
# 1시간 전부터 데이터 가져오기
df = get_historical_trades(symbol="BTCUSDT", limit=100)
if df is not None and len(df) > 0:
print(f"총 {len(df)}건의 체결 데이터\n")
print(df[['tradeTime_dt', 'price', 'size', 'side']].tail(10))
# 간단한 통계
print(f"\n--- 체결 통계 ---")
print(f"평균 가격: {df['price'].mean():.2f} USDT")
print(f"최고가: {df['price'].max():.2f} USDT")
print(f"최저가: {df['price'].min():.2f} USDT")
print(f"총 거래량: {df['size'].sum():.4f} BTC")
4단계: 실제 분석 프로젝트에 적용하기
저는 이 코드를 바탕으로 실제 트레이딩 봇의 체결 데이터를 저장하고 분석하는 시스템을 구축했습니다. 다음은 매일 자동으로 데이터를 수집하여 CSV 파일로 저장하는 예제입니다.
import requests
import csv
import os
from datetime import datetime, timedelta
import time
class BybitDataCollector:
"""Bybit 체결 데이터 수집기"""
def __init__(self, symbols=["BTCUSDT", "ETHUSDT"]):
self.base_url = "https://api.bybit.com/v5/market/history-trade"
self.symbols = symbols
self.rate_limit_delay = 0.2 # Rate Limit 방지 딜레이
def collect_trades(self, symbol, start_date, end_date):
"""
특정 기간의 모든 체결 데이터 수집
매개변수:
symbol: 거래 쌍
start_date: 시작 날짜 (datetime)
end_date: 종료 날짜 (datetime)
"""
all_trades = []
current_time = int(start_date.timestamp() * 1000)
end_time = int(end_date.timestamp() * 1000)
print(f"\n{symbol} 데이터 수집 시작: {start_date} ~ {end_date}")
while current_time < end_time:
params = {
"category": "spot",
"symbol": symbol,
"limit": 1000,
"startTime": current_time
}
try:
response = requests.get(self.base_url, params=params, timeout=10)
data = response.json()
if data["retCode"] != 0:
print(f"API 오류: {data['retMsg']}")
break
trades = data["result"]["list"]
if not trades:
print("더 이상 데이터가 없습니다.")
break
all_trades.extend(trades)
# 다음 요청을 위한 시간 업데이트
current_time = int(trades[-1]["tradeTime"]) + 1
# Rate Limit 방지
time.sleep(self.rate_limit_delay)
# 진행 상황 출력
progress = (current_time - int(start_date.timestamp() * 1000)) / \
(end_time - int(start_date.timestamp() * 1000)) * 100
print(f" 진행률: {min(progress, 100):.1f}% ({len(all_trades)}건 수집)")
except Exception as e:
print(f"오류 발생: {e}, 5초 후 재시도...")
time.sleep(5)
print(f"총 {len(all_trades)}건 수집 완료!")
return all_trades
def save_to_csv(self, trades, filename):
"""수집된 데이터를 CSV 파일로 저장"""
if not trades:
print("저장할 데이터가 없습니다.")
return
# 저장 폴더 확인
os.makedirs("data", exist_ok=True)
filepath = f"data/{filename}"
# CSV 작성
fieldnames = ["tradeId", "tradeTime", "price", "size", "side", "tickDirection"]
with open(filepath, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(trades)
print(f"✅ 데이터 저장 완료: {filepath}")
return filepath
사용 예시
if __name__ == "__main__":
collector = BybitDataCollector(symbols=["BTCUSDT"])
# 3일치 데이터 수집
end_date = datetime.now()
start_date = end_date - timedelta(days=3)
trades = collector.collect_trades("BTCUSDT", start_date, end_date)
if trades:
# CSV로 저장
date_str = datetime.now().strftime("%Y%m%d")
collector.save_to_csv(trades, f"btcusdt_trades_{date_str}.csv")
5단계: 선물(Futures) 데이터 가져오기
선물 거래소의 데이터를 가져올 때는 category 파라미터를 "linear"(USDT 선물) 또는 "inverse"(역계약 선물)로 변경하면 됩니다.
import requests
def get_futures_trades(symbol="BTCUSDT", contract_type="linear"):
"""
Bybit 선물 체결 데이터 가져오기
contract_type:
- "linear": USDT 선물 (가장 일반적)
- "inverse": 역계약 선물
"""
url = "https://api.bybit.com/v5/market/recent-trade"
params = {
"category": contract_type, # linear 또는 inverse
"symbol": symbol,
"limit": 10
}
response = requests.get(url, params=params)
data = response.json()
if data["retCode"] == 0:
return data["result"]["list"]
else:
print(f"오류: {data['retMsg']}")
return None
USDT 선물 데이터
print("=== BTC USDT 선물 ===")
futures_trades = get_futures_trades("BTCUSDT", "linear")
if futures_trades:
for trade in futures_trades[:3]:
print(f"가격: {trade['price']}, 수량: {trade['size']}")
역계약 선물 데이터
print("\n=== BTC 역계약 선물 ===")
inverse_trades = get_futures_trades("BTCUSD", "inverse")
if inverse_trades:
for trade in inverse_trades[:3]:
print(f"가격: {trade['price']}, 수량: {trade['size']}")
자주 발생하는 오류와 해결책
오류 1: "Invalid symbol" 에러
# ❌ 잘못된 예시
symbol="BTC/USDT" # 슬래시 사용 ❌
symbol="btcusdt" # 소문자 사용 ❌
✅ 올바른 예시
symbol="BTCUSDT" # 정확한 대문자 표기 ✅
symbol="ETHUSDT" # ETH도 동일하게 ✅
symbol="SOLUSDT" # 솔라나도 가능 ✅
원인: Bybit API는 정확한 심볼 표기를 요구합니다. BTC/USDT처럼 슬래시를 넣거나 소문자로 입력하면 오류가 발생합니다.
해결: Bybit 공식 문서에서 정확한 심볼 이름을 확인하세요. 현물은 BTCUSDT, 선물은 BTCUSDT(선물 카테고리) 형태입니다.
오류 2: Rate Limit 초과 (429 Too Many Requests)
import time
import requests
def safe_api_call_with_retry(url, params, max_retries=3):
"""Rate Limit을 고려한 안전한 API 호출"""
for attempt in range(max_retries):
try:
response = requests.get(url, params=params)
if response.status_code == 429:
# Rate Limit 초과 시 대기 후 재시도
wait_time = (attempt + 1) * 2 # 2초, 4초, 6초 대기
print(f"Rate Limit 초과. {wait_time}초 후 재시도 ({attempt+1}/{max_retries})")
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
print(f"요청 실패: {e}, 재시도 중...")
time.sleep(2)
return None
원인: Bybit Public API는 초당 10회, 분당 100회 제한이 있습니다. 이 제한을 초과하면 429 에러가 반환됩니다.
해결: 요청 사이에 0.1~0.2초 딜레이를 추가하고, 재시도 로직을 구현하세요. 대량 데이터 수집 시 Redis 큐나 스케줄러를 고려하세요.
오류 3: 빈 데이터 반환 또는 데이터 누락
import pandas as pd
from datetime import datetime, timedelta
def validate_and_fill_data(trades, expected_count=100):
"""
데이터 무결성 검증 및 보간
"""
if not trades:
print("⚠️ 경고: 데이터를 가져오지 못했습니다")
return None
df = pd.DataFrame(trades)
df['tradeTime'] = df['tradeTime'].astype(int)
df['price'] = df['price'].astype(float)
df['size'] = df['size'].astype(float)
# 시간순 정렬
df = df.sort_values('tradeTime')
# 시간 간격 이상 탐지 (1분 이상 비어있으면 경고)
df['time_diff'] = df['tradeTime'].diff()
large_gaps = df[df['time_diff'] > 60000] # 60초 = 60000ms
if len(large_gaps) > 0:
print(f"⚠️ 경고: {len(large_gaps)}건의 시간 간격 이상 발견")
print(" 해당 기간에 거래가 없었거나 API 호출 에러일 수 있습니다")
# 중복 제거
df = df.drop_duplicates(subset=['tradeId'])
print(f"✅ 최종 데이터: {len(df)}건 (중복 제거 후)")
return df
원인: 다음 세 가지 경우에 발생할 수 있습니다: ① 지정한 시간에 거래가 없음, ② API 응답 지연으로 데이터 누락, ③ Rate Limit으로 인한 요청 실패.
해결: 위 검증 함수를 사용하여 데이터 무결성을 확인하고, 누락 시 재요청하거나 데이터 수집 시간을 조정하세요.
오류 4: 타임스탬프 시간대 혼동
from datetime import datetime, timezone
❌ 잘못된 시간 해석
raw_timestamp = 1706691000000
wrong_time = datetime.fromtimestamp(raw_timestamp) # 로컬 시간대로 해석 ❌
✅ 올바른 해석
correct_time = datetime.fromtimestamp(raw_timestamp / 1000, tz=timezone.utc)
local_time = correct_time.astimezone() # 현재 로컬 시간대로 변환
print(f"원본 타임스탬프: {raw_timestamp}")
print(f"UTC 시간: {correct_time}")
print(f"로컬 시간: {local_time}")
원인: Bybit API는 모든 시간을 밀리초 단위 UTC 타임스탬프로 반환합니다. 이를 로컬 시간으로 변환하지 않으면 9시간 차이가 날 수 있습니다.
해결: 항상 datetime.fromtimestamp(ms / 1000, tz=timezone.utc) 패턴을 사용하고, 필요시 .astimezone()으로 변환하세요.
실전 활용 팁
- 데이터 캐싱: 같은 데이터를 반복 요청하지 않도록 로컬 파일이나 Redis에 캐싱하세요.
- 자동화: Linux의 cron 또는 Windows 작업 스케줄러로 매일 자동으로 데이터를 수집하게 설정하세요.
- 오류 로깅: 모든 API 호출 결과를 로그 파일에 기록하면 문제 발생 시 원인 파악이 빠릅니다.
- 대량 수집 시: 여러 날 데이터를 수집할 때는 분할 요청(하루씩 나누기)과 중간 저장으로 데이터 손실을 방지하세요.
결론
Bybit API를利用한 과거 체결 데이터 수집은 비교적 간단하지만, Rate Limit, 시간대 변환, 데이터 무결성 등 여러 함정이 있습니다. 이 튜토리얼에서 소개한 코드들을 기반으로 자신의 목적에 맞는 데이터 수집 시스템을 구축하시기 바랍니다.
특히 트레이딩 봇이나 자동매매 시스템을开发하시는 분들이라면, 수집한 데이터를 실시간 분석하고 수익 극대화 전략에 활용할 수 있습니다. Python과 Bybit API 조합은 퀀트 트레이딩 입문에 가장 좋은 선택입니다.
궁금한 점이 있으시면 댓글을 남겨주세요. 성공적인 트레이딩 시스템을 개발하시길 바랍니다!
📚 관련 튜토리얼:
👉 HolySheep AI 가입하고 무료 크레딧 받기