auto-stock/backend/app/services/swing/fill_sweep.py

"""체결가 미확인 TradeHistory 후속 보정 (스윙 스케줄러 사이클에서 호출).

[2026-10-08] 운영 관측: TradeHistory #432(풍산)·#433(현대차)의 executed_price 가 주문가
(현재가×1.005, 호가단위상 불가능한 값)로 남았다. 주문 직후의 체결 확인
(`trade_executor._verify_and_sync_execution_v3`)이 폴링 15초 안에 체결을 못 보거나 조회가
실패하면 그 뒤로 아무도 체결가를 채우지 않았다.

이제 주문 경로들은 확인 전 executed_price 를 **NULL** 로 기록한다(주문가 대체 금지). 이 모듈은
"당일 · COMPLETED · order_id 있음 · executed_price IS NULL · 일정 시간 경과" 행을 골라
KIS 체결조회(inquire-daily-ccld)로 실제 체결가·체결수량을 확인하고 db_worker 큐로 갱신한다.

규칙
  - DB 는 **읽기만**(호출부의 SELECT 전용 세션). 쓰기는 db_worker 큐(§3.2).
  - 사이클당 KIS 조회 상한·행별 재시도 간격·최대 시도 수를 둔다(모의 초당 2건 제한 대비).
  - 체결 0 · 미종결 부분체결은 갱신하지 않는다(상태 변경 없음 — 수동 확인 대상, warning).
  - SELL 손익은 기록 시점 추정치에서 진입가를 역산해 체결가로 재계산한다:
      p(추정가) = total_amount / order_quantity,
      entry = p − (net_profit_loss + commission) / order_quantity
    (모든 SELL 기록 경로가 total_amount=추정가×수량, net = (추정가−진입가)×수량 − commission
    으로 기록한다는 전제. 전제가 깨지면 손익은 갱신하지 않고 가격·금액만 갱신한다.)
  - [하드닝 R1 2026-10-09] 미확인 판정은 `unconfirmed_fill_clause_v2`(NULL + 레거시 '체결가=주문가 &
    slippage NULL'). BUY 의 ActiveTrade 진입가 보정은 entry_reason 으로 확정한 전략명(동등)만 —
    확정 불가면 보정하지 않고 warning. 일일 지표는 `get_fill_sweep_health_v2()`.
"""
from __future__ import annotations

import asyncio
import logging
import time
from datetime import datetime, time as dtime, timedelta

from sqlalchemy import select, or_, and_

from app.core.timeutil import now_kst
from app.models.models import ActiveTrade, TradeHistory
from app.services.utils import _is_domestic
from app.services.db_worker import DT_STRATEGY_NAMES, not_dt_strategy_clause

logger = logging.getLogger(__name__)

# 사이클당 최대 KIS 체결조회 수 / 주문 후 최소 경과(직후 확인 태스크와 경합 방지) /
# 같은 행 재조회 최소 간격 / 행별 최대 시도
FILL_SWEEP_MAX_LOOKUPS_PER_CYCLE = 2
FILL_SWEEP_MIN_AGE_SEC = 120
FILL_SWEEP_RETRY_INTERVAL_SEC = 180
FILL_SWEEP_MAX_ATTEMPTS = 5

# [하드닝 R4 2026-10-09] 종가 단일가(15:20~15:30) 주문 체결 확인.
# 동시호가 주문은 15:30 에야 체결되는데 sweep 은 장중(is_market_open, 15:30 미만)에만 돌았다 →
# 15:28 주문은 sweep 기회 0회로 executed_price 가 영구 NULL 이었다(운영 10/7 스윙 BUY 2건 15:22·15:28).
#   - 장 마감 직후 창(15:30 ~ +SWING_FILL_SWEEP_POST_CLOSE_MINUTES)에 스케줄러의 장외 분기가
#     `run_post_close_fill_sweep_v2` 로 같은 sweep 을 돌린다(당일 행 그대로 — 같은 날짜 조회).
#   - 동시호가 구간에 들어온 주문은 15:30 전엔 체결될 수 없으므로 조회하지 않는다(시도 횟수 보존).
_KRX_CLOSING_AUCTION_START = dtime(15, 20)
_KRX_MARKET_CLOSE = dtime(15, 30)
_DEFAULT_POST_CLOSE_MINUTES = 30
_DEFAULT_POST_CLOSE_INTERVAL_SEC = 60.0
_DEFAULT_POST_CLOSE_TIMEOUT_SEC = 60.0


