""" 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)