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-24 16:32:53 +09:00
parent f519f57c0f
commit 45b563f39c
1 changed files with 93 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
monitor.py
View File

@@ -31,44 +31,72 @@ class Monitor:
# ------------- Persistence -------------
def _load_buy_cooldown(self) -> dict:
if os.path.exists(self.cooldown_file):
"""load trade record file into nested dict {symbol:{'buy':{'datetime':dt,'signal':s},'sell':{...}}}"""
if not os.path.exists(self.cooldown_file):
return {}
try:
with open(self.cooldown_file, 'r', encoding='utf-8') as f:
data = json.load(f)
cooldown: dict[str, datetime] = {}
# [기존] 문자열 값, [신규] 객체 값 모두 지원
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": "...", "signal": "..."}
dt_str = value.get('datetime')
if isinstance(dt_str, str):
try:
cooldown[symbol] = datetime.fromisoformat(dt_str)
except Exception:
pass
signal = value.get('signal', '')
if isinstance(signal, str):
self.last_signal[symbol] = signal
return cooldown
raw = json.load(f)
except Exception as e:
print(f"Error loading cooldown data: {e}")
return {}
return {}
record: dict[str, dict] = {}
for symbol, value in raw.items():
# 신규 포맷: value has 'buy'/'sell'
if isinstance(value, dict) and ('buy' in value or 'sell' in value):
record[symbol] = {}
for side in ['buy', 'sell']:
side_val = value.get(side)
if isinstance(side_val, dict):
dt_iso = side_val.get('datetime')
sig = side_val.get('signal', '')
if dt_iso:
try:
dt_obj = datetime.fromisoformat(dt_iso)
except Exception:
dt_obj = None
else:
dt_obj = None
record[symbol][side] = {'datetime': dt_obj, 'signal': sig}
else:
# 구 포맷 처리 (매수만 기록)
try:
dt_obj = None
sig = ''
if isinstance(value, str):
dt_obj = datetime.fromisoformat(value)
elif isinstance(value, dict):
dt_iso = value.get('datetime')
sig = value.get('signal', '')
if dt_iso:
dt_obj = datetime.fromisoformat(dt_iso)
record.setdefault(symbol, {})['buy'] = {'datetime': dt_obj, 'signal': sig}
except Exception:
continue
# last_signal 채우기 (buy 기준)
for sym, sides in record.items():
if 'buy' in sides and sides['buy'].get('signal'):
self.last_signal[sym] = sides['buy']['signal']
return record
def _save_buy_cooldown(self) -> None:
"""save nested trade record structure"""
try:
# [신규] 포맷으로 저장
data: dict[str, dict] = {}
for symbol, dt in self.buy_cooldown.items():
data[symbol] = {
'datetime': dt.isoformat(),
'signal': self.last_signal.get(symbol, '')
for symbol, sides in self.buy_cooldown.items():
data[symbol] = {}
for side in ['buy', 'sell']:
info = sides.get(side)
if not info:
continue
dt_obj = info.get('datetime')
sig = info.get('signal', '')
data[symbol][side] = {
'datetime': dt_obj.isoformat() if isinstance(dt_obj, datetime) else '',
'signal': sig,
}
with open(self.cooldown_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
@@ -232,15 +260,16 @@ class Monitor:
else:
buy_amount = 300000
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]
last_buy_dt = self.buy_cooldown.get(symbol, {}).get('buy', {}).get('datetime')
if last_buy_dt and self.last_signal.get(symbol) == 'fall_6p':
time_diff = current_time - last_buy_dt
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]
last_buy_dt = self.buy_cooldown.get(symbol, {}).get('buy', {}).get('datetime')
if last_buy_dt:
time_diff = current_time - last_buy_dt
if time_diff.total_seconds() < 1800:
print(f"{symbol}: 매수 금지 중 (남은 시간: {1800 - time_diff.total_seconds():.0f}초)")
return False
@@ -271,7 +300,7 @@ class Monitor:
self.last_signal[symbol] = str(data['signal'].iloc[-1])
except Exception:
self.last_signal[symbol] = ''
self.buy_cooldown[symbol] = current_time
self.buy_cooldown.setdefault(symbol, {})['buy'] = {'datetime': current_time, 'signal': str(data['signal'].iloc[-1])}
# 매수를 저장함
self._save_buy_cooldown()
@@ -283,87 +312,39 @@ class Monitor:
return True
def sell_ticker(self, symbol: str, data: pd.DataFrame) -> bool:
"""Dev40(Deviation40) 매도 조건을 만족할 때만 매도 실행"""
try:
# 기존 로직 ---------------------------------------------------
#print('BUY: {}'.format(symbol))
#self.sendMsg('BUY: {}'.format(symbol))
check_5_week_lowest = False
# 5주봉이 20주봉이나 40주봉보다 아래에 있는지 체크
try:
# Convert hourly data to week-based rolling periods (5, 20, 40 weeks)
hours_in_week = 24 * 7 # 168 hours
period_5w = 5 * hours_in_week # 840 hours
period_20w = 20 * hours_in_week # 3,360 hours
period_40w = 40 * hours_in_week # 6,720 hours
if len(data) >= period_40w:
wma5 = data['Close'].rolling(window=period_5w).mean().iloc[-1]
wma20 = data['Close'].rolling(window=period_20w).mean().iloc[-1]
wma40 = data['Close'].rolling(window=period_40w).mean().iloc[-1]
# 5-week MA is the lowest among 5, 20, 40 week MAs
if (wma5 < wma20) and (wma5 < wma40):
check_5_week_lowest = True
except Exception:
# Ignore errors in MA calculation so as not to block trading logic
pass
# 최신 캔들의 시그널이 Dev40이 아니면 매도하지 않음
if data['signal'].iloc[-1] != 'deviation40':
return False
current_time = datetime.now()
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_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}초)")
return False
else:
if symbol in self.buy_cooldown:
time_diff = current_time - self.buy_cooldown[symbol]
if time_diff.total_seconds() < 1800:
print(f"{symbol}: 매수 금지 중 (남은 시간: {1800 - time_diff.total_seconds():.0f}초)")
# 최근 Dev40 매도로부터 20분(1200초) 쿨다운 적용
last_sell_dt = self.buy_cooldown.get(symbol, {}).get('sell', {}).get('datetime')
if last_sell_dt and self.last_signal.get(symbol) == 'deviation40':
time_diff = current_time - last_sell_dt
if time_diff.total_seconds() < 1200:
remain = 1200 - time_diff.total_seconds()
print(f"{symbol}: 매도 금지 중 (남은 시간: {remain:.0f}초)")
return False
buy_amount = 5100
if data['signal'].iloc[-1] == 'movingaverage':
buy_amount = 30000
elif data['signal'].iloc[-1] == 'deviation40':
buy_amount = 50000
elif data['signal'].iloc[-1] == 'deviation240':
buy_amount = 6000
elif data['signal'].iloc[-1] == 'deviation1440':
if symbol in ['BONK', 'PEPE', 'TON']:
buy_amount = 20000
else:
buy_amount = 30000
# heikin_ashi 조건 제거 완료
# 매도 수량/금액 산정 (예: 50,000 KRW 상당)
sell_amount = 50000 # KRW 기준 매도 총액 혹은 수량 설정
if data['signal'].iloc[-1] in ['movingaverage', 'deviation40', 'deviation240', 'deviation1440']:
if check_5_week_lowest:
buy_amount *= 4
_ = self.hts.buyCoinMarket(symbol, buy_amount)
# 실제 매도 실행 (HTS API)
_ = self.hts.sellCoinMarket(symbol, 0, sell_amount) # market 매도 (price 파라미터 미사용)
# 쿨다운 및 로그 저장
if self.cooldown_file is not None:
# 최근 매수 신호를 함께 기록하여 [신규] 포맷으로 저장
try:
self.last_signal[symbol] = str(data['signal'].iloc[-1])
except Exception:
self.last_signal[symbol] = ''
self.buy_cooldown[symbol] = current_time
self.last_signal[symbol] = 'deviation40'
self.buy_cooldown.setdefault(symbol, {})['sell'] = {'datetime': current_time, 'signal': 'deviation40'}
self._save_buy_cooldown()
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]))
print(f"{KR_COINS[symbol]} ({symbol}) [Dev40 매도], 현재가: {data['Close'].iloc[-1]:.4f}, 20분간 매도 금지 시작")
self.sendMsg("[KRW-COIN]" + "\n" + self.format_message('COIN', symbol, KR_COINS[symbol], data['Close'].iloc[-1], 'Dev40 매도'))
except Exception as e:
print(f"Error buying {symbol}: {str(e)}")
print(f"Error selling {symbol}: {str(e)}")
return False
return True