40만 원 기준 시뮬·dry-run 정합 및 hybrid 체결 엔진 통합.

초기 자금 GT_INITIAL_CASH_KRW=400000과 원화 한도 비율(알림·LIVE_ORDER·일한도·손실한도)을 맞추고, dry-run/live 체결을 sim_causal_hybrid(replay)와 동일 경로로 통합한다. 시뮬 리포트 갱신, Phase C 슈퍼바이저·매수매도 리허설 스크립트를 추가한다.

Co-authored-by: Cursor <cursoragent@cursor.com>

deepcoin/ops/live_trader.py

@@ -1,5 +1,7 @@
 """
 3단계: monitor_rules 발화 시 빗썸 실주문 (가드·로그).
+
+dry-run·live 체결 배분: 시뮬 sim_causal_hybrid 와 동일 (hybrid_sim_execution).
 """
 
 from __future__ import annotations
@@ -13,6 +15,7 @@ from typing import Any
 from config import (
     CHART_LOOKBACK_DAYS,
     COIN_NAME,
+    GT_INITIAL_CASH_KRW,
     LIVE_COOLDOWN_MIN,
     LIVE_DAILY_KRW_MAX,
     LIVE_DAILY_LOSS_LIMIT_KRW,
@@ -29,20 +32,29 @@ from deepcoin.matching.live_eval import evaluate_live_rules
 from deepcoin.matching.live_sizing import LivePositionState, live_sizing_enabled
 from deepcoin.matching.load_rules import load_monitor_rules
 from deepcoin.matching.position_sizing import (
-    compute_buy_amount_krw,
-    live_buy_asset_pct_scale,
     load_ev_wf_approved_rule_ids,
     top_leg_ids_by_forward_return,
 )
-from deepcoin.paths import resolve_ground_truth_file
 from deepcoin.ops.alert_message import build_rule_alert_message
+from deepcoin.ops.hybrid_sim_execution import (
+    hit_key,
+    plan_live_hit,
+    replay_paper_portfolio,
+    sort_hits_sim_order,
+)
 from deepcoin.ops.monitor import Monitor
-from deepcoin.paths import LIVE_TRADES_LOG, PAPER_FIRES_LOG
+from deepcoin.ops.paper_portfolio import PaperPortfolio
+from deepcoin.paths import (
+    LIVE_SIGNAL_HISTORY_JSON,
+    LIVE_TRADES_LOG,
+    PAPER_FIRES_LOG,
+    resolve_ground_truth_file,
+)
 
 
 class LiveTrader(Monitor):
     """
-    규칙 발화 시 실거래 실행. LIVE_TRADING_ENABLED=0 이면 주문 없음(드라이런 로그만).
+    규칙 발화 시 실거래 실행. LIVE_TRADING_ENABLED=0 이면 모의(sim hybrid)만.
     """
 
     def __init__(self) -> None:
@@ -57,8 +69,73 @@ class LiveTrader(Monitor):
         self._large_legs: set[int] = set()
         self._approved_rules: set[str] = set()
         self._position_state = LivePositionState.load()
+        self._paper = PaperPortfolio.load() if not LIVE_TRADING_ENABLED else None
+        self._live_signal_history: list[dict[str, Any]] = []
         self._ohlc_df = None
         self._load_sizing_context()
+        if self._paper_mode and self._paper.signal_history:
+            self._resync_paper_from_sim()
+        if LIVE_TRADING_ENABLED:
+            self._live_signal_history = self._load_live_signal_history()
+
+    @property
+    def _paper_mode(self) -> bool:
+        """dry-run: 모의 계좌·시뮬 hybrid 체결."""
+        return not LIVE_TRADING_ENABLED and self._paper is not None
+
+    def _load_live_signal_history(self) -> list[dict[str, Any]]:
+        """live 시뮬 정합용 발화 이력."""
+        if not LIVE_SIGNAL_HISTORY_JSON.is_file():
+            return []
+        try:
+            data = json.loads(LIVE_SIGNAL_HISTORY_JSON.read_text(encoding="utf-8"))
+            return list(data.get("signals") or [])
+        except (json.JSONDecodeError, OSError):
+            return []
+
+    def _save_live_signal_history(self) -> None:
+        """live 발화 이력 저장."""
+        LIVE_SIGNAL_HISTORY_JSON.parent.mkdir(parents=True, exist_ok=True)
+        LIVE_SIGNAL_HISTORY_JSON.write_text(
+            json.dumps(
+                {"signals": self._live_signal_history[-2000:]},
+                ensure_ascii=False,
+                indent=2,
+            ),
+            encoding="utf-8",
+        )
+
+    def _live_signal_seen(self, hit: dict[str, Any]) -> bool:
+        """live 이력에 동일 봉 발화가 있는지."""
+        dt, rid, side = hit_key(hit)
+        return any(
+            str(s["dt"]) == dt and str(s["rule_id"]) == rid and str(s["side"]) == side
+            for s in self._live_signal_history
+        )
+
+    def _append_live_signal(self, hit: dict[str, Any]) -> None:
+        """live 발화 이력 추가."""
+        if self._live_signal_seen(hit):
+            return
+        self._live_signal_history.append(
+            {
+                "dt": str(hit["dt"]),
+                "rule_id": str(hit["rule_id"]),
+                "side": str(hit["side"]),
+                "close": float(hit["close"]),
+            }
+        )
+
+    def _balances_for_trading(self) -> dict[str, dict[str, float]] | None:
+        """
+        dry-run: paper_portfolio만. live: 빗썸 API.
+        """
+        if self._paper_mode:
+            return self._paper.balances_dict()
+        try:
+            return self.load_balances_dict()
+        except Exception:
+            return None
 
     def _load_sizing_context(self) -> None:
         """GT leg·EV/WF 통과 규칙 캐시."""
@@ -77,12 +154,7 @@ class LiveTrader(Monitor):
             self._day_pnl_krw = 0.0
 
     def _append_log(self, record: dict[str, Any]) -> None:
-        """
-        live_trades.jsonl에 한 줄 append.
-
-        Args:
-            record: 로그 dict.
-        """
+        """live_trades.jsonl append."""
         LIVE_TRADES_LOG.parent.mkdir(parents=True, exist_ok=True)
         with LIVE_TRADES_LOG.open("a", encoding="utf-8") as f:
             f.write(json.dumps(record, ensure_ascii=False) + "\n")
@@ -95,11 +167,7 @@ class LiveTrader(Monitor):
         skip_reason: str = "",
         order_log: dict[str, Any] | None = None,
     ) -> None:
