kis_bot/kis_trader/execution/orphan_reconcile.py

"""
kis_trader/execution/orphan_reconcile.py — 봇 고아 포지션 복구 (장마감 전·후 배치)
================================================================================
실계좌 잔고(inquire-balance, 페이징)와 active_trades·orders 를 대조해
「봇이 매수했는데 active_trades 미기록」 고아만 복구한다.
이어서 GHOST_PURGE_ON_RECONCILE 시 「DB有·브로커0」 유령 행을 삭제한다.

- 기동: 전략 쓰레드 전 1회 (ORPHAN_RECONCILE_ON_STARTUP, 기본 ON)
- Pre-EOD: 활성 전략 중 가장 이른 EOD 시각 − N분 (기본 7분) 1회
- Post-EOD: 15:36~ 장마감 후 1회 (기존)
- 수동매수(orders BUY 없음) · MANUAL_HOLD_CODES · strategy=HOLDING 은 건드리지 않음.
- 매매 알림은 보내지 않음(로그 + 선택적 요약 MM).
"""
from __future__ import annotations

import datetime
import json
import logging
from typing import Any, Dict, List, Optional, Set, Tuple

from ..engine.strategy_eod import _STRATEGY_EOD_SPEC, parse_eod_hm, resolve_strategy_eod_params
from ..utils.env import get_env_bool, get_env_float, get_env_from_db, get_env_int
from ..utils.strategy_ids import canonical_strategy_id
from .kis_client import (
    broker_row_hldg_qty,
    broker_row_sellable_qty,
    cancelable_remainder_qty,
)

logger = logging.getLogger("kis_trader.orphan_reconcile")


def _current_is_mock_flag() -> int:
    """현재 KIS_MOCK → 1=모의 · 0=실전. 브로커 대조는 이 모드 행만."""
    from database import TradeDB
    return int(TradeDB.resolve_kis_is_mock(None))


def _row_matches_current_account(is_mock_val: Any, current: Optional[int] = None) -> bool:
    """active_trades.is_mock 이 현재 매매 모드와 같을 때만 브로커 대조 대상.

    NULL(미상)은 실전/모의 어느 쪽 잔고와도 맞춰 지우지 않음(오삭제 방지).
    """
    cur = _current_is_mock_flag() if current is None else int(current)
    if is_mock_val is None:
        return False
    try:
        return int(is_mock_val) == cur
    except (TypeError, ValueError):
        return False


# Pre-EOD 시각 산출 — (전략 ON 플래그, strategy_eod spec 키)
_PRE_EOD_STRATEGY_FLAGS: Tuple[Tuple[str, str, bool], ...] = (
    ("STRATEGY_BREAKOUT_ENABLED", "BREAKOUT", False),
    ("STRATEGY_MOMENTUM_ENABLED", "MOMENTUM", False),
    ("STRATEGY_SHORT_ENABLED", "TAIL", True),
)


def resolve_pre_eod_reconcile_hm() -> Tuple[int, int]:
    """
    활성 전략 EOD 시각 중 가장 이른 시각 − ORPHAN_RECONCILE_PRE_EOD_LEAD_MIN 분.
    활성 EOD 없으면 BREAKOUT 기본 15:15 기준.
    """
    lead = max(1, get_env_int("ORPHAN_RECONCILE_PRE_EOD_LEAD_MIN", 7))
    candidates: List[int] = []

    for flag, sid, def_on in _PRE_EOD_STRATEGY_FLAGS:
        if not get_env_bool(flag, def_on):
            continue
        spec = _STRATEGY_EOD_SPEC.get(sid)
        if spec is None:
            continue
        en_key, hm_key, def_en, def_hm, leg_key = spec
        params: Dict[str, Any] = {
            en_key: get_env_from_db(en_key, str(def_en)),
            hm_key: get_env_from_db(hm_key, def_hm),
        }
        if leg_key:
            params[leg_key] = get_env_from_db(leg_key, str(def_en))
        enabled, eod_hm = resolve_strategy_eod_params(params, sid)
        if not enabled:
            continue
        hh, mm = parse_eod_hm(eod_hm, def_hm)
        candidates.append(hh * 60 + mm)

    if not candidates:
        base_min = 15 * 60 + 15
    else:
        base_min = min(candidates)

    pre_min = max(9 * 60, base_min - lead)
    return pre_min // 60, pre_min % 60


