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-23 18:20:49 +09:00
parent 865a5c8680
commit 6cb939d690
31 changed files with 235 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
monitor.py
View File

@@ -18,13 +18,13 @@ import os
class Monitor:
"""자산(코인/주식/ETF) 모니터링 및 매수 실행 클래스"""
last_buy_signal = None
last_signal = None
cooldown_file = None
def __init__(self, cooldown_file='coins_buy_time.json') -> None:
self.hts = HTS()
# 최근 매수 신호 저장용(파일은 [신규] 포맷으로 저장)
self.last_buy_signal: dict[str, str] = {}
self.last_signal: dict[str, str] = {}
if cooldown_file is not None:
self.cooldown_file = cooldown_file
self.buy_cooldown = self._load_buy_cooldown()
@@ -45,16 +45,16 @@ class Monitor:
except Exception:
continue
elif isinstance(value, dict):
# [신규] 포맷: "SYMBOL": {"datetime": "...", "buy_signal": "..."}
# [신규] 포맷: "SYMBOL": {"datetime": "...", "signal": "..."}
dt_str = value.get('datetime')
if isinstance(dt_str, str):
try:
cooldown[symbol] = datetime.fromisoformat(dt_str)
except Exception:
pass
buy_signal = value.get('buy_signal', '')
if isinstance(buy_signal, str):
self.last_buy_signal[symbol] = buy_signal
signal = value.get('signal', '')
if isinstance(signal, str):
self.last_signal[symbol] = signal
return cooldown
except Exception as e:
print(f"Error loading cooldown data: {e}")
@@ -68,7 +68,7 @@ class Monitor:
for symbol, dt in self.buy_cooldown.items():
data[symbol] = {
'datetime': dt.isoformat(),
'buy_signal': self.last_buy_signal.get(symbol, '')
'signal': self.last_signal.get(symbol, '')
}
with open(self.cooldown_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
@@ -226,14 +226,14 @@ class Monitor:
pass
current_time = datetime.now()
if data['buy_signal'].iloc[-1] == 'fall_6p':
if data['signal'].iloc[-1] == 'fall_6p':
if data['Close'].iloc[-1] > 100:
buy_amount = 500000
else:
buy_amount = 300000
if symbol in self.buy_cooldown and symbol in self.last_buy_signal:
if self.last_buy_signal[symbol] == 'fall_6p':
if symbol in self.buy_cooldown and symbol in self.last_signal:
if self.last_signal[symbol] == 'fall_6p':
time_diff = current_time - self.buy_cooldown[symbol]
if time_diff.total_seconds() < 4000:
print(f"{symbol}: 매수 금지 중 (남은 시간: {600 - time_diff.total_seconds():.0f}초)")
@@ -246,20 +246,20 @@ class Monitor:
return False
buy_amount = 5100
if data['buy_signal'].iloc[-1] == 'movingaverage':
if data['signal'].iloc[-1] == 'movingaverage':
buy_amount = 30000
elif data['buy_signal'].iloc[-1] == 'deviation40':
elif data['signal'].iloc[-1] == 'deviation40':
buy_amount = 50000
elif data['buy_signal'].iloc[-1] == 'deviation240':
elif data['signal'].iloc[-1] == 'deviation240':
buy_amount = 6000
elif data['buy_signal'].iloc[-1] == 'deviation1440':
elif data['signal'].iloc[-1] == 'deviation1440':
if symbol in ['BONK', 'PEPE', 'TON']:
buy_amount = 20000
else:
buy_amount = 30000
# heikin_ashi 조건 제거 완료
if data['buy_signal'].iloc[-1] in ['movingaverage', 'deviation40', 'deviation240', 'deviation1440']:
if data['signal'].iloc[-1] in ['movingaverage', 'deviation40', 'deviation240', 'deviation1440']:
if check_5_week_lowest:
buy_amount *= 4
@@ -268,14 +268,14 @@ class Monitor:
if self.cooldown_file is not None:
# 최근 매수 신호를 함께 기록하여 [신규] 포맷으로 저장
try:
self.last_buy_signal[symbol] = str(data['buy_signal'].iloc[-1])
self.last_signal[symbol] = str(data['signal'].iloc[-1])
except Exception:
self.last_buy_signal[symbol] = ''
self.last_signal[symbol] = ''
self.buy_cooldown[symbol] = current_time
self._save_buy_cooldown()
print(f"{KR_COINS[symbol]} ({symbol}) [{data['buy_signal'].iloc[-1]}], 현재가: {data['Close'].iloc[-1]:.4f}, 20분간 매수 금지 시작")
self.sendMsg("[KRW-COIN]" + "\n" + self.format_message('COIN', symbol, KR_COINS[symbol], data['Close'].iloc[-1], data['buy_signal'].iloc[-1]))
print(f"{KR_COINS[symbol]} ({symbol}) [{data['signal'].iloc[-1]}], 현재가: {data['Close'].iloc[-1]:.4f}, 20분간 매수 금지 시작")
self.sendMsg("[KRW-COIN]" + "\n" + self.format_message('COIN', symbol, KR_COINS[symbol], data['Close'].iloc[-1], data['signal'].iloc[-1]))
except Exception as e:
print(f"Error buying {symbol}: {str(e)}")
return False
@@ -366,72 +366,72 @@ class Monitor:
return False
return True
def check_buy_point(self, symbol: str, data: pd.DataFrame, simulation: bool | None = None) -> pd.DataFrame:
def check_point(self, symbol: str, data: pd.DataFrame, simulation: bool | None = None) -> pd.DataFrame:
data = data.copy()
data['buy_signal'] = ''
data['buy_point'] = 0
if data['buy_point'].iloc[-1] != 1:
data['signal'] = ''
data['point'] = 0
if data['point'].iloc[-1] != 1:
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_signal'] = 'movingaverage'
data.at[data.index[i], 'buy_point'] = 1
if not simulation and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'movingaverage'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'movingaverage'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'movingaverage'
data.at[data.index[-1], 'point'] = 1
if data['Deviation40'].iloc[i - 1] < data['Deviation40'].iloc[i] and data['Deviation40'].iloc[i - 1] <= 90:
data.at[data.index[i], 'buy_signal'] = 'deviation40'
data.at[data.index[i], 'buy_point'] = 1
if not simulation and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation40'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'deviation40'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'deviation40'
data.at[data.index[-1], 'point'] = 1
if symbol not in ['BONK']:
if symbol in ['TRX']:
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 and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation240'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'deviation240'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'deviation240'
data.at[data.index[-1], 'point'] = 1
else:
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 and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation240'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'deviation240'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'deviation240'
data.at[data.index[-1], '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 and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'deviation1440'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'deviation1440'
data.at[data.index[-1], '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 and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'deviation1440'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'deviation1440'
data.at[data.index[-1], '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 and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'deviation1440'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'deviation1440'
data.at[data.index[-1], '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 and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'deviation1440'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'deviation1440'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'deviation1440'
data.at[data.index[-1], 'point'] = 1
try:
prev_low = data['Low'].iloc[i - 1]
@@ -440,18 +440,18 @@ class Monitor:
cond_close_drop = curr_close <= prev_low * 0.94
cond_low_drop = curr_low <= prev_low * 0.94
if cond_close_drop or cond_low_drop:
data.at[data.index[i], 'buy_signal'] = 'fall_6p'
data.at[data.index[i], 'buy_point'] = 1
if not simulation and data['buy_point'][-3:].sum() > 0:
data.at[data.index[-1], 'buy_signal'] = 'fall_6p'
data.at[data.index[-1], 'buy_point'] = 1
data.at[data.index[i], 'signal'] = 'fall_6p'
data.at[data.index[i], 'point'] = 1
if not simulation and data['point'][-3:].sum() > 0:
data.at[data.index[-1], 'signal'] = 'fall_6p'
data.at[data.index[-1], 'point'] = 1
except Exception:
pass
return data
# ------------- Formatting -------------
def format_message(self, market_type: str, symbol: str, symbol_name: str, close: float, buy_signal: str) -> str:
message = f"• 매수 [{market_type}] {symbol_name} ({symbol}): {buy_signal} "
def format_message(self, market_type: str, symbol: str, symbol_name: str, close: float, signal: str) -> str:
message = f"• 매수 [{market_type}] {symbol_name} ({symbol}): {signal} "
message += f"({'$' if market_type == 'US' else '₩'}{close:.4f})"
return message