def _settings_num_v2(name: str, default: float) -> float:
    try:
        from app.core.config import settings
        v = float(getattr(settings, name, default))
    except Exception as e:
        logger.warning(f"[FillSweep] 설정 {name} 읽기 실패 — 기본값 {default}: {type(e).__name__}: {e}")
        return float(default)
    return v if v > 0 else float(default)


def _post_close_window_v2(now: datetime) -> tuple[datetime, datetime]:
    start = now.replace(hour=_KRX_MARKET_CLOSE.hour, minute=_KRX_MARKET_CLOSE.minute,
                        second=0, microsecond=0)
    minutes = _settings_num_v2("SWING_FILL_SWEEP_POST_CLOSE_MINUTES", _DEFAULT_POST_CLOSE_MINUTES)
    return start, start + timedelta(minutes=minutes)


def post_close_sweep_active_v2(now: datetime | None = None) -> bool:
    """KRX 영업일 15:30 ~ 15:30+N분(장 마감 직후 체결 확인 창) 여부."""
    from app.core.market_schedule import is_krx_holiday
    now = now or now_kst()
    if now.weekday() >= 5 or is_krx_holiday(now.date()):
        return False
    start, end = _post_close_window_v2(now)
    return start <= now < end


def _auction_fill_pending_v2(created_at, now: datetime) -> bool:
    """동시호가(15:20~) 당일 주문인데 아직 15:30 전 — 체결될 수 없으므로 조회 보류."""
    if created_at is None or not hasattr(created_at, "time"):
        return False
    try:
        same_day = created_at.date() == now.date()
    except Exception as e:
        logger.warning(f"[FillSweep] created_at 비교 실패: {type(e).__name__}: {e}")
        return False
    return same_day and created_at.time() >= _KRX_CLOSING_AUCTION_START and now.time() < _KRX_MARKET_CLOSE


async def run_post_close_fill_sweep_v2(now: datetime | None = None) -> float | None:
    """장 마감 직후 체결 확인 1회 (스윙 스케줄러 장외 분기에서 호출).

    창 밖이면 None(호출부는 기존 대기 그대로). 창 안이면 sweep 을 돌리고 다음 호출까지의
    권장 대기 상한(초)을 반환한다 — 장외 분기의 최대 1800초 대기가 창을 건너뛰지 않게.
    DB 는 SELECT 전용 세션, 쓰기는 db_worker 큐(sweep 본체와 동일). 예외·시간초과는 warning
    후 흡수한다(체결가 보정은 리스크 가드가 아니다 — 다음 호출에서 재시도).
    """
    now = now or now_kst()
    if not post_close_sweep_active_v2(now):
        return None
    _, end = _post_close_window_v2(now)
    interval = _settings_num_v2("SWING_FILL_SWEEP_POST_CLOSE_INTERVAL_SEC", _DEFAULT_POST_CLOSE_INTERVAL_SEC)
    timeout = _settings_num_v2("SWING_FILL_SWEEP_TIMEOUT_SEC", _DEFAULT_POST_CLOSE_TIMEOUT_SEC)
    try:
        from app.core.database import AsyncSessionLocal
        # DB-READ-EXCEPTION: 장 마감 직후 체결가 미확인 TradeHistory/ActiveTrade SELECT 전용(commit 없음, 갱신은 db_task_queue 경유) — CLAUDE.md §3.2 ③-b 스윙 스케줄러 SELECT
        async with AsyncSessionLocal() as session:
            n = await asyncio.wait_for(sweep_unconfirmed_fills_v2(session), timeout=timeout)
        if n:
            logger.info(f"[FillSweep] 장 마감 후 체결가 보정 {n}건")
    except asyncio.TimeoutError:
        logger.warning(f"[FillSweep] 장 마감 후 체결가 보정 {timeout:.0f}초 초과 — 다음 호출에서 재시도")
    except Exception as e:
        logger.warning(f"[FillSweep] 장 마감 후 체결가 보정 실패: {type(e).__name__}: {e}")
    remain = (end - now).total_seconds()
    return max(15.0, min(interval, remain))


