kis_bot/kis_trader/backtest/optuna_study_store.py

#!/usr/bin/env python3
"""Optuna 스터디 결과 · 총 횟수 후처리 게이트 (kis_quant_db.optuna_study_result).

kis_optuna 공식 studies/trials 는 건드리지 않는다.
실매 봇 루프에 넣지 않는다.
"""
from __future__ import annotations

import json
import logging
import os
import socket
from typing import Any, Callable, Dict, Optional, Tuple

logger = logging.getLogger("optuna_study_store")

KIS_STUDY_TRIALS_ATTR = "kis_study_trials"

_DDL = """
CREATE TABLE IF NOT EXISTS optuna_study_result (
  study_name          VARCHAR(255) NOT NULL COMMENT 'Optuna study 이름 (kis_optuna 과 동일)',
  strategy            VARCHAR(32)  NOT NULL DEFAULT '' COMMENT 'tail|momentum|us_momentum|breakout|scalp',
  mode                VARCHAR(32)  NOT NULL DEFAULT '' COMMENT 'tpe 등',
  start_date          VARCHAR(10)  NULL COMMENT '백테 시작 YYYY-MM-DD',
  end_date            VARCHAR(10)  NULL COMMENT '백테 종료 YYYY-MM-DD',
  job_id              VARCHAR(64)  NULL COMMENT '웹 잡 id (참고)',
  symbol              VARCHAR(16)  NULL COMMENT '해외 종목핀일 때만',
  study_trials        INT          NOT NULL DEFAULT 0 COMMENT '스터디 총 시도 목표(COMPLETE+PRUNED+FAIL). 0이면 미설정',
  n_complete          INT          NOT NULL DEFAULT 0 COMMENT '끝난 시도 수 캐시(COMPLETE+PRUNED+FAIL, 후처리 게이트)',
  n_running           INT          NOT NULL DEFAULT 0 COMMENT '진행 중 trial 수 캐시 (표시용)',
  payload_json        LONGTEXT     NULL COMMENT 'results_all/gated/stable, mode_combo, postprocess_topn, overfit (웹 요약과 동일 키)',
  briefing_md         MEDIUMTEXT   NULL COMMENT '브리핑 마크다운',
  pp_needed           TINYINT      NOT NULL DEFAULT 0 COMMENT '1=목표 도달, 후처리 대기',
  pp_status           VARCHAR(16)  NOT NULL DEFAULT 'idle' COMMENT 'idle|running|done|error',
  pp_for_n            INT          NOT NULL DEFAULT 0 COMMENT '이 후처리가 반영한 n_complete',
  pp_lock_until       DATETIME     NULL COMMENT '락 만료 (죽은 워커 복구)',
  pp_lock_owner       VARCHAR(64)  NULL COMMENT '잠근 호스트/pid',
  pp_error            VARCHAR(500) NULL COMMENT '후처리 실패 메시지',
  leftover_note       VARCHAR(200) NULL COMMENT '웹 남은횟수 안내 캐시',
  updated_at          DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '행 갱신',
  created_at          DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '최초 생성',
  PRIMARY KEY (study_name),
  KEY idx_pp_needed (pp_needed, pp_status),
  KEY idx_job (job_id),
  KEY idx_updated (updated_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  COMMENT='Optuna 스터디별 웹 결과·후처리 산출 (파일 JSON 대체)'
"""


def _db():
    from database import TradeDB

    return TradeDB()


def ensure_optuna_study_result_table() -> None:
    db = _db()
    db.conn.execute_long(_DDL, timeout_sec=120)


def count_study_states(study: Any) -> Tuple[int, int, int]:
    """(COMPLETE 수, RUNNING 수, 끝난 시도 수).

    끝난 시도 = COMPLETE + PRUNED + FAIL.
    후처리·남은횟수 게이트는 「돌린 횟수」(끝난 시도) 기준.
    문제출제 오류(pruned: tp_max<tp 등)는 목표에서 빼지 않고 시도로 친다.
    """
    n_complete = 0
    n_running = 0
    n_finished = 0
    try:
        import optuna
        complete = optuna.trial.TrialState.COMPLETE
        running = optuna.trial.TrialState.RUNNING
        pruned = optuna.trial.TrialState.PRUNED
        fail = optuna.trial.TrialState.FAIL
    except Exception:
        complete = running = pruned = fail = None
    for t in list(getattr(study, "trials", None) or []):
        st = getattr(t, "state", None)
        if complete is not None and st == complete:
            n_complete += 1
            n_finished += 1
        elif running is not None and st == running:
            n_running += 1
        elif pruned is not None and st == pruned:
            n_finished += 1
        elif fail is not None and st == fail:
            n_finished += 1
    return n_complete, n_running, n_finished


