fc27e726f9
변경 사항 (Changes): 구문 오류(Syntax error) 및 토큰 낭비를 방지하기 위해 에이전트 쉘(Agent shell)과 파이썬 코드 스니펫에 다수의 신규 안전 규칙(Safety rules)을 추가함. 스키마 검증 및 적절한 SQL 포맷팅을 보장하기 위해 임시(Ad-hoc) 데이터베이스 쿼리 작성 가이드라인을 도입함. 코드 수정 후 UI 기능이 정상 작동하는지 확인하기 위해, 백테스트 웹 서비스 재시작 및 브라우저 검증에 대한 새로운 규칙을 구현함. 시스템 전반의 무결성(Integrity)을 유지하기 위해 실전 매매(Live trading), 웹 백테스팅, 파라미터 탐색(Parameter searches) 간의 일관성 검사(Consistency checks) 체계를 확립함. 기대 효과 (Impact): 이러한 개선 사항들은 트레이딩 시스템의 견고성(Robustness)과 신뢰성을 향상시키며, 에러 발생을 최소화하고 다양한 시스템 컴포넌트 간의 원활한 상호작용을 보장함.
461 lines
19 KiB
Python
461 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
kis_trader/engine/range_break_engine.py — 박스권 돌파(RANGE_BREAK) 공통 엔진
|
||
====================================================================
|
||
[SCAN] HTS momentum 조건(F·G·J) — 오늘 거래 활발 종목 풀
|
||
[TRIGGER] 횡보 박스 형성 → 거래량 폭발 양봉으로 박스 상단 돌파 시 진입
|
||
|
||
매도: 박스 실패(상단 재이탈) → 익절 → 어깨 → 손절 → 트레일 → EOD
|
||
(청산 본체는 breakout 과 동일 우선순위 — ``check_sell_signal_breakout_live`` 재사용)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from kis_trader.engine.range_break_env_keys import (
|
||
range_break_env_bool,
|
||
range_break_env_float,
|
||
range_break_env_int,
|
||
)
|
||
from kis_trader.strategies.breakout import (
|
||
check_sell_signal_breakout_live,
|
||
normalize_breakout_max_loss_krw,
|
||
)
|
||
from kis_trader.utils.trade_time import parse_trade_datetime as _t2dt
|
||
|
||
RANGE_BREAK_STRATEGY_ID = "RANGE_BREAK"
|
||
|
||
|
||
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 _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 _slot_key(candle_time: str, scan_interval_min: int = 1) -> str:
|
||
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 _day_running_high_low(candles: List[Dict], i: int, day: str) -> Tuple[float, float, float]:
|
||
running_high = float(candles[i]["high"])
|
||
running_low = float(candles[i]["low"])
|
||
day_open = float(candles[i]["open"])
|
||
for j in range(i, -1, -1):
|
||
if candles[j]["candle_time"][:8] != day:
|
||
break
|
||
running_high = max(running_high, float(candles[j]["high"]))
|
||
running_low = min(running_low, float(candles[j]["low"]))
|
||
day_open = float(candles[j]["open"])
|
||
return running_high, running_low, day_open
|
||
|
||
|
||
def _box_window(candles: List[Dict], i: int, box_lookback: int) -> List[Dict]:
|
||
"""신호봉 i 직전 box_lookback 개 확정봉 = 박스 구간."""
|
||
start = max(0, i - box_lookback)
|
||
return candles[start:i]
|
||
|
||
|
||
def eval_range_break_buy_at_index(
|
||
candles: List[Dict],
|
||
i: int,
|
||
params: Dict[str, Any],
|
||
day_open: Optional[float] = None,
|
||
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
|
||
"""박스권 돌파 매수 TRIGGER — 인덱스 i 신호봉."""
|
||
box_lookback = int(params.get("box_lookback_min", 30))
|
||
box_max_w = float(params.get("box_max_width_pct", 2.5))
|
||
box_min_w = float(params.get("box_min_width_pct", 0.3))
|
||
setup_vol_mult = float(params.get("setup_vol_max_mult", 0.8))
|
||
setup_bear_min = int(params.get("setup_bear_bars_min", 1))
|
||
vol_mult = float(params.get("vol_mult", 2.0))
|
||
vol_win = int(params.get("vol_window", 7))
|
||
break_margin = float(params.get("break_margin_pct", 0.0) or 0.0)
|
||
body_min = float(params.get("body_min_pct", 0.0) or 0.0)
|
||
time_start = int(params.get("time_start_hm", 1030))
|
||
time_end = int(params.get("time_end_hm", 1520))
|
||
min_price = float(params.get("min_price", 1000.0))
|
||
max_daily_chg = float(params.get("max_daily_chg", 25.0))
|
||
use_high_chase = _to_bool(params.get("use_high_chase_filter"), True)
|
||
high_chase_thr = float(params.get("high_chase_thr", 0.96))
|
||
vol_baseline_win = int(params.get("vol_baseline_win", 30))
|
||
|
||
need = max(box_lookback, vol_win, vol_baseline_win) + 3
|
||
if i < need or i >= len(candles):
|
||
return ("탈락-봉부족", "need=%d i=%d" % (need, i), None)
|
||
|
||
c = candles[i]
|
||
ct = str(c.get("candle_time", ""))
|
||
if len(ct) < 12:
|
||
return ("탈락-시간없음", ct, None)
|
||
hm = int(ct[8:12])
|
||
if hm < time_start or hm >= time_end:
|
||
return (None, None, None)
|
||
|
||
try:
|
||
op = float(c["open"])
|
||
hi = float(c["high"])
|
||
lo = float(c["low"])
|
||
cl = float(c["close"])
|
||
vol = float(c.get("volume", 0) or 0)
|
||
except Exception as e:
|
||
return ("탈락-캔들파싱", str(e), None)
|
||
|
||
if cl < min_price:
|
||
return ("탈락-최소가격", "%.0f" % cl, None)
|
||
|
||
box_arr = _box_window(candles, i, box_lookback)
|
||
if len(box_arr) < max(3, box_lookback // 2):
|
||
return ("탈락-박스창없음", "len=%d" % len(box_arr), None)
|
||
|
||
box_high = max(float(x["high"]) for x in box_arr)
|
||
box_low = min(float(x["low"]) for x in box_arr)
|
||
if box_low <= 0 or box_high <= box_low:
|
||
return ("탈락-박스무효", "high=%.0f low=%.0f" % (box_high, box_low), None)
|
||
|
||
box_width_pct = (box_high - box_low) / box_low * 100.0
|
||
if box_width_pct > box_max_w:
|
||
return ("탈락-박스넓음", "%.2f%% > %.2f%%" % (box_width_pct, box_max_w), None)
|
||
if box_width_pct < box_min_w:
|
||
return ("탈락-박스좁음", "%.2f%% < %.2f%%" % (box_width_pct, box_min_w), None)
|
||
|
||
box_vols = [float(x.get("volume", 0) or 0) for x in box_arr]
|
||
box_vol_avg = sum(box_vols) / len(box_vols) if box_vols else 0.0
|
||
base_start = max(0, i - box_lookback - vol_baseline_win)
|
||
base_vols = [
|
||
float(candles[k].get("volume", 0) or 0)
|
||
for k in range(base_start, i - box_lookback)
|
||
]
|
||
base_avg = sum(base_vols) / len(base_vols) if base_vols else box_vol_avg
|
||
if base_avg > 0 and box_vol_avg > base_avg * setup_vol_mult:
|
||
return (
|
||
"탈락-박스거래량과다",
|
||
"%.0f > %.0f×%.2f" % (box_vol_avg, base_avg, setup_vol_mult),
|
||
None,
|
||
)
|
||
|
||
bear_cnt = sum(
|
||
1 for x in box_arr
|
||
if float(x.get("close", 0) or 0) < float(x.get("open", 0) or 0)
|
||
)
|
||
if bear_cnt < setup_bear_min:
|
||
return ("탈락-음봉부족", "%d < %d" % (bear_cnt, setup_bear_min), None)
|
||
|
||
if cl <= op:
|
||
return ("탈락-음봉", "close<=open", None)
|
||
|
||
need_price = box_high * (1.0 + break_margin / 100.0)
|
||
if cl < need_price:
|
||
return (
|
||
"탈락-박스미돌파",
|
||
"close=%.0f < need=%.0f" % (cl, need_price),
|
||
None,
|
||
)
|
||
|
||
if body_min > 0 and op > 0:
|
||
body_pct = (cl - op) / op * 100.0
|
||
if body_pct < body_min:
|
||
return ("탈락-몸통부족", "%.2f%%" % body_pct, None)
|
||
|
||
vol_ref = box_vol_avg if box_vol_avg > 0 else 0.0
|
||
if vol_ref <= 0:
|
||
win_arr = [float(candles[k].get("volume", 0) or 0) for k in range(i - vol_win, i)]
|
||
vol_ref = sum(win_arr) / len(win_arr) if win_arr else 0.0
|
||
if vol_ref <= 0 or vol < vol_ref * vol_mult:
|
||
ratio = vol / vol_ref if vol_ref > 0 else 0.0
|
||
return ("탈락-거래량부족", "%.2fx < %.1fx" % (ratio, vol_mult), None)
|
||
|
||
day = ct[:8]
|
||
running_high, running_low, d_open = _day_running_high_low(candles, i, day)
|
||
if day_open is None:
|
||
day_open = d_open
|
||
if day_open > 0:
|
||
daily_chg = (cl - day_open) / day_open * 100.0
|
||
if daily_chg > max_daily_chg:
|
||
return ("탈락-이격과열", "%.1f%%" % daily_chg, None)
|
||
|
||
if use_high_chase and running_high > 0 and cl >= running_high * high_chase_thr:
|
||
return ("탈락-고점추격", "%.0f" % cl, None)
|
||
|
||
vol_ratio = vol / vol_ref if vol_ref > 0 else 0.0
|
||
return (None, None, {
|
||
"box_high": box_high,
|
||
"box_low": box_low,
|
||
"box_width_pct": round(box_width_pct, 3),
|
||
"vol_ratio": vol_ratio,
|
||
"bear_bars": bear_cnt,
|
||
"signal_candle_time": ct,
|
||
})
|
||
|
||
|
||
def check_buy_signal_range_break_live(
|
||
candles: List[Dict],
|
||
params: Dict[str, Any],
|
||
state: Dict[str, Any],
|
||
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
|
||
"""실매·백테 공용 — align: 신호봉 확정 후 다음 봉 시가 진입."""
|
||
live_align = _to_bool(params.get("live_backtest_align", True), True)
|
||
lookback = max(1, int(params.get("live_signal_lookback_bars", 1)))
|
||
confirmed = _confirmed_candles_only(candles)
|
||
if len(confirmed) < 10:
|
||
return ("탈락-봉부족", "확정봉 부족", None)
|
||
|
||
cooldown_min = float(params.get("cooldown_min", 30))
|
||
max_daily = int(params.get("max_daily", 1))
|
||
last_exit_dt = state.get("last_exit_dt")
|
||
if last_exit_dt is not None:
|
||
elapsed = (_t2dt(confirmed[-1]["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)
|
||
|
||
last_reject: Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]] = (
|
||
None, None, None,
|
||
)
|
||
if live_align:
|
||
entry_i = len(confirmed) - 1
|
||
for k in range(lookback):
|
||
signal_i = entry_i - 1 - k
|
||
if signal_i < 1:
|
||
break
|
||
reject, msg, sig = eval_range_break_buy_at_index(
|
||
confirmed, signal_i, params, None,
|
||
)
|
||
if reject:
|
||
if k == 0:
|
||
last_reject = (reject, msg, None)
|
||
continue
|
||
if sig:
|
||
ent = confirmed[entry_i]
|
||
entry_open = float(ent.get("open", 0) or 0)
|
||
if entry_open <= 0:
|
||
entry_open = float(ent.get("close", 0) or 0)
|
||
sig["entry_price"] = entry_open
|
||
sig["entry_bar_key"] = ent.get("candle_time")
|
||
sig["box_stop_line"] = float(sig.get("box_high", 0) or 0)
|
||
return (None, None, sig)
|
||
return last_reject
|
||
|
||
i = len(confirmed) - 1
|
||
return eval_range_break_buy_at_index(confirmed, i, params, None)
|
||
|
||
|
||
def check_sell_signal_range_break_live(
|
||
position: Dict[str, Any],
|
||
current_candle: Dict[str, Any],
|
||
params: Dict[str, Any],
|
||
is_eod: bool = False,
|
||
) -> Optional[Tuple[str, float]]:
|
||
"""박스 상단 재이탈 우선 → breakout 청산 체인."""
|
||
try:
|
||
lo = float(current_candle.get("low", current_candle["close"]))
|
||
cl = float(current_candle["close"])
|
||
except Exception:
|
||
return None
|
||
|
||
box_line = float(position.get("box_stop_line", 0) or 0)
|
||
if box_line > 0 and lo < box_line:
|
||
return ("box_fail", min(cl, box_line))
|
||
|
||
return check_sell_signal_breakout_live(position, current_candle, params, is_eod=is_eod)
|
||
|
||
|
||
def get_range_break_defaults_from_db(db=None, *, env_row: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""env_config + config_range_break 병합."""
|
||
own_db = None
|
||
r: Dict[str, Any] = {}
|
||
try:
|
||
if env_row is not None:
|
||
r = dict(env_row)
|
||
else:
|
||
if db is None:
|
||
from database import TradeDB
|
||
own_db = TradeDB()
|
||
db = own_db
|
||
if hasattr(db, "get_merged_env_snapshot"):
|
||
r = db.get_merged_env_snapshot()
|
||
elif hasattr(db, "get_latest_env"):
|
||
latest = db.get_latest_env()
|
||
r = dict((latest or {}).get("snapshot") or {})
|
||
else:
|
||
row = db.conn.execute(
|
||
"SELECT * FROM env_config ORDER BY id DESC LIMIT 1"
|
||
).fetchone()
|
||
r = dict(row) if row else {}
|
||
except Exception:
|
||
r = {}
|
||
finally:
|
||
if own_db is not None:
|
||
try:
|
||
own_db.close()
|
||
except Exception:
|
||
pass
|
||
|
||
sl = abs(range_break_env_float(r, "RANGE_BREAK_STOP_LOSS_PCT", 0.03))
|
||
tp = range_break_env_float(r, "RANGE_BREAK_TAKE_PROFIT_PCT", 0.10)
|
||
trail = range_break_env_float(r, "RANGE_BREAK_TRAIL_PCT", 0.015)
|
||
trail_arm = range_break_env_float(r, "RANGE_BREAK_TRAIL_ARM_PCT", 0.015)
|
||
shoulder_high = range_break_env_float(r, "RANGE_BREAK_SHOULDER_MIN_HIGH_PCT", 0.03)
|
||
shoulder_cut = range_break_env_float(r, "RANGE_BREAK_SHOULDER_CUT_PCT", 0.005)
|
||
cooldown_sec = range_break_env_int(r, "RANGE_BREAK_COOLDOWN_SEC", 1800)
|
||
|
||
return {
|
||
"box_lookback_min": range_break_env_int(r, "RANGE_BREAK_BOX_LOOKBACK_MIN", 30),
|
||
"box_max_width_pct": range_break_env_float(r, "RANGE_BREAK_BOX_MAX_WIDTH_PCT", 2.5),
|
||
"box_min_width_pct": range_break_env_float(r, "RANGE_BREAK_BOX_MIN_WIDTH_PCT", 0.3),
|
||
"setup_vol_max_mult": range_break_env_float(r, "RANGE_BREAK_SETUP_VOL_MAX_MULT", 0.8),
|
||
"setup_bear_bars_min": range_break_env_int(r, "RANGE_BREAK_SETUP_BEAR_BARS_MIN", 1),
|
||
"vol_mult": range_break_env_float(r, "RANGE_BREAK_VOL_MULT", 2.0),
|
||
"vol_window": range_break_env_int(r, "RANGE_BREAK_VOL_WIN", 7),
|
||
"break_margin_pct": range_break_env_float(r, "RANGE_BREAK_BREAK_MARGIN_PCT", 0.0),
|
||
"body_min_pct": range_break_env_float(r, "RANGE_BREAK_BODY_MIN_PCT", 0.0),
|
||
"time_start_hm": range_break_env_int(r, "RANGE_BREAK_TIME_START", 1030),
|
||
"time_end_hm": range_break_env_int(r, "RANGE_BREAK_TIME_END_HM", 1520),
|
||
"sl_pct": sl,
|
||
"tp_pct": tp,
|
||
"trail_pct": trail,
|
||
"trail_arm_pct": trail_arm,
|
||
"shoulder_min_high": shoulder_high,
|
||
"shoulder_cut_pct": shoulder_cut,
|
||
"max_hold_bars": range_break_env_int(r, "RANGE_BREAK_MAX_HOLD_BARS", 0),
|
||
"max_daily": range_break_env_int(r, "RANGE_BREAK_MAX_DAILY", 1),
|
||
"cooldown_min": cooldown_sec / 60.0,
|
||
"max_daily_chg": range_break_env_float(r, "RANGE_BREAK_MAX_DAILY_CHG", 25.0),
|
||
"min_price": range_break_env_float(r, "RANGE_BREAK_MIN_PRICE", 1000.0),
|
||
"high_chase_thr": range_break_env_float(r, "RANGE_BREAK_HIGH_CHASE_THR", 0.96),
|
||
"use_high_chase_filter": range_break_env_bool(r, "RANGE_BREAK_USE_HIGH_CHASE_FILTER", True),
|
||
"max_loss_krw": float(range_break_env_int(r, "RANGE_BREAK_MAX_LOSS_PER_TRADE_KRW", 200_000)),
|
||
"slot_money": float(range_break_env_int(r, "RANGE_BREAK_SLOT_MONEY", 200_000)),
|
||
"max_stocks": range_break_env_int(r, "RANGE_BREAK_MAX_STOCKS", 20),
|
||
"total_budget_krw": range_break_env_int(r, "RANGE_BREAK_TOTAL_BUDGET_KRW", 4_000_000),
|
||
"live_backtest_align": range_break_env_bool(r, "RANGE_BREAK_LIVE_BACKTEST_ALIGN", True),
|
||
"live_signal_lookback_bars": range_break_env_int(r, "RANGE_BREAK_LIVE_SIGNAL_LOOKBACK_BARS", 1),
|
||
"force_eod_exit": range_break_env_bool(r, "RANGE_BREAK_FORCE_EOD_EXIT", False),
|
||
"vol_baseline_win": range_break_env_int(r, "RANGE_BREAK_VOL_BASELINE_WIN", 30),
|
||
}
|
||
|
||
|
||
def range_break_ui_to_engine_params(ui: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""웹 UI(%) → 엔진 params."""
|
||
sl_ui = float(ui.get("sl_pct", 3.0))
|
||
max_loss = normalize_breakout_max_loss_krw(ui.get("max_loss_krw", 200_000))
|
||
slot_cap = float(ui.get("slot_money", 200_000))
|
||
from kis_trader.strategies.breakout import breakout_invest_amount_krw
|
||
slot = breakout_invest_amount_krw(max_loss, sl_ui, slot_cap)
|
||
|
||
def _ui_bool(key: str, default: bool = False) -> bool:
|
||
raw = ui.get(key)
|
||
if raw is None or raw == "":
|
||
return default
|
||
if isinstance(raw, bool):
|
||
return raw
|
||
return str(raw).strip().lower() in ("1", "true", "t", "y", "yes", "on")
|
||
|
||
return {
|
||
"box_lookback_min": int(ui.get("box_lookback_min", 30)),
|
||
"box_max_width_pct": float(ui.get("box_max_width_pct", 2.5)),
|
||
"box_min_width_pct": float(ui.get("box_min_width_pct", 0.3)),
|
||
"setup_vol_max_mult": float(ui.get("setup_vol_max_mult", 0.8)),
|
||
"setup_bear_bars_min": int(ui.get("setup_bear_bars_min", 1)),
|
||
"vol_mult": float(ui.get("vol_mult", 2.0)),
|
||
"vol_window": int(ui.get("vol_window", 7)),
|
||
"vol_baseline_win": int(ui.get("vol_baseline_win", 30)),
|
||
"break_margin_pct": float(ui.get("break_margin_pct", 0.0) or 0.0),
|
||
"body_min_pct": float(ui.get("body_min_pct", 0.0) or 0.0),
|
||
"time_start_hm": int(ui.get("time_start_hm", 1030)),
|
||
"time_end_hm": int(ui.get("time_end_hm", 1520)),
|
||
"cooldown_min": float(ui.get("cooldown_min", 30)),
|
||
"max_daily": int(ui.get("max_daily", 1)),
|
||
"max_daily_chg": float(ui.get("max_daily_chg", 25.0)),
|
||
"min_price": float(ui.get("min_price", 1000.0)),
|
||
"high_chase_thr": float(ui.get("high_chase_thr", 0.96)),
|
||
"use_high_chase_filter": _ui_bool("use_high_chase_filter", True),
|
||
"max_loss_krw": max_loss,
|
||
"slot_money": slot,
|
||
"stop_loss_pct": -abs(sl_ui) / 100.0,
|
||
"take_profit_pct": abs(float(ui.get("tp_pct", 10.0))) / 100.0,
|
||
"trail_pct": abs(float(ui.get("trail_pct", 1.5))) / 100.0,
|
||
"trail_arm_pct": abs(float(ui.get("trail_arm_pct", 1.5) or 0.0)) / 100.0,
|
||
"shoulder_min_high": abs(float(ui.get("shoulder_min_high_pct", 3.0))) / 100.0,
|
||
"shoulder_cut_pct": abs(float(ui.get("shoulder_cut_pct", 0.5))) / 100.0,
|
||
"max_hold_bars": int(float(ui.get("max_hold_bars", 0) or 0)),
|
||
"fee_rate": float(ui.get("fee_rate_pct", 0.015) or 0.015) / 100.0,
|
||
"sell_tax": float(ui.get("sell_tax_pct", 0.18) or 0.18) / 100.0,
|
||
"live_backtest_align": _ui_bool("live_backtest_align", True),
|
||
"live_signal_lookback_bars": int(ui.get("live_signal_lookback_bars", 1)),
|
||
"force_eod_exit": _ui_bool("force_eod_exit", False),
|
||
"portfolio_mode": True,
|
||
}
|
||
|
||
|
||
def range_break_min_bars_required(params: Dict[str, Any]) -> int:
|
||
"""백테 최소 봉 수 — 박스 lookback + 거래량 기준창."""
|
||
box_lb = int(params.get("box_lookback_min", 30))
|
||
vol_win = int(params.get("vol_window", 7))
|
||
vol_base = int(params.get("vol_baseline_win", 30))
|
||
align_extra = 1 if _to_bool(params.get("live_backtest_align", True), True) else 0
|
||
return max(box_lb, vol_win, vol_base) + 3 + align_extra
|
||
|
||
|
||
def range_break_scan_buy_at_bar(
|
||
candles: List[Dict],
|
||
bar_index: int,
|
||
params: Dict[str, Any],
|
||
state: Optional[Dict[str, Any]] = None,
|
||
day_open: Optional[float] = None,
|
||
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]], float, str]:
|
||
"""
|
||
백테 단일 봉 매수 스캔 — align: 신호봉 확정 후 다음 봉 시가 진입.
|
||
|
||
Returns:
|
||
reason, msg, signal, entry_price, entry_time(candle_time)
|
||
"""
|
||
i = bar_index
|
||
st = state or {}
|
||
sub = candles[: i + 1]
|
||
reason, msg, sig = check_buy_signal_range_break_live(sub, params, st)
|
||
if not sig:
|
||
return reason, msg, None, 0.0, ""
|
||
if i + 1 >= len(candles):
|
||
return ("탈락-진입봉없음", "다음 봉 없음", None, 0.0, "")
|
||
next_c = candles[i + 1]
|
||
if next_c["candle_time"][:8] != sub[-1]["candle_time"][:8]:
|
||
return ("탈락-일자넘김", "다음봉 일자 불일치", None, 0.0, "")
|
||
ep = float(next_c.get("open") or 0)
|
||
if ep <= 0:
|
||
ep = float(sig.get("entry_price") or sig.get("close") or 0)
|
||
sig["box_stop_line"] = float(sig.get("box_high", 0) or 0)
|
||
return None, None, sig, ep, str(next_c.get("candle_time") or "")
|
||
|
||
|
||
def run_range_break_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]:
|
||
from kis_trader.backtest.range_break_portfolio_backtest import (
|
||
run_range_break_backtest_portfolio,
|
||
)
|
||
return run_range_break_backtest_portfolio(
|
||
codes_candles, params, universe_by_slot=universe_by_slot,
|
||
ticks_by_code=ticks_by_code,
|
||
)
|