"""
MDD(Maximum Drawdown) 관리 - CLAUDE.md 섹션 6.3

- 포트폴리오 최고점 대비 하락률 추적
- -10% 돌파 시: 신규 진입 중단 + 기존 포지션 30% 축소
- -20% 돌파 시: 전체 포지션 50% 강제 축소 (레버리지/인버스 우선)
"""
import asyncio
import json
import logging
from typing import Dict, Optional

from app.core.config import settings
from app.core.json_helpers import _dumps
from app.core.log_throttle import warn_throttled
# [2026-09-27 §8.5 항목6] DT 전략명 정본은 db_worker (순환 import 없음 — db_worker 는
# 모듈 레벨에서 서비스 모듈을 import 하지 않는다).
from app.services.db_worker import DT_STRATEGY_NAMES as _DT_STRATEGY_NAMES

logger = logging.getLogger(__name__)

_MDD_REDIS_KEY = "mdd:state"


_PAGE_FULL_SENTINEL = "<balance-page-full>"
# MDD 갱신 정지 판정 기본값(초). swing 사이클 상한(진입 300 + sleep 60 + 청산 180 + 잔고 60 = 600)보다
# 커야 정상 사이클이 stale 로 오판되지 않는다. config.MDD_UPDATE_STALE_SEC 가 있으면 그것이 우선.
_MDD_UPDATE_STALE_SEC_DEFAULT = 900


def _stale_sec_v2() -> int:
    try:
        return int(getattr(settings, "MDD_UPDATE_STALE_SEC", _MDD_UPDATE_STALE_SEC_DEFAULT) or 0)
    except (TypeError, ValueError):
        return _MDD_UPDATE_STALE_SEC_DEFAULT


def _balance_page_full_v2(holdings) -> bool:
    """get_balance 첫 페이지가 가득 찼는지(연속조회 미구현 — 응답 밖 종목 누락 가능)."""
    rows = int(getattr(settings, "SWING_BALANCE_FULL_PAGE_ROWS", 20) or 0)
    return rows > 0 and len(holdings) >= rows


def _balance_incomplete_v2(balance, holdings) -> bool:
    """[하드닝 R9 2026-10-12] 잔고 응답이 불완전할 수 있는지(응답 밖 보유 종목 누락 가능).

    get_balance_all_v2 응답(_pages 키 있음)은 연속조회 결과라 `_partial` 이 True 일 때만 불완전이다
    (행 수가 페이지 한도 이상이어도 완전 응답이면 누락이 없다). `_pages` 가 없는 응답(구 get_balance
    형태)은 기존 휴리스틱(첫 페이지 가득 = 불완전)을 그대로 쓴다.
    """
    if isinstance(balance, dict) and "_pages" in balance:
        return bool(balance.get("_partial"))
    return _balance_page_full_v2(holdings)


def _reduction_net_pnl_v2(avg_price: float, current_price: float, sell_qty: int) -> dict:
    """[R4 2026-10-09] MDD 축소 매도의 비용 차감 순손익 (CLAUDE.md §5.1 순손익 기준).

    이전엔 net_profit_loss 에 비용 전 gross 추정치를 넣고 commission/profit_loss_rate 를 비웠다.
    시장 구분 정보가 없어 거래세는 DAY_TRADE_DEFAULT_TAX_MARKET 세율(DT 와 동일 규약)을 쓴다.
    체결가(executed_price)는 NULL 이라 current_price 기준 추정치다 — 확정되면 fill_sweep 이 보정.
    반환: net_profit_loss / commission(왕복 비용) / profit_loss_rate(비용 차감 %).
    """
    from app.services.trade_cost_core import net_realized_pnl
    if str(getattr(settings, "DAY_TRADE_DEFAULT_TAX_MARKET", "KOSPI")).upper() in ("KOSDAQ", "KQ"):
        tax_rate = float(settings.KR_SELL_TAX_RATE_KOSDAQ)
    else:
        tax_rate = float(settings.KR_SELL_TAX_RATE_KOSPI)
    pnl = net_realized_pnl(
        float(avg_price), float(current_price), int(sell_qty),
        float(settings.KR_COMMISSION_RATE), tax_rate,
    )
    return {
        "net_profit_loss": pnl["net_profit_loss"],
        "commission": pnl["total_cost"],
        "profit_loss_rate": pnl["net_profit_loss_rate"],
    }


_alert_tasks: set = set()  # fire-and-forget 태스크 참조 보관(GC 방지)


def _alert_retries_v2() -> int:
    try:
        return max(0, int(getattr(settings, "MDD_ALERT_SEND_RETRIES", 3)))
    except (TypeError, ValueError):
        return 3


def _alert_backoff_v2() -> float:
    try:
        return max(0.0, float(getattr(settings, "MDD_ALERT_RETRY_BACKOFF_SEC", 5.0)))
    except (TypeError, ValueError):
        return 5.0


async def _send_mdd_alert_v2(action: str, drawdown: float, reduce_pct: float) -> bool:
    """MDD 단계 전이 텔레그램 — 예외/False(전송 실패) 시 재시도, 끝내 실패하면 error 로그.

    notify_mdd_warning 이 bool 을 주면 그 값을, None(반환값 미사용 구현)이면 예외 없음=성공으로 본다.
    """
    retries = _alert_retries_v2()
    backoff = _alert_backoff_v2()
    last_err = "send returned False"
    for attempt in range(retries + 1):
        try:
            from app.services.telegram_notifier import telegram_notifier
            res = await telegram_notifier.notify_mdd_warning(
                action=action, drawdown=drawdown, reduce_pct=reduce_pct,
            )
            if res is not False:
                return True
            last_err = "send returned False"
        except Exception as e:
            last_err = f"{type(e).__name__}: {e}"
        if attempt < retries:
            await asyncio.sleep(backoff * (attempt + 1))
    logger.error(
        "[MDD] %s 텔레그램 알림 최종 실패(%d회 시도): %s — 운영자 수동 확인 필요",
        action, retries + 1, last_err,
    )
    return False


