"""자동매매 실행 엔진 (trade_executor.py)

흐름 요약:
  execute_auto_trade_background()            ← 추천 목록 수신
    └─ execute_auto_trade()                  ← 종목별 공통 필터 (가격·MDD·신호 등)
          ├─ _execute_buy()                  ← 매수 로직 (급등체크→트렌드→포지션사이징→주문)
          └─ _execute_sell()                 ← 매도 로직 (중복체크→주문→DB기록)
                └─ _verify_and_sync_execution()  ← 체결 확인·슬리피지 기록·부분체결 재주문
"""
import asyncio
import logging
import time
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, _ema,
)
from app.services.signal_types import BUY_KEYWORDS, SELL_KEYWORDS, STRONG_BUY_KEYWORDS, BLOCK_KEYWORDS

logger = logging.getLogger(__name__)

# ── 종목별 세마포어 — 동일 종목 동시 주문 방지 ────────────────────────────────
_order_semaphores: dict = {}
_SEMAPHORE_MAX_SIZE = 200


def _cleanup_semaphores() -> None:
    """세마포어 딕셔너리 크기가 임계값 초과 시 미사용 항목 제거.

    [동작] sem._value == 1 이면 현재 아무도 잡고 있지 않은 상태(unlock).
    장기 운영 시 거래된 적 있는 모든 종목 코드가 누적되는 것을 방지.
    """
    if len(_order_semaphores) <= _SEMAPHORE_MAX_SIZE:
        return  # 임계값 이하 → 정리 불필요
    to_delete = [
        code for code, sem in _order_semaphores.items()
        if sem._value == 1  # unlocked = 현재 사용 중 아님
    ]
    for code in to_delete:
        _order_semaphores.pop(code, None)
    if to_delete:
        logger.debug(f"[SEMAPHORE_CLEANUP] {len(to_delete)}개 세마포어 정리 (잔여: {len(_order_semaphores)})")


def _get_order_semaphore(stock_code: str) -> asyncio.Semaphore:
    """종목별 세마포어 반환. 없으면 생성. 크기 초과 시 미사용 항목 정리.

    [동작] 동일 종목에 대해 동시에 2개 이상의 주문이 발생하지 않도록
    asyncio.Semaphore(1) 으로 상호 배제. 코드 진입 시 lock → 완료 시 release.
    """
    if stock_code not in _order_semaphores:
        _cleanup_semaphores()  # 200개 초과 시 미사용 세마포어 정리
        _order_semaphores[stock_code] = asyncio.Semaphore(1)
    return _order_semaphores[stock_code]


# ── Redis 분산락 — 중복 주문 완전 방지 (TTL 60초) ────────────────────────────
_ORDER_LOCK_TTL = 60  # seconds


async def _acquire_order_lock(stock_code: str, order_type: str) -> bool:
    """Redis SETNX 기반 분산락 획득. 성공 시 True.

    [동작] Redis SET NX(NOT EXISTS) + EX(TTL 60초)로 원자적 락 설정.
    - 동일 종목·방향의 주문이 이미 처리 중이면 False → 호출측에서 SKIP.
    - Redis 미연결 시 True(락 없이 진행) — 이 경우 세마포어가 2차 방어.
    """
    try:
        from app.core.redis import get_redis
        client = get_redis()
        if not client:
            return True  # Redis 미연결 → 락 없이 진행 (세마포어가 대신 보호)
        key = f"order_lock:{stock_code}:{order_type}"
        result = await client.set(key, "1", nx=True, ex=_ORDER_LOCK_TTL)
        return result is True  # None이면 이미 키 존재 → 락 실패
    except Exception as e:
        logger.debug(f"[order_lock] 획득 오류 (패스): {e}")
        return True


async def _release_order_lock(stock_code: str, order_type: str) -> None:
    """Redis 분산락 해제.

    [동작] finally 블록에서 반드시 호출 — 주문 성공/실패 관계없이 락 해제.
    """
    try:
        from app.core.redis import delete_key
        await delete_key(f"order_lock:{stock_code}:{order_type}")
    except Exception:
        pass  # 해제 실패 시 TTL(60초) 만료 후 자동 해제됨


# ── 주문 재시도 — Exponential backoff (1→2→4초, 최대 3회) ───────────────────
async def _place_order_with_retry(
    is_domestic: bool,
    stock_code: str,
    order_type: str,
    quantity: int,
    price: float,
    exchange_code: str = "NASD",
    max_retries: int = 3,
) -> dict:
    """주문 실행 with exponential backoff. 시스템 오류에만 재시도.

    [동작]
    1. 국내/해외 분기 후 KIS REST API 호출
    2. rt_cd "0"(성공) 또는 "1"(PENDING) → 즉시 반환 (재시도 불필요)
    3. 그 외 오류: 1초→2초→4초 대기 후 최대 max_retries회 재시도
    4. 최대 재시도 소진 시 last_result(마지막 응답) 반환
    """
    last_result: dict = {}
    for attempt in range(max_retries):
        try:
            if is_domestic:
                # 국내 주식 주문 (KIS TTTC0012U / VTTC0012U)
                result = await korea_investment_api.place_order(stock_code, order_type, quantity, price)
            else:
                # 해외 주식 주문 (KIS TTTT1002U 등)
                result = await korea_investment_api.place_overseas_order(
                    stock_code, order_type, quantity, float(price), exchange_code
                )
            # rt_cd "0"=성공, "1"=PENDING → 재시도 불필요
            if result.get("rt_cd") in ("0", "1"):
                return result
            # rt_cd 오류 → 재시도 대상
            last_result = result
            if attempt < max_retries - 1:
                wait = 1 * (2 ** attempt)  # 1s → 2s → 4s (exponential backoff)
                logger.warning(
                    f"[ORDER_RETRY] {stock_code} {order_type} rt_cd={result.get('rt_cd')} "
                    f"→ {wait}초 후 재시도 ({attempt + 1}/{max_retries})"
                )
                await asyncio.sleep(wait)
        except Exception as e:
            # 네트워크 타임아웃 등 예외 → 재시도 대상
            last_result = {"rt_cd": "-1", "msg1": str(e), "odno": None}
            if attempt < max_retries - 1:
                wait = 1 * (2 ** attempt)
                logger.warning(f"[ORDER_RETRY] {stock_code} {order_type} 예외 ({attempt + 1}/{max_retries}), {wait}초 후 재시도: {type(e).__name__}: {e}")
                await asyncio.sleep(wait)
            else:
                logger.error(f"[ORDER_MAX_RETRY] {stock_code} {order_type} 최대 재시도 초과: {type(e).__name__}: {e}")
    return last_result or {"rt_cd": "-1", "msg1": "max_retries_exceeded", "odno": None}