# {trade_id: (attempts, last_try_monotonic)} — 당일 한정(날짜가 바뀌면 비운다)
_attempts: dict[int, tuple[int, float]] = {}
_attempts_day: str = ""
# [하드닝 R5 2026-10-09] 당일 갱신을 이미 큐잉한 TradeHistory id. 워커 반영 전(큐 지연·DLQ)에는 행이
# 여전히 '미확인'으로 조회되므로, 재조회·재큐잉(특히 비멱등 진입가 보정)을 막기 위해 당일 제외한다.
_queued_ids: set[int] = set()


def _reset_if_new_day(day_iso: str) -> None:
    global _attempts_day
    if _attempts_day != day_iso:
        _attempts.clear()
        _queued_ids.clear()
        _attempts_day = day_iso


def _exhausted_ids() -> list[int]:
    return [tid for tid, (n, _) in _attempts.items() if n >= FILL_SWEEP_MAX_ATTEMPTS]


def _due(trade_id: int, now_mono: float) -> bool:
    n, last = _attempts.get(trade_id, (0, 0.0))
    if n >= FILL_SWEEP_MAX_ATTEMPTS:
        return False
    return n == 0 or (now_mono - last) >= FILL_SWEEP_RETRY_INTERVAL_SEC


def _note_attempt(trade_id: int, now_mono: float) -> int:
    n, _ = _attempts.get(trade_id, (0, 0.0))
    n += 1
    _attempts[trade_id] = (n, now_mono)
    return n


def derive_sell_entry_price_v2(row) -> float | None:
    """기록 시점 추정치에서 진입가를 역산 (순수 함수). 불가하면 None."""
    try:
        q = int(row.order_quantity or 0)
        amt = float(row.total_amount or 0)
        if row.net_profit_loss is None or q <= 0 or amt <= 0:
            return None
        p = amt / q
        entry = p - (float(row.net_profit_loss) + float(row.commission or 0)) / q
    except (TypeError, ValueError, ZeroDivisionError):
        return None
    return entry if entry > 0 else None


# 진입가 보정 대상 TradeHistory.strategy_type — 스윙 소유만(DT 는 fill_reconciler, 수동·유입은
# ActiveTrade 진입가를 이 주문가로 기록하지 않는다).
_SWING_BUY_STRATEGY_TYPES = (None, "", "swing")


# [하드닝 R1 2026-10-09] 시딩(동기화 유입) ActiveTrade 식별 — swing/exit_service 와 같은 규약.
# 시딩 행의 entry_price 는 KIS 평단이라 주문가→체결가 보정 대상이 아니다.
_SEEDED_STRATEGY_NAMES = ("synced",)
_SEEDED_ENTRY_REASON_PREFIXES = ("동기화 유입", "외부 유입 포지션")


def _is_seeded_active_row_v2(r) -> bool:
    name = (getattr(r, "strategy_name", None) or "").strip()
    reason = (getattr(r, "entry_reason", None) or "").strip()
    return name in _SEEDED_STRATEGY_NAMES or reason.startswith(_SEEDED_ENTRY_REASON_PREFIXES)