def is_pre_eod_reconcile_window(
    now: Optional[datetime.datetime] = None,
    *,
    window_min: int = 3,
) -> bool:
    """Pre-EOD 고아복구 실행 윈도우 (기본 3분)."""
    if not get_env_bool("ORPHAN_RECONCILE_PRE_EOD_ENABLED", True):
        return False
    if not get_env_bool("ORPHAN_RECONCILE_ENABLED", True):
        return False
    t = now or datetime.datetime.now()
    pre_h, pre_m = resolve_pre_eod_reconcile_hm()
    cur = t.hour * 60 + t.minute
    start = pre_h * 60 + pre_m
    return start <= cur < start + max(1, window_min)


def parse_manual_hold_codes() -> Set[str]:
    """MANUAL_HOLD_CODES env — 쉼표/세미콜론 구분 종목코드."""
    manual_raw = str(get_env_from_db("MANUAL_HOLD_CODES", "") or "")
    return {
        c.strip() for c in manual_raw.replace(";", ",").split(",") if c.strip()
    }


def get_bot_bought_codes(db, *, account_mode: str = "current") -> Set[str]:
    """orders 테이블에 BUY 기록이 있는 종목코드.

    account_mode: current(현재 KIS_MOCK) | mock | live | all
    """
    codes: Set[str] = set()
    try:
        from database import TradeDB
        mode = (account_mode or "current").strip().lower()
        if mode in ("all", "both", "*"):
            cur = db.conn.execute("SELECT DISTINCT code FROM orders WHERE side='BUY'")
        else:
            if mode in ("mock", "1", "true", "모의"):
                flag = 1
            elif mode in ("live", "real", "0", "false", "실전"):
                flag = 0
            else:
                flag = int(TradeDB.resolve_kis_is_mock(None))
            cur = db.conn.execute(
                "SELECT DISTINCT code FROM orders WHERE side='BUY' AND is_mock=%s",
                (flag,),
            )
        for row in (cur.fetchall() or []):
            c = str(row.get("code") or "").strip()
            if c:
                codes.add(c)
    except Exception as e:
        logger.warning("orders BUY 코드 조회 실패: %s", e)
    return codes


def get_portfolio_origin_sets(db, account_mode: str = "current") -> Tuple[Set[str], Set[str]]:
    """(bot_bought_codes, manual_hold_codes) — backtest_web 보유·매도 탭과 동일."""
    return get_bot_bought_codes(db, account_mode=account_mode), parse_manual_hold_codes()


def _tracked_codes(db) -> Set[str]:
    """active_trades 에 한 건이라도 있는 종목코드 — 현재 KIS_MOCK 행만."""
    out: Set[str] = set()
    cur_mock = _current_is_mock_flag()
    try:
        cur = db.conn.execute(
            "SELECT DISTINCT code FROM active_trades WHERE is_mock=%s",
            (cur_mock,),
        )
        for row in (cur.fetchall() or []):
            c = str(row.get("code") or "").strip()
            if c:
                out.add(c)
    except Exception as e:
        logger.warning("active_trades 코드 조회 실패: %s", e)
    return out


def _latest_buy_order(db, code: str) -> Optional[Dict[str, Any]]:
    """해당 종목 최신 BUY orders 행 (복구 메타·체결가 참고). 현재 KIS_MOCK만."""
    try:
        from database import TradeDB
        mock_flag = int(TradeDB.resolve_kis_is_mock(None))
        cur = db.conn.execute(
            """
            SELECT * FROM orders
             WHERE side='BUY' AND code=%s AND is_mock=%s
             ORDER BY submitted_at DESC
             LIMIT 1
            """,
            (code, mock_flag),
        )
        row = cur.fetchone()
        return dict(row) if row else None
    except Exception as e:
        logger.warning("orders 최신 BUY 조회 실패 %s: %s", code, e)
        return None


