upbit-trader/core/monitor.py

"""트레일링 스탑 + 타임 스탑 감시 - 백그라운드 스레드에서 실행."""

import logging
import os
import time
from datetime import datetime

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초마다 체크

# 타임 스탑: 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"))


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"]
    buy_price = pos["buy_price"]
    drop_from_peak = (peak - current) / peak
    drop_from_buy = (buy_price - current) / buy_price  # 구매가 대비 하락률

    if drop_from_peak >= STOP_LOSS_PCT:
        reason = (
            f"트레일링스탑 | 최고가={peak:,.0f}원 → "
            f"현재={current:,.0f}원 ({drop_from_peak:.1%} 하락)"
        )
        return trader.sell(ticker, reason=reason)

    if drop_from_buy >= STOP_LOSS_PCT:
        reason = (
            f"스탑로스 | 매수가={buy_price:,.0f}원 → "
            f"현재={current:,.0f}원 ({drop_from_buy:.1%} 하락)"
        )
        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:.1f}시간 경과 후 "
        f"수익률={pnl_pct:+.1f}% (기준={TIME_STOP_MIN_GAIN_PCT:+.0f}% 미달)"
    )
    trader.sell(ticker, reason=reason)
    return True


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
    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"
    )

    # 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}초 | "
        f"트레일링스탑={STOP_LOSS_PCT:.1%} | "
        f"타임스탑={TIME_STOP_HOURS:.0f}h/{TIME_STOP_MIN_GAIN_PCT:+.0f}%"
    )
    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}")
        time.sleep(interval)