def resolve_entry_fill_target_v2(row, active_rows) -> tuple[str | None, str]:
    """BUY TradeHistory 행이 만든(갱신한) 스윙 ActiveTrade 의 strategy_name 을 확정 (순수 함수).

    [하드닝 R1 2026-10-09] 이전에는 `strategy_name_not=DT` 로 '가장 최근 비-DT active 행'을
    보정했다. 같은 종목에 시딩('synced') 행이나 다른 듀얼 전략 행이 있으면 엉뚱한 행의
    entry_price 가 (체결가−주문가)×수량/행수량 만큼 틀어졌다(운영: 103140·005380 은 스윙 행이
    이미 closed 이고 synced 행만 active — 구 경로라면 synced 행을 보정했다).

    매칭: 진입 시 executor 가 TradeHistory·ActiveTrade 양쪽에 **같은 entry_reason** 을 기록한다
    (추가매수도 ActiveTrade.entry_reason 을 최신 매수 사유로 덮어쓴다). 따라서
      1) 비-DT·비시딩·strategy_name 있는 active 행 중 entry_reason 이 같은 행을 찾고
      2) 그 strategy_name 이 하나로 확정되며
      3) db_worker 가 그 전략명으로 고를 행(같은 전략명 중 id 최대)이 바로 그 행일 때만
    전략명을 돌려준다. 하나라도 어긋나면 (None, 사유) — 보정하지 않는다(확정 불가 시 보정 금지).

    Returns:
        (strategy_name | None, 사유 문자열)
    """
    reason = (getattr(row, "entry_reason", None) or "").strip()
    if not reason:
        return None, "TradeHistory.entry_reason 없음"
    eligible = [
        r for r in (active_rows or [])
        if (getattr(r, "strategy_name", None) or "").strip()
        and (getattr(r, "strategy_name", None) or "") not in DT_STRATEGY_NAMES
        and not _is_seeded_active_row_v2(r)
    ]
    matched = [r for r in eligible if (getattr(r, "entry_reason", None) or "").strip() == reason]
    if not matched:
        return None, "entry_reason 일치하는 스윙 소유 active 행 없음(청산됐거나 사유 갱신됨)"
    names = {r.strategy_name for r in matched}
    if len(names) != 1:
        return None, f"entry_reason 일치 행의 전략명이 여러 개 {sorted(names)}"
    name = next(iter(names))
    same_name = [r for r in eligible if r.strategy_name == name]
    newest = max(same_name, key=lambda r: int(getattr(r, "id", 0) or 0))
    if newest not in matched:
        return None, f"전략명 {name} 의 최신 active 행(id={getattr(newest, 'id', None)})이 이 매수의 행이 아님"
    return name, "ok"


async def _load_active_rows_for_fill_v2(db, stock_code: str):
    """SELECT 전용 — 호출부 세션(스윙 스케줄러 exit_session, DB-READ-EXCEPTION)으로 읽는다."""
    return (await db.execute(
        select(ActiveTrade).where(
            ActiveTrade.stock_code == stock_code,
            ActiveTrade.status == "active",
            not_dt_strategy_clause(ActiveTrade.strategy_name),
        )
    )).scalars().all()


async def _enqueue_buy_entry_adjust_v2(row, filled_qty: int, filled_price: float, db=None) -> bool:
    """스윙 BUY 체결 확인 시 **이 매수가 만든** ActiveTrade.entry_price 를 체결가로 보정 (큐 경유).

    직후 확인(`_verify_and_sync_execution_v3`)이 이미 보정을 큐잉한 주문이면 건너뛴다(비멱등).
    대상 행은 `resolve_entry_fill_target_v2` 로 전략명을 확정해 `strategy_name`(동등) 으로 지정한다.
    확정할 수 없으면(db 없음 포함) 보정하지 않고 warning 만 남긴다 — 엉뚱한 행(시딩·타 전략)의
    진입가를 움직이면 손절선·손익이 틀어지기 때문.
    """
    from app.services.trade_executor import entry_fill_already_adjusted_v2, note_entry_fill_adjusted_v2
    from app.services.db_worker import enqueue_active_trade_upsert

    if getattr(row, "strategy_type", None) not in _SWING_BUY_STRATEGY_TYPES:
        return False
    if entry_fill_already_adjusted_v2(row.order_id):
        return False
    order_price = float(row.order_price or 0)
    if order_price <= 0 or filled_price <= 0 or filled_qty <= 0:
        return False
    if db is None:
        logger.warning(
            f"[FillSweep] {row.stock_code} #{getattr(row, 'id', None)} 진입가 보정 생략 — "
            f"ActiveTrade 조회 세션 없음(대상 행 확정 불가)"
        )
        return False
    try:
        active_rows = await _load_active_rows_for_fill_v2(db, row.stock_code)
    except Exception as e:
        logger.warning(
            f"[FillSweep] {row.stock_code} #{getattr(row, 'id', None)} 진입가 보정 생략 — "
            f"ActiveTrade 조회 실패: {type(e).__name__}: {e}"
        )
        return False
    target, why = resolve_entry_fill_target_v2(row, active_rows)
    if not target:
        logger.warning(
            f"[FillSweep] {row.stock_code} #{getattr(row, 'id', None)} ActiveTrade 진입가 보정 생략 — "
            f"{why} (주문가 {order_price:,.2f} → 체결가 {float(filled_price):,.2f}, 수동 확인)"
        )
        return False
    # [하드닝 R5 2026-10-09] 큐잉 **전에** 보정 완료로 표시한다(trade_executor 직후 확인 경로와 동일 순서).
    # entry_fill_adjust 는 비멱등(entry_price += 차이)이라, 태스크가 큐에 머무는/DLQ 재시도 중인 동안
    # 다음 sweep 이 같은 주문을 다시 보정하면 진입가·손절선이 누적 이동한다. 표시 후 큐잉이 실패하면
    # 보정이 유실될 뿐(손절선 불변) 이중 이동은 일어나지 않는다 — 더 안전한 쪽.
    note_entry_fill_adjusted_v2(row.order_id)
    await enqueue_active_trade_upsert({
        "stock_code": row.stock_code,
        "strategy_name": target,
        "entry_fill_adjust": {
            "order_price": order_price,
            "filled_price": float(filled_price),
            "filled_qty": int(filled_qty),
        },
    })
    return True


