fix(backtest): 22.8% 쓰레기스킵 → 5.2% (bt_candle_source 로더 통일)

## 근본원인 bt_candle_source._load_ticks_by_code_bulk / _ticks_for_bar_garbage 가 자체 SELECT 로 recv_ts/tick_time_raw 미포함 → bar_is_garbage 가 _lag_sec/recv_ts 없이 봉끝 폴백 → WS 봉의 76% 대량 컷. 엔진 로더(breakout_tick_loader.load_common_ticks_by_code) 는 이미 recv_ts + _lag_sec 완비였지만 두 로더가 발산 상태 → 이번에 사고 촉발. ## 수정 (C안: 로더 통일) - _load_ticks_by_code_bulk: breakout_tick_loader._fetch_ws_ticks_day_rows 위임. flat 리턴은 _ingest_flat_tick_rows (신설) 로 어댑팅. recv_ts + _lag_sec 모두 포함 → 엔진 로더와 100% 동일 SELECT/계산. - _ticks_for_bar_garbage: 단일 종목용으로 _load_ticks_by_code_bulk (codes_filter=[code]) 재사용. 별도 SELECT 제거. ## 검증 Optuna scalp TPE 10 trials · 2026-09-04 · Python 엔진: - Before: 슬롯 63671 · 쓰레기스킵 14495 (22.8%) · hole 7.4% · kis/ws pick 11% - After: 슬롯 63671 · 쓰레기스킵 3324 ( 5.2%) · hole 3.7% · kis/ws pick 20% 실매 스모크 (scripts/test_live_execution_validation.py): 최종 통과 · 완결 로그: logs/test_live_execution_validation_20260906_201535.log ## 문서 (rust_engine_parity_port_plan.md) - 부록 A: A6~A11 사전 정지 작업 추가 (wall-clock 정합·로더 통일·파케이·게이팅·뱃지) - 부록 C (신설): D안 로드맵 — Rust 세션이 현재 봉만 받고 틱/호가/프로그램매매 미전달. Python 정합 개선을 받아도 Rust 는 여전히 데이터 절반짜리 상태. P0 최우선 6단계. ## 룰 - CRITICAL §0 (봉 freeze · OHLC 폴백 금지) 준수 - 룰 19 (근본원인 딥다이브 · 3안 선보고) — B/C/D 안 제시 후 사용자 C안 선택 - 룰 25-1 (교차검증) — 두 로더 발산 재발 방지 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-06 20:18:11 +09:00
parent 152bf40a4c
commit d8ac1595c1
2 changed files with 178 additions and 61 deletions
--- a/kis_trader/backtest/bt_candle_source.py
+++ b/kis_trader/backtest/bt_candle_source.py
@@ -97,37 +97,50 @@ def _ws_ticks_table(market: Optional[str]) -> str:
    return "ws_ticks"