# ── 미체결 주문 1분 후 자동 취소 ─────────────────────────────────────────────
async def _auto_cancel_if_pending(order_id: str, stock_code: str, quantity: int, delay: int = 60) -> None:
    """delay초 후에도 PENDING 상태면 자동 취소.

    [동작] 주문 후 delay(기본 60초)를 기다린 뒤 KIS 취소 API 호출.
    이미 체결됐다면 취소 실패(msg1에 오류 메시지) → DEBUG 로그만 남기고 무시.
    호출 측에서 asyncio.create_task() 로 백그라운드 실행.
    """
    await asyncio.sleep(delay)  # 60초 대기 후 취소 시도
    try:
        result = await korea_investment_api.cancel_order(order_id, quantity)
        if result.get("rt_cd") == "0":
            logger.info(f"[AUTO_CANCEL] 미체결 자동 취소 완료: {stock_code} order_id={order_id}")
        else:
            # 이미 체결됐거나 취소 불가 상태 → 무시
            logger.debug(f"[AUTO_CANCEL] {stock_code} — {result.get('msg1')} (이미 체결됐을 수 있음)")
    except Exception as e:
        logger.debug(f"[AUTO_CANCEL] {stock_code} — {type(e).__name__}: {e}")


# ── 계좌 정보 30초 인메모리 캐시 ──────────────────────────────────────────────
_account_cache: dict = {"data": None, "ts": 0.0}
_ACCOUNT_CACHE_TTL = 30  # seconds


def _get_cached_account() -> tuple | None:
    """(available_cash, total_assets) 30초 캐시. 만료 시 None 반환.

    [동작] monotonic clock으로 TTL 검사. 만료되지 않았으면 캐시 반환,
    만료됐거나 데이터가 없으면 None → 호출측에서 portfolio_cache 재조회.
    """
    if _account_cache["data"] and time.monotonic() - _account_cache["ts"] < _ACCOUNT_CACHE_TTL:
        return _account_cache["data"]
    return None


def _set_account_cache(available_cash: float, total_assets: float) -> None:
    """계좌 캐시 갱신. 현재 시각(monotonic)을 타임스탬프로 저장."""
    _account_cache["data"] = (available_cash, total_assets)
    _account_cache["ts"] = time.monotonic()


# ── 지정가 조정 — 체결률 향상 ──────────────────────────────────────────────────
def _adjust_order_price(price: float, order_type: str, is_domestic: bool = True) -> float:
    """BUY +0.5% / SELL -0.5% 지정가 조정.

    국내: 원화 정수 반환 (KIS는 원화 단위 정수만 허용).
    해외: 소수점 2자리 반환 (USD 단위).
    """
    if order_type == "BUY":
        adjusted = price * 1.005
    else:
        adjusted = price * 0.995

    if is_domestic:
        return float(int(adjusted))  # 원화 정수
    return round(adjusted, 2)  # USD 소수점 2자리


# ── Kelly Criterion 포지션 사이징 ─────────────────────────────────────────────
async def _calculate_kelly_position(
    db: AsyncSession,
    stock_code: str,
    total_assets: float,
    available_cash: float,
    max_per_position: float = 0.04,
    lookback: int = 20,
) -> float:
    """Kelly Criterion 기반 최적 포지션 사이징 (Half-Kelly 적용).

    f* = (p*b - (1-p)) / b
    - p: 최근 N거래 승률 (이긴 거래 수 / 전체 거래 수)
    - b: 평균이익률 / 평균손실률 (Payoff Ratio)
    - Half-Kelly: f* / 2 → 변동성 감소 목적

    [동작]
    1. DB에서 최근 lookback회 매도 체결 기록 조회
    2. 5거래 미만이면 통계 신뢰 부족 → 0.0 반환 (호출측에서 v2 fallback)
    3. 반환값 = min(total_assets×clamped, available_cash, max_per_position 한도)
    """
    try:
        recent = await db.execute(
            select(TradeHistory)
            .where(
                TradeHistory.stock_code == stock_code,
                TradeHistory.order_type == "SELL",
                TradeHistory.status == "COMPLETED",
                TradeHistory.profit_loss_rate.isnot(None),
            )
            .order_by(desc(TradeHistory.executed_at))
            .limit(lookback)
        )
        trades = recent.scalars().all()
        if len(trades) < 5:
            return 0.0  # 데이터 부족 → v2 fallback

        rates = [float(t.profit_loss_rate) for t in trades]
        wins = [r for r in rates if r > 0]       # 수익 거래 목록
        losses = [abs(r) for r in rates if r < 0]  # 손실 거래 절대값 목록
        if not wins or not losses:
            return 0.0  # 전승 또는 전패 → Kelly 계산 불가

        p = len(wins) / len(rates)                           # 승률
        b = (sum(wins) / len(wins)) / (sum(losses) / len(losses))  # Payoff Ratio
        kelly_f = (p * b - (1 - p)) / b                      # Kelly 공식
        half_kelly = max(0.0, kelly_f / 2)                   # Half-Kelly (보수적)
        clamped = min(half_kelly, max_per_position)           # 종목당 최대 4% 상한 적용
        return min(total_assets * clamped, available_cash, total_assets * max_per_position)
    except Exception as e:
        logger.debug(f"[kelly] {stock_code} 계산 오류: {type(e).__name__}: {e}")
        return 0.0


# ── 체결 확인 + 부분 체결 처리 + 슬리피지 로깅 ───────────────────────────────
async def _verify_and_sync_execution(
    order_id: str,
    stock_code: str,
    stock_name: str,
    ordered_quantity: int,
    order_price: float,
    order_type: str,
    is_domestic: bool,
    exchange_code: str,
    db_trade_id: int,
    delay: int = 3,
) -> None:
    """주문 후 KIS 체결 내역 조회 → DB 업데이트 + 슬리피지 기록 + 부분 체결 재주문.

    [동작 흐름]
    1. delay초 대기 (체결 확인 전 KIS 서버 처리 시간 확보)
    2. KIS 체결 조회 API 호출 → filled_qty, filled_price 추출
    3. 슬리피지 계산: (체결가 - 주문가) / 주문가 × 100
    4. 별도 AsyncSessionLocal 세션으로 TradeHistory 업데이트
       - 완전 체결(filled_qty == ordered_quantity) → status = "COMPLETED"
       - 부분 체결(filled_qty < ordered_quantity) → status = "PARTIAL" + 잔여 재주문
    5. 포트폴리오 동기화 트리거

    주의: 호출측 db 세션과 분리된 별도 세션을 사용 (cross-session 충돌 방지).
    """
    await asyncio.sleep(delay)  # KIS 서버 체결 처리 대기

    if not order_id:
        # order_id 없으면 체결 조회 불가 → 포트폴리오 동기화만 수행
        await _delayed_sync_portfolio(0)
        return

    try:
        # KIS 체결 내역 조회 (TR: TTTC8001R / VTTC8001R)
        exec_result = await korea_investment_api.get_order_execution(order_id, is_domestic)
        filled_qty = int(exec_result.get("filled_qty", 0))
        filled_price = float(exec_result.get("filled_price", order_price) or order_price)

        # ── 슬리피지 계산: (체결가 - 주문가) / 주문가 × 100 ─────────────────
        slippage_pct = 0.0
        if order_price > 0 and filled_price > 0:
            slippage_pct = ((filled_price - order_price) / order_price) * 100
            logger.info(
                f"[SLIPPAGE] {stock_code} {order_type} "
                f"주문가={order_price:.2f} 체결가={filled_price:.2f} "
                f"슬리피지={slippage_pct:+.3f}%"
            )

        # ── DB 업데이트 (별도 세션으로 호출측 세션과 격리) ───────────────────
        if filled_qty > 0:
            async with AsyncSessionLocal() as session:
                row = await session.execute(
                    select(TradeHistory).where(TradeHistory.id == db_trade_id)
                )
                trade = row.scalar_one_or_none()
                if trade:
                    partial = filled_qty < ordered_quantity  # 부분 체결 여부
                    trade.executed_quantity = filled_qty
                    trade.executed_price = Decimal(str(filled_price))
                    trade.status = "PARTIAL" if partial else "COMPLETED"
                    trade.executed_at = _get_kst_now()
                    trade.slippage = Decimal(str(round(slippage_pct, 4)))
                    await session.commit()

                    # ── 부분 체결 → 미체결 수량 재주문 ──────────────────────
                    if partial:
                        remain = ordered_quantity - filled_qty
                        logger.warning(
                            f"[PARTIAL_FILL] {stock_code} {order_type} "
                            f"{filled_qty}/{ordered_quantity}주 체결 → 잔여 {remain}주 재주문"
                        )
                        # 재주문은 백그라운드 태스크로 분리 (현재 흐름 블록 방지)
                        asyncio.create_task(
                            _reorder_unfilled(
                                stock_code=stock_code,
                                stock_name=stock_name,
                                order_type=order_type,
                                quantity=remain,
                                price=order_price,
                                is_domestic=is_domestic,
                                exchange_code=exchange_code,
                            )
                        )

        await _delayed_sync_portfolio(0)  # KIS 잔고 기준으로 포트폴리오 재동기화

    except Exception as e:
        logger.warning(
            f"[VERIFY_EXEC] {stock_code} {order_type} 체결 확인 실패 — "
            f"{type(e).__name__}: {e} (포트폴리오 동기화만 실행)"
        )
        await _delayed_sync_portfolio(0)  # 체결 확인 실패해도 포트폴리오는 동기화