# ── [하드닝 R9 2026-10-12] 늦게 확정된 0체결·부분체결 스윙 BUY 의 ActiveTrade 원복 ─────────────
# 직후 확인(trade_executor._verify_and_sync_execution_v2)이 15초 폴링 동안 잔량 대기로 끝나면 확정은
# 여기 몫인데, 이전에는 0체결 확정을 warning 후 continue(CANCELLED 강등·ActiveTrade close 없음),
# 부분체결 확정은 TradeHistory·진입가 보정만 하고 ActiveTrade.quantity 를 주문수량으로 남겼다
# → 유령 active 행·수량 과대가 sync 유령 정리 전까지 분할매도 수량·보유 슬롯을 왜곡했다.
# 원복 정보는 (1) 같은 프로세스면 trade_executor 레지스트리(직후 확인이 재등록), (2) 재기동 등으로
# 없으면 TradeHistory 행(entry_reason·order_quantity)과 active 행으로 재구성한다.
# 멱등: 행 id 를 `_queued_ids` 에 먼저 넣고, close 는 이미 닫힌 행에 no-op, 축소는 절대값(=체결수량).
# Portfolio 는 스윙 주문 시점에 갱신하지 않으므로(KIS sync 가 진실) 되돌릴 것이 없다.


def _is_swing_buy_row_v2(row) -> bool:
    return (getattr(row, "order_type", None) == "BUY"
            and getattr(row, "strategy_type", None) in _SWING_BUY_STRATEGY_TYPES)


def reconstruct_buy_rollback_v2(row, active_rows, filled_qty: int) -> tuple[dict | None, str]:
    """TradeHistory BUY 행 + 그 종목 active 행들로 원복 정보를 재구성 (순수 함수).

    반환 (rollback | None, 사유). rollback 은 trade_executor 레지스트리와 같은 모양
    {stock_code, strategy_name, quantity, accumulate}.
    - 대상 행 확정은 진입가 보정과 같은 규칙(resolve_entry_fill_target_v2) — 확정 불가면 원복 안 함.
    - 행 수량 == 주문수량 → 신규 진입(accumulate False).
    - 행 수량 > 주문수량 → 추가매수(accumulate True — 원복 함수가 수동 확인 경보).
    - 행 수량 == 체결수량(< 주문수량) → 이미 원복됨(None, "already").
    - 그 외(행 수량 < 주문수량: 이미 일부 매도 등) → 역산 근거 없음(None).
    """
    ordered = int(getattr(row, "order_quantity", 0) or 0)
    if ordered <= 0:
        return None, "order_quantity 없음"
    target, why = resolve_entry_fill_target_v2(row, active_rows)
    if not target:
        return None, why
    same = [r for r in (active_rows or []) if getattr(r, "strategy_name", None) == target]
    newest = max(same, key=lambda r: int(getattr(r, "id", 0) or 0))
    at_qty = int(getattr(newest, "quantity", 0) or 0)
    rb = {
        "stock_code": row.stock_code,
        "strategy_name": target,
        "quantity": ordered,
        "accumulate": at_qty > ordered,
    }
    if at_qty >= ordered:
        return rb, "ok"
    if filled_qty > 0 and at_qty == int(filled_qty):
        return None, "already"
    return None, f"active 행 수량 {at_qty} < 주문수량 {ordered}(일부 매도 등) — 역산 근거 없음"


