테스트용 PR

test/binance_1m.py

@@ -1,35 +1,38 @@
-import requests
-import time
-import datetime
+"""
+이 모듈은 바이낸스 API에서 BTC/USDT의 과거 1분 봉 데이터를 가져옵니다.
+"""
 
-# 1년 전의 Unix Timestamp 계산 (밀리초 단위)
-one_year_ago = int((datetime.datetime.utcnow() - datetime.timedelta(days=365)).timestamp() * 1000)
+from datetime import datetime, timedelta
+import requests
 
-# 현재 시간의 Unix Timestamp (밀리초 단위)
-now = int(time.time() * 1000)
+# Binance API 엔드포인트 (상수는 대문자로)
+BINANCE_API_URL = "https://api.binance.com/api/v3/klines"
 
-# 바이낸스 API 엔드포인트
-url = "https://api.binance.com/api/v3/klines"
+# 1년 전과 현재 시간의 Unix Timestamp (밀리초 단위)
+ONE_YEAR_AGO = int((datetime.utcnow() - timedelta(days=365)).timestamp() * 1000)
+NOW = int(datetime.utcnow().timestamp() * 1000)
 
 # 요청 파라미터 설정
-params = {
-    "symbol": "BTCUSDT",  # 원하는 거래쌍
-    "interval": "1m",  # 1일 간격의 데이터
-    "startTime": one_year_ago,  # 1년 전
-    "endTime": now,  # 현재 시간
-    "limit": 1000  # 최대 1000개 데이터 (필요시 반복 요청)
+PARAMS = {
+    "symbol": "BTCUSDT",
+    "interval": "1m",
+    "startTime": ONE_YEAR_AGO,
+    "endTime": NOW,
+    "limit": 1000
 }
 
-# API 요청
-response = requests.get(url, params=params)
+def fetch_binance_data():
+    """Binance API에서 1년 전부터 현재까지의 1분 봉 데이터를 가져온다."""
+    response = requests.get(BINANCE_API_URL, params=PARAMS, timeout=10)
+    if response.status_code == 200:
+        data = response.json()
+        for candle in data:
+            timestamp = datetime.utcfromtimestamp(candle[0] / 1000)  # UTC 변환
+            open_price = candle[1]
+            close_price = candle[4]
+            print(f"{timestamp}: Open={open_price}, Close={close_price}")
+    else:
+        print(f"Error {response.status_code}: {response.text}")
 
-# 응답 데이터 확인
-if response.status_code == 200:
-    data = response.json()
-    for candle in data:
-        timestamp = datetime.datetime.utcfromtimestamp(candle[0] / 1000)
-        open_price = candle[1]
-        close_price = candle[4]
-        print(f"{timestamp}: Open={open_price}, Close={close_price}")
-else:
-    print(f"Error: {response.status_code}, {response.text}")
+if __name__ == "__main__":
+    fetch_binance_data()

test/upbit_1m.py

@@ -1,38 +1,42 @@
-import requests
-import datetime
+"""
+이 모듈은 Upbit API를 사용하여 BTC/KRW 1분 봉 데이터를 가져오는 기능을 제공합니다.
+"""
+
+from datetime import datetime, timedelta
 import time
+import requests
 
-# Upbit API 엔드포인트
-UPBIT_API_URL = "https://api.upbit.com/v1/candles/minutes/1"  # 1분봉 캔들 데이터
+# Upbit API 엔드포인트 (상수는 대문자로)
+UPBIT_API_URL = "https://api.upbit.com/v1/candles/minutes/1"
 
 # 심볼 설정 (BTC/KRW)
-symbol = "KRW-BTC"
+SYMBOL = "KRW-BTC"
 
-# 1년 전의 특정 날짜 기준으로 하루 동안의 데이터를 가져오기
-target_date = datetime.datetime(2024, 1, 1, 0, 0, 0)  # 원하는 날짜 설정 (UTC 기준)
-start_date = target_date
-end_date = start_date + datetime.timedelta(days=1)  # 하루치 데이터 수집
+# 1년 전 특정 날짜의 하루 동안 데이터를 가져오기
+TARGET_DATE = datetime(2024, 1, 1, 0, 0, 0)  # 원하는 날짜 설정 (UTC 기준)
+START_DATE = TARGET_DATE
+END_DATE = START_DATE + timedelta(days=1)  # 하루치 데이터 수집
 
-# Upbit API의 `to` 파라미터는 UTC 기준 RFC 3339 형식으로 설정해야 함
-to = end_date.strftime('%Y-%m-%dT%H:%M:%S')  # UTC 기준
+# Upbit API의 `to` 파라미터는 UTC 기준 RFC 3339 형식 사용
+TO = END_DATE.strftime('%Y-%m-%dT%H:%M:%S')  # UTC 기준
 
 # 수집된 데이터 리스트
 all_ohlcv = []
 
-print(f"📌 요청한 시작 날짜 (UTC): {start_date.strftime('%Y-%m-%d %H:%M:%S')}")
-print(f"📌 요청한 종료 날짜 (UTC): {end_date.strftime('%Y-%m-%d %H:%M:%S')}\n")
+print(f"📌 요청한 시작 날짜 (UTC): {START_DATE.strftime('%Y-%m-%d %H:%M:%S')}")
+print(f"📌 요청한 종료 날짜 (UTC): {END_DATE.strftime('%Y-%m-%d %H:%M:%S')}\n")
 
 while True:
     # API 요청 파라미터
     params = {
-        "market": symbol,
-        "to": to,
+        "market": SYMBOL,
+        "to": TO,
         "count": 200,  # 최대 200개 요청 가능
     }
 
     headers = {"accept": "application/json"}
 
-    response = requests.get(UPBIT_API_URL, params=params, headers=headers)
+    response = requests.get(UPBIT_API_URL, params=params, headers=headers, timeout=10)
 
     if response.status_code != 200:
         print(f"❌ API 요청 실패: {response.status_code}, {response.text}")
@@ -47,13 +51,16 @@
     ohlcv.reverse()
 
     # start_date 이전의 데이터는 필터링
-    filtered_ohlcv = [candle for candle in ohlcv if candle["candle_date_time_utc"] >= start_date.strftime('%Y-%m-%dT%H:%M:%SZ')]
+    filtered_ohlcv = [
+        candle for candle in ohlcv
+        if candle["candle_date_time_utc"] >= START_DATE.strftime('%Y-%m-%dT%H:%M:%SZ')
+    ]
 
     all_ohlcv.extend(filtered_ohlcv)
 
     # 가장 오래된 데이터의 시간을 다음 요청의 기준으로 설정 (UTC)
     if filtered_ohlcv:
-        to = filtered_ohlcv[0]["candle_date_time_utc"] + "Z"
+        TO = filtered_ohlcv[0]["candle_date_time_utc"] + "Z"
     else:
         break  # 더 이상 가져올 데이터가 없으면 종료
 
@@ -66,13 +73,26 @@
     first_candle = all_ohlcv[0]
     last_candle = all_ohlcv[-1]
 
-    print(f"📌 가장 과거 데이터: {first_candle['candle_date_time_kst']}, 시가: {first_candle['opening_price']}, 종가: {first_candle['trade_price']}")
-    print(f"📌 가장 최근 데이터: {last_candle['candle_date_time_kst']}, 시가: {last_candle['opening_price']}, 종가: {last_candle['trade_price']}\n")
+    print(
+        f"📌 가장 과거 데이터: {first_candle['candle_date_time_kst']}, "
+        f"시가: {first_candle['opening_price']}, 종가: {first_candle['trade_price']}"
+    )
+    print(
+        f"📌 가장 최근 데이터: {last_candle['candle_date_time_kst']}, "
+        f"시가: {last_candle['opening_price']}, 종가: {last_candle['trade_price']}\n"
+    )
 
     # 샘플 데이터 5개 출력
     print("📊 수집된 데이터 샘플 (최근 5개):")
     for candle in all_ohlcv[-5:]:
-        print(f"{candle['candle_date_time_kst']}, 시가: {candle['opening_price']}, 고가: {candle['high_price']}, 저가: {candle['low_price']}, 종가: {candle['trade_price']}, 거래량: {candle['candle_acc_trade_volume']}")
+        print(
+            f"{candle['candle_date_time_kst']}, "
+            f"시가: {candle['opening_price']}, "
+            f"고가: {candle['high_price']}, "
+            f"저가: {candle['low_price']}, "
+            f"종가: {candle['trade_price']}, "
+            f"거래량: {candle['candle_acc_trade_volume']}"
+        )
 
 else:
     print("❌ 데이터를 가져오지 못했습니다.")