암호화폐 캔들 데이터 확보 코드 . 파이썬 pybit

 

 

 

바이비트 암호화폐 캔들 데이터 처리 코드 . 파이썬 pybit 이용. 

 

 

개요 

암호화폐 거래소 바이비트 API 통신 위한 공식 파이썬 패키지 pybit 이용하여 파이썬에서  캔들 데이터 확보하기 위한 가장 기본적인 형태(사람이 사용하기 불편한 형태)에서 출발하여 점진적으로 더 유용한 형식으로 코드 발전시키는 과정, csv 파일로 저장 , 코드 정리 방법등  단계별로 모두 정리. 

 

본 글에서의 바이비트 API 버전 : 현재(2024. 08.21) 시점  최신 버전 API V5 . 

 

 

 

사전 필수 셋팅 

- 파이썬 개발환경 구축 되어있어야 함. 구축예 :  https://igotit.tistory.com/5761

- 파이썬에 pybit 설치되어있어야 함.  설치구문 :  pip install pybit 

 

사전 필수 개념 이해

 

- 본 글에서 pybit 이용하여 캔들 데이터 확보한다 함은 bybit 거래소 API v5 의 Get Kline 를  pybit 으로 편리하게 접근하는 것. 

 

-  API v5 의 Get Kline 의 request , response 이해하고 있어야 함. 

 

 

코드 1. 캔들 데이터 확보 기본 코드 

바이비트 거래소의 거래 종목 중 리니어 무기한 계약 인 BTCUSDT 의 캔들 데이터를 받기 위한 가장 기본적인 코드로 시작하자.  

 

아래 코드는 response = session.get_kline( ,,,) 으로 데이터 요청하여 수신된 데이터를 출력하는 초간단 코드이다. 

 

인자들

 

category :  BTCUSDT  종목이 속한  linear 기록,

 

 - 현물 (예: BTCUSDT ) 이면 spot, 인버스 종목 (예: BTCUSD )이면 inverse기록한다. 

 

 

symbol : 종목명을 대문자로 기록,

 

interval : 요청할 캔들의 시간 주기 5분봉에 해당하는 5 기록,

 - 기록가능한 값 : 1,3,5,15,30,60,120,240,360,720,D,M,W

 

start : 요청하는 데이터의 시작시간을 epoch 시간으로 밀리초 단위로 기록. 

 

end : 요청하는 데이터의 끝시간을 epoch 시간으로 밀리초 단위로 기록.

 

주의 : 반환 데이터는 end  시간부터 과거 시간순으로 제공됨. 

 

 

 

 

 

 

위 코드 실행하면 아래 화면처럼 VSCode 의 터미널 창에서  response 데이터들 출력된다.  

 

반환되는 캔들 데이터는 end 시간부터 과거 순으로 제공된다.  

 

 

 

코드 2.  start, end  인자 설정부 개선 

코드1에서 보면 가장 번잡스러운 지점은 캔들 데이터 요청하는 시간 기록하는 부분이다. 

 

특정 년월일시분초를 epoch 시간으로 수계산 하여 인자로 전달하지 않고  "2024-08-01 00:00:00"  처럼 사람이 익숙한 시간표현 형식으로 시구 간 설정하는 기능이 요구된다. 

 

utc 시간을 문자열 로 입력하면 epoch 밀리초 반환하는 함수를 만들자. 

 

관련 상세 정보 : 파이썬에서 시간처리 방법 

from datetime import datetime, timezone

def utc_datetime_to_epoch_ms(datetime_str):
    """
    UTC 날짜 및 시간 문자열을 밀리초 단위의 Epoch 시간으로 변환함.

    Args:
        datetime_str (str): UTC 날짜 및 시간 문자열, 포맷은 "YYYY-MM-DD HH:MM:SS"

    Returns:
        int: 밀리초 단위의 Epoch 시간
    """
    # 문자열을 datetime 객체로 변환 (UTC로 설정)
    dt_utc = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S")
    dt_utc = dt_utc.replace(tzinfo=timezone.utc)

    # Epoch 시간으로 변환 (초 단위)
    epoch_s = int(dt_utc.timestamp())

    # 밀리초 단위로 변환
    epoch_ms = epoch_s * 1000
    
    return epoch_ms

# 사용 예시
utc_datetime_str = "2024-08-01 00:00:00" 
epoch_ms = utc_datetime_to_epoch_ms(utc_datetime_str)
print(epoch_ms)

 

 