async def _resolve_buy_rollback_v2(row, db, filled_qty: int) -> dict | None:
    """레지스트리 우선, 없으면 TradeHistory 행으로 재구성. 확정 불가면 warning 후 None."""
    from app.services.trade_executor import _pop_buy_entry_rollback_v2
    rb = _pop_buy_entry_rollback_v2(row.order_id, row.stock_code)
    if rb:
        return rb
    if db is None:
        logger.warning(
            f"[FillSweep] {row.stock_code} #{getattr(row, 'id', None)} BUY 원복 정보 없음·조회 세션 없음 "
            f"— ActiveTrade 수동 확인 필요"
        )
        return None
    try:
        active_rows = await _load_active_rows_for_fill_v2(db, row.stock_code)
    except Exception as e:
        logger.warning(
            f"[FillSweep] {row.stock_code} #{getattr(row, 'id', None)} BUY 원복용 ActiveTrade 조회 실패 — "
            f"수동 확인 필요: {type(e).__name__}: {e}"
        )
        return None
    rb, why = reconstruct_buy_rollback_v2(row, active_rows, filled_qty)
    if rb is None and why != "already":
        logger.warning(
            f"[FillSweep] {row.stock_code} #{getattr(row, 'id', None)} BUY 체결 {filled_qty}/"
            f"{int(row.order_quantity or 0)}주 확정 — ActiveTrade 원복 생략: {why} (수동 확인)"
        )
    return rb


def unconfirmed_fill_clause_v2():
    """'체결가 미확인' TradeHistory 판정식 (모니터링 지표와 sweep 이 공유하는 단일 정의)."""
    return or_(
        TradeHistory.executed_price.is_(None),
        and_(
            TradeHistory.slippage.is_(None),
            TradeHistory.order_price.isnot(None),
            TradeHistory.executed_price == TradeHistory.order_price,
        ),
    )


# 일일 모니터링 지표 — 마지막 sweep 기준(프로세스 메모리, /health·로그용)
_last_stats: dict = {"day": "", "candidates": 0, "exhausted": 0, "queued": 0, "exhausted_warned": 0}


def get_fill_sweep_health_v2() -> dict:
    """당일 체결 미확인 지표: candidates=마지막 sweep 이 본 미확인 행 수(상한 20),
    exhausted=최대 시도 소진(수동 확인 대상) 행 수, queued=당일 보정 큐잉 누계."""
    return dict(_last_stats)


