Hwang
61c72a8a4c
변경 사항 ---- - _test_kiwoom_condition_list.py: 키움 웹소켓 조건검색 '목록조회' 기능을 단독으로 테스트하는 스크립트 추가 - _test_kiwoom_condition_realtime.py: 'momentum' 조건식을 실시간으로 등록하고 초기 매칭 종목 리스트 및 실시간 편입/이탈을 수신하는 테스트 스크립트 추가 - _verify_columnar_bitid.py, _verify_shared_e2e_breakout.py, _verify_shared_e2e.py: 공유 메모리 및 dict 간의 데이터 일관성을 검증하는 테스트 추가 영향 ---- - 신규 테스트 스크립트 추가로 키움 웹소켓 API의 기능 검증 및 안정성을 높임 - 기존 기능에 대한 영향 없음 Co-authored-by: Cursor <cursoragent@cursor.com>
280 lines
10 KiB
Python
280 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
kis_trader/backtest/param_search_updow_us.py — UPDOW 해외(US) 종목 파라미터 Grid Search
|
||
======================================================================================
|
||
- 국내용 ``param_search_updow.py`` 와 동일 엔진(``run_param_search_updow``)
|
||
- **차이**: 영문 티커(QQQM 등), ``holding_min_candles`` 해외 분봉(KIS), **코스피 레짐(069500) 미사용**, 어깨 그리드는 국내와 동일
|
||
|
||
실행:
|
||
cd /home/hoon/kis_bot
|
||
python3 kis_trader/backtest/param_search_updow_us.py --symbol QQQM --start 2026-05-01 --end 2026-05-29 --tf 15
|
||
python3 -m kis_trader.backtest.param_search_updow_us --symbol QQQM --exchange NASD --mode fast --save-top
|
||
|
||
옵션:
|
||
--symbol 해외 티커 (필수, 예: QQQM). --code 와 동일
|
||
--exchange 거래소 (기본 NASD → KIS 시세 API는 NAS 로 변환)
|
||
--start / --end / --tf / --mode / --max-combos / --min_trades / --top / --out / --save-top
|
||
→ 의미는 국내 CLI 와 동일
|
||
|
||
탐색 규모(기본):
|
||
fast : US 전용 경량 그리드 = 2,880조합 (limit_atr_mult 0.0~1.5 포함, 레짐 OFF).
|
||
1회 백테 ≈ 0.25초(60분 400봉) → **약 10~15분**. 캡은 env UPDOW_US_FAST_MAX_COMBOS.
|
||
※ 저변동 지수 ETF는 깊은 지정가(ATR×1.5)가 거의 체결 안 되므로 낮은 limit_atr_mult 가 핵심.
|
||
full : 54,000조합 (≈ 수 시간) — 정밀 탐색용.
|
||
|
||
사전 준비:
|
||
웹 UPDOW 탭 또는 API 로 해당 심볼 분봉을 ``holding_min_candles`` 에 수집해 두세요.
|
||
(market_type=US, exchange=NASD, symbol=QQQM)
|
||
"""
|
||
|
||
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, Tuple
|
||
|
||
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,
|
||
set_updow_stock_config,
|
||
)
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||
logger = logging.getLogger("param_search_updow_us")
|
||
|
||
|
||
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 _parse_symbol_exchange(args: argparse.Namespace) -> Tuple[str, str, str, str]:
|
||
"""(code, symbol, market_type, exchange) — DB·holding_min_candles 키는 code=symbol."""
|
||
raw = str(args.symbol or args.code or "").strip().upper()
|
||
if not raw:
|
||
raise ValueError("--symbol (또는 --code) 필수 (예: QQQM)")
|
||
ex_in = str(args.exchange or "NASD").strip().upper()
|
||
mt_in = str(args.market_type or "US").strip().upper()
|
||
mt, ex, sym = uhc.resolve_market_meta(raw, mt_in, ex_in, raw)
|
||
if mt != "US":
|
||
raise ValueError(
|
||
f"해외 전용 CLI 입니다. symbol={raw} → market_type={mt} (US 필요). "
|
||
"국내 6자리는 param_search_updow.py 를 사용하세요."
|
||
)
|
||
code = sym
|
||
return code, sym, mt, ex
|
||
|
||
|
||
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 해외(US) 파라미터 그리드 탐색 (레짐 OFF, holding_min_candles)"
|
||
)
|
||
ap.add_argument("--symbol", default="", help="해외 티커 (예: QQQM)")
|
||
ap.add_argument("--code", default="", help="--symbol 과 동일 (호환)")
|
||
ap.add_argument("--exchange", default="NASD", help="거래소 코드 (NASD/NYSE/AMEX)")
|
||
ap.add_argument("--market-type", default="US", dest="market_type", help="시장 (기본 US)")
|
||
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)")
|
||
ap.add_argument("--mode", choices=["fast", "full"], default="fast", help="fast|full")
|
||
ap.add_argument("--max-combos", type=int, default=-1, help="조합 상한 (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")
|
||
args = ap.parse_args(argv)
|
||
|
||
try:
|
||
code, symbol, market_type, exchange = _parse_symbol_exchange(args)
|
||
except ValueError as e:
|
||
logger.error("%s", e)
|
||
return 1
|
||
|
||
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분봉 없음 (%s ~ %s).\n"
|
||
" → 웹 UPDOW 탭에서 [US:%s] %s 분봉 수집 후 재시도.",
|
||
symbol,
|
||
exchange,
|
||
tf,
|
||
args.start,
|
||
args.end,
|
||
exchange,
|
||
symbol,
|
||
)
|
||
return 2
|
||
if len(candles) < 5:
|
||
logger.error("봉 부족: %s개 (최소 5)", len(candles))
|
||
return 3
|
||
|
||
base_cfg = udb.cfg_from_env_snapshot(snap)
|
||
base_cfg = load_updow_engine_cfg(db, code, base_cfg)
|
||
# 해외: 코스피 레짐(069500) 미적용 — 종목 설정에 60이 있어도 탐색 베이스는 OFF
|
||
base_cfg["regime_ma_bars"] = 0.0
|
||
base_cfg["regime_ma_ease_pct"] = 0.0
|
||
|
||
mode = str(args.mode or "fast").strip().lower()
|
||
if mode == "full":
|
||
grid = udb.default_param_grid_us(snap)
|
||
max_combos = 0 if args.max_combos < 0 else max(0, int(args.max_combos))
|
||
else:
|
||
grid = udb.default_param_grid_us_fast(snap)
|
||
# fast 기본 캡 = env UPDOW_US_FAST_MAX_COMBOS (기본 2880 ≈ 10~15분)
|
||
max_combos = (
|
||
udb.read_us_fast_max_combos(snap)
|
||
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=None,
|
||
stock_tf_min=tf,
|
||
max_combos=max_combos,
|
||
)
|
||
|
||
meta["market_type"] = market_type
|
||
meta["exchange"] = exchange
|
||
meta["symbol"] = symbol
|
||
meta["regime_proxy_code"] = None
|
||
meta["regime_1min_bars_loaded"] = 0
|
||
meta["regime_note"] = "US: KOSPI proxy regime disabled"
|
||
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]
|
||
|
||
logger.info(
|
||
"=== UPDOW US param search %s [%s:%s] tf=%s 봉=%s ===\n"
|
||
"그리드 조합=%s 백테=%s 통과=%s (min_trades=%s, 레짐=OFF)",
|
||
symbol,
|
||
market_type,
|
||
exchange,
|
||
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 {}
|
||
ac["regime_ma_bars"] = 0.0
|
||
ac["regime_ma_ease_pct"] = 0.0
|
||
nm = symbol
|
||
m = uhc.get_updow_stock_meta(db, code)
|
||
if m and m.get("name"):
|
||
nm = str(m["name"]).strip() or symbol
|
||
set_updow_stock_config(
|
||
db,
|
||
code,
|
||
nm,
|
||
ac,
|
||
tf_min=tf,
|
||
market_type=market_type,
|
||
exchange=exchange,
|
||
symbol=symbol,
|
||
)
|
||
logger.info(
|
||
"updow_stock_config 저장 (%s US:%s tf=%s max_hold=%s)",
|
||
code,
|
||
exchange,
|
||
tf,
|
||
int(float(ac.get("max_hold_bars", 16))),
|
||
)
|
||
|
||
payload: Dict[str, Any] = {
|
||
"code": code,
|
||
"symbol": symbol,
|
||
"market_type": market_type,
|
||
"exchange": exchange,
|
||
"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_us_{symbol}_{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)
|
||
return 0
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|