Files
auto-stock/backend/app/services/range_trading_strategy.py
shuim 9fab4018d4 fix: 프로그램 검증 루프 7회차 — 리스크·주문·청산·인프라 결함 일괄 수정 + 거래 확대 결정 반영
하네스 규칙 검사 전역 위반 117→0, 회차별 검증(code-verifier opus 4도메인)으로 Critical/High 해소.

주문·리스크
- MDD 강제축소 반복 매도(종목별 완료 집합·재시도 상한·날짜 롤오버·잔고 오류 보류·epoch·장외 미주문)
- KIS 거부 FAILED 기록, 결과 불명(전송 후) 재전송 금지·PENDING, 미전송 실패만 재시도(msg_cd 분류)
- 국내 전량·손실 매도 시장가, 부분 익절 직전 현재가, 지정가만 미체결 취소(cancel_order_v2: TR 0013U·조직번호) + 0체결 롤백
- 누적 4%·현금·레버리지(신규분 포함)·사이클 내 매수 누적, 수동매수 공용 가드(buy_guards), 수동매도 DT 차감·락
- 가드 우회 엔드포인트 차단(/recommendations/auto-trade 위임, US /execute 매수 410·매도 가드)
- 매수 슬롯=성공 매수, 평가 상한·회전, 업무거부/계좌 쿨다운, 프로세스 내 in-flight 주문 중복 차단
데이트레이딩
- 진입·청산 shield, 청산 진행 중 표식(소유 토큰), 2단계 청산 체크, 결과 불명 → 보유 감소 판정·비DT 차감 상한
- 전일 포지션 재시도(일시 실패 미산정), 분할매도 영속, ORB 구간 손절, /cycle 락, 사이클 내 현금 차감, 체결 수량 보정
스윙
- 손절 공백(판정 예외 시 하드 바닥), 단일 손절 소스·범위 검증, 매도 직전 KIS 잔고 상한(min(주문가능,보유)−DT)
- 결과 불명 체결 대조·장부 반영, 대체가(나이·진입 이후·ATR 캐시), 분할 단계 기록, 국면 실패 fail-closed
- 듀얼 플레이북·서킷브레이커 게이트, 평균회귀 우선·듀얼 독립, 시간손절 거래일
- 거래 확대(사용자 결정, 백테스트 미검증): RSI 46/35 정본 경로, 모멘텀 v2 ON, SIDEWAYS ADX AND, 트레일링 1차 익절 후, 20일 시간손절 ON
인프라·보안
- 계좌·손익 GET 인증, 세션 is_active·관리자 경로, 서명키 파생, realip, WS 토큰 마스킹
- KIS 토큰 즉시 실패(sleep 제거)·무효 시 캐시 삭제, 주문 5xx+JSON 본문 반환
- db_worker 종료 drain(멱등 재실행·커밋 불명 격리·예산·밴 커밋), stop_grace 40s, Redis 타임아웃·volatile-lru, /health DLQ·캐시
- KRX 휴장일, py3.11 비호환 f-string 수정 + Jenkins 이미지 내 compileall·헬스 실패 시 빌드 실패
하네스: rules_check 에 PEP 701 f-string 검사, 마커(allow-*), deploy 스모크 401 기준
CLAUDE.md: 실제 동작·플래그·백로그(§8.5-27~38) 갱신

backend 1451 passed / hooks 112 passed / tsc ok / py3.9 compile 0 errors

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 12:01:31 +09:00

