kis_bot/kis_trader/utils/live_portfolio_common.py

"""
실매 ↔ 포트폴리오 백테 정합 — 유니버스 슬롯·총한도·매수금액.
백테 ``breakout_portfolio_backtest`` / ``backtest_portfolio_common`` 과 동일 분모.
"""
from __future__ import annotations

import time
from datetime import datetime as dt
from typing import Any, Dict, List, Optional, Tuple

from ..backtest.backtest_portfolio_common import (
    STRATEGY_PORTFOLIO_KEYS,
    min_invest_ratio_of_slot,
    portfolio_exposure_krw,
    resolve_portfolio_params,
    target_qty_and_cost,
)
from ..utils.env import get_env_from_db, get_env_int
from ..utils.position_sizing import invest_qty_for_price

# 총한도·일1회 base 가드 스킵 (전략 자체 구현)
_PORTFOLIO_BUDGET_SKIP_DEFAULT = frozenset({"UPDOW"})


def _prof_add(prof: Optional[Dict[str, float]], key: str, t0: float) -> None:
    """LOOP_PROF 세분용 — 매매 로직 불변, ms 누적만."""
    if prof is None:
        return
    try:
        prof[key] = float(prof.get(key, 0.0) or 0.0) + (time.perf_counter() - t0) * 1000.0
    except Exception:
        pass


def portfolio_strategy_key(strategy_id: str) -> str:
    sid = (strategy_id or "").upper()
    return "TAIL" if sid == "SHORT" else sid


def _env_flag(key: str, default: bool) -> bool:
    raw = get_env_from_db(key, "")
    if raw in (None, "", "None"):
        return default
    return str(raw).strip().lower() in ("1", "true", "t", "y", "yes", "on")


def resolve_live_universe_history_source(
    strategy_id: str,
    *,
    universe_source: Optional[str] = None,
) -> str:
    """백테/조회가 읽을 이력 테이블.

    - ``ls_condition`` 유니버스 → ``ls_candidates_history``
    - 그 외 → ``target_candidates_history`` (키움/KIS)
    """
    sid = (strategy_id or "").upper()
    # 전략·전역 명시 오버라이드
    for key in (
        f"{sid}_LIVE_UNIVERSE_HISTORY_SOURCE",
        "LIVE_UNIVERSE_HISTORY_SOURCE",
    ):
        raw = get_env_from_db(key, "")
        if raw not in (None, "", "None"):
            s = str(raw).strip().lower()
            if s in ("ls", "ls_condition", "ls_afr"):
                return "ls"
            if s in ("kiwoom", "target", "kis", "condition"):
                return "kiwoom"

    src = str(universe_source or "").strip().lower()
    if not src:
        src = str(
            get_env_from_db(f"{sid}_UNIVERSE_SOURCE", "") or ""
        ).strip().lower()
    if src in ("ls_condition", "ls", "ls_afr"):
        return "ls"
    return "kiwoom"


def live_portfolio_budget_align_enabled(strategy_id: str) -> bool:
    sid = (strategy_id or "").upper()
    per_key = f"{sid}_LIVE_PORTFOLIO_BUDGET_ALIGN"
    if get_env_from_db(per_key, "") not in (None, "", "None"):
        return _env_flag(per_key, True)
    if sid in _PORTFOLIO_BUDGET_SKIP_DEFAULT:
        return False
    return _env_flag("LIVE_PORTFOLIO_BUDGET_ALIGN", True)


def resolve_live_scan_interval_min(strategy_id: str) -> int:
    sid = (strategy_id or "").upper()
    v = get_env_int(f"{sid}_SCAN_INTERVAL_MIN", 0)
    if v > 0:
        return max(1, int(v))
    return max(1, get_env_int("SCAN_INTERVAL_MIN", 1))


