""" 포트폴리오 API 엔드포인트 - GET /: 보유 종목 목록 조회 (캐시 활용) - GET /summary: 포트폴리오 요약 (총자산, 평가손익) - GET /sync: 실제 계좌와 동기화 - POST /sell: 수동 매도 (청산) """ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from typing import List, Optional from decimal import Decimal from pydantic import BaseModel import pytz from app.core.database import get_db, AsyncSessionLocal from app.models.models import Portfolio, TradeHistory, ActiveTrade, StockAnalysis, TradeJournal 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, # ① regime 파라미터 전달용 추가 ) from app.services.mdd_manager import mdd_manager from app.services.loss_tracker import loss_tracker from app.services.websocket_manager import websocket_manager import asyncio import logging from datetime import datetime, timedelta logger = logging.getLogger(__name__) # 한국 시간대 KST = pytz.timezone('Asia/Seoul') # Request/Response Models class SellRequest(BaseModel): stock_code: str quantity: int price: Optional[int] = None def _get_kst_now(): """현재 한국 시간 반환 (timezone-naive datetime)""" return datetime.now(KST).replace(tzinfo=None) # 단순 인메모리 캐시 (프로세스 단위). 다중 프로세스 환경이면 Redis 등 필요. class PortfolioCache: def __init__(self): self.holdings = [] # List[Portfolio] self.summary = None # Dict self.last_synced: datetime | None = None self.change_flag = False # 외부 이벤트로 변경 감지 (매수/매도 등) self.sync_lock = asyncio.Lock() self.min_interval = timedelta(seconds=15) # 최소 동기화 간격 self.last_attempt: datetime | None = None self.last_error: str | None = None # 실패 관리 (백오프) self.failure_count: int = 0 self.next_retry_at: datetime | None = None self.last_sync_success: bool = False self.last_trade_id: int | None = None def mark_changed(self): self.change_flag = True def should_sync(self) -> bool: now = datetime.now() # 최초 또는 변경 플래그 또는 최소간격 경과 후 오류 상태 재시도 if self.last_synced is None: return True if self.change_flag: return True # 오류 발생 시 백오프 기간에는 재시도 금지 if self.last_error: if self.next_retry_at and now < self.next_retry_at: return False # 백오프가 지났으면 재시도 허용 (최소 간격과 무관) return True if (now - self.last_synced) > timedelta(seconds=5): # 실시간: 5초마다 갱신 return True return False def _compute_backoff(self) -> timedelta: # 30s, 60s, 120s, max 5min base = 30 delay = base * (2 ** max(self.failure_count - 1, 0)) delay = min(delay, 300) return timedelta(seconds=delay) def record_failure(self, err: Exception | str): self.last_error = str(err) self.failure_count += 1 self.next_retry_at = datetime.now() + self._compute_backoff() self.last_sync_success = False def record_success(self): self.last_error = None self.failure_count = 0 self.next_retry_at = None self.last_sync_success = True def cache_summary(self, balance_data: dict = None): holdings = self.holdings total_stocks = len(holdings) total_value = sum(float(h.current_price or 0) * h.quantity for h in holdings) total_investment = sum(float(h.average_price or 0) * h.quantity for h in holdings) total_profit_loss = sum(float(h.profit_loss or 0) for h in holdings) total_profit_loss_rate = (total_profit_loss / total_investment * 100) if total_investment > 0 else 0 # KIS API output2에서 예수금 및 총자산 정보 추출 available_cash = 0.0 total_assets = 0.0 if balance_data: logger.info(f"🔍 KIS API balance response keys: {balance_data.keys()}") if "output2" in balance_data: output2 = balance_data["output2"][0] if isinstance(balance_data["output2"], list) else balance_data["output2"] logger.info(f"📊 output2 keys: {output2.keys()}") logger.info(f"💰 output2 data: {output2}") else: logger.warning("⚠️ No 'output2' in balance_data!") if balance_data and "output2" in balance_data: output2 = balance_data["output2"][0] if isinstance(balance_data["output2"], list) else balance_data["output2"] # 가급적 증권사 화면과 동일 기준 사용: # - 현금: D+2 정산 기준 예수금(d2_depo_amt) # - 총자산: 총평가금액(tot_evlu_amt) 우선, 없으면 유가평가+현금 계산 # 필드명이 환경에 따라 다를 수 있어 다중 키 체킹 def pick_float(d: dict, keys: list[str]) -> float: for k in keys: if k in d and d.get(k) not in (None, ""): try: return float(d.get(k)) except Exception: continue return 0.0 # 현금: D+2 정산액 우선 -> 예수금총액 -> 순자산금액 순 available_cash = pick_float(output2, [ "d2_depo_amt", # D+2 정산액 "d2_deposit", # 일부 응답 변형 키 "dnca_tot_amt", # 예수금 총액 "nass_amt" # 순자산금액 (fallback) ]) # 총자산: KIS API의 tot_evlu_amt 사용 (가장 정확) # tot_evlu_amt는 D+2 정산액 + 유가평가액을 포함한 실제 총자산 kis_total_assets = pick_float(output2, ["tot_evlu_amt", "tot_evlu_amt_smtl"]) if kis_total_assets > 0: # KIS API 값 우선 사용 total_assets = kis_total_assets calculated_assets = total_value + available_cash # 차이가 크면 로그 (정상 - D+2 정산액 때문) if abs(kis_total_assets - calculated_assets) > 1000: logger.debug( f"💰 총자산 - KIS API: {kis_total_assets:,.0f}원, " f"단순계산(현금+주식): {calculated_assets:,.0f}원, " f"차이: {(kis_total_assets - calculated_assets):,.0f}원 (D+2정산액 포함)" ) else: # KIS API 값이 없으면 계산 (fallback) total_assets = total_value + available_cash # 디버깅 로그(요약): 실제 사용된 값과 원본 일부 키 노출 try: logger.debug( "Portfolio summary calc - cash=%s, total_assets=%s, keys=%s", f"{available_cash:,.0f}", f"{total_assets:,.0f}", {k: output2.get(k) for k in [ "d2_depo_amt", "d2_deposit", "dnca_tot_amt", "nass_amt", "tot_evlu_amt" ] if k in output2} ) except Exception: pass # D+2 정산액 및 기타 상세 정보 추출 d2_settlement = 0.0 stock_value_from_api = 0.0 if balance_data and "output2" in balance_data: output2 = balance_data["output2"][0] if isinstance(balance_data["output2"], list) else balance_data["output2"] d2_settlement = pick_float(output2, ["nxdy_excc_amt", "d2_auto_rdpt_amt"]) stock_value_from_api = pick_float(output2, ["scts_evlu_amt", "evlu_amt_smtl_amt"]) self.summary = { "total_stocks": total_stocks, "total_value": total_value, "total_investment": total_investment, "total_profit_loss": total_profit_loss, "total_profit_loss_rate": total_profit_loss_rate, "available_cash": available_cash, "total_assets": total_assets, "d2_settlement": d2_settlement, # D+2 정산액 "stock_value_from_api": stock_value_from_api # API의 유가평가액 } portfolio_cache = PortfolioCache() from pydantic import BaseModel router = APIRouter() class PortfolioResponse(BaseModel): id: int stock_code: str stock_name: str = None quantity: int average_price: Decimal = None current_price: Decimal = None profit_loss: Decimal = None profit_loss_rate: Decimal = None class Config: from_attributes = True async def _get_latest_executed_trade_id(db: AsyncSession) -> int | None: try: result = await db.execute( select(func.max(TradeHistory.id)).where(TradeHistory.status.in_(['EXECUTED', 'COMPLETED'])) ) return result.scalar_one_or_none() except Exception as e: logger.warning(f"Failed to fetch latest trade id: {e}") return None @router.get("", response_model=List[PortfolioResponse]) async def get_portfolio(db: AsyncSession = Depends(get_db)): """포트폴리오 조회 (캐시 활용)""" # 거래 이력 기준으로 캐시 무효화 latest_trade_id = await _get_latest_executed_trade_id(db) if latest_trade_id and portfolio_cache.last_trade_id != latest_trade_id: portfolio_cache.mark_changed() if portfolio_cache.holdings and not portfolio_cache.should_sync(): logger.debug("Portfolio cache hit (holdings)") # 수량 0인 종목 필터링 (안전장치) return [h for h in portfolio_cache.holdings if h.quantity > 0] # 필요하면 동기화 시도 try: await sync_portfolio(db, force=False) except HTTPException: # 강제 동기화가 아닌 경우에는 오류를 사용자에게 전파하지 않고 캐시/빈 목록 반환 logger.warning("get_portfolio: sync failed, returning cached holdings") # 수량 0인 종목 필터링 (안전장치) return [h for h in portfolio_cache.holdings if h.quantity > 0] @router.get("/sync") async def sync_portfolio(db: AsyncSession = Depends(get_db), force: bool = True): """실제 계좌와 포트폴리오 동기화 (캐시 반영) force=False이면 캐시 정책에 따라 생략될 수 있음. """ if not force and not portfolio_cache.should_sync(): logger.debug("Skip sync: cache fresh & no change flag") return {"message": "Skipped (cache fresh)", "cached": True} # 락 획득에 30초 타임아웃 적용 (이전 sync hang 시 데드락 방지) try: await asyncio.wait_for(portfolio_cache.sync_lock.acquire(), timeout=30) except asyncio.TimeoutError: portfolio_cache.record_failure("Sync lock timeout (30s)") logger.error("Portfolio sync lock timeout — 이전 sync가 30초 이상 hang 중") if force: raise HTTPException(status_code=503, detail="Portfolio sync timeout") return {"message": "Sync timeout (using cache)", "cached": True} try: portfolio_cache.last_attempt = datetime.now() try: balance_data = await korea_investment_api.get_balance() # 디버깅: KIS API 전체 응답 로깅 logger.info(f"🔍 KIS API balance response keys: {balance_data.keys() if balance_data else 'None'}") # Balance API 오류 시 기존 DB 보유종목을 삭제하지 않고 캐시 유지 if not balance_data or balance_data.get('rt_cd') != '0': error_msg = balance_data.get('msg1', 'Unknown error') if balance_data else 'No response' logger.warning(f"⚠️ Balance API 오류 - 기존 포트폴리오 유지: {error_msg}") # 캐시가 비어있으면 DB에서 로드 if not portfolio_cache.holdings: result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) portfolio_cache.holdings = list(result.scalars().all()) logger.info(f"📦 DB에서 포트폴리오 로드: {len(portfolio_cache.holdings)}종목") return {"message": f"Balance API error: {error_msg}", "cached": True, "count": len(portfolio_cache.holdings)} portfolio_cache.record_success() if "output1" in balance_data and balance_data["output1"]: logger.info(f"🔍 output1 count: {len(balance_data['output1'])}") # 1. 오늘 발생한 모든 매매 이력 조회 (Ghost Stock 방지 - 시간 범위 확대) now = _get_kst_now() # 오늘의 시작 시간 (00:00:00) 계산 today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) # 오늘 포함 최근 24시간 (혹시 자정 근처 이슈 방지)이지만, 주로 '최근 행동'이 중요함. # 안전하게 '오늘 0시 이후'의 모든 거래를 가져옴. recent_point = today_start logger.info(f"🔎 Checking recent trades since {recent_point} for Ghost detection") # 복잡한 SQL 대신 단순 조회 후 애플리케이션 레벨에서 필터링 trades_stmt = select(TradeHistory).where( TradeHistory.executed_at >= recent_point, TradeHistory.status.in_(['EXECUTED', 'COMPLETED']) ).order_by(TradeHistory.executed_at.asc()) trade_result = await db.execute(trades_stmt) recent_trades = trade_result.scalars().all() # 종목별 최신 상태 추적 stock_status_map = {} for t in recent_trades: stock_status_map[t.stock_code] = t.order_type # 덮어쓰기 (마지막이 최신) logger.debug(f"Trade trace: {t.stock_name}({t.stock_code}) -> {t.order_type} at {t.executed_at}") # 최신 상태가 'SELL'인 종목 식별 ghost_candidates = {code for code, status in stock_status_map.items() if status == "SELL"} if ghost_candidates: logger.info(f"🚫 Ghost Stock Candidates (Last action today is SELL): {ghost_candidates}") # 기존 레코드 로드 (삭제 대신 업데이트) result = await db.execute(select(Portfolio)) existing_all = {p.stock_code: p for p in result.scalars().all()} processed_codes = set() new_holdings = [] if "output1" in balance_data: for item in balance_data["output1"]: stock_code = item.get("pdno") # 최근 최신 거래가 '매도'라면 KIS가 보내줘도 무시 (Ghost Stock 처리) # 만약 재매수했다면 latest_trades의 type이 'BUY'였을 것이므로 ghost_candidates에 없음. if stock_code in ghost_candidates: logger.info(f"👻 Skipping Ghost Stock: {stock_code} (Latest trade was SELL)") # DB에서도 명시적 삭제 if stock_code in existing_all: await db.delete(existing_all[stock_code]) logger.info(f"🗑️ Force deleted ghost stock from DB: {stock_code}") continue stock_name = item.get("prdt_name") quantity = int(item.get("hldg_qty", 0)) if quantity <= 0: logger.debug(f"Skipping {stock_name}({stock_code}) with 0 quantity") continue # KIS API 필드명 정확히 매핑 (로깅 추가) average_price = Decimal(str(item.get("pchs_avg_pric", 0))) # 🔥 실시간 현재가 조회 (잔고 조회의 prpr은 실시간이 아님) try: price_resp = await korea_investment_api.get_current_price(stock_code) if price_resp and "output" in price_resp: realtime_price = int(price_resp["output"].get("stck_prpr", 0)) if realtime_price > 0: current_price = Decimal(str(realtime_price)) else: current_price = Decimal(str(item.get("prpr", 0))) else: current_price = Decimal(str(item.get("prpr", 0))) except Exception as e: logger.warning(f"⚠️ 실시간 현재가 조회 실패 ({stock_code}): {e}, fallback to prpr") current_price = Decimal(str(item.get("prpr", 0))) # 평가손익 및 수익률 - KIS API에서 제공하는 값 우선 사용 profit_loss = Decimal(str(item.get("evlu_pfls_amt", 0))) profit_loss_rate = Decimal(str(item.get("evlu_pfls_rt", 0))) # 디버깅: 실제 KIS API 응답 로깅 logger.info( f"Portfolio sync - {stock_name}({stock_code}): " f"수량={quantity}, 평균단가={average_price}, 현재가={current_price}, " f"평가손익={profit_loss}, 수익률={profit_loss_rate}% | " f"원본 필드={item}" ) processed_codes.add(stock_code) if stock_code in existing_all: h = existing_all[stock_code] h.stock_name = stock_name h.quantity = quantity h.average_price = average_price h.current_price = current_price h.profit_loss = profit_loss h.profit_loss_rate = profit_loss_rate new_holdings.append(h) else: new_holding = Portfolio( stock_code=stock_code, stock_name=stock_name, quantity=quantity, average_price=average_price, current_price=current_price, profit_loss=profit_loss, profit_loss_rate=profit_loss_rate ) db.add(new_holding) new_holdings.append(new_holding) # 제거된 종목 반영 (DB에서 완전 삭제) for code, h in existing_all.items(): if code not in processed_codes: # 보호 로직: 최근 매수 종목이면 API 미반영 상태로 간주하고 유지 # (KIS API가 체결 직후 잔고 반영이 느릴 수 있음) last_action = stock_status_map.get(code) if last_action == "BUY" and h.quantity > 0: logger.info(f"🛡️ Preserving {h.stock_name}({code}) - API calls it missing, but recent BUY exists") new_holdings.append(h) continue logger.info(f"🗑️ Deleting {h.stock_name}({code}) from portfolio DB (not in KIS API response)") await db.delete(h) await db.commit() portfolio_cache.holdings = [h for h in new_holdings if h.quantity > 0] portfolio_cache.cache_summary(balance_data) # balance_data 전달 portfolio_cache.last_synced = datetime.now() portfolio_cache.change_flag = False portfolio_cache.last_trade_id = await _get_latest_executed_trade_id(db) logger.info(f"Portfolio synced: {len(portfolio_cache.holdings)} holdings") return {"message": "Portfolio synchronized", "count": len(portfolio_cache.holdings), "cached": False} except Exception as e: portfolio_cache.record_failure(e) logger.error(f"Portfolio sync failed: {e}") if force: # 명시적 강제 동기화 요청 시에는 오류 전파 raise HTTPException(status_code=502, detail=f"Failed to sync portfolio: {e}") # 비강제(백그라운드) 동기화 실패 시에는 캐시/빈 값 사용 return {"message": "Sync failed (using cache)", "cached": True, "error": str(e)} finally: portfolio_cache.sync_lock.release() @router.get("/summary") async def get_portfolio_summary(db: AsyncSession = Depends(get_db)): """포트폴리오 요약 (캐시 활용)""" # 거래 이력 기준으로 캐시 무효화 latest_trade_id = await _get_latest_executed_trade_id(db) if latest_trade_id and portfolio_cache.last_trade_id != latest_trade_id: portfolio_cache.mark_changed() if portfolio_cache.summary and not portfolio_cache.should_sync(): logger.debug("Portfolio cache hit (summary)") resp = dict(portfolio_cache.summary) resp.update({ "sync_status": "ok" if portfolio_cache.last_error is None else "error", "last_synced": portfolio_cache.last_synced.isoformat() if portfolio_cache.last_synced else None, "last_error": portfolio_cache.last_error, }) return resp try: await sync_portfolio(db, force=False) except HTTPException: logger.warning("get_portfolio_summary: sync failed, returning cached summary") base = portfolio_cache.summary or { "total_stocks": 0, "total_value": 0, "total_investment": 0, "total_profit_loss": 0, "total_profit_loss_rate": 0, "available_cash": 0, "total_assets": 0 } resp = dict(base) resp.update({ "sync_status": "ok" if portfolio_cache.last_error is None else "error", "last_synced": portfolio_cache.last_synced.isoformat() if portfolio_cache.last_synced else None, "last_error": portfolio_cache.last_error, }) return resp @router.post("/refresh") async def refresh_portfolio(db: AsyncSession = Depends(get_db)): """강제 캐시 무효화 및 재동기화""" portfolio_cache.change_flag = True return await sync_portfolio(db, force=True) async def check_and_execute_exit_trades(db: AsyncSession): """ 보유 종목의 손절/익절 조건을 체크하고 자동 매도 실행 CLAUDE.md 섹션 6.2 + 7.3 + 7.4 기반: - ATR 기반 손절: min(최저가 - 2.5×ATR, 진입가 -6%) ← -8% → -6% - 1차 익절(30%/50%): RSI 55~65 + BB 중앙선 상회 - 2차 익절(50%): RSI ≥ 65 + BB 상단선 도달 - 트레일링 스탑: 최고가 대비 -3% 하락 시 잔여 전량 청산 - 폴백: 고정 비율 손절 (STOP_LOSS_PERCENT) """ from app.core.config import settings from datetime import datetime try: await sync_portfolio(db, force=True) # MDD 업데이트 (async — thread-safe lock 내부) summary = await _get_portfolio_summary(db) total_assets = float(summary.get("total_assets", 0)) if summary else 0 mdd_action = await mdd_manager.update(total_assets) if mdd_action["action"] != "NONE": logger.warning(f"MDD 조치: {mdd_action}") result = await db.execute( select(Portfolio).where(Portfolio.quantity > 0) ) holdings = result.scalars().all() if not holdings: logger.debug("No holdings to check for exit conditions") return executed_exits = [] for holding in holdings: try: stock_code = holding.stock_code stock_name = holding.stock_name quantity = holding.quantity avg_price = float(holding.average_price or 0) # 실시간 가격 확인 price_resp = await korea_investment_api.get_current_price(stock_code) latest_price = float(holding.current_price or 0) if price_resp and price_resp.get("output"): latest_price = int(price_resp["output"].get("stck_prpr", latest_price)) if latest_price <= 0 or avg_price <= 0: continue profit_loss_rate = ((latest_price - avg_price) / avg_price) * 100 # ActiveTrade 조회 (분할 청산 상태 추적) at_result = await db.execute( select(ActiveTrade).where( ActiveTrade.stock_code == stock_code, ActiveTrade.status == "active" ) ) active_trade = at_result.scalar_one_or_none() remaining_pct = float(active_trade.remaining_pct) if active_trade and active_trade.remaining_pct else 1.0 # ATR 기반 청산 시도 (차트 데이터 필요) exit_signal = None 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 # ② 진입 후 최고가 (트레일링 스탑용) — 최근 10봉 최고값으로 근사 trailing_high = max(highs[-10:]) if highs else None # ① 시장 레짐 결정 # TODO: KOSPI 지수 일봉 기반 detect_market_regime() 으로 대체 예정 # 현재는 BULL 기본값 사용 (보수적 선택 — 1차 30% 익절) regime = MarketRegime.BULL 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 chart_err: logger.debug(f"ATR 청산 계산 실패 ({stock_code}): {chart_err}") should_exit = False exit_reason = "" sell_quantity = quantity order_price = latest_price if exit_signal: should_exit = True exit_reason = exit_signal.reason if exit_signal.action == "SELL_PARTIAL": sell_quantity = max(1, int(quantity * exit_signal.sell_pct)) logger.info(f"ATR 청산: {stock_name} - {exit_reason} (매도 {sell_quantity}주)") elif profit_loss_rate <= settings.STOP_LOSS_PERCENT: # 폴백: 고정 비율 손절 (-6%) should_exit = True exit_reason = f"고정 손절 ({profit_loss_rate:.2f}% <= {settings.STOP_LOSS_PERCENT}%)" elif profit_loss_rate >= settings.TAKE_PROFIT_PERCENT: should_exit = True exit_reason = f"고정 익절 ({profit_loss_rate:.2f}% >= {settings.TAKE_PROFIT_PERCENT}%)" if should_exit: logger.info(f"{exit_reason} - {stock_name}({stock_code})") try: order_result = await korea_investment_api.place_order( stock_code=stock_code, order_type="SELL", quantity=sell_quantity, price=order_price ) except Exception as e: logger.warning(f"지정가 실패, 시장가 대체: {e}") order_result = await korea_investment_api.place_order( stock_code=stock_code, order_type="SELL", quantity=sell_quantity, price=None ) executed_price = Decimal(str(order_result.get("output", {}).get("ord_prc", order_price))) total_amount = Decimal(str(executed_price)) * sell_quantity profit_loss = (float(executed_price) - avg_price) * sell_quantity trade = TradeHistory( stock_code=stock_code, stock_name=stock_name, order_type="SELL", order_price=Decimal(str(order_price)), order_quantity=sell_quantity, executed_price=executed_price, executed_quantity=sell_quantity, total_amount=total_amount, status="COMPLETED" if order_result.get("rt_cd") == "0" else "FAILED", order_id=order_result.get("odno"), message=f"자동 청산: {profit_loss_rate:.2f}%", exit_reason=f"{exit_reason}. 평균단가: {avg_price:,.0f}원, 매도가: {latest_price:,.0f}원, 손익: {profit_loss:,.0f}원", created_at=_get_kst_now(), executed_at=_get_kst_now() if order_result.get("rt_cd") == "0" else None ) db.add(trade) await db.commit() if order_result.get("rt_cd") == "0": # 연속 손실 추적 (섹션 7.4) — 인메모리 + DB 영속화 is_loss = profit_loss < 0 loss_tracker.record_trade_result(stock_code, is_loss) await loss_tracker.save_to_db(db, stock_code) # 분할 청산 시 ActiveTrade 잔여 비율 업데이트 if exit_signal and exit_signal.action == "SELL_PARTIAL" and active_trade: # ③ 정확한 잔여 비율 계산: # sell_pct = 현재 보유량 대비 비율이므로 # new_remaining = remaining_pct × (1 - sell_pct) # 예) remaining=0.70, sell_pct=0.50 → 0.70×0.50=0.35 (기존: 0.70-0.50=0.20 오류) new_remaining = remaining_pct * (1.0 - exit_signal.sell_pct) active_trade.remaining_pct = Decimal(str(round(new_remaining, 4))) await db.commit() else: # 전량 매도 holding.quantity = max(0, quantity - sell_quantity) holding.current_price = Decimal(str(latest_price)) if active_trade: active_trade.status = "closed" await db.commit() portfolio_cache.mark_changed() executed_exits.append({ "stock": stock_name, "reason": exit_reason, "profit_loss": profit_loss_rate, "sell_quantity": sell_quantity, }) logger.info(f"자동 매도 완료: {stock_name} {sell_quantity}주 @ {latest_price:,.0f}원") # 매매일지 백그라운드 생성 asyncio.create_task(_create_journal_background( trade_id=trade.id, stock_code=stock_code, stock_name=stock_name, order_type="SELL", executed_price=float(executed_price), quantity=sell_quantity, profit_loss=profit_loss, profit_loss_rate=profit_loss_rate, exit_reason=exit_reason, )) except Exception as e: logger.error(f"Exit trade error for {holding.stock_code}: {e}", exc_info=True) continue if executed_exits: logger.info(f"Executed {len(executed_exits)} exit trades: {executed_exits}") else: logger.debug("No exit conditions met for current holdings") except Exception as e: logger.error(f"check_and_execute_exit_trades error: {e}", exc_info=True) async def _create_journal_background( trade_id: int, stock_code: str, stock_name: str, order_type: str, executed_price: float, quantity: int, profit_loss: float, profit_loss_rate: float, exit_reason: str, ): """매도 체결 후 AI 매매일지 자동 생성 (백그라운드 태스크)""" try: import httpx from app.core.config import settings trade_info = { "stock_code": stock_code, "stock_name": stock_name, "order_type": order_type, "executed_price": executed_price, "quantity": quantity, "profit_loss": profit_loss, "profit_loss_rate": profit_loss_rate, "exit_reason": exit_reason, } content = None mood = None tags = [] # MCP 서버로 AI 일지 생성 요청 (실패 시 규칙 기반 폴백) try: async with httpx.AsyncClient() as client: resp = await client.post( f"{settings.MCP_SERVER_URL}/analyze", json={"analysis_type": "GENERATE_JOURNAL", "data": trade_info}, timeout=30.0, ) if resp.status_code == 200: result = resp.json().get("result", {}) content = result.get("content") or result.get("journal") or result.get("analysis") mood = result.get("mood") tags = result.get("tags", []) except Exception as mcp_err: logger.debug(f"MCP 일지 생성 실패 ({stock_code}): {mcp_err}") # 폴백: 규칙 기반 간단 일지 if not content: direction = "익절" if profit_loss_rate > 0 else "손절" sign = "+" if profit_loss > 0 else "" content = ( f"{stock_name}({stock_code}) 매도 체결.\n" f"매도가: {executed_price:,.0f}원 × {quantity}주\n" f"손익: {sign}{profit_loss:,.0f}원 ({sign}{profit_loss_rate:.2f}%)\n" f"사유: {exit_reason}" ) mood = "좋음" if profit_loss > 0 else "나쁨" tags = [direction, stock_code] # DB 저장 (별도 세션 사용 — 호출 시점의 세션은 이미 닫혔을 수 있음) async with AsyncSessionLocal() as session: journal = TradeJournal( trade_id=trade_id, stock_code=stock_code, stock_name=stock_name, journal_date=datetime.utcnow(), content=content, strategy_name="AUTO_TRADE", mood=mood, tags=tags if tags else None, ) session.add(journal) await session.commit() logger.info(f"매매일지 생성 완료: {stock_name} (trade_id={trade_id})") # WebSocket으로 프론트엔드에 즉시 알림 await websocket_manager.broadcast_json({ "type": "journal_created", "data": {"stock_code": stock_code, "stock_name": stock_name}, }) except Exception as e: logger.error(f"매매일지 생성 오류 ({stock_code}): {e}", exc_info=True) async def _get_portfolio_summary(db: AsyncSession) -> dict: """포트폴리오 요약 (MDD 계산용)""" try: result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) holdings = result.scalars().all() total_eval = sum(float(h.current_price or 0) * h.quantity for h in holdings) total_invest = sum(float(h.average_price or 0) * h.quantity for h in holdings) return {"total_assets": total_eval, "total_invest": total_invest} except Exception: return {"total_assets": 0, "total_invest": 0} @router.post("/sell") async def sell_stock( request: SellRequest, db: AsyncSession = Depends(get_db) ): """보유 종목 수동 매도 (청산 버튼)""" try: stock_code = request.stock_code quantity = request.quantity price = request.price logger.info(f"🔵 수동 매도 요청: {stock_code} {quantity}주 @ {price or '시장가'}") # 보유 여부 확인 result = await db.execute( select(Portfolio).where(Portfolio.stock_code == stock_code) ) holding = result.scalar_one_or_none() if not holding: raise HTTPException(status_code=404, detail=f"보유하지 않은 종목입니다: {stock_code}") if holding.quantity < quantity: raise HTTPException( status_code=400, detail=f"보유 수량 부족: {holding.quantity}주 보유 중, {quantity}주 매도 요청" ) # 매도 주문 실행 order_result = await korea_investment_api.place_order( stock_code=stock_code, order_type='SELL', quantity=quantity, price=price # None이면 시장가 ) if not order_result or order_result.get('rt_cd') != '0': error_msg = order_result.get('msg1', '알 수 없는 오류') if order_result else 'API 호출 실패' raise HTTPException(status_code=500, detail=f"매도 주문 실패: {error_msg}") # TradeHistory 기록 executed_price = price if price else holding.current_price # 체결가는 나중에 업데이트 profit_loss = (executed_price - holding.average_price) * quantity profit_loss_rate = ((executed_price / holding.average_price) - 1) * 100 trade = TradeHistory( stock_code=stock_code, stock_name=holding.stock_name, order_type='SELL', order_price=Decimal(str(price)) if price else None, order_quantity=quantity, executed_price=Decimal(str(executed_price)), executed_quantity=quantity, total_amount=Decimal(str(executed_price * quantity)), status='EXECUTED', order_id=order_result.get('output', {}).get('ODNO'), message=f"수동 청산: 손익 {profit_loss:+,.0f}원 ({profit_loss_rate:+.2f}%)", exit_reason=f"사용자 수동 청산 (손익: {profit_loss:+,.0f}원, 손익률: {profit_loss_rate:+.2f}%)", created_at=_get_kst_now(), executed_at=_get_kst_now() ) db.add(trade) await db.commit() # 포트폴리오 캐시 무효화 portfolio_cache.mark_changed() logger.info(f"✅ 수동 매도 완료: {holding.stock_name} {quantity}주 @ {executed_price:,.0f}원 (손익: {profit_loss:+,.0f}원)") # 매매일지 백그라운드 생성 asyncio.create_task(_create_journal_background( trade_id=trade.id, stock_code=stock_code, stock_name=holding.stock_name, order_type="SELL", executed_price=float(executed_price), quantity=quantity, profit_loss=float(profit_loss), profit_loss_rate=float(profit_loss_rate), exit_reason=f"수동 청산 (손익: {profit_loss:+,.0f}원, {profit_loss_rate:+.2f}%)", )) return { "success": True, "message": f"{holding.stock_name} {quantity}주 매도 주문 완료", "order_id": order_result.get('output', {}).get('ODNO'), "profit_loss": float(profit_loss), "profit_loss_rate": float(profit_loss_rate) } except HTTPException: raise except Exception as e: logger.error(f"Manual sell error: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"매도 처리 중 오류: {str(e)}")