init

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']}")
+                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, 'US'))
+                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분)