385 lines
12 KiB
Python
385 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
백테 틱 청산 공통 — 실매 폴링(STRATEGY_LOOP_SLEEP≈0.1초) 근사.
|
|
|
|
1분·N분봉 OHLC intrabar(open→high→low→close)는 익절/어깨를 손절보다 먼저
|
|
체결하는 낙관 편향을 만든다. 전 전략 백테·파람서치·Optuna는 ws_ticks 시간순 재생.
|
|
|
|
■ 절대규칙 (정합)
|
|
틱 청산/틱 진입 ON 이면 OHLC 폴백으로 체결·고점·손익 숫자를 **절대 변조하지 않는다**.
|
|
``*_TICK_FALLBACK_OHLC`` 체크가 ON이어도 ``use_tick_exit`` 경로에서는 무시한다.
|
|
(봉 high 선반영·EOD OHLC 강제 폴백 = 실매와 다른 엔진 — 금지)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Callable, Dict, List, Optional, Tuple
|
|
|
|
from kis_trader.utils.env import get_env_bool, get_env_float, get_env_int
|
|
|
|
|
|
SellFn = Callable[..., Optional[tuple]]
|
|
|
|
|
|
def _param_bool(
|
|
params: Optional[Dict[str, Any]],
|
|
param_key: str,
|
|
env_key: str,
|
|
default: bool,
|
|
) -> bool:
|
|
if params is not None and params.get(param_key) is not None:
|
|
s = str(params.get(param_key)).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 get_env_bool(env_key, default)
|
|
|
|
|
|
def parse_backtest_time(t: str) -> datetime:
|
|
from kis_trader.utils.trade_time import parse_trade_datetime
|
|
return parse_trade_datetime(t)
|
|
|
|
|
|
def _tick_time_to_ms(tick_time: str) -> int:
|
|
dt = parse_backtest_time(tick_time)
|
|
return int(dt.timestamp() * 1000)
|
|
|
|
|
|
def backtest_tick_poll_ms(
|
|
params: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
strategy_env: str = "",
|
|
default: int = 100,
|
|
) -> int:
|
|
"""틱 청산 폴링 간격(ms). 실매 STRATEGY_LOOP_SLEEP≈0.1초 정합 · 하한 50."""
|
|
if params is not None and params.get("backtest_tick_poll_ms") is not None:
|
|
try:
|
|
return max(50, int(float(params["backtest_tick_poll_ms"])))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if strategy_env:
|
|
v = get_env_int(strategy_env, 0)
|
|
if v > 0:
|
|
return max(50, int(v))
|
|
return max(50, int(get_env_int("BACKTEST_TICK_POLL_MS", default)))
|
|
|
|
|
|
def backtest_sell_slip_pct(
|
|
params: Optional[Dict[str, Any]] = None,
|
|
*,
|
|
strategy_env: str = "",
|
|
) -> float:
|
|
if params is not None and params.get("backtest_sell_slip_pct") is not None:
|
|
try:
|
|
return abs(float(params["backtest_sell_slip_pct"]))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if strategy_env:
|
|
return abs(float(get_env_float(strategy_env, 0.0)))
|
|
return abs(float(get_env_float("BACKTEST_SELL_SLIP_PCT", 0.0)))
|
|
|
|
|
|
def strategy_use_tick_exit(
|
|
params: Optional[Dict[str, Any]],
|
|
env_key: str,
|
|
*,
|
|
default: bool = True,
|
|
) -> bool:
|
|
return _param_bool(params, "backtest_use_tick_exit", env_key, default)
|
|
|
|
|
|
def strategy_tick_fallback_ohlc(
|
|
params: Optional[Dict[str, Any]],
|
|
env_key: str,
|
|
*,
|
|
default: bool = False,
|
|
) -> bool:
|
|
"""OHLC 폴백 플래그 조회.
|
|
|
|
기본 False. 틱 청산 ON 경로에서는 ``resolve_backtest_sell`` 이 이 값을 **무시**한다.
|
|
Optuna/파람서치는 호출측에서 False 고정.
|
|
"""
|
|
return _param_bool(params, "backtest_tick_fallback_ohlc", env_key, default)
|
|
|
|
|
|
def collect_minute_ticks(
|
|
ticks_by_code: Optional[Dict[str, Dict[str, List[Dict[str, Any]]]]],
|
|
code: str,
|
|
minute_key: str,
|
|
) -> List[Dict[str, Any]]:
|
|
if not ticks_by_code:
|
|
return []
|
|
bucket = ticks_by_code.get(code) or {}
|
|
try:
|
|
from kis_trader.backtest.shared_ticks import SharedBucketMapping
|
|
if isinstance(bucket, SharedBucketMapping):
|
|
return bucket.column_view_minute(minute_key)
|
|
except Exception:
|
|
pass
|
|
ticks = list(bucket.get(str(minute_key)[:12]) or [])
|
|
ticks.sort(key=lambda x: str(x.get("tick_time") or ""))
|
|
return ticks
|
|
|
|
|
|
def try_sell_on_ticks(
|
|
position: Dict[str, Any],
|
|
ticks: List[Dict[str, Any]],
|
|
params: Dict[str, Any],
|
|
sell_fn: SellFn,
|
|
*,
|
|
is_eod: bool = False,
|
|
entry_time: str = "",
|
|
poll_ms: int = 100,
|
|
slip_pct: float = 0.0,
|
|
low_mode: str = "current",
|
|
) -> Optional[Tuple[str, float, str, float]]:
|
|
"""
|
|
틱 시간순 청산 검사.
|
|
|
|
Returns:
|
|
(reason, fill_price, sell_time, hold_min) 또는 None
|
|
"""
|
|
try:
|
|
from kis_trader.backtest.shared_ticks import TickColumnView
|
|
except ImportError:
|
|
TickColumnView = None # type: ignore[misc, assignment]
|
|
if TickColumnView is not None and isinstance(ticks, TickColumnView):
|
|
return _try_sell_on_ticks_columnar(
|
|
position, ticks, params, sell_fn,
|
|
is_eod=is_eod, entry_time=entry_time,
|
|
poll_ms=poll_ms, slip_pct=slip_pct, low_mode=low_mode,
|
|
)
|
|
|
|
if not ticks:
|
|
return None
|
|
|
|
entry_key = str(entry_time or "")[:12]
|
|
try:
|
|
entry_dt = parse_backtest_time(entry_time or ticks[0].get("tick_time", entry_key))
|
|
except ValueError:
|
|
entry_dt = parse_backtest_time(entry_key)
|
|
|
|
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)
|
|
|
|
last_check_ms = -10**15
|
|
n = len(ticks)
|
|
poll = max(50, int(poll_ms))
|
|
slip = abs(float(slip_pct))
|
|
|
|
for idx, tick in enumerate(ticks):
|
|
tt = str(tick.get("tick_time") or "")
|
|
if len(tt) < 12:
|
|
continue
|
|
if entry_key and tt[:12] < entry_key:
|
|
continue
|
|
|
|
try:
|
|
tick_ms = _tick_time_to_ms(tt)
|
|
except ValueError:
|
|
continue
|
|
|
|
px = float(tick.get("price") or 0)
|
|
if px <= 0:
|
|
continue
|
|
|
|
mp = max(float(position.get("max_price", position["entry_price"])), px)
|
|
position["max_price"] = mp
|
|
|
|
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
|
|
|
|
if tick_ms - last_check_ms < poll:
|
|
continue
|
|
last_check_ms = tick_ms
|
|
|
|
candle = {
|
|
"high": mp,
|
|
"low": lo_sim,
|
|
"close": px,
|
|
"candle_time": tt[:12],
|
|
}
|
|
eod_here = bool(is_eod and idx == n - 1)
|
|
res = sell_fn(position, candle, params, is_eod=eod_here)
|
|
if not res:
|
|
continue
|
|
|
|
reason, _theoretical = res
|
|
fill_px = px
|
|
if slip > 0:
|
|
fill_px = px * (1.0 - slip / 100.0)
|
|
|
|
try:
|
|
sell_dt = parse_backtest_time(tt)
|
|
except ValueError:
|
|
sell_dt = parse_backtest_time(tt[:12])
|
|
hold_min = round((sell_dt - entry_dt).total_seconds() / 60.0, 1)
|
|
sell_time = tt[:14] if len(tt) >= 14 else tt[:12]
|
|
return reason, fill_px, sell_time, hold_min
|
|
|
|
return None
|
|
|
|
|
|
def _try_sell_on_ticks_columnar(
|
|
position: Dict[str, Any],
|
|
view: Any,
|
|
params: Dict[str, Any],
|
|
sell_fn: SellFn,
|
|
*,
|
|
is_eod: bool = False,
|
|
entry_time: str = "",
|
|
poll_ms: int = 100,
|
|
slip_pct: float = 0.0,
|
|
low_mode: str = "current",
|
|
) -> Optional[Tuple[str, float, str, float]]:
|
|
if len(view) == 0:
|
|
return None
|
|
from kis_trader.engine.whipsaw_filter import _tick_epoch_sec
|
|
|
|
owner = view.owner
|
|
_epoch = owner._epoch
|
|
_price = owner._price
|
|
_tick_time = owner._tick_time
|
|
|
|
poll = max(50, int(poll_ms))
|
|
slip = abs(float(slip_pct))
|
|
entry_key = str(entry_time or "")[:12]
|
|
entry_key_epoch = _tick_epoch_sec(entry_key) if entry_key else 0
|
|
|
|
if entry_time:
|
|
_entry_src = entry_time
|
|
else:
|
|
_fi = view.first_idx()
|
|
_entry_src = _tick_time[_fi].decode("utf-8") if _fi >= 0 else entry_key
|
|
try:
|
|
entry_dt = parse_backtest_time(_entry_src)
|
|
except ValueError:
|
|
entry_dt = parse_backtest_time(entry_key)
|
|
|
|
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)
|
|
|
|
last_check_ms = -10**15
|
|
n = len(view)
|
|
idx = -1
|
|
for i in view.iter_idx():
|
|
idx += 1
|
|
ts = int(_epoch[i])
|
|
if ts <= 0:
|
|
continue
|
|
if entry_key and ts < entry_key_epoch:
|
|
continue
|
|
px = float(_price[i])
|
|
if px <= 0:
|
|
continue
|
|
|
|
mp = max(float(position.get("max_price", position["entry_price"])), px)
|
|
position["max_price"] = mp
|
|
|
|
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
|
|
|
|
tick_ms = ts * 1000
|
|
if tick_ms - last_check_ms < poll:
|
|
continue
|
|
last_check_ms = tick_ms
|
|
|
|
tt = _tick_time[i].decode("utf-8")
|
|
candle = {
|
|
"high": mp,
|
|
"low": lo_sim,
|
|
"close": px,
|
|
"candle_time": tt[:12],
|
|
}
|
|
eod_here = bool(is_eod and idx == n - 1)
|
|
res = sell_fn(position, candle, params, is_eod=eod_here)
|
|
if not res:
|
|
continue
|
|
|
|
reason, _theoretical = res
|
|
fill_px = px
|
|
if slip > 0:
|
|
fill_px = px * (1.0 - slip / 100.0)
|
|
|
|
try:
|
|
sell_dt = parse_backtest_time(tt)
|
|
except ValueError:
|
|
sell_dt = parse_backtest_time(tt[:12])
|
|
hold_min = round((sell_dt - entry_dt).total_seconds() / 60.0, 1)
|
|
sell_time = tt[:14] if len(tt) >= 14 else tt[:12]
|
|
return reason, fill_px, sell_time, hold_min
|
|
|
|
return None
|
|
|
|
|
|
def resolve_backtest_sell(
|
|
position: Dict[str, Any],
|
|
bar: Dict[str, Any],
|
|
params: Dict[str, Any],
|
|
*,
|
|
is_eod: bool = False,
|
|
sell_fn: Optional[SellFn] = None,
|
|
low_mode: str = "current",
|
|
ticks: Optional[List[Dict[str, Any]]] = None,
|
|
use_tick_exit: bool = True,
|
|
tick_fallback_ohlc: bool = False,
|
|
poll_ms: int = 100,
|
|
slip_pct: float = 0.0,
|
|
) -> Optional[Tuple[str, float, str, float, str]]:
|
|
"""
|
|
한 봉 청산 — 틱 청산 ON 이면 틱만. OHLC 폴백으로 숫자 변조 금지.
|
|
|
|
Returns:
|
|
(reason, fill_price, sell_time, hold_min, exit_source)
|
|
exit_source: ws_ticks | ohlc_bar
|
|
"""
|
|
from kis_trader.engine.scalping_engine import check_sell_signal_backtest_bar
|
|
|
|
if sell_fn is None:
|
|
from kis_trader.engine.scalping_engine import check_sell_signal_live as sell_fn
|
|
|
|
ct = str(bar.get("candle_time") or "")
|
|
entry_time = str(position.get("entry_time") or "")
|
|
|
|
if use_tick_exit:
|
|
if ticks:
|
|
tick_res = try_sell_on_ticks(
|
|
position, ticks, params, sell_fn,
|
|
is_eod=is_eod, entry_time=entry_time,
|
|
poll_ms=poll_ms, slip_pct=slip_pct, low_mode=low_mode,
|
|
)
|
|
if tick_res:
|
|
reason, fill_px, sell_time, hold_min = tick_res
|
|
return reason, fill_px, sell_time, hold_min, "ws_ticks"
|
|
# 절대규칙: 틱 청산 ON → OHLC intrabar 폴백 무시 (tick_fallback_ohlc 체크 무관)
|
|
# EOD 플랫은 호출측에서 직전가/장마감만 처리.
|
|
return None
|
|
|
|
# 틱 청산 OFF = 레거시 순수 OHLC 경로 (폴백이 아님 — 봉 OHLC 가 본체)
|
|
res = check_sell_signal_backtest_bar(
|
|
position, bar, params, is_eod=is_eod, sell_fn=sell_fn, low_mode=low_mode,
|
|
)
|
|
if not res:
|
|
return None
|
|
reason, exit_price = res
|
|
try:
|
|
entry_dt = parse_backtest_time(entry_time)
|
|
sell_dt = parse_backtest_time(ct)
|
|
hold_min = round((sell_dt - entry_dt).total_seconds() / 60.0, 1)
|
|
except ValueError:
|
|
hold_min = 0.0
|
|
return reason, float(exit_price), ct, hold_min, "ohlc_bar"
|