#!/usr/bin/env python3
"""
컬럼 직접접근(TickColumnView) vs dict판 bit-identical 검증 — 1단계(momentum/whipsaw).

같은 프로세스에서 원본 dict 틱과 공유메모리 컬럼 뷰를 나란히 돌려,
아래 핫함수들의 산출물이 완전히 동일한지 확인한다:
  - whipsaw_filter.aggregate_ticks_subbars (via collect_whipsaw_ticks)
  - momentum_tick_replay.collect_minute_ticks
  - momentum_tick_replay.align_momentum_entry_from_ticks
  - momentum_tick_replay.try_momentum_sell_on_ticks (check_sell 은 결정적 스텁으로 격리)

데이터는 로더 계약(분 버킷 tick_time 오름차순)대로 생성한다.
"""
import os
import random
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import kis_trader.engine.momentum_tick_replay as mtr
import kis_trader.engine.tail_tick_replay as ttr
from kis_trader.backtest.shared_ticks import build_shared_ticks, TickColumnView
from kis_trader.engine.whipsaw_filter import (
    aggregate_ticks_subbars,
    collect_whipsaw_ticks,
)


def gen_ticks(seed=0):
    """{code: {minute: [tick,...]}} — 분 버킷은 tick_time 오름차순(로더 계약)."""
    rnd = random.Random(seed)
    data = {}
    codes = [f"{100000 + c}" for c in range(4)]
    for code in codes:
        bucket = {}
        # 09:00 부터 임의 분들 (분 경계 올바르게)
        for m in range(rnd.randint(3, 8)):
            total_min = 9 * 60 + m
            hh, mm = divmod(total_min, 60)
            minute = f"20260704{hh:02d}{mm:02d}"  # YYYYMMDDHHMM (12자리)
            n = rnd.randint(0, 40)
            ticks = []
            sec = 0
            for _ in range(n):
                sec += rnd.randint(0, 3)  # 오름차순 보장
                ss = min(59, sec)
                # 대부분 14자리, 가끔 12자리(초 생략). 로더는 len<12 를 드롭하므로 생성 안 함.
                r = rnd.random()
                if r < 0.08:
                    tt = minute  # 12자리
                else:
                    tt = f"{minute}{ss:02d}"  # 14자리
                px = 0.0 if rnd.random() < 0.05 else round(1000 + rnd.uniform(-50, 50), 1)
                vol = rnd.randint(0, 500)
                ticks.append({
                    "code": code,
                    "tick_time": tt,
                    "price": px,
                    "volume": vol,
                    "source": rnd.choice(["ws", "ws_recon", "rest"]),
                })
            # 로더 계약 재현: SQL ORDER BY tick_time (문자열) 로 정렬된 버킷을 준다.
            ticks.sort(key=lambda x: str(x["tick_time"]))
            bucket[minute] = ticks
        data[code] = bucket
    return data, codes


def materialize_view(view):
    """뷰를 (tick_time, price, volume) 튜플 리스트로 — 순서 확인용."""
    owner = view.owner
    out = []
    for i in view.iter_idx():
        out.append((
            owner._tick_time[i].decode("utf-8"),
            float(owner._price[i]),
            int(owner._volume[i]),
        ))
    return out


def dict_tuples(ticks):
    return [(str(t.get("tick_time") or ""), float(t.get("price") or 0), int(t.get("volume") or 0))
            for t in ticks]


