"""트레일링 스탑 + 타임 스탑 감시 - 백그라운드 스레드에서 실행.""" import logging import os import time from datetime import datetime import pyupbit from .market import get_current_price from . import trader logger = logging.getLogger(__name__) CHECK_INTERVAL = 10 # 10초마다 체크 # 타임 스탑: N시간 경과 후 수익률이 M% 미만이면 청산 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 = 7 # 최근 N개 40분봉으로 자연 진폭 계산 (≈5h, int(5*60/40)=7) ATR_MULT = 1.5 # 평균 진폭 × 배수 = 스탑 임계값 ATR_MIN_STOP = 0.010 # 최소 스탑 1.0% (너무 좁아지는 거 방지) ATR_MAX_STOP = 0.020 # 최대 스탑 2.0% (너무 넓어지는 거 방지) # ATR 캐시: 종목별 (스탑비율, 계산시각) — 40분마다 갱신 _atr_cache: dict[str, tuple[float, float]] = {} _ATR_CACHE_TTL = 2400 # 40분 def _resample_40m(df): """minute10 DataFrame → 40분봉으로 리샘플링.""" return ( df.resample("40min") .agg({"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) .dropna(subset=["close"]) ) def _get_adaptive_stop(ticker: str) -> float: """최근 ATR_CANDLES개 40분봉 평균 진폭 × ATR_MULT 로 적응형 스탑 비율 반환. 캐시(40분)를 활용해 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: fetch_n = (ATR_CANDLES + 2) * 4 # 40분봉 N개 = 10분봉 N*4개 df10 = pyupbit.get_ohlcv(ticker, interval="minute10", count=fetch_n) if df10 is None or df10.empty: return ATR_MIN_STOP df = _resample_40m(df10) if 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: """적응형 트레일링 스탑(최고가 기준) + 고정 스탑(매수가 기준) 체크.""" trader.update_peak(ticker, current) pos = trader.get_positions().get(ticker) if pos is None: return False peak = pos["peak_price"] buy_price = pos["buy_price"] stop_pct = _get_adaptive_stop(ticker) drop_from_peak = (peak - current) / peak drop_from_buy = (buy_price - current) / buy_price if drop_from_peak >= stop_pct: reason = ( f"트레일링스탑 | 최고가={peak:,.2f}원 → " f"현재={current:,.2f}원 ({drop_from_peak:.2%} 하락 | 스탑={stop_pct:.2%})" ) return trader.sell(ticker, reason=reason) if drop_from_buy >= stop_pct: reason = ( f"스탑로스 | 매수가={buy_price:,.2f}원 → " f"현재={current:,.2f}원 ({drop_from_buy:.2%} 하락 | 스탑={stop_pct:.2%})" ) return trader.sell(ticker, reason=reason) return False def _check_time_stop(ticker: str, pos: dict, current: float) -> bool: """타임 스탑 체크. 매도 시 True 반환. 조건: 보유 후 TIME_STOP_HOURS 경과 AND 수익률 < TIME_STOP_MIN_GAIN_PCT% """ entry_time = pos.get("entry_time") if entry_time is None: return False elapsed_hours = (datetime.now() - entry_time).total_seconds() / 3600 if elapsed_hours < TIME_STOP_HOURS: return False pnl_pct = (current - pos["buy_price"]) / pos["buy_price"] * 100 if pnl_pct >= TIME_STOP_MIN_GAIN_PCT: return False reason = ( f"타임스탑 | {elapsed_hours:.2f}시간 경과 후 " f"수익률={pnl_pct:+.2f}% (기준={TIME_STOP_MIN_GAIN_PCT:+.2f}% 미달)" ) trader.sell(ticker, reason=reason) return True def _check_shadow_position(ticker: str, spos: dict) -> None: """Shadow 포지션 청산 조건 체크 (트레일링 + 타임 스탑). 실제 포지션과 동일한 로직을 적용하되 주문 없이 결과만 기록. """ current = get_current_price(ticker) if current is None: return trader.update_shadow_peak(ticker, current) # 갱신 후 최신 값 재조회 spos = trader.get_shadow_positions().get(ticker) if spos is None: return buy_price = spos["buy_price"] peak = spos["peak_price"] entry_time = spos["entry_time"] stop_pct = _get_adaptive_stop(ticker) drop_from_peak = (peak - current) / peak elapsed_hours = (datetime.now() - entry_time).total_seconds() / 3600 pnl_pct = (current - buy_price) / buy_price * 100 reason = None if drop_from_peak >= stop_pct: reason = ( f"트레일링스탑 | 최고={peak:,.2f}→현재={current:,.2f}" f" ({drop_from_peak:.2%} | 스탑={stop_pct:.2%})" ) elif elapsed_hours >= TIME_STOP_HOURS and pnl_pct < TIME_STOP_MIN_GAIN_PCT: reason = ( f"타임스탑 | {elapsed_hours:.1f}h 경과 " f"수익률={pnl_pct:+.2f}% (기준={TIME_STOP_MIN_GAIN_PCT:+.2f}%)" ) if reason: trader.close_shadow(ticker, current, pnl_pct, reason) def _check_position(ticker: str, pos: dict) -> None: """단일 포지션 전체 체크 (트레일링 스탑 → 타임 스탑 순서).""" current = get_current_price(ticker) if current is None: return buy_price = pos["buy_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 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:,.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순위: 적응형 트레일링 스탑 if _check_trailing_stop(ticker, pos, current): return # 2순위: 타임 스탑 _check_time_stop(ticker, pos, current) def run_monitor(interval: int = CHECK_INTERVAL) -> None: """전체 포지션 감시 루프.""" logger.info( 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()) for ticker, pos in positions_snapshot.items(): try: _check_position(ticker, pos) except Exception as e: logger.error(f"모니터 오류 {ticker}: {e}") # Shadow 포지션 감시 (WF차단 종목 재활 추적) shadow_snapshot = trader.get_shadow_positions() for ticker, spos in shadow_snapshot.items(): try: _check_shadow_position(ticker, spos) except Exception as e: logger.error(f"Shadow 모니터 오류 {ticker}: {e}") time.sleep(interval)