def parse_study_trials_value(raw: Any) -> int:
    """빈값/0 = 미사용. 음수 없음."""
    if raw is None:
        return 0
    s = str(raw).strip()
    if not s or s.lower() in ("none", "null"):
        return 0
    try:
        n = int(float(s))
    except (TypeError, ValueError):
        return 0
    return max(0, n)


def resolve_cli_study_trials(cli_value: Any = None) -> int:
    """CLI > 프로세스 env. DB 전역 기본 200 넣지 않음."""
    n = parse_study_trials_value(cli_value)
    if n > 0:
        return n
    n = parse_study_trials_value(os.environ.get("KIS_OPTUNA_STUDY_TRIALS"))
    if n > 0:
        return n
    n = parse_study_trials_value(os.environ.get("PARAM_SEARCH_OPTUNA_STUDY_TRIALS"))
    if n > 0:
        return n
    try:
        from kis_trader.utils.env import get_env_from_db

        n = parse_study_trials_value(get_env_from_db("PARAM_SEARCH_OPTUNA_STUDY_TRIALS", "") or "")
    except Exception:
        n = 0
    return n


def apply_kis_study_trials(
    study: Any,
    *,
    n_trials: int,
    log: Optional[logging.Logger] = None,
    cli_study_trials: Optional[int] = None,
) -> int:
    """첫 워커가 user_attr 기록. 이후 워커는 저장된 목표 유지(더 작은 값으로 덮지 않음)."""
    lg = log or logger
    requested = parse_study_trials_value(
        cli_study_trials if cli_study_trials is not None else resolve_cli_study_trials()
    )
    fallback = max(1, int(n_trials or 0))
    stored = 0
    try:
        raw = (study.user_attrs or {}).get(KIS_STUDY_TRIALS_ATTR)
        stored = parse_study_trials_value(raw)
    except Exception:
        stored = 0
    if stored > 0:
        if requested > 0 and requested != stored:
            lg.warning(
                "⚠️ kis_study_trials=%s 유지 (요청 %s 무시 — 목표를 낮춰 조기 후처리하지 않음)",
                stored, requested,
            )
        return stored
    target = requested if requested > 0 else fallback
    try:
        study.set_user_attr(KIS_STUDY_TRIALS_ATTR, int(target))
    except Exception as exc:
        lg.warning("⚠️ kis_study_trials user_attr 저장 실패: %s", exc)
    lg.info("📌 스터디 총 횟수 목표 kis_study_trials=%s (이번 --trials=%s)", target, n_trials)
    return int(target)


def bind_study_trials(study: Any, *, n_trials: int, log: Optional[logging.Logger] = None) -> int:
    return apply_kis_study_trials(study, n_trials=n_trials, log=log)


def leftover_trials(study_trials: int, n_finished: int) -> int:
    """남은 횟수 = 목표 − 끝난 시도(COMPLETE+PRUNED+FAIL)."""
    t = int(study_trials or 0)
    c = int(n_finished or 0)
    if t <= 0:
        return 0
    return max(0, t - c)


def clamp_optimize_n_trials(
    study: Any,
    n_trials: int,
    *,
    log: Optional[logging.Logger] = None,
) -> int:
    """
    Optuna study.optimize(n_trials=N) 은 **추가** N회.
    study-trials 목표가 있으면 남은 횟수로 잘라서, 목표 초과 연타를 막는다.
    """
    lg = log or logger
    try:
        n = max(0, int(n_trials or 0))
    except (TypeError, ValueError):
        n = 0
    try:
        goal = parse_study_trials_value((study.user_attrs or {}).get(KIS_STUDY_TRIALS_ATTR))
    except Exception:
        goal = 0
    if goal <= 0:
        return max(1, n) if n > 0 else 0
    _c, _r, n_f = count_study_states(study)
    left = leftover_trials(goal, n_f)
    if left <= 0:
        lg.info(
            "📌 study-trials 목표 도달 — 추가 trial 생략 (끝난시도=%s / 목표=%s)",
            n_f, goal,
        )
        return 0
    if n <= 0 or n > left:
        lg.warning(
            "⚠️ --trials=%s → leftover %s 로 축소 (끝난시도=%s / study-trials=%s)",
            n, left, n_f, goal,
        )
        return left
    return n


