2025-06-08 [8] [v1.0.0 완료] 캔들스틱, 단순 이동 평균(SMA), 골든/데드 크로스

오늘은 두뇌를 만들기 위해 사용할 캔들스틱(과거 시세 데이터) 가져오는 방법에 대해 진행하겠습니다.
드디어 거의 끝이 보이고 있네요.

1. 캔들 스틱 가져오기

1). 캔들스틱 가져오는 함수

- get API 이므로 bithumb_api_client.py에 함수 추가

# --- 캔들스틱 데이터 가져오는 함수 ---
"""
빗썸 Public API를 사용하여 특정 암호화폐의 캔들스틱 데이터를 조회합니다.
:order_currency: 조회할 암호화폐 심볼 (예: "BTC", "ETH")
:payment_currency: 결제 통화 (기본값: "KRW")
:chart_intervals: 차트 간격 (예: "1m", "3m", "5m", "10m", "30m", "1h", "6h", "12h", "24h")
:return: 성공 시 캔들스틱 데이터 (list), 실패 시 None
"""
def get_candlestick_data(order_currency, payment_currency="KRW", chart_intervals="24h"):
    # Bithumb의 /public/candlestick API 호출
    endpoint = f"/public/candlestick/{order_currency.upper()}_{payment_currency.upper()}/{chart_intervals}"
    url = f"{BITHUMB_API_URL}{endpoint}"

    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()

        if data.get("status") == "0000":
            return data.get("data")
        else:
            print(f"Candlestick API 에러: {data.get('message')} (상태 코드: {data.get('status')})")
            return None
    except requests.exceptions.RequestException as e:
        print(f"HTTP 요청 에러 (Candlestick): {e}")
        return None
    except json.JSONDecodeError:
        print("JSON 응답 파싱 실패 (Candlestick)")
        return None
    except Exception as e:
        print(f"알 수 없는 에러 발생 (Candlestick): {e}")
        return None

 

2). 캔들스틱 테스트

- main부에 아래 항목 추가하여 테스트 진행

if __name__ == "__main__":
    print("\n--- 캔들스틱 데이터 테스트 (BTC/KRW, 24시간봉) ---")
    btc_candlestick = get_candlestick_data("BTC", "KRW", "24h")
    if btc_candlestick:
        view_day = 3
        print(f"최근 {view_day}일간의 캔들스틱 데이터:")
        for candle in btc_candlestick[view_day*-1:]:
            # 빗썸 캔들스틱 데이터 구조: [타임스탬프, 시가, 종가, 고가, 저가, 거래량]
            timestamp_ms = candle[0]
            open_price = candle[1]
            close_price = candle[2]
            high_price = candle[3]
            low_price = candle[4]
            volume = candle[5]

            # 타임스탬프를 사람이 읽을 수 있는 날짜로 변환 (ms 단위이므로 1000으로 나눠야 함)
            import datetime
            date = datetime.datetime.fromtimestamp(timestamp_ms / 1000).strftime('%Y-%m-%d')

            print(f"날짜: {date}, 시가: {open_price}, 종가: {close_price}, 고가: {high_price}, 저가: {low_price}, 거래량: {volume}")

- 아래와 같이 결과가 나오면 성공

 

2. 단순 이동 평균 (Simple Moving Average, SMA)

1). Pandas 모듈 설치

- 데이터 분석을 위해 Pandas 모듈 설치
- 터미널에 ' pip install pandas' 입력

 

2). technical_analyzer.py

- 기술적 지표 계산 로직을 위한 technical_analyzer.py를 생성한다.

о import 추가

- pandas 모듈을 추가하고 pd라고 지정 (pandas 대신 pd라는 이름으로 사용)

import pandas as pd

 

о 단순 이동 평균(SMA) 계산 함수

