"""스윙 트레이딩 청산 서비스 기존 portfolio.py의 check_and_execute_exit_trades() 로직을 기반으로 strategy_type='swing' 포지션만 처리하도록 분리한 서비스. 청산 기준 (CLAUDE.md 섹션 5.2, 6.3): - ATR 손절: 진입가 - 2.5*ATR (하드스탑과 비교하여 타이트한 쪽) - 고정비율 폴백 손절: -7% (ATR 계산 불가 시) - 1차 익절: RSI >= 55 OR 가격 > BB 중앙선 OR 수익 > 2*ATR -> 50% 매도 - 2차 익절: RSI >= 70 OR 가격 > BB 상단(99%) -> 전량 매도 - 트레일링스탑: 고점 - 1.5*ATR (잔여 50% 물량) 전략별 차별화 (CLAUDE.md 8.4): - 평균회귀/눌림목: RSI 기반 단계별 익절 - 모멘텀 돌파: +10%/+20% 단계별 30% 매도 + 넓은 트레일링 아키텍처 규칙: - DB 기록: db_worker 큐 경유 (enqueue_trade_history, enqueue_close_active_trade) - loss_tracker: 전량 매도 시 record_trade_result_safe() + save_to_db() - 매도 주문: safe_place_order() 사용 (세마포어+분산락 보호) """ import logging import time from datetime import datetime from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.core.config import settings from app.models.models import Portfolio, ActiveTrade from app.services.korea_investment_api import korea_investment_api from app.services.mean_reversion_strategy import ( calculate_atr, calculate_rsi, calculate_bollinger, check_exit_signal, MarketRegime, ) from app.services.market_analyzer import check_market_regime from app.services.mdd_manager import mdd_manager from app.services.loss_tracker import loss_tracker from app.services.db_worker import enqueue_trade_history, enqueue_close_active_trade from app.services.swing.position_manager import get_swing_holdings, get_swing_active_trade from app.services.utils import _is_domestic from app.core.json_helpers import _dumps logger = logging.getLogger(__name__) # 종목별 마지막 청산 시각(epoch) — 큐 지연으로 같은 분할매도가 반복되는 것을 막는 인메모리 가드 _recent_exits: dict[str, float] = {} EXIT_COOLDOWN_SEC = 180 def _check_momentum_exit( current_price: float, entry_price: float, atr: float, remaining_pct: float, trailing_high: float | None, ) -> "ExitSignal | None": """모멘텀 돌파 전용 청산 로직 (CLAUDE.md 8.4). 일반 평균회귀와 다른 익절 전략을 사용: - 이익을 최대한 키우기 위해 RSI 기반 익절 대신 수익률 기반 단계별 매도 - 손절은 평균회귀와 동일 (ATR 2.5배 vs 하드스탑 중 타이트한 쪽) 익절 단계: 1. +10% 이상: 30% 매도 (remaining_pct > 0.75일 때만) 2. +20% 이상: 추가 30% 매도 (remaining_pct > 0.45일 때만) 3. 잔량: 트레일링스탑 (고점 - 1.5*ATR) Args: current_price: 현재가 entry_price: 진입가 atr: ATR(14) 값 remaining_pct: 잔여 비율 (1.0=전량 보유) trailing_high: 진입 이후 최고가 (트레일링스탑 기준) Returns: ExitSignal | None: 청산 신호, 없으면 None Note: [CRITICAL] 손절 계산에서 stop_price = max(atr_stop, pct_stop) 사용. 이것은 가격 비교이므로 max()가 올바름 (높은 가격 = 타이트한 손절). CLAUDE.md에서 min()을 언급하는 것은 백분율 비교 기준. """ from app.services.mean_reversion_strategy import ExitSignal if entry_price <= 0: return None profit_pct = (current_price - entry_price) / entry_price * 100 # 손절: ATR 2.5배 vs 하드스탑 (CLAUDE.md 5.2) # stop_loss_pct는 음수 (-7.0 등), pct_stop은 진입가 대비 하드스탑 가격 # max()를 사용하는 이유: 가격 비교에서 높은 값 = 진입가에 가까운 = 타이트한 손절 stop_loss_pct = -abs(settings.STOP_LOSS_PERCENT) atr_stop = entry_price - 2.5 * atr pct_stop = entry_price * (1 + stop_loss_pct / 100) stop_price = max(atr_stop, pct_stop) if current_price <= stop_price: return ExitSignal( action="SELL_ALL", reason=f"모멘텀 손절 ({profit_pct:.1f}%, stop={stop_price:.0f})", sell_pct=1.0, ) # 트레일링스탑 (잔량에 대해) if trailing_high and trailing_high > entry_price: trail_stop = trailing_high - 1.5 * atr if current_price <= trail_stop and profit_pct > 0: return ExitSignal( action="SELL_ALL", reason=f"모멘텀 트레일링 (고점={trailing_high:.0f}, 트레일={trail_stop:.0f})", sell_pct=1.0, ) # 1차 익절: +10% → 30% 매도 (remaining > 0.65 = 아직 1차 매도 안 함) # 1차 매도 후 remaining = 1.0 - 0.30 = 0.70 → 다음 사이클에서 0.70 > 0.65 방지 # 따라서 0.75 기준 유지: 1.0 > 0.75 트리거, 매도 후 0.70 <= 0.75 → 재트리거 방지 if profit_pct >= 10.0 and remaining_pct > 0.75: return ExitSignal( action="SELL_PARTIAL", reason=f"모멘텀 1차 익절 +{profit_pct:.1f}% → 30% 매도", sell_pct=0.30, ) # 2차 익절: +20% → 30% 매도 # 1차 매도 후 remaining = 0.70, 2차 매도 후 remaining = 0.40 # 조건: remaining > 0.45 AND remaining <= 0.75 (1차와 겹치지 않게 상한 설정) if profit_pct >= 20.0 and 0.45 < remaining_pct <= 0.75: return ExitSignal( action="SELL_PARTIAL", reason=f"모멘텀 2차 익절 +{profit_pct:.1f}% → 30% 매도", sell_pct=0.30, ) return None async def check_and_execute_exit_v2(db: AsyncSession) -> list[dict]: """스윙 포지션 손절/익절 체크 및 자동 매도 실행. portfolio.py의 check_and_execute_exit_trades()와 동일한 로직이나 strategy_type='swing' 필터를 적용해 데이트레이딩 포지션을 제외함. 스케줄러에서 60초 주기로 호출. 처리 흐름 (종목별): 1. 스윙 보유 종목 조회 (get_swing_holdings) 2. 시장 국면 조회 (BULL/BEAR) 3. 실시간 현재가 조회 (KIS API) 4. ActiveTrade 조회 (분할 청산 상태) 5. 전략별 청산 분기: - MOMENTUM_BREAKOUT: _check_momentum_exit() - 평균회귀/눌림목/기본: check_exit_signal() 6. 고정비율 폴백 손절 (-7%) 7. 매도 수량 결정 (소량 분할 방지) 8. KIS API 매도 주문 실행 9. DB 기록 + loss_tracker 업데이트 Args: db: AsyncSession 인스턴스 Returns: list[dict]: 청산 실행 결과 목록 """ executed_exits = [] try: # strategy_type='swing' 포지션만 조회 _now_ts = time.time() holdings = await get_swing_holdings(db) if not holdings: logger.debug("[SwingExit] 스윙 보유 종목 없음") return executed_exits # 시장 국면 (동적 계산 — 기존 코드의 하드코딩 BULL 개선) try: regime_dict = await check_market_regime() regime_str = regime_dict.get("regime_200d", "bull") if isinstance(regime_dict, dict) else "bull" regime = MarketRegime.BULL if regime_str == "bull" else MarketRegime.BEAR except Exception: regime = MarketRegime.BULL # 폴백 for holding in holdings: try: stock_code = holding.stock_code stock_name = holding.stock_name or stock_code quantity = holding.quantity avg_price = float(holding.average_price or 0) market = getattr(holding, "market", None) or ( "KRX" if _is_domestic(stock_code) else "NASD" ) domestic = _is_domestic(stock_code, market) # ── 실시간 현재가 조회 ──────────────────────────────────── latest_price = float(holding.current_price or 0) if domestic: price_resp = await korea_investment_api.get_current_price(stock_code) if price_resp and price_resp.get("output"): latest_price = int(price_resp["output"].get("stck_prpr", latest_price)) else: price_resp = await korea_investment_api.get_overseas_price(stock_code, market) if price_resp and price_resp.get("output"): latest_price = float(price_resp["output"].get("last", latest_price)) if latest_price <= 0 or avg_price <= 0: continue # ── ActiveTrade 조회 (분할 청산 상태) ──────────────────── active_trade = await get_swing_active_trade(db, stock_code) # ActiveTrade upsert/close 시 항상 전략명을 명시해야 같은 종목의 DT 행을 # 잘못 건드리지 않는다(db_worker 는 전략명이 없으면 최신 행을 매칭). _at_strategy = (active_trade.strategy_name if active_trade else None) or "swing" remaining_pct = float(active_trade.remaining_pct) if active_trade and active_trade.remaining_pct is not None else 1.0 trailing_high = float(active_trade.highest_since_entry) if active_trade and active_trade.highest_since_entry else None # [S6] 트레일링스탑 고점(highest_since_entry) 매 사이클 갱신 if active_trade and latest_price > (trailing_high or 0): trailing_high = latest_price from app.services.db_worker import enqueue_active_trade_upsert await enqueue_active_trade_upsert({ "stock_code": stock_code, "strategy_name": _at_strategy, # DT 행 오염 방지 "highest_since_entry": trailing_high, }) # ── 전략별 청산 분기 ──────────────────────────────────── exit_signal = None _strategy = _at_strategy if domestic: try: chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=30) chart_data = chart_resp.get("output2", []) if len(chart_data) >= 20: closes = [float(d.get("stck_clpr", 0)) for d in reversed(chart_data)] highs = [float(d.get("stck_hgpr", 0)) for d in reversed(chart_data)] lows = [float(d.get("stck_lwpr", 0)) for d in reversed(chart_data)] atr = calculate_atr(highs, lows, closes) rsi = calculate_rsi(closes) bb = calculate_bollinger(closes) if atr and rsi is not None and bb: bb_upper, bb_middle, bb_lower = bb lowest_since = min(lows[-5:]) if lows else avg_price # 모멘텀 돌파 전용 청산 (CLAUDE.md 8.4) if _strategy == "MOMENTUM_BREAKOUT": exit_signal = _check_momentum_exit( latest_price, avg_price, atr, remaining_pct, trailing_high, ) else: # 평균회귀 / 눌림목 / 기본 스윙 exit_signal = check_exit_signal( regime=regime, rsi=rsi, close=latest_price, bb_upper=bb_upper, bb_middle=bb_middle, bb_lower=bb_lower, entry_price=avg_price, atr=atr, lowest_since_entry=lowest_since, remaining_pct=remaining_pct, trailing_high=trailing_high, ) except Exception as e: logger.debug(f"[SwingExit] ATR 계산 실패 ({stock_code}): {e}") # ── 고정 비율 폴백 손절 ─────────────────────────────────── if exit_signal is None: stop_loss_pct = abs(float(getattr(settings, "STOP_LOSS_PERCENT", 6.0))) # 적응형 모드(Crisis 등)의 손절값과 비교하여 타이트한 쪽 적용 try: from app.services.adaptive_strategy import get_current_adaptive_params _adaptive = get_current_adaptive_params() if not _adaptive.fallback_used: _adaptive_sl = abs(_adaptive.stop_loss_pct) stop_loss_pct = min(stop_loss_pct, _adaptive_sl) except Exception: pass # 적응형 전략 로드 실패 시 기본값 유지 profit_rate = ((latest_price - avg_price) / avg_price) * 100 if profit_rate <= -stop_loss_pct: from app.services.mean_reversion_strategy import ExitSignal exit_signal = ExitSignal( action="SELL_ALL", reason=f"고정비율 손절 ({profit_rate:.1f}%)", sell_pct=1.0, ) if not exit_signal: continue # [2026-09-23] 상태 저장(remaining_pct·Portfolio)이 db_worker 큐 경유라 # 큐 지연·DLQ 시 다음 사이클이 옛 값을 읽어 같은 분할매도를 반복한다. # 인메모리 쿨다운으로 직전 청산 직후 재매도를 차단한다. _last_exit = _recent_exits.get(stock_code) if _last_exit is not None and (_now_ts - _last_exit) < EXIT_COOLDOWN_SEC: logger.info( f"[SwingExit] {stock_code}: 직전 청산 {int(_now_ts - _last_exit)}초 전 " f"— 쿨다운({EXIT_COOLDOWN_SEC}초) 중 스킵" ) continue # ── 매도 수량 결정 (소량 분할 방지) ───────────────────── sell_pct = getattr(exit_signal, "sell_pct", 1.0) if sell_pct < 1.0 and quantity <= 3: # 3주 이하: 분할 매도 비효율 → 전량 매도 sell_qty = quantity logger.info(f"[SwingExit] {stock_code}: {quantity}주 소량 → 전량매도 (분할 비효율)") else: sell_qty = min(max(int(quantity * sell_pct), 1), quantity) if sell_qty <= 0: continue # ── KIS API 매도 주문 ───────────────────────────────────── logger.info( f"[SwingExit] 청산 신호 → {stock_code} {stock_name} " f"수량={sell_qty} 이유={exit_signal.reason}" ) # [C1] 분산락+세마포어 보호 주문 from app.services.trade_executor import safe_place_order order_resp = await safe_place_order( stock_code=stock_code, order_type="SELL", quantity=sell_qty, price=None if domestic else latest_price, is_domestic=domestic, exchange_code=market or "NASD", ) if not order_resp or order_resp.get("rt_cd") != "0": err = order_resp.get("msg1", "unknown") if order_resp else "no response" logger.error(f"[SwingExit] 매도 주문 실패 ({stock_code}): {err}") continue order_id = order_resp.get("output", {}).get("ODNO", "") profit_loss = (latest_price - avg_price) * sell_qty profit_loss_rate = ((latest_price - avg_price) / avg_price) * 100 # ── DB 기록 (db_task_queue 경유) ────────────────────────── await enqueue_trade_history({ "stock_code": stock_code, "stock_name": stock_name, "order_type": "SELL", "order_price": latest_price, "order_quantity": sell_qty, "executed_price": latest_price, "executed_quantity": sell_qty, "total_amount": latest_price * sell_qty, "status": "COMPLETED", "order_id": order_id, "exit_reason": exit_signal.reason, "net_profit_loss": profit_loss, "profit_loss_rate": profit_loss_rate, "strategy_type": "swing", }) _recent_exits[stock_code] = time.time() # 포트폴리오 화면 즉시 반영 (Portfolio 수량 차감 + 캐시) from app.services.portfolio_cache_bridge import reflect_sell await reflect_sell(stock_code, sell_qty, latest_price, remaining_qty=quantity - sell_qty) # 전량 청산이면 ActiveTrade 종료, 분할 매도면 remaining_pct 갱신 if sell_qty >= quantity: # strategy_name 을 비우면 db_worker 가 '가장 최근 active 행'을 닫는다 # → 같은 종목에 DT 행이 있으면 그쪽이 닫히고 스윙 행은 영구 잔존. await enqueue_close_active_trade( stock_code, strategy_name=_strategy or "swing" ) else: # 분할 매도 후 remaining_pct 갱신 (반복 트리거 방지) new_remaining = max(0.0, remaining_pct - sell_pct) from app.services.db_worker import enqueue_active_trade_upsert _at_data = { "stock_code": stock_code, "strategy_name": _strategy or "swing", "remaining_pct": new_remaining, # 잔량도 함께 갱신 — 누락 시 재기동 복원이 원래 수량으로 되살아나 # 보유량보다 많이 매도한다(2026-09-23 DT LG씨엔에스 사례와 동일 패턴) "quantity": quantity - sell_qty, "highest_since_entry": latest_price if latest_price > (trailing_high or 0) else trailing_high, } if active_trade is None: # [FIX] ActiveTrade 행이 없는 포지션(동기화 유입 등)은 upsert가 # NOT NULL(quantity/entry_price) 위반으로 INSERT 실패 → remaining_pct # 미저장 → 다음 사이클에 50% 매도 재실행(2026-09-22 두산에너빌리티). # 신규 생성에 필요한 필수 키를 함께 전달한다. _at_data.update({ "stock_name": stock_name, "quantity": quantity - sell_qty, "entry_price": avg_price, "current_price": latest_price, "strategy_name": "swing", "status": "active", "entry_reason": "동기화 유입 포지션 (부분매도 시 자동 생성)", }) await enqueue_active_trade_upsert(_at_data) # LossTracker 업데이트 — 전량매도 시에만 (분할매도 이중 카운트 방지) if sell_qty >= quantity: is_loss = profit_loss < 0 await loss_tracker.record_trade_result_safe( stock_code=stock_code, is_loss=is_loss, ) await loss_tracker.save_to_db(stock_code=stock_code) executed_exits.append({ "stock_code": stock_code, "stock_name": stock_name, "sell_quantity": sell_qty, "price": latest_price, "profit_loss": profit_loss, "profit_loss_rate": profit_loss_rate, "reason": exit_signal.reason, }) logger.info( f"[SwingExit] ✅ 청산 완료: {stock_code} {stock_name} " f"수익률={profit_loss_rate:.1f}% 이유={exit_signal.reason}" ) except Exception as e: logger.error(f"[SwingExit] 종목 처리 오류 ({holding.stock_code}): {e}", exc_info=True) except Exception as e: logger.error(f"[SwingExit] 전체 오류: {e}", exc_info=True) return executed_exits