#!/usr/bin/env python3
"""
LS 섀도 시세 유니버스 히스토리 수집 + 종목 히스토리 퀄리티 리포트 (단독 실행).

테이블: ``ls_universe_history``
  - target_candidates_history(키움 조건식)와 분리
  - 슬롯(기본 1분)마다 LS가 실제로 틱/봉을 쌓은 종목을 스냅샷

사용:
  # 오늘 틱으로 슬롯 백필 + 퀄리티 출력
  python3 scripts/collect_ls_universe_history.py --day 2026-07-27 --backfill --quality

  # 현재 시각 슬롯 1회 수집만
  python3 scripts/collect_ls_universe_history.py --collect

  # N초마다 수집 (Ctrl+C 종료)
  python3 scripts/collect_ls_universe_history.py --loop 60 --quality-every 5

  # 퀄리티만 (이미 쌓인 테이블/틱/봉 기준)
  python3 scripts/collect_ls_universe_history.py --day 2026-07-27 --quality
"""
from __future__ import annotations

import argparse
import os
import sys
import time
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Sequence, Tuple

_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _ROOT not in sys.path:
    sys.path.insert(0, _ROOT)


DDL = """
CREATE TABLE IF NOT EXISTS ls_universe_history (
    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
    slot_key VARCHAR(12) NOT NULL,
    scan_time VARCHAR(30) NOT NULL,
    code VARCHAR(20) NOT NULL,
    name VARCHAR(100) NOT NULL DEFAULT '',
    price DOUBLE NOT NULL DEFAULT 0,
    tick_count INT NOT NULL DEFAULT 0,
    candle_1m_count INT NOT NULL DEFAULT 0,
    last_tick_ts DATETIME(3) NULL,
    last_candle_dt VARCHAR(20) NULL,
    gap_1m_missing INT NOT NULL DEFAULT 0,
    source VARCHAR(16) NOT NULL DEFAULT 'ls_ws',
    note VARCHAR(200) NULL,
    UNIQUE KEY uq_ls_univ_slot_code (slot_key, code),
    INDEX idx_ls_univ_slot (slot_key),
    INDEX idx_ls_univ_code (code),
    INDEX idx_ls_univ_scan (scan_time)
) CHARACTER SET utf8mb4
"""


def _parse_day(s: str) -> str:
    s = (s or "").strip()
    if not s:
        return datetime.now().strftime("%Y-%m-%d")
    if len(s) == 8 and s.isdigit():
        return f"{s[:4]}-{s[5:7]}-{s[6:8]}" if False else f"{s[0:4]}-{s[4:6]}-{s[6:8]}"
    return s[:10]


def _day_compact(day: str) -> str:
    return day.replace("-", "")[:8]


