kis_bot/kis_trader/engine/limit_entry_common.py

"""
kis_trader/engine/limit_entry_common.py — ATR 지정가 진입(C안) 공통

꼬리(SHORT)·하락매수(UPDOW): 신호봉 확정 후 지정가 1회 → 유효 N봉 내 low 터치 시 체결,
미체결 시 취소(실매) / 백테 스킵.
"""
from __future__ import annotations

from typing import Any, Dict, List, Optional, Tuple

from ..utils.env import get_env_float, get_env_from_db, get_env_int

ENTRY_ALIGN = "align"
ENTRY_LIMIT_ATR = "limit_atr"


def short_entry_mode(params: Optional[Dict[str, Any]] = None) -> str:
    if params is not None and params.get("entry_mode") is not None:
        return str(params.get("entry_mode") or ENTRY_ALIGN).strip().lower()
    return str(
        get_env_from_db("TAIL_ENTRY_MODE", ENTRY_LIMIT_ATR) or ENTRY_LIMIT_ATR,
    ).strip().lower()


def updow_entry_mode(cfg: Optional[Dict[str, Any]] = None) -> str:
    if cfg is not None and cfg.get("entry_mode") is not None:
        return str(cfg.get("entry_mode") or ENTRY_ALIGN).strip().lower()
    return str(
        get_env_from_db("UPDOW_ENTRY_MODE", ENTRY_LIMIT_ATR) or ENTRY_LIMIT_ATR,
    ).strip().lower()


def is_limit_atr_entry(mode: str) -> bool:
    return str(mode or "").strip().lower() in (ENTRY_LIMIT_ATR, "limit", "atr_limit")


def _limit_mult(params: Optional[Dict[str, Any]], prefix: str) -> float:
    key = f"{prefix}_LIMIT_ATR_MULT"
    if params is not None and params.get("limit_atr_mult") is not None:
        # 0.0(저점 그대로 체결)도 유효값 — `or 1.5` 로 덮지 않도록 직접 float 변환
        raw = params.get("limit_atr_mult")
        if str(raw).strip() != "":
            try:
                return float(raw)
            except (TypeError, ValueError):
                pass
    return get_env_float(key, get_env_float("LIMIT_ATR_MULT_DEFAULT", 1.5))


def _limit_anchor(params: Optional[Dict[str, Any]], prefix: str) -> str:
    key = f"{prefix}_LIMIT_ANCHOR"
    if params is not None and params.get("limit_anchor") is not None:
        return str(params.get("limit_anchor") or "signal_low").strip().lower()
    return str(get_env_from_db(key, "signal_low") or "signal_low").strip().lower()


def _limit_valid_bars(params: Optional[Dict[str, Any]], prefix: str) -> int:
    key = f"{prefix}_LIMIT_VALID_BARS"
    if params is not None and params.get("limit_valid_bars") is not None:
        return max(1, int(params.get("limit_valid_bars") or 1))
    return max(1, get_env_int(key, 1))


def _limit_fill_slip_pct(params: Optional[Dict[str, Any]], prefix: str) -> float:
    key = f"{prefix}_LIMIT_FILL_SLIP_PCT"
    if params is not None and params.get("limit_fill_slip_pct") is not None:
        return float(params.get("limit_fill_slip_pct") or 0.0)
    return get_env_float(key, 0.0)


def tail_limit_params(params: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "mult": _limit_mult(params, "TAIL"),
        "anchor": _limit_anchor(params, "TAIL"),
        "valid_bars": _limit_valid_bars(params, "TAIL"),
        "fill_slip_pct": _limit_fill_slip_pct(params, "TAIL"),
    }


def updow_limit_params(cfg: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "mult": _limit_mult(cfg, "UPDOW"),
        "anchor": _limit_anchor(cfg, "UPDOW"),
        "valid_bars": _limit_valid_bars(cfg, "UPDOW"),
        "fill_slip_pct": _limit_fill_slip_pct(cfg, "UPDOW"),
    }


def floor_limit_price_krw(price: float) -> int:
    """지정가(원) — 정수 호가."""
    if price <= 0:
        return 0
    return max(1, int(price))


def resolve_limit_anchor_price(
    anchor: str,
    sig_bar: Dict[str, Any],
    candles: List[Dict[str, Any]],
    sig_i: int,
) -> float:
    """신호봉 기준 anchor 가격."""
    mode = (anchor or "signal_low").strip().lower()
    if mode == "signal_close":
        return float(sig_bar.get("close") or 0)
    if mode == "prev_close" and sig_i > 0:
        return float(candles[sig_i - 1].get("close") or 0)
    lo = float(sig_bar.get("low") or 0)
    if lo > 0:
        return lo
    return float(sig_bar.get("close") or 0)


def compute_atr_limit_price(
    anchor_px: float,
    atr: Optional[float],
    mult: float,
    *,
    min_price: float = 0.0,
) -> float:
    """anchor − ATR×mult 지정가 (매수 대기)."""
    if anchor_px <= 0:
        return 0.0
    a = float(atr or 0)
    if a <= 0:
        a = anchor_px * 0.01
    m = float(mult) if float(mult) > 0 else 1.5
    lp = anchor_px - a * m
    if min_price > 0 and lp < min_price:
        return 0.0
    if lp <= 0:
        return 0.0
    return lp


def limit_valid_until_bar_key(candles: List[Dict], sig_i: int, valid_bars: int) -> str:
    """체결 허용 **마지막 봉** 의 candle_time (이 봉까지 low≤지정가면 체결)."""
    j = sig_i + max(1, int(valid_bars))
    if j >= len(candles):
        j = len(candles) - 1
    return str(candles[j].get("candle_time") or "")[:12]


def limit_cancel_after_bar_key(valid_until_key: str) -> str:
    """
    이 키 **초과** 봉이 나오면 미체결 취소.

    valid_until=신호+1봉(3분 1개) → 그 다음 봉 시각부터 취소.
    """
    return str(valid_until_key or "")[:12]


def should_cancel_unfilled_limit(
    latest_bar_key: str,
    valid_until_key: str,
) -> bool:
    """최신 확정봉 시각이 유효 마지막 봉보다 크면 → 유효기간 종료, 취소."""
    lb = str(latest_bar_key or "")[:12]
    vu = str(valid_until_key or "")[:12]
    if not lb or not vu:
        return False
    return lb > vu


def try_limit_fill_on_bar(
    bar: Dict[str, Any],
    limit_price: float,
    fill_slip_pct: float = 0.0,
) -> Optional[float]:
    """백테: 해당 봉 low가 지정가 이하면 체결 (슬리피지는 불리하게만)."""
    if limit_price <= 0:
        return None
    lo = float(bar.get("low") or bar.get("close") or 0)
    if lo <= 0 or lo > limit_price:
        return None
    slip = float(fill_slip_pct or 0.0)
    if slip > 0:
        return limit_price * (1.0 + slip / 100.0)
    return limit_price


def merge_limit_into_signal(
    sig: Dict[str, Any],
    *,
    limit_price: float,
    valid_until_key: str,
    signal_bar_key: str,
) -> Dict[str, Any]:
    out = dict(sig)
    out["entry_mode"] = ENTRY_LIMIT_ATR
    out["limit_price"] = limit_price
    out["use_limit_buy"] = True
    out["valid_until_bar_key"] = valid_until_key
    out["signal_bar_key"] = signal_bar_key
    out["entry_price"] = limit_price
    return out