dsyoon/DeepCoin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
dsyoon
2025-05-01 10:18:47 +09:00
parent 52c3ca7e1c
commit f92b1881ec
2 changed files with 120 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
stock_monitor.py
View File

@@ -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)