이제 위 함수를 이용하여 시간 부분을 사람이 입력하기 편하게 하여 캔들 데이터 요청하는 코드. 

 

 

 

 

 

 

이제 위의 코드 이용하면 아래처럼 문자열 부분에 시작시간, 끝시간을 기록해 주면 되므로 편리하다. 

utc_datetime_start_str = "2024-08-01 00:00:00"

utc_datetime_end_str = "2024-08-21 00:00:00"

 

 

위 코드 2 의 문제점.

위 예제 코드는 8월1일 0시0분0초 부터 8월 21일 0시 0분 0초 까지의 5분 봉 데이터를 요청한 것이므로  이 기간 동안 5분 봉 수량은 5,761개이다.   그러나 위 코드 실행해 보면 지정한 end 시간 부터 과거 순으로 200개만 반환된다.  함수 get_kline( ,,, ) 의 인자로 limit 지정하지 않은 경우에는 기본 200개만 반환하기 때문이다. limit 지정 가능한 최대값은 1000 으로 제한되어있어서 1회 호출만으로는 1000개 이상 캔들 데이터 확보 불가하다.   

 

코드 3. get_kline() 반복호출 로 지정한 시 구간 내의 모든 캔들 확보 

상기 코드2의 문제점 해결하기 위하여, get_kline()을 반복적으로 호출하는 구조로 코드 구현하기로 한다.

 

반복 호출시 직전 호출에서 확보된 리스트의 마지막 캔들(시간적으로 과거) 보다 캔들 주기만큼 더 과거 시간을  인자 end에 재설정하는 방식이다. 

 

먼저 , 아래와 같은 함수를 정의하자. session.get_kline 을 반복적으로 호출하기 편리하게 한것. 함수 반환값에 session.get_kline() 호출결과 오류 점검용 retCode 를 반환하고 오류 없는 경우 캔들 데이터 있는 list 부분만 반환하고 있다. 

def cy_get_kline(category, symbol,interval_str, start_epoch_ms, end_epoch_ms):

    # 캔들 데이터  요청
    response = session.get_kline(
        category=category,
        symbol=symbol,
        interval=interval_str,  
        start=start_epoch_ms,  
        end=end_epoch_ms, 
        limit = 1000 # 요청가능 최대수량 1000
    )
    retCode = response["retCode"] # 0이면 정상 실행 . 그외 실패. 

    # session.get_kline() 호출결과 오류점검
    if retCode != 0 :
        print(f"retCode = {retCode}") #오류코드 출력.
        print(f"retMsg = {response['retMsg']}") #오류요인 출력. 

        return retCode      

    # 오류 없는 경우 candle list 확보.
    candle_list = response["result"]["list"]; 

    return retCode,candle_list

 

위 함수를 이용한 전체코드. 아래 코드에서 while 부분이 지정한 시구간에서 전체 캔들 확보 처리하고 있다. 

 

 

실행결과 화면

 

 

- 요청한 시구간에서 수집되어야 할 캔들수량 정확히 확보되었음을 알 수 있다. 

 

 

 

 

코드 4 .  캔들 데이터 파일 저장 기능 추가 

통상 캔들 데이터 수집하는 이유는 장기간의 데이터로  분석, 예측 등의 용도로 활용하는 경우가 많다. 데이터 수량이 방대한 경우 매번 필요할 때마다 거래소 API 호출하면 데이터 수집에 시간 소요되므로 파일이나 DB에 저장해 둔 데이터를 활용하는 것이 훨씬 효율적이다. 

 

파일 저장시 csv 파일형식이 많이 사용되며, 본 코드 4 에서는 csv 파일에 지정된 구간의 모든 캔들 데이터를 누적 저장하는 기능을 상기 코드 3에 추가 구현한다. 

 

앞의 코드3에서 candle_list 가 반복적으로 확보되는 시점마다 파일에 누적 저장하는 기능이 요구된다. 파일 저장을 위한 함수를 아래처럼 정의하자. 

 

함수의 마지막 인자는 첫 기록시점에만 헤더( timestamp, open, high, low, close, volume, turnover) 기록하기 위한 제어변수.

import csv
import os