class MDDManager:
    """포트폴리오 MDD 관리자 (싱글톤으로 사용)"""

    def __init__(self):
        self.peak_balance: float = 0.0
        self.is_paused: bool = False
        self.is_short_term_strategy_paused: bool = False  # MDD -10% 시 단기 전략 중단
        self.reduction_applied: Optional[str] = None  # None, "10pct", "20pct"
        self._reduction_executed_10pct: bool = False  # 중복 축소 실행 방지
        self._reduction_executed_20pct: bool = False
        self._lock = asyncio.Lock()  # 스케줄러-sync 동시 접근 방지
        # ── [항목7 2026-09-27] MDD 갱신 헬스 (fail-safe 판정용) ──────────────
        # 갱신 경로(swing/scheduler)가 record_update_failure/success 로 여기에 보고한다.
        # 상태를 mdd_manager 가 소유해야 can_open_new_position() 이 순환 import 없이
        # 판정할 수 있다(scheduler → mdd_manager 는 모듈 레벨 import 이므로 역방향 금지).
        self._update_failures: int = 0
        self._update_last_error: Optional[str] = None
        self._update_empty_account: bool = False
        self._blind_block_active: bool = False
        self._blind_block_hits: int = 0  # 차단 발동 누적 횟수(로그 레이트 리밋용)
        # ── [C-1 2026-10-02] 강제축소 진행 상태 (레벨별) ─────────────────────
        # 한 종목이라도 실패하면 executed 플래그를 리셋해 재시도하는데, 기존에는
        # 재시도가 '모든 보유종목'을 다시 30%/50% 매도해 성공 종목이 사이클마다 누적
        # 축소됐다(5사이클 82% 청산 재현). 레벨별 완료 집합으로 재시도를 실패 종목에
        # 한정한다. 완료 집합은 peak 갱신·reset() 시에만 초기화(§5.3 플래그 리셋과 동일)
        # 하고, 시도 횟수/포기 집합은 KST 날짜가 바뀌면 초기화한다.
        self._reduced_codes: Dict[str, set] = {}          # level -> {stock_code}
        self._reduction_attempts: Dict[str, Dict[str, int]] = {}  # level -> {code: 실패횟수}
        self._reduction_gave_up: Dict[str, set] = {}      # level -> {code} (당일 포기)
        self._reduction_progress_date: Optional[str] = None
        self._reduction_epoch: int = 0  # peak 갱신/reset 세대 (늦은 완료 기록 무효화)
        # [M-1 5회차] 전송 후 결과 불명 주문 — level -> {code: {"qty_before","sell_qty","cycles"}}
        # 다음 주기 재매도 전에 KIS 잔고(보유·주문가능 수량)로 체결 여부를 확인한다.
        self._reduction_unconfirmed: Dict[str, Dict[str, dict]] = {}
        # [3회차 2026-10-02] 진행 중 레벨(인메모리 전용 동시 실행 가드). executed 플래그는
        # 레벨 축소가 **완료**됐을 때만 True 로 둔다 — 루프 시작 시 True 로 두면 루프 중
        # save_to_redis 가 그 값을 영속해, 중간 재기동 시 남은 종목 축소가 영구 생략됐다.
        self._reduction_running: set = set()
        # [R1 2026-10-09] 마지막 관측 잔고 / reset 후 peak 재설정 대기 / 갱신 신선도(staleness)
        self._last_balance: float = 0.0
        self._rebase_peak_pending: bool = False
        self._update_last_ok_at = None  # datetime(KST). None 이면 생성 시각을 기준으로 삼는다
        self._created_at = self._now_kst_v2()

    @staticmethod
    def _spawn_mdd_alert_v2(action: str, drawdown: float, reduce_pct: float) -> None:
        """알림 태스크를 만들고 참조를 보관한다(GC 소실 방지). 실행 중 루프가 없으면 error."""
        try:
            task = asyncio.get_running_loop().create_task(
                _send_mdd_alert_v2(action, drawdown, reduce_pct)
            )
        except RuntimeError as e:
            logger.error("[MDD] %s 텔레그램 알림 예약 실패(이벤트 루프 없음): %s", action, e)
            return
        _alert_tasks.add(task)
        task.add_done_callback(_alert_tasks.discard)

    @staticmethod
    def _now_kst_v2():
        from datetime import datetime
        from app.core.market_schedule import KST
        return datetime.now(KST)

    def _state_snapshot_v2(self) -> dict:
        """[L-6 5회차] 영속 상태 스냅샷 (동기, I/O 없음 — Lock 안에서 호출 가능)."""
        return {
            "peak_balance": self.peak_balance,
            "is_paused": self.is_paused,
            "is_short_term_strategy_paused": self.is_short_term_strategy_paused,
            "reduction_applied": self.reduction_applied,
            "_reduction_executed_10pct": self._reduction_executed_10pct,
            "_reduction_executed_20pct": self._reduction_executed_20pct,
            **self._reduction_progress_state(),
        }

    async def _write_state_v2(self, key: str, state: dict) -> None:
        """[L-6 5회차] Redis 쓰기 — **Lock 밖에서만** 호출."""
        try:
            from app.core.redis import get_redis
            r = get_redis()
            if r:
                await r.set(key, _dumps(state))
        except Exception as e:
            logger.warning(f"[MDD] Redis 저장 실패 ({key}): {e}")

    async def save_to_redis(self) -> None:
        """MDD 상태를 Redis에 영속화 — 앱 재시작 시 복원용. (Lock 밖에서 호출)"""
        await self._write_state_v2(_MDD_REDIS_KEY, self._state_snapshot_v2())

    async def load_from_redis(self) -> bool:
        """Redis에서 MDD 상태 복원. 복원 성공 시 True 반환."""
        try:
            from app.core.redis import get_redis
            r = get_redis()
            if r:
                raw = await r.get(_MDD_REDIS_KEY)
                if raw:
                    state = json.loads(raw)
                    self.peak_balance = float(state.get("peak_balance", 0.0))
                    self.is_paused = bool(state.get("is_paused", False))
                    self.is_short_term_strategy_paused = bool(state.get("is_short_term_strategy_paused", False))
                    self.reduction_applied = state.get("reduction_applied")
                    self._reduction_executed_10pct = bool(state.get("_reduction_executed_10pct", False))
                    self._reduction_executed_20pct = bool(state.get("_reduction_executed_20pct", False))
                    self._restore_reduction_progress(state)
                    logger.info(
                        f"[MDD] Redis에서 복원: peak={self.peak_balance:,.0f}, "
                        f"paused={self.is_paused}, reduction={self.reduction_applied}"
                    )
                    return True
        except Exception as e:
            logger.warning(f"[MDD] Redis 복원 실패: {e}")
        return False

    async def update(self, current_balance: float) -> Dict:
        """MDD 업데이트 및 필요한 조치 반환 (thread-safe)

        Returns:
            {"action": "NONE"/"REDUCE_AND_PAUSE"/"FORCE_REDUCE",
             "drawdown": float, ...}
        """
        # [L-6 5회차] 락 안에서는 상태 변경 + 스냅샷만, Redis 쓰기는 락 밖에서.
        # (Redis 타임아웃 동안 can_open_new_position()·축소 루프가 락을 기다리던 문제)
        async with self._lock:
            result = self._update_locked_v2(current_balance)
            if current_balance > 0:
                self._last_balance = float(current_balance)
            snap = self._state_snapshot_v2()
        await self._write_state_v2(_MDD_REDIS_KEY, snap)
        return result

    def _update_locked_v2(self, current_balance: float) -> Dict:
        """update() 의 상태 전이 본체 (Lock 보유 중 호출, I/O 없음)."""
        if self._rebase_peak_pending and current_balance > 0:
            # [R1] reset() 시점에 잔고를 몰랐던 경우 — 첫 갱신 잔고를 새 기준 peak 로 삼는다
            self.peak_balance = float(current_balance)
            self._rebase_peak_pending = False
        if current_balance > self.peak_balance:
            self.peak_balance = current_balance
            self.is_paused = False
            self.is_short_term_strategy_paused = False
            self.reduction_applied = None
            self._reduction_executed_10pct = False
            self._reduction_executed_20pct = False
            self._clear_reduction_progress()

        if self.peak_balance <= 0:
            return {"action": "NONE", "drawdown": 0.0}

        drawdown_pct = ((current_balance - self.peak_balance) / self.peak_balance) * 100

        # -20% 돌파: 50% 강제 축소
        if drawdown_pct <= -20 and self.reduction_applied != "20pct":
            self.is_paused = True
            self.is_short_term_strategy_paused = True  # 단번에 -20% 도달 시에도 단기전략 차단
            self.reduction_applied = "20pct"
            logger.warning(f"MDD {drawdown_pct:.1f}% - 50% 강제 축소 + 신규진입 중단")
            # Telegram MDD 경고
            self._spawn_mdd_alert_v2("FORCE_REDUCE", drawdown_pct, 0.50)
            return {
                "action": "FORCE_REDUCE",
                "drawdown": drawdown_pct,
                "reduce_pct": 0.50,
                "prioritize_leverage": True,
            }

        # -10% 돌파: 30% 축소 + 신규진입 중단 + 단기전략 중단
        if drawdown_pct <= -10 and self.reduction_applied is None:
            self.is_paused = True
            self.is_short_term_strategy_paused = True  # CLAUDE.md 5.3: 단기 매매 전략 중단
            self.reduction_applied = "10pct"
            logger.warning(f"MDD {drawdown_pct:.1f}% - 30% 축소 + 신규진입 중단 + 단기전략 중단")
            # Telegram MDD 경고
            self._spawn_mdd_alert_v2("REDUCE_AND_PAUSE", drawdown_pct, 0.30)
            return {
                "action": "REDUCE_AND_PAUSE",
                "drawdown": drawdown_pct,
                "reduce_pct": 0.30,
                "short_term_strategy_paused": True,
            }

        # Redis 영속화는 호출부(update)가 락 밖에서 수행
        return {"action": "NONE", "drawdown": drawdown_pct}

    async def save_to_redis_v2(self, user_id: int) -> None:
        """[v2] 사용자 스코프 MDD 상태를 `mdd:state:{user_id}` 키로 영속화.

        save_to_redis()와 동일 직렬화 포맷 재사용, 키 네임스페이스만 사용자별로 분리.
        레거시 save_to_redis()는 불변(전역 "mdd:state" 키 그대로).
        """
        try:
            from app.core.redis import get_redis
            r = get_redis()
            if r:
                state = {
                    "peak_balance": self.peak_balance,
                    "is_paused": self.is_paused,
                    "is_short_term_strategy_paused": self.is_short_term_strategy_paused,
                    "reduction_applied": self.reduction_applied,
                    "_reduction_executed_10pct": self._reduction_executed_10pct,
                    "_reduction_executed_20pct": self._reduction_executed_20pct,
                    **self._reduction_progress_state(),
                }
                await r.set(f"{_MDD_REDIS_KEY}:{user_id}", _dumps(state))
        except Exception as e:
            logger.warning(f"[MDD v2] Redis 저장 실패 (user={user_id}): {e}")

    async def load_from_redis_v2(self, user_id: int) -> bool:
        """[v2] `mdd:state:{user_id}` 키에서 사용자 스코프 MDD 상태 복원."""
        try:
            from app.core.redis import get_redis
            r = get_redis()
            if r:
                raw = await r.get(f"{_MDD_REDIS_KEY}:{user_id}")
                if raw:
                    state = json.loads(raw)
                    self.peak_balance = float(state.get("peak_balance", 0.0))
                    self.is_paused = bool(state.get("is_paused", False))
                    self.is_short_term_strategy_paused = bool(state.get("is_short_term_strategy_paused", False))
                    self.reduction_applied = state.get("reduction_applied")
                    self._reduction_executed_10pct = bool(state.get("_reduction_executed_10pct", False))
                    self._reduction_executed_20pct = bool(state.get("_reduction_executed_20pct", False))
                    self._restore_reduction_progress(state)
                    logger.info(
                        f"[MDD v2] Redis에서 복원: user={user_id} peak={self.peak_balance:,.0f}, "
                        f"paused={self.is_paused}, reduction={self.reduction_applied}"
                    )
                    return True
        except Exception as e:
            logger.warning(f"[MDD v2] Redis 복원 실패 (user={user_id}): {e}")
        return False

    # ── [항목7] MDD 갱신 헬스 보고/조회 ────────────────────────────────────────
    # 기존 시그니처 변경 없이 새 메서드로만 추가한다.

    async def record_update_failure(self, error: str) -> int:
        """MDD 갱신 실패를 기록하고 누적 연속 실패 횟수를 반환.

        갱신 경로(swing/scheduler)의 except 절에서 호출한다.
        여기서 누적된 횟수가 MDD_BLIND_FAILURE_THRESHOLD 이상이면
        can_open_new_position() 이 신규 진입을 차단한다(청산은 영향 없음).
        """
        async with self._lock:
            self._update_failures += 1
            self._update_last_error = str(error)
            self._update_empty_account = False
            return self._update_failures

    async def record_update_success(self, empty_account: bool = False) -> None:
        """MDD 갱신 성공(또는 '정상 응답 + 빈 계좌')을 기록해 실패 카운터를 리셋.

        empty_account=True 는 tot_evlu_amt<=0(미입금 계좌)로 갱신을 스킵한 경우다.
        장애가 아니므로 실패로 계수하지 않고, 진입 차단 사유도 되지 않는다.
        """
        async with self._lock:
            self._update_failures = 0
            self._update_last_error = None
            self._update_empty_account = bool(empty_account)
            self._blind_block_active = False
            self._blind_block_hits = 0
            self._update_last_ok_at = self._now_kst_v2()

    def _is_update_stale_v2(self) -> bool:
        """[R1] 장중 마지막 갱신 성공 후 MDD_UPDATE_STALE_SEC 초과 여부 (Lock 보유 중 호출, I/O 없음).

        예외 없이 swing 사이클이 지연·정지하면 실패 카운터가 늘지 않아 낡은 drawdown 으로
        진입이 허용되던 구멍을 막는다. 기준 시각은 max(마지막 성공, 오늘 장 시작 09:00) —
        야간/휴장 후 첫 사이클 전에 DT 진입이 막히지 않게 한다. 장외에는 판정하지 않는다.
        """
        stale_sec = _stale_sec_v2()
        if stale_sec <= 0:
            return False
        from app.core.market_schedule import is_market_open
        if not is_market_open():
            return False
        now = self._now_kst_v2()
        ref = self._update_last_ok_at or self._created_at
        open_dt = now.replace(hour=9, minute=0, second=0, microsecond=0)
        if open_dt > ref:
            ref = open_dt
        return (now - ref).total_seconds() > stale_sec

    def _is_update_blind(self) -> bool:
        """MDD 갱신이 '눈이 먼' 상태인지 판정 (Lock 보유 중 호출, DB/IO 없음).

        - 빈 계좌(empty_account)는 차단 사유 아님 — 자산 0이면 애초에 매수 불가이고
          입금 직후 첫 사이클을 막을 이유가 없다.
        - 설정으로 끌 수 있으나 기본값은 True(config.MDD_BLIND_BLOCK_ENTRY).
        """
        if not settings.MDD_BLIND_BLOCK_ENTRY:
            return False
        # 빈 계좌는 '실패 카운터' 조건만 면제한다. 정체(stale) 판정은 그대로 수행해
        # 빈 계좌 응답 이후 swing 사이클이 멈춘 경우에도 fail-safe 가 작동한다.
        if not self._update_empty_account:
            threshold = max(1, int(settings.MDD_BLIND_FAILURE_THRESHOLD))
            if self._update_failures >= threshold:
                return True
        return self._is_update_stale_v2()

    def get_update_health(self) -> dict:
        """MDD 갱신 헬스 스냅샷 (동기·읽기 전용). 차단 여부 관측용."""
        return {
            "consecutive_failures": self._update_failures,
            "last_error": self._update_last_error,
            "empty_account": self._update_empty_account,
            "blind_block_enabled": bool(settings.MDD_BLIND_BLOCK_ENTRY),
            "blind_block_threshold": int(settings.MDD_BLIND_FAILURE_THRESHOLD),
            "blind_block_active": self._blind_block_active,
            "blind_block_hits": self._blind_block_hits,
        }

    async def can_open_new_position(self) -> bool:
        """신규 포지션 진입 가능 여부 (thread-safe)

        [항목7 2026-09-27] MDD 갱신이 연속 실패 중이면(= -10%/-20% 게이트가 작동
        불가) 신규 진입을 차단한다. 모든 진입 경로가 이 메서드를 await 하므로
        여기서 한 번에 적용된다. 시그니처 불변, 청산 경로에는 영향 없음.
        """
        async with self._lock:
            blind = self._is_update_blind()
            self._blind_block_active = blind
            if blind:
                self._blind_block_hits += 1
                if self._blind_block_hits == 1 or self._blind_block_hits % 20 == 0:
                    logger.error(
                        f"[MDD] 🚨 신규 진입 차단(fail-safe): MDD 갱신 연속 실패 "
                        f"{self._update_failures}회 (임계 {settings.MDD_BLIND_FAILURE_THRESHOLD}) — "
                        f"peak/drawdown 미갱신으로 -10%/-20% 게이트가 작동하지 않음(또는 장중 갱신 정지 "
                        f"{_stale_sec_v2()}초 초과). "
                        f"마지막 오류: {self._update_last_error} "
                        f"(누적 차단 {self._blind_block_hits}회, 청산은 정상 동작)"
                    )
                return False
            return not self.is_paused

    async def is_short_term_strategy_enabled(self) -> bool:
        """단기 전략 활성화 여부 (thread-safe) — MDD -10% 이상일 때만 활성화"""
        async with self._lock:
            return not self.is_short_term_strategy_paused

    async def reset(self):
        """MDD 상태 리셋 (수동 복구 시) — async + Lock으로 동시성 안전 보장

        [R1 2026-10-09] 플래그만 지우고 peak 를 유지하면 낙폭이 남아 있는 한 다음
        update() 가 REDUCE_AND_PAUSE 를 즉시 재발동해 전 종목 30% 를 또 매도했다.
        따라서 수동 리셋은 **peak 를 마지막 관측 잔고로 재설정**(re-base)한다 — 이후 다시
        -10% 가 되어야 재발동한다. 마지막 잔고를 모르면(재기동 직후) 다음 갱신 잔고로
        재설정한다. fail-safe 차단 카운터/플래그도 함께 초기화한다.
        fail-safe 만 풀려면 reset_fail_safe_v2() 를 쓴다(MDD 상태 불변).
        """
        await self.reset_v2()

    async def reset_v2(self, rebase_peak: bool = True) -> dict:
        """reset() 본체. rebase_peak=False 면 구 동작(peak 유지 — 축소 재발동 가능)."""
        async with self._lock:
            self.is_paused = False
            self.is_short_term_strategy_paused = False
            self.reduction_applied = None
            self._reduction_executed_10pct = False
            self._reduction_executed_20pct = False
            self._clear_reduction_progress()
            old_peak = self.peak_balance
            if rebase_peak:
                if self._last_balance > 0:
                    self.peak_balance = self._last_balance
                    self._rebase_peak_pending = False
                else:
                    self._rebase_peak_pending = True
            was_blind = self._blind_block_active or self._update_failures > 0
            self._update_failures = 0
            self._update_last_error = None
            self._update_empty_account = False
            self._blind_block_active = False
            self._blind_block_hits = 0
            self._update_last_ok_at = self._now_kst_v2()
            logger.warning(
                f"MDD 관리자 리셋 — peak {old_peak:,.0f} -> {self.peak_balance:,.0f} "
                f"(rebase={rebase_peak}, pending={self._rebase_peak_pending})"
            )
            if was_blind:
                logger.info("[MDD] fail-safe 차단 해제 — 갱신 실패 카운터/차단 플래그 초기화됨")
            result = {
                "old_peak": old_peak, "peak_balance": self.peak_balance,
                "rebase_peak": bool(rebase_peak), "rebase_pending": self._rebase_peak_pending,
            }
            snap = self._state_snapshot_v2()
        await self._write_state_v2(_MDD_REDIS_KEY, snap)  # [L-6] 락 밖에서
        return result

    async def reset_fail_safe_v2(self) -> None:
        """fail-safe(갱신 실패/정지) 차단만 해제. peak/pause/축소 상태는 건드리지 않는다."""
        async with self._lock:
            self._update_failures = 0
            self._update_last_error = None
            self._update_empty_account = False
            self._blind_block_active = False
            self._blind_block_hits = 0
            self._update_last_ok_at = self._now_kst_v2()

    # ── [C-1 2026-10-02] 강제축소 진행 상태 헬퍼 (기존 시그니처 불변, 신규 메서드만) ──

    @staticmethod
    def _today_kst() -> str:
        from datetime import datetime
        from app.core.market_schedule import KST
        return datetime.now(KST).date().isoformat()

    def _clear_reduction_progress(self) -> None:
        """완료/시도/포기 집합 전부 초기화 (Lock 보유 중 호출). peak 갱신·reset() 전용.

        세대(epoch)를 올려, 초기화 이전에 시작된 축소 루프가 뒤늦게 완료를 기록해
        다음 낙폭 이벤트에서 그 종목을 '이미 축소됨'으로 건너뛰는 일을 막는다.
        """
        self._reduced_codes = {}
        self._reduction_attempts = {}
        self._reduction_gave_up = {}
        self._reduction_progress_date = None
        self._reduction_unconfirmed = {}
        self._reduction_epoch = getattr(self, "_reduction_epoch", 0) + 1

    def _reduction_progress_state(self) -> dict:
        """Redis 직렬화용 진행 상태 (save_to_redis / save_to_redis_v2 공용)."""
        return {
            "reduced_codes": {lv: sorted(c) for lv, c in self._reduced_codes.items()},
            "reduction_attempts": {lv: dict(a) for lv, a in self._reduction_attempts.items()},
            "reduction_gave_up": {lv: sorted(c) for lv, c in self._reduction_gave_up.items()},
            "reduction_progress_date": self._reduction_progress_date,
            "reduction_unconfirmed": {lv: {c: dict(v) for c, v in m.items()}
                                      for lv, m in self._reduction_unconfirmed.items()},
        }

    def _restore_reduction_progress(self, state: dict) -> None:
        """Redis 상태에서 진행 상태 복원. 키가 없으면(구버전 상태) 빈 집합.

        재기동 후 완료 집합이 없으면 실패 종목 재시도 시 이미 축소한 종목을 다시
        매도하게 되므로, 복원 실패는 warning 으로 남긴다.
        """
        try:
            self._reduced_codes = {
                str(lv): set(map(str, codes))
                for lv, codes in (state.get("reduced_codes") or {}).items()
            }
            self._reduction_attempts = {
                str(lv): {str(c): int(n) for c, n in (att or {}).items()}
                for lv, att in (state.get("reduction_attempts") or {}).items()
            }
            self._reduction_gave_up = {
                str(lv): set(map(str, codes))
                for lv, codes in (state.get("reduction_gave_up") or {}).items()
            }
            self._reduction_progress_date = state.get("reduction_progress_date")
            self._reduction_unconfirmed = {
                str(lv): {str(c): dict(v) for c, v in (m or {}).items()}
                for lv, m in (state.get("reduction_unconfirmed") or {}).items()
            }
        except Exception as e:
            logger.warning(f"[MDD] 축소 진행 상태 복원 실패 — 빈 상태로 시작: {e}")
            self._reduced_codes = {}
            self._reduction_attempts = {}
            self._reduction_gave_up = {}
            self._reduction_progress_date = None

    def _roll_reduction_day_locked(self) -> None:
        """KST 날짜가 바뀌면 시도 횟수·포기 집합만 초기화 (Lock 보유 중 호출).

        완료 집합은 날짜로 초기화하지 않는다 — 낙폭이 이어지는 동안 날짜 리셋으로
        완료 종목을 다시 축소하면 C-1 누적 매도가 하루 주기로 재발한다.
        """
        today = self._today_kst()
        if self._reduction_progress_date != today:
            if self._reduction_attempts or self._reduction_gave_up:
                logger.warning(
                    f"[MDD] 날짜 변경({self._reduction_progress_date} → {today}) — "
                    f"축소 재시도 횟수·당일 포기 종목 초기화 "
                    f"(포기 종목: { {lv: sorted(c) for lv, c in self._reduction_gave_up.items()} })"
                )
            # 포기 종목이 있던 레벨은 executed 를 내려 새 날짜에 재시도되게 한다
            # (구버전 상태 복원 등으로 executed=True 와 포기 종목이 함께 남은 경우 방어).
            for _lv, _codes in self._reduction_gave_up.items():
                if _codes:
                    setattr(self, f"_reduction_executed_{_lv}", False)
            self._reduction_attempts = {}
            self._reduction_gave_up = {}
            self._reduction_progress_date = today

    def _try_begin_reduction_locked(self, action: str, log_prefix: str) -> Optional[str]:
        """축소 시작 판정 (Lock 보유 중 호출). 시작하면 레벨, 아니면 None.

        날짜 롤오버를 executed 체크보다 **먼저** 수행한다 — 포기 종목만 남은 레벨도
        새 날짜에는 재시도된다. 동시 실행은 인메모리 `_reduction_running` 으로만 막는다.
        """
        level = self.reduction_applied
        if level is None:
            logger.debug(f"{log_prefix} reduction_applied=None — 축소 불필요, 스킵")
            return None
        self._roll_reduction_day_locked()
        if level in self._reduction_running:
            logger.debug(f"{log_prefix} {action} 축소 진행 중 — 스킵")
            return None
        if getattr(self, f"_reduction_executed_{level}", False):
            logger.debug(f"{log_prefix} {action} 축소 이미 완료 — 스킵")
            return None
        self._reduction_running.add(level)
        return level

    async def _end_reduction_running(self, level: Optional[str]) -> None:
        if level is None:
            return
        async with self._lock:
            self._reduction_running.discard(level)

    async def _mark_reduction_unconfirmed_v2(
        self, level: str, epoch: int, stock_code: str, qty_before: int, sell_qty: int
    ) -> None:
        async with self._lock:
            if self._reduction_epoch != epoch:
                return
            self._reduction_unconfirmed.setdefault(level, {})[stock_code] = {
                "qty_before": int(qty_before), "sell_qty": int(sell_qty), "cycles": 0,
            }

    async def _resolve_unconfirmed_reduction_v2(
        self, level: str, epoch: int, h: dict, log_prefix: str
    ) -> Optional[str]:
        """[M-1 5회차] 결과 불명 주문의 판정 — None(대상 아님)/'sold'/'wait'/'not_sent'/'gave_up'.

        - 보유 < 주문 전 보유 → 'sold' (완료 처리; 판 수량은 _verdict_qty 로 조회)
        - 주문가능 < 보유 → 미체결 매도 주문이 살아 있음 → 'wait'
        - 주문가능 == 보유 == 주문 전 보유 → 전송 안 됨 → 'not_sent'(정상 재주문)
        - 주문가능 수량 필드가 없어 판정 불가 → 'wait', 연속 MDD_REDUCTION_MAX_RETRIES 주기면
          당일 포기 + error(수동 확인). 재매도는 하지 않는다.
        """
        code = h.get("pdno", "")
        rec = (self._reduction_unconfirmed.get(level) or {}).get(code)
        if rec is None:
            return None
        qty_now = int(h.get("hldg_qty", 0) or 0)
        _psbl_raw = h.get("ord_psbl_qty")
        ord_psbl = int(float(_psbl_raw)) if _psbl_raw not in (None, "") else None
        before = int(rec.get("qty_before", 0))
        async with self._lock:
            if self._reduction_epoch != epoch:
                return "wait"
            if qty_now < before:
                self._reduction_unconfirmed[level].pop(code, None)
                rec["sold_qty"] = before - qty_now
                self._reduction_unconfirmed.setdefault("_sold_" + level, {})[code] = rec
                self._reduced_codes.setdefault(level, set()).add(code)
                verdict = "sold"
            elif ord_psbl is not None and ord_psbl < qty_now:
                rec["cycles"] = int(rec.get("cycles", 0)) + 1
                verdict = "wait"
            elif ord_psbl is not None:
                self._reduction_unconfirmed[level].pop(code, None)
                verdict = "not_sent"
            else:
                rec["cycles"] = int(rec.get("cycles", 0)) + 1
                cap = max(1, int(settings.MDD_REDUCTION_MAX_RETRIES))
                if rec["cycles"] >= cap:
                    self._reduction_unconfirmed[level].pop(code, None)
                    self._reduction_gave_up.setdefault(level, set()).add(code)
                    verdict = "gave_up"
                else:
                    verdict = "wait"
        if verdict == "sold":
            logger.warning(f"{log_prefix} {code} 결과 불명 축소 주문 체결 확인(잔고 {before}→{qty_now}주) — 완료 처리")
        elif verdict == "not_sent":
            logger.warning(f"{log_prefix} {code} 결과 불명 축소 주문 미접수 확인(잔고·주문가능 불변) — 재주문")
        elif verdict == "gave_up":
            logger.error(
                f"{log_prefix} 🚨 {code} 결과 불명 축소 주문을 {rec['cycles']}주기 동안 확인 불가"
                f"(주문가능 수량 미제공) — 당일 재매도 안 함, 수동 확인 필요"
            )
        else:
            logger.warning(f"{log_prefix} {code} 결과 불명 축소 주문 확인 대기({rec['cycles']}주기) — 재매도 안 함")
        return verdict

    async def _begin_reduction_progress(self, level: str) -> tuple:
        """축소 루프 시작 시 (건너뛸 종목 집합, 세대) 스냅샷. Lock 보유 중 호출 금지."""
        async with self._lock:
            self._roll_reduction_day_locked()
            settled = set(self._reduced_codes.get(level, ())) | set(
                self._reduction_gave_up.get(level, ())
            )
            return settled, getattr(self, "_reduction_epoch", 0)

    async def _mark_reduction_done(self, level: str, stock_code: str, epoch: int) -> None:
        async with self._lock:
            if getattr(self, "_reduction_epoch", 0) != epoch:
                return  # 루프 도중 peak 갱신/reset — 새 세대에 옛 완료를 남기지 않는다
            self._reduced_codes.setdefault(level, set()).add(stock_code)
            self._reduction_attempts.get(level, {}).pop(stock_code, None)

    async def _mark_reduction_failed(
        self, level: str, stock_code: str, epoch: int, reason: str, log_prefix: str
    ) -> bool:
        """실패 1회 기록. 상한 도달 시 당일 포기 집합에 넣고 True 반환."""
        max_retries = max(1, int(settings.MDD_REDUCTION_MAX_RETRIES))
        async with self._lock:
            if getattr(self, "_reduction_epoch", 0) != epoch:
                return False
            att = self._reduction_attempts.setdefault(level, {})
            att[stock_code] = att.get(stock_code, 0) + 1
            n = att[stock_code]
            gave_up = n >= max_retries
            if gave_up:
                self._reduction_gave_up.setdefault(level, set()).add(stock_code)
        if gave_up:
            logger.error(
                f"{log_prefix} 🚨 강제축소 {stock_code} {n}회 연속 실패(상한 {max_retries}) — "
                f"당일 재시도 중단, 수동 확인 필요. 마지막 사유: {reason}"
            )
        else:
            logger.warning(
                f"{log_prefix} 강제축소 {stock_code} 실패 {n}/{max_retries}회 — "
                f"다음 주기에 이 종목만 재시도. 사유: {reason}"
            )
        return gave_up

    async def _finish_reduction_progress(
        self,
        executed_key: str,
        action: str,
        results: list,
        pending: list,
        log_prefix: str,
        level: Optional[str] = None,
        epoch: Optional[int] = None,
    ) -> None:
        """루프 종료 처리.

        executed 플래그는 **재시도 대상(실패·보류)도, 당일 포기 종목도 없을 때만** True.
        - pending 있음 → 다음 주기에 그 종목만 재시도(완료 종목은 settled 로 건너뜀)
        - 포기 종목만 남음 → executed=False 유지. 당일에는 settled 로 skip(주문 없음),
          KST 날짜가 바뀌면 시도 횟수가 초기화돼 재시도된다.
        """
        lv = level or self.reduction_applied or ""
        async with self._lock:
            if epoch is not None and epoch != self._reduction_epoch:
                # [M-1 4회차] 루프 도중 reset()/peak 갱신 — 옛 세대의 결과로 executed 를
                # 세우면 다음 -10% 이벤트의 축소가 생략된다. 상태를 건드리지 않는다.
                logger.warning(
                    f"{log_prefix} {action} 진행 중 MDD 상태가 리셋됨(세대 {epoch}→{self._reduction_epoch}) "
                    f"— 이번 결과로 executed/진행 상태를 갱신하지 않음 (이번 주기 {len(results)}종목 매도)"
                )
                return
            gave_up = sorted(self._reduction_gave_up.get(lv, ()))
            done = not pending and not gave_up
            setattr(self, executed_key, bool(done))
        if pending:
            logger.warning(
                f"{log_prefix} {action} 부분 완료: 이번 주기 {len(results)}종목 성공, "
                f"재시도 대상 {len(pending)}종목 {pending} — 완료 종목은 재매도하지 않음"
            )
        elif gave_up:
            warn_throttled(
                logger, f"mdd:gave_up:{lv}",
                "%s 🚨 %s 미완료 — 이번 주기 %d종목 성공, 당일 포기 %s "
                "(KST 날짜 변경 시 자동 재시도, 즉시 필요하면 수동 매도)",
                log_prefix, action, len(results), gave_up,
            )
        else:
            logger.warning(
                f"{log_prefix} {action} 레벨 축소 완료 (이번 주기 {len(results)}종목 매도, "
                f"누적 완료 {len(self._reduced_codes.get(lv, ()))}종목)"
            )


    async def execute_reduction(self, mdd_action: dict) -> list[dict]:
        """MDD 액션에 따라 포지션 축소 매도 실행.

        CLAUDE.md 5.3:
        - REDUCE_AND_PAUSE (-10%): 기존 포지션 30% 축소
        - FORCE_REDUCE (-20%): 50% 강제 축소, 레버리지/인버스 ETF 우선

        중복 실행 방지: 동일 reduction_applied 레벨에서 1회만 실행.

        [C-1 2026-10-02] 재시도 정책:
        - 종목 단위 완료 집합(`_reduced_codes[level]`)으로 이미 축소한 종목은 같은
          레벨에서 다시 매도하지 않는다. 일부 종목 실패 시 다음 주기에는 **실패 종목만**
          재시도한다(기존: 전 보유종목 재매도 → 성공 종목 누적 축소).
        - 완료 기록은 주문 접수(rt_cd=="0") 직후, 부기(TradeHistory·ActiveTrade)보다 먼저
          남긴다 — 부기 예외가 재매도를 유발하지 않게.
        - `order_lock_busy`(다른 경로가 같은 종목 주문 중)는 실패가 아닌 보류로 재시도.
        - 실패는 종목별로 계수해 `MDD_REDUCTION_MAX_RETRIES` 도달 시 당일 포기 + error.
        - 완료 집합은 Redis(`mdd:state`)에 영속 → 재기동 후에도 재축소하지 않는다.
          peak 갱신·reset() 시 초기화(§5.3 플래그 리셋과 동일), KST 날짜 변경 시에는
          시도 횟수·포기 집합만 초기화한다.
        - 수량 정책(불변): `max(int(보유 × 비율), 1)` — 1주 보유 종목은 30%/50% 단계에서도
          1주 전량 매도된다(0주 축소로 리스크가 남는 것보다 우선).
        """
        action = mdd_action.get("action", "NONE")
        if action == "NONE":
            return []

        reduce_pct = mdd_action.get("reduce_pct", 0.0)
        prioritize_leverage = mdd_action.get("prioritize_leverage", False)

        if reduce_pct <= 0:
            return []

        # [BUG FIX] executed_key를 Lock 밖(finally)에서도 참조하므로
        # Lock 진입 전에 None으로 초기화
        executed_key: Optional[str] = None

        # 중복 실행 방지: 이미 해당 레벨의 축소가 실행된 경우 스킵
        # [이슈2] reduction_applied=None 이면 축소 불필요(early return).
        # [3회차] 날짜 롤오버 → executed 체크 → 인메모리 진행 가드 순(_try_begin_reduction_locked)
        if not _reduction_market_open_v2("[MDD]", action):
            return []
        async with self._lock:
            level = self._try_begin_reduction_locked(action, "[MDD]")
        if level is None:
            return []
        executed_key = f"_reduction_executed_{level}"
        try:
            return await self._execute_reduction_impl(
                action, reduce_pct, prioritize_leverage, level, executed_key
            )
        finally:
            await self._end_reduction_running(level)

    async def _execute_reduction_impl(
        self, action: str, reduce_pct: float, prioritize_leverage: bool,
        level: str, executed_key: str,
    ) -> list[dict]:

        try:
            from app.services.korea_investment_api import korea_investment_api
            from app.services.db_worker import enqueue_trade_history

            # 현재 보유 종목 조회
            # [하드닝 R9 2026-10-12] 연속조회(get_balance_all_v2) — 첫 페이지 밖 보유 종목도 축소 대상(§8.5-37)
            balance = await korea_investment_api.get_balance_all_v2()
            # [C-2 3회차] 오류 응답(rt_cd≠0: EGW00201·40910000 등)은 output1 이 없다 —
            # '보유 없음'으로 처리하면 executed 가 서고 이 레벨 축소가 영구 생략됐다.
            if not _balance_response_ok(balance):
                logger.warning(
                    f"[MDD] {action} 잔고 조회 실패 응답 — 축소 보류, 다음 주기 재시도 "
                    f"(rt_cd={_rt_cd(balance)}, msg={_msg(balance)})"
                )
                return []
            holdings = balance.get("output1") or []
            settled, epoch = await self._begin_reduction_progress(level)
            if not holdings:
                logger.info(f"[MDD] {action} — 보유 종목 없음(정상 응답), 축소 불필요")
                await self._finish_reduction_progress(
                    executed_key, action, [], [], "[MDD]", level=level, epoch=epoch
                )
                await self.save_to_redis()
                return []

            # 레버리지/인버스 ETF 우선 정렬 (FORCE_REDUCE 시)
            LEVERAGE_KEYWORDS = {"레버리지", "2X", "3X", "인버스", "TIGER 200선물", "KODEX 인버스", "곱버스"}
            if prioritize_leverage:
                def _is_leverage(h):
                    name = h.get("prdt_name", "")
                    return any(kw in name for kw in LEVERAGE_KEYWORDS)
                holdings = sorted(holdings, key=lambda h: (not _is_leverage(h),))

            results = []
            skipped = 0
            pending: list[str] = []  # 다음 주기 재시도 대상 (실패·보류)
            if _balance_incomplete_v2(balance, holdings):
                # [R2 LOW] 첫 페이지만 조회 — 응답 밖 종목은 축소 대상에서 빠진다. 보이는 종목은
                # 축소하되 executed 를 세우지 않아 완료로 오인되지 않게 한다.
                logger.error(
                    f"[MDD] {action} 잔고 응답이 불완전({len(holdings)}행, "
                    f"_partial={balance.get('_partial')}) — 응답 밖 보유 종목은 "
                    f"축소되지 않았을 수 있음(수동 확인 필요)"
                )
                pending.append(_PAGE_FULL_SENTINEL)
            for h in holdings:
                # [M-1 4회차] 루프 도중 reset()/peak 갱신이면 남은 종목 매도를 중단한다.
                if self._reduction_epoch != epoch:
                    logger.warning(f"[MDD] {action} 진행 중 MDD 상태 리셋 감지 — 남은 종목 축소 중단")
                    break
                stock_code = h.get("pdno", "")
                stock_name = h.get("prdt_name", stock_code)
                quantity = int(h.get("hldg_qty", 0) or 0)
                current_price = float(h.get("prpr", 0) or 0)

                if quantity <= 0 or current_price <= 0:
                    skipped += 1
                    continue
                if stock_code in settled:
                    # [C-1] 이 레벨에서 이미 축소 완료(또는 당일 포기) — 재매도 금지
                    skipped += 1
                    continue
                # [M-1 5회차] 직전 주기 '전송 후 결과 불명' 주문 — 재매도 전에 잔고로 확인
                _verdict = await self._resolve_unconfirmed_reduction_v2(
                    level, epoch, h, "[MDD]"
                )
                if _verdict == "sold":
                    _uc_sold = int(_verdict_qty(h, self, level, stock_code))
                    await _sync_active_trade_after_reduction(
                        stock_code=stock_code, quantity=quantity + _uc_sold,
                        remaining_qty=quantity,
                    )
                    await _reflect_reduction(stock_code, _uc_sold, current_price, quantity)
                    skipped += 1
                    continue
                if _verdict == "wait":
                    pending.append(stock_code)
                    continue
                if _verdict == "gave_up":
                    skipped += 1
                    continue
                _ordered = False

                sell_qty = max(int(quantity * reduce_pct), 1)
                sell_qty = min(sell_qty, quantity)

                try:
                    from app.services.trade_executor import safe_place_order
                    order_resp = await safe_place_order(
                        stock_code=stock_code,
                        order_type="SELL",
                        quantity=sell_qty,
                        price=None,  # 시장가
                    )
                    if order_resp and order_resp.get("rt_cd") == "0":
                        # [C-1] 완료 기록을 부기보다 먼저 — 이후 예외가 재매도를 유발하지 않게
                        _ordered = True
                        await self._mark_reduction_done(level, stock_code, epoch)
                        await self.save_to_redis()
                        order_id = order_resp.get("output", {}).get("ODNO", "")
                        avg_price = float(h.get("pchs_avg_pric", current_price))
                        _pnl = _reduction_net_pnl_v2(avg_price, current_price, sell_qty)

                        # MDD 강제매도는 시스템적(systemic) 리스크 축소이며,
                        # 개별 종목의 전략 실패가 아니므로 loss_tracker에 기록하지 않음.
                        # 기록 시 양호한 종목이 MDD 축소로 인해 false ban될 수 있음.

                        await enqueue_trade_history({
                            "stock_code": stock_code,
                            "stock_name": stock_name,
                            "order_type": "SELL",
                            "order_price": current_price,
                            "order_quantity": sell_qty,
                            # [2026-10-08] 시장가 — 체결가는 swing/fill_sweep 체결조회가 채운다(그 전엔 NULL).
                            # total_amount·net_profit_loss 는 현재가 기준 추정치.
                            "executed_price": None,
                            "executed_quantity": sell_qty,
                            "total_amount": current_price * sell_qty,
                            "status": "COMPLETED",
                            "order_id": order_id,
                            "exit_reason": f"MDD {action} ({reduce_pct*100:.0f}% 축소)",
                            "net_profit_loss": _pnl["net_profit_loss"],
                            "commission": _pnl["commission"],
                            "profit_loss_rate": _pnl["profit_loss_rate"],
                            "strategy_type": "mdd_reduction",
                        })

                        # [FIX] 전량 매도 시 ActiveTrade 닫기 / 부분 매도 시 잔량 갱신
                        # [M8 2026-09-27] 전략명을 비우면 db_worker 가 '가장 최근 active 행'을
                        # 매칭하므로 같은 종목에 DT 행이 있으면 DT 행이 닫히고 스윙 행이 영구 잔존한다.
                        # → DT 인메모리 포지션이 있는 종목은 전략명을 명시해 처리한다.
                        remaining_qty = quantity - sell_qty
                        await _sync_active_trade_after_reduction(
                            stock_code=stock_code,
                            quantity=quantity,
                            remaining_qty=remaining_qty,
                        )

                        # [2026-09-23] 축소 체결분을 Portfolio·DT 인메모리에도 반영.
                        # 누락 시 다음 청산이 축소 전 수량으로 매도한다(과다매도).
                        await _reflect_reduction(stock_code, sell_qty, current_price, remaining_qty)

                        results.append({
                            "stock_code": stock_code,
                            "stock_name": stock_name,
                            "sell_quantity": sell_qty,
                            "reason": action,
                        })
                        logger.warning(
                            f"[MDD] {action} 매도: {stock_code} {stock_name} "
                            f"{sell_qty}주 (총 {quantity}주 중 {reduce_pct*100:.0f}%)"
                        )

                    elif _is_order_lock_busy(order_resp):
                        # [C-1] 다른 경로가 같은 종목을 주문 중 — 실패로 세지 않고 보류
                        pending.append(stock_code)
                        logger.warning(
                            f"[MDD] {action} {stock_code} 주문 락 점유 중(order_lock_busy) — "
                            f"보류, 다음 주기 재시도"
                        )
                    elif _is_transient_order_failure_v2(order_resp):
                        # [M-1 5회차] 전송 전 실패·레이트리밋·토큰 — 일시 장애. 재시도 횟수 미산정
                        # (토큰 장애 5분으로 축소를 당일 포기하지 않도록).
                        pending.append(stock_code)
                        logger.warning(
                            f"[MDD] {action} {stock_code} 일시 실패 "
                            f"(msg_cd={(order_resp or {}).get('msg_cd')}) — 미산정, 다음 주기 재시도"
                        )
                    elif _is_order_outcome_unknown_mdd_v2(order_resp):
                        # [M-1 5회차] 전송 후 결과 불명 — 팔렸을 수 있다. 다음 주기에 잔고로 확인할
                        # 때까지 재매도하지 않는다(누적 축소 방지).
                        await self._mark_reduction_unconfirmed_v2(level, epoch, stock_code, quantity, sell_qty)
                        pending.append(stock_code)
                        logger.error(
                            f"[MDD] {action} {stock_code} {sell_qty}주 주문 결과 불명 "
                            f"(msg_cd={(order_resp or {}).get('msg_cd')}) — 다음 주기 잔고 확인 전 재매도 안 함"
                        )
                    else:
                        _reason = (order_resp or {}).get("msg1") or "no_response"
                        if not await self._mark_reduction_failed(
                            level, stock_code, epoch, str(_reason), "[MDD]"
                        ):
                            pending.append(stock_code)
                    await asyncio.sleep(0.3)  # KIS API 레이트 리밋

                except Exception as e:
                    logger.error(f"[MDD] 축소 매도 실패 ({stock_code}): {e}")
                    if not _ordered:
                        if not await self._mark_reduction_failed(
                            level, stock_code, epoch, str(e), "[MDD]"
                        ):
                            pending.append(stock_code)

            # [C-1] 실패·보류 종목이 남으면 플래그를 내려 그 종목만 재시도
            await self._finish_reduction_progress(
                executed_key, action, results, pending, "[MDD]", level=level, epoch=epoch
            )
            await self.save_to_redis()
            return results

        except Exception as e:
            logger.error(f"[MDD] execute_reduction 실패: {e}", exc_info=True)
            # executed 는 완료 시에만 True — 여기서는 그대로 False(다음 주기 재시도)
            return []

    async def execute_reduction_v2(self, mdd_action: dict, kis, user_id: int) -> list[dict]:
        """[v2] 사용자 스코프 포지션 축소 매도 실행.

        execute_reduction()과 동일 로직이나 전역 korea_investment_api 대신 주입된
        kis(사용자 인스턴스)를 사용하고, trade_history/active_trade 큐 payload에
        user_id를 포함해 db_worker가 해당 사용자 레코드만 갱신하도록 한다.
        기존 execute_reduction()은 완전히 불변.
        """
        action = mdd_action.get("action", "NONE")
        if action == "NONE":
            return []

        reduce_pct = mdd_action.get("reduce_pct", 0.0)
        prioritize_leverage = mdd_action.get("prioritize_leverage", False)

        if reduce_pct <= 0:
            return []

        executed_key: Optional[str] = None

        if not _reduction_market_open_v2(f"[MDD v2] user={user_id}", action):
            return []
        async with self._lock:
            level = self._try_begin_reduction_locked(action, f"[MDD v2] user={user_id}")
        if level is None:
            return []
        executed_key = f"_reduction_executed_{level}"
        try:
            return await self._execute_reduction_v2_impl(
                action, reduce_pct, prioritize_leverage, level, executed_key, kis, user_id
            )
        finally:
            await self._end_reduction_running(level)

    async def _execute_reduction_v2_impl(
        self, action: str, reduce_pct: float, prioritize_leverage: bool,
        level: str, executed_key: str, kis, user_id: int,
    ) -> list[dict]:

        try:
            from app.services.db_worker import (
                enqueue_trade_history,
                enqueue_close_active_trade_v2,
                enqueue_active_trade_upsert,
            )

            # [하드닝 R9 2026-10-12] 연속조회(get_balance_all_v2) — §8.5-37
            balance = await kis.get_balance_all_v2()
            if not _balance_response_ok(balance):
                logger.warning(
                    f"[MDD v2] user={user_id} {action} 잔고 조회 실패 응답 — 축소 보류, 다음 주기 재시도 "
                    f"(rt_cd={_rt_cd(balance)}, msg={_msg(balance)})"
                )
                return []
            holdings = balance.get("output1") or []
            settled, epoch = await self._begin_reduction_progress(level)
            if not holdings:
                logger.info(f"[MDD v2] user={user_id} {action} — 보유 종목 없음(정상 응답), 축소 불필요")
                await self._finish_reduction_progress(
                    executed_key, action, [], [], f"[MDD v2] user={user_id}", level=level, epoch=epoch
                )
                await self.save_to_redis_v2(user_id)
                return []

            LEVERAGE_KEYWORDS = {"레버리지", "2X", "3X", "인버스", "TIGER 200선물", "KODEX 인버스", "곱버스"}
            if prioritize_leverage:
                def _is_leverage(h):
                    name = h.get("prdt_name", "")
                    return any(kw in name for kw in LEVERAGE_KEYWORDS)
                holdings = sorted(holdings, key=lambda h: (not _is_leverage(h),))

            results = []
            skipped = 0
            pending: list[str] = []  # 다음 주기 재시도 대상 (실패·보류)
            if _balance_incomplete_v2(balance, holdings):
                logger.error(
                    f"[MDD v2] user={user_id} {action} 잔고 응답이 불완전({len(holdings)}행, _partial={balance.get('_partial')}) — "
                    f"응답 밖 보유 종목은 축소되지 않았을 수 있음(수동 확인 필요)"
                )
                pending.append(_PAGE_FULL_SENTINEL)
            for h in holdings:
                if self._reduction_epoch != epoch:
                    logger.warning(f"[MDD v2] user={user_id} {action} 진행 중 MDD 상태 리셋 감지 — 남은 종목 축소 중단")
                    break
                stock_code = h.get("pdno", "")
                stock_name = h.get("prdt_name", stock_code)
                quantity = int(h.get("hldg_qty", 0) or 0)
                current_price = float(h.get("prpr", 0) or 0)

                if quantity <= 0 or current_price <= 0:
                    skipped += 1
                    continue
                if stock_code in settled:
                    # [C-1] 이 레벨에서 이미 축소 완료(또는 당일 포기) — 재매도 금지
                    skipped += 1
                    continue
                # [M-1 5회차] 직전 주기 '전송 후 결과 불명' 주문 — 재매도 전에 잔고로 확인
                _verdict = await self._resolve_unconfirmed_reduction_v2(
                    level, epoch, h, f"[MDD v2] user={user_id}"
                )
                if _verdict == "sold":
                    _uc_sold = int(_verdict_qty(h, self, level, stock_code))
                    await _sync_active_trade_after_reduction(
                        stock_code=stock_code, quantity=quantity + _uc_sold,
                        remaining_qty=quantity, user_id=user_id,
                    )
                    await _reflect_reduction(stock_code, _uc_sold, current_price, quantity)
                    skipped += 1
                    continue
                if _verdict == "wait":
                    pending.append(stock_code)
                    continue
                if _verdict == "gave_up":
                    skipped += 1
                    continue
                _ordered = False

                sell_qty = max(int(quantity * reduce_pct), 1)
                sell_qty = min(sell_qty, quantity)

                try:
                    # UserContext 조립 완료(트레이딩팀) — 종목별 동시 주문 락
                    # (_order_semaphores/Redis 분산락)을 거치도록 safe_place_order_v2 사용.
                    # SL/사이징 로직은 그대로, 주문 호출 경로만 교체.
                    from app.services.tenant.context import UserContext
                    from app.services.trade_executor import safe_place_order_v2

                    from app.core.config import settings as _settings

                    reduction_ctx = UserContext(
                        user_id=user_id,
                        username="",
                        mode=getattr(kis, "mode", _settings.KIS_MODE),
                        kis=kis,
                        mdd=self,
                        loss_tracker=None,
                    )
                    order_resp = await safe_place_order_v2(
                        reduction_ctx,
                        stock_code=stock_code,
                        order_type="SELL",
                        quantity=sell_qty,
                        price=None,  # 시장가
                    )
                    if order_resp and order_resp.get("rt_cd") == "0":
                        # [C-1] 완료 기록을 부기보다 먼저 — 이후 예외가 재매도를 유발하지 않게
                        _ordered = True
                        await self._mark_reduction_done(level, stock_code, epoch)
                        await self.save_to_redis_v2(user_id)
                        order_id = order_resp.get("output", {}).get("ODNO", "")
                        avg_price = float(h.get("pchs_avg_pric", current_price))
                        _pnl = _reduction_net_pnl_v2(avg_price, current_price, sell_qty)

                        # MDD 강제매도는 시스템적 리스크 축소 — loss_tracker 미기록 (execute_reduction과 동일 정책)
                        await enqueue_trade_history({
                            "user_id": user_id,
                            "stock_code": stock_code,
                            "stock_name": stock_name,
                            "order_type": "SELL",
                            "order_price": current_price,
                            "order_quantity": sell_qty,
                            # [2026-10-08] 시장가 — 체결가는 swing/fill_sweep 체결조회가 채운다(그 전엔 NULL).
                            # total_amount·net_profit_loss 는 현재가 기준 추정치.
                            "executed_price": None,
                            "executed_quantity": sell_qty,
                            "total_amount": current_price * sell_qty,
                            "status": "COMPLETED",
                            "order_id": order_id,
                            "exit_reason": f"MDD {action} ({reduce_pct*100:.0f}% 축소)",
                            "net_profit_loss": _pnl["net_profit_loss"],
                            "commission": _pnl["commission"],
                            "profit_loss_rate": _pnl["profit_loss_rate"],
                            "strategy_type": "mdd_reduction",
                        })

                        # [M8 2026-09-27] v1 과 동일 — DT 행 오폐기 방지 (전략명 명시)
                        remaining_qty = quantity - sell_qty
                        await _sync_active_trade_after_reduction(
                            stock_code=stock_code,
                            quantity=quantity,
                            remaining_qty=remaining_qty,
                            user_id=user_id,
                        )

                        # [2026-09-23] 축소 체결분을 Portfolio·DT 인메모리에도 반영.
                        # 누락 시 다음 청산이 축소 전 수량으로 매도한다(과다매도).
                        await _reflect_reduction(stock_code, sell_qty, current_price, remaining_qty)

                        results.append({
                            "stock_code": stock_code,
                            "stock_name": stock_name,
                            "sell_quantity": sell_qty,
                            "reason": action,
                        })
                        logger.warning(
                            f"[MDD v2] user={user_id} {action} 매도: {stock_code} {stock_name} "
                            f"{sell_qty}주 (총 {quantity}주 중 {reduce_pct*100:.0f}%)"
                        )

                    elif _is_order_lock_busy(order_resp):
                        # [C-1] 다른 경로가 같은 종목을 주문 중 — 실패로 세지 않고 보류
                        pending.append(stock_code)
                        logger.warning(
                            f"[MDD v2] user={user_id} {action} {stock_code} 주문 락 점유 중(order_lock_busy) — "
                            f"보류, 다음 주기 재시도"
                        )
                    elif _is_transient_order_failure_v2(order_resp):
                        # [M-1 5회차] 전송 전 실패·레이트리밋·토큰 — 일시 장애. 재시도 횟수 미산정
                        # (토큰 장애 5분으로 축소를 당일 포기하지 않도록).
                        pending.append(stock_code)
                        logger.warning(
                            f"[MDD v2] user={user_id} {action} {stock_code} 일시 실패 "
                            f"(msg_cd={(order_resp or {}).get('msg_cd')}) — 미산정, 다음 주기 재시도"
                        )
                    elif _is_order_outcome_unknown_mdd_v2(order_resp):
                        # [M-1 5회차] 전송 후 결과 불명 — 팔렸을 수 있다. 다음 주기에 잔고로 확인할
                        # 때까지 재매도하지 않는다(누적 축소 방지).
                        await self._mark_reduction_unconfirmed_v2(level, epoch, stock_code, quantity, sell_qty)
                        pending.append(stock_code)
                        logger.error(
                            f"[MDD v2] user={user_id} {action} {stock_code} {sell_qty}주 주문 결과 불명 "
                            f"(msg_cd={(order_resp or {}).get('msg_cd')}) — 다음 주기 잔고 확인 전 재매도 안 함"
                        )
                    else:
                        _reason = (order_resp or {}).get("msg1") or "no_response"
                        if not await self._mark_reduction_failed(
                            level, stock_code, epoch, str(_reason), f"[MDD v2] user={user_id}"
                        ):
                            pending.append(stock_code)
                    await asyncio.sleep(0.3)  # KIS API 레이트 리밋

                except Exception as e:
                    logger.error(f"[MDD v2] user={user_id} 축소 매도 실패 ({stock_code}): {e}")
                    if not _ordered:
                        if not await self._mark_reduction_failed(
                            level, stock_code, epoch, str(e), f"[MDD v2] user={user_id}"
                        ):
                            pending.append(stock_code)

            # [C-1] 실패·보류 종목이 남으면 플래그를 내려 그 종목만 재시도
            await self._finish_reduction_progress(
                executed_key, action, results, pending, f"[MDD v2] user={user_id}", level=level, epoch=epoch
            )
            await self.save_to_redis_v2(user_id)
            return results

        except Exception as e:
            logger.error(f"[MDD v2] user={user_id} execute_reduction_v2 실패: {e}", exc_info=True)
            return []


