#!/usr/bin/env python3
"""
스모크: 1M→N분 완전버킷 롤업 + confirm/merge volume upsert.

근본원인(2026-07-16 샘표): 불완전 롤업 삽입 + 동일 candle_time append 중복
→ RAM prior volume 왜곡 → 실매 vol 통과 / 백테 탈락.

실행:
  python3 -u scripts/smoke_candle_upsert_rollup.py
"""
from __future__ import annotations

import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from kis_trader.engine.candle_rollup import floor_candle_time_to_tf, rollup_1m_bars_to_tf
from kis_trader.ws.kis_ws import CandleAggregator


def bar(ct, o, h, l, c, v, src="ws"):
    return {
        "candle_time": ct,
        "open": o,
        "high": h,
        "low": l,
        "close": c,
        "volume": v,
        "source": src,
    }


def main() -> None:
    assert floor_candle_time_to_tf("202607160912", 3) == "202607160912"
    assert floor_candle_time_to_tf("202607160913", 3) == "202607160912"
    assert floor_candle_time_to_tf("202607160914", 3) == "202607160912"

    partial = [
        bar("202607160912", 100, 101, 99, 100, 100),
        bar("202607160913", 100, 102, 99, 101, 200),
    ]
    assert rollup_1m_bars_to_tf(partial, 3) == []

    full = partial + [bar("202607160914", 101, 110, 100, 105, 4226)]
    rolled = rollup_1m_bars_to_tf(full, 3)
    assert len(rolled) == 1
    assert rolled[0]["candle_time"] == "202607160912"
    assert rolled[0]["volume"] == 100 + 200 + 4226

    more = full + [
        bar("202607160915", 105, 106, 104, 105, 50),
        bar("202607160916", 105, 107, 104, 106, 60),
    ]
    assert len(rollup_1m_bars_to_tf(more, 3)) == 1

    agg = CandleAggregator(db=None, timeframes=[1, 3])
    code = "007540"
    assert agg.merge_confirmed_bars(
        code, 3,
        [bar("202607160912", 43000, 44000, 42000, 43500, 515, "rollup_1m")],
        log_tag="smoke_partial",
    ) == 1
    assert agg.merge_confirmed_bars(
        code, 3,
        [bar("202607160912", 43000, 44500, 42000, 43800, 4526, "rest")],
        log_tag="smoke_full",
    ) == 1
    buf = agg._confirmed[(code, 3)]
    assert len(buf) == 1 and buf[0]["volume"] == 4526
    assert agg.merge_confirmed_bars(
        code, 3,
        [bar("202607160912", 43000, 44000, 42000, 43700, 100, "ws")],
        log_tag="smoke_small",
    ) == 0
    assert buf[0]["volume"] == 4526

    agg2 = CandleAggregator(db=None, timeframes=[3])
    key = (code, 3)
    agg2.merge_confirmed_bars(
        code, 3,
        [bar("202607160912", 43000, 44000, 42000, 43500, 515, "rollup_1m")],
    )
    with agg2._lock:
        confirmed = agg2._confirm_current_bucket(key, {
            "candle_time": "202607160912",
            "open": 43000,
            "high": 44200,
            "low": 42000,
            "close": 43600,
            "volume": 800,
            "source": "ws",
        })
    assert len(agg2._confirmed[key]) == 1
    assert confirmed["volume"] == 800
    with agg2._lock:
        agg2._confirm_current_bucket(key, {
            "candle_time": "202607160912",
            "open": 43000,
            "high": 44100,
            "low": 42000,
            "close": 43400,
            "volume": 100,
            "source": "ws",
        })
    assert agg2._confirmed[key][0]["volume"] == 800

    # 진행 중 분봉은 REST/merge confirmed 에 넣지 않음 (장초 직전봉% 왜곡 방지)
    import datetime as _dt
    agg3 = CandleAggregator(db=None, timeframes=[1])
    code2 = "333050"
    frozen = _dt.datetime(2026, 7, 16, 9, 0, 34)
    # 전일 + 미완성 당일 09:00 을 넣으려 할 때 → 09:00 만 skip
    n = agg3.merge_confirmed_bars(
        code2, 1,
        [
            bar("202607151530", 5280, 5280, 5280, 5280, 960, "rest"),
            bar("202607160900", 5220, 5250, 5200, 5220, 10, "rest"),  # 진행분
        ],
        log_tag="smoke_skip_open",
        skip_incomplete_bucket=True,
        now=frozen,
    )
    assert n == 1
    buf3 = agg3._confirmed[(code2, 1)]
    assert len(buf3) == 1 and buf3[0]["candle_time"] == "202607151530"
    # 이미 들어간 진행분 purge
    agg3._confirmed[(code2, 1)].append(
        bar("202607160900", 5220, 5250, 5200, 5220, 10, "rest")
    )
    agg3.merge_confirmed_bars(
        code2, 1, [], log_tag="smoke_purge", skip_incomplete_bucket=True, now=frozen,
    )
    assert all(
        str(c["candle_time"])[:12] < "202607160900"
        for c in agg3._confirmed[(code2, 1)]
    )

    print("SMOKE_OK candle_upsert_rollup")


if __name__ == "__main__":
    main()