e1ac8d119b
- 프론트엔드 UI 업데이트 (backtest.html, backtest.js) 엔진 라디오 버튼 통합 관련 반영 - Rust 플러그인(kis_rust_core) 및 컴파일 소스코드 추가 - CLI 백테스트 스크립트 수정 및 최신화 - 기타 스크래치 테스트 스크립트, 로그 요약 마크다운(.md) 등 누락 파일 일괄 반영 - 추가적으로 아직 발견되지 않은 엣지 케이스나 렌더링 오류가 포함되어 있을 가능성이 있음
493 lines
19 KiB
Python
493 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
스캘핑·모멘텀 백테스트 공통 로더 — backtest_web / param_search 가
|
|
동일한 캔들·유니버스·손익 계산을 쓰도록 단일 진입점.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from kis_trader.backtest.backtest_portfolio_common import (
|
|
attach_scalp_trade_pnl,
|
|
backtest_slip_pct,
|
|
build_budget_warning,
|
|
fee_and_slot_from_env_row,
|
|
merge_portfolio_into_params,
|
|
min_invest_ratio_of_slot,
|
|
resolve_portfolio_params,
|
|
resolve_trigger_snapshots_for_backtest,
|
|
summarize_trades,
|
|
)
|
|
from kis_trader.backtest.breakout_tick_loader import (
|
|
load_common_ticks_by_code,
|
|
tick_coverage_stats,
|
|
)
|
|
from kis_trader.engine import scalping_engine as se
|
|
from kis_trader.engine.tick_exit_common import strategy_use_tick_exit
|
|
from kis_trader.utils.env import get_env_bool
|
|
|
|
SCALP_STRATEGY_ID = "SCALP"
|
|
MOMENTUM_STRATEGY_ID = "MOMENTUM"
|
|
|
|
|
|
def date_keys(start: str, end: str) -> Tuple[str, str, str, str]:
|
|
"""YYYY-MM-DD → candle_time 키 및 ymd."""
|
|
start_key = start.replace("-", "") + "0000"
|
|
end_key = end.replace("-", "") + "2359"
|
|
return start_key, end_key, start_key[:8], end_key[:8]
|
|
|
|
|
|
def resolve_scalp_universe(
|
|
start_ymd: str,
|
|
end_ymd: str,
|
|
*,
|
|
use_saved_history: bool,
|
|
strategy_id: str = SCALP_STRATEGY_ID,
|
|
history_source: str = "kiwoom",
|
|
) -> Tuple[Optional[Dict[str, List[str]]], str, int, int]:
|
|
"""
|
|
backtest_web 유니버스 해석과 동일.
|
|
|
|
Returns:
|
|
(universe_by_slot, source_label, history_slot_count, scan_interval_min)
|
|
"""
|
|
if use_saved_history and strategy_id:
|
|
try:
|
|
from kis_trader.database.db_manager import get_db as _get_ext_db
|
|
from kis_trader.backtest.universe_history_source import (
|
|
history_source_label,
|
|
resolve_backtest_universe_history_source,
|
|
)
|
|
|
|
debounce_sec = scalp_universe_exit_debounce_sec()
|
|
hs = resolve_backtest_universe_history_source(history_source)
|
|
history = _get_ext_db().get_universe_by_candle_time(
|
|
strategy_id=strategy_id,
|
|
start_ymd=start_ymd,
|
|
end_ymd=end_ymd,
|
|
exit_debounce_sec=debounce_sec,
|
|
history_source=hs,
|
|
)
|
|
if history:
|
|
return history, history_source_label(hs), len(history), 1
|
|
except Exception:
|
|
pass
|
|
return None, "all", 0, 1
|
|
|
|
|
|
def scalp_universe_exit_debounce_sec() -> int:
|
|
"""실매 ``CONDITION_EXIT_GRACE_SEC`` 정합 — 스냅샷 축소 시 N초 유지."""
|
|
from kis_trader.backtest.universe_timeline import universe_exit_debounce_sec_for_strategy
|
|
|
|
return universe_exit_debounce_sec_for_strategy("SCALP")
|
|
|
|
|
|
def scalp_backtest_candle_warmup_bars() -> int:
|
|
"""백테 RSI 등 warm-up — 실매 봉 버퍼와 같이 기간 시작 전 N봉 prepend."""
|
|
from kis_trader.utils.env import get_env_int
|
|
|
|
return max(0, int(get_env_int("SCALP_BACKTEST_CANDLE_WARMUP_BARS", 50)))
|
|
|
|
|
|
def prepend_scalp_candle_warmup(
|
|
db,
|
|
candles_by_code: Dict[str, List[Dict]],
|
|
period_start_key: str,
|
|
*,
|
|
warmup_bars: Optional[int] = None,
|
|
history_source: str = "kiwoom",
|
|
) -> int:
|
|
"""
|
|
``period_start_key``(YYYYMMDDHHMM) 이전 N봉을 종목별로 prepend.
|
|
RSI 판별용 — 포트폴리오 all_times 는 ``_backtest_period_start_key`` 로 기간만 사용.
|
|
"""
|
|
wb = (
|
|
scalp_backtest_candle_warmup_bars()
|
|
if warmup_bars is None
|
|
else max(0, int(warmup_bars))
|
|
)
|
|
if wb <= 0 or db is None or not period_start_key:
|
|
return 0
|
|
hs = str(history_source or "kiwoom").strip().lower()
|
|
if hs in ("ls", "ls_condition", "ls_ws"):
|
|
from kis_trader.backtest.ls_history_loaders import prepend_ls_candle_warmup
|
|
|
|
return prepend_ls_candle_warmup(
|
|
db, candles_by_code, str(period_start_key)[:12], wb,
|
|
)
|
|
ps = str(period_start_key)[:12]
|
|
total_prepended = 0
|
|
from kis_trader.backtest.bt_candle_source import fetch_ws_candles_warmup_before
|
|
for code, rows in list(candles_by_code.items()):
|
|
if not rows:
|
|
continue
|
|
first_period_idx = None
|
|
for i, r in enumerate(rows):
|
|
ct = str(r.get("candle_time") or "")
|
|
if ct >= ps:
|
|
first_period_idx = i
|
|
break
|
|
if first_period_idx is None:
|
|
continue
|
|
# 이미 기간 전 봉이 있으면 skip (idempotent)
|
|
if first_period_idx > 0:
|
|
continue
|
|
first_ct = str(rows[first_period_idx].get("candle_time") or "")
|
|
if not first_ct:
|
|
continue
|
|
warm_rows = fetch_ws_candles_warmup_before(
|
|
db, code, 1, first_ct, wb,
|
|
)
|
|
if not warm_rows:
|
|
continue
|
|
prefix = warm_rows
|
|
candles_by_code[code] = prefix + [dict(r) for r in rows]
|
|
total_prepended += len(prefix)
|
|
return total_prepended
|
|
|
|
|
|
def load_scalp_candles_by_code(
|
|
db,
|
|
start_key: str,
|
|
end_key: str,
|
|
rsi_period: int = 3,
|
|
*,
|
|
history_source: str = "kiwoom",
|
|
) -> Tuple[Dict[str, List[Dict]], int]:
|
|
"""1분봉 전 종목 로드 (+ 웜업). history_source=ls → ls_ws_candles.
|
|
|
|
반환은 ``(candles_by_code, total_rows)`` 만 — 호출부에 ``codes`` 리스트 없음.
|
|
웹 ``codes_analyzed`` 는 ``_codes_from_candles_map`` 으로 바인딩할 것.
|
|
"""
|
|
min_bars = int(rsi_period) + 5
|
|
hs = str(history_source or "kiwoom").strip().lower()
|
|
if hs in ("ls", "ls_condition", "ls_ws"):
|
|
from kis_trader.backtest.ls_history_loaders import load_ls_candles_by_code
|
|
|
|
candles_by_code, total_candles = load_ls_candles_by_code(
|
|
db, start_key, end_key, min_bars=min_bars,
|
|
)
|
|
prepend_scalp_candle_warmup(
|
|
db, candles_by_code, str(start_key)[:12], history_source="ls",
|
|
)
|
|
return candles_by_code, total_candles
|
|
|
|
from kis_trader.backtest.bt_candle_source import fetch_ws_candles_by_code_bulk
|
|
|
|
loaded = fetch_ws_candles_by_code_bulk(
|
|
db, 1, start_key, end_key,
|
|
confirmed_only=True,
|
|
)
|
|
candles_by_code: Dict[str, List[Dict]] = {}
|
|
total_candles = 0
|
|
for code, rows in loaded.items():
|
|
if len(rows) < min_bars:
|
|
continue
|
|
candles_by_code[code] = rows
|
|
total_candles += len(rows)
|
|
|
|
prepend_scalp_candle_warmup(db, candles_by_code, str(start_key)[:12])
|
|
return candles_by_code, total_candles
|
|
|
|
|
|
def _scalp_backtest_wants_ticks(params: Optional[Dict[str, Any]] = None) -> bool:
|
|
"""청산·진입 틱 재생이 필요한지 (기본 ON)."""
|
|
if strategy_use_tick_exit(params, "SCALP_BACKTEST_USE_TICK_EXIT", default=True):
|
|
return True
|
|
if params is not None and params.get("backtest_use_tick_entry") is not None:
|
|
return se._to_bool(params.get("backtest_use_tick_entry"), True)
|
|
return get_env_bool("SCALP_BACKTEST_USE_TICK_ENTRY", True)
|
|
|
|
|
|
def run_scalping_backtest_web_aligned(
|
|
candles_by_code: Dict[str, List[Dict]],
|
|
params: Dict[str, Any],
|
|
universe_by_slot: Optional[Dict[str, List[str]]],
|
|
*,
|
|
slot_money: float,
|
|
fee_rate: float,
|
|
sell_tax: float,
|
|
max_stocks: Optional[int] = None,
|
|
total_budget_krw: Optional[float] = None,
|
|
meta_out: Optional[Dict[str, Any]] = None,
|
|
mode: str = "reversal",
|
|
ticks_by_code: Optional[Dict[str, Dict[str, List[Dict]]]] = None,
|
|
) -> List[Dict]:
|
|
"""엔진 1회 + 웹과 동일 손익 부착 (reversal / momentum)."""
|
|
engine_params = dict(params)
|
|
engine_params["slot_money"] = float(slot_money)
|
|
if max_stocks is not None:
|
|
engine_params["max_stocks"] = int(max_stocks)
|
|
if total_budget_krw is not None:
|
|
tb = float(total_budget_krw)
|
|
engine_params["total_budget_krw"] = tb if tb > 0 else float(
|
|
int(engine_params.get("max_stocks") or 3) * slot_money
|
|
)
|
|
if universe_by_slot is not None:
|
|
engine_params.setdefault("scan_interval_min", 1)
|
|
engine_params.setdefault("portfolio_mode", True)
|
|
|
|
# 기간 시작 키 — 웜업 봉이 all_times 에 섞이지 않도록 (돌파·모멘텀과 동일)
|
|
_sk_w = str((meta_out or {}).get("start_key") or "")[:12]
|
|
if _sk_w:
|
|
engine_params["_backtest_period_start_key"] = _sk_w
|
|
_db_w = (meta_out or {}).get("db")
|
|
if _db_w is not None and str(mode).strip().lower() != "momentum":
|
|
_hs_w = str(
|
|
engine_params.get("_universe_history_source")
|
|
or engine_params.get("universe_history_source")
|
|
or (meta_out or {}).get("universe_history_source")
|
|
or "kiwoom"
|
|
).strip().lower()
|
|
prepend_scalp_candle_warmup(
|
|
_db_w, candles_by_code, _sk_w, history_source=_hs_w,
|
|
)
|
|
|
|
from kis_trader.backtest.backtest_env_timeline import attach_backtest_env_timeline_to_params
|
|
attach_backtest_env_timeline_to_params(engine_params, meta_out, "SCALP")
|
|
|
|
if str(mode).strip().lower() == "momentum":
|
|
from kis_trader.backtest import momentum_backtest_common as mbc
|
|
trades = mbc.run_momentum_backtest_web_aligned(
|
|
candles_by_code, engine_params, universe_by_slot,
|
|
slot_money=float(slot_money),
|
|
fee_rate=float(fee_rate),
|
|
sell_tax=float(sell_tax),
|
|
max_stocks=int(engine_params.get("max_stocks") or max_stocks or 3),
|
|
total_budget_krw=float(engine_params.get("total_budget_krw") or 0),
|
|
meta_out=meta_out,
|
|
)
|
|
else:
|
|
loaded_ticks: Dict[str, Dict[str, List[Dict]]] = dict(ticks_by_code or {})
|
|
tick_meta: Dict[str, Any] = {}
|
|
if _scalp_backtest_wants_ticks(engine_params):
|
|
if not loaded_ticks and meta_out is not None:
|
|
start_key = str(meta_out.get("start_key") or "")
|
|
end_key = str(meta_out.get("end_key") or "")
|
|
db = meta_out.get("db")
|
|
if db is None and start_key and end_key:
|
|
from kis_trader.backtest.backtest_portfolio_common import ensure_meta_db
|
|
|
|
db = ensure_meta_db(meta_out)
|
|
if db and start_key and end_key:
|
|
_hs_tick = str(
|
|
engine_params.get("_universe_history_source")
|
|
or engine_params.get("universe_history_source")
|
|
or (meta_out or {}).get("universe_history_source")
|
|
or "kiwoom"
|
|
).strip().lower()
|
|
if _hs_tick in ("ls", "ls_condition", "ls_ws"):
|
|
from kis_trader.backtest.ls_history_loaders import (
|
|
load_ls_ticks_by_code,
|
|
)
|
|
|
|
loaded_ticks, tick_rows = load_ls_ticks_by_code(
|
|
db, start_key, end_key, set(candles_by_code.keys()),
|
|
)
|
|
_tick_tbl = "ls_ws_ticks"
|
|
else:
|
|
loaded_ticks, tick_rows = load_common_ticks_by_code(
|
|
db, start_key, end_key, set(candles_by_code.keys()),
|
|
use_spill_fallback=(str(engine_params.get("backtest_use_spill_fallback", True)).lower() in ("1", "true", "t", "y", "yes", "on")),
|
|
)
|
|
_tick_tbl = "ws_ticks"
|
|
tick_meta = tick_coverage_stats(candles_by_code, loaded_ticks)
|
|
tick_meta["ws_tick_rows_loaded"] = tick_rows
|
|
tick_meta["tick_table"] = _tick_tbl
|
|
if tick_rows <= 0:
|
|
from kis_trader.utils.logger import get_logger as _get_logger
|
|
|
|
_get_logger("kis_trader.scalping_backtest").warning(
|
|
"⚠️ %s 데이터 없음 — SCALP 틱 청산/진입 스킵 "
|
|
"(틱 수집 후 재백테 권장, FALLBACK_OHLC=1 시 OHLC)",
|
|
_tick_tbl,
|
|
)
|
|
elif loaded_ticks:
|
|
if "_tick_meta_cached" in engine_params:
|
|
tick_meta = dict(engine_params["_tick_meta_cached"])
|
|
else:
|
|
tick_meta = tick_coverage_stats(candles_by_code, loaded_ticks)
|
|
engine_params["_tick_meta_cached"] = tick_meta
|
|
|
|
tick_meta["ws_tick_rows_loaded"] = sum(
|
|
len(lst) for cm in loaded_ticks.values() for lst in cm.values()
|
|
)
|
|
|
|
# 진입·프로그램 호가 스냅 — 모멘텀/돌파/꼬리 web_aligned 와 동일.
|
|
# 미로드 시 호가필터 ON + REJECT_IF_EMPTY → 전부「호가없음」0건.
|
|
# Optuna/그리드가 이미 _bt_orderbook_by_code 를 넣었으면 재사용(비었을 때만 DB).
|
|
ob_loaded, pg_loaded, snap_meta = resolve_trigger_snapshots_for_backtest(
|
|
candles_by_code,
|
|
engine_params,
|
|
strategy=SCALP_STRATEGY_ID,
|
|
meta_out=meta_out,
|
|
orderbook_by_code=engine_params.get("_bt_orderbook_by_code"),
|
|
program_by_code=engine_params.get("_bt_program_by_code"),
|
|
)
|
|
if ob_loaded is not None:
|
|
engine_params["_bt_orderbook_by_code"] = ob_loaded
|
|
if pg_loaded is not None:
|
|
engine_params["_bt_program_by_code"] = pg_loaded
|
|
|
|
rust_session = engine_params.get("_rust_session_id")
|
|
if rust_session:
|
|
try:
|
|
import kis_rust_core
|
|
p = engine_params
|
|
def get_float(k, default=0.0):
|
|
try: return float(p.get(k, default))
|
|
except: return float(default)
|
|
def get_int(k, default=0):
|
|
try: return int(p.get(k, default))
|
|
except: return int(default)
|
|
def get_bool(k, default=False):
|
|
v = p.get(k, default)
|
|
if isinstance(v, bool): return v
|
|
s = str(v).lower()
|
|
if s in ("1", "true", "yes", "t", "y"): return True
|
|
return False
|
|
|
|
sp = kis_rust_core.ScalpParams(
|
|
get_int("rsi_period", 14),
|
|
get_float("rsi_oversold", 30.0),
|
|
get_float("rsi_overbought", 70.0),
|
|
abs(get_float("sl_pct", 0.02)),
|
|
get_float("tp_pct", 0.05),
|
|
get_float("drop_rate", 0.0),
|
|
get_float("cooldown_min", 0.0),
|
|
get_int("max_daily", 10),
|
|
get_float("high_chase_thr", 1.2),
|
|
get_float("max_daily_chg", 25.0),
|
|
get_float("min_price", 1000.0),
|
|
get_float("vol_mult", 0.0),
|
|
get_bool("use_defense_filters", False),
|
|
get_bool("skip_hts", False),
|
|
get_int("time_start_hm", 900),
|
|
get_int("time_end_hm", 1520),
|
|
)
|
|
r_trades = kis_rust_core.run_engine_trial_scalp(rust_session, sp)
|
|
trades = [
|
|
{
|
|
"code": rt.code,
|
|
"entry_time": rt.buy_time,
|
|
"exit_time": rt.sell_time,
|
|
"entry": rt.buy_price,
|
|
"exit": rt.sell_price,
|
|
"pnl": 0,
|
|
"pnl_pct": rt.pnl_pct,
|
|
"reason": rt.reason,
|
|
"peak_price": rt.sell_price, # 간소화
|
|
"rsi": rt.rsi_entry,
|
|
"qty": None,
|
|
"entry_source": "rust_engine",
|
|
"ob_source": "",
|
|
"hold_min": 0,
|
|
}
|
|
for rt in r_trades
|
|
]
|
|
except Exception as e:
|
|
print(f"[ERROR] Rust run_engine_trial_scalp failed: {e}")
|
|
trades = []
|
|
else:
|
|
trades = se.run_scalping_backtest(
|
|
candles_by_code, engine_params, universe_by_slot=universe_by_slot,
|
|
ticks_by_code=loaded_ticks or None,
|
|
)
|
|
if meta_out is not None and tick_meta:
|
|
from kis_trader.backtest.breakout_tick_loader import enrich_tick_meta_with_traded_codes
|
|
tick_meta = enrich_tick_meta_with_traded_codes(
|
|
tick_meta, candles_by_code, loaded_ticks, trades,
|
|
)
|
|
meta_out["tick_backtest"] = tick_meta
|
|
if tick_meta.get("ws_tick_rows_loaded", 0) > 0:
|
|
meta_out["backtest_buy_source"] = "ws_ticks"
|
|
elif _scalp_backtest_wants_ticks(engine_params):
|
|
meta_out["backtest_buy_source"] = "ohlc_fallback"
|
|
else:
|
|
meta_out["backtest_buy_source"] = "align"
|
|
if meta_out is not None:
|
|
meta_out["universe_exit_debounce_sec"] = scalp_universe_exit_debounce_sec()
|
|
meta_out["candle_warmup_bars"] = scalp_backtest_candle_warmup_bars()
|
|
if snap_meta:
|
|
meta_out["trigger_snapshot_backtest"] = snap_meta
|
|
|
|
try:
|
|
from kis_trader.backtest.bt_post_filters import apply_bt_post_filters
|
|
trades = apply_bt_post_filters(
|
|
trades,
|
|
engine_params,
|
|
strategy="SCALP"
|
|
)
|
|
except Exception as e:
|
|
print(f"[ERROR] bt_post_filters failed: {e}")
|
|
|
|
attach_scalp_trade_pnl(
|
|
trades, fee_rate=fee_rate, sell_tax=sell_tax,
|
|
slip_pct=backtest_slip_pct(engine_params),
|
|
)
|
|
if meta_out is not None and str(mode).strip().lower() != "momentum":
|
|
skip_stats = engine_params.get("_portfolio_skip_stats") or {}
|
|
meta_out["skip_stats"] = dict(skip_stats)
|
|
meta_out["engine_params"] = engine_params
|
|
return trades
|
|
|
|
|
|
def resolve_scalp_portfolio_params(
|
|
env_row: Optional[Dict[str, Any]],
|
|
base_defaults: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
strategy: str = "SCALP",
|
|
slot_money: Optional[float] = None,
|
|
max_stocks: Optional[int] = None,
|
|
total_budget_krw: Optional[float] = None,
|
|
) -> Dict[str, Any]:
|
|
"""웹·파라서치 공통 포트폴리오 해석."""
|
|
return resolve_portfolio_params(
|
|
env_row,
|
|
base_defaults,
|
|
strategy=strategy,
|
|
slot_money=slot_money,
|
|
max_stocks=max_stocks,
|
|
total_budget_krw=total_budget_krw,
|
|
)
|
|
|
|
|
|
def merge_scalp_portfolio_into_params(
|
|
params: Dict[str, Any],
|
|
portfolio: Dict[str, Any],
|
|
) -> Dict[str, Any]:
|
|
return merge_portfolio_into_params(params, portfolio)
|
|
|
|
|
|
def build_scalp_budget_warning(
|
|
portfolio: Dict[str, Any],
|
|
skip_stats: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
strategy: str = "SCALP",
|
|
) -> Optional[str]:
|
|
ratio = min_invest_ratio_of_slot({}, strategy=strategy)
|
|
return build_budget_warning(portfolio, skip_stats, min_invest_ratio=ratio)
|
|
|
|
|
|
def summarize_scalp_trades(
|
|
trades: List[Dict],
|
|
*,
|
|
total_budget_krw: float,
|
|
period_days: int = 1,
|
|
) -> Dict[str, Any]:
|
|
return summarize_trades(
|
|
trades,
|
|
total_budget_krw=total_budget_krw,
|
|
period_days=period_days,
|
|
)
|
|
|
|
|
|
def fee_and_slot_from_env(
|
|
row: Optional[Dict[str, Any]],
|
|
*,
|
|
strategy: str = "SCALP",
|
|
) -> Tuple[float, float, float]:
|
|
return fee_and_slot_from_env_row(row, strategy=strategy)
|