cb7e5037a0
Changes: - Introduced the `e_min_chg_pct` parameter to define the minimum price change percentage compared to the previous day's close, enhancing the momentum trading strategy. - Updated various functions and classes to incorporate this new parameter, ensuring it is utilized in both backtesting and live trading scenarios. - Improved documentation and comments to clarify the purpose and usage of the new parameter across the codebase. Impact: - This addition allows for more precise control over trading conditions, potentially increasing the effectiveness of the momentum strategy while maintaining system integrity and performance.
588 lines
22 KiB
Python
588 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
momentum_hts_logic.py — HTS momentum 조건식(E∧F∧H∧I) 정합 TRIGGER·청산
|
||
====================================================================
|
||
|
||
[역할 분담 — SCAN vs TRIGGER vs 청산]
|
||
- SCAN (HTS ``momentum``): F∧H∧I∧K 등 → 후보 풀
|
||
· F: 가격대 · H: 1분 거래량 펄스 · I: 거래량증감 상위(SCAN만)
|
||
· K: 전일 종가 대비 최소 등락(예 0.2%) — TRIGGER ``e_min_chg_pct`` 와 대응
|
||
- TRIGGER (본 모듈): SCAN 후 **진입 타이밍**만 검사
|
||
· ``MOMENTUM_SKIP_HTS_SCAN_DUPES=true``: K·양봉·분봉거래량 중복 생략
|
||
· false: 전일종가+e_min_chg · 양봉 · 거래량 펄스(선택)
|
||
- 청산 (본 모듈): 래칫·어깨·트레일·손절·시간컷 (A안 돌파 추격)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from kis_trader.engine.momentum_env_keys import momentum_env_bool, momentum_env_float, momentum_env_int
|
||
from kis_trader.engine.orderbook_filter import orderbook_reject_for_entry
|
||
from kis_trader.engine.program_filter import program_reject_for_entry
|
||
from kis_trader.engine.whipsaw_filter import whipsaw_reject_for_signal
|
||
from kis_trader.utils.env import get_env_from_db
|
||
from kis_trader.utils.trade_time import parse_trade_datetime as _t2dt
|
||
|
||
|
||
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 resolve_prev_trading_day_open(
|
||
candles: List[Dict],
|
||
i: int,
|
||
day: str,
|
||
) -> Optional[float]:
|
||
"""
|
||
전일(직전 거래일) 시가 — 레거시 참고용 (현재 E 조건은 종가 사용).
|
||
|
||
1분봉에서 전일 **장 시작 구간** 첫 봉 open = 일봉 시가.
|
||
전일 오후 봉만 있으면(웜업 부족) 오후 open을 시가로 오인하므로 None 반환.
|
||
"""
|
||
from kis_trader.utils.env import get_env_int
|
||
|
||
# 전일 시가로 인정할 최대 HHMM.
|
||
# 기본 1000: 저유동 종목이 09:14 첫체결만 있어도 시가로 인정.
|
||
open_hm_max = max(900, int(get_env_int("MOMENTUM_PREV_DAY_OPEN_HM_MAX", 1000)))
|
||
|
||
prev_day: Optional[str] = None
|
||
prev_open: Optional[float] = None
|
||
earliest_hm: Optional[int] = None
|
||
for j in range(i - 1, -1, -1):
|
||
ct = str(candles[j].get("candle_time", ""))
|
||
d = ct[:8]
|
||
if not d or d >= day:
|
||
continue
|
||
if prev_day is None:
|
||
prev_day = d
|
||
if d != prev_day:
|
||
break
|
||
op = float(candles[j].get("open", 0) or 0)
|
||
if op > 0:
|
||
prev_open = op
|
||
hm = None
|
||
if len(ct) >= 12:
|
||
try:
|
||
hm = int(ct[8:12])
|
||
except (TypeError, ValueError):
|
||
hm = None
|
||
if hm is not None:
|
||
if earliest_hm is None or hm < earliest_hm:
|
||
earliest_hm = hm
|
||
if prev_open is None or prev_open <= 0:
|
||
return None
|
||
if earliest_hm is None or earliest_hm > open_hm_max:
|
||
return None
|
||
return prev_open
|
||
|
||
|
||
def resolve_prev_trading_day_close(
|
||
candles: List[Dict],
|
||
i: int,
|
||
day: str,
|
||
) -> Optional[float]:
|
||
"""
|
||
전일(직전 거래일) 종가 — HTS momentum E 조건 ``close > prev_close`` 확인용.
|
||
|
||
1분봉 역스캔에서 전일(day 이전) 마지막 봉 close = 일봉 종가.
|
||
전일 봉이 1개도 없으면 None 반환.
|
||
"""
|
||
prev_day: Optional[str] = None
|
||
prev_close: Optional[float] = None
|
||
for j in range(i - 1, -1, -1):
|
||
ct = str(candles[j].get("candle_time", ""))
|
||
d = ct[:8]
|
||
if not d or d >= day:
|
||
continue
|
||
if prev_day is None:
|
||
prev_day = d
|
||
# 역스캔 첫 번째로 만난 전일 봉 = 전일 마지막(최신) 봉 → 종가
|
||
cl = float(candles[j].get("close", 0) or 0)
|
||
if cl > 0:
|
||
prev_close = cl
|
||
break
|
||
return prev_close if (prev_close is not None and prev_close > 0) else None
|
||
|
||
|
||
def candles_have_prev_session_open(
|
||
candles: List[Dict],
|
||
day: str,
|
||
) -> bool:
|
||
"""기간일 ``day`` 기준 직전 거래일 **장시작 시가** 봉이 있는지 (레거시 호환)."""
|
||
if not candles:
|
||
return False
|
||
d = str(day or "")[:8]
|
||
if len(d) < 8:
|
||
return True
|
||
return resolve_prev_trading_day_open(candles, len(candles) - 1, d) is not None
|
||
|
||
|
||
def candles_have_prev_session_close(
|
||
candles: List[Dict],
|
||
day: str,
|
||
) -> bool:
|
||
"""기간일 ``day`` 기준 직전 거래일 **종가** 봉이 있는지 (E 조건 해석 가능)."""
|
||
if not candles:
|
||
return False
|
||
d = str(day or "")[:8]
|
||
if len(d) < 8:
|
||
return True
|
||
return resolve_prev_trading_day_close(candles, len(candles) - 1, d) is not None
|
||
|
||
|
||
def _volume_pulse_ok(
|
||
candles: List[Dict],
|
||
i: int,
|
||
params: Dict[str, Any],
|
||
) -> Tuple[bool, str]:
|
||
"""H/I 정신: 당일 거래량 펄스 — 직전 N봉 평균 × 배수 이상."""
|
||
vol_mult = float(params.get("mom_vol_mult", 1.0))
|
||
vol_win = int(params.get("mom_vol_win", 5))
|
||
if vol_mult <= 0:
|
||
return True, ""
|
||
vol = float(candles[i].get("volume", 0) or 0)
|
||
win = max(1, min(vol_win, i))
|
||
vols = [float(candles[k].get("volume", 0) or 0) for k in range(i - win, i)]
|
||
if not vols or sum(vols) <= 0:
|
||
return False, "거래량창없음"
|
||
avg = sum(vols) / len(vols)
|
||
if avg <= 0 or vol < avg * vol_mult:
|
||
ratio = vol / avg if avg > 0 else 0.0
|
||
return False, "%.2fx < %.1fx" % (ratio, vol_mult)
|
||
return True, ""
|
||
|
||
|
||
def _parse_ratchet_tiers(params: Dict[str, Any]) -> List[Tuple[float, float]]:
|
||
raw = params.get("ratchet_tiers")
|
||
if raw is None:
|
||
raw = get_env_from_db("MOMENTUM_RATCHET_TIERS", "")
|
||
if isinstance(raw, (list, tuple)):
|
||
pairs = list(raw)
|
||
else:
|
||
s = str(raw or "").strip()
|
||
if not s:
|
||
return []
|
||
pairs = []
|
||
for chunk in s.split(","):
|
||
chunk = chunk.strip()
|
||
if not chunk or ":" not in chunk:
|
||
continue
|
||
g, c = chunk.split(":", 1)
|
||
pairs.append((g, c))
|
||
tiers: List[Tuple[float, float]] = []
|
||
for g, c in pairs:
|
||
try:
|
||
gain = abs(float(g)) / 100.0
|
||
cut = abs(float(c)) / 100.0
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if gain <= 0 or cut <= 0:
|
||
continue
|
||
tiers.append((gain, cut))
|
||
tiers.sort(key=lambda x: x[0])
|
||
return tiers
|
||
|
||
|
||
def _shoulder_ratios(params: Dict[str, Any]) -> Tuple[float, float]:
|
||
"""어깨 발동·컷 비율 — params(비율) 또는 legacy 퍼센트."""
|
||
smh = float(params.get("shoulder_min_high", 0.005))
|
||
sc = float(params.get("shoulder_cut_pct", 0.003))
|
||
return max(0.0, smh), max(0.0, sc)
|
||
|
||
|
||
def _minutes_held(position: Dict[str, Any], candle: Dict[str, Any]) -> Optional[int]:
|
||
try:
|
||
e = _t2dt(position.get("entry_time") or position.get("buy_time", ""))
|
||
n = _t2dt(candle.get("candle_time", ""))
|
||
return max(0, int((n - e).total_seconds() / 60))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
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.05),
|
||
params.get("tp_max_pct", 0.08),
|
||
)
|
||
|
||
|
||
def resolve_momentum_skip_hts_scan_dupes(r: Optional[Dict[str, Any]] = None) -> bool:
|
||
"""
|
||
HTS momentum SCAN(kiwoom_condition) 사용 시 TRIGGER 중복 필터 생략 여부.
|
||
|
||
- ``MOMENTUM_SKIP_HTS_SCAN_DUPES`` 명시 → 그대로
|
||
- 미설정 → ``MOMENTUM_UNIVERSE_SOURCE`` 가 condition/kiwoom_condition 이면 True
|
||
"""
|
||
if r is None:
|
||
try:
|
||
from kis_trader.utils.env import get_strategy_env_dict
|
||
r = get_strategy_env_dict("MOMENTUM") or {}
|
||
except Exception:
|
||
r = {}
|
||
raw = r.get("MOMENTUM_SKIP_HTS_SCAN_DUPES")
|
||
if raw is not None and str(raw).strip() != "":
|
||
return _to_bool(raw, True)
|
||
# 엔진 defaults 에 이미 해석된 bool 이 있으면 universe fallback 금지
|
||
if "skip_hts_scan_dupes" in r and r.get("skip_hts_scan_dupes") is not None:
|
||
return _to_bool(r.get("skip_hts_scan_dupes"), False)
|
||
src = str(r.get("MOMENTUM_UNIVERSE_SOURCE") or "condition").strip().lower()
|
||
return src in ("kiwoom_condition", "condition")
|
||
|
||
|
||
def hts_trigger_defaults_from_row(r: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""env 행에서 HTS TRIGGER 전용 플래그."""
|
||
return {
|
||
"trigger_e_confirm": momentum_env_bool(r, "MOMENTUM_TRIGGER_E_CONFIRM", True),
|
||
"trigger_require_bull_bar": momentum_env_bool(r, "MOMENTUM_TRIGGER_REQUIRE_BULL_BAR", True),
|
||
"use_vol_trigger": momentum_env_bool(r, "MOMENTUM_USE_VOL_TRIGGER", True),
|
||
"use_rsi_filter": momentum_env_bool(r, "MOMENTUM_USE_RSI_FILTER", False),
|
||
}
|
||
|
||
|
||
def eval_momentum_hts_buy_at_index(
|
||
candles: List[Dict],
|
||
i: int,
|
||
params: Dict[str, Any],
|
||
state: Dict[str, Any],
|
||
*,
|
||
compute_rsi_series_fn=None,
|
||
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
|
||
"""
|
||
HTS momentum 조건식 정합 TRIGGER.
|
||
|
||
SCAN(E∧F∧H∧I)은 이미 통과한 종목만 후보 — 여기서는 진입 타이밍만 검사.
|
||
"""
|
||
if i < 1 or i >= len(candles):
|
||
return ("탈락-봉부족", "인덱스 부적절 (i=%d)" % i, None)
|
||
|
||
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", 1530)))
|
||
cooldown_min = float(params.get("cooldown_min", 10))
|
||
max_daily = int(params.get("max_daily", 5))
|
||
min_price = float(params.get("min_price", 1000.0))
|
||
use_defense = _to_bool(params.get("use_defense_filters"), True)
|
||
use_high_chase_f = _to_bool(params.get("use_high_chase_filter"), False)
|
||
high_chase_thr = float(params.get("high_chase_thr", 0.96))
|
||
trigger_e_confirm = _to_bool(params.get("trigger_e_confirm"), True)
|
||
trigger_bull_bar = _to_bool(params.get("trigger_require_bull_bar"), True)
|
||
use_vol_trigger = _to_bool(params.get("use_vol_trigger"), True)
|
||
use_rsi_filter = _to_bool(params.get("use_rsi_filter"), False)
|
||
rsi_period = int(params.get("rsi_period", 3))
|
||
rsi_min = float(params.get("mom_rsi_min", 50.0))
|
||
rsi_max = float(params.get("mom_rsi_max", 80.0))
|
||
|
||
c = candles[i]
|
||
day = c["candle_time"][:8]
|
||
hm = int(c["candle_time"][8:12])
|
||
op = float(c.get("open", 0) or 0)
|
||
cl = float(c.get("close", 0) or 0)
|
||
hi = float(c.get("high", cl) or cl)
|
||
|
||
# 해외 US: params._session_wrap_midnight 로 자정 넘김 RTH 허용 (국내는 기존과 동일)
|
||
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):
|
||
# 국내는 장중 루프에서 흔한 silent skip. 해외는 세션 키 오설정 디버깅용으로 탈락 표기.
|
||
if str(params.get("market") or "").strip().upper() == "US":
|
||
return (
|
||
"탈락-시간외",
|
||
"hm=%04d not in %d~%d (wrap=%s)"
|
||
% (hm, time_start_hm, time_end_hm, _wrap),
|
||
None,
|
||
)
|
||
return (None, None, None)
|
||
|
||
if use_defense and cl < min_price:
|
||
return ("탈락-최소가격", "%.0f < %.0f" % (cl, min_price), None)
|
||
|
||
last_exit_dt = state.get("last_exit_dt")
|
||
if last_exit_dt is not None:
|
||
elapsed = (_t2dt(c["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)
|
||
|
||
skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), False)
|
||
prev_close = resolve_prev_trading_day_close(candles, i, day)
|
||
|
||
# kiwoom momentum SCAN — E∧F∧H∧I 이미 통과, TRIGGER 는 타이밍·호가·휩쏘만
|
||
# ※ 해외(US)는 HTS SCAN 이 없으므로 skip_hts 숏컷으로 매 봉 시그널 내면 안 됨
|
||
# → V4 추격 패턴(돌파 OR 눌림재돌파) + RSI 하한.
|
||
if skip_hts:
|
||
is_us = str(params.get("market") or "").strip().upper() == "US"
|
||
if is_us:
|
||
from kis_trader.engine.momentum_chase_patterns import eval_momentum_chase_pattern
|
||
ok_pat, pat_name, metrics = eval_momentum_chase_pattern(candles, i, params)
|
||
if not ok_pat:
|
||
return (
|
||
"탈락-패턴",
|
||
"추격패턴 미충족 (%s) close=%.4f" % (pat_name or "?", cl),
|
||
None,
|
||
)
|
||
rsi_val: Optional[float] = None
|
||
if compute_rsi_series_fn is not None:
|
||
closes = [float(x["close"]) for x in candles]
|
||
ic = params.get("_indicator_cache")
|
||
if ic is not None and hasattr(ic, "rsi_at"):
|
||
rsi_val = ic.rsi_at(i, rsi_period)
|
||
else:
|
||
rsis = compute_rsi_series_fn(closes, rsi_period)
|
||
rsi_val = rsis[i] if i < len(rsis) else None
|
||
if rsi_val is None:
|
||
return ("탈락-RSI없음", "RSI 미계산", None)
|
||
if rsi_val < rsi_min:
|
||
return ("탈락-RSI약함", "RSI=%.1f < %.0f" % (rsi_val, rsi_min), None)
|
||
use_rsi_max = _to_bool(params.get("use_rsi_max_filter"), False)
|
||
if use_rsi_max and rsi_val > rsi_max:
|
||
return ("탈락-RSI과열", "RSI=%.1f > %.0f" % (rsi_val, rsi_max), None)
|
||
sig_hts: Dict[str, Any] = {
|
||
"signal": True,
|
||
"mode": "us_momentum_chase",
|
||
"pattern": pat_name or "chase",
|
||
"signal_candle_time": c.get("candle_time"),
|
||
"prev_day_close": prev_close,
|
||
}
|
||
if metrics:
|
||
sig_hts.update(metrics)
|
||
if rsi_val is not None:
|
||
sig_hts["rsi"] = rsi_val
|
||
return (None, None, sig_hts)
|
||
|
||
sig_hts = {
|
||
"signal": True,
|
||
"mode": "momentum_hts",
|
||
"pattern": "momentum_hts_scan",
|
||
"signal_candle_time": c.get("candle_time"),
|
||
"prev_day_close": prev_close,
|
||
}
|
||
ws_rej, ws_msg = whipsaw_reject_for_signal(
|
||
params, "MOMENTUM", signal_bar=c, current_price=cl,
|
||
)
|
||
if ws_rej:
|
||
return (ws_rej, ws_msg, None)
|
||
ob_rej, ob_msg = orderbook_reject_for_entry(params, "MOMENTUM", current_price=cl)
|
||
if ob_rej:
|
||
return (ob_rej, ob_msg, None)
|
||
prog_rej, prog_msg = program_reject_for_entry(params, "MOMENTUM", current_price=cl)
|
||
if prog_rej:
|
||
return (prog_rej, prog_msg, None)
|
||
return (None, None, sig_hts)
|
||
|
||
# ── E: 전일 종가 대비 최소 등락률 이상 (HTS K 조건 대응)
|
||
# e_min_chg_pct > 0 이면 "전일 종가 + X%" 이상이어야 진입 (HTS 0.2% = 0.2 입력).
|
||
# 0.0 = 전일 종가 초과만 확인 (기존 동작).
|
||
e_min_chg_pct = float(params.get("e_min_chg_pct", 0.0))
|
||
if trigger_e_confirm:
|
||
if prev_close is None or prev_close <= 0:
|
||
return ("탈락-전일종가없음", "전일 종가 미확인", None)
|
||
threshold = prev_close * (1.0 + e_min_chg_pct / 100.0)
|
||
if cl < threshold:
|
||
chg_pct = (cl - prev_close) / prev_close * 100.0
|
||
return (
|
||
"탈락-E미충족",
|
||
"등락 %.2f%% < %.1f%% (종가%.0f 전일종가%.0f)" % (chg_pct, e_min_chg_pct, cl, prev_close),
|
||
None,
|
||
)
|
||
|
||
# ── 양봉: 당일 매수세 확인 (선택) ─────────────────────────────────
|
||
if trigger_bull_bar and op > 0 and cl <= op:
|
||
return ("탈락-음봉", "양봉 미충족", None)
|
||
|
||
# ── 거래량 펄스: H/I 정신 (선택) ───────────────────────────────────
|
||
if use_vol_trigger:
|
||
vol_ok, vol_msg = _volume_pulse_ok(candles, i, params)
|
||
if not vol_ok:
|
||
return ("탈락-거래량", vol_msg, None)
|
||
|
||
# ── 고점추격 방지 (선택, 기본 OFF) ───────────────────────────────
|
||
if use_high_chase_f:
|
||
running_high = hi
|
||
for j in range(i, -1, -1):
|
||
if candles[j]["candle_time"][:8] != day:
|
||
break
|
||
running_high = max(running_high, float(candles[j].get("high", 0) or 0))
|
||
if running_high > 0 and cl >= running_high * high_chase_thr:
|
||
return (
|
||
"탈락-고점추격",
|
||
"현재가 %.0f ≥ 고가 %.0f × %.2f" % (cl, running_high, high_chase_thr),
|
||
None,
|
||
)
|
||
|
||
# ── RSI 필터 (선택, 기본 OFF — SCAN 이후 보조만) ───────────────────
|
||
rsi_val: Optional[float] = None
|
||
if use_rsi_filter and compute_rsi_series_fn is not None:
|
||
closes = [float(x["close"]) for x in candles]
|
||
ic = params.get("_indicator_cache")
|
||
if ic is not None and hasattr(ic, "rsi_at"):
|
||
rsi_val = ic.rsi_at(i, rsi_period)
|
||
else:
|
||
rsis = compute_rsi_series_fn(closes, rsi_period)
|
||
rsi_val = rsis[i] if i < len(rsis) else None
|
||
if rsi_val is None:
|
||
return ("탈락-RSI없음", "RSI 미계산", None)
|
||
if rsi_val < rsi_min:
|
||
return ("탈락-RSI약함", "RSI=%.1f < %.0f" % (rsi_val, rsi_min), None)
|
||
if rsi_val > rsi_max:
|
||
return ("탈락-RSI과열", "RSI=%.1f > %.0f" % (rsi_val, rsi_max), None)
|
||
|
||
sig: Dict[str, Any] = {
|
||
"signal": True,
|
||
"mode": "momentum_hts",
|
||
"pattern": "hts_e_confirm",
|
||
"signal_candle_time": c.get("candle_time"),
|
||
"prev_day_close": prev_close,
|
||
}
|
||
if rsi_val is not None:
|
||
sig["rsi"] = rsi_val
|
||
|
||
ws_rej, ws_msg = whipsaw_reject_for_signal(
|
||
params, "MOMENTUM",
|
||
signal_bar=c,
|
||
current_price=cl,
|
||
)
|
||
if ws_rej:
|
||
return (ws_rej, ws_msg, None)
|
||
ob_rej, ob_msg = orderbook_reject_for_entry(
|
||
params, "MOMENTUM", current_price=cl,
|
||
)
|
||
if ob_rej:
|
||
return (ob_rej, ob_msg, None)
|
||
prog_rej, prog_msg = program_reject_for_entry(
|
||
params, "MOMENTUM", current_price=cl,
|
||
)
|
||
if prog_rej:
|
||
return (prog_rej, prog_msg, None)
|
||
return (None, None, sig)
|
||
|
||
|
||
def check_sell_signal_momentum_hts_live(
|
||
position: Dict[str, Any],
|
||
current_candle: Dict[str, Any],
|
||
params: Dict[str, Any],
|
||
is_eod: bool = False,
|
||
) -> Optional[Tuple[str, float]]:
|
||
"""
|
||
HTS momentum 추세추격 청산 — 어깨·래칫·트레일 병행.
|
||
|
||
[청산 우선순위]
|
||
1순위 래칫컷 (설정 시)
|
||
2순위 어깨컷 (고점 대비 되돌림)
|
||
3순위 손절 (하드 스탑)
|
||
4순위 트레일컷 (추세 이익 보호)
|
||
5순위 시간컷
|
||
6순위 금액손실컷 (어깨·래칫 미발동 시)
|
||
7순위 익절 (tp_max 상한, 하드 캡)
|
||
8순위 장마감청산
|
||
"""
|
||
sl_pct = -abs(float(params.get("sl_pct", params.get("stop_loss_pct", 0.03))))
|
||
tp_pct = effective_tp_pct_from_params(params)
|
||
trail_pct = abs(float(params.get("trail_pct", 0.0) or 0.0))
|
||
trail_arm_pct = abs(float(params.get("trail_arm_pct", 0.0) or 0.0))
|
||
shoulder_min_high, shoulder_cut_pct = _shoulder_ratios(params)
|
||
ratchet_tiers = _parse_ratchet_tiers(params)
|
||
max_hold_bars = int(params.get("max_hold_bars", 0) or 0)
|
||
max_loss_krw = float(params.get("max_loss_krw", 200_000.0))
|
||
min_hold_sec = float(params.get("min_hold_sec", 30.0))
|
||
min_drop_pct = float(params.get("min_drop_pct_for_loss_cut", 0.015))
|
||
|
||
try:
|
||
hi = float(current_candle.get("high", current_candle["close"]))
|
||
lo = float(current_candle.get("low", current_candle["close"]))
|
||
cl = float(current_candle["close"])
|
||
except Exception:
|
||
return None
|
||
|
||
candle_time = current_candle.get("candle_time", "")
|
||
if candle_time and position.get("entry_time"):
|
||
try:
|
||
if (_t2dt(candle_time) - _t2dt(position["entry_time"])).total_seconds() < min_hold_sec:
|
||
return None
|
||
except Exception:
|
||
pass
|
||
|
||
max_price = max(float(position.get("max_price", position["entry_price"])), hi)
|
||
position["max_price"] = max_price
|
||
|
||
entry = float(position["entry_price"])
|
||
qty = int(position.get("qty", 1) or 1)
|
||
sl_line = entry * (1 + sl_pct)
|
||
tp_line = entry * (1 + tp_pct)
|
||
|
||
# 1순위 래칫
|
||
if ratchet_tiers and entry > 0:
|
||
peak_gain = (max_price - entry) / entry
|
||
cut_ratio = 0.0
|
||
for gain, cut in ratchet_tiers:
|
||
if peak_gain >= gain:
|
||
cut_ratio = cut
|
||
if cut_ratio > 0.0:
|
||
ratchet_line = max_price * (1.0 - cut_ratio)
|
||
if lo <= ratchet_line:
|
||
return ("래칫컷", ratchet_line)
|
||
|
||
# 2순위 어깨
|
||
shoulder_armed = entry > 0 and max_price >= entry * (1.0 + shoulder_min_high)
|
||
if shoulder_armed and shoulder_cut_pct > 0:
|
||
shoulder_line = max_price * (1.0 - shoulder_cut_pct)
|
||
if lo <= shoulder_line:
|
||
return ("어깨컷", shoulder_line)
|
||
|
||
# 3순위 손절
|
||
if lo <= sl_line:
|
||
return ("손절", sl_line)
|
||
|
||
# 4순위 트레일
|
||
if trail_pct > 0 and max_price > entry:
|
||
trail_arm_line = entry * (1.0 + trail_arm_pct)
|
||
if trail_arm_pct <= 0 or max_price >= trail_arm_line:
|
||
trail_line = max_price * (1.0 - trail_pct)
|
||
if lo <= trail_line:
|
||
return ("트레일컷", trail_line)
|
||
|
||
# 5순위 시간컷
|
||
if max_hold_bars > 0:
|
||
held = _minutes_held(position, current_candle)
|
||
if held is not None and held >= max_hold_bars:
|
||
return ("시간컷", cl)
|
||
|
||
# 6순위 금액손실컷
|
||
profit_val = (lo - entry) * qty
|
||
drop_pct = (entry - lo) / entry if entry > 0 else 0.0
|
||
if (
|
||
not shoulder_armed
|
||
and not ratchet_tiers
|
||
and profit_val <= -max_loss_krw
|
||
and drop_pct >= min_drop_pct
|
||
):
|
||
exit_px = entry - (max_loss_krw / qty) if qty > 0 else lo
|
||
return ("금액손실컷", exit_px)
|
||
|
||
# 7순위 익절 (하드 캡)
|
||
if hi >= tp_line:
|
||
return ("익절", tp_line)
|
||
|
||
# 8순위 장마감
|
||
if is_eod:
|
||
return ("장마감청산", cl)
|
||
|
||
return None
|