def make_study_goal_stop_callback(log: Optional[logging.Logger] = None):
    """trial 끝날 때마다 study-trials 도달이면 study.stop()."""
    lg = log or logger

    def _cb(study: Any, _trial: Any) -> None:
        try:
            goal = parse_study_trials_value((study.user_attrs or {}).get(KIS_STUDY_TRIALS_ATTR))
        except Exception:
            return
        if goal <= 0:
            return
        _c, _r, n_f = count_study_states(study)
        if n_f >= goal:
            lg.info(
                "📌 study-trials 도달 → optimize 중지 (끝난시도=%s / 목표=%s)",
                n_f, goal,
            )
            try:
                study.stop()
            except Exception:
                pass

    return _cb


def leftover_note(
    study_trials: int,
    n_finished: int,
    *,
    n_complete: Optional[int] = None,
) -> str:
    left = leftover_trials(study_trials, n_finished)
    t = int(study_trials or 0)
    done = int(n_finished or 0)
    ok = int(n_complete) if n_complete is not None else done
    if t <= 0:
        return ""
    if left > 0:
        return "남은 %s회 (시도 %s / 목표 %s · COMPLETE %s)" % (left, done, t, ok)
    return "목표 도달 (시도 %s / 목표 %s · COMPLETE %s)" % (done, t, ok)


def _owner() -> str:
    host = (socket.gethostname() or "host")[:40]
    return ("%s:%s" % (host, os.getpid()))[:64]


def _lock_minutes() -> int:
    try:
        from kis_trader.utils.env import get_env_int

        return max(1, int(get_env_int("OPTUNA_STUDY_PP_LOCK_MINUTES", 30)))
    except Exception:
        return 30


def load_row(study_name: str) -> Optional[Dict[str, Any]]:
    name = str(study_name or "").strip()
    if not name:
        return None
    try:
        ensure_optuna_study_result_table()
        cur = _db().conn.execute(
            "SELECT * FROM optuna_study_result WHERE study_name=%s",
            (name,),
        )
        row = cur.fetchone()
        return dict(row) if row else None
    except Exception as exc:
        logger.warning("⚠️ optuna_study_result 조회 실패: %s", exc)
        return None


def load_payload_dict(study_name: str) -> Optional[Dict[str, Any]]:
    row = load_row(study_name)
    if not row:
        return None
    raw = row.get("payload_json")
    if not raw:
        return None
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else None
    except Exception:
        return None


def load_phase1_study_for_job(job_id: str) -> Optional[str]:
    """step job_id → 1차 study_name (MariaDB). refine1 우선."""
    jid = str(job_id or "").strip()[:64]
    if not jid:
        return None
    try:
        ensure_optuna_study_result_table()
        cur = _db().conn.execute(
            "SELECT study_name FROM optuna_study_result "
            "WHERE job_id=%s AND study_name LIKE %s "
            "ORDER BY updated_at DESC LIMIT 1",
            (jid, "%refine1%"),
        )
        row = cur.fetchone()
        if row:
            name = str(row.get("study_name") or "").strip()
            if name:
                return name
        cur = _db().conn.execute(
            "SELECT study_name FROM optuna_study_result "
            "WHERE job_id=%s ORDER BY updated_at DESC LIMIT 1",
            (jid,),
        )
        row = cur.fetchone()
        if row:
            return str(row.get("study_name") or "").strip() or None
    except Exception as exc:
        logger.warning("⚠️ load_phase1_study_for_job 실패 job=%s: %s", jid, exc)
    return None


def phase1_payload_ready(study_name: str) -> bool:
    """1차 Top10 narrow 입력용 payload 가 DB에 있는지."""
    name = str(study_name or "").strip()
    if not name:
        return False
    data = load_payload_dict(name)
    return payload_has_rows(data)


def upsert_counts(
    *,
    study_name: str,
    strategy: str = "",
    mode: str = "",
    start: str = "",
    end: str = "",
    job_id: str = "",
    symbol: str = "",
    study_trials: int = 0,
    n_complete: int = 0,
    n_running: int = 0,
    leftover_n_complete: Optional[int] = None,
) -> None:
    """n_complete = 끝난 시도(게이트). leftover_n_complete = COMPLETE 실수(안내)."""
    name = str(study_name or "").strip()
    if not name:
        return
    note = leftover_note(
        study_trials,
        n_complete,
        n_complete=leftover_n_complete if leftover_n_complete is not None else None,
    )
    needed = 1 if leftover_trials(study_trials, n_complete) > 0 else 0
    jid = (job_id or os.environ.get("OPTUNA_WEB_JOB_ID") or "")[:64] or None
    sym = (symbol or "")[:16] or None
    try:
        ensure_optuna_study_result_table()
        _db().conn.execute(
            """
            INSERT INTO optuna_study_result (
              study_name, strategy, mode, start_date, end_date, job_id, symbol,
              study_trials, n_complete, n_running, leftover_note, pp_needed
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON DUPLICATE KEY UPDATE
              strategy=IF(VALUES(strategy)='', strategy, VALUES(strategy)),
              mode=IF(VALUES(mode)='', mode, VALUES(mode)),
              start_date=COALESCE(VALUES(start_date), start_date),
              end_date=COALESCE(VALUES(end_date), end_date),
              job_id=COALESCE(VALUES(job_id), job_id),
              symbol=COALESCE(VALUES(symbol), symbol),
              study_trials=IF(VALUES(study_trials)>0, VALUES(study_trials), study_trials),
              n_complete=VALUES(n_complete),
              n_running=VALUES(n_running),
              leftover_note=VALUES(leftover_note),
              pp_needed=VALUES(pp_needed)
            """,
            (
                name,
                str(strategy or "")[:32],
                str(mode or "")[:32],
                (start or "")[:10] or None,
                (end or "")[:10] or None,
                jid,
                sym,
                int(study_trials or 0),
                int(n_complete or 0),
                int(n_running or 0),
                note[:200] if note else None,
                int(needed),
            ),
        )
    except Exception as exc:
        logger.warning("⚠️ optuna_study_result upsert 실패: %s", exc)


def payload_has_rows(data: Optional[Dict[str, Any]]) -> bool:
    if not isinstance(data, dict):
        return False
    if data.get("results_gated") or data.get("results_all") or data.get("results") or data.get("results_stable"):
        return True
    mc = data.get("mode_combo") or {}
    return bool(isinstance(mc, dict) and (mc.get("params") or mc.get("backtest")))


def payload_row_count(data: Optional[Dict[str, Any]]) -> int:
    if not isinstance(data, dict):
        return 0
    return max(
        len(list(data.get("results_all") or [])),
        len(list(data.get("results") or [])),
        len(list(data.get("results_gated") or [])),
        len(list(data.get("results_stable") or [])),
    )


def _payload_has_postprocess(data: Optional[Dict[str, Any]]) -> bool:
    if not isinstance(data, dict):
        return False
    topn = data.get("postprocess_topn") or {}
    if isinstance(topn, dict) and topn.get("postprocess_by_anchor"):
        return True
    mc = data.get("mode_combo") or {}
    return bool(isinstance(mc, dict) and (mc.get("params") or mc.get("backtest")))


