Bitget 선물 1분봉 (1min candle, ohlc) 데이터 수집 코드 (파이썬)

목차

퀀트 트레이딩 시스템을 개발하고 적용하기 위해서는 아이디어가 잘 작동하는지 확인하는 작업이 필요하다. 나같은 회로 개발자들은 Testbench 또는 시뮬레이션이라 부르는데, 이 쪽 분야에서는 백테스트(Backtest)라는 용어가 흔히 사용된다.

이 Backtest를 하기 위해서 다양한 데이터가 필요한데, 일반적으로 차트 분석을 기반으로 하는 시스템 트레이딩은 캔들(시고저종)과 거래량이 주요 지표이다. 이 글을 검색해서 왔다면 뭐 다 아는 얘기일테니 서론은 이만하고 바로 코드를 예시로 적어보겠다.

import configparser
import hmac
import base64
import json
import time
import os
import requests
import pandas as pd
from datetime import datetime, timedelta
import pytz
import hashlib

# Configurations
config = configparser.ConfigParser()
config.read('config.ini')

try:
    API_KEY = config.get('bitget', 'API_KEY')
    API_SECRET_KEY = config.get('bitget', 'API_SECRET')
except configparser.NoOptionError as e:
    print(f"Config error: {e}")
    API_KEY = 'your_api_key_here'
    API_SECRET_KEY = 'your_secret_key_here'

API_BASE_URL = 'https://api.bitget.com'


# Helper functions
def get_timestamp():
    return int(time.time() * 1000)


def get_utc_times_from_seoul(start_date, end_date):
    if isinstance(start_date, str):
        start_date = datetime.strptime(start_date, '%Y-%m-%d')
    if isinstance(end_date, str):
        end_date = datetime.strptime(end_date, '%Y-%m-%d')

    seoul_tz = pytz.timezone('Asia/Seoul')
    start_seoul = seoul_tz.localize(datetime(start_date.year, start_date.month, start_date.day, 0, 0, 0))
    end_seoul = seoul_tz.localize(datetime(end_date.year, end_date.month, end_date.day, 0, 0, 0))

    start_utc = start_seoul.astimezone(pytz.utc)
    end_utc = end_seoul.astimezone(pytz.utc)

    return start_utc, end_utc


def create_signature(secret_key, timestamp, method, request_path, body):
    prehash = f'{timestamp}{method.upper()}{request_path}{body}'
    mac = hmac.new(secret_key.encode('utf-8'), prehash.encode('utf-8'), hashlib.sha256).hexdigest()
    return mac


def create_headers(api_key, signature, timestamp):
    return {
        'ACCESS-KEY': api_key,
        'ACCESS-SIGN': signature,
        'ACCESS-TIMESTAMP': str(timestamp),
        'Content-Type': 'application/json'
    }


def send_request(method, endpoint, params=None, body=''):
    timestamp = get_timestamp()
    request_path = endpoint + parse_params_to_str(params)
    signature = create_signature(API_SECRET_KEY, timestamp, method, request_path, body)
    headers = create_headers(API_KEY, signature, timestamp)

    url = API_BASE_URL + request_path
    try:
        if method == 'GET':
            response = requests.get(url, headers=headers)
        else:
            response = requests.post(url, headers=headers, data=body)

        response.raise_for_status()
        return response.json().get('data', [])
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        return None


def parse_params_to_str(params):
    if not params:
        return ''
    return '?' + '&'.join(f'{key}={value}' for key, value in sorted(params.items()))


def write_to_csv(data, filename, symbol):
    df = pd.DataFrame(data, columns=['ts', 'open_price', 'high_price', 'low_price', 'close_price', 'baseVol', 'usdtVol'])
    df['timebar_1m'] = pd.to_datetime(df['ts'].astype(float), unit='ms').dt.tz_localize('UTC') \
        .dt.tz_convert('Asia/Seoul').dt.strftime('%Y-%m-%dT%H:%M:%SZ')
    df['volume'] = df['usdtVol'].astype(float)
    df.drop(columns=['baseVol', 'usdtVol'], inplace=True)
    df.insert(0, 'symbol', symbol)
    df = df[['timebar_1m', 'symbol', 'open_price', 'high_price', 'low_price', 'close_price', 'volume']]
    df.to_csv(filename, index=False)
    print(f'Saved data to {filename}')