def _reduction_market_open_v2(log_prefix: str, action: str) -> bool:
    """[L-1 4회차] 강제축소 주문은 정규장(휴장일·주말·장외 제외)에만. 판정·상태는 유지된다.

    swing/scheduler 도 장 미오픈이면 사이클을 건너뛰지만, 사이클이 15:30 을 넘기거나
    다른 호출부(v2)가 부를 수 있으므로 주문 직전에 한 번 더 확인한다.
    """
    from app.core.market_schedule import is_market_open
    if is_market_open():
        return True
    warn_throttled(
        logger, "mdd:reduction_market_closed",
        "%s %s 강제축소 보류 — 장 운영시간 아님(휴장일/주말/장외). 개장 후 재시도", log_prefix, action,
    )
    return False


def _verdict_qty(h: dict, mgr, level: str, code: str) -> int:
    """'sold' 판정 직후 실제 감소 수량 (판정 시 기록, 1회 소비)."""
    rec = (mgr._reduction_unconfirmed.get("_sold_" + level) or {}).pop(code, None) or {}
    return int(rec.get("sold_qty", 0))


def _is_transient_order_failure_v2(resp) -> bool:
    """전송 전 실패·레이트리밋·토큰 무효 — trade_executor 의 재전송 안전 판정과 같은 정의."""
    from app.services.trade_executor import _is_retryable_order_failure
    return isinstance(resp, dict) and resp.get("rt_cd") != "0" and _is_retryable_order_failure(resp)


