joungmin/upbit-trader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: ATR adaptive trailing stop and 2-decimal formatting

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
joungmin
2026-03-01 10:32:24 +09:00
parent 83a229dd26
commit b0f0b3e82a
2 changed files with 74 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
core/monitor.py
View File

@@ -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())