508 lines
25 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""박스권(SIDEWAYS) 레인지 매매 전략 — CLAUDE.md §8.5 항목 9 전반부
⚠️ **백테스트 미검증 — `RANGE_TRADE_ENABLED=True` 전환 금지 (CLAUDE.md §8.1-4)**
현재 상태: 코드만 존재. 실계좌·모의계좌 어느 쪽에서도 검증되지 않았다.
`settings.RANGE_TRADE_ENABLED` 는 기본 False 이며, False 인 동안
`swing/entry_service.scan_range_trading_signals()` 는 함수 진입 즉시 `[]` 를 반환한다
(기존 동작 완전 불변).
**켜기 전에 반드시 통과해야 하는 검증 항목**
1. SIDEWAYS 구간 라벨링 정확도 — `detect_market_regime()` 이 SIDEWAYS 로 판정한
구간(2023-03~2026-03)을 추출하고, 그 구간에서만 이 전략을 돌린 결과를 측정한다.
SIDEWAYS 표본이 60거래일 미만이면 통계적 의미 없음 → 판정 불가로 간주.
2. In-sample 백테스트(2023-03~2025-03): 승률 ≥ 50%, Profit Factor ≥ 1.3,
평균 보유 5~15거래일, MDD ≥ -8%(더 얕아야 함), 거래비용 0.5%(왕복) 반영 후 기준.
3. Walk-forward (6개월 학습 / 3개월 검증, 최소 4구간): 모든 검증 구간에서
PF ≥ 1.0, 어느 구간도 MDD -10% 초과 금지. 한 구간이라도 깨지면 불합격.
4. 박스 이탈 손절의 실효성: 전체 거래 중 `box_break` 손절 비중과 그때의 평균 손실률을
따로 집계. 평균 손실이 -RANGE_TRADE_STOP_LOSS_PERCENT 보다 크면(더 많이 잃으면)
박스 이탈 판정 임계값(RANGE_TRADE_BOX_BREAK_PCT)을 재조정해야 한다.
5. 파라미터 민감도: RANGE_TRADE_LOOKBACK_DAYS(15/20/25),
RANGE_TRADE_MIN_WIDTH_PCT(5/6/8), RANGE_TRADE_RSI_ENTRY(35/40/45) 그리드에서
PF 가 1.0 아래로 떨어지는 조합이 절반을 넘으면 과최적화로 판단 → 폐기.
6. 기존 전략과의 중복/상관: 같은 날 평균회귀·눌림목과 동일 종목을 잡는 비율이 30%를
넘으면 신규 신호원으로서의 가치가 없음 → 배선 재검토.
7. 미검증 상태로 스케줄러에 배선하지 말 것(swing/scheduler.py 는 본 작업 범위 밖이며
현재 호출부가 없다 — 보고서의 후속 작업 항목 참조).
─────────────────────────────────────────────────────────────────────────────
전략 개요 (순수 로직 — DB·네트워크 접근 없음)
진입: 성립된 박스권의 **하단 지지 반등**
- 박스 성립 검증 (detect_range_box): 최근 N일 고/저 레인지 폭이 일정 범위,
상단·하단을 각각 min_touches 회 이상 접촉, 접촉 순서가 교대(왕복 이력)해야 한다.
- ADX 상한으로 추세장 배제 (박스권 전제 = 추세 없음).
- 가격이 박스 하단 zone 안 + BB 하단 부근(%B) + RSI 과매도.
- **반등 확인 필수**: RSI 상승 전환(rsi > rsi_prev) 또는 양봉. 둘 다 불가하면 진입 거부.
- 거래량 ≥ 20일 평균 × 배수 (지지 반등의 수급 확인).
청산: 박스 상단 접근 시 익절 / 박스 이탈 시 즉시 손절
- 1) 박스 하단 하향 이탈 → 즉시 전량 (레인지 전략의 최대 리스크)
- 2) 일반 손절 (기본%·ATR 중 타이트한 쪽) → 전량
- 3) 박스 상단 zone 도달 / BB 상단(99%) / RSI 전량임계 / 트레일링 → 전량
- 4) 박스 중단(mid) 통과 또는 RSI 부분임계 → 50% 부분 익절 (수수료 가드 후)
손절 (CLAUDE.md §5.2):
stop_price = max(atr_stop, pct_stop) ← 가격 기준에서 **높은 값이 타이트한 쪽**
atr_stop = entry_price − RANGE_TRADE_ATR_STOP_MULTIPLIER × ATR
pct_stop = entry_price × (1 + stop_loss_pct/100), stop_loss_pct = -abs(...) 보장
박스 하단 이탈 손절은 위 stop_price 와 **별개의 OR 트리거**로 동작한다
(박스 하단이 stop_price 보다 낮을 수도, 높을 수도 있어 max() 에 섞지 않는다).
모든 파라미터는 `app.core.config.settings` 의 `RANGE_TRADE_*` 설정에서 읽는다
(하드코딩 금지 — 함수 인자로 오버라이드 가능하게 두어 백테스트 그리드 탐색에 사용).
"""
import logging
from dataclasses import dataclass
from typing import Dict, List, Optional
from app.core.log_throttle import warn_throttled # noqa: E402
logger = logging.getLogger(__name__)
# ActiveTrade.strategy_name 에 기록되는 값.
# ⚠️ mdd_manager.SWING_STRATEGY_NAMES 에 이 값이 추가되어야 MDD 축소가 반영된다
# (본 작업 범위 밖 — 보고서 후속 작업 항목 참조).
RANGE_TRADE_STRATEGY_NAME = "RANGE_TRADE"
@dataclass
class RangeBox:
"""성립이 검증된 박스권."""
high: float # 박스 상단 (lookback 구간 최고가)
low: float # 박스 하단 (lookback 구간 최저가)
mid: float # 박스 중앙
width_pct: float # 박스 폭 / 중앙 × 100
top_touches: int # 상단 접촉 봉 수
bottom_touches: int # 하단 접촉 봉 수
alternations: int # 상단↔하단 접촉 교대 횟수 (왕복 이력)
lookback: int # 판정에 사용한 봉 수
position_pct: float # 현재가의 박스 내 위치 (0=하단, 100=상단)
@dataclass
class RangeExitSignal:
"""청산 신호. mean_reversion_strategy.ExitSignal 과 동일한 필드 구성
(action/reason/sell_pct) — swing/exit_service 의 소비 코드와 duck-typing 호환."""
action: str # "SELL_PARTIAL" | "SELL_ALL"
reason: str
sell_pct: float # 0.5 = 50%, 1.0 = 전량
def _cfg(name: str, override: Optional[float], default: float) -> float:
"""설정값 조회: 명시 오버라이드 > settings.RANGE_TRADE_* > 하드 폴백.
하드 폴백은 config.py 가 아직 해당 키를 갖지 않는 경우(구버전 이미지)에만 쓰인다.
"""
if override is not None:
return float(override)
try:
from app.core.config import settings
val = getattr(settings, name, None)
if val is not None:
return float(val)
except Exception as _swallowed:
warn_throttled(logger, "range_trading_strategy:103", "[RangeTrade] settings.%s 조회 실패(기본값 사용): %s", name, _swallowed)
return float(default)
# ═══════════════════════════════════════════════════════════════════════════
# 1. 박스권 성립 검증
# ═══════════════════════════════════════════════════════════════════════════
def detect_range_box(
highs: List[float],
lows: List[float],
closes: List[float],
lookback: Optional[int] = None,
min_width_pct: Optional[float] = None,
max_width_pct: Optional[float] = None,
touch_tolerance_pct: Optional[float] = None,
min_touches: Optional[int] = None,
min_alternations: Optional[int] = None,
box_break_pct: Optional[float] = None,
) -> Optional[RangeBox]:
"""최근 lookback 봉이 '박스권'인지 검증하고 박스를 반환. 실패 시 None.
검증 조건 (모두 AND):
① 데이터 충분: len ≥ lookback
② 박스 폭: min_width_pct ≤ (high-low)/mid×100 ≤ max_width_pct
- 너무 좁으면 왕복 이익이 거래비용(왕복 ~0.5%)을 못 넘고,
너무 넓으면 '박스'가 아니라 추세/고변동 구간이다.
③ 상단·하단 접촉 각각 min_touches 회 이상 (tolerance 이내 접근을 접촉으로 간주)
④ 접촉 순서 교대 min_alternations 회 이상 → 레인지 안에서 실제로 왕복한 이력
(모든 하단 접촉이 앞쪽에만 몰려 있으면 하락 후 횡보일 뿐 박스가 아니다)
⑤ 현재 종가가 박스를 이탈하지 않았음 (상·하단 ±box_break_pct 이내)
"""
_lookback = int(_cfg("RANGE_TRADE_LOOKBACK_DAYS", lookback, 20))
_min_w = _cfg("RANGE_TRADE_MIN_WIDTH_PCT", min_width_pct, 6.0)
_max_w = _cfg("RANGE_TRADE_MAX_WIDTH_PCT", max_width_pct, 20.0)
_tol = _cfg("RANGE_TRADE_TOUCH_TOLERANCE_PCT", touch_tolerance_pct, 2.0)
_min_touch = int(_cfg("RANGE_TRADE_MIN_TOUCHES", min_touches, 2))
_min_alt = int(_cfg("RANGE_TRADE_MIN_ALTERNATIONS", min_alternations, 2))
_break = _cfg("RANGE_TRADE_BOX_BREAK_PCT", box_break_pct, 1.0)
if _lookback <= 0:
return None
if len(highs) < _lookback or len(lows) < _lookback or len(closes) < 1:
return None
win_h = highs[-_lookback:]
win_l = lows[-_lookback:]
box_high = max(win_h)
box_low = min(win_l)
if box_high <= 0 or box_low <= 0 or box_high <= box_low:
return None
mid = (box_high + box_low) / 2.0
if mid <= 0:
return None
width_pct = (box_high - box_low) / mid * 100.0
if width_pct < _min_w or width_pct > _max_w:
logger.debug(
"[RangeBox] 폭 조건 미충족: width=%.2f%% (허용 %.1f~%.1f%%)",
width_pct, _min_w, _max_w,
)
return None
# ③④ 접촉 및 교대 판정 (시간 순서 유지)
top_line = box_high * (1.0 - _tol / 100.0)
bottom_line = box_low * (1.0 + _tol / 100.0)
events: List[str] = []
top_touches = 0
bottom_touches = 0
for h, l in zip(win_h, win_l):
is_top = h >= top_line
is_bottom = l <= bottom_line
if is_top and not is_bottom:
top_touches += 1
events.append("T")
elif is_bottom and not is_top:
bottom_touches += 1
events.append("B")
elif is_top and is_bottom:
# 한 봉이 박스 전체를 커버 = 변동성 급증, 박스 판정 무효
logger.debug("[RangeBox] 단일 봉이 박스 전폭 커버 — 박스 무효")
return None
if top_touches < _min_touch or bottom_touches < _min_touch:
logger.debug(
"[RangeBox] 접촉 부족: top=%d, bottom=%d (각 %d회 이상 필요)",
top_touches, bottom_touches, _min_touch,
)
return None
alternations = 0
prev = None
for e in events:
if prev is not None and e != prev:
alternations += 1
prev = e
if alternations < _min_alt:
logger.debug(
"[RangeBox] 왕복 이력 부족: alternations=%d (%d회 이상 필요) events=%s",
alternations, _min_alt, "".join(events),
)
return None
# ⑤ 현재 종가의 박스 이탈 여부
close = closes[-1]
if close < box_low * (1.0 - _break / 100.0):
logger.debug("[RangeBox] 하단 이탈 상태 (close=%.1f < low=%.1f) — 박스 무효", close, box_low)
return None
if close > box_high * (1.0 + _break / 100.0):
logger.debug("[RangeBox] 상단 이탈 상태 (close=%.1f > high=%.1f) — 박스 무효", close, box_high)
return None
position_pct = (close - box_low) / (box_high - box_low) * 100.0
return RangeBox(
high=box_high,
low=box_low,
mid=mid,
width_pct=width_pct,
top_touches=top_touches,
bottom_touches=bottom_touches,
alternations=alternations,
lookback=_lookback,
position_pct=position_pct,
)
# ═══════════════════════════════════════════════════════════════════════════
# 2. 진입 판정
# ═══════════════════════════════════════════════════════════════════════════
def check_range_buy_signal(
highs: List[float],
lows: List[float],
closes: List[float],
volumes: List[int],
current_close: float,
current_volume: int,
rsi: float,
bb_upper: float,
bb_middle: float,
bb_lower: float,
adx: Optional[float],
rsi_prev: Optional[float] = None,
opens: Optional[List[float]] = None,
box: Optional[RangeBox] = None,
rsi_entry: Optional[float] = None,
bb_pct_b_max: Optional[float] = None,
entry_zone_pct: Optional[float] = None,
max_adx: Optional[float] = None,
volume_ratio: Optional[float] = None,
) -> Optional[Dict]:
"""박스권 하단 지지 반등 매수 신호. 조건 미충족 시 None.
반환 dict 키는 mean_reversion_strategy.check_buy_signal() 과 동일한 관례를 따른다
(signal/strategy/description) + 레인지 전용 필드(box_high/box_low/target_price).
Args:
adx: 일봉 ADX(14). **None 이면 진입 거부**(추세장 배제 판정 불가 = fail-closed).
rsi_prev: N봉 전 RSI. RSI 상승 전환 확인용.
opens: 시가 리스트. 양봉 확인용. rsi_prev·opens 둘 다 없으면 반등 확인 불가 → 거부.
"""
_rsi_entry = _cfg("RANGE_TRADE_RSI_ENTRY", rsi_entry, 40.0)
_pct_b_max = _cfg("RANGE_TRADE_BB_PCT_B_MAX", bb_pct_b_max, 0.25)
_zone = _cfg("RANGE_TRADE_ENTRY_ZONE_PCT", entry_zone_pct, 25.0)
_max_adx = _cfg("RANGE_TRADE_MAX_ADX", max_adx, 20.0)
_vol_ratio = _cfg("RANGE_TRADE_VOLUME_RATIO", volume_ratio, 1.0)
if current_close <= 0 or rsi is None:
return None
# ── ① 추세장 배제 (ADX 상한) ──────────────────────────────────────────
if adx is None:
logger.debug("[RangeBuy] ADX 없음 — 추세 판정 불가로 진입 거부(fail-closed)")
return None
if adx >= _max_adx:
logger.debug("[RangeBuy] ADX(%.1f) >= %.1f — 추세장, 레인지 진입 거부", adx, _max_adx)
return None
# ── ② 박스 성립 검증 ────────────────────────────────────────────────
_box = box if box is not None else detect_range_box(highs, lows, closes)
if _box is None:
return None
# ── ③ 박스 하단 zone 안인지 ──────────────────────────────────────────
if _box.position_pct > _zone:
logger.debug(
"[RangeBuy] 박스 내 위치 %.1f%% > %.1f%% — 하단 지지대 아님",
_box.position_pct, _zone,
)
return None
# ── ④ BB 하단 부근 (%B) ─────────────────────────────────────────────
bb_width = bb_upper - bb_lower
pct_b = (current_close - bb_lower) / bb_width if bb_width > 0 else 0.5
if pct_b > _pct_b_max:
logger.debug("[RangeBuy] %%B(%.2f) > %.2f — BB 하단 부근 아님", pct_b, _pct_b_max)
return None
# ── ⑤ RSI 과매도 ────────────────────────────────────────────────────
if rsi >= _rsi_entry:
logger.debug("[RangeBuy] RSI(%.1f) >= %.1f — 과매도 아님", rsi, _rsi_entry)
return None
# ── ⑥ 반등 확인 필수 (RSI 상승 전환 OR 양봉) ─────────────────────────
rsi_turning_up = rsi_prev is not None and rsi > rsi_prev
bullish_candle = False
if opens and len(opens) >= 1 and len(closes) >= 1:
bullish_candle = closes[-1] > opens[-1]
if not (rsi_turning_up or bullish_candle):
logger.debug(
"[RangeBuy] 반등 미확인 — 진입 거부 (rsi=%.1f, rsi_prev=%s, 양봉=%s)",
rsi, f"{rsi_prev:.1f}" if rsi_prev is not None else "N/A", bullish_candle,
)
return None
# ── ⑦ 거래량 확인 ───────────────────────────────────────────────────
if volumes and len(volumes) >= 20:
avg_vol_20 = sum(volumes[-20:]) / 20.0
if avg_vol_20 > 0 and current_volume < avg_vol_20 * _vol_ratio:
logger.debug(
"[RangeBuy] 거래량 부족: %d < %.0f×%.2f",
current_volume, avg_vol_20, _vol_ratio,
)
return None
elif current_volume <= 0:
return None
# 목표가: 박스 상단과 BB 상단 중 보수적(낮은) 쪽
target_price = min(_box.high, bb_upper) if bb_upper > 0 else _box.high
reversal_desc = []
if rsi_turning_up:
reversal_desc.append(f"RSI반등({rsi_prev:.1f}→{rsi:.1f})")
if bullish_candle:
reversal_desc.append("양봉")
return {
"signal": "BUY",
"strategy": RANGE_TRADE_STRATEGY_NAME,
"confidence": 0.55, # 미검증 전략 — 보수적 고정값 (신뢰도 상향은 백테스트 후)
"box_high": _box.high,
"box_low": _box.low,
"box_mid": _box.mid,
"box_width_pct": _box.width_pct,
"position_pct": _box.position_pct,
"target_price": target_price,
"description": (
f"박스권 하단 지지 반등: 박스[{_box.low:.0f}~{_box.high:.0f}] "
f"폭{_box.width_pct:.1f}% 접촉(T{_box.top_touches}/B{_box.bottom_touches}) "
f"왕복{_box.alternations} 위치{_box.position_pct:.0f}% "
f"RSI({rsi:.1f})<{_rsi_entry:.0f} %B({pct_b:.2f}) ADX({adx:.1f})<{_max_adx:.0f} "
f"{'+'.join(reversal_desc)}"
),
}
# ═══════════════════════════════════════════════════════════════════════════
# 3. 손절가 계산 (CLAUDE.md §5.2)
# ═══════════════════════════════════════════════════════════════════════════
def calculate_range_stop_loss(
entry_price: float,
atr: float,
box_low: float,
atr_multiplier: Optional[float] = None,
stop_loss_percent: Optional[float] = None,
box_break_pct: Optional[float] = None,
) -> Dict[str, float]:
"""레인지 전략 손절가.
CLAUDE.md §5.2: 기본 SL 과 ATR SL 중 **타이트한 쪽**.
가격 기준 비교에서는 더 **높은** 손절가가 타이트하므로 `max(atr_stop, pct_stop)`.
(수익률 기준으로 비교할 때의 `min()` 과 방향이 반대인 점 주의)
박스 하단 이탈 손절(box_break_stop)은 별개의 OR 트리거로 반환한다 —
max() 에 섞으면 박스가 넓을 때 손절이 과도하게 느슨해지거나(하단이 낮음),
박스가 좁을 때 정상 노이즈에서 조기 손절(하단이 높음)될 수 있다.
"""
_atr_mult = _cfg("RANGE_TRADE_ATR_STOP_MULTIPLIER", atr_multiplier, 2.0)
_sl_pct_raw = _cfg("RANGE_TRADE_STOP_LOSS_PERCENT", stop_loss_percent, 5.0)
_break = _cfg("RANGE_TRADE_BOX_BREAK_PCT", box_break_pct, 1.0)
stop_loss_pct = -abs(_sl_pct_raw) # 부호 가드 — 양수 입력 시 전종목 매도 방지
atr_stop = entry_price - _atr_mult * max(0.0, float(atr or 0.0))
pct_stop = entry_price * (1.0 + stop_loss_pct / 100.0)
stop_price = max(atr_stop, pct_stop) # 높은 값 = 타이트한 손절
box_break_stop = box_low * (1.0 - _break / 100.0) if box_low > 0 else 0.0
return {
"stop_price": stop_price,
"atr_stop": atr_stop,
"pct_stop": pct_stop,
"stop_loss_pct": stop_loss_pct,
"box_break_stop": box_break_stop,
}
# ═══════════════════════════════════════════════════════════════════════════
# 4. 청산 판정
# ═══════════════════════════════════════════════════════════════════════════
def check_range_exit_signal(
rsi: float,
close: float,
bb_upper: float,
bb_middle: float,
bb_lower: float,
entry_price: float,
atr: float,
box_high: float,
box_low: float,
remaining_pct: float = 1.0,
trailing_high: Optional[float] = None,
rsi_partial_exit: Optional[float] = None,
rsi_full_exit: Optional[float] = None,
exit_zone_pct: Optional[float] = None,
trailing_atr_mult: Optional[float] = None,
min_profit_pct: Optional[float] = None,
) -> Optional[RangeExitSignal]:
"""레인지 청산 판정. 우선순위대로 평가한다.
1. 박스 하단 하향 이탈 → 즉시 전량 (레인지 전략의 핵심 리스크)
2. 일반 손절 max(atr_stop, pct_stop) → 전량
3. 박스 상단 zone / BB 상단(99%) / RSI 전량임계 / 트레일링 → 전량
4. 박스 중앙 상회 또는 RSI 부분임계 → 50% 부분 익절 (수수료 가드 통과 시)
"""
_rsi_partial = _cfg("RANGE_TRADE_RSI_PARTIAL_EXIT", rsi_partial_exit, 55.0)
_rsi_full = _cfg("RANGE_TRADE_RSI_FULL_EXIT", rsi_full_exit, 65.0)
_exit_zone = _cfg("RANGE_TRADE_EXIT_ZONE_PCT", exit_zone_pct, 80.0)
_trail_mult = _cfg("RANGE_TRADE_TRAILING_ATR_MULT", trailing_atr_mult, 1.5)
_min_profit = _cfg("RANGE_TRADE_MIN_PROFIT_PCT", min_profit_pct, 0.5)
if close <= 0 or entry_price <= 0:
return None
stops = calculate_range_stop_loss(entry_price, atr, box_low)
# ── 1. 박스 하단 이탈 → 즉시 전량 손절 ──────────────────────────────
if stops["box_break_stop"] > 0 and close <= stops["box_break_stop"]:
return RangeExitSignal(
action="SELL_ALL",
reason=(
f"박스 이탈 손절 (close={close:.0f} <= 박스하단이탈선"
f"={stops['box_break_stop']:.0f}, box_low={box_low:.0f})"
),
sell_pct=1.0,
)
# ── 2. 일반 손절 (기본%·ATR 중 타이트한 쪽) ─────────────────────────
if close <= stops["stop_price"]:
return RangeExitSignal(
action="SELL_ALL",
reason=(
f"손절 (close={close:.0f} <= stop={stops['stop_price']:.0f}, "
f"ATR기준={stops['atr_stop']:.0f}, {stops['stop_loss_pct']:.1f}%기준"
f"={stops['pct_stop']:.0f})"
),
sell_pct=1.0,
)
profit_pct = (close - entry_price) / entry_price * 100.0
# ── 3. 전량 익절 조건 ──────────────────────────────────────────────
full_reasons: List[str] = []
if box_high > box_low:
pos_pct = (close - box_low) / (box_high - box_low) * 100.0
if pos_pct >= _exit_zone:
full_reasons.append(f"박스상단zone(위치{pos_pct:.0f}%>={_exit_zone:.0f}%)")
if bb_upper > 0 and close >= bb_upper * 0.99:
full_reasons.append(f"BB상단({bb_upper:.0f})도달")
if rsi is not None and rsi >= _rsi_full:
full_reasons.append(f"RSI({rsi:.1f})>={_rsi_full:.0f}")
if trailing_high is not None and trailing_high > entry_price and atr > 0:
trail_stop = trailing_high - _trail_mult * atr
if close < trail_stop:
full_reasons.append(f"트레일링(close={close:.0f}<{trail_stop:.0f})")
# 익절은 수수료(왕복 ~0.5%)를 커버할 때만. 트레일링은 손실 보호이므로 예외.
_has_trailing = any(r.startswith("트레일링") for r in full_reasons)
if full_reasons and (profit_pct >= _min_profit or _has_trailing):
return RangeExitSignal(
action="SELL_ALL",
reason=f"전량청산: {' OR '.join(full_reasons)}",
sell_pct=1.0,
)
# ── 4. 부분 익절 (50%) ─────────────────────────────────────────────
if remaining_pct > 0.50 and profit_pct >= _min_profit:
partial_reasons: List[str] = []
if rsi is not None and rsi >= _rsi_partial:
partial_reasons.append(f"RSI({rsi:.1f})>={_rsi_partial:.0f}")
if bb_middle > 0 and close > bb_middle:
partial_reasons.append(f"BB중앙({bb_middle:.0f})상회")
if box_high > box_low and close > (box_high + box_low) / 2.0:
partial_reasons.append("박스중앙상회")
if partial_reasons:
return RangeExitSignal(
action="SELL_PARTIAL",
reason=f"1차 익절(50%): {' OR '.join(partial_reasons)}",
sell_pct=0.50,
)
return None