kis_bot/kis_trader/strategies/updown_box_stock_cfg.py

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