async def sweep_unconfirmed_fills_v2(db) -> int:
    """체결가 미확인 행 보정. 반환 = 갱신 큐잉 건수.

    호출부는 SELECT 전용 세션을 넘긴다(commit 없음).
    """
    from app.services.korea_investment_api import korea_investment_api
    from app.services.trade_executor import _fill_update_fields_v2
    from app.services.db_worker import enqueue_update_trade_history

    now = now_kst()
    _reset_if_new_day(now.date().isoformat())
    day_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    # created_at 은 naive KST 로 저장된다(now_kst 기본값과 동일 규약)
    cutoff = now - timedelta(seconds=FILL_SWEEP_MIN_AGE_SEC)
    if getattr(day_start, "tzinfo", None) is not None:
        day_start = day_start.replace(tzinfo=None)
        cutoff = cutoff.replace(tzinfo=None)

    if _last_stats.get("day") != now.date().isoformat():
        _last_stats.update(day=now.date().isoformat(), candidates=0, exhausted=0, queued=0,
                           exhausted_warned=0)
    rows = (await db.execute(
        select(TradeHistory)
        .where(
            TradeHistory.status == "COMPLETED",
            # [하드닝 R1 2026-10-09] 미확인 = 체결가 NULL(현 규약) 또는 레거시 규약
            # (체결가=주문가 대체 + slippage NULL — 체결 확인은 항상 slippage 를 채운다).
            # 운영 10/1~: 스윙 SELL 15/15·BUY 3/12 가 후자로 남아 손익·슬리피지가 주문가 기준이었다.
            unconfirmed_fill_clause_v2(),
            TradeHistory.order_id.isnot(None),
            TradeHistory.order_id != "",
            TradeHistory.created_at >= day_start,
            TradeHistory.created_at <= cutoff,
            or_(TradeHistory.currency_code.is_(None), TradeHistory.currency_code == "KRW"),
            TradeHistory.order_type.in_(("BUY", "SELL")),
            # DT 체결 보정은 daytrading/fill_reconciler 소관 — 이중 보정 방지(NULL 은 포함)
            or_(TradeHistory.strategy_type.is_(None),
                TradeHistory.strategy_type.notin_(DT_STRATEGY_NAMES)),
            # 시도 소진 행이 상위 20건을 영구 점유해 새 행이 밀리지 않도록 제외
            # 큐잉 완료(워커 반영 대기) 행도 제외 — 재보정·상위 20건 점유 방지
            TradeHistory.id.notin_((_exhausted_ids() + sorted(_queued_ids)) or [-1]),
        )
        .order_by(TradeHistory.created_at.asc())
        .limit(20)
    )).scalars().all()
    _last_stats["candidates"] = len(rows)
    _last_stats["exhausted"] = len(_exhausted_ids())
    if _last_stats["exhausted"] > int(_last_stats.get("exhausted_warned", 0)):
        # 소진 건수가 늘 때만 경보(2분 주기 로그 반복 방지)
        _last_stats["exhausted_warned"] = _last_stats["exhausted"]
        logger.warning(
            f"[FillSweep] 체결 미확인 시도 소진 {_last_stats['exhausted']}건(당일) — "
            f"executed_price/slippage 수동 확인 필요: ids={sorted(_exhausted_ids())[:10]}"
        )
    if not rows:
        return 0

    mono = time.monotonic()
    lookups = 0
    queued = 0
    for row in rows:
        if lookups >= FILL_SWEEP_MAX_LOOKUPS_PER_CYCLE:
            break
        if row.id in _queued_ids or not _due(row.id, mono):
            continue
        if _auction_fill_pending_v2(getattr(row, "created_at", None), now):
            continue  # 종가 동시호가 주문 — 15:30 체결 전 조회는 시도만 소진한다(장 마감 후 창에서 확인)
        if not _is_domestic(row.stock_code):
            continue  # 국내 체결조회 전용(해외는 currency_code 가 비어 있는 행이 있어 코드로도 거른다)
        lookups += 1
        n = _note_attempt(row.id, mono)
        info = await korea_investment_api.get_order_execution(str(row.order_id), True)
        filled_qty = int((info or {}).get("filled_qty", 0) or 0)
        filled_price = float((info or {}).get("filled_price", 0) or 0)
        remaining = (info or {}).get("remaining_qty")
        ordered = int(row.order_quantity or 0)

        if not (info or {}).get("ok"):
            logger.warning(
                f"[FillSweep] {row.stock_code} #{row.id} 체결조회 실패 ({n}/{FILL_SWEEP_MAX_ATTEMPTS}) "
                f"order_id={row.order_id}"
            )
            continue
        if (
            filled_qty <= 0 and _is_swing_buy_row_v2(row) and info.get("found")
            and (remaining == 0 or info.get("cancelled"))
        ):
            # [하드닝 R9] 0체결 + 잔량 소멸 확정 = 매수 불성립 → CANCELLED + ActiveTrade close.
            from app.services.trade_executor import (
                _mark_trade_cancelled_v2, _rollback_buy_active_trade_v2,
            )
            _queued_ids.add(row.id)  # 먼저 표시 — 이후 예외여도 당일 재처리 안 함(멱등)
            _attempts.pop(row.id, None)
            await _mark_trade_cancelled_v2(str(row.order_id), row.id, filled_qty=0)
            _rb = await _resolve_buy_rollback_v2(row, db, 0)
            _res = await _rollback_buy_active_trade_v2(str(row.order_id), _rb, 0)
            queued += 1
            _last_stats["queued"] = int(_last_stats.get("queued", 0)) + 1
            logger.warning(
                f"[FillSweep] {row.stock_code} #{row.id} BUY 0체결 소멸 확정 — CANCELLED 강등, "
                f"ActiveTrade 원복={_res} (order_id={row.order_id})"
            )
            continue
        if filled_qty <= 0 or filled_price <= 0:
            logger.warning(
                f"[FillSweep] {row.stock_code} #{row.id} {row.order_type} 체결 미확인 "
                f"(체결 {filled_qty}주, 주문행발견={info.get('found')}, 잔량={remaining}) "
                f"({n}/{FILL_SWEEP_MAX_ATTEMPTS}) — 갱신 안 함, 소진 시 수동 확인"
            )
            continue
        if filled_qty < ordered and remaining != 0:
            logger.warning(
                f"[FillSweep] {row.stock_code} #{row.id} 부분체결 {filled_qty}/{ordered}주 미종결 "
                f"(잔량={remaining}) — 종결 후 재시도"
            )
            continue
        filled_qty = min(filled_qty, ordered) if ordered > 0 else filled_qty

        ref = float(row.order_price or 0) or (float(row.total_amount or 0) / ordered if ordered else 0.0)
        slippage = ((filled_price - ref) / ref * 100) if ref > 0 else 0.0
        basis = None
        if row.order_type == "SELL":
            entry = derive_sell_entry_price_v2(row)
            if entry is not None:
                basis = {
                    "entry_price": entry,
                    "market": "KOSPI",
                    "is_domestic": True,
                    "net_rate": False,
                }
            else:
                logger.warning(
                    f"[FillSweep] {row.stock_code} #{row.id} SELL 진입가 역산 불가 — 체결가·금액만 갱신"
                )
        fields = _fill_update_fields_v2(row.order_type, filled_qty, filled_price, slippage, basis)
        fields.pop("executed_at", None)  # 실제 체결 시각을 모른다 — 기존 값 유지
        # 큐잉 전에 표시 — 이후 단계가 예외로 빠져도 같은 행을 당일 다시 보정하지 않는다
        _queued_ids.add(row.id)
        _attempts.pop(row.id, None)
        await enqueue_update_trade_history(row.id, fields)
        if _is_swing_buy_row_v2(row) and ordered > 0 and filled_qty < ordered:
            # [하드닝 R9] 부분체결 확정 → ActiveTrade.quantity 를 체결수량으로(절대값 — 멱등).
            # 진입가 보정보다 먼저 큐잉해야 보정 분모(행 수량)가 체결수량이 된다(직후 확인과 같은 순서).
            from app.services.trade_executor import _rollback_buy_active_trade_v2
            _rb = await _resolve_buy_rollback_v2(row, db, filled_qty)
            await _rollback_buy_active_trade_v2(str(row.order_id), _rb, filled_qty)
        elif row.order_type == "BUY":
            # 전량 체결 확정 — 남은 원복 정보 폐기(레지스트리 정리)
            from app.services.trade_executor import _pop_buy_entry_rollback_v2
            _pop_buy_entry_rollback_v2(row.order_id, row.stock_code)
        if row.order_type == "BUY":
            await _enqueue_buy_entry_adjust_v2(row, filled_qty, filled_price, db)
        queued += 1
        _last_stats["queued"] = int(_last_stats.get("queued", 0)) + 1
        logger.info(
            f"[FillSweep] {row.stock_code} #{row.id} {row.order_type} 체결 확인 — "
            f"{filled_qty}주 @ {filled_price:,.2f} (기준가 {ref:,.2f}, 슬리피지 {slippage:+.3f}%)"
        )
    return queued