def _parse_order_meta(row: Dict[str, Any]) -> Dict[str, Any]:
    """orders.raw_json 에서 stop/target 등 복구."""
    meta: Dict[str, Any] = {}
    raw = row.get("raw_json")
    if not raw:
        return meta
    try:
        d = json.loads(raw) if isinstance(raw, str) else raw
        if isinstance(d, dict):
            meta = d
    except Exception:
        pass
    return meta


def _num(v, default: float = 0.0) -> float:
    try:
        return float(str(v or 0).replace(",", "").strip())
    except Exception:
        return default


def _broker_qty(broker: Dict[str, Any], code: str) -> int:
    """보유수량. 없으면 매도가능(구맵). 매도가능 0이어도 보유>0 이면 0이 아님."""
    br = (broker or {}).get(code) or (broker or {}).get(str(code).strip())
    if not br:
        return 0
    hldg = broker_row_hldg_qty(br)
    if hldg > 0:
        return hldg
    return broker_row_sellable_qty(br)


def _orphan_skip_sell_reasons() -> Optional[List[str]]:
    """
    재등록 금지 sell_reason 키워드 목록.
    env 비우면 None → 매수 이후 **모든** 청산 히스토리를 재등록 금지로 본다.
    기본: 장마감청산·eod (모의잔고 지연으로 EOD 후 고아복구가 유령 재등록하던 케이스).
    """
    raw = str(
        get_env_from_db(
            "ORPHAN_SKIP_SELL_REASONS",
            "장마감청산,eod",
        )
        or ""
    ).strip()
    if not raw:
        return None
    return [x.strip() for x in raw.replace(";", ",").split(",") if x.strip()]


def _find_sell_after_buy(
    db,
    *,
    code: str,
    strategy: str,
    buy_date: Optional[str],
    is_mock: Optional[int] = None,
) -> Optional[Dict[str, Any]]:
    """
    같은 code+strategy(+is_mock) 에서 buy_date(없으면 당일) 이후 청산 히스토리 1건.
    ORPHAN_SKIP_SELL_REASONS 가 있으면 해당 사유만, 비면 모든 청산.
    """
    if not get_env_bool("ORPHAN_SKIP_AFTER_SELL_HISTORY", True):
        return None
    code = str(code or "").strip()
    sid = canonical_strategy_id(str(strategy or "").strip() or "SCALP")
    if not code:
        return None
    reasons = _orphan_skip_sell_reasons()
    buy_ts = str(buy_date or "").strip()
    day = datetime.datetime.now().strftime("%Y-%m-%d")
    mock_flag = _current_is_mock_flag() if is_mock is None else int(is_mock)
    try:
        if buy_ts:
            sql = (
                "SELECT code, strategy, sell_reason, sell_date, buy_date, qty, is_mock "
                "FROM trade_history "
                "WHERE code=%s AND strategy=%s AND is_mock=%s "
                "AND sell_date IS NOT NULL AND sell_date<>'' AND sell_date>=%s "
                "ORDER BY sell_date DESC LIMIT 20"
            )
            rows = db.conn.execute(sql, (code, sid, mock_flag, buy_ts)).fetchall() or []
        else:
            sql = (
                "SELECT code, strategy, sell_reason, sell_date, buy_date, qty, is_mock "
                "FROM trade_history "
                "WHERE code=%s AND strategy=%s AND is_mock=%s AND sell_date LIKE %s "
                "ORDER BY sell_date DESC LIMIT 20"
            )
            rows = db.conn.execute(sql, (code, sid, mock_flag, day + "%")).fetchall() or []
    except Exception as e:
        logger.warning("청산 히스토리 조회 실패 %s/%s: %s", sid, code, e)
        return None

    for row in rows:
        d = dict(row)
        reason = str(d.get("sell_reason") or "").strip()
        # 유령정리(0원) 이력은 재등록 금지·청산후삭제 트리거로 쓰지 않음
        if reason.lower().startswith("ghost_purge"):
            continue
        if reasons is None:
            return d
        reason_l = reason.lower()
        for kw in reasons:
            if kw.lower() in reason_l or kw in reason:
                return d
    return None


