auto-stock/backend/app/services/daytrading/exit_service.py

"""데이트레이딩 청산 서비스

당일 포지션의 청산 조건 체크 및 실행:
  - 5분봉 ATR 기반 손절/익절 (day_trading_strategy.check_day_trade_exit 재활용)
  - 15:20 EOD 강제 청산 (CLAUDE.md 7.3: 당일 청산 원칙)
  - 일일 손실 한도(-3%) 초과 시 즉시 전체 청산
  - 분할 매도 지원: 목표가 달성 시 50% 매도, 잔량 트레일링스탑

청산 우선순위:
  1. 일일 손실 한도 초과 → 전체 강제 청산
  2. EOD 15:20 도달 → 전체 강제 청산
  3. 개별 포지션 체크:
     - 손절: ATR 기반 또는 하드스탑
     - 시간 손절: 진입 30분 경과 후 언제든 -0.3% 미만
     - 과매수 청산: RSI >= 70
     - 분할 익절: 목표가 +2R 달성 시 50% 매도

아키텍처 규칙:
  - CLAUDE.md 3.1: DB I/O 금지 — 인메모리 상태만 수정, 체결 후 db_task_queue 경유
  - loss_tracker 연동: 전량 매도 시 record_trade_result_safe() + save_to_db() 호출
"""
import asyncio
import contextvars
import logging
import time
from datetime import datetime, timedelta, time as _dtime

import pytz

from app.core.config import settings
from app.core.log_throttle import warn_throttled
from app.core.market_schedule import is_eod_flat_time, is_market_open
from app.services.korea_investment_api import korea_investment_api
from app.services.day_trading_strategy import (
    parse_minute_candles, aggregate_to_5min,
    check_day_trade_exit, is_daily_loss_exceeded,
    is_time_stop_exit_reason,
)
from app.services.daytrading.position_manager import day_trading_state
from app.services.daytrading import fill_reconciler
from app.services.db_worker import enqueue_trade_history, enqueue_close_active_trade
from app.services.loss_tracker import loss_tracker
from app.services.trade_cost_core import net_realized_pnl

logger = logging.getLogger(__name__)

# 한국 표준시(KST) 타임존
_KST = pytz.timezone("Asia/Seoul")

# [M1] 전일(stale) 포지션 강제청산으로 발생한 실현손익 누적 (당일 집계에서 제외된 금액).
# 당일 손실 한도 판단에는 쓰지 않고 추적/로그 용도로만 사용한다.
_stale_realized_pnl: float = 0.0

# [H3/H-A 2026-10-02] 이번 프로세스에서 stale 매도 **주문이 접수된**(rt_cd=="0") (날짜, 종목).
# 재주문 금지용. 실패는 여기 넣지 않는다 — 아래 _stale_sell_failures 로 제한 재시도한다.
_stale_sell_attempted: set[tuple[str, str]] = set()
# [H-A] stale 매도 실패 이력 {(날짜, 종목): (실패 횟수, 마지막 실패 monotonic)}
_stale_sell_failures: dict[tuple[str, str], tuple[int, float]] = {}
# [H-A/H-B] 종목별 마지막 청산 주문 결과: "accepted" | "rejected" | "unknown"
_last_exit_order_outcome: dict[str, str] = {}
# [M-4] stale 직전 시도의 주문 수량·시각 (불명 주문 매칭 범위)
_stale_last_qty: dict[tuple[str, str], int] = {}
_stale_last_since: dict[tuple[str, str], datetime] = {}
# [L-1] 보유 0 정리를 이미 큐잉한 (날짜, 종목)
_stale_cleaned: set[tuple[str, str]] = set()

# [H4 2026-10-02] 청산 체크 처리량·상한 — settings 로 이관(값 동일):
#   DAY_TRADE_EXIT_CHECK_CONCURRENCY(4) / DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC(15) /
#   DAY_TRADE_EXIT_CHECK_TIMEOUT_SEC(45) / DAY_TRADE_TOTAL_ASSETS_REFRESH_SEC(300).
# 종전에는 포지션마다 당일 전체 분봉을 **순차** 조회해(오후 종목당 10~12페이지 × 약 3초)
# 포지션 몇 개만으로 스케줄러 _guard(STEP_TIMEOUT=180)를 넘겨 청산 체크 전체가 취소됐다.
# 손절선 판정은 현재가 1회 조회로 먼저 끝내고, 분봉 기반 판정은 제한 동시성 + 포지션별
# 상한으로 돌린다. 동시성은 entry_service 선조회(8)보다 낮게 잡아 KIS 세마포어(8)를
# 진입 스캔과 나눠 쓴다.
# 최악치(기본값, N=DAY_TRADE_MAX_POSITIONS=5 → ceil(5/4)=2 웨이브):
#   총자산 조회 15 + 손절 선판정 2×(15+5)=40 + 분봉 판정 2×45=90 = 145초
#   + close_stale_positions(잔고 15 + 종목당 현재가·주문) — stale 이 있으면 180초에
#   근접할 수 있다. 초과 시 _guard 가 취소하지만 주문~반영은 shield 로 완주한다.
_total_assets_refreshed_at: float = 0.0


def get_stale_realized_pnl() -> float:
    """전일 포지션 강제청산으로 실현된 손익 누적값(당일 daily_pnl 에서 제외된 금액).

    하위호환용 원시 누적값. 날짜 경계 처리는 가산 시점(_add_stale_realized_pnl)에만 하므로,
    날짜가 바뀐 뒤 첫 가산 전에는 전일 값이 남아 보일 수 있다 — 표시용은 _v2 를 쓸 것.
    """
    return _stale_realized_pnl


# [R7 2026-10-10] _stale_realized_pnl 이 귀속된 KST 날짜. 날짜가 바뀌면 0 으로 되돌린다
# (종전에는 프로세스 수명 동안 무한 누적돼 '전일 누적'·/close-all 응답이 여러 날의 합이 됐다).
# 매매 판정에는 쓰이지 않는 추적·표시 값이다.
_stale_realized_pnl_date: str | None = None


def _roll_stale_realized_pnl_day() -> None:
    """날짜가 바뀌었으면 stale 누적값을 0 으로 초기화하고 날짜를 오늘로 맞춘다."""
    global _stale_realized_pnl, _stale_realized_pnl_date
    today = _today_key()
    if _stale_realized_pnl_date != today:
        if _stale_realized_pnl_date is not None and _stale_realized_pnl != 0.0:
            logger.info(
                f"[DayExit] 전일·고아 청산 손익 누적 일자 변경 {_stale_realized_pnl_date}→{today}: "
                f"{_stale_realized_pnl:+.0f}원 → 0 초기화"
            )
        if _stale_realized_pnl_date is not None:
            _stale_realized_pnl = 0.0
        _stale_realized_pnl_date = today


def _add_stale_realized_pnl(delta: float) -> float:
    """당일(KST) stale 누적값에 delta 를 더하고 새 누적값을 반환한다."""
    global _stale_realized_pnl
    _roll_stale_realized_pnl_day()
    _stale_realized_pnl += delta
    return _stale_realized_pnl


def get_stale_realized_pnl_v2() -> float:
    """당일(KST) 전일·고아 청산 실현손익 누적값. 날짜가 바뀌면 0 에서 다시 센다."""
    _roll_stale_realized_pnl_day()
    return _stale_realized_pnl


# [R2-3 2026-10-09] 국내 ETF/ETN 매도에는 증권거래세가 없다. DT 포지션에는 상품 구분이 없어
# 종목명으로 판별한다(운용사 브랜드 + 공백으로 시작, 또는 'ETN' 포함). 판별 실패(이름 없음·
# 신규 브랜드)는 주식 세율을 그대로 쓴다 — 순손익을 **낮게**(손실 쪽으로) 보는 기존 동작.
# ⚠️ 해외·채권형 ETF 의 매매차익 배당소득세(15.4%)는 원천징수라 여기서 모델링하지 않는다.
_ETF_BRAND_PREFIXES = (
    "KODEX", "TIGER", "PLUS", "KIWOOM", "RISE", "ACE", "SOL", "HANARO", "KOSEF",
    "ARIRANG", "KBSTAR", "TIMEFOLIO", "WOORI", "1Q", "BNK", "FOCUS", "TRUSTON",
    "KTOP", "UNICORN", "VITA", "WON", "TREX", "DAISHIN343", "마이티", "에셋플러스",
    "파워", "히어로즈", "마이다스", "KCGI", "HK",
)


def _is_etf_or_etn_name_impl(stock_name) -> bool:
    """종목명으로 국내 ETF/ETN 여부 판별 (브랜드 + 공백 접두 또는 'ETN' 토큰)."""
    name = str(stock_name or "").strip()
    if not name:
        return False
    upper = name.upper()
    if "ETN" in upper.split() or upper.endswith(" ETN") or " ETN(" in upper:
        return True
    head = upper.split(" ", 1)
    if len(head) < 2:
        return False
    # 브랜드 토큰 + 공백일 때만 인정(주식명은 대개 공백이 없다). ETN 발행사(삼성·신한·TRUE 등)는
    # 이름에 'ETN' 이 붙어 위에서 걸린다.
    return head[0] in _ETF_BRAND_PREFIXES


def _dt_sell_tax_rate(pos: dict) -> float:
    """DT 매도 거래세율 — 포지션에 시장 구분이 없으면 DAY_TRADE_DEFAULT_TAX_MARKET(스윙 기본과 동일).

    [R2-3] 국내 ETF/ETN(종목명 판별)은 증권거래세 0.
    """
    if _is_etf_or_etn_name_impl((pos or {}).get("stock_name")):
        return 0.0
    market = str((pos or {}).get("market") or settings.DAY_TRADE_DEFAULT_TAX_MARKET or "KOSPI").upper()
    if market == "KOSDAQ":
        return float(settings.KR_SELL_TAX_RATE_KOSDAQ)
    return float(settings.KR_SELL_TAX_RATE_KOSPI)


def dt_net_pnl(entry_price: float, exit_price: float, quantity: int, pos: dict) -> dict:
    """[2026-10-01] DT 실현손익(비용 차감). 스윙 집행 경로와 같은 비용 정의(trade_cost_core).

    매수·매도 수수료(KR_COMMISSION_RATE) + 매도 거래세(KR_SELL_TAX_RATE_*)를 뺀다.
    반환 키: gross_profit_loss / total_cost / net_profit_loss / net_profit_loss_rate
    """
    return net_realized_pnl(
        entry_price, exit_price, quantity,
        float(settings.KR_COMMISSION_RATE), _dt_sell_tax_rate(pos),
    )


def _sell_fill_pnl_impl(
    entry_price: float, avg_price: float, signal_price: float,
    filled_qty: int, quantity: int, pos: dict,
) -> dict:
    """[H-R1-2 2026-10-09] 매도 체결 확인 결과 → 손익·거래대금 (순수 계산).

    - 전량 체결(filled ≥ quantity): 종전과 동일 — 주문 수량 전체를 체결가로.
    - 미종결 부분체결(0 < filled < quantity): 체결분은 실제 체결가, 미체결분은 **신호가 기준 추정**
      (청산 접수 시 recorded_net 과 같은 근거)으로 합산한다. 예전에는 미체결분까지 체결가로 계상했다.
      미체결분을 0 으로 빼면 일일 손실한도(daily_pnl)가 그만큼 덜 잡히므로(손실 과소) 추정치를 남긴다.
      거래대금·체결수량은 실제 체결분만.
    두 구간을 수량 가중 혼합 청산가 하나로 묶어 dt_net_pnl 을 1회 호출한다(비용 반올림이 구간별로
    두 번 일어나지 않게 — 전량 체결 시 종전 값과 비트 동일).
    """
    q = max(0, int(quantity))
    f = max(0, min(int(filled_qty), q))
    if f >= q or f <= 0:
        r = dt_net_pnl(entry_price, avg_price, q, pos)
        return {
            "net_profit_loss": r["net_profit_loss"], "total_cost": r["total_cost"],
            "net_profit_loss_rate": r["net_profit_loss_rate"],
            "total_amount": float(avg_price) * q, "executed_quantity": q, "is_partial": False,
        }
    _sig = float(signal_price) if signal_price and float(signal_price) > 0 else float(avg_price)
    blended_exit = (float(avg_price) * f + _sig * (q - f)) / q
    r = dt_net_pnl(entry_price, blended_exit, q, pos)
    return {
        "net_profit_loss": r["net_profit_loss"], "total_cost": r["total_cost"],
        "net_profit_loss_rate": r["net_profit_loss_rate"],
        "total_amount": float(avg_price) * f, "executed_quantity": f, "is_partial": True,
    }


def _fallback_stop_pct() -> float:
    """[M7] stop_loss 가 없을 때 사용할 폴백 손절 비율.

    기존 하드코딩 0.06(-6%)은 전략 하드캡(ORB 3%, VWAP/RSI_BB 3.5%)의 2배로 느슨했다.
    가장 보수적(타이트)한 하드캡을 사용해 폴백이 실제 전략보다 헐거워지지 않게 한다.
    position_manager.restore_from_db() 의 폴백(DAY_TRADE_STOP_LOSS_PCT)과도 일치.
    """
    return min(
        settings.DAY_TRADE_STOP_LOSS_PCT,
        settings.DAY_TRADE_ORB_STOP_LOSS_PCT,
        settings.DAY_TRADE_REVERSION_STOP_LOSS_PCT,
    )


def _fallback_stop_price(entry_price: float) -> float:
    """진입가 기준 폴백 손절가."""
    return entry_price * (1 - _fallback_stop_pct())


def _fallback_target_price(entry_price: float, stop_loss: float) -> float:
    """진입가·손절가 기준 폴백 목표가 (DAY_TRADE_TARGET_MULTIPLIER 정합)."""
    risk = entry_price - stop_loss
    if risk <= 0:
        risk = entry_price * _fallback_stop_pct()
    return entry_price + risk * settings.DAY_TRADE_TARGET_MULTIPLIER


def _kst_now() -> datetime:
    """현재 KST 시간을 timezone-naive datetime으로 반환.

    entry_time과 비교 시 timezone 불일치 방지를 위해 tzinfo 제거.

    Returns:
        datetime: 현재 KST 시간 (timezone-naive)
    """
    return datetime.now(_KST).replace(tzinfo=None)


# [R2-1 2026-10-09] KRX 종가 단일가 매매 구간(시장 구조 상수 — 전략 파라미터 아님).
# 이 구간에 낸 시장가 주문은 15:30 에 일괄 체결된다.
_KRX_CLOSING_AUCTION_START = _dtime(15, 20)
_KRX_MARKET_CLOSE = _dtime(15, 30)
# 종가 단일가 재확인 백그라운드 태스크 (GC 방지·테스트 대기용)
_auction_fill_tasks: set = set()


def _in_closing_auction_impl(now: datetime) -> bool:
    """now(KST naive) 가 종가 단일가 구간(15:20 ≤ t < 15:30)인가."""
    t = now.time()
    return _KRX_CLOSING_AUCTION_START <= t < _KRX_MARKET_CLOSE


def _auction_recheck_params_impl() -> tuple[float, int, float]:
    """(장 마감 후 대기 초, 재확인 횟수, 재확인 간격 초).

    settings 에 DAY_TRADE_FILL_AUCTION_RECHECK_{DELAY_SEC,ATTEMPTS,INTERVAL_SEC} 가 있으면 그 값,
    없으면 기본 30초/3회/30초(config.py 미편입 — 같은 라운드에서 config 를 고치지 않았다).
    """
    _delay = max(0.0, float(getattr(settings, "DAY_TRADE_FILL_AUCTION_RECHECK_DELAY_SEC", 30.0)))
    _attempts = max(1, int(getattr(settings, "DAY_TRADE_FILL_AUCTION_RECHECK_ATTEMPTS", 3)))
    _interval = max(0.0, float(getattr(settings, "DAY_TRADE_FILL_AUCTION_RECHECK_INTERVAL_SEC", 30.0)))
    return _delay, _attempts, _interval


def _seconds_until_auction_recheck_impl(now: datetime) -> float:
    """장 마감(15:30) + 재확인 대기(_auction_recheck_params_impl) 까지 남은 초 (지났으면 0)."""
    _buf = _auction_recheck_params_impl()[0]
    target = datetime.combine(now.date(), _KRX_MARKET_CLOSE) + timedelta(seconds=_buf)
    return max(0.0, (target - now).total_seconds())


def _apply_round_trip_streak_v2(round_trip_net: float) -> None:
    """[R2-4] 전량 청산 시 왕복 순손익으로 당일 연패 판정 (trade_lock 안에서 호출).

    position_manager.update_pnl(is_full_close=True) 의 판정 규칙과 동일:
    손실 → +1, 이익 → 0 리셋, 0 → 중립. daily_pnl 은 호출부가 update_pnl(..., False) 로 더한다.
    """
    if round_trip_net < 0:
        day_trading_state.loss_streak_today += 1
    elif round_trip_net > 0:
        day_trading_state.loss_streak_today = 0


# [HL2-2 2026-10-09] 매도 0체결 복원 포지션의 '이미 기록된 승패' 표식.
# 전량 청산 접수 시 _finalize_exit_impl 이 당일 연패(update_pnl)·loss_tracker 를 기록하고,
# 그 매도가 0체결·부분 종결로 복원되면 기록은 되돌리지 않는다(§8.5-27). 복원 포지션이 다시
# 전량 청산될 때 같은 왕복을 또 세면 거래 1건이 3연패/42일 밴 카운트를 2 올린다.
# 값: True(손실로 기록됨) / False(이익·비손실로 기록됨) / 키 없음(기록 안 됨).
# ⚠️ 인메모리 전용 — 재기동하면 사라져 재청산 시 다시 기록된다(보수적 방향).
_RESTORED_STREAK_MARK_KEY = "restored_streak_is_loss"
_RESTORED_LOSS_TRACKER_MARK_KEY = "restored_loss_tracker_is_loss"
# _finalize_exit_impl → _on_unfilled 로 넘기는 스냅샷 전용 키(포지션에는 남기지 않는다)
_FULL_EXIT_RECORD_KEY = "_full_exit_record"

# [R6-1 2026-10-10] 당일 매수분 고아 표식. _detect_dt_orphans_impl 이 당일 창의 DT 원장 순체결이
# 양수인 고아에 붙인다. 이 포지션은 실보유·타전략 차감·재시도 관리는 전일·고아(stale) 경로를
# 그대로 타되, 실현손익은 **당일** daily_pnl·loss_streak_today 에 넣는다(§7 이 당일 집계 제외를
# 허용하는 것은 '전일' 포지션뿐 — 당일 사고 판 손실이 -3% 한도·3연패 쿨다운을 우회하던 결함).
_ORPHAN_TODAY_KEY = "orphan_today"
# 당일 고아 청산 사유 접두사 — position_manager._STALE_EXIT_REASON_PREFIXES("DT 고아 잔고 청산")와
# **겹치지 않아야** 재기동 원장 재구성(_rebuild_daily_state_from_rows)이 당일 손익에 포함한다.
_ORPHAN_TODAY_EXIT_REASON_PREFIX = "DT 당일 고아 잔고 청산"


def _is_orphan_today_position(pos) -> bool:
    """[R6-1] 당일 매수분 고아(당일 손익 집계 대상)인가."""
    return bool((pos or {}).get(_ORPHAN_TODAY_KEY))


def _should_record_outcome_impl(prev_mark, is_loss: bool) -> bool:
    """같은 왕복에 대해 승패를 (다시) 기록해야 하는가.

    - 이전 기록 없음(None) → 기록
    - 이전과 같은 결과 → 이미 셌으므로 기록 안 함(이중 집계 방지)
    - 이전=비손실, 이번=손실 → 기록(손실을 놓치지 않는 보수적 방향)
    - 이전=손실, 이번=비손실 → 기록 안 함(이미 센 손실을 유지 — 보수적 방향)
    """
    if prev_mark is None:
        return True
    return bool(is_loss) and prev_mark is False


def _read_outcome_mark_impl(src: dict | None, key: str):
    """포지션 dict 의 승패 표식(True/False/None). bool 이 아니면 None(기록 안 됨으로 간주)."""
    if not isinstance(src, dict):
        return None
    v = src.get(key)
    return v if isinstance(v, bool) else None


# [2026-10-01] 진입 시각 신뢰 구간 (KST 시각). DT 진입은 정규장 안에서만 일어나므로
# 이 밖의 entry_time 은 기준 불일치(예: 과거 UTC 저장 행 — KST 09:05 → UTC 00:05)로 본다.
_ENTRY_TIME_VALID_FROM = (9, 0)
_ENTRY_TIME_VALID_TO = (15, 30)

# 1분봉 라벨(stck_cntg_hour)이 "분 시작"인지 "분 종료"인지 응답 규약이 확정되지 않았다.
# 라벨 >= 진입시각 + 60초 인 봉만 쓰면 두 해석 모두에서 봉 전체가 진입 이후 체결이다.
_POST_ENTRY_CANDLE_LAG_SEC = 60


def _normalize_entry_time(entry_time) -> datetime | None:
    """entry_time 을 KST naive datetime 으로 정규화. 신뢰할 수 없으면 None.

    - 인메모리 진입(entry_service): `datetime.now(KST).replace(tzinfo=None)` (KST naive)
    - DB 복원(position_manager.restore_from_db): `ActiveTrade.created_at` (모델 기본값 now_kst
      = KST naive). 단 2026-09-23 이전 행은 UTC 로 저장된 것이 섞여 있다(timeutil 참조).
    """
    if entry_time is None or not isinstance(entry_time, datetime):
        return None
    if entry_time.tzinfo is not None:
        entry_time = entry_time.astimezone(_KST).replace(tzinfo=None)
    hm = (entry_time.hour, entry_time.minute)
    if hm < _ENTRY_TIME_VALID_FROM or hm > _ENTRY_TIME_VALID_TO:
        return None
    if entry_time > _kst_now():
        return None
    return entry_time


def _candle_datetime(candle) -> datetime | None:
    """MinuteCandle(date=YYYYMMDD, time=HHMMSS, KST) → naive datetime. 형식 불량이면 None."""
    d = getattr(candle, "date", "") or ""
    t = getattr(candle, "time", "") or ""
    if len(d) != 8 or len(t) < 4:
        return None
    try:
        return datetime(
            int(d[:4]), int(d[4:6]), int(d[6:8]),
            int(t[:2]), int(t[2:4]), int(t[4:6]) if len(t) >= 6 else 0,
        )
    except (ValueError, TypeError):
        return None


def _post_entry_candle_high(candles, entry_time) -> float | None:
    """진입 **이후에 시작한** 분봉들의 최고가. 산출 불가/해당 봉 없음이면 None (순수 함수).

    [2026-10-01 버그 수정] 예전에는 `max(c.high for c in candles_5m[-3:])` 로 최근 5분봉 3개의
    고가를 그대로 트레일링 기준 고점에 넣었다. 진입 후 15분 이내에는 **진입 전 봉의 고가**가
    섞여, 실제로 오르지 않았는데 +1% 트레일링 트리거가 켜지거나 `고점 − 1.5×ATR` 손절선이
    부풀려졌다(운영 실측: 트레일링 6건이 "고점 +1.1~2.2%" 기록 후 −0.13~−0.81% 로 종료).

    보수적 처리:
      - **진입이 속한 봉은 제외한다.** 봉 고가가 진입 전/후 어느 쪽 체결인지 알 수 없다.
        진입 직후 구간의 고점은 30초 주기의 현재가 관측과 진입 시 초기값(=진입가)으로만 잡는다
        (고점을 과소 추정 → 트레일링이 늦게 켜질 뿐, 하드 손절·시간손절은 영향 없음).
      - 5분봉 대신 1분봉을 쓰고, 라벨 >= 진입시각+60초 인 봉만 인정한다(라벨 규약 무관 안전).
      - 날짜까지 포함한 datetime 으로 비교해 전일 캔들 혼입(CLAUDE.md §7)을 차단한다.
        날짜·시각이 비어 있거나 형식 불량인 봉은 위치를 알 수 없으므로 제외한다.
      - entry_time 이 없거나 신뢰할 수 없으면 None → 호출부는 현재가만 쓴다.
    """
    entry = _normalize_entry_time(entry_time)
    if entry is None:
        return None
    threshold = entry + timedelta(seconds=_POST_ENTRY_CANDLE_LAG_SEC)
    best: float | None = None
    for c in candles or []:
        cdt = _candle_datetime(c)
        if cdt is None or cdt < threshold:
            continue
        h = float(getattr(c, "high", 0) or 0)
        if h > 0 and (best is None or h > best):
            best = h
    return best


