From f92b1881ecc1d9d12a075c559c7d72a9c66da37c Mon Sep 17 00:00:00 2001 From: dsyoon Date: Thu, 1 May 2025 10:18:47 +0900 Subject: [PATCH] init --- config.py | 4 +- stock_monitor.py | 178 +++++++++++++++++++++++++++++++---------------- 2 files changed, 120 insertions(+), 62 deletions(-) diff --git a/config.py b/config.py index 70c91d5..68985a0 100644 --- a/config.py +++ b/config.py @@ -10,7 +10,7 @@ STOCK_TELEGRAM_CHAT_ID = '574661323' # 볼린저 밴드 설정 BOLLINGER_PERIOD = 20 # 볼린저 밴드 기간 BOLLINGER_STD = 2 # 표준편차 승수 -ALERT_THRESHOLD = 0.10 # 하단 밴드 대비 10% 근접 시 알림 +BOLLINGER_THRESHOLD = 0.10 # 하단 밴드 대비 10% 근접 시 알림 BUY_THRESHOLD = 0.15 KR_COINS = { @@ -171,7 +171,6 @@ KR_ETFS = { "034020.KS": "두산에너빌리티 / 원전,친환경", "160550.KQ": "NEW / 미디어,콘텐츠", "089980.KQ": "상아프론테크 / 2차전지,소재", - "066970.KQ": "엘앤에프 / 2차전지 소재", "131970.KQ": "테스나 / 반도체 테스트", "036930.KQ": "주성엔지니어링 / 반도체 장비", "078600.KQ": "대주전자재료 / 2차전지 소재", @@ -182,7 +181,6 @@ KR_ETFS = { "068760.KQ": "셀트리온제약 / 바이오,제약", "032500.KQ": "케이엠더블유 / 5G,통신장비", "178320.KQ": "서진시스템 / 5G,전장,통신", - "091990.KQ": "셀트리온헬스케어 / 바이오,유통", "095700.KQ": "제넥신 / 바이오,백신", "084370.KQ": "유진테크 / 반도체 장비", "069080.KQ": "웹젠 / 게임", diff --git a/stock_monitor.py b/stock_monitor.py index dc56c8e..040e8f0 100644 --- a/stock_monitor.py +++ b/stock_monitor.py @@ -38,23 +38,42 @@ def calculate_bollinger_bands(data): return data -def check_bollinger_bands(symbol, data): - if len(data) < BOLLINGER_PERIOD: +def calculate_technical_indicators(data): + # 볼린저 밴드 계산 + data = calculate_bollinger_bands(data) + + # RSI 계산 (14일 기준) + delta = data['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 + data['RSI'] = 100 - (100 / (1 + rs)) + + # MACD 계산 + exp1 = data['Close'].ewm(span=12, adjust=False).mean() + exp2 = data['Close'].ewm(span=26, adjust=False).mean() + data['MACD'] = exp1 - exp2 + data['Signal'] = data['MACD'].ewm(span=9, adjust=False).mean() + + # 이동평균선 + data['MA5'] = data['Close'].rolling(window=5).mean() + data['MA20'] = data['Close'].rolling(window=20).mean() + data['MA60'] = data['Close'].rolling(window=60).mean() + + # 거래량 이동평균 + data['Volume_MA5'] = data['Volume'].rolling(window=5).mean() + + return data + +def check_buy_signals(symbol, data): + if len(data) < 60: # 최소 60일치 데이터 필요 return None - - # 과거 10개 봉에서 ALERT_THRESHOLD 아래로 빠진 적이 있는지 체크 - check = False - for i in range(-1, -2, -1): - past = data.iloc[i] - upper_band = past['Upper'] - lower_band = past['Lower'] - price = past['Close'] - distance = (price - lower_band) / (upper_band - lower_band) - if distance < ALERT_THRESHOLD: - check = True - break - + latest = data.iloc[-1] + prev = data.iloc[-2] + + # 볼린저 밴드 신호 + bb_signal = False if isinstance(latest['Upper'], float): upper_band = latest['Upper'] lower_band = latest['Lower'] @@ -63,20 +82,70 @@ def check_bollinger_bands(symbol, data): upper_band = latest['Upper'].iloc[0] lower_band = latest['Lower'].iloc[0] current_price = latest['Close'].iloc[0] + distance = (current_price - lower_band) / (upper_band - lower_band) - - buy = False - if check and BUY_THRESHOLD < distance: - buy = True - + bb_signal = distance < BOLLINGER_THRESHOLD + + # RSI 과매도 신호 (RSI < 30) + rsi_signal = latest['RSI'].iloc[0] < 30 + + # MACD 신호 (MACD가 시그널 라인을 상향 돌파) + macd_signal = (prev['MACD'].iloc[0] < prev['Signal'].iloc[0]) and (latest['MACD'].iloc[0] > latest['Signal'].iloc[0]) + + # 이동평균선 골든크로스 임박 또는 발생 + ma_signal = (prev['MA5'].iloc[0] < prev['MA20'].iloc[0]) and (latest['MA5'].iloc[0] >= latest['MA20'].iloc[0]) + + # 거래량 증가 신호 (5일 평균 대비 150% 이상) + volume_signal = latest['Volume'].iloc[0] > (latest['Volume_MA5'].iloc[0] * 1.5) + + # 종합 신호 + buy_signals = { + 'bb_signal': bb_signal, + 'rsi_signal': rsi_signal, + 'macd_signal': macd_signal, + 'ma_signal': ma_signal, + 'volume_signal': volume_signal + } + + # 최소 3개 이상의 신호가 동시에 발생할 때 매수 신호로 간주 + signal_count = sum(1 for signal in buy_signals.values() if signal) + return { 'symbol': symbol, 'price': current_price, 'lower_band': lower_band, 'distance': distance, - 'buy': buy + 'rsi': latest['RSI'].iloc[0], + 'macd': latest['MACD'].iloc[0], + 'signal_line': latest['Signal'].iloc[0], + 'buy_signals': buy_signals, + 'signal_count': signal_count, + 'buy': signal_count >= 3 } +def format_message(info, market_type): + message = "" + if info['buy']: + message += '🛒 ' + + message += f"[{market_type}] {info['name']} ({info['symbol']}) " + message += f"현재가: {'$' if market_type == 'US' else '₩'}{info['price']:.2f}\n" + + # 매수 신호 상세 정보 + if any(info['buy_signals'].values()): + message += "📊 매수 신호:\n" + if info['buy_signals']['bb_signal']: + message += "- 볼린저 밴드 하단 근접 (근접도: {:.1f}%)\n".format(info['distance'] * 100) + if info['buy_signals']['rsi_signal']: + message += f"- RSI 과매도 구간 (RSI: {info['rsi']:.1f})\n" + if info['buy_signals']['macd_signal']: + message += "- MACD 골든크로스\n" + if info['buy_signals']['ma_signal']: + message += "- 이동평균선 골든크로스\n" + if info['buy_signals']['volume_signal']: + message += "- 거래량 급증\n" + + return message def get_coin_data(symbol, retries=3): for attempt in range(retries): @@ -119,7 +188,7 @@ def get_stock_data(symbol, retries=3): for attempt in range(retries): try: end = datetime.now() - start = end - timedelta(days=60) + start = end - timedelta(days=300) data = yf.download( symbol, start=start.strftime('%Y-%m-%d'), @@ -141,33 +210,28 @@ def get_stock_data(symbol, retries=3): def monitor_us_stocks(): message = "" - - # 미국 주식 모니터링 print("US Stocks {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) + for symbol in US_STOCKS: data = get_stock_data(symbol) if data is not None and not data.empty: try: - data = calculate_bollinger_bands(data) - info = check_bollinger_bands(symbol, data) + data = calculate_technical_indicators(data) + info = check_buy_signals(symbol, data) info['name'] = US_STOCKS[symbol] - print(" - {} ({}): {:.2f} ({:.2f})".format(info['name'], symbol, info['price'], info['distance'])) - - if info['buy']: - message += '🛒' - if info['distance'] < ALERT_THRESHOLD: - message += "🔔" - message += "[{}] {} ({}) 현재가: ${:.2f}, 근접도: {:.2f}%\n".format('US', info['name'], info['symbol'], info['price'], info['distance'] * 100) + print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['signal_count']}") + + if info['buy'] or any(info['buy_signals'].values()): + message += format_message(info, 'US') except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") - else: - print(f"Data for {symbol} is empty or None.") time.sleep(0.5) - - try: - send_stock_telegram_message(message) - except Exception as e: - print(f"Error sending Telegram message: {str(e)}") + + if message: + try: + send_stock_telegram_message(message) + except Exception as e: + print(f"Error sending Telegram message: {str(e)}") return @@ -181,16 +245,13 @@ def monitor_kr_stocks(): data = get_stock_data(symbol) if data is not None and not data.empty: try: - data = calculate_bollinger_bands(data) - info = check_bollinger_bands(symbol, data) + data = calculate_technical_indicators(data) + info = check_buy_signals(symbol, data) info['name'] = KR_ETFS[symbol] - print(" - {} ({}): {:.2f} ({:.2f})".format(info['name'], symbol, info['price'], info['distance'])) - - if info['buy']: - message += '🛒' - if info['distance'] < ALERT_THRESHOLD: - message += "🔔" - message += "[{}] {} ({}) 현재가: ${:.2f}, 근접도: {:.2f}%\n".format('KR', info['name'], info['symbol'], info['price'], info['distance'] * 100) + print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['signal_count']}") + + if info['buy'] or any(info['buy_signals'].values()): + message += format_message(info, 'KR') except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") @@ -215,17 +276,12 @@ def monitor_coins(): data = get_coin_data(symbol) if data is not None and not data.empty: try: - data = calculate_bollinger_bands(data) - info = check_bollinger_bands(symbol, data) + data = calculate_technical_indicators(data) + info = check_buy_signals(symbol, data) info['name'] = KR_COINS[symbol] - print(" - {} ({}): {:.2f} ({:.2f})".format(info['name'], symbol, info['price'], info['distance'])) - - message += "· {} ({}) 현재가: ₩{}, 근접도: {:.2f}%".format(info['name'], info['symbol'], info['price'], info['distance'] * 100) - if info['buy']: - message += ' (🛒)' - if info['distance'] < ALERT_THRESHOLD: - message += "(🔔)" - message += '\n' + print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['signal_count']}") + + message += format_message(info, 'KR') except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") else: @@ -241,6 +297,10 @@ def monitor_coins(): def run_schedule(): + monitor_kr_stocks() + monitor_us_stocks() + monitor_coins() + # 코인 모니터링 스케줄 (매시간 1분, 11분, 21분, 31분, 41분, 51분) for minute in [1, 11, 21, 31, 41, 51]: schedule.every().hour.at(f":{minute:02d}").do(monitor_coins)