# --- 단순 이동 평균(SMA) 계산 함수 추가
"""
주어진 캔들스틱 데이터로 단순 이동평균(SMA)을 계산
:candlestick_data: 빗썸 API로부터 받은 캔들스틱 데이터 (리스트의 리스트)
:period: 이동평균을 계산할 기간 (기본값: 5)
:return: 성공 시 이동평균선 데이터 (pandas Series), 실패 시 None
"""
def calculate_sma(candlestick_data, period=5):
    if not candlestick_data or len(candlestick_data) < period:
        print(f"⚠️ SMA 계산 실패: 데이터가 부족합니다. (필요: {period}, 보유: {len(candlestick_data)})")
        return None

    try:
        # 1. 캔들스틱 데이터를 pandas DataFrame으로 변환
        # 빗썸 캔들스틱 데이터 구조: [타임스탬프, 시가, 종가, 고가, 저가, 거래량]
        df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])

        # 2. 숫자형 데이터로 변환 (API 응답이 문자열일 수 있으므로)
        df['close'] = pd.to_numeric(df['close'])

        # 3. '종가(close)'를 기준으로 이동평균 계산
        # rolling(window=period)는 기간만큼 데이터를 묶어주고, .mean()은 그 평균을 계산
        sma = df['close'].rolling(window=period).mean()

        return sma

    except Exception as e:
        print(f"⚠️ SMA 계산 중 오류 발생: {e}")
        return None

 

о SMA 함수 테스트 진행

if __name__ == '__main__':
    # 테스트용 캔들스틱 데이터
    # 실제 데이터는 [타임스탬프, 시가, 종가, 고가, 저가, 거래량] 순서
    period = 5
    test_data = [
        [1, 100, 105, 110, 95, 1000],  # 종가: 105
        [2, 105, 110, 115, 100, 1200],  # 종가: 110
        [3, 110, 115, 120, 105, 1100],  # 종가: 115
        [4, 115, 120, 125, 110, 1300],  # 종가: 120
        [5, 120, 125, 130, 115, 1400],  # 종가: 125 -> 5일 평균: (105+110+115+120+125)/5 = 115
        [6, 125, 130, 135, 120, 1500],  # 종가: 130 -> 5일 평균: (110+115+120+125+130)/5 = 120
    ]

    print("--- SMA 계산 테스트 (기간: 5) ---")
    sma_series = calculate_sma(test_data, period=period)
    if sma_series is not None:
        print("계산된 5일 이동평균선:")
        print(sma_series)

- 아래와 같이 출력되면 성공
- 0~3의 경우 data가 5일치 축적되지 않아 NaN으로 출력

 

3. 중간 테스트

- main_trader.py에서 두 모듈을 불러와 실제 데이터로 SMA를 구한다.

1). import 추가

import bithumb_api_client           # API 호출 담당
import technical_analyzer           # 기술적 지표 계산 담당

 

2). main부 제작

if __name__ == "__main__":
    # 1. 대상 코인 설정
    target_coin = "BTC"

    # 2. 빗썸에서 캔들스틱 데이터 가져오기 (24시간봉)
    print(f"--- {target_coin} 캔들스틱 데이터 가져오기 ---")
    candlestick_data = bithumb_api_client.get_candlestick_data(target_coin, chart_intervals="24h")

    if candlestick_data:
        # 3. 기술적 분석기로 이동평균선 계산
        print("\n--- 5일 및 20일 이동평균선 계산 ---")
        sma_5 = technical_analyzer.calculate_sma(candlestick_data, period=5)
        sma_20 = technical_analyzer.calculate_sma(candlestick_data, period=20)

        if sma_5 is not None and sma_20 is not None:
            # 4. 계산 결과 확인 (최근 5개 값만 출력)
            print("\n--- 최근 5일간의 종가 및 이동평균선 ---")
            # 깔끔한 데이터를 위해 pandas DataFrame으로 만들기
            import pandas as pd

            df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])
            df['close'] = pd.to_numeric(df['close'])  # 종가를 숫자형으로 변환
            df['SMA5'] = sma_5
            df['SMA20'] = sma_20

            print(df[['close', 'SMA5', 'SMA20']].tail(5))

            # TODO: 골든/데드 크로스 판단 로직 구현

- 아래와 같이 출력되면 성공

 

4. 골든/데드 크로스

- 위에서 구한 데이터를 이용하여 골든/데드 크로스 신호를 판단한다.

1). 크로스 신호 판단 함수

- main_trader.py에 다음 함수 추가