async def check_and_exit_positions() -> list[dict]:
    """활성 데이트레이딩 포지션의 청산 조건 체크 및 실행.

    인메모리 positions 순회 → 5분봉 ATR 기반 청산 판단.
    스케줄러에서 30초 주기로 호출.

    체크 순서:
      1. 일일 손실 한도 초과 → 전체 강제 청산
      2. EOD 15:20 → 전체 강제 청산
      3. 개별 포지션별 check_day_trade_exit() 호출

    Returns:
        list[dict]: 청산 실행 결과 목록
    """
    day_trading_state.reset_if_new_day()
    exits = []

    if not day_trading_state.positions:
        # 인메모리가 비어도 DB에 전일 미청산 포지션이 남아 있을 수 있다(자정 reset 이후)
        return await close_stale_positions()

    # [L5 2026-10-02] 일일 손실한도 분모 확보 — last_total_assets 는 scan_and_enter 에서만
    # 갱신돼, 진입 스캔이 돌지 않는 구간(MDD 차단·마감 후·재기동 직후)에는 0 이었다.
    # is_daily_loss_exceeded() 는 total_assets<=0 이면 무조건 False → 청산 측 -3% 한도 무력화.
    await _ensure_total_assets_for_exit_impl()

    # 일일 손실 한도 초과 시 전체 강제 청산
    # [S5] 미실현 손익도 포함하여 일일 손실 한도 판단
    unrealized_pnl = 0.0
    for _code, _pos in day_trading_state.positions.items():
        # [R1-4 2026-10-09] 전일(stale) 포지션의 미실현(오버나이트 갭 포함)은 당일 한도에 넣지 않는다.
        # 실현손익도 당일에 넣지 않는 규칙(§7, count_daily_pnl=False)과 같은 기준이다. 넣으면
        # 갭 하락만으로 -3% 가 오발동해 당일 정상 포지션까지 force_close_all 로 청산된다.
        if _is_stale_position(_pos):
            continue
        _entry = _pos.get("entry_price", 0)
        _qty = _pos.get("quantity", 0)
        _cur = _pos.get("current_price", _entry)
        if _entry > 0 and _qty > 0:
            unrealized_pnl += (_cur - _entry) * _qty
    total_daily_pnl = day_trading_state.daily_pnl + unrealized_pnl
    if is_daily_loss_exceeded(day_trading_state.last_total_assets, total_daily_pnl):
        logger.warning(
            f"[DayExit] 일일 손실 한도 초과 → 전체 강제 청산 "
            f"(실현={day_trading_state.daily_pnl:.0f}, 미실현={unrealized_pnl:.0f})"
        )
        return await force_close_all(reason="일일 손실 한도 초과")

    # EOD 강제 청산 (15:20)
    if is_eod_flat_time():
        logger.info("[DayExit] EOD (15:20) → 전체 강제 청산")
        return await force_close_all(reason="EOD 강제 청산")

    # [안전장치 2026-09-23] 전일 미청산 포지션 즉시 청산.
    # 데이트레이딩은 당일 청산이 원칙인데 15:20 EOD가 실패/누락되면(2026-09-22 LG씨엔에스)
    # 포지션이 하룻밤 갭 리스크에 노출된 채 다음 날 일반 청산 로직을 타게 된다.
    stale_exits = await close_stale_positions()
    if stale_exits:
        exits.extend(stale_exits)
        if not day_trading_state.positions:
            return exits

    # [H4 2026-10-02] 1단계 — 손절선 선판정(현재가 1회 조회, 제한 동시성).
    # 분봉 전체 조회가 느려도 손절은 먼저 집행된다(§8.1-11 청산 우선).
    codes_snapshot = list(day_trading_state.positions.keys())
    fast, _fast_failed = await _run_exit_checks_bounded_v2_impl(
        codes_snapshot, _fast_stop_check_impl, float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC) + 5.0, "손절 선판정",
    )
    handled: set[str] = set()
    for code, res in fast.items():
        if not res:
            continue
        _done, _exit = res
        if _done:
            handled.add(code)
        if _exit:
            exits.append(_exit)

    # 2단계 — 분봉 기반 전체 판정(트레일링·시간손절·목표가·RSI). 포지션별 상한 + 제한 동시성.
    remaining_codes = [
        c for c in codes_snapshot
        if c not in handled and c in day_trading_state.positions
    ]
    full, _full_failed = await _run_exit_checks_bounded_v2_impl(
        remaining_codes, _check_single_position, float(settings.DAY_TRADE_EXIT_CHECK_TIMEOUT_SEC), "청산 판정",
    )
    for code in remaining_codes:
        if full.get(code):
            exits.append(full[code])
    # [R3-3] 2단계 실패·판정 불가 종목은 연속 실패로 센다.
    # [R4-3 2026-10-09] 1단계에서 '처리됨'으로 빠진 종목도 그 처리가 실패(손절 주문 KIS 거부·예외)
    # 였으면 실패로 센다 — 예전에는 handled 종목이 _full_failed 에 없어 카운터가 0 으로 리셋됐다.
    # (1단계 현재가 미확보 등 handled 가 아닌 종목은 2단계 결과가 우선한다.)
    _update_exit_check_health_impl(
        codes_snapshot, set(_full_failed) | (set(_fast_failed) & handled), "청산 판정",
    )

    return exits


async def check_stops_only_v2() -> list[dict]:
    """[M4 2026-10-02] 손절 전용 체크 — 09:00~09:30 ORB 수집 구간에서 호출.

    ORB 분기가 `continue` 로 청산 체크를 건너뛰어, 그 30분 동안 전일 미청산 포지션과
    재기동 복원 포지션의 손절이 전혀 판정되지 않았다. 이 함수는 진입·익절 규칙을
    건드리지 않고 **전일 포지션 청산 + 손절선 선판정**만 수행한다.
    """
    day_trading_state.reset_if_new_day()
    exits: list[dict] = list(await close_stale_positions() or [])
    codes = list(day_trading_state.positions.keys())
    if not codes:
        return exits
    fast, _fast_failed = await _run_exit_checks_bounded_v2_impl(
        codes, _fast_stop_check_impl, float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC) + 5.0, "ORB 구간 손절 판정",
    )
    for res in fast.values():
        if res and res[1]:
            exits.append(res[1])
    # [R3-3] ORB 구간은 선판정이 유일한 손절 판정 — 현재가 미확보도 실패로 센다.
    _update_exit_check_health_impl(codes, _fast_failed, "ORB 구간 손절 판정")
    return exits


async def _run_exit_checks_bounded_impl(codes, fn, timeout: float, label: str) -> dict:
    """[H4] 포지션별 청산 판정을 제한 동시성 + 포지션별 상한으로 실행. {code: 결과}.

    주문~상태 반영은 _execute_exit_impl 이 shield 하므로, 상한 초과로 판정 코루틴이
    취소돼도 이미 전송된 주문의 반영은 끝까지 진행된다.
    """
    out, _failed = await _run_exit_checks_bounded_v2_impl(codes, fn, timeout, label)
    return out


# [R3-3 2026-10-09] 판정 실패 수집기 — 판정 단계마다 새 set 을 ContextVar 로 내려보낸다.
# gather 가 만든 자식 태스크는 컨텍스트를 복사하므로 같은 set 을 본다(동시 사이클 간 섞이지 않음).
_exit_failed_cv: contextvars.ContextVar = contextvars.ContextVar("dt_exit_failed_codes", default=None)
# [R3-3] 종목별 '판정 실패' 연속 사이클 수 / 이번 연속 구간에서 경보를 이미 보낸 종목
_exit_check_failures: dict[str, int] = {}
_exit_stall_alerted: set[str] = set()
# [R3-3] 경보 기준 연속 사이클 수 기본값(config 미편입 — settings.DAY_TRADE_EXIT_CHECK_FAIL_ALERT_CYCLES 우선)
_EXIT_CHECK_FAIL_ALERT_CYCLES_DEFAULT = 4


def _mark_exit_check_failed_impl(stock_code: str) -> None:
    """[R3-3] 현재 판정 단계의 수집기에 '판정 실패(손절 미판정)'를 기록. 수집기 없으면 무시."""
    _c = _exit_failed_cv.get()
    if _c is not None:
        _c.add(stock_code)


async def _run_exit_checks_bounded_v2_impl(codes, fn, timeout: float, label: str) -> tuple[dict, set]:
    """[R3-3] _run_exit_checks_bounded_impl + 판정 실패 종목 집합. ({code: 결과}, {실패 code}).

    실패 = 포지션별 상한 초과·예외, 또는 판정 함수가 _mark_exit_check_failed_impl 로 알린
    '판정 불가'(폴백 손절 현재가 미확보 등). 예전에는 전부 삼키고 정상 반환해 스케줄러가
    '청산 체크 성공'으로 기록했다 — 손절이 멈춰도 경보가 없었다.
    """
    out: dict = {}
    failed: set = set()
    if not codes:
        return out, failed
    sem = asyncio.Semaphore(max(1, int(settings.DAY_TRADE_EXIT_CHECK_CONCURRENCY)))

    async def _one(code: str) -> None:
        async with sem:
            try:
                out[code] = await asyncio.wait_for(fn(code), timeout=timeout)
            except asyncio.TimeoutError:
                failed.add(code)
                logger.warning(
                    f"[DayExit] ⏱️ {label} {timeout:.0f}초 초과 ({code}) — 이번 사이클 스킵, "
                    f"다음 사이클 재판정"
                )
            except Exception as e:
                failed.add(code)
                logger.error(f"[DayExit] {label} 오류 ({code}): {e}", exc_info=True)

    _tok = _exit_failed_cv.set(failed)
    try:
        await asyncio.gather(*(_one(c) for c in codes))
    finally:
        _exit_failed_cv.reset(_tok)
    return out, failed


def _exit_check_fail_alert_cycles_impl() -> int:
    try:
        return max(1, int(getattr(settings, "DAY_TRADE_EXIT_CHECK_FAIL_ALERT_CYCLES",
                                  _EXIT_CHECK_FAIL_ALERT_CYCLES_DEFAULT)))
    except (TypeError, ValueError):
        return _EXIT_CHECK_FAIL_ALERT_CYCLES_DEFAULT


# [R4-3 2026-10-09] 종목별 청산 매도 주문 '거부'(rejected) 연속 횟수 / 이번 연속 구간 경보 보낸 종목.
# 접수(accepted) 1회면 리셋. 전일(stale) 경로는 _stale_sell_failures 가 따로 센다(여기서 제외).
_exit_reject_streak: dict[str, int] = {}
_exit_reject_alerted: set[str] = set()


def _note_exit_order_rejected_impl(
    stock_code: str, quantity: int, msg_cd, msg1, reason: str,
) -> None:
    """[R4-3] DT 청산 주문 거부 1건 기록 — 판정 실패 수집기(헬스)에 넣고, N 회 연속이면
    텔레그램(DT_EXIT_REJECT). 예전에는 logger.error 만 남고 헬스는 '성공'으로 리셋돼,
    손절선에 닿은 포지션의 시장가 매도가 계속 거부돼도 15:20 EOD 재시도 실패 전까지 경보가 없었다.
    매매 로직(재시도 주기·수량)은 바꾸지 않는다."""
    _mark_exit_check_failed_impl(stock_code)
    n = _exit_reject_streak.get(stock_code, 0) + 1
    _exit_reject_streak[stock_code] = n
    _limit = _exit_check_fail_alert_cycles_impl()
    if n >= _limit and stock_code not in _exit_reject_alerted:
        _exit_reject_alerted.add(stock_code)
        _alert_once_v2(
            "DT_EXIT_REJECT", f"{stock_code}:{time.monotonic():.0f}",
            f"[데이트레이딩] {stock_code} 청산 매도 주문이 {n}회 연속 거부됨 — 손절·청산이 집행되지 "
            f"않고 있습니다. 수량={int(quantity or 0)}주, msg_cd={msg_cd}, msg1={msg1}, 사유={reason}. "
            f"계좌(주문가능수량·계좌 거부 코드)를 확인하고 필요하면 수동 청산하세요.",
        )


def _note_exit_order_accepted_impl(stock_code: str) -> None:
    """[R4-3] 청산 주문 접수 → 거부 연속 횟수·경보 상태 해제."""
    _exit_reject_streak.pop(stock_code, None)
    _exit_reject_alerted.discard(stock_code)


def _update_exit_check_health_impl(codes, failed: set, label: str) -> None:
    """[R3-3] 이번 사이클 판정 결과로 종목별 연속 실패 수 갱신 + N 사이클 연속이면 텔레그램(DT_EXIT_STALL).

    청산된(인메모리에 없는) 종목은 카운터를 지운다. 성공 1회면 0 으로 되돌리고 경보 상태도 푼다
    (다시 N 사이클 연속 실패하면 다시 경보).
    """
    _limit = _exit_check_fail_alert_cycles_impl()
    for code in list(codes or []):
        if code not in day_trading_state.positions:
            _exit_check_failures.pop(code, None)
            _exit_stall_alerted.discard(code)
            continue
        if code not in failed:
            _exit_check_failures.pop(code, None)
            _exit_stall_alerted.discard(code)
            continue
        n = _exit_check_failures.get(code, 0) + 1
        _exit_check_failures[code] = n
        if n >= _limit and code not in _exit_stall_alerted:
            _exit_stall_alerted.add(code)
            _qty = int((day_trading_state.positions.get(code) or {}).get("quantity", 0) or 0)
            _alert_once_v2(
                "DT_EXIT_STALL", f"{code}:{time.monotonic():.0f}",
                f"[데이트레이딩] {code}({_qty}주) 청산 판정({label})이 {n}사이클 연속 실패 — 손절·트레일링·"
                f"시간손절이 멈춘 상태일 수 있습니다(KIS 시세·분봉 조회 장애 추정). 계좌를 확인하고 "
                f"필요하면 수동 청산하세요.",
            )
    for code in [c for c in _exit_check_failures if c not in day_trading_state.positions]:
        _exit_check_failures.pop(code, None)
        _exit_stall_alerted.discard(code)
    # [R4-3] 청산돼 인메모리에서 사라진 종목의 거부 카운터 정리
    for code in [c for c in _exit_reject_streak if c not in day_trading_state.positions]:
        _exit_reject_streak.pop(code, None)
        _exit_reject_alerted.discard(code)


def get_exit_check_health_v2() -> dict:
    """[R3-3] 청산 판정 헬스 — {"failing": {code: 연속 실패 사이클}, "stalled": [경보 기준 이상 code],
    "alert_cycles": N}. 스케줄러가 _record_dt_exit_result_v2(False) 판단에 쓸 수 있다."""
    _limit = _exit_check_fail_alert_cycles_impl()
    failing = {c: n for c, n in _exit_check_failures.items() if c in day_trading_state.positions}
    return {
        "failing": failing,
        "stalled": sorted(c for c, n in failing.items() if n >= _limit),
        "alert_cycles": _limit,
        # [R4-3] 종목별 청산 주문 거부 연속 횟수
        "order_rejects": {
            c: n for c, n in _exit_reject_streak.items() if c in day_trading_state.positions
        },
    }


def _effective_stop_loss(pos: dict) -> float:
    """포지션의 실효 손절가 — _check_single_position 과 같은 규칙(없거나 0 이하면 폴백)."""
    entry_price = float(pos.get("entry_price", 0) or 0)
    _sl_raw = pos.get("stop_loss")
    if _sl_raw is not None and float(_sl_raw) > 0:
        return float(_sl_raw)
    return _fallback_stop_price(entry_price)


async def _fast_stop_check_impl(stock_code: str) -> tuple[bool, dict | None]:
    """[H4] 현재가 1회 조회로 손절선(하드/폴백)만 판정. (처리됨 여부, 청산 결과).

    check_day_trade_exit 의 '손절' 조건(현재가 ≤ 손절가)과 같은 판정이다 — 입력 가격만
    분봉 종가 대신 현재가 시세다. 미도달·조회 실패면 (False, None) → 분봉 기반 판정으로 넘긴다.
    """
    pos = day_trading_state.get_position(stock_code)
    if not pos:
        return True, None
    # [H-B] 청산 주문 진행 중 → '처리됨'으로 보고 2단계 판정도 건너뛰게 한다
    if day_trading_state.is_exit_in_progress(stock_code):
        return True, None
    # [R1-4] 전일 포지션은 close_stale_positions 전용 — 일반 경로는 실보유·타전략 차감 없이
    # 행 수량 전량을 팔고 전일 손익을 당일 daily_pnl·연패에 넣는다(§7 금지). '처리됨'으로 넘긴다.
    if _is_stale_position(pos):
        _warn_stale_in_general_path(stock_code, "손절 선판정")
        return True, None
    entry_price = float(pos.get("entry_price", 0) or 0)
    if entry_price <= 0 or int(pos.get("quantity", 0) or 0) <= 0:
        return False, None
    stop_loss = _effective_stop_loss(pos)

    current_price = 0.0
    try:
        resp = await asyncio.wait_for(
            korea_investment_api.get_current_price(stock_code), timeout=float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC),
        )
        current_price = float(resp.get("output", {}).get("stck_prpr", 0) or 0) if resp else 0.0
    except asyncio.TimeoutError:
        logger.warning(
            f"[DayExit] 손절 선판정 현재가 {settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC:.0f}초 초과 ({stock_code}) "
            f"— 분봉 판정으로 넘김"
        )
        _mark_exit_check_failed_impl(stock_code)  # [R3-3] (2단계가 있으면 그쪽 결과가 우선)
        return False, None
    except Exception as e:
        logger.warning(f"[DayExit] 손절 선판정 현재가 조회 실패 ({stock_code}): {e} — 분봉 판정으로 넘김")
        _mark_exit_check_failed_impl(stock_code)
        return False, None
    if current_price <= 0:
        _mark_exit_check_failed_impl(stock_code)
        return False, None

    async with day_trading_state.trade_lock:
        if stock_code in day_trading_state.positions:
            day_trading_state.positions[stock_code]["current_price"] = current_price
    _remember_price_impl(stock_code, current_price)  # [R3-4]
    _rest_price_seen[stock_code] = (current_price, _kst_now())  # [R1-9] 2단계 재사용

    if current_price > stop_loss:
        return False, None

    loss_pct = (current_price - entry_price) / entry_price * 100
    reason = (
        f"손절 {loss_pct:.1f}% (현재={current_price:.0f} ≤ 손절가={stop_loss:.0f}, 현재가 선판정)"
    )
    logger.warning(f"[DayExit] ⚠️ {stock_code} {reason}")
    try:
        result = await _execute_exit(stock_code, pos, current_price, reason, sell_pct=1.0)
    except Exception as e:
        logger.error(f"[DayExit] 손절 선판정 주문 실행 실패 ({stock_code}): {e}", exc_info=True)
        _mark_exit_check_failed_impl(stock_code)  # [R4-3] 손절 미집행 → 헬스에 실패로
        result = None
    # 주문 실패여도 이번 사이클 분봉 판정에서 같은 손절을 재주문하지 않는다(다음 사이클 재시도).
    return True, result


# [R1-9 2026-10-09] 손절 선판정(1단계)이 REST 로 받은 현재가 {code: (가격, 관측 KST naive)}.
_rest_price_seen: dict[str, tuple[float, datetime]] = {}

# [R3-4 2026-10-09] KIS 로 실제 관측한 현재가 {code: (가격, time.monotonic())} — 폴백 손절 대체가의
# 나이 판정용. pos['current_price'] 는 영속화·복원되는 값이라 시각을 싣지 않고 여기 따로 둔다
# (재기동 후엔 비어 있어 대체가 없음 = 판정 불가로 처리된다).
_price_seen_mono: dict[str, tuple[float, float]] = {}
# 대체가 최대 나이 기본값(초, config 미편입 — settings.DAY_TRADE_FALLBACK_PRICE_MAX_AGE_SEC 우선)
_DT_FALLBACK_PRICE_MAX_AGE_SEC_DEFAULT = 180.0


def _remember_price_impl(stock_code: str, price: float) -> None:
    """[R3-4] KIS 실관측 현재가를 monotonic 시각과 함께 기록."""
    try:
        p = float(price)
    except (TypeError, ValueError):
        return
    if p > 0:
        _price_seen_mono[stock_code] = (p, time.monotonic())


def _dt_fallback_price_max_age_impl() -> float:
    try:
        v = float(getattr(settings, "DAY_TRADE_FALLBACK_PRICE_MAX_AGE_SEC", _DT_FALLBACK_PRICE_MAX_AGE_SEC_DEFAULT))
    except (TypeError, ValueError):
        v = _DT_FALLBACK_PRICE_MAX_AGE_SEC_DEFAULT
    return v if v > 0 else _DT_FALLBACK_PRICE_MAX_AGE_SEC_DEFAULT


def _fresher_rest_price_impl(stock_code: str, candles) -> float | None:
    """[R1-9] 2단계 판정가로 1단계 REST 현재가를 쓸지. 쓸 값 또는 None(분봉 종가 사용).

    조건: (a) 같은 사이클 관측(나이 ≤ 선판정 상한 + 분봉 판정 상한) (b) 마지막 1분봉 라벨 + 60초
    이후의 관측 — 라벨이 분 시작/종료 어느 규약이어도 그 봉이 끝난 뒤의 가격이라 분봉 종가보다
    신선하다. 라벨을 알 수 없으면 (a) 만으로 판단한다.
    """
    _rp = _rest_price_seen.get(stock_code)
    if not _rp or _rp[0] <= 0:
        return None
    price, seen_at = _rp
    _max_age = float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC) + float(settings.DAY_TRADE_EXIT_CHECK_TIMEOUT_SEC)
    if (_kst_now() - seen_at).total_seconds() > _max_age:
        return None
    last_dt = None
    for c in candles or []:
        cdt = _candle_datetime(c)
        if cdt is not None and (last_dt is None or cdt > last_dt):
            last_dt = cdt
    if last_dt is not None and seen_at < last_dt + timedelta(seconds=_POST_ENTRY_CANDLE_LAG_SEC):
        return None
    return float(price)


async def _ensure_total_assets_for_exit_impl() -> float:
    """[L5] 청산 측 일일 손실한도 분모(총자산) 확보.

    순서: 최근 값이 유효(>0, 갱신 주기 이내)면 그대로 → KIS 잔고 조회 → 마지막 값 →
    Redis 일일 안전장치 저장분(`daytrading:daily:{오늘}`.last_total_assets) → 없으면 warning.
    """
    global _total_assets_refreshed_at
    now = time.monotonic()
    cur = float(day_trading_state.last_total_assets or 0)
    if cur > 0 and _total_assets_refreshed_at and (now - _total_assets_refreshed_at) < float(settings.DAY_TRADE_TOTAL_ASSETS_REFRESH_SEC):
        return cur

    fresh = 0.0
    try:
        bal = await asyncio.wait_for(
            korea_investment_api.get_balance(), timeout=float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC),
        )
        summary = ((bal or {}).get("output2") or [{}])[0] if isinstance(bal, dict) else {}
        fresh = float((summary or {}).get("tot_evlu_amt", 0) or 0)
    except asyncio.TimeoutError:
        logger.warning(f"[DayExit] 총자산 조회 {settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC:.0f}초 초과 — 이전 값 사용")
    except Exception as e:
        logger.warning(f"[DayExit] 총자산 조회 실패 — 이전 값 사용: {e}")
    # 실패해도 다음 갱신 주기까지 재조회하지 않는다(매 사이클 실패 호출 방지)
    _total_assets_refreshed_at = now

    if fresh > 0:
        day_trading_state.last_total_assets = fresh
        return fresh
    if cur > 0:
        return cur

    restored = 0.0
    try:
        from app.core.redis import get_json
        from app.services.daytrading.position_manager import _today_kst
        data = await get_json(f"daytrading:daily:{_today_kst()}")
        if data and str(data.get("date") or "") == _today_kst().isoformat():
            restored = float(data.get("last_total_assets", 0) or 0)
    except Exception as e:
        logger.warning(f"[DayExit] 총자산 Redis 복원 실패: {e}")
    if restored > 0:
        day_trading_state.last_total_assets = restored
        logger.warning(f"[DayExit] 총자산 KIS 미확보 → Redis 저장값 {restored:,.0f}원 사용")
        return restored

    # 다음 사이클에 즉시 재시도하도록 타임스탬프를 되돌린다(값이 아예 없는 상태)
    _total_assets_refreshed_at = 0.0
    warn_throttled(
        logger, "dt_exit:total_assets_missing",
        f"[DayExit] ⚠️ 총자산 미확보 — 청산 측 일일 손실한도"
        f"({float(settings.DAY_TRADE_DAILY_LOSS_LIMIT) * 100:+.1f}%) 판정 불가 상태 "
        f"(KIS·이전 값·Redis 모두 없음)",
    )
    return 0.0


# [R3-1 2026-10-09] 청산 RSI 워밍업 — 진입 판정과 같은 시계열 기준으로 맞춘다.
# 진입(entry_service._fetch_candles_impl)은 전일 봉을 붙인 시계열로 RSI(14)를 계산하는데
# 청산은 당일 봉만 써서, 당일 5분봉 15개 미만(약 10:10 이전)에는 rsi_wilder 가 None 이라
# '과매수 청산'이 구조적으로 발동하지 않았고 이후에도 진입/청산 RSI 시드가 달랐다.
# 전일 봉은 장중에 바뀌지 않으므로 종목별로 하루 1회만 확보해 재사용한다(KIS 호출 최소화).
# 확보 실패·빈 결과면 _RSI_WARMUP_RETRY_SEC 마다 재시도하고, 그동안은 종전대로 당일 봉만 쓴다.
# 값 {종목: (당일 영업일자, 확보 시각 monotonic, 전일 1분봉 리스트)}
_rsi_warmup_prev_cache: dict[str, tuple[str, float, list]] = {}
_RSI_WARMUP_RETRY_SEC: float = 300.0        # 운영 상수(전략 파라미터 아님) — 빈 결과 재시도 간격
_RSI_WARMUP_FETCH_TIMEOUT_SEC: float = 15.0  # 워밍업 조회 상한 — 포지션별 판정 상한(45s) 안에서 끝나게


def _prev_session_candles_impl(candles: list, today_date: str) -> list:
    """`today_date` 보다 이전 영업일자의 1분봉만 남긴다(영업일자 빈 봉 제외)."""
    out = []
    for c in candles or []:
        d = getattr(c, "date", "") or ""
        if d and d < today_date:
            out.append(c)
    return out


async def _rsi_warmup_prev_candles_impl(stock_code: str, today_date: str) -> list:
    """[R3-1] 청산 RSI 워밍업용 전일 1분봉. 실패해도 예외를 올리지 않고 [] 를 돌려준다.

    순서: 모듈 캐시(당일 1회) → 진입 스캔 캐시(`entry_service._candle_cache`, 같은 시계열)
    → `get_minute_chart_full_day_v2(include_prev_day=True)` 직접 조회.
    """
    if not today_date:
        return []
    now = time.monotonic()
    cached = _rsi_warmup_prev_cache.get(stock_code)
    if cached and cached[0] == today_date:
        if cached[2]:
            return cached[2]
        if (now - cached[1]) < _RSI_WARMUP_RETRY_SEC:
            return []

    prev: list = []
    # 1) 진입 스캔 캐시 재사용 — 네트워크 없음
    try:
        from app.services.daytrading import entry_service as _ens
        ent = _ens._candle_cache.get(stock_code)
        if ent:
            prev = _prev_session_candles_impl(ent[1], today_date)
    except Exception as e:
        logger.warning(f"[DayExit] 진입 분봉 캐시 참조 실패 ({stock_code}): {e}")
        prev = []

    # 2) 직접 조회 (진입과 같은 API·같은 옵션)
    if not prev:
        try:
            raw = await asyncio.wait_for(
                korea_investment_api.get_minute_chart_full_day_v2(
                    stock_code, include_prev_day=True
                ),
                timeout=_RSI_WARMUP_FETCH_TIMEOUT_SEC,
            )
            prev = _prev_session_candles_impl(parse_minute_candles(raw or []), today_date)
        except Exception as e:
            warn_throttled(
                logger, f"dt_exit:rsi_warmup:{stock_code}",
                f"[DayExit] 청산 RSI 워밍업 조회 실패 ({stock_code}): {e!r} "
                f"— 당일 봉만으로 RSI 판정({_RSI_WARMUP_RETRY_SEC:.0f}초 후 재시도)",
            )
            prev = []

    # 당일 외 항목 정리(무한 증식 방지)
    for k in [k for k, v in _rsi_warmup_prev_cache.items() if v[0] != today_date]:
        _rsi_warmup_prev_cache.pop(k, None)
    _rsi_warmup_prev_cache[stock_code] = (today_date, now, prev)
    if not prev:
        warn_throttled(
            logger, f"dt_exit:rsi_warmup_empty:{stock_code}",
            f"[DayExit] 청산 RSI 워밍업 전일 봉 0개 ({stock_code}) — 당일 5분봉 15개 미만 구간은 "
            f"과매수 청산(RSI≥{float(settings.DAY_TRADE_RSI_EXIT_OVERBOUGHT):g}) 판정 불가",
        )
    return prev


def _exit_signal_candles_5m_impl(today_candles: list, prev_candles: list) -> list:
    """청산 판정용 5분봉 = 전일 워밍업 + 당일 봉 (진입 판정 시계열과 같은 구성).

    `check_day_trade_exit` 에서 이 시계열이 쓰이는 곳은 `_rsi`(워밍업 의도)와
    `_atr`(내부에서 `_last_session_candles` 로 당일 세션만 사용 — 갭 TR 미포함)뿐이므로
    ATR·트레일링은 당일 봉 기준 그대로다. 현재가·고점은 호출부가 당일 봉으로 따로 구한다.
    """
    if not prev_candles:
        return aggregate_to_5min(today_candles)
    return aggregate_to_5min(list(prev_candles) + list(today_candles))