def slot_key_from_dt(when: dt, scan_interval_min: int = 1) -> str:
    """백테 ``_bt_slot_key`` / ``_slot_key`` 와 동일."""
    iv = max(1, int(scan_interval_min))
    date = when.strftime("%Y%m%d")
    hm = when.hour * 100 + when.minute
    total_min = (hm // 100) * 60 + (hm % 100)
    slot_min = (total_min // iv) * iv
    slot_hm = (slot_min // 60) * 100 + (slot_min % 60)
    return date + str(slot_hm).zfill(4)


def resolve_live_total_budget_krw(
    strategy_id: str,
    max_stocks: int,
    slot_money: float,
) -> float:
    pkey = portfolio_strategy_key(strategy_id)
    keys = STRATEGY_PORTFOLIO_KEYS.get(pkey, STRATEGY_PORTFOLIO_KEYS["SCALP"])
    env_row: Dict[str, Any] = {}
    for k in keys.get("total_budget", ()):
        v = get_env_int(k, 0)
        if v > 0:
            env_row[k] = v
            break
    for k in keys.get("slot", ()):
        v = get_env_int(k, 0)
        if v > 0:
            env_row[k] = v
    for k in keys.get("max_stocks", ()):
        v = get_env_int(k, 0)
        if v > 0:
            env_row[k] = v
    pf = resolve_portfolio_params(
        env_row if env_row else None,
        strategy=pkey,
        slot_money=float(slot_money),
        max_stocks=int(max_stocks),
    )
    return float(pf["total_budget_krw"])


def resolve_live_min_invest_ratio(strategy_id: str) -> float:
    return min_invest_ratio_of_slot({}, strategy=portfolio_strategy_key(strategy_id))


def live_max_daily(strategy_id: str) -> int:
    sid = (strategy_id or "").upper()
    v = get_env_int(f"{sid}_MAX_DAILY", 0)
    if v > 0:
        return int(v)
    if sid == "SHORT":
        return get_env_int("TAIL_MAX_DAILY", 0)
    return 0


def live_daily_entry_count(
    db: Any,
    code: str,
    today: str,
    strategy_id: str,
    *,
    prof: Optional[Dict[str, float]] = None,
) -> int:
    try:
        from .today_trades_cache import get_today_trades_cached

        t0 = time.perf_counter()
        trades, from_cache, db_ms = get_today_trades_cached(db, today)
        if from_cache:
            _prof_add(prof, "guard_trades_cache_ms", t0)
            if prof is not None:
                prof["guard_trades_hit"] = float(prof.get("guard_trades_hit", 0) or 0) + 1.0
        else:
            if prof is not None:
                prof["guard_trades_db_ms"] = float(prof.get("guard_trades_db_ms", 0) or 0) + float(db_ms or 0)
                prof["guard_trades_n"] = float(prof.get("guard_trades_n", 0) or 0) + 1.0
        sid = (strategy_id or "").upper()
        return len([
            t for t in trades
            if t.get("code") == code
            and str(t.get("strategy", "")).upper().startswith(sid)
        ])
    except Exception:
        return 0


def resolve_live_buy_qty(
    curr_price: float,
    holdings: Dict[str, Dict],
    strategy_id: str,
    slot_money: float,
    *,
    max_stocks: int,
    invest_cap: Optional[float] = None,
) -> Tuple[int, float, Optional[str]]:
    """
    포트폴리오 백테 ``target_qty_and_cost`` + ``min_invest_ratio`` + ``total_budget`` 정합.
    Returns: (qty, invest_krw, reject_reason)
    """
    if curr_price <= 0:
        return 0, 0.0, "가격0"
    cap = float(invest_cap if invest_cap is not None else slot_money)
    if cap <= 0:
        return 0, 0.0, "투자캡0"
    exposure = portfolio_exposure_krw(holdings)
    total_budget = resolve_live_total_budget_krw(strategy_id, max_stocks, slot_money)
    remaining = max(0.0, total_budget - exposure)
    ratio = resolve_live_min_invest_ratio(strategy_id)
    target_qty, target_cost = target_qty_and_cost(curr_price, cap)
    min_required = target_cost * ratio
    if target_qty < 1 or remaining < min_required:
        return 0, 0.0, "소액/잔여부족"
    invest = min(cap, remaining, target_cost)
    qty = invest_qty_for_price(curr_price, invest)
    if qty < 1:
        return 0, 0.0, "수량0"
    cost = qty * curr_price
    if cost < min_required:
        return 0, 0.0, "소액"
    return qty, invest, None


def live_portfolio_entry_reject(
    db: Any,
    holdings: Dict[str, Dict],
    strategy_id: str,
    code: str,
    slot_money: float,
    max_stocks: int,
    *,
    prof: Optional[Dict[str, float]] = None,
) -> Optional[str]:
    """일일한도·총한도 — ``None`` 이면 통과."""
    t0 = time.perf_counter()
    enabled = live_portfolio_budget_align_enabled(strategy_id)
    _prof_add(prof, "guard_env_ms", t0)
    if not enabled:
        return None
    t1 = time.perf_counter()
    md = live_max_daily(strategy_id)
    _prof_add(prof, "guard_env_ms", t1)
    if md > 0:
        today = dt.now().strftime("%Y%m%d")
        t2 = time.perf_counter()
        cnt = live_daily_entry_count(db, code, today, strategy_id, prof=prof)
        _prof_add(prof, "guard_daily_ms", t2)
        if cnt >= md:
            return f"탈락-일일한도 daily={cnt}/{md}"
    t3 = time.perf_counter()
    total = resolve_live_total_budget_krw(strategy_id, max_stocks, slot_money)
    if total > 0 and portfolio_exposure_krw(holdings) >= total - 1e-6:
        _prof_add(prof, "guard_budget_ms", t3)
        return "탈락-총한도"
    _prof_add(prof, "guard_budget_ms", t3)
    return None


def live_portfolio_budget_full(
    holdings: Dict[str, Dict],
    strategy_id: str,
    slot_money: float,
    max_stocks: int,
) -> bool:
    if not live_portfolio_budget_align_enabled(strategy_id):
        return False
    total = resolve_live_total_budget_krw(strategy_id, max_stocks, slot_money)
    if total <= 0:
        return False
    return portfolio_exposure_krw(holdings) >= total - 1e-6