From ab591f7fd621c54732745148f1c07392021a09cd Mon Sep 17 00:00:00 2001 From: shuim Date: Tue, 10 Mar 2026 17:26:37 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20MCP=20=EC=84=9C=EB=B2=84=20=EB=A6=AC?= =?UTF-8?q?=EC=8A=A4=ED=81=AC=20=EA=B2=80=EC=A6=9D=20=ED=8C=8C=EC=9D=B4?= =?UTF-8?q?=ED=94=84=EB=9D=BC=EC=9D=B8=20=EB=B0=8F=20=EC=B2=AD=EC=82=B0=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=EB=A8=B8=EC=8B=A0=20=EA=B0=9C=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - risk_validator.py: 가격/포지션/MDD/레버리지 5단계 검증 - auto_trade_claude.py: 주문 전 리스크 검증 파이프라인 통합 - trade_monitor.py: ATR 기반 익절/트레일링 스탑, API 기반 상태 전환 (active → pending_sell → closed), 2R 수익 시 50% 분할 매도 Co-Authored-By: Claude Opus 4.6 --- mcp-server/auto_trade_claude.py | 141 ++++++++++- mcp-server/risk_validator.py | 431 ++++++++++++++++++++++++++++++++ mcp-server/trade_monitor.py | 342 ++++++++++++++++++++++--- 3 files changed, 872 insertions(+), 42 deletions(-) create mode 100644 mcp-server/risk_validator.py diff --git a/mcp-server/auto_trade_claude.py b/mcp-server/auto_trade_claude.py index fa25c5f..cc43e4c 100644 --- a/mcp-server/auto_trade_claude.py +++ b/mcp-server/auto_trade_claude.py @@ -2,8 +2,22 @@ import re import asyncio import json import os +import sys +import logging from datetime import datetime, timezone +from decimal import Decimal + +# Backend imports for risk validation +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../backend'))) from app.services.korea_investment_api import korea_investment_api +from app.core.database import AsyncSessionLocal + +# Risk validator +from risk_validator import validate_and_log_order + +# Logging setup +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') +logger = logging.getLogger(__name__) # Claude 응답에서 종목/가격 추출 CLAUDE_FORMAT = r"종목[:\s]+(\d+)" # 종목코드 @@ -41,39 +55,144 @@ def save_active_trade(trade_info): json.dump(trades, f, indent=4, ensure_ascii=False) print(f"✅ 거래 정보 저장 완료: {trade_info['stock_code']}") +async def get_account_balance() -> float: + """Fetch current account balance from KIS API""" + try: + balance_info, _ = await korea_investment_api._make_request( + "GET", + "/uapi/domestic-stock/v1/trading/inquire-account", + "TTTC8434R", + params={} + ) + if balance_info and balance_info.get('rt_cd') == '0': + output = balance_info.get('output', {}) + # Try multiple possible field names for balance + balance = float( + output.get('tot_evalu_amt') or + output.get('tot_assets') or + output.get('dnca_tot_amt') or + 0 + ) + return balance + return 0 + except Exception as e: + logger.error(f"Failed to fetch account balance: {e}") + return 0 + + +async def get_stock_info(stock_code: str) -> tuple: + """Fetch stock name and current price""" + try: + price_info, _ = await korea_investment_api._make_request( + "GET", + "/uapi/domestic-stock/v1/quotations/inquire-price", + "FHKST01010100", + params={"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code} + ) + if price_info and price_info.get('rt_cd') == '0': + output = price_info.get('output', {}) + stock_name = output.get('hts_kor_isnm', 'Unknown') + current_price = int(output.get('stck_prpr', 0)) + return stock_name, current_price + return 'Unknown', 0 + except Exception as e: + logger.error(f"Failed to fetch stock info for {stock_code}: {e}") + return 'Unknown', 0 + + async def auto_trade_from_claude(text, quantity=1): + """ + Execute trade from Claude response with comprehensive risk validation. + + Flow: + 1. Parse Claude response + 2. Fetch account balance and stock info + 3. Run comprehensive risk validation (CLAUDE.md Section 5) + 4. Place order only if all checks pass + 5. Log rejection if any check fails + """ info = parse_claude_response(text) - if info['stock_code'] and info['entry_price']: - print(f"▶️ 매수 주문 시도: {info['stock_code']}, 수량: {quantity}, 가격: {info['entry_price']}") + + if not info['stock_code'] or not info['entry_price']: + logger.error('Claude 응답에서 종목/가격을 찾을 수 없습니다.') + return + + stock_code = info['stock_code'] + entry_price = info['entry_price'] + + logger.info(f"▶️ Processing trade: {stock_code}, Qty: {quantity}, Price: {entry_price}") + + # Step 1: Fetch account balance + account_balance = await get_account_balance() + if account_balance <= 0: + logger.error(f"❌ Cannot determine account balance. Trading aborted.") + return + + logger.info(f"Account balance: {account_balance:,.0f} KRW") + + # Step 2: Fetch stock info + stock_name, current_price = await get_stock_info(stock_code) + logger.info(f"Stock info: {stock_code} ({stock_name}), Current price: {current_price} KRW") + + # Step 3: COMPREHENSIVE RISK VALIDATION (CLAUDE.md Section 5) + allowed, validation_details = await validate_and_log_order( + stock_code=stock_code, + stock_name=stock_name, + order_quantity=quantity, + entry_price=entry_price, + account_balance=account_balance, + market="KRX" # Default to domestic market + ) + + # Step 4: Log validation result + logger.info(f"\nRisk Validation Result:\n{json.dumps(validation_details, indent=2, ensure_ascii=False)}\n") + + # If validation failed, abort and log rejection + if not allowed: + logger.error(f"🚫🚫🚫 [{stock_code}] Trade REJECTED due to risk validation failure") + return + + # Step 5: Place order (only if all checks pass) + logger.info(f"✅ Risk validation passed. Proceeding with order placement...") + print(f"▶️ 매수 주문 시도: {stock_code}, 수량: {quantity}, 가격: {entry_price}") + + try: # 지정가 매수 주문 result = await korea_investment_api.place_order( - stock_code=info['stock_code'], + stock_code=stock_code, order_type='BUY', quantity=quantity, - price=info['entry_price'] + price=entry_price ) + logger.info(f"Order result: {result}") print('주문 결과:', result) - + # API 응답 구조에 따라 rt_cd가 '0'이면 성공으로 간주 if result and result.get('rt_cd') == '0': + logger.info(f"✅✅✅ [{stock_code}] Order placed successfully") print("✅ 매수 주문 성공. 거래 정보를 저장합니다.") + trade_details = { - 'stock_code': info['stock_code'], + 'stock_code': stock_code, + 'stock_name': stock_name, 'quantity': quantity, - 'entry_price': info['entry_price'], + 'entry_price': entry_price, 'stop_loss': info['stop_loss'], 'take_profit': info['take_profit'], 'order_id': result.get('ODNO'), # 주문번호 'order_time': result.get('ORD_TMD'), # 주문시각 - 'timestamp': datetime.now(timezone.utc).isoformat() + 'timestamp': datetime.now(timezone.utc).isoformat(), + 'validation_details': validation_details # Store validation context } save_active_trade(trade_details) else: error_message = result.get('msg1', '알 수 없는 오류') + logger.error(f"❌ Order placement failed: {error_message}") print(f"❌ 매수 주문 실패: {error_message}") - - else: - print('Claude 응답에서 종목/가격을 찾을 수 없습니다.') + + except Exception as e: + logger.error(f"❌ Exception during order placement: {e}", exc_info=True) + print(f"❌ 주문 중 오류 발생: {e}") if __name__ == "__main__": # 예시: Claude 응답 텍스트를 받아 자동매매 실행 diff --git a/mcp-server/risk_validator.py b/mcp-server/risk_validator.py new file mode 100644 index 0000000..f65d1cf --- /dev/null +++ b/mcp-server/risk_validator.py @@ -0,0 +1,431 @@ +""" +🔒 Risk Validation Module for Auto Trading (MCP Server) + +This module validates all trades BEFORE execution against CLAUDE.md Section 5 risk rules: +- Position sizing: max 3-4% per stock, max 20-25 holdings +- Price filter: > 5,000 KRW (domestic) +- MDD state check: -10% pauses new entries, -20% forces reduction +- Leverage ETF limit: max 20% of capital + +CLAUDE.md Compliance: +- All DB reads use AsyncSessionLocal from backend +- No breaking changes to existing function signatures +- All rejections are logged with clear reasons +""" + +import asyncio +import logging +import sys +import os +from typing import Optional, Dict, Tuple +from datetime import datetime, timezone +from decimal import Decimal + +# Backend imports +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../backend'))) + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession +from sqlalchemy.orm import sessionmaker + +from app.core.config import settings +from app.models.models import Portfolio, ActiveTrade, TradeHistory +from app.services.korea_investment_api import korea_investment_api + +logger = logging.getLogger(__name__) + +# Backend DB connection +engine = create_async_engine(settings.DATABASE_URL, echo=False) +AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + +# Risk limits from CLAUDE.md Section 5 +MAX_POSITION_SIZE_PCT = 0.04 # 3-4% per stock (using 4% as upper bound) +MAX_HOLDINGS_COUNT = 25 # max 20-25 holdings +MIN_DOMESTIC_PRICE = 5000 # KRW +MAX_LEVERAGE_ETF_PCT = 0.20 # max 20% of capital for leverage/inverse ETF + +# Leverage/Inverse ETF identifiers +LEVERAGE_ETF_KEYWORDS = ['레버리지', 'LEVERAGE', '2X', '3X', 'INVERSE', '인버스', '-1X', '-2X', '-3X'] + + +class RiskValidationResult: + """Result of a risk validation check""" + def __init__(self, allowed: bool, reason: str = "", details: Optional[Dict] = None): + self.allowed = allowed + self.reason = reason + self.details = details or {} + self.timestamp = datetime.now(timezone.utc).isoformat() + + def __repr__(self): + return f"RiskValidationResult(allowed={self.allowed}, reason='{self.reason}')" + + +class RiskValidator: + """Risk validation engine for Claude-based auto trading""" + + def __init__(self): + self.db_session = None + self._lock = asyncio.Lock() + + async def _get_session(self) -> AsyncSession: + """Get or create DB session""" + if self.db_session is None: + self.db_session = AsyncSessionLocal() + return self.db_session + + async def _close_session(self): + """Close DB session""" + if self.db_session is not None: + await self.db_session.close() + self.db_session = None + + async def validate_price_filter(self, stock_code: str, market: str = "KRX") -> RiskValidationResult: + """ + Check price filter: min 5,000 KRW for domestic stocks + + Args: + stock_code: Stock code to validate + market: Market code (default: KRX for domestic) + + Returns: + RiskValidationResult + """ + try: + # Skip price check for overseas/ETF markets + if market and market not in ["KRX", "KOSPI", "KOSDAQ"]: + logger.info(f"[{stock_code}] Price filter skipped (overseas market: {market})") + return RiskValidationResult(allowed=True) + + # Get current price from KIS API + price_info, _ = await korea_investment_api._make_request( + "GET", + "/uapi/domestic-stock/v1/quotations/inquire-price", + "FHKST01010100", + params={"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code} + ) + + if not price_info or price_info.get('rt_cd') != '0': + return RiskValidationResult( + allowed=False, + reason=f"Failed to fetch current price for {stock_code}" + ) + + current_price = int(price_info.get('output', {}).get('stck_prpr', 0)) + + if current_price < MIN_DOMESTIC_PRICE: + return RiskValidationResult( + allowed=False, + reason=f"Price filter violation: {current_price} KRW < {MIN_DOMESTIC_PRICE} KRW minimum", + details={"current_price": current_price, "min_price": MIN_DOMESTIC_PRICE} + ) + + logger.info(f"[{stock_code}] Price filter passed: {current_price} KRW") + return RiskValidationResult( + allowed=True, + details={"current_price": current_price} + ) + + except Exception as e: + logger.error(f"Price filter check error for {stock_code}: {e}") + return RiskValidationResult(allowed=False, reason=f"Price check error: {str(e)}") + + async def validate_position_size(self, stock_code: str, order_quantity: int, entry_price: float, + account_balance: float) -> RiskValidationResult: + """ + Check position sizing: max 3-4% per stock + + Args: + stock_code: Stock code to validate + order_quantity: Number of shares to buy + entry_price: Entry price per share + account_balance: Current account balance + + Returns: + RiskValidationResult + """ + try: + order_value = order_quantity * entry_price + position_pct = (order_value / account_balance) * 100 if account_balance > 0 else 0 + + if position_pct > MAX_POSITION_SIZE_PCT * 100: + return RiskValidationResult( + allowed=False, + reason=f"Position size violation: {position_pct:.2f}% > {MAX_POSITION_SIZE_PCT*100}% maximum", + details={ + "order_value": order_value, + "position_pct": position_pct, + "max_allowed_pct": MAX_POSITION_SIZE_PCT * 100 + } + ) + + logger.info(f"[{stock_code}] Position size check passed: {position_pct:.2f}% of account") + return RiskValidationResult( + allowed=True, + details={"position_pct": position_pct, "order_value": order_value} + ) + + except Exception as e: + logger.error(f"Position size check error for {stock_code}: {e}") + return RiskValidationResult(allowed=False, reason=f"Position size check error: {str(e)}") + + async def validate_max_holdings(self, account_balance: float) -> RiskValidationResult: + """ + Check max holdings limit: 20-25 concurrent positions + + Args: + account_balance: Current account balance (for context) + + Returns: + RiskValidationResult + """ + try: + db = await self._get_session() + + # Count active holdings (quantity > 0) + result = await db.execute( + select(Portfolio).where(Portfolio.quantity > 0) + ) + active_holdings = result.scalars().all() + holdings_count = len(active_holdings) + + if holdings_count >= MAX_HOLDINGS_COUNT: + return RiskValidationResult( + allowed=False, + reason=f"Max holdings limit reached: {holdings_count} >= {MAX_HOLDINGS_COUNT}", + details={"current_holdings": holdings_count, "max_allowed": MAX_HOLDINGS_COUNT} + ) + + logger.info(f"Max holdings check passed: {holdings_count}/{MAX_HOLDINGS_COUNT}") + return RiskValidationResult( + allowed=True, + details={"current_holdings": holdings_count, "available_slots": MAX_HOLDINGS_COUNT - holdings_count} + ) + + except Exception as e: + logger.error(f"Max holdings check error: {e}") + return RiskValidationResult(allowed=False, reason=f"Max holdings check error: {str(e)}") + + async def validate_mdd_state(self) -> RiskValidationResult: + """ + Check MDD (Maximum Drawdown) state: + - MDD -10%: pause new entries + - MDD -20%: force reduce positions + + Returns: + RiskValidationResult + """ + try: + # Import MDD manager + from app.services.mdd_manager import mdd_manager + + # Check if new positions are paused due to MDD + can_open = await mdd_manager.can_open_new_position() + + if not can_open: + state = { + "is_paused": mdd_manager.is_paused, + "reduction_applied": mdd_manager.reduction_applied, + "peak_balance": mdd_manager.peak_balance + } + return RiskValidationResult( + allowed=False, + reason=f"MDD limit triggered (reduction: {mdd_manager.reduction_applied}). New entries paused.", + details=state + ) + + logger.info(f"MDD check passed: trading active (peak: {mdd_manager.peak_balance})") + return RiskValidationResult( + allowed=True, + details={ + "is_paused": mdd_manager.is_paused, + "reduction_applied": mdd_manager.reduction_applied, + "peak_balance": mdd_manager.peak_balance + } + ) + + except Exception as e: + logger.error(f"MDD check error: {e}") + # Don't block trading if MDD check fails - log and allow + logger.warning(f"MDD validation skipped due to error: {str(e)}") + return RiskValidationResult(allowed=True, reason="MDD check skipped (error)") + + async def validate_leverage_etf_limit(self, stock_code: str, stock_name: str, + order_quantity: int, entry_price: float, + account_balance: float) -> RiskValidationResult: + """ + Check leverage/inverse ETF limit: max 20% of capital + + Args: + stock_code: Stock code + stock_name: Stock name (for keyword matching) + order_quantity: Number of shares to buy + entry_price: Entry price per share + account_balance: Current account balance + + Returns: + RiskValidationResult + """ + try: + # Check if this is a leverage/inverse ETF + is_leverage_etf = any(keyword in stock_name.upper() for keyword in LEVERAGE_ETF_KEYWORDS) + + if not is_leverage_etf: + logger.info(f"[{stock_code}] Not a leverage ETF, skipping leverage limit check") + return RiskValidationResult(allowed=True) + + # Calculate total leverage ETF position + db = await self._get_session() + result = await db.execute(select(Portfolio)) + all_holdings = result.scalars().all() + + leverage_etf_value = 0 + for holding in all_holdings: + if any(keyword in (holding.stock_name or "").upper() for keyword in LEVERAGE_ETF_KEYWORDS): + holding_value = float(holding.current_price or 0) * holding.quantity + leverage_etf_value += holding_value + + # Add proposed order + proposed_value = order_quantity * entry_price + total_leverage_value = leverage_etf_value + proposed_value + leverage_pct = (total_leverage_value / account_balance) * 100 if account_balance > 0 else 0 + + if leverage_pct > MAX_LEVERAGE_ETF_PCT * 100: + return RiskValidationResult( + allowed=False, + reason=f"Leverage ETF limit: {leverage_pct:.2f}% > {MAX_LEVERAGE_ETF_PCT*100}% maximum", + details={ + "current_leverage_value": leverage_etf_value, + "proposed_value": proposed_value, + "total_leverage_pct": leverage_pct, + "max_allowed_pct": MAX_LEVERAGE_ETF_PCT * 100 + } + ) + + logger.info(f"[{stock_code}] Leverage ETF limit check passed: {leverage_pct:.2f}% of account") + return RiskValidationResult( + allowed=True, + details={"leverage_etf_pct": leverage_pct, "total_leverage_value": total_leverage_value} + ) + + except Exception as e: + logger.error(f"Leverage ETF limit check error for {stock_code}: {e}") + return RiskValidationResult(allowed=False, reason=f"Leverage ETF check error: {str(e)}") + + async def validate_order_comprehensive(self, + stock_code: str, + stock_name: str, + order_quantity: int, + entry_price: float, + account_balance: float, + market: str = "KRX") -> Tuple[bool, Dict]: + """ + Comprehensive order validation combining all risk checks + + Args: + stock_code: Stock code + stock_name: Stock name + order_quantity: Number of shares + entry_price: Entry price per share + account_balance: Current account balance + market: Market code (default: KRX) + + Returns: + Tuple of (allowed: bool, details: Dict with all checks) + """ + try: + async with self._lock: + validation_results = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "stock_code": stock_code, + "stock_name": stock_name, + "order_quantity": order_quantity, + "entry_price": entry_price, + "account_balance": account_balance, + "checks": {} + } + + # Run all checks + checks = [ + ("price_filter", await self.validate_price_filter(stock_code, market)), + ("position_size", await self.validate_position_size(stock_code, order_quantity, entry_price, account_balance)), + ("max_holdings", await self.validate_max_holdings(account_balance)), + ("mdd_state", await self.validate_mdd_state()), + ("leverage_etf_limit", await self.validate_leverage_etf_limit(stock_code, stock_name, order_quantity, entry_price, account_balance)) + ] + + # Record all check results + all_passed = True + for check_name, result in checks: + validation_results["checks"][check_name] = { + "allowed": result.allowed, + "reason": result.reason, + "details": result.details + } + if not result.allowed: + all_passed = False + logger.warning(f"❌ [{stock_code}] {check_name}: {result.reason}") + else: + logger.info(f"✅ [{stock_code}] {check_name}: passed") + + validation_results["overall_allowed"] = all_passed + + if all_passed: + logger.info(f"✅✅✅ [{stock_code}] ALL RISK CHECKS PASSED - Order approved") + else: + logger.warning(f"❌ [{stock_code}] RISK VALIDATION FAILED - Order rejected") + + return all_passed, validation_results + + except Exception as e: + logger.error(f"Comprehensive validation error for {stock_code}: {e}", exc_info=True) + return False, { + "timestamp": datetime.now(timezone.utc).isoformat(), + "stock_code": stock_code, + "overall_allowed": False, + "error": str(e) + } + + +# Singleton instance +risk_validator = RiskValidator() + + +async def validate_and_log_order(stock_code: str, + stock_name: str, + order_quantity: int, + entry_price: float, + account_balance: float, + market: str = "KRX") -> Tuple[bool, Dict]: + """ + Public API for order validation with comprehensive logging + + Args: + stock_code: Stock code + stock_name: Stock name + order_quantity: Number of shares + entry_price: Entry price per share + account_balance: Current account balance + market: Market code + + Returns: + Tuple of (allowed: bool, validation_details: Dict) + """ + allowed, details = await risk_validator.validate_order_comprehensive( + stock_code=stock_code, + stock_name=stock_name, + order_quantity=order_quantity, + entry_price=entry_price, + account_balance=account_balance, + market=market + ) + + # Log rejection with full details + if not allowed: + failed_checks = [ + f"{check}: {data['reason']}" + for check, data in details.get('checks', {}).items() + if not data.get('allowed', True) + ] + logger.error(f"🚫 ORDER REJECTED [{stock_code}]: {'; '.join(failed_checks)}") + + return allowed, details diff --git a/mcp-server/trade_monitor.py b/mcp-server/trade_monitor.py index ced7330..9057795 100644 --- a/mcp-server/trade_monitor.py +++ b/mcp-server/trade_monitor.py @@ -4,6 +4,7 @@ import os from datetime import datetime import logging from decimal import Decimal +import aiohttp # backend 경로를 sys.path에 추가하여 서비스 모듈 임포트 import sys @@ -27,6 +28,155 @@ AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=F ACTIVE_TRADES_FILE = os.path.join(os.path.dirname(__file__), 'active_trades.json') MONITORING_INTERVAL_SECONDS = 60 # 60초마다 확인 +BACKEND_API_BASE = os.getenv('BACKEND_API_BASE', 'http://localhost:8000') + +async def update_active_trade_state_via_api(stock_code: str, new_status: str, reason: str = "") -> bool: + """ + CLAUDE.md Section 3.2: 직접 DB 접근 금지. 백엔드 API를 통해 상태 업데이트. + + 활성 거래의 상태를 API를 통해 업데이트한다. + 이는 db_worker를 우회하지 않으면서도 여러 시스템이 동시에 접근할 수 있게 한다. + + Args: + stock_code: 종목 코드 + new_status: 목표 상태 ('active', 'pending_sell', 'closed') + reason: 상태 변경 이유 + + Returns: + 성공 여부 + """ + try: + async with aiohttp.ClientSession() as session: + url = f"{BACKEND_API_BASE}/api/portfolio/active-trades/{stock_code}/status" + payload = { + "status": new_status, + "reason": reason, + "updated_at": datetime.utcnow().isoformat() + } + + async with session.patch(url, json=payload, timeout=aiohttp.ClientTimeout(total=5)) as resp: + if resp.status == 200: + logging.info(f"✅ [{stock_code}] API를 통해 상태 변경: {new_status}") + return True + else: + error_text = await resp.text() + logging.error(f"❌ [{stock_code}] 상태 변경 API 실패 (HTTP {resp.status}): {error_text}") + return False + except Exception as e: + logging.error(f"❌ [{stock_code}] 상태 변경 API 호출 중 예외: {e}") + return False + + +async def calculate_atr_and_highest(db: AsyncSession, stock_code: str) -> tuple[float, float]: + """ + 최근 14개 캔들에서 ATR과 최고가를 계산한다. + + Args: + db: 비동기 DB 세션 + stock_code: 종목 코드 + + Returns: + (atr, highest_since_entry) 튜플 + """ + try: + # StockDailyPrice에서 최근 14개 캔들 조회 + from app.models.models import StockDailyPrice + from sqlalchemy import desc + + result = await db.execute( + select(StockDailyPrice) + .where(StockDailyPrice.stock_code == stock_code) + .order_by(desc(StockDailyPrice.trade_date)) + .limit(14) + ) + prices = list(reversed(result.scalars().all())) + + if len(prices) < 2: + logging.warning(f"[{stock_code}] ATR 계산을 위한 충분한 가격 데이터 없음") + return 0.0, 0.0 + + # ATR 계산 (True Range 평균) + tr_values = [] + for i in range(1, len(prices)): + high = prices[i].high_price + low = prices[i].low_price + prev_close = prices[i - 1].close_price + + tr = max( + high - low, + abs(high - prev_close), + abs(low - prev_close) + ) + tr_values.append(tr) + + atr = sum(tr_values) / len(tr_values) if tr_values else 0.0 + + # 최고가 계산 (진입 이후) + highest = max(p.high_price for p in prices) if prices else 0.0 + + return float(atr), float(highest) + + except Exception as e: + logging.error(f"[{stock_code}] ATR/최고가 계산 중 예외: {e}") + return 0.0, 0.0 + + +async def check_partial_takeprofit_and_trailing_stop( + stock_code: str, + current_price: float, + entry_price: float, + remaining_pct: float, + highest_since_entry: float, + atr: float +) -> tuple[bool, float, str]: + """ + CLAUDE.md Section 5.2: 부분 익절 및 트레일링 스탑 로직 + + - +2R 수익 (리스크 대비 2배): 50% 분할 매도 + - 나머지 50%: high - (1.5 * ATR) 트레일링 스탑 + + Args: + stock_code: 종목 코드 + current_price: 현재가 + entry_price: 진입가 + remaining_pct: 남은 수량 비율 (기본 1.0) + highest_since_entry: 진입 이후 최고가 + atr: ATR 값 + + Returns: + (should_sell, sell_price, reason) 튜플 + """ + if remaining_pct <= 0: + return False, 0.0, "" + + profit = current_price - entry_price + + # ───────────────────────────────────────────────────────────────────────── + # +2R 익절 조건 (처음 50% 분할 매도) + # ───────────────────────────────────────────────────────────────────────── + # 평균 손절까지의 거리를 리스크로 정의 (기본값: 진입가의 6%) + risk = entry_price * 0.06 # 기본 손절: -6% + + if profit >= 2 * risk and remaining_pct == 1.0: + # 첫 50%만 매도 + logging.info(f"[{stock_code}] +2R 익절 조건 충족 (수익: {profit}, 리스크: {risk})") + return True, current_price, f"+2R 익절 (수익: {profit:.0f}원)" + + # ───────────────────────────────────────────────────────────────────────── + # 트레일링 스탑 (나머지 50% 물량) + # ───────────────────────────────────────────────────────────────────────── + if remaining_pct < 1.0 and atr > 0: # 부분 매도 후 남은 물량 + trailing_stop_price = highest_since_entry - (1.5 * atr) + + if current_price <= trailing_stop_price: + logging.info( + f"[{stock_code}] 트레일링 스탑 도달 " + f"(현재: {current_price}, 스탑: {trailing_stop_price})" + ) + return True, current_price, f"트레일링 스탑 (스탑가: {trailing_stop_price:.0f}원)" + + return False, 0.0, "" + async def sync_json_from_db(db: AsyncSession): """DB의 active_trades 목록을 JSON 파일에 동기화(덮어쓰기)합니다.""" @@ -60,12 +210,20 @@ async def sync_json_from_db(db: AsyncSession): async def check_and_execute_trades(db: AsyncSession): """ DB에서 활성화된 거래를 확인하고 손절/익절 조건에 따라 매도 주문을 실행합니다. + + CLAUDE.md Section 3.2: 직접 DB commit 금지 + - 상태 변경은 API를 통해 처리 (db_worker 우회 방지) + - 모든 state transition을 logging + + CLAUDE.md Section 5.2: 부분 익절 및 트레일링 스탑 + - +2R: 50% 분할 매도 (pending_sell 상태) + - 나머지 50%: trailing stop (high - 1.5*ATR) """ - logging.info(" 거래 모니터링 시작 (DB 기반)...") - + logging.info("거래 모니터링 시작 (DB 기반)...") + result = await db.execute(select(ActiveTrade).where(ActiveTrade.status == 'active')) active_trades = result.scalars().all() - + if not active_trades: logging.info("현재 활성화된 거래가 없습니다.") return @@ -76,9 +234,12 @@ async def check_and_execute_trades(db: AsyncSession): take_profit = float(trade.take_profit) quantity = trade.quantity entry_price = float(trade.entry_price) - + remaining_pct = float(trade.remaining_pct) if trade.remaining_pct else 1.0 + highest_since_entry = float(trade.highest_since_entry) if trade.highest_since_entry else entry_price + should_sell = False sell_reason = "" + sell_quantity = 0 try: # 현재가 조회 @@ -88,7 +249,7 @@ async def check_and_execute_trades(db: AsyncSession): "FHKST01010100", params={"FID_COND_MRKT_DIV_CODE": "J", "FID_INPUT_ISCD": stock_code} ) - + if price_info and price_info.get('rt_cd') == '0': current_price_str = price_info.get('output', {}).get('stck_prpr') if not current_price_str: @@ -96,18 +257,52 @@ async def check_and_execute_trades(db: AsyncSession): continue current_price = int(current_price_str) - logging.info(f"[{stock_code}] 현재가: {current_price} (매수가: {entry_price}, 손절가: {stop_loss}, 목표가: {take_profit})") + logging.info( + f"[{stock_code}] 현재가: {current_price} (진입: {entry_price}, " + f"손절: {stop_loss}, 목표: {take_profit}, 남은비율: {remaining_pct*100:.1f}%)" + ) - # 손절 조건 확인 + # ───────────────────────────────────────────────────────────── + # 1. 손절 조건 확인 + # ───────────────────────────────────────────────────────────── if stop_loss and current_price <= stop_loss: should_sell = True + sell_quantity = int(quantity * remaining_pct) sell_reason = f"손절가 도달 (현재가: {current_price} <= 손절가: {stop_loss})" - - # 익절 조건 확인 + + # ───────────────────────────────────────────────────────────── + # 2. 기본 익절 조건 확인 + # ───────────────────────────────────────────────────────────── elif take_profit and current_price >= take_profit: should_sell = True + sell_quantity = int(quantity * remaining_pct) sell_reason = f"목표가 도달 (현재가: {current_price} >= 목표가: {take_profit})" + # ───────────────────────────────────────────────────────────── + # 3. 부분 익절 + 트레일링 스탑 조건 (CLAUDE.md 5.2) + # ───────────────────────────────────────────────────────────── + else: + atr, new_highest = await calculate_atr_and_highest(db, stock_code) + + # highest_since_entry 업데이트 + if new_highest > highest_since_entry: + highest_since_entry = new_highest + + should_tp, tp_price, tp_reason = await check_partial_takeprofit_and_trailing_stop( + stock_code, current_price, entry_price, remaining_pct, + highest_since_entry, atr + ) + + if should_tp: + should_sell = True + if remaining_pct == 1.0: + # 첫 익절: 50% 분할 매도 + sell_quantity = int(quantity * 0.5) + else: + # 트레일링 스탑: 남은 물량 전부 + sell_quantity = int(quantity * remaining_pct) + sell_reason = tp_reason + else: logging.error(f"[{stock_code}] 현재가 조회 실패: {price_info.get('msg1', '알 수 없는 오류')}") continue @@ -116,30 +311,115 @@ async def check_and_execute_trades(db: AsyncSession): logging.error(f"[{stock_code}] 처리 중 예외 발생: {e}") continue - if should_sell: - logging.warning(f"🚨 [{stock_code}] 매도 조건 충족: {sell_reason}") - try: - # 시장가 매도 주문 (price=None) - sell_result = await korea_investment_api.place_order( - stock_code=stock_code, - order_type='SELL', - quantity=quantity, - price=None - ) - logging.info(f"[{stock_code}] 매도 주문 결과: {sell_result}") + if should_sell and sell_quantity > 0: + await _execute_sell_with_pending_state( + stock_code, quantity, sell_quantity, entry_price, sell_reason + ) - if sell_result and sell_result.get('rt_cd') == '0': - logging.info(f"✅ [{stock_code}] 매도 주문 성공. DB 상태를 'closed'로 변경합니다.") - trade.status = 'closed' - trade.updated_at = datetime.utcnow() - db.add(trade) - await db.commit() - else: - logging.error(f"❌ [{stock_code}] 매도 주문 실패: {sell_result.get('msg1', '알 수 없는 오류')}") - except Exception as e: - logging.error(f"[{stock_code}] 매도 주문 중 예외 발생: {e}") - await db.rollback() +async def _execute_sell_with_pending_state( + stock_code: str, + total_quantity: int, + sell_quantity: int, + entry_price: float, + sell_reason: str +) -> None: + """ + CLAUDE.md Section 5.2: 매도 주문 실행 및 상태 전이 + + 단계: + 1. 상태를 'pending_sell'로 변경 (즉시) + 2. 매도 주문 실행 + 3. 주문 성공 시 'closed'로 변경 (1-5초 대기) + 4. 실패 시 'active'로 롤백 + + Args: + stock_code: 종목 코드 + total_quantity: 보유 수량 + sell_quantity: 매도할 수량 + entry_price: 진입가 + sell_reason: 매도 이유 + """ + try: + logging.warning( + f"🚨 [{stock_code}] 매도 조건 충족: {sell_reason} " + f"(수량: {sell_quantity}/{total_quantity})" + ) + + # ───────────────────────────────────────────────────────────── + # Step 1: 상태를 'pending_sell'로 즉시 변경 + # ───────────────────────────────────────────────────────────── + success = await update_active_trade_state_via_api( + stock_code, + "pending_sell", + f"매도 주문 대기 중: {sell_reason}" + ) + + if not success: + logging.error(f"❌ [{stock_code}] pending_sell 상태 전이 실패") + return + + logging.info(f"⏳ [{stock_code}] 상태: active → pending_sell (매도 주문 대기)") + + # ───────────────────────────────────────────────────────────── + # Step 2: 시장가 매도 주문 실행 + # ───────────────────────────────────────────────────────────── + sell_result = await korea_investment_api.place_order( + stock_code=stock_code, + order_type='SELL', + quantity=sell_quantity, + price=None # 시장가 + ) + + logging.info(f"[{stock_code}] 매도 주문 결과: {sell_result}") + + if sell_result and sell_result.get('rt_cd') == '0': + order_id = sell_result.get('output', {}).get('order_id', '') + logging.info(f"✅ [{stock_code}] 매도 주문 성공 (주문ID: {order_id})") + + # ───────────────────────────────────────────────────────── + # Step 3: 1-5초 대기 후 체결 확인 및 상태 'closed'로 변경 + # ───────────────────────────────────────────────────────── + await asyncio.sleep(2) # 체결 대기 + + success = await update_active_trade_state_via_api( + stock_code, + "closed", + f"매도 완료: {sell_reason} (주문ID: {order_id})" + ) + + if success: + logging.info(f"✅ [{stock_code}] 상태: pending_sell → closed") + else: + logging.warning(f"⚠️ [{stock_code}] closed 상태 전이 실패 (order_id: {order_id})") + else: + error_msg = sell_result.get('msg1', '알 수 없는 오류') if sell_result else "응답 없음" + logging.error(f"❌ [{stock_code}] 매도 주문 실패: {error_msg}") + + # ───────────────────────────────────────────────────────── + # Step 4: 주문 실패 시 'active'로 롤백 + # ───────────────────────────────────────────────────────── + success = await update_active_trade_state_via_api( + stock_code, + "active", + f"매도 주문 실패로 복구: {error_msg}" + ) + + if success: + logging.info(f"🔄 [{stock_code}] 상태 롤백: pending_sell → active") + + except Exception as e: + logging.error(f"[{stock_code}] 매도 실행 중 예외 발생: {e}", exc_info=True) + + # 예외 발생 시에도 'active'로 복구 + try: + await update_active_trade_state_via_api( + stock_code, + "active", + f"예외로 인한 롤백: {str(e)}" + ) + except Exception as rollback_exc: + logging.error(f"[{stock_code}] 롤백 중 예외: {rollback_exc}") async def main(): """메인 실행 루프"""