""" 포트폴리오 API 엔드포인트 - GET /: 보유 종목 목록 조회 (캐시 활용) - GET /summary: 포트폴리오 요약 (총자산, 평가손익) - GET /sync: 실제 계좌와 동기화 - POST /sell: 수동 매도 (청산) """ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm.exc import StaleDataError 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 from app.core.timeutil import now_kst 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"] # 예수금 우선순위 (2026-09-23): # _ord_psbl_cash — 매수가능조회로 받은 주문가능현금(당일 매도대금 포함) # prvs_rcdl_excc_amt — 가수도정산금액(주문가능). 데이트레이딩 엔진도 이 값을 쓴다 # nxdy_excc_amt — 익일정산금액 # dnca_tot_amt — 예수금총액(D+2 기준). 당일 매수가 많으면 음수가 되므로 최후순위 # 음수/0은 건너뛰고 첫 양수를 사용한다(화면에 예수금 -913,965원이 뜨던 문제). _cash_keys = [ "_ord_psbl_cash", "prvs_rcdl_excc_amt", "nxdy_excc_amt", "d2_depo_amt", "d2_deposit", "dnca_tot_amt", "nass_amt", ] available_cash = 0.0 for _k in _cash_keys: _v = _pick_float(output2, [_k]) if _v > 0: available_cash = _v if _k != "_ord_psbl_cash": logger.debug(f"예수금 필드 선택: {_k}={_v:,.0f}") break 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 # 최근 수동 매도 {code: (qty, ts)} — KIS 잔고 반영 지연으로 동기화가 판 종목을 # 되살리는 것 방지 (5분 TTL) self.recent_sells: dict[str, tuple[int, datetime]] = {} def note_sell(self, stock_code: str, quantity: int) -> None: self.recent_sells[stock_code] = (quantity, datetime.now()) def recently_sold_qty(self, stock_code: str, ttl_sec: int = 300) -> int: rec = self.recent_sells.get(stock_code) if not rec: return 0 qty, ts = rec if (datetime.now() - ts).total_seconds() > ttl_sec: self.recent_sells.pop(stock_code, None) return 0 return qty 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} # 다른 sync가 진행 중이면 비강제 호출은 기다리지 않고 캐시 반환 # (KIS 지연 시 요청이 30초씩 쌓여 화면 전체가 느려지던 문제, 2026-09-23) if not force and portfolio_cache.sync_lock.locked(): return {"message": "Sync in progress (using cache)", "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 = {} # 주문가능현금(당일 매도대금 포함) → output2에 주입해 available_cash로 사용 try: if balance_data and balance_data.get("rt_cd") == "0" and balance_data.get("output2"): _buyable = await korea_investment_api.get_buyable_cash() if _buyable > 0: _o2 = balance_data["output2"] (_o2[0] if isinstance(_o2, list) else _o2)["_ord_psbl_cash"] = _buyable except Exception as _bc_err: logger.debug(f"주문가능현금 조회 실패 (예수금총액 폴백): {_bc_err}") # 디버깅: KIS API 전체 응답 로깅 logger.info(f"🔍 KIS API balance response keys: {balance_data.keys() if balance_data else 'None'}") # 데이트레이딩 엔진이 보유 중인 종목 → strategy_type='daytrading' 태깅용. # (국내 output1 블록 안에서 정의하면 해외 루프에서 NameError → sync 전체 실패) try: from app.services.daytrading.position_manager import day_trading_state as _dts _dt_qty = { code: int(pos.get("quantity", 0) or 0) for code, pos in _dts.positions.items() } except Exception: _dt_qty = {} _dt_codes = set(_dt_qty) # 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") # 최근 5분 내 수동 매도한 수량 이하로만 남아 있으면 KIS 반영 지연으로 판단 _sold_qty = portfolio_cache.recently_sold_qty(stock_code) if _sold_qty and int(item.get("hldg_qty", 0)) <= _sold_qty: logger.info(f"👻 최근 수동 매도 종목 스킵: {stock_code} (KIS 잔고 반영 대기)") if stock_code in existing_all: await db.delete(existing_all[stock_code]) continue 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" # 전량이 DT 포지션일 때만 태깅(혼재 시 스윙 청산 대상 유지) if stock_code in _dt_codes and quantity <= _dt_qty.get(stock_code, 0): h.strategy_type = "daytrading" 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", strategy_type=( "daytrading" if quantity <= _dt_qty.get(stock_code, 0) else "swing" ), ) 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 # 같은 종목을 스윙+DT로 동시 보유하면 Portfolio 1행에 합쳐지므로, # 전량이 DT 포지션일 때만 태깅한다(부분 혼재 시 스윙 청산 대상 유지). if stock_code in _dt_codes and quantity <= _dt_qty.get(stock_code, 0): h.strategy_type = "daytrading" 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, strategy_type=( "daytrading" if quantity <= _dt_qty.get(stock_code, 0) else "swing" ), ) 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 의존성) 경로에서만 호출되므로 # 실시간 콜백/스케줄러 직접 호출과는 성격이 다름. # ActiveTrade 잔량 정합 — 같은 종목에 active 행이 1개뿐일 때만 실제 보유량으로 교정. # (분할매도·수동매도 누락으로 DB 잔량이 실제와 달라지면 다음 청산이 과다매도) try: from app.core.config import settings as _cfg # 모듈 상단 미임포트(함수 지역 사용) _held = {h.stock_code: h.quantity for h in new_holdings if h.quantity > 0} _at_rows = (await db.execute( select(ActiveTrade).where(ActiveTrade.status == "active") )).scalars().all() _by_code: dict[str, list] = {} for _row in _at_rows: _by_code.setdefault(_row.stock_code, []).append(_row) for _code, _rows in _by_code.items(): if len(_rows) != 1: continue # 스윙+DT 혼재 등 → 자동 교정하지 않음 _row = _rows[0] _actual = _held.get(_code, 0) if _actual == 0: _row.status = "closed" logger.info(f"🧹 ActiveTrade 정리: {_code} 보유 없음 → closed") elif int(_row.quantity or 0) != _actual: logger.info( f"🧹 ActiveTrade 잔량 교정: {_code} {_row.quantity} → {_actual}" ) _row.quantity = _actual # 외부 유입 포지션(다른 인스턴스/증권사 앱 매수 등)은 ActiveTrade 행이 없어 # 손절·분할익절·트레일링 관리를 전혀 받지 못한다(2026-09-22 카카오뱅크 -13%). # 보유 중인데 active 행이 없으면 현재 평균단가를 진입가로 보고 생성한다. # 데이트레이딩 태깅 종목은 제외 — 엔진이 당일 청산 대상으로 복원해버린다. _managed = set(_by_code) for _h in new_holdings: if _h.quantity <= 0 or _h.stock_code in _managed: continue if (_h.strategy_type or "swing") == "daytrading": continue _avg = float(_h.average_price or 0) if _avg <= 0: continue _cur = float(_h.current_price or _avg) db.add(ActiveTrade( user_id=getattr(_h, "user_id", None), stock_code=_h.stock_code, stock_name=_h.stock_name, quantity=_h.quantity, entry_price=Decimal(str(_avg)), current_price=Decimal(str(_cur)), stop_loss=Decimal(str(round(_avg * (1 - abs(_cfg.STOP_LOSS_PERCENT) / 100), 2))), status="active", remaining_pct=Decimal("1.0"), highest_since_entry=Decimal(str(max(_avg, _cur))), strategy_name=_h.strategy_type or "swing", entry_reason="외부 유입 포지션 — 동기화 시 자동 등록(진입가=평균단가)", currency_code=_h.currency_code or "KRW", exchange_code=_h.exchange_code, )) logger.info( f"🆕 ActiveTrade 자동 등록: {_h.stock_name}({_h.stock_code}) " f"{_h.quantity}주 @ {_avg:,.0f} — 외부 유입 포지션" ) except Exception as _rec_err: logger.warning(f"ActiveTrade 정합 실패(무시): {_rec_err}") 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 StaleDataError as e: # 수동 매도(행 삭제)와 동시 실행 시 발생 — 실패로 기록하지 않고 다음 주기에 재시도 # (2026-09-22 14:33/14:42 수동 청산 중 sync 전체가 실패하던 문제) await db.rollback() portfolio_cache.mark_changed() logger.warning(f"Portfolio sync 충돌(동시 매도) — 다음 주기 재시도: {e}") return {"message": "Sync skipped (concurrent update)", "cached": True} 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": now_kst(), "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 original_avg_price = float(holding.average_price or 0) # 매도 주문 실행 # [2026-09-23] safe_place_order 경유 — 종목별 세마포어 + Redis 분산락으로 # 스윙/데이트레이딩 자동 청산과 동시에 매도되는 것을 막는다. from app.services.trade_executor import safe_place_order order_result = await safe_place_order( stock_code=stock_code, order_type='SELL', quantity=quantity, price=price, # None이면 시장가 is_domestic=True, ) 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 기록 # 체결가는 나중에 업데이트. current_price 가 NULL 이면 평균단가로 폴백 — # None 연산으로 TypeError 가 나면 주문은 이미 나간 뒤라 상태 갱신이 전부 누락된다. _exec_raw = price or holding.current_price or holding.average_price or 0 executed_price = Decimal(str(_exec_raw)) _avg_dec = Decimal(str(original_avg_price)) if original_avg_price > 0 else None profit_loss = float((executed_price - _avg_dec) * quantity) if _avg_dec else 0.0 profit_loss_rate = float((executed_price / _avg_dec - 1) * 100) if _avg_dec else 0.0 # 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: # 전량 매도 → 수량 0 (행 삭제는 동시 실행 중인 sync 와 StaleDataError 충돌을 # 일으킨다. 수량 0 행은 조회에서 제외되고 다음 sync 가 KIS 기준으로 정리한다) holding.quantity = 0 holding.current_price = Decimal(str(executed_price)) else: # 분할 매도 → 수량 차감 holding.quantity -= quantity holding.current_price = Decimal(str(executed_price)) await db.commit() # ActiveTrade 정리 — 누락 시 유령 active 행이 남아, 같은 종목을 다시 매수했을 때 # 옛 remaining_pct(예: 0.50)가 살아나 1차 익절이 건너뛰어진다. # (2026-09-23 확인: 삼아알미늄·LG 등 수동 청산 종목이 active 상태로 잔존) from app.services.db_worker import ( enqueue_close_active_trade as _enq_close_at, enqueue_active_trade_upsert as _enq_at_upsert, ) try: if quantity >= original_holding_qty: await _enq_close_at(stock_code, strategy_name=None) else: await _enq_at_upsert({ "stock_code": stock_code, "quantity": original_holding_qty - quantity, "current_price": float(executed_price), }) except Exception as _at_err: logger.warning(f"[SELL] ActiveTrade 정리 실패 {stock_code}: {_at_err}") # 데이트레이딩 엔진이 같은 종목을 보유 중이면 인메모리 잔량도 줄인다. # (누락 시 엔진이 매도된 수량을 그대로 들고 있다고 믿고 다음 사이클에 재매도) try: from app.services.daytrading.position_manager import day_trading_state as _dts async with _dts.trade_lock: _dt_pos = _dts.positions.get(stock_code) if _dt_pos: _left = max(0, int(_dt_pos.get("quantity", 0) or 0) - quantity) if _left <= 0: _dts.remove_position(stock_code) logger.info(f"[SELL] DT 포지션 제거(수동 매도): {stock_code}") else: _dt_pos["quantity"] = _left logger.info(f"[SELL] DT 포지션 잔량 갱신(수동 매도): {stock_code} → {_left}주") except Exception as _dt_err: logger.warning(f"[SELL] DT 인메모리 반영 실패 {stock_code}: {_dt_err}") # ✅ 인메모리 캐시 즉시 반영 (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.note_sell(stock_code, quantity) await portfolio_cache.apply_sell(stock_code, quantity, float(executed_price), original_avg_price) 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 갱신 quantity: Optional[int] = None # 실제 보유 수량으로 교정 (DB 잔량 드리프트 정정용) @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 now_kst() # 부분 매도 시 잔여 비율 갱신 (CLAUDE.md 5.2: +2R 50% 분할 매도 후 잔량 추적 유지) if request.remaining_pct is not None: active_trade.remaining_pct = Decimal(str(round(request.remaining_pct, 4))) if request.quantity is not None: if request.quantity < 0: raise HTTPException(status_code=400, detail="quantity는 0 이상이어야 합니다") logger.info( f"[TradeMonitor API] {stock_code} ActiveTrade 수량 교정: " f"{active_trade.quantity} → {request.quantity}" ) active_trade.quantity = request.quantity 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)}" )