""" kis_trader/strategies/updown_box_stock_cfg.py — UPDOWN 박스권 **종목별 파라미터 오버라이드** ================================================================================================= [역할] - 박스 전략은 기본 **글로벌**(env ``UPDOWN_BOX_*``)로 모든 watchlist 종목을 돌린다. - 다만 특정 종목(예: 변동성이 다른 우량주)은 손으로 핀(pin)해 **종목별 파라미터**를 쓰고 싶을 수 있다. → 이 테이블(``updown_box_stock_cfg``)이 그 **선택적 오버라이드**를 담는다. - 값이 NULL(미설정)인 항목은 **글로벌 값을 그대로 상속**한다 (부분 오버라이드 가능). [설계 — 구 ``updow_stock_config`` 와 분리하는 이유] - ``updow_stock_config`` 는 구 하락매수(updow_buy) 엔진 전용 컬럼 구조 + "행 추가(latest-row)" 방식이라, 박스 파라미터를 섞으면 의미·쓰기 충돌이 난다. 그래서 박스 전용 **코드 UNIQUE upsert** 테이블을 둔다. [키] - ``box_cfg_to_env_patch`` 의 키맵과 동일한 박스 파라미터를 종목별로 저장. - ``ratchet_tiers`` 만 문자열('' = 래칫 OFF), 나머지는 DOUBLE. 모든 수치는 하드코딩 금지 — 글로벌 기본은 env(``UPDOWN_BOX_*``)에서 로드, 여기선 종목별 override 만. """ from __future__ import annotations import logging from datetime import datetime as dt from typing import Any, Dict, List, Optional logger = logging.getLogger("kis_trader.updown_box_stock_cfg") # 종목별 오버라이드 가능한 박스 파라미터 (DOUBLE) — ratchet_tiers 는 문자열로 별도 처리 BOX_NUMERIC_KEYS = ( "near_low_pct", "range_max_pct", "bb_bw_max", "ma_slope_tol_pct", "body_drop_min_pct", "breach_buffer_pct", "panic_from_high_pct", "stop_loss_pct", "tp_pct", "shoulder_min_high_pct", "shoulder_cut_pct", "max_hold_bars", "lookback_bars", "breakout_follow", ) BOX_STR_KEYS = ("ratchet_tiers",) BOX_ALL_KEYS = tuple(BOX_NUMERIC_KEYS) + tuple(BOX_STR_KEYS) def _trade_db_core(db: Any): """TradeDBExt 이면 ``.raw``, 아니면 그대로 (다른 cfg 모듈과 동일 규약).""" return getattr(db, "raw", db) def _ddl() -> str: cols = [ "id BIGINT AUTO_INCREMENT PRIMARY KEY", "code VARCHAR(20) NOT NULL", "name VARCHAR(50) NOT NULL DEFAULT ''", ] for k in BOX_NUMERIC_KEYS: cols.append(f"{k} DOUBLE NULL") for k in BOX_STR_KEYS: cols.append(f"{k} VARCHAR(64) NULL") cols.append("updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP") cols.append("UNIQUE KEY uq_updown_box_stock_code (code)") body = ",\n ".join(cols) return ( "CREATE TABLE IF NOT EXISTS updown_box_stock_cfg (\n " + body + "\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 " "COMMENT='UPDOWN 박스권 종목별 파라미터 오버라이드(NULL=글로벌 상속)'" ) def ensure_updown_box_stock_cfg_table(db: Any) -> None: """``updown_box_stock_cfg`` 테이블 생성 + 신규 키 컬럼 마이그레이션(있으면 무시).""" raw = _trade_db_core(db) try: raw.conn.execute(_ddl()) except Exception as e: logger.warning("updown_box_stock_cfg 생성 경고: %s", e) # 구버전 테이블에 신규 박스 키가 빠져 있으면 추가 for k in BOX_NUMERIC_KEYS: try: raw.conn.execute(f"ALTER TABLE updown_box_stock_cfg ADD COLUMN {k} DOUBLE NULL") except Exception: pass for k in BOX_STR_KEYS: try: raw.conn.execute(f"ALTER TABLE updown_box_stock_cfg ADD COLUMN {k} VARCHAR(64) NULL") except Exception: pass try: raw.conn.commit() except Exception: pass def get_box_overrides(db: Any, code: str) -> Dict[str, Any]: """종목별 박스 오버라이드 — **설정된(NULL 아님) 키만** 반환. 없으면 {}.""" ensure_updown_box_stock_cfg_table(db) code = str(code or "").strip() if not code: return {} raw = _trade_db_core(db) try: row = raw.conn.execute( "SELECT * FROM updown_box_stock_cfg WHERE code=%s LIMIT 1", (code,) ).fetchone() except Exception as e: logger.warning("get_box_overrides 조회 실패 %s: %s", code, e) return {} if not row: return {} d = dict(row) out: Dict[str, Any] = {} for k in BOX_NUMERIC_KEYS: v = d.get(k) if v is not None: try: out[k] = float(v) except (TypeError, ValueError): pass for k in BOX_STR_KEYS: v = d.get(k) if v is not None: # 빈 문자열('')도 유효 — 래칫 OFF 의 명시적 표현 out[k] = str(v) return out def set_box_overrides(db: Any, code: str, name: str, overrides: Dict[str, Any]) -> bool: """종목별 박스 오버라이드 upsert. - ``overrides`` 에 담긴 키만 갱신. 값이 None 또는 '' (숫자 키) 이면 해당 키는 **NULL(글로벌 상속)** 로 클리어. - ``ratchet_tiers`` 는 '' 도 유효값(래칫 OFF)으로 저장하고, None 이면 NULL(글로벌 상속). """ ensure_updown_box_stock_cfg_table(db) code = str(code or "").strip() if not code: return False raw = _trade_db_core(db) set_cols: List[str] = ["name"] set_vals: List[Any] = [str(name or code)[:50]] for k in BOX_NUMERIC_KEYS: if k not in overrides: continue v = overrides.get(k) if v is None or (isinstance(v, str) and not str(v).strip()): set_cols.append(k) set_vals.append(None) # NULL = 글로벌 상속 else: try: set_cols.append(k) set_vals.append(float(v)) except (TypeError, ValueError): set_cols.append(k) set_vals.append(None) for k in BOX_STR_KEYS: if k not in overrides: continue v = overrides.get(k) if v is None: set_cols.append(k) set_vals.append(None) # 미설정 = 글로벌 상속 else: set_cols.append(k) set_vals.append(str(v).strip()) # '' = 래칫 OFF (명시) # INSERT ... ON DUPLICATE KEY UPDATE (code UNIQUE) insert_cols = ["code"] + set_cols insert_vals = [code] + set_vals placeholders = ", ".join(["%s"] * len(insert_cols)) update_clause = ", ".join([f"{c}=VALUES({c})" for c in set_cols]) sql = ( f"INSERT INTO updown_box_stock_cfg ({', '.join(insert_cols)}) " f"VALUES ({placeholders}) " f"ON DUPLICATE KEY UPDATE {update_clause}, updated_at=NOW()" ) try: raw.conn.execute(sql, insert_vals) raw.conn.commit() logger.info("📌 updown_box_stock_cfg 저장: %s (%d키)", code, len(set_cols) - 1) return True except Exception as e: logger.warning("set_box_overrides 저장 실패 %s: %s", code, e) return False def delete_box_overrides(db: Any, code: str) -> bool: """종목별 오버라이드 삭제 → 글로벌로 복귀.""" ensure_updown_box_stock_cfg_table(db) code = str(code or "").strip() if not code: return False raw = _trade_db_core(db) try: raw.conn.execute("DELETE FROM updown_box_stock_cfg WHERE code=%s", (code,)) raw.conn.commit() return True except Exception as e: logger.warning("delete_box_overrides 실패 %s: %s", code, e) return False def list_box_stock_cfg(db: Any) -> List[Dict[str, Any]]: """오버라이드가 등록된 종목 목록 [{code, name, n_overrides}, ...].""" ensure_updown_box_stock_cfg_table(db) raw = _trade_db_core(db) try: rows = raw.conn.execute( "SELECT * FROM updown_box_stock_cfg ORDER BY code" ).fetchall() except Exception: return [] out: List[Dict[str, Any]] = [] for r in (rows or []): d = dict(r) n = sum(1 for k in BOX_ALL_KEYS if d.get(k) is not None) out.append({"code": str(d.get("code") or ""), "name": str(d.get("name") or ""), "n_overrides": n}) return out