#!/usr/bin/env python3 """ kis_trader/backtest/param_search_updown_box.py — UPDOWN 박스권 글로벌 파라미터 Grid Search ======================================================================================================== [전략 = UPDOWN 박스권(횡보 하단 되돌림 매수)] - 엔진: ``kis_trader.engine.updown_box`` (판별 evaluate_box · 진입 box_entry_signal · 청산 eval_box_exit_at_price · 백테 run_backtest_box) — **실매와 100% 동일 단일 소스**. - 파라미터는 **글로벌**(env_config ``UPDOWN_BOX_*``) — 종목별 행 없음. 여러 종목을 같은 파라미터로 돌려 **합산 성과**가 가장 좋은 조합을 찾는다(실매 운영과 정합). - 분봉 데이터: ``holding_min_candles`` (``holding_bot.get_stored_min_candles``) — 기본 15분. [관련 CLI] UPDOW (구 하락매수): kis_trader/backtest/param_search_updow.py UPDOWN 박스권 : kis_trader/backtest/param_search_updown_box.py ← 이 파일 실행: cd /home/hoon/kis_bot # 종목 직접 지정 python3 kis_trader/backtest/param_search_updown_box.py --codes 005930,000660,035720 --tf 15 \ --start 2026-01-01 --end 2026-06-30 --mode fast # watchlist active 종목 자동 사용 python3 -m kis_trader.backtest.param_search_updown_box --from-watchlist --tf 15 --mode full --top 30 # 1위 → env_config UPDOWN_BOX_* 적용 python3 kis_trader/backtest/param_search_updown_box.py --codes 005930 --tf 15 --apply 1 옵션: --codes 쉼표 구분 6자리 종목코드 (--from-watchlist 와 택일) --from-watchlist updown_watchlist active 종목 자동 사용 --start/--end 기간 (기본 최근 365일) --tf 분봉 (기본 env UPDOWN_SCAN_TF_MIN=15) --mode fast(핵심4축) | full (기본 fast) --max-combos 조합 상한 (기본 fast=800, full=0=무제한) --min-trades 합산 최소 거래 수 필터 (기본 3) --top 상위 N개 출력·저장 (기본 30) --out 결과 JSON 경로 (선택) --apply N N번째 결과를 env_config UPDOWN_BOX_* 에 적용 (총손익>0 일 때만) --allow-non-positive-pnl total_pnl≤0 이어도 적용 """ from __future__ import annotations import argparse import json import logging import os import sys from datetime import datetime, timedelta from typing import Any, Dict, List, Optional HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(os.path.dirname(HERE)) if ROOT not in sys.path: sys.path.insert(0, ROOT) from kis_trader.utils.legacy_root import ensure_legacy_root # noqa: E402 ensure_legacy_root() import holding_bot as hb # noqa: E402 from database import TradeDB # noqa: E402 from kis_trader.engine import updown_box as bx # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(message)s") logging.getLogger("TradeDB").setLevel(logging.WARNING) logger = logging.getLogger("param_search_updown_box") def _db() -> TradeDB: return TradeDB() def _results_dir() -> str: d = os.path.join(HERE, "results") os.makedirs(d, exist_ok=True) return d def _resolve_codes(db: TradeDB, args) -> List[str]: """--codes 또는 --from-watchlist 로 종목 리스트 해석.""" if args.from_watchlist: try: from kis_trader.strategies.updown_watchlist import ( ensure_updown_watchlist_table, list_active_watchlist, ) ensure_updown_watchlist_table(db) rows = list_active_watchlist(db) codes = [str(r.get("code") or "").strip() for r in rows] return [c for c in codes if len(c) == 6 and c.isdigit()] except Exception as e: logger.error("watchlist 조회 실패: %s", e) return [] raw = str(args.codes or "").strip() out: List[str] = [] for part in raw.replace(" ", "").split(","): if len(part) == 6 and part.isdigit(): out.append(part) return out def _apply_env_patch(db: TradeDB, patch: Dict[str, Any]) -> int: """UPDOWN_BOX_* env_config 패치 INSERT (최신 스냅샷에 머지).""" try: snap = db.get_merged_env_snapshot() or {} for k, v in patch.items(): snap[k] = str(v) eid = db.insert_env_snapshot(snap) if eid is None: logger.error("env_config INSERT 실패") return 8 logger.info("✅ env_config INSERT id=%s keys=%s", eid, list(patch.keys())) return 0 except Exception as e: logger.error("env 적용 실패: %s", e) return 8 def main(argv: Optional[List[str]] = None) -> int: from kis_trader.backtest.param_search_dates import resolve_param_search_range default_start, today = resolve_param_search_range("UPDOWN", lookback_days=365) ap = argparse.ArgumentParser(description="UPDOWN 박스권 글로벌 파라미터 그리드 탐색") ap.add_argument("--codes", default="", help="쉼표 구분 6자리 종목코드") ap.add_argument("--from-watchlist", action="store_true", help="watchlist active 종목 사용") ap.add_argument("--start", default=default_start, help="시작일 YYYY-MM-DD (거래일 보정)") ap.add_argument("--end", default=today, help="종료일 YYYY-MM-DD (주말·휴장이면 이전 장운영일)") ap.add_argument("--tf", type=int, default=-1, help="분봉 (기본 env UPDOWN_SCAN_TF_MIN=15)") ap.add_argument("--mode", choices=["fast", "full"], default="fast") ap.add_argument("--max-combos", type=int, default=-1, help="조합 상한 (0=무제한)") ap.add_argument("--min-trades", type=int, default=3, help="합산 최소 거래 수 필터") ap.add_argument("--rank-by", choices=["alpha", "pnl"], default="alpha", help="정렬 기준 alpha=B&H초과수익(기본) | pnl=절대손익") ap.add_argument("--keep-negative-alpha", action="store_true", help="기본값(미지정)은 base_cfg 기준 α≤0 종목(추세주)을 universe에서 제외") ap.add_argument("--top", type=int, default=30, help="상위 N개 출력·저장") ap.add_argument("--out", default="", help="결과 JSON 경로") ap.add_argument("--apply", nargs="?", const=1, type=int, default=None, metavar="N", help="N번째 결과를 env_config UPDOWN_BOX_* 에 적용") ap.add_argument("--allow-non-positive-pnl", action="store_true", help="total_pnl≤0 이어도 적용") args = ap.parse_args(argv) db = _db() try: codes = _resolve_codes(db, args) if not codes: logger.error("종목 없음 — --codes 005930,000660 또는 --from-watchlist 지정") return 1 if int(args.tf) >= 0: tf = int(args.tf) else: from kis_trader.utils.env import get_env_int tf = get_env_int("UPDOWN_SCAN_TF_MIN", 15) if tf not in hb.KIWOOM_MINUTE_TICS: logger.error("tf=%s 허용 목록 아님: %s", tf, hb.KIWOOM_MINUTE_TICS) return 1 base_cfg = bx.get_box_cfg_from_env() candles_by_code: Dict[str, List[Dict[str, Any]]] = {} for code in codes: cs = hb.get_stored_min_candles(db, code, args.start, args.end, tf_min=tf) if cs and len(cs) >= int(base_cfg.get("min_bars", 20)) + 2: candles_by_code[code] = cs else: logger.warning("⚠️ %s 분봉 부족(%s개) → 제외", code, len(cs) if cs else 0) if not candles_by_code: logger.error("백테 가능한 종목 0 (%s분봉 수집 필요: %s ~ %s)", tf, args.start, args.end) return 2 # === 추세종목(α≤0) 자동 제외 === # 박스전략은 "그냥 들고 있는 것(B&H)"보다 나아야 의미가 있다. base_cfg 기준으로 # 각 종목 백테 → 봇 총손익 ≤ 단순보유 인 종목은 박스권 대상이 아니므로 universe 에서 뺀다. # (--keep-negative-alpha 로 끌 수 있음) if not args.keep_negative_alpha: kept: Dict[str, List[Dict[str, Any]]] = {} for code, cs in candles_by_code.items(): bt = bx.run_backtest_box(cs, base_cfg) a = float(bt.get("alpha_pct", bt.get("total_pnl", 0)) or 0) bh = float(bt.get("buy_hold_pct", 0) or 0) if a > 0: kept[code] = cs logger.info(" ✔ %s 채택 (봇 %.2f%% vs B&H %.2f%% → α=%.2f%%)", code, float(bt.get("total_pnl", 0)), bh, a) else: logger.warning(" ✘ %s 제외 — 추세주 (봇 %.2f%% ≤ B&H %.2f%% → α=%.2f%%)", code, float(bt.get("total_pnl", 0)), bh, a) if not kept: logger.error("α>0 종목 0 — 박스전략 적합 종목이 없습니다 (--keep-negative-alpha 로 강제 가능)") return 2 candles_by_code = kept mode = str(args.mode or "fast").strip().lower() grid = bx.default_box_param_grid(mode) if mode == "full": max_combos = 0 if args.max_combos < 0 else max(0, int(args.max_combos)) else: max_combos = 800 if args.max_combos < 0 else max(0, int(args.max_combos)) results, meta = bx.run_param_search_box_multi( candles_by_code, grid=grid, base_cfg=base_cfg, min_trades_total=max(0, int(args.min_trades)), max_combos=max_combos, rank_by=str(args.rank_by), ) meta["search_mode"] = mode meta["tf"] = tf meta["start"] = args.start meta["end"] = args.end logger.info( "=== UPDOWN 박스권 param search tf=%s 종목=%d ===\n" "그리드 조합=%s 백테=%s 통과=%s (min_trades=%s)", tf, len(candles_by_code), meta.get("cartesian_product"), meta.get("backtests_run"), meta.get("passed"), args.min_trades, ) logger.info("--- 그리드 축 ---") for k, vals in (meta.get("grid_axes") or {}).items(): logger.info(" [%s] %s", k, vals) logger.info("정렬기준=%s (alpha=B&H초과수익)", meta.get("rank_by")) top_n = max(1, int(args.top)) top = results[:top_n] for i, row in enumerate(top, 1): logger.info( "#%s α=%s%% pnl=%s%% (B&H %s%%) pf=%s win=%s%% trades=%s cfg=%s", i, row.get("alpha_pct"), row.get("total_pnl"), row.get("buy_hold_pct"), row.get("pf"), row.get("win_rate"), row.get("n_trades"), row.get("apply_cfg"), ) payload: Dict[str, Any] = { "strategy": "UPDOWN_BOX", "codes": list(candles_by_code.keys()), "tf": tf, "start": args.start, "end": args.end, "meta": meta, "top": top, } out_path = args.out.strip() if not out_path: ts = datetime.now().strftime("%Y%m%d_%H%M%S") out_path = os.path.join(_results_dir(), f"updown_box_param_{tf}m_{ts}.json") with open(out_path, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) logger.info("💾 결과 저장: %s", out_path) if args.apply is not None: rank = max(1, int(args.apply)) if rank > len(top): logger.error("순번 %s 유효하지 않음 (1~%s)", rank, len(top)) return 5 row = top[rank - 1] # alpha 정렬이면 α(B&H 초과수익)>0, pnl 정렬이면 절대손익>0 을 적용 가드로 사용 if str(args.rank_by) == "alpha": guard = float(row.get("alpha_pct") or 0) gname = "alpha" else: guard = float(row.get("total_pnl") or 0) gname = "total_pnl" if guard <= 0 and not args.allow_non_positive_pnl: logger.warning("%s=%s ≤ 0 → 미적용 (--allow-non-positive-pnl 로 강제)", gname, guard) return 6 patch = bx.box_cfg_to_env_patch(row.get("apply_cfg") or {}) if not patch: logger.error("apply_cfg 비어 있음") return 7 return _apply_env_patch(db, patch) return 0 finally: db.close() if __name__ == "__main__": raise SystemExit(main())