# ── 부분 체결 후 미체결 수량 재주문 ──────────────────────────────────────────
async def _reorder_unfilled(
    stock_code: str,
    stock_name: str,
    order_type: str,
    quantity: int,
    price: float,
    is_domestic: bool,
    exchange_code: str = "NASD",
) -> None:
    """부분 체결 잔여 수량 재주문 (1회 한정)."""
    try:
        result = await _place_order_with_retry(
            is_domestic=is_domestic,
            stock_code=stock_code,
            order_type=order_type,
            quantity=quantity,
            price=price,
            exchange_code=exchange_code,
            max_retries=2,
        )
        if result.get("rt_cd") == "0":
            logger.info(f"[REORDER] {stock_code} {order_type} 미체결 {quantity}주 재주문 성공")
        else:
            logger.warning(f"[REORDER] {stock_code} {order_type} 미체결 재주문 실패: {result.get('msg1')}")
    except Exception as e:
        logger.error(f"[REORDER_ERROR] {stock_code} {order_type}: {type(e).__name__}: {e}")


async def _execute_buy(
    stock: dict,
    db: AsyncSession,
    holding,
    current_price: float,
    signals: list,
    score_val: float,
    is_domestic: bool,
    exchange_code: str,
    stock_code: str,
    stock_name: str,
) -> dict:
    """BUY 주문 실행 로직 — 선체크 → semaphore → 주문 → TradeHistory 기록 → 체결 확인."""

    # ── 1. 3일 급등 체크 (ATR 기반 동적 임계값) ───────────────────────────────
    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:
            closes = (
                [float(item.get("stck_clpr", 0)) for item in chart_data[-4:]]
                if is_domestic
                else [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
                )
                try:
                    highs = (
                        [float(item.get("stck_hgpr", 0)) for item in chart_data[-4:]]
                        if is_domestic
                        else [float(item.get("high", 0)) for item in chart_data[-4:]]
                    )
                    lows = (
                        [float(item.get("stck_lwpr", 0)) for item in chart_data[-4:]]
                        if is_domestic
                        else [float(item.get("low", 0)) for item in chart_data[-4:]]
                    )
                    daily_ranges = [
                        abs(highs[i] - lows[i]) for i in range(len(closes))
                        if closes[i] > 0 and highs[i] > 0 and lows[i] > 0
                    ]
                    atr_pct = (
                        (sum(daily_ranges) / len(daily_ranges) / price_3days_ago * 100)
                        if daily_ranges and price_3days_ago > 0 else 0
                    )
                    surge_threshold = max(10.0, atr_pct * 2)
                except Exception:
                    surge_threshold = 10.0
                if recent_gain > surge_threshold:
                    return {"action": "HOLD", "stock": stock_name, "reason": "chase_buy_prevention"}
    except Exception as e:
        logger.debug(f"[SURGE_CHECK] {stock_code} 차트 조회 실패, 우회: {type(e).__name__}: {e}")

    # ── 2. 트렌드 필터 (MA + EMA 병행) ───────────────────────────────────────
    if TREND_FILTER_ENABLED:
        try:
            ma20 = stock.get("ma20")
            ma60 = stock.get("ma60")
            ema20 = ema60 = None
            closes_local: list = []

            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)
                    ema20 = _ema(closes_local, TREND_MA_SHORT)
                    ema60 = _ema(closes_local, TREND_MA_LONG)

            ma_ok = bool(
                ma20 and ma60
                and float(ma20) >= float(ma60) * (1 - TREND_TOLERANCE_PCT)
                and current_price >= float(ma20) * (1 - TREND_TOLERANCE_PCT)
            )
            ema_ok = bool(
                ema20 and ema60
                and ema20 >= ema60 * (1 - TREND_TOLERANCE_PCT)
                and current_price >= ema20 * (1 - TREND_TOLERANCE_PCT)
            ) if ema20 and ema60 else False

            if not (ma_ok or ema_ok):
                return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter"}
        except Exception as e:
            logger.debug(f"[TREND_FILTER] {stock_code}: {type(e).__name__}: {e}")
            return {"action": "HOLD", "stock": stock_name, "reason": "trend_filter_error"}

    # ── 3. 현금·자산 조회 (30초 캐시 활용) ──────────────────────────────────
    # KIS API 호출을 줄이기 위해 인메모리 캐시(TTL 30초) 우선 사용
    cached = _get_cached_account()
    if cached:
        available_cash, total_assets = cached
    else:
        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
            _set_account_cache(available_cash, total_assets)
        except Exception:
            available_cash, total_assets = 0.0, 0.0

    daily_cutoff = _get_kst_now().replace(hour=0, minute=0, second=0, microsecond=0)  # 오늘 00:00:00 KST

    # ── 4. PENDING 포함 오늘 중복 주문 차단 ──────────────────────────────────
    # 오늘 이미 BUY 주문이 있으면(COMPLETED 또는 PENDING 상태) 재진입 방지
    pending_check = await db.execute(
        select(TradeHistory).where(
            TradeHistory.stock_code == stock_code,
            TradeHistory.order_type == "BUY",
            TradeHistory.status.in_(["EXECUTED", "COMPLETED", "PENDING"]),
            TradeHistory.created_at >= daily_cutoff,
        )
    )
    if pending_check.scalars().first():
        return {"action": "SKIP", "stock": stock_name, "reason": "already_ordered_today"}

    # ── 5. 일 매수 횟수 제한 (settings.MAX_DAILY_BUYS_PER_STOCK) ─────────────
    # 종목당 당일 최대 매수 횟수를 초과하면 추가 매수 차단
    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,
        )
    )
    if len(daily_buy_check.scalars().all()) >= settings.MAX_DAILY_BUYS_PER_STOCK:
        return {"action": "HOLD", "stock": stock_name, "reason": "daily_trade_limit"}

    # ── 6. 쿨다운 체크 (ATR 기반 동적) ──────────────────────────────────────
    # ATR%가 높을수록(고변동) 쿨다운 단축 → max(15, 30 - atr_pct×3)분
    # 예: ATR% = 5% → 쿨다운 = max(15, 30-15) = 15분
    #     ATR% = 1% → 쿨다운 = max(15, 30-3)  = 27분
    atr_val = float(stock.get("atr", 0) or 0)
    if atr_val > 0 and current_price > 0:
        atr_pct = atr_val / current_price * 100
        cooldown_min = max(15, int(30 - atr_pct * 3))
    else:
        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()
    # 마지막 매수 이후 쿨다운 시간이 지나지 않았으면 HOLD
    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"}

    # ── 7. 포지션 사이징 (Kelly 우선, v2 fallback) ────────────────────────────
    # 우선순위: Half-Kelly → calculate_position_size_v2 → available_cash×4%
    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 = await _calculate_kelly_position(
        db, stock_code, total_assets, available_cash
    )
    if invest_amount <= 0:
        # Kelly 데이터 부족(5거래 미만) → 전략 v2 fallback
        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:
        # v2도 0 반환 시(현금 부족 등) → 가용현금의 4% 최소 배분
        invest_amount = available_cash * 0.04

    # 투자금액 → 주수 환산 (Decimal 사용으로 부동소수점 오차 방지)
    quantity = int(Decimal(str(invest_amount)) / Decimal(str(current_price)))
    if quantity <= 0:
        return {"action": "SKIP", "reason": "quantity_zero", "stock": stock_name}

    # ── 8. Redis 분산락 획득 ──────────────────────────────────────────────────
    # 여러 워커/프로세스가 동시에 같은 종목을 매수하는 것을 방지
    if not await _acquire_order_lock(stock_code, "BUY"):
        return {"action": "SKIP", "stock": stock_name, "reason": "order_lock_busy"}

    # ── 9. 주문 실행 (종목별 세마포어 + 재시도) ───────────────────────────────
    # 세마포어: 단일 프로세스 내 동일 종목 동시 주문 방지
    # 재시도: exponential backoff (1→2→4초, 최대 3회)
    try:
        order_price = _adjust_order_price(current_price, "BUY", is_domestic)  # +0.5% 체결률 향상
        async with _get_order_semaphore(stock_code):
            order_result = await _place_order_with_retry(
                is_domestic=is_domestic,
                stock_code=stock_code,
                order_type="BUY",
                quantity=quantity,
                price=order_price,
                exchange_code=exchange_code,
            )

        total_amount = Decimal(str(current_price)) * quantity
        entry_reason = f"매수 신호: {', '.join(signals[:3])}. 점수: {score_val}점"
        success = order_result.get("rt_cd") == "0"   # KIS 성공 코드
        is_pending = order_result.get("rt_cd") == "1"  # 주문접수(체결 미확인)
        status = "COMPLETED" if success else ("PENDING" if is_pending else "FAILED")

        # TradeHistory DB 기록 (flush로 id 확보 후 commit)
        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.flush()   # auto-increment id 확보 (commit 전에 id 필요)
        trade_id = trade.id
        await db.commit()

        if success:
            # 인메모리 포트폴리오 캐시 즉시 반영 (UI 실시간 업데이트)
            try:
                from app.api.endpoints.portfolio import portfolio_cache
                portfolio_cache.apply_buy(stock_code, stock_name, quantity, current_price)
            except Exception:
                pass
            # 3초 후 KIS 체결 확인 → 슬리피지 기록 + 부분체결 재주문 (백그라운드)
            asyncio.create_task(
                _verify_and_sync_execution(
                    order_id=order_result.get("odno", ""),
                    stock_code=stock_code,
                    stock_name=stock_name,
                    ordered_quantity=quantity,
                    order_price=order_price,
                    order_type="BUY",
                    is_domestic=is_domestic,
                    exchange_code=exchange_code,
                    db_trade_id=trade_id,
                    delay=3,
                )
            )
        elif is_pending:
            # PENDING(주문접수) 상태이면 60초 후 미체결 자동 취소 시도
            order_id_val = order_result.get("odno")
            if order_id_val:
                asyncio.create_task(_auto_cancel_if_pending(order_id_val, stock_code, quantity))

        return {
            "action": "BUY",
            "stock": stock_name,
            "quantity": quantity,
            "price": current_price,
            "success": success,
        }

    except ValueError as e:
        logger.error(f"[BUY_VALIDATION] {stock_code}: {e}")
        return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)}
    except Exception as e:
        logger.error(f"[BUY_ERROR] {stock_code}: {type(e).__name__}: {e}", exc_info=True)
        return {"action": "BUY_FAILED", "stock": stock_name, "error": str(e)}
    finally:
        await _release_order_lock(stock_code, "BUY")