def get_futures_list():
    endpoint = '/api/v2/mix/market/tickers'
    params = {'productType': 'USDT-FUTURES'}
    return [item['symbol'] for item in send_request('GET', endpoint, params) if 'symbol' in item]


def fetch_history_candles(symbol, start_time, end_time, limit=200):
    endpoint = '/api/v2/mix/market/history-candles'
    params = {
        'symbol': symbol,
        'productType': 'usdt-futures',
        'granularity': '1m',
        'startTime': start_time,
        'endTime': end_time,
        'limit': limit
    }
    return send_request('GET', endpoint, params)


def fetch_and_save_data(symbol, start_date, end_date, symbol_path):
    date = start_date
    while date <= end_date:
        start_utc, end_utc = get_utc_times_from_seoul(date, date)
        start_time = int(start_utc.timestamp() * 1000)

        next_date = date + timedelta(days=1)
        start_utc_next, _ = get_utc_times_from_seoul(next_date, next_date)
        end_time = int(start_utc_next.timestamp() * 1000)

        all_data = []
        current_start_time = start_time
        while current_start_time < end_time:
            partial_end_time = min(current_start_time + 200 * 60 * 1000, end_time)
            ohlcv = fetch_history_candles(symbol, current_start_time, partial_end_time)
            if ohlcv:
                all_data.extend(ohlcv)
                current_start_time = partial_end_time
            else:
                print(f'{symbol}: No more data available for {date.strftime("%Y-%m-%d")}')
                break

        if all_data:
            filename = os.path.join(symbol_path, date.strftime("%Y%m%d") + ".csv")
            write_to_csv(all_data, filename, symbol)
        else:
            print(f'{symbol}: No data available for {date.strftime("%Y-%m-%d")}')
        date += timedelta(days=1)


def main():
    symbol_list = get_futures_list()
    remove_list = ['BTCUSDT','ETHUSDT','XRPUSDT']
    symbol_list = [symbol for symbol in symbol_list if symbol not in remove_list]

    start_date_input = input(
        "Enter the start date in 'YYYYMMDD' format (leave blank for 3 days ago/ 1 for 2019-07-10): ")
    if start_date_input:
        start_date = datetime.strptime('20190710', '%Y%m%d') if start_date_input == '1' else datetime.strptime(
            start_date_input, '%Y%m%d')
    else:
        start_date = datetime.now() - timedelta(days=3)

    end_date_input = input("Enter the end date in 'YYYYMMDD' format (leave blank for yesterday): ")
    end_date = datetime.strptime(end_date_input, '%Y%m%d') if end_date_input else datetime.now() - timedelta(days=1)

    for symbol in symbol_list:
        symbol_path = os.path.join('.\\BACK_DATA', symbol)
        if not os.path.exists(symbol_path):
            os.makedirs(symbol_path)
        fetch_and_save_data(symbol, start_date, end_date, symbol_path)


if __name__ == "__main__":
    main()

언어는 요즘 누구나 쉽게 쓰는 파이썬으로 작성되었다.

필요한 pkg는 알아서 설치하면 되고, 개인적으로 conda만 적용해도 위의 코드는 무난히 실행된다.

remove_list: 추출하지 말아야할 심볼

symbol_path: 심볼이 저장될 path. 기본적으로 BACK_DATA 아래 저장된다.

실행하면 시작 날짜를 입력하는데 1을 입력하면 비트겟 선물 거래소 시작일인 20190710부터 시작한다.

끝나는 날짜는 입력하지 않고 엔터키로 넘기면 어제 날짜가 입력된다.

각 파일의 이름은 날짜로 설정되며 각각의 심볼의 폴더에 csv 저장된다.

나중엔 BACK_DATA 폴더의 파일 이름을 확인해서 기존에 수집된 데이터는 제외하고 신규 DATA만 수집하는 옵션을 추가할 생각이다. 개인적으로 사용하기에는 어차피 일괄적으로 뽑은 날짜가 있기 때문에 큰 문제는 없을 듯하다.