def main():
    fails = 0
    checks = 0

    for seed in range(40):
        data, codes = gen_ticks(seed)
        store = build_shared_ticks(data)
        if store is None:
            print("build_shared_ticks 반환 None — 폴백(테스트 불가)")
            return 1
        try:
            shared = dict(store.attach_mapping())  # {code: SharedBucketMapping}

            for code in codes:
                bucket = data[code]
                minutes = list(bucket.keys())

                # ── 1) collect_minute_ticks: dict vs 컬럼 ───────────────
                for mk in minutes:
                    d = mtr.collect_minute_ticks(data, code, mk)
                    v = mtr.collect_minute_ticks(shared, code, mk)
                    assert not isinstance(d, TickColumnView)
                    assert isinstance(v, TickColumnView) or len(bucket[mk]) == 0
                    dt = dict_tuples(d)
                    vt = materialize_view(v) if isinstance(v, TickColumnView) else dict_tuples(v)
                    checks += 1
                    if dt != vt:
                        fails += 1
                        print(f"[collect_minute] mismatch seed={seed} code={code} mk={mk}")
                        print("  dict:", dt[:5], "...")
                        print("  col :", vt[:5], "...")

                # ── 2) collect_whipsaw_ticks + aggregate: dict vs 컬럼 ──
                for mk in minutes:
                    for lb in (60, 90, 180):
                        dticks = collect_whipsaw_ticks(data, code, mk, lb, 1)
                        vticks = collect_whipsaw_ticks(shared, code, mk, lb, 1)
                        for sb in (10, 30, 60):
                            da = aggregate_ticks_subbars(dticks, sb, lb)
                            va = aggregate_ticks_subbars(vticks, sb, lb)
                            checks += 1
                            if da != va:
                                fails += 1
                                print(f"[aggregate] mismatch seed={seed} code={code} mk={mk} lb={lb} sb={sb}")
                                print("  dict:", da[:2])
                                print("  col :", va[:2])

                # ── 2b) cap_by_tick_time_le: dict 컴프리헨션 vs 뷰 캡핑 ──
                for mk in minutes:
                    d = mtr.collect_minute_ticks(data, code, mk)
                    v = mtr.collect_minute_ticks(shared, code, mk)
                    if not isinstance(v, TickColumnView):
                        continue
                    for sk in (f"{mk}00", f"{mk}15", f"{mk}30", f"{mk}59", mk):
                        cap_d = [tk for tk in d if str(tk.get("tick_time") or "")[:14] <= sk[:14]]
                        cap_v = v.cap_by_tick_time_le(sk[:14])
                        checks += 1
                        if dict_tuples(cap_d) != materialize_view(cap_v):
                            fails += 1
                            print(f"[cap] mismatch seed={seed} code={code} mk={mk} sk={sk}")
                            print("  dict:", dict_tuples(cap_d)[:4])
                            print("  col :", materialize_view(cap_v)[:4])

                # ── 2c) TAIL: collect_bar_ticks / limit_fill / align (dict vs 컬럼) ──
                for mk in minutes:
                    for tf in (1, 3, 5):
                        db_bar = ttr.collect_bar_ticks(data, code, mk, tf)
                        vw_bar = ttr.collect_bar_ticks(shared, code, mk, tf)
                        # 순서·값 동일
                        vt = materialize_view(vw_bar) if isinstance(vw_bar, TickColumnView) else dict_tuples(vw_bar)
                        checks += 1
                        if dict_tuples(db_bar) != vt:
                            fails += 1
                            print(f"[tail.collect_bar] seed={seed} code={code} mk={mk} tf={tf}")
                        for lp in (0.0, 900.0, 1000.0, 1050.0, 99999.0):
                            for slip in (0.0, 0.1):
                                rd = ttr.try_limit_fill_from_ticks(db_bar, lp, slip)
                                rv = ttr.try_limit_fill_from_ticks(vw_bar, lp, slip)
                                checks += 1
                                if rd != rv:
                                    fails += 1
                                    print(f"[tail.limit_fill] seed={seed} code={code} mk={mk} tf={tf} lp={lp} slip={slip}: {rd} vs {rv}")
                        for fo in (0.0, 777.0):
                            rd = ttr.align_entry_price_from_ticks(db_bar, fo)
                            rv = ttr.align_entry_price_from_ticks(vw_bar, fo)
                            checks += 1
                            if rd != rv:
                                fails += 1
                                print(f"[tail.align] seed={seed} code={code} mk={mk} tf={tf} fo={fo}: {rd} vs {rv}")

                # ── 3) align_momentum_entry_from_ticks: dict vs 컬럼 ────
                for mk in minutes:
                    for min_tt in ("", f"{mk}30", f"{mk}05"):
                        rd = mtr.align_momentum_entry_from_ticks(data, code, mk, 999.0, {}, min_tick_time=min_tt)
                        rv = mtr.align_momentum_entry_from_ticks(shared, code, mk, 999.0, {}, min_tick_time=min_tt)
                        checks += 1
                        if rd != rv:
                            fails += 1
                            print(f"[align_entry] mismatch seed={seed} code={code} mk={mk} min_tt={min_tt}")
                            print("  dict:", rd, " col:", rv)

                # ── 4) try_momentum_sell_on_ticks: dict vs 컬럼 (check_sell 스텁) ──
                recorded = {"candles": []}

                def fake_check_sell(position, candle, params, *, is_eod=False):
                    recorded["candles"].append((candle["candle_time"], round(candle["high"], 3),
                                                round(candle["low"], 3), round(candle["close"], 3), is_eod))
                    # 결정적 트리거: close 가 entry_price*1.01 이상이면 매도
                    if candle["close"] >= position["entry_price"] * 1.01:
                        return ("어깨컷", candle["close"])
                    return None

                orig = mtr.check_sell_signal_momentum_live
                mtr.check_sell_signal_momentum_live = fake_check_sell
                try:
                    for mk in minutes:
                        for entry_time in (f"{mk}00", f"{mk}10", f"{mk}30"):
                            for is_eod in (False, True):
                                pos_d = {"entry_price": 1000.0, "max_price": 1000.0, "entry_time": entry_time}
                                pos_v = {"entry_price": 1000.0, "max_price": 1000.0, "entry_time": entry_time}
                                d_ticks = mtr.collect_minute_ticks(data, code, mk)
                                v_ticks = mtr.collect_minute_ticks(shared, code, mk)

                                recorded["candles"] = []
                                rd = mtr.try_momentum_sell_on_ticks(pos_d, d_ticks, {}, is_eod=is_eod, entry_time=entry_time)
                                cand_d = list(recorded["candles"])

                                recorded["candles"] = []
                                rv = mtr.try_momentum_sell_on_ticks(pos_v, v_ticks, {}, is_eod=is_eod, entry_time=entry_time)
                                cand_v = list(recorded["candles"])

                                checks += 1
                                if rd != rv or cand_d != cand_v or pos_d.get("max_price") != pos_v.get("max_price"):
                                    fails += 1
                                    print(f"[sell] mismatch seed={seed} code={code} mk={mk} entry={entry_time} eod={is_eod}")
                                    print("  ret dict:", rd, " col:", rv)
                                    print("  maxp dict:", pos_d.get("max_price"), " col:", pos_v.get("max_price"))
                                    print("  candles dict:", cand_d[:3], "...(", len(cand_d), ")")
                                    print("  candles col :", cand_v[:3], "...(", len(cand_v), ")")
                finally:
                    mtr.check_sell_signal_momentum_live = orig
        finally:
            store.unlink()

    print(f"\n총 검증 {checks}건, 불일치 {fails}건")
    return 0 if fails == 0 else 2


if __name__ == "__main__":
    sys.exit(main())