def _is_order_outcome_unknown_mdd_v2(resp) -> bool:
    from app.services.trade_executor import _is_order_outcome_unknown_v2
    return _is_order_outcome_unknown_v2(resp)


def _rt_cd(resp):
    return resp.get("rt_cd") if isinstance(resp, dict) else None


def _msg(resp):
    return (resp.get("msg_cd"), resp.get("msg1")) if isinstance(resp, dict) else repr(resp)


def _balance_response_ok(balance) -> bool:
    """잔고 응답이 '정상'인지. rt_cd 가 있고 '0' 이 아니거나 output1 이 list 가 아니면 False."""
    if not isinstance(balance, dict):
        return False
    if _rt_cd(balance) not in (None, "0"):
        return False
    return isinstance(balance.get("output1"), list)


def _is_order_lock_busy(order_resp) -> bool:
    """safe_place_order(_v2) 가 분산락 획득 실패 시 돌려주는 응답인지 판정."""
    return bool(order_resp) and order_resp.get("msg1") == "order_lock_busy"


# 싱글톤 인스턴스
mdd_manager = MDDManager()


# [M8] 비-데이트레이딩(스윙 계열) ActiveTrade.strategy_name 값 전수 목록
# db_worker 는 strategy_name 동등 비교만 지원하므로 '!= day_trade' 를 표현할 수 없다.
# → 비-DT 후보 전략명을 개별로 지정해 DT 행을 절대 건드리지 않는다.
#
# ⚠️ 경고: 이 목록은 `ActiveTrade.strategy_name` 에 기록되는 **모든 비-DT 값**을 덮어야
#   한다. 하나라도 빠지면 MDD 축소가 그 행에 반영되지 않아(db_worker 가 '필수 키 누락'
#   으로 스킵) quantity/remaining_pct 가 축소 전 값으로 남고, 다음 청산이 과다매도 →
#   주문거부를 일으킨다. 전량 축소 경로에서는 해당 행이 영구히 status='active' 로 남는다.
#
# [조사 방법 — 전략명 추가/변경 시 반드시 재실행]
#   1) grep -rn 'strategy_name' backend/app --include='*.py' \
#        | grep -E "strategy_name(\"|')?\]?\s*[:=]"
#      → enqueue_active_trade_upsert / ActiveTrade(...) 로 들어가는 리터럴 수집
#   2) grep -rn 'strategy_type' backend/app --include='*.py' | grep -E "strategy_type.*[:=]"
#      → trade_executor 는 `stock.get("strategy_name") or stock.get("strategy_type","swing")`,
#        portfolio 동기화는 `strategy_name=_h.strategy_type or "swing"` 이므로
#        Portfolio.strategy_type 값도 그대로 ActiveTrade.strategy_name 이 된다.
#   3) 수집값에서 DT 계열(DT_STRATEGY_NAMES)만 제외하고 나머지 전부를 여기에 넣는다.
#
# 2026-09-27 조사 결과:
#   "swing"              — trade_executor.py:1105 기본값 / swing/entry_service.py:288
#                          / portfolio.py:873, :1317
#   "MOMENTUM_BREAKOUT"  — swing/entry_service.py:291
#   "PULLBACK_BUY"       — swing/entry_service.py:313
#   "manual"             — manual_buy.py:156(+:168), portfolio_v2.py:382(+:402),
#                          portfolio.py:1536 → Portfolio.strategy_type='manual' 경유
#   "us_swing"           — us_market_scheduler.py:374, :577 (Portfolio.strategy_type)
#   "us_mean_reversion"  — us_market_scheduler.py:306, :406 (ActiveTrade 직접)
#   "synced"             — portfolio.py sync_portfolio 외부 유입 포지션 시딩(§8.5 항목4③)
#
# [2026-09-27 §8.5 항목6] 아래 목록은 **참조/문서용**으로만 남는다.
#   축소 반영 경로는 열거 대신 payload 의 `strategy_name_not=DT_STRATEGY_NAMES`
#   (= strategy_name IS NULL OR NOT IN (DT)) 한 건으로 처리하므로, 목록에서 빠진
#   전략명이 있어도 누락되지 않는다(NULL·미등재 전략명까지 자동 포함).
SWING_STRATEGY_NAMES = (
    "swing",
    "MOMENTUM_BREAKOUT",
    "PULLBACK_BUY",
    "manual",
    "us_swing",
    "us_mean_reversion",
    "synced",
)

