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-07 13:15:19 +09:00
parent f861487107
commit 59ebb15b6c
3 changed files with 129 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
config.py
View File

@@ -32,6 +32,7 @@ KR_COINS = {
"LINK": "Chainlink",
"ONDO": "ONDO",
"PEPE": "PEPE",
"POL": "POL",
"SEI": "SEI",
"SHIB": "Shiba Inu",
"STORJ": "Storj",

174
stock_monitor.py
View File

@@ -74,46 +74,6 @@ def send_coin_telegram_message(message_list, header):
return
def buy_ticker(symbol, data):
try:
# 매수 금지 시간 확인 (20분)
current_time = datetime.now()
if symbol in buy_cooldown:
time_diff = current_time - buy_cooldown[symbol]
if time_diff.total_seconds() < 1200: # 20분 = 1200초
print(f"{symbol}: 매수 금지 중 (남은 시간: {1200 - time_diff.total_seconds():.0f}초)")
return False
BUY_AMOUNT = 6000
if data['buy_signal'].iloc[-1] == 'movingaverage':
BUY_AMOUNT = 70000
elif data['buy_signal'].iloc[-1] == 'deviation40':
BUY_AMOUNT = 40000
elif data['buy_signal'].iloc[-1] == 'deviation240':
BUY_AMOUNT = 6000
_ = hts.buyCoinMarket(symbol, BUY_AMOUNT)
# 매수 성공 시 금지 시간 설정 및 파일에 저장
buy_cooldown[symbol] = current_time
save_buy_cooldown(buy_cooldown)
print(f"{KR_COINS[symbol]} ({symbol}): {data['Close'].iloc[-1]:.2f}: 매수 완료, 20분간 매수 금지 시작")
try:
pool = Pool(12)
pool.map(send_coin_msg, ["[KRW-COIN]" + "\n" + format_message('COIN', symbol, KR_COINS[symbol], data['Close'].iloc[-1], data['buy_signal'].iloc[-1])])
except Exception as e:
print(f"Error sending Telegram message: {str(e)}")
return True
except Exception as e:
print(f"Error buying {symbol}: {str(e)}")
return False
return
def send_stock_msg(text):
stock_client = telegram.Bot(token=STOCK_TELEGRAM_BOT_TOKEN)
@@ -196,34 +156,51 @@ def calculate_technical_indicators(data):
return data
def check_buy_point(data, simulation=None):
"""
# 매수 포인트 탐지 및 표시
if simulation:
recent_data = data
else:
# recent_data의 복사본 생성
recent_data = data.tail(10).copy()
def buy_ticker(symbol, data):
try:
# 매수 금지 시간 확인 (20분)
current_time = datetime.now()
if symbol in buy_cooldown:
time_diff = current_time - buy_cooldown[symbol]
if time_diff.total_seconds() < 1200: # 20분 = 1200초
print(f"{symbol}: 매수 금지 중 (남은 시간: {1200 - time_diff.total_seconds():.0f}초)")
return False
# 'buy_point' 열 초기화
recent_data['buy_point'] = 0
BUY_AMOUNT = 6000
# FutureWarning 해결
if recent_data['buy_point'].iloc[-1] != 1:
# 코드 계속
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
if data['buy_signal'].iloc[-1] == 'movingaverage':
BUY_AMOUNT = 10000
elif data['buy_signal'].iloc[-1] == 'deviation40':
BUY_AMOUNT = 15000
elif data['buy_signal'].iloc[-1] == 'deviation240':
BUY_AMOUNT = 6000
elif data['buy_signal'].iloc[-1] == 'deviation1440':
if symbol in ['BONK', 'PEPE', 'TON']:
BUY_AMOUNT = 30000
else:
BUY_AMOUNT = 50000
if not simulation:
if recent_data['buy_point'][-10:-1].sum() > 0:
recent_data.at[recent_data.index[-1], 'buy_point'] = 1
_ = hts.buyCoinMarket(symbol, BUY_AMOUNT)
return recent_data
"""
# 매수 성공 시 금지 시간 설정 및 파일에 저장
buy_cooldown[symbol] = current_time
save_buy_cooldown(buy_cooldown)
print(f"{KR_COINS[symbol]} ({symbol}): {data['Close'].iloc[-1]:.4f}: 매수 완료, 20분간 매수 금지 시작")
try:
pool = Pool(12)
pool.map(send_coin_msg, ["[KRW-COIN]" + "\n" + format_message('COIN', symbol, KR_COINS[symbol], data['Close'].iloc[-1], data['buy_signal'].iloc[-1])])
except Exception as e:
print(f"Error sending Telegram message: {str(e)}")
except Exception as e:
print(f"Error buying {symbol}: {str(e)}")
return False
return True
def check_buy_point(symbol, data, simulation=None):
# 매수 포인트 탐지 및 표시
# 데이터 복사본 생성하여 SettingWithCopyWarning 방지
@@ -256,20 +233,67 @@ def check_buy_point(data, simulation=None):
data.at[data.index[-1], 'buy_signal'] = 'deviation40'
data.at[data.index[-1], 'buy_point'] = 1
if symbol not in ['BONK']:
# BONK는 240 체크하지 않음
if symbol in ['TRX']:
# Deviation240(이격도 240) 기반 매수 조건: 90 이하에서 상승 전환
if data['Deviation240'].iloc[i - 1] < data['Deviation240'].iloc[i] and data['Deviation240'].iloc[i - 1] <= 98:
data.at[data.index[i], 'buy_signal'] = 'deviation240'
data.at[data.index[i], 'buy_point'] = 1
if not simulation:
if data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation240'
data.at[data.index[-1], 'buy_point'] = 1
else:
# Deviation240(이격도 240) 기반 매수 조건: 90 이하에서 상승 전환
if data['Deviation240'].iloc[i - 1] < data['Deviation240'].iloc[i] and data['Deviation240'].iloc[i - 1] <= 90:
data.at[data.index[i], 'buy_signal'] = 'deviation240'
data.at[data.index[i], 'buy_point'] = 1
if not simulation:
if data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation240'
data.at[data.index[-1], 'buy_point'] = 1
# Deviation240(이격도 240) 기반 매수 조건: 90 이하에서 상승 전환
if data['Deviation240'].iloc[i - 1] < data['Deviation240'].iloc[i] and data['Deviation240'].iloc[i - 1] <= 90:
data.at[data.index[i], 'buy_signal'] = 'deviation240'
data.at[data.index[i], 'buy_point'] = 1
if not simulation:
if data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation240'
data.at[data.index[-1], 'buy_point'] = 1
if symbol in ['TON']:
if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 89:
data.at[data.index[i], 'buy_signal'] = 'deviation1440'
data.at[data.index[i], 'buy_point'] = 1
if not simulation:
if data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
elif symbol in ['XRP']:
if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 90:
data.at[data.index[i], 'buy_signal'] = 'deviation1440'
data.at[data.index[i], 'buy_point'] = 1
if not simulation:
if data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
elif symbol in ['BONK']:
if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 76:
data.at[data.index[i], 'buy_signal'] = 'deviation1440'
data.at[data.index[i], 'buy_point'] = 1
if not simulation:
if data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
else:
if data['Deviation1440'].iloc[i - 1] < data['Deviation1440'].iloc[i] and data['Deviation1440'].iloc[i - 1] <= 80:
data.at[data.index[i], 'buy_signal'] = 'deviation1440'
data.at[data.index[i], 'buy_point'] = 1
if not simulation:
if data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
return data
def format_message(market_type, symbol, symbol_name, close, buy_signal):
message = f"매수 [{market_type}] {symbol_name} ({symbol}): {buy_signal} "
message += f"현재가: {'$' if market_type == 'US' else '₩'}{close:.2f}, "
message += f"현재가: {'$' if market_type == 'US' else '₩'}{close:.4f}, "
return message
@@ -277,7 +301,7 @@ 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"
message += f"현재가: {'$' if market_type == 'US' else '₩'}{info['price']:.4f} \n"
return message
@@ -432,7 +456,7 @@ def monitor_us_stocks():
if data is not None and not data.empty:
try:
data = calculate_technical_indicators(data)
recent_data = check_buy_point(data) # Changed to check_buy_point
recent_data = check_buy_point(symbol, data) # Changed to check_buy_point
if recent_data['buy_point'].iloc[-1] != 1:
continue
print(f" - {US_STOCKS[symbol]} ({symbol}): {recent_data['Close'].iloc[-1]:.2f}")
@@ -463,7 +487,7 @@ def monitor_kr_stocks():
if data is not None and not data.empty:
try:
data = calculate_technical_indicators(data)
recent_data = check_buy_point(data) # Changed to check_buy_point
recent_data = check_buy_point(symbol, data) # Changed to check_buy_point
if recent_data['buy_point'].iloc[-1] != 1:
continue
print(f" - {KR_ETFS[symbol]} ({symbol}): {recent_data['Close'].iloc[-1]:.2f}")
@@ -503,7 +527,7 @@ def monitor_coins():
if data is not None and not data.empty:
try:
data = calculate_technical_indicators(data)
recent_data = check_buy_point(data) # Changed to check_buy_point
recent_data = check_buy_point(symbol, data) # Changed to check_buy_point
if recent_data['buy_point'].iloc[-1] != 1:
continue

34
stock_simulation.py
View File

@@ -69,7 +69,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)
data = check_buy_point(data, simulation=True) # 한 번만 계산
data = check_buy_point(symbol, data, simulation=True) # 한 번만 계산
print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}")
print(f"총 데이터 수: {len(data)}")
@@ -145,7 +145,7 @@ def analyze_bottom_period(symbol: str, interval_minutes: int, days: int = 90):
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)
data = check_buy_point(symbol, data, simulation=True)
print(f"데이터 기간: {data.index[0]} ~ {data.index[-1]}")
print(f"총 데이터 수: {len(data)}")
@@ -162,10 +162,12 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
ma_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'movingaverage')])
dev40_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation40')])
dev240_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation240')])
dev1440_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation1440')])
print(f" - MA 신호: {ma_signals}")
print(f" - Dev40 신호: {dev40_signals}")
print(f" - Dev240 신호: {dev240_signals}")
print(f" - Dev1440 신호: {dev1440_signals}")
# 서브플롯 생성 (가격 + Deviation)
fig, (ax1, ax2) = plt.subplots(2, 1, sharex=True, figsize=(15, 8), height_ratios=[3, 1])
@@ -283,6 +285,16 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
for time in dev240_buy_points.index:
ax1.axvline(x=time, color='blue', linestyle='--', alpha=0.5, linewidth=1)
# Deviation1440 기반 매수 포인트
dev1440_buy_points = data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation1440')]
scatter_dev1440_buy_points = None
if len(dev1440_buy_points) > 0:
scatter_dev1440_buy_points = ax1.scatter(dev1440_buy_points.index, dev1440_buy_points['Close'],
facecolors='none', edgecolors='purple', linestyle='--',
linewidth=2, s=200, zorder=10, label='Dev1440 매수 포인트')
for time in dev1440_buy_points.index:
ax1.axvline(x=time, color='purple', linestyle='--', alpha=0.5, linewidth=1)
# 마우스 오버 기능 추가 (이동평균선 매수 포인트)
@@ -309,6 +321,14 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
))
cursor_dev240.connect("remove", lambda sel: sel.annotation.set_visible(False))
# 마우스 오버 기능 추가 (Deviation1440 매수 포인트)
if scatter_dev1440_buy_points is not None:
cursor_dev1440 = mplcursors.cursor(scatter_dev1440_buy_points, hover=True)
cursor_dev1440.connect("add", lambda sel: sel.annotation.set_text(
f'Dev1440 매수신호\n날짜: {matplotlib.dates.num2date(sel.target[0]).replace(tzinfo=None).strftime("%Y-%m-%d %H:%M")}\n가격: {sel.target[1]:.2f}'
))
cursor_dev1440.connect("remove", lambda sel: sel.annotation.set_visible(False))
# 모든 봉에 마우스 오버 기능 추가
@@ -352,6 +372,8 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
plot_lines.append(scatter_dev40_buy_points)
if scatter_dev240_buy_points is not None:
plot_lines.append(scatter_dev240_buy_points)
if scatter_dev1440_buy_points is not None:
plot_lines.append(scatter_dev1440_buy_points)
# zip 길이가 짧은 쪽에 맞춰 매핑
for leg_handle, orig in zip(legend_handles, plot_lines):
@@ -481,8 +503,8 @@ def run_simulation(symbol: str, interval_minutes: int, days: int = 30):
if __name__ == "__main__":
interval = 60
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']
#target_coins = ['ADA','APE','ARB','BONK','HBAR','LINK','ONDO','PEPE','SEI','SHIB','STORJ','SUI','TON','TRX','WLD','XLM','XRP']
target_coins = ['POL']
# 그래프 표시 여부 설정 (성능 향상을 위해 기본값은 False)
show_graphs = True # True로 설정하면 각 코인마다 그래프 표시
@@ -501,18 +523,20 @@ if __name__ == "__main__":
# 그래프 없이 빠른 분석만 수행
data = fetch_price_history(symbol, interval, days)
data = calculate_technical_indicators(data)
data = check_buy_point(data, simulation=True)
data = check_buy_point(symbol, data, simulation=True)
# 매수 신호 통계만 출력
total_buy_signals = len(data[data['buy_point'] == 1])
ma_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'movingaverage')])
dev40_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation40')])
dev240_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation240')])
dev1440_signals = len(data[(data['buy_point'] == 1) & (data['buy_signal'] == 'deviation1440')])
print(f"총 매수 신호: {total_buy_signals}")
print(f" - MA 신호: {ma_signals}")
print(f" - Dev40 신호: {dev40_signals}")
print(f" - Dev240 신호: {dev240_signals}")
print(f" - Dev1440 신호: {dev1440_signals}")
except Exception as e:
print(f"Error analyzing {symbol}: {str(e)}")