async def _check_single_position(stock_code: str) -> dict | None:
    """단일 포지션의 청산 조건 체크.

    5분봉 데이터를 조회하여 check_day_trade_exit()로 청산 신호를 판단.
    최고가(highest_price)를 갱신하여 트레일링스탑 정확도 향상.

    Args:
        stock_code: 종목 코드

    Returns:
        dict | None: 청산 실행 결과 또는 None (청산 조건 미충족)
    """
    pos = day_trading_state.get_position(stock_code)
    if not pos:
        return None
    # [H-B] 청산 주문 진행 중이면 판정하지 않는다(인메모리 수량이 아직 반영 전)
    if day_trading_state.is_exit_in_progress(stock_code):
        return None
    # [R1-4] 전일 포지션은 close_stale_positions 전용(실보유·타전략 차감·당일 손익 제외)
    if _is_stale_position(pos):
        _warn_stale_in_general_path(stock_code, "분봉 판정")
        return None

    entry_price = pos.get("entry_price", 0)
    # [M7] stop_loss / target_price 는 None 뿐 아니라 0 으로 저장될 수 있다.
    # `or` 패턴은 0 도 폴백으로 넘기지만 의도가 모호하므로 명시적으로 None/비정상값을 검사한다.
    # (target_price=0이면 day_trading_strategy.py의 목표가 도달 조건이 항상 참이 되어 즉시 오매도됨)
    _sl_raw = pos.get("stop_loss")
    stop_loss = float(_sl_raw) if (_sl_raw is not None and float(_sl_raw) > 0) else _fallback_stop_price(entry_price)
    _tp_raw = pos.get("target_price")
    target_price = (
        float(_tp_raw) if (_tp_raw is not None and float(_tp_raw) > 0)
        else _fallback_target_price(entry_price, stop_loss)
    )
    highest_price = pos.get("highest_price", entry_price)
    entry_time = pos.get("entry_time")

    # [H4] try 범위를 "분봉 조회 + 지표/신호 계산"까지로 한정한다.
    # 예전 코드는 _execute_exit 까지 하나의 try 안에 있고 로그가 debug 였기 때문에
    # 분봉 조회 실패 시 손절·트레일링·시간손절이 전부 미적용된 채 15:20 까지 방치되고,
    # 주문 실패조차 로그에 보이지 않았다.
    exit_signal = None
    current_price = 0.0
    try:
        raw = await korea_investment_api.get_minute_chart_full_day(stock_code)
        candles = parse_minute_candles(raw)
        candles_5m = aggregate_to_5min(candles)
    except Exception as e:
        logger.warning(f"[DayExit] 5분봉 조회 실패 ({stock_code}): {e} — 폴백 손절만 판정")
        return await _check_fallback_stop_loss(stock_code, pos, stop_loss)

    if not candles_5m or len(candles_5m) < 5:
        # 분봉이 부족하면 ATR/RSI 기반 판정이 불가 → 손절선만 지킨다(방치 금지)
        logger.warning(
            f"[DayExit] 5분봉 부족 ({stock_code}): {len(candles_5m or [])}개 — 폴백 손절만 판정"
        )
        return await _check_fallback_stop_loss(stock_code, pos, stop_loss)

    try:
        current_price = float(candles_5m[-1].close)
        # [R1-9] 1단계 REST 현재가가 마지막 1분봉보다 확실히 뒤의 관측이면 그 값을 쓴다
        # (분봉은 1분 버킷 캐시라 최대 1분 묵을 수 있다 — 낡은 종가로 신선한 현재가를 덮어쓰지 않게).
        _rest_px = _fresher_rest_price_impl(stock_code, candles)
        if _rest_px is not None:
            current_price = _rest_px

        # highest_price 갱신 — 진입 이후에 시작한 1분봉 고가만 반영 (30초 주기 보완).
        # [2026-10-01] 예전 `candles_5m[-3:]` 는 진입 전 봉 고가가 섞여 트레일링 고점을 오염시켰다.
        # entry_time 없음/불신 → None → 현재가만 사용(고점 과대 추정보다 과소 추정이 안전).
        # 기존 highest_price 와 max → 단조 증가 유지.
        candle_high = _post_entry_candle_high(candles, entry_time)
        if candle_high is None and entry_time is not None and _normalize_entry_time(entry_time) is None:
            logger.warning(
                f"[DayExit] entry_time 불신 ({stock_code}, entry_time={entry_time}) "
                f"— 트레일링 고점은 현재가만으로 갱신"
            )
        new_highest = max(current_price, candle_high or 0.0, highest_price)
        async with day_trading_state.trade_lock:
            if stock_code in day_trading_state.positions:
                if new_highest > highest_price:
                    day_trading_state.positions[stock_code]["highest_price"] = new_highest
                # [S5] 미실현 손익 계산용 현재가 갱신
                day_trading_state.positions[stock_code]["current_price"] = current_price
        _remember_price_impl(stock_code, current_price)  # [R3-4]
        if new_highest > highest_price:
            highest_price = new_highest

        # [R3-1] RSI 판정 시계열만 진입과 같은 워밍업 구성으로 맞춘다(임계값 불변).
        # 현재가·고점은 위에서 당일 봉으로 이미 확정했고, ATR 은 _atr 이 당일 세션만 쓴다.
        _today_date = (getattr(candles[-1], "date", "") or "") if candles else ""
        _prev_warm = await _rsi_warmup_prev_candles_impl(stock_code, _today_date)
        candles_5m_signal = _exit_signal_candles_5m_impl(candles, _prev_warm)

        # 청산 신호 체크
        current_time = _kst_now().time()
        exit_signal = check_day_trade_exit(
            entry_price=entry_price,
            current_price=current_price,
            stop_loss=stop_loss,
            target_price=target_price,
            highest_since_entry=highest_price,
            candles_5min=candles_5m_signal,
            current_time=current_time,
            entry_time=entry_time,
        )
    except Exception as e:
        logger.warning(f"[DayExit] 청산 지표 계산 실패 ({stock_code}): {e} — 폴백 손절만 판정")
        return await _check_fallback_stop_loss(stock_code, pos, stop_loss)

    if not exit_signal or current_price <= 0:
        return None

    # 주문 실행은 try 밖(별도 try) — 실패를 조회 실패와 구분해 error 로 남긴다.
    try:
        return await _execute_exit(
            stock_code, pos, current_price, exit_signal.reason,
            sell_pct=exit_signal.sell_pct,
        )
    except Exception as e:
        logger.error(f"[DayExit] 청산 주문 실행 실패 ({stock_code}): {e}", exc_info=True)
        return None


async def _check_fallback_stop_loss(
    stock_code: str, pos: dict, stop_loss: float,
) -> dict | None:
    """[H4] 분봉 미확보 시 최소 안전장치 — 손절선 이탈만 판정해 전량 청산.

    swing/exit_service.py 의 고정비율 폴백 손절과 같은 취지. 분봉이 없으면
    트레일링/시간손절/RSI 청산은 판정할 수 없으므로 "손절선"만 지킨다.
    현재가는 KIS 현재가 조회 → 인메모리 최신가(current_price) 순으로 확보하고,
    둘 다 실패하면 logger.error 로 남긴다(손절 미적용을 침묵시키지 않는다).

    Args:
        stock_code: 종목 코드
        pos: 포지션 정보
        stop_loss: 적용할 손절가 (0 이하이면 진입가 기준 폴백 산출)

    Returns:
        dict | None: 청산 결과 또는 None (손절 미도달·판정 불가)
    """
    entry_price = float(pos.get("entry_price", 0) or 0)
    if entry_price <= 0 or int(pos.get("quantity", 0) or 0) <= 0:
        return None
    if day_trading_state.is_exit_in_progress(stock_code):  # [H-B]
        return None
    if stop_loss <= 0:
        stop_loss = _fallback_stop_price(entry_price)

    current_price = 0.0
    try:
        resp = await korea_investment_api.get_current_price(stock_code)
        current_price = float(resp.get("output", {}).get("stck_prpr", 0) or 0) if resp else 0.0
    except Exception as e:
        logger.warning(f"[DayExit] 폴백 손절 현재가 조회 실패 ({stock_code}): {e}")
    _fresh_from_kis = current_price > 0
    if not _fresh_from_kis:
        # [R3-4] 대체가는 나이 상한 안의 최근 관측가만 쓴다. 예전에는 시각 없는 pos['current_price']
        # 를 써서, 수십 분 전 가격이 손절가보다 높으면 '미도달'로 조용히 통과했다.
        _max_age = _dt_fallback_price_max_age_impl()
        _cached = _price_seen_mono.get(stock_code)
        _age = (time.monotonic() - _cached[1]) if _cached else None
        if _cached and _cached[0] > 0 and _age is not None and 0 <= _age <= _max_age:
            current_price = float(_cached[0])
            logger.warning(
                f"[DayExit] 폴백 손절 대체가 사용 ({stock_code}): {current_price:.0f} (나이 {_age:.0f}초 ≤ {_max_age:.0f}초)"
            )
        else:
            # 판정 불가 — '손절 미도달'과 구분한다(error + 연속 실패 집계 → N 사이클이면 텔레그램).
            _mark_exit_check_failed_impl(stock_code)
            _error_throttled(
                f"dt_exit:fallback_no_price:{stock_code}",
                f"[DayExit] 🚨 폴백 손절 판정 불가 ({stock_code}) — KIS 현재가 실패, 대체가 "
                + (f"나이 {_age:.0f}초 > {_max_age:.0f}초" if _age is not None else "없음")
                + " — 손절 미적용 상태, 수동 확인 필요",
            )
            return None

    async with day_trading_state.trade_lock:
        if stock_code in day_trading_state.positions:
            day_trading_state.positions[stock_code]["current_price"] = current_price
    if _fresh_from_kis:
        _remember_price_impl(stock_code, current_price)

    if current_price > stop_loss:
        return None

    loss_rate = (current_price - entry_price) / entry_price * 100
    logger.warning(
        f"[DayExit] ⚠️ 폴백 손절 발동 ({stock_code}): 현재가={current_price:.0f} "
        f"<= 손절={stop_loss:.0f} ({loss_rate:.2f}%) — 분봉 미확보"
    )
    try:
        return await _execute_exit(
            stock_code, pos, current_price,
            reason=f"폴백 손절(분봉 미확보) {loss_rate:.1f}%",
            sell_pct=1.0,
        )
    except Exception as e:
        logger.error(f"[DayExit] 폴백 손절 주문 실패 ({stock_code}): {e}", exc_info=True)
        return None


async def _execute_exit(
    stock_code: str, pos: dict, price: float, reason: str,
    sell_pct: float = 1.0,
) -> dict | None:
    """실제 매도 주문 실행 및 상태 업데이트.

    처리 흐름:
      1. 매도 수량 계산 (분할 매도 시 최소 1주 보장)
      2. safe_place_order()로 시장가 매도 (세마포어+분산락 보호)
      3. 인메모리 상태 업데이트 (trade_lock 보호)
      4. 전량 매도 시 loss_tracker 연동
      5. DB 기록 (db_task_queue 경유)

    Args:
        stock_code: 종목 코드
        pos: 포지션 정보 딕셔너리
        price: 현재가 (매도 기준가)
        reason: 청산 사유 (로그 및 DB 기록용)
        sell_pct: 매도 비율 (1.0=전량, 0.5=50% 분할 매도)

    Returns:
        dict | None: 청산 결과 (stock_code, price, quantity, profit_loss 등)
    """
    return await _execute_exit_impl(stock_code, pos, price, reason, sell_pct, is_stale=False)


async def _execute_exit_stale(
    stock_code: str, pos: dict, price: float, reason: str,
) -> dict | None:
    """[M1] 전일 미청산(stale) 포지션 전용 청산 경로.

    _execute_exit 과 동일하게 주문·상태반영·DB 기록을 수행하지만,
    실현손익을 **당일** daily_pnl / loss_streak_today 에 가산하지 않는다.
    (전일 손실이 당일 손실 한도(-3%)를 오발동시켜 정상 포지션까지 전량 청산되고
     신규 진입이 종일 차단되던 문제 — 2026-09-27 수정)
    """
    return await _execute_exit_impl(stock_code, pos, price, reason, 1.0, is_stale=True)


async def _execute_exit_impl(
    stock_code: str, pos: dict, price: float, reason: str,
    sell_pct: float, is_stale: bool,
) -> dict | None:
    """매도 주문 실행 + 상태 반영 공통 구현 (신규 함수 — 기존 시그니처 불변).

    is_stale=True 이면 당일 손익 집계(update_pnl)를 건너뛴다.
    """
    total_qty = pos.get("quantity", 0)

    if total_qty <= 0 or price <= 0:
        return None

    # [P1] 분할 매도 수량 계산
    if sell_pct >= 1.0:
        quantity = total_qty
    else:
        quantity = max(1, int(total_qty * sell_pct))
        if quantity >= total_qty:
            quantity = total_qty

    is_full_exit = (quantity >= total_qty)

    # [H-B 2026-10-02] 청산 진행 중 등록 — 같은 종목에 주문이 이미 나가 있으면 내지 않는다.
    # 판정 코루틴이 상한(wait_for)으로 취소돼도 shield 된 주문은 계속 진행되는데, 인메모리
    # 수량은 finalize 전까지 그대로라 2단계/다음 사이클이 같은 물량을 다시 팔았다(4주 → 4+4).
    # 해제는 인메모리 반영 직후(정상) 또는 shield 작업의 finally(실패·불명)가 보장한다.
    async with day_trading_state.trade_lock:
        if stock_code in day_trading_state.exits_in_progress:
            warn_throttled(
                logger, "dt_exit:exit_in_progress",
                f"[DayExit] {stock_code} 청산 주문 진행 중 — 중복 매도 방지로 이번 요청 건너뜀 ({reason})",
            )
            return None
        # [L-8] 판정 시점의 포지션 객체가 아직 그대로인지 — 그 사이 청산·교체됐으면 내지 않는다.
        # 전일(stale) 경로는 DB 에만 있는 행·수량 축소 사본을 넘기므로, 인메모리가 비었거나
        # 같은 날짜 경계 밖(전일) 포지션일 때만 허용한다.
        _cur = day_trading_state.positions.get(stock_code)
        if is_stale:
            if _cur is not None and _cur is not pos and not _is_stale_position(_cur):
                logger.warning(f"[DayExit] {stock_code} 전일 청산 요청 — 당일 포지션과 겹쳐 건너뜀")
                return None
        elif _cur is not pos:
            logger.warning(
                f"[DayExit] {stock_code} 판정 이후 포지션이 청산/교체됨 — 주문하지 않는다 ({reason})"
            )
            return None
        elif _is_stale_position(pos):
            # [R1-4] 공통 종단 가드 — 전일 포지션은 is_stale=True 경로(close_stale_positions)만 판다.
            # 일반 경로는 실보유·타전략 차감 없이 행 수량 전량을 팔고 전일 손익을 당일에 넣는다.
            _warn_stale_in_general_path(stock_code, f"일반 청산({reason})")
            return None
        day_trading_state.exits_in_progress.add(stock_code)
        _tok = object()
        _exit_tokens[stock_code] = _tok

    # [H4 2026-10-02] 주문 전송 ~ 상태 반영 전체를 shield 한다.
    # 예전에는 finalize 만 shield 돼 있어, 청산 체크가 포지션별 wait_for·스케줄러 _guard
    # 로 취소되는 시점이 safe_place_order 대기 중이면 "브로커는 주문을 받았는데 응답을
    # 잃어 인메모리는 그대로" → 다음 사이클이 같은 물량을 다시 판다(과다매도·스윙 물량 매도).
    return await asyncio.shield(_place_and_finalize_exit_guarded_impl(
        stock_code, pos, price, reason, sell_pct, quantity, total_qty, is_full_exit, is_stale,
        _tok,
    ))


# 진행 중 표식의 소유 토큰 {code: token}. finalize 가 먼저 해제한 뒤 같은 종목의 새 청산이
# 표식을 다시 잡았을 때, 앞선 작업의 finally 가 새 표식을 지우지 못하게 한다.
_exit_tokens: dict[str, object] = {}


def _is_stale_position(pos: dict) -> bool:
    et = (pos or {}).get("entry_time")
    return hasattr(et, "date") and et.date() < _kst_now().date()


def _warn_stale_in_general_path(stock_code: str, label: str) -> None:
    """[R1-4] 일반 청산 경로가 전일 포지션을 만났다 — 건너뛰고 stale 경로에 맡긴다(로그 스로틀)."""
    warn_throttled(
        logger, f"dt_exit:stale_in_general:{stock_code}",
        f"[DayExit] {stock_code} 전일 미청산 포지션 — {label}·EOD·일일한도 경로에서 제외, "
        f"close_stale_positions(실보유·타전략 차감, 당일 손익 제외)만 청산한다",
    )


def _release_exit_in_progress(stock_code: str, delay: float = 0.0, token=None) -> None:
    """[H-B] 청산 진행 중 표식 해제. delay>0 이면 그만큼 뒤에 해제(결과 불명 보류).

    token 을 주면 그 토큰이 아직 현재 소유자일 때만 해제한다(앞선 작업이 뒤 작업 표식을
    지우는 경합 방지). finalize(현재 소유자 문맥)는 token 없이 즉시 해제한다.
    """
    def _do() -> None:
        if token is not None and _exit_tokens.get(stock_code) is not token:
            return
        _exit_tokens.pop(stock_code, None)
        day_trading_state.exits_in_progress.discard(stock_code)

    if delay <= 0:
        _do()
        return
    try:
        asyncio.get_running_loop().call_later(delay, _do)
    except RuntimeError:
        _do()


async def _place_and_finalize_exit_guarded_impl(
    stock_code: str, pos: dict, price: float, reason: str, sell_pct: float,
    quantity: int, total_qty: int, is_full_exit: bool, is_stale: bool,
    token=None,
) -> dict | None:
    """[H-B] 주문~반영 + 진행 중 표식 해제 보장 (shield 대상)."""
    _hold = 0.0
    try:
        return await _place_and_finalize_exit_impl(
            stock_code, pos, price, reason, sell_pct, quantity, total_qty, is_full_exit, is_stale,
        )
    finally:
        if _last_exit_order_outcome.get(stock_code) == "unknown":
            _hold = float(settings.DAY_TRADE_EXIT_UNKNOWN_HOLD_SEC)
            logger.error(
                f"[DayExit] 🚨 {stock_code} 매도 주문 결과 불명(접수 미확인) — {_hold:.0f}초 동안 "
                f"같은 종목 재매도 보류. 다음 시도 직전에 당일 주문내역·실보유를 다시 확인한다"
            )
        if token is None or _exit_tokens.get(stock_code) is token:
            _release_exit_in_progress(stock_code, _hold, token=token)


# ── [H-1 2026-10-02] 주문 응답 분류 ─────────────────────────────────────────────
# korea_investment_api.place_order 가 예외를 msg_cd 로 구분해 준다:
#   전송 전 실패(요청이 나가지 않음) → 재매도 안전 = rejected
#   전송 후 응답 유실(ReadTimeout·연결 끊김·5xx 재시도 초과) → 접수 여부 불명 = unknown
# [L-2 r5] safe_place_order 가 붙이는 ORDER_LOCK_BUSY(분산락 미획득)·KIS_PRESEND_EXCEPTION
# (place_order 호출 전 예외)도 전송 전이다 — trade_executor.SAFE_ORDER_MSG_* 와 같은 값.
# (순환 import 를 피하려고 리터럴로 둔다. 값이 바뀌면 이 집합도 함께 바꿀 것.)
_PRESEND_MSG_CDS = frozenset({
    "TOKEN_UNAVAILABLE", "KIS_PRESEND_CONNECT", "KIS_PRESEND_CIRCUIT_OPEN",
    "KIS_PRESEND_EXCEPTION", "ORDER_LOCK_BUSY",
})
_TRANSPORT_UNKNOWN_MSG_CD = "KIS_TRANSPORT_UNKNOWN"
_RATE_LIMIT_MSG_CDS = frozenset({"EGW00201"})
# safe_place_order 가 분산락을 못 잡으면 주문을 보내지 않고 이 msg1 로 돌려준다(전송 전).
_ORDER_LOCK_BUSY_MSG = "order_lock_busy"
# [H-2] 결과 불명 주문이 아직 확인되지 않은 종목
#   {(날짜, code): {"qty": 주문수량, "since": datetime, "total_qty": 주문 시 DT 보유,
#                   "reason"/"price"/"sell_pct": 원 주문의 청산 사유·판정가·매도비율 (R4-2)}}
# [L-1 r5] 키에 날짜를 넣고 날짜가 바뀌면 정리한다(_prune_exit_state_for_today).
_unresolved_exit_unknown: dict[tuple[str, str], dict] = {}
# [M-3 r5] 불명 확인용 조회(잔고·비DT 수량) 연속 실패 횟수 {(날짜, code): n}, 경보 보낸 키
_unknown_lookup_failures: dict[tuple[str, str], int] = {}
_unknown_lookup_alerted: set[tuple[str, str]] = set()
_alert_tasks: set = set()
# 종목별 마지막 주문 응답 요약(분류·msg_cd·msg1) — stale 재시도 산정용
_last_exit_order_detail: dict[str, dict] = {}
# 이 프로세스가 접수 확인한(finalize 한) DT 매도 주문번호 {(날짜, 정규화 ODNO)} — 불명 주문
# 매칭에서 제외한다(이미 반영한 우리 주문을 다른 불명 주문으로 다시 세지 않게).
_finalized_sell_odnos: set[tuple[str, str]] = set()


def _today_key() -> str:
    return _kst_now().date().isoformat()


def _norm_odno(v) -> str:
    s_ = str(v or "").strip()
    return s_.lstrip("0") or ("0" if s_ else "")


def _prune_exit_state_for_today() -> None:
    """[L-1 r5] 날짜가 바뀐 불명·ODNO·조회실패 상태를 버린다(전일 상태가 당일 판정에 새지 않게)."""
    today = _today_key()
    for d in (_unresolved_exit_unknown, _unknown_lookup_failures):
        for k in [k for k in d if k[0] != today]:
            d.pop(k, None)
    for coll in (_finalized_sell_odnos, _unknown_lookup_alerted):
        for k in [k for k in coll if k[0] != today]:
            coll.discard(k)


def _classify_order_response(resp) -> str:
    """주문 응답 → "accepted" | "rejected" | "unknown".

    rejected = 전송 전 실패(msg_cd ∈ _PRESEND_MSG_CDS, 분산락 busy) 또는 KIS 업무 응답(rt_cd
    가 "0"/"-1" 이 아닌 값). msg_cd 없는 rt_cd "-1"(래퍼가 감싼 예외)·전송 불명은 unknown.
    """
    if not isinstance(resp, dict) or resp.get("rt_cd") is None:
        return "unknown"
    rt = str(resp.get("rt_cd"))
    mc = str(resp.get("msg_cd") or "")
    if rt == "0":
        return "accepted"
    if mc in _PRESEND_MSG_CDS or str(resp.get("msg1") or "") == _ORDER_LOCK_BUSY_MSG:
        return "rejected"
    if mc == _TRANSPORT_UNKNOWN_MSG_CD or rt == "-1":
        return "unknown"
    return "rejected"


def _is_transient_order_failure(detail: dict | None) -> bool:
    """[M-3] 횟수에 세지 않는 일시 실패 — 레이트리밋·전송 전·전송 불명·응답 없음."""
    if not detail:
        return True
    if detail.get("outcome") == "unknown":
        return True
    mc = str(detail.get("msg_cd") or "")
    m1 = str(detail.get("msg1") or "")
    return (
        mc in _PRESEND_MSG_CDS or mc in _RATE_LIMIT_MSG_CDS or mc == _TRANSPORT_UNKNOWN_MSG_CD
        or any(c in m1 for c in _RATE_LIMIT_MSG_CDS) or m1 == _ORDER_LOCK_BUSY_MSG
    )


def _row_is_live_or_filled_sell(row: dict, stock_code: str) -> bool:
    """[M-4] 당일 주문내역 행이 '이 종목의 유효한 매도 접수'인지.

    거부(rjct_qty>0)·취소(cncl_yn=Y)이면서 체결 0 인 행은 접수로 보지 않는다
    (fill_reconciler.is_terminal_fill 과 같은 필드 기준).
    """
    if not isinstance(row, dict):
        return False
    if str(row.get("pdno", stock_code) or stock_code) != stock_code:
        return False
    if str(row.get("sll_buy_dvsn_cd", "")) != "01":
        return False
    try:
        oq = int(float(row.get("ord_qty", 0) or 0))
        cq = int(float(row.get("tot_ccld_qty", 0) or 0))
        rj = int(float(row.get("rjct_qty", 0) or 0))
    except (TypeError, ValueError):
        return False
    if oq <= 0:
        return False
    cancelled = str(row.get("cncl_yn", "") or "").upper() == "Y"
    if cq == 0 and (rj > 0 or cancelled):
        return False
    return True


def _find_accepted_sell_row(rows, stock_code: str, qty: int | None, since) -> dict | None:
    """[H-2/M-4] 주문 수량·시각 범위까지 맞는 유효 매도 행. since=None 이면 시각 무관."""
    _since_hms = None
    if since is not None:
        _w = float(settings.DAY_TRADE_EXIT_UNKNOWN_MATCH_WINDOW_SEC)
        _since_hms = (since - timedelta(seconds=_w)).strftime("%H%M%S")
    for r in rows or []:
        if not _row_is_live_or_filled_sell(r, stock_code):
            continue
        if qty is not None:
            try:
                if int(float(r.get("ord_qty", 0) or 0)) != int(qty):
                    continue
            except (TypeError, ValueError):
                continue
        if _since_hms is not None:
            t = str(r.get("ord_tmd", "") or "")
            if t and t < _since_hms:
                continue
        odno = _norm_odno(r.get("odno", ""))
        if odno and (_today_key(), odno) in _finalized_sell_odnos:
            continue
        return r
    return None


async def _lookup_todays_orders_impl(stock_code: str) -> list | None:
    try:
        return await asyncio.wait_for(
            korea_investment_api.get_todays_orders(stock_code),
            timeout=float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC),
        )
    except asyncio.TimeoutError:
        logger.warning(f"[DayExit] 당일 주문내역 조회 시간초과 ({stock_code}) — 접수 미확인")
    except Exception as e:
        logger.warning(f"[DayExit] 당일 주문내역 조회 실패 ({stock_code}): {e} — 접수 미확인")
    return None


