diff --git a/kis_trader/backtest/breakout_portfolio_backtest.py b/kis_trader/backtest/breakout_portfolio_backtest.py index 9e98041..c95795f 100644 --- a/kis_trader/backtest/breakout_portfolio_backtest.py +++ b/kis_trader/backtest/breakout_portfolio_backtest.py @@ -32,7 +32,7 @@ from kis_trader.strategies.breakout import ( normalize_breakout_max_loss_krw, ) from kis_trader.strategies.base import is_strategy_eod_bar -from kis_trader.engine.tail_engine import compute_atr_series +from kis_trader.engine.atr_series import compute_atr_series def _entry_atr_at(ctx: Dict[str, Any], idx: int) -> float: diff --git a/kis_trader/engine/atr_series.py b/kis_trader/engine/atr_series.py new file mode 100644 index 0000000..fe2e170 --- /dev/null +++ b/kis_trader/engine/atr_series.py @@ -0,0 +1,37 @@ +#!/usr/bin/env python3 +""" +kis_trader/engine/atr_series.py — ATR 시리즈 (엔진·전략 공통, 순환 import 방지) +""" +from __future__ import annotations + +from typing import Dict, List, Optional + + +def compute_atr_series(candles: List[Dict], period: int = 14) -> List[Optional[float]]: + """ + ATR(Average True Range) 변동성 지표 계산기 (엔진 내부용). + + ■ RMA(Wilder's Smoothing) 방식 — TradingView 기본과 동일 (2026-06 SMA→RMA 전환). + ATR_t = (ATR_{t-1} × (period-1) + TR_t) / period + · 첫 ATR(인덱스 period)은 SMA(TR[1..period])로 시드. + · SMA 대비: 급락(큰 TR)이 14봉 지나도 '계단식 급락' 없이 완만히 감쇠 → + 급변장에서 손절/목표가가 덜 출렁임. + """ + atr_list: List[Optional[float]] = [None] * len(candles) + if len(candles) < period + 1: + return atr_list + trs = [0.0] * len(candles) + for i in range(1, len(candles)): + hi = float(candles[i]["high"]) + lo = float(candles[i]["low"]) + prev_cl = float(candles[i - 1]["close"]) + trs[i] = max(hi - lo, abs(hi - prev_cl), abs(lo - prev_cl)) + + # 첫 ATR(인덱스 period): TR[1..period] 단순 평균으로 시드 (Wilder 초기값) + prev_atr = sum(trs[1:period + 1]) / period + atr_list[period] = prev_atr + # 이후: Wilder RMA 누적 감쇠 (이전 ATR×(n-1) + 오늘 TR) / n + for i in range(period + 1, len(candles)): + prev_atr = (prev_atr * (period - 1) + trs[i]) / period + atr_list[i] = prev_atr + return atr_list diff --git a/kis_trader/engine/momentum_engine.py b/kis_trader/engine/momentum_engine.py index 37eb6fc..f9cccda 100644 --- a/kis_trader/engine/momentum_engine.py +++ b/kis_trader/engine/momentum_engine.py @@ -32,7 +32,7 @@ from kis_trader.engine.momentum_env_keys import ( momentum_env_int, _legacy_float, ) -from kis_trader.strategies.base import is_strategy_eod_bar +from kis_trader.engine.strategy_eod import is_strategy_eod_bar from kis_trader.utils.env import get_env_int MOMENTUM_STRATEGY_ID = "MOMENTUM" diff --git a/kis_trader/engine/strategy_eod.py b/kis_trader/engine/strategy_eod.py new file mode 100644 index 0000000..1ebfe80 --- /dev/null +++ b/kis_trader/engine/strategy_eod.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +""" +kis_trader/engine/strategy_eod.py — 전략 EOD 시각 판정 (실매·백테·엔진 공통) +===================================================================== +strategies.base 와 engine 간 순환 import 방지용 — 엔진은 이 모듈만 import. +""" +from __future__ import annotations + +from datetime import datetime as dt +from typing import Any, Dict, Tuple + + +def parse_eod_hm(raw: str, default: str = "15:25") -> Tuple[int, int]: + """EOD 시각 문자열 → (시, 분). ``1515`` / ``15:15`` 모두 허용.""" + s = str(raw or default).strip() + if not s or s.lower() == "none": + s = default + if ":" in s: + parts = s.split(":", 1) + try: + return int(parts[0]), int(parts[1]) + except (ValueError, TypeError): + pass + if len(s) == 4 and s.isdigit(): + return int(s[:2]), int(s[2:]) + try: + hh, mm = [int(x) for x in s.split(":")] + return hh, mm + except Exception: + return 15, 25 + + +def is_live_eod_now( + enabled: bool, + eod_hm: str, + now: dt, + *, + default_hm: str = "15:25", +) -> bool: + """실매 EOD 당일청산 시각 도달 여부.""" + if not enabled: + return False + eod_hh, eod_mm = parse_eod_hm(eod_hm, default_hm) + return (now.hour > eod_hh) or (now.hour == eod_hh and now.minute >= eod_mm) + + +# 전략별 EOD env 키 — 실매·백테·파라서치 공통 +_STRATEGY_EOD_SPEC: Dict[str, Tuple[str, str, bool, str, str]] = { + "BREAKOUT": ("BREAKOUT_EOD_ENABLED", "BREAKOUT_EOD_HM", True, "15:15", ""), + "MOMENTUM": ("MOMENTUM_EOD_ENABLED", "MOMENTUM_EOD_HM", True, "15:25", "MOMENTUM_FORCE_EOD_EXIT"), + "TAIL": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:25", "force_eod_exit"), + "SHORT": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:25", "force_eod_exit"), +} + + +def _params_truthy_bool(val: Any, default: bool) -> bool: + if val is None or val == "" or val == "None": + return default + if isinstance(val, bool): + return val + return str(val).strip().lower() in ("1", "true", "t", "y", "yes", "on") + + +def resolve_strategy_eod_params( + params: Dict[str, Any], + strategy_id: str, +) -> Tuple[bool, str]: + """params → (eod_enabled, eod_hm). UI ``eod_enabled``/``eod_hm`` 우선, 없으면 env 키.""" + sid = str(strategy_id or "").strip().upper() + if sid == "SHORT": + sid = "TAIL" + spec = _STRATEGY_EOD_SPEC.get(sid) + if spec is None: + return False, "15:25" + en_key, hm_key, def_en, def_hm, leg_key = spec + + if "eod_enabled" in params: + enabled = _params_truthy_bool(params.get("eod_enabled"), def_en) + elif en_key in params: + enabled = _params_truthy_bool(params.get(en_key), def_en) + elif leg_key and leg_key in params: + enabled = _params_truthy_bool(params.get(leg_key), def_en) + else: + enabled = def_en + + raw_hm = params.get("eod_hm") + if raw_hm not in (None, "", "None"): + eod_hm = str(raw_hm).strip() + elif params.get(hm_key) not in (None, "", "None"): + eod_hm = str(params.get(hm_key)).strip() + else: + eod_hm = def_hm + return enabled, eod_hm + + +def is_backtest_eod_bar( + candle_time: str, + enabled: bool, + eod_hm: str, + *, + default_hm: str = "15:25", +) -> bool: + """백테 1분봉/스캔키 — 실매 ``is_live_eod_now`` 와 동일 시각 기준.""" + if not enabled: + return False + eod_hh, eod_mm = parse_eod_hm(eod_hm, default_hm) + t = str(candle_time).strip() + if len(t) < 12: + return False + try: + bar_hh = int(t[8:10]) + bar_mm = int(t[10:12]) + except (ValueError, TypeError): + return False + return (bar_hh > eod_hh) or (bar_hh == eod_hh and bar_mm >= eod_mm) + + +def is_strategy_eod_bar( + candle_time: str, + params: Dict[str, Any], + strategy_id: str, +) -> bool: + """전략 params + 봉시각 → EOD 청산 여부 (실매와 동일 키·시각).""" + sid = str(strategy_id or "").strip().upper() + if sid == "SHORT": + sid = "TAIL" + spec = _STRATEGY_EOD_SPEC.get(sid) + if spec is None: + return False + _, _, _, def_hm, _ = spec + enabled, eod_hm = resolve_strategy_eod_params(params, sid) + return is_backtest_eod_bar(candle_time, enabled, eod_hm, default_hm=def_hm) diff --git a/kis_trader/engine/tail_engine.py b/kis_trader/engine/tail_engine.py index 90e41b6..c69dc16 100644 --- a/kis_trader/engine/tail_engine.py +++ b/kis_trader/engine/tail_engine.py @@ -67,7 +67,8 @@ from kis_trader.engine.tail_tick_replay import ( tail_timeframe_min, try_limit_fill_on_bar_with_ticks, ) -from kis_trader.strategies.base import is_strategy_eod_bar +from kis_trader.engine.atr_series import compute_atr_series +from kis_trader.engine.strategy_eod import is_strategy_eod_bar from kis_trader.engine.tail_env_keys import ( tail_env_bool, tail_env_float, @@ -374,36 +375,6 @@ def compute_sma_series(closes: List[float], period: int = 20) -> List[Optional[f return sma_list -def compute_atr_series(candles: List[Dict], period: int = 14) -> List[Optional[float]]: - """ - ATR(Average True Range) 변동성 지표 계산기 (엔진 내부용). - - ■ RMA(Wilder's Smoothing) 방식 — TradingView 기본과 동일 (2026-06 SMA→RMA 전환). - ATR_t = (ATR_{t-1} × (period-1) + TR_t) / period - · 첫 ATR(인덱스 period)은 SMA(TR[1..period])로 시드. - · SMA 대비: 급락(큰 TR)이 14봉 지나도 '계단식 급락' 없이 완만히 감쇠 → - 급변장에서 손절/목표가가 덜 출렁임. - """ - atr_list: List[Optional[float]] = [None] * len(candles) - if len(candles) < period + 1: - return atr_list - trs = [0.0] * len(candles) - for i in range(1, len(candles)): - hi = float(candles[i]["high"]) - lo = float(candles[i]["low"]) - prev_cl = float(candles[i - 1]["close"]) - trs[i] = max(hi - lo, abs(hi - prev_cl), abs(lo - prev_cl)) - - # 첫 ATR(인덱스 period): TR[1..period] 단순 평균으로 시드 (Wilder 초기값) - prev_atr = sum(trs[1:period + 1]) / period - atr_list[period] = prev_atr - # 이후: Wilder RMA 누적 감쇠 (이전 ATR×(n-1) + 오늘 TR) / n - for i in range(period + 1, len(candles)): - prev_atr = (prev_atr * (period - 1) + trs[i]) / period - atr_list[i] = prev_atr - return atr_list - - def _t2dt(t: str) -> datetime: """candle_time 문자열 → datetime.""" return datetime.strptime(t, "%Y%m%d%H%M") diff --git a/kis_trader/strategies/base.py b/kis_trader/strategies/base.py index a446dad..7ee58b7 100644 --- a/kis_trader/strategies/base.py +++ b/kis_trader/strategies/base.py @@ -61,126 +61,13 @@ def _is_non_stock(name: str, code: str) -> bool: return False -def parse_eod_hm(raw: str, default: str = "15:25") -> Tuple[int, int]: - """EOD 시각 문자열 → (시, 분). ``1515`` / ``15:15`` 모두 허용.""" - s = str(raw or default).strip() - if not s or s.lower() == "none": - s = default - if ":" in s: - parts = s.split(":", 1) - try: - return int(parts[0]), int(parts[1]) - except (ValueError, TypeError): - pass - if len(s) == 4 and s.isdigit(): - return int(s[:2]), int(s[2:]) - try: - hh, mm = [int(x) for x in s.split(":")] - return hh, mm - except Exception: - return 15, 25 - - -def is_live_eod_now( - enabled: bool, - eod_hm: str, - now: dt, - *, - default_hm: str = "15:25", -) -> bool: - """실매 EOD 당일청산 시각 도달 여부.""" - if not enabled: - return False - eod_hh, eod_mm = parse_eod_hm(eod_hm, default_hm) - return (now.hour > eod_hh) or (now.hour == eod_hh and now.minute >= eod_mm) - - -# 전략별 EOD env 키 — 실매·백테·파라서치 공통 -_STRATEGY_EOD_SPEC: Dict[str, Tuple[str, str, bool, str, str]] = { - "BREAKOUT": ("BREAKOUT_EOD_ENABLED", "BREAKOUT_EOD_HM", True, "15:15", ""), - "MOMENTUM": ("MOMENTUM_EOD_ENABLED", "MOMENTUM_EOD_HM", True, "15:25", "MOMENTUM_FORCE_EOD_EXIT"), - "TAIL": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:25", "force_eod_exit"), - "SHORT": ("TAIL_EOD_ENABLED", "TAIL_EOD_HM", True, "15:25", "force_eod_exit"), -} - - -def _params_truthy_bool(val: Any, default: bool) -> bool: - if val is None or val == "" or val == "None": - return default - if isinstance(val, bool): - return val - return str(val).strip().lower() in ("1", "true", "t", "y", "yes", "on") - - -def resolve_strategy_eod_params( - params: Dict[str, Any], - strategy_id: str, -) -> Tuple[bool, str]: - """params → (eod_enabled, eod_hm). UI ``eod_enabled``/``eod_hm`` 우선, 없으면 env 키.""" - sid = str(strategy_id or "").strip().upper() - if sid == "SHORT": - sid = "TAIL" - spec = _STRATEGY_EOD_SPEC.get(sid) - if spec is None: - return False, "15:25" - en_key, hm_key, def_en, def_hm, leg_key = spec - - if "eod_enabled" in params: - enabled = _params_truthy_bool(params.get("eod_enabled"), def_en) - elif en_key in params: - enabled = _params_truthy_bool(params.get(en_key), def_en) - elif leg_key and leg_key in params: - enabled = _params_truthy_bool(params.get(leg_key), def_en) - else: - enabled = def_en - - raw_hm = params.get("eod_hm") - if raw_hm not in (None, "", "None"): - eod_hm = str(raw_hm).strip() - elif params.get(hm_key) not in (None, "", "None"): - eod_hm = str(params.get(hm_key)).strip() - else: - eod_hm = def_hm - return enabled, eod_hm - - -def is_backtest_eod_bar( - candle_time: str, - enabled: bool, - eod_hm: str, - *, - default_hm: str = "15:25", -) -> bool: - """백테 1분봉/스캔키 — 실매 ``is_live_eod_now`` 와 동일 시각 기준.""" - if not enabled: - return False - eod_hh, eod_mm = parse_eod_hm(eod_hm, default_hm) - t = str(candle_time).strip() - if len(t) < 12: - return False - try: - bar_hh = int(t[8:10]) - bar_mm = int(t[10:12]) - except (ValueError, TypeError): - return False - return (bar_hh > eod_hh) or (bar_hh == eod_hh and bar_mm >= eod_mm) - - -def is_strategy_eod_bar( - candle_time: str, - params: Dict[str, Any], - strategy_id: str, -) -> bool: - """전략 params + 봉시각 → EOD 청산 여부 (실매와 동일 키·시각).""" - sid = str(strategy_id or "").strip().upper() - if sid == "SHORT": - sid = "TAIL" - spec = _STRATEGY_EOD_SPEC.get(sid) - if spec is None: - return False - _, _, _, def_hm, _ = spec - enabled, eod_hm = resolve_strategy_eod_params(params, sid) - return is_backtest_eod_bar(candle_time, enabled, eod_hm, default_hm=def_hm) +from ..engine.strategy_eod import ( + is_backtest_eod_bar, + is_live_eod_now, + is_strategy_eod_bar, + parse_eod_hm, + resolve_strategy_eod_params, +) class BaseStrategy(ABC, threading.Thread): diff --git a/kis_trader/strategies/breakout.py b/kis_trader/strategies/breakout.py index 6099657..a480039 100644 --- a/kis_trader/strategies/breakout.py +++ b/kis_trader/strategies/breakout.py @@ -41,7 +41,7 @@ _bo_logger = get_logger("kis_trader.breakout") from ..utils.position_sizing import invest_qty_for_price from ..engine.ema_trend_filter import eval_ema_uptrend_reject from ..engine.scalping_engine import check_sell_signal_backtest_bar -from ..engine.tail_engine import compute_atr_series +from ..engine.atr_series import compute_atr_series from ..engine.whipsaw_filter import whipsaw_reject_for_signal from ..engine.orderbook_filter import orderbook_reject_for_entry from ..engine.program_filter import program_reject_for_entry