# --- 골든/데드 크로스 신호 판단 함수 ---
"""
주어진 DataFrame의 최근 데이터를 바탕으로 골든/데드 크로스 신호를 판단
:df: 'SMA5'와 'SMA20' 컬럼이 포함된 pandas DataFrame
:return: "BUY"(골든크로스), "SELL"(데드크로스), "HOLD"(교차 없음) 중 하나의 문자열
"""
def check_cross_signal(df):
    # 마지막 두 개의 데이터 (어제와 오늘)를 가져오기
    last_row = df.iloc[-1]  # 오늘
    prev_row = df.iloc[-2]  # 어제

    # 오늘과 어제의 5일, 20일 이동평균선 값
    sma5_today = last_row['SMA5']
    sma20_today = last_row['SMA20']
    sma5_yesterday = prev_row['SMA5']
    sma20_yesterday = prev_row['SMA20']

    # 골든 크로스: 5일선이 20일선을 아래에서 위로 돌파
    if sma5_yesterday < sma20_yesterday and sma5_today > sma20_today:
        return "BUY"  # 매수 신호

    # 데드 크로스: 5일선이 20일선을 위에서 아래로 돌파
    elif sma5_yesterday > sma20_yesterday and sma5_today < sma20_today:
        return "SELL"  # 매도 신호

    else:
        return "HOLD"  # 유지 (교차 없음)

 

2). 테스트 진행

- 3 -> 2)에서 작성한 main에 아르 내용을 추가하여 테스트를 진행한다.

# 골든/데드 크로스 판단 로직 구현
if len(df) > 20:
    signal = check_cross_signal(df)  # 위에서 만든 함수 호출

    print("\n--- 📈 매매 신호 판단 📉 ---")
    if signal == "BUY":
        print(f"[{target_coin}] 🟢 골든 크로스 발생! 매수 신호입니다!")
        # TODO: 여기에 portfolio_manager.record_vportfolio_trade('buy', ...) 호출 로직 추가!
    elif signal == "SELL":
        print(f"[{target_coin}] 🔴 데드 크로스 발생! 매도 신호입니다!")
        # TODO: 여기에 portfolio_manager.record_vportfolio_trade('sell', ...) 호출 로직 추가!
    else:
        print(f"[{target_coin}] ⚪️ 유지 신호입니다. (교차 없음)")
else:
    print("\n데이터가 충분하지 않아 매매 신호를 판단할 수 없습니다.")

- 아래와 같이 출력되면 성공

 

5. 종합 테스트

- 이때까지 진행한 소스를 이용해서 종합 테스트를 진행할 수 있게 되었습니다.
- main_trader.py을 수정하여 종합 테스트를 진행해보겠습니다.

1). import

import time                         # Loop에 사용

import virtual_portfolio_manager    # 가상 자산 관리 담당
import bithumb_api_client           # API 호출 담당
import technical_analyzer           # 기술적 지표 계산 담당
import trade_history                # 거래 내역 담당

 

2). main부 변경

- check_cross_signal(df)는 그대로 두고 main을 변경하여 테스트를 진행한다.

