#!/usr/bin/env python3 """Simulate load/save of fills across a fake restart — no network.""" from __future__ import annotations import json import sys import tempfile import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.session_persist import ( # noqa: E402 SessionState, SessionStore, trim_for_ui, utc_day, ) def test_same_day_resume() -> None: with tempfile.TemporaryDirectory() as td: path = Path(td) / "session_state.json" store = SessionStore(path) now = time.time() st = store.load_or_rotate(now=now) assert st.day == utc_day(now) assert st.fills == [] # Record fills like the bot would t0 = now - 3600 # 1h ago for i in range(5): side = "buy" if i % 2 == 0 else "sell" st.fills.append( { "id": f"f{i}", "ts": t0 + i * 60, "side": side, "price": 80000 + i, "size_btc": 0.001, "fee_usd": 0.0, "mid_at_fill": 80000 + i, "notional_usd": 80.0, } ) if side == "buy": st.buy_qty += 0.001 st.buy_notional += 80.0 else: st.sell_qty += 0.001 st.sell_notional += 80.0 st.day_start_equity = 1000.0 st.day_start_ts = t0 store.save(st) # Fake restart: new store instance, load again store2 = SessionStore(path) st2 = store2.load_or_rotate(now=now + 10) assert st2.day == st.day assert len(st2.fills) == 5 assert st2.fills_buy == 3 assert st2.fills_sell == 2 assert abs(st2.day_start_equity - 1000.0) < 1e-9 assert [f["id"] for f in st2.fills] == ["f0", "f1", "f2", "f3", "f4"] print("OK same_day_resume fills=5 equity0=1000") def test_new_day_rotates() -> None: with tempfile.TemporaryDirectory() as td: path = Path(td) / "session_state.json" store = SessionStore(path) # Pretend yesterday yesterday_ts = time.time() - 86400 st = SessionState( day=utc_day(yesterday_ts), day_start_equity=999.0, day_start_ts=yesterday_ts, fills=[ { "id": "old", "ts": yesterday_ts, "side": "buy", "price": 1, "size_btc": 1, "fee_usd": 0, "mid_at_fill": 1, "notional_usd": 1, } ], ) store.save(st) now = time.time() st2 = store.load_or_rotate(now=now) assert st2.day == utc_day(now) assert st2.fills == [] assert st2.day_start_equity == 0.0 archive = path.with_name(f"session_state_{utc_day(yesterday_ts)}.json") assert archive.exists(), f"missing archive {archive}" archived = json.loads(archive.read_text()) assert len(archived["fills"]) == 1 print("OK new_day_rotates archived", archive.name) def test_trim_keeps_5h() -> None: now = time.time() fills = [] # 10 fills over 6 hours + 3 older than 5h for i in range(13): # i=0 at 6h ago, i=12 at now ts = now - (6 * 3600) + i * (6 * 3600 / 12) fills.append({"id": str(i), "ts": ts, "side": "buy", "price": 1, "size_btc": 0.01}) ui = trim_for_ui(fills, now=now, retain_hours=5.0, max_fills=500) assert all(now - f["ts"] <= 5 * 3600 + 1 for f in ui) assert len(ui) < len(fills) assert len(ui) >= 8 # most of the 5h window print(f"OK trim_keeps_5h ui={len(ui)}/{len(fills)}") def test_trim_caps_500() -> None: now = time.time() fills = [ {"id": str(i), "ts": now - 60, "side": "buy", "price": 1, "size_btc": 0.01} for i in range(600) ] ui = trim_for_ui(fills, now=now, retain_hours=5.0, max_fills=500) assert len(ui) == 500 assert ui[0]["id"] == "100" assert ui[-1]["id"] == "599" print("OK trim_caps_500") def test_seed_from_jsonl_no_invent() -> None: with tempfile.TemporaryDirectory() as td: td = Path(td) path = td / "session_state.json" jsonl = td / "fills_live.jsonl" now = time.time() day = utc_day(now) rows = [ { "id": "a1", "ts": now - 100, "side": "buy", "price": 80000, "size_btc": 0.001, "notional_usd": 80, "mid_at_fill": 80001, }, { "id": "a2", "ts": now - 50, "side": "sell", "price": 80010, "size_btc": 0.001, "notional_usd": 80.01, "mid_at_fill": 80005, }, # other day — must be ignored { "id": "old", "ts": now - 90000, "side": "buy", "price": 1, "size_btc": 1, "notional_usd": 1, "mid_at_fill": 1, }, ] with jsonl.open("w") as fh: for r in rows: fh.write(json.dumps(r) + "\n") store = SessionStore(path) st = store.load_or_rotate(now=now) n = store.seed_from_jsonl(st, jsonl, now=now) assert n == 2, n assert [f["id"] for f in st.fills] == ["a1", "a2"] assert st.fills_buy == 1 and st.fills_sell == 1 # Second seed must not duplicate n2 = store.seed_from_jsonl(st, jsonl, now=now) assert n2 == 0 # Append a new jsonl row (simulates old bot still writing) → merge with jsonl.open("a") as fh: fh.write( json.dumps( { "id": "a3", "ts": now - 10, "side": "buy", "price": 80020, "size_btc": 0.002, "notional_usd": 160.04, "mid_at_fill": 80015, } ) + "\n" ) n3 = store.seed_from_jsonl(st, jsonl, now=now) assert n3 == 1 and [f["id"] for f in st.fills] == ["a1", "a2", "a3"] store.save(st) st2 = SessionStore(path).load_or_rotate(now=now) assert len(st2.fills) == 3 print("OK seed_from_jsonl_no_invent day=", day) def test_boot_sync_no_fake_fill_logic() -> None: """Document the LiveEngine contract: first refresh must not append a fill.""" # Pure logic mirror of live_engine.refresh_account boot branch prev = 0.0 boot_synced = False fills = [] inv = 0.00139 # open position at restart if not boot_synced: prev = inv boot_synced = True elif abs(inv - prev) > 1e-9: fills.append({"side": "buy", "size": abs(inv - prev)}) assert boot_synced and prev == inv and fills == [] # Real delta after boot inv2 = inv + 0.0005 if not boot_synced: prev = inv2 boot_synced = True elif abs(inv2 - prev) > 1e-9: fills.append({"side": "buy", "size": abs(inv2 - prev)}) prev = inv2 assert len(fills) == 1 and abs(fills[0]["size"] - 0.0005) < 1e-12 print("OK boot_sync_no_fake_fill_logic") if __name__ == "__main__": test_same_day_resume() test_new_day_rotates() test_trim_keeps_5h() test_trim_caps_500() test_seed_from_jsonl_no_invent() test_boot_sync_no_fake_fill_logic() print("ALL PASS")