async def _fetch_non_dt_active_qty_impl(stock_code: str) -> int | None:
    """[H-2 r5] 같은 종목의 비-DT(스윙·유입·legacy NULL) active ActiveTrade 수량 합. 실패면 None.

    결과 불명 매도의 재매도 상한을 정하려면 계좌 보유분 중 **DT 몫**을 알아야 하는데,
    DB 없이 얻을 수 있는 소스가 없다: Portfolio 캐시(portfolio_cache_bridge)는 stock_code
    UNIQUE 로 스윙+DT 가 합산된 한 행이고(§3.5), 스윙 보유 인메모리 캐시는 존재하지 않는다
    (swing/position_manager 는 전부 DB 세션 함수). close_stale_positions 와 같은 소스·같은
    필터(not_dt_strategy_clause)를 쓴다. 결과 불명 이후에만 호출되는 드문 경로다.
    """
    try:
        from app.core.database import AsyncSessionLocal
        from app.models.models import ActiveTrade
        from app.services.db_worker import not_dt_strategy_clause
        from sqlalchemy import select as _select

        # DB-READ-EXCEPTION: §3.2 ③ 과 같은 읽기 전용 SELECT 1건, commit/flush/add 없음.
        # 쓰기는 db_worker 경유. 결과 불명 매도의 재매도 상한 확인에서만 호출된다.
        async with AsyncSessionLocal() as _db:
            _rows = (await _db.execute(
                _select(ActiveTrade).where(
                    ActiveTrade.stock_code == stock_code,
                    ActiveTrade.status == "active",
                    not_dt_strategy_clause(ActiveTrade.strategy_name),
                )
            )).scalars().all()
        return sum(int(r.quantity or 0) for r in _rows)
    except Exception as e:
        logger.warning(f"[DayExit] 비-DT 보유 수량 조회 실패({stock_code}): {e}")
        return None


async def _resolve_unknown_exit_v2_impl(
    stock_code: str, qty: int, since, total_qty: int,
) -> tuple[str, dict | None, int]:
    """[H-2 r5] 결과 불명 매도 판정 → (status, 접수 응답 dict|None, DT 매도가능수량).

    status:
      "accepted"      — 계좌의 DT 몫이 주문 수량만큼 줄었다(접수·대기·체결 모두 주문가능수량에서
                        빠진다). 당일 주문내역에 맞는 행이 있으면 그 ODNO 로, 없으면 ODNO 없이 반영.
      "not_accepted"  — DT 몫이 줄지 않았다 → 주문이 접수되지 않음. 재매도 상한 = DT 매도가능수량.
      "lookup_failed" — 잔고 또는 비-DT 수량 조회 실패 → 판단 불가(재매도 금지, 보류 유지).

    **판정의 근거는 잔고**다. `get_todays_orders` 는 조회 실패에도 `[]` 를 돌려줘
    "실패"와 "없음"을 응답 형태로 구분할 수 없다(korea_investment_api 쪽 계약) → 주문내역은
    ODNO 를 얻는 보조 소스로만 쓰고, 행이 있어도 잔고 감소가 없으면 접수로 보지 않는다
    (같은 수량·시각의 스윙 매도를 DT 접수로 오인하지 않게).
      DT 매도가능수량 = max(0, min(주문가능, 보유) − 비-DT active 수량)
    """
    h, _complete = await _fetch_kis_holdings_v3_impl()
    if h is not None and _holding_unknown_v2(h, _complete, stock_code):
        # [R3-1] 응답에 없는데 첫 페이지가 가득 참 → 0주(=접수)로 확정하지 않는다(판단 불가·보류).
        logger.warning(
            f"[DayExit] {stock_code} 결과 불명 매도 판정 보류 — KIS 잔고 첫 페이지 가득 참·종목 미발견"
            f"(다음 페이지에 있을 수 있음)"
        )
        return "lookup_failed", None, 0
    nd = await _fetch_non_dt_active_qty_impl(stock_code) if h is not None else None
    if h is None or nd is None:
        return "lookup_failed", None, 0
    _hh = h.get(stock_code) or {}
    hldg = int(_hh.get("hldg", 0) or 0)
    sellable = int(_hh.get("sellable", 0) or 0)
    dt_sellable = max(0, min(sellable, hldg) - int(nd))
    # [C-1 r9] v3 가 '보유 기준 DT 몫 온전 여부'를 보도록 판정 근거를 남긴다(KIS 재조회 없이)
    _last_unknown_resolve_snapshot[stock_code] = (hldg, sellable, int(nd))
    if dt_sellable <= max(0, int(total_qty) - int(qty)):
        rows = await _lookup_todays_orders_impl(stock_code)
        row = _find_accepted_sell_row(rows, stock_code, qty, since)
        odno = str((row or {}).get("odno", "") or "")
        logger.warning(
            f"[DayExit] {stock_code} 결과 불명 매도 → 접수로 판정: DT 매도가능 {dt_sellable}주 "
            f"(보유 {hldg}·주문가능 {sellable} − 비DT {nd}) ≤ 주문 전 {total_qty} − {qty}"
            f" / 주문내역 ODNO={odno or '미확인'}"
        )
        return "accepted", {"rt_cd": "0", "msg1": "resolved_by_holdings", "output": {"ODNO": odno}}, dt_sellable
    return "not_accepted", None, dt_sellable


def _note_unknown_lookup_failure(stock_code: str) -> None:
    """[M-3 r5] 불명 확인 조회 실패 — error(레이트리밋) + N회 연속이면 텔레그램 수동 청산 경보."""
    key = (_today_key(), stock_code)
    n = _unknown_lookup_failures.get(key, 0) + 1
    _unknown_lookup_failures[key] = n
    _error_throttled(
        f"dt_exit:unknown_lookup_failed:{stock_code}",
        f"[DayExit] 🚨 {stock_code} 직전 매도 결과 불명 + 잔고/비DT 수량 조회 실패({n}회 연속) — "
        f"재매도 보류 중(손절 미집행). 수동 확인 필요",
    )
    _limit = int(settings.DAY_TRADE_EXIT_UNKNOWN_LOOKUP_ALERT_COUNT)
    if n >= _limit and key not in _unknown_lookup_alerted:
        _unknown_lookup_alerted.add(key)
        try:
            from app.services.telegram_notifier import telegram_notifier
            _t = asyncio.get_running_loop().create_task(asyncio.wait_for(
                telegram_notifier.notify_system_warning(
                    "DT_EXIT_UNKNOWN",
                    f"{stock_code} 데이트레이딩 매도 결과 불명 + 확인 조회 {n}회 연속 실패 — "
                    f"손절이 보류된 상태입니다. 계좌를 확인하고 수동 청산하세요.",
                ),
                timeout=10.0,
            ))
            _alert_tasks.add(_t)
            _t.add_done_callback(_alert_tasks.discard)
        except Exception as e:
            logger.error(f"[DayExit] 수동 청산 경보 발송 실패({stock_code}): {e}")


async def _fetch_dt_ledger_net_qty_impl(stock_code: str) -> int | None:
    """[C-1 r9] DT 원장(TradeHistory strategy_type='daytrading') 순체결수량. 실패면 None.

    창은 DAY_TRADE_ORPHAN_LOOKBACK_DAYS(전일 포지션 경로도 덮도록 당일만이 아니다).
    집계는 _dt_net_qty_from_history — BUY 는 COMPLETED 만, SELL 은 FAILED/REJECTED 외 전부
    (PENDING 포함)를 차감한다. 결과 불명 매도의 재매도 상한에서만 호출되는 드문 경로다.
    """
    try:
        from app.core.database import AsyncSessionLocal
        from app.models.models import TradeHistory
        from sqlalchemy import select as _select

        _since = datetime.combine(
            _kst_now().date() - timedelta(days=int(settings.DAY_TRADE_ORPHAN_LOOKBACK_DAYS)),
            datetime.min.time(),
        )
        # DB-READ-EXCEPTION: §3.2 ③-c 와 같은 읽기 전용 SELECT 1건, commit/flush/add 없음.
        # 쓰기는 db_worker 경유. 결과 불명 매도의 재매도 상한 확인에서만 호출된다.
        async with AsyncSessionLocal() as _db:
            _rows = (await _db.execute(
                _select(TradeHistory).where(
                    TradeHistory.stock_code == stock_code,
                    TradeHistory.strategy_type == "daytrading",
                    TradeHistory.created_at >= _since,
                )
            )).scalars().all()
        return int(_dt_net_qty_from_history(_rows).get(stock_code, 0))
    except Exception as e:
        logger.error(f"[DayExit] DT 원장 순체결 조회 실패({stock_code}): {e}")
        return None


# [C-1 r9] v2 판정 근거 스냅샷 {code: (hldg, sellable, 비DT)} — v3 가 즉시 pop 한다
_last_unknown_resolve_snapshot: dict[str, tuple[int, int, int]] = {}

# [C-1 r9] 재매도 보류 상태(판정 불가) — lookup_failed 와 같은 처리(보류·N회 경보)
_UNKNOWN_HOLD_STATUSES = ("lookup_failed", "ambiguous", "ledger_failed", "orders_failed", "ledger_nonpositive")


async def _lookup_todays_orders_v2_impl(stock_code: str) -> list | None:
    """[C-1 r9 2차] 당일 주문내역 — **조회 실패·시간초과면 None**(get_todays_orders_v2 계약).

    v1(_lookup_todays_orders_impl)은 실패에도 [] 를 받아 "주문 없음"과 구분할 수 없었다.
    호출부는 None 을 판정 보류(재매도 금지)로 취급한다.
    """
    try:
        return await asyncio.wait_for(
            korea_investment_api.get_todays_orders_v2(stock_code),
            timeout=float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC),
        )
    except asyncio.TimeoutError:
        logger.warning(f"[DayExit] 당일 주문내역(v2) 조회 시간초과 ({stock_code}) — 판정 보류")
    except Exception as e:
        logger.warning(f"[DayExit] 당일 주문내역(v2) 조회 실패 ({stock_code}): {e} — 판정 보류")
    return None


def _find_accepted_sell_row_strict_v2(rows, stock_code: str, qty: int, since) -> dict | None:
    """[C-1 r9 2차] 불명 주문 등록 시각 이후·수량 일치·유효(체결>0 또는 접수 상태) 매도 행.

    _find_accepted_sell_row 와 달리 **주문시각(ord_tmd)이 없거나 since 가 없으면 매칭하지 않는다**
    (시각 근거 없는 같은 수량 행 — 스윙 매도 등 — 을 DT 접수로 오인하지 않게). since 에서
    DAY_TRADE_EXIT_UNKNOWN_MATCH_WINDOW_SEC 만큼만 앞당겨 시계 오차를 허용한다.
    """
    if since is None:
        return None
    _w = float(settings.DAY_TRADE_EXIT_UNKNOWN_MATCH_WINDOW_SEC)
    _since_hms = (since - timedelta(seconds=_w)).strftime("%H%M%S")
    for r in rows or []:
        if not _row_is_live_or_filled_sell(r, stock_code):
            continue
        try:
            if int(float(r.get("ord_qty", 0) or 0)) != int(qty):
                continue
        except (TypeError, ValueError):
            continue
        t = str(r.get("ord_tmd", "") or "")
        if not t or t < _since_hms:
            continue
        odno = _norm_odno(r.get("odno", ""))
        if odno and (_today_key(), odno) in _finalized_sell_odnos:
            continue
        return r
    return None


async def _resolve_unknown_exit_v3_impl(
    stock_code: str, qty: int, since, total_qty: int,
) -> tuple[str, dict | None, int]:
    """[C-1 r9, 2차 2026-10-12] v2 판정 + HTS·수동 보유분(ActiveTrade 없음) 재매도 방지.

    v2 는 "DT 몫 = min(주문가능, 보유) − 비-DT active" 로 본다. ActiveTrade 없는 수동 보유분 M 이
    있으면 원 주문이 체결됐어도 not_accepted 가 되고 재시도가 수동 물량을 판다. 결과 불명 SELL 은
    TradeHistory 행을 남기지 않아 원장도 매수 수량 그대로다(원장만으로는 못 막는다).
    v2 가 not_accepted 일 때 순서대로:
      1) 당일 주문내역 v2(실패면 None). None → "orders_failed"(보류). 등록 시각 이후·수량 일치·유효
         매도 행이 있으면 **온전 여부와 무관하게** accepted.
      2) 행 없음 + 보유 기준 DT 몫(보유 − 비DT) < 주문 전 수량 → "ambiguous"(보류).
      3) 원장 순체결 실패 → "ledger_failed", ≤ 0 → "ledger_nonpositive"(둘 다 보류 + error·경보).
      4) 그 외(주문내역 [] = KIS 미접수 실증, 원장 > 0) → 재매도, 상한 min(DT 매도가능, 원장, q).
    ⚠️ 잔여: 주문내역이 첫 페이지에 안 보이면(당일 주문 다수) 4) 로 떨어져 수동 물량을 최대 q 주
    팔 수 있다. 같은 수량·등록 이후 스윙 매도는 1) 에서 DT 접수로 오인될 수 있다(과소매도 쪽).
    상한 계산은 try 밖. 보류 상태들은 _UNKNOWN_HOLD_STATUSES.
    """
    st, resp, dt_sellable = await _resolve_unknown_exit_v2_impl(stock_code, qty, since, total_qty)
    _snap = _last_unknown_resolve_snapshot.pop(stock_code, None)
    if st != "not_accepted":
        return st, resp, dt_sellable
    # 1) 주문내역(v2) — 조회 실패(None)면 보류. 행이 있으면 온전 판정과 무관하게 접수 확정.
    rows = await _lookup_todays_orders_v2_impl(stock_code)
    if rows is None:
        return "orders_failed", None, 0
    row = _find_accepted_sell_row_strict_v2(rows, stock_code, qty, since)
    if row is not None:
        odno = str(row.get("odno", "") or "")
        logger.warning(
            f"[DayExit] {stock_code} 결과 불명 매도 → 당일 주문내역 일치 행(등록 이후·{qty}주, "
            f"ODNO={odno or '미확인'})으로 접수 판정 (잔고상 DT 매도가능 {dt_sellable}주)"
        )
        return "accepted", {"rt_cd": "0", "msg1": "resolved_by_orders", "output": {"ODNO": odno}}, dt_sellable
    # 2) 행 없음 + 보유 기준 DT 몫이 온전하지 않음 → 판정 불가(보류)
    intact = _snap is None or (int(_snap[0]) - int(_snap[2])) >= int(total_qty)
    if not intact:
        logger.error(
            f"[DayExit] 🚨 {stock_code} 결과 불명 매도 판정 불가 — 보유 {_snap[0]}주 − 비DT {_snap[2]}주가 "
            f"주문 전 DT {total_qty}주보다 적은데 DT 매도가능 {dt_sellable}주가 주문 후 기대치"
            f"({max(0, int(total_qty) - int(qty))}주)보다 많고 주문내역 일치 행도 없음. "
            f"ActiveTrade 없는 수동·HTS 보유분일 수 있어 재매도 보류"
        )
        return "ambiguous", None, 0
    # 3) 원장 — 실패·순체결 ≤ 0 이면 보류(과다매도보다 보류, §3.5)
    ledger = await _fetch_dt_ledger_net_qty_impl(stock_code)
    if ledger is None:
        return "ledger_failed", None, 0
    if ledger <= 0:
        logger.error(
            f"[DayExit] 🚨 {stock_code} 결과 불명 매도 — DT 원장 순체결 {ledger}주(≤0)라 DT 몫을 확인할 수 "
            f"없음. 재매도 보류(원장 반영 지연이면 다음 확인에서 해소)"
        )
        return "ledger_nonpositive", None, 0
    # 4) 주문내역 [] 실증(KIS 가 접수한 적 없음) + 원장 > 0 → 재매도 허용, 상한 min(DT 매도가능, 원장, q)
    capped = min(int(dt_sellable), int(ledger), int(qty))
    if capped < dt_sellable:
        logger.warning(
            f"[DayExit] {stock_code} 결과 불명 재매도 상한 축소: DT 매도가능 {dt_sellable} → "
            f"min(원장 {ledger}, 원 주문 {qty}) = {capped}주"
        )
    return "not_accepted", None, capped


def _note_unknown_hold_v2(stock_code: str, status: str) -> None:
    """[C-1 r9] 보류 사유별 로그. 조회 실패 계열은 기존 N회 연속 텔레그램 경보를 공유한다."""
    if status == "orders_failed":
        _error_throttled(
            f"dt_exit:unknown_orders_failed:{stock_code}",
            f"[DayExit] 🚨 {stock_code} 직전 매도 결과 불명 + 당일 주문내역 조회 실패 — 재매도 보류 중",
        )
    if status == "ledger_nonpositive":
        _error_throttled(
            f"dt_exit:unknown_ledger_nonpos:{stock_code}",
            f"[DayExit] 🚨 {stock_code} 직전 매도 결과 불명 + DT 원장 순체결 ≤ 0 — 재매도 보류 중. 계좌 확인 필요",
        )
    if status == "ambiguous":
        _error_throttled(
            f"dt_exit:unknown_ambiguous:{stock_code}",
            f"[DayExit] 🚨 {stock_code} 직전 매도 결과 불명 + 접수 여부 판정 불가(수동 보유분 혼재 의심) — "
            f"재매도 보류 중. 계좌 확인 필요",
        )
    _note_unknown_lookup_failure(stock_code)


async def _place_and_finalize_exit_impl(
    stock_code: str, pos: dict, price: float, reason: str, sell_pct: float,
    quantity: int, total_qty: int, is_full_exit: bool, is_stale: bool,
) -> dict | None:
    """[H4] 매도 주문 + 체결 이후 상태 반영 (취소로부터 shield 되는 구간).

    주문 결과를 `_last_exit_order_outcome[stock_code]` 에 남긴다:
      accepted(rt_cd=="0" 또는 불명 후 당일 주문내역에서 접수 확인) /
      rejected(전송 전 실패·KIS 업무 거부) /
      unknown(전송 후 응답 유실·DAY_TRADE_EXIT_ORDER_TIMEOUT_SEC 초과 — 접수 미확인).

    불명이면 잔고(주문가능·보유 − 비-DT active)로 접수 여부를 판정하고(_resolve_unknown_exit_v2_impl),
    판정이 안 되면 DAY_TRADE_EXIT_UNKNOWN_HOLD_SEC 동안 같은 종목 재매도를 보류한 뒤 재시도
    직전에 다시 판정한다. 판정 불가(조회 실패)가 이어지면 손절이 보류된 채로 남는다 →
    DAY_TRADE_EXIT_UNKNOWN_LOOKUP_ALERT_COUNT 회 연속이면 텔레그램 수동 청산 경보.

    ⚠️ [L-4] **장 마감 직전(대략 15:29:40 이후) 불명**이면 보류·재확인 주기가 15:30 폐장을
    넘긴다. 그 경우 이 포지션은 **오버나이트로 남고**, 다음 영업일 자정 reset 이후
    `close_stale_positions()`(ActiveTrade 기준 전일 포지션 청산 — 실보유·타전략 차감 확인 포함)가
    장중 첫 사이클에 처리한다. 인메모리 불명 상태(_unresolved_exit_unknown)는 날짜 키라서
    다음 날 판정에 새지 않는다(_prune_exit_state_for_today).
    """
    from app.services.trade_executor import safe_place_order

    _prune_exit_state_for_today()
    _ukey = (_today_key(), stock_code)

    # [H-2] 직전 불명 주문이 미해결이면 새 주문 전에 다시 확인한다(잔고 기준).
    _prev = _unresolved_exit_unknown.get(_ukey)
    if _prev is not None:
        # [C-1 r9] v3 — 수동·HTS 보유분 혼재 시 not_accepted 오판 방지 + DT 원장 순체결 상한
        _st, _resp, _dt_sellable = await _resolve_unknown_exit_v3_impl(
            stock_code, int(_prev["qty"]), _prev["since"],
            int(_prev.get("total_qty", total_qty)),
        )
        if _st in _UNKNOWN_HOLD_STATUSES:
            # 판단 불가 — 재매도하지 않고 보류를 이어간다(guarded finally 가 HOLD 뒤 해제)
            _last_exit_order_outcome[stock_code] = "unknown"
            _note_unknown_hold_v2(stock_code, _st)
            return None
        _unknown_lookup_failures.pop(_ukey, None)
        if _st == "accepted":
            _unresolved_exit_unknown.pop(_ukey, None)
            _last_exit_order_outcome[stock_code] = "accepted"
            if not is_stale:
                _note_exit_order_accepted_impl(stock_code)  # [R4-3]
            _pq = min(int(_prev["qty"]), int(total_qty))
            # [R4-2 2026-10-09] 확정되는 것은 **원 주문**이다 — 이번 사이클의 판정(사유·가격)이
            # 아니라 등록 때 저장한 원 사유·가격·매도비율로 반영한다. 이번 사유로 확정하면
            # 손절이 시간손절로 기록돼 loss_tracker 에서 빠지거나(§5.1) 그 반대가 되고,
            # TradeHistory exit_reason·(ODNO 미발견 시) 순손익도 틀린다.
            # 옛 엔트리(키 없음)는 이번 호출 값으로 폴백(구 동작).
            _o_reason = _prev.get("reason") or reason
            try:
                _o_price = float(_prev.get("price") or 0) or price
            except (TypeError, ValueError):
                logger.warning(f"[DayExit] {stock_code} 불명 주문 원 가격 파싱 실패 — 이번 판정가 사용")
                _o_price = price
            _o_sell_pct = _prev.get("sell_pct", sell_pct)
            if _o_reason != reason:
                logger.warning(
                    f"[DayExit] {stock_code} 직전 결과 불명 매도 접수 확정 — 원 사유 '{_o_reason}'·"
                    f"가격 {_o_price:,.0f} 로 반영(이번 판정 '{reason}'·{price:,.0f} 아님)"
                )
            return await _finalize_after_accept_impl(
                stock_code, pos, _o_price, _o_reason, _o_sell_pct if _pq < total_qty else 1.0,
                _resp, _pq, total_qty, _pq >= total_qty, is_stale,
            )
        # not_accepted — 재매도 수량을 DT 매도가능수량으로 상한(스윙 물량 보호)
        if _dt_sellable <= 0:
            _unresolved_exit_unknown.pop(_ukey, None)
            _last_exit_order_outcome[stock_code] = "rejected"
            _error_throttled(
                f"dt_exit:unknown_no_sellable:{stock_code}",
                f"[DayExit] 🚨 {stock_code} 직전 매도 결과 불명 + DT 매도가능 0 — 재매도하지 않는다"
                f"(실보유·ActiveTrade 수동 확인)",
            )
            if not is_stale:
                # [R4-3] 매도 불가 상태도 청산 미집행 — 헬스·연속 거부 경보에 반영
                _note_exit_order_rejected_impl(
                    stock_code, quantity, "DT_SELLABLE_ZERO", "직전 결과 불명 + DT 매도가능 0", reason,
                )
            return None
        if _dt_sellable < quantity:
            logger.warning(
                f"[DayExit] {stock_code} 직전 주문 결과 불명 — DT 매도가능 {_dt_sellable}주로 "
                f"매도 수량 축소({quantity}→{_dt_sellable})"
            )
            quantity = _dt_sellable
            is_full_exit = quantity >= total_qty
        _unresolved_exit_unknown.pop(_ukey, None)

    _last_exit_order_outcome[stock_code] = "unknown"
    _since = _kst_now()
    _timeout = float(settings.DAY_TRADE_EXIT_ORDER_TIMEOUT_SEC)
    order_resp = None
    try:
        order_resp = await asyncio.wait_for(
            safe_place_order(
                stock_code=stock_code,
                order_type="SELL",
                quantity=quantity,
                price=None,  # 시장가
            ),
            timeout=_timeout,
        )
    except asyncio.TimeoutError:
        logger.error(f"[DayExit] 매도 주문 응답 {_timeout:.0f}초 초과 ({stock_code}) — 결과 불명")

    outcome = _classify_order_response(order_resp)
    _r = order_resp if isinstance(order_resp, dict) else {}
    _last_exit_order_detail[stock_code] = {
        "outcome": outcome, "msg_cd": _r.get("msg_cd"), "msg1": _r.get("msg1"),
    }
    if outcome == "unknown":
        logger.error(
            f"[DayExit] 매도 주문 결과 불명 ({stock_code}): rt_cd={_r.get('rt_cd')} "
            f"msg_cd={_r.get('msg_cd')} msg1={_r.get('msg1')} — 당일 주문내역 확인"
        )
        _st, resolved, _ = await _resolve_unknown_exit_v2_impl(
            stock_code, quantity, _since, total_qty,
        )
        if _st != "accepted":
            # 아직 반영 전일 수 있다(주문가능수량 갱신 지연) → 기록만 하고 보류. 재시도 직전에 다시 확인.
            _unresolved_exit_unknown[_ukey] = {
                "qty": quantity, "since": _since, "total_qty": total_qty,
                # [R4-2] 접수 확정 시 원 주문 기준으로 반영하기 위한 원 사유·가격·매도비율
                "reason": reason, "price": price, "sell_pct": sell_pct,
            }
            if _st == "lookup_failed":
                _note_unknown_lookup_failure(stock_code)
            return None
        order_resp = resolved
        outcome = "accepted"
    if outcome == "rejected":
        _last_exit_order_outcome[stock_code] = "rejected"
        logger.error(
            f"[DayExit] 매도 주문 실패 ({stock_code}): msg_cd={_r.get('msg_cd')} "
            f"{_r.get('msg1', 'unknown')}"
        )
        if not is_stale:
            # [R4-3] 거부 연속 횟수 → 헬스 실패 + N 회째 텔레그램. 전일 경로는 _stale_sell_failures 담당.
            _note_exit_order_rejected_impl(
                stock_code, quantity, _r.get("msg_cd"), _r.get("msg1"), reason,
            )
        return None
    _last_exit_order_outcome[stock_code] = "accepted"
    if not is_stale:
        _note_exit_order_accepted_impl(stock_code)  # [R4-3]
    return await _finalize_after_accept_impl(
        stock_code, pos, price, reason, sell_pct, order_resp, quantity, total_qty,
        is_full_exit, is_stale,
    )


async def _finalize_after_accept_impl(
    stock_code: str, pos: dict, price: float, reason: str, sell_pct: float,
    order_resp: dict, quantity: int, total_qty: int, is_full_exit: bool, is_stale: bool,
) -> dict | None:
    """접수 확인된 매도의 상태 반영. [L-2] 반영 중 예외면 인메모리 최소 반영 + 불명 처리."""
    _odno = _norm_odno(((order_resp or {}).get("output") or {}).get("ODNO", ""))
    if _odno:
        _finalized_sell_odnos.add((_today_key(), _odno))
    # [2026-09-23] 여기서부터는 이미 체결된 주문의 상태 반영이다. 스케줄러의 단계
    # 타임아웃(_guard)이 이 지점을 취소하면 "브로커는 팔았는데 인메모리는 그대로" 상태가
    # 되어 다음 사이클이 같은 물량을 다시 판다 → shield 로 취소로부터 보호한다.
    try:
        if is_stale:
            return await asyncio.shield(_finalize_exit_stale(
                stock_code, pos, price, reason, order_resp, quantity, total_qty,
            ))
        return await asyncio.shield(_finalize_exit(
            stock_code, pos, price, reason, sell_pct, order_resp, quantity, total_qty, is_full_exit
        ))
    except Exception as e:
        logger.error(
            f"[DayExit] 🚨 {stock_code} 매도 접수 후 상태 반영 실패: {e} — 인메모리 최소 반영, "
            f"재매도 보류(결과 불명 처리)", exc_info=True,
        )
        async with day_trading_state.trade_lock:
            _cur = day_trading_state.positions.get(stock_code)
            if _cur is not None and int(_cur.get("quantity", 0) or 0) == int(total_qty):
                if is_full_exit:
                    day_trading_state.remove_position(stock_code)
                else:
                    _cur["quantity"] = int(total_qty) - int(quantity)
                    _cur["target_price"] = float("inf")
                    _cur["partial_sold"] = True
            if is_full_exit:
                day_trading_state.closed_today.add(stock_code)
        _last_exit_order_outcome[stock_code] = "unknown"
        return None


async def _finalize_exit(
    stock_code: str, pos: dict, price: float, reason: str, sell_pct: float,
    order_resp: dict, quantity: int, total_qty: int, is_full_exit: bool,
) -> dict | None:
    """체결 이후 상태 반영 (인메모리·DB·캐시). 취소되면 안 되는 구간.

    당일 포지션 전용 — 실현손익을 당일 daily_pnl / loss_streak_today 에 반영한다.
    (전일 포지션은 _finalize_exit_stale 사용 — M1)
    """
    return await _finalize_exit_impl(
        stock_code, pos, price, reason, sell_pct, order_resp,
        quantity, total_qty, is_full_exit, count_daily_pnl=True,
    )


async def _finalize_exit_stale(
    stock_code: str, pos: dict, price: float, reason: str,
    order_resp: dict, quantity: int, total_qty: int,
) -> dict | None:
    """[M1] 전일(stale) 포지션 체결 이후 상태 반영.

    당일 손익 집계(daily_pnl·loss_streak_today)에는 가산하지 않고,
    별도 누적값(_stale_realized_pnl)과 로그로만 추적한다.
    DB TradeHistory 기록은 동일하게 수행되므로 실제 손익 추적은 유지된다.
    """
    return await _finalize_exit_impl(
        stock_code, pos, price, reason, 1.0, order_resp,
        quantity, total_qty, True, count_daily_pnl=False,
    )


async def _finalize_exit_impl(
    stock_code: str, pos: dict, price: float, reason: str, sell_pct: float,
    order_resp: dict, quantity: int, total_qty: int, is_full_exit: bool,
    count_daily_pnl: bool = True,
) -> dict | None:
    """체결 이후 상태 반영 공통 구현 (신규 함수 — 기존 시그니처 불변).

    count_daily_pnl=False 이면 당일 손익/연속손실 집계를 건너뛰고
    전일 손익 누적값에만 더한다(전일 포지션 강제청산 경로).
    """
    entry_price = pos.get("entry_price", 0)
    stock_name = pos.get("stock_name", stock_code)
    order_id = order_resp.get("output", {}).get("ODNO", "")
    # [R6-1] 당일 매수분 고아는 stale 경로(count_daily_pnl=False)로 들어와도 당일 손익·연패에 넣는다.
    # 인메모리·closed_today·체결 0 복원 처리는 원래 플래그(count_daily_pnl)를 그대로 따른다.
    _count_today = bool(count_daily_pnl) or _is_orphan_today_position(pos)
    # [2026-10-01] 순손익(수수료·거래세 차감)으로 기록·집계한다.
    # 일일 손실한도(DAY_TRADE_DAILY_LOSS_LIMIT)·당일 연패·loss_tracker 가 모두 이 값을 본다
    # → 비용만큼 한도가 **약간 더 일찍** 발동한다(보수적 방향, 의도). 손익분기(gross≈0)
    # 청산도 순손익으로는 손실이다.
    _pnl = dt_net_pnl(entry_price, price, quantity, pos)
    gross_profit_loss = _pnl["gross_profit_loss"]
    trade_cost = _pnl["total_cost"]
    profit_loss = _pnl["net_profit_loss"]
    profit_loss_rate = _pnl["net_profit_loss_rate"] if entry_price > 0 else 0.0

    # 인메모리 상태 업데이트 (DB 직접 접근 금지)
    _recent_dt_sell_finalized[stock_code] = time.monotonic()
    # [R2-4 2026-10-09] 왕복(분할 익절 포함) 순손익. 분할 매도 레그의 순손익을 포지션에
    # 누적(realized_net)해 두고, 전량 청산 때 마지막 레그와 합쳐 승패(당일 연패·loss_tracker)를
    # 판정한다. 예전에는 마지막 레그만 봐 +1.5R 분할 후 잔량 -0.1% 청산이 '손실'로 셌다.
    # ⚠️ 인메모리 전용 — 재기동하면 누적이 사라져 마지막 레그 기준(구 동작)으로 돌아간다.
    _prior_realized = 0.0
    # [R4-1 2026-10-09] 분할 레그가 0체결로 종결되면 _on_unfilled 가 되돌릴 수 있도록
    # 분할 전 목표가·partial_sold 를 보관한다(아래에서 positions[code] 를 inf/True 로 바꾸기 전).
    _pre_partial_state: dict | None = None
    async with day_trading_state.trade_lock:
        _mem_pos = day_trading_state.positions.get(stock_code)
        if not is_full_exit and _mem_pos is not None:
            _pre_partial_state = {
                "target_price": _mem_pos.get("target_price"),
                "partial_sold": bool(_mem_pos.get("partial_sold", False)),
            }
        _src = _mem_pos if _mem_pos is not None else (pos or {})
        try:
            _prior_realized = float(_src.get("realized_net", 0.0) or 0.0)
        except (TypeError, ValueError):
            logger.warning(f"[DayExit] {stock_code} realized_net 파싱 실패 — 마지막 레그 기준 판정")
            _prior_realized = 0.0
        # [HL2-2] 0체결 복원 포지션이면 같은 왕복의 승패가 이미 기록돼 있다(표식).
        _prev_streak_mark = _read_outcome_mark_impl(_src, _RESTORED_STREAK_MARK_KEY)
        _prev_lt_mark = _read_outcome_mark_impl(_src, _RESTORED_LOSS_TRACKER_MARK_KEY)
        _streak_mark = _prev_streak_mark
        if is_full_exit:
            day_trading_state.remove_position(stock_code)
            # [2026-10-08] DT 전용 청산 집계(엔진 카드 '청산 N') — 인메모리에 없던 전일·고아 포지션 포함
            day_trading_state.dt_exited_today.add(stock_code)
            if not count_daily_pnl:
                # [M10] DB에서만 발견된 전일 포지션은 인메모리에 없어 remove_position()의
                # closed_today 추가가 일어나지 않는다 → 강제청산한 종목이 같은 날
                # scan_and_enter 후보로 되살아난다. 명시적으로 재진입 금지 집합에 넣는다.
                day_trading_state.closed_today.add(stock_code)
        else:
            # 분할 매도: 잔량 유지, target_price 제거(트레일링만 적용)
            remaining = total_qty - quantity
            day_trading_state.positions[stock_code]["quantity"] = remaining
            day_trading_state.positions[stock_code]["target_price"] = float("inf")
            day_trading_state.positions[stock_code]["partial_sold"] = True
            day_trading_state.positions[stock_code]["realized_net"] = _prior_realized + profit_loss
        if _count_today:
            if not count_daily_pnl:
                logger.warning(
                    f"[DayExit] 당일 매수분 고아 청산 실현손익은 당일 집계에 포함: {stock_code} "
                    f"{profit_loss:+.0f}원 (일일 손실한도·당일 연패 반영)"
                )
            if is_full_exit and (_prior_realized != 0.0 or _prev_streak_mark is not None):
                # daily_pnl 은 이 레그만 더하고, 연패는 왕복 순손익으로 판정한다.
                day_trading_state.update_pnl(profit_loss, is_full_close=False)
                _rt_streak = _prior_realized + profit_loss
                if _rt_streak != 0.0 and _should_record_outcome_impl(_prev_streak_mark, _rt_streak < 0):
                    _apply_round_trip_streak_v2(_rt_streak)
                    _streak_mark = _rt_streak < 0
                elif _rt_streak != 0.0:
                    logger.info(
                        f"[DayExit] {stock_code} 매도 미체결 복원 포지션 재청산 — 당일 연패는 이미 "
                        f"기록됨(이전 손실={_prev_streak_mark}) → 왕복 {_rt_streak:+.0f}원 재집계 생략"
                    )
            else:
                day_trading_state.update_pnl(profit_loss, is_full_close=is_full_exit)
                if is_full_exit and profit_loss != 0.0:
                    _streak_mark = profit_loss < 0
        else:
            _stale_now = _add_stale_realized_pnl(profit_loss)
            logger.warning(
                f"[DayExit] 전일 포지션 실현손익은 당일 집계 제외: {stock_code} "
                f"{profit_loss:+.0f}원 (오늘 처리한 전일·고아 누적={_stale_now:+.0f}원)"
            )

    # [M-2 2026-10-02] 인메모리 반영이 끝났으므로 청산 진행 중 표식을 **지금** 해제한다.
    # 예전에는 아래 꼬리 단계(Redis 저장·loss_tracker·포트폴리오 캐시) 뒤에 해제돼, 그중
    # 하나가 멈추면 표식이 영구 잔류 → EOD 강제청산이 잔량을 계속 건너뛰었다.
    _release_exit_in_progress(stock_code)

    # DB 기록 (큐 경유) — 원장·ActiveTrade 를 꼬리 단계보다 먼저 남긴다
    await _bounded_finalize_step(enqueue_trade_history({
        "stock_code": stock_code,
        "stock_name": stock_name,
        "order_type": "SELL",
        "order_price": price,
        "order_quantity": quantity,
        "executed_price": price,
        "executed_quantity": quantity,
        "total_amount": price * quantity,
        "status": "COMPLETED",
        "order_id": order_id,
        "exit_reason": reason,
        "commission": trade_cost,
        "net_profit_loss": profit_loss,
        "profit_loss_rate": profit_loss_rate,
        "strategy_type": "daytrading",
    }), stock_code, "TradeHistory 큐잉")
    # [R2-1 2026-10-09] loss_tracker 기록 시점 결정.
    # 15:20~15:30 은 KRX 종가 단일가 구간이라 그때 낸 시장가 매도는 15:30 에 체결된다.
    # 주문 직후 체결 확인(수 초)은 체결 0·미종결만 보고 끝나므로, 이 구간 매도는 15:30 이후
    # 재확인한 **실제 체결가 기준 왕복 순손익**으로 loss_tracker 승패를 기록한다(_on_settled).
    # 재확인이 실패하면 지금 값(신호가 기준)으로 기록한다 — 구 동작으로 폴백.
    _record_loss_now = False
    _on_settled = None
    _auction = False
    if is_full_exit and not is_time_stop_exit_reason(reason):
        _auction = bool(
            settings.DAY_TRADE_FILL_CHECK_ENABLED and order_id
            and _in_closing_auction_impl(_kst_now())
        )
        if _auction:
            async def _on_settled(leg_net):
                if leg_net is None:
                    logger.warning(
                        f"[DayExit] {stock_code} 종가 단일가 매도 전량 체결 미확인(체결 0·부분) — "
                        f"loss_tracker 기록 보류(잔량이 청산될 때 기록)"
                    )
                    return
                _rt = _prior_realized + float(leg_net)
                if not _should_record_outcome_impl(_prev_lt_mark, _rt < 0):
                    logger.info(
                        f"[DayExit] {stock_code} 매도 미체결 복원 포지션 재청산 — loss_tracker 는 이미 "
                        f"기록됨(이전 손실={_prev_lt_mark}) → 왕복 {_rt:+.0f}원 재기록 생략"
                    )
                    return
                logger.info(
                    f"[DayExit] {stock_code} 종가 단일가 체결 확인 후 연속손실 기록: "
                    f"왕복 순손익={_rt:+.0f}원 (신호가 기준 {_prior_realized + profit_loss:+.0f}원)"
                )

                async def _rec():
                    await loss_tracker.record_trade_result_safe(stock_code, _rt < 0)
                    await loss_tracker.save_to_db(stock_code=stock_code)

                await _bounded_finalize_step(_rec(), stock_code, "loss_tracker 기록(체결 확인 후)")
        else:
            _record_loss_now = True
    # [2026-10-01] 실제 체결가 확인 — 백그라운드(루프 비차단). 확인되면 원장·당일 손익 보정.
    _scheduled = None
    _snap_pos = dict(pos)
    if _pre_partial_state is not None:
        _snap_pos[_PRE_PARTIAL_STATE_KEY] = _pre_partial_state
    # [HL2-2] 전량 청산 접수 때 기록한(할) 승패와 그 시점 왕복 누적을 스냅샷에 싣는다.
    # 0체결·부분 종결로 복원되면 _on_unfilled 가 이 값으로 복원 포지션에 표식·realized_net 을 남긴다.
    _lt_will_record = False
    _lt_mark = _prev_lt_mark
    if _record_loss_now:
        _lt_is_loss = (_prior_realized + profit_loss) < 0
        _lt_will_record = _should_record_outcome_impl(_prev_lt_mark, _lt_is_loss)
        if _lt_will_record:
            _lt_mark = _lt_is_loss
    if is_full_exit:
        _snap_pos[_FULL_EXIT_RECORD_KEY] = {
            "streak": _streak_mark if _count_today else None,
            "loss_tracker": _lt_mark,
            "prior_realized": _prior_realized,
        }
    try:
        _scheduled = _schedule_sell_fill_reconcile_v2(
            stock_code=stock_code, pos=_snap_pos, order_id=order_id, signal_price=price,
            quantity=quantity, entry_price=entry_price, recorded_net=profit_loss,
            count_daily_pnl=count_daily_pnl, is_partial=not is_full_exit,
            on_settled=_on_settled, closing_auction=_auction,
        )
    except Exception as e:
        logger.warning(f"[DayExit] {stock_code} 체결가 확인 예약 실패: {e}")
    if _on_settled is not None and _scheduled is None:
        # 재확인 예약 실패 → 지연 기록 주체가 없다. 지금 값으로 기록(구 동작).
        logger.warning(f"[DayExit] {stock_code} 종가 단일가 재확인 예약 실패 — 신호가 기준으로 연속손실 기록")
        _record_loss_now = True
    if is_full_exit:
        await _bounded_finalize_step(
            enqueue_close_active_trade(stock_code, strategy_name="day_trade"),
            stock_code, "ActiveTrade close 큐잉",
        )
    else:
        # 분할 매도 후 ActiveTrade 잔량 갱신 — 누락 시 재기동 복원에서 원래 수량이 되살아나
        # 다음 청산이 보유량보다 많이 매도한다(2026-09-23 LG씨엔에스 4주 보유 중 7주 매도).
        from app.services.db_worker import enqueue_active_trade_upsert
        # [M1 2026-10-02] 분할매도 상태 영속화 — remaining_pct(<1.0)·highest_since_entry 를
        # 함께 남긴다(둘 다 기존 ORM 컬럼 → db_worker payload 계약 무변경, 선택 키).
        # restore_from_db 가 remaining_pct<1.0 을 보고 목표가 분할 재발동을 막고,
        # 트레일링 고점을 이어받는다. Numeric(5,2) 반올림으로 1.00 이 되지 않게 0.99 상한.
        _remaining = total_qty - quantity
        _remaining_pct = min(0.99, round(_remaining / total_qty, 2)) if total_qty > 0 else 0.99
        _highest = float(pos.get("highest_price") or 0) or None
        _payload = {
            "stock_code": stock_code,
            "strategy_name": "day_trade",
            "quantity": _remaining,
            "remaining_pct": _remaining_pct,
        }
        if _highest:
            _payload["highest_since_entry"] = _highest
        await _bounded_finalize_step(
            enqueue_active_trade_upsert(_payload), stock_code, "ActiveTrade 잔량 큐잉",
        )

    # [F9 2026-09-28] 일일 안전장치(daily_pnl/loss_streak_today/closed_today) 영속화.
    # 값이 바뀐 직후에 저장한다 — 재기동해도 일일 -3% 한도·3연패 쿨다운·당일
    # 재진입 차단이 풀리지 않게 한다(락 밖에서 호출: Redis I/O 로 락을 잡지 않는다).
    # [M-2] Redis 무응답이면 상한에서 끊는다(core/redis socket_timeout 은 인프라 소유).
    await _bounded_finalize_step(day_trading_state.save_daily_state(), stock_code, "일일 안전장치 저장")

    # [P0] loss_tracker 연동 — 전량 매도 시 연속 손실 추적
    # [2026-10-01] 시간손절은 **중립**(손실로도 이익으로도 세지 않음 → 기록 자체를 생략).
    # 기회비용 청산까지 3연패 → CONSECUTIVE_LOSS_BAN_DAYS 진입금지에 넣으면 그 밴이
    # 스윙 진입(swing/entry_service)까지 막는다. 하드/폴백 손절·일일한도·EOD·트레일링 등
    # 나머지 청산은 기존대로 기록한다. (당일 연패 쿨다운 loss_streak_today 는 위 update_pnl 이
    # 그대로 센다 — DT 전용·당일 한정 스로틀이라 스윙에 번지지 않는다.)
    if is_full_exit:
        if is_time_stop_exit_reason(reason):
            logger.info(
                f"[DayExit] 시간손절은 연속손실 집계 제외(중립): {stock_code} "
                f"순손익={profit_loss:+.0f}원 — {settings.CONSECUTIVE_LOSS_BAN_COUNT}연패/"
                f"{settings.CONSECUTIVE_LOSS_BAN_DAYS}일 밴 카운트 변화 없음"
            )
        elif _record_loss_now:
            # [R2-4] 왕복 순손익(분할 익절 레그 포함)으로 승패 판정
            _round_trip = _prior_realized + profit_loss
            is_loss = _round_trip < 0
            if _prior_realized != 0.0:
                logger.info(
                    f"[DayExit] {stock_code} 왕복 순손익 {_round_trip:+.0f}원 "
                    f"(분할 {_prior_realized:+.0f} + 잔량 {profit_loss:+.0f}) 기준 승패 판정"
                )

            async def _record_loss():
                # record_trade_result() 후 반드시 save_to_db() (§5.1)
                await loss_tracker.record_trade_result_safe(stock_code, is_loss)
                await loss_tracker.save_to_db(stock_code=stock_code)

            # [HL2-2] 0체결 복원 포지션 재청산이면 같은 왕복을 다시 세지 않는다.
            if _should_record_outcome_impl(_prev_lt_mark, is_loss):
                await _bounded_finalize_step(_record_loss(), stock_code, "loss_tracker 기록")
            else:
                logger.info(
                    f"[DayExit] {stock_code} 매도 미체결 복원 포지션 재청산 — loss_tracker 는 이미 "
                    f"기록됨(이전 손실={_prev_lt_mark}) → 왕복 {_round_trip:+.0f}원 재기록 생략"
                )

    # 포트폴리오 화면에서 즉시 제거/차감
    from app.services.portfolio_cache_bridge import reflect_sell
    await _bounded_finalize_step(
        reflect_sell(stock_code, quantity, price, remaining_qty=max(0, int(total_qty) - int(quantity))),
        stock_code, "포트폴리오 캐시 반영",
    )

    sell_label = f"{int(sell_pct * 100)}%" if sell_pct < 1.0 else "전량"
    logger.info(
        f"[DayExit] ✅ 청산({sell_label}): {stock_code} {stock_name} "
        f"수량={quantity}/{total_qty} 순수익률={profit_loss_rate:.2f}% "
        f"(총손익={gross_profit_loss:+.0f} 비용={trade_cost:.0f} 순손익={profit_loss:+.0f}원) 이유={reason}"
    )

    return {
        "stock_code": stock_code,
        "stock_name": stock_name,
        "price": price,
        "quantity": quantity,
        "profit_loss": profit_loss,            # 순손익(비용 차감)
        "profit_loss_rate": profit_loss_rate,  # 순수익률(진입 원금 대비 %)
        "gross_profit_loss": gross_profit_loss,
        "trade_cost": trade_cost,
        "reason": reason,
        "sell_pct": sell_pct,
        "order_id": order_id,
    }


async def _bounded_finalize_step(coro, stock_code: str, label: str) -> None:
    """[M-2] 체결 후 꼬리 단계 1개를 상한(DAY_TRADE_EXIT_FINALIZE_STEP_TIMEOUT_SEC)으로 실행.

    한 단계가 멈추거나 실패해도 나머지 단계와 청산 결과 반환은 계속된다(error 로그).
    """
    _t = float(settings.DAY_TRADE_EXIT_FINALIZE_STEP_TIMEOUT_SEC)
    try:
        await asyncio.wait_for(coro, timeout=_t)
    except asyncio.TimeoutError:
        logger.error(f"[DayExit] 🚨 {stock_code} 청산 후 '{label}' {_t:.0f}초 초과 — 건너뜀(수동 확인)")
    except Exception as e:
        logger.error(f"[DayExit] 🚨 {stock_code} 청산 후 '{label}' 실패: {e}", exc_info=True)


def _schedule_sell_fill_reconcile(
    stock_code: str, pos: dict, order_id: str, signal_price: float, quantity: int,
    entry_price: float, recorded_net: float, count_daily_pnl: bool,
):
    """매도 실제 체결가 확인 예약 — 구 시그니처 유지용 위임(_schedule_sell_fill_reconcile_v2)."""
    return _schedule_sell_fill_reconcile_v2(
        stock_code=stock_code, pos=pos, order_id=order_id, signal_price=signal_price,
        quantity=quantity, entry_price=entry_price, recorded_net=recorded_net,
        count_daily_pnl=count_daily_pnl,
    )


def _schedule_sell_fill_reconcile_v2(
    stock_code: str, pos: dict, order_id: str, signal_price: float, quantity: int,
    entry_price: float, recorded_net: float, count_daily_pnl: bool,
    is_partial: bool = False, on_settled=None, closing_auction: bool = False,
):
    """매도 실제 체결가 확인을 백그라운드로 예약 (실시간 루프 비차단).

    [R2-1/R2-4 2026-10-09] v2 추가 인자:
      - is_partial: 분할 매도 레그면 체결가 보정 차이(delta)를 잔량 포지션의 realized_net
        (왕복 순손익 누적)에도 반영한다.
      - on_settled(leg_net | None): 이 레그의 **확정** 순손익을 한 번 넘긴다(전량 체결 확인 →
        실제 체결가 기준, 확인 불가 → 신호가 기준 recorded_net, 체결 0·부분 → None).
      - closing_auction: 15:20~15:30 종가 단일가 주문. 즉시 확인이 종결이 아니면 15:30 이후
        (DAY_TRADE_FILL_AUCTION_RECHECK_*) 다시 확인한다.

    확인되면:
      - TradeHistory(order_id) 의 체결가·비용·순손익·수익률·슬리피지(%)를 db_worker 로 갱신
      - 기록했던 순손익과의 차이를 당일 daily_pnl 에 보정(전일 stale 청산은 _stale_realized_pnl —
        §7 당일 미반영 규칙 유지). 당일 연패·loss_tracker 판정은 청산 시점 값 그대로 둔다.
    확인 실패면 신호가 기록이 그대로 남는다(fill_reconciler 가 warning).
    """
    from app.core.timeutil import now_kst as _now_kst
    _order_date = _now_kst().date().isoformat()  # 주문일(KST) — 확인이 늦어져도 이 날짜로 조회
    _settled = [False]
    # [R6-1] 당일 매수분 고아: stale 경로(count_daily_pnl=False — 인메모리 복원 없이 재청산 대기)를
    # 유지하되 손익 보정은 당일 daily_pnl 로 보낸다(_finalize_exit_impl 의 당일 집계와 대칭).
    _orphan_today = (not count_daily_pnl) and _is_orphan_today_position(pos)

    async def _settle(leg_net) -> None:
        """on_settled 를 최대 1회 호출 (예외는 warning — 흐름 비차단)."""
        if _settled[0] or on_settled is None:
            _settled[0] = True
            return
        _settled[0] = True
        try:
            await on_settled(leg_net)
        except Exception as e:
            logger.warning(f"[DayExit] {stock_code} 매도 확정 콜백 실패: {type(e).__name__}: {e}", exc_info=True)

    def _adjust_partial_realized_impl(delta: float) -> None:
        """분할 레그 체결가 보정분을 잔량 포지션의 왕복 누적(realized_net)에 반영 (trade_lock 안)."""
        if not is_partial or not delta:
            return
        _cur = day_trading_state.positions.get(stock_code)
        if _cur is not None and _cur.get("partial_sold"):
            _cur["realized_net"] = float(_cur.get("realized_net", 0.0) or 0.0) + float(delta)

    async def _on_fill(filled_qty: int, avg_price: float) -> None:
        if filled_qty < quantity:
            logger.warning(
                f"[DayExit] 매도 부분체결 확인(미종결): {stock_code} {filled_qty}/{quantity}주 "
                f"order_id={order_id} — 체결분은 체결가, 미체결 {int(quantity) - int(filled_qty)}주는 "
                f"신호가 기준 추정으로 손익 반영. 인메모리 수량은 주문 수량 기준(수동 확인 필요)"
            )
        # [H-R1-2 2026-10-09] 미종결 부분체결이면 미체결분을 체결가로 계상하지 않는다
        # (_sell_fill_pnl_impl — 체결분 실측 + 미체결분 신호가 추정, 거래대금·체결수량은 체결분만).
        new = _sell_fill_pnl_impl(entry_price, avg_price, signal_price, filled_qty, quantity, pos)
        delta = new["net_profit_loss"] - recorded_net
        slippage_pct = ((avg_price - signal_price) / signal_price * 100) if signal_price > 0 else 0.0
        from decimal import Decimal
        from app.services.db_worker import enqueue_update_trade_history_by_order_id
        _th_upd = {
            # 식별 키(db_worker 가 setattr 하지 않음): KIS ODNO 는 일별 순번이라 날짜·구분으로 좁힌다
            "order_date": _order_date,
            "order_type": "SELL",
            "strategy_type": "daytrading",
            "executed_price": Decimal(str(round(avg_price, 2))),
            "total_amount": Decimal(str(round(new["total_amount"], 2))),
            "commission": Decimal(str(new["total_cost"])),
            "net_profit_loss": Decimal(str(round(new["net_profit_loss"], 2))),
            "profit_loss_rate": Decimal(str(round(new["net_profit_loss_rate"], 4))) if entry_price > 0 else None,
            "slippage": Decimal(str(round(slippage_pct, 4))),
        }
        if new["is_partial"]:
            _th_upd["executed_quantity"] = int(new["executed_quantity"])
        await enqueue_update_trade_history_by_order_id(order_id, _th_upd)
        if delta:
            if count_daily_pnl:
                async with day_trading_state.trade_lock:
                    day_trading_state.adjust_realized_pnl(delta)
                    _adjust_partial_realized_impl(delta)
                await day_trading_state.save_daily_state()
            elif _orphan_today:
                async with day_trading_state.trade_lock:
                    day_trading_state.adjust_realized_pnl(delta)
                await day_trading_state.save_daily_state()
            else:
                _add_stale_realized_pnl(delta)
        logger.info(
            f"[DayExit] 매도 체결가 확인: {stock_code} 신호가={signal_price:,.0f} "
            f"체결가={avg_price:,.2f} (슬리피지 {slippage_pct:+.3f}%) "
            f"순손익 {recorded_net:+.0f}→{new['net_profit_loss']:+.0f}원 "
            f"({'당일' if (count_daily_pnl or _orphan_today) else '전일'} 손익 보정 {delta:+.0f}원)"
        )
        await _settle(new["net_profit_loss"] if filled_qty >= quantity else None)

    async def _on_unfilled(filled_qty: int, avg_price: float, unfilled: int) -> None:
        """[2026-10-08] 접수됐던 매도가 **종결(취소·거부·잔량 0)됐는데 미체결분이 있다**.

        _finalize_exit_impl 은 접수 시점에 인메모리 제거·ActiveTrade close·TradeHistory COMPLETED
        를 이미 반영했다(10/06 KB금융 고아의 원인). 여기서 되돌릴 수 있는 것만 되돌린다:
          - 당일 포지션: 미체결 수량을 인메모리에 되살려 손절·EOD 관리 재개(재시도)
          - 전일·고아 포지션: 재주문 차단 표식을 지우고 [R8-1] 미체결 수량으로 ActiveTrade(day_trade,
            active)를 되살려 close_stale_positions 가 다시 잡게 한다. 접수 때 행이 닫히고 reflect_sell 이
            Portfolio 수량을 0 으로 깎아, 표식만 지우면 전일 경로(active 행)·고아 경로(Portfolio 수량>0)
            어느 쪽에도 걸리지 않았고 다음 sync 가 'daytrading' 태그를 풀어 스윙/external 로 넘겼다.
          - 당일/전일 손익을 실제 체결분 기준으로 보정, TradeHistory 수량·상태 보정
        loss_tracker·당일 연패 기록은 되돌리지 않는다(§8.5-27, 보수적 방향).
        """
        _recent_dt_sell_finalized.pop(stock_code, None)
        _today = _kst_now().date().isoformat()
        _stale_sell_attempted.discard((_today, stock_code))
        _stale_cleaned.discard((_today, stock_code))
        if filled_qty > 0 and avg_price > 0:
            new_net = dt_net_pnl(entry_price, avg_price, filled_qty, pos)
            _net, _cost = new_net["net_profit_loss"], new_net["total_cost"]
        else:
            _net, _cost = 0.0, 0.0
        delta = _net - recorded_net
        restored_mem = False
        _restored_snapshot: dict | None = None
        async with day_trading_state.trade_lock:
            if count_daily_pnl:
                day_trading_state.adjust_realized_pnl(delta)
                _adjust_partial_realized_impl(delta)
                # [R4-1] 분할 레그가 체결 0으로 종결 → 분할 자체가 없었던 것이므로 목표가·partial_sold
                # 도 분할 전으로 되돌린다(수량만 되살리면 원 수량 전체가 트레일링 전용이 된다).
                # 부분체결(filled_qty>0)이면 분할이 일어난 것이므로 현행(inf/True) 유지.
                _pre = (pos or {}).get(_PRE_PARTIAL_STATE_KEY) if (is_partial and filled_qty <= 0) else None
                _cur = day_trading_state.positions.get(stock_code)
                # [HL2-2] 전량 청산 레그의 기록 정보(승패 표식·접수 시점 왕복 누적)
                _full_rec = None if is_partial else (pos or {}).get(_FULL_EXIT_RECORD_KEY)
                if _cur is not None:
                    _cur["quantity"] = int(_cur.get("quantity", 0) or 0) + int(unfilled)
                    _restore_pre_partial_state_impl(stock_code, _cur, _pre)
                    _apply_full_exit_record_impl(stock_code, _cur, _full_rec, _net, use_prior=False)
                    restored_mem = True
                    _restored_snapshot = dict(_cur)
                elif not day_trading_state.is_exit_in_progress(stock_code):
                    _np = dict(pos)
                    _np.pop(_PRE_PARTIAL_STATE_KEY, None)
                    _np.pop(_FULL_EXIT_RECORD_KEY, None)
                    _np["quantity"] = int(unfilled)
                    _np["sell_unfilled_restored"] = True
                    _restore_pre_partial_state_impl(stock_code, _np, _pre)
                    _apply_full_exit_record_impl(stock_code, _np, _full_rec, _net, use_prior=True)
                    day_trading_state.add_position(stock_code, _np)
                    restored_mem = True
                    _restored_snapshot = dict(_np)
                if filled_qty <= 0:
                    day_trading_state.dt_exited_today.discard(stock_code)
            elif _orphan_today:
                # [R6-1] 인메모리로 직접 되살리지 않는다(손절가 없는 근거 행 사본). 손익만 당일 기준으로
                # 되돌린다. [R8-1] 대신 아래에서 ActiveTrade(day_trade, active)를 되살린다(당일 생성 →
                # close_stale_positions 의 당일 행 복원 경로가 KIS 실보유 − 타전략 확인 후 인메모리로 복원).
                day_trading_state.adjust_realized_pnl(delta)
                _restored_snapshot = _stale_unfilled_restore_snapshot_impl(
                    stock_code, pos, unfilled, orphan_today=True,
                )
                if _restored_snapshot is not None:
                    # 당일 DT 전량 청산 기록(dt_exited_today) 뒤 보유 재개 — 복원 보류(restore_block)를 푼다.
                    # add_position 의 표식과 같은 의미(재기동 원장 재구성도 순체결>0 이면 reheld 로 본다).
                    day_trading_state.dt_reheld_today.add(stock_code)
            else:
                _add_stale_realized_pnl(delta)
                # [R8-1] 전일·고아 잔량 — ActiveTrade 를 원 진입일(오늘 이전)로 되살려 다음 사이클
                # close_stale_positions 의 전일 경로가 다시 잡게 한다.
                _restored_snapshot = _stale_unfilled_restore_snapshot_impl(
                    stock_code, pos, unfilled, orphan_today=False,
                )
        if _restored_snapshot is not None:
            # [R1-5 2026-10-09] ActiveTrade(day_trade)도 되살린다. 전량 매도 접수 때
            # _finalize_exit_impl 이 행을 닫았고(분할이면 잔량으로 줄였고), 인메모리만 복원하면
            # 재기동 시 restore_from_db 가 이 포지션을 찾지 못해 고아·관리 공백이 된다.
            await _bounded_finalize_step(
                _enqueue_restored_active_trade_impl(stock_code, _restored_snapshot),
                stock_code, "미체결 복원 ActiveTrade 큐잉",
            )
        from decimal import Decimal
        from app.services.db_worker import enqueue_update_trade_history_by_order_id
        _upd = {
            "order_date": _order_date,
            "order_type": "SELL",
            "strategy_type": "daytrading",
            "executed_quantity": int(filled_qty),
            "commission": Decimal(str(_cost)),
            "net_profit_loss": Decimal(str(round(_net, 2))),
            "status": "CANCELLED" if filled_qty <= 0 else "COMPLETED",
        }
        if filled_qty > 0 and avg_price > 0:
            _upd["executed_price"] = Decimal(str(round(avg_price, 2)))
            _upd["total_amount"] = Decimal(str(round(avg_price * filled_qty, 2)))
        await enqueue_update_trade_history_by_order_id(order_id, _upd)
        if count_daily_pnl or _orphan_today:
            await day_trading_state.save_daily_state()
        if _kst_now().time() >= _KRX_MARKET_CLOSE:
            # [R3-2] 장 마감 후 복원 = 더 이상 장중 재시도가 없다 → 오버나이트 확정. 잔존 점검이 먼저
            # '청산 확인'을 남겼을 수 있으므로 여기서 직접 텔레그램(DT_EOD)으로 알린다.
            _alert_once_v2(
                "DT_EOD", f"post_close_unfilled:{stock_code}:{order_id}",
                f"[데이트레이딩] 장 마감 후 매도 미체결 확인: {stock_code} 체결 {filled_qty}/{quantity}주, "
                f"미체결 {unfilled}주(order_id={order_id}) — 오버나이트 보유. 다음 영업일 장중 첫 사이클에 "
                f"전일 포지션 청산이 시도됩니다. 계좌를 확인하세요.",
            )
        logger.error(
            f"[DayExit] 🚨 매도 미체결 종결: {stock_code} 체결 {filled_qty}/{quantity}주 "
            f"(미체결 {unfilled}주) order_id={order_id} — "
            + (f"인메모리 포지션 {'복원' if restored_mem else '복원 실패'}(손절·EOD 재시도), "
               f"당일 손익 {delta:+.0f}원 보정"
               if count_daily_pnl else
               f"전일/고아 포지션 재청산 대기(close_stale_positions, ActiveTrade "
               f"{'복원' if _restored_snapshot is not None else '복원 실패 — 수동 확인'}), "
               f"{'당일' if _orphan_today else '전일'} 손익 {delta:+.0f}원 보정")
            + ". loss_tracker·연패 기록은 되돌리지 않음"
        )
        await _settle(None)

    async def _on_result(info: dict) -> None:
        filled = int(info.get("filled_qty", 0) or 0)
        avg = float(info.get("avg_price", 0) or 0)
        if info.get("terminal") and filled < quantity:
            await _on_unfilled(filled, avg, int(quantity) - filled)
            return
        if filled > 0 and avg > 0:
            await _on_fill(filled, avg)

    if on_settled is None and not closing_auction:
        # [2026-10-08] v2 — 체결 0·미체결 종결도 돌려받는다(v1 은 체결>0 일 때만 콜백).
        return fill_reconciler.schedule_v2(stock_code, order_id, "SELL", int(quantity), _on_result)
    if not settings.DAY_TRADE_FILL_CHECK_ENABLED:
        return None

    def _is_final(info: dict | None) -> bool:
        if info is None:
            return False
        return bool(info.get("terminal")) or int(info.get("filled_qty", 0) or 0) >= int(quantity)

    async def _run() -> None:
        info = None
        try:
            info = await fill_reconciler.fetch_fill_v2(stock_code, order_id, "SELL", int(quantity))
            if closing_auction and not _is_final(info):
                _wait = _seconds_until_auction_recheck_impl(_kst_now())
                _, _attempts, _interval = _auction_recheck_params_impl()
                logger.info(
                    f"[DayExit] {stock_code} 종가 단일가 매도(order_id={order_id}) 체결 미종결 — "
                    f"{_wait:.0f}초 뒤(장 마감 후) 재확인 최대 {_attempts}회"
                )
                await asyncio.sleep(_wait)
                for i in range(_attempts):
                    _i = await fill_reconciler.fetch_fill_v2(stock_code, order_id, "SELL", int(quantity))
                    if _i is not None:
                        info = _i
                    if _is_final(info):
                        break
                    if i < _attempts - 1:
                        await asyncio.sleep(_interval)
            if info is not None:
                await _on_result(info)
        except Exception as e:
            logger.warning(
                f"[DayExit] {stock_code} 매도 체결 확인·반영 실패 order_id={order_id}: "
                f"{type(e).__name__}: {e}", exc_info=True,
            )
        finally:
            if not _settled[0]:
                # 체결 확인 불가 → 신호가 기준 값으로 확정(구 동작 폴백)
                logger.warning(
                    f"[DayExit] {stock_code} 매도 체결 확인 불가(order_id={order_id}) — "
                    f"신호가 기준 순손익 {recorded_net:+.0f}원으로 확정"
                )
                await _settle(recorded_net)

    try:
        task = asyncio.get_running_loop().create_task(_run())
    except RuntimeError as e:
        logger.warning(f"[DayExit] {stock_code} 매도 체결 확인 예약 실패: {e}")
        return None
    _auction_fill_tasks.add(task)
    task.add_done_callback(_auction_fill_tasks.discard)
    return task


async def _enqueue_restored_active_trade_impl(stock_code: str, pos: dict) -> None:
    """[R1-5] 매도 미체결로 인메모리에 되살린 포지션을 ActiveTrade(day_trade, active)로 영속화.

    db_worker upsert 는 active day_trade 행이 있으면 갱신(분할 미체결 → 수량을 복원 후 수량으로),
    없으면 새로 만든다(전량 미체결 → 접수 때 닫힌 행 대신 새 active 행). 쓰기는 큐 경유(§3.1).
    target_price 가 inf(분할 후 트레일링 전용)면 take_profit 을 넣지 않고 remaining_pct<1 로 남겨
    restore_from_db 가 목표가 분할을 재발동하지 않게 한다.
    """
    from app.services.db_worker import enqueue_active_trade_upsert

    qty = int(pos.get("quantity", 0) or 0)
    entry = float(pos.get("entry_price", 0) or 0)
    if qty <= 0 or entry <= 0:
        logger.error(
            f"[DayExit] 🚨 {stock_code} 미체결 복원 ActiveTrade 생략(수량 {qty}·진입가 {entry}) — 수동 확인 필요"
        )
        return
    payload: dict = {
        "stock_code": stock_code,
        "stock_name": pos.get("stock_name") or stock_code,
        "strategy_name": "day_trade",
        "quantity": qty,
        "entry_price": entry,
        "status": "active",
    }
    _sl = pos.get("stop_loss")
    if _sl is not None and 0 < float(_sl) < float("inf"):
        payload["stop_loss"] = float(_sl)
    _tp = pos.get("target_price")
    if _tp is not None and 0 < float(_tp) < float("inf"):
        payload["take_profit"] = float(_tp)
    _hi = pos.get("highest_price")
    if _hi is not None and 0 < float(_hi) < float("inf"):
        payload["highest_since_entry"] = float(_hi)
    if pos.get("partial_sold"):
        payload["remaining_pct"] = 0.99
    else:
        # [R4-1] 분할 레그 0체결 복원 — 분할 때 기록한 remaining_pct(<1.0)를 1.0 으로 되돌려야
        # restore_from_db 가 목표가 분할을 막지 않는다(분할 이력 없음 = 원 수량 전체).
        payload["remaining_pct"] = 1.0
    if pos.get("reason"):
        payload["entry_reason"] = str(pos.get("reason"))
    # [HL2-1 2026-10-09] 원 진입 시각·매수 ODNO 를 싣는다(둘 다 기존 ORM 컬럼 — 선택 키).
    # 전량 미체결 복원은 새 active 행을 만드는데, 없으면 created_at=복원 시각(now_kst 기본값)이
    # 되어 재기동 후 entry_time(=created_at)이 리셋 → 30분 시간손절·트레일링 최소 보유가 늦게
    # 발동하고, order_id 가 NULL 이라 R3 의 ODNO 영속화가 이 경로에서 다시 빠진다.
    _ct = _restored_created_at_impl(pos.get("entry_time"))
    if _ct is not None:
        payload["created_at"] = _ct
    else:
        logger.warning(
            f"[DayExit] {stock_code} 미체결 복원 ActiveTrade — 원 진입 시각 없음/형식 불량"
            f"({pos.get('entry_time')!r}) → created_at 은 복원 시각(시간손절 시계 리셋 가능)"
        )
    _oid = str(pos.get("order_id") or "").strip()
    if _oid:
        payload["order_id"] = _oid
    await enqueue_active_trade_upsert(payload)


def _restored_created_at_impl(entry_time) -> datetime | None:
    """복원 ActiveTrade.created_at 값 — 원 진입 시각을 KST naive 로(모델 기본값 now_kst 와 같은 기준).

    datetime 이 아니거나 미래 시각이면 None(호출부가 키를 넣지 않아 모델 기본값을 쓴다).
    정규장 밖 시각도 그대로 둔다 — restore 쪽 _elapsed_minutes/_normalize_entry_time 이 판정한다.
    """
    if not isinstance(entry_time, datetime):
        return None
    if entry_time.tzinfo is not None:
        entry_time = entry_time.astimezone(_KST).replace(tzinfo=None)
    if entry_time > _kst_now():
        return None
    return entry_time


def _stale_unfilled_restore_snapshot_impl(
    stock_code: str, pos: dict, unfilled: int, orphan_today: bool,
) -> dict | None:
    """[R8-1 2026-10-10] 전일·고아 매도 0체결/부분 종결 → ActiveTrade 복원용 포지션 사본(없으면 None).

    _enqueue_restored_active_trade_impl 에 넘길 dict 를 만든다. 수량은 **미체결분만**이다
    (접수 때 sell_qty 는 이미 KIS 실보유 − 타전략(§3.5)·주문가능 상한으로 잘렸다 — 그 이상은 DT 몫이
    아니다). 다음 close_stale_positions 가 매도 시점에 다시 KIS 실보유 − 타전략으로 상한을 건다.

    created_at(=entry_time) 규칙:
      - 전일·과거 고아(orphan_today=False): 원 진입 시각. 반드시 **오늘 이전**이어야 전일 경로에
        잡힌다 — 시각이 없거나 오늘 이후면 전일 15:30(KST)으로 대체(warning). 오늘 날짜로 들어가면
        당일 행 복원 경로가 손절가 없는 사본을 '당일 진입'으로 되살린다.
      - 당일 고아(orphan_today=True): 원 진입 시각(당일) → 당일 행 복원 경로(인메모리 관리 재개).
    """
    try:
        qty = int(unfilled or 0)
    except (TypeError, ValueError):
        qty = 0
    if qty <= 0:
        return None
    snap = dict(pos or {})
    for _k in (_PRE_PARTIAL_STATE_KEY, _FULL_EXIT_RECORD_KEY, "orphan", _ORPHAN_TODAY_KEY):
        snap.pop(_k, None)
    snap["quantity"] = qty
    snap["strategy"] = "day_trade"
    snap["sell_unfilled_restored"] = True
    # 분할 이력 없음(전일·고아 매도는 항상 전량 레그) — remaining_pct=1.0 로 복원된다.
    snap.pop("partial_sold", None)
    if not orphan_today:
        today = _kst_now().date()
        _ct = _restored_created_at_impl(snap.get("entry_time"))
        if _ct is None or _ct.date() >= today:
            _fallback = datetime.combine(today - timedelta(days=1), _KRX_MARKET_CLOSE)
            logger.warning(
                f"[DayExit] {stock_code} 전일·고아 미체결 복원 — 원 진입 시각 "
                f"{snap.get('entry_time')!r} 이 오늘 이전이 아니거나 없음 → created_at={_fallback} 로 "
                f"기록(전일 경로 재청산 보장, 실제 진입일 아님)"
            )
            snap["entry_time"] = _fallback
        else:
            snap["entry_time"] = _ct
    return snap


# ── [2026-10-08] DT 고아 잔고 감지 ────────────────────────────────────────────────
# 동기: 10/06 KB금융(105560) EOD 시장가 매도가 **접수(rt_cd=0)** 되자 _finalize_exit_impl 이
# 체결 확인 없이 인메모리 제거 + ActiveTrade close + TradeHistory COMPLETED 를 남겼다. 이 구조에서는
# 실제 0체결(취소/미체결)이면 잔고가 남는다.
# [정정 2026-10-09] 105560 의 잔여 3주가 "10/06 EOD 0체결"이라는 서술은 **운영 DB 와 맞지 않는다**:
#   - ActiveTrade id=178 'synced' 3주(평단 171,058.33)가 10/01 08:55 — 그날 DT 매수(13:44) **이전** 에 생성.
#   - ActiveTrade id=177(9/30 day_trade)이 수량 6 으로 기록됐는데 9/30 DT 매수 TradeHistory 는 3주
#     → 9/30 이전부터 DT 외 3주가 있었고 DT 행에 합산된 것으로 보인다(추정).
#   - TradeHistory(daytrading) BUY 3+3+3 / SELL 3+3+3 = 순 0. 10/06 SELL(order_id 0000031910)은
#     slippage NULL = 체결 확인이 안 된 채 COMPLETED(체결 여부는 KIS 주문내역 미확인 — 추정 불가).
#   → 잔여 3주는 DT 0체결분이 아니라 기존 외부/동기화 보유분일 가능성이 높다(추정). 그래서 아래
#     감지는 TradeHistory 순보유 0 인 105560 을 **팔지 않고** 수동 확인 경보만 낸다(R1-1).
# Portfolio 행은 sync 가 'daytrading' 태그를 유지한 채 수량만 갱신했고, 이후
#   - restore_from_db / close_stale_positions 는 **active** ActiveTrade 만 보고(행이 closed),
#   - 스윙 경로(execute_auto_trade)는 daytrading 소유 행을 매도하지 않으며(§6.4),
#   - sync_portfolio 는 'daytrading' 행에 ActiveTrade 를 시딩하지 않는다
# → 아무도 손절을 걸지 않는 고아가 됐다. 아래 감지는 close_stale_positions 의 읽기 전용
# 세션(§3.2 ③) 안에서만 수행하고, 매도는 기존 stale 경로(실보유·타전략 차감·재시도·
# count_daily_pnl=False·closed_today)를 그대로 탄다.

# [R4-1 2026-10-09] 분할 레그 체결 확인용 포지션 스냅샷에 싣는 "분할 전 상태" 키
# ({"target_price": 분할 전 목표가, "partial_sold": 분할 전 값}). 체결 0 종결 시 복원에 쓴다.
_PRE_PARTIAL_STATE_KEY = "_pre_partial_state"


def _apply_full_exit_record_impl(
    stock_code: str, target: dict, rec, filled_net: float, use_prior: bool,
) -> None:
    """[HL2-2] 전량 청산 레그가 0체결·부분 종결로 복원될 때 복원 포지션에 남길 상태.

    - 승패 표식: 접수 시점에 당일 연패·loss_tracker 에 기록한 결과(재청산 시 이중 집계 방지)
    - realized_net: 체결된 레그의 순손익(filled_net)을 왕복 누적에 더한다 — 빠지면 다음 전량
      청산의 왕복 판정(_prior_realized + 마지막 레그)에서 그 레그가 누락된다.
      use_prior=True 면 접수 시점 누적(prior_realized)을 기준으로, False 면 대상의 현재값에 더한다.
    rec 가 없으면(분할 레그·구 스냅샷) 아무것도 하지 않는다.
    """
    if not isinstance(rec, dict):
        return
    for _key, _src_key in (
        (_RESTORED_STREAK_MARK_KEY, "streak"),
        (_RESTORED_LOSS_TRACKER_MARK_KEY, "loss_tracker"),
    ):
        _v = rec.get(_src_key)
        if isinstance(_v, bool):
            target[_key] = _v
        elif use_prior:
            target.pop(_key, None)
    if use_prior:
        try:
            _base = float(rec.get("prior_realized", 0.0) or 0.0)
        except (TypeError, ValueError):
            logger.warning(f"[DayExit] {stock_code} 복원 prior_realized 파싱 실패 — 0 기준")
            _base = 0.0
    else:
        try:
            _base = float(target.get("realized_net", 0.0) or 0.0)
        except (TypeError, ValueError):
            logger.warning(f"[DayExit] {stock_code} 복원 realized_net 파싱 실패 — 0 기준")
            _base = 0.0
    target["realized_net"] = _base + float(filled_net or 0.0)
    if filled_net:
        logger.warning(
            f"[DayExit] {stock_code} 매도 부분체결 복원 — 체결 레그 순손익 {filled_net:+.0f}원을 "
            f"왕복 누적에 반영(realized_net={target['realized_net']:+.0f}원)"
        )


def _restore_pre_partial_state_impl(stock_code: str, target: dict, pre) -> None:
    """[R4-1] 체결 0 으로 종결된 분할 레그 — 목표가·partial_sold 를 분할 전 값으로 되돌린다."""
    if not isinstance(pre, dict):
        return
    _tp = pre.get("target_price")
    if _tp is not None:
        target["target_price"] = _tp
    target["partial_sold"] = bool(pre.get("partial_sold", False))
    logger.warning(
        f"[DayExit] {stock_code} 분할 매도 체결 0 종결 — 목표가({_tp})·partial_sold"
        f"({target['partial_sold']}) 분할 전 상태로 복원(목표가 분할 재시도 가능)"
    )


# 최근 DT 매도 반영 시각 {code: monotonic} — 반영 직후 Portfolio/KIS 가 아직 옛 수량일 때
# 고아로 오판하지 않게 DAY_TRADE_ORPHAN_GRACE_SEC 동안 보류한다. 0체결 확인 시 해제.
_recent_dt_sell_finalized: dict[str, float] = {}


def _orphan_blocked_reason(stock_code: str, today_key: str) -> str | None:
    """고아/복원 판정을 보류해야 하면 사유, 아니면 None."""
    return _orphan_blocked_reason_v2(stock_code, today_key, block_unresolved_unknown=True)


def _orphan_blocked_reason_v2(
    stock_code: str, today_key: str, block_unresolved_unknown: bool,
) -> str | None:
    """[R1-2 2026-10-09] 고아/복원 판정 보류 사유.

    block_unresolved_unknown=False(고아 후보 전용): 직전 고아 매도가 **결과 불명**이어도 후보에서
    빼지 않는다. 빼면 `_place_and_finalize_exit_impl` 의 불명 재판정(잔고 기준)에 다시 도달하지
    못해 `_unresolved_exit_unknown` 이 당일 내내 풀리지 않고 고아가 오버나이트로 남았다.
    재판정·재매도 상한은 stale 경로(_stale_sell_failures 백오프 → _resolve_unknown_exit_v2_impl)가
    그대로 맡는다. 진행 중 표식(HOLD) 동안은 아래 '청산 주문 진행 중'이 계속 막는다.
    """
    if stock_code in day_trading_state.positions:
        return "인메모리 보유"
    if day_trading_state.is_exit_in_progress(stock_code):
        return "청산 주문 진행 중"
    if block_unresolved_unknown and (today_key, stock_code) in _unresolved_exit_unknown:
        return "결과 불명 매도 미해결"
    _t = _recent_dt_sell_finalized.get(stock_code)
    if _t is not None and time.monotonic() - _t < float(settings.DAY_TRADE_ORPHAN_GRACE_SEC):
        return "최근 DT 매도 반영(유예)"
    return None


# [R1-3/6/7] 수동 개입 경보를 보낸 (날짜, key) — 하루 1회만 텔레그램(로그는 _error_throttled)
_manual_alerted: set[tuple[str, str]] = set()


def _alert_manual_intervention_impl(key: str, msg: str) -> None:
    """[R1-3/6/7] 자동 처리 주체가 없는 잔고 → logger.error(스로틀) + 텔레그램(하루 1회).

    warn_throttled 경고는 놓치기 쉬웠다(KB금융 105560 3주가 10/01부터 DT·스윙 어느 쪽도
    관리하지 않는 상태로 남았다). 발송 실패는 error 로그만 남기고 흐름을 막지 않는다.
    """
    _error_throttled(f"dt_exit:manual:{key}", msg)
    _k = (_today_key(), key)
    for k in [k for k in _manual_alerted if k[0] != _k[0]]:
        _manual_alerted.discard(k)
    if _k in _manual_alerted:
        return
    _manual_alerted.add(_k)
    try:
        from app.services.telegram_notifier import telegram_notifier
        _t = asyncio.get_running_loop().create_task(asyncio.wait_for(
            telegram_notifier.notify_system_warning("DT_MANUAL_CHECK", msg), timeout=10.0,
        ))
        _alert_tasks.add(_t)
        _t.add_done_callback(_alert_tasks.discard)
    except Exception as e:
        logger.error(f"[DayExit] 수동 확인 경보 발송 실패({key}): {e}")


def _dt_net_qty_from_history(rows) -> dict[str, int]:
    """[R1-1] TradeHistory(daytrading) 행 → 종목별 DT 순보유 추정(BUY 체결 − SELL 체결). 순수 함수.

    보수적 집계(과다매도 방지 쪽으로 기운다):
      - BUY 는 status=='COMPLETED' 만, 체결수량(없으면 주문수량)으로 센다.
      - SELL 은 FAILED/REJECTED 를 뺀 전부를 센다 — PENDING(결과 불명)도 팔린 것으로 본다.
        0체결 종결은 fill_reconciler 가 CANCELLED·executed_quantity=0 으로 바꿔 두므로 0 으로 들어간다.
    창 밖에서 산 물량을 창 안에서 판 경우 음수가 될 수 있다 → 호출부가 0 하한으로 쓴다(매도 안 함).
    """
    out: dict[str, int] = {}
    for r in rows or []:
        code = getattr(r, "stock_code", None)
        if not code:
            continue
        ot = str(getattr(r, "order_type", "") or "").upper()
        st = str(getattr(r, "status", "") or "").upper()
        eq = getattr(r, "executed_quantity", None)
        q = int(eq if eq is not None else (getattr(r, "order_quantity", 0) or 0))
        if ot == "BUY":
            if st == "COMPLETED":
                out[code] = out.get(code, 0) + q
        elif ot == "SELL":
            if st not in ("FAILED", "REJECTED"):
                out[code] = out.get(code, 0) - q
    return out


def _row_created_date_v2(row):
    """[R6-1] 행의 created_at(KST naive 저장) 날짜. 없거나 파싱 불가면 None."""
    _c = getattr(row, "created_at", None)
    if isinstance(_c, datetime):
        return _c.date()
    return None


def _last_completed_dt_buy_at_v2(rows, today=None) -> dict:
    """[R6-1] {(code, is_today): 마지막 COMPLETED DT BUY created_at}. 순수 함수.

    today 를 주지 않으면 KST 오늘을 쓴다. 고아 청산 사유·entry_time 의 '진입일'을 근거
    ActiveTrade 행(과거 행일 수 있음)이 아니라 실제 매수 체결 기준으로 쓰기 위한 것이다.
    """
    _today = today or _kst_now().date()
    out: dict = {}
    for r in rows or []:
        code = getattr(r, "stock_code", None)
        if not code:
            continue
        if str(getattr(r, "order_type", "") or "").upper() != "BUY":
            continue
        if str(getattr(r, "status", "") or "").upper() != "COMPLETED":
            continue
        _c = getattr(r, "created_at", None)
        if not isinstance(_c, datetime):
            continue
        key = (code, _c.date() == _today)
        if key not in out or _c > out[key]:
            out[key] = _c
    return out


async def _detect_dt_orphans_impl(db, active_dt_rows, stale: dict, today) -> tuple[dict, list]:
    """(고아 후보 {code: pos}, 복원 대상 ActiveTrade 행 목록).

    호출부(close_stale_positions)의 읽기 전용 세션을 그대로 쓴다 — SELECT 만, commit 없음.
    - 복원: 당일 생성 active day_trade 행인데 인메모리에 없고 당일 DT 전량 청산 기록도 없는 종목
      (restore_block_reason_v2 — [R2 하드닝] closed_today 는 재진입 차단 집합이라 근거로 쓰지 않는다).
    - 고아: Portfolio.strategy_type ∈ DT 이고 수량>0 인데 인메모리·active DT 행이 없는 종목.
      근거로 DAY_TRADE_ORPHAN_LOOKBACK_DAYS 이내 DT ActiveTrade 행(상태 무관)이 있어야 하며,
      매도 상한 = min(Portfolio 수량, 그 행 수량, TradeHistory DT 순보유).

    [R1-1 2026-10-09] **정상 체결로 닫힌 DT 행은 근거가 아니다.** Portfolio 'daytrading' 태그는
    같은 종목의 수동·HTS 매수분이 합산돼도 해제되지 않는다(§3.5 — 행이 stock_code UNIQUE).
    예전에는 최근 DT 행이 있기만 하면 그 수량만큼 팔아, DT 3주 정상 청산 후 남은 수동 10주에서
    3주를 매일 시장가 매도했다. 이제 같은 창의 TradeHistory(strategy_type='daytrading')
    BUY−SELL 순체결수량(_dt_net_qty_from_history)이 양수여야만 고아로 본다 — 0체결 종결된 매도는
    fill_reconciler 가 CANCELLED·체결 0 으로 바꾸므로 그 몫만 양수로 남는다.
    근거가 없거나 순보유 0 이면 팔지 않고 **logger.error + 텔레그램**(수동 확인, R1-3/6/7).
    """
    from app.models.models import ActiveTrade, Portfolio, TradeHistory
    from app.services.db_worker import DT_STRATEGY_NAMES
    from sqlalchemy import select as _select

    _tk = today.isoformat()
    restores: list = []
    active_codes = set()
    for r in active_dt_rows or []:
        active_codes.add(r.stock_code)
        _c = r.created_at.date() if getattr(r, "created_at", None) else None
        if _c is None or _c < today or r.stock_code in stale:
            continue
        # [R2 하드닝 2026-10-09] closed_today(재진입 차단)가 아니라 DT 전량 청산 기록으로 판단한다.
        # 불명 매수 접수 확정·매도 0체결 종결 복원 종목은 closed_today 에 남아 있어도 보유 중이다.
        _rb = day_trading_state.restore_block_reason_v2(r.stock_code)
        if _rb is not None:
            warn_throttled(
                logger, f"dt_exit:restore_block:{r.stock_code}",
                f"[DayExit] ⚠️ {r.stock_code}: 당일 active DT 행({int(r.quantity or 0)}주)이 인메모리에 없으나 "
                f"{_rb} — 복원 보류(close 큐 반영 지연이면 정상, 실보유가 남아 있으면 수동 확인)",
            )
            continue
        if _orphan_blocked_reason(r.stock_code, _tk) is not None:
            continue
        if int(r.quantity or 0) > 0:
            restores.append(r)

    pf_rows = (await db.execute(
        _select(Portfolio).where(
            Portfolio.strategy_type.in_(DT_STRATEGY_NAMES),
            Portfolio.quantity > 0,
        )
    )).scalars().all()
    cands: dict[str, object] = {}
    for p in pf_rows or []:
        code = getattr(p, "stock_code", None)
        if not code or code in stale or code in active_codes:
            continue
        if int(getattr(p, "quantity", 0) or 0) <= 0:
            continue
        if _orphan_blocked_reason_v2(code, _tk, block_unresolved_unknown=False) is not None:
            continue
        cands[code] = p
    if not cands:
        return {}, restores

    _since = datetime.combine(
        today - timedelta(days=int(settings.DAY_TRADE_ORPHAN_LOOKBACK_DAYS)), datetime.min.time(),
    )
    ev_rows = (await db.execute(
        _select(ActiveTrade).where(
            ActiveTrade.stock_code.in_(list(cands)),
            ActiveTrade.strategy_name.in_(DT_STRATEGY_NAMES),
            ActiveTrade.created_at >= _since,
        ).order_by(ActiveTrade.created_at.desc())
    )).scalars().all()
    latest: dict[str, object] = {}
    for r in ev_rows or []:
        if r.stock_code not in latest:
            latest[r.stock_code] = r
    # [R1-1] 같은 창의 DT 원장 — 순체결수량이 양수인 종목만 DT 몫이 남아 있다고 본다.
    th_rows = (await db.execute(
        _select(TradeHistory).where(
            TradeHistory.stock_code.in_(list(cands)),
            TradeHistory.strategy_type == "daytrading",
            TradeHistory.created_at >= _since,
        )
    )).scalars().all()
    net_by_code = _dt_net_qty_from_history(th_rows)
    # [R6-1 2026-10-10] 당일 창(KST 오늘 생성 행)만의 DT 순체결 — 양수면 '당일 매수분' 고아다.
    # 그 손익은 전일 포지션 예외(§7 count_daily_pnl=False)가 아니라 당일 집계 대상이다.
    # created_at 이 없는 행은 당일로 보지 않는다(근거 불명 → 기존 동작).
    today_rows = [r for r in th_rows or [] if _row_created_date_v2(r) == today]
    today_net_by_code = _dt_net_qty_from_history(today_rows)
    last_buy_at = _last_completed_dt_buy_at_v2(th_rows, today)

    _lb = int(settings.DAY_TRADE_ORPHAN_LOOKBACK_DAYS)
    orphans: dict[str, dict] = {}
    for code, p in cands.items():
        pf_qty = int(getattr(p, "quantity", 0) or 0)
        ev = latest.get(code)
        ev_qty = int(getattr(ev, "quantity", 0) or 0) if ev is not None else 0
        th_net = max(0, int(net_by_code.get(code, 0)))
        if ev is None or ev_qty <= 0 or th_net <= 0:
            _why = (
                f"최근 {_lb}일 DT 진입 근거(ActiveTrade) 없음" if ev is None or ev_qty <= 0
                else f"최근 {_lb}일 DT 원장 순체결 {int(net_by_code.get(code, 0))}주(매수=매도 — DT 몫 없음)"
            )
            _alert_manual_intervention_impl(
                f"orphan_unowned:{code}",
                f"[DayExit] 🚨 {code}: Portfolio 'daytrading' {pf_qty}주인데 인메모리·active DT 행이 없고 "
                f"{_why} — DT 는 자동 청산하지 않는다. 스윙 청산도 'daytrading' 태그 행은 건너뛰므로 "
                f"손절 주체가 없는 잔고일 수 있다(수동 매수 합산·sync 'synced' 행 공존 포함). "
                f"수동 청산 또는 Portfolio 태그 재지정 필요",
            )
            continue
        _cap = min(pf_qty, ev_qty, th_net)
        _entry = float(getattr(ev, "entry_price", 0) or 0) or float(getattr(p, "average_price", 0) or 0)
        # [R6-1] 당일 DT 순매수가 남아 있으면 당일 포지션 — 청산 손익을 당일 한도·연패에 넣는다.
        _today_net = int(today_net_by_code.get(code, 0))
        _is_today = _today_net > 0
        # 진입 시각은 근거 ActiveTrade 생성시각(과거 행일 수 있다)이 아니라 실제 DT 매수 체결 행 기준.
        # 당일 고아면 당일 마지막 매수, 아니면 오늘 이전 마지막 매수 → 없으면 근거 행 시각(구 동작).
        _buy_at = last_buy_at.get((code, True)) if _is_today else last_buy_at.get((code, False))
        _entry_time = _buy_at or getattr(ev, "created_at", None)
        orphans[code] = {
            "stock_name": getattr(p, "stock_name", None) or getattr(ev, "stock_name", None) or code,
            "strategy": "day_trade",
            "entry_price": _entry,
            "quantity": _cap,
            "highest_price": float(getattr(ev, "highest_since_entry", 0) or 0) or _entry,
            "entry_time": _entry_time,
            "reason": "",
            "order_id": "",
            "orphan": True,
            _ORPHAN_TODAY_KEY: _is_today,
        }
        logger.warning(
            f"[DayExit] ⚠️ DT 고아 잔고 감지: {code} Portfolio 'daytrading' {pf_qty}주, 최근 DT 행 "
            f"(status={getattr(ev, 'status', '?')}, 수량 {ev_qty}, 생성 {getattr(ev, 'created_at', None)}), "
            f"DT 원장 순체결 {th_net}주(당일 {_today_net}주) — 인메모리·active 행 없음 → 전량 청산 대상"
            f"(상한 {_cap}주, {'당일 매수분 — 당일 손익 집계' if _is_today else '전일 이전 — 당일 손익 제외'})"
        )
    return orphans, restores


async def _restore_lost_today_positions_v2_impl(rows, kis_holdings: dict, other_qty: dict) -> int:
    """[C-2 r9] 판정 수량을 min(주문가능, 보유) 로 — 접수·미체결 DT 매도 물량을 복원해 재매도하지 않게."""
    from app.services.daytrading.position_manager import _cap_hldg_by_sellable_v2

    _capped = {c: _cap_hldg_by_sellable_v2(v) for c, v in (kis_holdings or {}).items()}
    return await _restore_lost_today_positions_impl(rows, _capped, other_qty)


async def _restore_lost_today_positions_impl(rows, kis_holdings: dict, other_qty: dict) -> int:
    """당일 active 행 → 인메모리 복원 (restore_from_db 와 같은 변환). 복원 수 반환."""
    from app.services.daytrading.position_manager import _active_trade_row_to_position

    n = 0
    for r in rows:
        code = r.stock_code
        qty = int(r.quantity or 0)
        _h = kis_holdings.get(code) or {}
        hldg = int(_h.get("hldg", 0) or 0)
        dt_avail = hldg - int(other_qty.get(code, 0) or 0)
        if qty <= 0 or dt_avail < qty:
            warn_throttled(
                logger, f"dt_exit:restore_skip:{code}",
                f"[DayExit] ⚠️ {code}: 당일 active DT 행({qty}주)이 인메모리에 없으나 실보유 {hldg}주 − "
                f"타전략 {int(other_qty.get(code, 0) or 0)}주 = {dt_avail}주 < 행 수량 — 복원 보류(수동 확인)",
            )
            continue
        async with day_trading_state.trade_lock:
            _rb = None if code in day_trading_state.positions else day_trading_state.restore_block_reason_v2(code)
            if code in day_trading_state.positions or _rb is not None:
                if _rb is not None:
                    warn_throttled(
                        logger, f"dt_exit:restore_block:{code}",
                        f"[DayExit] ⚠️ {code}: 당일 active DT 행 복원 직전 {_rb} — 복원 보류",
                    )
                continue
            day_trading_state.add_position(code, _active_trade_row_to_position(r))
        n += 1
        logger.warning(
            f"[DayExit] ⚠️ 당일 DT 포지션 인메모리 유실 → ActiveTrade 에서 복원: {code} {qty}주 "
            f"(실보유 {hldg}주) — 손절·트레일링·EOD 관리 재개"
        )
    return n


async def close_stale_positions() -> list[dict]:
    """진입일이 오늘이 아닌(전일 미청산) 데이트레이딩 포지션을 시장가 전량 청산.

    EOD(15:20) 강제청산이 실패하거나 스케줄러가 그 구간을 놓친 경우의 안전장치.
    장중에만 동작한다(시장가 주문 필요). 자정 reset 이후 전일 포지션은 DB 에만 남으므로
    이 경로가 오버나이트 보유를 끝내는 유일한 장치다.

    재시도 규칙 [H-A 2026-10-02]:
      - 주문이 접수(rt_cd=="0")되면 같은 날·같은 프로세스에서 다시 주문하지 않는다.
      - 명시적 거부(EGW00201 등)는 DAY_TRADE_STALE_SELL_RETRY_BACKOFF_SEC 간격으로
        최대 DAY_TRADE_STALE_SELL_MAX_RETRIES 회까지 재시도한다.
      - 결과 불명(응답 없음·예외·시간초과)이면 재시도 전에 당일 주문내역에서 이 종목
        매도 접수 여부를 확인하고, 접수돼 있으면 재주문하지 않는다.
      - 이중 매도는 매 시도마다 다시 읽는 KIS 주문가능수량(ord_psbl_qty)으로도 막힌다.
      - 매도 수량 = min(행 수량, 주문가능수량, 보유수량 − 타 전략 보유분).
      - 주문 없이 ActiveTrade 만 닫는 정리는 **KIS 보유수량(hldg_qty)=0** 일 때만 한다.

    Returns:
        list[dict]: 청산 실행 결과 목록
    """
    if not is_market_open():
        return []

    today = _kst_now().date()
    results: list[dict] = []

    # [2026-09-23] 인메모리 positions 만 순회하면 자정 reset_if_new_day() 이후에는
    # 대상이 사라져 안전장치가 무력해진다 → DB(ActiveTrade)에서도 전일 포지션을 찾는다.
    stale: dict[str, dict] = {}
    for stock_code in list(day_trading_state.positions.keys()):
        pos = day_trading_state.get_position(stock_code)
        entry_time = (pos or {}).get("entry_time")
        entry_date = entry_time.date() if hasattr(entry_time, "date") else None
        if pos and entry_date is not None and entry_date < today:
            stale[stock_code] = pos

    # [M9 / CLAUDE.md §3.1 예외 — 읽기 전용, 쓰기는 db_worker(db_task_queue) 경유]
    # SELECT 만 수행하며 commit()/flush()/add() 호출이 없다. 세션 수명은 아래
    # async with 블록으로 최소화하고, 실패는 warning 으로 운영에 노출한다.
    # 실제 청산 기록은 아래 _execute_exit_stale → db_task_queue 경유로만 이뤄진다.
    # [H3 2026-10-02] 같은 세션에서 비-DT(스윙·유입 등) active 행 수량도 읽는다(§3.5 차감용).
    #   조회 실패면 None 으로 남겨 **매도하지 않는다**(과다매도 = 스윙 물량 매도 방지).
    other_strategy_qty: dict[str, int] | None = None
    restore_rows: list = []
    try:
        from app.core.database import AsyncSessionLocal
        from app.models.models import ActiveTrade
        from app.services.db_worker import not_dt_strategy_clause
        from sqlalchemy import select as _select

        async with AsyncSessionLocal() as _db:
            _rows = (await _db.execute(
                _select(ActiveTrade).where(
                    ActiveTrade.strategy_name == "day_trade",
                    ActiveTrade.status == "active",
                )
            )).scalars().all()
            for _row in _rows:
                _created = _row.created_at.date() if _row.created_at else None
                if _created is None or _created >= today or _row.stock_code in stale:
                    continue
                _entry = float(_row.entry_price or 0)
                stale[_row.stock_code] = {
                    "stock_name": _row.stock_name or _row.stock_code,
                    "strategy": "day_trade",
                    "entry_price": _entry,
                    "quantity": int(_row.quantity or 0),
                    "highest_price": float(_row.highest_since_entry or _entry),
                    "entry_time": _row.created_at,
                    "reason": "",
                    "order_id": _row.order_id or "",
                }
            other_strategy_qty = {}
            if stale:
                _other_rows = (await _db.execute(
                    _select(ActiveTrade).where(
                        ActiveTrade.stock_code.in_(list(stale.keys())),
                        ActiveTrade.status == "active",
                        not_dt_strategy_clause(ActiveTrade.strategy_name),
                    )
                )).scalars().all()
                for _o in _other_rows:
                    other_strategy_qty[_o.stock_code] = (
                        other_strategy_qty.get(_o.stock_code, 0) + int(_o.quantity or 0)
                    )
            # [2026-10-08] DT 고아 잔고 / 당일 active 행 유실 감지 — 같은 읽기 전용 세션.
            # 위 쿼리 순서(①DT active ②비-DT)는 바꾸지 않고 뒤에 덧붙인다.
            if settings.DAY_TRADE_ORPHAN_DETECT_ENABLED:
                _orphans, _restores = await _detect_dt_orphans_impl(_db, _rows, stale, today)
                stale.update(_orphans)
                restore_rows = _restores
                _extra = [c for c in list(_orphans) + [r.stock_code for r in _restores]
                          if c not in other_strategy_qty]
                if _extra:
                    _other_rows2 = (await _db.execute(
                        _select(ActiveTrade).where(
                            ActiveTrade.stock_code.in_(_extra),
                            ActiveTrade.status == "active",
                            not_dt_strategy_clause(ActiveTrade.strategy_name),
                        )
                    )).scalars().all()
                    for _o in _other_rows2:
                        other_strategy_qty[_o.stock_code] = (
                            other_strategy_qty.get(_o.stock_code, 0) + int(_o.quantity or 0)
                        )
    except Exception as _db_err:
        logger.warning(
            f"[DayExit] 전일 포지션 DB 조회 실패(인메모리만 처리, 타 전략 보유분 미확인 → "
            f"매도 보류): {_db_err}"
        )
        other_strategy_qty = None

    stale = {
        c: p for c, p in stale.items() if int((p or {}).get("quantity", 0) or 0) > 0
    }
    if not stale and not restore_rows:
        return results

    # [H3] KIS 실보유 조회 — 실패하면 None → 매도하지 않는다(warning).
    kis_holdings, _holdings_complete = await _fetch_kis_holdings_v3_impl()

    # [2026-10-08] 당일 active 행이 있는데 인메모리에서 사라진 포지션 → 복원(정상 관리 재개).
    # 실보유에서 타 전략 몫을 뺀 DT 가용이 행 수량 이상일 때만(이미 팔린 행을 되살려 스윙
    # 물량을 파는 것을 막는다). 판정 불가(조회 실패)면 복원하지 않는다 — 다음 사이클 재시도.
    if restore_rows and kis_holdings is not None and other_strategy_qty is not None:
        # [R3-1] 첫 페이지가 가득 차 응답에 없는 종목은 '보유 0'이 아니라 판정 불가 → 이번 회차 보류.
        _judgeable = []
        for _r in restore_rows:
            if _holding_unknown_v2(kis_holdings, _holdings_complete, _r.stock_code):
                _error_throttled(
                    f"dt_exit:restore_page_full:{_r.stock_code}",
                    f"[DayExit] 🚨 {_r.stock_code}: 당일 active DT 행이 인메모리에 없으나 KIS 잔고 첫 페이지가 "
                    f"가득 차 실보유를 확인할 수 없음 — 복원 보류(손절 공백, 수동 확인 필요)",
                )
                continue
            _judgeable.append(_r)
        if _judgeable:
            await _restore_lost_today_positions_v2_impl(_judgeable, kis_holdings, other_strategy_qty)
    if not stale:
        return results

    for stock_code, pos in stale.items():
        entry_time = pos.get("entry_time")
        entry_date = entry_time.date() if hasattr(entry_time, "date") else None
        _attempt_key = (today.isoformat(), stock_code)

        if day_trading_state.is_exit_in_progress(stock_code):  # [H-B]
            continue
        _already_accepted = _attempt_key in _stale_sell_attempted

        row_qty = int(pos.get("quantity", 0) or 0)
        _mem_pos = day_trading_state.positions.get(stock_code)
        _mem_is_other = _mem_pos is not None and _mem_pos is not pos
        if _mem_is_other:
            # 같은 종목에 당일 DT 포지션과 전일 DB 행이 공존 — close_active_trade 가
            # '가장 최근 day_trade 행'을 닫으므로 당일 행을 잘못 닫을 수 있다. 자동 처리 보류.
            _error_throttled(
                "dt_exit:stale_coexist",
                f"[DayExit] 🚨 {stock_code}: 당일 DT 포지션과 전일 미청산 DB 행이 공존 — "
                f"자동 청산·정리를 보류한다(수동 확인 필요)",
            )
            continue
        if kis_holdings is None:
            logger.warning(
                f"[DayExit] ⚠️ 전일 포지션 {stock_code} 매도 보류 — KIS 실보유 조회 실패 "
                f"(실보유 미확인 상태로 팔면 스윙 물량을 팔 수 있다). 다음 사이클 재시도"
            )
            continue
        if other_strategy_qty is None:
            logger.warning(
                f"[DayExit] ⚠️ 전일 포지션 {stock_code} 매도 보류 — 타 전략 보유분 조회 실패"
                f"(§3.5 차감 불가). 다음 사이클 재시도"
            )
            continue

        if _holding_unknown_v2(kis_holdings, _holdings_complete, stock_code):
            # [R3-1] 응답에 없는데 첫 페이지가 가득 참 → '보유 0'(=매도 없이 ActiveTrade 정리)으로
            # 해석하면 실보유가 남은 채 행이 닫혀 손절 주체가 없는 고아가 된다. 매도·정리 모두 보류.
            _alert_manual_intervention_impl(
                f"stale_page_full:{stock_code}",
                f"[DayExit] 🚨 전일 DT 포지션 {stock_code}({row_qty}주): KIS 잔고 첫 페이지가 가득 차"
                f"(≥{int(getattr(settings, 'SWING_BALANCE_FULL_PAGE_ROWS', 20) or 0)}행) 실보유를 확인할 수 "
                f"없음 — 자동 매도·정리 보류(ActiveTrade 유지). get_balance 연속조회 미구현(§8.5-37), "
                f"계좌 확인 후 수동 청산 필요",
            )
            continue
        _h = kis_holdings.get(stock_code) or {"hldg": 0, "sellable": 0}
        hldg_qty = int(_h.get("hldg", 0) or 0)
        sellable_qty = int(_h.get("sellable", 0) or 0)
        other_qty = int(other_strategy_qty.get(stock_code, 0) or 0)
        # [M-A] 타 전략 차감은 **보유수량** 기준으로 한다. 주문가능수량(ord_psbl_qty)에는 이미
        # 대기 중인 스윙 매도분이 빠져 있어 거기서 또 빼면 이중 차감된다. 주문가능수량은 상한으로만.
        dt_available = max(0, min(sellable_qty, hldg_qty - other_qty))
        sell_qty = min(row_qty, dt_available)

        _is_orphan = bool(pos.get("orphan"))
        if _is_orphan and (hldg_qty <= 0 or (sell_qty <= 0 and not _already_accepted)):
            # 고아 후보는 닫을 ActiveTrade 가 없다 — 실보유 0(이미 팔림·Portfolio 미동기화)이거나
            # DT 가용 0(타 전략 몫)이면 아무것도 하지 않는다.
            if hldg_qty <= 0:
                warn_throttled(
                    logger, f"dt_exit:orphan_nothing:{stock_code}",
                    f"[DayExit] DT 고아 후보 {stock_code}: KIS 보유 0주 — 매도 대상 없음"
                    f"(Portfolio 미동기화 추정)",
                )
            else:
                # [R1-3/7] 실보유가 있는데 DT 가용 0 = 'daytrading' 태그 행을 DT 는 타 전략 몫으로,
                # 스윙은 DT 소유로 보고 서로 미루는 관리 공백일 수 있다 → 수동 개입 경보.
                _alert_manual_intervention_impl(
                    f"orphan_no_dt_available:{stock_code}",
                    f"[DayExit] 🚨 DT 고아 후보 {stock_code}: KIS 보유 {hldg_qty}주(주문가능 "
                    f"{sellable_qty}주) − 비-DT active 행 {other_qty}주 → DT 가용 {dt_available}주 — "
                    f"DT 는 팔지 않는다. Portfolio 태그가 'daytrading' 이면 스윙 청산도 건너뛰므로 "
                    f"손절 주체가 없을 수 있다. 수동 청산 또는 Portfolio 태그 재지정 필요",
                )
            continue

        if hldg_qty <= 0 and _attempt_key not in _stale_cleaned:
            # [L-1] 실보유 0 — 이미 팔렸다(접수된 매도의 체결 포함). 주문 없이 ActiveTrade 만 정리.
            sell_qty = 0
        elif _already_accepted:
            # [H3/H-A] 이번 프로세스에서 이미 매도 주문이 **접수된** 종목은 재주문 금지.
            # 예전에는 매도 성공 후 ActiveTrade close 가 큐에서 반영되기 전(또는 엉뚱한 행이
            # 닫혀) 다음 사이클 DB 조회가 같은 행을 다시 stale 로 잡아 **반복 매도**했다.
            warn_throttled(
                logger, "dt_exit:stale_already_attempted",
                f"[DayExit] 전일 포지션 {stock_code} 은 이번 프로세스에서 매도 주문이 이미 접수됨 — "
                f"재주문하지 않는다(체결 후 보유 0 이 되면 ActiveTrade 를 정리)",
            )
            continue
        elif hldg_qty <= 0:
            continue  # 이미 정리함

        if sell_qty <= 0:
            if hldg_qty <= 0:
                # 실보유 0 — 이미 팔렸다. 주문 없이 ActiveTrade(day_trade)만 정리
                _stale_sell_attempted.add(_attempt_key)
                _stale_cleaned.add(_attempt_key)
                logger.warning(
                    f"[DayExit] ⚠️ 전일 포지션 {stock_code} KIS 보유 0주 (행 {row_qty}주) "
                    f"— 매도 없이 ActiveTrade(day_trade)만 닫는다"
                )
                async with day_trading_state.trade_lock:
                    if day_trading_state.positions.get(stock_code) is pos:
                        day_trading_state.remove_position(stock_code)
                    day_trading_state.closed_today.add(stock_code)
                await day_trading_state.save_daily_state()
                await enqueue_close_active_trade(stock_code, strategy_name="day_trade")
            else:
                # 보유는 있는데 DT 가용이 0 — 대기 매도·유령 비-DT 행 등으로 판단 불가.
                # 행을 닫으면 실보유가 있는데 추적이 끊기므로 유지하고 사람에게 넘긴다.
                _error_throttled(
                    f"dt_exit:stale_no_dt_available:{stock_code}",
                    f"[DayExit] 🚨 전일 포지션 {stock_code}: KIS 보유 {hldg_qty}주(주문가능 "
                    f"{sellable_qty}주) − 타전략 {other_qty}주 → DT 가용 0, 행 {row_qty}주 — "
                    f"매도·정리 보류(ActiveTrade 유지). 수동 확인 필요",
                )
            continue

        # [H-A] 실패 이력이 있으면 백오프·최대 횟수·결과 불명 확인
        _fail = _stale_sell_failures.get(_attempt_key)
        if _fail is not None:
            _cnt, _last = _fail
            _max = int(settings.DAY_TRADE_STALE_SELL_MAX_RETRIES)
            _wait = float(settings.DAY_TRADE_STALE_SELL_RETRY_BACKOFF_SEC)
            if _cnt >= _max:
                # [M-3] 업무 거부가 한도에 닿아도 당일 포기하지 않는다 — 긴 간격으로 계속 시도
                _wait = float(settings.DAY_TRADE_STALE_SELL_EXHAUSTED_RETRY_SEC)
                _error_throttled(
                    f"dt_exit:stale_retry_exhausted:{stock_code}",
                    f"[DayExit] 🚨 전일 포지션 {stock_code} 매도 업무 거부 {_cnt}회(한도 {_max}회) — "
                    f"{_wait:.0f}초 간격으로 계속 재시도한다. 수동 청산 권장(오버나이트 보유 위험)",
                )
            if time.monotonic() - _last < _wait:
                continue
            if _last_exit_order_outcome.get(stock_code) == "unknown":
                _lq = int(_stale_last_qty.get(_attempt_key) or 0)
                # [L-1 r5] 주문내역 행만으로는 스윙 매도와 구분이 안 된다 → 잔고 기준 DT 가용이
                # 직전 주문 수량만큼 줄었을 때만 접수로 본다(줄지 않았으면 재매도 — 수량 상한은 위
                # dt_available 이 이미 걸려 있다).
                _reduced = dt_available <= max(0, row_qty - _lq)
                _acc = _reduced and await _stale_sell_accepted_today_impl(
                    stock_code, qty=_lq or None,
                    since=_stale_last_since.get(_attempt_key),
                )
                if _acc:
                    _stale_sell_attempted.add(_attempt_key)
                    logger.error(
                        f"[DayExit] 🚨 전일 포지션 {stock_code}: 직전 주문 응답은 불명이었으나 "
                        f"당일 매도 접수가 확인됨 — 재주문하지 않는다(체결 후 보유 0 이면 다음 "
                        f"사이클이 ActiveTrade 를 정리)"
                    )
                    continue

        sell_pos = pos
        if sell_qty < row_qty:
            logger.warning(
                f"[DayExit] ⚠️ 전일 포지션 {stock_code} 수량 불일치 — 행 {row_qty}주, "
                f"KIS 보유 {hldg_qty}주(주문가능 {sellable_qty}주) − 타전략 {other_qty}주 "
                f"= DT 가용 {dt_available}주 → {sell_qty}주만 매도"
            )
            sell_pos = dict(pos)
            sell_pos["quantity"] = sell_qty

        price = 0.0
        try:
            price_resp = await korea_investment_api.get_current_price(stock_code)
            price = float(price_resp.get("output", {}).get("stck_prpr", 0) or 0) if price_resp else 0.0
        except Exception as e:
            logger.warning(f"[DayExit] 전일 포지션 현재가 조회 실패 ({stock_code}): {e}")
        if price <= 0:
            price = float(pos.get("entry_price", 0) or 0)
        if price <= 0:
            continue

        if _is_orphan and _is_orphan_today_position(pos):
            # [R6-1] 당일 매수분 — 사유 접두사를 분리해 재기동 원장 재구성이 당일 손익에 포함하게 한다
            # (position_manager._STALE_EXIT_REASON_PREFIXES 와 겹치지 않음). 손익은 당일 집계.
            _exit_reason = (
                f"{_ORPHAN_TODAY_EXIT_REASON_PREFIX} (진입일 {entry_date or '미상'}, 인메모리·active 행 "
                f"없음 — 당일 매수 후 포지션 기록 유실 추정, 당일 손익 집계)"
            )
            logger.warning(
                f"[DayExit] ⚠️ DT 당일 고아 잔고 즉시 청산: {stock_code} "
                f"(진입일={entry_date or '미상'}, 수량={sell_qty}/{row_qty}, KIS 보유 {hldg_qty}·"
                f"주문가능 {sellable_qty} − 타전략 {other_qty}) — 당일 손익·연패 반영"
            )
        elif _is_orphan:
            _exit_reason = (
                f"DT 고아 잔고 청산 (진입일 {entry_date or '미상'}, 인메모리·active 행 없음 — "
                f"EOD/청산 매도 미체결 추정)"
            )
            logger.warning(
                f"[DayExit] ⚠️ DT 고아 잔고 즉시 청산: {stock_code} "
                f"(진입일={entry_date or '미상'}, 수량={sell_qty}/{row_qty}, KIS 보유 {hldg_qty}·"
                f"주문가능 {sellable_qty} − 타전략 {other_qty})"
            )
        else:
            _exit_reason = f"전일 미청산 포지션 강제 청산 (진입일 {entry_date}, EOD 누락)"
            logger.warning(
                f"[DayExit] ⚠️ 전일 미청산 포지션 즉시 청산: {stock_code} "
                f"(진입일={entry_date}, 수량={sell_qty}/{row_qty})"
            )
        # [M10] 청산 성공/실패와 무관하게 당일 재진입은 금지한다. 그 사이
        # scan_and_enter 가 같은 종목을 다시 매수하면 전일 포지션이 되살아난다.
        async with day_trading_state.trade_lock:
            day_trading_state.closed_today.add(stock_code)
        # [F9] closed_today 변경분 영속화 — 재기동 후에도 당일 재진입을 막는다
        await day_trading_state.save_daily_state()

        # [M1] 전일 손익은 당일 daily_pnl / loss_streak_today 에 가산하지 않는다.
        #   [R6-1] 단 당일 매수분 고아(orphan_today)는 _finalize_exit_impl 이 당일 집계에 넣는다.
        _last_exit_order_outcome.pop(stock_code, None)
        _last_exit_order_detail.pop(stock_code, None)
        _stale_last_qty[_attempt_key] = sell_qty
        _stale_last_since[_attempt_key] = _kst_now()
        result = None
        try:
            result = await _execute_exit_stale(
                stock_code, sell_pos, price, reason=_exit_reason,
            )
        except Exception as e:
            logger.error(f"[DayExit] 전일 포지션 {stock_code} 매도 예외: {e}", exc_info=True)
        _outcome = _last_exit_order_outcome.get(stock_code)
        if result or _outcome == "accepted":
            # [H-A] 시도 기록은 **접수 성공 시에만** 남긴다
            _stale_sell_attempted.add(_attempt_key)
            _stale_sell_failures.pop(_attempt_key, None)
            if result:
                results.append(result)
        else:
            # [M-3] 일시 실패(레이트리밋·전송 전·전송 불명)는 횟수에 세지 않는다 — 업무 거부만 센다
            _detail = _last_exit_order_detail.get(stock_code)
            _prev_cnt = _stale_sell_failures.get(_attempt_key, (0, 0.0))[0]
            _transient = _is_transient_order_failure(_detail)
            _cnt = _prev_cnt if _transient else _prev_cnt + 1
            _stale_sell_failures[_attempt_key] = (_cnt, time.monotonic())
            logger.error(
                f"[DayExit] 🚨 전일 포지션 {stock_code} 매도 실패 "
                f"({'일시 실패 — 미산정' if _transient else '업무 거부'}, 누적 {_cnt}/"
                f"{int(settings.DAY_TRADE_STALE_SELL_MAX_RETRIES)}회, 결과={_outcome or '미상'}, "
                f"msg_cd={(_detail or {}).get('msg_cd')}) — 백오프 뒤 재시도"
            )
        await asyncio.sleep(0.3)  # KIS 레이트 리밋

    return results


