hoon/kis_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
844d32f39cb37a34db07ad129570d93259e81cc4
kis_bot/kis_trader/engine/tail_engine.py
Hwang 78edb75e01 feat: Add new files and enhance backtesting functionality 
Changes:
- Introduced new files for strategy definitions and study names.
- Enhanced `backtest_web.py` with functions to handle integer display prices and trade data formatting.
- Updated backtesting logic to incorporate end-of-day (EOD) parameters for breakout and momentum strategies.
- Added EOD configuration options in the database and parameter search files.

Impact:
- These changes improve the modularity and usability of the backtesting framework, allowing for better integration of EOD strategies and clearer trade data presentation.
2026-07-06 19:11:34 +09:00

2025 lines
84 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
kis_trader/engine/tail_engine.py — 꼬리잡기 백테스트·실매매 공통 엔진
====================================================
백테스트(backtest_web), 파라미터 탐색(tail_param_search), 실매매(kis_trader TailCatchStrategy)가
모두 동일한 진입/청산 계산식과 '고급 방어 로직'을 쓰도록 통합된 단일 소스 엔진.
■ SCAN vs TRIGGER (돌파 전략과 동일 원칙)
[SCAN — HTS 조건검색, 널넬하게]
**실매 권장: 키움 ``tail`` (``SHORT_UNIVERSE_SOURCE=kiwoom_condition``)**
A) [일] 시가→종가 -10% ~ -1.5% (당일 약세)
B) 체결강도 85% ~ 400%
C) 3봉전 대비 거래량 180% ~ 2000%
F) [일] 저가 대비 종가 +1% ~ +8% (꼬리 회복 구간)
→ ``KiwoomConditionSearchManager`` + ``CONDITION_SHORT_NAME=tail``
**레거시 KIS ``condition`` (REST 폴링)**
A) 1봉 등락률 -10% ~ -1.5% (직전봉 종가 대비)
B/C 동일 축 — ``ConditionSearchManager`` / ``tail``
→ ``target_candidates_history`` (strategy_id=SHORT) 에 스냅샷 저장.
[TRIGGER — 본 엔진, 엄격하게 — HTS A/B/C 는 조건검색에서 이미 통과]
반전 패턴 OR(망치·핀바·장악·관통·하라미·도지·샛별) + 당일 회복률·3분봉 회복 위치,
(선택) 신호봉 거래량 폭증(``TAIL_VOL_MULT``×N봉평균, 0=OFF) + RSI·MA20,
고점추격·피뢰침, 시간대/쿨다운. 패턴별 ``TAIL_PATTERN_*`` env 로 ON/OFF.
``TAIL_SKIP_HTS_SCAN_DUPES=true`` (기본) 이면 TRIGGER 에서 HTS A 와 겹치는
**당일 시가→저점 낙폭** 은 건너뜀 → 이중 필터로 거래가 사라지는 현상 방지.
■ 엔진 공통 로직 (백테·실매 동일)
매도 우선순위 (V4):
1) 트레일(어깨컷) — max_price 갱신 후 되돌림(저가로 터치 판정, 체결=매도선), 발동 수익% 충족 시 **손절·금액손실보다 우선**
2) ATR 캡 적용 익절 / 3) ATR 캡 적용 손절 / 4) 금액손실컷(트레일 미발동 구간만) / 5) 장마감
ATR 목표·손절: 배수 × ATR 후 ``TAIL_ATR_*_MIN/MAX_PCT`` % 상·하한 캡 (잡주 과대 목표가 방지).
백테 청산: N분 OHLC → N회 intrabar 체크 (``check_sell_signal_backtest_bar``, session_low 모드).
진입·지표는 N분봉 그대로.
■ 진입 모드 (``TAIL_ENTRY_MODE``)
- ``align``: 신호봉 확정 → 다음 3분봉 시가 시장가(레거시)
- ``limit_atr``(기본): 신호봉 확정 → anchor−ATR×mult 지정가 → ``TAIL_LIMIT_VALID_BARS`` 봉 내
low 터치 시 체결, 미체결 시 다음 봉부터 취소(실매) / 백테 스킵
"""
from datetime import datetime
from typing import List, Dict, Any, Optional, Tuple
from kis_trader.engine.limit_entry_common import (
compute_atr_limit_price,
is_limit_atr_entry,
limit_valid_until_bar_key,
resolve_limit_anchor_price,
short_entry_mode,
tail_limit_params,
try_limit_fill_on_bar,
)
from kis_trader.engine.indicator_cache import (
attach_indicator_caches_to_params,
get_indicator_cache_from_params,
)
from kis_trader.engine.whipsaw_filter import inject_whipsaw_ticks_into_params, whipsaw_reject_for_signal
from kis_trader.backtest.trigger_snapshot_loader import inject_trigger_snapshots_into_params
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.scalping_engine import check_sell_signal_backtest_bar
from kis_trader.engine.tail_tick_replay import (
align_entry_price_from_ticks,
collect_bar_ticks,
tail_backtest_wants_tick_replay,
tail_timeframe_min,
try_limit_fill_on_bar_with_ticks,
)
from kis_trader.strategies.base import is_strategy_eod_bar
from kis_trader.engine.tail_env_keys import (
tail_env_bool,
tail_env_float,
tail_env_int,
)
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:
"""봉 시각을 N분 단위 슬롯 키로 변환.
신봇 기준: caller 가 ``TradeDBExt.get_universe_by_candle_time()`` 으로
1분 캔들 시각 키 dict 를 만들어 주입하므로 기본값은 1분(= passthrough).
``--fallback-universe`` 시뮬레이션은 caller 가 ``scan_interval_min=5`` 를
명시해 5분 버킷팅으로 사용.
"""
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 get_tail_defaults_from_db(db=None) -> Dict[str, Any]:
"""
env_config 최신 행에서 꼬리잡기 관련 값과 고급 방어 로직 값을 전부 로드.
백테스트·파라미터서치·실매매가 동일 DB 값을 쓰도록 단일 소스.
"""
own_db = None
r: Dict[str, Any] = {}
try:
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 {}
if r:
# TAIL_* 키만 사용 (레거시 MIN_DROP_RATE 등 폴백 없음)
min_drop = tail_env_float(r, "TAIL_MIN_DROP_RATE", 0.03)
min_rec = tail_env_float(r, "TAIL_MIN_RECOVERY_RATIO", 0.5)
tail_ratio = tail_env_float(r, "TAIL_RATIO_MIN", 1.5)
tail_pct = tail_env_float(r, "TAIL_PCT_MIN", 0.003)
sl_pct = abs(tail_env_float(r, "TAIL_STOP_LOSS_PCT", -0.03))
tp_pct = tail_env_float(r, "TAIL_TAKE_PROFIT_PCT", 0.05)
shoulder_high = tail_env_float(r, "TAIL_SHOULDER_MIN_HIGH_PCT", 0.005)
shoulder_cut = tail_env_float(r, "TAIL_SHOULDER_CUT_PCT", 0.003)
cooldown_sec = tail_env_int(r, "TAIL_COOLDOWN_SEC", 900)
rsi_period = tail_env_int(r, "TAIL_RSI_PERIOD", 14)
rsi_threshold = tail_env_float(r, "TAIL_RSI_THRESHOLD", 78.0)
max_rec_3m = tail_env_float(r, "TAIL_MAX_RECOVERY_3M", 0.8)
high_chase = tail_env_float(r, "TAIL_HIGH_CHASE_THR", 0.96)
time_start = tail_env_int(r, "TAIL_TIME_START", 930)
time_end = tail_env_int(r, "TAIL_TIME_END", 1500)
max_daily = tail_env_int(r, "TAIL_MAX_DAILY", 3)
min_price = tail_env_float(r, "TAIL_MIN_PRICE", 1000.0)
max_daily_change = tail_env_float(r, "TAIL_MAX_DAILY_CHG", 20.0)
ma20_max_above = tail_env_float(r, "TAIL_MA20_MAX_ABOVE_PCT", 3.0)
stop_atr_mult = tail_env_float(r, "TAIL_STOP_ATR_MULT", 2.0)
target_atr_mult = tail_env_float(r, "TAIL_TARGET_ATR_MULT", 2.5)
atr_sl_min_pct = _read_tail_pct_from_row(r, "TAIL_ATR_SL_MIN_PCT", "TAIL_ATR_SL_MIN_PCT", 0.8) * 100.0
atr_sl_max_pct = _read_tail_pct_from_row(r, "TAIL_ATR_SL_MAX_PCT", "TAIL_ATR_SL_MAX_PCT", 6.0) * 100.0
atr_tp_min_pct = _read_tail_pct_from_row(r, "TAIL_ATR_TP_MIN_PCT", "TAIL_ATR_TP_MIN_PCT", 0.5) * 100.0
atr_tp_max_pct = _read_tail_pct_from_row(r, "TAIL_ATR_TP_MAX_PCT", "TAIL_ATR_TP_MAX_PCT", 3.0) * 100.0
max_loss_krw = tail_env_int(r, "TAIL_MAX_LOSS_KRW", 200000)
_min_drop_loss = r.get("TAIL_MIN_DROP_FOR_LOSS_CUT")
min_drop_pct_for_loss_cut = 0.015
if _min_drop_loss not in (None, ""):
v = float(_min_drop_loss)
min_drop_pct_for_loss_cut = v / 100.0 if v >= 1 else v
risk_pct = float(r.get("RISK_PCT_PER_TRADE") or 0.01)
kelly_mult = float(r.get("KELLY_MULTIPLIER") or 0.25)
min_hold_sec = float(r.get("MIN_HOLD_AFTER_BUY_SEC") or 30.0)
capital = float(r.get("BACKTEST_CAPITAL") or 100000000.0)
skip_hts_scan_dupes = tail_env_bool(r, "TAIL_SKIP_HTS_SCAN_DUPES", True)
use_intraday_drop = tail_env_bool(r, "TAIL_USE_INTRADAY_DROP", False)
use_ma20_filter = tail_env_bool(r, "TAIL_USE_MA20_FILTER", False)
use_rsi_filter = tail_env_bool(r, "TAIL_USE_RSI_FILTER", True)
use_daily_range_f = tail_env_bool(r, "TAIL_USE_DAILY_RANGE_FILTER", True)
use_high_chase_f = tail_env_bool(r, "TAIL_USE_HIGH_CHASE_FILTER", True)
bar_chg_min_pct = tail_env_float(r, "TAIL_BAR_CHG_MIN_PCT", -10.0)
bar_chg_max_pct = tail_env_float(r, "TAIL_BAR_CHG_MAX_PCT", -1.5)
tail_vol_mult = tail_env_float(r, "TAIL_VOL_MULT", 0.0)
tail_vol_win = tail_env_int(r, "TAIL_VOL_WIN", 5)
max_stocks = tail_env_int(r, "TAIL_MAX_STOCKS", 3)
total_budget_krw = tail_env_int(r, "TAIL_TOTAL_BUDGET_KRW", 0)
slot_money = tail_env_int(r, "TAIL_SLOT_MONEY", 3_000_000)
short_max_buy = tail_env_int(r, "TAIL_MAX_BUY_AMOUNT", 0)
from kis_trader.utils.env import get_env_from_db
ratchet_tiers = str(r.get("TAIL_RATCHET_TIERS") or get_env_from_db("TAIL_RATCHET_TIERS", "") or "").strip()
max_hold_bars = tail_env_int(r, "TAIL_MAX_HOLD_BARS", 0)
eod_enabled = tail_env_bool(r, "TAIL_EOD_ENABLED", True)
eod_hm = str(r.get("TAIL_EOD_HM") or "15:25").strip() or "15:25"
# ws_ticks 진입가 재생 기본 ON — tail_tick_replay·모멘텀·돌파와 동일 (실매 체결 정합, env=0 일 때만 OFF)
backtest_use_tick_db = tail_env_bool(r, "TAIL_BACKTEST_USE_TICK_DB", True)
backtest_tick_fallback_ohlc = tail_env_bool(r, "TAIL_BACKTEST_TICK_FALLBACK_OHLC", True)
trail_pct = abs(tail_env_float(r, "TAIL_TRAIL_PCT", 0.0))
trail_arm_pct = abs(tail_env_float(r, "TAIL_TRAIL_ARM_PCT", 0.0))
_pat = _load_tail_pattern_params_from_row(r)
else:
min_drop, min_rec = 0.03, 0.5
tail_ratio, tail_pct = 1.5, 0.003
sl_pct, tp_pct = 0.03, 0.05
shoulder_high, shoulder_cut = 0.005, 0.003
cooldown_sec, rsi_period, rsi_threshold = 900, 14, 78.0
max_rec_3m, high_chase = 0.8, 0.96
time_start, time_end, max_daily = 930, 1500, 3
min_price, max_daily_change, ma20_max_above = 1000.0, 20.0, 3.0
stop_atr_mult, target_atr_mult = 2.0, 2.5
atr_sl_min_pct, atr_sl_max_pct = 0.8, 6.0
atr_tp_min_pct, atr_tp_max_pct = 0.5, 3.0
max_loss_krw, risk_pct, kelly_mult = 200000, 0.01, 0.25
min_drop_pct_for_loss_cut = 0.015
min_hold_sec, capital = 30.0, 100000000.0
skip_hts_scan_dupes, use_intraday_drop = True, False
use_ma20_filter, use_rsi_filter = False, True
use_daily_range_f, use_high_chase_f = True, True
bar_chg_min_pct, bar_chg_max_pct = -10.0, -1.5
tail_vol_mult, tail_vol_win = 0.0, 5
max_stocks, total_budget_krw, slot_money, short_max_buy = 3, 0, 3_000_000, 0
ratchet_tiers, max_hold_bars = "", 0
eod_enabled, eod_hm = True, "15:25"
backtest_use_tick_db, backtest_tick_fallback_ohlc = False, True
trail_pct, trail_arm_pct = 0.0, 0.0
_pat = _load_tail_pattern_params_from_row({})
except Exception:
min_drop, min_rec = 0.03, 0.5
tail_ratio, tail_pct = 1.5, 0.003
sl_pct, tp_pct = 0.03, 0.05
shoulder_high, shoulder_cut = 0.015, 0.03
cooldown_sec, rsi_period, rsi_threshold = 900, 14, 78.0
max_rec_3m, high_chase = 0.8, 0.96
time_start, time_end, max_daily = 930, 1500, 3
min_price, max_daily_change, ma20_max_above = 1000.0, 20.0, 3.0
stop_atr_mult, target_atr_mult = 2.5, 5.0
atr_sl_min_pct, atr_sl_max_pct = 0.8, 6.0
atr_tp_min_pct, atr_tp_max_pct = 0.5, 5.0
max_loss_krw, risk_pct, kelly_mult = 200000, 0.01, 0.25
min_drop_pct_for_loss_cut = 0.015
min_hold_sec, capital = 30.0, 100000000.0
skip_hts_scan_dupes, use_intraday_drop = True, False
use_ma20_filter, use_rsi_filter = False, True
use_daily_range_f, use_high_chase_f = True, True
bar_chg_min_pct, bar_chg_max_pct = -10.0, -1.5
tail_vol_mult, tail_vol_win = 0.0, 5
max_stocks, total_budget_krw, slot_money, short_max_buy = 3, 0, 3_000_000, 0
ratchet_tiers, max_hold_bars = "", 0
eod_enabled, eod_hm = True, "15:25"
backtest_use_tick_db, backtest_tick_fallback_ohlc = False, True
trail_pct, trail_arm_pct = 0.0, 0.0
_pat = _load_tail_pattern_params_from_row({})
finally:
if own_db is not None:
try:
own_db.close()
except Exception:
pass
return {
"min_drop_rate": min_drop,
"min_recovery_ratio": min_rec,
"max_rec_3m": max_rec_3m,
"tail_ratio_min": tail_ratio,
"tail_pct_min": tail_pct,
"sl_pct": sl_pct,
"tp_pct": tp_pct,
"shoulder_min_high": shoulder_high,
"shoulder_cut_pct": shoulder_cut,
"ratchet_tiers": ratchet_tiers,
"max_hold_bars": max_hold_bars,
"rsi_period": rsi_period,
"rsi_threshold": rsi_threshold,
"high_chase_thr": high_chase,
"cooldown_min": cooldown_sec // 60,
"time_start_hm": time_start,
"time_end_hm": time_end,
"max_daily": max_daily,
# 고급 방어 파라미터 반환
"min_price": min_price,
"max_daily_change": max_daily_change,
"ma20_max_above": ma20_max_above,
"stop_atr_mult": stop_atr_mult,
"target_atr_mult": target_atr_mult,
"atr_sl_min_pct": atr_sl_min_pct,
"atr_sl_max_pct": atr_sl_max_pct,
"atr_tp_min_pct": atr_tp_min_pct,
"atr_tp_max_pct": atr_tp_max_pct,
"max_loss_krw": max_loss_krw,
"min_drop_pct_for_loss_cut": min_drop_pct_for_loss_cut,
"risk_pct": risk_pct,
"kelly_mult": kelly_mult,
"min_hold_sec": min_hold_sec,
"capital": capital,
# SCAN/TRIGGER 분리 플래그
"skip_hts_scan_dupes": skip_hts_scan_dupes,
"use_intraday_drop": use_intraday_drop,
"use_ma20_filter": use_ma20_filter,
"use_rsi_filter": use_rsi_filter,
"use_daily_range_filter": use_daily_range_f,
"use_high_chase_filter": use_high_chase_f,
"bar_chg_min_pct": bar_chg_min_pct,
"bar_chg_max_pct": bar_chg_max_pct,
"tail_vol_mult": tail_vol_mult,
"tail_vol_win": tail_vol_win,
"max_stocks": max_stocks,
"total_budget_krw": total_budget_krw,
"slot_money": slot_money,
"short_max_buy_amount": short_max_buy,
"portfolio_mode": True,
"entry_mode": short_entry_mode(
{"entry_mode": r.get("TAIL_ENTRY_MODE")} if r else None
),
"limit_atr_mult": tail_limit_params(
{
"limit_atr_mult": tail_env_float(r, "TAIL_LIMIT_ATR_MULT", 1.5) if r else None,
"limit_anchor": (r.get("TAIL_LIMIT_ANCHOR") if r else None),
"limit_valid_bars": tail_env_int(r, "TAIL_LIMIT_VALID_BARS", 1) if r else None,
"limit_fill_slip_pct": tail_env_float(r, "TAIL_LIMIT_FILL_SLIP_PCT", 0.0) if r else None,
}
).get("mult", 1.5),
"limit_anchor": tail_limit_params(
{
"limit_atr_mult": tail_env_float(r, "TAIL_LIMIT_ATR_MULT", 1.5) if r else None,
"limit_anchor": (r.get("TAIL_LIMIT_ANCHOR") if r else None),
}
).get("anchor", "signal_low"),
"limit_valid_bars": int(tail_limit_params(
{"limit_valid_bars": tail_env_int(r, "TAIL_LIMIT_VALID_BARS", 1) if r else None}
).get("valid_bars", 1)),
"limit_fill_slip_pct": float(tail_limit_params(
{"limit_fill_slip_pct": tail_env_float(r, "TAIL_LIMIT_FILL_SLIP_PCT", 0.0) if r else None}
).get("fill_slip_pct", 0.0)),
"backtest_use_tick_db": backtest_use_tick_db,
"backtest_tick_fallback_ohlc": backtest_tick_fallback_ohlc,
# 체결량 상한(진입봉 거래량×N%) — 0=OFF. 실매 IOC 미체결 근사 (DB·웹·CLI 공통).
"backtest_vol_fill_cap_pct": tail_env_float(r, "TAIL_BACKTEST_VOL_FILL_CAP_PCT", 0.0) if r else 0.0,
"trail_pct": trail_pct,
"trail_arm_pct": trail_arm_pct,
"eod_enabled": eod_enabled,
"eod_hm": eod_hm,
"force_eod_exit": eod_enabled,
**_pat,
}
def compute_rsi_series(closes: List[float], period: int = 14) -> List[Optional[float]]:
"""RSI 시리즈 (Wilder 스무딩)."""
rsi_list: List[Optional[float]] = [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 compute_sma_series(closes: List[float], period: int = 20) -> List[Optional[float]]:
"""단순 이동평균(SMA) 계산기 (엔진 내부용)."""
sma_list: List[Optional[float]] = [None] * len(closes)
if len(closes) < period:
return sma_list
running_sum = sum(closes[:period])
sma_list[period - 1] = running_sum / period
for i in range(period, len(closes)):
running_sum += closes[i] - closes[i - period]
sma_list[i] = running_sum / period
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")
def _pct_to_frac(v: Any, default_pct: float) -> float:
"""퍼센트 숫자(1.5=1.5%%) 또는 소수(0.015) → 비율 소수."""
if v is None or v == "":
v = default_pct
x = float(v)
return x / 100.0 if x >= 0.2 else x
def _read_tail_pct_from_row(r: Dict[str, Any], db_key: str, env_key: str, default_pct: float) -> float:
"""env_config 행 → 비율 소수. 없으면 get_env_float 폴백."""
raw = r.get(db_key) if r else None
if raw not in (None, ""):
return _pct_to_frac(raw, default_pct)
try:
from kis_trader.utils.env import get_env_float
return _pct_to_frac(get_env_float(env_key, default_pct), default_pct)
except Exception:
return _pct_to_frac(default_pct, default_pct)
def compute_tail_atr_prices(
entry_price: float,
atr_value: Optional[float],
params: Dict[str, Any],
) -> Tuple[float, float]:
"""
꼬리잡기 ATR 손절/익절가 — 배수 적용 후 % 상·하한 캡 (UPDOW ``resolve_atr_exit_pcts`` 와 동일 철학).
:return: (stop_price, target_price)
"""
if entry_price <= 0:
return entry_price * 0.97, entry_price * 1.03
atr = float(atr_value) if atr_value and float(atr_value) > 0 else entry_price * 0.01
stop_mult = float(params.get("stop_atr_mult", 2.5))
target_mult = float(params.get("target_atr_mult", 5.0))
sl_pct = (atr * stop_mult) / entry_price
tp_pct = (atr * target_mult) / entry_price
sl_min = _pct_to_frac(params.get("atr_sl_min_pct", 0.8), 0.8)
sl_max = _pct_to_frac(params.get("atr_sl_max_pct", 6.0), 6.0)
tp_min = _pct_to_frac(params.get("atr_tp_min_pct", 0.5), 0.5)
tp_max = _pct_to_frac(params.get("atr_tp_max_pct", 5.0), 5.0)
sl_pct = min(max(sl_pct, sl_min), sl_max)
tp_pct = min(max(tp_pct, tp_min), tp_max)
stop_p = entry_price * (1.0 - sl_pct)
target_p = entry_price * (1.0 + tp_pct)
return stop_p, target_p
def _load_tail_pattern_params_from_row(r: Optional[Dict[str, Any]]) -> Dict[str, Any]:
"""반전 패턴 ON/OFF 및 형태 임계값 — env ``TAIL_PATTERN_*`` 단일 소스."""
row = r or {}
return {
"pattern_hammer": tail_env_bool(row, "TAIL_PATTERN_HAMMER", True),
"pattern_pin": tail_env_bool(row, "TAIL_PATTERN_PIN", False),
"pattern_engulfing": tail_env_bool(row, "TAIL_PATTERN_ENGULFING", False),
"pattern_piercing": tail_env_bool(row, "TAIL_PATTERN_PIERCING", False),
"pattern_harami": tail_env_bool(row, "TAIL_PATTERN_HARAMI", False),
"pattern_doji": tail_env_bool(row, "TAIL_PATTERN_DOJI", False),
"pattern_morning_star": tail_env_bool(row, "TAIL_PATTERN_MORNING_STAR", False),
"pin_close_upper_ratio": tail_env_float(row, "TAIL_PIN_CLOSE_UPPER_RATIO", 0.66),
"pin_max_upper_tail_ratio": tail_env_float(row, "TAIL_PIN_MAX_UPPER_TAIL_RATIO", 0.35),
"engulf_min_body_ratio": tail_env_float(row, "TAIL_ENGULF_MIN_BODY_RATIO", 1.0),
"piercing_penetrate_ratio": tail_env_float(row, "TAIL_PIERCING_PENETRATE_RATIO", 0.5),
"harami_max_body_ratio": tail_env_float(row, "TAIL_HARAMI_MAX_BODY_RATIO", 0.5),
"doji_body_max_ratio": tail_env_float(row, "TAIL_DOJI_BODY_MAX_RATIO", 0.15),
"morning_star_body_max_ratio": tail_env_float(row, "TAIL_MORNING_STAR_BODY_MAX_RATIO", 0.35),
"candle_lookback": tail_env_int(row, "TAIL_CANDLE_LOOKBACK", 3),
}
def _bar_ohlc(c: Dict) -> Tuple[float, float, float, float]:
return float(c["open"]), float(c["high"]), float(c["low"]), float(c["close"])
def _body_parts(op: float, hi: float, lo: float, cl: float) -> Dict[str, float]:
"""봉 몸통·꼬리 분해 (도지: 몸통 0 이면 range 기준 최소값)."""
body_top = max(op, cl)
body_bot = min(op, cl)
body_len = body_top - body_bot
rng = hi - lo if hi > lo else 0.0
if body_len <= 0:
body_len = max(rng * 0.001, 1e-6) if rng > 0 else 1.0
lower_tail = max(0.0, body_bot - lo) if lo > 0 else 0.0
upper_tail = max(0.0, hi - body_top) if hi > 0 else 0.0
return {
"body_top": body_top,
"body_bot": body_bot,
"body_len": body_len,
"range": rng,
"lower_tail": lower_tail,
"upper_tail": upper_tail,
"is_bullish": cl >= op,
"is_bearish": cl < op,
}
def _hammer_tail_metrics(
op: float, hi: float, lo: float, cl: float, lo_ref: float,
) -> Tuple[float, float, float]:
parts = _body_parts(op, hi, lo, cl)
tail_len = parts["lower_tail"]
body_len = parts["body_len"]
lo_use = lo if lo > 0 else lo_ref
tail_ratio = tail_len / body_len if body_len > 0 else 0.0
tail_pct = tail_len / lo_use if lo_use > 0 and tail_len > 0 else 0.0
return tail_ratio, tail_pct, lo_use
def _detect_hammer_pattern(
candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
"""망치(하단 꼬리) — 신호봉 또는 lookback 봉에서 탐색."""
rsi_period = int(params.get("rsi_period", 14))
lookback = int(params.get("candle_lookback", 3))
tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
tail_pct_min = float(params.get("tail_pct_min", 0.003))
lo_ref = float(candles[i]["low"]) if float(candles[i]["low"]) > 0 else float(candles[i]["high"])
for j in range(i, max(i - lookback, rsi_period) - 1, -1):
op, hi, lo, cl = _bar_ohlc(candles[j])
body_top = max(op, cl)
body_bot = min(op, cl)
body_len = body_top - body_bot if body_top > body_bot else 1.0
tail_len = body_bot - lo if lo > 0 else 0.0
lo_use = lo if lo > 0 else lo_ref
if tail_len <= 0:
continue
tail_ratio = tail_len / body_len if body_len > 0 else 0.0
tail_pct = tail_len / lo_use if lo_use > 0 else 0.0
if tail_ratio >= tail_ratio_min and tail_pct >= tail_pct_min:
return True, {
"pattern": "hammer",
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"pattern_bar_idx": j,
}
return False, {}
def _detect_pin_pattern(
candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
"""스트릭트 핀바 — 하단 꼬리 + 종가 상단 1/3 + 윗꼬리 짧음."""
rsi_period = int(params.get("rsi_period", 14))
lookback = int(params.get("candle_lookback", 3))
tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
tail_pct_min = float(params.get("tail_pct_min", 0.003))
pin_close_upper = float(params.get("pin_close_upper_ratio", 0.66))
pin_upper_max = float(params.get("pin_max_upper_tail_ratio", 0.35))
lo_ref = float(candles[i]["low"]) if float(candles[i]["low"]) > 0 else float(candles[i]["high"])
for j in range(i, max(i - lookback, rsi_period) - 1, -1):
op, hi, lo, cl = _bar_ohlc(candles[j])
parts = _body_parts(op, hi, lo, cl)
if parts["range"] <= 0:
continue
tail_ratio, tail_pct, _ = _hammer_tail_metrics(op, hi, lo, cl, lo_ref)
if tail_ratio < tail_ratio_min or tail_pct < tail_pct_min:
continue
close_pos = (cl - lo) / parts["range"]
if close_pos < pin_close_upper:
continue
if parts["body_len"] > 0 and parts["upper_tail"] / parts["body_len"] > pin_upper_max:
continue
return True, {
"pattern": "pin",
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"pattern_bar_idx": j,
"close_pos": close_pos,
}
return False, {}
def _detect_engulfing_pattern(
candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
"""불리시 장악형 — 전봉 음봉 몸통을 현재 양봉이 완전 삼킴."""
if i < 1:
return False, {}
engulf_min = float(params.get("engulf_min_body_ratio", 1.0))
po, ph, pl, pc = _bar_ohlc(candles[i - 1])
co, ch, cl, cc = _bar_ohlc(candles[i])
prev = _body_parts(po, ph, pl, pc)
curr = _body_parts(co, ch, cl, cc)
if not prev["is_bearish"] or not curr["is_bullish"]:
return False, {}
if co > pc or cc < po:
return False, {}
if curr["body_len"] < prev["body_len"] * engulf_min:
return False, {}
tail_ratio, tail_pct, _ = _hammer_tail_metrics(co, ch, cl, cc, pl if pl > 0 else ch)
return True, {
"pattern": "engulfing",
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"pattern_bar_idx": i,
}
def _detect_piercing_pattern(
candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
"""관통형 — 갭다운 후 전봉 몸통 중간 이상 회복(완전 장악 전)."""
if i < 1:
return False, {}
penetrate = float(params.get("piercing_penetrate_ratio", 0.5))
po, ph, pl, pc = _bar_ohlc(candles[i - 1])
co, ch, cl, cc = _bar_ohlc(candles[i])
prev = _body_parts(po, ph, pl, pc)
curr = _body_parts(co, ch, cl, cc)
if not prev["is_bearish"] or not curr["is_bullish"]:
return False, {}
if co >= pc:
return False, {}
midpoint = prev["body_bot"] + prev["body_len"] * penetrate
if cc <= midpoint:
return False, {}
if cc >= po:
return False, {}
tail_ratio, tail_pct, _ = _hammer_tail_metrics(co, ch, cl, cc, pl if pl > 0 else ch)
return True, {
"pattern": "piercing",
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"pattern_bar_idx": i,
}
def _detect_harami_pattern(
candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
"""불리시 하라미 — 큰 음봉 안에 작은 양봉(몸통 포함)."""
if i < 1:
return False, {}
harami_max = float(params.get("harami_max_body_ratio", 0.5))
po, ph, pl, pc = _bar_ohlc(candles[i - 1])
co, ch, cl, cc = _bar_ohlc(candles[i])
prev = _body_parts(po, ph, pl, pc)
curr = _body_parts(co, ch, cl, cc)
if not prev["is_bearish"]:
return False, {}
if curr["body_top"] > prev["body_top"] or curr["body_bot"] < prev["body_bot"]:
return False, {}
if curr["body_len"] > prev["body_len"] * harami_max:
return False, {}
if not curr["is_bullish"]:
return False, {}
tail_ratio, tail_pct, _ = _hammer_tail_metrics(co, ch, cl, cc, pl if pl > 0 else ch)
return True, {
"pattern": "harami",
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"pattern_bar_idx": i,
}
def _detect_doji_pattern(
candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
"""저점 도지 — 몸통 극소 + 하단 꼬리(망치형) + 종가 중상단."""
doji_max = float(params.get("doji_body_max_ratio", 0.15))
tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
tail_pct_min = float(params.get("tail_pct_min", 0.003))
pin_close_upper = float(params.get("pin_close_upper_ratio", 0.66))
op, hi, lo, cl = _bar_ohlc(candles[i])
parts = _body_parts(op, hi, lo, cl)
if parts["range"] <= 0:
return False, {}
real_body = abs(cl - op)
if real_body / parts["range"] > doji_max:
return False, {}
tail_ratio, tail_pct, _ = _hammer_tail_metrics(op, hi, lo, cl, lo if lo > 0 else hi)
if tail_ratio < tail_ratio_min or tail_pct < tail_pct_min:
return False, {}
close_pos = (cl - lo) / parts["range"]
if close_pos < pin_close_upper:
return False, {}
return True, {
"pattern": "doji",
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"pattern_bar_idx": i,
"close_pos": close_pos,
}
def _detect_morning_star_pattern(
candles: List[Dict], i: int, params: Dict[str, Any],
) -> Tuple[bool, Dict[str, Any]]:
"""샛별형(3봉) — 장음봉 + 작은 별 + 양봉이 1봉 몸통 중간 돌파."""
if i < 2:
return False, {}
star_max = float(params.get("morning_star_body_max_ratio", 0.35))
o0, h0, l0, c0 = _bar_ohlc(candles[i - 2])
o1, h1, l1, c1 = _bar_ohlc(candles[i - 1])
o2, h2, l2, c2 = _bar_ohlc(candles[i])
bear = _body_parts(o0, h0, l0, c0)
star = _body_parts(o1, h1, l1, c1)
bull = _body_parts(o2, h2, l2, c2)
if not bear["is_bearish"] or bear["body_len"] <= 0:
return False, {}
if star["body_len"] > bear["body_len"] * star_max:
return False, {}
midpoint = (bear["body_top"] + bear["body_bot"]) / 2.0
if not bull["is_bullish"] or c2 <= midpoint:
return False, {}
tail_ratio, tail_pct, _ = _hammer_tail_metrics(o2, h2, l2, c2, l2 if l2 > 0 else h2)
return True, {
"pattern": "morning_star",
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"pattern_bar_idx": i,
}
def eval_tail_reversal_pattern(
candles: List[Dict],
i: int,
params: Dict[str, Any],
) -> Tuple[bool, str, Dict[str, Any]]:
"""
TRIGGER 반전 패턴 OR 평가.
``TAIL_PATTERN_*`` 가 모두 OFF 이면 망치만(기존 동작) 검사.
"""
checks = []
if _to_bool(params.get("pattern_hammer"), True):
checks.append(("hammer", _detect_hammer_pattern))
if _to_bool(params.get("pattern_pin"), False):
checks.append(("pin", _detect_pin_pattern))
if _to_bool(params.get("pattern_engulfing"), False):
checks.append(("engulfing", _detect_engulfing_pattern))
if _to_bool(params.get("pattern_piercing"), False):
checks.append(("piercing", _detect_piercing_pattern))
if _to_bool(params.get("pattern_harami"), False):
checks.append(("harami", _detect_harami_pattern))
if _to_bool(params.get("pattern_doji"), False):
checks.append(("doji", _detect_doji_pattern))
if _to_bool(params.get("pattern_morning_star"), False):
checks.append(("morning_star", _detect_morning_star_pattern))
if not checks:
checks.append(("hammer", _detect_hammer_pattern))
for name, fn in checks:
ok, metrics = fn(candles, i, params)
if ok:
return True, name, metrics
active = ",".join(n for n, _ in checks)
return False, active, {}
def _tail_volume_spike_ok(
candles: List[Dict],
i: int,
params: Dict[str, Any],
) -> Tuple[bool, str]:
"""
TRIGGER — 신호봉 거래량이 직전 N봉 평균 대비 배수 이상인지 (모멘텀 mom_vol_mult 와 동일 패턴).
``tail_vol_mult`` ≤ 0 이면 OFF (기존 백테·실매 동작 유지).
"""
vol_mult = float(params.get("tail_vol_mult", 0) or 0)
vol_win = int(params.get("tail_vol_win", 5) or 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 _eval_tail_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]]]:
"""신호봉 인덱스 ``i`` 에서 꼬리잡기 매수 조건 평가 (백테스트 루프와 동일 시점)."""
if i < 19 or i >= len(candles):
return ("탈락-데이터", f"인덱스 부적절 (i={i})", None)
c = candles[i]
day = c["candle_time"][:8]
hm = int(c["candle_time"][8:12])
op = float(c["open"])
hi = float(c["high"])
lo = float(c["low"])
cl = float(c["close"])
# 엔진 파라미터 로드
min_drop_rate = float(params.get("min_drop_rate", 0.03))
min_recovery_ratio = float(params.get("min_recovery_ratio", 0.5))
max_rec_3m = float(params.get("max_rec_3m", 0.8))
tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
tail_pct_min = float(params.get("tail_pct_min", 0.003))
rsi_period = int(params.get("rsi_period", 14))
rsi_threshold = float(params.get("rsi_threshold", 78))
high_chase_thr = float(params.get("high_chase_thr", 0.96))
time_start_hm = int(params.get("time_start_hm", 930))
time_end_hm = int(params.get("time_end_hm", 1500))
cooldown_min = float(params.get("cooldown_min", 15))
max_daily = int(params.get("max_daily", 3))
# 방어 로직 파라미터
min_price = float(params.get("min_price", 1000.0))
max_daily_change = float(params.get("max_daily_change", 20.0))
ma20_max_above = float(params.get("ma20_max_above", 3.0))
# SCAN/TRIGGER 분리 — 조건검색 유니버스 사용 시 HTS A/B/C 중복 필터 생략
skip_hts = _to_bool(params.get("skip_hts_scan_dupes"), True)
use_intraday_drop = _to_bool(params.get("use_intraday_drop"), False)
use_ma20 = _to_bool(params.get("use_ma20_filter"), False)
use_rsi = _to_bool(params.get("use_rsi_filter"), True)
use_daily_range = _to_bool(params.get("use_daily_range_filter"), True)
use_high_chase = _to_bool(params.get("use_high_chase_filter"), True)
bar_chg_min_pct = float(params.get("bar_chg_min_pct", -10.0))
bar_chg_max_pct = float(params.get("bar_chg_max_pct", -1.5))
if hm < time_start_hm or hm > time_end_hm:
return (None, None, None) # 시간대 탈락
if state.get("daily_cnt", 0) >= max_daily:
return (None, None, 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 cl <= 0 or cl < min_price:
return ("탈락-가격", f"현재가 부적절 (현재 {cl:,.0f}원, 최소 {min_price:,.0f}원)", None)
# 당일 누적 OHLC 및 피뢰침 검사
running_open = op
running_high = hi
running_low = lo if lo > 0 else hi
for j in range(i - 1, -1, -1):
if candles[j]["candle_time"][:8] != day:
break
running_open = float(candles[j]["open"])
running_high = max(running_high, float(candles[j]["high"]))
lj = float(candles[j]["low"])
if lj > 0:
running_low = min(running_low, lj)
if cl <= 0 or running_open <= 0:
return (None, None, None)
# 일일 변동폭(피뢰침) 검사 — TRIGGER 선택 필터
if use_daily_range and running_low > 0:
range_change_pct = (running_high - running_low) / running_low * 100
if range_change_pct > max_daily_change:
return ("탈락-피뢰침 급등주", f"일일 변동폭 {range_change_pct:.1f}% > {max_daily_change:.0f}%", None)
# HTS A (봉 등락률) — 조건검색 미사용(전종목) 백테 시에만 TRIGGER 에서 재검사
if not skip_hts and i >= 1:
prev_cl = float(candles[i - 1]["close"])
if prev_cl > 0:
bar_chg = (cl - prev_cl) / prev_cl * 100.0
if bar_chg < bar_chg_min_pct or bar_chg > bar_chg_max_pct:
return (
"탈락-봉등락",
f"봉등락 {bar_chg:.2f}% (HTS A: {bar_chg_min_pct:.1f}~{bar_chg_max_pct:.1f}%)",
None,
)
# 당일 시가→저점 낙폭 — HTS A 와 다른 축; ``TAIL_USE_INTRADAY_DROP=true`` 일 때만
if use_intraday_drop:
drop = (running_open - running_low) / running_open
if drop < min_drop_rate:
return (
"탈락-낙폭",
f"낙폭 {drop*100:.2f}% < {min_drop_rate*100:.1f}% (시가 {running_open:,.0f} → 저점 {running_low:,.0f})",
None,
)
# 회복률 검사
day_range = running_high - running_low
rec_day = (cl - running_low) / day_range if day_range > 0 else 0
if rec_day < min_recovery_ratio:
return (
"탈락-회복률",
f"회복률 {rec_day*100:.1f}% < {min_recovery_ratio*100:.0f}% (저점 {running_low:,.0f} → 현재 {cl:,.0f})",
None,
)
# 반전 패턴 OR (망치·핀바·장악·관통·하라미·도지·샛별) — env TAIL_PATTERN_* 로 ON/OFF
pat_ok, pat_active, pat_metrics = eval_tail_reversal_pattern(candles, i, params)
if not pat_ok:
return (
"탈락-패턴",
f"반전패턴 미충족 (활성: {pat_active})",
None,
)
tail_ratio = float(pat_metrics.get("tail_ratio", 0.0))
tail_pct = float(pat_metrics.get("tail_pct", 0.0))
pattern_name = str(pat_metrics.get("pattern", pat_active))
# TRIGGER — 신호봉 거래량 폭증 (HTS C는 스캔 시점; 매수 직전 재확인)
vol_ok, vol_msg = _tail_volume_spike_ok(candles, i, params)
if not vol_ok:
return ("탈락-거래량", f"신호봉 거래량 {vol_msg}", None)
# 3분봉 내 회복 위치 상한 검사
c_range = hi - lo if hi > lo else 0
rec_3m = (cl - lo) / c_range if c_range > 0 else 0
if not (min_recovery_ratio <= rec_3m <= max_rec_3m):
return (
"탈락-회복3분",
f"3분봉 회복률 {rec_3m*100:.1f}% (기준 {min_recovery_ratio*100:.0f}~{max_rec_3m*100:.0f}%)",
None,
)
# 고점 추격 방지 — TRIGGER 선택 필터
if use_high_chase and cl >= running_high * high_chase_thr:
return (
"탈락-피뢰침 고점추격",
f"현재가 {cl:,.0f} ≥ 고점대비 {high_chase_thr*100:.0f}%",
None,
)
# RSI 검사 — TRIGGER 선택 필터 (HTS B 체결강도와 별개)
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(closes, rsi_period)
rsi_val = rsis[i] if i < len(rsis) else None
if use_rsi and (rsi_val is None or rsi_val >= rsi_threshold):
return (
"탈락-RSI",
(f"RSI {rsi_val:.1f}" if rsi_val is not None else "RSI None") + f" ≥ {rsi_threshold:.0f}",
None,
)
# MA20 방어 — TRIGGER 선택 필터 (기본 OFF: 역배열 컷이 신호를 과하게 줄임)
if use_ma20:
ma20 = sum(closes[i - 19:i + 1]) / 20.0
if cl < ma20:
return ("탈락-MA20", f"현재가 {cl:,.0f} < MA20 {ma20:,.0f} (역배열)", None)
if ma20 > 0 and cl > ma20 * (1 + ma20_max_above / 100):
return ("탈락-MA20초과", f"MA20 대비 {ma20_max_above:.0f}% 이격 초과", None)
sig = {
"signal": True,
"pattern": pattern_name,
"tail_ratio": tail_ratio,
"tail_pct": tail_pct,
"recovery_pos": rec_3m,
"rsi_val": rsi_val,
"atr_calc_val": None,
"signal_candle_time": c.get("candle_time"),
}
return _tail_signal_with_whipsaw(params, candles, i, sig)
def _tail_signal_with_whipsaw(
params: Dict[str, Any],
candles: List[Dict],
i: int,
sig: Dict[str, Any],
) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
"""꼬리 신호 dict 반환 직전 — 휩쏘(TRIGGER) 공통 (실매·백테·파라서치)."""
c = candles[i]
cl = float(c.get("close") or 0)
ws_rej, ws_msg = whipsaw_reject_for_signal(
params, "TAIL",
signal_bar=c,
current_price=cl,
)
if ws_rej:
return (ws_rej, ws_msg, None)
ob_rej, ob_msg = orderbook_reject_for_entry(
params, "TAIL", current_price=cl,
)
if ob_rej:
return (ob_rej, ob_msg, None)
prog_rej, prog_msg = program_reject_for_entry(
params, "TAIL", current_price=cl,
)
if prog_rej:
return (prog_rej, prog_msg, None)
return (None, None, sig)
def check_buy_signal_live(
candles: List[Dict],
params: Dict[str, Any],
state: Dict[str, Any],
) -> tuple:
"""
실시간 꼬리잡기 매수 신호.
- ``live_backtest_align=True`` (기본): 백테와 동일 — **직전 확정봉=신호봉**,
**현재 확정봉=진입봉** (신호봉 조건 충족 후 다음 3분봉 시가 진입).
- ``live_backtest_align=False``: 구버전 — 마지막 확정봉 1개만 검사.
candles: 3분봉 (candle_time, open, high, low, close, volume)
state: { "last_exit_dt": datetime|None, "daily_cnt": int }
"""
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) < 20:
return ("탈락-데이터", f"확정봉 부족 (len={len(confirmed)} < 20)", 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 < 19:
break
reject, msg, sig = _eval_tail_buy_at_index(
confirmed, signal_i, params, state,
)
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")
return (None, None, sig)
return last_reject
i = len(confirmed) - 1
return _eval_tail_buy_at_index(confirmed, i, params, state)
def _tail_ratchet_tiers(params: Dict[str, Any]) -> List[Tuple[float, float]]:
"""래칫 단계 트레일 — 돌파 ``_breakout_ratchet_tiers`` 와 동일 형식.
env/params ``"2:1.5,5:1.0"`` → +2% 수익부터 고점 대비 1.5% 되돌림컷.
비어 있으면 [] (= OFF, 단일 어깨컷 사용).
"""
raw = params.get("ratchet_tiers")
if raw is None:
from kis_trader.utils.env import get_env_from_db
raw = get_env_from_db("TAIL_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 _tail_max_hold_minutes(params: Dict[str, Any]) -> int:
"""시간컷: 최대 보유 **분** (돌파 BREAKOUT_MAX_HOLD_BARS 와 동일 — 이름은 bars, 단위는 분). 0=OFF."""
v = params.get("max_hold_bars")
if v is None:
from kis_trader.utils.env import get_env_int
v = get_env_int("TAIL_MAX_HOLD_BARS", 0)
try:
return max(0, int(float(v)))
except (TypeError, ValueError):
return 0
def _tail_minutes_held(position: Dict[str, Any], current_candle: Dict[str, Any]) -> Optional[int]:
"""진입 이후 경과 분 — 시간컷 판정."""
et = position.get("entry_time") or position.get("buy_time")
ct = current_candle.get("candle_time")
if not et or not ct:
return None
try:
e_dt = _t2dt(str(et)[:12])
c_dt = _t2dt(str(ct)[:12])
return max(0, int((c_dt - e_dt).total_seconds() // 60))
except Exception:
return None
def check_sell_signal_live(
position: Dict[str, Any],
current_candle: Dict[str, Any],
params: Dict[str, Any],
is_eod: bool = False,
) -> Optional[tuple]:
"""
실시간 및 백테스트 공통 청산 조건.
[V4]: 트레일(어깨) 저가 기준 · 수익 구간 우선 · ATR 캡 익절/손절 · 금액손실컷(트레일 미발동만).
position: entry_price, entry_time(YYYYMMDDHHMM), stop, target, max_price, qty
current_candle: high, low, close, candle_time
반환: (reason_str, exit_price) 또는 None
"""
shoulder_min_high = float(params.get("shoulder_min_high", 0.005))
shoulder_cut_pct = float(params.get("shoulder_cut_pct", 0.003))
max_loss_krw = int(params.get("max_loss_krw", 200000))
min_hold_sec = float(params.get("min_hold_sec", 30.0))
hi = float(current_candle.get("high", current_candle["close"]))
lo = float(current_candle.get("low", current_candle["close"]))
cl = float(current_candle["close"])
candle_time = current_candle.get("candle_time", "")
max_p = max(float(position.get("max_price", 0) or 0), hi)
ep = float(position["entry_price"])
stop = float(position["stop"])
target = float(position["target"])
qty = int(position.get("qty", 1) or 1)
# 최소 보유 시간 검사 (너무 짧으면 청산 무시)
if candle_time and position.get("entry_time"):
try:
entry_dt = _t2dt(position["entry_time"])
curr_dt = _t2dt(candle_time)
if (curr_dt - entry_dt).total_seconds() < min_hold_sec:
return None
except Exception:
pass
reason = None
exit_price = cl
profit_val = (lo - ep) * qty
drop_pct = (ep - lo) / ep if ep > 0 else 0.0
min_drop_pct = float(params.get("min_drop_pct_for_loss_cut", 0.015))
# 1순위: 트레일 — 돌파와 동일: 래칫 문자열 있으면 **단일 어깨 대체**, 없으면 shoulder_min/cut
ratchet_tiers = _tail_ratchet_tiers(params)
trail_armed = False
if ratchet_tiers and ep > 0:
peak_gain = (max_p - ep) / ep
cut_ratio = 0.0
for gain_thr, cut in ratchet_tiers:
if peak_gain >= gain_thr:
cut_ratio = cut
if cut_ratio > 0.0:
trail_armed = True
ratchet_line = max_p * (1.0 - cut_ratio)
if lo > 0 and lo <= ratchet_line:
reason = "래칫컷"
exit_price = ratchet_line
else:
trail_armed = ep > 0 and max_p >= ep * (1.0 + shoulder_min_high)
trail_stop_px = max_p * (1.0 - shoulder_cut_pct) if trail_armed else 0.0
if trail_armed and lo > 0 and lo <= trail_stop_px:
reason = "어깨컷"
exit_price = trail_stop_px
# 2순위: ATR 캡 익절
if not reason and hi >= target:
reason = "익절"
exit_price = target
# 3순위: ATR 캡 손절
if not reason and lo > 0 and lo <= stop:
reason = "손절"
exit_price = stop
# 4순위: 보조 트레일링 — 래칫/어깨(1순위)와 별도. TAIL_TRAIL_PCT>0 일 때만 (돌파 BREAKOUT_TRAIL_* 와 동일)
# trail_arm_pct: 고점이 진입×(1+arm) 도달 후에만 무장 (0=즉시 무장). 어깨 미발동 구간의 팝 후 되돌림 보호.
if not reason:
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))
if trail_pct > 0 and ep > 0:
trail_arm_line = ep * (1.0 + trail_arm_pct)
if max_p > ep and max_p >= trail_arm_line:
trail_line = max_p * (1.0 - trail_pct)
if lo > 0 and lo <= trail_line:
reason = "트레일컷"
exit_price = trail_line
# 5순위: 금액 손실컷 — 1순위 어깨/래칫 미발동(수익% 문턱 못 넘음)일 때만
if (
not reason
and not trail_armed
and profit_val <= -max_loss_krw
and drop_pct >= min_drop_pct
):
reason = "금액손실컷"
exit_price = ep - (max_loss_krw / qty) if qty > 0 else lo
# 6순위: 시간컷 — 돌파와 동일(분 단위), 기본 OFF. 어깨·손절·익절 뒤, 장마감 앞
if not reason:
max_hold_min = _tail_max_hold_minutes(params)
if max_hold_min > 0:
held = _tail_minutes_held(position, current_candle)
if held is not None and held >= max_hold_min:
reason = "시간컷"
exit_price = cl
# 7순위: 장 마감 강제 청산
if not reason and is_eod:
reason = "장마감"
exit_price = cl
if reason:
return (reason, exit_price)
return None
def resolve_tail_invest_amount_krw(params: Dict[str, Any]) -> int:
"""
실매 ``resolve_invest_amount_krw`` + ``TailCatchStrategy`` 종목당 상한과 동일.
1회 매수 금액 = slot_money (MAX_LOSS/sl 로 투자금을 줄이지 않음).
"""
slot = int(float(params.get("slot_money") or params.get("invest_amount") or 0))
if slot <= 0:
try:
from kis_trader.utils.env import get_env_int
slot = get_env_int("TAIL_SLOT_MONEY", 3_000_000)
except Exception:
slot = 3_000_000
hard_cap = int(float(params.get("short_max_buy_amount") or params.get("per_stock_cap") or 0))
if hard_cap <= 0:
try:
from kis_trader.utils.env import get_env_int
hard_cap = get_env_int("TAIL_MAX_BUY_AMOUNT", 0)
except Exception:
hard_cap = 0
if hard_cap > 0 and slot > hard_cap:
return hard_cap
return slot
def _portfolio_exposure_krw(portfolio: Dict[str, Dict[str, Any]]) -> float:
"""동시 보유 매입금 합 (실매 UPDOW 노출·SHORT 총한도와 동일 개념)."""
total = 0.0
for p in portfolio.values():
total += float(p.get("entry_price") or 0) * int(p.get("qty") or 1)
return total
def _tail_max_stocks_from_params(params: Dict[str, Any]) -> int:
for key in ("max_stocks", "short_max_stocks"):
v = params.get(key)
if v not in (None, "", 0):
return max(1, int(v))
try:
from kis_trader.utils.env import get_env_int
n = get_env_int("TAIL_MAX_STOCKS", 3)
return max(1, int(n))
except Exception:
return 3
def _tail_total_budget_from_params(params: Dict[str, Any]) -> float:
"""0 이하면 호출측에서 max_stocks×slot_money 로 유도."""
for key in ("total_budget_krw", "short_total_budget_krw"):
v = params.get(key)
if v not in (None, ""):
try:
return float(v)
except (TypeError, ValueError):
pass
try:
from kis_trader.utils.env import get_env_int
cap = get_env_int("TAIL_TOTAL_BUDGET_KRW", 0)
if cap > 0:
return float(cap)
except Exception:
pass
return 0.0
def _tail_min_invest_ratio_of_slot(params: Dict[str, Any]) -> float:
"""1회 slot 대비 최소 투입 비율 — 미만이면 매수 스킵 (실매: 풀 slot 주문 후 예수금 부족 시만 축소)."""
v = params.get("min_invest_ratio_of_slot")
if v not in (None, ""):
return max(0.01, min(1.0, float(v)))
try:
from kis_trader.utils.env import get_env_float
return max(0.01, min(1.0, get_env_float("TAIL_MIN_INVEST_RATIO_OF_SLOT", 0.90)))
except Exception:
return 0.90
def _tail_target_qty_and_cost(entry_price: float, slot_money: float) -> Tuple[int, float]:
"""실매와 동일: 1회 투자금(slot) 기준 목표 수량·매입금."""
from kis_trader.utils.position_sizing import invest_qty_for_price
qty = invest_qty_for_price(entry_price, slot_money)
if qty < 1:
return 0, 0.0
return qty, qty * entry_price
def _vol_capped_qty(qty: int, candle: Dict[str, Any], cap_pct: float) -> int:
"""진입봉 거래량×cap_pct% 로 체결 가능 주수 제한 (실매 유동성 제약 근사).
cap_pct<=0 → 제한 없음. 거래량 정보 없으면(0) 기존 동작 유지(보수적). 반환 0=미체결.
"""
if cap_pct <= 0:
return qty
try:
vol = int(float(candle.get("volume") or 0))
except (TypeError, ValueError):
vol = 0
if vol <= 0:
return qty
fillable = int(vol * cap_pct / 100.0)
return min(qty, fillable)
def _buy_priority_key(
code: str,
uni_codes: Optional[List[str]],
) -> Tuple[int, str]:
"""실매 후보 순회 순서 근사 — 유니버스 편입 순서(index), 없으면 code 정렬."""
if uni_codes is None:
return (0, code)
try:
return (uni_codes.index(code), code)
except ValueError:
return (999999, code)
def _universe_codes_at(
t: str,
slot_key: str,
universe_timeline: Optional[Any],
universe_by_slot: Optional[Dict[str, List[str]]],
) -> Optional[List[str]]:
"""그 시각(봉 마감초) 유효 유니버스 코드 리스트 (돌파·모멘텀 공통 로직).
- ``universe_timeline`` (초단위, 실매 get_universe_at 정합) 우선 — 봉 마감(HH:MM:59)
직전 최신 조건검색 스냅샷. strict lag(1분 지연) 없이 실매와 동일 시점 조회.
- 없으면 1분 슬롯(``universe_by_slot``) 폴백. 둘 다 없으면 None(전종목·필터없음).
"""
if universe_timeline is not None:
return universe_timeline.codes_at(str(t)[:12] + "59")
if universe_by_slot is not None:
return universe_by_slot.get(slot_key, [])
return None
def run_tail_backtest_portfolio(
candles_by_code: 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]:
"""
시각순 포트폴리오 백테스트 — 실매 BaseStrategy 제약 근사.
- 모든 종목 봉을 ``candle_time`` 순으로 처리 (종목별 독립 합산 아님)
- ``max_stocks``: 동시 보유 종목 수 (SHORT_MAX_STOCKS / MAX_STOCKS)
- ``total_budget_krw``: 동시 보유 매입금 합 상한 (0 → max_stocks×slot_money)
- 1시각(봉)당 신규 매수 1건 (실매 1루프 1매수)
- 1회 투자금 = ``slot_money`` (실매 resolve_invest_amount_krw)
"""
rsi_period = int(params.get("rsi_period", 14))
min_bars = rsi_period + 5
max_stocks = _tail_max_stocks_from_params(params)
slot_money = resolve_tail_invest_amount_krw(params)
total_budget = _tail_total_budget_from_params(params)
if total_budget <= 0:
total_budget = float(max_stocks * slot_money)
min_invest_ratio = _tail_min_invest_ratio_of_slot(params)
skipped_micro_buys = 0
# 체결량 상한(유동성 제약 근사) — 진입봉 거래량×N% 까지만 체결. 0=OFF(동작 불변).
# 실매 시장가 IOC 가 호가에 있는 만큼만 체결되는 것을 봉단위로 근사 (소형주 미체결↑).
vol_fill_cap_pct = float(params.get("backtest_vol_fill_cap_pct", 0) or 0)
skipped_vol_unfilled = 0
tick_tf = tail_timeframe_min(params)
use_ticks = bool(ticks_by_code) and tail_backtest_wants_tick_replay(params)
tick_fill_stats = {"ws_ticks": 0, "ohlc_low": 0, "ohlc_open": 0}
attach_indicator_caches_to_params(params, candles_by_code)
# 종목별 컨텍스트
ctx_by_code: Dict[str, Dict[str, Any]] = {}
all_times_set = set()
for code, raw_rows in candles_by_code.items():
if len(raw_rows) < min_bars:
continue
candles = [dict(r) for r in raw_rows]
closes = [float(c["close"]) for c in candles]
ctx_by_code[code] = {
"code": code,
"candles": candles,
"closes": closes,
"rsis": compute_rsi_series(closes, rsi_period),
"ma20s": compute_sma_series(closes, 20),
"atrs": compute_atr_series(candles, 14),
"time_index": {c["candle_time"]: idx for idx, c in enumerate(candles)},
"last_exit_dt": {},
"daily_cnt": {},
"pending_entry": None,
"pending_limit": None,
}
for c in candles:
all_times_set.add(c["candle_time"])
all_times = sorted(all_times_set)
portfolio: Dict[str, Dict[str, Any]] = {}
all_trades: List[Dict] = []
# 초단위 유니버스 타임라인 (실매 get_universe_at 정합, 돌파·모멘텀 공통). 없으면 1분 슬롯 폴백.
universe_timeline = params.get("_universe_timeline")
for t in all_times:
slot_key = _slot_key(t, params.get("scan_interval_min", 1))
uni_codes = _universe_codes_at(t, slot_key, universe_timeline, universe_by_slot)
uni_set = set(uni_codes) if uni_codes is not None else None
# ── Phase 0a: ATR 지정가 체결 (유효 봉 low ≤ limit) ──
for code, ctx in ctx_by_code.items():
pl = ctx.get("pending_limit")
if not pl or code in portfolio:
continue
idx = ctx["time_index"].get(t)
if idx is None:
continue
sig_i = int(pl.get("signal_i", -1))
if idx <= sig_i:
continue
vu = str(pl.get("valid_until") or "")[:12]
if vu and str(t)[:12] > vu:
ctx["pending_limit"] = None
continue
c = ctx["candles"][idx]
bar_ticks = (
collect_bar_ticks(ticks_by_code, code, t, tick_tf) if use_ticks else []
)
fill, fill_src = try_limit_fill_on_bar_with_ticks(
c,
float(pl.get("limit_price") or 0),
float(pl.get("fill_slip") or 0),
ticks=bar_ticks,
params=params,
)
if not fill or fill <= 0:
continue
if fill_src in tick_fill_stats:
tick_fill_stats[fill_src] += 1
if len(portfolio) >= max_stocks:
continue
exposure = _portfolio_exposure_krw(portfolio)
remaining = max(0.0, total_budget - exposure)
target_qty, target_cost = _tail_target_qty_and_cost(fill, float(slot_money))
min_required = target_cost * min_invest_ratio
if target_qty < 1 or remaining < min_required:
skipped_micro_buys += 1
continue
invest = min(float(slot_money), remaining, target_cost)
qty = int(invest / fill)
if qty < 1:
skipped_micro_buys += 1
continue
# 유동성 제약: 진입봉 거래량×cap% 까지만 체결 (실매 IOC 미체결 근사)
capped = _vol_capped_qty(qty, c, vol_fill_cap_pct)
if capped < qty:
if capped < 1:
skipped_vol_unfilled += 1
continue
qty = capped
cost = qty * fill
if cost < min_required or exposure + cost > total_budget + 1e-6:
skipped_micro_buys += 1
continue
portfolio[code] = {
"entry_price": fill,
"entry_time": t,
"stop": pl["stop"],
"target": pl["target"],
"max_price": fill,
"session_low": fill,
"qty": qty,
}
ctx["pending_limit"] = None
break
# ── Phase 0b: 예약 진입 (align — 다음 봉 시가) ──
pending_codes = [
code for code, ctx in ctx_by_code.items()
if ctx.get("pending_entry") and ctx["pending_entry"].get("entry_time") == t
]
pending_codes.sort(
key=lambda c: _buy_priority_key(c, uni_codes),
)
for code in pending_codes:
ctx = ctx_by_code[code]
pe = ctx.pop("pending_entry", None)
if not pe or code in portfolio:
continue
if len(portfolio) >= max_stocks:
break
entry_price = float(pe["entry_price"])
if use_ticks:
bar_ticks = collect_bar_ticks(ticks_by_code, code, t, tick_tf)
entry_price, align_src = align_entry_price_from_ticks(bar_ticks, entry_price)
if align_src in tick_fill_stats:
tick_fill_stats[align_src] += 1
if entry_price <= 0:
continue
exposure = _portfolio_exposure_krw(portfolio)
remaining = max(0.0, total_budget - exposure)
target_qty, target_cost = _tail_target_qty_and_cost(entry_price, float(slot_money))
min_required = target_cost * min_invest_ratio
if target_qty < 1 or remaining < min_required:
skipped_micro_buys += 1
continue
invest = min(float(slot_money), remaining, target_cost)
qty = int(invest / entry_price)
if qty < 1:
skipped_micro_buys += 1
continue
# 유동성 제약: 진입봉 거래량×cap% 까지만 체결 (실매 IOC 미체결 근사)
if vol_fill_cap_pct > 0:
_eidx = ctx["time_index"].get(t)
_ebar = ctx["candles"][_eidx] if _eidx is not None else {}
capped = _vol_capped_qty(qty, _ebar, vol_fill_cap_pct)
if capped < qty:
if capped < 1:
skipped_vol_unfilled += 1
continue
qty = capped
cost = qty * entry_price
if cost < min_required:
skipped_micro_buys += 1
continue
if exposure + cost > total_budget + 1e-6:
skipped_micro_buys += 1
continue
portfolio[code] = {
"entry_price": entry_price,
"entry_time": t,
"stop": pe["stop"],
"target": pe["target"],
"max_price": entry_price,
"session_low": entry_price,
"qty": qty,
}
break # 1시각 1매수
# ── Phase 1: 보유 종목 청산 ──
for code in list(portfolio.keys()):
ctx = ctx_by_code.get(code)
if ctx is None:
continue
idx = ctx["time_index"].get(t)
if idx is None:
continue
candles = ctx["candles"]
c = candles[idx]
day = t[:8]
hi = float(c["high"])
lo = float(c["low"])
cl = float(c["close"])
op = float(c["open"])
is_eod = is_strategy_eod_bar(candles[idx]["candle_time"], params, "TAIL")
pos = portfolio[code]
max_p = max(float(pos.get("max_price", 0) or 0), hi)
hp = float(c.get("holding_peak") or 0)
if hp > 0:
max_p = max(max_p, hp)
pos["max_price"] = max_p
cur_c_info = {
"open": op,
"high": hi,
"low": lo,
"close": cl,
"candle_time": t,
}
res = check_sell_signal_backtest_bar(
pos,
cur_c_info,
params,
is_eod=is_eod,
sell_fn=check_sell_signal_live,
low_mode="session_low",
)
if not res:
continue
reason, exit_price = res
all_trades.append({
"code": code,
"entry_time": pos["entry_time"],
"exit_time": t,
"entry": round(pos["entry_price"]),
"exit": round(exit_price),
"pnl": 0,
"reason": reason,
"hold_min": 0,
"peak_price": round(max_p),
"qty": pos.get("qty", 1),
})
ctx["last_exit_dt"][day] = _t2dt(t)
ctx["daily_cnt"][day] = ctx["daily_cnt"].get(day, 0) + 1
del portfolio[code]
# ── Phase 2: 신규 매수 신호 (다음 봉 시가 진입 예약) ──
if len(portfolio) >= max_stocks:
continue
exposure = _portfolio_exposure_krw(portfolio)
if exposure >= total_budget - 1e-6:
continue
time_start_hm = int(params.get("time_start_hm", 930))
time_end_hm = int(params.get("time_end_hm", 1500))
hm = int(t[8:12])
candidates: List[Tuple[Tuple[int, str], str, Dict[str, Any]]] = []
for code, ctx in ctx_by_code.items():
if code in portfolio or ctx.get("pending_entry") or ctx.get("pending_limit"):
continue
idx = ctx["time_index"].get(t)
if idx is None:
continue
candles = ctx["candles"]
c = candles[idx]
day = t[:8]
cl = float(c["close"])
if uni_set is not None:
if code not in uni_set:
continue
if cl <= 0 or hm < time_start_hm or hm > time_end_hm:
continue
if ctx["daily_cnt"].get(day, 0) >= int(params.get("max_daily", 3)):
continue
if day in ctx["last_exit_dt"]:
elapsed = (_t2dt(t) - ctx["last_exit_dt"][day]).total_seconds() / 60
if elapsed < float(params.get("cooldown_min", 15)):
continue
eval_params = dict(params)
ic = get_indicator_cache_from_params(params, code)
if ic is not None:
eval_params["_indicator_cache"] = ic
inject_whipsaw_ticks_into_params(
eval_params,
ticks_by_code=ticks_by_code,
code=code,
bar_candle_time=t,
strategy="TAIL",
tf_min=tick_tf,
)
inject_trigger_snapshots_into_params(
eval_params,
orderbook_by_code=orderbook_by_code,
program_by_code=program_by_code,
code=code,
bar_candle_time=t,
)
if universe_by_slot is not None:
eval_params.setdefault("skip_hts_scan_dupes", True)
else:
eval_params.setdefault("skip_hts_scan_dupes", False)
state = {
"daily_cnt": ctx["daily_cnt"].get(day, 0),
"last_exit_dt": ctx["last_exit_dt"].get(day),
}
reject, _msg, sig = _eval_tail_buy_at_index(candles, idx, eval_params, state)
if reject or not sig:
continue
atr = ctx["atrs"][idx] if ctx["atrs"][idx] is not None else cl * 0.01
pri = _buy_priority_key(code, uni_codes)
if is_limit_atr_entry(short_entry_mode(params)):
lp_cfg = tail_limit_params(params)
sig_bar = candles[idx]
anchor_px = resolve_limit_anchor_price(
lp_cfg["anchor"], sig_bar, candles, idx,
)
min_px = float(params.get("min_price", 1000.0))
limit_px = compute_atr_limit_price(
anchor_px, atr, lp_cfg["mult"], min_price=min_px,
)
if limit_px <= 0:
continue
stop_p, target_p = compute_tail_atr_prices(limit_px, atr, params)
vu = limit_valid_until_bar_key(candles, idx, lp_cfg["valid_bars"])
candidates.append((pri, code, {
"pending_limit": True,
"signal_i": idx,
"limit_price": limit_px,
"valid_until": vu,
"fill_slip": lp_cfg["fill_slip_pct"],
"stop": stop_p,
"target": target_p,
}))
continue
if idx + 1 >= len(candles):
continue
next_c = candles[idx + 1]
if next_c["candle_time"][:8] != day:
continue
entry_price = float(next_c["open"])
if entry_price <= 0:
entry_price = cl
stop_p, target_p = compute_tail_atr_prices(entry_price, atr, params)
candidates.append((pri, code, {
"entry_time": next_c["candle_time"],
"entry_price": entry_price,
"stop": stop_p,
"target": target_p,
}))
if not candidates:
continue
candidates.sort(key=lambda x: x[0])
_pri, pick_code, pe = candidates[0]
if pe.get("pending_limit"):
ctx_by_code[pick_code]["pending_limit"] = pe
else:
ctx_by_code[pick_code]["pending_entry"] = pe
skip_stats: Dict[str, Any] = {}
if skipped_micro_buys:
skip_stats["skipped_micro_buys"] = skipped_micro_buys
if skipped_vol_unfilled:
skip_stats["skipped_vol_unfilled"] = skipped_vol_unfilled
if use_ticks:
skip_stats["tick_entry_sources"] = dict(tick_fill_stats)
if skip_stats:
params["_portfolio_skip_stats"] = skip_stats
return all_trades
def run_tail_backtest(
candles_by_code: 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]:
"""
백테스트 1회 실행. (backtest_web 및 tail_param_search 호출용)
기본: ``portfolio_mode=true`` → 시각순 포트폴리오 (실매 MAX_STOCKS·총한도·slot_money).
``portfolio_mode=false`` → 레거시 종목별 독립 루프.
"""
if _to_bool(params.get("portfolio_mode"), True):
return run_tail_backtest_portfolio(
candles_by_code, params, universe_by_slot,
ticks_by_code=ticks_by_code,
orderbook_by_code=orderbook_by_code,
program_by_code=program_by_code,
)
# ── 레거시: 종목별 독립 시뮬 ──
tick_tf = tail_timeframe_min(params)
use_ticks = bool(ticks_by_code) and tail_backtest_wants_tick_replay(params)
# 파라미터 준비
min_drop_rate = float(params.get("min_drop_rate", 0.03))
min_recovery_ratio = float(params.get("min_recovery_ratio", 0.5))
max_rec_3m = float(params.get("max_rec_3m", 0.8))
tail_ratio_min = float(params.get("tail_ratio_min", 1.5))
tail_pct_min = float(params.get("tail_pct_min", 0.003))
shoulder_min_high = float(params.get("shoulder_min_high", 0.005))
shoulder_cut_pct = float(params.get("shoulder_cut_pct", 0.003))
rsi_period = int(params.get("rsi_period", 14))
rsi_threshold = float(params.get("rsi_threshold", 78))
high_chase_thr = float(params.get("high_chase_thr", 0.96))
time_start_hm = int(params.get("time_start_hm", 930))
time_end_hm = int(params.get("time_end_hm", 1500))
cooldown_min = float(params.get("cooldown_min", 15))
max_daily = int(params.get("max_daily", 3))
# 방어 로직 (동적 계산용)
stop_atr_mult = float(params.get("stop_atr_mult", 2.5))
target_atr_mult = float(params.get("target_atr_mult", 8.0))
max_loss_krw = int(params.get("max_loss_krw", 200000))
risk_pct = float(params.get("risk_pct", 0.01))
kelly_mult = float(params.get("kelly_mult", 0.25))
capital = float(params.get("capital", 100000000.0))
static_sl_pct = abs(float(params.get("sl_pct", 0.03)))
all_trades: List[Dict] = []
for code, candles in candles_by_code.items():
if len(candles) < rsi_period + 5:
continue
# 벡터 연산으로 지표 선행 계산 (백테스트 속도 최적화)
closes = [float(c["close"]) for c in candles]
rsis = compute_rsi_series(closes, rsi_period)
ma20s = compute_sma_series(closes, 20)
atrs = compute_atr_series(candles, 14)
position = None
last_exit_dt: Dict[str, datetime] = {}
daily_cnt: Dict[str, int] = {}
cur_day = None
running_open, running_high, running_low = 0.0, 0.0, 0.0
i = rsi_period + 1
while i < len(candles):
c = candles[i]
day = c["candle_time"][:8]
hm = int(c["candle_time"][8:12])
op = float(c["open"])
hi = float(c["high"])
lo = float(c["low"])
cl = float(c["close"])
# 일일 변수 초기화 및 갱신
if day != cur_day:
cur_day = day
running_open = op
running_high = hi
running_low = lo if lo > 0 else hi
else:
running_high = max(running_high, hi)
if lo > 0:
running_low = min(running_low, lo)
is_eod = is_strategy_eod_bar(c["candle_time"], params, "TAIL")
# ── 1. 청산 검사 (포지션 보유 중일 때) ──
if position is not None:
max_p = max(position["max_price"], hi)
hp = float(c.get("holding_peak") or 0)
if hp > 0:
max_p = max(max_p, hp)
position["max_price"] = max_p
cur_c_info = {
"open": op,
"high": hi,
"low": lo,
"close": cl,
"candle_time": c["candle_time"],
}
res = check_sell_signal_backtest_bar(
position,
cur_c_info,
params,
is_eod=is_eod,
sell_fn=check_sell_signal_live,
low_mode="session_low",
)
if res:
reason, exit_price = res
all_trades.append({
"code": code,
"entry_time": position["entry_time"],
"exit_time": c["candle_time"],
"entry": round(position["entry_price"]),
"exit": round(exit_price),
"pnl": 0,
"reason": reason,
"hold_min": 0,
"peak_price": round(max_p),
"qty": position.get("qty", 1),
})
last_exit_dt[day] = _t2dt(c["candle_time"])
daily_cnt[day] = daily_cnt.get(day, 0) + 1
position = None
i += 1
continue
# ── 2. 매수 검사 (포지션 없을 때, 유니버스 시뮬레이션 시 해당 슬롯 후보만) ──
if universe_by_slot is not None:
# 신봇 기본: 1분봉 == 슬롯 키 (TradeDBExt.get_universe_by_candle_time 키 포맷).
slot_key = _slot_key(c["candle_time"], params.get("scan_interval_min", 1))
if code not in universe_by_slot.get(slot_key, []):
i += 1
continue
if cl <= 0 or running_open <= 0 or hm < time_start_hm or hm > time_end_hm:
i += 1
continue
if daily_cnt.get(day, 0) >= max_daily:
i += 1
continue
if day in last_exit_dt:
elapsed = (_t2dt(c["candle_time"]) - last_exit_dt[day]).total_seconds() / 60
if elapsed < cooldown_min:
i += 1
continue
eval_params = dict(params)
if universe_by_slot is not None:
eval_params.setdefault("skip_hts_scan_dupes", True)
else:
eval_params.setdefault("skip_hts_scan_dupes", False)
state = {
"daily_cnt": daily_cnt.get(day, 0),
"last_exit_dt": last_exit_dt.get(day),
}
reject, _msg, sig = _eval_tail_buy_at_index(candles, i, eval_params, state)
if reject or not sig:
i += 1
continue
atr = atrs[i] if atrs[i] is not None else cl * 0.01
# ── 3. 매수 — limit_atr: 유효 봉 내 지정가 터치 / align: 다음 봉 시가 ──
if is_limit_atr_entry(short_entry_mode(params)):
lp_cfg = tail_limit_params(params)
sig_bar = candles[i]
anchor_px = resolve_limit_anchor_price(
lp_cfg["anchor"], sig_bar, candles, i,
)
min_px = float(params.get("min_price", 1000.0))
limit_px = compute_atr_limit_price(
anchor_px, atr, lp_cfg["mult"], min_price=min_px,
)
if limit_px <= 0:
i += 1
continue
vu = limit_valid_until_bar_key(candles, i, lp_cfg["valid_bars"])
filled = False
for j in range(i + 1, min(i + 1 + lp_cfg["valid_bars"], len(candles))):
if candles[j]["candle_time"][:8] != day:
break
bar_ticks = (
collect_bar_ticks(
ticks_by_code, code, candles[j]["candle_time"], tick_tf,
)
if use_ticks else []
)
fp, _src = try_limit_fill_on_bar_with_ticks(
candles[j], limit_px, lp_cfg["fill_slip_pct"],
ticks=bar_ticks, params=params,
)
if fp and fp > 0:
entry_price = fp
entry_time = candles[j]["candle_time"]
stop_p, target_p = compute_tail_atr_prices(entry_price, atr, params)
from_risk = (capital * risk_pct * kelly_mult) / static_sl_pct if static_sl_pct > 0 else capital
from_cap = max_loss_krw / static_sl_pct if static_sl_pct > 0 else capital
invest_amount = min(from_risk, from_cap)
calc_qty = max(1, int(invest_amount / entry_price))
position = {
"entry_price": entry_price,
"entry_time": entry_time,
"stop": stop_p,
"target": target_p,
"max_price": entry_price,
"session_low": entry_price,
"qty": calc_qty,
}
filled = True
i = j + 1
break
if not filled:
i += 1
continue
if i + 1 >= len(candles):
i += 1
continue
next_c = candles[i + 1]
if next_c["candle_time"][:8] != day:
i += 1
continue
entry_price = float(next_c["open"])
if entry_price <= 0:
entry_price = cl
if use_ticks:
bar_ticks = collect_bar_ticks(
ticks_by_code, code, next_c["candle_time"], tick_tf,
)
entry_price, _align_src = align_entry_price_from_ticks(bar_ticks, entry_price)
stop_p, target_p = compute_tail_atr_prices(entry_price, atr, params)
# 포지션 사이징 로직 (Risk % 및 Max Loss 반영)
from_risk = (capital * risk_pct * kelly_mult) / static_sl_pct if static_sl_pct > 0 else capital
from_cap = max_loss_krw / static_sl_pct if static_sl_pct > 0 else capital
invest_amount = min(from_risk, from_cap)
calc_qty = max(1, int(invest_amount / entry_price))
position = {
"entry_price": entry_price,
"entry_time": next_c["candle_time"],
"stop": stop_p,
"target": target_p,
"max_price": entry_price,
"session_low": entry_price,
"qty": calc_qty,
}
i += 1 # 진입 봉 건너뜀
continue
return all_trades
Reference in New Issue View Git Blame Copy Permalink