async def _execute_sell(
    stock: dict,
    db: AsyncSession,
    holding,
    current_price: float,
    signals: list,
    is_domestic: bool,
    exchange_code: str,
    stock_code: str,
    stock_name: str,
) -> dict:
    """SELL 주문 실행 로직 — 중복체크 → semaphore → 주문 → TradeHistory → 체결 확인.

    [동작 흐름]
    1. 현재 손익률 계산 (현재가 vs 평균매입가)
    2. ATR 기반 동적 손절선·트레일링스탑 재계산
    3. Redis 분산락 획득 → 중복 매도 방지
    4. 30초 이내 동일 종목 SELL 기록 확인 (FAILED 제외)
    5. 지정가 -0.5% 조정 후 KIS SELL 주문
    6. TradeHistory DB 기록 (flush → id 확보 → commit)
    7. 성공 시: 포트폴리오 캐시 즉시 제거 + 체결 확인 백그라운드 태스크
    """

    quantity = holding.quantity  # 보유 전량 매도
    _avg = float(holding.average_price or 0)
    # 손익률: (현재가 - 평균매입가) / 평균매입가 × 100
    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)

    # ── ATR 기반 동적 손절/트레일링 스탑 조정 ────────────────────────────────
    # ATR 손절선 = 평균매입가 - 2.5×ATR (기본 손절보다 유리한 쪽 사용)
    # 트레일링스탑 = 고점 - 1.5×ATR (수익 실현 구간에서만 적용)
    atr = float(stock.get("atr", 0) or 0)
    _avg_price = float(holding.average_price or 0)
    exit_reason_detail = f"손익률 {profit_loss_rate:.2f}%"
    if atr > 0 and _avg_price > 0:
        atr_stop_price = _avg_price - 2.5 * atr              # ATR 기반 손절가
        atr_stop_pct = ((atr_stop_price / _avg_price) - 1) * 100
        stop_loss = max(stop_loss, atr_stop_pct)              # 기본 손절 vs ATR 손절 중 유리한 쪽

        high_price = float(stock.get("high_price") or current_price)
        trailing_stop_price = high_price - 1.5 * atr          # 고점 기준 트레일링스탑
        if current_price <= trailing_stop_price and profit_loss_rate > 0:
            exit_reason_detail = f"ATR 트레일링 스탑: {current_price:.0f} ≤ {trailing_stop_price:.0f} | {exit_reason_detail}"

    # Redis 분산락 획득 — 여러 워커가 동시에 같은 종목을 매도하는 것을 방지
    if not await _acquire_order_lock(stock_code, "SELL"):
        return {"action": "SKIP", "stock": stock_name, "reason": "order_lock_busy"}

    try:
        # ── 1. 중복 주문 체크 (30초 이내, FAILED 제외) ───────────────────────
        # FAILED 주문은 제외하여 오류 이후 재매도 허용
        recent_cutoff = _get_kst_now() - timedelta(seconds=30)
        duplicate_check = await db.execute(
            select(TradeHistory).where(
                TradeHistory.stock_code == stock_code,
                TradeHistory.order_type == "SELL",
                TradeHistory.status.in_(["COMPLETED", "PARTIAL", "PENDING"]),
                TradeHistory.created_at >= recent_cutoff,
            )
        )
        if duplicate_check.scalar_one_or_none():
            return {"action": "SKIP", "reason": "duplicate_order", "stock": stock_name}

        # ── 2. 주문 실행 (종목별 세마포어 + 재시도) ──────────────────────────
        order_price = _adjust_order_price(current_price, "SELL", is_domestic)  # -0.5% 체결률 향상
        async with _get_order_semaphore(stock_code):
            order_result = await _place_order_with_retry(
                is_domestic=is_domestic,
                stock_code=stock_code,
                order_type="SELL",
                quantity=quantity,
                price=order_price,
                exchange_code=exchange_code,
            )

        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"
        is_pending = order_result.get("rt_cd") == "1"
        status = "COMPLETED" if success else ("PENDING" if is_pending else "FAILED")

        # TradeHistory DB 기록
        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"자동매도: {exit_reason_detail}",
            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.flush()   # auto-increment id 확보
        trade_id = trade.id
        await db.commit()

        if success:
            # 인메모리 포트폴리오 캐시에서 해당 종목 즉시 제거 (UI 실시간 반영)
            try:
                from app.api.endpoints.portfolio import portfolio_cache
                avg = float(holding.average_price or current_price)
                portfolio_cache.apply_sell(stock_code, quantity, current_price, avg)
                portfolio_cache.holdings = [
                    h for h in portfolio_cache.holdings if h.stock_code != stock_code
                ]
            except Exception:
                pass
            # 3초 후 KIS 체결 확인 (슬리피지 기록 포함)
            asyncio.create_task(
                _verify_and_sync_execution(
                    order_id=order_result.get("odno", ""),
                    stock_code=stock_code,
                    stock_name=stock_name,
                    ordered_quantity=quantity,
                    order_price=order_price,
                    order_type="SELL",
                    is_domestic=is_domestic,
                    exchange_code=exchange_code,
                    db_trade_id=trade_id,
                    delay=3,
                )
            )
        elif is_pending:
            # PENDING 상태 → 60초 후 미체결 자동 취소 시도
            order_id_val = order_result.get("odno")
            if order_id_val:
                asyncio.create_task(_auto_cancel_if_pending(order_id_val, stock_code, quantity))

        return {
            "action": "SELL",
            "stock": stock_name,
            "quantity": quantity,
            "price": current_price,
            "profit_loss": profit_loss_rate,
            "success": success,
        }

    except ValueError as e:
        logger.error(f"[SELL_VALIDATION] {stock_code}: {e}")
        return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)}
    except Exception as e:
        logger.error(f"[SELL_ERROR] {stock_code}: {type(e).__name__}: {e}", exc_info=True)
        return {"action": "SELL_FAILED", "stock": stock_name, "error": str(e)}
    finally:
        await _release_order_lock(stock_code, "SELL")