def purge_active_after_sell_history(db) -> Dict[str, Any]:
    """
    active_trades 행의 buy_date 이후에 같은 전략 청산 히스토리가 있으면 DB에서 제거.
    (EOD 청산 후 모의잔고 지연 → 고아복구가 다시 넣은 유령 정리)
    현재 KIS_MOCK 과 같은 is_mock 행만 대상.
    """
    out: Dict[str, Any] = {"purged": [], "failed": [], "skipped": []}
    if not get_env_bool("ORPHAN_PURGE_ACTIVE_AFTER_SELL_HISTORY", True):
        out["msg"] = "ORPHAN_PURGE_ACTIVE_AFTER_SELL_HISTORY=false"
        return out
    cur_mock = _current_is_mock_flag()
    try:
        rows = db.conn.execute(
            "SELECT code, name, strategy, current_qty, buy_date, is_mock "
            "FROM active_trades WHERE is_mock=%s",
            (cur_mock,),
        ).fetchall() or []
    except Exception as e:
        out["error"] = str(e)
        return out

    manual_hold = parse_manual_hold_codes()
    for row in rows:
        d = dict(row)
        code = str(d.get("code") or "").strip()
        strategy = str(d.get("strategy") or "").strip()
        name = str(d.get("name") or code)
        if not code:
            continue
        if code in manual_hold or strategy.upper() == "HOLDING":
            out["skipped"].append({"code": code, "strategy": strategy, "why": "protected"})
            continue
        hit = _find_sell_after_buy(
            db,
            code=code,
            strategy=strategy,
            buy_date=str(d.get("buy_date") or "") or None,
            is_mock=cur_mock,
        )
        if not hit:
            continue
        try:
            db.delete_active_trade(code=code, strategy=strategy or None)
            out["purged"].append({
                "code": code,
                "name": name,
                "strategy": strategy,
                "qty": int(d.get("current_qty") or 0),
                "sell_reason": hit.get("sell_reason"),
                "sell_date": hit.get("sell_date"),
            })
            logger.warning(
                "🧹 [청산후유령정리] [%s] %s %s — 히스토리 %s @%s → active 삭제",
                strategy,
                name,
                code,
                hit.get("sell_reason"),
                hit.get("sell_date"),
            )
        except Exception as e:
            out["failed"].append({"code": code, "strategy": strategy, "error": str(e)})
            logger.exception("청산후 유령 active 삭제 실패 %s", code)
    out["purged_count"] = len(out["purged"])
    return out


