From b0f0b3e82a0e0b2bda52d8c92f5eff1956fc765c Mon Sep 17 00:00:00 2001 From: joungmin Date: Sun, 1 Mar 2026 10:32:24 +0900 Subject: [PATCH] feat: ATR adaptive trailing stop and 2-decimal formatting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - monitor.py: replace fixed 1.5% stop with ATR-based adaptive stop recent 5x 1h candles avg range × 1.5 mult, clamped 1.0%~4.0% 10min cache per ticker to minimize API calls all log numbers formatted to 2 decimal places - notify.py: apply 2 decimal places to price, P&L, fee, cum_profit Co-Authored-By: Claude Sonnet 4.6 --- core/monitor.py | 90 +++++++++++++++++++++++++++++++++++-------------- core/notify.py | 18 +++++----- 2 files changed, 74 insertions(+), 34 deletions(-) diff --git a/core/monitor.py b/core/monitor.py index d7bb22d..6e636cd 100644 --- a/core/monitor.py +++ b/core/monitor.py @@ -5,43 +5,82 @@ import os import time from datetime import datetime +import pyupbit + from .market import get_current_price from . import trader logger = logging.getLogger(__name__) -STOP_LOSS_PCT = float(os.getenv("STOP_LOSS_PCT", "5")) / 100 # 최고가 대비 -N% → 트레일링 스탑 -CHECK_INTERVAL = 10 # 10초마다 체크 +CHECK_INTERVAL = 10 # 10초마다 체크 # 타임 스탑: N시간 경과 후 수익률이 M% 미만이면 청산 -TIME_STOP_HOURS = float(os.getenv("TIME_STOP_HOURS", "24")) +TIME_STOP_HOURS = float(os.getenv("TIME_STOP_HOURS", "24")) TIME_STOP_MIN_GAIN_PCT = float(os.getenv("TIME_STOP_MIN_GAIN_PCT", "3")) +# ATR 기반 적응형 트레일링 스탑 파라미터 +ATR_CANDLES = 5 # 최근 N개 1h봉으로 자연 진폭 계산 +ATR_MULT = 1.5 # 평균 진폭 × 배수 = 스탑 임계값 +ATR_MIN_STOP = 0.010 # 최소 스탑 1.0% (너무 좁아지는 거 방지) +ATR_MAX_STOP = 0.040 # 최대 스탑 4.0% (너무 넓어지는 거 방지) + +# ATR 캐시: 종목별 (스탑비율, 계산시각) — 10분마다 갱신 +_atr_cache: dict[str, tuple[float, float]] = {} +_ATR_CACHE_TTL = 600 # 10분 + + +def _get_adaptive_stop(ticker: str) -> float: + """최근 ATR_CANDLES개 1h봉 평균 진폭 × ATR_MULT 로 적응형 스탑 비율 반환. + + 캐시(10분)를 활용해 API 호출 최소화. + 계산 실패 시 ATR_MIN_STOP 반환. + """ + now = time.time() + cached = _atr_cache.get(ticker) + if cached and (now - cached[1]) < _ATR_CACHE_TTL: + return cached[0] + + try: + df = pyupbit.get_ohlcv(ticker, interval="minute60", count=ATR_CANDLES + 2) + if df is None or len(df) < ATR_CANDLES: + return ATR_MIN_STOP + ranges = (df["high"] - df["low"]) / df["low"] + avg_range = ranges.iloc[-ATR_CANDLES:].mean() + stop = float(max(ATR_MIN_STOP, min(ATR_MAX_STOP, avg_range * ATR_MULT))) + except Exception as e: + logger.debug(f"[ATR] {ticker} 계산 실패: {e}") + stop = ATR_MIN_STOP + + _atr_cache[ticker] = (stop, now) + return stop + def _check_trailing_stop(ticker: str, pos: dict, current: float) -> bool: - """트레일링 스탑(최고가 기준) + 고정 스탑(매수가 기준) 체크. 매도 시 True 반환.""" + """적응형 트레일링 스탑(최고가 기준) + 고정 스탑(매수가 기준) 체크.""" trader.update_peak(ticker, current) pos = trader.get_positions().get(ticker) if pos is None: return False - peak = pos["peak_price"] + peak = pos["peak_price"] buy_price = pos["buy_price"] - drop_from_peak = (peak - current) / peak - drop_from_buy = (buy_price - current) / buy_price # 구매가 대비 하락률 + stop_pct = _get_adaptive_stop(ticker) - if drop_from_peak >= STOP_LOSS_PCT: + drop_from_peak = (peak - current) / peak + drop_from_buy = (buy_price - current) / buy_price + + if drop_from_peak >= stop_pct: reason = ( - f"트레일링스탑 | 최고가={peak:,.0f}원 → " - f"현재={current:,.0f}원 ({drop_from_peak:.1%} 하락)" + f"트레일링스탑 | 최고가={peak:,.2f}원 → " + f"현재={current:,.2f}원 ({drop_from_peak:.2%} 하락 | 스탑={stop_pct:.2%})" ) return trader.sell(ticker, reason=reason) - if drop_from_buy >= STOP_LOSS_PCT: + if drop_from_buy >= stop_pct: reason = ( - f"스탑로스 | 매수가={buy_price:,.0f}원 → " - f"현재={current:,.0f}원 ({drop_from_buy:.1%} 하락)" + f"스탑로스 | 매수가={buy_price:,.2f}원 → " + f"현재={current:,.2f}원 ({drop_from_buy:.2%} 하락 | 스탑={stop_pct:.2%})" ) return trader.sell(ticker, reason=reason) @@ -66,8 +105,8 @@ def _check_time_stop(ticker: str, pos: dict, current: float) -> bool: return False reason = ( - f"타임스탑 | {elapsed_hours:.1f}시간 경과 후 " - f"수익률={pnl_pct:+.1f}% (기준={TIME_STOP_MIN_GAIN_PCT:+.0f}% 미달)" + f"타임스탑 | {elapsed_hours:.2f}시간 경과 후 " + f"수익률={pnl_pct:+.2f}% (기준={TIME_STOP_MIN_GAIN_PCT:+.2f}% 미달)" ) trader.sell(ticker, reason=reason) return True @@ -80,20 +119,21 @@ def _check_position(ticker: str, pos: dict) -> None: return buy_price = pos["buy_price"] - pnl = (current - buy_price) / buy_price * 100 - peak = pos["peak_price"] + pnl = (current - buy_price) / buy_price * 100 + peak = pos["peak_price"] drop_from_peak = (peak - current) / peak - drop_from_buy = (buy_price - current) / buy_price + drop_from_buy = (buy_price - current) / buy_price + stop_pct = _get_adaptive_stop(ticker) entry_time = pos.get("entry_time", datetime.now()) elapsed_hours = (datetime.now() - entry_time).total_seconds() / 3600 logger.info( - f"[감시] {ticker} 현재={current:,.0f} | 매수가={buy_price:,.0f} | 최고={peak:,.0f} | " - f"수익률={pnl:+.1f}% | peak하락={drop_from_peak:.1%} | buy하락={drop_from_buy:.1%} | " - f"보유={elapsed_hours:.1f}h" + f"[감시] {ticker} 현재={current:,.2f} | 매수가={buy_price:,.2f} | 최고={peak:,.2f} | " + f"수익률={pnl:+.2f}% | peak하락={drop_from_peak:.2%} | buy하락={drop_from_buy:.2%} | " + f"스탑={stop_pct:.2%} | 보유={elapsed_hours:.2f}h" ) - # 1순위: 트레일링 스탑 + # 1순위: 적응형 트레일링 스탑 if _check_trailing_stop(ticker, pos, current): return @@ -104,9 +144,9 @@ def _check_position(ticker: str, pos: dict) -> None: def run_monitor(interval: int = CHECK_INTERVAL) -> None: """전체 포지션 감시 루프.""" logger.info( - f"모니터 시작 | 체크={interval}초 | " - f"트레일링스탑={STOP_LOSS_PCT:.1%} | " - f"타임스탑={TIME_STOP_HOURS:.0f}h/{TIME_STOP_MIN_GAIN_PCT:+.0f}%" + f"모니터 시작 | 체크={interval}초 | ATR×{ATR_MULT} " + f"(최소={ATR_MIN_STOP:.2%} / 최대={ATR_MAX_STOP:.2%}) | " + f"타임스탑={TIME_STOP_HOURS:.0f}h/{TIME_STOP_MIN_GAIN_PCT:+.2f}%" ) while True: positions_snapshot = dict(trader.get_positions()) diff --git a/core/notify.py b/core/notify.py index 8f2a9dc..bef2068 100644 --- a/core/notify.py +++ b/core/notify.py @@ -38,9 +38,9 @@ def notify_buy( ) _send( f"📈 [매수] {ticker}\n" - f"가격: {price:,.0f}원\n" - f"수량: {amount}\n" - f"투자금: {invested_krw:,}원\n" + f"가격: {price:,.2f}원\n" + f"수량: {amount:.8f}\n" + f"투자금: {invested_krw:,.2f}원\n" f"{budget_line}" ) @@ -54,10 +54,10 @@ def notify_sell( cum_emoji = "💚" if cum_profit >= 0 else "🔴" _send( f"{trade_emoji} [매도] {ticker}\n" - f"가격: {price:,.0f}원\n" + f"가격: {price:,.2f}원\n" f"수익률: {pnl_pct:+.2f}%\n" - f"실손익: {krw_profit:+,.0f}원 (수수료 {fee_krw:,.0f}원)\n" - f"{cum_emoji} 누적손익: {cum_profit:+,.0f}원\n" + f"실손익: {krw_profit:+,.2f}원 (수수료 {fee_krw:,.2f}원)\n" + f"{cum_emoji} 누적손익: {cum_profit:+,.2f}원\n" f"사유: {reason}" ) @@ -120,9 +120,9 @@ def notify_status( emoji = "📈" if pnl >= 0 else "📉" lines.append( f"{emoji} {ticker}\n" - f" 현재가: {current:,.0f}원\n" + f" 현재가: {current:,.2f}원\n" f" 수익률: {pnl:+.2f}%\n" - f" 최고가 대비: -{drop:.1f}%\n" - f" 보유: {elapsed:.1f}h" + f" 최고가 대비: -{drop:.2f}%\n" + f" 보유: {elapsed:.2f}h" ) _send("\n".join(lines))