kis_bot/kis_trader/engine/dbband_env_keys.py

"""
kis_trader/engine/dbband_env_keys.py — 더블 볼린저(DBBAND) env 키 단일 정의
=============================================================================
캐시 리엔 더블 BB(20/2 + 20/3) + 추세 MA 필터 — config_dbband · dbband_engine · 웹 · 파라서치 공통.
"""
from __future__ import annotations

from typing import Any, Dict, Optional

# config_dbband 전용 키
DBBAND_CONFIG_KEYS = frozenset({
    "DBBAND_BB_PERIOD",
    "DBBAND_BB_INNER_STD",
    "DBBAND_BB_OUTER_STD",
    "DBBAND_TREND_MA_PERIOD",
    "DBBAND_USE_TREND_FILTER",
    "DBBAND_SIDE_MODE",
    "DBBAND_ENTRY_VALID_BARS",
    "DBBAND_ENTRY_MODE",
    "DBBAND_STOP_MODE",
    "DBBAND_STOP_BUFFER_PCT",
    "DBBAND_STOP_LOSS_PCT",
    "DBBAND_TP_MODE",
    "DBBAND_TAKE_PROFIT_PCT",
    "DBBAND_RR_RATIO",
    "DBBAND_EXIT_MODE",
    "DBBAND_SHOULDER_MIN_HIGH_PCT",
    "DBBAND_SHOULDER_CUT_PCT",
    "DBBAND_TRAIL_PCT",
    "DBBAND_TRAIL_ARM_PCT",
    "DBBAND_TIME_START",
    "DBBAND_TIME_END",
    "DBBAND_COOLDOWN_SEC",
    "DBBAND_MAX_DAILY",
    "DBBAND_MIN_PRICE",
    "DBBAND_SLOT_MONEY",
    "DBBAND_MAX_STOCKS",
    "DBBAND_TOTAL_BUDGET_KRW",
    "DBBAND_MAX_BUY_AMOUNT",
    "DBBAND_MAX_HOLD_BARS",
    "DBBAND_TIMEFRAME",
    "DBBAND_MIN_INVEST_RATIO_OF_SLOT",
    "DBBAND_FORCE_EOD_EXIT",
    "DBBAND_LIVE_MIN_CANDLES",
    "DBBAND_LIVE_SIGNAL_LOOKBACK_BARS",
    "DBBAND_UNIVERSE_SOURCE",
    "DBBAND_GRID_BB_PERIOD0",
    "DBBAND_GRID_BB_PERIOD1",
    "DBBAND_GRID_INNER_STD0",
    "DBBAND_GRID_INNER_STD1",
    "DBBAND_GRID_OUTER_STD0",
    "DBBAND_GRID_OUTER_STD1",
    "DBBAND_GRID_TREND_MA0",
    "DBBAND_GRID_TREND_MA1",
    "DBBAND_GRID_SL0",
    "DBBAND_GRID_SL1",
    "DBBAND_GRID_TP0",
    "DBBAND_GRID_TP1",
    "DBBAND_GRID_RR0",
    "DBBAND_GRID_RR1",
})


def _row_val(row: Dict[str, Any], key: str, default: Any = None) -> Any:
    v = row.get(key)
    if v not in (None, "", "None"):
        return v
    return default


def dbband_env_float(row: Dict[str, Any], key: str, default: float) -> float:
    v = _row_val(row, key)
    if v is None:
        return float(default)
    try:
        return float(v)
    except (TypeError, ValueError):
        return float(default)


def dbband_env_int(row: Dict[str, Any], key: str, default: int) -> int:
    v = _row_val(row, key)
    if v is None:
        return int(default)
    try:
        return int(float(v))
    except (TypeError, ValueError):
        return int(default)


def dbband_env_bool(row: Dict[str, Any], key: str, default: bool) -> bool:
    v = _row_val(row, key)
    if v is None:
        return default
    s = str(v).strip().lower()
    if s in ("1", "true", "t", "y", "yes", "on"):
        return True
    if s in ("0", "false", "f", "n", "no", "off", ""):
        return False
    return default


