#!/usr/bin/env python3 """ kis_trader/engine/scalping_engine.py — 스캘핑(SCALP reversal) 백테·파서치·실매 공통 엔진 ==================================================================== SCALP reversal 전용. 모멘텀(MOMENTUM)은 ``momentum_engine.py`` 로 완전 분리됨. ■ 라이브·백테 검증 원칙 (뇌동 분기 금지) - SCALP 진입/청산: 웹·CLI·실매가 ``check_buy_signal_*`` / ``check_sell_signal_live`` 동일 경로. - MOMENTUM: ``momentum_engine`` 전용 (SCALP 와 분리). ■ SCALP reversal 공통 로직 (A안 — RSI V자 + scalp_re HTS SCAN) - SCAN: 키움 ``CONDITION_SCALP_KIWOOM_NAME`` (기본 scalp_re). - ``SCALP_SKIP_HTS_SCAN_DUPES=true`` (kiwoom_condition 기본): TRIGGER 에 낙폭·RSI·되돌림 중복 검사 생략. - 매수(reversal): 시간대, 쿨다운, 일일 진입 횟수, RSI 과매도/과매수, 되돌림(음봉->양봉), 낙폭, 거래량. - 매수(macd): HTS C — MACD+Stochastic [12,26,5,3,3] 골든크로스 (``SCALP_USE_MACD_CROSS=true``). - 매수 방어: 고점추격 방지(high_chase_thr), 급등주(max_daily_chg), 최소 가격(min_price). - 매도(V4, tail_engine 동일): 1순위 어깨컷, 2순위 익절, 3순위 손절, 4순위 금액손실컷, 5순위 장마감청산. - 백테 SCALP 청산: 1분 OHLC → N회 intrabar (``check_sell_signal_backtest_bar``). 캔들 형식: list of dict with keys candle_time(YYYYMMDDHHMI), open, high, low, close, volume """ from datetime import datetime from typing import List, Dict, Any, Optional, Tuple, Callable from kis_trader.utils.env import get_env_int def _frac_from_env_keys(r: Optional[dict], keys: Tuple[str, ...], default: float) -> float: """env_config 행에서 비율(0.003=0.3%) 값 로드. 키 순서대로 fallback.""" if r: for k in keys: v = r.get(k) if v not in (None, ""): try: return float(v) except (ValueError, TypeError): continue return float(default) def resolve_effective_tp_pct(tp_pct: float, tp_max_pct: float) -> float: """익절 % — tp_max_pct 상한 적용 (꼬리 TAIL_ATR_TP_MAX_PCT 와 동일 개념). tp_pct=2.5%, tp_max=2% → 실제 2% 목표가. tp_max=0 이면 상한 미적용(OFF). """ tp = abs(float(tp_pct)) cap = abs(float(tp_max_pct)) if cap > 0: return min(tp, cap) return tp def effective_tp_pct_from_params(params: Dict[str, Any]) -> float: """params['tp_pct'] + params['tp_max_pct'] → 진입·청산용 실효 익절 비율.""" return resolve_effective_tp_pct( params.get("tp_pct", 0.015), params.get("tp_max_pct", 0.02), ) def _to_bool(v: Any, default: bool = True) -> bool: if v is None: return default if isinstance(v, bool): return v s = str(v).strip().lower() if s in ("1", "true", "t", "y", "yes", "on"): return True if s in ("0", "false", "f", "n", "no", "off", ""): return False return default def resolve_scalp_skip_hts_scan_dupes(r: Optional[Dict[str, Any]] = None) -> bool: """ HTS scalp_re SCAN(kiwoom_condition) 사용 시 TRIGGER 중복 필터 생략 여부. - ``SCALP_SKIP_HTS_SCAN_DUPES`` 명시(true/false) → 그대로 - 미설정 → ``SCALP_UNIVERSE_SOURCE`` 가 condition/kiwoom_condition 이면 True - ranking 등 전종목 소스 → False (TRIGGER 에서 HTS A/B/C 재검사) """ if r is None: try: from kis_trader.utils.env import get_strategy_env_dict r = get_strategy_env_dict("SCALP") or {} except Exception: r = {} raw = r.get("SCALP_SKIP_HTS_SCAN_DUPES") if raw is not None and str(raw).strip() != "": return _to_bool(raw, True) # 엔진 defaults dict(이미 skip_hts_scan_dupes 해석됨)를 넘긴 경우 — # env 키 없음 → universe fallback(True) 로 덮어쓰지 않음 (실매 false 고정 재발 방지) if "skip_hts_scan_dupes" in r and r.get("skip_hts_scan_dupes") is not None: return _to_bool(r.get("skip_hts_scan_dupes"), False) src = str(r.get("SCALP_UNIVERSE_SOURCE") or "condition").strip().lower() return src in ("kiwoom_condition", "condition") # DB 기본값 로드 (백테스트/param_search가 동일한 값 사용하도록 단일 소스) def get_scalping_defaults_from_db(*, env_row: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ config_scalp + env_config 병합에서 스캘핑 관련 기본값 로드. 백테스트 API·param_search·실매(get_env_from_db)가 동일 merged 소스를 사용. env_row 가 있으면 DB 조회 생략 (웹 백테 env 타임라인용). """ try: if env_row is not None: r = dict(env_row) else: from kis_trader.utils.env import get_strategy_env_dict r = get_strategy_env_dict("SCALP") if not r: from database import TradeDB db = TradeDB() if hasattr(db, "get_merged_env_snapshot"): r = db.get_merged_env_snapshot() else: latest = db.get_latest_env() if hasattr(db, "get_latest_env") else None r = dict((latest or {}).get("snapshot") or {}) db.close() if r: # SCALP_COOLDOWN_SEC(초) → cooldown_min(분). 실매매와 동일 키 사용 sec = r.get("SCALP_COOLDOWN_SEC") or r.get("REENTRY_COOLDOWN_SEC") or "600" cooldown_min = max(0, int(float(sec)) // 60) fee_pct = float(r.get("FEE_RATE_PCT") or 0.015) tax_pct = float(r.get("SELL_TAX_RATE_PCT") or 0.18) slot = float(r.get("SLOT_MONEY_DEFAULT") or 3_000_000) # ──────── 방어 로직 (SCALP_* 전용 키 우선 → 공용/타전략 fallback) ──────── # HIGH_CHASE_THR·MOMENTUM_* 를 먼저 읽으면 Optuna/웹 apply 값이 무시됨. high_chase_thr = float( r.get("SCALP_HIGH_PRICE_CHASE_THRESHOLD") or r.get("HIGH_CHASE_THR") or r.get("HIGH_PRICE_CHASE_THRESHOLD") or 0.96, ) max_daily_chg = float( r.get("SCALP_MAX_DAILY_CHANGE_PCT") or r.get("MAX_DAILY_CHG") or r.get("MAX_DAILY_CHANGE_PCT") or 20.0, ) min_price = float( r.get("SCALP_MIN_PRICE") or r.get("MOMENTUM_MIN_PRICE") or r.get("MIN_PRICE_TAIL") or 1000.0, ) max_loss_krw = int( float( r.get("SCALP_MAX_LOSS_PER_TRADE_KRW") or r.get("MOMENTUM_MAX_LOSS_PER_TRADE_KRW") or r.get("MAX_LOSS_PER_TRADE_KRW") or 200000, ), ) _min_drop_loss = r.get("SCALP_MIN_DROP_PCT_FOR_LOSS_CUT") min_drop_pct_for_loss_cut = 0.015 if _min_drop_loss not in (None, ""): v = float(_min_drop_loss) min_drop_pct_for_loss_cut = v / 100.0 if v >= 1 else v min_margin = float( r.get("SCALP_MIN_PROFIT_PCT") or r.get("MOMENTUM_MIN_PROFIT_PCT") or 0.2, ) rsi_overbought = float(r.get("SCALP_RSI_OVERBOUGHT") or 75.0) use_defense_filters = _to_bool(r.get("SCALP_USE_DEFENSE_FILTERS"), True) use_macd_cross = _to_bool(r.get("SCALP_USE_MACD_CROSS"), False) macd_fast = int(float(r.get("SCALP_MACD_FAST") or 12)) macd_slow = int(float(r.get("SCALP_MACD_SLOW") or 26)) macd_signal = int(float(r.get("SCALP_MACD_SIGNAL") or 5)) stoch_k_period = int(float(r.get("SCALP_STOCH_K_PERIOD") or 5)) stoch_d_period = int(float(r.get("SCALP_STOCH_D_PERIOD") or 3)) stoch_slow = int(float(r.get("SCALP_STOCH_SLOW") or 3)) skip_hts_scan_dupes = resolve_scalp_skip_hts_scan_dupes(r) shoulder_min_high = _frac_from_env_keys( r, ("SCALP_SHOULDER_MIN_HIGH_PCT", "SHOULDER_MIN_HIGH_PCT"), 0.005, ) shoulder_cut_pct = _frac_from_env_keys( r, ("SCALP_SHOULDER_CUT_PCT", "SHOULDER_CUT_PCT"), 0.003, ) _mhs = r.get("SCALP_MIN_HOLD_SEC") or r.get("MIN_HOLD_SEC") min_hold_sec = float(_mhs) if _mhs not in (None, "") else 30.0 tp_max_pct = _frac_from_env_keys(r, ("SCALP_TP_MAX_PCT",), 0.02) max_stocks = int(float(r.get("SCALP_MAX_STOCKS") or r.get("MAX_STOCKS") or 3)) total_budget_krw = float(r.get("SCALP_TOTAL_BUDGET_KRW") or 0) portfolio_mode = _to_bool(r.get("SCALP_PORTFOLIO_MODE"), True) rsi_period = int(float( r.get("SCALP_RSI_PERIOD") or r.get("RSI_PERIOD") or 3, )) sl_pct = abs(float(r.get("SCALP_STOP_LOSS_PCT") or 0.015)) tp_pct = abs(float(r.get("SCALP_TAKE_PROFIT_PCT") or 0.015)) rsi_oversold = float(r.get("SCALP_RSI_OVERSOLD") or 25.0) drop_rate = float(r.get("SCALP_MIN_DROP_RATE") or 0.015) require_reversal_candle = _to_bool(r.get("SCALP_REQUIRE_REVERSAL_CANDLE"), True) _vm = r.get("VOL_MULTIPLIER") vol_mult = float(_vm) if _vm not in (None, "") else 0.0 time_start_hm = int(float(r.get("SCALP_TIME_START") or r.get("TIME_START") or 900)) time_end_hm = int(float(r.get("SCALP_TIME_END") or r.get("TIME_END") or 1530)) max_daily = int(float(r.get("SCALP_MAX_DAILY") or r.get("MAX_DAILY") or 3)) # EOD — 실매 기존 15:25 장마감청산과 동일 (strategy_eod SCALP) eod_enabled = _to_bool(r.get("SCALP_EOD_ENABLED"), True) if r.get("SCALP_EOD_ENABLED") in (None, "", "None") and r.get("force_eod_exit") not in (None, "", "None"): eod_enabled = _to_bool(r.get("force_eod_exit"), True) eod_hm = str(r.get("SCALP_EOD_HM") or "15:25").strip() or "15:25" # 신호=T−1 확정봉, 진입=형성 중 봉 T 시가(첫 틱) — 모멘텀/꼬리 BT 정합 # DB 신규 컬럼 NULL→'' 이면 _to_bool('')=False 가 되므로 빈값은 미설정 취급 _la = r.get("SCALP_LIVE_BACKTEST_ALIGN") live_backtest_align = ( True if _la in (None, "") else _to_bool(_la, True) ) _lb = r.get("SCALP_LIVE_SIGNAL_LOOKBACK_BARS") live_signal_lookback_bars = int(float(_lb or 1)) _lf = r.get("SCALP_LIVE_ALIGN_USE_FORMING_BAR") live_align_use_forming_bar = ( True if _lf in (None, "") else _to_bool(_lf, True) ) backtest_use_tick_entry = _to_bool( r.get("SCALP_BACKTEST_USE_TICK_ENTRY"), True, ) backtest_tick_fallback_ohlc = _to_bool( r.get("SCALP_BACKTEST_TICK_FALLBACK_OHLC"), False, ) else: cooldown_min, fee_pct, tax_pct, slot = 10, 0.015, 0.18, 300_000.0 high_chase_thr, max_daily_chg, min_price = 0.96, 20.0, 1000.0 max_loss_krw, min_margin, rsi_overbought = 200000, 0.2, 75.0 min_drop_pct_for_loss_cut = 0.015 use_defense_filters = True use_macd_cross = False macd_fast, macd_slow, macd_signal = 12, 26, 5 stoch_k_period, stoch_d_period, stoch_slow = 5, 3, 3 skip_hts_scan_dupes = True shoulder_min_high, shoulder_cut_pct, min_hold_sec = 0.005, 0.003, 30.0 tp_max_pct = 0.02 max_stocks, total_budget_krw, portfolio_mode = 3, 0.0, True rsi_period = 3 sl_pct, tp_pct = 0.015, 0.015 rsi_oversold, drop_rate = 25.0, 0.015 require_reversal_candle = True vol_mult = 0.0 time_start_hm, time_end_hm, max_daily = 900, 1530, 3 eod_enabled, eod_hm = True, "15:25" live_backtest_align, live_signal_lookback_bars = True, 1 live_align_use_forming_bar = True backtest_use_tick_entry, backtest_tick_fallback_ohlc = True, False except Exception: cooldown_min, fee_pct, tax_pct, slot = 10, 0.015, 0.18, 300_000.0 high_chase_thr, max_daily_chg, min_price = 0.96, 20.0, 1000.0 max_loss_krw, min_margin, rsi_overbought = 200000, 0.2, 75.0 min_drop_pct_for_loss_cut = 0.015 use_defense_filters = True use_macd_cross = False macd_fast, macd_slow, macd_signal = 12, 26, 5 stoch_k_period, stoch_d_period, stoch_slow = 5, 3, 3 skip_hts_scan_dupes = True shoulder_min_high, shoulder_cut_pct, min_hold_sec = 0.005, 0.003, 30.0 tp_max_pct = 0.02 max_stocks, total_budget_krw, portfolio_mode = 3, 0.0, True rsi_period = 3 sl_pct, tp_pct = 0.015, 0.015 rsi_oversold, drop_rate = 25.0, 0.015 require_reversal_candle = True vol_mult = 0.0 time_start_hm, time_end_hm, max_daily = 900, 1530, 3 eod_enabled, eod_hm = True, "15:25" live_backtest_align, live_signal_lookback_bars = True, 1 live_align_use_forming_bar = True backtest_use_tick_entry, backtest_tick_fallback_ohlc = True, False return { "cooldown_min": cooldown_min, "time_start_hm": time_start_hm, "time_end_hm": time_end_hm, "time_start": time_start_hm, "time_end": time_end_hm, "fee_rate": fee_pct / 100, "sell_tax": tax_pct / 100, "slot_money": slot, "rsi_period": rsi_period, "rsi_oversold": rsi_oversold, "sl_pct": sl_pct, "tp_pct": tp_pct, "drop_rate": drop_rate, "require_reversal_candle": require_reversal_candle, "vol_mult": vol_mult, "trail_trigger": 0.007, "trail_stop": 0.004, "max_daily": max_daily, "high_chase_thr": high_chase_thr, "max_daily_chg": max_daily_chg, "min_price": min_price, "max_loss_krw": max_loss_krw, "min_drop_pct_for_loss_cut": min_drop_pct_for_loss_cut, "min_margin": min_margin / 100, "rsi_overbought": rsi_overbought, "use_defense_filters": use_defense_filters, "use_macd_cross": use_macd_cross, "macd_fast": macd_fast, "macd_slow": macd_slow, "macd_signal": macd_signal, "stoch_k_period": stoch_k_period, "stoch_d_period": stoch_d_period, "stoch_slow": stoch_slow, "skip_hts_scan_dupes": skip_hts_scan_dupes, "shoulder_min_high": shoulder_min_high, "shoulder_cut_pct": shoulder_cut_pct, "min_hold_sec": min_hold_sec, "tp_max_pct": tp_max_pct, "max_stocks": max_stocks, "total_budget_krw": total_budget_krw, "portfolio_mode": portfolio_mode, "eod_enabled": eod_enabled, "eod_hm": eod_hm, # 레거시 별칭 — 포트폴리오/단건 엔진 force_eod 경로 호환 "force_eod_exit": eod_enabled, "live_backtest_align": live_backtest_align, "live_signal_lookback_bars": live_signal_lookback_bars, "live_align_use_forming_bar": live_align_use_forming_bar, "backtest_use_tick_entry": backtest_use_tick_entry, "backtest_tick_fallback_ohlc": backtest_tick_fallback_ohlc, } def compute_rsi_series(closes: list, period: int = 3) -> list: """RSI 시리즈 계산 (Wilder 스무딩). backtest_web과 동일.""" rsi_list = [None] * len(closes) if len(closes) < period + 1: return rsi_list deltas = [closes[i] - closes[i - 1] for i in range(1, len(closes))] gains = [max(d, 0) for d in deltas] losses = [max(-d, 0) for d in deltas] avg_gain = sum(gains[:period]) / period avg_loss = sum(losses[:period]) / period for i in range(period, len(closes)): idx = i - 1 if i > period: avg_gain = (avg_gain * (period - 1) + gains[idx]) / period avg_loss = (avg_loss * (period - 1) + losses[idx]) / period rs = avg_gain / avg_loss if avg_loss > 0 else float("inf") rsi_val = 100 - (100 / (1 + rs)) if avg_loss > 0 else 100.0 rsi_list[i] = rsi_val return rsi_list def _ema_series(values: List[float], period: int) -> List[Optional[float]]: """지수이동평균(EMA) 시리즈. 초기값은 SMA.""" n = len(values) out: List[Optional[float]] = [None] * n if period <= 0 or n < period: return out mult = 2.0 / (period + 1) ema = sum(values[:period]) / period out[period - 1] = ema for i in range(period, n): ema = (values[i] - ema) * mult + ema out[i] = ema return out def _sma_optional_series(values: List[Optional[float]], period: int) -> List[Optional[float]]: """None 을 건너뛰지 않고, 윈도우 내 유효값만으로 SMA.""" n = len(values) out: List[Optional[float]] = [None] * n if period <= 0: return out for i in range(period - 1, n): window = [v for v in values[i - period + 1 : i + 1] if v is not None] if len(window) == period: out[i] = sum(window) / period return out def compute_macd_stoch_hts_lines( candles: List[Dict], macd_fast: int = 12, macd_slow: int = 26, macd_signal: int = 5, stoch_k_period: int = 5, stoch_d_period: int = 3, stoch_slow: int = 3, ) -> Tuple[ List[Optional[float]], List[Optional[float]], List[Optional[float]], List[Optional[float]], List[Optional[float]], List[Optional[float]], ]: """ HTS 조건식 C — MACD+Stochastic [12,26,5,3,3] 복합선. - fast_line = MACD + Slow%K - slow_line = Signal + Slow%D - 골든크로스: fast 가 slow 를 0봉전(현재봉) 상향돌파 Returns: macd_line, signal_line, slow_k, slow_d, fast_combined, slow_combined """ n = len(candles) closes = [float(c["close"]) for c in candles] highs = [float(c["high"]) for c in candles] lows = [float(c["low"]) for c in candles] ema_fast = _ema_series(closes, macd_fast) ema_slow = _ema_series(closes, macd_slow) macd_line: List[Optional[float]] = [None] * n for i in range(n): if ema_fast[i] is not None and ema_slow[i] is not None: macd_line[i] = ema_fast[i] - ema_slow[i] macd_vals = [v for v in macd_line if v is not None] signal_line: List[Optional[float]] = [None] * n if len(macd_vals) >= macd_signal: sig_sub = _ema_series(macd_vals, macd_signal) idx = 0 for i in range(n): if macd_line[i] is not None: if idx < len(sig_sub): signal_line[i] = sig_sub[idx] idx += 1 raw_k: List[Optional[float]] = [None] * n for i in range(stoch_k_period - 1, n): hh = max(highs[i - stoch_k_period + 1 : i + 1]) ll = min(lows[i - stoch_k_period + 1 : i + 1]) if hh == ll: raw_k[i] = 50.0 else: raw_k[i] = (closes[i] - ll) / (hh - ll) * 100.0 slow_k = _sma_optional_series(raw_k, stoch_slow) slow_d = _sma_optional_series(slow_k, stoch_d_period) fast_combined: List[Optional[float]] = [None] * n slow_combined: List[Optional[float]] = [None] * n for i in range(n): if macd_line[i] is not None and slow_k[i] is not None: fast_combined[i] = macd_line[i] + slow_k[i] if signal_line[i] is not None and slow_d[i] is not None: slow_combined[i] = signal_line[i] + slow_d[i] return macd_line, signal_line, slow_k, slow_d, fast_combined, slow_combined def _is_macd_stoch_golden_cross( fast_combined: List[Optional[float]], slow_combined: List[Optional[float]], i: int, ) -> bool: """0봉전 골든크로스: fast 가 slow 를 상향돌파.""" if i < 1: return False f0, f1 = fast_combined[i], fast_combined[i - 1] s0, s1 = slow_combined[i], slow_combined[i - 1] if None in (f0, f1, s0, s1): return False return f0 > s0 and f1 <= s1 def _macd_min_bars(params: Dict[str, Any]) -> int: """MACD+Stoch 계산에 필요한 최소 봉 수.""" macd_slow = int(params.get("macd_slow", 26)) macd_signal = int(params.get("macd_signal", 5)) stoch_k = int(params.get("stoch_k_period", 5)) stoch_slow = int(params.get("stoch_slow", 3)) stoch_d = int(params.get("stoch_d_period", 3)) return macd_slow + macd_signal + stoch_k + stoch_slow + stoch_d + 5 def _apply_scalp_defense_and_vol( candles: List[Dict], i: int, params: Dict[str, Any], cl: float, lo: float, day: str, ) -> Tuple[str, Optional[str], Optional[str]]: """ 방어 필터 + 거래량 배수 (reversal·macd 공통). Returns: ("ok", None, None) — 통과 ("reject", code, msg) — 탈락 ("skip", None, None) — 당일 시가 없음 (신호 없음) """ drop_rate = float(params.get("drop_rate", 0.015)) vol_mult = float(params.get("vol_mult", 0)) high_chase_thr = float(params.get("high_chase_thr", 0.96)) max_daily_chg = float(params.get("max_daily_chg", 20.0)) min_price = float(params.get("min_price", 1000.0)) use_defense_filters = _to_bool(params.get("use_defense_filters", True), True) skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), False) if use_defense_filters and cl < min_price: return ("reject", "탈락-최소가격", "%.0f < %.0f" % (cl, min_price)) running_open = float(candles[i]["open"]) running_low = lo running_high = float(candles[i]["high"]) for j in range(i - 1, -1, -1): if candles[j]["candle_time"][:8] != day: break running_open = float(candles[j]["open"]) running_low = min(running_low, float(candles[j]["low"])) running_high = max(running_high, float(candles[j]["high"])) if running_open <= 0: return ("skip", None, None) # HTS scalp_re SCAN(kiwoom) — B/C/F/D 이미 통과. 낙폭·고점·급등·분봉거래량 중복 생략 if use_defense_filters and not skip_hts: dr = (running_open - running_low) / running_open if dr < drop_rate: return ("reject", "탈락-낙폭", "%.2f%% < %.1f%%(SCALP_MIN_DROP_RATE)" % (dr * 100, drop_rate * 100)) if cl >= running_high * high_chase_thr: return ("reject", "탈락-고점추격", "현재가 %.0f >= 고가 %.0f * %.2f" % (cl, running_high, high_chase_thr)) if running_low > 0: daily_chg_pct = (running_high - running_low) / running_low * 100 if daily_chg_pct > max_daily_chg: return ("reject", "탈락-급등주", "일일변동 %.1f%% > %.0f%%" % (daily_chg_pct, max_daily_chg)) if not skip_hts and vol_mult > 0: volumes = [float(x.get("volume", 0)) for x in candles] vol = volumes[i] if i < len(volumes) else 0 win = max(1, min(20, i)) vol_avg = sum(volumes[i - win : i]) / win if vol_avg > 0 and vol < vol_avg * vol_mult: return ("reject", "탈락-거래량", "%.0f < 평균%.0f × %.1f" % (vol, vol_avg, vol_mult)) return ("ok", None, None) def _t2dt(t: str) -> datetime: """candle_time / 실매 buy_time → datetime (공통 파서).""" from kis_trader.utils.trade_time import parse_trade_datetime return parse_trade_datetime(t) def _slot_key(candle_time: str, scan_interval_min: int = 1) -> str: """ 봉 시각을 N분 단위 슬롯 키로 변환. 신봇 기준: * 실매매는 10초 주기 REST 폴링 + 변동 tick 마다 초단위 event_time 저장. * 백테스트 caller(``param_search``/``tail_param_search``)가 ``TradeDBExt.get_universe_by_candle_time()`` 으로 **1분 캔들 시각 키** (YYYYMMDDHHMM) 를 가진 dict 를 만들어 엔진에 주입한다. * 엔진은 그 dict 를 분단위(scan_interval_min=1) 키로 바로 조회. 과거 호환: * ``--fallback-universe`` 시뮬레이션 경로는 여전히 5분 버킷팅을 쓸 수 있도록 ``scan_interval_min`` 파라미터는 남겨 둔다. """ date = candle_time[:8] hm = int(candle_time[8:12]) total_min = (hm // 100) * 60 + (hm % 100) slot_min = (total_min // scan_interval_min) * scan_interval_min slot_hm = (slot_min // 60) * 100 + (slot_min % 60) return date + str(slot_hm).zfill(4) def build_universe_simulation( codes_candles: Dict[str, List[Dict]], top_n: int = 20, min_score: float = 4.0, scan_interval_min: int = 5, ) -> Dict[str, List[str]]: """ 과거 1분봉만으로 '5분마다 강도 순 상위 N종목' 유니버스를 흉내 냄. kiwoom_universe_scanner의 개미털기 점수(drop_rate·회복률)를 1분봉 기준으로 근사. 실매매의 target_candidates 5분 갱신과 동일한 타이밍으로 백테스트 제한용. Returns: slot_key(YYYYMMDDHHMM) → 해당 슬롯에 매수 검사 허용할 종목 코드 리스트 """ slot_codes_scores: Dict[str, List[Tuple[str, float, float]]] = {} # slot -> [(code, score, vol)] for code, rows in codes_candles.items(): if len(rows) < 2: continue candles = [dict(r) for r in rows] # 날짜별로 묶어서 당일 시가/고/저/종가 누적 by_day: Dict[str, List[Dict]] = {} for c in candles: day = c["candle_time"][:8] if day not in by_day: by_day[day] = [] by_day[day].append(c) for day, day_candles in by_day.items(): day_candles.sort(key=lambda x: x["candle_time"]) day_open = float(day_candles[0]["open"]) running_high = max(float(c["high"]) for c in day_candles) running_low = min(float(c["low"]) for c in day_candles) vol_sum = sum(float(c.get("volume", 0)) for c in day_candles) # 5분 단위 슬롯: 해당 슬롯 시작 시점까지의 데이터로 점수 계산 (슬롯 시작 직전 봉까지) market_open_min = 9 * 60 # 09:00 = 540분 seen_slots = set() for j, c in enumerate(day_candles): ct = c["candle_time"] hm = int(ct[8:12]) total_min = (hm // 100) * 60 + (hm % 100) slot_min = (total_min // scan_interval_min) * scan_interval_min slot_hm = (slot_min // 60) * 100 + (slot_min % 60) slot_key = day + str(slot_hm).zfill(4) if slot_key in seen_slots: continue seen_slots.add(slot_key) # as_of: 이 슬롯에 쓰일 마지막 봉 (슬롯 시작 직전 봉, 장시작 09:00 이상) slot_min_val = (slot_hm // 100) * 60 + (slot_hm % 100) as_of_min = max(market_open_min, slot_min_val - 1) as_of_hm = (as_of_min // 60) * 100 + (as_of_min % 60) as_of_str = day + str(as_of_hm).zfill(4) up_to = [x for x in day_candles if x["candle_time"] <= as_of_str] if not up_to: continue o = float(up_to[0]["open"]) hi = max(float(x["high"]) for x in up_to) lo = min(float(x["low"]) for x in up_to) cl = float(up_to[-1]["close"]) drop_rate = (o - lo) / o if o > 0 else 0 rng = hi - lo recovery_pos = (cl - lo) / rng if rng > 0 else 0 score = (drop_rate * 100) if (drop_rate >= 0.03 and recovery_pos >= 0.5) else 0.0 vol_part = sum(float(x.get("volume", 0)) for x in up_to) if slot_key not in slot_codes_scores: slot_codes_scores[slot_key] = [] slot_codes_scores[slot_key].append((code, score, vol_part)) # 슬롯별 상위 top_n, min_score 이상만 universe_by_slot: Dict[str, List[str]] = {} for slot_key, lst in slot_codes_scores.items(): lst = [(c, s, v) for c, s, v in lst if s >= min_score] lst.sort(key=lambda x: (-x[1], -x[2])) universe_by_slot[slot_key] = [x[0] for x in lst[:top_n]] return universe_by_slot # ────────────────────────────────────────────────────────────────────────────── # [레거시] 모멘텀 유니버스 시뮬 — momentum_engine.build_universe_simulation_momentum 위임 # ────────────────────────────────────────────────────────────────────────────── def build_universe_simulation_momentum( codes_candles: Dict[str, List[Dict]], top_n: int = 20, min_score: float = 4.0, scan_interval_min: int = 5, ) -> Dict[str, List[str]]: """[폐기] ``momentum_engine.build_universe_simulation_momentum`` 사용.""" from kis_trader.engine.momentum_engine import build_universe_simulation_momentum as _impl return _impl(codes_candles, top_n=top_n, min_score=min_score, scan_interval_min=scan_interval_min) def run_scalping_backtest( codes_candles: Dict[str, List[Dict]], params: Dict[str, Any], universe_by_slot: Optional[Dict[str, List[str]]] = None, ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]] = None, ) -> List[Dict]: """ 종목별 캔들에 대해 스캘핑 백테스트 실행. 실매매와 동일한 규칙(방어로직 포함) 적용. universe_by_slot이 주어지면, 5분마다 해당 슬롯의 후보 종목에서만 매수 신호를 검사 (실매매의 target_candidates 5분 갱신과 동일한 유니버스 시뮬레이션). """ if _to_bool(params.get("use_rust"), False): from kis_trader.utils.logger import get_logger get_logger("kis_trader.scalping_engine").info("🚀 Rust 엔진(Experimental)으로 스캘핑 시뮬레이션을 실행합니다.") return run_scalping_backtest_rust_experimental(codes_candles, params) if _to_bool(params.get("portfolio_mode"), True): from kis_trader.backtest.scalping_portfolio_backtest import run_scalping_backtest_portfolio return run_scalping_backtest_portfolio( codes_candles, params, universe_by_slot=universe_by_slot, mode="reversal", ticks_by_code=ticks_by_code, ) rsi_period = int(params.get("rsi_period", 3)) rsi_oversold = float(params.get("rsi_oversold", 25)) rsi_overbought = float(params.get("rsi_overbought", 75.0)) # 롱 스캘핑: 손절·익절 폭은 양의 비율이어야 함. 폼/DB에서 음수(-1.2%)가 넘어오면 # stop 이 진입가 위로 뒤집혀 같은 봉에서 '손절'로 오표시되며 청산가가 역전되는 버그 발생. sl_pct = abs(float(params.get("sl_pct", 0.015))) tp_pct = effective_tp_pct_from_params(params) drop_rate = float(params.get("drop_rate", 0.015)) slot_money = float(params.get("slot_money", 300_000)) fee_rate = float(params.get("fee_rate", 0.00015)) sell_tax = float(params.get("sell_tax", 0.0018)) cooldown_min = float(params.get("cooldown_min", 10)) trail_trigger = float(params.get("trail_trigger", 0)) trail_stop = float(params.get("trail_stop", 0.004)) time_start_hm = int(params.get("time_start_hm", 900)) time_end_hm = int(params.get("time_end_hm", 1530)) max_daily = int(params.get("max_daily", 3)) vol_mult = float(params.get("vol_mult", 0)) # 방어 로직 파라미터 high_chase_thr = float(params.get("high_chase_thr", 0.96)) max_daily_chg = float(params.get("max_daily_chg", 20.0)) min_price = float(params.get("min_price", 1000.0)) max_loss_krw = float(params.get("max_loss_krw", 200000.0)) min_margin = float(params.get("min_margin", 0.002)) use_defense_filters = _to_bool(params.get("use_defense_filters", True), True) use_macd_cross = _to_bool(params.get("use_macd_cross", False), False) from kis_trader.engine.strategy_eod import is_strategy_eod_bar from kis_trader.engine.tick_exit_common import ( backtest_sell_slip_pct, backtest_tick_poll_ms, collect_minute_ticks, resolve_backtest_sell, strategy_tick_fallback_ohlc, strategy_use_tick_exit, ) use_tick_exit = bool(ticks_by_code) and strategy_use_tick_exit( params, "SCALP_BACKTEST_USE_TICK_EXIT", default=True, ) tick_fallback_ohlc = strategy_tick_fallback_ohlc( params, "SCALP_BACKTEST_TICK_FALLBACK_OHLC", default=False, ) tick_poll_ms = backtest_tick_poll_ms(params, strategy_env="SCALP_BACKTEST_POLL_MS") tick_sell_slip = backtest_sell_slip_pct(params, strategy_env="SCALP_BACKTEST_SELL_SLIP_PCT") all_trades: List[Dict] = [] for code, rows in codes_candles.items(): if len(rows) < rsi_period + 5: continue candles = [dict(r) for r in rows] closes = [float(c["close"]) for c in candles] volumes = [float(c.get("volume", 0)) for c in candles] rsis = compute_rsi_series(closes, rsi_period) macd_combined = _macd_lines_from_params(candles, params) if use_macd_cross else None position: Optional[Dict] = None last_exit_dt: Dict[str, datetime] = {} daily_cnt: Dict[str, int] = {} cur_day = None running_open = 0.0 running_high = 0.0 running_low = 0.0 for i in range(rsi_period + 1, len(candles)): c = candles[i] day = c["candle_time"][:8] hm = int(c["candle_time"][8:12]) cl = float(c["close"]) lo = float(c["low"]) hi = float(c["high"]) vol = volumes[i] if i < len(volumes) else 0 # 당일 고가/저가 누적 (피뢰침 방지용) if day != cur_day: cur_day = day running_open = float(c["open"]) running_high = hi running_low = lo else: running_high = max(running_high, hi) running_low = min(running_low, lo) # 실매 SCALP_EOD_HM(기본 15:25) 과 동일 — force_eod 마지막봉만 의존 금지 is_eod = is_strategy_eod_bar(c["candle_time"], params, "SCALP") # ── 포지션 보유 중: 청산 체크 ── if position is not None: # 진입봉(같은 candle_time)에서는 청산 금지: # 백테스트가 이미 알고 있는 봉의 고/저를 즉시 사용하는 look-ahead를 방지. if c["candle_time"] == position["entry_time"]: continue # 틱 있으면 resolve_backtest_sell — 없으면 기존 OHLC N회 청산 reason = None exit_price = cl sell_time = c["candle_time"] if use_tick_exit: minute_ticks = collect_minute_ticks( ticks_by_code, code, c["candle_time"], ) res5 = resolve_backtest_sell( position, c, params, is_eod=is_eod, sell_fn=check_sell_signal_live, low_mode="current", ticks=minute_ticks, use_tick_exit=use_tick_exit, tick_fallback_ohlc=tick_fallback_ohlc, poll_ms=tick_poll_ms, slip_pct=tick_sell_slip, ) if res5: reason, exit_price, sell_time, _hold, _src = res5 else: # check_sell_signal_backtest_bar — 1분봉 N회 청산(실매 current_price 정렬) reason, exit_price = check_sell_signal_backtest_bar( position, c, params, is_eod, ) or (None, cl) if reason: qty = position["qty"] buy_amt = position["entry_price"] * qty sell_amt = exit_price * qty pnl = ( sell_amt - buy_amt - buy_amt * fee_rate - sell_amt * fee_rate - sell_amt * sell_tax ) hold_min = int( (_t2dt(sell_time or c["candle_time"]) - _t2dt(position["entry_time"])).total_seconds() / 60 ) all_trades.append({ "code": code, "buy_time": position["entry_time"], "sell_time": sell_time or c["candle_time"], "buy_price": position["entry_price"], "sell_price": round(exit_price, 2), "qty": qty, "pnl": round(pnl), "profit_rate": round( (exit_price - position["entry_price"]) / position["entry_price"] * 100, 2 ), "hold_min": hold_min, "sell_reason": reason, "rsi_entry": round(position["rsi"], 1), }) last_exit_dt[day] = _t2dt(sell_time or c["candle_time"]) position = None continue # ── 포지션 없음: 매수 신호 (유니버스 시뮬레이션 시 해당 슬롯 후보만 검사) ── if universe_by_slot is not None: # 신봇 기본: 1분봉 == 슬롯 키. simulation fallback 은 caller 가 # ``scan_interval_min`` 을 params 에 명시해 5분 슬롯으로 바꿀 수 있음. slot_key = _slot_key(c["candle_time"], params.get("scan_interval_min", 1)) if code not in universe_by_slot.get(slot_key, []): continue bt_state = { "last_exit_dt": last_exit_dt.get(day), "daily_cnt": daily_cnt.get(day, 0), } st = _apply_buy_state_filters( candles, i, params, bt_state, ) if st[2] is None: continue reject, _, sig = _eval_scalp_buy_at_index(candles, i, params, macd_combined=macd_combined) if reject or not sig: continue rsi = sig.get("rsi") if rsi is None: continue # 다음 봉 시가에 진입 if i + 1 >= len(candles): continue next_c = candles[i + 1] if next_c["candle_time"][:8] != day: continue entry_price = float(next_c["open"]) if entry_price <= 0: continue # 포지션 크기 계산 (최대 금액 손실 컷 기반) invest_amount = slot_money if max_loss_krw > 0 and sl_pct > 0: invest_limit = max_loss_krw / sl_pct invest_amount = min(invest_limit, slot_money) qty = max(1, int(invest_amount / entry_price)) stop = entry_price * (1 - sl_pct) target = entry_price * (1 + tp_pct) position = { "entry_price": entry_price, "entry_time": next_c["candle_time"], "qty": qty, "stop": stop, "target": target, "max_price": entry_price, "rsi": rsi, } daily_cnt[day] = daily_cnt.get(day, 0) + 1 all_trades.sort(key=lambda x: x["sell_time"]) return all_trades # ── 실시간 봇용: 단일 시점 매수/매도 판단 (백테스트와 100% 동일 규칙) ────────────── def _confirmed_candles_only(candles: List[Dict]) -> List[Dict]: """확정봉만 사용 (미확정 봉 제외). 없으면 원본 유지.""" confirmed = [ c for c in candles if _to_bool(c.get("is_confirmed", 1), True) ] return confirmed if confirmed else list(candles) def _apply_buy_state_filters( candles: List[Dict], i: int, params: Dict[str, Any], state: Dict[str, Any], ) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]: """시간대·쿨다운·일일횟수 등 상태 필터 (신호봉 인덱스 i 기준 — 백테스트와 동일).""" c = candles[i] time_start_hm = int(params.get("time_start_hm", 900)) time_end_hm = int(params.get("time_end_hm", 1530)) cooldown_min = float(params.get("cooldown_min", 10)) max_daily = int(params.get("max_daily", 3)) hm = int(c["candle_time"][8:12]) if hm < time_start_hm or hm >= time_end_hm: return (None, None, None) last_exit_dt = state.get("last_exit_dt") if last_exit_dt is not None: elapsed = (_t2dt(c["candle_time"]) - last_exit_dt).total_seconds() / 60 if elapsed < cooldown_min: return (None, None, None) if state.get("daily_cnt", 0) >= max_daily: return (None, None, None) return (None, None, {"ok": True}) def _macd_lines_from_params( candles: List[Dict], params: Dict[str, Any], ) -> Tuple[List[Optional[float]], List[Optional[float]]]: """종목당 1회 MACD+Stoch 복합선 계산 (백테스트 루프용).""" _, _, _, _, fast_combined, slow_combined = compute_macd_stoch_hts_lines( candles, macd_fast=int(params.get("macd_fast", 12)), macd_slow=int(params.get("macd_slow", 26)), macd_signal=int(params.get("macd_signal", 5)), stoch_k_period=int(params.get("stoch_k_period", 5)), stoch_d_period=int(params.get("stoch_d_period", 3)), stoch_slow=int(params.get("stoch_slow", 3)), ) return fast_combined, slow_combined def _eval_scalp_buy_at_index( candles: List[Dict], i: int, params: Dict[str, Any], macd_combined: Optional[Tuple[List[Optional[float]], List[Optional[float]]]] = None, ) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]: """SCALP 진입 모드 분기: MACD 골든크로스 vs RSI reversal → 호가필터.""" if _to_bool(params.get("use_macd_cross", False), False): reject, msg, sig = _eval_macd_golden_buy_at_index( candles, i, params, macd_combined=macd_combined, ) else: reject, msg, sig = _eval_reversal_buy_at_index(candles, i, params) if reject or not sig: return (reject, msg, sig) cl = float(candles[i].get("close") or 0) # 휩쏘 — 모멘텀과 동일 순서(휩쏘 → 호가). SCALP_WHIPSAW_* / 주입틱·실매 WS. from kis_trader.engine.whipsaw_filter import whipsaw_reject_for_signal ws_rej, ws_msg = whipsaw_reject_for_signal( params, "SCALP", signal_bar=candles[i], current_price=cl, ) if ws_rej: return (ws_rej, ws_msg, None) # 호가필터 — ORDERBOOK_FILTER_ENABLED / params['_orderbook_filter_enabled'] # 실매 기본 OFF 면 차단 없음(수집만). 파람서치 --orderbook-filter on 시 차단. from kis_trader.engine.orderbook_filter import orderbook_reject_for_entry ob_rej, ob_msg = orderbook_reject_for_entry(params, "SCALP", current_price=cl) if ob_rej: return (ob_rej, ob_msg, None) return (None, None, sig) def _eval_macd_golden_buy_at_index( candles: List[Dict], i: int, params: Dict[str, Any], macd_combined: Optional[Tuple[List[Optional[float]], List[Optional[float]]]] = None, ) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]: """ HTS C — MACD+Stochastic [12,26,5,3,3] 골든크로스 (0봉전 상향돌파). SCAN 은 K/L/제외만 두고, 진입 타이밍은 TRIGGER 에서 검사 (SCAN vs TRIGGER 분리). """ min_bars = _macd_min_bars(params) if i < 1 or i >= len(candles) or len(candles) < min_bars: return ("탈락-봉부족", "MACD+Stoch 계산 봉 %d/%d" % (len(candles), min_bars), None) c = candles[i] day = c["candle_time"][:8] cl = float(c["close"]) lo = float(c["low"]) status, rej_code, rej_msg = _apply_scalp_defense_and_vol(candles, i, params, cl, lo, day) if status == "reject": return (rej_code, rej_msg, None) if status == "skip": return (None, None, None) if macd_combined is not None: fast_combined, slow_combined = macd_combined else: fast_combined, slow_combined = _macd_lines_from_params(candles, params) if not _is_macd_stoch_golden_cross(fast_combined, slow_combined, i): f0 = fast_combined[i] s0 = slow_combined[i] return ( "탈락-MACD골든X", "fast=%s slow=%s (0봉전 상향돌파 아님)" % ( "%.2f" % f0 if f0 is not None else "N/A", "%.2f" % s0 if s0 is not None else "N/A", ), None, ) rsi_period = int(params.get("rsi_period", 3)) closes = [float(x["close"]) for x in candles] rsis = compute_rsi_series(closes, rsi_period) rsi = rsis[i] if i < len(rsis) else None return (None, None, {"signal": True, "rsi": rsi, "entry_mode": "macd_golden"}) def _eval_reversal_buy_at_index( candles: List[Dict], i: int, params: Dict[str, Any], ) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]: """ 봉 i 에서 reversal 매수 신호 판정 (시간·쿨다운 제외). ``run_scalping_backtest`` 의 신호봉 검사와 동일 규칙. """ if i < 1 or i >= len(candles): return ("탈락-봉부족", "인덱스 범위 밖", None) rsi_period = int(params.get("rsi_period", 3)) rsi_oversold = float(params.get("rsi_oversold", 25)) rsi_overbought = float(params.get("rsi_overbought", 75.0)) drop_rate = float(params.get("drop_rate", 0.015)) vol_mult = float(params.get("vol_mult", 0)) high_chase_thr = float(params.get("high_chase_thr", 0.96)) max_daily_chg = float(params.get("max_daily_chg", 20.0)) min_price = float(params.get("min_price", 1000.0)) use_defense_filters = _to_bool(params.get("use_defense_filters", True), True) require_reversal_candle = bool(params.get("require_reversal_candle", True)) skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), False) c = candles[i] day = c["candle_time"][:8] cl = float(c["close"]) lo = float(c["low"]) prev_c = candles[i - 1] status, rej_code, rej_msg = _apply_scalp_defense_and_vol(candles, i, params, cl, lo, day) if status == "reject": return (rej_code, rej_msg, None) if status == "skip": return (None, None, None) closes = [float(x["close"]) for x in candles] rsis = compute_rsi_series(closes, rsi_period) rsi = rsis[i] if i < len(rsis) else None # kiwoom scalp_re SCAN — HTS B/C/F/D 통과 후 진입 타이밍만 (RSI V자·되돌림 중복 금지) if skip_hts: return (None, None, {"signal": True, "rsi": rsi, "entry_mode": "reversal_hts_scan"}) if rsi is None: return ("탈락-RSI없음", "RSI 미계산 (봉 축적 중)", None) if rsi <= 0.0: return ("탈락-RSI무효", "RSI3=0.0 (봉 부족, 계산 불가)", None) if rsi > rsi_overbought: return ("탈락-RSI과열", "RSI3=%.1f > %.0f" % (rsi, rsi_overbought), None) if rsi > rsi_oversold: return ("탈락-RSI조건", "RSI3=%.1f (과매도<%.0f 아님)" % (rsi, rsi_oversold), None) if require_reversal_candle: prev_bear = float(prev_c["close"]) < float(prev_c["open"]) curr_bull = cl > float(c["open"]) if not (prev_bear and curr_bull): return ("탈락-되돌림없음", "prev_bear=%s curr_bull=%s" % (prev_bear, curr_bull), None) return (None, None, {"signal": True, "rsi": rsi, "entry_mode": "reversal"}) def check_buy_signal_live( candles: List[Dict], params: Dict[str, Any], state: Dict[str, Any], ) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]: """ 실시간 reversal 매수 신호. - ``live_backtest_align=True`` (기본): - 신호봉 = 직전 확정봉 (T−1) - 진입봉 = 형성 중 봉 T (``live_align_use_forming_bar`` 기본 True) → ``entry_price`` = 진입봉 시가(실매·BT 첫 틱/시가 정합) - forming 없으면 구 폴백: 신호=confirmed[-2], 진입=confirmed[-1] open - ``live_backtest_align=False``: 구버전 — 마지막 1봉만 검사. """ if len(candles) < 4: return ("탈락-봉부족", "확정봉 4개 미만", None) live_align = _to_bool(params.get("live_backtest_align", True), True) lookback = max(1, int(params.get("live_signal_lookback_bars", 1))) use_forming = _to_bool(params.get("live_align_use_forming_bar", True), True) forming: Optional[Dict[str, Any]] = None if live_align and use_forming and candles: last = candles[-1] if last.get("is_confirmed") in (0, False, "0", "false"): forming = last confirmed = _confirmed_candles_only(candles) if len(confirmed) < 4: return ("탈락-봉부족", "확정봉 4개 미만", None) last_reject: Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]] = ( None, None, None, ) if live_align: # forming 있으면: 신호=confirmed[-1], 진입=forming open (BT next open 과 동일) # forming 없으면: 신호=confirmed[-2], 진입=confirmed[-1] open (구 폴백) if forming is not None: entry_bar = forming signal_base_i = len(confirmed) - 1 else: entry_bar = confirmed[-1] signal_base_i = len(confirmed) - 2 for k in range(lookback): signal_i = signal_base_i - k if signal_i < 1: break st = _apply_buy_state_filters(confirmed, signal_i, params, state) if st[2] is None: continue reject, msg, sig = _eval_scalp_buy_at_index(confirmed, signal_i, params) if reject: if k == 0: last_reject = (reject, msg, None) continue if sig: entry_open = float(entry_bar.get("open", 0) or 0) if entry_open <= 0: entry_open = float(entry_bar.get("close", 0) or 0) sig["entry_price"] = entry_open sig["entry_bar_key"] = entry_bar.get("candle_time") return (None, None, sig) return last_reject # 구버전: 마지막 확정봉 1개만 신호봉으로 검사 i = len(confirmed) - 1 st = _apply_buy_state_filters(confirmed, i, params, state) if st[2] is None: return (None, None, None) return _eval_scalp_buy_at_index(confirmed, i, params) def _intrabar_exit_prices( open_: float, high: float, low: float, close: float, n_checks: int, ) -> List[float]: """1분 OHLC → 실매 10초 폴링 흉내 가격 경로 (open→high→low→close). **high·low 앵커는 반드시 경유** — 선형 보간만 쓰면 고가/저가를 빗나가 어깨컷이 어긋난다. """ n = max(2, int(n_checks)) anchors: List[float] = [] for px in (float(open_), float(high), float(low), float(close)): if not anchors or px != anchors[-1]: anchors.append(px) if len(anchors) == 1: return [anchors[0]] * n if n <= len(anchors): return anchors[:n] # 앵커(o,h,l,c) 사이에 보조 점 삽입해 n개 맞춤 seg_count = len(anchors) - 1 extra_total = n - len(anchors) extra_per_seg = [extra_total // seg_count] * seg_count for i in range(extra_total % seg_count): extra_per_seg[i] += 1 out: List[float] = [anchors[0]] for si in range(seg_count): a, b = anchors[si], anchors[si + 1] extras = extra_per_seg[si] for j in range(1, extras + 1): t = j / (extras + 1) out.append(a + (b - a) * t) if out[-1] != b: out.append(b) return out[:n] if len(out) >= n else out + [out[-1]] * (n - len(out)) def check_sell_signal_backtest_bar( position: Dict[str, Any], candle: Dict[str, Any], params: Dict[str, Any], is_eod: bool = False, sell_fn: Optional[Callable[..., Optional[tuple]]] = None, low_mode: str = "current", ) -> Optional[tuple]: """백테 전용 청산 — 1분·N분 OHLC를 N회 가격 체크로 쪼개 실매와 정렬. 실매 폴링 흉내: - ``low_mode="current"`` (스캘핑·모멘텀): ``low=close=current_price`` - ``low_mode="session_low"`` (꼬리잡기): ``low=누적 session_low``, ``close=current_price`` ``sell_fn`` 미지정 시 ``check_sell_signal_live`` (scalping_engine). ``BACKTEST_EXIT_CHECKS_PER_BAR`` (기본 6) = 1 이면 구 OHLC 1회 체크. """ if sell_fn is None: sell_fn = check_sell_signal_live n_checks = get_env_int("BACKTEST_EXIT_CHECKS_PER_BAR", 6) if n_checks <= 1: return sell_fn(position, candle, params, is_eod) o = float(candle.get("open", candle["close"])) h = float(candle.get("high", candle["close"])) l = float(candle.get("low", candle["close"])) c = float(candle["close"]) ct = candle.get("candle_time", "") prices = _intrabar_exit_prices(o, h, l, c, n_checks) result: Optional[tuple] = None session_low: Optional[float] = None if str(low_mode).strip().lower() == "session_low": ep = float(position.get("entry_price", 0) or 0) session_low = float(position.get("session_low", ep) or ep) for idx, px in enumerate(prices): mp = float(position.get("max_price", 0) or 0) if px > mp: position["max_price"] = px mp = px if session_low is not None: session_low = min(session_low, px) position["session_low"] = session_low lo_sim = session_low else: lo_sim = px sim = { "open": o, "high": mp, "low": lo_sim, "close": px, "candle_time": ct, } eod_here = bool(is_eod and idx == len(prices) - 1) result = sell_fn(position, sim, params, is_eod=eod_here) if result: return result return result def check_sell_signal_live( position: Dict[str, Any], current_candle: Dict[str, Any], params: Dict[str, Any], is_eod: bool = False, ) -> Optional[tuple]: """ 실시간 봇 및 백테스트 공용: 보유 포지션 청산 (tail_engine V4 어깨컷과 동일 우선순위). position: { "entry_price", "entry_time", "qty", "stop", "target", "max_price" } current_candle: { "high", "low", "close", "candle_time"(optional) } 반환: (reason_str, exit_price) 또는 None [청산 우선순위 — 꼬리잡기 tail_engine.check_sell_signal_live 와 동일] 1순위 어깨컷: max_price 갱신 후 shoulder_cut_pct 되돌림 (저가 lo 기준, 매도선 체결) 2순위 익절 (target) 3순위 손절 (stop) 4순위 금액손실컷 — 어깨 미발동(수익 문턱 미도달)일 때만 5순위 장마감청산 """ shoulder_min_high = float(params.get("shoulder_min_high", 0.005)) shoulder_cut_pct = float(params.get("shoulder_cut_pct", 0.003)) max_loss_krw = float(params.get("max_loss_krw", 200000.0)) min_hold_sec = float(params.get("min_hold_sec", 30.0)) hi = float(current_candle.get("high", current_candle["close"])) lo = float(current_candle.get("low", current_candle["close"])) cl = float(current_candle["close"]) candle_time = current_candle.get("candle_time", "") max_p = max(float(position.get("max_price", 0) or 0), hi) position["max_price"] = max_p ep = float(position["entry_price"]) stop = float(position["stop"]) target = float(position["target"]) qty = int(position.get("qty", 1) or 1) if (not is_eod) and candle_time and position.get("entry_time"): try: entry_dt = _t2dt(position["entry_time"]) curr_dt = _t2dt(candle_time) if (curr_dt - entry_dt).total_seconds() < min_hold_sec: return None except Exception: pass reason = None exit_price = cl profit_val = (lo - ep) * qty drop_pct = (ep - lo) / ep if ep > 0 else 0.0 min_drop_pct = float(params.get("min_drop_pct_for_loss_cut", 0.015)) trail_armed = ep > 0 and max_p >= ep * (1.0 + shoulder_min_high) trail_stop_px = max_p * (1.0 - shoulder_cut_pct) if trail_armed else 0.0 trail_hit = trail_armed and lo > 0 and lo <= trail_stop_px if trail_hit: reason = "어깨컷" exit_price = trail_stop_px elif hi >= target: reason = "익절" exit_price = target elif lo > 0 and lo <= stop: reason = "손절" exit_price = stop elif ( not trail_armed and profit_val <= -max_loss_krw and drop_pct >= min_drop_pct ): reason = "금액손실컷" exit_price = ep - (max_loss_krw / qty) if qty > 0 else lo elif is_eod: reason = "장마감청산" exit_price = cl if reason: return (reason, exit_price) return None def run_scalping_backtest_rust_experimental( codes_candles: Dict[str, List[Dict]], params: Dict[str, Any], ) -> List[Dict]: """ Rust 엔진 (kis_rust_core) 을 통한 초고속 스캘핑 백테스트 (실험). """ try: import kis_rust_core from kis_rust_core import ScalpParams, CandleData except ImportError as e: from kis_trader.utils.logger import get_logger get_logger("kis_trader.scalping_engine").error(f"Rust core import failed: {e}") return [] use_defense_filters = str(params.get("use_defense_filters", True)).strip().lower() in ("1", "true", "t", "y", "yes", "on") # skip_hts_scan_dupes 는 HTS 조건검색 엔진 쓸 때 낙폭/RSI 중복을 끌지 여부. skip_hts = False if "skip_hts_scan_dupes" in params: skip_hts = str(params.get("skip_hts_scan_dupes")).strip().lower() in ("1", "true", "t", "y", "yes", "on") else: src = str(params.get("SCALP_UNIVERSE_SOURCE", "condition")).strip().lower() skip_hts = src in ("kiwoom_condition", "condition") rp = ScalpParams( rsi_period=int(params.get("rsi_period", 3)), rsi_oversold=float(params.get("rsi_oversold", 25.0)), rsi_overbought=float(params.get("rsi_overbought", 75.0)), sl_pct=abs(float(params.get("sl_pct", 0.015))), tp_pct=effective_tp_pct_from_params(params), drop_rate=float(params.get("drop_rate", 0.015)), cooldown_min=float(params.get("cooldown_min", 10.0)), max_daily=int(params.get("max_daily", 3)), high_chase_thr=float(params.get("high_chase_thr", 0.96)), max_daily_chg=float(params.get("max_daily_chg", 20.0)), min_price=float(params.get("min_price", 1000.0)), vol_mult=float(params.get("vol_mult", 0.0)), use_defense_filters=use_defense_filters, skip_hts=skip_hts, time_start_hm=int(params.get("time_start_hm", 900)), time_end_hm=int(params.get("time_end_hm", 1530)) ) all_trades = [] for code, rows in codes_candles.items(): if not rows: continue # 캔들 변환 rust_candles = [] for r in rows: rust_candles.append(CandleData( str(r["candle_time"]), float(r["open"]), float(r["high"]), float(r["low"]), float(r["close"]), float(r.get("volume", 0)), float(r.get("rsi", 50.0)) # TODO: rust 내부에서 rsi 계산하도록 변경 필요 )) res = kis_rust_core.run_scalp_backtest_fast(code, rust_candles, rp) for t in res: all_trades.append({ "code": t.code, "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, "entry": t.buy_price, "exit": t.sell_price, "exit_reason": t.reason, "qty": 1, # 임시 "pnl": 0, # PnL 부착부에서 재계산 "profit_rate": round(t.pnl_pct, 2), "hold_min": 0, "sell_reason": t.reason, "rsi_entry": round(t.rsi_entry, 1), "is_rust_core": True, }) return all_trades