def save_payload(
    *,
    study_name: str,
    out_data: Dict[str, Any],
    n_complete: int = 0,
    briefing_text: Optional[str] = None,
    pp_status: str = "idle",
    pp_error: Optional[str] = None,
    job_id: str = "",
) -> None:
    name = str(study_name or "").strip()
    if not name or not isinstance(out_data, dict):
        return
    payload = json.dumps(out_data, ensure_ascii=False)
    brief = briefing_text
    if brief is None:
        bp = str(out_data.get("briefing_md_path") or "")
        if bp and os.path.isfile(bp):
            try:
                with open(bp, encoding="utf-8") as f:
                    brief = f.read()
            except Exception:
                brief = None
        md = str(out_data.get("briefing_md") or "")
        if brief is None and md and str(md).endswith(".md") and os.path.isfile(md):
            try:
                with open(md, encoding="utf-8") as f:
                    brief = f.read()
            except Exception:
                brief = None
    strat = str(out_data.get("strategy") or "")[:32]
    mode = str(out_data.get("mode") or "")[:32]
    start = (str(out_data.get("start") or "")[:10] or None)
    end = (str(out_data.get("end") or "")[:10] or None)
    jid = (job_id or os.environ.get("OPTUNA_WEB_JOB_ID") or "")[:64] or None
    sym = str(out_data.get("symbol") or "")[:16] or None
    st_goal = parse_study_trials_value(out_data.get("optuna_study_trials"))
    # 게이트용 = 끝난 시도. COMPLETE만 쓰지 말 것(pruned 있으면 이어돌리기 오판).
    n_done = int(
        n_complete
        or out_data.get("optuna_n_finished")
        or out_data.get("optuna_trials_completed")
        or out_data.get("optuna_n_complete")
        or 0
    )
    n_ok = int(out_data.get("optuna_n_complete") or 0)
    st = str(pp_status or "idle")[:16]
    needed = 0 if st == "done" else (1 if leftover_trials(st_goal, n_done) > 0 else 0)
    note = leftover_note(st_goal, n_done, n_complete=n_ok if n_ok else None)
    try:
        ensure_optuna_study_result_table()
        _db().conn.execute(
            """
            INSERT INTO optuna_study_result (
              study_name, strategy, mode, start_date, end_date, job_id, symbol,
              study_trials, n_complete, leftover_note, payload_json, briefing_md,
              pp_status, pp_error, pp_for_n, pp_needed
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON DUPLICATE KEY UPDATE
              strategy=IF(VALUES(strategy)='', strategy, VALUES(strategy)),
              mode=IF(VALUES(mode)='', mode, VALUES(mode)),
              start_date=COALESCE(VALUES(start_date), start_date),
              end_date=COALESCE(VALUES(end_date), end_date),
              job_id=COALESCE(VALUES(job_id), job_id),
              symbol=COALESCE(VALUES(symbol), symbol),
              study_trials=IF(VALUES(study_trials)>0, VALUES(study_trials), study_trials),
              n_complete=IF(VALUES(n_complete)>0, VALUES(n_complete), n_complete),
              leftover_note=COALESCE(VALUES(leftover_note), leftover_note),
              payload_json=VALUES(payload_json),
              briefing_md=COALESCE(VALUES(briefing_md), briefing_md),
              pp_status=VALUES(pp_status),
              pp_error=VALUES(pp_error),
              pp_for_n=IF(VALUES(pp_status)='done', VALUES(pp_for_n), pp_for_n),
              pp_needed=IF(VALUES(pp_status)='done', 0, VALUES(pp_needed)),
              pp_lock_until=IF(VALUES(pp_status)='done' OR VALUES(pp_status)='error', NULL, pp_lock_until),
              pp_lock_owner=IF(VALUES(pp_status)='done' OR VALUES(pp_status)='error', NULL, pp_lock_owner)
            """,
            (
                name, strat, mode, start, end, jid, sym,
                int(st_goal or 0), n_done, (note[:200] if note else None), payload, brief,
                st, (str(pp_error)[:500] if pp_error else None),
                n_done if st == "done" else 0, needed,
            ),
        )
    except Exception as exc:
        logger.warning("⚠️ optuna_study_result payload 저장 실패: %s", exc)


def ingest_out_data(
    data: Dict[str, Any],
    *,
    job_id: str = "",
    replace: bool = False,
) -> Dict[str, Any]:
    """JSON/out_data 한 건을 optuna_study_result 에 넣는다. 더 빈약한 파일로 덮지 않음."""
    if not isinstance(data, dict):
        return {"ok": False, "reason": "not_dict"}
    name = str(data.get("optuna_study_name") or data.get("study_name") or "").strip()
    if not name:
        return {"ok": False, "reason": "no_study_name"}
    old = load_payload_dict(name)
    if old and not replace:
        old_n = payload_row_count(old)
        new_n = payload_row_count(data)
        old_pp = _payload_has_postprocess(old)
        new_pp = _payload_has_postprocess(data)
        if old_pp and not new_pp and old_n >= new_n:
            return {"ok": True, "skipped": "keep_richer", "study_name": name}
        if old_n > new_n and not new_pp:
            return {"ok": True, "skipped": "keep_more_rows", "study_name": name}
    n_done = int(
        data.get("optuna_n_finished")
        or data.get("optuna_trials_completed")
        or 0
    )
    n_ok = int(data.get("optuna_n_complete") or 0)
    if n_done <= 0:
        n_done = n_ok or payload_row_count(data) or 0
    st_goal = parse_study_trials_value(data.get("optuna_study_trials"))
    upsert_counts(
        study_name=name,
        strategy=str(data.get("strategy") or ""),
        mode=str(data.get("mode") or ""),
        start=str(data.get("start") or ""),
        end=str(data.get("end") or ""),
        job_id=job_id,
        symbol=str(data.get("symbol") or ""),
        study_trials=st_goal,
        n_complete=n_done,
        n_running=0,
        leftover_n_complete=n_ok if n_ok else None,
    )
    pp_done = _payload_has_postprocess(data)
    save_payload(
        study_name=name,
        out_data=data,
        n_complete=n_done,
        pp_status="done" if pp_done else "idle",
        job_id=job_id,
    )
    return {"ok": True, "study_name": name, "n": payload_row_count(data), "pp": pp_done}