async def execute_auto_trade_background(recommendations_response: dict, db: AsyncSession):
    """recommendations_response dict 에서 추천 종목을 꺼내 자동매매 실행.

    [동작 흐름]
    1. 시장 국면(BEAR_MARKET_FILTER) 조회 → 최대 거래 수·포지션 배율 결정
    2. recommendations_by_rule 그룹을 단일 unique_stocks dict로 정규화
       - 동일 종목이 여러 룰에 걸리면 우선순위(rule_priority) 낮은 쪽 유지
    3. (rule_priority ASC, score DESC)로 정렬 → 상위 max_trades개 처리
    4. 종목별 execute_auto_trade() 호출 (0.5초 간격 → KIS API 레이트 리밋 준수)
    """
    if not isinstance(recommendations_response, dict):
        logger.error("[AUTO_TRADE_BG] recommendations_response is not a dict: %r", recommendations_response)
        return

    try:
        total_processed = 0
        trade_results = []

        # 시장 국면 조회 (BEAR 시 max_trades 축소, position_multiplier 감소)
        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()}  # 현재 보유 종목 집합

        # recommendations_by_rule 정규화 (list 또는 dict 모두 처리)
        rec_groups = recommendations_response.get("recommendations_by_rule", [])
        if isinstance(rec_groups, dict):
            normalized_list = []
            for name, group in rec_groups.items():
                if isinstance(group, dict):
                    normalized_list.append(group)
                else:
                    normalized_list.append({"rule_name": name, "stocks": group})
            rec_groups = normalized_list

        # 종목 코드 기준으로 중복 제거 (rule_priority 낮은 쪽 = 더 중요한 규칙 우선)
        unique_stocks: dict = {}
        for rule_group in rec_groups:
            if not isinstance(rule_group, dict):
                logger.warning("[AUTO_TRADE_BG] Invalid rule_group: %r", rule_group)
                continue
            stocks_list = rule_group.get("stocks")
            if not isinstance(stocks_list, list):
                logger.warning("[AUTO_TRADE_BG] Invalid stocks list: %r", stocks_list)
                continue
            for stock in stocks_list:
                if not isinstance(stock, dict):
                    logger.warning("[AUTO_TRADE_BG] Skipping non-dict stock entry: %r", stock)
                    continue
                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  # 더 높은 우선순위 룰로 교체

        # 우선순위↑(숫자↓), 점수↓ 순으로 정렬 후 max_trades개만 처리
        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]:
            if not isinstance(stock, dict):
                logger.warning("[AUTO_TRADE_BG] Skipping invalid stock entry: %r", stock)
                continue
            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)  # KIS API 레이트 리밋: 초당 2회 이하 권장

            except Exception as e:
                logger.error(f"[AUTO_TRADE_BG] {stock.get('stock_code')} 처리 오류: {type(e).__name__}: {e}")
                continue

    except Exception as e:
        logger.error(
            f"[AUTO_TRADE_BG] 전체 오류: {type(e).__name__}: {e} — "
            f"recommendations_response keys={list(recommendations_response.keys()) if isinstance(recommendations_response, dict) else '?'}"
        )


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"[PORTFOLIO_SYNC] 백그라운드 동기화 실패: {type(e).__name__}: {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"[PORTFOLIO_SYNC] 실행 실패: {type(inner_e).__name__}: {inner_e}")
    except Exception as e:
        logger.error(f"[PORTFOLIO_SYNC] 딜레이 동기화 실패: {type(e).__name__}: {e}")