def save_to_csv(file_name, data, write_header=False):
    """
    CSV 파일에 데이터를 누적 저장
    """
    with open(file_name, mode='a', newline='') as file:
        writer = csv.writer(file)
        
        # 처음 파일을 생성할 때 헤더를 추가
        if write_header:
            writer.writerow(["timestamp", "open", "high", "low", "close", "volume", "turnover"])
        
        for row in data:
            writer.writerow(row)

 

위 함수 활용하여 csv  파일로 저장하는 전체 코드.

 

 

실행결과 

- 파일 열어보면 의도한 대로 잘 저장해 준다. 

 

 

저장된 csv 파일을 읽어서 캔들 챠트 그려보면 아래와 같다. 

 

 

동일 구간의 실제 바이비트 거래소의 USDT 5분봉 보면 아래 그림과 같고, 본 코드의 캔들 데이터 수집이 정확하게 이뤄지고 있음을 확인할 수 있다. 바이빗 웹에서 보이는 5분 봉은 최대 약 1주일 간 만 볼 수 있다. 19일 ~ 23 일 구간을 위 챠트와 비교해보면 동일함을 알 수 있다. 

 

 

 

참고 : 저장된 csv 파일을 오픈하여 챠트 그리는 방법.  

plotly . 웹으로 챠트 표현

 

 

기타. 

위 코드로 1년치 (2023-08-21 ~ 2024-08-21) 1분봉 저장해 보면 처리 완료까지 약 2분 소요되며, 저장된 파일 용량은 33Mbyte, 캔들수량은 527,041개 이다. 

 

 

 

 

 

 

 

코드 5. 사람 실수 방지 및 코드 구조 정리  

앞의 코드 4 와 기능은 동일하면서 , 본 코드 활용하여 주기적(예 1주일 단위로 1주일 동안 새로 생성된 데이터 수집 등 )으로 데이터를 확보 처리를 해야 하는 경우 사람의 실수가 등장할 수 있는 부분을 최소화해 보자. 

 

- 파일명 자동 생성 . 코드 4에서는 파일명을 사람이 기록하게 되어있는데 파일명 작성 규칙이 있다고 해도 필연 실수 발생하기 쉬운 지점이다. 자동으로 파일명 생성하게 한다.

 

파일명 형식예 : bybit_linear_BTCUSDT_M5_2024.08.01_00.00.00_2024.08.21_00.00.00.csv 

 

- 캔들봉의 주기 (1분봉, 5분봉, 1시간봉.. ) 설정 및 각 봉의 시간주기에 해당하는 밀리초 확보가능한 데이터 형식 도입. 이를 위하여 아래와 같은 enum 정의한다.  사용할 때는 이름 M1, M3, H1, .. 식으로 사람이 쉽게 식별 가능하게 하고 파일 저장 시에 이 이름이 파일명에 명시되게 하여 해당 파일의 캔들의 시간 주기를 바로 식별할 수 있도록 한다. 

 

from enum import Enum

class KlineInterval(Enum):
    """Kline 간격을 문자열 형식으로 정의하고, 밀리초 단위 시간 간격을 계산하는 Enum 클래스"""
    M1 = "1"      # 1분
    M3 = "3"      # 3분
    M5 = "5"      # 5분
    M15 = "15"    # 15분
    M30 = "30"    # 30분
    H1 = "60"     # 1시간
    H2 = "120"    # 2시간
    H4 = "240"    # 4시간
    H6 = "360"    # 6시간
    H12 = "720"   # 12시간
    D1 = "D"      # 1일
    W1 = "W"      # 1주

    def __str__(self):
        return self.value

    def to_milliseconds(self):
        """Enum 멤버에 해당하는 밀리초를 반환"""
        intervals_in_minutes = {
            KlineInterval.M1: 1,
            KlineInterval.M3: 3,
            KlineInterval.M5: 5,
            KlineInterval.M15: 15,
            KlineInterval.M30: 30,
            KlineInterval.H1: 60,
            KlineInterval.H2: 120,
            KlineInterval.H4: 240,
            KlineInterval.H6: 360,
            KlineInterval.H12: 720,
            KlineInterval.D1: 1440,
            KlineInterval.W1: 10080
        }
        return intervals_in_minutes[self] * 60 * 1000

    @classmethod
    def from_string(cls, interval_str):
        for interval in cls:
            if interval.value == interval_str:
                return interval
        raise ValueError(f"Invalid interval: {interval_str}")

    @classmethod
    def is_valid(cls, interval_str):
        try:
            cls.from_string(interval_str)
            return True
        except ValueError:
            return False


