From f7784bb5bc12289402d684f8fe02bf50154e36db Mon Sep 17 00:00:00 2001 From: dsyoon Date: Mon, 4 Aug 2025 08:50:36 +0900 Subject: [PATCH] init --- stock_monitor.py | 138 ++++++++++++++++++++++++----------------------- 1 file changed, 72 insertions(+), 66 deletions(-) diff --git a/stock_monitor.py b/stock_monitor.py index bf66de2..d98ed92 100644 --- a/stock_monitor.py +++ b/stock_monitor.py @@ -82,6 +82,7 @@ def calculate_technical_indicators(data): # 이동평균선 data['MA5'] = data['Close'].rolling(window=5).mean() data['MA20'] = data['Close'].rolling(window=20).mean() + data['MA40'] = data['Close'].rolling(window=40).mean() data['MA60'] = data['Close'].rolling(window=60).mean() # 거래량 이동평균 @@ -90,42 +91,42 @@ def calculate_technical_indicators(data): return data -def get_daily_bollinger_distance(data): - """일봉 기준 볼린저 밴드 하단 근접도를 계산하여 distance, 하단 밴드, 종가를 반환한다.""" - # 데이터 인덱스가 datetime이 아니면 변환 - df = data.copy() - if not isinstance(df.index, pd.DatetimeIndex): - if 'datetime' in df.columns: - df = df.set_index('datetime') - # 일봉 리샘플링 - daily = df.resample('D').agg({'Open': 'first', - 'High': 'max', - 'Low': 'min', - 'Close': 'last', - 'Volume': 'sum'}).dropna() +def check_ma_alert(symbol, data): + """1시간봉 기준 이동평균선 조건 알림 + - 5봉 이동평균(MA5) 상승 + - 20봉 이동평균(MA20) 상승 + - 40봉 이동평균(MA40)이 하락세에서 상승세로 전환되는 시점 (직전 기울기 < 0 and 현재 기울기 ≥ 0) + """ + # 40 이동평균선의 기울기를 계산하기 위해 최소 41개 캔들이 필요합니다. + if len(data) < 41: + return None - # 볼린저 밴드 계산 - daily = calculate_bollinger_bands(daily) + # 이동평균선 값 추출 + ma5_current, ma5_prev = data['MA5'].iloc[-1], data['MA5'].iloc[-2] + ma20_current, ma20_prev = data['MA20'].iloc[-1], data['MA20'].iloc[-2] + ma40_current = data['MA40'].iloc[-1] + ma40_prev1, ma40_prev2 = data['MA40'].iloc[-2], data['MA40'].iloc[-3] - if len(daily) < BOLLINGER_PERIOD: - return None, None, None + # 조건 계산 + up5 = ma5_current > ma5_prev + up20 = ma20_current > ma20_prev + slope_prev = ma40_prev1 - ma40_prev2 # 직전 기울기 (음수: 하락) + slope_now = ma40_current - ma40_prev1 # 현재 기울기 + turning = (slope_prev < 0) and (slope_now >= 0) - latest_daily = daily.iloc[-1] + alert = up5 and up20 and turning - # 볼린저 밴드 값이 존재하는지 확인 - if pd.isna(latest_daily['Upper']) or pd.isna(latest_daily['Lower']): - return None, None, None + return { + 'symbol': symbol, + 'price': data['Close'].iloc[-1], + 'alert': alert, + 'details': { + 'up5': up5, + 'up20': up20, + 'ma40_turning': turning + } + } - upper_band = latest_daily['Upper'] - lower_band = latest_daily['Lower'] - current_price = latest_daily['Close'] - - # 밴드 폭이 0이면 계산 불가 - if upper_band - lower_band == 0: - return None, None, None - - distance = (current_price - lower_band) / (upper_band - lower_band) - return distance, lower_band, current_price def check_buy_signals(symbol, data): if len(data) < 60: # 최소 60일치 데이터 필요 @@ -134,25 +135,18 @@ def check_buy_signals(symbol, data): latest = data.iloc[-1] prev = data.iloc[-2] - # 볼린저 밴드 신호 (일봉 기준) - distance, lower_band, current_price = get_daily_bollinger_distance(data) - - if distance is None: - # 일봉 볼린저 계산이 불가능한 경우 기존 주기 데이터를 사용 - if isinstance(latest['Upper'], float): - upper_band = latest['Upper'] - lower_band = latest['Lower'] - current_price = latest['Close'] - else: - upper_band = latest['Upper'].iloc[0] - lower_band = latest['Lower'].iloc[0] - current_price = latest['Close'].iloc[0] - - if upper_band - lower_band != 0: - distance = (current_price - lower_band) / (upper_band - lower_band) - else: - distance = 1 # 밴드폭이 0이면 신호 미충족으로 간주 - + # 볼린저 밴드 신호 + bb_signal = False + if isinstance(latest['Upper'], float): + upper_band = latest['Upper'] + lower_band = latest['Lower'] + current_price = latest['Close'] + else: + 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) bb_signal = distance < BOLLINGER_THRESHOLD # U자 반등 후 이전 고점 돌파 여부 계산 (BREAKOUT) @@ -302,6 +296,14 @@ def format_message(info, market_type): message = message.replace("{count}", str(count)) return message + +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" + return message + def get_coin_data(symbol, retries=3): for attempt in range(retries): try: @@ -378,13 +380,14 @@ def monitor_us_stocks(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - info = check_buy_signals(symbol, data) + info = check_ma_alert(symbol, data) + if info is None: + continue info['name'] = US_STOCKS[symbol] - print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['signal_count']}") - - if info['buy']: - #if info['buy'] or any(info['buy_signals'].values()): - message_list.append(format_message(info, 'US')) + print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']}") + + if info['alert']: + message_list.append(format_ma_message(info, 'US')) except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") time.sleep(0.5) @@ -411,12 +414,14 @@ def monitor_kr_stocks(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - info = check_buy_signals(symbol, data) + info = check_ma_alert(symbol, data) + if info is None: + continue info['name'] = KR_ETFS[symbol] - print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['signal_count']}") + print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']}") - if info['buy']: - message_list.append(format_message(info, 'KR')) + if info['alert']: + message_list.append(format_ma_message(info, 'KR')) except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") @@ -448,13 +453,14 @@ def monitor_coins(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - info = check_buy_signals(symbol, data) + info = check_ma_alert(symbol, data) + if info is None: + continue info['name'] = KR_COINS[symbol] - print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['signal_count']}") + print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']}") - if info['buy']: - #if info['buy'] or any(info['buy_signals'].values()): - message_list.append(format_message(info, 'KR')) + if info['alert']: + message_list.append(format_ma_message(info, 'KR')) except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") else: @@ -472,7 +478,7 @@ def monitor_coins(): def run_schedule(): # 코인 모니터링 스케줄 (매시간 1분, 11분, 21분, 31분, 41분, 51분) - for minute in [4, 14, 24, 34,44, 54]: + for minute in [4, 34]: schedule.every().hour.at(f":{minute:02d}").do(monitor_coins) # 미국 주식 모니터링 스케줄 (매일 저녁 5시 20분)