def params_to_dbband_env_patch(p: Dict[str, Any]) -> Dict[str, str]:
    """파라서치·웹 JSON params → DBBAND env 패치 (문자열)."""
    patch: Dict[str, str] = {}
    if not p:
        return patch

    def _set(k: str, v: Any) -> None:
        if v is not None and v != "":
            patch[k] = str(v)

    mapping = (
        ("bb_period", "DBBAND_BB_PERIOD"),
        ("bb_inner_std", "DBBAND_BB_INNER_STD"),
        ("bb_outer_std", "DBBAND_BB_OUTER_STD"),
        ("trend_ma_period", "DBBAND_TREND_MA_PERIOD"),
        ("use_trend_filter", "DBBAND_USE_TREND_FILTER"),
        ("side_mode", "DBBAND_SIDE_MODE"),
        ("entry_valid_bars", "DBBAND_ENTRY_VALID_BARS"),
        ("entry_mode", "DBBAND_ENTRY_MODE"),
        ("stop_mode", "DBBAND_STOP_MODE"),
        ("stop_buffer_pct", "DBBAND_STOP_BUFFER_PCT"),
        ("sl_pct", "DBBAND_STOP_LOSS_PCT"),
        ("tp_mode", "DBBAND_TP_MODE"),
        ("tp_pct", "DBBAND_TAKE_PROFIT_PCT"),
        ("rr_ratio", "DBBAND_RR_RATIO"),
        ("exit_mode", "DBBAND_EXIT_MODE"),
        ("shoulder_min_high", "DBBAND_SHOULDER_MIN_HIGH_PCT"),
        ("shoulder_cut_pct", "DBBAND_SHOULDER_CUT_PCT"),
        ("trail_pct", "DBBAND_TRAIL_PCT"),
        ("trail_arm_pct", "DBBAND_TRAIL_ARM_PCT"),
        ("time_start_hm", "DBBAND_TIME_START"),
        ("time_end_hm", "DBBAND_TIME_END"),
        ("max_daily", "DBBAND_MAX_DAILY"),
        ("min_price", "DBBAND_MIN_PRICE"),
        ("slot_money", "DBBAND_SLOT_MONEY"),
        ("max_stocks", "DBBAND_MAX_STOCKS"),
        ("total_budget_krw", "DBBAND_TOTAL_BUDGET_KRW"),
        ("max_hold_bars", "DBBAND_MAX_HOLD_BARS"),
        ("timeframe", "DBBAND_TIMEFRAME"),
        ("min_invest_ratio_of_slot", "DBBAND_MIN_INVEST_RATIO_OF_SLOT"),
    )
    for js_k, env_k in mapping:
        if js_k in p and p[js_k] is not None:
            _set(env_k, p[js_k])

    if "use_trend_filter" in p:
        _set("DBBAND_USE_TREND_FILTER", str(p["use_trend_filter"]).lower())
    if "cooldown_min" in p and p["cooldown_min"] is not None:
        _set("DBBAND_COOLDOWN_SEC", int(float(p["cooldown_min"]) * 60))
    if "sl_pct" in p:
        _set("DBBAND_STOP_LOSS_PCT", abs(float(p["sl_pct"])))
    if "force_eod_exit" in p:
        _set("DBBAND_FORCE_EOD_EXIT", str(p["force_eod_exit"]).lower())
    return patch


def web_body_to_dbband_env_patch(body: Dict[str, Any]) -> Dict[str, str]:
    """웹 saveDbBandConfig POST → DBBAND env 패치."""
    if not isinstance(body, dict):
        return {}

    def _get(key: str) -> Any:
        v = body.get(key)
        if v is None or v == "":
            return None
        return v

    p: Dict[str, Any] = {}
    for k in (
        "bb_period", "bb_inner_std", "bb_outer_std", "trend_ma_period",
        "side_mode", "entry_valid_bars", "entry_mode", "stop_mode",
        "stop_buffer_pct", "tp_mode", "rr_ratio",
        "cooldown_min", "max_daily", "min_price", "slot_money",
        "max_stocks", "total_budget_krw", "max_hold_bars", "timeframe",
        "min_invest_ratio_of_slot",
    ):
        v = _get(k)
        if v is not None:
            p[k] = v

    sl = _get("sl_pct")
    if sl is not None:
        p["sl_pct"] = float(sl)
    tp = _get("tp_pct")
    if tp is not None:
        p["tp_pct"] = float(tp)
    smh = _get("shoulder_min_high")
    if smh is not None:
        p["shoulder_min_high"] = float(smh)
    sc = _get("shoulder_cut_pct")
    if sc is not None:
        p["shoulder_cut_pct"] = float(sc)
    tr = _get("trail_pct")
    if tr is not None:
        p["trail_pct"] = float(tr)
    tra = _get("trail_arm_pct")
    if tra is not None:
        p["trail_arm_pct"] = float(tra)

    utf = _get("use_trend_filter")
    if utf is not None:
        p["use_trend_filter"] = utf
    if "time_start" in body and body["time_start"] not in (None, ""):
        p["time_start_hm"] = int(float(body["time_start"]))
    if "time_end" in body and body["time_end"] not in (None, ""):
        p["time_end_hm"] = int(float(body["time_end"]))
    if "force_eod_exit" in body:
        p["force_eod_exit"] = body.get("force_eod_exit")

    return params_to_dbband_env_patch(p)