def migrate_optuna_json_dir(
    results_dir: str,
    *,
    jobs_dir: str = "",
) -> Dict[str, Any]:
    """results/*.json + 웹 잡 result_json 을 테이블로 이관."""
    from pathlib import Path

    seen = set()
    ok_n = 0
    skip_n = 0
    err_n = 0
    paths = []
    rd = Path(results_dir)
    if rd.is_dir():
        paths.extend(sorted(rd.glob("optuna_*.json"), key=lambda p: p.stat().st_mtime))
    jd = Path(jobs_dir) if jobs_dir else None
    if jd and jd.is_dir():
        for jp in jd.glob("*.json"):
            try:
                meta = json.loads(jp.read_text(encoding="utf-8"))
            except Exception:
                continue
            rp = str((meta or {}).get("result_json") or "")
            if rp and Path(rp).is_file():
                paths.append(Path(rp))
            for extra in list((meta or {}).get("result_jsons") or []):
                if extra and Path(str(extra)).is_file():
                    paths.append(Path(str(extra)))
    for p in paths:
        try:
            key = str(p.resolve())
        except Exception:
            key = str(p)
        if key in seen or str(p.name).endswith("_from_db.json"):
            continue
        seen.add(key)
        try:
            data = json.loads(p.read_text(encoding="utf-8"))
        except Exception as exc:
            err_n += 1
            logger.warning("⚠️ migrate JSON 읽기 실패 %s: %s", p, exc)
            continue
        out = ingest_out_data(data, job_id="")
        if out.get("skipped"):
            skip_n += 1
        elif out.get("ok"):
            ok_n += 1
        else:
            err_n += 1
            logger.warning("⚠️ migrate skip %s: %s", p.name, out.get("reason"))
    return {"ok": True, "ingested": ok_n, "skipped": skip_n, "errors": err_n, "files": len(seen)}


def try_lock_postprocess(study_name: str) -> bool:
    name = str(study_name or "").strip()
    if not name:
        return False
    owner = _owner()
    mins = _lock_minutes()
    try:
        ensure_optuna_study_result_table()
        cur = _db().conn.execute(
            f"""
            UPDATE optuna_study_result
            SET pp_status='running',
                pp_lock_owner=%s,
                pp_lock_until=DATE_ADD(NOW(), INTERVAL {int(mins)} MINUTE),
                pp_error=NULL
            WHERE study_name=%s
              AND pp_status<>'running'
              AND (pp_lock_until IS NULL OR pp_lock_until < NOW())
            """,
            (owner, name),
        )
        return int(getattr(cur, "rowcount", 0) or 0) > 0
    except Exception as exc:
        logger.warning("⚠️ 후처리 락 실패: %s", exc)
        return False


def set_study_trials_target(study_name: str, n: int, storage_url: Optional[str] = None) -> None:
    """확정: 목표를 현재 완료 수로 내린다."""
    name = str(study_name or "").strip()
    target = max(0, int(n or 0))
    if not name:
        return
    try:
        ensure_optuna_study_result_table()
        _db().conn.execute(
            """
            UPDATE optuna_study_result
            SET study_trials=%s,
                leftover_note=%s,
                pp_needed=0
            WHERE study_name=%s
            """,
            (target, leftover_note(target, target), name),
        )
    except Exception as exc:
        logger.warning("⚠️ study_trials 확정 실패: %s", exc)
    if storage_url:
        try:
            import optuna

            st = optuna.load_study(study_name=name, storage=storage_url)
            st.set_user_attr(KIS_STUDY_TRIALS_ATTR, target)
        except Exception as exc:
            logger.warning("⚠️ kis_study_trials user_attr 확정 실패: %s", exc)


