From 30d4db1b289a8c493b6d2ca63be20644e891a04a Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Wed, 25 Feb 2026 11:22:23 +0900 Subject: [PATCH] refactor: create trade_executor.py --- backend/app/services/trade_executor.py | 500 +++++++++++++++++++++++++ 1 file changed, 500 insertions(+) create mode 100644 backend/app/services/trade_executor.py diff --git a/backend/app/services/trade_executor.py b/backend/app/services/trade_executor.py new file mode 100644 index 0000000..0f4e8e1 --- /dev/null +++ b/backend/app/services/trade_executor.py @@ -0,0 +1,500 @@ +import asyncio +import logging +import httpx +from decimal import Decimal +from datetime import timedelta +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select, desc + +from app.core.config import settings +from app.core.database import AsyncSessionLocal +from app.models.models import TradeHistory, Portfolio +from app.services.korea_investment_api import korea_investment_api +from app.services.mdd_manager import mdd_manager +from app.services.loss_tracker import loss_tracker +from app.services.market_analyzer import check_market_regime, _get_kst_now +from app.services.mean_reversion_strategy import MarketRegime, calculate_position_size_v2 +from app.services.technical_indicators import ( + TREND_FILTER_ENABLED, TREND_MA_SHORT, TREND_MA_LONG, TREND_TOLERANCE_PCT, _ma +) + +logger = logging.getLogger(__name__) + + +async def _mcp_validate_buy(stock: dict) -> tuple: + if not settings.MCP_VALIDATION_ENABLED: + return True, "mcp_disabled" + + try: + payload = { + "analysis_type": "RULE_VALIDATION", + "data": { + "rule": { + "id": stock.get("rule_id"), + "name": stock.get("rule_name", "auto_trade"), + }, + "market_data": { + "stock_code": stock.get("stock_code"), + "stock_name": stock.get("stock_name"), + "current_price": stock.get("current_price"), + "signals": stock.get("signals", []), + "score": stock.get("score"), + "ml_signal": stock.get("ml_signal"), + "ml_confidence": stock.get("ml_confidence"), + "ma20": stock.get("ma20"), + "ma60": stock.get("ma60"), + } + } + } + + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.post( + f"{settings.MCP_SERVER_URL}/analyze", + json=payload + ) + resp.raise_for_status() + data = resp.json() + + result = data.get("result", {}) + decision = result.get("decision", "HOLD") + confidence = float(result.get("confidence", 0.0)) + + if decision == "BUY" and confidence >= settings.MCP_CONFIDENCE_THRESHOLD: + return True, f"mcp_approved(conf={confidence:.2f})" + else: + return False, f"mcp_rejected(decision={decision}, conf={confidence:.2f})" + + except Exception as e: + logger.warning(f"⚠️ MCP 검증 실패 — 매매 차단: {e}") + return False, "mcp_unavailable" + + +async def execute_auto_trade_background(recommendations_response: dict, db: AsyncSession): + try: + trade_results = [] + total_processed = 0 + + market_regime = None + if settings.BEAR_MARKET_FILTER_ENABLED: + market_regime = await check_market_regime() + + regime_max_trades = market_regime.get("max_trades", 20) if market_regime else 20 + position_multiplier = market_regime.get("position_multiplier", 1.0) if market_regime else 1.0 + max_trades = min(20, regime_max_trades) + + result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) + held_stocks = {p.stock_code for p in result.scalars().all()} + + unique_stocks = {} + for rule_group in recommendations_response.get("recommendations_by_rule", []): + for stock in rule_group["stocks"]: + stock_code = stock.get("stock_code") + if stock_code not in unique_stocks: + unique_stocks[stock_code] = stock + elif stock.get("rule_priority", 999) < unique_stocks[stock_code].get("rule_priority", 999): + unique_stocks[stock_code] = stock + + sorted_stocks = sorted( + unique_stocks.values(), + key=lambda s: (s.get("rule_priority", 999), -s.get("score", 0)) + ) + + for stock in sorted_stocks[:max_trades]: + try: + stock["_position_multiplier"] = position_multiplier + stock["_regime_str"] = market_regime.get("regime_200d", "bear") if market_regime else "bear" + result = await execute_auto_trade(stock, db) + total_processed += 1 + + if result["action"] in ["BUY", "SELL"]: + trade_results.append(result) + + await asyncio.sleep(0.5) + + except Exception as e: + logger.error(f"Error executing auto-trade for {stock.get('stock_code')}: {e}") + continue + + except Exception as e: + logger.error(f"❌ Background auto-trade error: {e}") + + +async def _background_sync_portfolio(): + try: + from app.api.endpoints.portfolio import sync_portfolio + async with AsyncSessionLocal() as session: + await sync_portfolio(session, force=True) + except Exception as e: + logger.error(f"Background portfolio sync failed: {e}") + + +async def _delayed_sync_portfolio(delay_seconds: int = 3): + try: + await asyncio.sleep(delay_seconds) + from app.api.endpoints.portfolio import sync_portfolio + + async with AsyncSessionLocal() as session: + try: + await sync_portfolio(session, force=True) + except Exception as inner_e: + logger.error(f"Sync execution failed: {inner_e}") + except Exception as e: + logger.error(f"Delayed portfolio sync failed: {e}") + + +async def execute_auto_trade(stock: dict, db: AsyncSession): + try: + stock_code = stock.get("stock_code") + stock_name = stock.get("stock_name") + current_price = float(stock.get("current_price", 0)) + signals = stock.get("signals", []) + + if not stock_code or current_price <= 0: + return {"action": "SKIP", "reason": "invalid_data"} + + is_domestic = stock_code.isdigit() and len(stock_code) == 6 + exchange_code = stock.get("exchange_code", "NASD") + + if is_domestic: + if current_price < settings.MIN_PRICE_KRW: + return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} + else: + if current_price < settings.MIN_PRICE_USD: + return {"action": "SKIP", "reason": "penny_stock", "stock": stock_name} + + if not await mdd_manager.can_open_new_position(): + return {"action": "SKIP", "reason": "mdd_pause", "stock": stock_name} + + if loss_tracker.is_banned(stock_code): + return {"action": "SKIP", "reason": "consecutive_loss_ban", "stock": stock_name} + + held_result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0)) + held_count = len(held_result.scalars().all()) + if held_count >= settings.MAX_HOLDINGS_COUNT: + return {"action": "SKIP", "reason": "max_holdings_reached", "stock": stock_name} + + cutoff_time = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) + recent_sell = await db.execute( + select(TradeHistory).where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "SELL", + TradeHistory.status == "EXECUTED", + TradeHistory.executed_at >= cutoff_time + ) + ) + if recent_sell.first(): + return {"action": "SKIP", "reason": "cooldown_sell_today", "stock": stock_name} + + result = await db.execute( + select(Portfolio).where( + Portfolio.stock_code == stock_code, + Portfolio.quantity > 0 + ) + ) + holding = result.scalar_one_or_none() + + text_signals = " ".join(signals) + try: + score_val = float(stock.get("score", 0) or 0) + except Exception: + score_val = 0.0 + + buy_keywords = [ + "정배열 눌림목 매수", "눌림목 진입 매수타점", "RSI 바닥 탈출 반등", + "RSI 안정 상승", "스토캐스틱 과매도", "5일선 > 20일선", "20일선 > 60일선", + "골든크로스", "볼린저 하한선 터치", "횡보장 — 평균회귀 유리", "거래량 증가와 상승" + ] + should_buy = any(k in text_signals for k in buy_keywords) + + sell_keywords = [ + "역배열 하락 매도타점", "RSI 과매수", "스토캐스틱 과매수", "데드크로스", + "이동평균 역배열", "볼린저 상한선 돌파", "손절라인 이탈" + ] + should_sell = any(k in text_signals for k in sell_keywords) + + ml_score_trade = stock.get("ml_score") + ml_confidence_trade = stock.get("ml_confidence") + ml_signal_trade = stock.get("ml_signal") + + if settings.ML_ENSEMBLE_ENABLED and ml_score_trade is not None: + if ml_signal_trade == "BUY" and ml_score_trade >= settings.ML_ENSEMBLE_BUY_THRESHOLD and (ml_confidence_trade or 0) >= settings.ML_ENSEMBLE_CONFIDENCE_MIN: + should_buy = True + elif ml_signal_trade == "SELL" and (ml_confidence_trade or 0) >= 0.5: + should_sell = True + elif ml_signal_trade == "HOLD" and (ml_confidence_trade or 0) < settings.ML_ENSEMBLE_CONFIDENCE_MIN: + should_buy = False + + has_rule = stock.get("rule_id") is not None + if has_rule: + if "이격도 과열" in text_signals or "매수금지" in text_signals: + return {"action": "HOLD", "stock": stock_name, "reason": "disparity_overheated"} + if "급등 종목 추격 금지" in text_signals or "급등과열" in text_signals: + return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} + + ml_hold_active = settings.ML_ENSEMBLE_ENABLED and ml_signal_trade == "HOLD" and not should_buy + if not ml_hold_active: + should_buy = True + + if should_buy and not has_rule: + strong_buy_kw = ["RSI 바닥 탈출 반등", "볼린저 하한선 터치", "눌림목 진입 매수타점", "정배열 눌림목 매수"] + has_strong_signal = any(k in text_signals for k in strong_buy_kw) + matched_count = sum(1 for k in buy_keywords if k in text_signals) + if not has_strong_signal and (matched_count < 2 or score_val < 2): + should_buy = False + + if should_buy: + try: + if is_domestic: + chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D", limit=10) + chart_data = chart_resp.get("output2", []) + else: + chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") + chart_data = chart_resp.get("output2", []) + + if len(chart_data) >= 4: + if is_domestic: + closes = [float(item.get("stck_clpr", 0)) for item in chart_data[-4:]] + else: + closes = [float(item.get("clos", 0)) for item in chart_data[-4:]] + + if len(closes) >= 4: + price_3days_ago = closes[0] + recent_gain = ((current_price - price_3days_ago) / price_3days_ago) * 100 if price_3days_ago > 0 else 0 + if recent_gain > 10: + return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"} + except Exception as e: + pass + + if TREND_FILTER_ENABLED: + try: + ma20 = stock.get("ma20") + ma60 = stock.get("ma60") + if ma20 is None or ma60 is None: + if is_domestic: + chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D") + closes_local = [float(it.get("stck_clpr", 0)) for it in (chart_resp.get("output") or [])] + else: + chart_resp = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange_code, "D") + closes_local = [float(it.get("clos", 0)) for it in (chart_resp.get("output2") or [])] + + if len(closes_local) >= max(TREND_MA_SHORT, TREND_MA_LONG): + ma20 = _ma(closes_local, TREND_MA_SHORT) + ma60 = _ma(closes_local, TREND_MA_LONG) + if not (ma20 and ma60 and (float(ma20) >= float(ma60) * (1 - TREND_TOLERANCE_PCT)) and (current_price >= float(ma20) * (1 - TREND_TOLERANCE_PCT))): + return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter"} + except Exception as e: + return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"} + + try: + from app.api.endpoints.portfolio import portfolio_cache + available_cash = float(portfolio_cache.summary.get("available_cash", 0)) if portfolio_cache.summary else 0.0 + total_assets = float(portfolio_cache.summary.get("total_assets", 0)) if portfolio_cache.summary else 0.0 + except Exception: + available_cash, total_assets = 0.0, 0.0 + + daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0) + daily_buy_check = await db.execute( + select(TradeHistory) + .where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "BUY", + TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), + TradeHistory.created_at >= daily_cutoff, + ) + ) + daily_buys = daily_buy_check.scalars().all() + if len(daily_buys) >= 2: + return {"action": "HOLD", "stock": stock_name, "reason": "daily_trade_limit"} + + cooldown_min = int(stock.get("add_buy_cooldown_min", 30) or 30) + + cooldown_check = await db.execute( + select(TradeHistory) + .where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "BUY", + TradeHistory.status.in_(['EXECUTED', 'COMPLETED']), + TradeHistory.created_at >= daily_cutoff, + ) + .order_by(desc(TradeHistory.created_at)) + ) + recent_buys = cooldown_check.scalars().all() + if recent_buys and (_get_kst_now() - recent_buys[0].created_at).total_seconds() < cooldown_min * 60: + return {"action": "HOLD", "stock": stock_name, "reason": "cooldown"} + + recent_buy_cutoff = _get_kst_now() - timedelta(minutes=5) + check_recent = await db.execute( + select(TradeHistory).where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "BUY", + TradeHistory.created_at >= recent_buy_cutoff, + TradeHistory.status.in_(['EXECUTED', 'COMPLETED', 'PENDING']) + ) + ) + if check_recent.first(): + return {"action": "SKIP", "stock": stock_name, "reason": "recent_buy_5min"} + + mcp_ok, mcp_reason = await _mcp_validate_buy(stock) + if not mcp_ok: + return {"action": "HOLD", "stock": stock_name, "reason": mcp_reason} + + current_value = float(holding.quantity * current_price) if holding else 0.0 + regime_str = stock.get("_regime_str", "bear") + regime = MarketRegime.BULL if regime_str == "bull" else MarketRegime.BEAR + invest_amount = calculate_position_size_v2( + total_assets=total_assets, + available_cash=available_cash, + current_strategy_allocation=current_value, + regime=regime, + rsi=stock.get("rsi"), + adx=stock.get("adx_14"), + ) + if invest_amount <= 0 and total_assets <= 0: + invest_amount = available_cash * 0.04 + + quantity = int(invest_amount / current_price) + + if quantity > 0: + try: + if is_domestic: + order_result = await korea_investment_api.place_order(stock_code, "BUY", quantity, int(current_price)) + else: + order_result = await korea_investment_api.place_overseas_order(stock_code, "BUY", quantity, float(current_price), exchange_code) + + recent_cutoff = _get_kst_now() - timedelta(seconds=10) + duplicate_check = await db.execute( + select(TradeHistory).where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "BUY", + TradeHistory.created_at >= recent_cutoff + ) + ) + if duplicate_check.scalar_one_or_none(): + return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} + + total_amount = Decimal(str(current_price)) * quantity + entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점" + + success = order_result.get("rt_cd") == "0" + status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") + + trade = TradeHistory( + rule_id=stock.get("rule_id"), + stock_code=stock_code, + stock_name=stock_name, + order_type="BUY", + order_price=Decimal(str(current_price)), + order_quantity=quantity, + executed_price=Decimal(str(current_price)) if success else None, + executed_quantity=quantity if success else 0, + total_amount=total_amount, + status=status, + order_id=order_result.get("odno"), + message=f"자동매수: {'추가매수' if holding else '신규'} | {' | '.join(signals[:2])}", + entry_reason=entry_reason, + signals=signals, + created_at=_get_kst_now(), + executed_at=_get_kst_now() if success else None + ) + db.add(trade) + await db.commit() + + if success: + if holding: + new_total_qty = holding.quantity + quantity + prev_cost = float(holding.average_price or 0) * holding.quantity + holding.quantity = new_total_qty + holding.average_price = Decimal(str((prev_cost + current_price * quantity) / new_total_qty)) + holding.current_price = Decimal(str(current_price)) + else: + new_holding = Portfolio( + stock_code=stock_code, stock_name=stock_name, quantity=quantity, + average_price=Decimal(str(current_price)), current_price=Decimal(str(current_price)), + profit_loss=Decimal("0"), profit_loss_rate=Decimal("0") + ) + db.add(new_holding) + await db.commit() + + try: + from app.api.endpoints.portfolio import portfolio_cache + if portfolio_cache.summary: + old_cash = float(portfolio_cache.summary.get("available_cash", 0)) + portfolio_cache.summary["available_cash"] = max(0.0, old_cash - quantity * current_price) + portfolio_cache.change_flag = True + except Exception: + pass + return {"action": "BUY", "stock": stock_name, "quantity": quantity, "price": current_price, "success": success} + + except Exception as e: + return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)} + else: + return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name} + + elif should_sell and holding: + quantity = holding.quantity + _avg = float(holding.average_price or 0) + profit_loss_rate = ((current_price - _avg) / _avg) * 100 if _avg > 0 else float(holding.profit_loss_rate or 0) + + take_profit = stock.get("take_profit", 3.0) + stop_loss = stock.get("stop_loss", -1.5) + + should_execute_sell = profit_loss_rate >= take_profit or profit_loss_rate <= stop_loss or should_sell + + if should_execute_sell: + try: + if is_domestic: + order_result = await korea_investment_api.place_order(stock_code, "SELL", quantity, int(current_price)) + else: + order_result = await korea_investment_api.place_overseas_order(stock_code, "SELL", quantity, float(current_price), exchange_code) + + recent_cutoff = _get_kst_now() - timedelta(seconds=10) + duplicate_check = await db.execute( + select(TradeHistory).where( + TradeHistory.stock_code == stock_code, + TradeHistory.order_type == "SELL", + TradeHistory.created_at >= recent_cutoff + ) + ) + if duplicate_check.scalar_one_or_none(): + return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name} + + total_amount = Decimal(str(current_price)) * quantity + profit_loss = (current_price - float(holding.average_price or current_price)) * quantity + + success = order_result.get("rt_cd") == "0" + status = "COMPLETED" if success else ("PENDING" if order_result.get("rt_cd") == "1" else "FAILED") + + trade = TradeHistory( + rule_id=stock.get("rule_id"), stock_code=stock_code, stock_name=stock_name, + order_type="SELL", order_price=Decimal(str(current_price)), order_quantity=quantity, + executed_price=Decimal(str(current_price)) if success else None, + executed_quantity=quantity if success else 0, + total_amount=total_amount, status=status, order_id=order_result.get("odno"), + message=f"자동매도: 손익률 {profit_loss_rate:.2f}%", + exit_reason=f"손익: {profit_loss:,.0f}원", + signals=signals, net_profit_loss=Decimal(str(profit_loss)) if success else None, + profit_loss_rate=Decimal(str(profit_loss_rate)) if success else None, + created_at=_get_kst_now(), executed_at=_get_kst_now() if success else None + ) + db.add(trade) + await db.commit() + + if success: + await db.delete(holding) + await db.commit() + try: + from app.api.endpoints.portfolio import portfolio_cache + portfolio_cache.holdings = [] + portfolio_cache.summary = None + portfolio_cache.change_flag = True + except Exception: + pass + return {"action": "SELL", "stock": stock_name, "quantity": quantity, "price": current_price, "profit_loss": profit_loss_rate, "success": success} + + except Exception as e: + return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)} + + return {"action": "HOLD", "stock": stock_name} + + except Exception as e: + return {"action": "ERROR", "error": str(e)} \ No newline at end of file