OKX 거래소 API로加密货币历史行情数据获取：加密货币历史行情回测教程

⚠️ 주의: 제목의 "加密货币历史行情数据获取"은 검색 최적화를 위한 키워드입니다. 본문부터 모든 내용은 한국어로만 작성됩니다.

서론: 왜加密货币历史行情数据获取가 필요한가?

저는 지난 3개월간 비트코인 선물 전략의 과거 데이터 검증 작업을 진행했습니다.最初は日本の取引所のAPIを使おうとしましたが言語の壁と手数料の問題で苦労しました.다행히 HolySheep AI의 전역 API 게이트웨이를 통해 OKX, Binance, Bybit 등 주요 거래소의 데이터를统一的 방식으로 수집할 수 있었습니다.

이 튜토리얼에서는:

• OKX REST API로 역사적 시세 데이터 가져오기
• 수집된 데이터로 트레이딩 전략 백테스트 수행하기
• HolySheep AI로 시장 분석 자동화하기

를 순서대로 진행하겠습니다.

1. OKX API 설정

1.1 API 키 발급

1. OKX 거래소 회원가입 및 실명인증
2. マイページ → API管理 → API 키 생성
3. 권한 설정: Read Only (데이터 조회만 필요)
4. Passphrase, API Key, Secret Key 안전한 곳에 보관

1.2 필수 라이브러리 설치

# Python 3.8 이상 권장
pip install requests pandas numpy matplotlib python-dotenv

백테스트용 추가 라이브러리
pip install backtrader ta

HolySheep AI 연동용
pip install openai

2. OKX REST API로 역사적 시세 데이터 수집

2.1 기초 데이터 조회 함수

import requests
import pandas as pd
import time
from datetime import datetime, timedelta
from dotenv import load_dotenv
import os

.env 파일에서 API 키 로드 (보안 권장)
load_dotenv()
OKX_API_KEY = os.getenv("OKX_API_KEY", "your_okx_api_key")
OKX_SECRET_KEY = os.getenv("OKX_SECRET_KEY", "your_okx_secret_key")
OKX_PASSPHRASE = os.getenv("OKX_PASSPHRASE", "your_passphrase")

OKX API 기본 URL
OKX_BASE_URL = "https://www.okx.com"