-        """
-        Phase C dry-run: 모든 발화·스킵 사유·모의 금액을 paper_fires.jsonl에 기록.
-
-        금요일 `07_phase_c_paper_report.py`로 forward 수익률(참고) 집계.
-        """
+        """Phase C paper_fires.jsonl."""
         PAPER_FIRES_LOG.parent.mkdir(parents=True, exist_ok=True)
         row = {
             "ts": datetime.now().isoformat(timespec="seconds"),
@@ -112,21 +180,21 @@ class LiveTrader(Monitor):
             "skip_reason": skip_reason or "",
             "live_enabled": bool(LIVE_TRADING_ENABLED),
             "order_message": (order_log or {}).get("message", ""),
+            "sizing": "sim_causal_hybrid",
         }
         with PAPER_FIRES_LOG.open("a", encoding="utf-8") as f:
             f.write(json.dumps(row, ensure_ascii=False) + "\n")
 
     def _can_trade(self, rule_id: str, planned_krw: float | None = None) -> tuple[bool, str]:
         """
-        일·쿨다운·손실 한도 검사.
-
-        Args:
-            rule_id: 규칙 ID.
-
-        Returns:
-            (허용 여부, 사유).
+        쿨다운(1봉=3분) + live 일한도. dry-run은 일한도만 생략.
         """
         self._reset_day_if_needed()