def _error_throttled(key: str, msg: str) -> None:
    """[Low] 같은 key 의 error 로그를 레이트리밋(30초 사이클마다 반복되는 경고 억제)."""
    from app.core.log_throttle import should_log
    ok, suppressed = should_log(key)
    if ok:
        logger.error(msg + (f" (+{suppressed}건 억제)" if suppressed else ""))


async def _stale_sell_accepted_today_impl(stock_code: str, qty: int | None = None, since=None) -> bool:
    """[H-A/M-4] 당일 주문내역에 이 종목의 **유효한** 매도 접수가 있는지.

    거부·취소(체결 0) 행은 접수로 보지 않고, 주문 수량(qty)·주문 시각(since − 허용오차 이후)
    으로 범위를 좁힌다(같은 날 스윙 매도·거부 행을 DT 접수로 오인하던 결함). 조회 실패·없음 →
    False — 그 경우 재시도 주문 수량은 다시 읽은 주문가능수량(ord_psbl_qty)이 제한한다.
    """
    rows = await _lookup_todays_orders_impl(stock_code)
    return _find_accepted_sell_row(rows, stock_code, qty, since) is not None


def _balance_rows_page_full_v2(n_rows: int) -> bool:
    """[R3-1] get_balance 첫 페이지가 가득 찼는가 — 응답 밖(다음 페이지)에 종목이 있을 수 있다.

    get_balance 는 연속조회 없이 첫 페이지만 읽는다(모의 20행/실전 50행, CLAUDE.md §8.5-37).
    판정 기준은 스윙(swing/exit_service)·MDD(mdd_manager._balance_page_full_v2)와 같은
    SWING_BALANCE_FULL_PAGE_ROWS 다(경로마다 기준이 다르면 같은 계좌를 다르게 해석한다).
    """
    rows = int(getattr(settings, "SWING_BALANCE_FULL_PAGE_ROWS", 20) or 0)
    return rows > 0 and int(n_rows) >= rows


def _holding_unknown_v2(holdings: dict | None, complete: bool, stock_code: str) -> bool:
    """[R3-1] 이 종목의 실보유를 '판정 불가'로 봐야 하는가.

    조회 실패(None)이거나, 응답에 없는데 페이지가 가득 찬 경우(다음 페이지에 있을 수 있음).
    응답에 있는 종목은 페이지가 가득 차도 수량을 신뢰한다(같은 종목 행은 연속돼 나온다는 전제 —
    추정, KIS 응답 정렬 규약 미확인).
    """
    if holdings is None:
        return True
    return (not complete) and (stock_code not in holdings)


async def _fetch_kis_holdings_v2_impl() -> dict[str, dict] | None:
    """[H3/M-A] KIS 계좌 실보유 {종목코드: {"hldg": 보유수량, "sellable": 주문가능수량}}.

    조회 실패면 None (warning). 빈 목록은 "보유 0"(정상 응답)이다. `ord_psbl_qty` 가 없으면
    주문가능수량 = 보유수량으로 본다. 주문가능수량에는 대기 중인 매도 주문분이 이미 빠져 있다.

    [R3-1 2026-10-09] 첫 페이지가 가득 차(응답 밖 종목 누락 가능) 완전성을 보장할 수 없으면
    **None(실보유 미확인)** 을 돌려준다. 호출부가 '응답에 없음 = 0주'로 해석해 고아 정리·불명
    매도 접수 오판·불명 매수 미접수 오판을 하던 결함을 막는 보수적 계약이다. 종목별로 판정할 수
    있는 exit_service 내부 호출부는 _fetch_kis_holdings_v3_impl 을 쓴다.
    """
    holdings, complete = await _fetch_kis_holdings_v3_impl()
    if holdings is not None and not complete:
        _error_throttled(
            "dt_exit:holdings_page_full",
            f"[DayExit] KIS 잔고 첫 페이지가 가득 참(≥{int(getattr(settings, 'SWING_BALANCE_FULL_PAGE_ROWS', 20) or 0)}행) "
            f"— 응답 밖 종목 누락 가능, 실보유 미확인으로 처리(get_balance 연속조회 미구현, CLAUDE.md §8.5-37)",
        )
        return None
    return holdings


async def _fetch_kis_holdings_v3_impl() -> tuple[dict[str, dict] | None, bool]:
    """[R3-1] (_fetch_kis_holdings_v2_impl 과 같은 dict | None, 완전성).

    완전성 False = 첫 페이지가 가득 차 응답 밖 종목이 있을 수 있다. 이때 응답에 **없는** 종목은
    0주가 아니라 '판정 불가'로 다뤄야 한다(_holding_unknown_v2).
    """
    try:
        bal = await asyncio.wait_for(
            korea_investment_api.get_balance(), timeout=float(settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC),
        )
    except asyncio.TimeoutError:
        logger.warning(f"[DayExit] KIS 잔고 조회 {settings.DAY_TRADE_EXIT_FAST_STOP_TIMEOUT_SEC:.0f}초 초과 — 실보유 미확인")
        return None, False
    except Exception as e:
        logger.warning(f"[DayExit] KIS 잔고 조회 실패 — 실보유 미확인: {e}")
        return None, False
    if not isinstance(bal, dict):
        logger.warning(f"[DayExit] KIS 잔고 응답 형식 이상({type(bal).__name__}) — 실보유 미확인")
        return None, False
    if bal.get("rt_cd") not in (None, "0"):
        logger.warning(f"[DayExit] KIS 잔고 조회 오류 응답: {bal.get('msg1')} — 실보유 미확인")
        return None, False
    rows = bal.get("output1")
    if not isinstance(rows, list):
        logger.warning("[DayExit] KIS 잔고 응답에 output1 없음 — 실보유 미확인")
        return None, False
    out: dict[str, dict] = {}
    for h in rows:
        if not isinstance(h, dict):
            continue
        code = str(h.get("pdno", "") or "")
        if not code:
            continue
        try:
            hq = max(0, int(float(h.get("hldg_qty", 0) or 0)))
            sq = hq
            if h.get("ord_psbl_qty") not in (None, ""):
                sq = max(0, min(hq, int(float(h.get("ord_psbl_qty") or 0))))
        except (TypeError, ValueError):
            logger.warning(f"[DayExit] KIS 잔고 수량 파싱 실패 ({code}) — 실보유 미확인")
            return None, False
        cur = out.setdefault(code, {"hldg": 0, "sellable": 0})
        cur["hldg"] += hq
        cur["sellable"] += sq
    return out, not _balance_rows_page_full_v2(len(rows))


async def force_close_all(reason: str = "강제 청산") -> list[dict]:
    """모든 데이트레이딩 포지션 강제 청산.

    호출 시점:
      - EOD 15:20 (CLAUDE.md 7.3)
      - 일일 손실 한도(-3%) 초과
      - 수동 API 호출

    현재가 조회 실패 시 진입가를 대체가로 사용.
    KIS API 레이트 리밋 방지를 위해 종목 간 0.3초 대기.

    Args:
        reason: 강제 청산 사유 (로그 및 DB 기록용)

    Returns:
        list[dict]: 청산 실행 결과 목록
    """
    results = []
    failed_codes: list[str] = []
    codes = list(day_trading_state.positions.keys())
    stale_skipped = 0

    for stock_code in codes:
        pos = day_trading_state.get_position(stock_code)
        if not pos:
            continue
        # [R1-4] 전일 포지션은 일반 경로(_execute_exit, count_daily_pnl=True)로 팔지 않는다.
        # 아래에서 close_stale_positions 로 넘긴다(실보유·타전략 차감, 당일 손익 제외).
        if _is_stale_position(pos):
            _warn_stale_in_general_path(stock_code, f"강제 청산({reason})")
            stale_skipped += 1
            continue
        # [H-B] 진행 중 청산 주문이 있으면 건너뛴다. 진행 중 표식은 shield 작업의 finally 가
        # 해제하므로(결과 불명이면 DAY_TRADE_EXIT_UNKNOWN_HOLD_SEC 뒤) 다음 EOD 사이클(30초)에
        # 남은 포지션을 다시 잡는다 — 영구 skip 되지 않는다.
        if day_trading_state.is_exit_in_progress(stock_code):
            logger.warning(f"[DayExit] 강제 청산: {stock_code} 청산 주문 진행 중 — 이번 회차 건너뜀")
            continue
        try:
            # 현재가 조회
            price_resp = await korea_investment_api.get_current_price(stock_code)
            price = float(
                price_resp.get("output", {}).get("stck_prpr", 0) or 0
            ) if price_resp else 0.0

            if price <= 0:
                price = float(pos.get("entry_price", 0))

            result = await _execute_exit(stock_code, pos, price, reason)
            if result:
                results.append(result)
            else:
                # [R2-5] 예외 없이 None(KIS 거부·결과 불명 보류 등)도 실패로 집계해 재시도·경보한다.
                logger.error(f"[DayExit] 강제 청산 미체결 반환 ({stock_code}) — 재시도 대상")
                failed_codes.append(stock_code)

            await asyncio.sleep(0.3)  # KIS API 레이트 리밋

        except Exception as e:
            logger.error(f"[DayExit] 강제 청산 오류 ({stock_code}): {e}", exc_info=True)
            failed_codes.append(stock_code)

    # S4 fix: 실패 종목 1회 재시도
    if failed_codes:
        logger.warning(f"[DayExit] 강제 청산 실패 {len(failed_codes)}건 — 2초 후 재시도: {failed_codes}")
        await asyncio.sleep(2.0)
        still_failed: list[str] = []
        for stock_code in failed_codes:
            pos = day_trading_state.get_position(stock_code)
            if not pos:
                continue
            if day_trading_state.is_exit_in_progress(stock_code):
                # 결과 불명 보류 등 — 이번 회차는 건드리지 않는다(다음 사이클·장 마감 후 잔존 점검이 잡는다)
                continue
            try:
                price_resp = await korea_investment_api.get_current_price(stock_code)
                price = float(
                    price_resp.get("output", {}).get("stck_prpr", 0) or 0
                ) if price_resp else 0.0
                if price <= 0:
                    price = float(pos.get("entry_price", 0))
                result = await _execute_exit(stock_code, pos, price, reason)
                if result:
                    results.append(result)
                elif day_trading_state.get_position(stock_code):
                    logger.critical(f"[DayExit] 강제 청산 재시도 미체결 반환 ({stock_code}) — 포지션 미청산 위험!")
                    still_failed.append(stock_code)
                await asyncio.sleep(0.3)
            except Exception as e:
                logger.critical(f"[DayExit] 강제 청산 재시도 실패 ({stock_code}): {e} — 포지션 미청산 위험!", exc_info=True)
                still_failed.append(stock_code)
        if still_failed:
            logger.critical(f"[DayExit] 최종 미청산 종목: {still_failed} — 수동 청산 필요!")
            # [R2-5] critical 로그만으로는 15:30 이후 오버나이트 보유를 아무도 모른다 → 텔레그램.
            _desc = ", ".join(
                f"{c}({int((day_trading_state.get_position(c) or {}).get('quantity', 0) or 0)}주)"
                for c in still_failed
            )
            _alert_once_v2(
                "DT_EOD", f"force_close:{reason}:{','.join(sorted(still_failed))}",
                f"[데이트레이딩] '{reason}' 재시도 후에도 미청산: {_desc} — 다음 사이클에 재시도하지만 "
                f"장 마감(15:30) 전에 계좌를 확인하고 필요하면 수동 청산하세요.",
            )

    # [R2-5] EOD 구간이면 장 마감 직후 잔존 DT 포지션 점검을 하루 1회 예약한다.
    if is_eod_flat_time():
        _schedule_post_close_residual_check_impl()

    if stale_skipped:
        try:
            results.extend(await close_stale_positions() or [])
        except Exception as e:
            logger.error(f"[DayExit] 강제 청산 중 전일 포지션 청산 실패: {e} — 수동 확인 필요", exc_info=True)

    logger.info(f"[DayExit] 강제 청산 완료: {len(results)}건 ({reason})")
    return results


# ─────────────────────────────────────────────────────────────────────────────
# [R2-5 2026-10-09] EOD 미청산 경보
# ─────────────────────────────────────────────────────────────────────────────
# (날짜, tag:key) — 하루 1회만 텔레그램
_once_alerted: set[tuple[str, str]] = set()
# 장 마감 후 잔존 점검을 예약한 날짜
_post_close_check_scheduled: set[str] = set()
_post_close_tasks: set = set()


def _alert_once_v2(tag: str, key: str, msg: str) -> None:
    """logger.critical + 텔레그램 notify_system_warning(tag)(하루 1회/키). 발송 실패는 error 로그만."""
    logger.critical(f"[DayExit] 🚨 {msg}")
    _k = (_today_key(), f"{tag}:{key}")
    for k in [k for k in _once_alerted if k[0] != _k[0]]:
        _once_alerted.discard(k)
    if _k in _once_alerted:
        return
    _once_alerted.add(_k)
    try:
        from app.services.telegram_notifier import telegram_notifier
        _t = asyncio.get_running_loop().create_task(asyncio.wait_for(
            telegram_notifier.notify_system_warning(tag, msg), timeout=10.0,
        ))
        _alert_tasks.add(_t)
        _t.add_done_callback(_alert_tasks.discard)
    except Exception as e:
        logger.error(f"[DayExit] {tag} 경보 발송 실패({key}): {e}")


def check_post_close_residual_v2() -> list[str]:
    """장 마감 후 인메모리 DT 포지션이 남아 있으면 텔레그램 경보(DT_EOD). 잔존 종목 코드 반환.

    §7 'EOD 전량 청산' 이 지켜지지 않은 상태 — 다음 영업일 close_stale_positions 까지 손절 없이
    오버나이트로 남는다. 전일(stale) 포지션도 포함해 센다.
    """
    residual = sorted(day_trading_state.positions.keys())
    if not residual:
        logger.info("[DayExit] 장 마감 후 잔존 DT 포지션 없음 — EOD 전량 청산 확인")
        return []
    _desc = ", ".join(
        f"{c}({int((day_trading_state.positions.get(c) or {}).get('quantity', 0) or 0)}주"
        f"{', 청산 진행/불명' if day_trading_state.is_exit_in_progress(c) else ''})"
        for c in residual
    )
    _alert_once_v2(
        "DT_EOD", f"post_close:{','.join(residual)}",
        f"[데이트레이딩] 장 마감 후 미청산 DT 포지션 {len(residual)}종목: {_desc} — 오버나이트 보유 중. "
        f"다음 영업일 장중 첫 사이클에 전일 포지션 청산이 시도됩니다. 계좌를 확인하세요.",
    )
    return residual


# [R3-2] 종가 재확인 1회 조회의 여유 시간(초) — fetch_fill_v2 왕복 상한 추정(config 미편입)
_AUCTION_RECHECK_FETCH_SLACK_SEC = 30.0


def _auction_recheck_wait_cap_impl() -> float:
    """[R3-2] 진행 중인 종가 재확인 태스크를 기다릴 최대 초 — 재확인 횟수×(간격+조회 여유)+여유."""
    _, _attempts, _interval = _auction_recheck_params_impl()
    return float(_attempts) * (float(_interval) + _AUCTION_RECHECK_FETCH_SLACK_SEC) + _AUCTION_RECHECK_FETCH_SLACK_SEC


async def _await_auction_rechecks_impl() -> None:
    """[R3-2] 진행 중인 종가 단일가 매도 재확인 태스크가 끝날 때까지 대기(상한 있음).

    상한을 넘기면 error 를 남기고 점검을 진행한다(남은 재확인이 나중에 복원하면 _on_unfilled 가
    장 마감 후 복원 경보를 직접 보낸다 — 이중 안전).
    """
    pending = [t for t in list(_auction_fill_tasks) if not t.done()]
    if not pending:
        return
    _cap = _auction_recheck_wait_cap_impl()
    _done, _not_done = await asyncio.wait(pending, timeout=_cap)
    if _not_done:
        logger.error(
            f"[DayExit] 장 마감 후 잔존 점검: 종가 매도 재확인 {len(_not_done)}건이 {_cap:.0f}초 안에 끝나지 않음 "
            f"— 점검을 진행한다(이후 0체결 복원 시 별도 경보)"
        )


def _schedule_post_close_residual_check_impl() -> None:
    """EOD 구간 첫 호출에서 '장 마감 + DAY_TRADE_FILL_AUCTION_RECHECK_DELAY_SEC' 뒤 잔존 점검 1회 예약."""
    _day = _today_key()
    if _day in _post_close_check_scheduled:
        return
    for d in [d for d in _post_close_check_scheduled if d != _day]:
        _post_close_check_scheduled.discard(d)

    async def _run() -> None:
        await asyncio.sleep(_seconds_until_auction_recheck_impl(_kst_now()))
        # [R3-2] 종가 단일가 매도 재확인(_auction_fill_tasks)도 같은 시각에 깨어나 KIS 조회를 마친 뒤에야
        # 0체결·부분체결 포지션을 되살린다. 그보다 먼저 점검하면 '잔존 없음'을 확정해 버린다 →
        # 진행 중인 재확인이 끝날 때까지(상한: 재확인 횟수×(간격+조회 여유)+여유) 기다린 뒤 점검한다.
        await _await_auction_rechecks_impl()
        try:
            check_post_close_residual_v2()
        except Exception as e:
            logger.error(f"[DayExit] 장 마감 후 잔존 점검 실패: {e} — 수동 확인 필요", exc_info=True)

    try:
        task = asyncio.get_running_loop().create_task(_run())
    except RuntimeError as e:
        logger.error(f"[DayExit] 장 마감 후 잔존 점검 예약 실패: {e}")
        return
    _post_close_check_scheduled.add(_day)
    _post_close_tasks.add(task)
    task.add_done_callback(_post_close_tasks.discard)