""" kis_trader/engine/updown_box.py — 박스권(횡보) 판별 엔진 (UPDOWN 전용) ================================================================================ [목적] "이 종목이 지금 박스권(횡보)인가?" 를 분봉(기본 15분)으로 판정하고, 박스의 하단/상단/높이/품질점수를 산출한다. SCAN(watchlist 충전)·TRIGGER(하단 되돌림 진입) 양쪽이 공유하는 **단일 소스** — 백테=실매 동일 로직. [박스권이란? — 투자 용어] - 박스권(Box / Range, 횡보): 가격이 일정 범위(상단·하단) 안에서 오르내리며 뚜렷한 추세 없이 갇혀 있는 구간. "지지선(하단)에서 사고 저항선(상단)에서 판다"가 박스 매매의 기본. 우리 전략은 **하단 되돌림 매수(dip-buy)** 에 쓴다. [판별 3축 — 모두 만족해야 박스] 1) 좁은 범위 : (box_high - box_low) / box_low × 100 ≤ ``UPDOWN_BOX_RANGE_MAX_PCT`` → 박스가 너무 넓으면(변동성 큼) 박스권이 아니다. 2) 밴드 스퀴즈 : 볼린저밴드 폭(BandWidth) ≤ ``UPDOWN_BOX_BB_BW_MAX`` · BandWidth(%) = (상단밴드 - 하단밴드) / 중심선(MA) × 100 · 변동성이 수축(squeeze)하면 횡보 신호. 3) 추세 없음 : |MA 기울기| ≤ ``UPDOWN_BOX_MA_SLOPE_TOL_PCT`` · 기울기(%) = (MA_now - MA_prev) / MA_prev × 100 · 0 에 가까울수록 추세가 없다(횡보). [하드코딩 금지] - 모든 임계값은 ``get_env_*`` 로 DB/Env 로드 (변수명 대문자 스네이크). """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional, Tuple from ..utils.env import get_env_float, get_env_from_db, get_env_int logger = logging.getLogger("kis_trader.updown_box") # ---------------------------------------------------------------------- # env 파라미터 (그리드·실매 공통 — 종목 행 없으면 글로벌) # ---------------------------------------------------------------------- def get_box_cfg_from_env() -> Dict[str, Any]: """박스 판별/진입 파라미터 글로벌 기본값 (env_config ``UPDOWN_BOX_*``). 파라서치 그리드 끝값·실매 기본 모두 이 함수를 단일 소스로 쓴다. ※ ``ratchet_tiers`` 는 문자열(다단 래칫), 그 외는 숫자. """ return { # 박스 정의에 쓸 분봉 개수 (lookback) "lookback_bars": float(get_env_int("UPDOWN_BOX_LOOKBACK_BARS", 20)), # 박스 최대 높이(%) — 좁을수록 타이트한 박스 "range_max_pct": get_env_float("UPDOWN_BOX_RANGE_MAX_PCT", 8.0), # 볼린저밴드 "bb_period": float(get_env_int("UPDOWN_BOX_BB_PERIOD", 20)), "bb_std": get_env_float("UPDOWN_BOX_BB_STD", 2.0), "bb_bw_max": get_env_float("UPDOWN_BOX_BB_BW_MAX", 12.0), # 이동평균 기울기 "ma_period": float(get_env_int("UPDOWN_BOX_MA_PERIOD", 20)), "ma_slope_tol_pct": get_env_float("UPDOWN_BOX_MA_SLOPE_TOL_PCT", 0.5), # 진입: 박스 하단에서 이 % 이내로 내려와야 (싸게 줍기) "near_low_pct": get_env_float("UPDOWN_BOX_NEAR_LOW_PCT", 2.0), # 진입: 직전봉 음봉(하락 바디) 요구 여부 (1=요구). 되돌림에서 줍기. "require_down_bar": get_env_float("UPDOWN_BOX_REQUIRE_DOWN_BAR", 1.0), # 진입: 직전봉 최소 음봉 몸통 하락률(%) (0=무시) "body_drop_min_pct": get_env_float("UPDOWN_BOX_BODY_DROP_MIN_PCT", 0.2), # 판별 최소 봉 수 (이보다 적으면 판별 불가) "min_bars": float(get_env_int("UPDOWN_BOX_MIN_BARS", 20)), # ── 청산 스택 ──────────────────────────────────────── # 박스 하단 이탈 손절 여유(%) — 노이즈 이탈 방지 (price < low×(1-buf/100)) "breach_buffer_pct": get_env_float("UPDOWN_BOX_BREACH_BUFFER_PCT", 0.3), # 패닉 손절: 보유 중 고점 대비 이 %p 급락하면 무조건 컷 (진입가 무관) "panic_from_high_pct": get_env_float("UPDOWN_PANIC_FROM_HIGH_PCT", 3.0), # 하드 손절(%) — 진입가 대비. (박스/패닉이 안 걸릴 때 최종 방어) "stop_loss_pct": get_env_float("UPDOWN_BOX_STOP_LOSS_PCT", 2.5), # 익절(%) — 진입가 대비 (박스 상단 도달 기대치 근방) "tp_pct": get_env_float("UPDOWN_BOX_TP_PCT", 3.0), # 돌파 추종(breakout follow): ON(1)이면 보유 고점이 박스 상단을 위로 돌파한 뒤에는 # 익절(tp) 청산을 끄고 래칫(트레일)만으로 추종 → 추세를 끝까지 따라간다. # 횡보 중엔 평소대로 tp로 짧게, '확실히 오르는' 돌파 종목은 덜 먹어도 안전하게 추종. # (손절/패닉/박스이탈은 돌파 후에도 그대로 유지) "breakout_follow": float(get_env_int("UPDOWN_BOX_BREAKOUT_FOLLOW", 0)), # 래칫(어깨컷): 고점이 진입가 대비 +X% 도달 시 무장(arm) "shoulder_min_high_pct": get_env_float("UPDOWN_BOX_SHOULDER_MIN_HIGH_PCT", 1.0), # 래칫(어깨컷): 무장 후 고점 대비 -Y% 하락 시 컷 "shoulder_cut_pct": get_env_float("UPDOWN_BOX_SHOULDER_CUT_PCT", 1.0), # 다단 래칫 — TAIL·BREAKOUT 과 동일 형식 "수익%:컷%,…" (예 "2:1.5,5:1.0"). # 값이 있으면 단일 어깨컷 대신 다단 트레일 사용. 빈 값="" → OFF(단일 어깨컷). "ratchet_tiers": str(get_env_from_db("UPDOWN_BOX_RATCHET_TIERS", "") or ""), # 보유 한도(봉) — 이만큼 지나면 무조건 청산 "max_hold_bars": float(get_env_int("UPDOWN_BOX_MAX_HOLD_BARS", 64)), } def parse_box_ratchet_tiers(cfg: Optional[Dict[str, Any]] = None) -> List[Tuple[float, float]]: """다단 래칫 문자열 → [(gain비율, cut비율), …] (TAIL ``_tail_ratchet_tiers`` 와 동일). "2:1.5,5:1.0" → [(0.02, 0.015), (0.05, 0.01)] = +2%부터 고점대비 1.5%컷, +5%부터 1.0%컷. 비어 있으면 [] (= OFF → 단일 어깨컷 사용). """ raw = (cfg or {}).get("ratchet_tiers") if raw is None: raw = get_env_from_db("UPDOWN_BOX_RATCHET_TIERS", "") if isinstance(raw, (list, tuple)): pairs = list(raw) else: s = str(raw or "").strip() if not s: return [] pairs = [] for chunk in s.split(","): chunk = chunk.strip() if not chunk or ":" not in chunk: continue g, c = chunk.split(":", 1) pairs.append((g, c)) tiers: List[Tuple[float, float]] = [] for g, c in pairs: try: gain = abs(float(g)) / 100.0 cut = abs(float(c)) / 100.0 except (TypeError, ValueError): continue if gain <= 0 or cut <= 0: continue tiers.append((gain, cut)) tiers.sort(key=lambda x: x[0]) return tiers # ---------------------------------------------------------------------- # 기초 계산기 (순수 함수) # ---------------------------------------------------------------------- def _sma(values: List[float], period: int) -> Optional[float]: """단순이동평균(SMA) — 최근 period 개. 부족하면 None.""" if period <= 0 or len(values) < period: return None window = values[-period:] return sum(window) / float(period) def _stddev(values: List[float], period: int) -> Optional[float]: """표준편차(모집단) — 볼린저밴드용. 최근 period 개.""" if period <= 0 or len(values) < period: return None window = values[-period:] m = sum(window) / float(period) var = sum((v - m) ** 2 for v in window) / float(period) return var ** 0.5 def bollinger_bandwidth(closes: List[float], period: int, std_mult: float) -> Optional[float]: """볼린저밴드 폭(BandWidth, %) = (상단-하단)/중심선 × 100. 값이 작을수록 변동성 수축(스퀴즈)=횡보 가능성. None 이면 데이터 부족. """ ma = _sma(closes, period) sd = _stddev(closes, period) if ma is None or sd is None or ma <= 0: return None upper = ma + std_mult * sd lower = ma - std_mult * sd return (upper - lower) / ma * 100.0 def ma_slope_pct(closes: List[float], period: int) -> Optional[float]: """MA 기울기(%) = (MA_now - MA_prev) / MA_prev × 100. 직전 봉 기준 MA 와 현재 MA 를 비교. 0 에 가까우면 추세 없음(횡보). None 이면 데이터 부족. """ if len(closes) < period + 1: return None ma_now = _sma(closes, period) ma_prev = _sma(closes[:-1], period) if ma_now is None or ma_prev is None or ma_prev <= 0: return None return (ma_now - ma_prev) / ma_prev * 100.0 def compute_box(candles: List[Dict[str, Any]], lookback: int) -> Optional[Dict[str, float]]: """최근 lookback 봉의 박스 경계 산출. box_low = 최근 lookback 봉 중 최저가(저가의 최소) box_high = 최근 lookback 봉 중 최고가(고가의 최대) Returns: {box_low, box_high, box_mid, range_pct} 또는 None(데이터 부족). """ if lookback <= 0 or len(candles) < lookback: return None window = candles[-lookback:] lows = [float(c.get("low", c.get("close", 0)) or 0) for c in window] highs = [float(c.get("high", c.get("close", 0)) or 0) for c in window] lows = [v for v in lows if v > 0] highs = [v for v in highs if v > 0] if not lows or not highs: return None box_low = min(lows) box_high = max(highs) if box_low <= 0 or box_high <= box_low: return None box_mid = (box_low + box_high) / 2.0 range_pct = (box_high - box_low) / box_low * 100.0 return { "box_low": box_low, "box_high": box_high, "box_mid": box_mid, "range_pct": range_pct, } # ---------------------------------------------------------------------- # 박스 판별 (SCAN/TRIGGER 공용) # ---------------------------------------------------------------------- def evaluate_box( candles: List[Dict[str, Any]], cfg: Optional[Dict[str, float]] = None, ) -> Dict[str, Any]: """종목이 박스권인지 판정 + 박스 정보·품질점수 반환. Returns dict: is_box : bool — 3축(범위·스퀴즈·기울기) 모두 만족 box_low/high/mid/range_pct bb_bw : 볼린저 밴드폭(%) (None 가능) ma_slope : MA 기울기(%) (None 가능) box_score : 품질점수(높을수록 좋은 박스 — eviction 우선순위·정렬용) reason : 탈락 사유(디버그) """ c = cfg or get_box_cfg_from_env() out: Dict[str, Any] = { "is_box": False, "box_low": 0.0, "box_high": 0.0, "box_mid": 0.0, "range_pct": 0.0, "bb_bw": None, "ma_slope": None, "box_score": 0.0, "reason": "", } lookback = int(c.get("lookback_bars", 20)) min_bars = int(c.get("min_bars", 20)) if len(candles) < max(min_bars, lookback): out["reason"] = f"봉부족({len(candles)}<{max(min_bars, lookback)})" return out box = compute_box(candles, lookback) if not box: out["reason"] = "박스경계 산출불가" return out out.update(box) closes = [float(x.get("close", 0) or 0) for x in candles] closes = [v for v in closes if v > 0] bb_period = int(c.get("bb_period", 20)) bb_std = float(c.get("bb_std", 2.0)) bb_bw = bollinger_bandwidth(closes, bb_period, bb_std) out["bb_bw"] = bb_bw ma_period = int(c.get("ma_period", 20)) slope = ma_slope_pct(closes, ma_period) out["ma_slope"] = slope # ── 3축 판별 ─────────────────────────────────────────────── range_max = float(c.get("range_max_pct", 8.0)) bw_max = float(c.get("bb_bw_max", 12.0)) slope_tol = float(c.get("ma_slope_tol_pct", 0.5)) if box["range_pct"] > range_max: out["reason"] = f"범위과대({box['range_pct']:.1f}>{range_max:.1f}%)" return out if bb_bw is not None and bb_bw > bw_max: out["reason"] = f"밴드폭과대({bb_bw:.1f}>{bw_max:.1f}%)" return out if slope is not None and abs(slope) > slope_tol: out["reason"] = f"추세존재(|{slope:.2f}|>{slope_tol:.2f}%)" return out out["is_box"] = True # 품질점수: 박스가 좁고(range_pct↓)·스퀴즈 강하고(bb_bw↓)·평평할수록(|slope|↓) 高. # 정규화 단순화 — 작을수록 좋은 3지표의 역수 합 (eviction/정렬 우선순위용). rp = max(0.1, box["range_pct"]) bw = max(0.1, bb_bw if bb_bw is not None else range_max) sl = max(0.01, abs(slope) if slope is not None else slope_tol) out["box_score"] = round((1.0 / rp) * 50 + (1.0 / bw) * 30 + (1.0 / sl) * 5, 3) return out def near_box_low(price: float, box_low: float, box_high: float, near_pct: float) -> bool: """현재가가 박스 하단에서 near_pct% 이내인가 (싸게 줍는 진입 구역). near = box_low × (1 + near_pct/100). price ≤ near 이면 True. 하단 이탈(price < box_low)도 '이내'로 본다(진입 판정은 호출부에서 음봉 등과 조합). """ if box_low <= 0 or box_high <= box_low or price <= 0: return False threshold = box_low * (1.0 + max(0.0, near_pct) / 100.0) return price <= threshold def box_low_breached(price: float, box_low: float, buffer_pct: float = 0.0) -> bool: """박스 하단 이탈 여부 (구조적 손절 트리거). price < box_low × (1 - buffer_pct/100) 이면 박스가 깨진 것 → 손절 신호. buffer_pct 는 노이즈 이탈 방지용 여유(기본 0). """ if box_low <= 0 or price <= 0: return False floor_px = box_low * (1.0 - max(0.0, buffer_pct) / 100.0) return price < floor_px # ---------------------------------------------------------------------- # 진입 신호 (SCAN 통과 후 TRIGGER — 박스 하단 되돌림 매수) # ---------------------------------------------------------------------- def box_entry_signal( candles: List[Dict[str, Any]], cfg: Optional[Dict[str, float]] = None, ) -> Tuple[bool, str, Dict[str, Any]]: """박스 하단 되돌림 매수 신호. 조건(모두 만족): 1) 박스권일 것 (``evaluate_box.is_box``) 2) 직전 확정봉(candles[-2]) 종가가 박스 하단 near_low_pct 이내 (싸게) 3) (옵션) 직전봉이 음봉이고 몸통 하락률 ≥ body_drop_min_pct (되돌림) 진입가는 최신 확정봉(candles[-1]) 시가 — 백테 '다음봉 시가' 와 동일. Returns: (ok, reason, info) — info 에 box_low/high, entry_ref 등. """ c = cfg or get_box_cfg_from_env() info: Dict[str, Any] = {} if len(candles) < 2: return (False, "봉부족(<2)", info) box = evaluate_box(candles[:-1], c) # 신호 시점까지의 확정봉으로 박스 판정 info.update(box) if not box.get("is_box"): return (False, f"비박스:{box.get('reason','')}", info) box_low = float(box["box_low"]) box_high = float(box["box_high"]) sig_bar = candles[-2] # 신호봉 (직전 확정봉) ent_bar = candles[-1] # 진입봉 (최신 확정봉) — 시가 진입 sig_close = float(sig_bar.get("close", 0) or 0) sig_open = float(sig_bar.get("open", sig_close) or sig_close) ent_open = float(ent_bar.get("open", ent_bar.get("close", 0)) or 0) info["entry_ref"] = ent_open near_pct = float(c.get("near_low_pct", 2.0)) if not near_box_low(sig_close, box_low, box_high, near_pct): return (False, f"하단미근접(c={sig_close:.0f}/low={box_low:.0f})", info) require_down = float(c.get("require_down_bar", 1.0)) >= 0.5 if require_down: if not (sig_close < sig_open and sig_open > 0): return (False, "직전봉 음봉아님", info) body_min = float(c.get("body_drop_min_pct", 0.0)) if body_min > 0: drop_pct = (sig_open - sig_close) / sig_open * 100.0 if drop_pct < body_min: return (False, f"음봉약함({drop_pct:.2f}<{body_min:.2f}%)", info) if ent_open <= 0: return (False, "진입봉 시가 0", info) return (True, "박스하단되돌림", info) # ---------------------------------------------------------------------- # 청산 스택 (백테=실매 공용 — 단일 가격레벨 평가기) # ---------------------------------------------------------------------- def eval_box_exit_at_price( avg: float, max_price: float, px: float, *, box_low: float, breach_buffer_pct: float, panic_from_high_pct: float, sl_eff: float, tp_eff: float, shoulder_min_high: float, shoulder_cut_pct: float, ratchet_tiers: Optional[List[Tuple[float, float]]] = None, box_high: float = 0.0, breakout_follow: bool = False, ) -> Tuple[Optional[Tuple[str, float]], float]: """UPDOWN 박스 청산 — 단일 가격레벨 우선순위 평가 (백테 intrabar = 실매 틱 동일). 우선순위: 1) 박스 하단 이탈 (구조 붕괴 → 즉시 손절) 2) 패닉(고점 대비 -X%) (무조건 긴급컷, 진입가 무관) 3) 래칫 트레일 (tiers 있으면 다단 래칫컷 / 없으면 단일 어깨컷) 4) 익절 (+tp_eff) ← **돌파 추종 ON + 박스 상단 돌파 시 이 단계만 스킵** 5) 하드 손절 (-sl_eff) 돌파 추종(breakout_follow): 보유 고점(max_p)이 box_high 를 위로 넘어선 뒤에는 익절(4)을 끄고 래칫(3)이 천장 역할 → 추세를 끝까지 따라간다. 손절·패닉·박스이탈(1,2,5)은 유지. Args: sl_eff, tp_eff, shoulder_* : 소수 비율(0.025 = 2.5%) ratchet_tiers : [(gain비율, cut비율), …] (TAIL/BREAKOUT 동일). 있으면 단일 어깨 대체. box_high : 박스 상단가(0이면 돌파판정 불가 → tp 정상 작동) breakout_follow : True 면 박스 상단 돌파 후 tp 해제(래칫 추종) Returns: ((reason, exit_px) 또는 None, 갱신된 max_price) """ if avg <= 0 or px <= 0: return None, max_price max_p = max(float(max_price or avg), px) # 박스 상단 돌파 여부 — 보유 중 고점이 박스 천장을 위로 넘었으면 '추세 전환'으로 본다. broke_out = bool(breakout_follow) and box_high > 0 and max_p > box_high # 1순위: 박스 하단 이탈 (구조적) if box_low > 0 and box_low_breached(px, box_low, breach_buffer_pct): return ("박스이탈", float(px)), max_p # 2순위: 패닉 — 보유 고점 대비 급락 (진입가 무관, 무조건) if panic_from_high_pct > 0 and max_p > 0: panic_floor = max_p * (1.0 - panic_from_high_pct / 100.0) if px <= panic_floor: return (f"패닉(고점-{panic_from_high_pct:.1f}%)", float(px)), max_p # 3순위: 래칫 트레일 — TAIL/BREAKOUT 과 동일하게 tiers 있으면 다단 래칫(단일 어깨 대체) if ratchet_tiers: peak_gain = (max_p - avg) / avg cut_ratio = 0.0 for gain_thr, cut in ratchet_tiers: if peak_gain >= gain_thr: cut_ratio = cut # 더 높은 수익 단계일수록 타이트한 컷으로 갱신 if cut_ratio > 0.0: ratchet_line = max_p * (1.0 - cut_ratio) if px <= ratchet_line: return ("래칫컷", float(ratchet_line)), max_p else: # 단일 어깨컷 (이익 무장 후 트레일) trail_armed = max_p >= avg * (1.0 + shoulder_min_high) if trail_armed: trail_stop_px = max_p * (1.0 - shoulder_cut_pct) if px <= trail_stop_px: return ("어깨컷", float(trail_stop_px)), max_p pnl_pct = (px - avg) / avg # 4순위: 익절 — 단, 박스 상단 돌파(추종 ON) 상태면 tp 를 끄고 래칫(3)이 천장 역할. if pnl_pct >= tp_eff and not broke_out: return (f"익절(+{pnl_pct * 100:.2f}/{tp_eff * 100:.2f}%)", float(px)), max_p # 5순위: 하드 손절 if pnl_pct <= -sl_eff: return (f"손절({pnl_pct * 100:.2f}/{sl_eff * 100:.2f}%)", float(px)), max_p return None, max_p # ---------------------------------------------------------------------- # 백테스트 러너 (실매와 동일 진입/청산 로직 공유) # ---------------------------------------------------------------------- def run_box_backtest_rust_experimental( candles: List[Dict[str, Any]], cfg: Dict[str, float] ) -> Dict[str, Any]: try: import kis_rust_core except ImportError: raise RuntimeError("kis_rust_core module is not available") def get_float(k, default=0.0): try: return float(cfg.get(k, default)) except: return float(default) def get_int(k, default=0): try: return int(cfg.get(k, default)) except: return int(default) rust_params = kis_rust_core.BoxParams( get_int("lookback_bars", 20), get_float("range_max_pct", 8.0), get_int("bb_period", 20), get_float("bb_std", 2.0), get_float("bb_bw_max", 12.0), get_int("ma_period", 20), get_float("ma_slope_tol_pct", 0.5), get_float("stop_loss_pct", 2.5), get_float("tp_pct", 3.0), get_int("max_hold_bars", 64), ) rust_candles = [] # 보통 단일 종목 테스트 target_code = "000000" # dummy if candles and "code" in candles[0]: target_code = candles[0]["code"] for c in candles: rc = kis_rust_core.CandleData( c.get("candle_time", ""), float(c.get("open", 0)), float(c.get("high", 0)), float(c.get("low", 0)), float(c.get("close", 0)), float(c.get("volume", 0)), float(c.get("rsi", 0.0)), ) rust_candles.append(rc) trades_rs = kis_rust_core.run_box_backtest_fast(target_code, rust_candles, rust_params) # 박스권 결과 포맷으로 변환 trades = [] total_pnl = 0.0 wins = 0 for t in trades_rs: trades.append({ "buy_time": t.buy_time, "sell_time": t.sell_time, "buy_price": t.buy_price, "sell_price": t.sell_price, "entry_time": t.buy_time, "exit_time": t.sell_time, "buy": t.buy_price, "sell": t.sell_price, "reason": t.reason, "pnl_pct": t.pnl_pct, "exit_reason": t.reason }) total_pnl += t.pnl_pct if t.pnl_pct > 0: wins += 1 n_trades = len(trades) win_rate = (wins / n_trades * 100) if n_trades > 0 else 0.0 return { "trades": trades, "n_trades": n_trades, "pf": 0.0, "win_rate": win_rate, "total_pnl": total_pnl } def run_backtest_box( candles: List[Dict[str, Any]], cfg: Optional[Dict[str, float]] = None, fee_rate: float = 0.015 / 100, sell_tax: float = 0.18 / 100, ) -> Dict[str, Any]: """분봉 리스트(시간 오름차순)로 박스 전략 백테스트. 진입: ``box_entry_signal`` (박스+하단근접+직전음봉), 진입가=해당봉 시가. 청산: ``eval_box_exit_at_price`` (박스이탈→패닉→어깨컷→익절→손절) + 보유한도. 봉내 체결: ``_intrabar_exit_prices`` 로 실매 틱과 동일하게 근사. Returns: 거래목록·요약(PF/승률/총손익) dict. """ from .scalping_engine import _intrabar_exit_prices c = cfg or get_box_cfg_from_env() if c.get("use_rust", 0) == 1 or c.get("use_rust") == True: return run_box_backtest_rust_experimental(candles, c) n = len(candles) trades: List[Dict[str, Any]] = [] if n < int(c.get("min_bars", 20)) + 2: return {"trades": [], "n_trades": 0, "pf": 0.0, "win_rate": 0.0, "total_pnl": 0.0} sl_eff = max(0.0001, float(c.get("stop_loss_pct", 2.5)) / 100.0) tp_eff = max(0.0001, float(c.get("tp_pct", 3.0)) / 100.0) smh = max(0.0, float(c.get("shoulder_min_high_pct", 1.0)) / 100.0) scut = max(0.0, float(c.get("shoulder_cut_pct", 1.0)) / 100.0) panic = max(0.0, float(c.get("panic_from_high_pct", 3.0))) breach_buf = max(0.0, float(c.get("breach_buffer_pct", 0.3))) max_hold = int(float(c.get("max_hold_bars", 64))) tiers = parse_box_ratchet_tiers(c) # 다단 래칫(있으면 단일 어깨 대체) bo_follow = bool(float(c.get("breakout_follow", 0) or 0)) # 박스 상단 돌파 추종 n_checks = max(1, get_env_int("BACKTEST_EXIT_CHECKS_PER_BAR", 6)) pos: Optional[Dict[str, Any]] = None lookback = int(c.get("lookback_bars", 20)) i = lookback while i < n: bar = candles[i] o = float(bar.get("open", bar.get("close", 0)) or 0) h = float(bar.get("high", bar.get("close", 0)) or 0) l = float(bar.get("low", bar.get("close", 0)) or 0) cl = float(bar.get("close", 0) or 0) if cl <= 0: i += 1 continue if pos is None: # 진입 판정: 신호봉=i-1, 진입봉=i (candles[:i+1] 의 마지막이 진입봉) ok, _reason, info = box_entry_signal(candles[: i + 1], c) if ok and o > 0: pos = { "entry_i": i, "avg": o, "max_price": o, "box_low": float(info.get("box_low", 0) or 0), "box_high": float(info.get("box_high", 0) or 0), } i += 1 continue # 보유 중 — 봉내 체결 우선순위 평가 avg = float(pos["avg"]) max_p = max(float(pos["max_price"]), h) box_low = float(pos["box_low"]) box_high = float(pos.get("box_high", 0) or 0) prices = [cl] if n_checks <= 1 else _intrabar_exit_prices(o, h, l, cl, n_checks) exit_res: Optional[Tuple[str, float]] = None for px in prices: res, max_p = eval_box_exit_at_price( avg, max_p, float(px), box_low=box_low, breach_buffer_pct=breach_buf, panic_from_high_pct=panic, sl_eff=sl_eff, tp_eff=tp_eff, shoulder_min_high=smh, shoulder_cut_pct=scut, ratchet_tiers=tiers, box_high=box_high, breakout_follow=bo_follow, ) if res: exit_res = res break pos["max_price"] = max_p bars_held = i - int(pos["entry_i"]) if exit_res is None and bars_held >= max_hold and bars_held > 0: exit_res = (f"보유한도({max_hold}봉)", cl) if exit_res: reason, exit_px = exit_res buy_cost = avg * (1.0 + fee_rate) sell_rev = exit_px * (1.0 - fee_rate - sell_tax) pnl_pct = (sell_rev - buy_cost) / buy_cost * 100.0 ent_bar = candles[int(pos["entry_i"])] trades.append({ "entry_i": int(pos["entry_i"]), "exit_i": i, "entry_time": str(ent_bar.get("candle_time") or ent_bar.get("candle_time_str") or ""), "exit_time": str(bar.get("candle_time") or bar.get("candle_time_str") or ""), "buy": avg, "sell": exit_px, "reason": reason, "pnl_pct": round(pnl_pct, 4), "bars_held": bars_held, }) pos = None i += 1 wins = [t for t in trades if t["pnl_pct"] > 0] losses = [t for t in trades if t["pnl_pct"] <= 0] gross_win = sum(t["pnl_pct"] for t in wins) gross_loss = abs(sum(t["pnl_pct"] for t in losses)) pf = (gross_win / gross_loss) if gross_loss > 0 else (gross_win if gross_win > 0 else 0.0) total_pnl = sum(t["pnl_pct"] for t in trades) # === B&H(단순보유) 벤치마크 === # "그냥 처음부터 끝까지 들고만 있었으면?" — 봇이 첫 매수 판정을 시작하는 봉(lookback)의 # 시가에 1주 사서 마지막 봉 종가에 판 수익률. 거래와 동일한 수수료·세금을 적용해 공정 비교. # 횡보(박스권) 종목은 B&H가 0% 근처일 것이므로, 봇의 total_pnl 이 이를 얼마나 초과(alpha) # 했는지가 전략의 실효 판단 기준이 된다. buy_hold_pct = 0.0 if n > lookback: bh_buy = float(candles[lookback].get("open", candles[lookback].get("close", 0)) or 0) bh_sell = float(candles[-1].get("close", 0) or 0) if bh_buy > 0 and bh_sell > 0: bh_cost = bh_buy * (1.0 + fee_rate) bh_rev = bh_sell * (1.0 - fee_rate - sell_tax) buy_hold_pct = (bh_rev - bh_cost) / bh_cost * 100.0 return { "trades": trades, "n_trades": len(trades), "pf": round(pf, 3), "win_rate": round(len(wins) / len(trades) * 100.0, 2) if trades else 0.0, "total_pnl": round(total_pnl, 4), "buy_hold_pct": round(buy_hold_pct, 4), # 단순보유 수익률(같은 구간·수수료) "alpha_pct": round(total_pnl - buy_hold_pct, 4), # 봇 초과수익(전략 실효) } # ---------------------------------------------------------------------- # 라이브 신호 (백테=실매 동일 — updow_buy.check_*_live 와 동일 시그니처 패턴) # ---------------------------------------------------------------------- def _norm_candle_time_key(c: Dict[str, Any]) -> str: ct = c.get("candle_time") or c.get("candle_time_str", "") if isinstance(ct, str) and len(ct) == 19 and " " in ct: ct = ct.replace("-", "").replace(" ", "").replace(":", "")[:12] return str(ct or "").strip()[:12] def check_buy_signal_box_live( candles: List[Dict[str, Any]], cfg: Optional[Dict[str, float]] = None, *, last_fired_entry_key: Optional[str] = None, ) -> Tuple[Optional[str], str, Optional[Dict[str, Any]]]: """라이브 매수 — ``box_entry_signal`` 과 동일 (백테 ``run_backtest_box`` 정합). Returns: (reject_code, message, signal_or_none) """ c = cfg or get_box_cfg_from_env() min_need = max(2, int(c.get("min_bars", 20))) if len(candles) < min_need: return ("탈락-봉부족", f"확정봉 {len(candles)}개 (최소 {min_need})", None) if len(candles) < 2: return ("탈락-봉부족", "신호·진입 봉 2개 이상 필요", None) ent_bar = candles[-1] ent_key = _norm_candle_time_key(ent_bar) if not ent_key: return ("탈락-봉시각", "진입봉 candle_time 비어 있음", None) if last_fired_entry_key and last_fired_entry_key == ent_key: return ("탈락-중복진입봉", f"이번 진입봉({ent_key})에 이미 주문 시도함", None) ok, reason, info = box_entry_signal(candles, c) if not ok: return ("탈락-비신호", reason, None) entry_open = float(info.get("entry_ref", 0) or ent_bar.get("open", 0) or 0) if entry_open <= 0: return ("탈락-시가없음", "진입봉 시가 0", None) sl_pct = float(c.get("stop_loss_pct", 2.5)) / 100.0 tp_pct = float(c.get("tp_pct", 3.0)) / 100.0 box_low = float(info.get("box_low", 0) or 0) sig = { "entry_price": entry_open, "updow_entry_bar_key": ent_key, "signal_candle_key": _norm_candle_time_key(candles[-2]), "stop_price": entry_open * (1.0 - sl_pct), "target_price": entry_open * (1.0 + tp_pct), "sl_pct": sl_pct * 100.0, "tp_pct": tp_pct * 100.0, "box_low": box_low, "box_high": float(info.get("box_high", 0) or 0), "box_score": float(info.get("box_score", 0) or 0), "atr_entry": 0.0, } return (None, f"박스하단되돌림 → 시가 {entry_open:.0f} (low={box_low:.0f})", sig) def check_sell_signal_box_live( *, buy_price: float, candles: List[Dict[str, Any]], cfg: Optional[Dict[str, float]] = None, entry_bar_key: str, buy_time_str: str, current_price: float, max_price: float = 0.0, box_low: float = 0.0, box_high: float = 0.0, ) -> Optional[Tuple[str, float]]: """라이브 매도 — ``eval_box_exit_at_price`` (박스이탈→패닉→어깨→익절→손절) + 보유한도. ``box_low``/``box_high`` 는 진입 시점 박스 경계( watchlist / entry_features 저장값). ``box_high`` + cfg.breakout_follow 로 박스 상단 돌파 추종(익절 해제→래칫 추종) 판정. """ from ..strategies.updow_buy import estimate_updow_entry_bar_index if buy_price <= 0 or not candles: return None c = cfg or get_box_cfg_from_env() sl_eff = max(0.0001, float(c.get("stop_loss_pct", 2.5)) / 100.0) tp_eff = max(0.0001, float(c.get("tp_pct", 3.0)) / 100.0) smh = max(0.0, float(c.get("shoulder_min_high_pct", 1.0)) / 100.0) scut = max(0.0, float(c.get("shoulder_cut_pct", 1.0)) / 100.0) panic = max(0.0, float(c.get("panic_from_high_pct", 3.0))) breach_buf = max(0.0, float(c.get("breach_buffer_pct", 0.3))) max_hold = max(1, int(float(c.get("max_hold_bars", 64)))) last = candles[-1] try: c_now = float(last.get("close", 0) or 0) h_now = float(last.get("high", c_now) or c_now) except (TypeError, ValueError): return None if c_now <= 0: return None px = float(current_price) if current_price > 0 else c_now mp = max(float(max_price or buy_price), h_now, px) j = estimate_updow_entry_bar_index(candles, entry_bar_key, buy_time_str) last_idx = len(candles) - 1 if j >= 0 and last_idx <= j: return None bars_held = 0 if j < 0 else (last_idx - j) bl = float(box_low or 0) bh = float(box_high or 0) bo_follow = bool(float(c.get("breakout_follow", 0) or 0)) res, _ = eval_box_exit_at_price( buy_price, mp, px, box_low=bl, breach_buffer_pct=breach_buf, panic_from_high_pct=panic, sl_eff=sl_eff, tp_eff=tp_eff, shoulder_min_high=smh, shoulder_cut_pct=scut, ratchet_tiers=parse_box_ratchet_tiers(c), box_high=bh, breakout_follow=bo_follow, ) if res: return res if j >= 0 and bars_held >= max_hold and bars_held > 0: return (f"보유한도({max_hold}봉)", c_now) return None # ---------------------------------------------------------------------- # 파라서치 그리드 (env ``UPDOWN_BOX_GRID_*`` — 하드코딩 0) # ---------------------------------------------------------------------- def _grid_vals(env_key: str, default_csv: str) -> List[float]: from ..utils.env import get_env_from_db s = (get_env_from_db(env_key, default_csv) or default_csv).strip() out: List[float] = [] for part in s.split(","): part = part.strip() if not part: continue try: out.append(float(part)) except ValueError: pass return out or [float(x) for x in default_csv.split(",") if x.strip()] def _grid_strs(env_key: str, default_semicolon: str) -> List[str]: """문자열 그리드 후보(다단 래칫) — 세미콜론(;) 구분. 빈 문자열 후보="" (OFF) 포함 가능. 예 ``";2:1.5,5:1.0"`` → ["", "2:1.5,5:1.0"] (OFF + 1프리셋). 래칫 내부는 쉼표라 구분자는 ;. """ from ..utils.env import get_env_from_db raw = get_env_from_db(env_key, default_semicolon) raw = default_semicolon if raw is None else str(raw) out: List[str] = [] seen = set() for chunk in raw.split(";"): c = chunk.strip() if c not in seen: seen.add(c) out.append(c) return out or [""] def default_box_param_grid(mode: str = "fast") -> Dict[str, List[Any]]: """박스권 UPDOWN 그리드 축. mode=fast 는 축 수·값 축소. ``ratchet_tiers`` 는 문자열 후보(다단 래칫) — 빈 값="" 은 단일 어깨컷(OFF). """ if mode == "full": return { "near_low_pct": _grid_vals("UPDOWN_BOX_GRID_NEAR0", "1.0,1.5,2.0,2.5,3.0"), "tp_pct": _grid_vals("UPDOWN_BOX_GRID_TP0", "2.0,2.5,3.0,3.5,4.0,5.0"), "stop_loss_pct": _grid_vals("UPDOWN_BOX_GRID_SL0", "2.0,2.5,3.0,3.5"), "shoulder_min_high_pct": _grid_vals("UPDOWN_BOX_GRID_SHOULDER_ARM0", "0.5,1.0,1.5,2.0"), "shoulder_cut_pct": _grid_vals("UPDOWN_BOX_GRID_SHOULDER_CUT0", "0.5,1.0,1.5"), "panic_from_high_pct": _grid_vals("UPDOWN_BOX_GRID_PANIC0", "2.0,2.5,3.0,4.0"), "range_max_pct": _grid_vals("UPDOWN_BOX_GRID_RANGE0", "6.0,8.0,10.0"), # 다단 래칫 후보: OFF(단일 어깨컷) + 2 프리셋 "ratchet_tiers": _grid_strs("UPDOWN_BOX_GRID_RATCHET0", ";1.5:1.0,3:0.7;2:1.5,5:1.0"), # 돌파추종 후보: 0=OFF(횡보 익절형) / 1=ON(박스 상단 돌파 후 tp 해제·래칫 추종) "breakout_follow": _grid_vals("UPDOWN_BOX_GRID_BREAKOUT0", "0,1"), } # fast — 핵심 4축 + 래칫(OFF/1프리셋) + 돌파추종(OFF/ON) return { "near_low_pct": _grid_vals("UPDOWN_BOX_GRID_NEAR_FAST", "1.5,2.0,2.5"), "tp_pct": _grid_vals("UPDOWN_BOX_GRID_TP_FAST", "2.5,3.0,3.5,4.0"), "stop_loss_pct": _grid_vals("UPDOWN_BOX_GRID_SL_FAST", "2.0,2.5,3.0"), "shoulder_cut_pct": _grid_vals("UPDOWN_BOX_GRID_SHOULDER_CUT_FAST", "0.8,1.0,1.2"), "ratchet_tiers": _grid_strs("UPDOWN_BOX_GRID_RATCHET_FAST", ";2:1.5,5:1.0"), # 돌파추종 0/1 — 종목 성격(횡보 vs 돌파)에 맞는 쪽을 α 기준으로 자동 선택 "breakout_follow": _grid_vals("UPDOWN_BOX_GRID_BREAKOUT_FAST", "0,1"), } def run_param_search_box( candles: List[Dict[str, Any]], *, grid: Optional[Dict[str, List[float]]] = None, base_cfg: Optional[Dict[str, float]] = None, min_trades: int = 1, max_combos: int = 800, rank_by: str = "alpha", min_alpha: Optional[float] = None, ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """박스 전략 그리드 탐색 — ``run_backtest_box`` 단일 소스. rank_by: "alpha"=B&H 초과수익(α) 우선 정렬(기본) | "pnl"=절대 총손익 우선. min_alpha: None=필터 없음 | 값 지정 시 α≤min_alpha 조합 제외(0.0 이면 단순보유만 못한 조합 제거). """ from itertools import product g = grid or default_box_param_grid("fast") base = dict(base_cfg or get_box_cfg_from_env()) keys = list(g.keys()) combos = list(product(*[g[k] for k in keys])) if max_combos > 0 and len(combos) > max_combos: combos = combos[:max_combos] results: List[Dict[str, Any]] = [] for combo in combos: cfg = dict(base) apply_cfg: Dict[str, Any] = {} for k, v in zip(keys, combo): if k == "ratchet_tiers": cfg[k] = str(v) apply_cfg[k] = str(v) else: cfg[k] = float(v) apply_cfg[k] = float(v) bt = run_backtest_box(candles, cfg) if bt["n_trades"] < min_trades: continue alpha = float(bt.get("alpha_pct", bt["total_pnl"])) if min_alpha is not None and alpha <= float(min_alpha): continue results.append({ "apply_cfg": apply_cfg, "n_trades": bt["n_trades"], "pf": bt["pf"], "win_rate": bt["win_rate"], "total_pnl": bt["total_pnl"], "buy_hold_pct": bt.get("buy_hold_pct", 0.0), "alpha_pct": round(alpha, 4), }) if str(rank_by).lower() == "pnl": results.sort(key=lambda r: (-float(r["total_pnl"]), -float(r["pf"]), -int(r["n_trades"]))) else: results.sort(key=lambda r: (-float(r["alpha_pct"]), -float(r["pf"]), -int(r["n_trades"]))) meta = { "cartesian_product": len(combos), "backtests_run": len(combos), "passed": len(results), "rank_by": str(rank_by).lower(), "grid_axes": {k: g[k] for k in keys}, } return results, meta def run_param_search_box_multi( candles_by_code: Dict[str, List[Dict[str, Any]]], *, grid: Optional[Dict[str, List[float]]] = None, base_cfg: Optional[Dict[str, float]] = None, min_trades_total: int = 3, max_combos: int = 800, rank_by: str = "alpha", rust_session_id: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """여러 종목 분봉을 같은 글로벌 파라미터로 백테 → 조합별 합산 성과. 글로벌 튜닝용 — 종목별 ``run_backtest_box`` 결과를 합산(총손익 합·총거래·통합 PF). 실매가 글로벌 ``UPDOWN_BOX_*`` 단일 파라미터로 모든 watchlist 종목을 돌리는 것과 정합. rank_by: "alpha"=합산 B&H 초과수익(α) 우선 정렬(기본) | "pnl"=합산 절대손익 우선. 추세종목(B&H 압도)이 절대손익으로 1위를 오염시키는 것을 막기 위해 기본은 α 정렬. """ from itertools import product g = grid or default_box_param_grid("fast") base = dict(base_cfg or get_box_cfg_from_env()) keys = list(g.keys()) combos = list(product(*[g[k] for k in keys])) if max_combos > 0 and len(combos) > max_combos: combos = combos[:max_combos] results: List[Dict[str, Any]] = [] for combo in combos: cfg = dict(base) apply_cfg: Dict[str, Any] = {} for k, v in zip(keys, combo): if k == "ratchet_tiers": # 다단 래칫은 문자열 축 cfg[k] = str(v) apply_cfg[k] = str(v) else: cfg[k] = float(v) apply_cfg[k] = float(v) tot_pnl = 0.0 tot_bh = 0.0 tot_trades = 0 gross_win = 0.0 gross_loss = 0.0 wins = 0 per_code: Dict[str, float] = {} per_code_alpha: Dict[str, float] = {} for code, candles in candles_by_code.items(): if rust_session_id: try: import kis_rust_core def get_float(k, default=0.0): try: return float(cfg.get(k, default)) except: return float(default) def get_int(k, default=0): try: return int(cfg.get(k, default)) except: return int(default) bp = kis_rust_core.BoxParams( get_int("time_start_hm", 900), get_int("time_end_hm", 1520), get_float("near_low_pct", 0.02), get_float("range_max_pct", 0.08), get_float("bb_bw_max", 15.0), get_float("ma_slope_tol_pct", 0.02), get_float("body_drop_min_pct", 0.0), get_float("breach_buffer_pct", 0.015), get_float("panic_from_high_pct", 0.08), get_float("stop_loss_pct", 0.03), get_float("tp_pct", 0.05), get_float("shoulder_min_high_pct", 0.005), get_float("shoulder_cut_pct", 0.003), get_int("lookback_bars", 20), get_int("max_hold_bars", 0), str(cfg.get("ratchet_tiers", "")), float(cfg.get("breakout_follow", 0.0)) > 0, get_int("max_daily", 10), ) r_trades = kis_rust_core.run_engine_trial_box(rust_session_id, bp) # _convert_rust_trades 코드가 필요하지만, 우선 pnl 등만 필요하므로 단순 계산 tot_pnl_code = sum(t["pnl_pct"] for t in r_trades) bt = { "total_pnl": tot_pnl_code, "buy_hold_pct": (candles[-1]["close"]/candles[0]["close"] - 1.0)*100.0 if candles else 0.0, "n_trades": len(r_trades), "trades": r_trades, "alpha_pct": tot_pnl_code - ((candles[-1]["close"]/candles[0]["close"] - 1.0)*100.0 if candles else 0.0) } except Exception as e: import logging logging.getLogger("kis_trader.engine").warning(f"Rust engine failed for updown_box: {e}") bt = run_backtest_box(candles, cfg) else: bt = run_backtest_box(candles, cfg) tot_pnl += float(bt["total_pnl"]) tot_bh += float(bt.get("buy_hold_pct", 0.0)) tot_trades += int(bt["n_trades"]) per_code[code] = float(bt["total_pnl"]) per_code_alpha[code] = float(bt.get("alpha_pct", bt["total_pnl"])) for t in bt["trades"]: p = float(t["pnl_pct"]) if p > 0: gross_win += p wins += 1 else: gross_loss += abs(p) if tot_trades < min_trades_total: continue pf = (gross_win / gross_loss) if gross_loss > 0 else (gross_win if gross_win > 0 else 0.0) tot_alpha = tot_pnl - tot_bh results.append({ "apply_cfg": apply_cfg, "n_trades": tot_trades, "pf": round(pf, 3), "win_rate": round(wins / tot_trades * 100.0, 2) if tot_trades else 0.0, "total_pnl": round(tot_pnl, 4), "buy_hold_pct": round(tot_bh, 4), "alpha_pct": round(tot_alpha, 4), "per_code_pnl": {k: round(v, 3) for k, v in per_code.items()}, "per_code_alpha": {k: round(v, 3) for k, v in per_code_alpha.items()}, }) if str(rank_by).lower() == "pnl": results.sort(key=lambda r: (-float(r["total_pnl"]), -float(r["pf"]), -int(r["n_trades"]))) else: results.sort(key=lambda r: (-float(r["alpha_pct"]), -float(r["pf"]), -int(r["n_trades"]))) meta = { "cartesian_product": len(combos), "backtests_run": len(combos) * max(1, len(candles_by_code)), "passed": len(results), "rank_by": str(rank_by).lower(), "codes": list(candles_by_code.keys()), "grid_axes": {k: g[k] for k in keys}, } return results, meta def box_cfg_to_env_patch(cfg: Dict[str, Any]) -> Dict[str, Any]: """그리드 apply_cfg → env_config ``UPDOWN_BOX_*`` 키 맵 (ratchet_tiers 는 문자열).""" key_map = { "near_low_pct": "UPDOWN_BOX_NEAR_LOW_PCT", "range_max_pct": "UPDOWN_BOX_RANGE_MAX_PCT", "bb_bw_max": "UPDOWN_BOX_BB_BW_MAX", "ma_slope_tol_pct": "UPDOWN_BOX_MA_SLOPE_TOL_PCT", "body_drop_min_pct": "UPDOWN_BOX_BODY_DROP_MIN_PCT", "breach_buffer_pct": "UPDOWN_BOX_BREACH_BUFFER_PCT", "panic_from_high_pct": "UPDOWN_PANIC_FROM_HIGH_PCT", "stop_loss_pct": "UPDOWN_BOX_STOP_LOSS_PCT", "tp_pct": "UPDOWN_BOX_TP_PCT", "shoulder_min_high_pct": "UPDOWN_BOX_SHOULDER_MIN_HIGH_PCT", "shoulder_cut_pct": "UPDOWN_BOX_SHOULDER_CUT_PCT", "ratchet_tiers": "UPDOWN_BOX_RATCHET_TIERS", "max_hold_bars": "UPDOWN_BOX_MAX_HOLD_BARS", "lookback_bars": "UPDOWN_BOX_LOOKBACK_BARS", "breakout_follow": "UPDOWN_BOX_BREAKOUT_FOLLOW", } patch: Dict[str, Any] = {} for k, env_k in key_map.items(): if k in cfg: patch[env_k] = cfg[k] # ratchet_tiers 는 문자열, 그 외 숫자 return patch