#!/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())