def _slot_key_from_dt(dt: datetime, slot_min: int = 1) -> str:
    slot_min = max(1, int(slot_min))
    m = (dt.minute // slot_min) * slot_min
    return dt.strftime("%Y%m%d%H") + f"{m:02d}"


def _ensure_table(db) -> None:
    db.conn.execute(DDL)


def _name_map(db, codes: Sequence[str]) -> Dict[str, str]:
    out: Dict[str, str] = {}
    if not codes:
        return out
    # target_candidates 에 있으면 이름 재사용
    try:
        cols = [r["Field"] for r in db.conn.execute("SHOW COLUMNS FROM target_candidates").fetchall()]
        if "code" in cols and "name" in cols:
            # chunk
            codes = [str(c) for c in codes]
            for i in range(0, len(codes), 200):
                chunk = codes[i : i + 200]
                ph = ",".join(["%s"] * len(chunk))
                rows = db.conn.execute(
                    f"SELECT code, name FROM target_candidates WHERE code IN ({ph})",
                    tuple(chunk),
                ).fetchall()
                for r in rows:
                    out[str(r["code"])] = str(r.get("name") or "")
    except Exception:
        pass
    return out


def _market_1m_slots(day: str) -> List[str]:
    """정규장 09:00~15:29 1분 슬롯 키 목록."""
    d = _day_compact(day)
    out: List[str] = []
    t0 = datetime.strptime(d + "0900", "%Y%m%d%H%M")
    t1 = datetime.strptime(d + "1530", "%Y%m%d%H%M")
    cur = t0
    while cur < t1:
        out.append(cur.strftime("%Y%m%d%H%M"))
        cur += timedelta(minutes=1)
    return out


def collect_slot(db, *, slot_key: Optional[str] = None, slot_min: int = 1) -> int:
    """현재(또는 지정) 슬롯에 LS 틱이 있는 종목을 ls_universe_history 에 upsert."""
    _ensure_table(db)
    now = datetime.now()
    sk = slot_key or _slot_key_from_dt(now, slot_min)
    # 슬롯 구간
    try:
        slot_dt = datetime.strptime(sk, "%Y%m%d%H%M")
    except ValueError:
        slot_dt = now.replace(second=0, microsecond=0)
    slot_end = slot_dt + timedelta(minutes=max(1, slot_min))
    ts0 = slot_dt.strftime("%Y-%m-%d %H:%M:%S")
    ts1 = slot_end.strftime("%Y-%m-%d %H:%M:%S")
    day = slot_dt.strftime("%Y-%m-%d")

    rows = db.conn.execute(
        "SELECT a.code AS code, a.tick_count AS tick_count, a.last_tick_ts AS last_tick_ts, "
        "b.price AS last_price "
        "FROM ("
        "  SELECT code, COUNT(*) AS tick_count, MAX(ts) AS last_tick_ts "
        "  FROM ls_ws_ticks WHERE ts >= %s AND ts < %s "
        "  GROUP BY code"
        ") a "
        "LEFT JOIN ls_ws_ticks b ON b.code = a.code AND b.ts = a.last_tick_ts",
        (ts0, ts1),
    ).fetchall()
    if not rows:
        return 0

    codes = [str(r["code"]) for r in rows]
    names = _name_map(db, codes)
    scan_time = now.strftime("%Y-%m-%d %H:%M:%S")
    n = 0
    for r in rows:
        code = str(r["code"])
        # 당일 1분봉 수 + 최신 봉
        cinfo = db.conn.execute(
            "SELECT COUNT(*) n, MAX(datetime) mx FROM ls_ws_candles "
            "WHERE code=%s AND tf_min=1 AND datetime LIKE %s",
            (code, f"{day}%"),
        ).fetchone()
        c_n = int((cinfo or {}).get("n") or 0)
        c_mx = (cinfo or {}).get("mx")
        # 간단 갭: 장 시작~해당 슬롯까지 기대 1분봉 − 보유
        expect = sum(1 for s in _market_1m_slots(day) if s <= sk)
        gap = max(0, expect - c_n) if expect > 0 else 0
        price = float(r.get("last_price") or 0)
        db.conn.execute(
            "INSERT INTO ls_universe_history "
            "(slot_key, scan_time, code, name, price, tick_count, candle_1m_count, "
            " last_tick_ts, last_candle_dt, gap_1m_missing, source, note) "
            "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
            "ON DUPLICATE KEY UPDATE "
            "scan_time=VALUES(scan_time), name=VALUES(name), price=VALUES(price), "
            "tick_count=VALUES(tick_count), candle_1m_count=VALUES(candle_1m_count), "
            "last_tick_ts=VALUES(last_tick_ts), last_candle_dt=VALUES(last_candle_dt), "
            "gap_1m_missing=VALUES(gap_1m_missing), note=VALUES(note)",
            (
                sk,
                scan_time,
                code,
                names.get(code, ""),
                price,
                int(r.get("tick_count") or 0),
                c_n,
                r.get("last_tick_ts"),
                str(c_mx or "")[:20] or None,
                int(gap),
                "ls_ws",
                f"slot_min={slot_min}",
            ),
        )
        n += 1
    return n


def backfill_day(db, day: str, *, slot_min: int = 1) -> Tuple[int, int]:
    """해당 일의 ls_ws_ticks 를 분 슬롯으로 묶어 히스토리 upsert. (slots, rows)"""
    _ensure_table(db)
    d = _day_compact(day)
    ts0 = f"{day} 00:00:00"
    ts1 = f"{day} 23:59:59.999"
    slots = db.conn.execute(
        "SELECT DATE_FORMAT(ts, '%%Y%%m%%d%%H%%i') sk FROM ls_ws_ticks "
        "WHERE ts >= %s AND ts <= %s GROUP BY sk ORDER BY sk",
        (ts0, ts1),
    ).fetchall()
    # DATE_FORMAT minute is exact; for slot_min>1 floor in python
    seen = set()
    slot_list: List[str] = []
    for r in slots:
        sk_raw = str(r["sk"])
        if len(sk_raw) < 12:
            continue
        if slot_min <= 1:
            sk = sk_raw[:12]
        else:
            dt = datetime.strptime(sk_raw[:12], "%Y%m%d%H%M")
            sk = _slot_key_from_dt(dt, slot_min)
        if sk not in seen:
            seen.add(sk)
            slot_list.append(sk)
    total_rows = 0
    for sk in slot_list:
        if not sk.startswith(d):
            continue
        total_rows += collect_slot(db, slot_key=sk, slot_min=slot_min)
    return len(slot_list), total_rows


def quality_report(db, day: str, *, top: int = 20, since: str = "00:00:00") -> None:
    """LS 틱/봉/히스토리/키움후보 대비 퀄리티 출력."""
    _ensure_table(db)
    d = _day_compact(day)
    start_ts = f"{day} {since}"
    end_ts = f"{day} 23:59:59.999"
    s_compact = since.replace(":", "")
    while len(s_compact) < 6:
        s_compact += "0"
    start_dt_str = f"{d}{s_compact[:6]}"
    start_hm_str = f"{d}{s_compact[:4]}"

    print("=" * 72)
    print(f"LS 종목 히스토리 퀄리티  — day={day} (분석 기준 시각: {since} ~ 23:59:59)")
    print("=" * 72)

    print(f"\n[1. 실매매 틱 / 캔들 / 호가 적재 현황 — {day} {since} 이후]")
    tick_ls = db.conn.execute(
        "SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(ts) mn, MAX(ts) mx "
        "FROM ls_ws_ticks WHERE ts >= %s AND ts <= %s",
        (start_ts, end_ts),
    ).fetchone()
    print(f" 🔸 [LS 틱   (ls_ws_ticks)]     n={tick_ls['n']:>9,d} | codes={tick_ls['codes']:>4} | range={tick_ls['mn']} ~ {tick_ls['mx']}")

    try:
        tick_kis = db.conn.execute(
            "SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(tick_time) mn, MAX(tick_time) mx "
            "FROM ws_ticks WHERE tick_time >= %s AND tick_time <= %s",
            (start_dt_str, f"{d}235959"),
        ).fetchone()
        print(f" 🔸 [KIS 틱  (ws_ticks)]        n={tick_kis['n']:>9,d} | codes={tick_kis['codes']:>4} | range={tick_kis['mn']} ~ {tick_kis['mx']}")
    except Exception as e:
        print(f" 🔸 [KIS 틱  (ws_ticks)]        조회스킵({e})")

    cndl_ls = db.conn.execute(
        "SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(datetime) mn, MAX(datetime) mx "
        "FROM ls_ws_candles WHERE tf_min=1 AND datetime >= %s AND datetime <= %s",
        (start_ts, end_ts),
    ).fetchone()
    print(f" 🔸 [LS 분봉 (ls_ws_candles)]   n={cndl_ls['n']:>9,d} | codes={cndl_ls['codes']:>4} | range={cndl_ls['mn']} ~ {cndl_ls['mx']}")

    try:
        cndl_kis = db.conn.execute(
            "SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(candle_time) mn, MAX(candle_time) mx "
            "FROM ws_candles WHERE timeframe=1 AND candle_time >= %s AND candle_time <= %s",
            (start_hm_str, f"{d}2359"),
        ).fetchone()
        print(f" 🔸 [KIS 분봉 (ws_candles)]      n={cndl_kis['n']:>9,d} | codes={cndl_kis['codes']:>4} | range={cndl_kis['mn']} ~ {cndl_kis['mx']}")
    except Exception as e:
        print(f" 🔸 [KIS 분봉 (ws_candles)]      조회스킵({e})")

    try:
        ob_ls = db.conn.execute(
            "SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(snap_time) mn, MAX(snap_time) mx "
            "FROM ls_ws_orderbook WHERE snap_time >= %s AND snap_time <= %s",
            (start_dt_str, f"{d}235959"),
        ).fetchone()
        print(f" 🔸 [LS 호가 (ls_ws_orderbook)] n={ob_ls['n']:>9,d} | codes={ob_ls['codes']:>4} | range={ob_ls['mn']} ~ {ob_ls['mx']}")
    except Exception as e:
        print(f" 🔸 [LS 호가 (ls_ws_orderbook)] 조회스킵({e})")

    try:
        ob_kis = db.conn.execute(
            "SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(snap_time) mn, MAX(snap_time) mx "
            "FROM ws_orderbook WHERE snap_time >= %s AND snap_time <= %s",
            (start_dt_str, f"{d}235959"),
        ).fetchone()
        print(f" 🔸 [KIS 호가 (ws_orderbook)]    n={ob_kis['n']:>9,d} | codes={ob_kis['codes']:>4} | range={ob_kis['mn']} ~ {ob_kis['mx']}")
    except Exception as e:
        print(f" 🔸 [KIS 호가 (ws_orderbook)]    조회스킵({e})")

    hist = db.conn.execute(
        "SELECT COUNT(*) n, COUNT(DISTINCT code) codes, COUNT(DISTINCT slot_key) slots "
        "FROM ls_universe_history WHERE slot_key LIKE %s",
        (f"{d}%",),
    ).fetchone()
    print(
        f"[ls_universe_history] rows={hist['n']:,} codes={hist['codes']} "
        f"slots={hist['slots']}"
    )

    # 종목별 틱/봉
    per = db.conn.execute(
        "SELECT code, COUNT(*) ticks, MIN(ts) mn, MAX(ts) mx "
        "FROM ls_ws_ticks WHERE ts >= %s AND ts <= %s "
        "GROUP BY code ORDER BY ticks DESC LIMIT %s",
        (f"{day} 00:00:00", f"{day} 23:59:59.999", int(top)),
    ).fetchall()
    print(f"\n--- 틱 상위 {top} 종목 ---")
    print(f"{'code':8} {'ticks':>8} {'candle_1m':>9} {'gap_est':>7} {'last_tick'}")
    expect_full = len(_market_1m_slots(day))
    now_sk = _slot_key_from_dt(datetime.now(), 1)
    if now_sk.startswith(d):
        expect_now = sum(1 for s in _market_1m_slots(day) if s <= now_sk)
    else:
        expect_now = expect_full
    for r in per:
        code = str(r["code"])
        cn = db.conn.execute(
            "SELECT COUNT(*) n FROM ls_ws_candles WHERE code=%s AND tf_min=1 AND datetime LIKE %s",
            (code, f"{day}%"),
        ).fetchone()["n"]
        gap = max(0, int(expect_now) - int(cn))
        print(
            f"{code:8} {int(r['ticks']):8d} {int(cn):9d} {gap:7d} {r['mx']}"
        )

    # 키움 후보 히스토리와 교집합 (오늘)
    try:
        kw = db.conn.execute(
            "SELECT COUNT(DISTINCT code) n FROM target_candidates_history "
            "WHERE slot_key LIKE %s",
            (f"{d}%",),
        ).fetchone()["n"]
        both = db.conn.execute(
            "SELECT COUNT(DISTINCT h.code) n FROM ls_universe_history h "
            "INNER JOIN target_candidates_history t "
            "  ON t.code=h.code AND t.slot_key LIKE %s "
            "WHERE h.slot_key LIKE %s",
            (f"{d}%", f"{d}%"),
        ).fetchone()["n"]
        ls_only = db.conn.execute(
            "SELECT COUNT(DISTINCT code) n FROM ls_universe_history WHERE slot_key LIKE %s",
            (f"{d}%",),
        ).fetchone()["n"]
        print(f"\n--- 키움 후보(history) vs LS 히스토리 ---")
        print(f"키움 distinct codes(today): {kw}")
        print(f"LS history distinct:       {ls_only}")
        print(f"교집합(대략):              {both}")
    except Exception as e:
        print(f"(키움 교차 비교 스킵: {e})")

    print("\n[2. 증권사 간 시세 및 레이텐시(수신 지연) 3자 비교 실측]")
    def _fmt(val):
        return f"{val:.1f}ms" if val is not None else "N/A"

    try:
        v1 = db.conn.execute(
            "SELECT COUNT(*) n, AVG(ABS(diff_pct)) avg_abs, MAX(ABS(diff_pct)) max_abs, "
            "AVG(kis_age_ms) kis_ms, AVG(kiwoom_age_ms) kw_ms "
            "FROM ws_price_validation WHERE ts >= %s AND ts <= %s AND diff_pct IS NOT NULL",
            (start_ts, end_ts),
        ).fetchone()
        if v1 and v1["n"] > 0:
            print(f" 1) KIS ↔ 키움 시세 갭 실측 (ws_price_validation, 표본: {v1['n']:,}건)")
            print(f"    👉 가격 괴리율: 평균 {v1['avg_abs']:.4f}% (최대 {v1['max_abs']:.4f}%)")
            print(f"    👉 수신 지연(Age): KIS 평균 {_fmt(v1['kis_ms'])} vs 키움 평균 {_fmt(v1['kw_ms'])}")
        else:
            print(" 1) KIS ↔ 키움 (ws_price_validation): 당일 실측 표본 없음")
    except Exception as e:
        print(f" (ws_price_validation 스킵: {e})")

    try:
        v2 = db.conn.execute(
            "SELECT COUNT(*) n, "
            "AVG(ABS(diff_kis_ls_pct)) kis_ls_avg, MAX(ABS(diff_kis_ls_pct)) kis_ls_max, "
            "AVG(ABS(diff_kw_ls_pct)) kw_ls_avg, MAX(ABS(diff_kw_ls_pct)) kw_ls_max, "
            "AVG(kis_age_ms) kis_ms, AVG(kiwoom_age_ms) kw_ms, AVG(ls_age_ms) ls_ms "
            "FROM ws_price_validation_ls "
            "WHERE ts >= %s AND ts <= %s AND diff_kis_ls_pct IS NOT NULL",
            (start_ts, end_ts),
        ).fetchone()
        if v2 and v2["n"] > 0:
            print(f" 2) KIS ↔ 키움 ↔ LS 삼각 시세 갭 실측 (ws_price_validation_ls, 표본: {v2['n']:,}건)")
            print(f"    👉 KIS  vs LS 가격 괴리율: 평균 {v2['kis_ls_avg']:.4f}% (최대 {v2['kis_ls_max']:.4f}%)")
            print(f"    👉 키움 vs LS 가격 괴리율: 평균 {v2['kw_ls_avg']:.4f}% (최대 {v2['kw_ls_max']:.4f}%)")
            print(f"    👉 수신 지연(Age): KIS 평균 {_fmt(v2['kis_ms'])} vs 키움 평균 {_fmt(v2['kw_ms'])} vs LS 평균 {_fmt(v2['ls_ms'])}")
        else:
            print(" 2) KIS ↔ 키움 ↔ LS (ws_price_validation_ls): 당일 실측 표본 없음")
    except Exception as e:
        print(f" (ws_price_validation_ls 스킵: {e})")

    # 최근 슬롯 커버
    recent = db.conn.execute(
        "SELECT slot_key, COUNT(*) n FROM ls_universe_history "
        "WHERE slot_key LIKE %s GROUP BY slot_key ORDER BY slot_key DESC LIMIT 8",
        (f"{d}%",),
    ).fetchall()
    print("\n--- 최근 슬롯 종목수 ---")
    for r in recent:
        print(f"  {r['slot_key']}: {r['n']} codes")
    print("=" * 72)


def main() -> int:
    ap = argparse.ArgumentParser(description="LS universe history collect + quality")
    ap.add_argument("--day", default="", help="YYYY-MM-DD (기본=오늘)")
    ap.add_argument("--slot-min", type=int, default=1, help="슬롯 분 (기본 1)")
    ap.add_argument("--collect", action="store_true", help="현재 슬롯 1회 수집")
    ap.add_argument("--backfill", action="store_true", help="해당일 틱→슬롯 백필")
    ap.add_argument("--quality", action="store_true", help="퀄리티 리포트 출력")
    ap.add_argument("--loop", type=int, default=0, help="N초마다 --collect 반복 (0=OFF)")
    ap.add_argument("--quality-every", type=int, default=0, help="루프 N회마다 quality")
    ap.add_argument("--top", type=int, default=20, help="퀄리티 상위 종목 수")
    ap.add_argument("--since", default="00:00:00", help="이 시각(HH:MM:SS) 이후부터 분석 (기본 00:00:00)")
    args = ap.parse_args()

    day = _parse_day(args.day)
    from database import TradeDB

    db = TradeDB()
    try:
        _ensure_table(db)
        print(f"[ok] ls_universe_history ready | day={day} slot_min={args.slot_min}")

        if args.backfill:
            ns, nr = backfill_day(db, day, slot_min=max(1, args.slot_min))
            print(f"[backfill] slots={ns} upsert_rows≈{nr}")

        if args.collect and not args.loop:
            n = collect_slot(db, slot_min=max(1, args.slot_min))
            print(f"[collect] upsert codes={n} slot={_slot_key_from_dt(datetime.now(), args.slot_min)}")

        if args.quality and not args.loop:
            quality_report(db, day, top=max(5, args.top), since=args.since)

        if args.loop and args.loop > 0:
            i = 0
            print(f"[loop] every {args.loop}s — Ctrl+C to stop")
            while True:
                n = collect_slot(db, slot_min=max(1, args.slot_min))
                sk = _slot_key_from_dt(datetime.now(), args.slot_min)
                print(f"[{datetime.now():%H:%M:%S}] collect slot={sk} codes={n}")
                i += 1
                if args.quality_every > 0 and i % args.quality_every == 0:
                    quality_report(db, day, top=max(5, args.top), since=args.since)
                time.sleep(max(5, args.loop))

        # 아무 플래그 없으면 backfill+quality 기본
        if not any([args.collect, args.backfill, args.quality, args.loop]):
            ns, nr = backfill_day(db, day, slot_min=max(1, args.slot_min))
            print(f"[backfill] slots={ns} upsert_rows≈{nr}")
            quality_report(db, day, top=max(5, args.top), since=args.since)
    finally:
        try:
            db.close()
        except Exception:
            pass
    return 0


if __name__ == "__main__":
    raise SystemExit(main())