def purge_ghost_active_trades(order_mgr, broker: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    DB(active_trades) 有 · 브로커 0주 → 유령 행 삭제.

    - 현재 KIS_MOCK 과 같은 is_mock 행만 대상 (모의 보유를 실전 잔고로 지우지 않음).
    - MANUAL_HOLD_CODES · strategy=HOLDING(장기) 은 보호(삭제 안 함).
    - 잔고 API 실패 시 삭제하지 않음(오판 방지).
    """
    out: Dict[str, Any] = {
        "purged": [], "skipped_manual": [], "skipped_holding": [],
        "skipped_other_account": [], "failed": [],
    }
    if not get_env_bool("GHOST_PURGE_ON_RECONCILE", True):
        out["msg"] = "GHOST_PURGE_ON_RECONCILE=false"
        return out

    db = order_mgr.db
    cur_mock = _current_is_mock_flag()
    if broker is None:
        broker = order_mgr.get_broker_holdings(force=True)
        if not getattr(order_mgr, "_holdings_last_fetch_ok", False):
            out["error"] = "실계좌 잔고 조회 실패"
            return out

    cancelable = None
    if get_env_bool("INQUIRE_PSBL_RVSECNCL_BEFORE_GHOST", True):
        try:
            cancelable = order_mgr.get_cancelable_sells(force=True)
        except Exception as e:
            logger.warning("유령정리 전 정정취소가능 조회 예외: %s", e)
            cancelable = None
        if cancelable is None:
            out["error"] = "정정취소가능 조회 실패 — 유령삭제 스킵"
            logger.warning("🧹 [유령잔고] 정정취소가능조회 실패 → 이번 회차 삭제 안 함")
            return out

    manual_hold = parse_manual_hold_codes()
    try:
        rows = db.conn.execute(
            "SELECT code, name, strategy, current_qty, is_mock FROM active_trades"
        ).fetchall() or []
    except Exception as e:
        out["error"] = str(e)
        logger.warning("유령잔고 조회 실패: %s", e)
        return out

    for row in rows:
        d = dict(row)
        code = str(d.get("code") or "").strip()
        if not code:
            continue
        name = str(d.get("name") or code)
        strategy = str(d.get("strategy") or "").strip()
        if not _row_matches_current_account(d.get("is_mock"), cur_mock):
            out["skipped_other_account"].append({
                "code": code, "name": name, "strategy": strategy,
                "is_mock": d.get("is_mock"),
            })
            continue
        if code in manual_hold:
            out["skipped_manual"].append({"code": code, "name": name, "strategy": strategy})
            continue
        # 장기 홀딩봇 포지션은 당일 EOD 대상이 아님 — 브로커 0이어도 수동 확인 전 보존
        if strategy.upper() == "HOLDING":
            out["skipped_holding"].append({"code": code, "name": name})
            continue
        if _broker_qty(broker, code) > 0:
            continue
        if cancelable_remainder_qty(cancelable or {}, code) > 0:
            logger.info(
                "⏸ [유령스킵] [%s] %s %s — 정정취소가능 매도 잔량 있음",
                strategy, name, code,
            )
            continue
        try:
            pending = db.get_pending_sell_order(strategy, code)
        except Exception:
            pending = None
        if pending:
            logger.info(
                "⏸ [유령스킵] [%s] %s %s — DB 미체결 매도 ODNO=%s",
                strategy, name, code, pending.get("ord_no"),
            )
            continue
        try:
            # OrderManager 와 동일: trade_history 에 ghost_purge 남긴 뒤 정리
            recorded = False
            if hasattr(order_mgr, "_record_ghost_purge_history"):
                recorded = bool(
                    order_mgr._record_ghost_purge_history(
                        code,
                        strategy or "",
                        sell_reason="ghost_purge:broker_zero",
                    )
                )
            else:
                db.delete_active_trade(code=code, strategy=strategy or None)
            out["purged"].append({
                "code": code,
                "name": name,
                "strategy": strategy,
                "qty": int(d.get("current_qty") or 0),
                "history_recorded": recorded,
            })
            logger.warning(
                "🧹 [유령잔고삭제] [%s] %s %s — 브로커 0주 → 정리%s",
                strategy, name, code,
                " (trade_history 기록)" if recorded else "",
            )
        except Exception as e:
            out["failed"].append({"code": code, "name": name, "error": str(e)})
            logger.exception("유령잔고 삭제 실패 %s", code)

    out["purged_count"] = len(out["purged"])
    out["failed_count"] = len(out["failed"])
    out["skipped_other_account_count"] = len(out["skipped_other_account"])
    if out["skipped_other_account"]:
        logger.info(
            "🧹 [유령스킵] 다른 계좌모드(is_mock≠현재=%s) %d건 — 삭제 안 함",
            cur_mock, len(out["skipped_other_account"]),
        )
    if out["purged"]:
        try:
            order_mgr.invalidate_holdings_cache()
        except Exception:
            pass
    return out


def sync_active_trades_with_broker(order_mgr, broker: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    개장 전(08:35~09:15) 또는 고아복구 시 DB(active_trades)와 증권사 실계좌 잔고를 동기화.
    - 기업 이벤트(액면분할, 무상증자 등) 또는 수기 매입으로 계좌 매수단가/수량이 변경된 경우,
      DB의 avg_buy_price, current_qty, stop_price, target_price, max_price 를 비율에 맞춰 보정(Sync).
    - 오진입 손절 및 트레이링 스톱 착시 방지.
    """
    out: Dict[str, Any] = {"synced": [], "failed": []}
    if not get_env_bool("ORPHAN_RECONCILE_ENABLED", True):
        out["msg"] = "ORPHAN_RECONCILE_ENABLED=false"
        return out

    db = order_mgr.db
    if broker is None:
        broker = order_mgr.get_broker_holdings(force=True)
        if not getattr(order_mgr, "_holdings_last_fetch_ok", False):
            out["error"] = "실계좌 잔고 조회 실패 — 안전을 위해 동기화 스킵"
            logger.warning("🔄 [계좌동기화] 실계좌 잔고 조회 실패 — 안전을 위해 스킵")
            return out

    manual_hold = parse_manual_hold_codes()
    min_diff_px = get_env_float("SYNC_PRICE_DIFF_MIN", 0.5)
    cur_mock = _current_is_mock_flag()

    try:
        rows = db.conn.execute(
            "SELECT * FROM active_trades WHERE is_mock=%s",
            (cur_mock,),
        ).fetchall() or []
    except Exception as e:
        out["error"] = str(e)
        logger.warning("🔄 [계좌동기화] active_trades 조회 실패: %s", e)
        return out

    now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    for row in rows:
        d = dict(row)
        code = str(d.get("code") or "").strip()
        if not code:
            continue
        name = str(d.get("name") or code)
        strategy = str(d.get("strategy") or "").strip()

        # 수동 보호 및 장기 투자 포지션은 개입 차단
        if code in manual_hold or strategy.upper() == "HOLDING":
            continue

        br = broker.get(code)
        if not br:
            # 브로커에 0주인 종목은 purge_ghost_active_trades 가 처리하므로 여기선 스킵
            continue

        br_qty = int(br.get("qty") or 0)
        br_avg = float(br.get("avg_price") or 0.0)
        db_qty = int(d.get("current_qty") or 0)
        db_avg = float(d.get("avg_buy_price") or 0.0)

        if br_qty <= 0 or br_avg <= 0:
            continue

        qty_diff = (br_qty != db_qty)
        price_diff = abs(br_avg - db_avg) >= min_diff_px

        if not (qty_diff or price_diff):
            continue

        # 단가/수량 변경 감지! 비율(ratio) 계산하여 손절가·익절가·고점(max_price) 비례 조정
        ratio = (br_avg / db_avg) if (db_avg > 0) else 1.0
        old_stop = float(d.get("stop_price") or 0.0)
        old_target = float(d.get("target_price") or 0.0)
        old_max = float(d.get("max_price") or 0.0)

        new_stop = round(old_stop * ratio, 2) if old_stop > 0 else 0.0
        new_target = round(old_target * ratio, 2) if old_target > 0 else 0.0
        new_max = max(br_avg, round(old_max * ratio, 2)) if old_max > 0 else br_avg
        new_inv = br_avg * br_qty

        try:
            if strategy:
                db.conn.execute(
                    "UPDATE active_trades SET avg_buy_price=%s, current_qty=%s, target_qty=%s, "
                    "stop_price=%s, target_price=%s, max_price=%s, total_invested=%s, updated_at=%s "
                    "WHERE code=%s AND strategy=%s",
                    (br_avg, br_qty, br_qty, new_stop, new_target, new_max, new_inv, now_str, code, strategy)
                )
            else:
                db.conn.execute(
                    "UPDATE active_trades SET avg_buy_price=%s, current_qty=%s, target_qty=%s, "
                    "stop_price=%s, target_price=%s, max_price=%s, total_invested=%s, updated_at=%s "
                    "WHERE code=%s",
                    (br_avg, br_qty, br_qty, new_stop, new_target, new_max, new_inv, now_str, code)
                )
            
            msg = (
                f"🔄 [계좌 싱크 로봇] 원장 변동 동기화! {name}({code}) [{strategy}]\n"
                f"  · 수량: {db_qty}주 ➔ {br_qty}주\n"
                f"  · 매입단가: {db_avg:,.0f}원 ➔ {br_avg:,.0f}원 (비율 {ratio:.4f})\n"
                f"  · 손절가 보정: {old_stop:,.0f}원 ➔ {new_stop:,.0f}원\n"
                f"  · 고점(max) 보정: {old_max:,.0f}원 ➔ {new_max:,.0f}원"
            )
            logger.info(msg)
            out["synced"].append({
                "code": code, "name": name, "strategy": strategy,
                "old_qty": db_qty, "new_qty": br_qty,
                "old_avg": db_avg, "new_avg": br_avg,
            })
            # 알림 발송
            if hasattr(order_mgr, "notifier") and order_mgr.notifier and hasattr(order_mgr.notifier, "send_message"):
                try:
                    order_mgr.notifier.send_message(msg)
                except Exception:
                    pass
            elif hasattr(order_mgr, "send_telegram_msg"):
                try:
                    order_mgr.send_telegram_msg(msg)
                except Exception:
                    pass
        except Exception as e:
            out["failed"].append({"code": code, "name": name, "error": str(e)})
            logger.exception("🔄 [계좌동기화] UPDATE 실패 %s", code)

    if out["synced"]:
        try:
            order_mgr.invalidate_holdings_cache()
        except Exception:
            pass
        out["synced_count"] = len(out["synced"])
    else:
        out["synced_count"] = 0
    return out


def reconcile_orphan_positions(order_mgr) -> Dict[str, Any]:
    """
    장마감 후 1회 호출 — 봇 고아만 active_trades 에 upsert.
    이어서 GHOST_PURGE_ON_RECONCILE 시 유령(DB有·브로커0) 삭제.

    Returns:
        reconciled, skipped_tracked, skipped_manual, skipped_no_order,
        skipped_after_sell, failed, ghost_purged, ghost_purged_count,
        stale_purged, stale_purged_count
    """
    result: Dict[str, Any] = {
        "reconciled": [],
        "skipped_tracked": [],
        "skipped_manual": [],
        "skipped_no_order": [],
        "skipped_after_sell": [],
        "failed": [],
        "ghost_purged": [],
        "stale_purged": [],
        "synced": [],
    }

    if not get_env_bool("ORPHAN_RECONCILE_ENABLED", True):
        result["msg"] = "ORPHAN_RECONCILE_ENABLED=false"
        return result

    db = order_mgr.db
    broker = order_mgr.get_broker_holdings(force=True)
    if not getattr(order_mgr, "_holdings_last_fetch_ok", False):
        result["error"] = "실계좌 잔고 조회 실패"
        logger.warning("🧩 [고아복구] 잔고 API 실패 — 스킵")
        return result

    # EOD 청산 후 잘못 다시 들어간 active 유령부터 정리
    stale = purge_active_after_sell_history(db)
    result["stale_purged"] = list(stale.get("purged") or [])
    result["stale_purged_count"] = int(stale.get("purged_count") or 0)
    if stale.get("error"):
        result["stale_error"] = stale["error"]

    tracked = _tracked_codes(db)
    bot_bought, manual_hold = get_portfolio_origin_sets(db)

    for code, br in (broker or {}).items():
        c = str(code).strip()
        qty = int((br or {}).get("qty") or 0)
        if qty <= 0:
            continue
        name = str((br or {}).get("name") or c)

        if c in manual_hold:
            result["skipped_manual"].append({"code": c, "name": name, "qty": qty})
            continue
        if c in tracked:
            result["skipped_tracked"].append({"code": c, "name": name, "qty": qty})
            continue
        if c not in bot_bought:
            result["skipped_no_order"].append({"code": c, "name": name, "qty": qty})
            continue

        order_row = _latest_buy_order(db, c)
        if not order_row:
            result["skipped_no_order"].append({"code": c, "name": name, "qty": qty})
            continue

        meta = _parse_order_meta(order_row)
        strategy = canonical_strategy_id(
            str(order_row.get("strategy_id") or meta.get("strategy_id") or "SCALP")
        )
        buy_date = (
            str(order_row.get("filled_at") or order_row.get("submitted_at") or "").strip()
            or None
        )

        # 당일/매수 이후 장마감청산 등 히스토리 있으면 재등록 금지
        # (모의·잔고 API 지연으로 브로커 qty>0 이어도 EOD 후 유령 복구 방지)
        closed = _find_sell_after_buy(
            db, code=c, strategy=strategy, buy_date=buy_date,
            is_mock=_current_is_mock_flag(),
        )
        if closed:
            result["skipped_after_sell"].append({
                "code": c,
                "name": name,
                "qty": qty,
                "strategy": strategy,
                "sell_reason": closed.get("sell_reason"),
                "sell_date": closed.get("sell_date"),
            })
            logger.info(
                "⏭ [고아복구스킵] %s(%s) [%s] — 이미 청산(%s @%s), 재등록 안 함",
                name,
                c,
                strategy,
                closed.get("sell_reason"),
                closed.get("sell_date"),
            )
            continue

        filled_px = _num(order_row.get("filled_avg_price"))
        if filled_px <= 0:
            filled_px = _num(order_row.get("price") or meta.get("price_ref"))
        br_avg = _num((br or {}).get("avg_price"))
        avg_buy = br_avg if br_avg > 0 else filled_px
        cur_px = _num((br or {}).get("current_price")) or avg_buy
        stop_px = _num(meta.get("stop_price"))
        target_px = _num(meta.get("target_price"))
        atr_entry = _num(meta.get("atr_entry"))

        trade_data = {
            "code": c,
            "name": str(order_row.get("name") or meta.get("name") or name),
            "strategy": strategy,
            "avg_buy_price": avg_buy,
            "current_price": cur_px,
            "stop_price": stop_px,
            "target_price": target_px,
            "max_price": max(avg_buy, cur_px),
            "atr_entry": atr_entry,
            "target_qty": qty,
            "current_qty": qty,
            "total_invested": avg_buy * qty,
            "status": "HOLDING",
            "buy_date": buy_date,
            "size_class": meta.get("size_class") or "",
            "entry_features": meta.get("entry_features") or {},
            "is_mock": _current_is_mock_flag(),
        }

        try:
            ok = db.upsert_trade(trade_data)
            if ok:
                result["reconciled"].append({
                    "code": c,
                    "name": trade_data["name"],
                    "strategy": strategy,
                    "qty": qty,
                    "avg_buy": avg_buy,
                })
                logger.info(
                    "🧩 [고아복구] %s(%s) [%s] %d주 @ %.0f → active_trades",
                    trade_data["name"], c, strategy, qty, avg_buy,
                )
            else:
                result["failed"].append({"code": c, "name": name, "error": "upsert_trade 실패"})
        except Exception as e:
            result["failed"].append({"code": c, "name": name, "error": str(e)})
            logger.exception("고아복구 upsert 실패 %s", c)

    try:
        order_mgr.invalidate_holdings_cache()
    except Exception:
        pass

    # 실계좌 평단가·수량 동기화 (액면조정·수기매입 등 보정)
    sync = sync_active_trades_with_broker(order_mgr, broker=broker)
    result["synced"] = list(sync.get("synced") or [])
    result["synced_count"] = int(sync.get("synced_count") or 0)
    if sync.get("error"):
        result["sync_error"] = sync["error"]
    for f in sync.get("failed") or []:
        result["failed"].append(f)

    # 유령(DB有·브로커0) — 동일 잔고 스냅샷으로 삭제 (추가 REST 없음)
    ghost = purge_ghost_active_trades(order_mgr, broker=broker)
    result["ghost_purged"] = list(ghost.get("purged") or [])
    result["ghost_purged_count"] = int(ghost.get("ghost_purged_count") or ghost.get("purged_count") or 0)
    if ghost.get("error"):
        result["ghost_error"] = ghost["error"]
    for f in ghost.get("failed") or []:
        result["failed"].append(f)

    result["reconciled_count"] = len(result["reconciled"])
    result["failed_count"] = len(result["failed"])
    result["skipped_after_sell_count"] = len(result["skipped_after_sell"])
    logger.info(
        "🧩 [고아복구] 완료 — 복구 %d / 단가·수량동기화 %d / 청산후스킵 %d / 청산후유령정리 %d / "
        "유령삭제 %d / 실패 %d / 보호(수동) %d / 이미추적 %d / 주문없음 %d",
        len(result["reconciled"]),
        result["synced_count"],
        len(result["skipped_after_sell"]),
        result["stale_purged_count"],
        result["ghost_purged_count"],
        len(result["failed"]),
        len(result["skipped_manual"]),
        len(result["skipped_tracked"]),
        len(result["skipped_no_order"]),
    )
    return result