async def execute_auto_trade(stock: dict, db: AsyncSession):
    """종목 단건 매매 판단 및 실행.

    공통 필터(가격·MDD·손실추적·보유수·당일매도쿨다운·신호분석)를 통과한 후
    _execute_buy / _execute_sell 에 위임한다.

    [필터 체크 순서]
    1. 유효성 검사 (stock_code, current_price)
    2. 국내/해외 구분 (stock_code 6자리 숫자 → 국내)
    3. 가격 필터 (국내 5,000원↑ / 미국 $4↑)
    4. MDD 일시정지 여부 (MDD < -10% 시 신규진입 중단)
    5. 연속손실 밴 여부 (3회 연속 손실 → 30일 진입 금지)
    6. 최대 보유 종목 수 초과 여부 (settings.MAX_HOLDINGS_COUNT)
    7. 당일 매도 후 재매수 쿨다운 (당일 SELL COMPLETED 기록 존재 시 재매수 금지)
    8. 신호 분석: BUY_KEYWORDS / SELL_KEYWORDS / BLOCK_KEYWORDS 매칭
    9. ML 가중치 통합 (ML_ENSEMBLE_ENABLED 시 rule 40% + ML 60% 합산)
    10. ATR 강제 손절 (신호 없어도 현재가 ≤ avg - 2.5×ATR 이면 강제 매도)
    11. _execute_buy 또는 _execute_sell 위임
    """
    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"}

        # 국내(6자리 숫자) vs 해외 판별
        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}

        # ── MDD / 손실추적 / 보유수 ──────────────────────────────────────────
        if not await mdd_manager.can_open_new_position():
            # MDD -10% 이상 낙폭 → 신규 진입 중단
            return {"action": "SKIP", "reason": "mdd_pause", "stock": stock_name}

        if loss_tracker.is_banned(stock_code):
            # 3회 연속 손실 종목 → 30거래일 진입 금지
            return {"action": "SKIP", "reason": "consecutive_loss_ban", "stock": stock_name}

        held_result = await db.execute(select(Portfolio).where(Portfolio.quantity > 0))
        if len(held_result.scalars().all()) >= settings.MAX_HOLDINGS_COUNT:
            # 동시 보유 20종목 한도 초과
            return {"action": "SKIP", "reason": "max_holdings_reached", "stock": stock_name}

        # ── 당일 매도 후 재매수 쿨다운 ───────────────────────────────────────
        # 당일 이미 SELL 체결된 종목은 당일 재매수 금지 (섹션 4.3)
        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.in_(["COMPLETED", "PARTIAL"]),
                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()  # None이면 신규매수, 있으면 추가매수/매도 판단

        # ── 신호 분석 (Enum 기반) ─────────────────────────────────────────────
        text_signals = " ".join(signals)
        try:
            score_val = float(stock.get("score", 0) or 0)
        except Exception:
            score_val = 0.0

        # "거래량부족" 신호 제거 후 매수 판단 (거래량 부족 단독으로는 매수 차단)
        signals_without_volume_weak = [s for s in signals if "거래량부족" not in s]
        text_signals_filtered = " ".join(signals_without_volume_weak)

        should_buy = any(k in text_signals_filtered for k in BUY_KEYWORDS)
        if not signals_without_volume_weak and signals:
            should_buy = False  # 거래량부족 신호만 있으면 매수 금지

        should_sell = any(k in text_signals for k in SELL_KEYWORDS)

        # ── ML 가중치 통합 (기본: 규칙 30% + ML 70%) ─────────────────────────
        # weighted_score = score × RULE_WEIGHT + ml_score × ML_WEIGHT
        # ML BUY: weighted_score ≥ threshold AND confidence ≥ min → should_buy = True
        # ML SELL: confidence ≥ 0.5 → should_sell = True
        # ML HOLD: weighted_score 낮으면 should_buy = False 강제
        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:
            try:
                weighted_score = (
                    score_val * settings.ML_ENSEMBLE_RULE_WEIGHT
                    + float(ml_score_trade) * settings.ML_ENSEMBLE_ML_WEIGHT
                )
                if (
                    ml_signal_trade == "BUY"
                    and weighted_score >= 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 weighted_score < settings.ML_ENSEMBLE_BUY_THRESHOLD * 0.7
                ):
                    should_buy = False  # ML이 HOLD 강하게 권고 시 매수 억제
            except Exception as e:
                logger.debug(f"[ML_WEIGHT] {stock_code} 가중치 계산 오류: {e}")

        has_rule = stock.get("rule_id") is not None
        if has_rule:
            # 규칙 기반 종목: BLOCK 신호만 없으면 should_buy = True (규칙 자체가 매수 신호)
            if any(b in text_signals for b in BLOCK_KEYWORDS):
                return {"action": "HOLD", "stock": stock_name, "reason": "disparity_overheated"}

            ml_hold_active = settings.ML_ENSEMBLE_ENABLED and ml_signal_trade == "HOLD" and not should_buy
            if not ml_hold_active:
                should_buy = True
        else:
            # 비규칙 종목: BLOCK 신호 있으면 과열 HOLD
            if any(b in text_signals for b in BLOCK_KEYWORDS):
                return {"action": "HOLD", "stock": stock_name, "reason": "overheat_no_rule"}

        if should_buy and not has_rule:
            # 비규칙 종목: 강한 신호 OR (신호 2개 이상 AND 점수 2점 이상) 필수
            has_strong_signal = any(k in text_signals_filtered for k in STRONG_BUY_KEYWORDS)
            matched_count = sum(1 for k in BUY_KEYWORDS if k in text_signals_filtered)
            if not has_strong_signal and (matched_count < 2 or score_val < 2):
                should_buy = False

        # ── ATR 기반 강제 손절 (신호 없어도 발동) ────────────────────────────
        # 현재가가 평균매입가 - 2.5×ATR 이하이면 매도 신호 없어도 강제 손절
        if holding and not should_sell:
            atr = float(stock.get("atr", 0) or 0)
            avg_price = float(holding.average_price or 0)
            if atr > 0 and avg_price > 0:
                atr_stop_price = avg_price - 2.5 * atr
                if current_price <= atr_stop_price:
                    should_sell = True
                    signals = list(signals) + [f"ATR 손절: {current_price:.0f} ≤ {atr_stop_price:.0f}"]

        # ── 매매 실행 위임 ────────────────────────────────────────────────────
        if should_buy:
            return await _execute_buy(
                stock, db, holding, current_price, signals, score_val,
                is_domestic, exchange_code, stock_code, stock_name,
            )

        if should_sell and holding:
            return await _execute_sell(
                stock, db, holding, current_price, signals,
                is_domestic, exchange_code, stock_code, stock_name,
            )

        return {"action": "HOLD", "stock": stock_name}

    except ValueError as e:
        logger.error(f"[TRADE_VALIDATION] {stock.get('stock_code')}: {e}")
        return {"action": "ERROR", "error": str(e)}
    except Exception as e:
        logger.error(f"[TRADE_ERROR] {stock.get('stock_code')}: {type(e).__name__}: {e}", exc_info=True)
        return {"action": "ERROR", "error": str(e)}