if __name__ == "__main__":
    # --- 봇 설정 ---
    target_coin = "BTC"
    check_interval_seconds = 60  # 체크 간격 (초 단위)

    # --- 봇 시작 ---
    trade_history.set_trade_mode(is_virtual=True)
    my_vportfolio = virtual_portfolio_manager.load_vportfolio()

    print("=" * 50)
    print("🚀 가상 자동매매 봇을 시작합니다! 🚀")
    print(f"대상 코인: {target_coin}, 체크 간격: {check_interval_seconds}초")
    print("봇을 중지하려면 Ctrl+C 를 누르세요.")
    print("=" * 50)

    while True:
        try:
            # --- 1. 시장 데이터 분석 및 신호 생성 ---
            print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 시장 데이터 분석 시작...")
            candlestick_data = bithumb_api_client.get_candlestick_data(target_coin,
                                                                       chart_intervals="1h")  # 예시: 1시간봉으로 변경

            signal = "HOLD"  # 기본 신호는 유지
            if candlestick_data and len(candlestick_data) > 20:
                import pandas as pd

                df = pd.DataFrame(candlestick_data, columns=['timestamp', 'open', 'close', 'high', 'low', 'volume'])
                df['close'] = pd.to_numeric(df['close'])
                df['SMA5'] = technical_analyzer.calculate_sma(candlestick_data, period=5)
                df['SMA20'] = technical_analyzer.calculate_sma(candlestick_data, period=20)

                signal = check_cross_signal(df)
            else:
                print("데이터가 충분하지 않아 매매 신호를 판단할 수 없습니다.")

            # --- 2. 신호에 따른 액션 수행 ---
            print(f"-> 현재 신호: {signal}")

            if signal == "BUY":
                # TODO: 이미 코인을 보유하고 있다면 추가 매수 안 하는 로직 추가하면 더 좋음
                print(f"[{target_coin}] 🟢 골든 크로스 발생! 매수 주문을 시도합니다.")
                cash_to_invest = my_vportfolio['cash'] * 0.1  # 가용 현금의 10% 투자

                # 현재가 조회
                ticker = bithumb_api_client.get_ticker_info(target_coin)
                if ticker:
                    current_price = float(ticker['closing_price'])
                    quantity_to_buy = cash_to_invest / current_price

                    # 가상 매수 실행
                    updated_portfolio = virtual_portfolio_manager.record_vportfolio_trade(my_vportfolio, "buy",
                                                                                          target_coin, quantity_to_buy,
                                                                                          current_price)
                    if updated_portfolio: my_vportfolio = updated_portfolio

            elif signal == "SELL":
                # TODO: 수익실현/손절매 로직과 통합하면 더 좋음
                print(f"[{target_coin}] 🔴 데드 크로스 발생! 매도 주문을 시도합니다.")
                quantity_to_sell = 0
                for coin in my_vportfolio["coins_owned"]:
                    if coin["symbol"] == target_coin.upper():
                        quantity_to_sell = coin["quantity"]
                        break

                if quantity_to_sell > 0:
                    ticker = bithumb_api_client.get_ticker_info(target_coin)
                    if ticker:
                        current_price = float(ticker['closing_price'])
                        # 가상 매도 실행
                        updated_portfolio = virtual_portfolio_manager.record_vportfolio_trade(my_vportfolio, "sell",
                                                                                              target_coin,
                                                                                              quantity_to_sell,
                                                                                              current_price)
                        if updated_portfolio: my_vportfolio = updated_portfolio
                else:
                    print(f"-> {target_coin} 보유 수량이 없어 매도하지 않습니다.")

            # --- 3. 다음 사이클까지 대기 ---
            print(f"\n분석 완료. {check_interval_seconds}초 후 다음 분석을 시작합니다...")
            time.sleep(check_interval_seconds)

        except KeyboardInterrupt:
            # 사용자가 Ctrl+C를 눌러서 프로그램을 중단시킬 때
            print("\n\n사용자에 의해 프로그램이 중단되었습니다. 최종 포트폴리오 상태를 저장합니다...")
            virtual_portfolio_manager.save_vportfolio(my_vportfolio)
            print("프로그램을 종료합니다.")
            break  # while 루프 탈출

        except Exception as e:
            # 예기치 못한 다른 에러 발생 시
            print(f"⚠️ 예상치 못한 오류 발생: {e}")
            print("60초 후 재시도합니다...")
            time.sleep(60)

- 다음과 같이 출력되며 반복되면 성공

- 테스트를 위해 현재 신호를 "BUY"로 설정하고 테스트 진행하면 다음과 같이 출력됩니다.

- 파이참에서 해당 루프 종료는 '정지' 버튼을 클릭하거나 Ctrl+F2를 누르면 됩니다.

 

드디어 이 단계까지 왔네요.
이제부터 본인이 생각하는 방법을 구현하여 기능을 강화하면 되는데요!
다음에 무엇을 진행하고 어떻게 개선할지는 고민을 해보도록 하겠습니다ㅎㅎ

 

다음 단계

• 시야 확장
• 전략 최적화
• 필요 시 GUI 구현

 

끝! 끝! 끝!

v1.0.0 끝!!!!!!!