=======================================================
활용예 

# 문자열로부터 Enum 멤버를 얻는 방법
interval = KlineInterval.from_string("5")
print(interval)  # 출력: KlineInterval.M5

# 밀리초 단위 시간 간격 얻기
milliseconds = interval.to_milliseconds()
print(milliseconds)  # 출력: 300000 (5분)

# 유효성 검사
print(KlineInterval.is_valid("60"))  # 출력: True
print(KlineInterval.is_valid("7"))   # 출력: False

 

 

위 enum 적용 및 파일명 자동 생성부 추가하고 , 가독성 향상시킨 코드4의 개선 최종 코드. 기능은 4와 동일함. 

 

코드 에서 사용자가 기록해줘야 하는 부분은 라인 80~87 까지가 전부이다. 여기에 기록된 값을 재료로 사용하여 파일명은 자동 생성 처리된다. 

 

 

 

 

 

코드6. 모듈화 

 

- 코드 5 까지는 모든 함수들이 1개의 파일에 다 몰려 있는 구조인데, 시험이 완료되고 다듬어진 함수들을 보면 범용성 있기에 별도의 파일에 정의해 두고 필요한 곳에서 호출하여 사용하는 식으로 활용하게 된다. 

 

- 본 코드 6에서는 코드5 의 main() 이 작성된 파이썬 파일에 있던 모든 함수들을 별도의 파일로 만들고 이를 main() 이 작성된 파일에서는 함수들이 보이지 않게 처리해 뒀다. 

 

- 기능적으로 코드 6 은 코드 5와 완전히 동일하므로 코드 실행 결과도 동일하다. 

 

- 코드 모듈화 방식은 프로젝트 초기 단계부터 파일 분리, 클래스의 완전무결한  정의, 상속 관계 완벽... 이런 식으로  구현하는 방식은 적합하지 않다. 이런 방식이 가능한 경우는  이미 모듈화 코딩 내공 높은 사람이 그 동한 해 왔던 유사 분야의 코딩할 때나 이런 방식이 가능하다. 내공 높은 사람이라도 이전에 경험하지 못한 새로운 영역 작업하는 경우는 일단 실행되는 코드 작성부터 시작한다. 

 

처음에는 모듈화 되지 않은 것이라도 1개의 파일에  잡탕 방식으로 먼저 타이핑 시작해도 된다.

단 시작만 잡탕으로 시작하는 것이지 프로젝트 진행하면서 코드 구조 정리하고, 함수들이 잘 분리되고 다른 곳에서도 활용성이 생기는 것이 확인된 시점에 모듈화 하는 방식도 매우 실용적인 방법론이고 좋은 결과를 준다.  

 

 

- 아래 코드에서 앞부분에 from import 부분들이 다른 파일에 정의된 함수 및 클래스 로딩 처리. 

 

 

 

모듈화 잘된 파일을 만들어 두는 장소예. 

 

- 사람마다 본인 소스 파일 관리방식은 천차만별인데, 본 예에서는 모듈화 잘된 것들 보관폴더인 CySDK_PYTHON 하위에 파일들 만들었다. 아래그림 붉은 박스. 

- 당연하게도, 야무진 꿈을 꾸며 매진했던 많은 응용 프로젝트들이  항상 좋은 결과를 주는 것은 아니다.

 

프로젝트에서 희망했던 목표가 달성되지 못했다고, 그 과정에서 피땀 흘려가며 만든 것들도 같이 쓰레기통에 처박히는 것은 더더욱 나쁜 결과를 만들게 된다. 

 

나의 시간, 피와 땀, 노력이 투입된 것들을 지속적으로 일관되고 체계적으로 저장 가능한 "가치 누적 저장 형식" 을 구축함이 타당하다.     

 

 

 

 

데이터 베이스에 저장하기 

 

앞의 코드는 데이터를 CSV 파일로 저장하는 예이나, 전체 코드 골격 유지하면서 SQLite DB 에 저장 가능하다. 

별도 정리함. https://igotit.tistory.com/5810

암호화폐 과거 캔들 데이터 DB 에 저장하기 . 파이썬 코드

 

 

---

첫 등록 : 2024.08.21

최종 수정 : 2024.09.01

단축 주소 : https://igotit.tistory.com/5766

---