-def _ticks_for_bar_garbage(
-    db,
-    code: str,
-    start_key: str,
-    end_key: str,
-    *,
-    market: Optional[str] = None,
-) -> Optional[List[Dict[str, Any]]]:
-    from kis_trader.engine.feed_fallback import candle_garbage_fallback_enabled
+def _ingest_flat_tick_rows(
+    rows: List[Dict[str, Any]],
+    target: Dict[str, List[Dict[str, Any]]],
+) -> int:
+    """SELECT 행 → 종목별 flat 리스트 (봉 pick 쓰레기검사용).

-    if not candle_garbage_fallback_enabled():
-        return None
-    sk, ek = _tick_time_bounds(start_key, end_key)
-    mk = (market or "").strip().upper()
-    table = _ws_ticks_table(mk if mk in ("US", "KR") else "KR")
-    try:
-        if mk in ("US", "KR"):
-            rows = db.conn.execute(
-                f"SELECT tick_time, source, tick_time_raw FROM {table} "
-                "WHERE market=%s AND code=%s AND tick_time >= %s AND tick_time <= %s",
-                (mk, str(code).strip(), sk, ek),
-            ).fetchall()
-        else:
-            rows = db.conn.execute(
-                f"SELECT tick_time, source, tick_time_raw FROM {table} "
-                "WHERE market=%s AND code=%s AND tick_time >= %s AND tick_time <= %s",
-                ("KR", str(code).strip(), sk, ek),
-            ).fetchall()
-        return [dict(r) for r in (rows or [])]
-    except Exception:
-        return None
+    ``breakout_tick_loader._ingest_tick_rows`` 는 minute_key 버킷 딕트를 만들지만,
+    bt_candle_source 소비자 (``dedupe_by_read_pairs``) 는 flat 리스트를 원한다.
+    → 같은 컬럼 (``recv_ts`` + ``_lag_sec``) 을 유지하되 리턴 형식만 다르게.
+    """
+    from kis_trader.backtest.breakout_tick_loader import _tick_row_lag_seconds
+
+    n = 0
+    for r in rows or []:
+        code = str(r.get("code") or "").strip()
+        if not code:
+            continue
+        tt = str(r.get("tick_time") or "")[:14]
+        if len(tt) < 12:
+            continue
+        kw_lag = _tick_row_lag_seconds(r)
+        # recv_ts: bar_is_garbage 가 wall-clock 정합 판정에 사용 (docs/정합성.md §9)
+        # 문자열로 보존. None 이면 봉끝 폴백.
+        _recv_raw = r.get("recv_ts")
+        _recv_str = ""
+        if _recv_raw is not None:
+            try:
+                _recv_str = (
+                    _recv_raw.strftime("%Y-%m-%d %H:%M:%S")
+                    if hasattr(_recv_raw, "strftime")
+                    else str(_recv_raw)
+                )
+            except Exception:
+                _recv_str = str(_recv_raw)
+        target[code].append({
+            "code": code,
+            "tick_time": tt,
+            "source": r.get("source") or "",
+            "tick_time_raw": str(r.get("tick_time_raw") or ""),
+            "_lag_sec": kw_lag,
+            "recv_ts": _recv_str,
+        })
+        n += 1
+    return n


 def _load_ticks_by_code_bulk(
@@ -138,49 +151,64 @@ def _load_ticks_by_code_bulk(
    market: Optional[str] = None,
    codes_filter: Optional[Sequence[str]] = None,
 ) -> Optional[Dict[str, List[Dict[str, Any]]]]:
-    """쓰레기 검사용 틱 — 기간 1~2쿼리. OFF 면 None (봉만 dedupe)."""
+    """쓰레기 검사용 틱 — 로더 통일 (2026-09-06 C안).
+
+    이전(자체 SELECT): ``tick_time, source, tick_time_raw`` 만 → recv_ts/_lag_sec 미포함
+    → ``bar_is_garbage`` 가 봉끝 폴백 → **22.8% 대량 컷** (실매와 다른 기준).
+
+    현재: ``breakout_tick_loader._fetch_ws_ticks_day_rows`` 위임 →
+    엔진 로더와 **동일 SELECT + 동일 lag 계산 + 동일 recv_ts** →
+    쓰레기 판정도 wall-clock 정합.
+
+    - KR → ``ws_ticks``  /  US → ``ws_ticks_us``
+    - OFF 면 None (봉만 dedupe)
+    - 리턴 형식은 flat ``Dict[code, List[tick]]`` (bt_candle_source 소비자 호환)
+    """
    from kis_trader.engine.feed_fallback import candle_garbage_fallback_enabled

    if not candle_garbage_fallback_enabled():
        return None
-    sk, ek = _tick_time_bounds(start_key, end_key)
-    mk = (market or "").strip().upper()
-    out: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
-    want: List[str] = []
-    if codes_filter:
-        want = [str(c).strip() for c in codes_filter if str(c).strip()]
-    in_sql = ""
-    in_params: List[str] = []
-    if want:
-        in_sql = " AND code IN (" + ",".join(["%s"] * len(want)) + ")"
-        in_params = want

-    def _pull(table: str, mkt: str) -> int:
-        rows = db.conn.execute(
-            f"SELECT code, tick_time, source, tick_time_raw FROM {table} "
-            "WHERE market=%s AND tick_time >= %s AND tick_time <= %s"
-            + in_sql,
-            (mkt, sk, ek, *in_params),
-        ).fetchall()
-        n = 0
-        for r in rows or []:
-            code = str(r.get("code") or "").strip()
-            if not code:
+    from kis_trader.backtest.breakout_tick_loader import (
+        _candle_keys_to_tick_range,
+        _fetch_ws_ticks_day_rows,
+        _iter_tick_day_chunks,
+        _ws_ticks_table as _bl_ws_ticks_table,
+    )
+
+    mk = (market or "").strip().upper()
+    tt_start, tt_end = _candle_keys_to_tick_range(start_key, end_key)
+
+    want: Optional[set] = None
+    if codes_filter:
+        want = {str(c).strip() for c in codes_filter if str(c).strip()}
+
+    out: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
+
+    def _pull(table: str, mkt_str: str) -> int:
+        total = 0
+        for chunk_s, chunk_e in _iter_tick_day_chunks(tt_start, tt_end):
+            try:
+                rows = _fetch_ws_ticks_day_rows(table, mkt_str, chunk_s, chunk_e, want)
+            except Exception as e:
+                logger.warning(
+                    "쓰레기검사 틱 로드 실패 (table=%s day=%s): %s",
+                    table, chunk_s[:8], e,
+                )
                continue
-            out[code].append(dict(r))
-            n += 1
-        return n
+            total += _ingest_flat_tick_rows(rows, out)
+        return total

    try:
        n_kr = n_us = 0
        if mk == "US":
-            n_us = _pull("ws_ticks_us", "US")
+            n_us = _pull(_bl_ws_ticks_table("US"), "US")
        elif mk == "KR":
-            n_kr = _pull("ws_ticks", "KR")
+            n_kr = _pull(_bl_ws_ticks_table("KR"), "KR")
        else:
-            n_kr = _pull("ws_ticks", "KR")
+            n_kr = _pull(_bl_ws_ticks_table("KR"), "KR")
            try:
-                n_us = _pull("ws_ticks_us", "US")
+                n_us = _pull(_bl_ws_ticks_table("US"), "US")
            except Exception:
                n_us = 0
        logger.info("📥 틱 bulk 쓰레기검사용: KR=%s US=%s 종목=%s", n_kr, n_us, len(out))
@@ -190,6 +218,34 @@ def _load_ticks_by_code_bulk(
        return None


+def _ticks_for_bar_garbage(
+    db,
+    code: str,
+    start_key: str,
+    end_key: str,
+    *,
+    market: Optional[str] = None,
+) -> Optional[List[Dict[str, Any]]]:
+    """단일 종목 쓰레기 검사용 틱 — 로더 통일 (bulk 재사용, 2026-09-06 C안).
+
+    이전엔 별도 SELECT (``tick_time, source, tick_time_raw`` 만) → recv_ts/_lag_sec 없음.
+    이제 ``_load_ticks_by_code_bulk`` (breakout_tick_loader 위임) 를 코드 1개로 호출 →
+    동일 정합.
+    """
+    from kis_trader.engine.feed_fallback import candle_garbage_fallback_enabled
+
+    if not candle_garbage_fallback_enabled():
+        return None
+    if not str(code or "").strip():
+        return None
+    result = _load_ticks_by_code_bulk(
+        db, start_key, end_key, market=market, codes_filter=[str(code).strip()],
+    )
+    if result is None:
+        return None
+    return result.get(str(code).strip(), [])
+
+
 def list_ws_candle_codes(
    db,
    timeframe: int,