shuim
a57c720559
- portfolio_cache_bridge 신설: 매매 엔진 체결 시 Portfolio(db_worker 큐) + 인메모리 캐시 즉시 갱신. DT 매수는 strategy_type='daytrading'으로 태깅되어 보유 종목 패널의 '데이트레이딩' 그룹에 표시되고 스윙 청산 로직(strategy_type='swing' 대상)에서 제외됨. (기존엔 KIS 동기화로 'swing'으로 유입돼 DT 포지션이 안 보이고 스윙 엔진이 같이 관리하는 문제) - DT 매도·스윙 자동 매도도 캐시에서 즉시 제거/차감 + 최근매도 기록(동기화 되살림 방지). - 동기화 시 DT 엔진 보유 종목은 'daytrading'으로 태깅. - 프론트: 수동 매도 성공 시 낙관적 갱신으로 서버 재조회 전 즉시 목록에서 제거. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
400 lines
19 KiB
Python
400 lines
19 KiB
Python
"""스윙 트레이딩 청산 서비스
|
|
|
|
기존 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
|
|
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__)
|
|
|
|
|
|
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' 포지션만 조회
|
|
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)
|
|
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,
|
|
"highest_since_entry": trailing_high,
|
|
})
|
|
|
|
# ── 전략별 청산 분기 ────────────────────────────────────
|
|
exit_signal = None
|
|
_strategy = active_trade.strategy_name if active_trade else "swing"
|
|
|
|
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
|
|
|
|
# ── 매도 수량 결정 (소량 분할 방지) ─────────────────────
|
|
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",
|
|
})
|
|
|
|
# 포트폴리오 화면 즉시 반영 (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:
|
|
await enqueue_close_active_trade(stock_code, strategy_name=None)
|
|
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,
|
|
"remaining_pct": new_remaining,
|
|
"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
|