# 데이트레이딩 계열 ActiveTrade.strategy_name 값 — 정본은 db_worker.
#   "day_trade"  — daytrading/entry_service.py, exit_service.py
#   "daytrading" — Portfolio.strategy_type='daytrading' 이 동기화로 그대로 들어가는 경우
# [2026-09-27 §8.5 항목6] 이중 정의 제거 — 정본은 db_worker.DT_STRATEGY_NAMES
# (모듈 상단에서 import, not_dt_strategy_clause 와 동일 목록 공유).
# db_worker 는 모듈 레벨에서 어떤 서비스 모듈도 import 하지 않으므로
# (asyncio/sqlalchemy/core.database/models.models 뿐) 순환 참조가 없다.
# 재노출: 기존 `from app.services.mdd_manager import DT_STRATEGY_NAMES` 호출부 호환 유지.
DT_STRATEGY_NAMES = _DT_STRATEGY_NAMES


def _dt_position_qty(stock_code: str) -> int:
    """데이트레이딩 인메모리 포지션 수량 (조회 실패 시 0). DB 접근 없음."""
    try:
        from app.services.daytrading.position_manager import day_trading_state
        pos = day_trading_state.positions.get(stock_code)
        return int((pos or {}).get("quantity", 0) or 0)
    except Exception as e:
        # 조회 실패를 0 으로 삼키되 반드시 기록한다. 호출부는 0 이어도 전략명 없는
        # close/upsert 를 하지 않고 비-DT 필터를 쓰므로 DT 행 오폐기는 일어나지 않는다.
        logger.warning(f"[MDD] DT 인메모리 수량 조회 실패 ({stock_code}) — 0 으로 간주: {e}")
        return 0


