hoon/kis_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
feat/kiwoom-ws-validation
kis_bot/kis_trader/backtest/tail_param_search.py
Hwang 336d637b72 feat(param-search): Add new evaluation functions for breakout, momentum, and tail parameter combinations 
Changes:
- Added `apply_params_to_db` function to streamline parameter application to the database.
- Introduced `evaluate_breakout_param_combo`, `evaluate_momentum_param_combo`, and `evaluate_tail_param_combo` functions to enhance the evaluation of parameter combinations for respective strategies.
- Updated `requirements.txt` to include `optuna==4.2.1` for improved optimization capabilities.

Impact:
- These additions improve the modularity and efficiency of parameter evaluations across different trading strategies, facilitating better optimization and backtesting processes.
2026-07-06 02:18:14 +09:00

1451 lines
69 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
kis_trader/backtest/tail_param_search.py — 꼬리잡기 백테스트 파라미터 자동 탐색 (Grid Search)
==============================================================================================
tail_engine 을 직접 임포트해 run_tail_backtest 호출. 기본값은 DB(env_config) 단일 소스.
실매매·백테스트·파라서치가 동일한 env 값을 사용해 결과 예측 가능.
[V3 통합]: MA20, 피뢰침, ATR 동적 배수, 리스크 비율 등 고급 방어 로직 변수 탐색 추가.
[성능 최적화]: 멀티프로세싱(ProcessPool) 및 Heapq 기반 메모리 최적화 적용.
실행:
cd /home/hoon/kis_bot
python3 kis_trader/backtest/tail_param_search.py
# 또는
python3 -m kis_trader.backtest.tail_param_search --mode full --apply
옵션:
--start 시작일 (기본: 오늘-7일)
--end 종료일 (기본: 오늘)
--mode 탐색 모드: fast / coarse / fine / full / massive (기본: fast ≈768조합·~10–20분·포트폴리오)
--timeframe ws_candles 분봉 (3·5·15·60, 기본 3 — backtest_web 꼬리 탭과 동일)
--slot-money 1회투자금(원). 미지정 시 DB (웹 tl_slot)
--max-stocks 동시보유 종목. 미지정 시 DB SHORT_MAX_STOCKS
--total-budget 총 운용한도(원). 0/미지정 시 동시×1회투자 (웹 tl_total_budget)
--top 상위 N개 출력·JSON 저장 (기본: 5000)
--min_trades 최소 거래 건수 필터 (기본: 1)
--min_win_rate 승률 하한 (기본: 40.0).
--min_pf Profit Factor 하한 (기본: 1.0).
--max-combos 백테 조합 상한 (기본 5000, 0=무제한·fine/full 비추)
--apply [N] 1위(또는 N위) 결과를 DB에 적용. N 생략 시 1.
--from-file --apply N 과 함께 사용 시, 최근 결과 JSON에서 N번째 적용 (탐색 생략).
⚠️ 그리드 축 env 오버라이드 (콤마 구분, DB/env_config 또는 셸 export):
TAIL_GRID_{FAST|COARSE|FINE|FULL|MASSIVE}_{축이름}
예) TAIL_GRID_COARSE_MAX_DAILY_CHG=15,20,25,30
예) TAIL_GRID_FAST_LIMIT_ATR_MULT=1.5,2.0
⚠️ 파라서치 진입 모드 (체결 수 확보): TAIL_PARAM_SEARCH_ENTRY_MODE=align|limit_atr
미설정 시 DB TAIL_ENTRY_MODE 사용. limit_atr 은 거래 수가 적을 수 있음.
UPDOW(직전봉 하락·다음봉 시가 매수, holding_min_candles):
kis_trader/backtest/param_search_updow.py
위치 이관 (2026-04 기준):
backtest_scalping/tail_param_search.py → kis_trader/backtest/tail_param_search.py
- ROOT = kis_bot 프로젝트 루트 (__file__ 기준 3단계 위)
- 결과 저장: kis_trader/backtest/results/search_tail_{mode}_YYYYMMDD_HHMMSS.json
(``search_momentum_*`` / ``search_breakout_*`` 와 동일 패턴 · 매 실행 신규 파일)
- 쓰기 실패 시: ~/.kis_bot_search_results/ 에 동일 파일명 폴백
- --from-file /--apply-ai: 최신 search_tail_*.json 우선 → 구 tail_search_*.json →
레거시 tail_param_result.json (kis_trader/backtest·backtest_scalping)
"""
import sys
import os
import time
import json
import signal
import logging
import argparse
import itertools
import heapq
from datetime import datetime, timedelta
from concurrent.futures import as_completed
from typing import List, Dict, Any, Tuple, Optional, Set
# 프로젝트 루트 경로 추가 (database, tail_engine 등 임포트용)
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)
if HERE not in sys.path:
sys.path.insert(0, HERE)
from database import TradeDB
from kis_trader.engine import tail_engine as te
from kis_trader.backtest import tail_backtest_common as tbc
from kis_trader.backtest.backtest_portfolio_common import (
merge_param_search_apply_source,
portfolio_env_patch,
session_env_patch,
)
from kis_trader.backtest.param_search_cli_common import (
add_portfolio_cli_args,
add_search_filter_cli_args,
apply_session_to_fixed,
combo_passes_search_filters,
format_session_hm,
search_json_meta,
)
from kis_trader.backtest.param_search_pool import (
ParamSearchSharedPayload,
ParamSearchProgressETA,
assert_parent_alive,
cap_combos_uniform,
cap_combos_uniform_lazy,
iter_pool_chunk_results,
managed_process_pool,
param_search_chunk_plan,
param_search_worker_budget_line,
try_acquire_run_lock,
worker_shared_get,
)
from kis_trader.engine.indicator_cache import attach_indicator_caches_to_params
from kis_trader.utils.env import get_env_bool, get_env_float, get_env_from_db, get_env_int
logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger("tail_param_search")
DEFAULT_MAX_COMBOS = 5000 # coarse 그리드 전체(~3500) + fine/full 상한
FAST_MAX_COMBOS = 256 # fast 16축 → 균등샘플 (각 축 값별 marginal 스캔 · ~15–30초)
TAIL_BACKTEST_STRATEGY_ID = tbc.TAIL_STRATEGY_ID # backtest_web 과 동일
# ─────────────────────────────────────────────────────────────────────────────
# 결과 디렉터리 (신규 위치 우선, 구 경로 fallback)
# ─────────────────────────────────────────────────────────────────────────────
def _results_dir_for_write() -> str:
d = os.path.join(HERE, "results")
os.makedirs(d, exist_ok=True)
return d
def _results_dirs_for_read() -> List[str]:
"""타임스탬프 JSON 탐색 — 신규 results · 홈 폴백 · 구 backtest_scalping/results."""
seen: Set[str] = set()
out: List[str] = []
for d in (
os.path.join(HERE, "results"),
os.path.join(os.path.expanduser("~"), ".kis_bot_search_results"),
os.path.join(ROOT, "backtest_scalping", "results"),
):
if d and d not in seen:
seen.add(d)
out.append(d)
return out
def _legacy_tail_param_fixed_paths() -> List[str]:
"""구버전 고정 파일명 (--from-file 레거시 호환)."""
return [
os.path.join(HERE, "tail_param_result.json"),
os.path.join(ROOT, "backtest_scalping", "tail_param_result.json"),
]
def _latest_tail_json_path() -> Optional[str]:
"""가장 최근 수정 시각인 search_tail_*.json 또는 tail_search_*.json 경로."""
prefixes = ("search_tail_", "tail_search_")
best_path: Optional[str] = None
best_mtime = -1.0
for d in _results_dirs_for_read():
if not os.path.isdir(d):
continue
try:
names = os.listdir(d)
except OSError:
continue
for f in names:
if not f.endswith(".json"):
continue
if not any(f.startswith(p) for p in prefixes):
continue
p = os.path.join(d, f)
try:
m = os.path.getmtime(p)
except OSError:
continue
if m > best_mtime:
best_mtime = m
best_path = p
return best_path
def _find_tail_result_json() -> Optional[str]:
"""--from-file / --apply / apply_ai 가 읽을 JSON (최신 타임스탬프 파일 우선)."""
ts_path = _latest_tail_json_path()
if ts_path:
return ts_path
for p in _legacy_tail_param_fixed_paths():
if os.path.isfile(p):
return p
return None
def _parse_csv_floats(env_key: str, fallback: List[float]) -> List[float]:
"""env/DB 콤마 구분 float 리스트 → 그리드 축."""
raw = get_env_from_db(env_key, "")
if not raw or str(raw).strip() in ("", "None"):
return list(fallback)
out: List[float] = []
for part in str(raw).split(","):
part = part.strip().split("#", 1)[0].strip()
if not part:
continue
try:
out.append(float(part))
except ValueError:
continue
return out if out else list(fallback)
def _parse_csv_strings(env_key: str, fallback: List[str]) -> List[str]:
"""env 콤마 문자열 → 그리드 축 (entry_mode 등)."""
raw = get_env_from_db(env_key, "")
if raw in (None, "", "None"):
return list(fallback)
out: List[str] = []
for part in str(raw).split(","):
part = part.strip().split("#", 1)[0].strip().lower()
if part:
out.append(part)
return out if out else list(fallback)
def _parse_semi_strings(env_key: str, fallback: List[str]) -> List[str]:
"""env 세미콜론(;) 구분 문자열 → 그리드 축.
tier 프리셋처럼 값 자체에 콤마가 들어가는 경우용(예: '30000:50,100000:35').
예) TAIL_GRID_FAST_TRAIL_TIERS='off;20000:40,50000:30;30000:50,100000:35'
"""
raw = get_env_from_db(env_key, "")
if raw in (None, "", "None"):
return list(fallback)
out: List[str] = []
for part in str(raw).split(";"):
part = part.strip()
if part:
out.append(part)
return out if out else list(fallback)
def _parse_csv_bools(env_key: str, fallback: List[bool]) -> List[bool]:
"""env 콤마 bool 리스트 → 그리드 축 (패턴 ON/OFF 등). 'true/1/on' = True."""
raw = get_env_from_db(env_key, "")
if raw in (None, "", "None"):
return list(fallback)
out: List[bool] = []
for part in str(raw).split(","):
part = part.strip().split("#", 1)[0].strip().lower()
if not part:
continue
out.append(part in ("1", "true", "t", "y", "yes", "on"))
return out if out else list(fallback)
def _format_param_cell(value: Any, col_w: int) -> str:
"""그리드 표시: 문자열·bool은 그대로, 숫자는 .4g."""
if isinstance(value, bool):
return f"{str(value):>{col_w}}"
if isinstance(value, str):
return f"{value:>{col_w}}"
return f"{value:>{col_w}.4g}"
def _parse_csv_ints(env_key: str, fallback: List[int]) -> List[int]:
raw = get_env_from_db(env_key, "")
if not raw or str(raw).strip() in ("", "None"):
return list(fallback)
out: List[int] = []
for part in str(raw).split(","):
part = part.strip().split("#", 1)[0].strip()
if not part:
continue
try:
out.append(int(float(part)))
except ValueError:
continue
return out if out else list(fallback)
def _snap_float(snap: Optional[Dict[str, Any]], key: str, default: float) -> float:
"""env_config 스냅샷 → float (그리드 끝값)."""
if snap and key in snap and snap[key] is not None:
raw = str(snap[key]).split("#", 1)[0].strip()
if raw:
try:
return float(raw)
except (TypeError, ValueError):
pass
try:
return float(get_env_from_db(key, default) or default)
except (TypeError, ValueError):
return float(default)
def _tail_grids(mode: str) -> Dict[str, List[Any]]:
"""
꼬리 파라서치 그리드 — TAIL_GRID_{MODE}_* env 로 축 오버라이드 (돌파 BREAKOUT_GRID_* 와 동일 패턴).
fast: entry_mode(align/limit_atr) + limit_atr_mult — 틱 OFF·3분봉 OHLC (모멘텀과 동일).
"""
m = (mode or "fast").strip().lower()
p = f"TAIL_GRID_{m.upper()}_"
if m == "fast":
# 16축 — tail_vol_mult×4, atr_tp_max×4 + 청산방식(ratchet)·RSI·반전패턴 (균등샘플 cap)
return {
"entry_mode": _parse_csv_strings(f"{p}ENTRY_MODE", ["align", "limit_atr"]),
"max_daily_change": _parse_csv_floats(f"{p}MAX_DAILY_CHG", [15.0, 20.0, 25.0, 30.0]),
"min_drop_rate": _parse_csv_floats(f"{p}MIN_DROP_RATE", [0.02, 0.03]),
"min_recovery_ratio": _parse_csv_floats(f"{p}MIN_RECOVERY_RATIO", [0.25, 0.4, 0.5]),
"tail_ratio_min": _parse_csv_floats(f"{p}TAIL_RATIO_MIN", [1.0, 1.5]),
"max_rec_3m": _parse_csv_floats(f"{p}MAX_REC_3M", [0.70, 0.80, 0.90]),
"shoulder_min_high": _parse_csv_floats(f"{p}SHOULDER_MIN_HIGH", [0.003, 0.005]),
"shoulder_cut_pct": _parse_csv_floats(f"{p}SHOULDER_CUT_PCT", [0.002, 0.003]),
"stop_atr_mult": _parse_csv_floats(f"{p}STOP_ATR_MULT", [1.5, 2.0]),
"target_atr_mult": _parse_csv_floats(f"{p}TARGET_ATR_MULT", [1.5, 2.0]),
"atr_tp_max_pct": _parse_csv_floats(f"{p}ATR_TP_MAX_PCT", [1.0, 1.5, 2.0, 3.0]),
"tail_vol_mult": _parse_csv_floats(f"{p}VOL_MULT", [0.0, 1.5, 2.0, 2.5]),
"tail_vol_win": _parse_csv_floats(f"{p}VOL_WIN", [5.0]),
"limit_atr_mult": _parse_csv_floats(f"{p}LIMIT_ATR_MULT", [1.0, 1.5, 2.0]),
# ── 청산방식 탐색: ratchet_tiers ──────────────────────────────
# off = 단일 어깨컷(shoulder_*) 사용 / 나머지 = 다단 래칫 트레일.
# "off" 포함으로 shoulder_min_high·shoulder_cut_pct 축이 비로소
# 의미를 갖는다(래칫 ON 조합에선 어깨컷은 엔진상 무시됨).
"ratchet_tiers": _parse_csv_strings(
f"{p}RATCHET_TIERS",
["off", "0.5:0.3,1.0:0.25,2.0:0.2", "1.0:0.4,2.0:0.3"],
),
# ── 과매매 제어: 종목당 일일 매수횟수 · 재진입 콜다운(분) ───────
# 실매 정합 핵심 — 현재 DB(20회·1분)는 같은 종목 연타로 PF 왜곡.
# 파람서치가 "적게 사는 게 PF↑"를 찾으면 DB에 반영된다.
"max_daily": _parse_csv_ints(f"{p}MAX_DAILY", [2, 3, 5]),
"cooldown_min": _parse_csv_floats(f"{p}COOLDOWN_MIN", [3.0, 5.0, 15.0]),
# ── 진입 캔들 등락 필터(%) — 신호봉이 이 구간 안일 때만 매수 ─────
# bar_chg_min: 하한(과대 급락 제외) / bar_chg_max: 상한(최소 하락폭)
"bar_chg_min_pct": _parse_csv_floats(f"{p}BAR_CHG_MIN", [-10.0, -7.0]),
"bar_chg_max_pct": _parse_csv_floats(f"{p}BAR_CHG_MAX", [-1.0, -1.5, -2.0]),
# ── RSI 과열 진입 거부 기준(%) ────────────────────────────────
"rsi_threshold": _parse_csv_floats(f"{p}RSI_THRESHOLD", [78.0, 82.0, 85.0]),
# ── 반전 캔들 패턴 ON/OFF (hammer 는 base 고정, 추가 패턴만 탐색) ─
# pin: 핀바(긴 아래꼬리 단일봉, 꼬리잡기와 직결)
# engulfing/piercing: 2봉 반전 — 신호 다양성·표본 확대용
"pattern_pin": _parse_csv_bools(f"{p}PATTERN_PIN", [False, True]),
"pattern_engulfing": _parse_csv_bools(f"{p}PATTERN_ENGULFING", [False, True]),
"pattern_piercing": _parse_csv_bools(f"{p}PATTERN_PIERCING", [False, True]),
# ── 호가필터(2단계 조인트) 축 — 기본 단일값=동작 불변 ──────────
# max_spread_pct : 스프레드 상한(%) ↑완화
# min_bid_ask_ratio : 총매수/총매도 잔량비 하한(꼬리지지) ↓완화
"max_spread_pct": _parse_csv_floats(f"{p}MAX_SPREAD_PCT", [0.45]),
"min_bid_ask_ratio": _parse_csv_floats(f"{p}MIN_BID_ASK_RATIO", [0.85]),
# ── 당일 누적손익 트레일 익절(%) — 0=OFF(기본·동작 불변) ─────────
# 고점 대비 N% 되돌리면 그날 신규매수 중단(실매 daily_profit_halt 동일).
# 일(日) 단위 이벤트라 표본=거래일 수 → 긴 기간에서만 유의미(강건 구간 검증).
# 스윕하려면 env TAIL_GRID_FAST_TRAIL_DROP_PCT=0,30,40 + DB
# DAILY_PROFIT_TRAIL_ARM_KRW(발동 최소수익) 설정 필요.
# ── 당일 누적손익 다단계 트레일(ratchet 철학) — 벌수록 타이트 ──────
# 'off'=트레일 미사용(기준군) / '수익원:컷%,…'=다단 trail.
# peak 가 높은 구간에 들수록 되돌림 컷%를 좁혀 이익 보존(워런 버핏식).
# tier 가 켜지면 단일 trail_drop_pct·arm 은 무시됨. 일(日)단위 표본.
# env TAIL_GRID_FAST_TRAIL_TIERS 로 세미콜론(;) 구분 override.
"trail_tiers": _parse_semi_strings(f"{p}TRAIL_TIERS", [
"off",
"5000:40,10000:30,20000:20", # 보수: 일찍·타이트하게 사수
"30000:50,100000:35,300000:25", # 중도
"50000:60,200000:45,500000:30", # 공격: 많이 벌 때만 강하게 사수
]),
# 단일 트레일(tier='off' 일 때만 의미) — 하위호환·기준 비교용 단일값.
"trail_drop_pct": _parse_csv_floats(f"{p}TRAIL_DROP_PCT", [0.0]),
"trail_arm_krw": _parse_csv_floats(f"{p}TRAIL_ARM_KRW", [10000.0]),
}
if m == "coarse":
return {
"max_daily_change": _parse_csv_floats(f"{p}MAX_DAILY_CHG", [15.0, 20.0, 25.0, 30.0]),
"min_drop_rate": _parse_csv_floats(f"{p}MIN_DROP_RATE", [0.02, 0.03]),
"min_recovery_ratio": _parse_csv_floats(f"{p}MIN_RECOVERY_RATIO", [0.4, 0.45, 0.5]),
"tail_ratio_min": _parse_csv_floats(f"{p}TAIL_RATIO_MIN", [1.0, 1.5]),
"max_rec_3m": _parse_csv_floats(f"{p}MAX_REC_3M", [0.85, 0.9]),
"shoulder_min_high": _parse_csv_floats(f"{p}SHOULDER_MIN_HIGH", [0.002, 0.003, 0.005]),
"shoulder_cut_pct": _parse_csv_floats(f"{p}SHOULDER_CUT_PCT", [0.002, 0.003]),
"stop_atr_mult": _parse_csv_floats(f"{p}STOP_ATR_MULT", [1.5, 2.0]),
"target_atr_mult": _parse_csv_floats(f"{p}TARGET_ATR_MULT", [1.5, 2.0]),
"atr_tp_max_pct": _parse_csv_floats(f"{p}ATR_TP_MAX_PCT", [0.8, 1.0, 1.2]),
"max_loss_krw": _parse_csv_floats(f"{p}MAX_LOSS_KRW", [100000.0, 200000.0]),
"limit_atr_mult": _parse_csv_floats(f"{p}LIMIT_ATR_MULT", [1.0, 1.5, 2.0]),
}
# fine / full / massive — 기존 PARAM_GRIDS 폴백 + env 축만 덮어쓰기
base = dict(PARAM_GRIDS.get(m) or {})
if not base:
return _tail_grids("fast")
overrides = {
"max_daily_change": _parse_csv_floats(f"{p}MAX_DAILY_CHG", base.get("max_daily_change", [20.0, 25.0])),
"limit_atr_mult": _parse_csv_floats(f"{p}LIMIT_ATR_MULT", [1.0, 1.5, 2.0]),
}
for axis, env_suffix, fb_key in (
("min_drop_rate", "MIN_DROP_RATE", "min_drop_rate"),
("min_recovery_ratio", "MIN_RECOVERY_RATIO", "min_recovery_ratio"),
("tail_ratio_min", "TAIL_RATIO_MIN", "tail_ratio_min"),
("max_rec_3m", "MAX_REC_3M", "max_rec_3m"),
("shoulder_min_high", "SHOULDER_MIN_HIGH", "shoulder_min_high"),
("shoulder_cut_pct", "SHOULDER_CUT_PCT", "shoulder_cut_pct"),
("stop_atr_mult", "STOP_ATR_MULT", "stop_atr_mult"),
("target_atr_mult", "TARGET_ATR_MULT", "target_atr_mult"),
):
if fb_key in base or env_suffix:
overrides[axis] = _parse_csv_floats(
f"{p}{env_suffix}",
base.get(fb_key, overrides.get(axis, [0.03])),
)
merged = dict(base)
merged.update(overrides)
return merged
# ─────────────────────────────────────────────────────────────────────────────
# 파라미터 그리드 정의 (fine/full/massive 폴백 — fast/coarse 는 _tail_grids)
# ─────────────────────────────────────────────────────────────────────────────
PARAM_GRIDS = {
# ─────────────────────────────────────────────────────────────────────
# [FAST] ~10–15분 — 포트폴리오 ON · limit_atr_mult×3 포함 시 ~768조합 (2^8×3)
# 진입 4축 + 청산 4축, atr_tp_max·회복률 3단계는 DB base 고정 (coarse에서 탐색)
# ─────────────────────────────────────────────────────────────────────
"fine": {
"min_drop_rate": [0.015, 0.02, 0.025, 0.03, 0.04],
"min_recovery_ratio": [0.35, 0.4, 0.45, 0.5, 0.6],
"tail_ratio_min": [1.0, 1.2, 1.5, 1.8, 2.0],
"tail_pct_min": [0.001, 0.002, 0.003, 0.005],
"stop_atr_mult": [1.5, 2.0, 2.5],
"target_atr_mult": [1.5, 2.0, 2.5, 3.0],
"atr_sl_min_pct": [0.5, 0.8],
"atr_sl_max_pct": [0.8, 1.0, 1.2],
"atr_tp_min_pct": [0.3, 0.5, 0.8],
"atr_tp_max_pct": [0.8, 1.0, 1.2],
"shoulder_min_high": [0.002, 0.0025, 0.003, 0.004, 0.005],
"shoulder_cut_pct": [0.0015, 0.002, 0.0025, 0.003, 0.004, 0.005],
"max_rec_3m": [0.85, 0.9, 0.95],
"sl_pct": [0.015, 0.02, 0.025],
"tp_pct": [0.03, 0.04, 0.05],
"rsi_threshold": [72, 75, 78, 82],
"max_loss_krw": [150000, 200000],
"min_drop_pct_for_loss_cut": [0.01, 0.015, 0.02],
},
"full": {
"min_drop_rate": [0.015, 0.02, 0.025, 0.03, 0.04, 0.05],
"min_recovery_ratio": [0.35, 0.4, 0.5, 0.6],
"tail_ratio_min": [1.0, 1.2, 1.5, 1.8, 2.0],
"tail_pct_min": [0.001, 0.002, 0.003],
"stop_atr_mult": [1.5, 2.0, 2.5, 3.0],
"target_atr_mult": [1.5, 2.0, 2.5, 3.0],
"atr_sl_min_pct": [0.5, 0.8],
"atr_sl_max_pct": [0.8, 1.0, 1.2, 1.5],
"atr_tp_min_pct": [0.3, 0.5, 0.8],
"atr_tp_max_pct": [0.8, 1.0, 1.2, 1.5],
"shoulder_min_high": [0.002, 0.003, 0.004, 0.005, 0.007],
"shoulder_cut_pct": [0.0015, 0.002, 0.003, 0.004, 0.005],
"max_rec_3m": [0.85, 0.9, 0.95],
"sl_pct": [0.015, 0.02, 0.025, 0.03],
"tp_pct": [0.03, 0.04, 0.05, 0.06],
"ma20_max_above": [3.0, 5.0],
"max_daily_change": [15.0, 20.0, 25.0],
"max_loss_krw": [100000, 200000],
"min_drop_pct_for_loss_cut": [0.01, 0.015, 0.02, 0.025],
},
"massive": {
"min_drop_rate": [0.015, 0.02, 0.025, 0.03, 0.04, 0.05],
"min_recovery_ratio": [0.35, 0.4, 0.45, 0.5, 0.6],
"tail_ratio_min": [1.0, 1.2, 1.5, 1.8, 2.0],
"tail_pct_min": [0.001, 0.002, 0.003, 0.005],
"stop_atr_mult": [1.5, 2.0, 2.5, 3.0],
"target_atr_mult": [1.5, 2.0, 2.5, 3.0],
"atr_sl_min_pct": [0.5, 0.8, 1.0],
"atr_sl_max_pct": [0.8, 1.0, 1.2, 1.5],
"atr_tp_min_pct": [0.3, 0.5, 0.8],
"atr_tp_max_pct": [0.8, 1.0, 1.2, 1.5],
"shoulder_min_high": [0.002, 0.0025, 0.003, 0.004, 0.005, 0.007],
"shoulder_cut_pct": [0.0015, 0.002, 0.0025, 0.003, 0.004, 0.005],
"max_rec_3m": [0.85, 0.9, 0.95],
"ma20_max_above": [2.0, 3.0, 5.0],
"max_daily_change": [15.0, 20.0, 25.0],
"sl_pct": [0.015, 0.02, 0.025, 0.03],
"tp_pct": [0.03, 0.04, 0.05, 0.06],
"risk_pct": [0.005, 0.01, 0.02],
"kelly_mult": [0.15, 0.25, 0.5],
"max_loss_krw": [100000, 200000],
"min_drop_pct_for_loss_cut": [0.01, 0.015, 0.02],
},
}
TAIL_GRID_AXIS_HINTS_KO: Dict[str, str] = {
"min_drop_rate": "당일 낙폭(시가→저점) 하한",
"min_recovery_ratio": "저점 대비 회복률 하한 (낮을수록 매수 많음)",
"tail_ratio_min": "망치 꼬리/몸통 비율 하한",
"tail_pct_min": "꼬리 길이 최소 비율",
"max_rec_3m": "3분봉 회복 위치 상한 (높을수록 매수 많음)",
"stop_atr_mult": "ATR 손절 배수 (실제 손절=배수×ATR 후 %캡)",
"target_atr_mult": "ATR 익절 배수 (실제 익절=배수×ATR 후 %캡)",
"atr_sl_min_pct": "ATR 손절 % 하한",
"atr_sl_max_pct": "ATR 손절 % 상한",
"atr_tp_min_pct": "ATR 익절 % 하한",
"atr_tp_max_pct": "ATR 익절 % 상한 (잡주 과대 목표 방지) — fast 그리드",
"tail_vol_mult": "TRIGGER 거래량 — 신호봉 vol ≥ N봉평균×배수 (0=OFF)",
"tail_vol_win": "거래량 평균 창(봉 수) — tail_vol_mult>0 일 때",
"shoulder_min_high": "어깨 발동: 진입가 대비 최소 수익(0.003=0.3%) 후 max_price 추적",
"shoulder_cut_pct": "어깨 폭: 갱신된 max_price 대비 되돌림(0.002=0.2%, 매도선=max×(1-cut))",
"ratchet_tiers": "청산방식: off=단일 어깨컷 / '수익%:컷%,…'=다단 래칫 트레일",
"trail_tiers": "당일손익 다단 트레일 'off'/'수익원:컷%,…' — 벌수록 컷% 타이트(이익 보존), 일단위 표본",
"trail_drop_pct": "당일손익 단일 트레일(고점대비 %, 0=OFF) — tier='off' 일 때만 의미",
"trail_arm_krw": "단일 트레일 발동 최소수익(원) — tier='off' & drop>0 일 때만 의미",
"max_daily": "종목당 일일 매수 횟수 상한 (낮을수록 과매매↓·실매정합↑)",
"cooldown_min": "청산 후 동일종목 재진입 콜다운(분) — 연타 방지",
"bar_chg_min_pct": "진입 신호봉 등락 하한(%) — 과대 급락 제외",
"bar_chg_max_pct": "진입 신호봉 등락 상한(%) — 최소 하락폭 요구",
"rsi_threshold": "RSI 과열 진입 거부 기준(%) — 낮을수록 매수 적음",
"pattern_pin": "핀바(긴 아래꼬리 단일봉) 진입 패턴 ON/OFF",
"pattern_engulfing": "장악형(2봉 반전) 진입 패턴 ON/OFF",
"pattern_piercing": "관통형(2봉 반전) 진입 패턴 ON/OFF",
"max_loss_krw": "1회 최대 금액 손실(원, 트레일 미발동 시)",
"limit_atr_mult": "ATR 지정가 깊이(anchor−ATR×배수) — limit_atr 전용, 클수록 체결↓",
"entry_mode": "진입: align=다음봉 시가 / limit_atr=지정가(틱 체결 재생)",
"max_daily_change": "급등주 필터(당일 등락률 상한, %)",
"trail_pct": "보조 트레일 폭(고점 대비, 0=OFF). 래칫 ON이면 1단계보다 arm이 커야 효과",
"trail_arm_pct": "보조 트레일 무장: 진입×(1+arm) 도달 후 (0=즉시)",
"max_spread_pct": "호가 스프레드 상한(%) — kiwoom_0d 본체 재계산(6/25~)",
"min_bid_ask_ratio": "총매수/총매도 잔량비 하한(꼬리지지) — kiwoom_0d 본체 재계산(6/25~)",
}
def evaluate_tail_param_combo(
combo: Dict[str, Any],
*,
base_params: Dict[str, Any],
candles_by_code: Dict[str, List[Dict]],
fee_rate: float,
sell_tax: float,
min_trades: int,
min_win_rate: float,
min_pf: float,
universe_by_slot: Optional[Dict[str, List[str]]] = None,
slot_money: float = 3_000_000.0,
max_stocks: int = 3,
total_budget_krw: float = 9_000_000.0,
period_days: int = 1,
cache_holder: Optional[Dict[str, Any]] = None,
ticks_by_code: Any = None,
orderbook_by_code: Any = None,
program_by_code: Any = None,
log_verdict_by_code: Any = None,
) -> Optional[Dict[str, Any]]:
"""
단일 파라미터 조합 백테 — Grid 워커·Optuna objective 공통.
min_trades·승률·PF 게이트 미통과 시 None.
"""
test_params = dict(base_params)
test_params.update(combo)
if "max_spread_pct" in combo and combo.get("max_spread_pct") is not None:
test_params["_ob_max_spread_pct"] = float(combo["max_spread_pct"])
if "min_bid_ask_ratio" in combo and combo.get("min_bid_ask_ratio") is not None:
test_params["_ob_min_bid_ask_ratio"] = float(combo["min_bid_ask_ratio"])
# 당일손익 트레일 익절 — tier(우선) 또는 단일 drop 이 켜졌을 때만 시뮬 게이트 ON.
_tier_combo = str(combo.get("trail_tiers") or "").strip().lower()
_tier_on = _tier_combo not in ("", "off")
if _tier_on:
test_params["_backtest_daily_profit_trail"] = True
test_params.setdefault("daily_profit_mode", "trailing")
test_params["daily_trail_tiers"] = str(combo["trail_tiers"]).strip()
elif float(combo.get("trail_drop_pct") or 0) > 0:
test_params["_backtest_daily_profit_trail"] = True
test_params.setdefault("daily_profit_mode", "trailing")
test_params["daily_trail_drop_pct"] = float(combo["trail_drop_pct"])
if combo.get("trail_arm_krw") is not None:
test_params["daily_trail_arm_krw"] = float(combo["trail_arm_krw"])
else:
_arm_env = os.environ.get("TAIL_GRID_TRAIL_ARM_KRW", "")
if str(_arm_env).strip() != "":
test_params["daily_trail_arm_krw"] = float(_arm_env)
if cache_holder:
test_params.update(cache_holder)
if log_verdict_by_code:
test_params["_backtest_log_verdict_by_code"] = log_verdict_by_code
meta: Dict[str, Any] = {}
trades = tbc.run_tail_backtest_web_aligned(
candles_by_code, test_params, universe_by_slot,
slot_money=slot_money, fee_rate=fee_rate, sell_tax=sell_tax,
max_stocks=max_stocks, total_budget_krw=total_budget_krw,
ticks_by_code=ticks_by_code,
orderbook_by_code=orderbook_by_code,
program_by_code=program_by_code,
meta_out=meta,
)
stats = tbc.summarize_tail_trades(
trades,
total_budget_krw=total_budget_krw,
period_days=period_days,
)
total_trades = stats["total_trades"]
if total_trades < min_trades:
return None
win_rate = stats["win_rate"]
total_pnl = stats["total_pnl"]
pf = float(stats.get("pf") or 0)
if not combo_passes_search_filters(
win_rate=win_rate, pf=pf,
min_win_rate=min_win_rate, min_pf=min_pf,
):
return None
return {
"params": combo,
"apply_cfg": {**base_params, **combo},
"total_trades": total_trades,
"win_rate": win_rate,
"total_pnl": total_pnl,
"wins": stats["wins"],
"losses": stats["losses"],
"pf": stats["pf"],
"bot_pct": stats["bot_pct"],
"daily_avg_pct": stats["daily_avg_pct"],
"avg_hold_min": stats["avg_hold_min"],
"sell_reasons": _count_tail_sell_reasons(trades),
"skipped_micro_buys": int(
(meta.get("skip_stats") or {}).get("skipped_micro_buys") or 0
),
}
def evaluate_param_chunk(
param_chunk: List[Dict[str, Any]],
base_params: Dict[str, Any],
candles_by_code: Optional[Dict[str, List[Dict]]],
fee_rate: float,
sell_tax: float,
min_trades: int,
min_win_rate: float,
min_pf: float,
top_n: int,
universe_by_slot: Optional[Dict[str, List[str]]] = None,
slot_money: float = 3_000_000.0,
max_stocks: int = 3,
total_budget_krw: float = 9_000_000.0,
period_days: int = 1,
) -> List[Tuple[float, Dict]]:
"""
워커 프로세스에서 실행될 백테스트 평가 함수.
universe_by_slot이 있으면 유니버스 히스토리(5분마다 후보)만 매수 검사.
시각순 포트폴리오·총한도·소액매수 스킵은 tail_backtest_common 과 웹과 동일.
"""
shared = worker_shared_get()
ticks_preloaded = None
orderbook_preloaded = None
program_preloaded = None
log_verdict_preloaded = None
if shared:
if candles_by_code is None:
candles_by_code = shared.get("candles_by_code") or {}
if universe_by_slot is None:
universe_by_slot = shared.get("universe_by_slot")
ticks_preloaded = shared.get("ticks_by_code")
# ws_ticks 공유메모리(opt-in): descriptor 로 read-only attach (워커당 1회 재사용).
if not ticks_preloaded:
_desc = shared.get("ticks_shared_descriptor")
if _desc:
_tm = shared.get("_ticks_mapping_cache")
if _tm is None:
from kis_trader.backtest.shared_ticks import SharedTicksMapping
_tm = SharedTicksMapping(_desc)
shared["_ticks_mapping_cache"] = _tm
ticks_preloaded = _tm
orderbook_preloaded = shared.get("orderbook_by_code")
program_preloaded = shared.get("program_by_code")
log_verdict_preloaded = shared.get("log_verdict_by_code")
if candles_by_code is None:
candles_by_code = {}
cache_holder: Dict[str, Any] = {}
attach_indicator_caches_to_params(cache_holder, candles_by_code)
local_heap = []
for combo in param_chunk:
assert_parent_alive()
result_pkg = evaluate_tail_param_combo(
combo,
base_params=base_params,
candles_by_code=candles_by_code,
fee_rate=fee_rate,
sell_tax=sell_tax,
min_trades=min_trades,
min_win_rate=min_win_rate,
min_pf=min_pf,
universe_by_slot=universe_by_slot,
slot_money=slot_money,
max_stocks=max_stocks,
total_budget_krw=total_budget_krw,
period_days=period_days,
cache_holder=cache_holder,
ticks_by_code=ticks_preloaded,
orderbook_by_code=orderbook_preloaded,
program_by_code=program_preloaded,
log_verdict_by_code=log_verdict_preloaded,
)
if result_pkg is None:
continue
win_rate = result_pkg["win_rate"]
total_pnl = result_pkg["total_pnl"]
if len(local_heap) < top_n:
heapq.heappush(local_heap, (win_rate, total_pnl, id(result_pkg), result_pkg))
else:
heapq.heappushpop(local_heap, (win_rate, total_pnl, id(result_pkg), result_pkg))
return local_heap
def run_search(
start: str,
end: str,
mode: str,
top_n: int,
min_trades: int,
min_win_rate: float,
min_pf: float,
sort_by: str = "pnl",
use_fallback_universe: bool = False,
max_combos: int = 0,
timeframe: int = 3,
time_start_hm: Optional[int] = None,
time_end_hm: Optional[int] = None,
slot_money: Optional[float] = None,
max_stocks: Optional[int] = None,
total_budget_krw: Optional[float] = None,
orderbook_filter: str = "off",
) -> bool:
"""탐색 실행. 결과가 있어서 JSON 저장까지 했으면 True, 조건 만족 조합 없이 조기 return 시 False."""
db = TradeDB()
try:
# 1. Base Environment Parameters 로드 (backtest_web _get_tail_defaults_for_backtest 와 동일)
base_params = te.get_tail_defaults_from_db(db)
if time_start_hm is not None:
base_params["time_start_hm"] = int(time_start_hm)
if time_end_hm is not None:
base_params["time_end_hm"] = int(time_end_hm)
# ── 호가필터 ON/OFF (2단계 워크플로) ───────────────────────────
# off(기본): 1단계 — 호가 게이트 없이 코어 파라미터만 순수 탐색.
# on : 2단계 — kiwoom_0d 본체로 스프레드·꼬리지지(총잔량비) 실제 적용.
# auto : env/DB 의 *_ORDERBOOK_FILTER_ENABLED 값을 그대로 따름.
_ob_mode = (orderbook_filter or "off").strip().lower()
if _ob_mode == "off":
base_params["_orderbook_filter_enabled"] = False
elif _ob_mode == "on":
base_params["_orderbook_filter_enabled"] = True
_ob_filter_on = bool(base_params.get("_orderbook_filter_enabled")) or _ob_mode == "auto"
logger.info(
"📌 호가필터: %s (%s)",
_ob_mode.upper(),
"적용" if _ob_filter_on else "스킵 — 코어 파라미터 순수 탐색",
)
row = db.conn.execute("SELECT * FROM env_config ORDER BY id DESC LIMIT 1").fetchone()
r = dict(row) if row else {}
fee_rate, sell_tax, _slot_from_fee = tbc.fee_and_slot_from_env_row(r)
portfolio = tbc.resolve_tail_portfolio_params(
r,
base_params,
slot_money=slot_money if slot_money is not None else _slot_from_fee,
max_stocks=max_stocks,
total_budget_krw=total_budget_krw,
)
slot_money = float(portfolio["slot_money"])
max_stocks = int(portfolio["max_stocks"])
total_budget_krw = float(portfolio["total_budget_krw"])
tbc.merge_tail_portfolio_into_params(base_params, portfolio)
base_params["capital"] = float(
r.get("BACKTEST_CAPITAL") or base_params.get("capital") or 100_000_000.0
)
period_days = max(
1,
(datetime.strptime(end, "%Y-%m-%d") - datetime.strptime(start, "%Y-%m-%d")).days + 1,
)
tail_tf = int(timeframe)
if tail_tf not in tbc.VALID_TIMEFRAMES:
logger.error("❌ timeframe 은 3·5·15·60 중 하나여야 합니다 (backtest_web 과 동일)")
return False
start_key, end_key, start_ymd, end_ymd = tbc.date_keys(start, end)
use_saved_history = not use_fallback_universe
universe_by_slot, universe_source, universe_history_slots, scan_interval_min = (
tbc.resolve_tail_universe(start_ymd, end_ymd, use_saved_history=use_saved_history)
)
if use_fallback_universe:
print("📌 [유니버스] --fallback-universe: 저장 이력 무시 → ws_candles 전 종목")
elif universe_source == "history":
avg = (
sum(len(v) for v in universe_by_slot.values()) / max(1, universe_history_slots)
if universe_by_slot else 0
)
print(
f"✅ 유니버스: SHORT 저장 이력 (웹백테 동일) | "
f"{universe_history_slots:,}슬롯 · 평균 {avg:.1f}종목"
)
else:
print("📌 [유니버스] 저장 이력 없음 → ws_candles 전 종목 (웹 폴백과 동일)")
base_params = dict(base_params)
base_params["scan_interval_min"] = scan_interval_min
base_params["timeframe"] = tail_tf
base_params.setdefault("force_eod_exit", False)
base_params.setdefault("backtest_tick_fallback_ohlc", True)
# ── 틱재생(ws_ticks) — 기본 ON (실매 체결 정합) ──────────────────────
# 3분봉 OHLC(open→high→low→close 가정)는 손절보다 익절을 먼저 체결하는
# 낙관 편향을 만든다. 모멘텀·돌파와 동일하게 tail_tick_replay 기본값(ON)을
# 그대로 따르되, params/env(TAIL_BACKTEST_USE_TICK_DB)로 오버라이드 가능.
from kis_trader.engine.tail_tick_replay import tail_backtest_use_tick_db as _tail_use_tick
base_params.setdefault("backtest_use_tick_db", _tail_use_tick(None))
if base_params.get("backtest_use_tick_db"):
logger.info("📌 틱재생(ws_ticks): ON — 실매 체결 정합 모드 (OHLC 낙관편향 제거)")
logger.info(
f"📅 데이터 로드: {start} ~ {end} | TF={tail_tf} | "
f"유니버스={universe_source} | 매수시간 "
f"{base_params.get('time_start_hm', 930):04d}-{base_params.get('time_end_hm', 1500):04d}"
)
rsi_period = int(base_params.get("rsi_period", 14))
candles_by_code, total_candles, has_holding_peak = tbc.load_tail_candles_by_code(
db, start_key, end_key, tail_tf, rsi_period=rsi_period,
)
if not candles_by_code:
logger.info("❌ 백테스트할 데이터가 없습니다.")
return False
ticks_by_code: Dict[str, Dict[str, List[Dict]]] = {}
tick_backtest_meta: Dict[str, Any] = {}
from kis_trader.engine.tail_tick_replay import tail_backtest_wants_tick_replay
from kis_trader.backtest.tail_tick_loader import load_tail_ticks_by_code, tick_coverage_stats
_tick_probe = dict(base_params)
_tick_probe["entry_mode"] = "align"
if tail_backtest_wants_tick_replay(_tick_probe):
_tick_db = TradeDB()
try:
ticks_by_code, tick_rows = load_tail_ticks_by_code(
_tick_db, start_key, end_key, set(candles_by_code.keys()),
)
finally:
_tick_db.close()
else:
tick_rows = 0
if tick_rows > 0:
tick_backtest_meta = tick_coverage_stats(candles_by_code, ticks_by_code)
tick_backtest_meta["ws_tick_rows_loaded"] = tick_rows
cov = tick_backtest_meta.get("tick_bar_coverage_pct", 0)
logger.info(
"✅ ws_ticks %s건 | 3분봉 커버리지 %s%% (%s/%s종목)",
f"{tick_rows:,}",
cov,
tick_backtest_meta.get("tick_codes_with_data", 0),
tick_backtest_meta.get("tick_codes_total", 0),
)
elif tail_backtest_wants_tick_replay(_tick_probe):
logger.warning("⚠️ ws_ticks 없음 — OHLC 폴백 (WS_TICK_SAVE_ENABLED 후 재탐색)")
_pre_grid = _tail_grids(mode)
# kiwoom_0d 본체 재계산 — 필터 ON + 호가필터 축 **실제 스윕**(값 2개↑) 일 때만.
_ob_axes = ("max_spread_pct", "min_bid_ask_ratio")
_ob_sweeping = any(len(set(_pre_grid.get(k) or [])) > 1 for k in _ob_axes)
if _ob_filter_on and _ob_sweeping:
base_params["backtest_use_kiwoom_body_snapshot"] = True
base_params["_backtest_use_kiwoom_body"] = True
_ob_axis_vals = {k: _pre_grid.get(k) for k in _ob_axes if len(set(_pre_grid.get(k) or [])) > 1}
logger.info(
"📌 호가필터 스윕 활성 → kiwoom_0d 본체 재계산 "
f"(축 {_ob_axis_vals}, 본체 없는 날은 log_backfill 폴백)"
)
elif _ob_filter_on:
logger.info("📌 호가필터 ON · 스윕 없음 → 본체 재계산 생략(판정 재생 경로)")
orderbook_by_code: Dict[str, Dict[str, List[Dict]]] = {}
program_by_code: Dict[str, Dict[str, List[Dict]]] = {}
log_verdict_by_code: Dict[str, Dict[str, List[Dict]]] = {}
trigger_snap_meta: Dict[str, Any] = {}
try:
from kis_trader.backtest.trigger_snapshot_loader import load_trigger_snapshots_by_code
orderbook_by_code, program_by_code, trigger_snap_meta = load_trigger_snapshots_by_code(
db, start_key, end_key, set(candles_by_code.keys()),
engine_params=base_params, strategy="TAIL",
)
log_verdict_by_code = trigger_snap_meta.get("log_verdict_by_code") or {}
ob_rows = int(trigger_snap_meta.get("ws_orderbook_rows_loaded") or 0)
pg_rows = int(trigger_snap_meta.get("ws_program_rows_loaded") or 0)
lv_rows = int(trigger_snap_meta.get("log_verdict_rows") or 0)
logger.info(
"✅ TRIGGER 스냅샷 ws_orderbook %s건 | ws_program %s건 | log_backfill 판정 %s건",
f"{ob_rows:,}", f"{pg_rows:,}", f"{lv_rows:,}",
)
if ob_rows <= 0 and pg_rows <= 0:
logger.warning(
"⚠️ TRIGGER 스냅샷 없음 — 호가·프로그램 필터 스킵 (실매 수집 후 재탐색)"
)
except Exception as _snap_ex:
logger.debug("trigger snapshot 로드 스킵: %s", _snap_ex)
logger.info(
f"📦 종목: {len(candles_by_code)}개 (ws_candles 전체) | "
f"캔들: {total_candles:,}개 | portfolio_mode=ON"
)
logger.info(
f"💼 포트폴리오: 1회 {slot_money:,.0f}원 | 동시 {max_stocks}종 | "
f"총한도 {total_budget_krw:,.0f}원 (웹백테 동일)"
)
if portfolio.get("budget_warning"):
logger.warning(f"💰 {portfolio['budget_warning']}")
# 4. Grid 조합 생성
grid = _tail_grids(mode)
if not grid:
logger.error(f"❌ 알 수 없는 모드: {mode}")
return False
if "entry_mode" not in grid:
_search_entry = get_env_from_db("TAIL_PARAM_SEARCH_ENTRY_MODE", "")
if _search_entry not in (None, "", "None"):
base_params["entry_mode"] = str(_search_entry).strip().lower()
if mode == "fast":
print(
f"📌 [fast] TAIL_GRID_FAST_* · 22축(거래량·atr_tp_max·청산방식·RSI·반전패턴·"
f"과매매제어·봉등락) · →{FAST_MAX_COMBOS}샘플 · "
f"ws_ticks={'ON' if tick_rows > 0 else 'OFF'}"
)
if mode == "coarse":
print(
"📌 [coarse] TAIL_GRID_COARSE_* env · max_daily_chg 포함 (기본 ~4천조합, cap 5000)"
)
if "entry_mode" in grid:
print(f"📌 진입모드 그리드: {grid['entry_mode']} (조합별 align/limit_atr + 틱 재생)")
else:
print(
f"📌 진입모드: {base_params.get('entry_mode', 'align')} "
f"(TAIL_PARAM_SEARCH_ENTRY_MODE)"
)
keys = list(grid.keys())
axes = list(grid.values())
# 데카르트곱 전체를 RAM 에 펼치지 않는다 (호가 스윕 시 수천만 → OOM 방지).
dict_combos, total_grid, max_combos_cap, dropped_by_cap = cap_combos_uniform_lazy(
keys,
axes,
mode,
strategy_env_prefix="TAIL",
default_fast=FAST_MAX_COMBOS,
default_other=DEFAULT_MAX_COMBOS,
max_combos_override=max_combos,
)
sampled = len(dict_combos)
logger.info(f"🔍 탐색 모드: {mode.upper()} | 그리드 조합: {total_grid:,} → 백테: {sampled:,}")
if mode == "fast":
logger.info(f"📌 [fast] {total_grid:,}→{max_combos_cap}균등샘플")
if dropped_by_cap:
logger.info(f" (max-combos={max_combos_cap} 균등 샘플, 제외 {dropped_by_cap:,}개)")
for k in keys:
hint = TAIL_GRID_AXIS_HINTS_KO.get(k)
if hint:
logger.info(f" [{k}] {hint}")
logger.info(f"📌 1위 정렬 기준: {'총손익 최대 (수익 나는 조합 우선)' if sort_by == 'pnl' else '승률 최대'}")
# ── ws_ticks 공유메모리 — 워커별 사본 대신 1벌 공유 (momentum 과 동일) ────────
# 기본 ON(검증 완료). ON 이면 컬럼(numpy)로 shared_memory 에 1벌 올리고 워커는
# read-only attach → 메모리 N배 제거 + 워커 상한 해제. 엔진 핫루프는 TickColumnView
# 로 배열 직접 읽어 dict 경로와 bit-identical. 끄려면 TAIL_PARAM_SEARCH_SHARED_TICKS=0.
# numpy/shm 미지원·빌드 실패 시 자동 폴백.
shared_tick_store = None
if get_env_bool("TAIL_PARAM_SEARCH_SHARED_TICKS", True) and ticks_by_code:
from kis_trader.backtest.shared_ticks import (
build_shared_ticks,
shared_ticks_available,
)
if shared_ticks_available():
shared_tick_store = build_shared_ticks(ticks_by_code)
if shared_tick_store is not None:
import atexit as _atexit
_atexit.register(shared_tick_store.unlink)
logger.info("📦 ws_ticks 공유메모리 ON — 워커 attach(read-only), 사본 제거")
ticks_by_code = {}
import gc as _gc
_gc.collect()
try:
import ctypes as _ctypes
_ctypes.CDLL("libc.so.6").malloc_trim(0)
except Exception:
pass
else:
logger.info("⚠️ ws_ticks 공유메모리 build 실패 — 기존 경로 폴백")
else:
logger.info("⚠️ numpy/shared_memory 미지원 — 기존 경로 폴백")
# 5. 멀티프로세싱을 위한 청크 분할 (메모리 및 부하 분산)
shared = ParamSearchSharedPayload({
"candles_by_code": candles_by_code,
"universe_by_slot": universe_by_slot,
"ticks_by_code": ticks_by_code,
"ticks_shared_descriptor": (shared_tick_store.descriptor() if shared_tick_store else None),
"orderbook_by_code": orderbook_by_code,
"program_by_code": program_by_code,
"log_verdict_by_code": log_verdict_by_code,
"trigger_snapshot_meta": trigger_snap_meta,
})
payload_bytes = shared.estimate_bytes()
n_cpu = os.cpu_count() or 4
_cpu_frac = get_env_float("PARAM_SEARCH_CPU_FRAC", 0.8)
max_workers, chunk_size, _ = param_search_chunk_plan(sampled, payload_bytes)
if ticks_by_code:
tick_cap = get_env_int("TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_TICKS", 3)
if tick_cap > 0:
max_workers = min(max_workers, tick_cap)
logger.info(
"📌 ws_ticks payload — 워커 상한 %d (TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_TICKS)",
max_workers,
)
# 공유메모리 틱은 사본이 없어 OOM 위험이 없다 → 기본 상한 없음(CPU/메모리 계획대로).
if shared_tick_store is not None:
shared_cap = get_env_int("TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_SHARED_TICKS", 0)
if shared_cap > 0 and max_workers > shared_cap:
max_workers = shared_cap
logger.info(
"📌 ws_ticks 공유메모리 — 워커 상한 %d (TAIL_PARAM_SEARCH_MAX_WORKERS_WITH_SHARED_TICKS)",
max_workers,
)
chunks = [dict_combos[i:i + chunk_size] for i in range(0, len(dict_combos), chunk_size)]
logger.info(param_search_worker_budget_line(payload_bytes))
logger.info(
f"⚙️ 멀티프로세싱 시작 (코어: {n_cpu}, 워커: {max_workers}, CPU {_cpu_frac*100:.0f}%) | "
f"청크: {len(chunks):,}개 (청크당 ~{chunk_size}조합)"
)
start_time = time.time()
global_heap = []
progress_eta = ParamSearchProgressETA(len(chunks), max_workers)
with managed_process_pool(max_workers, shared_payload=shared) as executor:
def _submit(chunk: List[Dict[str, Any]]):
return executor.submit(
evaluate_param_chunk, chunk, base_params, None,
fee_rate, sell_tax, min_trades, min_win_rate, min_pf, top_n, None,
slot_money, max_stocks, total_budget_krw, period_days,
)
logger.info(f"⏳ 청크 처리 중… (청크당 최대 {chunk_size:,}개 조합, 완료되는 대로 진행률·ETA 출력)")
processed = 0
use_carriage_return = sys.stdout.isatty()
for local_results in iter_pool_chunk_results(
executor, chunks, _submit, max_workers=max_workers,
):
processed += 1
for i, item in enumerate(local_results):
wr, pnl, _, result_pkg = item
if sort_by == "pnl":
unique_item = (-pnl, wr, (processed, i), result_pkg)
else:
unique_item = (wr, pnl, (processed, i), result_pkg)
if len(global_heap) < top_n:
heapq.heappush(global_heap, unique_item)
else:
heapq.heappushpop(global_heap, unique_item)
# ETA — 워밍업 1파도 제외 누적 평균 (param_search_pool.ParamSearchProgressETA)
progress = (processed / len(chunks)) * 100
elapsed_so_far = time.time() - start_time
eta_str = ParamSearchProgressETA.format_sec(
progress_eta.remaining_sec(processed, elapsed_so_far),
)
elapsed_str = ParamSearchProgressETA.format_elapsed(elapsed_so_far)
eta_msg = f" | 경과: {elapsed_str} | 남은시간: {eta_str}"
line = f"⏳ 진행률: {progress:.1f}% ({processed:,}/{len(chunks):,} 청크 완료){eta_msg}"
if use_carriage_return:
print(f"\r{line}", end="", flush=True)
else:
logger.info(line)
if use_carriage_return:
print(flush=True) # 줄바꿈으로 진행률 줄 마무리
# 워커 종료 후 공유메모리 즉시 해제 (atexit 는 크래시 대비 이중 안전장치).
if shared_tick_store is not None:
shared_tick_store.unlink()
shared_tick_store = None
elapsed = time.time() - start_time
# 5. 결과 정렬 및 출력
if not global_heap:
logger.info("⚠️ 조건을 만족하는 조합이 없습니다. (min_trades를 낮추거나 기간을 늘려보세요)")
print("📌 DB 미적용. 기존 설정 유지.")
return False
# sort_by pnl → (-pnl, wr) 최소힙 → heappop 순이 이미 [best pnl, ..., worst] 이므로 reverse 금지
# sort_by win_rate → (wr, pnl) 최소힙 → heappop 순은 [low wr, ..., high wr] 이므로 reverse 필요
best_results = [heapq.heappop(global_heap)[3] for _ in range(len(global_heap))]
if sort_by == "win_rate":
best_results.reverse()
# sl_pct(손절%)별 상위 보장 → 한 값만 상위 독점 방지, 동점이면 손절 낮은 쪽(보수적) 1위
if "sl_pct" in keys and best_results:
sl_vals = sorted(set(r["params"]["sl_pct"] for r in best_results))
per_sl = max(1, top_n // len(sl_vals))
by_sl = {}
for r in best_results:
v = r["params"]["sl_pct"]
if v not in by_sl:
by_sl[v] = []
if len(by_sl[v]) < per_sl:
by_sl[v].append(r)
best_results = []
for v in sl_vals:
best_results.extend(by_sl.get(v, []))
best_results.sort(key=lambda r: (-r["total_pnl"], r["params"]["sl_pct"], -r["win_rate"]))
logger.info(f"✅ 손절(sl_pct)별 상위 {per_sl}개씩 보장 → {len(best_results)}건 (동점 시 손절 낮은 쪽 1위)")
filtered = [
r for r in best_results
if combo_passes_search_filters(
win_rate=float(r.get("win_rate") or 0),
pf=float(r.get("pf") or 0),
min_win_rate=min_win_rate,
min_pf=min_pf,
)
]
if filtered:
best_results = filtered
order_msg = "수익→승률 순" if sort_by == "pnl" else "승률→수익 순"
logger.info(
f"✅ 승률≥{min_win_rate}% · PF≥{min_pf} {len(best_results)}건 중 {order_msg} 상위"
)
else:
logger.info(
f"⚠️ 승률≥{min_win_rate}% · PF≥{min_pf} 없음 → 차악(상위) 적용"
)
# 손익 마이너스인 조합 제외 (수익 나는 것만 표시·저장)
profitable = [r for r in best_results if r["total_pnl"] > 0]
if profitable:
best_results = profitable
logger.info(f"✅ 총손익 플러스만 사용: {len(best_results)}건 (손실 조합 제외)")
else:
logger.info(f"⚠️ 수익 나는 조합 없음 → 손실 최소 순으로 표시")
order_label = "수익" if sort_by == "pnl" else "승률"
hdr_keys = [k for k in keys if k in (best_results[0]["params"] if best_results else {})]
col_w = max((len(k) for k in hdr_keys), default=6) + 2
sep_w = len(hdr_keys) * (col_w + 2) + 50
print(f"\n✅ 탐색 완료! 총 소요 시간: {elapsed:.1f}초")
print(f"\n{'='*min(sep_w, 100)}")
print(f" 🏆 꼬리잡기 {order_label} TOP {min(top_n, len(best_results))}")
print(f"{'='*min(sep_w, 100)}")
hdr = " ".join(f"{k:>{col_w}}" for k in hdr_keys)
print(f"{hdr} | {'손익(원)':>12} {'수익%':>7} {'승률':>6} {'거래':>5} {'PF':>5}")
print("-" * min(sep_w + 10, 110))
for i, res in enumerate(best_results[:top_n]):
p = res["params"]
row = " ".join(_format_param_cell(p.get(k, ""), col_w) for k in hdr_keys)
pf = res.get("pf", 0) or 0
bot_pct = res.get("bot_pct", 0) or 0
print(
f"{row} | {res['total_pnl']:>+12,.0f} {bot_pct:>+6.2f}% "
f"{res['win_rate']:>5.1f}% {res['total_trades']:>5} {pf:>5.2f}"
)
best = best_results[0]
bp = best["params"]
print(f"""
╔══════════════════════════════════════════╗
║ 🏆 1위 최적 파라미터 ║
╠══════════════════════════════════════════╣""")
for k in hdr_keys:
v = bp.get(k, "")
print(f"║ {k:<28s} : {str(v):>6} ║")
print(f"""╠══════════════════════════════════════════╣
║ 총 손익 : {best['total_pnl']:>+12,.0f} 원 ║
║ 수익률(한도) : {best.get('bot_pct', 0):>+6.2f}% ║
║ 승률 : {best['win_rate']:>6.1f}% ║
║ 총 거래 : {best['total_trades']:>5} 건 ║
║ Profit Factor : {best.get('pf', 0):>5.2f} ║
╚══════════════════════════════════════════╝""")
reasons = best.get("sell_reasons") or {}
if reasons:
parts = ", ".join(f"{k}={v}" for k, v in sorted(reasons.items(), key=lambda x: -x[1]))
print(f"🏅 1위 청산분포: {parts}")
# 6. JSON 파일로 결과 저장 (매 실행 새 파일 · momentum/breakout 과 동일 정책)
out_data = {
"mode": mode,
"start": start,
"end": end,
"timeframe": tail_tf,
"universe_source": universe_source,
"universe_history_slots": universe_history_slots,
"slot_money": int(slot_money),
"max_stocks": max_stocks,
"total_budget_krw": int(total_budget_krw),
"portfolio_mode": True,
"budget_warning": portfolio.get("budget_warning"),
"backtest_days": period_days,
"web_backtest_aligned": True,
"min_trades": min_trades,
"min_win_rate": min_win_rate,
"min_pf": min_pf,
"grid_keys": keys,
"cartesian_product": total_grid,
"sampled_backtests": sampled,
"dropped_by_combo_cap": dropped_by_cap,
"max_combos": max_combos_cap,
"grid_axis_hints": {k: TAIL_GRID_AXIS_HINTS_KO[k] for k in keys if k in TAIL_GRID_AXIS_HINTS_KO},
"holding_peak_in_candles": has_holding_peak,
"tested_combos": sampled,
"elapsed_sec": round(elapsed, 1),
"results": best_results[:top_n],
}
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
out_name = f"search_tail_{mode}_{ts}.json"
out_dir = _results_dir_for_write()
out_path = os.path.join(out_dir, out_name)
try:
with open(out_path, "w", encoding="utf-8") as f:
json.dump(out_data, f, indent=2, ensure_ascii=False)
except OSError:
fb = os.path.join(os.path.expanduser("~"), ".kis_bot_search_results")
os.makedirs(fb, exist_ok=True)
out_path = os.path.join(fb, out_name)
with open(out_path, "w", encoding="utf-8") as f:
json.dump(out_data, f, indent=2, ensure_ascii=False)
logger.warning("⚠️ results/ 쓰기 권한 없음 → 폴백 저장: %s", out_path)
logger.info("💾 결과 저장: %s", out_path)
return True
except Exception as e:
logger.error(f"❌ 탐색 중 오류 발생: {e}", exc_info=True)
return False
finally:
db.close()
def _merge_tail_apply_source(
target: Dict[str, Any],
search_meta: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""tail JSON → apply 소스 (공통 merge 사용)."""
return merge_param_search_apply_source(target, search_meta)
def _count_tail_sell_reasons(trades: List[Dict[str, Any]]) -> Dict[str, int]:
"""청산 사유별 건수 — tail trades 는 reason 또는 sell_reason."""
out: Dict[str, int] = {}
for t in trades or []:
k = str(t.get("sell_reason") or t.get("reason") or "unknown")
out[k] = out.get(k, 0) + 1
return out
def _tail_params_to_env_map(p: Dict[str, Any]) -> Dict[str, str]:
"""꼬리 파라서치·웹 save_config 와 동일 — TAIL_* 키만."""
from kis_trader.engine.tail_env_keys import params_to_tail_env_patch
env_map = params_to_tail_env_patch(p)
if "risk_pct" in p:
env_map["RISK_PCT_PER_TRADE"] = str(p["risk_pct"])
if "kelly_mult" in p:
env_map["KELLY_MULTIPLIER"] = str(p["kelly_mult"])
return env_map
def apply_from_json(apply_idx: int):
"""
저장된 JSON 결과에서 N위 조합을 추출하여 env_config에 즉시 반영
"""
out_file = _find_tail_result_json()
if not out_file:
logger.error(
"❌ 결과를 찾을 수 없습니다. 후보:\n"
" - kis_trader/backtest/results/search_tail_*.json (최신)\n"
" - 구 형식 tail_search_*.json 또는 tail_param_result.json"
)
return
with open(out_file, "r", encoding="utf-8") as f:
data = json.load(f)
results = data.get("results", [])
if not results:
logger.error("❌ JSON에 저장된 결과가 없습니다.")
return
if apply_idx < 1 or apply_idx > len(results):
logger.error(f"❌ 유효하지 않은 순위입니다. (1~{len(results)} 사이 입력)")
return
target = results[apply_idx - 1]
if target.get("total_pnl", 0) <= 0:
logger.warning(f"⚠️ {apply_idx}번째 결과는 총손익 ≤ 0 (조건 미충족). DB 미적용. 기존 설정 유지.")
print("⚠️ 해당 순위는 총손익 ≤ 0 (조건 미충족). DB 미적용. 기존 설정 유지.")
return
merged = _merge_tail_apply_source(target, data)
env_map = _tail_params_to_env_map(merged)
db = TradeDB()
try:
latest = db.get_latest_env()
snap = dict(latest["snapshot"]) if latest else {}
snap.update(env_map)
env_id = db.insert_env_snapshot(snap)
logger.info(f"\n🚀 [자동 반영 완료] {apply_idx}위 조합을 DB에 적용했습니다. (env_id: {env_id})")
logger.info(f"📂 파일: {out_file}")
logger.info(f"적용된 값: {json.dumps(env_map, indent=2, ensure_ascii=False)}")
logger.info("실매매 봇(kis_trader/main.py TailCatchStrategy)이 다음 루프부터 이 설정을 사용하여 매매를 시작합니다.")
finally:
db.close()
def apply_params_to_db(p: dict):
"""
병합된 꼬리 설정 dict → env_config 반영.
(_merge_tail_apply_source 결과 또는 apply_from_json 과 동일 형태)
"""
env_map = _tail_params_to_env_map(dict(p or {}))
if not env_map:
logger.warning("DB 적용할 env 키가 없습니다.")
return
db = TradeDB()
try:
latest = db.get_latest_env()
snap = dict(latest["snapshot"]) if latest else {}
snap.update(env_map)
db.insert_env_snapshot(snap)
finally:
db.close()
def main():
today = datetime.now().strftime("%Y-%m-%d")
week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
parser = argparse.ArgumentParser(description="꼬리잡기 V3 백테스트 파라미터 최적화 (Grid Search)")
parser.add_argument("--start", default=week_ago, help="시작일 (YYYY-MM-DD, 기본 7일)")
parser.add_argument("--end", default=today, help="종료일 (YYYY-MM-DD)")
parser.add_argument("--timeframe", "--tf", default=3, type=int, dest="timeframe",
help="ws_candles 분봉 3·5·15·60 (backtest_web 꼬리 탭과 동일, 기본 3)")
add_portfolio_cli_args(parser)
parser.add_argument("--mode", default="fast", choices=["fast", "coarse", "fine", "full", "massive"],
help="탐색 모드 (fast≈768조합·포트폴리오·~10–20분, coarse≈10k)")
parser.add_argument("--max-combos", type=int, default=-1,
help=f"백테 조합 상한 (기본 {DEFAULT_MAX_COMBOS}, 0=무제한·수억 조합 모드 비추)")
parser.add_argument("--top", default=5000, type=int, help="메모리에 유지·JSON 저장할 상위 N개 (기본 5000)")
parser.add_argument("--min_trades", default=3, type=int,
help="최소 거래 건수 (1건 플러스 과최적화 방지, 기본 3)")
add_search_filter_cli_args(parser)
parser.add_argument("--apply", nargs="?", const=1, type=int, default=None, metavar="N",
help="N번째 결과를 DB에 적용 (기본 1위). --from-file 과 함께 쓰면 재탐색 없이 즉시 적용")
parser.add_argument("--from-file", action="store_true", help="탐색 생략하고 기존 저장된 JSON에서 적용")
parser.add_argument("--sort-by", default="pnl", choices=["pnl", "win_rate"],
help="1위 기준: pnl=총손익 최대(기본), win_rate=승률 최대")
parser.add_argument("--fallback-universe", action="store_true", dest="fallback_universe",
help="저장 이력 무시, 전체 종목으로 매수 검사 (조합별 거래 수 확대·느림)")
parser.add_argument("--use-universe-history", action="store_true", dest="use_universe_history",
help="SHORT 유니버스 저장 이력만 매수 (기본 동작과 동일, 명시용)")
parser.add_argument(
"--orderbook-filter", default="off", choices=["off", "on", "auto"],
dest="orderbook_filter",
help="호가필터: off=1단계(코어만·기본) / on=2단계(kiwoom 본체 스프레드·꼬리지지) / auto=DB값",
)
parser.add_argument("--apply-ai", action="store_true", dest="apply_ai",
help="Gemini가 수익·승률 기준으로 하나 골라 DB 적용. --from-file 과 함께 쓰면 탐색 없이 최근 JSON만 사용; 그 외에는 탐색 완료 후 방금 생성된 JSON으로 적용")
args = parser.parse_args()
start = args.start
end = args.end
# 탐색 없이 최근 JSON으로만 AI 적용 (--from-file --apply-ai)
if args.apply_ai and args.from_file:
import param_apply_ai
param_apply_ai.apply_ai_tail()
return
if args.from_file and args.apply is not None:
apply_from_json(args.apply)
return
# SIGTERM 도 KeyboardInterrupt 로 전환 (systemd·운영자 kill 대응)
def _sigterm_to_kbd(_sig, _frm):
raise KeyboardInterrupt("SIGTERM 수신 → 워커 정리 후 종료")
try:
signal.signal(signal.SIGTERM, _sigterm_to_kbd)
except Exception:
pass
run_lock = try_acquire_run_lock("tail_param_search")
if run_lock is None:
logger.error(
"⛔ 이미 실행 중인 tail_param_search 가 있습니다.\n"
" ps -ef | grep tail_param_search\n"
" pkill -f 'tail_param_search.py' 후 재실행하세요."
)
sys.exit(2)
try:
if args.max_combos >= 0:
max_combos = max(0, int(args.max_combos))
elif args.mode == "fast":
max_combos = FAST_MAX_COMBOS
else:
max_combos = DEFAULT_MAX_COMBOS
# 기본: SHORT 유니버스 저장 이력 (backtest_web universe=history). 전종목은 --fallback-universe.
use_fallback = bool(args.fallback_universe)
if args.use_universe_history:
use_fallback = False
had_results = run_search(
start, end, args.mode, args.top,
args.min_trades, args.min_win_rate, args.min_pf,
sort_by=args.sort_by, use_fallback_universe=use_fallback,
max_combos=max_combos,
timeframe=args.timeframe,
time_start_hm=args.time_start,
time_end_hm=args.time_end,
slot_money=args.slot_money,
max_stocks=args.max_stocks,
total_budget_krw=args.total_budget,
orderbook_filter=args.orderbook_filter,
)
except KeyboardInterrupt as e:
print(f"\n⛔ {e} — 미완료 결과 없이 종료합니다.", flush=True)
sys.exit(130)
finally:
run_lock.release()
if args.apply is not None:
if _find_tail_result_json():
apply_from_json(args.apply)
else:
logger.warning("결과가 없어 DB 적용을 건너뜁니다.")
# 탐색에서 조건 만족 조합이 있었을 때만 방금 저장된 JSON으로 AI 적용 (없으면 기존 설정 유지)
if args.apply_ai and had_results:
import param_apply_ai
param_apply_ai.apply_ai_tail()
elif args.apply_ai and not had_results:
print("📌 이번 탐색에서 조건 만족 조합 없음 → apply_ai 스킵. DB 미적용. 기존 설정 유지.")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink