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-10 23:13:29 +09:00
parent 5a73d20402
commit 22287c2c2a
2 changed files with 45 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
coins_buy_time.json
View File

@@ -1,5 +1,5 @@
{
"BONK": "2025-08-07T07:44:02.345835",
"APE": "2025-08-09T14:22:02.089619",
"PEPE": "2025-08-09T14:52:08.398420"
"BONK": {"datetime": "2025-08-07T07:44:02.345835", "buy_signal": "movingaverage"},
"APE": {"datetime": "2025-08-09T14:22:02.089619", "buy_signal": "movingaverage"},
"PEPE": {"datetime": "2025-08-09T14:52:08.398420", "buy_signal": "movingaverage"}
}

52
monitor.py
View File

@@ -19,10 +19,13 @@ import os
class Monitor:
"""자산(코인/주식/ETF) 모니터링 및 매수 실행 클래스"""
last_buy_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] = {}
if cooldown_file is not None:
self.cooldown_file = cooldown_file
self.buy_cooldown = self._load_buy_cooldown()
@@ -34,8 +37,25 @@ class Monitor:
with open(self.cooldown_file, 'r', encoding='utf-8') as f:
data = json.load(f)
cooldown: dict[str, datetime] = {}
for symbol, time_str in data.items():
cooldown[symbol] = datetime.fromisoformat(time_str)
# [기존] 문자열 값, [신규] 객체 값 모두 지원
for symbol, value in data.items():
if isinstance(value, str):
# [기존] 포맷: "SYMBOL": "2025-08-07T07:44:02.345835"
try:
cooldown[symbol] = datetime.fromisoformat(value)
except Exception:
continue
elif isinstance(value, dict):
# [신규] 포맷: "SYMBOL": {"datetime": "...", "buy_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
return cooldown
except Exception as e:
print(f"Error loading cooldown data: {e}")
@@ -44,9 +64,13 @@ class Monitor:
def _save_buy_cooldown(self) -> None:
try:
data: dict[str, str] = {}
# [신규] 포맷으로 저장
data: dict[str, dict] = {}
for symbol, dt in self.buy_cooldown.items():
data[symbol] = dt.isoformat()
data[symbol] = {
'datetime': dt.isoformat(),
'buy_signal': self.last_buy_signal.get(symbol, '')
}
with open(self.cooldown_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
@@ -135,11 +159,12 @@ class Monitor:
else:
buy_amount = 300000
if symbol in self.buy_cooldown:
time_diff = current_time - self.buy_cooldown[symbol]
if time_diff.total_seconds() < 600:
print(f"{symbol}: 매수 금지 중 (남은 시간: {600 - time_diff.total_seconds():.0f}초)")
return False
if symbol in self.buy_cooldown and symbol in self.last_buy_signal:
if self.last_buy_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}초)")
return False
else:
if symbol in self.buy_cooldown:
time_diff = current_time - self.buy_cooldown[symbol]
@@ -163,6 +188,11 @@ class Monitor:
_ = self.hts.buyCoinMarket(symbol, buy_amount)
if self.cooldown_file is not None:
# 최근 매수 신호를 함께 기록하여 [신규] 포맷으로 저장
try:
self.last_buy_signal[symbol] = str(data['buy_signal'].iloc[-1])
except Exception:
self.last_buy_signal[symbol] = ''
self.buy_cooldown[symbol] = current_time
self._save_buy_cooldown()
@@ -249,8 +279,10 @@ class Monitor:
try:
prev_low = data['Low'].iloc[i - 1]
curr_close = data['Close'].iloc[i]
curr_low = data['Low'].iloc[i]
cond_close_drop = curr_close <= prev_low * 0.94
if cond_close_drop:
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: