[데이터수집] ccxt로 binance 1분봉 실전 수집기 만들기

 

전략 개발하든, 백테스트를 하든, 결국 시작은 가격데이터 확보 및 적재이다.
ccxt 모듈을 활용하여 binance의 과거 가격데이터를,
from ~ to 기준 분봉단위로 수집하는 실전용 클래스를 소개한다.

그냥 코드만 던지는 건 재미없고, 실전에서 부딪히는 3가지 핵심 포인트 중심으로 소개.

 

 

 

ccxt는 이스라엘 개발자 Igor Kroitor가 만든 오픈소스 프로젝트로,

크립토 데이터 수집의 사실상 표준으로 자리잡았다.

 

거래소마다 API도 제각각인데,

ccxt는 이것을 통일된 방식으로 감싸주는 라이브러리이다.

그래서 binance든 업비트든 똑같은 코드로 호출할 수 있다.

데이터 수집 입문용으로는 거의 정석.

 

초기엔 개인 프로젝트 였던 것이,

현재는 오히려 거래소 공식문서에서 ccxt 사용법을 소개할 정도.

업비트도 지원한지 5년이 넘었구만

 

 

0. ccxt - binance 기본 사용법

ccxt 사용법은 매우매우 간단하다.

단순 가격조회는 API key조차 필요없다.

import ccxt    # ccxt 호출
import time

# spot = ccxt.binance()   # binance 현물
perp = ccxt.binanceusdm() # binance 선물

since = int(time.time() * 1000) - 60 * 1000  # 약 1000분전으로 시간 설정

# 실제 함수 호출
ohlcv = perp.fetch_ohlcv('ETHUSDT', '1m', since=since, limit=1000)

print(ohlcv[:5])

 

(출력)

[[1750496580000, 2437.69, 2437.69, 2435.8, 2435.91, 5063.953], [1750496640000, 2435.91, 2436.25, 2434.93, 2435.47, 2928.102], [1750496700000, 2435.47, 2437.58, 2435.3, 2435.93, 4288.201], [1750496760000, 2435.93, 2436.5, 2435.57, 2436.5, 1331.868], [1750496820000, 2436.5, 2436.5, 2434.51, 2434.52, 2228.517]]

 

사실상 fetch_ohlcv 한 줄로 입수가 가능하다.

 

fetch_ohlcv()는 다음과 같은 인수를 받는다:

1. symbol: 'BTCUSDT', 'ETHUSDT' 등
2. 봉 단위: 1분봉('1m'), 1시간봉('1h'), 1일봉('1d') 등
3. 조회 시작 시간: UTC 기준 밀리세컨드 timestamp
4. limit: 한 번에 조회할 봉의 수 (최대 1,000개)

---

1. SPOT과 PERP의 구분

먼저 binance에서는 같은 'BTCUSDT' 쌍이라고 하더라도,

현물시장(Spot)과 선물시장(Perpetual Futures)이 별도로 존재한다.

 

Perpetual은 만기일이 없는 무기한 선물(perpetual contract)이기 때문에,

perp이라는 이름이 붙는다.

 

class CCXTPriceFetcher(PriceBatchFetcher):
    def __init__(self):
        self.spot = ccxt.binance()     # binance 현물(spot) 시장
        self.perp = ccxt.binanceusdm() # binance 선물(perp) 시장

    def fetch_ohlcv(self, symbol: str, from_date: str, to_date: str, interval: int,
        venue: str = "perp") -> List[AssetPriceRecordModel]:

        exchange = self.perp if venue == "perp" else self.spot
        market_symbol = symbol.replace("_PERP", "") if symbol.endswith("_PERP") else symbol
        
    # ...

 

내가 실제 사용하는 실전용 조회함수는

다음과 같은 인수를 받는다:

1.  symbol: 'BTCUSDT' (현물), 'ETHUSDT_PERP' (선물) 등
2. from_date: '2025-06-01' 
   (timestamp고 뭐고 모르겠고 그냥 일자기준으로 넣을래)
3. to_date: '2025-06-22'
4. interval: 1분단위 숫자
5. venue: 'spot' or 'perp'

나는 선물 symbol을 'ETHUSDT_PERP' 과 같이 관리하고,

실제 ccxt 요청 시에는 '_PERP' 을 제거해서 호출한다.

 

솔직히 선물에 접미어를 붙이지 말고,

현물에 '_SPOT'을 붙였어야 했는데.

