import time from datetime import datetime from dateutil.relativedelta import relativedelta import pandas as pd import yfinance as yf import matplotlib.pyplot as plt import matplotlib.dates import mplcursors plt.rcParams['font.family'] ='AppleGothic' plt.rcParams['axes.unicode_minus'] =False from config import * from stock_monitor import calculate_technical_indicators, detect_turnaround_signal, get_coin_more_data, check_buy_point # 비트/알트코인 KRW 마켓 식별: 문자열 "-KRW" 포함 여부로 간단 구분 INTERVAL_MAP = { 60: "60m", # 1시간 (yfinance) 240: "4h", # 4시간 (yfinance) } BITHUMB_MAX_COUNT = 3000 # API 최대 3000 캔들 def fetch_price_history(symbol: str, interval_minutes: int, days: int = 30) -> pd.DataFrame: """최근 `days`일 데이터(캔들)를 가져온다. 코인(-KRW)은 빗썸, 그 외 yfinance.""" if symbol in KR_COINS: bong_count = 3000 return get_coin_more_data(symbol, interval_minutes, bong_count=bong_count) # -------- 주식/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 analyze_bottom_period(symbol: str, interval_minutes: int, days: int = 90): """저점 기간(6월 22일~7월 9일) 분석""" data = fetch_price_history(symbol, interval_minutes, days) data = calculate_technical_indicators(data) print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}") print(f"총 데이터 수: {len(data)}") # 저점 기간 필터링 (6월 22일~7월 9일) bottom_start = pd.Timestamp('2025-06-22') bottom_end = pd.Timestamp('2025-07-09') bottom_data = data[(data.index >= bottom_start) & (data.index <= bottom_end)] if len(bottom_data) == 0: print("저점 기간 데이터가 없습니다.") return print(f"\n저점 기간 데이터: {bottom_data.index[0]} ~ {bottom_data.index[-1]}") print(f"저점 기간 데이터 수: {len(bottom_data)}") # 저점 기간의 기술적 지표 분석 print("\n=== 저점 기간 기술적 지표 분석 ===") # 1. 가격 분석 min_price = bottom_data['Low'].min() max_price = bottom_data['High'].max() avg_price = bottom_data['Close'].mean() print(f"최저가: {min_price:.4f}") print(f"최고가: {max_price:.4f}") print(f"평균가: {avg_price:.4f}") print(f"가격 변동폭: {((max_price - min_price) / min_price * 100):.2f}%") # 3. 볼린저 밴드 분석 bb_lower_min = bottom_data['Lower'].min() bb_upper_max = bottom_data['Upper'].max() print(f"\n볼린저 밴드 분석:") print(f"하단 밴드 최저: {bb_lower_min:.4f}") print(f"상단 밴드 최고: {bb_upper_max:.4f}") # 4. 거래량 분석 volume_avg = bottom_data['Volume'].mean() volume_max = bottom_data['Volume'].max() print(f"\n거래량 분석:") print(f"평균 거래량: {volume_avg:.0f}") print(f"최대 거래량: {volume_max:.0f}") # 5. 실제 저점 찾기 actual_bottom_idx = bottom_data['Low'].idxmin() actual_bottom_price = bottom_data.loc[actual_bottom_idx, 'Low'] actual_bottom_date = actual_bottom_idx print(f"\n실제 저점:") print(f"날짜: {actual_bottom_date}") print(f"가격: {actual_bottom_price:.4f}") # 실제 저점에서 RSI 출력 제거 # print(f"RSI: {bottom_data.loc[actual_bottom_idx, 'RSI']:.2f}") print(f"볼린저 하단 대비: {((actual_bottom_price - bottom_data.loc[actual_bottom_idx, 'Lower']) / bottom_data.loc[actual_bottom_idx, 'Lower'] * 100):.2f}%") # 6. 매수 신호 분석 print(f"\n=== 매수 신호 분석 ===") # 현재 매수 조건으로 저점 기간에서 매수 신호가 몇 개 발생하는지 확인 alerts = [] debug_info = [] for i in range(len(bottom_data)): slice_df = data.iloc[:data.index.get_loc(bottom_data.index[i]) + 1] info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes) if info: debug_info.append({ 'date': bottom_data.index[i], 'price': bottom_data['Close'].iloc[i], 'alert': info['alert'], 'details': info['details'] }) if info['alert']: alerts.append((bottom_data.index[i], bottom_data['Close'].iloc[i])) print(f"저점 기간 매수 신호 수: {len(alerts)}") if alerts: print("매수 신호 발생 시점:") for date, price in alerts: print(f" {date}: {price:.4f}") return bottom_data, alerts def run_simulation(symbol: str, interval_minutes: int, days: int = 30): data = fetch_price_history(symbol, interval_minutes) data = calculate_technical_indicators(data) data = check_buy_point(data, simulation=True) print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}") print(f"총 데이터 수: {len(data)}") # 파라미터 후보군 (표준화된 기술적 분석 기준) param_candidates = [ {'rsi_oversold': 32, 'bb_distance': 0.06, 'near_low_tolerance': 0.01, 'volume_multiplier': 2.0}, # 기본 설정 {'rsi_oversold': 30, 'bb_distance': 0.05, 'near_low_tolerance': 0.008, 'volume_multiplier': 2.5}, # 엄격한 설정 {'rsi_oversold': 28, 'bb_distance': 0.04, 'near_low_tolerance': 0.005, 'volume_multiplier': 3.0}, # 매우 엄격한 설정 ] alerts = [] for params in param_candidates: alerts.clear() for i in range(len(data)): slice_df = data.iloc[: i + 1] info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes, params=params) if info and info['alert']: alerts.append((slice_df.index[-1], slice_df['Close'].iloc[-1])) print(f"\n총 매수 신호 수: {len(alerts)}") # 서브플롯 생성 fig, ax1 = plt.subplots(figsize=(15, 8)) fig.suptitle(f"{symbol} - 시뮬레이션 {interval_minutes}분봉", fontsize=14) # 메인 차트 (가격, 이동평균선, 볼린저 밴드) line_close = ax1.plot(data.index, data["Close"], label="종가", color="black", linewidth=1.5)[0] line_ma5 = ax1.plot(data.index, data["MA5"], label="MA5", color="red", linewidth=1)[0] line_ma20 = ax1.plot(data.index, data["MA20"], label="MA20", color="blue", linewidth=1)[0] line_ma40 = ax1.plot(data.index, data["MA40"], label="MA40", color="green", linewidth=1)[0] line_ma120 = ax1.plot(data.index, data["MA120"], label="MA120", color="purple", linewidth=1)[0] line_ma200 = ax1.plot(data.index, data["MA200"], label="MA200", color="brown", linewidth=1)[0] line_ma240 = ax1.plot(data.index, data["MA240"], label="MA240", color="black", linewidth=1)[0] line_ma720 = ax1.plot(data.index, data["MA720"], label="MA720", color="cyan", linewidth=1)[0] line_ma1440 = ax1.plot(data.index, data["MA1440"], label="MA1440", color="magenta", linewidth=1)[0] # Bollinger Bands line_upper = ax1.plot(data.index, data["Upper"], label="볼린저 Upper", color="grey", linestyle="--", linewidth=1)[0] line_lower = ax1.plot(data.index, data["Lower"], label="볼린저 Lower", color="grey", linestyle="--", linewidth=1)[0] ax1.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1) # 매수 신호 표시 scatter_buy = None if alerts: times, prices = zip(*alerts) scatter_buy = ax1.scatter(times, prices, facecolors='none', edgecolors='red', linewidths=2, s=150, zorder=6, label='매수신호') for time in times: ax1.axvline(x=time, color='red', linestyle='--', alpha=0.3) # 매수 포인트 탐지 및 표시 # 'buy_point' 열 추가 data['buy_point'] = 0 for i in range(1, len(data)): if all(data[f'MA{n}'].iloc[i] < data['MA720'].iloc[i] for n in [5, 20, 40, 120, 200, 240]) and \ all(data[f'MA{n}'].iloc[i] > data[f'MA{n}'].iloc[i-1] for n in [5, 20, 40, 120, 200, 240]) and \ data['MA720'].iloc[i] < data['MA1440'].iloc[i]: data.at[data.index[i], 'buy_point'] = 1 # 매수 포인트를 빨간 동그라미로 표시 buy_points = data[data['buy_point'] == 1] scatter_buy_points = ax1.scatter(buy_points.index, buy_points['Close'], color='red', s=100, zorder=5, label='매수 포인트') # 마우스 오버 기능 추가 cursor = mplcursors.cursor(scatter_buy_points, hover=True) cursor.connect("add", lambda sel: sel.annotation.set_text( f'날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}' )) cursor.connect("remove", lambda sel: sel.annotation.set_visible(False)) # 매수 신호에도 마우스 오버 기능 추가 if scatter_buy is not None: cursor2 = mplcursors.cursor(scatter_buy, hover=True) cursor2.connect("add", lambda sel: sel.annotation.set_text( f'매수신호\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}' )) cursor2.connect("remove", lambda sel: sel.annotation.set_visible(False)) # 모든 봉에 마우스 오버 기능 추가 cursor3 = mplcursors.cursor(line_close, hover=True) cursor3.connect("add", lambda sel: sel.annotation.set_text( f'종가\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}' )) cursor3.connect("remove", lambda sel: sel.annotation.set_visible(False)) ax1.set_ylabel("가격") ax1.legend(loc='upper left', fontsize=10) ax1.grid(True, linestyle='--', alpha=0.5) plt.tight_layout() plt.show() if __name__ == "__main__": interval = 60 days = 90 # 분석 기간을 90일로 늘림 (6월~8월 데이터 포함) #target_coins = ['ADA','APE','ARB','BONK','HBAR','LINK','ONDO','PEPE','SEI','SHIB','STORJ','SUI','TON','TRX','WLD','XLM','XRP'] target_coins = ['APE'] for symbol in target_coins: print(f"\n=== {symbol} 저점 기간 분석 시작 ===") try: # 저점 기간 분석 bottom_data, alerts = analyze_bottom_period(symbol, interval, days) # 전체 기간 시뮬레이션 print(f"\n=== {symbol} 전체 기간 시뮬레이션 ===") run_simulation(symbol, interval, days) except Exception as e: print(f"Error analyzing {symbol}: {str(e)}")