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-08-04 21:36:29 +09:00
parent 8f542085bc
commit 3f6ca0ce47
3 changed files with 295 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
stock_monitor.py
View File

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