초기 설계가 아쉽다.

 

기본 시장(venue)은 선물( 'perp' )로 지정해뒀다.

롱도 치고 숏도 치고, 양방향 트레이딩을 위해서이다.

---

2. K-조선에서는 UTC 안 씁니다

ccxt의 since 인수는 UTC 기준의 ms timestamp를 요구한다.

하지만 우리는 익숙한 KST 기준 날짜 문자열로 입력하고 싶다.

그래서 다음과 같이 변환한다.

from_utc = int((pd.to_datetime(from_date) - timedelta(hours=9)).timestamp() * 1000)
to_utc = int((pd.to_datetime(to_date) - timedelta(hours=9)).timestamp() * 1000)

 

아 모르겠고 9시간 빠르대

• '2025-06-22' → 2025-06-21 15:00:00 UTC로 바뀜
• ccxt에 던질 때는 UTC로 변환 ( - timedelta(hours=9)),
• 반환 값은 다시 KST로 변환 ( + timedelta(hours=9)).

df["trade_date"] = pd.to_datetime(df["timestamp"], unit="ms") + timedelta(hours=9)

 

이 부분을 처리하지 않으면

한국 기준 날짜가 어긋나는 대참사가 발생한다.

특히 binance, upbit 등 여러 거래소 가격데이터 적재 시,

동일시각 가격이 다르게 입력될 수 있음.

---

3. 1,000개 봉은 16시간 밖에 안 되요

1분봉으로 확보하면 5분이든 1시간이든 일봉이든 만들 수 있지만,

1시간봉으로 확보하면 1분봉, 5분봉은 만들 수 없다.

그래서 데이터 확보는 1분 기준으로 진행한다.

 

ccxt의 fetch_ohlcv()는 호출당 최대 1,000개 봉만 반환한다.
즉, 1분봉 기준으로는 약 16.7시간치만 확보 가능하다.

 

저는 1년 치 쌓고 싶은데요?

 

그래서 긴 시간을 수집하려면 from~to 기간 사이 루프가 필요하다.

all_data = []
since = from_utc

while since < to_utc:
    ohlcv = exchange.fetch_ohlcv(market_symbol, tf, since=since, limit=1000)
    if not ohlcv:
        break
    all_data.extend(ohlcv)
    since = ohlcv[-1][0] + 60_000  # 1m 간격
    # time.sleep(5) # 너무 오랜기간 요청하면 Too Many Requests 뜸

 

한 번에 너무 많은 기간을 요청하면 IP가 밴 당할 수 있다.
그래서 time.sleep()을 함께 쓰면 좋다.

 

---

4. 마무리: DataFrame 변환 및 반환

가져온 데이터는 pandas로 변환하고,
내부 프로젝트 기준에 맞춰 정리한다.

        df = pd.DataFrame(all_data, columns=["timestamp", "open", "high", "low", "close", "volume"])
        df["trade_date"] = pd.to_datetime(df["timestamp"], unit="ms") + timedelta(hours=9)
        df["asset_id"] = symbol
        df["open_interest"] = None

        df = df[[
            "trade_date", "asset_id", "open", "high", "low", "close", "volume", "open_interest"
        ]]
        df = df[(df["trade_date"] >= from_date) & (df["trade_date"] <= to_date)]

        records = [
            AssetPriceRecordModel(
                trade_date=row["trade_date"],
                asset_id=row["asset_id"],
                open=row["open"],
                high=row["high"],
                low=row["low"],
                close=row["close"],
                volume=row["volume"],
                open_interest=row["open_interest"],
            )
            for _, row in df.iterrows()
        ]
        l = get_logger(self)
        l.info(f"Successfully read price records from CCXT.")
        return records

 

매일 새벽 4시마다 전일 데이터를 적재하여
전략 개발, 백테스트, 시각화에 활용 중이다.

 

---

풀 코드는 아래 GitHub에서 확인 가능:

https://github.com/reposer/minstack/blob/main/infrastructure/price_fetcher/ccxt_price_fetcher.py

minstack/infrastructure/price_fetcher/ccxt_price_fetcher.py at main · reposer/minstack

 

다음에는

1. 현재 풀코드 기준으로 소개하지 못한 나머지 처리 로직
2. 수집한 데이터를 MySQL DB에 저장하는 파이프라인
3. 전체 구조를 클린 아키텍쳐로 정리 방법

계획중.