From 3f6ca0ce477208b88b40d61ecf9b7832868af8b8 Mon Sep 17 00:00:00 2001 From: dsyoon Date: Mon, 4 Aug 2025 21:36:29 +0900 Subject: [PATCH] init --- config.py | 3 + stock_monitor.py | 190 +++++++++++++++++++++++++++++------- stock_monitor_1.py | 236 ++++++++++++++++++++++++++------------------- 3 files changed, 295 insertions(+), 134 deletions(-) diff --git a/config.py b/config.py index 92c80cb..1e0a63a 100644 --- a/config.py +++ b/config.py @@ -16,6 +16,9 @@ BREAKOUT_LOOKBACK = 30 # U자 반등 후 돌파 판단에 사용할 과거 캔 BREAKOUT_WEEK_LOOKBACK = 42 # 4시간봉 1주일 ≒ 42개 BREAKOUT_WEEK_LIMIT = 0.05 # 1주일 대비 5% 미만 상승 조건 +# 볼린저 밴드 squeeze 탐지 임계값 (밴드폭/중심선) +SQUEEZE_THRESHOLD = 0.04 # 4% 이하 + # 장기간 저항선 돌파 감지 설정 RESISTANCE_LOOKBACK = 120 # 저항선 판단을 위한 과거 캔들 수 (예: 120개) RESISTANCE_BREAK_THRESHOLD = 0.01 # 저항선 대비 1% 이상 돌파 시 신호 diff --git a/stock_monitor.py b/stock_monitor.py index 6bc7553..acd5246 100644 --- a/stock_monitor.py +++ b/stock_monitor.py @@ -1,3 +1,5 @@ +import pandas as pd + import yfinance as yf import pandas as pd from datetime import datetime, timedelta @@ -17,6 +19,7 @@ def send_coin_msg(text): asyncio.run(coin_client.send_message(chat_id=COIN_TELEGRAM_CHAT_ID, text=text)) return + def send_coin_telegram_message(message_list, header): pStr = header + "\n" for i, message in enumerate(message_list): @@ -33,17 +36,19 @@ def send_coin_telegram_message(message_list, header): return + def send_stock_msg(text): stock_client = telegram.Bot(token=STOCK_TELEGRAM_BOT_TOKEN) asyncio.run(stock_client.send_message(chat_id=STOCK_TELEGRAM_CHAT_ID, text=text)) return + def send_stock_telegram_message(message_list, header): - pStr = header+"\n" + pStr = header + "\n" for i, message in enumerate(message_list): pStr += message - if i+1 % 20 == 0: + if i + 1 % 20 == 0: pool = Pool(12) pool.map(send_stock_msg, [pStr]) pStr = '' @@ -54,6 +59,7 @@ def send_stock_telegram_message(message_list, header): return + def calculate_bollinger_bands(data): data['MA'] = data['Close'].rolling(window=BOLLINGER_PERIOD).mean() data['STD'] = data['Close'].rolling(window=BOLLINGER_PERIOD).std() @@ -65,29 +71,29 @@ def calculate_bollinger_bands(data): 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['MA40'] = data['Close'].rolling(window=40).mean() data['MA60'] = data['Close'].rolling(window=60).mean() - + # 거래량 이동평균 data['Volume_MA5'] = data['Volume'].rolling(window=5).mean() - + return data @@ -132,10 +138,14 @@ def check_ma_alert(symbol, data, interval=0): def check_buy_signals(symbol, data): if len(data) < 60: # 최소 60일치 데이터 필요 return None - + latest = data.iloc[-1] prev = data.iloc[-2] - + + # 8월 3일 이후의 데이터만 고려 + if latest.name <= pd.Timestamp('2025-08-03'): + return None + # 볼린저 밴드 신호 bb_signal = False if isinstance(latest['Upper'], float): @@ -146,7 +156,7 @@ def check_buy_signals(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) bb_signal = distance < BOLLINGER_THRESHOLD @@ -154,12 +164,12 @@ def check_buy_signals(symbol, data): breakout_signal = False if len(data) >= max(BREAKOUT_LOOKBACK, BREAKOUT_WEEK_LOOKBACK) + 1: # ① U자 형태 확인 - window_close = data['Close'].iloc[-BREAKOUT_LOOKBACK-1:-1] + window_close = data['Close'].iloc[-BREAKOUT_LOOKBACK - 1:-1] prev_high = window_close.max() prev_low = window_close.min() # ② 1주일(42캔들) 전 가격 대비 5% 이상 상승하지 않았는지 체크 - price_week_ago = data['Close'].iloc[-BREAKOUT_WEEK_LOOKBACK-1] + price_week_ago = data['Close'].iloc[-BREAKOUT_WEEK_LOOKBACK - 1] if price_week_ago > 0: week_change = (current_price - price_week_ago) / price_week_ago else: @@ -167,27 +177,28 @@ def check_buy_signals(symbol, data): # ③ 조건 종합: U자+돌파 && 주간 상승률 ≤ 5% if ( - prev_high > 0 and (prev_high - prev_low) / prev_high > BUY_THRESHOLD and current_price > prev_high - and week_change <= BREAKOUT_WEEK_LIMIT + prev_high > 0 and (prev_high - prev_low) / prev_high > BUY_THRESHOLD and current_price > prev_high + and week_change <= BREAKOUT_WEEK_LIMIT ): breakout_signal = True # 장기간 저항선 돌파 여부 계산 (LONG RESISTANCE BREAKOUT) long_breakout_signal = False if len(data) >= RESISTANCE_LOOKBACK + 1: - resistance_level = data['Close'].iloc[-RESISTANCE_LOOKBACK-1:-1].max() - previous_closes = data['Close'].iloc[-RESISTANCE_LOOKBACK-1:-1] + resistance_level = data['Close'].iloc[-RESISTANCE_LOOKBACK - 1:-1].max() + previous_closes = data['Close'].iloc[-RESISTANCE_LOOKBACK - 1:-1] # 과거 구간에서 저항선 이상으로 종가가 한번도 올라가지 않은 경우 + 현재 가격이 저항선 돌파 if (previous_closes <= resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD)).all() and \ - current_price > resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD): + current_price > resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD): long_breakout_signal = True - + # RSI 과매도 신호 (RSI < 30) if not isinstance(latest['Upper'], float): 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]) + 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]) @@ -219,7 +230,8 @@ def check_buy_signals(symbol, data): 'signal_line': latest['Signal'].iloc[0], 'buy_signals': buy_signals, 'signal_count': signal_count, - 'buy': long_breakout_signal or breakout_signal or ((bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) + 'buy': long_breakout_signal or breakout_signal or ( + (bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) } else: rsi_signal = latest['RSI'] < 30 @@ -257,17 +269,19 @@ def check_buy_signals(symbol, data): 'signal_line': latest['Signal'], 'buy_signals': buy_signals, 'signal_count': signal_count, - 'buy': long_breakout_signal or breakout_signal or ((bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) + 'buy': long_breakout_signal or breakout_signal or ( + (bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) } + 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}, " - + # 매수 신호 상세 정보 count = 0 if any(info['buy_signals'].values()): @@ -305,6 +319,7 @@ def format_ma_message(info, market_type): message += f"현재가: {'$' if market_type == 'US' else '₩'}{info['price']:.2f} \n" return message + def get_coin_data(symbol, interval=240, retries=3): for attempt in range(retries): try: @@ -348,10 +363,10 @@ def get_kr_stock_data(symbol, retries=3): try: end = datetime.now() start = end - timedelta(days=300) - + # FinanceDataReader를 사용하여 한국 주식 데이터 가져오기 data = fdr.DataReader(symbol, start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d')) - + if not data.empty: # FinanceDataReader의 컬럼명을 yfinance 형식으로 변환 data = data.rename(columns={ @@ -372,10 +387,11 @@ def get_kr_stock_data(symbol, retries=3): continue return None + def monitor_us_stocks(): message_list = [] print("US Stocks {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) - + for symbol in US_STOCKS: data = get_kr_stock_data(symbol) if data is not None and not data.empty: @@ -386,13 +402,13 @@ def monitor_us_stocks(): continue info['name'] = US_STOCKS[symbol] 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) - + if len(message_list) > 0: try: send_stock_telegram_message(message_list, header="[US-STOCK]") @@ -411,7 +427,7 @@ def monitor_kr_stocks(): # .KS 접미사 제거 clean_symbol = symbol.replace('.KS', '') data = get_kr_stock_data(clean_symbol) - + if data is not None and not data.empty: try: data = calculate_technical_indicators(data) @@ -428,10 +444,10 @@ def monitor_kr_stocks(): print(f"Error processing data for {symbol}: {str(e)}") else: print(f"Data for {symbol} is empty or None.") - + # 각 심볼 처리 후 1초 대기 time.sleep(1) - + except Exception as e: print(f"Unexpected error processing {symbol}: {str(e)}") continue @@ -460,7 +476,8 @@ def monitor_coins(): if info is None: continue info['name'] = KR_COINS[symbol] - print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})") + print( + f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})") if info['alert']: message_list.append(format_ma_message(info, 'KR')) @@ -480,7 +497,8 @@ def monitor_coins(): if info is None: continue info['name'] = KR_COINS[symbol] - print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})") + print( + f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})") if info['alert']: message_list.append(format_ma_message(info, 'KR')) @@ -498,8 +516,109 @@ def monitor_coins(): return -def run_schedule(): +# ---------------------- +# Indicator utilities +# ---------------------- +def calculate_bollinger_bands(data: pd.DataFrame, period: int = 20, std: int = 2): + data = data.copy() + data['MA'] = data['Close'].rolling(window=period).mean() + data['STD'] = data['Close'].rolling(window=period).std() + data['Upper'] = data['MA'] + std * data['STD'] + data['Lower'] = data['MA'] - std * data['STD'] + return data + + +def calculate_technical_indicators(data: pd.DataFrame): + """Add MA5/20/40, RSI14, MACD+Hist, Bollinger""" + data = calculate_bollinger_bands(data) + + # RSI(14) + delta = data['Close'].diff() + gain = delta.where(delta > 0, 0).rolling(14).mean() + loss = (-delta.where(delta < 0, 0)).rolling(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['MACD_Hist'] = data['MACD'] - data['Signal'] + + # Moving averages + data['MA5'] = data['Close'].rolling(5).mean() + data['MA20'] = data['Close'].rolling(20).mean() + data['MA40'] = data['Close'].rolling(40).mean() + + return data + +# ---------------------- +# Turnaround Detector v6 +# ---------------------- + +def detect_turnaround_signal(symbol, data, interval=0): + if len(data) < 7: + return None + + cur = data.iloc[-1] + prev = data.iloc[-2] + recent = data.iloc[-7:-1] + + # ①~② 국지 바닥 + 하단 터치 + is_bottom = prev.Low == recent['Low'].min() + touch_lower = prev.Close <= prev.Lower * 1.01 + + # ③ 첫 반등 + MA5 돌파 + rebound = (cur.Close >= prev.Close * 1.005) and (cur.Close >= cur.Lower * 1.01) + ma5_break = (prev.Close <= prev.MA5) and (cur.Close > cur.MA5) + ma20_below = cur.Close < cur.MA20 + + # ④ RSI 회복 + rsi_recover = (prev.RSI < 40) and (cur.RSI > 40) and (cur.RSI > prev.RSI) + + # ⑤ MACD 히스토그램 양전환 + hist_cross = (prev.MACD_Hist <= 0) and (cur.MACD_Hist > 0) + + # ⑥ MA5-MA20 골든크로스 (이번 봉) (or DMI 사용) + cross_ma = (prev.MA5 <= prev.MA20) and (cur.MA5 > cur.MA20) + + # ⑦ 장기 추세 아직 하락 + down_trend = cur.MA20 < cur.MA40 + + # 최근 저점 탐지 강화 + recent_low = data['Low'].iloc[-10:].min() + is_recent_low = cur.Low <= recent_low * 1.005 + + # 추가 조건: MA5가 MA20을 상향 돌파 + ma5_cross_ma20 = (prev.MA5 <= prev.MA20) and (cur.MA5 > cur.MA20) + + # 매수 신호 조건 (저점 + 단기 MA 돌파 + RSI 회복) + # 매수 신호 조건: 전봉 하단 밴드 터치 + 반등 + MA5 돌파 + RSI 회복 + alert = touch_lower and rebound and ma5_break and rsi_recover + + return { + "symbol": symbol, + "price": float(cur.Close), + "alert": alert, + "details": { + "interval": interval, + "is_bottom": is_bottom, + "touch": touch_lower, + "rebound": rebound, + "ma5_break": ma5_break, + "ma20_below": ma20_below, + "rsi_recover": rsi_recover, + "hist_cross": hist_cross, + "cross_ma": cross_ma, + "down_trend": down_trend, + }, + } + + + +def run_schedule(): # 코인 모니터링 스케줄 (매시간 1분, 11분, 21분, 31분, 41분, 51분) for minute in [4, 14, 24, 34, 44, 54]: schedule.every().hour.at(f":{minute:02d}").do(monitor_coins) @@ -520,5 +639,4 @@ def run_schedule(): if __name__ == "__main__": - run_schedule() diff --git a/stock_monitor_1.py b/stock_monitor_1.py index bf66de2..46ea7a4 100644 --- a/stock_monitor_1.py +++ b/stock_monitor_1.py @@ -17,6 +17,7 @@ def send_coin_msg(text): asyncio.run(coin_client.send_message(chat_id=COIN_TELEGRAM_CHAT_ID, text=text)) return + def send_coin_telegram_message(message_list, header): pStr = header + "\n" for i, message in enumerate(message_list): @@ -33,17 +34,19 @@ def send_coin_telegram_message(message_list, header): return + def send_stock_msg(text): stock_client = telegram.Bot(token=STOCK_TELEGRAM_BOT_TOKEN) asyncio.run(stock_client.send_message(chat_id=STOCK_TELEGRAM_CHAT_ID, text=text)) return + def send_stock_telegram_message(message_list, header): - pStr = header+"\n" + pStr = header + "\n" for i, message in enumerate(message_list): pStr += message - if i+1 % 20 == 0: + if i + 1 % 20 == 0: pool = Pool(12) pool.map(send_stock_msg, [pStr]) pStr = '' @@ -54,6 +57,7 @@ def send_stock_telegram_message(message_list, header): return + def calculate_bollinger_bands(data): data['MA'] = data['Close'].rolling(window=BOLLINGER_PERIOD).mean() data['STD'] = data['Close'].rolling(window=BOLLINGER_PERIOD).std() @@ -65,106 +69,101 @@ def calculate_bollinger_bands(data): 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['MA40'] = data['Close'].rolling(window=40).mean() data['MA60'] = data['Close'].rolling(window=60).mean() - + # 거래량 이동평균 data['Volume_MA5'] = data['Volume'].rolling(window=5).mean() - + 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, interval=0): + """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': { + 'interval': interval, + '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일치 데이터 필요 return None - + 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) breakout_signal = False if len(data) >= max(BREAKOUT_LOOKBACK, BREAKOUT_WEEK_LOOKBACK) + 1: # ① U자 형태 확인 - window_close = data['Close'].iloc[-BREAKOUT_LOOKBACK-1:-1] + window_close = data['Close'].iloc[-BREAKOUT_LOOKBACK - 1:-1] prev_high = window_close.max() prev_low = window_close.min() # ② 1주일(42캔들) 전 가격 대비 5% 이상 상승하지 않았는지 체크 - price_week_ago = data['Close'].iloc[-BREAKOUT_WEEK_LOOKBACK-1] + price_week_ago = data['Close'].iloc[-BREAKOUT_WEEK_LOOKBACK - 1] if price_week_ago > 0: week_change = (current_price - price_week_ago) / price_week_ago else: @@ -172,27 +171,28 @@ def check_buy_signals(symbol, data): # ③ 조건 종합: U자+돌파 && 주간 상승률 ≤ 5% if ( - prev_high > 0 and (prev_high - prev_low) / prev_high > BUY_THRESHOLD and current_price > prev_high - and week_change <= BREAKOUT_WEEK_LIMIT + prev_high > 0 and (prev_high - prev_low) / prev_high > BUY_THRESHOLD and current_price > prev_high + and week_change <= BREAKOUT_WEEK_LIMIT ): breakout_signal = True # 장기간 저항선 돌파 여부 계산 (LONG RESISTANCE BREAKOUT) long_breakout_signal = False if len(data) >= RESISTANCE_LOOKBACK + 1: - resistance_level = data['Close'].iloc[-RESISTANCE_LOOKBACK-1:-1].max() - previous_closes = data['Close'].iloc[-RESISTANCE_LOOKBACK-1:-1] + resistance_level = data['Close'].iloc[-RESISTANCE_LOOKBACK - 1:-1].max() + previous_closes = data['Close'].iloc[-RESISTANCE_LOOKBACK - 1:-1] # 과거 구간에서 저항선 이상으로 종가가 한번도 올라가지 않은 경우 + 현재 가격이 저항선 돌파 if (previous_closes <= resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD)).all() and \ - current_price > resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD): + current_price > resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD): long_breakout_signal = True - + # RSI 과매도 신호 (RSI < 30) if not isinstance(latest['Upper'], float): 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]) + 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]) @@ -224,7 +224,8 @@ def check_buy_signals(symbol, data): 'signal_line': latest['Signal'].iloc[0], 'buy_signals': buy_signals, 'signal_count': signal_count, - 'buy': long_breakout_signal or breakout_signal or ((bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) + 'buy': long_breakout_signal or breakout_signal or ( + (bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) } else: rsi_signal = latest['RSI'] < 30 @@ -262,17 +263,19 @@ def check_buy_signals(symbol, data): 'signal_line': latest['Signal'], 'buy_signals': buy_signals, 'signal_count': signal_count, - 'buy': long_breakout_signal or breakout_signal or ((bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) + 'buy': long_breakout_signal or breakout_signal or ( + (bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal))) } + 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}, " - + # 매수 신호 상세 정보 count = 0 if any(info['buy_signals'].values()): @@ -302,10 +305,19 @@ def format_message(info, market_type): message = message.replace("{count}", str(count)) return message -def get_coin_data(symbol, retries=3): + +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, interval=240, retries=3): for attempt in range(retries): try: - url = "https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=3000".format(240, symbol) + url = "https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=3000".format(interval, symbol) headers = {"accept": "application/json"} response = requests.get(url, headers=headers) json_data = json.loads(response.text) @@ -345,10 +357,10 @@ def get_kr_stock_data(symbol, retries=3): try: end = datetime.now() start = end - timedelta(days=300) - + # FinanceDataReader를 사용하여 한국 주식 데이터 가져오기 data = fdr.DataReader(symbol, start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d')) - + if not data.empty: # FinanceDataReader의 컬럼명을 yfinance 형식으로 변환 data = data.rename(columns={ @@ -369,26 +381,28 @@ def get_kr_stock_data(symbol, retries=3): continue return None + def monitor_us_stocks(): message_list = [] print("US Stocks {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) - + for symbol in US_STOCKS: data = get_kr_stock_data(symbol) 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, 0) + 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) - + if len(message_list) > 0: try: send_stock_telegram_message(message_list, header="[US-STOCK]") @@ -407,25 +421,27 @@ def monitor_kr_stocks(): # .KS 접미사 제거 clean_symbol = symbol.replace('.KS', '') data = get_kr_stock_data(clean_symbol) - + 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, 0) + 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)}") else: print(f"Data for {symbol} is empty or None.") - + # 각 심볼 처리 후 1초 대기 time.sleep(1) - + except Exception as e: print(f"Unexpected error processing {symbol}: {str(e)}") continue @@ -444,17 +460,42 @@ def monitor_coins(): print("KRW COINs {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) for symbol in KR_COINS: - data = get_coin_data(symbol) + # 1시간 + interval = 60 + data = get_coin_data(symbol, interval=interval) 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, interval) + 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']} ({info['details']['interval']})") - 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: + print(f"Data for {symbol} is empty or None.") + time.sleep(0.5) + + # 4시간 + interval = 240 + data = get_coin_data(symbol, interval=interval) + if data is not None and not data.empty: + try: + data = calculate_technical_indicators(data) + info = check_ma_alert(symbol, data, interval) + if info is None: + continue + info['name'] = KR_COINS[symbol] + print( + f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})") + + 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: @@ -469,10 +510,10 @@ def monitor_coins(): return -def run_schedule(): +def run_schedule(): # 코인 모니터링 스케줄 (매시간 1분, 11분, 21분, 31분, 41분, 51분) - for minute in [4, 14, 24, 34,44, 54]: + for minute in [4, 14, 24, 34, 44, 54]: schedule.every().hour.at(f":{minute:02d}").do(monitor_coins) # 미국 주식 모니터링 스케줄 (매일 저녁 5시 20분) @@ -491,5 +532,4 @@ def run_schedule(): if __name__ == "__main__": - run_schedule()