def should_run_postprocess(n_finished: int, study_trials: int) -> bool:
    """목표 도달 = 끝난 시도(COMPLETE+PRUNED+FAIL) ≥ kis_study_trials."""
    t = int(study_trials or 0)
    if t <= 0:
        return True
    return int(n_finished or 0) >= t


def finalize_optuna_export(
    study: Any,
    *,
    out_data: Dict[str, Any],
    out_path: str,
    strategy: str,
    mode: str,
    enrich_fn: Callable[[], None],
    log: Optional[logging.Logger] = None,
    symbol: str = "",
) -> bool:
    """optimize 이후. 시도 목표 미달이면 후처리 스킵, 도달이면 락 후 기존 enrich."""
    lg = log or logger
    name = str(getattr(study, "study_name", "") or out_data.get("optuna_study_name") or "")
    n_c, n_r, n_f = count_study_states(study)
    try:
        target = parse_study_trials_value((study.user_attrs or {}).get(KIS_STUDY_TRIALS_ATTR))
    except Exception:
        target = 0
    if target <= 0:
        target = max(1, int(out_data.get("optuna_n_trials_requested") or 0) or n_f or n_c)
    start = str(out_data.get("start") or "")
    end = str(out_data.get("end") or "")
    # DB n_complete 컬럼 = 끝난 시도(게이트·남은횟수). COMPLETE 실수는 out_data 에.
    upsert_counts(
        study_name=name,
        strategy=str(out_data.get("strategy") or strategy or ""),
        mode=str(out_data.get("mode") or mode or ""),
        start=start,
        end=end,
        symbol=str(out_data.get("symbol") or symbol or ""),
        study_trials=target,
        n_complete=n_f,
        n_running=n_r,
        leftover_n_complete=n_c,
    )
    out_data["optuna_study_trials"] = target
    out_data["optuna_n_complete"] = n_c
    out_data["optuna_n_finished"] = n_f
    try:
        from kis_trader.backtest.optuna_common import annotate_optuna_period_daily_avg
        annotate_optuna_period_daily_avg(out_data)
    except Exception:
        pass
    row = load_row(name) or {}
    if str(row.get("pp_status") or "") == "done" and row.get("payload_json") and int(row.get("pp_for_n") or 0) >= target:
        lg.info("📌 후처리 이미 완료 (pp_for_n=%s, 목표=%s) — 스킵", row.get("pp_for_n"), target)
        return False
    if not should_run_postprocess(n_f, target):
        save_payload(
            study_name=name,
            out_data=out_data,
            n_complete=n_f,
            pp_status="idle",
        )
        lg.info(
            "📌 스터디 총 횟수 미달 — 후처리 스킵 "
            "(시도 %s / 목표 %s · COMPLETE %s). 이어 돌리기 또는 확정.",
            n_f, target, n_c,
        )
        return False
    if not try_lock_postprocess(name):
        lg.info("📌 다른 워커가 후처리 중/완료 — 이 프로세스는 후처리 생략")
        return False
    try:
        enrich_fn()
        save_payload(
            study_name=name,
            out_data=out_data,
            n_complete=n_f,
            pp_status="done",
        )
        lg.info(
            "📌 optuna_study_result payload 저장 study=%s 시도=%s COMPLETE=%s",
            name, n_f, n_c,
        )
        return True
    except Exception as exc:
        save_payload(
            study_name=name,
            out_data=out_data,
            n_complete=n_f,
            pp_status="error",
            pp_error=str(exc),
        )
        raise


def flags_for_web(study_name: str) -> Dict[str, Any]:
    row = load_row(study_name) or {}
    t = int(row.get("study_trials") or 0)
    # n_complete 컬럼 = 끝난 시도(게이트용). COMPLETE 실수는 leftover_note 참고.
    done = int(row.get("n_complete") or 0)
    left = leftover_trials(t, done)
    pp = str(row.get("pp_status") or "idle")
    has_payload = bool(row.get("payload_json"))
    note = row.get("leftover_note") or leftover_note(t, done)
    return {
        "study_name": study_name,
        "study_trials": t,
        "n_complete": done,
        "n_running": int(row.get("n_running") or 0),
        "leftover_trials": left,
        "leftover_note": note,
        "pp_status": pp,
        "pp_needed": int(row.get("pp_needed") or 0),
        "has_payload": has_payload,
        "can_continue": left > 0,
        "can_confirm": left > 0,
    }