async def _sync_active_trade_after_reduction(
    stock_code: str,
    quantity: int,
    remaining_qty: int,
    user_id: Optional[int] = None,
) -> None:
    """MDD 축소 체결을 ActiveTrade 에 반영 (전략 귀속을 명시).

    db_worker 의 close_active_trade / upsert_active_trade 는 strategy_name 이
    비어 있으면 '가장 최근 active 행'을 매칭한다. 같은 종목에 데이트레이딩 행이
    있으면 DT 행이 닫히거나 잔량이 덮어써지고 스윙 행은 영구 잔존한다.

    처리 규칙:
      - DT 인메모리 포지션 없음 → 기존 동작 유지(전략명 없이 최신 행 1건)
      - DT 인메모리 포지션 있음 →
          * 전량 축소: 비-DT 행 + DT 행을 각각 닫음(모두 청산됐으므로)
          * 부분 축소: 비-DT 행만 갱신 (DT 잔량은 DT 자체 청산 경로가 갱신)

    [2026-09-27 §8.5 항목6] 전략명 열거(SWING_STRATEGY_NAMES 루프) → payload 필터
    1건으로 대체. 열거 방식은 (a) 목록에 없는/NULL 전략명 행을 놓치고,
    (b) 미존재 전략명마다 db_worker 의 '필수 키 누락' 경고 로그를 남겼다.
      - 비-DT 대상: strategy_name_not=DT_STRATEGY_NAMES (NULL 포함, DT 만 제외)
      - DT 대상   : strategy_names=DT_STRATEGY_NAMES (IN)

    시그니처 변경 없이 기존 enqueue_* / enqueue_db_write 인자 범위 내에서만 처리한다.
    """
    from app.services.db_worker import (
        enqueue_active_trade_upsert,
        enqueue_db_write,
    )

    dt_qty = _dt_position_qty(stock_code)

    async def _close_filtered(**filters) -> None:
        """전략명 확장 필터(strategy_names / strategy_name_not)로 close 태스크 큐잉.

        enqueue_close_active_trade 시그니처는 strategy_name 만 받으므로
        (시그니처 불변 원칙) 범용 enqueue_db_write 로 payload 를 직접 넘긴다.
        """
        data: dict = {"stock_code": stock_code, **filters}
        if user_id is not None:
            data["user_id"] = user_id
        await enqueue_db_write({"type": "close_active_trade", "data": data})

    async def _upsert_non_dt(ratio: float, qty: int) -> None:
        """비-DT(스윙 계열) 최신 active 행 1건만 잔량 갱신. DT 행은 절대 건드리지 않음."""
        payload: dict = {
            "stock_code": stock_code,
            # [하드닝 R3] remaining_pct 불변 (익절 단계 보존) — quantity 만 갱신
            "quantity": qty,
            "strategy_name_not": DT_STRATEGY_NAMES,
        }
        if user_id is not None:
            payload["user_id"] = user_id
        await enqueue_active_trade_upsert(payload)

    if remaining_qty <= 0:
        if dt_qty > 0:
            # 전량 청산 — 비-DT 행과 DT 행을 모두 닫는다(대상 범위 불변).
            await _close_filtered(strategy_name_not=DT_STRATEGY_NAMES)
            await _close_filtered(strategy_names=list(DT_STRATEGY_NAMES))
            logger.warning(
                f"[MDD] {stock_code}: DT {dt_qty}주 병존 — 비-DT/DT ActiveTrade 각각 종료"
            )
        else:
            # [R5] 전략명 없는 close 는 '최근 active 행'을 닫아 고아 DT 행을 오폐기한다.
            # MDD 축소 대상은 스윙 계열이므로 비-DT 필터(NULL 포함)로 한정한다.
            await _close_filtered(strategy_name_not=DT_STRATEGY_NAMES)
        return

    # 부분 축소: quantity 만 갱신하고 remaining_pct(익절 단계)는 유지한다 [하드닝 R3]
    sell_ratio = (remaining_qty / quantity) if quantity > 0 else 1.0
    if dt_qty > 0:
        # 스윙 행의 잔량 = 전체 잔량 − DT 잔량 (DT 는 체결분만큼 인메모리에서 차감됨)
        _sold = max(0, quantity - remaining_qty)
        _dt_left = max(0, dt_qty - _sold)
        _swing_left = max(0, remaining_qty - _dt_left)
        # [M1 2026-09-27] 배율은 반드시 '스윙 수량 기준' 으로 계산한다.
        # 총수량(스윙+DT) 기준 sell_ratio 를 곱하면 스윙 행의 remaining_pct 가
        # 과소평가되어(예: 정답 0.875 대신 0.70) _check_momentum_exit 의
        # 1차 익절 게이트(remaining_pct > 0.75)가 영구 미발동한다.
        _swing_entry_qty = max(0, quantity - dt_qty)
        if _swing_entry_qty > 0:
            _swing_ratio = min(1.0, max(0.0, _swing_left / _swing_entry_qty))
        else:
            _swing_ratio = 1.0
        if _swing_left <= 0:
            # [M1-guard 2026-09-27] _close_filtered() 는 '관리 대상에서 제거'라는 과도한 부작용이
            # 있으므로 인메모리 DT 수량을 신뢰할 수 있을 때만 허용한다.
            # quantity 는 KIS 잔고(hldg_qty), dt_qty 는 인메모리 → dt_qty > quantity 면
            # 인메모리가 잔고보다 큼 = stale 확정(외부 경로로 DT 분이 이미 매도된 경우).
            # 이때 _swing_left 는 허위 0 이므로 close 하면 실제 보유 중인 스윙 행이
            # status='closed' 가 되어 손절·트레일링 관리가 사라진다.
            if dt_qty > quantity:
                # 보수적 처리: 잔고 전량을 스윙분으로 간주하고 잔량만 갱신(close 금지).
                # 아무것도 안 하면 quantity 가 축소 전 값으로 남아 다음 청산이 과다매도한다.
                await _upsert_non_dt(sell_ratio, remaining_qty)
                logger.warning(
                    f"[MDD] {stock_code}: DT 인메모리 수량이 잔고를 초과 — stale 판단 "
                    f"(잔고 quantity={quantity}, 인메모리 dt_qty={dt_qty}, 매도 {_sold}주) → "
                    f"스윙 ActiveTrade close 생략, 잔량 {remaining_qty}주·배율 {sell_ratio:.4f} 로 갱신"
                )
                return
            # 스윙분은 전량 소진, 잔량은 전부 DT — quantity=0 유령 active 행을 남기지 않는다.
            await _close_filtered(strategy_name_not=DT_STRATEGY_NAMES)
            logger.warning(
                f"[MDD] {stock_code}: 스윙분 전량 축소(잔량 {remaining_qty}주는 DT) — "
                f"스윙 ActiveTrade 종료, DT 행은 DT 청산 경로에 위임"
            )
            return
        await _upsert_non_dt(_swing_ratio, _swing_left)
        logger.warning(
            f"[MDD] {stock_code}: DT {dt_qty}주 병존 — 스윙 ActiveTrade 만 잔량 갱신 "
            f"(스윙 잔량={_swing_left}/{_swing_entry_qty}, 배율={_swing_ratio:.4f}, "
            f"DT 잔량={_dt_left}, DT 행은 DT 청산 경로에 위임)"
        )
    else:
        # [R5] 전략명 없는 upsert 금지 — DT 행이 덮어써지고 스윙 행이 유령화된다.
        await _upsert_non_dt(sell_ratio, remaining_qty)


