diff --git a/config.py b/config.py index 91ecde6..c5d29ac 100644 --- a/config.py +++ b/config.py @@ -32,6 +32,7 @@ KR_COINS = { "LINK": "Chainlink", "ONDO": "ONDO", "PEPE": "PEPE", + "POL": "POL", "SEI": "SEI", "SHIB": "Shiba Inu", "STORJ": "Storj", diff --git a/stock_monitor.py b/stock_monitor.py index 0341427..b784284 100644 --- a/stock_monitor.py +++ b/stock_monitor.py @@ -74,46 +74,6 @@ def send_coin_telegram_message(message_list, header): return -def buy_ticker(symbol, data): - try: - # 매수 금지 시간 확인 (20분) - current_time = datetime.now() - if symbol in buy_cooldown: - time_diff = current_time - buy_cooldown[symbol] - if time_diff.total_seconds() < 1200: # 20분 = 1200초 - print(f"{symbol}: 매수 금지 중 (남은 시간: {1200 - time_diff.total_seconds():.0f}초)") - return False - - BUY_AMOUNT = 6000 - - if data['buy_signal'].iloc[-1] == 'movingaverage': - BUY_AMOUNT = 70000 - elif data['buy_signal'].iloc[-1] == 'deviation40': - BUY_AMOUNT = 40000 - elif data['buy_signal'].iloc[-1] == 'deviation240': - BUY_AMOUNT = 6000 - - _ = hts.buyCoinMarket(symbol, BUY_AMOUNT) - - # 매수 성공 시 금지 시간 설정 및 파일에 저장 - buy_cooldown[symbol] = current_time - save_buy_cooldown(buy_cooldown) - print(f"{KR_COINS[symbol]} ({symbol}): {data['Close'].iloc[-1]:.2f}: 매수 완료, 20분간 매수 금지 시작") - - try: - pool = Pool(12) - pool.map(send_coin_msg, ["[KRW-COIN]" + "\n" + format_message('COIN', symbol, KR_COINS[symbol], data['Close'].iloc[-1], data['buy_signal'].iloc[-1])]) - except Exception as e: - print(f"Error sending Telegram message: {str(e)}") - - return True - - except Exception as e: - print(f"Error buying {symbol}: {str(e)}") - return False - return - - def send_stock_msg(text): stock_client = telegram.Bot(token=STOCK_TELEGRAM_BOT_TOKEN) @@ -196,34 +156,51 @@ def calculate_technical_indicators(data): return data -def check_buy_point(data, simulation=None): - """ - # 매수 포인트 탐지 및 표시 - if simulation: - recent_data = data - else: - # recent_data의 복사본 생성 - recent_data = data.tail(10).copy() +def buy_ticker(symbol, data): + try: + # 매수 금지 시간 확인 (20분) + current_time = datetime.now() + if symbol in buy_cooldown: + time_diff = current_time - buy_cooldown[symbol] + if time_diff.total_seconds() < 1200: # 20분 = 1200초 + print(f"{symbol}: 매수 금지 중 (남은 시간: {1200 - time_diff.total_seconds():.0f}초)") + return False - # 'buy_point' 열 초기화 - recent_data['buy_point'] = 0 + BUY_AMOUNT = 6000 - # FutureWarning 해결 - if recent_data['buy_point'].iloc[-1] != 1: - # 코드 계속 - for i in range(1, len(recent_data)): - if all(recent_data[f'MA{n}'].iloc[i] < recent_data['MA720'].iloc[i] for n in [5, 20, 40, 120, 200, 240]) and \ - all(recent_data[f'MA{n}'].iloc[i] > recent_data[f'MA{n}'].iloc[i-1] for n in [5, 20, 40, 120, 200, 240]) and \ - recent_data['MA720'].iloc[i] < recent_data['MA1440'].iloc[i]: - recent_data.at[recent_data.index[i], 'buy_point'] = 1 + if data['buy_signal'].iloc[-1] == 'movingaverage': + BUY_AMOUNT = 10000 + elif data['buy_signal'].iloc[-1] == 'deviation40': + BUY_AMOUNT = 15000 + elif data['buy_signal'].iloc[-1] == 'deviation240': + BUY_AMOUNT = 6000 + elif data['buy_signal'].iloc[-1] == 'deviation1440': + if symbol in ['BONK', 'PEPE', 'TON']: + BUY_AMOUNT = 30000 + else: + BUY_AMOUNT = 50000 - if not simulation: - if recent_data['buy_point'][-10:-1].sum() > 0: - recent_data.at[recent_data.index[-1], 'buy_point'] = 1 + _ = hts.buyCoinMarket(symbol, BUY_AMOUNT) - return recent_data - """ + # 매수 성공 시 금지 시간 설정 및 파일에 저장 + buy_cooldown[symbol] = current_time + save_buy_cooldown(buy_cooldown) + print(f"{KR_COINS[symbol]} ({symbol}): {data['Close'].iloc[-1]:.4f}: 매수 완료, 20분간 매수 금지 시작") + + try: + pool = Pool(12) + pool.map(send_coin_msg, ["[KRW-COIN]" + "\n" + format_message('COIN', symbol, KR_COINS[symbol], data['Close'].iloc[-1], data['buy_signal'].iloc[-1])]) + except Exception as e: + print(f"Error sending Telegram message: {str(e)}") + + except Exception as e: + print(f"Error buying {symbol}: {str(e)}") + return False + + return True + +def check_buy_point(symbol, data, simulation=None): # 매수 포인트 탐지 및 표시 # 데이터 복사본 생성하여 SettingWithCopyWarning 방지 @@ -256,20 +233,67 @@ def check_buy_point(data, simulation=None): data.at[data.index[-1], 'buy_signal'] = 'deviation40' data.at[data.index[-1], 'buy_point'] = 1 + if symbol not in ['BONK']: + # BONK는 240 체크하지 않음 + + if symbol in ['TRX']: + # Deviation240(이격도 240) 기반 매수 조건: 90 이하에서 상승 전환 + if data['Deviation240'].iloc[i - 1] < data['Deviation240'].iloc[i] and data['Deviation240'].iloc[i - 1] <= 98: + data.at[data.index[i], 'buy_signal'] = 'deviation240' + data.at[data.index[i], 'buy_point'] = 1 + if not simulation: + if data['buy_point'][-3:].sum() > 0: + data.at[data.index[-1], 'buy_signal'] = 'deviation240' + data.at[data.index[-1], 'buy_point'] = 1 + else: + # Deviation240(이격도 240) 기반 매수 조건: 90 이하에서 상승 전환 + if data['Deviation240'].iloc[i - 1] < data['Deviation240'].iloc[i] and data['Deviation240'].iloc[i - 1] <= 90: + data.at[data.index[i], 'buy_signal'] = 'deviation240' + data.at[data.index[i], 'buy_point'] = 1 + if not simulation: + if data['buy_point'][-3:].sum() > 0: + data.at[data.index[-1], 'buy_signal'] = 'deviation240' + data.at[data.index[-1], 'buy_point'] = 1 + # Deviation240(이격도 240) 기반 매수 조건: 90 이하에서 상승 전환 - if data['Deviation240'].iloc[i - 1] < data['Deviation240'].iloc[i] and data['Deviation240'].iloc[i - 1] <= 90: - data.at[data.index[i], 'buy_signal'] = 'deviation240' - data.at[data.index[i], 'buy_point'] = 1 - if not simulation: - if data['buy_point'][-3:].sum() > 0: - data.at[data.index[-1], 'buy_signal'] = 'deviation240' - data.at[data.index[-1], 'buy_point'] = 1 + if symbol in ['TON']: + if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 89: + data.at[data.index[i], 'buy_signal'] = 'deviation1440' + data.at[data.index[i], 'buy_point'] = 1 + if not simulation: + if data['buy_point'][-3:].sum() > 0: + data.at[data.index[-1], 'buy_signal'] = 'deviation1440' + data.at[data.index[-1], 'buy_point'] = 1 + elif symbol in ['XRP']: + if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 90: + data.at[data.index[i], 'buy_signal'] = 'deviation1440' + data.at[data.index[i], 'buy_point'] = 1 + if not simulation: + if data['buy_point'][-3:].sum() > 0: + data.at[data.index[-1], 'buy_signal'] = 'deviation1440' + data.at[data.index[-1], 'buy_point'] = 1 + elif symbol in ['BONK']: + if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 76: + data.at[data.index[i], 'buy_signal'] = 'deviation1440' + data.at[data.index[i], 'buy_point'] = 1 + if not simulation: + if data['buy_point'][-3:].sum() > 0: + data.at[data.index[-1], 'buy_signal'] = 'deviation1440' + data.at[data.index[-1], 'buy_point'] = 1 + else: + if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 80: + data.at[data.index[i], 'buy_signal'] = 'deviation1440' + data.at[data.index[i], 'buy_point'] = 1 + if not simulation: + if data['buy_point'][-3:].sum() > 0: + data.at[data.index[-1], 'buy_signal'] = 'deviation1440' + data.at[data.index[-1], 'buy_point'] = 1 return data def format_message(market_type, symbol, symbol_name, close, buy_signal): message = f"매수 [{market_type}] {symbol_name} ({symbol}): {buy_signal} " - message += f"현재가: {'$' if market_type == 'US' else '₩'}{close:.2f}, " + message += f"현재가: {'$' if market_type == 'US' else '₩'}{close:.4f}, " return message @@ -277,7 +301,7 @@ def format_ma_message(info, market_type): """MA 알림 메시지 생성""" prefix = '상승 ' if info.get('alert') else '' message = prefix + f"[{market_type}] {info['name']} ({info['symbol']}) " - message += f"현재가: {'$' if market_type == 'US' else '₩'}{info['price']:.2f} \n" + message += f"현재가: {'$' if market_type == 'US' else '₩'}{info['price']:.4f} \n" return message @@ -432,7 +456,7 @@ def monitor_us_stocks(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - recent_data = check_buy_point(data) # Changed to check_buy_point + recent_data = check_buy_point(symbol, data) # Changed to check_buy_point if recent_data['buy_point'].iloc[-1] != 1: continue print(f" - {US_STOCKS[symbol]} ({symbol}): {recent_data['Close'].iloc[-1]:.2f}") @@ -463,7 +487,7 @@ def monitor_kr_stocks(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - recent_data = check_buy_point(data) # Changed to check_buy_point + recent_data = check_buy_point(symbol, data) # Changed to check_buy_point if recent_data['buy_point'].iloc[-1] != 1: continue print(f" - {KR_ETFS[symbol]} ({symbol}): {recent_data['Close'].iloc[-1]:.2f}") @@ -503,7 +527,7 @@ def monitor_coins(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - recent_data = check_buy_point(data) # Changed to check_buy_point + recent_data = check_buy_point(symbol, data) # Changed to check_buy_point if recent_data['buy_point'].iloc[-1] != 1: continue diff --git a/stock_simulation.py b/stock_simulation.py index c814ba0..cbcc9d7 100644 --- a/stock_simulation.py +++ b/stock_simulation.py @@ -69,7 +69,7 @@ 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) - data = check_buy_point(data, simulation=True) # 한 번만 계산 + data = check_buy_point(symbol, data, simulation=True) # 한 번만 계산 print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}") print(f"총 데이터 수: {len(data)}") @@ -145,7 +145,7 @@ def analyze_bottom_period(symbol: str, interval_minutes: int, days: int = 90): 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) + data = check_buy_point(symbol, data, simulation=True) print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}") print(f"총 데이터 수: {len(data)}") @@ -162,10 +162,12 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30): ma_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'movingaverage')]) dev40_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation40')]) dev240_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation240')]) + dev1440_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation1440')]) print(f" - MA 신호: {ma_signals}") print(f" - Dev40 신호: {dev40_signals}") print(f" - Dev240 신호: {dev240_signals}") + print(f" - Dev1440 신호: {dev1440_signals}") # 서브플롯 생성 (가격 + Deviation) fig, (ax1, ax2) = plt.subplots(2, 1, sharex=True, figsize=(15, 8), height_ratios=[3, 1]) @@ -283,6 +285,16 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30): for time in dev240_buy_points.index: ax1.axvline(x=time, color='blue', linestyle='--', alpha=0.5, linewidth=1) + # Deviation1440 기반 매수 포인트 + dev1440_buy_points = data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation1440')] + scatter_dev1440_buy_points = None + if len(dev1440_buy_points) > 0: + scatter_dev1440_buy_points = ax1.scatter(dev1440_buy_points.index, dev1440_buy_points['Close'], + facecolors='none', edgecolors='purple', linestyle='--', + linewidth=2, s=200, zorder=10, label='Dev1440 매수 포인트') + for time in dev1440_buy_points.index: + ax1.axvline(x=time, color='purple', linestyle='--', alpha=0.5, linewidth=1) + # 마우스 오버 기능 추가 (이동평균선 매수 포인트) @@ -309,6 +321,14 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30): )) cursor_dev240.connect("remove", lambda sel: sel.annotation.set_visible(False)) + # 마우스 오버 기능 추가 (Deviation1440 매수 포인트) + if scatter_dev1440_buy_points is not None: + cursor_dev1440 = mplcursors.cursor(scatter_dev1440_buy_points, hover=True) + cursor_dev1440.connect("add", lambda sel: sel.annotation.set_text( + f'Dev1440 매수신호\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}' + )) + cursor_dev1440.connect("remove", lambda sel: sel.annotation.set_visible(False)) + # 모든 봉에 마우스 오버 기능 추가 @@ -352,6 +372,8 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30): plot_lines.append(scatter_dev40_buy_points) if scatter_dev240_buy_points is not None: plot_lines.append(scatter_dev240_buy_points) + if scatter_dev1440_buy_points is not None: + plot_lines.append(scatter_dev1440_buy_points) # zip 길이가 짧은 쪽에 맞춰 매핑 for leg_handle, orig in zip(legend_handles, plot_lines): @@ -481,8 +503,8 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30): 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'] + #target_coins = ['ADA','APE','ARB','BONK','HBAR','LINK','ONDO','PEPE','SEI','SHIB','STORJ','SUI','TON','TRX','WLD','XLM','XRP'] + target_coins = ['POL'] # 그래프 표시 여부 설정 (성능 향상을 위해 기본값은 False) show_graphs = True # True로 설정하면 각 코인마다 그래프 표시 @@ -501,18 +523,20 @@ if __name__ == "__main__": # 그래프 없이 빠른 분석만 수행 data = fetch_price_history(symbol, interval, days) data = calculate_technical_indicators(data) - data = check_buy_point(data, simulation=True) + data = check_buy_point(symbol, data, simulation=True) # 매수 신호 통계만 출력 total_buy_signals = len(data[data['buy_point'] == 1]) ma_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'movingaverage')]) dev40_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation40')]) dev240_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation240')]) + dev1440_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation1440')]) print(f"총 매수 신호: {total_buy_signals}") print(f" - MA 신호: {ma_signals}") print(f" - Dev40 신호: {dev40_signals}") print(f" - Dev240 신호: {dev240_signals}") + print(f" - Dev1440 신호: {dev1440_signals}") except Exception as e: print(f"Error analyzing {symbol}: {str(e)}")