refactor: create trade_executor.py

This commit is contained in:
2026-02-25 11:22:23 +09:00
parent 9ed4506fec
commit 30d4db1b28

View File

@@ -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)}