+        last = self._rule_last_unix.get(rule_id, 0.0)
+        if time.time() - last < LIVE_COOLDOWN_MIN * 60:
+            return False, f"규칙 쿨다운({LIVE_COOLDOWN_MIN}분)"
+        if self._paper_mode:
+            return True, ""
         if self._day_trades >= LIVE_MAX_TRADES_PER_DAY:
             return False, "일 최대 거래 수 초과"
         need = float(planned_krw if planned_krw is not None else LIVE_ORDER_KRW)
@@ -134,147 +202,103 @@ class LiveTrader(Monitor):
             return False, "일 주문 한도 초과"
         if self._day_pnl_krw <= -abs(LIVE_DAILY_LOSS_LIMIT_KRW):
             return False, "일 손실 한도 초과"
-        last = self._rule_last_unix.get(rule_id, 0.0)
-        if time.time() - last < LIVE_COOLDOWN_MIN * 60:
-            return False, f"규칙 쿨다운({LIVE_COOLDOWN_MIN}분)"
         return True, ""
 
     def _load_ohlc_df(self) -> None:
-        """drawdown tier용 3m OHLC 캐시."""
+        """drawdown tier용 3m OHLC."""
         try:
             frames = load_frames_from_db(self, SYMBOL, lookback_days=CHART_LOOKBACK_DAYS)
             self._ohlc_df = frames.get(MATCH_PRIMARY_INTERVAL)
         except Exception:
             self._ohlc_df = None
 