async def _reflect_reduction(
    stock_code: str, sell_qty: int, price: float, remaining_qty: int
) -> None:
    """MDD 축소 체결을 Portfolio 캐시/DB 와 데이트레이딩 인메모리 포지션에 반영.

    ActiveTrade 잔량은 호출측에서 upsert 하므로 여기서는 Portfolio 와 인메모리만 다룬다.
    """
    try:
        from app.services.portfolio_cache_bridge import reflect_sell
        await reflect_sell(stock_code, sell_qty, price, remaining_qty=remaining_qty)
    except Exception as e:
        logger.warning(f"[MDD] Portfolio 반영 실패 ({stock_code}): {e}")

    # [항목11 2026-09-27] 인메모리 차감 규칙을 _sync_active_trade_after_reduction 과 통일.
    #   ActiveTrade 측 규칙: _dt_left = max(0, dt_qty - _sold)  (= DT 우선 차감)
    #                        단, dt_qty > quantity(KIS 잔고) 면 stale 로 보고 close 를 생략.
    # 기존 코드는 DT 우선 차감(max(0, dt-sell))까지는 같았지만 stale 케이스를 가드하지
    # 않아, 인메모리 잔량이 실제 잔고(remaining_qty)를 초과한 채 남았다.
    #   예) 잔고 quantity=10, 인메모리 dt_qty=15, 매도 3주 → 잔고 7주인데 인메모리 12주
    #       → 다음 DT 청산이 12주 매도를 시도해 과다매도/주문거부.
    # 따라서 (1) DT 우선 차감, (2) 실제 잔고 잔량으로 상한 클램프, (3) 음수 금지.
    _dt_daily_state_dirty = False
    try:
        from app.services.daytrading.position_manager import day_trading_state
        async with day_trading_state.trade_lock:
            pos = day_trading_state.positions.get(stock_code)
            if pos:
                _dt_now = max(0, int(pos.get("quantity", 0) or 0))
                _balance_before = max(0, sell_qty + remaining_qty)  # = KIS 잔고(quantity)
                _stale = _dt_now > _balance_before
                if _stale:
                    # 인메모리가 잔고보다 큼 = 이미 외부 경로로 매도된 분이 반영 안 됨.
                    # 여기서 또 sell_qty 를 빼면 실제 잔고와 더 벌어지므로, 중복 차감 없이
                    # 실제 잔량으로 클램프만 한다(_sync 의 보수적 stale 처리와 동일 취지).
                    _left = min(_dt_now, remaining_qty)
                    logger.warning(
                        f"[MDD] {stock_code}: DT 인메모리 수량이 잔고를 초과 — stale 판단 "
                        f"(잔고 {_balance_before}주, 인메모리 {_dt_now}주, 매도 {sell_qty}주) → "
                        f"중복 차감 생략, 실제 잔량 {remaining_qty}주로 클램프"
                    )
                else:
                    # DT 우선 차감 (ActiveTrade 의 _dt_left 계산과 동일 규칙)
                    _left = max(0, _dt_now - sell_qty)
                    # 잔고 잔량 초과 방지 (이론상 불가하나 이중 방어)
                    _left = min(_left, remaining_qty)
                if _left <= 0:
                    day_trading_state.remove_position(stock_code)
                    # remove_position() 이 closed_today 에 추가한다 → 영속화 필요
                    _dt_daily_state_dirty = True
                    logger.warning(
                        f"[MDD] DT 포지션 전량 축소: {stock_code} "
                        f"(인메모리 {_dt_now}주 → 0, 매도 {sell_qty}주, 잔고 잔량 {remaining_qty}주)"
                    )
                else:
                    pos["quantity"] = _left
                    logger.warning(
                        f"[MDD] DT 포지션 잔량 갱신: {stock_code} {_dt_now} → {_left}주 "
                        f"(매도 {sell_qty}주, 잔고 잔량 {remaining_qty}주)"
                    )
    except Exception as e:
        logger.warning(f"[MDD] DT 인메모리 반영 실패 ({stock_code}): {e}")

    # [F9 2026-09-28] MDD 축소가 DT 포지션을 닫으면 closed_today 가 바뀌므로 영속화한다.
    # 이 경로에 저장이 없으면 축소 직후 재기동 시 당일 재진입 차단이 풀려, MDD -10%/-20%
    # 구간에서 방금 강제 축소한 종목을 그날 다시 매수할 수 있다.
    # - trade_lock **해제 후** 호출 — Redis I/O 로 락을 잡지 않는다(§3.1 실시간 계층 보호)
    # - 저장 실패가 축소 자체를 실패시키지 않도록 예외를 삼키되 warning 으로 남긴다
    if _dt_daily_state_dirty:
        try:
            from app.services.daytrading.position_manager import day_trading_state as _dts_save
            await _dts_save.save_daily_state()
        except Exception as e:
            logger.warning(
                f"[MDD] DT 일일 안전장치 저장 실패 ({stock_code}): {e} — 재기동 시 "
                f"당일 재진입 차단이 풀릴 수 있다"
            )