#!/usr/bin/env python3 """ kis_trader/backtest/param_search_updow.py — Updow(직전봉 하락 → 다음봉 시가 매수) 파라미터 Grid Search ======================================================================================================== - 엔진: ``kis_trader.strategies.updow_buy.run_param_search_updow`` - **종목 파라미터(베이스)**: ``updow_stock_config`` (없으면 env ``UPDOW_*`` 폴백) - **그리드 끝값**: env ``UPDOW_GRID_*`` (어깨 ``UPDOW_GRID_SHOULDER_*`` 포함) → ``updow_buy.default_param_grid`` / ``default_param_grid_web_fast`` - **분봉 데이터**: ``holding_min_candles`` (``holding_bot.get_stored_min_candles``) [관련 CLI (전략별)] BREAKOUT : kis_trader/backtest/param_search_breakout.py UPDOW : kis_trader/backtest/param_search_updow.py ← 이 파일 (국내 6자리) UPDOW US : kis_trader/backtest/param_search_updow_us.py (해외 QQQM 등, 레짐 OFF) TAIL : kis_trader/backtest/tail_param_search.py 실행: cd /home/hoon/kis_bot python3 kis_trader/backtest/param_search_updow.py --code 005930 --start 2025-01-01 --end 2026-05-01 --tf 60 python3 -m kis_trader.backtest.param_search_updow --code 005930 --tf 3 --top 20 --save-top 옵션: --code 종목코드 (필수, 6자리) --start 시작일 YYYY-MM-DD (기본: 오늘-365일) --end 종료일 (기본: 오늘) --tf 분봉 단위 (기본: env UPDOW_TF_MIN, 없으면 60). 허용: holding_bot.KIWOOM_MINUTE_TICS --mode 탐색 모드 fast|full (기본 fast) --max-combos 백테 조합 상한 (기본: fast=1200, full=0=무제한) --min_trades 그리드 결과 필터 최소 거래 수 (기본 1) --top 상위 N개만 stdout + JSON (기본 30) --out 결과 JSON 경로 (미지정 시 kis_trader/backtest/results/updow_param___.json) --save-top 탐색 1위 ``apply_cfg`` → ``updow_stock_config`` INSERT (종목 code 필수) --apply N번째 결과 DB 적용 (기본 1, 총손익>0 일 때만). --from-file 시 최근 JSON --from-file --apply N 과 함께: 탐색 생략, 해당 종목 최근 updow_param__*.json 에서 적용 --env --from-file/--apply 시 env_config 전역 UPDOW_* 도 함께 갱신 (기본: updow_stock_config 만) --json --from-file 시 JSON 경로 직접 지정 (미지정 시 code·tf 기준 최근 파일) 예시: # 탐색 python3 kis_trader/backtest/param_search_updow.py \\ --code 005930 --tf 15 --start 2026-05-01 --end 2026-05-31 --mode fast # 저장된 JSON 1위 → updow_stock_config (종목별, code 필수) python3 kis_trader/backtest/param_search_updow.py --code 005930 --tf 15 --from-file --apply 1 """ 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) import holding_bot as hb # noqa: E402 from database import TradeDB # noqa: E402 from kis_trader.strategies import updow_buy as udb # noqa: E402 from kis_trader.strategies import updow_holding_cfg as uhc # noqa: E402 from kis_trader.strategies.updow_holding_cfg import ( # noqa: E402 ensure_updow_backtest_tables, load_updow_engine_cfg, migrate_updow_from_holding, set_updow_stock_config, ) from kis_trader.backtest.backtest_portfolio_common import resolve_portfolio_params # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(message)s") logger = logging.getLogger("param_search_updow") def _db() -> TradeDB: db = TradeDB() ensure_updow_backtest_tables(db) return db def _results_dir() -> str: d = os.path.join(HERE, "results") os.makedirs(d, exist_ok=True) return d def _results_dirs_for_read() -> List[str]: """읽기용: results → 홈 폴백.""" return [ os.path.join(HERE, "results"), os.path.join(os.path.expanduser("~"), ".kis_bot_search_results"), ] def _latest_updow_json(code: str, tf: Optional[int] = None) -> Optional[str]: """종목(·분봉) 기준 가장 최근 updow_param_*.json.""" code = str(code).strip() if tf is not None and int(tf) >= 0: prefix = f"updow_param_{code}_{int(tf)}m_" else: prefix = f"updow_param_{code}_" best_path, best_mtime = None, -1.0 for d in _results_dirs_for_read(): if not os.path.isdir(d): continue for fname in os.listdir(d): if not (fname.startswith(prefix) and fname.endswith(".json")): continue p = os.path.join(d, fname) try: m = os.path.getmtime(p) except OSError: continue if m > best_mtime: best_mtime = m best_path = p return best_path def _apply_updow_rank( data: Dict[str, Any], rank: int, *, apply_env: bool = False, allow_non_positive_pnl: bool = False, ) -> int: """JSON payload 또는 탐색 직후 top[] — rank(1-based) → updow_stock_config.""" from kis_trader.backtest.param_search_apply_snapshot import ( # noqa: WPS433 _apply_updow_row, _apply_updow_stock, ) top = data.get("top") or [] if rank < 1 or rank > len(top): logger.error("순번 %s 유효하지 않음 (1~%s)", rank, len(top)) return 5 row = top[rank - 1] pnl = int(row.get("total_pnl") or 0) if pnl <= 0 and not allow_non_positive_pnl: logger.warning( "total_pnl=%s ≤ 0 → DB 미적용. 적용하려면 --allow-non-positive-pnl", pnl, ) return 6 tf = int(data.get("tf") or 60) patch = _apply_updow_row(row, tf=tf) if not patch: logger.error("apply_cfg 가 비어 있습니다.") return 7 code_h = str(data.get("code") or "").strip() if apply_env: db = _db() try: latest = db.get_latest_env() snap = dict(latest["snapshot"]) if latest else {} snap.update(patch) eid = db.insert_env_snapshot(snap) if eid: logger.info("UPDOW env_config INSERT id=%s keys=%s", eid, list(patch.keys())) else: logger.error("env_config INSERT 실패") return 8 finally: db.close() if code_h: _apply_updow_stock(code_h, row, tf=tf, dry_run=False) else: logger.warning("JSON 에 code 없음 → updow_stock_config 건너뜀") return 9 return 0 def _apply_from_latest_json( code: str, rank: int, *, tf: Optional[int] = None, json_path: str = "", apply_env: bool = False, allow_non_positive_pnl: bool = False, ) -> int: """code(·tf) 기준 최근 JSON에서 rank번째 apply.""" path = json_path.strip() if not path: path = _latest_updow_json(code, tf) if not path or not os.path.isfile(path): hint = f"updow_param_{code}_" if tf is not None and int(tf) >= 0: hint = f"updow_param_{code}_{int(tf)}m_" logger.error("JSON 없음: %s*.json (results 또는 ~/.kis_bot_search_results)", hint) return 2 logger.info("📂 %s 에서 %s번째 적용", path, rank) try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) except Exception as e: logger.error("JSON 로드 실패: %s", e) return 3 json_code = str(data.get("code") or "").strip() if json_code and json_code != str(code).strip(): logger.warning( "CLI --code=%s 와 JSON code=%s 불일치 → JSON 종목(%s) 기준 적용", code, json_code, json_code, ) code = json_code data["code"] = code return _apply_updow_rank( data, rank, apply_env=apply_env, allow_non_positive_pnl=allow_non_positive_pnl, ) def main(argv: List[str] | None = None) -> int: today = datetime.now().strftime("%Y-%m-%d") default_start = (datetime.now() - timedelta(days=365)).strftime("%Y-%m-%d") ap = argparse.ArgumentParser(description="Updow 파라미터 그리드 탐색 (holding_min_candles + env_config)") ap.add_argument("--code", required=True, help="6자리 종목코드 (--from-file 시 최근 JSON 필터용)") 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="분봉 (기본 -1 = env UPDOW_TF_MIN, 없으면 60)") ap.add_argument("--mode", choices=["fast", "full"], default="fast", help="탐색 모드 (fast=경량, full=전체)") ap.add_argument("--max-combos", type=int, default=-1, help="백테 조합 상한 (0=무제한, 기본 fast=1200/full=0)") ap.add_argument("--min_trades", type=int, default=1, help="최소 거래 건수 필터") ap.add_argument("--top", type=int, default=30, help="상위 N개 출력·저장") ap.add_argument("--out", default="", help="JSON 저장 경로 (선택)") ap.add_argument("--save-top", action="store_true", help="탐색 1위를 updow_stock_config에 INSERT") ap.add_argument("--apply", nargs="?", const=1, type=int, default=None, metavar="N", help="N번째 결과 DB 적용 (기본 1). --from-file 시 최근 JSON에서 적용") ap.add_argument("--from-file", action="store_true", help="--apply N 과 함께: 탐색 생략, updow_param__*.json 최근 파일에서 적용") ap.add_argument("--json", default="", help="--from-file 시 JSON 경로 직접 지정 (미지정=최근 파일)") ap.add_argument("--env", action="store_true", help="적용 시 env_config 전역 UPDOW_* 도 함께 갱신 (기본: updow_stock_config 만)") ap.add_argument("--allow-non-positive-pnl", action="store_true", help="total_pnl≤0 이더라도 적용") args = ap.parse_args(argv) code = str(args.code).strip() if len(code) != 6 or not code.isdigit(): logger.error("--code 는 6자리 숫자여야 합니다: %r", code) return 1 if args.from_file: if args.apply is None or int(args.apply) < 1: logger.error("--from-file 는 --apply N 과 함께 사용하세요 (예: --from-file --apply 1)") return 1 tf_filter = int(args.tf) if int(args.tf) >= 0 else None return _apply_from_latest_json( code, int(args.apply), tf=tf_filter, json_path=args.json, apply_env=bool(args.env), allow_non_positive_pnl=bool(args.allow_non_positive_pnl), ) db = _db() try: latest = db.get_latest_env() snap = dict(latest["snapshot"]) if latest else {} tf_default = udb.read_updow_tf_min(snap) tf = int(args.tf) if int(args.tf) >= 0 else tf_default if tf not in hb.KIWOOM_MINUTE_TICS: logger.error("tf=%s 는 허용 목록에 없습니다: %s", tf, hb.KIWOOM_MINUTE_TICS) return 1 candles = hb.get_stored_min_candles(db, code, args.start, args.end, tf_min=tf) if not candles: logger.error("%s 분봉 데이터 없음 (%s ~ %s). 키움 분봉 수집 후 재시도.", tf, args.start, args.end) return 2 if len(candles) < 5: logger.error("봉 부족: %s개 (최소 5)", len(candles)) return 3 base_cfg = udb.cfg_from_env_snapshot(snap) migrate_updow_from_holding(db, code, base_cfg) base_cfg = load_updow_engine_cfg(db, code, base_cfg) proxy = str((snap or {}).get("UPDOW_KOSPI_1MIN_PROXY_CODE") or "069500").strip() regime_candles = hb.get_stored_min_candles(db, proxy, args.start, args.end, tf_min=1) or [] mode = str(args.mode or "fast").strip().lower() if mode == "full": grid = udb.default_param_grid(snap) max_combos = 0 if args.max_combos < 0 else max(0, int(args.max_combos)) else: grid = udb.default_param_grid_web_fast(snap) max_combos = 1200 if args.max_combos < 0 else max(0, int(args.max_combos)) results, meta = udb.run_param_search_updow( candles, grid=grid, min_trades=max(0, int(args.min_trades)), base_cfg=base_cfg, env_snapshot=snap, regime_candles=regime_candles or None, stock_tf_min=tf, max_combos=max_combos, ) meta["regime_proxy_code"] = proxy meta["regime_1min_bars_loaded"] = len(regime_candles) meta["search_mode"] = mode meta["combo_cap"] = max_combos meta["tf"] = tf meta["code"] = code meta["start"] = args.start meta["end"] = args.end meta["candle_count"] = len(candles) top_n = max(1, int(args.top)) top = results[:top_n] updow_portfolio = resolve_portfolio_params(snap, strategy="UPDOW") meta["portfolio"] = updow_portfolio meta["slot_money"] = float(updow_portfolio["slot_money"]) meta["max_stocks"] = int(updow_portfolio["max_stocks"]) meta["total_budget_krw"] = float(updow_portfolio["total_budget_krw"]) tb = float(updow_portfolio["total_budget_krw"]) for row in top: pnl = row.get("total_pnl") if pnl is not None and tb > 0: row["bot_pct"] = round(float(pnl) / tb * 100, 2) logger.info( "=== Updow param search %s tf=%s 봉=%s ===\n그리드 조합=%s 백테=%s 통과=%s (min_trades=%s)", code, tf, len(candles), meta.get("cartesian_product"), meta.get("backtests_run"), meta.get("passed"), args.min_trades, ) hints = meta.get("grid_axis_hints") or {} if hints: logger.info("--- 그리드 축(탐색 중 바뀌는 값) ---") for k, txt in hints.items(): logger.info(" [%s] %s", k, txt) for i, row in enumerate(top, 1): p = row.get("params") or {} logger.info( "#%s pnl=%s win=%s%% trades=%s pf=%s hold=%s params=%s", i, row.get("total_pnl"), row.get("win_rate"), row.get("total_trades"), row.get("pf"), row.get("avg_hold"), p, ) if args.save_top: if not results: logger.error("--save-top 지정했으나 통과한 결과가 없습니다.") return 4 ac = results[0].get("apply_cfg") or {} nm = code m = uhc.get_updow_stock_meta(db, code) if m and m.get("name"): nm = str(m["name"]).strip() or code set_updow_stock_config(db, code, nm, ac, tf_min=tf) logger.info( "updow_stock_config 저장 (%s, tf=%s, max_hold=%s)", code, tf, int(float(ac.get("max_hold_bars", 16))), ) payload: Dict[str, Any] = { "code": code, "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"updow_param_{code}_{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 and int(args.apply) >= 1: rc = _apply_updow_rank( payload, int(args.apply), apply_env=bool(args.env), allow_non_positive_pnl=bool(args.allow_non_positive_pnl), ) if rc != 0: return rc return 0 finally: db.close() if __name__ == "__main__": raise SystemExit(main())