""" 포트폴리오 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, Field 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 from app.services.utils import _is_domestic 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 = Field(gt=0, description="매도 수량 (1 이상)") price: Optional[int] = Field(None, gt=0, description="지정가 (None이면 시장가)") def _get_kst_now(): """현재 한국 시간 반환 (timezone-naive datetime)""" return datetime.now(KST).replace(tzinfo=None) def _pick_float(d: dict, keys: list) -> float: """Helper to extract float value from dict by key priority.""" for k in keys: if k in d and d.get(k) not in (None, ""): try: return float(d[k]) except Exception: continue return 0.0 def _calculate_portfolio_values(holdings: list) -> dict: """Calculate total value, investment, and profit/loss from holdings. Returns dict with total_stocks, total_value, total_investment, total_profit_loss, total_profit_loss_rate, domestic_value, overseas_value_usd. """ total_stocks = len(holdings) # 국내/해외 분리하여 평가금액 계산 domestic_value = sum( float(h.current_price or 0) * h.quantity for h in holdings if _is_domestic(h.stock_code, getattr(h, "market", None)) ) overseas_value_usd = sum( float(h.current_price or 0) * h.quantity for h in holdings if not _is_domestic(h.stock_code, getattr(h, "market", None)) ) # 국내/해외 투자금을 분리 계산 (통화 혼합 방지) domestic_investment = sum( float(h.average_price or 0) * h.quantity for h in holdings if _is_domestic(h.stock_code, getattr(h, "market", None)) ) overseas_investment_usd = sum( float(h.average_price or 0) * h.quantity for h in holdings if not _is_domestic(h.stock_code, getattr(h, "market", None)) ) # 국내 손익 (KRW 단위) domestic_profit_loss = domestic_value - domestic_investment # 해외 손익 (USD 단위) — 원화 환산은 cache_summary()에서 환율 적용 시 수행 overseas_profit_loss_usd = overseas_value_usd - overseas_investment_usd # 하위 호환: total_investment 은 국내만 (기존 호출처 동작 유지) total_investment = domestic_investment total_profit_loss = domestic_profit_loss return { "total_stocks": total_stocks, "domestic_value": domestic_value, "overseas_value_usd": overseas_value_usd, "domestic_investment": domestic_investment, "overseas_investment_usd": overseas_investment_usd, "domestic_profit_loss": domestic_profit_loss, "overseas_profit_loss_usd": overseas_profit_loss_usd, "total_investment": total_investment, "total_profit_loss": total_profit_loss, } def _extract_balance_data(balance_data: dict) -> dict: """Extract cash and total assets from KIS API balance response. Returns dict with available_cash, total_assets, kis_total_assets, output2. """ available_cash = 0.0 total_assets = 0.0 kis_total_assets = 0.0 output2 = {} 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"] available_cash = _pick_float(output2, [ "dnca_tot_amt", "d2_depo_amt", "d2_deposit", "nass_amt" ]) kis_total_assets = _pick_float(output2, ["tot_evlu_amt", "tot_evlu_amt_smtl"]) if kis_total_assets > 0: total_assets = kis_total_assets try: logger.debug( "Portfolio balance extract - 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 return { "available_cash": available_cash, "total_assets": total_assets, "kis_total_assets": kis_total_assets, "output2": output2, } def _extract_detailed_balance(balance_data: dict) -> dict: """Extract D+2 settlement and stock value details from KIS API response.""" d2_settlement = 0.0 stock_value_from_api = 0.0 bfdy_total_assets = 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"]) bfdy_total_assets = _pick_float(output2, ["bfdy_tot_asst_evlu_amt"]) return { "d2_settlement": d2_settlement, "stock_value_from_api": stock_value_from_api, "bfdy_total_assets": bfdy_total_assets, } def _extract_overseas_balance(overseas_balance_data: dict) -> dict: """Extract overseas holdings total from KIS API response.""" overseas_total_krw = 0.0 if overseas_balance_data and "output2" in overseas_balance_data: ovrs_out2 = overseas_balance_data["output2"] if isinstance(ovrs_out2, list): ovrs_out2 = ovrs_out2[0] if ovrs_out2 else {} # tot_evlu_amt: 해외주식 평가금액 합계 (원화) overseas_total_krw = _pick_float(ovrs_out2, ["tot_evlu_amt", "frcr_evlu_tota"]) if overseas_total_krw > 0: logger.debug(f"💵 해외 잔고 합산: {overseas_total_krw:,.0f}원") return {"overseas_total_krw": overseas_total_krw} # 단순 인메모리 캐시 (프로세스 단위). 다중 프로세스 환경이면 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.summary_lock = asyncio.Lock() # summary 수정 보호용 락 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 get_summary_copy(self) -> dict | None: """summary의 스냅샷 반환 (락 없이, 읽기 전용). summary가 수정 중일 수 있으므로 복사본을 반환.""" if self.summary is None: return None return dict(self.summary) async def apply_buy(self, stock_code: str, stock_name: str, quantity: int, price: float): """매수 체결 즉시 캐시 반영 — KIS 재동기화 전까지 summary 수치 정확하게 유지""" async with self.summary_lock: if self.summary is None: self.mark_changed() return cost = quantity * price # 예수금 차감 self.summary["available_cash"] = max(0.0, float(self.summary.get("available_cash", 0)) - cost) # 주식 평가금액 증가 self.summary["total_value"] = float(self.summary.get("total_value", 0)) + cost # 매입금액 증가 self.summary["total_investment"] = float(self.summary.get("total_investment", 0)) + cost # 총 자산 = 예수금 + 주식평가 (재계산) self.summary["total_assets"] = self.summary["available_cash"] + self.summary["total_value"] # 평가손익 재계산 inv = self.summary["total_investment"] val = self.summary["total_value"] self.summary["total_profit_loss"] = val - inv self.summary["total_profit_loss_rate"] = ((val - inv) / inv * 100) if inv > 0 else 0 self.summary["total_stocks"] = len([h for h in self.holdings if h.quantity > 0]) self.mark_changed() async def apply_sell(self, stock_code: str, quantity: int, price: float, avg_price: float): """매도 체결 즉시 캐시 반영""" async with self.summary_lock: if self.summary is None: self.mark_changed() return proceeds = quantity * price cost = quantity * avg_price # 예수금 증가 self.summary["available_cash"] = float(self.summary.get("available_cash", 0)) + proceeds # 주식 평가금액 감소 self.summary["total_value"] = max(0.0, float(self.summary.get("total_value", 0)) - proceeds) # 매입금액 감소 self.summary["total_investment"] = max(0.0, float(self.summary.get("total_investment", 0)) - cost) # 총 자산 = 예수금 + 주식평가 self.summary["total_assets"] = self.summary["available_cash"] + self.summary["total_value"] # 평가손익 재계산 inv = self.summary["total_investment"] val = self.summary["total_value"] self.summary["total_profit_loss"] = val - inv self.summary["total_profit_loss_rate"] = ((val - inv) / inv * 100) if inv > 0 else 0 self.summary["total_stocks"] = len([h for h in self.holdings if h.quantity > 0]) self.mark_changed() 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) > self.min_interval: # min_interval(15초)마다 갱신 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 async def cache_summary(self, balance_data: dict = None, overseas_balance_data: dict = None, exchange_rate: float = 0.0): """포트폴리오 요약 캐시 업데이트. Args: balance_data: 국내 잔고 KIS API 응답 overseas_balance_data: 해외 잔고 KIS API 응답 exchange_rate: USD/KRW 환율 (해외 종목 원화 환산용, 0이면 미제공) """ # 1. Calculate portfolio values from holdings holdings = self.holdings portfolio_values = _calculate_portfolio_values(holdings) domestic_value = portfolio_values["domestic_value"] overseas_value_usd = portfolio_values["overseas_value_usd"] domestic_investment = portfolio_values["domestic_investment"] overseas_investment_usd = portfolio_values["overseas_investment_usd"] # 해외 원화 환산 (환율 미제공 시 overseas_balance_data output2의 tot_evlu_amt 사용) overseas_value_krw = overseas_value_usd * exchange_rate if exchange_rate > 0 else 0.0 overseas_investment_krw = overseas_investment_usd * exchange_rate if exchange_rate > 0 else 0.0 total_value = domestic_value + overseas_value_krw # 손익 계산: 국내(KRW) + 해외(환율 적용 KRW) 분리 후 합산 domestic_profit_loss = domestic_value - domestic_investment overseas_profit_loss_krw = overseas_value_krw - overseas_investment_krw total_investment = domestic_investment + overseas_investment_krw total_profit_loss = domestic_profit_loss + overseas_profit_loss_krw total_profit_loss_rate = (total_profit_loss / total_investment * 100) if total_investment > 0 else 0 # 2. Extract balance data from KIS API balance_info = _extract_balance_data(balance_data) available_cash = balance_info["available_cash"] total_assets = balance_info["total_assets"] kis_total_assets = balance_info["kis_total_assets"] output2 = balance_info["output2"] # Log discrepancies if kis_total_assets > 0: calculated_assets = total_value + available_cash 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: total_assets = total_value + available_cash # 3. Extract detailed balance info (D+2, stock value, previous day total) detail_info = _extract_detailed_balance(balance_data) d2_settlement = detail_info["d2_settlement"] stock_value_from_api = detail_info["stock_value_from_api"] bfdy_total_assets = detail_info["bfdy_total_assets"] # 4. Extract overseas balance overseas_info = _extract_overseas_balance(overseas_balance_data) overseas_total_krw = overseas_info["overseas_total_krw"] if overseas_total_krw > 0: total_assets += overseas_total_krw # 5. Calculate daily profit/loss today_profit_loss = (total_assets - bfdy_total_assets) if bfdy_total_assets > 0 else 0.0 today_profit_loss_rate = (today_profit_loss / bfdy_total_assets * 100) if bfdy_total_assets > 0 else 0.0 # 6. Build and store summary summary_data = { "total_stocks": portfolio_values["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, "stock_value_from_api": stock_value_from_api, "bfdy_total_assets": bfdy_total_assets, "today_profit_loss": today_profit_loss, "today_profit_loss_rate": today_profit_loss_rate, # 해외 관련 추가 필드 "overseas_value_usd": overseas_value_usd, "overseas_value_krw": overseas_value_krw if exchange_rate > 0 else overseas_total_krw, "exchange_rate": exchange_rate, } # summary 수정을 락으로 보호 async with self.summary_lock: self.summary = summary_data portfolio_cache = PortfolioCache() from pydantic import BaseModel router = APIRouter() class PortfolioResponse(BaseModel): id: int stock_code: str stock_name: Optional[str] = None quantity: int average_price: Optional[Decimal] = None current_price: Optional[Decimal] = None profit_loss: Optional[Decimal] = None profit_loss_rate: Optional[Decimal] = None market: str = "KRX" strategy_type: str = "swing" 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, overseas_balance_data = await asyncio.gather( korea_investment_api.get_balance(), korea_investment_api.get_overseas_balance(), return_exceptions=True, ) if isinstance(balance_data, Exception): logger.warning(f"⚠️ 국내 잔고 조회 실패: {balance_data}") balance_data = {} if isinstance(overseas_balance_data, Exception): logger.warning(f"⚠️ 해외 잔고 조회 실패: {overseas_balance_data}") overseas_balance_data = {} # 디버깅: 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)}종목") # summary는 절대 None으로 덮어쓰지 않는다 — 마지막 성공 summary를 그대로 보존한다. # 단, 최초 동기화부터 실패하여 summary가 아직 한 번도 채워진 적 없다면(None) # 현금 계산이 영영 막히지 않도록 holdings 기반 부분값(available_cash=0)이라도 채워둔다. if portfolio_cache.summary is None: try: await portfolio_cache.cache_summary() # available_cash=0이지만 holdings는 즉시 반영 logger.info("📦 summary 최초 미존재 — holdings 기반 부분값으로 채움 (available_cash=0)") except Exception as summary_err: logger.warning(f"⚠️ 부분 summary 생성 실패 (summary는 None 유지): {summary_err}") 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) recent_point = today_start logger.info(f"🔎 Checking recent trades since {recent_point} for Ghost detection") 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"} # 부분 매도 보정: KIS 잔고에서 여전히 수량이 있는 종목은 ghost에서 제외 if ghost_candidates and "output1" in balance_data: still_holding = { item.get("pdno") for item in balance_data["output1"] if int(item.get("hldg_qty", 0)) > 0 } partial_sell_excluded = ghost_candidates & still_holding if partial_sell_excluded: logger.info(f"↩️ 부분매도 종목은 ghost 제외 (잔고 있음): {partial_sell_excluded}") ghost_candidates -= still_holding 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: # ── 현재가 병렬 조회 (N+1 순차 → asyncio.gather) ────────────────── _prefetch_codes = [ item.get("pdno") for item in balance_data["output1"] if item.get("pdno") not in ghost_candidates and int(item.get("hldg_qty", 0)) > 0 ] if _prefetch_codes: _price_results = await asyncio.gather( *[korea_investment_api.get_current_price(c) for c in _prefetch_codes], return_exceptions=True, ) _realtime_price_map = dict(zip(_prefetch_codes, _price_results)) else: _realtime_price_map = {} # ────────────────────────────────────────────────────────────────── for item in balance_data["output1"]: stock_code = item.get("pdno") 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 average_price = Decimal(str(item.get("pchs_avg_pric", 0))) # 실시간 현재가 (병렬 조회 결과 참조) price_resp = _realtime_price_map.get(stock_code) try: if price_resp and not isinstance(price_resp, Exception) and "output" in price_resp: realtime_price = int(price_resp["output"].get("stck_prpr", 0)) current_price = Decimal(str(realtime_price)) if realtime_price > 0 else Decimal(str(item.get("prpr", 0))) else: if isinstance(price_resp, Exception): logger.warning(f"⚠️ 실시간 현재가 조회 실패 ({stock_code}): {price_resp}, fallback to prpr") 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))) 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 h.market = "KRX" 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, market="KRX", ) db.add(new_holding) new_holdings.append(new_holding) # ── 해외 잔고 처리 ───────────────────────────────────────────────── if overseas_balance_data and overseas_balance_data.get("rt_cd") == "0" and "output1" in overseas_balance_data: ovrs_items = overseas_balance_data["output1"] or [] # 해외 현재가 병렬 조회 _ovrs_codes = [ item.get("ovrs_pdno") for item in ovrs_items if item.get("ovrs_pdno") and int(item.get("ord_psbl_qty", 0)) > 0 ] if _ovrs_codes: _ovrs_price_results = await asyncio.gather( *[korea_investment_api.get_overseas_price(c, item.get("ovrs_excg_cd", "NASD")) for c, item in zip(_ovrs_codes, ovrs_items)], return_exceptions=True, ) _ovrs_price_map = dict(zip(_ovrs_codes, _ovrs_price_results)) else: _ovrs_price_map = {} for item in ovrs_items: stock_code = item.get("ovrs_pdno") if not stock_code: continue quantity = int(item.get("ord_psbl_qty", 0)) if quantity <= 0: continue stock_name = item.get("ovrs_item_name", stock_code) exchange_code = item.get("ovrs_excg_cd", "NASD") average_price = Decimal(str(item.get("pchs_avg_pric", 0))) # 실시간 현재가 price_resp = _ovrs_price_map.get(stock_code) try: if price_resp and not isinstance(price_resp, Exception) and "output" in price_resp: realtime_price = float(price_resp["output"].get("last", 0)) current_price = Decimal(str(realtime_price)) if realtime_price > 0 else Decimal(str(item.get("now_pric2", 0))) else: current_price = Decimal(str(item.get("now_pric2", 0))) except Exception: current_price = Decimal(str(item.get("now_pric2", 0))) profit_loss = Decimal(str(item.get("evlu_pfls_amt", 0))) profit_loss_rate = Decimal(str(item.get("evlu_pfls_rt2", 0))) logger.info( f"[해외] Portfolio sync - {stock_name}({stock_code}/{exchange_code}): " f"수량={quantity}, 평균단가={average_price}, 현재가={current_price}" ) 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 h.market = exchange_code 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, market=exchange_code, ) db.add(new_holding) new_holdings.append(new_holding) # ────────────────────────────────────────────────────────────────── # 제거된 종목 반영 (DB에서 완전 삭제) for code, h in existing_all.items(): if code not in processed_codes: 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) # TODO: db_worker 마이그레이션 필요 — 읽기+업데이트가 같은 트랜잭션에서 # 이뤄져야 하므로(Portfolio SELECT → UPDATE/DELETE → commit) 큐로 분리가 어려움. # sync_portfolio()는 API 엔드포인트(get_db 의존성) 경로에서만 호출되므로 # 실시간 콜백/스케줄러 직접 호출과는 성격이 다름. await db.commit() portfolio_cache.holdings = [h for h in new_holdings if h.quantity > 0] await portfolio_cache.cache_summary(balance_data, overseas_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 = portfolio_cache.get_summary_copy() or {} else: try: await sync_portfolio(db, force=False) except HTTPException: logger.warning("get_portfolio_summary: sync failed, returning cached summary") summary_copy = portfolio_cache.get_summary_copy() resp = summary_copy 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, } # 누적 실현 손익 (TradeHistory SELL 체결 기준) try: realized_result = await db.execute( select(func.sum(TradeHistory.net_profit_loss)).where( TradeHistory.order_type == "SELL", TradeHistory.status.in_(["EXECUTED", "COMPLETED"]), TradeHistory.net_profit_loss.isnot(None), ) ) cumulative_realized_pl = float(realized_result.scalar() or 0) except Exception as e: logger.warning(f"누적 실현손익 조회 실패: {e}") cumulative_realized_pl = 0.0 resp["cumulative_realized_pl"] = cumulative_realized_pl 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 async def _sync_portfolio_background(): """백그라운드 포트폴리오 KIS 동기화 (비차단). 이미 동기화 중이면 건너뜀. [DB 직접 접근 예외 사유] sync_portfolio()는 KIS API → Portfolio READ/UPSERT → ActiveTrade UPSERT의 복합 트랜잭션이 한 세션 내에서 원자적으로 실행되어야 합니다. db_worker 큐는 단순 INSERT/UPDATE 태스크 단위를 처리하므로 이 복합 조회+쓰기 흐름을 분리하면 일관성이 깨집니다. 따라서 이 함수에 한해 AsyncSessionLocal 직접 사용을 허용합니다. (실시간 WebSocket 콜백이 아닌 asyncio.create_task 백그라운드 태스크에서 호출) """ if portfolio_cache.sync_lock.locked(): return try: async with AsyncSessionLocal() as db: await sync_portfolio(db, force=False) except Exception as e: logger.debug(f"Background portfolio sync: {e}") def _build_combined_resp() -> dict: """portfolio_cache에서 combined 응답 딕셔너리 생성""" 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 { "holdings": [h for h in portfolio_cache.holdings if h.quantity > 0], "summary": resp, } @router.get("/combined") async def get_portfolio_combined(db: AsyncSession = Depends(get_db)): """보유종목 목록 + 요약을 반환. 응답 우선순위: ① 신선한 캐시 → 즉시 반환 ② 캐시 만료 + 매매 변경 없음 → 기존 캐시 즉시 반환 + 백그라운드 KIS 갱신 ③ 서버 재시작 등 캐시 비어있음 → DB 즉시 반환 + 백그라운드 KIS 갱신 ④ 매매 발생(change_flag) 또는 DB도 비어있음 → KIS API 동기 호출 (필수) """ 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 portfolio_cache.summary and not portfolio_cache.should_sync(): logger.debug("Portfolio cache hit (combined)") return _build_combined_resp() # ② 캐시 만료, 매매 변경 없음 → stale 데이터 즉시 반환 + 백그라운드 갱신 if portfolio_cache.holdings and portfolio_cache.summary and not portfolio_cache.change_flag: logger.debug("Portfolio cache stale (no trade) → returning cached, syncing in background") asyncio.create_task(_sync_portfolio_background()) return _build_combined_resp() # ③ 캐시 비어있음 (서버 재시작) → DB에서 즉시 로드 + 백그라운드 KIS 갱신 if not portfolio_cache.holdings: result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) db_holdings = list(result.scalars().all()) if db_holdings: portfolio_cache.holdings = db_holdings await portfolio_cache.cache_summary() # available_cash=0이지만 holdings는 즉시 표시 logger.info(f"Portfolio cold-start: DB {len(db_holdings)}종목 즉시 반환, KIS 갱신 백그라운드 시작") asyncio.create_task(_sync_portfolio_background()) return _build_combined_resp() # ④ 매매 발생 또는 DB도 비어있음 → KIS API 동기 호출 try: await sync_portfolio(db, force=False) except HTTPException: logger.warning("get_portfolio_combined: sync failed, returning cached data") return _build_combined_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): """[DEPRECATED] 스윙 포지션 손절/익절 체크. .. deprecated:: services/swing/exit_service.py의 check_and_execute_exit_v2()로 대체됨. swing_scheduler가 호출. 이 함수는 더 이상 활성 스케줄러에서 호출되지 않음. 하위 호환을 위해 함수는 유지. CLAUDE.md 섹션 6.2 + 7.3 + 7.4 기반: - ATR 기반 손절: min(최저가 - 2.5×ATR, 진입가 -6%) - 1차 익절(30%/50%): RSI 55~65 + BB 중앙선 상회 - 2차 익절(50%): RSI ≥ 65 + BB 상단선 도달 - 트레일링 스탑: 최고가 대비 -3% 하락 시 잔여 전량 청산 - 폴백: 고정 비율 손절 (STOP_LOSS_PERCENT) """ import warnings warnings.warn( "check_and_execute_exit_trades is deprecated. " "Use services/swing/exit_service.check_and_execute_exit_v2() instead.", DeprecationWarning, stacklevel=2, ) from app.core.config import settings from datetime import datetime try: # 매매가 발생했거나 캐시가 비어있을 때만 KIS 잔고 조회 (API 호출 최소화) latest_trade_id = await _get_latest_executed_trade_id(db) if latest_trade_id != portfolio_cache.last_trade_id or not portfolio_cache.holdings: logger.info(f"🔄 New trade detected (prev={portfolio_cache.last_trade_id}, cur={latest_trade_id}) — syncing balance") await sync_portfolio(db, force=True) else: logger.debug("⏭️ No new trade since last sync — skipping get_balance()") # 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}") # CRITICAL: MDD 액션에 따른 실제 포지션 축소 매도 실행 try: reduction_results = await mdd_manager.execute_reduction(mdd_action) if reduction_results: logger.warning(f"MDD 강제 축소 완료: {len(reduction_results)}종목 매도") except Exception as e: logger.error(f"MDD 강제 축소 실행 실패: {e}", exc_info=True) 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) 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 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 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 trailing_high = max(highs[-10:]) if highs else None 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}") else: # 해외: 해외 차트로 ATR 계산 시도 try: chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, market) chart_data = chart_resp.get("output2", []) if len(chart_data) >= 20: closes = [float(d.get("clos", 0)) for d in reversed(chart_data)] highs = [float(d.get("high", 0)) for d in reversed(chart_data)] lows = [float(d.get("low", 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 trailing_high = max(highs[-10:]) if highs else None 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}주)") # [이슈9 연동] STOP_LOSS_PERCENT는 양수(6.0)로 저장됨 — 손절 비교 시 음수 변환 elif profit_loss_rate <= -abs(settings.STOP_LOSS_PERCENT): should_exit = True exit_reason = f"고정 손절 ({profit_loss_rate:.2f}% <= -{abs(settings.STOP_LOSS_PERCENT):.1f}%)" 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}) [{'국내' if domestic else '해외/' + market}]") # ── 주문 실행 (국내/해외 분기) ─────────────────────────── if domestic: 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 ) else: order_result = await korea_investment_api.place_overseas_order( symbol=stock_code, order_type="SELL", quantity=sell_quantity, price=order_price, exchange_code=market, ) # 해외 주문 응답은 ord_prc 대신 ovrs_ord_unpr 사용 _out = order_result.get("output", {}) _raw_price = _out.get("ord_prc") or _out.get("ovrs_ord_unpr") or order_price executed_price = Decimal(str(_raw_price)) total_amount = Decimal(str(executed_price)) * sell_quantity profit_loss = (float(executed_price) - avg_price) * sell_quantity # TradeHistory DB 기록 (db_worker 큐 경유 — CLAUDE.md 3.2 준수) from app.services.db_worker import enqueue_trade_history _trade_status = "COMPLETED" if order_result.get("rt_cd") == "0" else "FAILED" await enqueue_trade_history({ "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": _trade_status, "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, }) if order_result.get("rt_cd") == "0": # 연속 손실 추적 (섹션 7.4) — 인메모리 + DB 영속화 is_loss = profit_loss < 0 await loss_tracker.record_trade_result_safe(stock_code, is_loss) await loss_tracker.save_to_db(db, stock_code) # 분할 청산 시 ActiveTrade 잔여 비율 업데이트 # TODO: db_worker 마이그레이션 필요 — 읽기+업데이트가 같은 트랜잭션이라 큐 분리 어려움 if exit_signal and exit_signal.action == "SELL_PARTIAL" and sell_quantity < quantity: new_remaining = remaining_pct * (1.0 - exit_signal.sell_pct) # 잔여 수량 즉시 반영 (다음 사이클 원래 수량 기준 재매도 방지) holding.quantity = quantity - sell_quantity holding.current_price = Decimal(str(latest_price)) if active_trade: active_trade.remaining_pct = Decimal(str(round(new_remaining, 4))) else: # [FIX] ActiveTrade 없는 포지션(동기화 유입)은 remaining_pct를 # 저장할 곳이 없어 부분매도가 매 사이클 반복됨 → 행 생성 db.add(ActiveTrade( stock_code=stock_code, stock_name=stock_name, quantity=quantity - sell_quantity, entry_price=Decimal(str(avg_price)), current_price=Decimal(str(latest_price)), status="active", remaining_pct=Decimal(str(round(new_remaining, 4))), highest_since_entry=Decimal(str(latest_price)), strategy_name="swing", entry_reason="동기화 유입 포지션 (부분매도 시 자동 생성)", currency_code="KRW" if domestic else "USD", )) 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}원") # 매매일지 백그라운드 생성 — trade_id 미확보이므로 0 전달 asyncio.create_task(_create_journal_background( trade_id=0, 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 일지 생성 요청 (실패 시 규칙 기반 폴백) _mcp_headers = {"X-API-Key": settings.MCP_API_KEY} if settings.MCP_API_KEY else {} try: async with httpx.AsyncClient(headers=_mcp_headers) 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 저장 — CLAUDE.md 3.2: db_worker 큐 경유 from app.services.db_worker import enqueue_trade_journal_raw await enqueue_trade_journal_raw({ "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, }) 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}주 매도 요청" ) # ✅ delete/commit 후 접근 불가를 대비해 미리 캡처 stock_name = holding.stock_name original_holding_qty = holding.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 # CLAUDE.md 3.2: db_worker 큐 경유 from app.services.db_worker import enqueue_trade_history await enqueue_trade_history({ "stock_code": stock_code, "stock_name": stock_name, "order_type": "SELL", "order_price": float(price) if price else None, "order_quantity": quantity, "executed_price": float(executed_price), "executed_quantity": quantity, "total_amount": float(executed_price * quantity), "status": "COMPLETED", "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}%)", "net_profit_loss": profit_loss, "profit_loss_rate": profit_loss_rate, "strategy_type": "manual", "created_at": _get_kst_now(), "executed_at": _get_kst_now(), }) # ✅ Portfolio DB 즉시 반영 (매도 후 보유종목에서 즉시 제거) if quantity >= original_holding_qty: # 전량 매도 → 레코드 삭제 await db.delete(holding) else: # 분할 매도 → 수량 차감 holding.quantity -= quantity holding.current_price = Decimal(str(executed_price)) await db.commit() # ✅ 인메모리 캐시 즉시 반영 (KIS 재동기화 전에도 화면에서 즉시 제거) if quantity >= original_holding_qty: portfolio_cache.holdings = [ h for h in portfolio_cache.holdings if h.stock_code != stock_code ] else: for h in portfolio_cache.holdings: if h.stock_code == stock_code: h.quantity -= quantity break portfolio_cache.mark_changed() logger.info(f"✅ 수동 매도 완료: {stock_name} {quantity}주 @ {executed_price:,.0f}원 (손익: {profit_loss:+,.0f}원)") # 매매일지 백그라운드 생성 — trade_id 미확보이므로 0 전달 asyncio.create_task(_create_journal_background( trade_id=0, stock_code=stock_code, stock_name=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"{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)}") class ActiveTradeStatusUpdateRequest(BaseModel): """활성 거래 상태 업데이트 요청""" status: str # 'active', 'pending_sell', 'closed' reason: Optional[str] = "" updated_at: Optional[str] = None remaining_pct: Optional[float] = None # 부분 매도 후 잔여 비율 (예: 0.5). 제공 시 ActiveTrade.remaining_pct 갱신 @router.patch("/active-trades/{stock_code}/status") async def update_active_trade_status( stock_code: str, request: ActiveTradeStatusUpdateRequest, db: AsyncSession = Depends(get_db) ): """ 활성 거래의 상태를 업데이트합니다. CLAUDE.md Section 3.2: trade_monitor에서 직접 DB commit 대신 이 API를 호출 - status: 'active' | 'pending_sell' | 'closed' - reason: 상태 변경 이유 - updated_at: 업데이트 시각 Args: stock_code: 종목 코드 request: 상태 업데이트 요청 db: DB 세션 Returns: 성공 결과 또는 에러 """ try: # 활성 거래 조회 result = await db.execute( select(ActiveTrade).where( ActiveTrade.stock_code == stock_code, ActiveTrade.status != 'closed' ) ) active_trade = result.scalar_one_or_none() if not active_trade: raise HTTPException( status_code=404, detail=f"활성 거래를 찾을 수 없음: {stock_code}" ) # 상태 유효성 검증 valid_statuses = {'active', 'pending_sell', 'closed'} if request.status not in valid_statuses: raise HTTPException( status_code=400, detail=f"유효하지 않은 상태: {request.status}. 허용값: {valid_statuses}" ) # 상태 변경 기록 (로깅) old_status = active_trade.status new_status = request.status reason = request.reason or "" logger.info( f"[TradeMonitor API] 활성 거래 상태 변경: " f"{stock_code} {old_status} → {new_status} (이유: {reason})" ) # 상태 업데이트 active_trade.status = new_status active_trade.updated_at = datetime.fromisoformat(request.updated_at) if request.updated_at else datetime.utcnow() # 부분 매도 시 잔여 비율 갱신 (CLAUDE.md 5.2: +2R 50% 분할 매도 후 잔량 추적 유지) if request.remaining_pct is not None: active_trade.remaining_pct = Decimal(str(round(request.remaining_pct, 4))) db.add(active_trade) await db.commit() await db.refresh(active_trade) return { "success": True, "stock_code": stock_code, "old_status": old_status, "new_status": new_status, "reason": reason, "updated_at": active_trade.updated_at.isoformat(), "remaining_pct": float(active_trade.remaining_pct) if active_trade.remaining_pct is not None else None } except HTTPException: raise except Exception as e: logger.error( f"[TradeMonitor API] 상태 업데이트 실패 ({stock_code}): {e}", exc_info=True ) await db.rollback() raise HTTPException( status_code=500, detail=f"상태 업데이트 중 오류: {str(e)}" )