-    def _resolve_buy_amount_krw(self, hit: dict[str, Any]) -> float:
-        """
-        총자산·현금·EV/WF·leg 티어로 매수 원화 산출.
-
-        GT_SIGNAL_CAUSAL=1 이면 시뮬 sim_primary(hybrid, enhanced=False)와 동일 tier·weight.
-
-        Args:
-            hit: evaluate_live_rules 항목.
-
-        Returns:
-            매수 원화.
-        """
-        rid = hit["rule_id"]
-        if rid not in self._approved_rules:
-            return 0.0
-        price = float(hit["close"])
-        cash = 0.0
-        qty = 0.0
-        try:
-            bal = self.load_balances_dict()
-            sym = bal.get(SYMBOL, {})
-            cash = float(sym.get("available_krw") or sym.get("krw") or 0)
-            qty = float(sym.get("balance") or 0)
-        except Exception:
-            return 0.0
-        if live_sizing_enabled():
-            if self._ohlc_df is None:
-                self._load_ohlc_df()
-            return self._position_state.plan_buy_amount_krw(
-                hit["dt"],
-                price,
-                cash,
-                qty,
-                self._ohlc_df,
-                enhanced=False,
-                fee_rate=TRADING_FEE_RATE,
-            )
-        scale = live_buy_asset_pct_scale(
-            rid,
-            hit["dt"],
-            self._gt_trades,
-            approved_rules=self._approved_rules,
-            large_legs=self._large_legs,
+    def _resync_paper_from_sim(self) -> None:
+        """기존 paper 잔고를 sim_causal_hybrid replay 로 맞춤."""
+        if self._ohlc_df is None:
+            self._load_ohlc_df()
+        if self._ohlc_df is None or getattr(self._ohlc_df, "empty", True):
+            return
+        replayed, _ = replay_paper_portfolio(
+            self._paper.signal_history,
+            self._ohlc_df,
+            approved_buy_rules=self._approved_rules,
         )
-        return compute_buy_amount_krw(
-            cash,
-            qty,
-            price,
-            1.0,
-            1.0,
-            asset_pct_scale=scale,
-            fee_rate=TRADING_FEE_RATE,
+        self._paper.cash_krw = replayed.cash_krw
+        self._paper.qty = replayed.qty
+        self._paper.qty_by_leg = dict(replayed.qty_by_leg)
+        self._paper.current_leg_id = replayed.current_leg_id
+        self._paper.save()
+
+    def _sim_plan(self, hit: dict[str, Any]) -> Any:
+        """시뮬 hybrid 배분 1건."""
+        if self._ohlc_df is None:
+            self._load_ohlc_df()
+        if self._paper_mode:
+            hist = list(self._paper.signal_history)
+        else:
+            hist = list(self._live_signal_history)
+        return plan_live_hit(
+            hist,
+            hit,
+            self._ohlc_df,
+            approved_buy_rules=self._approved_rules,
         )
 
-    def _execute_order(self, hit: dict[str, Any]) -> dict[str, Any]:
-        """
-        매수·매도 주문 실행 또는 드라이런.
-
-        Args:
-            hit: evaluate_live_rules 항목.
-
-        Returns:
-            로그용 결과 dict.
-        """
+    def _execute_live_order(self, hit: dict[str, Any], plan: Any) -> dict[str, Any]:
+        """실거래: 시뮬 planned 금액·수량으로 API 주문."""
         side = hit["side"]
         price = float(hit["close"])
-        if side == "buy":
-            amount_krw = self._resolve_buy_amount_krw(hit)
-            if amount_krw <= 0:
-                return {
-                    "ts": datetime.now().isoformat(timespec="seconds"),
-                    "rule_id": hit["rule_id"],
-                    "side": side,
-                    "signal_dt": hit["dt"],
-                    "price": price,
-                    "amount_krw": 0,
-                    "live_enabled": LIVE_TRADING_ENABLED,
-                    "ok": False,
-                    "message": "매수 스킵(EV/WF·leg·현금)",
-                }
-        else:
-            amount_krw = float(LIVE_ORDER_KRW)
         record: dict[str, Any] = {
             "ts": datetime.now().isoformat(timespec="seconds"),
             "rule_id": hit["rule_id"],
             "side": side,
             "signal_dt": hit["dt"],
             "price": price,
-            "amount_krw": amount_krw,
-            "live_enabled": LIVE_TRADING_ENABLED,
+            "amount_krw": plan.amount_krw,
+            "live_enabled": True,
             "ok": False,
-            "message": "",
+            "message": plan.message,
+            "sizing": "sim_causal_hybrid",
         }
-
-        if not LIVE_TRADING_ENABLED:
-            record["message"] = "dry_run (LIVE_TRADING_ENABLED=0)"
-            record["ok"] = True
+        if not plan.ok:
             return record
 
         try:
             if side == "buy":
-                ok = self.buyCoinMarket(SYMBOL, int(amount_krw), count=None)
+                ok = self.buyCoinMarket(SYMBOL, int(plan.amount_krw), count=None)
                 record["ok"] = bool(ok)
                 record["message"] = "buyCoinMarket" if ok else "buy failed"
             elif side == "sell":
                 bal = self.load_balances_dict().get(SYMBOL, {})
-                qty = float(bal.get("balance") or 0)
-                if qty <= 0:
+                held = float(bal.get("balance") or 0)
+                if held <= 0:
                     record["message"] = "보유 없음"
                 else:
-                    gross = qty * price
-                    record["amount_krw"] = round(gross, 0)
-                    ok = self.sellCoinMarket(SYMBOL, int(price), qty)
-                    record["ok"] = bool(ok)
-                    record["message"] = f"sell qty={qty}" if ok else "sell failed"
-                    if record["ok"] and live_sizing_enabled():
-                        fee = gross * TRADING_FEE_RATE
-                        self._position_state.record_sell(
-                            gross, fee, full_close=True
+                    sell_qty = min(float(plan.sell_qty), held)
+                    if sell_qty <= 0:
+                        record["message"] = "매도 수량 0"
+                    else:
+                        gross = sell_qty * price
+                        record["amount_krw"] = round(gross, 0)
+                        ok = self.sellCoinMarket(SYMBOL, int(price), sell_qty)
+                        record["ok"] = bool(ok)
+                        record["sell_qty"] = sell_qty
+                        record["message"] = (
+                            f"sell qty={sell_qty:.4f}" if ok else "sell failed"
                         )
-                        self._position_state.save()
+                        if record["ok"] and live_sizing_enabled():
+                            fee = gross * TRADING_FEE_RATE
+                            self._position_state.record_sell(
+                                gross, fee, full_close=(sell_qty >= held * 0.999)
+                            )
+                            self._position_state.save()
             else:
                 record["message"] = f"unknown side {side}"
         except Exception as exc:
             record["message"] = str(exc)
 
         if record["ok"]:
-            spent = float(record.get("amount_krw") or amount_krw)
+            spent = float(record.get("amount_krw") or plan.amount_krw)
             self._day_spent_krw += spent
             self._day_trades += 1
             self._rule_last_unix[hit["rule_id"]] = time.time()
@@ -282,74 +306,158 @@ class LiveTrader(Monitor):
                 fee = spent * TRADING_FEE_RATE
                 self._position_state.record_buy(hit["dt"], price, spent, fee)
                 self._position_state.save()
+            self._append_live_signal(hit)
+            self._save_live_signal_history()
         return record
 
+    def _process_paper_batch(self, new_hits: list[dict[str, Any]]) -> None:
+        """
+        dry-run: 신규 발화를 이력에 넣고 시뮬 전체 재생 후 알림.
+        """
+        if not new_hits:
+            return
+        if self._ohlc_df is None:
+            self._load_ohlc_df()
+        for hit in new_hits:
+            self._paper.append_signal(hit)
+
+        replayed, results = replay_paper_portfolio(
+            self._paper.signal_history,
+            self._ohlc_df,
+            approved_buy_rules=self._approved_rules,
+        )
+        self._paper.cash_krw = replayed.cash_krw
+        self._paper.qty = replayed.qty
+        self._paper.qty_by_leg = dict(replayed.qty_by_leg)
+        self._paper.current_leg_id = replayed.current_leg_id
+
+        for hit in new_hits:
+            key = hit_key(hit)
+            res = results.get(key)
+            if res is None:
+                self._paper.mark_processed(hit["rule_id"], hit["dt"])
+                continue
+            log = {
+                "ok": res.ok,
+                "message": res.message,
+                "amount_krw": res.amount_krw,
+                "sell_qty": res.sell_qty,
+            }
+            self._append_paper_fire(
+                hit, res.amount_krw, res.ok, "" if res.ok else res.message, log
+            )
+            self._paper.mark_processed(hit["rule_id"], hit["dt"])
+            print(f"  [{hit['side']}] {hit['rule_id']} @ {hit['dt']}")
+            print(f"    order: {res.message} ok={res.ok}")
+            if not res.ok:
+                continue
+            self._rule_last_unix[hit["rule_id"]] = time.time()
+            post_balances = self._paper.balances_dict()
+            msg = build_rule_alert_message(
+                hit,
+                post_balances,
+                trade_krw=res.amount_krw,
+                trade_qty=res.sell_qty if hit["side"] == "sell" else None,
+            )
+            sym = post_balances.get(SYMBOL, {})
+            msg += (
+                f"\n[모의잔고·체결후] 현금 {_fmt_paper_krw(sym.get('krw', 0))} · "
+                f"보유 {float(sym.get('balance', 0)):.4f} {SYMBOL}"
+            )
+            msg += f"\n[체결] {res.message}"
+            self._send_coin_msg(msg)
+        self._paper.save()
+
     def run_once(self) -> None:
-        """1회: 규칙 평가 → (허용 시) 주문 → 텔레그램."""
+        """1회: 규칙 평가 → 시뮬 hybrid 체결 → 텔레그램."""
         rules = load_monitor_rules()
         print(
             f"[06] {datetime.now():%Y-%m-%d %H:%M:%S} "
             f"{COIN_NAME} live={'ON' if LIVE_TRADING_ENABLED else 'OFF'} "
-            f"rules={len(rules)}"
+            f"rules={len(rules)} · sim=hybrid · bar={MATCH_PRIMARY_INTERVAL}m"
         )
         if not rules:
             print("  monitor_rules 없음")
             return
 
-        fired = evaluate_live_rules(rules)
-        balances = None
-        try:
-            balances = self.load_balances_dict()
-        except Exception:
-            pass
-
+        fired = evaluate_live_rules(rules, force_refresh=True)
         if not fired:
             print("  발화 없음")
             return
 
-        for hit in fired:
+        if self._paper_mode:
+            print(
+                f"  [paper] 현금 ₩{self._paper.cash_krw:,.0f} · "
+                f"보유 {self._paper.qty:.4f} {SYMBOL} "
+                f"(초기 ₩{GT_INITIAL_CASH_KRW:,.0f})"
+            )
+
+        new_paper_hits: list[dict[str, Any]] = []
+        for hit in sort_hits_sim_order(fired):
             rid = hit["rule_id"]
-            if hit["side"] == "buy" and hit["rule_id"] not in self._approved_rules:
+            if self._paper_mode and self._paper.already_processed(rid, hit["dt"]):
+                print(f"  [{hit['side']}] {rid} @ {hit['dt']} (이미 처리)")
+                continue
+            if LIVE_TRADING_ENABLED and self._live_signal_seen(hit):
+                continue
+
+            if hit["side"] == "buy" and rid not in self._approved_rules:
                 print(f"  [{hit['side']}] {rid} @ {hit['dt']}")
                 print("    skip: EV/WF 미통과 규칙")
-                self._append_paper_fire(
-                    hit, 0.0, False, "EV/WF 미통과 규칙"
-                )
+                if self._paper_mode:
+                    self._append_paper_fire(hit, 0.0, False, "EV/WF 미통과 규칙")
+                    self._paper.mark_processed(rid, hit["dt"])
                 continue
-            planned = (
-                self._resolve_buy_amount_krw(hit)
-                if hit["side"] == "buy"
-                else float(LIVE_ORDER_KRW)
-            )
-            ok, reason = self._can_trade(rid, planned)
-            print(f"  [{hit['side']}] {rid} @ {hit['dt']}")
+
+            plan_preview = self._sim_plan(hit)
+            ok, reason = self._can_trade(rid, plan_preview.amount_krw)
             if not ok:
+                print(f"  [{hit['side']}] {rid} @ {hit['dt']}")
                 print(f"    skip: {reason}")
-                self._append_paper_fire(hit, planned, False, reason)
+                if self._paper_mode:
+                    self._append_paper_fire(
+                        hit, plan_preview.amount_krw, False, reason
+                    )
+                    self._paper.mark_processed(rid, hit["dt"])
                 continue
-            if hit["side"] == "buy" and planned <= 0:
-                print("    skip: 매수금액 0")
-                self._append_paper_fire(hit, 0.0, False, "매수금액 0")
+
+            if self._paper_mode:
+                new_paper_hits.append(hit)
                 continue
-            log = self._execute_order(hit)
-            self._append_paper_fire(hit, planned, True, "", log)
+
+            print(f"  [{hit['side']}] {rid} @ {hit['dt']}")
+            log = self._execute_live_order(hit, plan_preview)
             self._append_log(log)
             print(f"    order: {log['message']} ok={log['ok']}")
-            msg = build_rule_alert_message(hit, balances)
-            if log["ok"]:
-                msg += f"\n[체결] {log['message']}"
-            else:
-                msg += f"\n[실패] {log['message']}"
+            if not log["ok"]:
+                continue
+            balances = self._balances_for_trading()
+            msg = build_rule_alert_message(
+                hit,
+                balances,
+                trade_krw=float(log.get("amount_krw") or 0),
+                trade_qty=float(log.get("sell_qty") or 0) or None,
+            )
+            if balances:
+                sym = balances.get(SYMBOL, {})
+                msg += (
+                    f"\n[잔고] 현금 {_fmt_paper_krw(sym.get('krw', 0))} · "
+                    f"보유 {float(sym.get('balance', 0)):.4f} {SYMBOL}"
+                )
+            msg += f"\n[체결] {log['message']}"
             self._send_coin_msg(msg)
 
-    def run_loop(self, sleep_sec: int) -> None:
-        """
-        상시 루프.
+        if self._paper_mode and new_paper_hits:
+            self._process_paper_batch(new_paper_hits)
 
-        Args:
-            sleep_sec: 대기 초.
-        """
+    def run_loop(self, sleep_sec: int) -> None:
+        """상시 루프."""
         print(f"[06] 실거래 루프 시작 · sleep={sleep_sec}s")
         while True:
             self.run_once()
             time.sleep(sleep_sec)
+
+
+def _fmt_paper_krw(value: float) -> str:
+    """원화 표시."""
+    return f"₩{float(value):,.0f}"