def get_klines(symbol="BTC-USDT", timeframe="1H", limit=100, after=None):
    """
    OKX에서 캔들스틱(ohlcv) 데이터 조회
    
    Args:
        symbol: 거래쌍 (예: BTC-USDT, ETH-USDT)
        timeframe: 시간프레임 (1m, 5m, 1H, 4H, 1D)
        limit: 조회 개수 (최대 100)
        after: 페이지네이션용 타임스탬프
    
    Returns:
        DataFrame: 날짜,시가,고가,저가,종가,거래량
    """
    endpoint = "/api/v5/market/candles"
    url = f"{OKX_BASE_URL}{endpoint}"
    
    params = {
        "instId": symbol,
        "bar": timeframe,
        "limit": limit
    }
    
    if after:
        params["after"] = after
    
    headers = {
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.get(url, headers=headers, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        if data.get("code") != "0":
            print(f"API 오류: {data.get('msg')}")
            return None
        
        candles = data.get("data", [])
        
        if not candles:
            print("데이터가 없습니다.")
            return None
        
        # 데이터 가공: [timestamp, open, high, low, close, volume]
        df = pd.DataFrame(candles, columns=[
            "timestamp", "open", "high", "low", "close", "volume",
            "quote_vol", "num_trades", "buy_vol", "confirm"
        ])
        
        # 숫자형 변환
        for col in ["open", "high", "low", "close", "volume"]:
            df[col] = pd.to_numeric(df[col], errors='coerce')
        
        # 타임스탬프 변환
        df['datetime'] = pd.to_datetime(df['timestamp'].astype(float), unit='ms')
        df = df.sort_values('datetime').reset_index(drop=True)
        
        return df[['datetime', 'open', 'high', 'low', 'close', 'volume']]
    
    except requests.exceptions.RequestException as e:
        print(f"네트워크 오류: {e}")
        return None

사용 예시
if __name__ == "__main__":
    btc_data = get_klines("BTC-USDT", "1H", 100)
    
    if btc_data is not None:
        print(f"BTC-USDT 최근 {len(btc_data)}개 데이터 조회 성공")
        print(btc_data.tail())
        
        # CSV 저장
        btc_data.to_csv("btc_1h.csv", index=False)
        print("btc_1h.csv 파일로 저장 완료")

2.2 과거 데이터 대량 수집 (폴백테스트용)

import requests
import pandas as pd
from datetime import datetime, timedelta
import time

def get_historical_klines(symbol="BTC-USDT", timeframe="1H", 
                          start_date=None, end_date=None,
                          max_requests_per_minute=20):
    """
    지정한 기간의 모든 истори 데이터 수집
    
    Args:
        symbol: 거래쌍
        timeframe: 시간프레임
        start_date: 시작일 (YYYY-MM-DD)
        end_date: 종료일 (YYYY-MM-DD)
        max_requests_per_minute: 분당 요청 제한 (OKX는 20회/분)
    
    Returns:
        DataFrame: 전체 기간 데이터
    """
    all_data = []
    
    # 종료일이 없으면 현재 시각으로 설정
    if end_date:
        end_dt = datetime.strptime(end_date, "%Y-%m-%d")
    else:
        end_dt = datetime.utcnow()
    
    # 시작일이 없으면 3개월 전으로 설정
    if start_date:
        start_dt = datetime.strptime(start_date, "%Y-%m-%d")
    else:
        start_dt = end_dt - timedelta(days=90)
    
    # 종료 타임스탬프 (밀리초)
    end_timestamp = int(end_dt.timestamp() * 1000)
    start_timestamp = int(start_dt.timestamp() * 1000)
    
    current_after = end_timestamp
    
    request_count = 0
    minute_start = time.time()
    
    print(f"{symbol} {timeframe} 데이터 수집 시작")
    print(f"기간: {start_date or start_dt.strftime('%Y-%m-%d')} ~ {end_date or end_dt.strftime('%Y-%m-%d')}")
    
    while current_after > start_timestamp:
        # 분당 요청 수 제한
        request_count += 1
        elapsed = time.time() - minute_start
        
        if request_count >= max_requests_per_minute:
            if elapsed < 60:
                sleep_time = 60 - elapsed
                print(f"속도 제한 도달. {sleep_time:.1f}초 대기...")
                time.sleep(sleep_time)
                minute_start = time.time()
                request_count = 0
            else:
                minute_start = time.time()
                request_count = 0
        
        url = "https://www.okx.com/api/v5/market/candles"
        params = {
            "instId": symbol,
            "bar": timeframe,
            "limit": 100,
            "after": current_after  # 이 시간 이전 데이터 조회
        }
        
        try:
            response = requests.get(url, params=params, timeout=10)
            
            if response.status_code == 429:
                print("속도 제한 초과. 10초 대기...")
                time.sleep(10)
                continue
            
            data = response.json()
            
            if data.get("code") != "0":
                print(f"API 오류: {data.get('msg')}")
                break
            
            candles = data.get("data", [])
            
            if not candles:
                print("더 이상 데이터가 없습니다.")
                break
            
            # 데이터 저장
            for candle in candles:
                ts = int(candle[0])
                if ts < start_timestamp:
                    break
                all_data.append({
                    "timestamp": ts,
                    "datetime": pd.to_datetime(ts, unit='ms'),
                    "open": float(candle[1]),
                    "high": float(candle[2]),
                    "low": float(candle[3]),
                    "close": float(candle[4]),
                    "volume": float(candle[5])
                })
            
            # 다음 페이지 조회를 위한 after 값 갱신
            current_after = int(candles[-1][0]) - 1
            
            print(f"수집 중... {len(all_data)}개 레코드 (마지막: {pd.to_datetime(current_after, unit='ms')})")
            
            # 서버 부하 방지
            time.sleep(0.1)
        
        except Exception as e:
            print(f"오류 발생: {e}")
            time.sleep(5)
    
    if all_data:
        df = pd.DataFrame(all_data)
        df = df.sort_values('datetime').reset_index(drop=True)
        print(f"\n총 {len(df)}개 레코드 수집 완료!")
        return df
    else:
        print("수집된 데이터가 없습니다.")
        return None

사용 예시: 최근 6개월 BTC 1시간봉 데이터
if __name__ == "__main__":
    historical_btc = get_historical_klines(
        symbol="BTC-USDT",
        timeframe="1H",
        end_date="2025-01-15",
        start_date="2024-07-15"
    )
    
    if historical_btc is not None:
        historical_btc.to_csv("btc_historical_6months.csv", index=False)
        print(f"\n파일 저장 완료: btc_historical_6months.csv")
        print(f"데이터 범위: {historical_btc['datetime'].min()} ~ {historical_btc['datetime'].max()}")

3. 백테스트 시스템 구축

3.1 간단한 RSI 기반 트레이딩 전략

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime

class Backtester:
    """
    단순 이동평균 교차 및 RSI 기반 백테스트 엔진
    """
    
    def __init__(self, initial_balance=10000):
        self.initial_balance = initial_balance
        self.balance = initial_balance
        self.position = 0  # 보유 수량
        self.trades = []
        self.equity_curve = []
        
    def calculate_indicators(self, df):
        """기술적 지표 계산"""
        # 단순 이동평균
        df['sma_fast'] = df['close'].rolling(window=20).mean()
        df['sma_slow'] = df['close'].rolling(window=50).mean()
        
        # RSI 계산
        delta = df['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        df['rsi'] = 100 - (100 / (1 + rs))
        
        return df
    
    def run_strategy(self, df, strategy="sma_crossover"):
        """
        전략 실행
        
        Args:
            df: 시세 데이터
            strategy: 'sma_crossover' 또는 'rsi'
        """
        df = self.calculate_indicators(df.copy())
        
        initial_price = df['close'].iloc[0]
        
        for i, row in df.iterrows():
            current_price = row['close']
            
            # 현재 수익률 계산
            current_equity = self.balance + (self.position * current_price)
            self.equity_curve.append({
                'datetime': row['datetime'],
                'equity': current_equity
            })
            
            if strategy == "sma_crossover":
                self._sma_crossover_signal(row, i, df)
            elif strategy == "rsi":
                self._rsi_signal(row, i, df)
        
        return self.get_results()
    
    def _sma_crossover_signal(self, row, i, df):
        """이동평균 교차 신호"""
        if i < 50:
            return
            
        current_price = row['close']
        sma_fast = row['sma_fast']
        sma_slow = row['sma_slow']
        
        # 골든크로스 (매수 신호)
        if sma_fast > sma_slow and df.iloc[i-1]['sma_fast'] <= df.iloc[i-1]['sma_slow']:
            if self.position == 0:
                buy_amount = self.balance * 0.95  # 5% 현금 보유
                self.position = buy_amount / current_price
                self.balance -= buy_amount
                self.trades.append({
                    'datetime': row['datetime'],
                    'type': 'BUY',
                    'price': current_price,
                    'amount': self.position,
                    'reason': 'Golden Cross'
                })
        
        # 데드크로스 (매도 신호)
        elif sma_fast < sma_slow and df.iloc[i-1]['sma_fast'] >= df.iloc[i-1]['sma_slow']:
            if self.position > 0:
                sell_amount = self.position * current_price
                self.balance += sell_amount
                self.trades.append({
                    'datetime': row['datetime'],
                    'type': 'SELL',
                    'price': current_price,
                    'amount': self.position,
                    'reason': 'Dead Cross',
                    'profit': sell_amount - (self.trades[-1]['price'] * self.position) if self.trades else 0
                })
                self.position = 0
    
    def _rsi_signal(self, row, i, df):
        """RSI 반등 전략 (RSI < 30 매수, RSI > 70 매도)"""
        if i < 20:
            return
        
        current_price = row['close']
        rsi = row['rsi']
        
        # 과매도 구간 → 매수
        if rsi < 30 and self.position == 0:
            buy_amount = self.balance * 0.95
            self.position = buy_amount / current_price
            self.balance -= buy_amount
            self.trades.append({
                'datetime': row['datetime'],
                'type': 'BUY',
                'price': current_price,
                'amount': self.position,
                'reason': f'RSI Oversold ({rsi:.1f})'
            })
        
        # 과매수 구간 → 매도
        elif rsi > 70 and self.position > 0:
            last_buy = None
            for trade in reversed(self.trades):
                if trade['type'] == 'BUY':
                    last_buy = trade
                    break
            
            sell_amount = self.position * current_price
            self.balance += sell_amount
            profit = 0
            if last_buy:
                profit = sell_amount - (last_buy['price'] * self.position)
            
            self.trades.append({
                'datetime': row['datetime'],
                'type': 'SELL',
                'price': current_price,
                'amount': self.position,
                'reason': f'RSI Overbought ({rsi:.1f})',
                'profit': profit
            })
            self.position = 0
    
    def get_results(self):
        """백테스트 결과 요약"""
        trades_df = pd.DataFrame(self.trades)
        
        # 수익률 계산
        total_return = ((self.balance + self.position * (trades_df['price'].iloc[-1] if len(trades_df) > 0 else 0)) 
                        - self.initial_balance) / self.initial_balance * 100
        
        # 승률 계산
        if len(trades_df) > 0:
            buy_trades = trades_df[trades_df['type'] == 'BUY']
            sell_trades = trades_df[trades_df['type'] == 'SELL']
            
            profitable_trades = trades_df[trades_df.get('profit', 0) > 0]
            win_rate = len([t for t in self.trades if t.get('profit', 0) > 0]) / max(len([t for t in self.trades if t.get('profit', 0) != 0]), 1) * 100
        else:
            win_rate = 0
        
        return {
            'initial_balance': self.initial_balance,
            'final_balance': self.balance + self.position * (trades_df['price'].iloc[-1] if len(trades_df) > 0 else 0),
            'total_return': total_return,
            'total_trades': len(self.trades),
            'win_rate': win_rate,
            'trades': trades_df,
            'equity_curve': pd.DataFrame(self.equity_curve)
        }
    
    def plot_results(self, results):
        """결과 시각화"""
        equity_df = results['equity_curve']
        
        plt.figure(figsize=(14, 10))
        
        # equity 차트
        plt.subplot(3, 1, 1)
        plt.plot(equity_df['datetime'], equity_df['equity'], label='Equity', color='blue')
        plt.axhline(y=self.initial_balance, color='gray', linestyle='--', label='Initial')
        plt.title('Equity Curve')
        plt.xlabel('Date')
        plt.ylabel('Balance (USDT)')
        plt.legend()
        plt.grid(True)
        
        # 거래 내역
        trades_df = results['trades']
        if len(trades_df) > 0:
            plt.subplot(3, 1, 2)
            buy_trades = trades_df[trades_df['type'] == 'BUY']
            sell_trades = trades_df[trades_df['type'] == 'SELL']
            
            plt.scatter(buy_trades['datetime'], buy_trades['price'], 
                       marker='^', color='green', s=100, label='Buy', zorder=5)
            plt.scatter(sell_trades['datetime'], sell_trades['price'], 
                       marker='v', color='red', s=100, label='Sell', zorder=5)
            plt.title('Trade Points')
            plt.legend()
            plt.grid(True)
        
        plt.tight_layout()
        plt.savefig('backtest_results.png', dpi=150)
        plt.show()

실행 예시
if __name__ == "__main__":
    # 데이터 로드
    df = pd.read_csv("btc_historical_6months.csv")
    
    # 백테스트 실행
    bt = Backtester(initial_balance=10000)
    results = bt.run_strategy(df, strategy="sma_crossover")
    
    # 결과 출력
    print("\n" + "="*50)
    print("백테스트 결과 요약")
    print("="*50)
    print(f"초기 자본: ${results['initial_balance']:,.2f}")
    print(f"최종 자본: ${results['final_balance']:,.2f}")
    print(f"총 수익률: {results['total_return']:.2f}%")
    print(f"총 거래 수: {results['total_trades']}")
    print(f"승률: {results['win_rate']:.1f}%")
    
    # 차트 그리기
    bt.plot_results(results)
    print("\n결과 차트 저장: backtest_results.png")

4. HolySheep AI로 시장 분석 자동화

저는 백테스트 결과를 HolySheep AI에 전달하여 자동 분석 리포트를 생성합니다.아래는 실제 제가 사용 중인 코드입니다:

import os
from openai import OpenAI

HolySheep AI API 설정
⚠️ 반드시 https://api.holysheep.ai/v1 을 사용하세요
client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

def generate_analysis_report(backtest_results, crypto_name="비트코인"):
    """
    HolySheep AI를 사용하여 백테스트 결과를 분석하고 리포트 생성
    
    Args:
        backtest_results: Backtester.get_results() 반환값
        crypto_name: 암호화폐 이름
    
    Returns:
        str: AI가 생성한 분석 리포트
    """
    
    # 결과 데이터를 요약하여 프롬프트에 포함
    summary_prompt = f"""
    다음은 {crypto_name} 트레이딩 전략 백테스트 결과입니다:
    
    - 초기 자본: ${backtest_results['initial_balance']:,.2f}
    - 최종 자본: ${backtest_results['final_balance']:,.2f}
    - 총 수익률: {backtest_results['total_return']:.2f}%
    - 총 거래 수: {backtest_results['total_trades']}회
    - 승률: {backtest_results['win_rate']:.1f}%
    
    거래 내역:
    {backtest_results['trades'].to_string() if len(backtest_results['trades']) > 0 else '거래 없음'}
    
    위 결과를 바탕으로 다음 항목을 분석해주세요:
    1. 전략의 강점과 약점
    2. 개선 가능한 포인트
    3. 리스크 분석
    4. 향후 투자 전략 제안
    
    한국어로 상세하게 작성해주세요.
    """
    
    try:
        response = client.chat.completions.create(
            model="gpt-4.1",  # HolySheep에서 제공하는 모델
            messages=[
                {
                    "role": "system", 
                    "content": "당신은 숙련된 암호화폐 트레이딩 분석 전문가입니다."
                },
                {
                    "role": "user", 
                    "content": summary_prompt
                }
            ],
            temperature=0.7,
            max_tokens=2000
        )
        
        report = response.choices[0].message.content
        
        # 사용량 정보 출력 (HolySheep 대시보드에서도 확인 가능)
        if hasattr(response, 'usage'):
            print(f"토큰 사용량: {response.usage.total_tokens}")
            print(f"추정 비용: ${response.usage.total_tokens / 1_000_000 * 8:.4f}")  # GPT-4.1 기준
        
        return report
    
    except Exception as e:
        print(f"AI 분석 중 오류 발생: {e}")
        return None

def generate_trading_signal(df_recent, holy_client):
    """
    최근 데이터 기반 매매 신호 생성
    
    Args:
        df_recent: 최근 시세 데이터
        holy_client: HolySheep AI 클라이언트
    
    Returns:
        dict: 매매 신호 및 이유
    """
    
    # 최근 24시간 데이터 요약
    recent_summary = f"""
    최근 24시간 {len(df_recent)}개 봉 데이터:
    - 시가: ${df_recent['open'].iloc[0]:,.2f}
    - 종가: ${df_recent['close'].iloc[-1]:,.2f}
    - 고가: ${df_recent['high'].max():,.2f}
    - 저가: ${df_recent['low'].min():,.2f}
    - 평균 거래량: {df_recent['volume'].mean():,.2f}
    - 변동률: {((df_recent['close'].iloc[-1] / df_recent['open'].iloc[0]) - 1) * 100:.2f}%
    """
    
    prompt = f"""
    {recent_summary}
    
    위 데이터를 분석하여:
    1. 현재 시장 분위기 (강세/약세/중립)
    2. 단기 매매 신호 (매수/매도/관망)
    3. 주요 참고 사항
    
    을 JSON 형식으로 답변해주세요:
    {{"sentiment": "...", "signal": "...", "reason": "..."}}
    """
    
    try:
        response = holy_client.chat.completions.create(
            model="gpt-4.1",
            messages=[
                {"role": "system", "content": "당신은 실시간 암호화폐 시장 분석 전문가입니다."},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"},
            temperature=0.3,
            max_tokens=500
        )
        
        import json
        result = json.loads(response.choices[0].message.content)
        return result
    
    except Exception as e:
        print(f"신호 생성 오류: {e}")
        return None

실행 예시
if __name__ == "__main__":
    # 백테스트 결과 로드
    results = {
        'initial_balance': 10000,
        'final_balance': 12450,
        'total_return': 24.5,
        'total_trades': 28,
        'win_rate': 64.3,
        'trades': pd.DataFrame([
            {'datetime': '2024-08-01', 'type': 'BUY', 'price': 65000, 'amount': 0.15},
            {'datetime': '2024-08-05', 'type': 'SELL', 'price': 68500, 'amount': 0.15, 'profit': 525}
        ])
    }
    
    # 분석 리포트 생성
    print("HolySheep AI가 백테스트 결과를 분석 중입니다...")
    report = generate_analysis_report(results, "비트코인")
    
    if report:
        print("\n" + "="*50)
        print("AI 분석 리포트")
        print("="*50)
        print(report)
        
        # 리포트 저장
        with open("analysis_report.txt", "w", encoding="utf-8") as f:
            f.write(report)
        print("\n리포트 저장: analysis_report.txt")

5. 전체 데이터 파이프라인

#!/usr/bin/env python3
"""
OKX + HolySheep AI 통합 트레이딩 분석 시스템
作者: HolySheep AI 기술 블로그
"""

import pandas as pd
import schedule
import time
import os
from datetime import datetime

모듈 import
from okx_api import get_klines, get_historical_klines
from backtester import Backtester
from holysheep_analysis import generate_analysis_report, generate_trading_signal, client

def daily_analysis():
    """매일 실행되는 분석 루틴"""
    print(f"\n{'='*50}")
    print(f"분석 시작: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print('='*50)
    
    # 1단계: 최신 데이터 수집
    print("\n[1/4] BTC-USDT 데이터 수집 중...")
    btc_data = get_klines("BTC-USDT", "1H", 100)
    
    if btc_data is None:
        print("데이터 수집 실패. 다음 실행 때 재시도...")
        return
    
    # 2단계: 백테스트 실행
    print("\n[2/4] 백테스트 실행 중...")
    bt = Backtester(initial_balance=10000)
    results = bt.run_strategy(btc_data, strategy="sma_crossover")
    
    print(f"   - 수익률: {results['total_return']:.2f}%")
    print(f"   - 거래 수: {results['total_trades']}")
    
    # 3단계: HolySheep AI 분석
    print("\n[3/4] HolySheep AI 분석 요청 중...")
    report = generate_analysis_report(results, "비트코인")
    
    if report:
        print("   - 분석 완료!")
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        with open(f"reports/report_{timestamp}.txt", "w", encoding="utf-8") as f:
            f.write(report)
    
    # 4단계: 실시간 신호 생성
    print("\n[4/4] 실시간 신호 생성 중...")
    signal = generate_trading_signal(btc_data, client)
    
    if signal:
        print(f"   - 시장 분위기: {signal.get('sentiment', 'N/A')}")
        print(f"   - 매매 신호: {signal.get('signal', 'N/A')}")
        print(f"   - 이유: {signal.get('reason', 'N/A')[:100]}...")
    
    print(f"\n분석 완료: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

def main():
    """메인 실행 함수"""
    # 리포트 저장 디렉토리 생성
    os.makedirs("reports", exist_ok=True)
    
    print("OKX + HolySheep AI 트레이딩 분석 시스템 시작")
    print(" HolySheep AI API: https://api.holysheep.ai/v1")
    
    # 즉시 1회 실행
    daily_analysis()
    
    # 매일 오전 9시, 오후 6시에 실행
    schedule.every().day.at("09:00").do(daily_analysis)
    schedule.every().day.at("18:00").do(daily_analysis)
    
    print("\n스케줄러 실행 중... (Ctrl+C로 종료)")
    
    while True:
        schedule.run_pending()
        time.sleep(60)

if __name__ == "__main__":
    main()

자주 발생하는 오류와 해결책

# 요청 사이에 딜레이 추가
import time

def safe_api_call():
    for i in range(10):
        response = requests.get(url)
        if response.status_code == 429:
            print("속도 제한. 5초 대기...")
            time.sleep(5)
        else:
            return response.json()
    raise Exception("API 호출 실패")

# 올바른 설정 확인
import os
from openai import OpenAI

❌ 잘못된 설정
client = OpenAI(api_key="...", base_url="https://api.openai.com/v1")

✅ 올바른 설정
client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # HolySheep 전용 URL
)

# 명시적 타입 변환
df['close'] = df['close'].astype(float)
df['volume'] = pd.to_numeric(df['volume'], errors='coerce')

결측치 처리
df = df.dropna(subset=['close', 'volume'])

# 결측치 확인 및 처리
print(df.isnull().sum())

결측치 앞뒤 값으로 보간
df = df.interpolate(method='linear')

또는 결측치가 많은 구간 제거
df = df.dropna()

다시 지표 계산
df = bt.calculate_indicators(df)

# UTC로 통일하여 처리
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
df['datetime'] = df['datetime'].dt.tz_convert('Asia/Seoul')  # 한국 시간대

시각화 시 시간대 명시
plt.gca().xaxis.set_major_formatter(
    mdates.DateFormatter('%Y-%m-%d %H:%M', tz='Asia/Seoul')
)

거래소 API 비교