import math import requests import pandas as pd import yfinance as yf import matplotlib.pyplot as plt plt.rcParams['font.family'] ='AppleGothic' plt.rcParams['axes.unicode_minus'] =False from config import * from stock_monitor import calculate_technical_indicators, detect_turnaround_signal # 비트/알트코인 KRW 마켓 식별: 문자열 "-KRW" 포함 여부로 간단 구분 INTERVAL_MAP = { 60: "60m", # 1시간 (yfinance) 240: "4h", # 4시간 (yfinance) } BITHUMB_MAX_COUNT = 3000 # API 최대 3000 캔들 def fetch_coin_history_bithumb(symbol: str, interval_minutes: int, days: int) -> pd.DataFrame: """빗썸 API를 이용해 최근 `days`일 코인 데이터 수집 (interval 60 / 240)""" if interval_minutes not in (60, 240): raise ValueError("Bithumb API only supports 60 or 240 minutes in this helper") minutes = interval_minutes count = int(math.ceil(days * 24 * 60 / minutes)) + 10 # 여유분 10 캔들 count = min(count, BITHUMB_MAX_COUNT) url = f"https://api.bithumb.com/v1/candles/minutes/{minutes}?market=KRW-{symbol}&count={count}" res = requests.get(url, timeout=5) res.raise_for_status() raw = res.json() if not isinstance(raw, list) or len(raw) == 0: raise RuntimeError("Empty response from Bithumb API") df_temp = pd.DataFrame(raw) # API 반환: [timestamp, open, close, high, low, volume] 순 df_temp = df_temp.sort_index(ascending=False) # 최신순, 뒤집어서 역순 전달 data = pd.DataFrame() # data.columns = ['datetime', 'open', 'close', 'high', 'low', 'volume'] # data['datetime'] = pd.to_datetime(data_temp['candle_date_time_kst']) data['datetime'] = pd.to_datetime(df_temp['candle_date_time_kst'], format='%Y-%m-%dT%H:%M:%S') data['Open'] = df_temp['opening_price'] data['Close'] = df_temp['trade_price'] data['High'] = df_temp['high_price'] data['Low'] = df_temp['low_price'] data['Volume'] = df_temp['candle_acc_trade_volume'] data = data.set_index('datetime') data = data.astype(float) data["datetime"] = data.index data = data.set_index("datetime").sort_index() # 시간 오름차순 return data def fetch_price_history(symbol: str, interval_minutes: int, days: int = 7) -> pd.DataFrame: """최근 `days`일 데이터(캔들)를 가져온다. 코인(-KRW)은 빗썸, 그 외 yfinance.""" if symbol in KR_COINS: base_symbol = symbol.replace("-KRW", "") return fetch_coin_history_bithumb(base_symbol, interval_minutes, days) # -------- 주식/ETF/해외코인 (yfinance) -------- if interval_minutes not in INTERVAL_MAP: raise ValueError("interval must be 60 or 240") interval_str = INTERVAL_MAP[interval_minutes] df = yf.download( tickers=symbol, period=f"{days}d", interval=interval_str, progress=False, ) if df.empty: raise RuntimeError("No data fetched. Check symbol or interval support.") return df def run_simulation(symbol: str, interval_minutes: int, days: int = 7): data = fetch_price_history(symbol, interval_minutes, days) data = calculate_technical_indicators(data) alerts = [] # (timestamp, price) # 시계열 순회하며 알림 조건 체크 for i in range(len(data)): slice_df = data.iloc[: i + 1] info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes) if info and info["alert"]: alerts.append((slice_df.index[-1], slice_df["Close"].iloc[-1])) # 모든 매수 신호를 표시 # 기존 필터 제거하여 전체 기간 매수 신호 사용 # Plot plt.figure(figsize=(12, 6)) plt.plot(data.index, data["Close"], label="종가", color="black") plt.plot(data.index, data["MA5"], label="MA5", color="orange", linewidth=1) plt.plot(data.index, data["MA20"], label="MA20", color="blue", linewidth=1) plt.plot(data.index, data["MA40"], label="MA40", color="green", linewidth=1) # Bollinger Bands plt.plot(data.index, data["Upper"], label="볼린저 Upper", color="grey", linestyle="--", linewidth=1) plt.plot(data.index, data["Lower"], label="볼린저 Lower", color="grey", linestyle="--", linewidth=1) plt.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1) if alerts: times, prices = zip(*alerts) plt.scatter(times, prices, facecolors='none', edgecolors='red', linewidths=2, s=150, zorder=6, label='매수신호') plt.title(f"{symbol} – 시뮬레이션 {interval_minutes}분봉 (최근 {days}일)") plt.xlabel("날짜") plt.ylabel("가격") plt.legend() plt.grid(True) plt.tight_layout() plt.show() if __name__ == "__main__": symbol = 'WLD' interval = 60 days = 7 run_simulation(symbol, interval, days)