#!/usr/bin/env python3 """2026-08-14 실체결 vs filter_eval 근접 통계 (인덱스 친화).""" from __future__ import annotations import sys from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from database import TradeDB from kis_trader.utils.env import get_env_from_db, invalidate_merged_env_cache def ts14(raw: Any) -> str: s = str(raw or "").strip() s = s.replace("-", "").replace(":", "").replace(" ", "").replace("T", "") if len(s) < 8: return "" return (s + "000000")[:14] def parse14(st: str) -> Optional[datetime]: if len(st) < 14: return None try: return datetime.strptime(st[:14], "%Y%m%d%H%M%S") except ValueError: return None def main() -> None: invalidate_merged_env_cache() db = TradeDB() try: extra_keys = [ "WS_ORDERBOOK_TICK_MAX_AGE_SEC", "SLOT_MONEY_DEFAULT", "MOMENTUM_SLOT_MONEY", "BREAKOUT_SLOT_MONEY", "SCALP_SLOT_MONEY", "TAIL_SLOT_MONEY", "MOMENTUM_ORDERBOOK_ENTRY_ASK_MAX_MULT", "BREAKOUT_ORDERBOOK_ENTRY_ASK_MAX_MULT", "SCALP_ORDERBOOK_ENTRY_ASK_MAX_MULT", "TAIL_ORDERBOOK_ENTRY_ASK_MAX_MULT", "LIVE_OB_PROVIDER", "WS_TRIGGER_EVAL_SAVE_ENABLED", "MOMENTUM_ORDERBOOK_COLLECT_ENABLED", "BREAKOUT_ORDERBOOK_COLLECT_ENABLED", "KIWOOM_WS_ORDERBOOK_ENABLED", "LS_WS_UH1_ENABLED", "LS_CONDITION_ORDERBOOK", ] print("=== extra env ===") for k in extra_keys: print(f" {k}={get_env_from_db(k, '')!r}") day = "20260814" like = day + "%" trades = [ dict(r) for r in db.conn.execute( "SELECT id, code, name, strategy, buy_price, qty, buy_date, sell_date, sell_reason " "FROM trade_history WHERE buy_date LIKE %s OR buy_date LIKE %s " "ORDER BY buy_date", ("2026-08-14%", like), ).fetchall() ] opens = [ dict(r) for r in db.conn.execute( "SELECT code, name, strategy, avg_buy_price AS buy_price, current_qty AS qty, buy_date " "FROM active_trades WHERE buy_date LIKE %s OR buy_date LIKE %s", ("2026-08-14%", like), ).fetchall() ] print(f"\nclosed buys 0814={len(trades)} open buys 0814={len(opens)}") fe_n = db.conn.execute( "SELECT COUNT(*) AS n FROM ws_orderbook WHERE source=%s AND snap_time LIKE %s", ("filter_eval", like), ).fetchone() print("filter_eval 0814 n=", dict(fe_n)["n"]) rej_n = db.conn.execute( "SELECT COUNT(*) AS n FROM ws_orderbook WHERE source=%s AND snap_time LIKE %s " "AND reject_code IS NOT NULL AND reject_code <> %s", ("filter_eval", like, ""), ).fetchone() print("filter_eval reject 0814 n=", dict(rej_n)["n"]) pass_n = db.conn.execute( "SELECT COUNT(*) AS n FROM ws_orderbook WHERE source=%s AND snap_time LIKE %s " "AND (reject_code IS NULL OR reject_code = %s)", ("filter_eval", like, ""), ).fetchone() print("filter_eval pass 0814 n=", dict(pass_n)["n"]) by_st = db.conn.execute( "SELECT strategy, " "SUM(CASE WHEN reject_code IS NOT NULL AND reject_code <> %s THEN 1 ELSE 0 END) AS rej, " "COUNT(*) AS n FROM ws_orderbook WHERE source=%s AND snap_time LIKE %s " "GROUP BY strategy", ("", "filter_eval", like), ).fetchall() print("filter_eval by strategy:") for r in by_st: d = dict(r) print(f" {d.get('strategy')}: n={d.get('n')} rej={d.get('rej')}") buckets = { "fe_pm3": 0, "fe_pm30": 0, "fe_pm180": 0, "fe_none180": 0, "fe_rej_then_buy": 0, "fe_pass_then_buy": 0, "body_pm5": 0, } examples: List[str] = [] all_buys = list(trades) + list(opens) for t in all_buys: code = str(t.get("code") or "").strip() b14 = ts14(t.get("buy_date")) dt = parse14(b14) if not dt: continue t0 = (dt - timedelta(seconds=180)).strftime("%Y%m%d%H%M%S") t1 = (dt + timedelta(seconds=180)).strftime("%Y%m%d%H%M%S") rows = [ dict(r) for r in db.conn.execute( "SELECT snap_time, reject_code, reject_msg, bid_qty_l3, ask_qty_l3, source, strategy " "FROM ws_orderbook WHERE code=%s AND snap_time>=%s AND snap_time<=%s " "AND source=%s ORDER BY snap_time", (code, t0, t1, "filter_eval"), ).fetchall() ] t0b = (dt - timedelta(seconds=5)).strftime("%Y%m%d%H%M%S") t1b = (dt + timedelta(seconds=5)).strftime("%Y%m%d%H%M%S") body5 = db.conn.execute( "SELECT COUNT(*) AS n FROM ws_orderbook WHERE code=%s AND snap_time>=%s AND snap_time<=%s " "AND source=%s", (code, t0b, t1b, "kiwoom_0d"), ).fetchone() if dict(body5)["n"] > 0: buckets["body_pm5"] += 1 if not rows: buckets["fe_none180"] += 1 if len(examples) < 12: examples.append( f"NO_EVAL {t.get('strategy')} {code} {t.get('name')} buy={b14} " f"body±5s={dict(body5)['n']}" ) continue best = None best_abs = 1e9 for row in rows: dtr = parse14(str(row.get("snap_time") or "")) if not dtr: continue ad = abs((dtr - dt).total_seconds()) if ad < best_abs: best_abs = ad best = row if best is None: buckets["fe_none180"] += 1 continue if best_abs <= 3: buckets["fe_pm3"] += 1 if best_abs <= 30: buckets["fe_pm30"] += 1 buckets["fe_pm180"] += 1 rej = str(best.get("reject_code") or "").strip() if rej: buckets["fe_rej_then_buy"] += 1 examples.append( f"REJ_BUY Δ={best_abs:.0f}s {t.get('strategy')} {code} {t.get('name')} " f"buy={b14} eval={best.get('snap_time')} {rej} {best.get('reject_msg')} " f"L3 {best.get('bid_qty_l3')}/{best.get('ask_qty_l3')}" ) else: buckets["fe_pass_then_buy"] += 1 print("\n=== 0814 매수 vs filter_eval ===") print("buys total", len(all_buys)) for k, v in buckets.items(): print(f" {k}={v}") print("\n=== 탈락기록 후 매수 / 평가없음 샘플 ===") for e in examples: print(" ", e) print("\n=== 0814 매도 사유 ===") reasons = db.conn.execute( "SELECT sell_reason, COUNT(*) AS n FROM trade_history " "WHERE sell_date LIKE %s OR sell_date LIKE %s GROUP BY sell_reason ORDER BY n DESC", ("2026-08-14%", like), ).fetchall() for r in reasons: d = dict(r) print(f" {d.get('sell_reason')!r}: {d.get('n')}") finally: db.close() if __name__ == "__main__": main()