diff --git a/src/config.py b/src/config.py index ad400c1..2fb7a0b 100644 --- a/src/config.py +++ b/src/config.py @@ -106,3 +106,21 @@ class Config: volume_multiplier=self.volume_multiplier, )) + +# ── OOS 사후보정용 비용 모델 ────────────────────────────────────── +COST_MODEL = { + "taker_fee_bps": 4.0, # Binance USDⓈ-M Futures VIP 0 + "maker_fee_bps": 2.0, # 향후 limit TP 도입 대비 + # MTF bot 주문 타입 (현재 전부 MARKET = taker) + "entry_order_type": "taker", + "sl_order_type": "taker", + "tp_order_type": "taker", +} + +# 3개 프리셋 시나리오 (확장 금지, 이 셋으로 고정) +COST_SCENARIOS = { + "fees_only": {"slippage_bps_per_side": 0.0, "funding_bps_per_8h": 0.0}, + "realistic": {"slippage_bps_per_side": 1.0, "funding_bps_per_8h": 1.0}, + "pessimistic": {"slippage_bps_per_side": 3.0, "funding_bps_per_8h": 2.0}, +} + diff --git a/src/mtf_bot.py b/src/mtf_bot.py index 70cb18a..795b19f 100644 --- a/src/mtf_bot.py +++ b/src/mtf_bot.py @@ -390,6 +390,21 @@ class TriggerStrategy: _MTF_TRADE_DIR = Path("data/trade_history") +# ── 킬스위치 상수 (bot.py와 동일 기준) ── +_FAST_KILL_STREAK = 8 # 연속 손실 N회 → 즉시 중단 +_SLOW_KILL_WINDOW = 15 # 최근 N거래 PF 산출 +_SLOW_KILL_PF_THRESHOLD = 0.75 # PF < 이 값이면 중단 + + +def _tail_lines(path: Path, n: int) -> list[str]: + """파일의 마지막 n줄을 읽는다.""" + try: + with open(path) as f: + lines = f.readlines() + return lines[-n:] + except Exception: + return [] + class ExecutionManager: """ @@ -408,6 +423,11 @@ class ExecutionManager: self._sl_price: Optional[float] = None self._tp_price: Optional[float] = None self._atr_at_entry: Optional[float] = None + # ── 킬스위치 ── + self._killed: bool = False + self._trade_history: list[dict] = [] + self._restore_trade_history() + self._restore_kill_switch() def execute(self, signal: str, current_price: float, atr_value: Optional[float]) -> Optional[Dict]: """ @@ -424,6 +444,10 @@ class ExecutionManager: if signal == "HOLD": return None + if self._killed: + logger.warning(f"[ExecutionManager] 킬스위치 발동 상태 — 신규 진입 차단 (신호={signal})") + return None + if self.current_position is not None: logger.debug( f"[ExecutionManager] 포지션 중복 차단: " @@ -503,6 +527,10 @@ class ExecutionManager: # JSONL에 기록 self._save_trade(reason, exit_price, pnl_bps) + # ── 킬스위치: 거래 이력 추가 + 판정 ── + self._append_trade_history(pnl_bps) + self._check_kill_switch() + # ── 실주문 (프로덕션 전환 시 주석 해제) ── # if self.current_position == "LONG": # await self.exchange.create_market_sell_order(symbol, amount) @@ -540,6 +568,91 @@ class ExecutionManager: except Exception as e: logger.warning(f"[ExecutionManager] 거래 기록 저장 실패: {e}") + # ── 킬스위치 ────────────────────────────────────────────────────── + + def _trade_history_path(self) -> Path: + _MTF_TRADE_DIR.mkdir(parents=True, exist_ok=True) + return _MTF_TRADE_DIR / f"mtf_{self.symbol.replace('/', '').replace(':', '').lower()}.jsonl" + + def _restore_trade_history(self) -> None: + """부팅 시 JSONL에서 최근 N건의 pnl_bps를 복원한다.""" + path = self._trade_history_path() + if not path.exists(): + return + try: + tail_n = max(_FAST_KILL_STREAK, _SLOW_KILL_WINDOW) + lines = _tail_lines(path, tail_n) + for line in lines: + line = line.strip() + if line: + record = json.loads(line) + self._trade_history.append({"pnl_bps": record.get("pnl_bps", 0.0)}) + logger.info( + f"[ExecutionManager] 거래 이력 복원: {len(self._trade_history)}건" + ) + except Exception as e: + logger.warning(f"[ExecutionManager] 거래 이력 복원 실패: {e}") + + def _restore_kill_switch(self) -> None: + """부팅 시 리셋 플래그 확인 후, 이력 기반으로 킬스위치 소급 검증.""" + reset_key = f"RESET_KILL_SWITCH_MTF_{self.symbol.replace('/', '').replace(':', '').upper()}" + if os.environ.get(reset_key, "").lower() == "true": + logger.info(f"[ExecutionManager] 킬스위치 수동 해제 감지 ({reset_key}=True)") + self._killed = False + return + if self._check_kill_switch(silent=True): + logger.warning("[ExecutionManager] 부팅 시 킬스위치 조건 충족 — 신규 진입 차단") + + def _append_trade_history(self, pnl_bps: float) -> None: + """킬스위치 판단용 거래 이력에 추가한다.""" + self._trade_history.append({"pnl_bps": round(pnl_bps, 1)}) + max_window = max(_FAST_KILL_STREAK, _SLOW_KILL_WINDOW) + if len(self._trade_history) > max_window * 2: + self._trade_history = self._trade_history[-max_window:] + + def _check_kill_switch(self, silent: bool = False) -> bool: + """킬스위치 조건을 검사하고, 발동 시 True를 반환한다. + + Fast Kill: 최근 8연속 손실 (pnl_bps < 0) + Slow Kill: 최근 15거래 PF < 0.75 + """ + trades = self._trade_history + if not trades: + return False + + # Fast Kill + if len(trades) >= _FAST_KILL_STREAK: + recent = trades[-_FAST_KILL_STREAK:] + if all(t["pnl_bps"] < 0 for t in recent): + reason = f"Fast Kill ({_FAST_KILL_STREAK}연속 손실)" + self._trigger_kill_switch(reason, silent) + return True + + # Slow Kill + if len(trades) >= _SLOW_KILL_WINDOW: + recent = trades[-_SLOW_KILL_WINDOW:] + gross_profit = sum(t["pnl_bps"] for t in recent if t["pnl_bps"] > 0) + gross_loss = abs(sum(t["pnl_bps"] for t in recent if t["pnl_bps"] < 0)) + if gross_loss > 0: + pf = gross_profit / gross_loss + if pf < _SLOW_KILL_PF_THRESHOLD: + reason = f"Slow Kill (최근 {_SLOW_KILL_WINDOW}거래 PF={pf:.2f})" + self._trigger_kill_switch(reason, silent) + return True + + return False + + def _trigger_kill_switch(self, reason: str, silent: bool = False) -> None: + """킬스위치 발동: 상태 변경 + 로그.""" + self._killed = True + msg = ( + f"[MTF KILL SWITCH] {self.symbol} 신규 진입 중단\n" + f"사유: {reason}\n" + f"기존 포지션 SL/TP는 정상 작동합니다.\n" + f"해제: RESET_KILL_SWITCH_MTF_{self.symbol.replace('/', '').replace(':', '').upper()}=True 후 재시작" + ) + logger.error(msg) + def get_position_info(self) -> Dict: """현재 포지션 정보 반환.""" return { diff --git a/tests/test_mtf_bot.py b/tests/test_mtf_bot.py index b7c915d..d40a1d6 100644 --- a/tests/test_mtf_bot.py +++ b/tests/test_mtf_bot.py @@ -421,3 +421,107 @@ class TestExecutionManager: # TP/SL = 2.3/1.5 = 1.533... expected_rr = round(2.3 / 1.5, 2) assert result["risk_reward"] == expected_rr + + +# ═══════════════════════════════════════════════════════════════════ +# Test 5: ExecutionManager 킬스위치 +# ═══════════════════════════════════════════════════════════════════ + + +class TestExecutionManagerKillSwitch: + """ExecutionManager 킬스위치 (Fast Kill / Slow Kill) 테스트.""" + + def _make_em(self) -> ExecutionManager: + """JSONL 복원 없이 깨끗한 ExecutionManager 생성.""" + em = ExecutionManager.__new__(ExecutionManager) + em.symbol = "TESTUSDT" + em.current_position = None + em._entry_price = None + em._entry_ts = None + em._sl_price = None + em._tp_price = None + em._atr_at_entry = None + em._killed = False + em._trade_history = [] + return em + + def test_fast_kill_triggers_after_8_consecutive_losses(self): + """8연속 손실 시 Fast Kill 발동.""" + em = self._make_em() + for _ in range(8): + em._append_trade_history(-50.0) + assert em._check_kill_switch() is True + assert em._killed is True + + def test_fast_kill_not_triggered_with_7_losses(self): + """7연속 손실은 Fast Kill 미발동.""" + em = self._make_em() + for _ in range(7): + em._append_trade_history(-50.0) + assert em._check_kill_switch() is False + assert em._killed is False + + def test_fast_kill_broken_by_single_win(self): + """연속 손실 중 1회 수익이 있으면 Fast Kill 미발동.""" + em = self._make_em() + for _ in range(4): + em._append_trade_history(-50.0) + em._append_trade_history(10.0) # 중간에 수익 + for _ in range(3): + em._append_trade_history(-50.0) + assert em._check_kill_switch() is False + + def test_slow_kill_triggers_when_pf_below_threshold(self): + """최근 15거래 PF < 0.75 시 Slow Kill 발동.""" + em = self._make_em() + # 12패 (-100 bps each) + 3승 (+50 bps each) + # gross_profit=150, gross_loss=1200, PF=0.125 + for _ in range(12): + em._append_trade_history(-100.0) + for _ in range(3): + em._append_trade_history(50.0) + assert em._check_kill_switch() is True + assert em._killed is True + + def test_slow_kill_not_triggered_when_pf_above_threshold(self): + """PF > 0.75면 Slow Kill 미발동.""" + em = self._make_em() + # 7패 (-100 each) + 8승 (+100 each) + # gross_profit=800, gross_loss=700, PF=1.14 + for _ in range(7): + em._append_trade_history(-100.0) + for _ in range(8): + em._append_trade_history(100.0) + assert em._check_kill_switch() is False + + def test_killed_state_blocks_new_entry(self): + """킬스위치 발동 후 신규 진입이 차단된다.""" + em = self._make_em() + em._killed = True + result = em.execute("EXECUTE_LONG", 2.0, 0.01) + assert result is None + + def test_killed_state_allows_existing_sl_tp(self): + """킬스위치 발동 후에도 기존 포지션의 청산은 정상 동작.""" + em = self._make_em() + # 먼저 포지션 진입 + em.execute("EXECUTE_SHORT", 2.0, 0.01) + # 킬스위치 발동 + em._killed = True + # 청산은 정상 동작해야 함 + em.close_position("SL 히트", exit_price=2.015, pnl_bps=-75.0) + assert em.current_position is None + + def test_kill_switch_integrated_with_close(self): + """close_position 호출 시 자동으로 킬스위치 판정이 실행된다.""" + em = self._make_em() + # 7번 손실 기록 + for _ in range(7): + em._append_trade_history(-50.0) + # 포지션 진입 후 8번째 손실로 청산 + em.execute("EXECUTE_SHORT", 2.0, 0.01) + em.close_position("SL 히트", exit_price=2.015, pnl_bps=-75.0) + # 8연패 → Fast Kill 발동 + assert em._killed is True + # 다음 진입 차단 + assert em.execute("EXECUTE_LONG", 1.95, 0.01) is None