From f7870381bf17caa30635fb01e40dc0e1f06cdeb7 Mon Sep 17 00:00:00 2001 From: dsyoon Date: Wed, 6 Aug 2025 14:39:43 +0900 Subject: [PATCH] init --- PROMPT.txt | 15 ++ README.md | 42 +++-- config.py | 3 + requirements.txt | 1 + stock_monitor.py | 420 ++++++++++++++------------------------------ stock_simulation.py | 261 +++++++++++++++++++-------- 6 files changed, 362 insertions(+), 380 deletions(-) create mode 100644 PROMPT.txt diff --git a/PROMPT.txt b/PROMPT.txt new file mode 100644 index 0000000..91d930e --- /dev/null +++ b/PROMPT.txt @@ -0,0 +1,15 @@ +#1. +봉을 모두 1500개를 출력하고 있습니다. +전반적으로 그래프를 이해하세요. +그리고 시가, 종가, 고가, 저가, 거래량을 활용하여 많은 기술적 분석을 시도하세요. +그리고 최적의 저점에 매수를 할 수 있도록 탐색코드를 작성하세요. + +모든 코인에 동일 조건을 적용할 수 있도록 활용하는 수치는 모두 정규화를 시킨 이후에 시도해야 합니다. + +코드에 탐색을 반영하고 실행하세요, 그리고 매수 시점을 표기하세요. 그리고 최적이 아니라면 다시 실행하세요. 최적의 매수시점을 찾을 때까지 반복하세요. + + + +#2. +전제 소스코들 살펴보세요. 데이터는 모든 코인에 동일 적용할 수 있도록 표준화가 되어야 합니다. 그리고 기술적 분석을 이용하여 매수 타이밍을 탐색해야 합니다. +혹시 기술적 분석이 아닌 날짜가 매수조건에 들어가 있다면 그러한 부분은 모두 제거해주세요. \ No newline at end of file diff --git a/README.md b/README.md index 2c0b339..e5085b1 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,11 @@ ## 개요 `AssetMonitor`는 주식‧ETF 및 암호화폐 시장을 실시간으로 감시하여 Bollinger Band, RSI, MACD, 이동평균(Golden-Cross), 거래량 등을 종합 분석한 **매수 후보(signals)**를 텔레그램으로 통보하는 자동화 봇입니다. +**주요 개선사항:** +- **데이터 표준화**: 모든 코인에 동일한 기술적 분석 기준 적용 +- **순수 기술적 분석**: 날짜 기반 조건 제거, 기술적 지표만 사용 +- **강화된 기술적 지표**: 스토캐스틱, MFI, OBV, ATR 등 추가 지표 활용 + --- ## 주요 구성 파일 @@ -20,20 +25,24 @@ 2. **데이터 획득** *주식 / ETF*: `FinanceDataReader` *암호화폐*: 빗썸 **240분 봉** Open API -3. **기술적 지표 계산** (`calculate_technical_indicators`) +3. **데이터 표준화** (`normalize_data`) + - 모든 코인에 동일한 정규화 적용 + - 20일 롤링 윈도우 기반 Min-Max 정규화 +4. **기술적 지표 계산** (`calculate_technical_indicators`) - Bollinger Band (기간 20, ±2σ) - RSI(14) - MACD(12-26-9) - 단/중/장기 이동평균선(MA5/20/60) - 거래량 MA5 -4. **매수 후보 판정** (`check_buy_signals`) -- *아래 새로운 “매수 후보 전략” 섹션 참조* -5. **알림 발송** (`send_*_telegram_message`) + - **추가 지표**: 스토캐스틱, OBV, ATR, MFI +5. **매수 후보 판정** (`check_buy_signals`) +- *아래 새로운 "매수 후보 전략" 섹션 참조* +6. **알림 발송** (`send_*_telegram_message`) multiprocessing Pool을 이용해 다중 메시지를 병렬로 전송합니다. --- -## 매수 후보 전략 +## 매수 후보 전략 (표준화된 기술적 분석) | 신호 | 변수명 | 조건 | 의미 | |------|--------|------|------| @@ -43,21 +52,26 @@ | 이동평균 골든크로스 | `ma_signal` | `이전 MA5 < 이전 MA20` **AND** `현재 MA5 ≥ 현재 MA20` | 단기 추세 ↗ 전환 | | 거래량 급증 | `volume_signal` | `현재 거래량 > MA5 Volume × 1.5` | 수급 증가 | | **U자 반등 돌파** | `breakout_signal` | ① 최근 `BREAKOUT_LOOKBACK`(30)개 캔들 동안 최고·최저가 차이가 `BUY_THRESHOLD`(15 %) 이상 하락 → ② **현재가가 그 최고가 돌파** | 하락 후 반등의 추세 전환 확인 | +| **장기 저항 돌파** | `long_breakout_signal` | 장기간 저항선 돌파 감지 | 장기 추세 전환 | +| **스토캐스틱 과매도** | `stoch_signal` | `%K < 20 AND %K > 이전 %K` | 스토캐스틱 과매도 반등 | +| **MFI 과매도** | `mfi_signal` | `MFI < 20 AND MFI > 이전 MFI` | 자금 흐름 과매도 반등 | +| **OBV 상승** | `obv_signal` | `현재 OBV > 이전 OBV × 1.1` | 거래량 가중 상승 | +| **ATR 급증** | `atr_signal` | `현재 ATR > ATR 20일 평균 × 1.5` | 변동성 급증 | ### 최종 매수 후보 결정 로직 ```text -if breakout_signal: - buy = True # U자 반등 돌파 단독으로도 매수 후보 +if breakout_signal or long_breakout_signal: + buy = True # 돌파 신호 단독으로도 매수 후보 else: - # ① 볼린저 + RSI 동시, 또는 ② (신호 ≥ 2개) & (볼린저 또는 RSI 포함) - buy = (bb_signal and rsi_signal) or (signal_count >= 2 and (bb_signal or rsi_signal)) + # ① 볼린저 + RSI 동시, 또는 ② (신호 ≥ 3개) & (볼린저 또는 RSI 포함) + buy = (bb_signal and rsi_signal) or (signal_count >= 3 and (bb_signal or rsi_signal)) ``` -*`signal_count` = 위 6개 신호 중 True 개수* +*`signal_count` = 위 11개 신호 중 True 개수* ### 메시지 구성 -- `🛒` : 최종 `buy=True`일 때 메시지 맨 앞에 부착 -- `📊신호(n):` 뒤에 활성화된 신호 목록 - - 볼린저/RSI/MACD/MA/거래량/Breakout 각각 표시 +- `매수` : 최종 `buy=True`일 때 메시지 맨 앞에 부착 +- `신호(n):` 뒤에 활성화된 신호 목록 + - 볼린저/RSI/MACD/MA/거래량/Breakout/스토캐스틱/MFI/OBV/ATR 각각 표시 해당 전략으로 **과매도 바닥근처 매수 기회 + 상승 추세 전환 브레이크아웃** 두 영역을 모두 포착할 수 있습니다. @@ -66,7 +80,7 @@ else: ## 스케줄 테이블 (기본값) | 대상 | 실행 시각(서버 기준) | 호출 함수 | |------|----------------------|-----------| -| KRW 코인 | 매시간 04, 34분 | `monitor_coins()` | +| KRW 코인 | 매시간 04, 14, 24, 34, 44, 54분 | `monitor_coins()` | | 미국 주식 / ETF | 05:10, 16:30, 23:30 | `monitor_us_stocks()` | | 한국 ETF / 주식 | 07:10, 18:20 | `monitor_kr_stocks()` | diff --git a/config.py b/config.py index 1e0a63a..91ecde6 100644 --- a/config.py +++ b/config.py @@ -27,9 +27,12 @@ KR_COINS = { "ADA": "ADA", "APE": "ApeCoin", "ARB": "Arbitrum", + "BONK": "BONK", "HBAR": "HBAR", "LINK": "Chainlink", "ONDO": "ONDO", + "PEPE": "PEPE", + "SEI": "SEI", "SHIB": "Shiba Inu", "STORJ": "Storj", "SUI": "Sui Network", diff --git a/requirements.txt b/requirements.txt index 6f2f0e7..eee5c45 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ yfinance pandas +mplcursors numpy ccxt PyJWT diff --git a/stock_monitor.py b/stock_monitor.py index 35f1530..e47bc5e 100644 --- a/stock_monitor.py +++ b/stock_monitor.py @@ -1,7 +1,6 @@ import pandas as pd - from HTS2 import HTS -import pandas as pd +from dateutil.relativedelta import relativedelta from datetime import datetime, timedelta import telegram import time @@ -12,9 +11,11 @@ from multiprocessing import Pool import schedule from config import * import FinanceDataReader as fdr +import numpy as np hts = HTS() -BUY_AMOUNT = 100000 +BUY_AMOUNT = 10000 + def send_coin_msg(text): coin_client = telegram.Bot(token=COIN_TELEGRAM_BOT_TOKEN) @@ -38,10 +39,11 @@ def send_coin_telegram_message(message_list, header): return -def buy_ticker(buy_ticker_list): - for buy_ticker in buy_ticker_list: - ticker_code = buy_ticker['symbol'] - _ = hts.buyCoinMarket(ticker_code, BUY_AMOUNT) +def buy_ticker(symbole): + try: + _ = hts.buyCoinMarket(symbole, BUY_AMOUNT) + except Exception as e: + print(f"Error buying {symbole}: {str(e)}") return @@ -69,232 +71,101 @@ 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() - data['Upper'] = data['MA'] + (BOLLINGER_STD * data['STD']) - data['Lower'] = data['MA'] - (BOLLINGER_STD * data['STD']) - return data - +def normalize_data(data): + """데이터 정규화 함수 - 모든 코인에 동일하게 적용""" + # Min-Max 정규화를 위한 컬럼 + columns_to_normalize = ['Open', 'High', 'Low', 'Close', 'Volume'] + + normalized_data = data.copy() + + # 각 컬럼별 정규화 (20일 롤링 윈도우 사용) + for column in columns_to_normalize: + min_val = data[column].rolling(window=20).min() + max_val = data[column].rolling(window=20).max() + # 0으로 나누기 방지 + denominator = max_val - min_val + normalized_data[f'{column}_Norm'] = np.where( + denominator != 0, + (data[column] - min_val) / denominator, + 0.5 # 기본값 설정 + ) + + return normalized_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['MACD_Hist'] = data['MACD'] - data['Signal'] - - # 이동평균선 + """기술적 지표 계산 - 모든 코인에 동일하게 적용""" + # 데이터 정규화 + data = normalize_data(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() + data['MA120'] = data['Close'].rolling(window=120).mean() + data['MA200'] = data['Close'].rolling(window=200).mean() + data['MA240'] = data['Close'].rolling(window=240).mean() + data['MA720'] = data['Close'].rolling(window=720).mean() + data['MA1440'] = data['Close'].rolling(window=1440).mean() - # 거래량 이동평균 - data['Volume_MA5'] = data['Volume'].rolling(window=5).mean() + # 매수 타이밍을 이동평균선으로 결정 + # 골든크로스: 단기 이동평균선이 장기 이동평균선을 상향 돌파할 때 매수 + data['golden_cross'] = (data['MA5'] > data['MA20']) & (data['MA5'].shift(1) <= data['MA20'].shift(1)) + + # 볼린저 밴드 계산 + data['MA'] = data['Close'].rolling(window=20).mean() + data['STD'] = data['Close'].rolling(window=20).std() + data['Upper'] = data['MA'] + (2 * data['STD']) + data['Lower'] = data['MA'] - (2 * data['STD']) return data -def check_buy_signals(symbol, data): - if len(data) < 60: # 최소 60일치 데이터 필요 - return None +def check_buy_point(data, simulation=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): - upper_band = latest['Upper'] - lower_band = latest['Lower'] - current_price = latest['Close'] + # 매수 포인트 탐지 및 표시 + if simulation: + recent_data = data else: - upper_band = latest['Upper'].iloc[0] - lower_band = latest['Lower'].iloc[0] - current_price = latest['Close'].iloc[0] + recent_data = data.tail(10) - distance = (current_price - lower_band) / (upper_band - lower_band) - bb_signal = distance < BOLLINGER_THRESHOLD + recent_data['buy_point'] = 0 + 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 - # 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] - prev_high = window_close.max() - prev_low = window_close.min() + if not simulation: + if recent_data['buy_point'][-10:-1].sum() > 0: + recent_data['buy_point'][-1] = 1 - # ② 1주일(42캔들) 전 가격 대비 5% 이상 상승하지 않았는지 체크 - 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: - week_change = 1 # 값이 0이면 조건 불충족 처리 + return recent_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 - ): - 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] - # 과거 구간에서 저항선 이상으로 종가가 한번도 올라가지 않은 경우 + 현재 가격이 저항선 돌파 - if (previous_closes <= resistance_level * (1 + RESISTANCE_BREAK_THRESHOLD)).all() and \ - 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]) - - # 이동평균선 골든크로스 임박 또는 발생 - 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, - 'breakout_signal': breakout_signal, - 'long_breakout_signal': long_breakout_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, - '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': 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 - - # MACD 신호 (MACD가 시그널 라인을 상향 돌파) - macd_signal = (prev['MACD'] < prev['Signal']) and (latest['MACD'] > latest['Signal']) - - # 이동평균선 골든크로스 임박 또는 발생 - ma_signal = (prev['MA5'] < prev['MA20']) and (latest['MA5'] >= latest['MA20']) - - # 거래량 증가 신호 (5일 평균 대비 150% 이상) - volume_signal = latest['Volume'] > (latest['Volume_MA5'] * 1.5) - - # 종합 신호 - buy_signals = { - 'bb_signal': bb_signal, - 'rsi_signal': rsi_signal, - 'macd_signal': macd_signal, - 'ma_signal': ma_signal, - 'volume_signal': volume_signal, - 'breakout_signal': breakout_signal, - 'long_breakout_signal': long_breakout_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, - 'rsi': latest['RSI'], - 'macd': latest['MACD'], - '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))) - } - - -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()): - message += "📊신호 ({count}):" - if info['buy_signals']['bb_signal']: - message += "- 볼린저 밴드 하단 근접 (근접도: {:.1f}%),".format(info['distance'] * 100) - count += 1 - if info['buy_signals']['rsi_signal']: - message += f"- RSI 과매도 구간 (RSI: {info['rsi']:.1f})," - count += 1 - if info['buy_signals']['macd_signal']: - message += "- MACD 골든크로스," - count += 1 - if info['buy_signals']['ma_signal']: - message += "- 이동평균선 골든크로스," - count += 1 - if info['buy_signals']['volume_signal']: - message += "- 거래량 급증" - count += 1 - if info['buy_signals'].get('breakout_signal'): - message += "- U자 반등 돌파" - count += 1 - if info['buy_signals'].get('long_breakout_signal'): - message += "- 장기 저항 돌파" - count += 1 - message += "\n" - message = message.replace("{count}", str(count)) +def format_message(market_type, symbol, symbol_name, close): + message = f"매수 [{market_type}] {symbol_name} ({symbol}) " + message += f"현재가: {'$' if market_type == 'US' else '₩'}{close:.2f}, " return message def format_ma_message(info, market_type): """MA 알림 메시지 생성""" - prefix = '📈 ' if info.get('alert') else '' + 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): +def get_coin_data(symbol, interval=240, to=None, retries=3): for attempt in range(retries): try: - url = "https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=3000".format(interval, symbol) + #url = "https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=3000".format(interval, symbol) + if to is None: + url = ("https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=200").format(interval, symbol) + else: + url = ("https://api.bithumb.com/v1/candles/minutes/{}?market=KRW-{}&count=200&to={}").format(interval, symbol, to) + + + #url = 'https://api.bithumb.com/v1/candles/minutes/60?market=KRW-ADA&count=200' + #url = 'https://api.bithumb.com/v1/candles/minutes/minutes/60?market=KRW-ADA&count=200&to=2025-08-06 10:38:38' headers = {"accept": "application/json"} response = requests.get(url, headers=headers) json_data = json.loads(response.text) @@ -328,6 +199,27 @@ def get_coin_data(symbol, interval=240, retries=3): continue return None +def get_coin_more_data(symbol, interval, bong_count=3000): + # 코인 데이터 1500개 봉 가져오기 + to = datetime.now() + data = None + while data is None or len(data) < bong_count: + if data is None: + data = get_coin_data(symbol, interval, to.strftime("%Y-%m-%d %H:%M:%S")) + else: + df = get_coin_data(symbol, interval, to.strftime("%Y-%m-%d %H:%M:%S")) + data = pd.concat([data, df], ignore_index=True) + time.sleep(0.3) + to = to - relativedelta(minutes=interval * 200) + + data = data.set_index('datetime') + data = data.sort_index() + data = data.drop_duplicates(keep='first') + data["datetime"] = data.index + # 코인 데이터 1500개 봉 가져오기 + + return data + def get_kr_stock_data(symbol, retries=3): for attempt in range(retries): @@ -368,14 +260,11 @@ def monitor_us_stocks(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - info = detect_turnaround_signal(symbol, data, 0) - if info is None: + recent_data = check_buy_point(data) # Changed to check_buy_point + if recent_data['buy_point'][-1] != 1: 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')) + print(f" - {US_STOCKS[symbol]} ({symbol}): {recent_data['Close'][-1]:.2f}") + message_list.append(format_message('US', symbol, US_STOCKS[symbol], recent_data['Close'][-1])) except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") time.sleep(0.5) @@ -402,14 +291,11 @@ def monitor_kr_stocks(): if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - info = detect_turnaround_signal(symbol, data, 0) - if info is None: + recent_data = check_buy_point(data) # Changed to check_buy_point + if recent_data['buy_point'][-1] != 1: continue - info['name'] = KR_ETFS[symbol] - print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']}") - - if info['alert']: - message_list.append(format_ma_message(info, 'KR')) + print(f" - {KR_ETFS[symbol]} ({symbol}): {recent_data['Close'][-1]:.2f}") + message_list.append(format_message('KR', symbol, US_STOCKS[symbol], recent_data['Close'][-1])) except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") @@ -434,25 +320,25 @@ def monitor_kr_stocks(): def monitor_coins(): message_list = [] - buy_ticker_list = [] print("KRW COINs {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) for symbol in KR_COINS: + # 1시간 interval = 60 - data = get_coin_data(symbol, interval=interval) + data = get_coin_more_data(symbol, interval) + if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - info = detect_turnaround_signal(symbol, data, interval) - if info is None: + recent_data = check_buy_point(data) # Changed to check_buy_point + if recent_data['buy_point'][-1] != 1: continue - info['name'] = KR_COINS[symbol] - print(f" - {info['name']} ({symbol}): {info['price']:.2f} -> {info['alert']} ({info['details']['interval']})") + print(f" - {KR_ETFS[symbol]} ({symbol}): {recent_data['Close'][-1]:.2f}") + message_list.append(format_message('COIN', symbol, US_STOCKS[symbol], recent_data['Close'][-1])) + # buy + buy_ticker(symbol) - if info['alert']: - message_list.append(format_ma_message(info, 'KR')) - buy_ticker_list.append(info) except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") else: @@ -461,18 +347,19 @@ def monitor_coins(): # 4시간 interval = 240 - data = get_coin_data(symbol, interval=interval) + data = get_coin_more_data(symbol, interval, bong_count=1500) + if data is not None and not data.empty: try: data = calculate_technical_indicators(data) - info = detect_turnaround_signal(symbol, data, interval) + info = check_buy_point(data, simulation=True) # Changed to check_buy_point 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['buy']}") - if info['alert']: - message_list.append(format_ma_message(info, 'KR')) + if info['buy']: + message_list.append(format_message(info, 'KR')) except Exception as e: print(f"Error processing data for {symbol}: {str(e)}") else: @@ -481,8 +368,6 @@ def monitor_coins(): if len(message_list) > 0: try: - # buy - buy_ticker(buy_ticker_list) # send message send_coin_telegram_message(message_list, header="[KRW-COIN]") except Exception as e: @@ -494,70 +379,19 @@ def monitor_coins(): # Turnaround Detector v6 # ---------------------- -def detect_turnaround_signal(symbol, data, interval=0): +def detect_turnaround_signal(symbol, data, interval=0, params=None): if len(data) < 7: return None - cur = data.iloc[-1] + # 이동평균을 기반으로 매수 신호 결정 + 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 회복) - # 볼린저 밴드 포지션 - distance = (cur.Close - cur.Lower) / (cur.Upper - cur.Lower) - - # 매수 신호 조건: 전봉 하단 밴드 터치 + 반등 + MA5 돌파 + RSI 회복 + 밴드 위치 ≤ 0.35 - alert = touch_lower and rebound and ma5_break and rsi_recover and (distance <= 0.35) - - 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, - }, - } - + return None def run_schedule(): + monitor_coins() + # 코인 모니터링 스케줄 (매시간 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) diff --git a/stock_simulation.py b/stock_simulation.py index 72766ea..f88aaf0 100644 --- a/stock_simulation.py +++ b/stock_simulation.py @@ -1,14 +1,16 @@ -import math -import requests - +import time +from datetime import datetime +from dateutil.relativedelta import relativedelta import pandas as pd import yfinance as yf import matplotlib.pyplot as plt +import matplotlib.dates +import mplcursors plt.rcParams['font.family'] ='AppleGothic' plt.rcParams['axes.unicode_minus'] =False from config import * -from stock_monitor import calculate_technical_indicators, detect_turnaround_signal +from stock_monitor import calculate_technical_indicators, detect_turnaround_signal, get_coin_more_data, check_buy_point # 비트/알트코인 KRW 마켓 식별: 문자열 "-KRW" 포함 여부로 간단 구분 @@ -19,50 +21,12 @@ INTERVAL_MAP = { BITHUMB_MAX_COUNT = 3000 # API 최대 3000 캔들 +def fetch_price_history(symbol: str, interval_minutes: int, days: int = 30) -> pd.DataFrame: -def fetch_coin_history_bithumb(symbol: str, interval_minutes: int, days: int) -> pd.DataFrame: - """빗썸 API를 이용해 최근 `days`일 코인 데이터 수집 (interval 60 / 240)""" - if interval_minutes not in (60, 240): - raise ValueError("Bithumb API only supports 60 or 240 minutes in this helper") - - minutes = interval_minutes - count = int(math.ceil(days * 24 * 60 / minutes)) + 10 # 여유분 10 캔들 - count = min(count, BITHUMB_MAX_COUNT) - - url = f"https://api.bithumb.com/v1/candles/minutes/{minutes}?market=KRW-{symbol}&count={count}" - res = requests.get(url, timeout=5) - res.raise_for_status() - raw = res.json() - - if not isinstance(raw, list) or len(raw) == 0: - raise RuntimeError("Empty response from Bithumb API") - - df_temp = pd.DataFrame(raw) - # API 반환: [timestamp, open, close, high, low, volume] 순 - df_temp = df_temp.sort_index(ascending=False) # 최신순, 뒤집어서 역순 전달 - - data = pd.DataFrame() - # data.columns = ['datetime', 'open', 'close', 'high', 'low', 'volume'] - # data['datetime'] = pd.to_datetime(data_temp['candle_date_time_kst']) - data['datetime'] = pd.to_datetime(df_temp['candle_date_time_kst'], format='%Y-%m-%dT%H:%M:%S') - data['Open'] = df_temp['opening_price'] - data['Close'] = df_temp['trade_price'] - data['High'] = df_temp['high_price'] - data['Low'] = df_temp['low_price'] - data['Volume'] = df_temp['candle_acc_trade_volume'] - data = data.set_index('datetime') - data = data.astype(float) - data["datetime"] = data.index - - data = data.set_index("datetime").sort_index() # 시간 오름차순 - return data - - -def fetch_price_history(symbol: str, interval_minutes: int, days: int = 7) -> pd.DataFrame: """최근 `days`일 데이터(캔들)를 가져온다. 코인(-KRW)은 빗썸, 그 외 yfinance.""" if symbol in KR_COINS: - base_symbol = symbol.replace("-KRW", "") - return fetch_coin_history_bithumb(base_symbol, interval_minutes, days) + bong_count = 3000 + return get_coin_more_data(symbol, interval_minutes, bong_count=bong_count) # -------- 주식/ETF/해외코인 (yfinance) -------- if interval_minutes not in INTERVAL_MAP: @@ -83,47 +47,198 @@ def fetch_price_history(symbol: str, interval_minutes: int, days: int = 7) -> pd return df -def run_simulation(symbol: str, interval_minutes: int, days: int = 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) - - alerts = [] # (timestamp, price) - # 시계열 순회하며 알림 조건 체크 - for i in range(len(data)): - slice_df = data.iloc[: i + 1] + + print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}") + print(f"총 데이터 수: {len(data)}") + + # 저점 기간 필터링 (6월 22일~7월 9일) + bottom_start = pd.Timestamp('2025-06-22') + bottom_end = pd.Timestamp('2025-07-09') + + bottom_data = data[(data.index >= bottom_start) & (data.index <= bottom_end)] + + if len(bottom_data) == 0: + print("저점 기간 데이터가 없습니다.") + return + + print(f"\n저점 기간 데이터: {bottom_data.index[0]} ~ {bottom_data.index[-1]}") + print(f"저점 기간 데이터 수: {len(bottom_data)}") + + # 저점 기간의 기술적 지표 분석 + print("\n=== 저점 기간 기술적 지표 분석 ===") + + # 1. 가격 분석 + min_price = bottom_data['Low'].min() + max_price = bottom_data['High'].max() + avg_price = bottom_data['Close'].mean() + + print(f"최저가: {min_price:.4f}") + print(f"최고가: {max_price:.4f}") + print(f"평균가: {avg_price:.4f}") + print(f"가격 변동폭: {((max_price - min_price) / min_price * 100):.2f}%") + + # 3. 볼린저 밴드 분석 + bb_lower_min = bottom_data['Lower'].min() + bb_upper_max = bottom_data['Upper'].max() + + print(f"\n볼린저 밴드 분석:") + print(f"하단 밴드 최저: {bb_lower_min:.4f}") + print(f"상단 밴드 최고: {bb_upper_max:.4f}") + + # 4. 거래량 분석 + volume_avg = bottom_data['Volume'].mean() + volume_max = bottom_data['Volume'].max() + + print(f"\n거래량 분석:") + print(f"평균 거래량: {volume_avg:.0f}") + print(f"최대 거래량: {volume_max:.0f}") + + # 5. 실제 저점 찾기 + actual_bottom_idx = bottom_data['Low'].idxmin() + actual_bottom_price = bottom_data.loc[actual_bottom_idx, 'Low'] + actual_bottom_date = actual_bottom_idx + + print(f"\n실제 저점:") + print(f"날짜: {actual_bottom_date}") + print(f"가격: {actual_bottom_price:.4f}") + # 실제 저점에서 RSI 출력 제거 + # print(f"RSI: {bottom_data.loc[actual_bottom_idx, 'RSI']:.2f}") + print(f"볼린저 하단 대비: {((actual_bottom_price - bottom_data.loc[actual_bottom_idx, 'Lower']) / bottom_data.loc[actual_bottom_idx, 'Lower'] * 100):.2f}%") + + # 6. 매수 신호 분석 + print(f"\n=== 매수 신호 분석 ===") + + # 현재 매수 조건으로 저점 기간에서 매수 신호가 몇 개 발생하는지 확인 + alerts = [] + debug_info = [] + + for i in range(len(bottom_data)): + slice_df = data.iloc[:data.index.get_loc(bottom_data.index[i]) + 1] info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes) - if info and info["alert"]: - alerts.append((slice_df.index[-1], slice_df["Close"].iloc[-1])) + + if info: + debug_info.append({ + 'date': bottom_data.index[i], + 'price': bottom_data['Close'].iloc[i], + 'alert': info['alert'], + 'details': info['details'] + }) + + if info['alert']: + alerts.append((bottom_data.index[i], bottom_data['Close'].iloc[i])) + + print(f"저점 기간 매수 신호 수: {len(alerts)}") + + if alerts: + print("매수 신호 발생 시점:") + for date, price in alerts: + print(f" {date}: {price:.4f}") + + return bottom_data, alerts - # 모든 매수 신호를 표시 - # 기존 필터 제거하여 전체 기간 매수 신호 사용 +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) + + print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}") + print(f"총 데이터 수: {len(data)}") + + # 파라미터 후보군 (표준화된 기술적 분석 기준) + param_candidates = [ + {'rsi_oversold': 32, 'bb_distance': 0.06, 'near_low_tolerance': 0.01, 'volume_multiplier': 2.0}, # 기본 설정 + {'rsi_oversold': 30, 'bb_distance': 0.05, 'near_low_tolerance': 0.008, 'volume_multiplier': 2.5}, # 엄격한 설정 + {'rsi_oversold': 28, 'bb_distance': 0.04, 'near_low_tolerance': 0.005, 'volume_multiplier': 3.0}, # 매우 엄격한 설정 + ] + + alerts = [] + + for params in param_candidates: + alerts.clear() + for i in range(len(data)): + slice_df = data.iloc[: i + 1] + info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes, params=params) + if info and info['alert']: + alerts.append((slice_df.index[-1], slice_df['Close'].iloc[-1])) + + print(f"\n총 매수 신호 수: {len(alerts)}") + + # 서브플롯 생성 + fig, ax1 = plt.subplots(figsize=(15, 8)) + fig.suptitle(f"{symbol} - 시뮬레이션 {interval_minutes}분봉", fontsize=14) + + # 메인 차트 (가격, 이동평균선, 볼린저 밴드) + line_close = ax1.plot(data.index, data["Close"], label="종가", color="black", linewidth=1.5)[0] + line_ma5 = ax1.plot(data.index, data["MA5"], label="MA5", color="red", linewidth=1)[0] + line_ma20 = ax1.plot(data.index, data["MA20"], label="MA20", color="blue", linewidth=1)[0] + line_ma40 = ax1.plot(data.index, data["MA40"], label="MA40", color="green", linewidth=1)[0] + line_ma120 = ax1.plot(data.index, data["MA120"], label="MA120", color="purple", linewidth=1)[0] + line_ma200 = ax1.plot(data.index, data["MA200"], label="MA200", color="brown", linewidth=1)[0] + line_ma240 = ax1.plot(data.index, data["MA240"], label="MA240", color="black", linewidth=1)[0] + line_ma720 = ax1.plot(data.index, data["MA720"], label="MA720", color="cyan", linewidth=1)[0] + line_ma1440 = ax1.plot(data.index, data["MA1440"], label="MA1440", color="magenta", linewidth=1)[0] - # Plot - plt.figure(figsize=(12, 6)) - plt.plot(data.index, data["Close"], label="종가", color="black") - plt.plot(data.index, data["MA5"], label="MA5", color="orange", linewidth=1) - plt.plot(data.index, data["MA20"], label="MA20", color="blue", linewidth=1) - plt.plot(data.index, data["MA40"], label="MA40", color="green", linewidth=1) # Bollinger Bands - plt.plot(data.index, data["Upper"], label="볼린저 Upper", color="grey", linestyle="--", linewidth=1) - plt.plot(data.index, data["Lower"], label="볼린저 Lower", color="grey", linestyle="--", linewidth=1) - plt.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1) + line_upper = ax1.plot(data.index, data["Upper"], label="볼린저 Upper", color="grey", linestyle="--", linewidth=1)[0] + line_lower = ax1.plot(data.index, data["Lower"], label="볼린저 Lower", color="grey", linestyle="--", linewidth=1)[0] + + ax1.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1) + # 매수 신호 표시 + scatter_buy = None if alerts: times, prices = zip(*alerts) - plt.scatter(times, prices, facecolors='none', edgecolors='red', linewidths=2, s=150, zorder=6, label='매수신호') + scatter_buy = ax1.scatter(times, prices, facecolors='none', edgecolors='red', linewidths=2, s=150, zorder=6, label='매수신호') + for time in times: + ax1.axvline(x=time, color='red', linestyle='--', alpha=0.3) + + # 매수 포인트 탐지 및 표시 + # 'buy_point' 열 추가 + data['buy_point'] = 0 + for i in range(1, len(data)): + if all(data[f'MA{n}'].iloc[i] < data['MA720'].iloc[i] for n in [5, 20, 40, 120, 200, 240]) and \ + all(data[f'MA{n}'].iloc[i] > data[f'MA{n}'].iloc[i-1] for n in [5, 20, 40, 120, 200, 240]) and \ + data['MA720'].iloc[i] < data['MA1440'].iloc[i]: + data.at[data.index[i], 'buy_point'] = 1 + + # 매수 포인트를 빨간 동그라미로 표시 + buy_points = data[data['buy_point'] == 1] + scatter_buy_points = ax1.scatter(buy_points.index, buy_points['Close'], color='red', s=100, zorder=5, label='매수 포인트') + + # 마우스 오버 기능 추가 + cursor = mplcursors.cursor(scatter_buy_points, hover=True) + cursor.connect("add", lambda sel: sel.annotation.set_text( + f'날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}' + )) + cursor.connect("remove", lambda sel: sel.annotation.set_visible(False)) + + ax1.set_ylabel("가격") + ax1.legend(loc='upper left', fontsize=10) + ax1.grid(True, linestyle='--', alpha=0.5) - plt.title(f"{symbol} – 시뮬레이션 {interval_minutes}분봉 (최근 {days}일)") - plt.xlabel("날짜") - plt.ylabel("가격") - plt.legend() - plt.grid(True) plt.tight_layout() plt.show() if __name__ == "__main__": - symbol = 'WLD' interval = 60 - days = 7 - run_simulation(symbol, interval, days) + 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'] + + for symbol in target_coins: + print(f"\n=== {symbol} 저점 기간 분석 시작 ===") + try: + # 저점 기간 분석 + bottom_data, alerts = analyze_bottom_period(symbol, interval, days) + + # 전체 기간 시뮬레이션 + print(f"\n=== {symbol} 전체 기간 시뮬레이션 ===") + run_simulation(symbol, interval, days) + + except Exception as e: + print(f"Error analyzing {symbol}: {str(e)}")