0713fb1e11
Sell position if held for TIME_STOP_HOURS (default 24h) with less than TIME_STOP_MIN_GAIN_PCT (default +3%) gain. Frees up capital for fresh momentum opportunities. Priority: trailing stop (-10%) checked first, then time stop. Both thresholds configurable via .env. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""트레일링 스탑 + 타임 스탑 감시 - 백그라운드 스레드에서 실행."""
|
|
|
|
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 = 0.10 # 최고가 대비 -10% → 트레일링 스탑
|
|
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"]
|
|
drop_from_peak = (peak - current) / peak
|
|
|
|
if drop_from_peak >= STOP_LOSS_PCT:
|
|
reason = (
|
|
f"트레일링스탑 | 최고가={peak:,.0f}원 → "
|
|
f"현재={current:,.0f}원 ({drop_from_peak:.1%} 하락)"
|
|
)
|
|
trader.sell(ticker, reason=reason)
|
|
return True
|
|
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
|
|
|
|
pnl = (current - pos["buy_price"]) / pos["buy_price"] * 100
|
|
peak = pos["peak_price"]
|
|
drop_from_peak = (peak - current) / peak
|
|
entry_time = pos.get("entry_time", datetime.now())
|
|
elapsed_hours = (datetime.now() - entry_time).total_seconds() / 3600
|
|
|
|
logger.info(
|
|
f"[감시] {ticker} 현재={current:,.0f} | 최고={peak:,.0f} | "
|
|
f"하락={drop_from_peak:.1%} | 수익률={pnl:+.1f}% | "
|
|
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:.0%} | "
|
|
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)
|