hoon/kis_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
kis_bot/kis_trader/engine/momentum_engine.py
Your Name e1ac8d119b chore: 작업 중 발생한 부수적 변경 사항 및 누락된 파일 전체 커밋 
- 프론트엔드 UI 업데이트 (backtest.html, backtest.js) 엔진 라디오 버튼 통합 관련 반영
- Rust 플러그인(kis_rust_core) 및 컴파일 소스코드 추가
- CLI 백테스트 스크립트 수정 및 최신화
- 기타 스크래치 테스트 스크립트, 로그 요약 마크다운(.md) 등 누락 파일 일괄 반영
- 추가적으로 아직 발견되지 않은 엣지 케이스나 렌더링 오류가 포함되어 있을 가능성이 있음
2026-09-06 17:04:50 +09:00

850 lines
35 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
kis_trader/engine/momentum_engine.py — 모멘텀(HTS momentum 조건식) 전용 엔진
============================================================================
스캘핑 reversal(SCALP)과 완전 분리. 백테스트·파라서치·MomentumStrategy 공통.
[전략 컨셉 — SCAN vs TRIGGER vs 청산]
- SCAN: 키움 ``momentum`` 조건검색 (E∧F∧H∧I) → target_candidates_history
- TRIGGER: ``momentum_hts_logic`` — 전일시가 돌파 유지(E) + 양봉·거래량 펄스
※ 구 1분 돌파/눌림재돌파·RSI50·EMA 패턴 TRIGGER 는 폐기
- 청산: ``momentum_hts_logic`` — 래칫·어깨·트레일·손절·시간컷
"""
from __future__ import annotations
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, Tuple
from kis_trader.engine.momentum_hts_logic import (
check_sell_signal_momentum_hts_live,
eval_momentum_hts_buy_at_index,
hts_trigger_defaults_from_row,
resolve_momentum_skip_hts_scan_dupes,
)
from kis_trader.engine.momentum_env_keys import (
momentum_env_bool,
momentum_env_float,
momentum_env_int,
_legacy_float,
)
from kis_trader.engine.strategy_eod import is_strategy_eod_bar
from kis_trader.utils.env import get_env_from_db, get_env_int
from kis_trader.utils.trade_time import parse_trade_datetime as _t2dt
MOMENTUM_STRATEGY_ID = "MOMENTUM"
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 _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 compute_rsi_series(closes: list, period: int = 3) -> list:
"""RSI 시리즈 (Wilder)."""
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 resolve_effective_tp_pct(tp_pct: float, tp_max_pct: float) -> float:
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:
return resolve_effective_tp_pct(
params.get("tp_pct", 0.015),
params.get("tp_max_pct", 0.02),
)
import os
try:
import kis_rust_core
except ImportError:
kis_rust_core = None
def run_momentum_backtest_rust_experimental(
codes_candles: Dict[str, List[Dict]],
params: Dict[str, Any]
) -> List[Dict]:
"""모멘텀 매매 Rust 엔진 실험용 브릿지"""
if kis_rust_core is None:
raise RuntimeError("kis_rust_core module is not available")
def get_float(k, default=0.0):
try: return float(params.get(k, default))
except: return float(default)
def get_int(k, default=0):
try: return int(params.get(k, default))
except: return int(default)
def get_bool(k, default=False):
v = params.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
rust_params = kis_rust_core.MomentumParams(
get_int("time_start_hm", 900),
get_int("time_end_hm", 1530),
get_float("cooldown_min", 10.0),
get_int("max_daily", 5),
get_float("min_price", 1000.0),
get_bool("use_defense_filters", True),
get_bool("use_high_chase_filter", False),
get_float("high_chase_thr", 0.96),
get_bool("trigger_e_confirm", True),
get_bool("trigger_require_bull_bar", True),
get_bool("use_vol_trigger", True),
get_bool("use_rsi_filter", False),
get_bool("use_rsi_max_filter", False),
get_int("rsi_period", 3),
get_float("mom_rsi_min", 50.0),
get_float("mom_rsi_max", 80.0),
get_bool("use_ema_filter", False),
get_int("ema_fast_period", 9),
get_int("ema_slow_period", 21),
get_float("e_min_chg_pct", 0.0),
get_float("mom_vol_mult", 1.2),
get_int("mom_vol_win", 5),
get_bool("skip_hts_scan_dupes", False),
get_float("sl_pct", 0.03),
get_float("tp_pct", 0.08),
get_float("trail_pct", 0.02),
get_float("trail_arm_pct", 0.01),
get_float("shoulder_min_high", 0.005),
get_float("shoulder_cut_pct", 0.003),
get_int("max_hold_bars", 0),
get_float("max_loss_krw", 200000.0),
get_float("min_hold_sec", 30.0),
get_float("min_drop_pct_for_loss_cut", 0.015),
)
all_trades = []
for code, candles_dict in codes_candles.items():
rust_candles = []
for c in candles_dict:
rc = kis_rust_core.CandleData(
c.get("candle_time", ""),
float(c.get("open", 0)),
float(c.get("high", 0)),
float(c.get("low", 0)),
float(c.get("close", 0)),
float(c.get("volume", 0)),
float(c.get("rsi", 0.0)),
)
rust_candles.append(rc)
trades = kis_rust_core.run_momentum_backtest_fast(code, rust_candles, rust_params)
for t in trades:
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,
"profit_rate": t.pnl_pct,
"qty": 1,
"pnl": 0,
"sell_reason": t.reason,
"max_price": t.max_price,
"rsi_entry": t.rsi_entry,
"strategy": "MOMENTUM"
})
return all_trades
def get_momentum_defaults_from_db(db=None, *, env_row: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""env_config + config_momentum 병합 → 엔진 params dict."""
own_db = None
r: Dict[str, Any] = {}
try:
if env_row is not None:
r = dict(env_row)
else:
# SCALP 와 동일 — RAM TTL. 매 루프 full snapshot 금지(장중 DB 락 팽창).
try:
from kis_trader.utils.env import get_strategy_env_dict
r = get_strategy_env_dict("MOMENTUM") or {}
except Exception:
r = {}
if not r:
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 {}
rsi_period = momentum_env_int(r, "MOMENTUM_RSI_PERIOD", 3)
mom_rsi_min = momentum_env_float(r, "MOMENTUM_RSI_MIN", 50.0)
mom_rsi_max = momentum_env_float(r, "MOMENTUM_RSI_MAX", 80.0)
mom_vol_mult = momentum_env_float(r, "MOMENTUM_VOL_MULT", 1.05)
mom_vol_win = momentum_env_int(r, "MOMENTUM_VOL_WIN", 5)
mom_time_end = momentum_env_int(r, "MOMENTUM_TIME_END_HM", 1530)
# HTS K: 전일 종가 대비 최소 등락률(%). 0=단순 종가 초과만. HTS 기본 0.2.
e_min_chg_pct = momentum_env_float(r, "MOMENTUM_E_MIN_CHG_PCT", 0.2)
sl_pct = abs(_legacy_float(
r, "MOMENTUM_STOP_LOSS_PCT", ("SCALP_STOP_LOSS_PCT",), 0.03,
))
tp_pct = _legacy_float(r, "MOMENTUM_TAKE_PROFIT_PCT", ("SCALP_TAKE_PROFIT_PCT",), 0.08)
tp_max = _legacy_float(r, "MOMENTUM_TP_MAX_PCT", ("SCALP_TP_MAX_PCT",), 0.10)
shoulder_high = _legacy_float(
r, "MOMENTUM_SHOULDER_MIN_HIGH_PCT",
("SCALP_SHOULDER_MIN_HIGH_PCT", "SHOULDER_MIN_HIGH_PCT"), 0.005,
)
shoulder_cut = _legacy_float(
r, "MOMENTUM_SHOULDER_CUT_PCT",
("SCALP_SHOULDER_CUT_PCT", "SHOULDER_CUT_PCT"), 0.003,
)
ratchet_tiers = str(
r.get("MOMENTUM_RATCHET_TIERS")
or get_env_from_db("MOMENTUM_RATCHET_TIERS", "")
or ""
).strip()
trail_pct = abs(momentum_env_float(r, "MOMENTUM_TRAIL_PCT", 0.02))
trail_arm_pct = abs(momentum_env_float(r, "MOMENTUM_TRAIL_ARM_PCT", 0.01))
max_hold_bars = momentum_env_int(r, "MOMENTUM_MAX_HOLD_BARS", 0)
max_daily = momentum_env_int(r, "MOMENTUM_MAX_DAILY", 5)
min_price = momentum_env_float(r, "MOMENTUM_MIN_PRICE", 1000.0)
max_daily_chg = momentum_env_float(r, "MOMENTUM_MAX_DAILY_CHG", 20.0)
high_chase = momentum_env_float(r, "MOMENTUM_HIGH_CHASE_THR", 0.96)
mom_max_open = momentum_env_float(r, "MOMENTUM_MAX_FROM_OPEN_PCT", 999.0)
mom_min_open = momentum_env_float(r, "MOMENTUM_MIN_FROM_OPEN_PCT", -999.0)
max_loss_krw = momentum_env_int(r, "MOMENTUM_MAX_LOSS_PER_TRADE_KRW", 200_000)
min_drop_loss = r.get("MOMENTUM_MIN_DROP_PCT_FOR_LOSS_CUT") or r.get("SCALP_MIN_DROP_PCT_FOR_LOSS_CUT")
min_drop_pct_for_loss_cut = 0.015
if min_drop_loss not in (None, "", "None"):
v = float(min_drop_loss)
min_drop_pct_for_loss_cut = v / 100.0 if v >= 1 else v
cooldown_sec = momentum_env_int(r, "MOMENTUM_COOLDOWN_SEC", 600)
time_start = momentum_env_int(r, "MOMENTUM_TIME_START", 830)
time_end = momentum_env_int(r, "MOMENTUM_TIME_END", mom_time_end)
skip_hts = resolve_momentum_skip_hts_scan_dupes(r)
use_defense = momentum_env_bool(r, "MOMENTUM_USE_DEFENSE_FILTERS", True)
use_high_chase_f = momentum_env_bool(r, "MOMENTUM_USE_HIGH_CHASE_FILTER", False)
use_daily_range_f = momentum_env_bool(r, "MOMENTUM_USE_DAILY_RANGE_FILTER", False)
use_rsi_filter = momentum_env_bool(r, "MOMENTUM_USE_RSI_FILTER", False)
# V4 TRIGGER — 웹백테/CLI가 이 dict 만 보면 DB와 동일 (누락 시 HTML·코드 기본값으로 어긋남)
use_ema_filter = momentum_env_bool(r, "MOMENTUM_USE_EMA_FILTER", False)
use_rsi_max_filter = momentum_env_bool(r, "MOMENTUM_USE_RSI_MAX_FILTER", False)
ema_fast_period = momentum_env_int(r, "MOMENTUM_EMA_FAST_PERIOD", 9)
ema_slow_period = momentum_env_int(r, "MOMENTUM_EMA_SLOW_PERIOD", 21)
pattern_breakout = momentum_env_bool(r, "MOMENTUM_PATTERN_BREAKOUT", True)
pattern_pullback = momentum_env_bool(r, "MOMENTUM_PATTERN_PULLBACK", True)
chase_lookback_min = momentum_env_int(r, "MOMENTUM_CHASE_LOOKBACK_MIN", 10)
pullback_lookback_min = momentum_env_int(r, "MOMENTUM_PULLBACK_LOOKBACK_MIN", 15)
pullback_min_pct = momentum_env_float(r, "MOMENTUM_PULLBACK_MIN_PCT", 0.3)
pullback_max_pct = momentum_env_float(r, "MOMENTUM_PULLBACK_MAX_PCT", 3.0)
setup_vol_max_mult = momentum_env_float(r, "MOMENTUM_SETUP_VOL_MAX_MULT", 0.8)
setup_bear_bars_min = momentum_env_int(r, "MOMENTUM_SETUP_BEAR_BARS_MIN", 1)
hts_flags = hts_trigger_defaults_from_row(r)
slot_money = momentum_env_int(r, "MOMENTUM_SLOT_MONEY", 3_000_000)
max_stocks = momentum_env_int(r, "MOMENTUM_MAX_STOCKS", 3)
total_budget = momentum_env_int(r, "MOMENTUM_TOTAL_BUDGET_KRW", 0)
min_hold_sec = momentum_env_float(r, "MOMENTUM_MIN_HOLD_SEC", 30.0)
live_align = momentum_env_bool(r, "MOMENTUM_LIVE_BACKTEST_ALIGN", True)
lookback_bars = momentum_env_int(r, "MOMENTUM_LIVE_SIGNAL_LOOKBACK_BARS", 1)
# ALIGN 시 형성 중 봉(T)을 진입봉으로 — BT portfolio 와 동일 (확정봉만 쓰면 1봉 지연)
use_forming = momentum_env_bool(r, "MOMENTUM_LIVE_ALIGN_USE_FORMING_BAR", True)
# last_exit 분 floor — 신호봉 candle_time 과 시계 정합 (wall-clock 초 단위면 쿨다운 과다)
cd_floor = momentum_env_bool(r, "MOMENTUM_COOLDOWN_USE_CANDLE_FLOOR", True)
cd_engine_only = momentum_env_bool(r, "MOMENTUM_COOLDOWN_ENGINE_ONLY", True)
skip_pre_sub = momentum_env_bool(r, "MOMENTUM_BACKTEST_SKIP_PRE_SUBSCRIBE", False)
force_eod = momentum_env_bool(r, "MOMENTUM_FORCE_EOD_EXIT", False)
eod_enabled = momentum_env_bool(r, "MOMENTUM_EOD_ENABLED", True)
if r.get("MOMENTUM_EOD_ENABLED") in (None, "", "None") and r.get("MOMENTUM_FORCE_EOD_EXIT") not in (None, "", "None"):
eod_enabled = force_eod
eod_hm = str(r.get("MOMENTUM_EOD_HM") or "15:20").strip() or "15:20"
portfolio_mode = True
# 수익구간 호가매도 (기본 OFF)
exit_ob_enabled = momentum_env_bool(r, "MOMENTUM_EXIT_OB_ENABLED", False)
exit_ob_ratio_min = momentum_env_float(r, "MOMENTUM_EXIT_OB_RATIO_MIN", 0.4)
exit_ob_ma_window = momentum_env_int(r, "MOMENTUM_EXIT_OB_MA_WINDOW", 5)
exit_ob_min_profit_pct = momentum_env_float(r, "MOMENTUM_EXIT_OB_MIN_PROFIT_PCT", 0.005)
exit_ob_min_hold_bars = momentum_env_int(r, "MOMENTUM_EXIT_OB_MIN_HOLD_BARS", 3)
# 손절호가 (기본 OFF)
stop_ob_enabled = momentum_env_bool(r, "MOMENTUM_STOP_OB_ENABLED", False)
stop_ob_ratio_min = momentum_env_float(r, "MOMENTUM_STOP_OB_RATIO_MIN", 0.4)
stop_ob_ma_window = momentum_env_int(r, "MOMENTUM_STOP_OB_MA_WINDOW", 5)
stop_ob_min_loss_pct = momentum_env_float(r, "MOMENTUM_STOP_OB_MIN_LOSS_PCT", 0.003)
stop_ob_min_hold_bars = momentum_env_int(r, "MOMENTUM_STOP_OB_MIN_HOLD_BARS", 2)
except Exception:
rsi_period, mom_rsi_min, mom_rsi_max = 3, 50.0, 80.0
mom_vol_mult, mom_vol_win, mom_time_end = 1.2, 5, 1530
e_min_chg_pct = 0.2
sl_pct, tp_pct, tp_max = 0.03, 0.08, 0.10
shoulder_high, shoulder_cut = 0.005, 0.003
ratchet_tiers, trail_pct, trail_arm_pct = "", 0.02, 0.01
max_hold_bars, max_daily = 0, 5
min_price, max_daily_chg, high_chase = 1000.0, 20.0, 0.96
mom_max_open, mom_min_open = 999.0, -999.0
max_loss_krw, min_drop_pct_for_loss_cut = 200_000, 0.015
cooldown_sec, time_start, time_end = 600, 830, 1530
skip_hts, use_defense = True, True
use_high_chase_f, use_daily_range_f = False, False
use_rsi_filter = False
use_ema_filter, use_rsi_max_filter = False, False
ema_fast_period, ema_slow_period = 9, 21
pattern_breakout, pattern_pullback = True, True
chase_lookback_min, pullback_lookback_min = 10, 15
pullback_min_pct, pullback_max_pct = 0.3, 3.0
setup_vol_max_mult, setup_bear_bars_min = 0.8, 1
hts_flags = hts_trigger_defaults_from_row({})
slot_money, max_stocks, total_budget = 3_000_000, 3, 0
min_hold_sec, live_align, lookback_bars = 30.0, True, 1
use_forming, cd_floor, cd_engine_only = True, True, True
skip_pre_sub = False
force_eod, portfolio_mode = False, True
eod_enabled, eod_hm = True, "15:20"
exit_ob_enabled, exit_ob_ratio_min, exit_ob_ma_window = False, 0.4, 5
exit_ob_min_profit_pct, exit_ob_min_hold_bars = 0.005, 3
stop_ob_enabled, stop_ob_ratio_min, stop_ob_ma_window = False, 0.4, 5
stop_ob_min_loss_pct, stop_ob_min_hold_bars = 0.003, 2
finally:
if own_db is not None:
try:
own_db.close()
except Exception:
pass
return {
"rsi_period": rsi_period,
"mom_rsi_min": mom_rsi_min,
"mom_rsi_max": mom_rsi_max,
"mom_vol_mult": mom_vol_mult,
"mom_vol_win": mom_vol_win,
"mom_time_end_hm": mom_time_end,
"e_min_chg_pct": e_min_chg_pct,
"mom_max_from_open_pct": mom_max_open,
"mom_min_from_open_pct": mom_min_open,
"sl_pct": sl_pct,
"tp_pct": tp_pct,
"tp_max_pct": tp_max,
"shoulder_min_high": shoulder_high,
"shoulder_cut_pct": shoulder_cut,
"ratchet_tiers": ratchet_tiers,
"trail_pct": trail_pct,
"trail_arm_pct": trail_arm_pct,
"max_hold_bars": max_hold_bars,
"max_daily": max_daily,
"min_price": min_price,
"max_daily_chg": max_daily_chg,
"high_chase_thr": high_chase,
"max_loss_krw": float(max_loss_krw),
"min_drop_pct_for_loss_cut": min_drop_pct_for_loss_cut,
"cooldown_min": cooldown_sec / 60.0,
"time_start_hm": time_start,
"time_end_hm": time_end,
"skip_hts_scan_dupes": skip_hts,
"use_defense_filters": use_defense,
"use_high_chase_filter": use_high_chase_f,
"use_daily_range_filter": use_daily_range_f,
"use_rsi_filter": use_rsi_filter,
"use_ema_filter": use_ema_filter,
"use_rsi_max_filter": use_rsi_max_filter,
"ema_fast_period": ema_fast_period,
"ema_slow_period": ema_slow_period,
"pattern_breakout": pattern_breakout,
"pattern_pullback": pattern_pullback,
"chase_lookback_min": chase_lookback_min,
"pullback_lookback_min": pullback_lookback_min,
"pullback_min_pct": pullback_min_pct,
"pullback_max_pct": pullback_max_pct,
"setup_vol_max_mult": setup_vol_max_mult,
"setup_bear_bars_min": setup_bear_bars_min,
**hts_flags,
"slot_money": float(slot_money),
"max_stocks": max_stocks,
"total_budget_krw": float(total_budget),
"min_hold_sec": min_hold_sec,
"live_backtest_align": live_align,
"live_signal_lookback_bars": lookback_bars,
"live_align_use_forming_bar": use_forming,
"cooldown_use_candle_floor": cd_floor,
"cooldown_engine_only": cd_engine_only,
"backtest_skip_pre_subscribe": skip_pre_sub,
"eod_enabled": eod_enabled,
"eod_hm": eod_hm,
"force_eod_exit": eod_enabled,
"portfolio_mode": portfolio_mode,
"scan_interval_min": 1,
# 수익구간 호가매도
"exit_ob_enabled": exit_ob_enabled,
"exit_ob_ratio_min": exit_ob_ratio_min,
"exit_ob_ma_window": exit_ob_ma_window,
"exit_ob_min_profit_pct": exit_ob_min_profit_pct,
"exit_ob_min_hold_bars": exit_ob_min_hold_bars,
# 손절호가
"stop_ob_enabled": stop_ob_enabled,
"stop_ob_ratio_min": stop_ob_ratio_min,
"stop_ob_ma_window": stop_ob_ma_window,
"stop_ob_min_loss_pct": stop_ob_min_loss_pct,
"stop_ob_min_hold_bars": stop_ob_min_hold_bars,
}
def eval_momentum_buy_at_index(
candles: List[Dict],
i: int,
params: Dict[str, Any],
state: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
"""HTS momentum 조건식 정합 TRIGGER — ``momentum_hts_logic`` 위임."""
return eval_momentum_hts_buy_at_index(
candles, i, params, state,
compute_rsi_series_fn=compute_rsi_series,
)
def _confirmed_candles_only(candles: List[Dict]) -> List[Dict]:
out: List[Dict] = []
for c in candles:
if c.get("is_confirmed") in (0, False, "0", "false"):
continue
out.append(c)
if len(out) < len(candles) * 0.5:
return list(candles)
return out
def check_buy_signal_momentum_live(
candles: List[Dict],
params: Dict[str, Any],
state: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
"""실매·백테 공용 모멘텀 진입.
live_backtest_align=True:
- 신호봉 = 직전 확정봉 (T-1)
- 진입봉 = 형성 중 봉 T (``live_align_use_forming_bar`` 기본 True) 또는
forming 없으면 마지막 확정봉 (구버전 폴백)
"""
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) < 6:
return ("탈락-봉부족", "확정봉 6개 미만", None)
last_reject: Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]] = (
None, None, None,
)
if live_align:
# forming 있으면: 신호=confirmed[-1], 진입=forming open (BT idx=T 와 동일)
# 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
reject, msg, sig = eval_momentum_buy_at_index(
confirmed, signal_i, params, state,
)
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
i = len(confirmed) - 1
return eval_momentum_buy_at_index(confirmed, i, params, state)
def check_sell_signal_momentum_live(
position: Dict[str, Any],
current_candle: Dict[str, Any],
params: Dict[str, Any],
is_eod: bool = False,
) -> Optional[Tuple[str, float]]:
"""HTS momentum 추세추격 청산 — ``momentum_hts_logic`` 위임 (어깨컷 폐기)."""
return check_sell_signal_momentum_hts_live(
position, current_candle, params, is_eod=is_eod,
)
def _intrabar_exit_prices(
open_: float, high: float, low: float, close: float, n_checks: int,
) -> List[float]:
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]
seg_count = len(anchors) - 1
extras = n - len(anchors)
per_seg = extras // seg_count
rem = extras % seg_count
out = [anchors[0]]
for si in range(seg_count):
a, b = anchors[si], anchors[si + 1]
extras_here = per_seg + (1 if si < rem else 0)
for j in range(1, extras_here + 1):
t = j / (extras_here + 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_momentum_backtest_bar(
position: Dict[str, Any],
candle: Dict[str, Any],
params: Dict[str, Any],
is_eod: bool = False,
) -> Optional[Tuple[str, float]]:
"""백테: 1분 OHLC를 N회 가격 체크로 분할 → ``check_sell_signal_momentum_live`` 호출.
실매 MomentumStrategy 와 동일 청산 함수·우선순위. intrabar 순서만 분봉 OHLC로 근사.
"""
n_checks = get_env_int("BACKTEST_EXIT_CHECKS_PER_BAR", 6)
if n_checks <= 1:
return check_sell_signal_momentum_live(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[str, float]] = None
for idx, px in enumerate(prices):
mp = float(position.get("max_price", 0) or 0)
if px > mp:
position["max_price"] = px
mp = px
sim = {"open": o, "high": mp, "low": px, "close": px, "candle_time": ct}
eod_here = bool(is_eod and idx == len(prices) - 1)
result = check_sell_signal_momentum_live(position, sim, params, is_eod=eod_here)
if result:
return result
return result
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]]:
"""모멘텀 SCAN 유니버스 시뮬레이션 (HTS scalp 조건검색 근사)."""
slot_codes_scores: Dict[str, List[Tuple[str, float, float]]] = {}
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]
by_day.setdefault(day, []).append(c)
for day, day_candles in by_day.items():
day_candles.sort(key=lambda x: x["candle_time"])
market_open_min = 9 * 60
seen_slots = set()
for c in 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)
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 len(up_to) < 2:
continue
o = float(up_to[0]["open"])
hi = max(float(x["high"]) for x in up_to)
cl = float(up_to[-1]["close"])
if o <= 0 or hi <= 0 or cl <= 0:
continue
from_open_pct = (cl / o - 1) * 100
if from_open_pct <= 0.3:
momentum_score = 0.0
elif from_open_pct > 20.0:
momentum_score = 0.0
else:
momentum_score = from_open_pct * 1.5
chase_ratio = cl / hi
chase_score = max(0.0, (chase_ratio - 0.85) * 50)
up_count = sum(1 for x in up_to if float(x["close"]) > float(x["open"]))
uptrend_score = (up_count / len(up_to)) * 5.0
closes = [float(x["close"]) for x in up_to]
rsis = compute_rsi_series(closes, 3)
last_rsi = rsis[-1] if rsis and rsis[-1] is not None else 0.0
if 50 <= last_rsi <= 70:
rsi_score = (last_rsi - 50) / 4.0
elif 70 < last_rsi <= 80:
rsi_score = 5.0 - (last_rsi - 70) / 2.0
else:
rsi_score = 0.0
total_score = momentum_score + chase_score + uptrend_score + rsi_score
vol_sum = sum(float(x.get("volume", 0)) for x in up_to)
slot_codes_scores.setdefault(slot_key, []).append((code, total_score, vol_sum))
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
def run_momentum_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,
orderbook_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
program_by_code: Optional[Dict[str, Dict[str, List[Any]]]] = None,
) -> List[Dict]:
"""모멘텀 백테스트 — portfolio_mode 시 포트폴리오 모듈 위임."""
if params.get("use_rust", False):
return run_momentum_backtest_rust_experimental(codes_candles, params)
if _to_bool(params.get("portfolio_mode"), True):
from kis_trader.backtest.momentum_portfolio_backtest import run_momentum_backtest_portfolio
return run_momentum_backtest_portfolio(
codes_candles, params, universe_by_slot=universe_by_slot,
ticks_by_code=ticks_by_code,
orderbook_by_code=orderbook_by_code,
program_by_code=program_by_code,
)
rsi_period = int(params.get("rsi_period", 3))
sl_pct = abs(float(params.get("sl_pct", 0.015)))
tp_pct = effective_tp_pct_from_params(params)
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))
time_start_hm = int(params.get("time_start_hm", 900))
time_end_hm = int(params.get("mom_time_end_hm", params.get("time_end_hm", 1430)))
max_loss_krw = float(params.get("max_loss_krw", 200_000.0))
all_trades: List[Dict] = []
for code, rows in codes_candles.items():
if len(rows) < max(rsi_period + 5, 6):
continue
candles = [dict(r) for r in rows]
position: Optional[Dict] = None
last_exit_dt: Dict[str, datetime] = {}
daily_cnt: Dict[str, int] = {}
cur_day = None
from kis_trader.engine.momentum_tick_replay import (
align_momentum_entry_from_ticks,
momentum_live_align_enabled,
resolve_momentum_sell_for_bar,
)
live_align = momentum_live_align_enabled(params)
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"])
if day != cur_day:
cur_day = day
is_eod = is_strategy_eod_bar(c["candle_time"], params, "MOMENTUM")
if position is not None:
if str(position["entry_time"])[:12] == str(c["candle_time"])[:12]:
continue
sell_res = resolve_momentum_sell_for_bar(
position, c, params,
is_eod=is_eod,
ticks_by_code=ticks_by_code,
code=code,
)
if sell_res:
reason, exit_price, sell_time_key, hold_min, _exit_src = sell_res
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
)
all_trades.append({
"code": code,
"buy_time": position["entry_time"],
"sell_time": sell_time_key,
"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.get("rsi", 0), 1),
"strategy": MOMENTUM_STRATEGY_ID,
})
last_exit_dt[day] = _t2dt(sell_time_key)
position = None
continue
if universe_by_slot is not None:
slot_key = _slot_key(c["candle_time"], int(params.get("scan_interval_min", 1)))
if code not in universe_by_slot.get(slot_key, []):
continue
# 해외 US: 자정 넘김 RTH (_session_wrap_midnight) — 국내 경로는 기존과 동일
from kis_trader.utils.session_hm import hm_in_trading_window
_wrap = bool(params.get("_session_wrap_midnight"))
if not hm_in_trading_window(hm, time_start_hm, time_end_hm, wrap_midnight=_wrap):
continue
if live_align:
if i < 6:
continue
signal_idx = i - 1
entry_bar_time = c["candle_time"]
entry_open = float(c["open"])
if entry_open <= 0:
continue
else:
if i < 5:
continue
signal_idx = i
if i + 1 >= len(candles):
continue
next_c = candles[i + 1]
if next_c["candle_time"][:8] != day:
continue
entry_bar_time = next_c["candle_time"]
entry_open = float(next_c["open"])
if entry_open <= 0:
continue
eval_params = dict(params)
if "skip_hts_scan_dupes" not in eval_params:
eval_params["skip_hts_scan_dupes"] = resolve_momentum_skip_hts_scan_dupes()
state = {
"daily_cnt": daily_cnt.get(day, 0),
"last_exit_dt": last_exit_dt.get(day),
}
reject, _msg, sig = eval_momentum_buy_at_index(
candles, signal_idx, eval_params, state,
)
if reject or not sig:
continue
rsi = float(sig.get("rsi") or 0)
entry_price, entry_time_key, _src = align_momentum_entry_from_ticks(
ticks_by_code, code, entry_bar_time, entry_open, params,
)
invest_amount = slot_money
if max_loss_krw > 0 and sl_pct > 0:
invest_amount = min(max_loss_krw / sl_pct, slot_money)
qty = int(invest_amount / entry_price)
if qty < 1:
continue
position = {
"entry_price": entry_price,
"entry_time": entry_time_key,
"qty": qty,
"stop": entry_price * (1 - sl_pct),
"target": entry_price * (1 + tp_pct),
"max_price": entry_price,
"rsi": rsi,
}
daily_cnt[day] = daily_cnt.get(day, 0) + 1
return all_trades
Reference in New Issue View Git Blame Copy Permalink