2025-06-01 [4] bithumb_api_client.py

이제까지의 테스트를 끝으로 본격적으로 매크로를 만들어보려고 합니다.

하지만 중간중간 함수 Test 진행과 가상 Money를 이용한 동작 Test는 계속 진행할 예정입니다.

 

1. Bithumb API 함수 제작

- Module import

import requests
import time
import uuid
import jwt  # PyJWT Module
import json

# --- API_KEY, SECRET_KEY 가져오기 ---
try:
    from config import api_key, api_secret
except ImportError:
    print("주의: config.py 파일을 찾을 수 없습니다. API 키를 사용하려면 해당 파일을 생성하고 api_key와 api_secret 변수를 설정해주세요.")
    api_key = None  # 기본값으로 None 설정
    api_secret = None  # 기본값으로 None 설정

- URL 설정

# --- Set URL ---
BITHUMB_API_URL = "https://api.bithumb.com"

- 종목 정보 가져오는 함수 제작 (get_ticker_info)

# --- 종목 정보를 가져오는 함수 ---
"""
order_currency : 종목 이름 (예: "BTC", "ETH")
payment_currency : 화폐 단위 (기본값: "KRW")
return
    - success : Ticker 정보
    - failure : None
"""
def get_ticker_info(order_currency, payment_currency="KRW"):
    endpoint = f"/public/ticker/{order_currency.upper()}_{payment_currency.upper()}"
    url = f"{BITHUMB_API_URL}{endpoint}"

    try:
        response = requests.get(url)
        response.raise_for_status()  # HTTP 오류 발생 시 예외 발생 ('200 OK'가 아닌 경우)
        data = response.json()

        """
        data 정보 data.get('key')를 통해 특정 value 획득 가능
        status : return code
        data
            opening_price           : 당일 시가
            closing_price           : 가장 최근 체결 가격 (사실상 현재가)
            min_price               : 당일 저가
            max_price               : 당일 고가
            units_traded            : 당일 거래량
            acc_trade_value         : 당일 거래 금액
            prev_closing_price      : 전일 종가
            units_traded_24H        : 최근 24시간 거래량
            acc_trade_value_24H     : 최근 24시간 거래 금액
            fluctate_24H            : 최근 24시간 변동 금액 (현재가 - 24시간 전 가격)
            fluctate_rate_24H       : 최근 24시간 변동률 (%) = (현재가 - 24시간 전 가격) / 24시간
            date                    : 데이터 기준 시간 (timestamp) (ms)
        """
        if data.get("status") == "0000":  # Bithumb API get success
            return data.get("data")
        else:   # Bithumb API get fail
            print(f"Ticker API 에러: {data.get('message')} (상태 코드: {data.get('status')})")
            return None
    except requests.exceptions.RequestException as e:   # HTTP request fail
        print(f"HTTP 요청 에러 (Ticker): {e}")
        return None
    except json.JSONDecodeError:    # JSON parsing fail
        print("JSON 응답 파싱 실패 (Ticker)")
        return None
    except Exception as e:  # Unknown error
        print(f"알 수 없는 에러 발생 (Ticker): {e}")
        return None

- 계정 정보 가져오는 함수 제작

# --- 사용자의 계정 정보를 가져오는 함수 ---
"""
return
    - success : 계정 정보
    - failure : None
"""
def get_account_info():
    if not api_key or not api_secret:
        print("API 키 또는 시크릿 키가 설정되지 않았습니다. config.py 파일을 확인해주세요.")
        return None

    endpoint = "/v1/accounts"
    url = f"{BITHUMB_API_URL}{endpoint}"

    # JWT payload 생성
    payload = {
        'access_key': api_key,
        'nonce': str(uuid.uuid4()),  # 매번 고유한 값 생성
        'timestamp': str(round(time.time() * 1000))  # 현재 시간을 밀리초 단위로 변환하여 문자열로
    }

    # JWT 토큰 생성 (HS256 알고리즘 사용)
    jwt_token = jwt.encode(payload, api_secret, algorithm="HS256")
    authorization_token = f'Bearer {jwt_token}'

    headers = {
        'Authorization': authorization_token
    }

    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()  # HTTP 오류 발생 시 예외 발생

        data = response.json()
        return data  # 성공 시 전체 date 반환

    except requests.exceptions.HTTPError as e:
        print(f"HTTP 에러 (Account Info): {e.response.status_code} - {e.response.text}")
        return None
    except requests.exceptions.RequestException as e:
        print(f"HTTP 요청 에러 (Account Info): {e}")
        return None
    except Exception as e:
        print(f"알 수 없는 에러 발생 (Account Info): {e}")
        return None

 

2. 함수 테스트 부분 제작

# --- 테스트용으로 함수를 호출하는 부분 ---
"""
if __name__ == "__main__": 이 파일을 main으로 실행했을 때 실행되는 부분
"""
if __name__ == "__main__":
    # 만약 config.py가 없거나 API 키가 없을 경우를 위한 안내
    if not api_key or not api_secret:
        print("\n[알림] API 키가 설정되지 않아 Private API 테스트는 건너뛰었습니다.")
        print("Private API 기능을 테스트하려면 config.py 파일에 api_key와 api_secret을 설정해주세요.")
        exit(255)
    test_order_currency = "BTC"
    test_payment_currency = "KRW"
    print(f"--- Ticker 정보 테스트 ({test_order_currency}/{test_payment_currency}) ---")
    btc_ticker = get_ticker_info(test_order_currency, test_payment_currency)
    if btc_ticker:
        print(f"{test_order_currency} 현재가: {btc_ticker.get('closing_price')} {test_payment_currency}")
        # print(json.dumps(btc_ticker, indent=4, ensure_ascii=False)) # 전체 데이터 보고 싶으면 주석 해제

    print("\n--- 계정 정보 테스트 ---")
    account_data = get_account_info()
    if account_data:
        print("계정 정보를 성공적으로 가져왔습니다.")
        # 특정 data 출력
        for i in range(len(account_data)):
            if account_data[i].get("currency") == "ETH":
                print(json.dumps(account_data[i], indent=4, ensure_ascii=False))
        #print(json.dumps(account_data, indent=4, ensure_ascii=False)) # 전체 데이터 보고 싶으면 주석 해제

 

 

3. 결과 확인

아래와 같이 정보가 출력되면 테스트 결과 정상입니다.

"ETH"(이더리움)을 보유하고있지 않으면 계정 정보가 출력되지 않을 수 있습니다. (주석 확인)

 

이런 정보들을 잘 조합하면 "오늘 시가 대비 얼마나 올랐나?", "24시간 변동률이 특정 수치를 넘었나?" 같은 다양한 시장 분석 조건이나 매매 전략의 신호로 활용 가능하다.

 

다음 단계

• 매매 전략을 세우고, 그 전략에 따라 판단하는 로직 구현

 

끝!