Phase C dry-run·문서화·DB 증분 저장 및 운영 env 동기화

- 1분봉 다운로드 제외, MONITOR_PERSIST로 05/06 수집 시 coins.db INSERT
- Phase C paper_fires 로그·07 모의 리포트, hybrid 시뮬 산출물·reference 문서 갱신
- .env Phase C(LIVE=0), bootstrap dotenv override=True

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

deepcoin/ops/live_trader.py

@@ -37,7 +37,7 @@ from deepcoin.matching.position_sizing import (
 from deepcoin.paths import resolve_ground_truth_file
 from deepcoin.ops.alert_message import build_rule_alert_message
 from deepcoin.ops.monitor import Monitor
-from deepcoin.paths import LIVE_TRADES_LOG
+from deepcoin.paths import LIVE_TRADES_LOG, PAPER_FIRES_LOG
 
 
 class LiveTrader(Monitor):
@@ -87,6 +87,35 @@ class LiveTrader(Monitor):
         with LIVE_TRADES_LOG.open("a", encoding="utf-8") as f:
             f.write(json.dumps(record, ensure_ascii=False) + "\n")
 
+    def _append_paper_fire(
+        self,
+        hit: dict[str, Any],
+        planned_krw: float,
+        would_trade: bool,
+        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 수익률(참고) 집계.
+        """
+        PAPER_FIRES_LOG.parent.mkdir(parents=True, exist_ok=True)
+        row = {
+            "ts": datetime.now().isoformat(timespec="seconds"),
+            "signal_dt": hit.get("dt"),
+            "rule_id": hit.get("rule_id"),
+            "side": hit.get("side"),
+            "close": float(hit.get("close") or 0),
+            "planned_krw": round(float(planned_krw), 0),
+            "would_trade": bool(would_trade),
+            "skip_reason": skip_reason or "",
+            "live_enabled": bool(LIVE_TRADING_ENABLED),
+            "order_message": (order_log or {}).get("message", ""),
+        }
+        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]:
         """
         일·쿨다운·손실 한도 검사.
@@ -283,6 +312,9 @@ class LiveTrader(Monitor):
             if hit["side"] == "buy" and hit["rule_id"] 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 미통과 규칙"
+                )
                 continue
             planned = (
                 self._resolve_buy_amount_krw(hit)
@@ -293,11 +325,14 @@ class LiveTrader(Monitor):
             print(f"  [{hit['side']}] {rid} @ {hit['dt']}")
             if not ok:
                 print(f"    skip: {reason}")
+                self._append_paper_fire(hit, planned, False, reason)
                 continue
             if hit["side"] == "buy" and planned <= 0:
                 print("    skip: 매수금액 0")
+                self._append_paper_fire(hit, 0.0, False, "매수금액 0")
                 continue
             log = self._execute_order(hit)
+            self._append_paper_fire(hit, planned, True, "", log)
             self._append_log(log)
             print(f"    order: {log['message']} ok={log['ok']}")
             msg = build_rule_alert_message(hit, balances)

deepcoin/ops/monitor.py

@@ -415,18 +415,56 @@ class Monitor(HTS):
         bars_per_day = max((24 * 60) // max(interval, 1), 1)
         return bars_per_day * lookback_days + DB_ROW_WARMUP_BARS
 
-    def get_coin_saved_data(
+    def persist_api_candles_to_db(
         self,
         symbol: str,
         interval: int,
         data: pd.DataFrame,
         db_path: str = DB_PATH,
+    ) -> tuple[int, int]:
+        """
+        API로 받은 봉을 coins.db에 증분 INSERT합니다 (01_download.append_data와 동일).
+
+        dry-run·05·06·live_eval이 load_frames_from_db 할 때마다 최신 봉이 쌓입니다.
+
+        Returns:
+            (추가 행 수, 스킵 행 수)
+        """
+        if not MONITOR_PERSIST_CANDLES or data is None or data.empty:
+            return 0, 0
+
+        from deepcoin.data.downloader import (
+            append_data,
+            get_last_timestamp,
+            months_for_interval,
+            prune_before_cutoff,
+        )
+
+        if not isinstance(data.index, pd.DatetimeIndex):
+            data = data.copy()
+            data.index = pd.to_datetime(data.index)
+        data = data.sort_index()
+
+        last_ts = get_last_timestamp(symbol, interval, db_path=db_path)
+        inserted, skipped = append_data(
+            symbol, interval, data, last_ts=last_ts, db_path=db_path
+        )
+        if inserted > 0:
+            months = months_for_interval(interval, DOWNLOAD_MONTHS)
+            prune_before_cutoff(symbol, interval, months, db_path=db_path)
+        return inserted, skipped
+
+    def read_candles_from_db(
+        self,
+        symbol: str,
+        interval: int,
+        db_path: str = DB_PATH,
         max_rows: int = DB_READ_LIMIT_DEFAULT,
     ) -> pd.DataFrame:
         """
-        coins.db에서 저장된 봉을 읽고, API로 받은 최신 봉을 DB에 반영합니다.
+        coins.db에서 저장된 봉을 읽습니다.
 
-        scripts/01_download.py로 미리 적재해 두면 장기 MA 계산에 유리합니다.
+        scripts/01_download.py 또는 persist_api_candles_to_db로 적재된 데이터.
         """
         conn = sqlite3.connect(db_path)
         cursor = conn.cursor()
@@ -440,33 +478,6 @@ class Monitor(HTS):
             f"CREATE INDEX IF NOT EXISTS {table_name}_idx ON {table_name}(CODE, ymdhms)"
         )
 
-        for i in range(1, len(data)):
-            ymdhms = data["datetime"].iloc[-i].strftime("%Y-%m-%d %H:%M:%S")
-            cursor.execute(
-                f"SELECT 1 FROM {table_name} WHERE CODE = ? AND ymdhms = ?",
-                (symbol, ymdhms),
-            )
-            if not cursor.fetchone():
-                cursor.execute(
-                    f"INSERT INTO {table_name} "
-                    "(CODE, NAME, ymdhms, ymd, hms, Close, Open, High, Low, Volume) "
-                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-                    (
-                        symbol,
-                        KR_COINS[symbol],
-                        ymdhms,
-                        data["datetime"].iloc[-i].strftime("%Y%m%d"),
-                        data["datetime"].iloc[-i].strftime("%H%M%S"),
-                        data["Close"].iloc[-i],
-                        data["Open"].iloc[-i],
-                        data["High"].iloc[-i],
-                        data["Low"].iloc[-i],
-                        data["Volume"].iloc[-i],
-                    ),
-                )
-            else:
-                break
-
         cursor.execute(
             f"SELECT Open, Close, High, Low, Volume, ymdhms AS datetime "
             f"FROM (SELECT Open, Close, High, Low, Volume, ymdhms "
@@ -491,26 +502,42 @@ class Monitor(HTS):
         df["datetime"] = df.index
         return df
 
+    def get_coin_saved_data(
+        self,
+        symbol: str,
+        interval: int,
+        data: pd.DataFrame,
+        db_path: str = DB_PATH,
+        max_rows: int = DB_READ_LIMIT_DEFAULT,
+    ) -> pd.DataFrame:
+        """하위 호환: API 봉 저장 후 DB에서 읽기."""
+        self.persist_api_candles_to_db(symbol, interval, data, db_path=db_path)
+        return self.read_candles_from_db(
+            symbol, interval, db_path=db_path, max_rows=max_rows
+        )
+
     def get_coin_some_data(
         self, symbol: str, interval: int, db_max_rows: int | None = None
     ) -> pd.DataFrame:
         """
         WLD 시세: API 최신 봉 + coins.db 과거 봉 + 1분봉 최신 1개를 합칩니다.
 
-        DB가 비어 있으면 API·1분봉만 사용합니다. 과거 적재는 scripts/01_download.py 실행.
+        MONITOR_PERSIST_CANDLES=1 이면 API 청크를 즉시 coins.db에 INSERT합니다.
         """
         data = self.get_coin_data(symbol, interval)
         if data is None or data.empty:
             return pd.DataFrame()
 
+        self.persist_api_candles_to_db(symbol, interval, data)
+
         data_1 = self.get_coin_data(symbol, interval=1)
         if data_1 is not None and not data_1.empty:
             data_1 = data_1.copy()
             data_1.at[data_1.index[-1], "Volume"] = data_1["Volume"].iloc[-1] * 60
 
         row_limit = DB_READ_LIMIT_DEFAULT if db_max_rows is None else int(db_max_rows)
-        saved_data = self.get_coin_saved_data(
-            symbol, interval, data, max_rows=row_limit
+        saved_data = self.read_candles_from_db(
+            symbol, interval, max_rows=row_limit
         )
         parts = [data]
         if saved_data is not None and not saved_data.empty: