"""Unit tests for P3-DATA-ENGINE-V1 (spec-required coverage list). Covers: timestamp parsing, decimal price parsing, bid/ask validation, volume handling, spread calculation, canonical serialization, hash calculation, bar-period calculation, aggregation, partial-bar handling, checkpoint state transitions, lock state transitions. Hand-computed truth anchors: ts(2023-01-02 03:04:05.100 UTC) = 1672628645100 ms "100.000000" * 1e6 = 100_000_000 ; "100.000100" = 100_000_100 """ import os import sys import tempfile import time sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from tests.harness import Suite, eq, expect # noqa: E402 from engine.util import ( # noqa: E402 parse_timestamp_dotted, apply_tz_offset, in_ts_valid_window, parse_price_to_micro, parse_volume_token, canonical_json_sha256, sha256_bytes, ) from engine.versions import PRICE_SCALE, TIMEFRAMES # noqa: E402 from engine.canonical import ( # noqa: E402 serialize_tick, tick_chunk_content_id, bar_file_content_id, serialize_bars, ) from engine.checkpoint import ( # noqa: E402 validate_transition, CheckpointTransitionError, write_checkpoint, load_checkpoint, CheckpointCorrupt, STATUS_INITIALIZED, STATUS_RUNNING, STATUS_CHUNK_COMMITTED, STATUS_PAUSED, STATUS_FAILED, STATUS_COMPLETED, STATUS_RESUME_BLOCKED, ) from engine.lock import LockHandle, lock_status # noqa: E402 from engine.aggregate import Aggregator # noqa: E402 from engine.config import default_config, config_sha256 # noqa: E402 NOW_MS = int(time.time() * 1000) # --------------------------------------------------------------------------- ts_suite = Suite("unit_timestamp_parsing") @ts_suite.case("dotted_ms_primary") def _(): v, err = parse_timestamp_dotted("2023.01.02 03:04:05.100") return eq(v, 1672628645100, "epoch-ms") + (err,) @ts_suite.case("dotted_s_secondary_ms0") def _(): v, err = parse_timestamp_dotted("2023.01.02 03:04:05") return eq(v, 1672628645000, "epoch-ms(sec-only)") + (err,) @ts_suite.case("tz_offset_applied") def _(): v, _ = parse_timestamp_dotted("2023.01.02 03:04:05.100") shifted = apply_tz_offset(v, 0) return (shifted == 1672628645100, "offset 0 must keep identity: %r" % shifted) @ts_suite.case("tz_offset_minus_120") def _(): v, _ = parse_timestamp_dotted("2023.01.02 03:04:05.100") shifted = apply_tz_offset(v, 120) return eq(shifted, 1672628645100 - 120 * 60000, "tz-120") @ts_suite.case("rejects_bad_shapes") def _(): bad = ["2023.01.02", "2023-01-02 03:04:05", "2023.01.02 03:04:05.12", "2023.01.02T03:04:05.100", "00.01.02 03:04:05.100", "2023.13.02 03:04:05.100", "2023.01.32 03:04:05.100", "2023.01.02 24:04:05.100", "2023.01.02 03:60:05.100", "2023.01.02 03:04:60.100", ""] for b in bad: v, err = parse_timestamp_dotted(b) if err is None: return False, "accepted bad timestamp %r -> %r" % (b, v) return True, "" @ts_suite.case("range_window") def _(): ok = in_ts_valid_window(1672628645100, NOW_MS) ok2 = in_ts_valid_window(946684800000, NOW_MS) # exactly 2000-01-01 ok3 = in_ts_valid_window(946684799999, NOW_MS) # 1 ms before ok4 = in_ts_valid_window(NOW_MS + 8 * 86400000, NOW_MS) # beyond +7d return (ok and ok2 and (not ok3) and (not ok4), "ok=%s ok2=%s ok3=%s ok4=%s" % (ok, ok2, ok3, ok4)) # --------------------------------------------------------------------------- price_suite = Suite("unit_price_parsing") @price_suite.case("decimal_exact") def _(): a, e = parse_price_to_micro("100.000000", PRICE_SCALE) b, e2 = parse_price_to_micro("100.000100", PRICE_SCALE) c, e3 = parse_price_to_micro("1.5", PRICE_SCALE) d, e4 = parse_price_to_micro("0.000001", PRICE_SCALE) return (a == 100000000 and b == 100000100 and c == 1500000 and d == 1 and e is None and e2 is None and e3 is None and e4 is None, "%r %r %r %r" % (a, b, c, d)) @price_suite.case("precision_rule") def _(): v, e = parse_price_to_micro("1.1234567", PRICE_SCALE) return (v is None and e == "MALFORMED_PRICE_PRECISION", "value=%r err=%r" % (v, e)) @price_suite.case("non_numeric_and_signed") def _(): for tok in ("abc", "-1.5", "+1.5", "1,5", "1 .5", "1.5.5", ""): v, e = parse_price_to_micro(tok, PRICE_SCALE) if e != "MALFORMED_PRICE_PARSE": return False, "token %r -> %r/%r" % (tok, v, e) return True, "" @price_suite.case("volume_rules") def _(): v, e = parse_volume_token("123") v2, e2 = parse_volume_token("0") v3, e3 = parse_volume_token("-5") v4, e4 = parse_volume_token("1.5") v5, e5 = parse_volume_token("abc") return (v == 123 and e is None and v2 == 0 and e2 is None and e3 == "MALFORMED_VOLUME" and e4 == "MALFORMED_VOLUME" and e5 == "MALFORMED_VOLUME", "%r/%r %r/%r %r/%r %r/%r %r/%r" % (v, e, v2, e2, v3, e3, v4, e4, v5, e5)) @price_suite.case("bid_ask_spread_integer") def _(): # 100.000100 - 100.000000 -> 100 micro units bid = 100000000 ask = 100000100 return eq(ask - bid, 100, "spread_u") # --------------------------------------------------------------------------- ser_suite = Suite("unit_serialization_hash") @ser_suite.case("canonical_tick_line") def _(): line = serialize_tick(1672628645100, 100000000, 100000100, 1) return eq(line, b"1672628645100|100000000|100000100|1\n", "tick ser") @ser_suite.case("chunk_content_id_ordered") def _(): recs = [(1672628645100, 100000000, 100000100, 1), (1672628645200, 100000100, 100000200, 1)] id1 = tick_chunk_content_id(recs) id2 = tick_chunk_content_id(recs) id3 = tick_chunk_content_id(list(reversed(recs))) return (id1 == id2 and id1 != id3, "stable=%s order-sensitive=%s" % (id1 == id2, id1 != id3)) @ser_suite.case("bar_serialization_fixed_fields") def _(): # full 17-field CBS_V1 row; serialization selects the 14 canonical fields row = (0, 1, 60000, 120000, 200000100, 200000600, 200000000, 200000600, 2, 2, 100, 200, 300, 150, True, 1, 2) line = serialize_bars(row) expected = (b"0|1|200000100|200000600|200000000|200000600|" b"2|2|100|200|300|1|1|2\n") return eq(line, expected, "bar ser") @ser_suite.case("hash_sha256_smoke") def _(): h1 = sha256_bytes(b"abc") return (len(h1) == 64 and h1 == sha256_bytes(b"abc") and h1 != sha256_bytes(b"abd"), h1) @ser_suite.case("canonical_json_stable") def _(): x = {"b": 1, "a": [3, 1]} return (canonical_json_sha256(x) == canonical_json_sha256(x) and canonical_json_sha256({"b": 1, "a": [3, 1]}) == canonical_json_sha256(x), "canonical JSON hash must be stable") # --------------------------------------------------------------------------- bar_suite = Suite("unit_bar_period_and_aggregation") @bar_suite.case("period_id_floor") def _(): for tf, period in TIMEFRAMES.items(): ts = 1672628645100 pid = ts // period start = pid * period end = start + period if not (start <= ts < end): return False, "%s: boundary violated" % tf if end - start != period: return False, "%s: period width" % tf return True, "M1/M5/M15/M30/H1 boundaries ok" @bar_suite.case("aggregation_ohlc_ints") def _(): cfg = default_config("C:\\unused\\src.csv", output_root="C:\\unused\\out", source_tz_offset_minutes=0) ticks = [ (60000, 100000000, 100000100, 100, 1, 1, 0), # mid 200000100 (61000, 100000200, 100000400, 200, 1, 2, 1), # mid 200000600 (120000, 100000300, 100000500, 200, 1, 3, 2), # mid 200000800 ] agg = Aggregator(cfg) for t in ticks: agg.consume(t) rows = agg.finish_eof("M1") if len(rows) != 2: return False, "expected 2 M1 bars, got %d" % len(rows) b0 = rows[0] # bar_idx, period_id, start,end, o,h,l,c, ticks, vol, smin,smax,ssum,savg ok = (b0[0] == 0 and b0[1] == 1 and b0[2] == 60000 and b0[3] == 120000 and b0[4] == 200000100 and b0[5] == 200000600 and b0[6] == 200000100 and b0[7] == 200000600 and b0[8] == 2 and b0[9] == 2 and b0[10] == 100 and b0[11] == 200 and b0[12] == 300 and b0[13] == 150 and b0[14] is True and b0[15] == 1 and b0[16] == 2) b1 = rows[1] ok1 = (b1[1] == 2 and b1[4] == 200000800 and b1[8] == 1 and b1[14] is False) return (ok and ok1, "b0=%r b1=%r" % (b0[:14], b1[:14])) @bar_suite.case("no_empty_bar_inflation") def _(): cfg = default_config("C:\\unused\\src.csv", output_root="C:\\unused\\out", source_tz_offset_minutes=0) # same-period ticks -> exactly ONE bar per timeframe (no fabricated bars) agg = Aggregator(cfg) agg.consume((0, 100000000, 100000100, 100, 1, 1, 0)) agg.consume((59000, 100000100, 100000200, 100, 1, 2, 1)) same_period = 0 for tf in cfg["timeframes"]: rows = agg.finish_eof(tf) same_period += len(rows) # far-apart ticks -> M1 bars exist only in periods 0 and 1440, no fillers agg2 = Aggregator(cfg) agg2.consume((0, 100000000, 100000100, 100, 1, 1, 0)) agg2.consume((86400000, 100000000, 100000100, 100, 1, 2, 1)) rows_m1 = agg2.finish_eof("M1") periods = [r[1] for r in rows_m1] return (same_period == 5 and len(rows_m1) == 2 and periods == [0, 1440], "same_period=%d periods=%r" % (same_period, periods)) @bar_suite.case("partial_bar_is_final_false") def _(): cfg = default_config("C:\\unused\\src.csv", output_root="C:\\unused\\out", source_tz_offset_minutes=0) agg = Aggregator(cfg) agg.consume((60000, 100000000, 100000100, 100, 1, 1, 0)) agg.consume((61000, 100000100, 100000200, 100, 1, 2, 1)) rows = agg.finish_eof("M1") return (len(rows) == 1 and rows[0][14] is False and rows[0][8] == 2, "EOF partial bar is_final=false") @bar_suite.case("carry_continuity_guard") def _(): cfg = default_config("C:\\unused\\src.csv", output_root="C:\\unused\\out", source_tz_offset_minutes=0) agg = Aggregator(cfg) agg.consume((60000, 100000000, 100000100, 100, 1, 1, 0)) # a resumed stream with ts BEFORE the carried last_ts must trip the guard try: agg.consume((59000, 100000000, 100000100, 100, 1, 2, 1)) except ValueError: return True, "order violation detected" return False, "order violation NOT detected" # --------------------------------------------------------------------------- ckpt_suite = Suite("unit_checkpoint_states") @ckpt_suite.case("legal_transitions") def _(): legal = [(STATUS_INITIALIZED, STATUS_RUNNING), (STATUS_RUNNING, STATUS_CHUNK_COMMITTED), (STATUS_CHUNK_COMMITTED, STATUS_PAUSED), (STATUS_PAUSED, STATUS_RUNNING), (STATUS_RUNNING, STATUS_PAUSED), (STATUS_CHUNK_COMMITTED, STATUS_COMPLETED), (STATUS_PAUSED, STATUS_FAILED), (STATUS_FAILED, STATUS_RESUME_BLOCKED)] for a, b in legal: try: validate_transition(a, b) except CheckpointTransitionError: return False, "legal transition rejected: %s -> %s" % (a, b) return True, "all legal transitions accepted" @ckpt_suite.case("illegal_transitions") def _(): illegal = [(STATUS_COMPLETED, STATUS_RUNNING), (STATUS_COMPLETED, STATUS_CHUNK_COMMITTED), (STATUS_RESUME_BLOCKED, STATUS_PAUSED), (STATUS_INITIALIZED, STATUS_CHUNK_COMMITTED)] for a, b in illegal: try: validate_transition(a, b) return False, "illegal transition accepted: %s -> %s" % (a, b) except CheckpointTransitionError: pass return True, "all illegal transitions rejected" @ckpt_suite.case("atomic_write_and_hash_verify") def _(): with tempfile.TemporaryDirectory() as td: p = os.path.join(td, "state", "checkpoint.json") payload = {"schema_version": "CP_V1", "status": STATUS_INITIALIZED, "x": {"a": 1}} write_checkpoint(p, payload) loaded = load_checkpoint(p) ok = loaded["status"] == STATUS_INITIALIZED and loaded["x"]["a"] == 1 return (ok and "checkpoint_hash" in loaded, "checkpoint round-trip ok=%s" % ok) @ckpt_suite.case("corruption_detected") def _(): with tempfile.TemporaryDirectory() as td: p = os.path.join(td, "state", "checkpoint.json") write_checkpoint(p, {"status": STATUS_INITIALIZED, "payload": 1}) with open(p, "r", encoding="utf-8") as fh: text = fh.read() text = text.replace(STATUS_INITIALIZED, STATUS_PAUSED) # tamper content with open(p, "w", encoding="utf-8") as fh: fh.write(text) try: load_checkpoint(p) except CheckpointCorrupt: return True, "tampered checkpoint rejected" return False, "tampered checkpoint accepted" # --------------------------------------------------------------------------- lock_suite = Suite("unit_lock_state_transitions") @lock_suite.case("acquire_refuse_live") def _(): with tempfile.TemporaryDirectory() as td: p = os.path.join(td, "state", "lock") h1 = LockHandle(p, "RUN-A", 111) h1.acquire() age = lock_status(p) h2 = LockHandle(p, "RUN-B", 222) try: h2.acquire() except Exception: refused = True else: refused = False h1.release() ok_released = not os.path.exists(p) return (refused and ok_released and age["_stale"] is False, "refused=%s released=%s" % (refused, ok_released)) @lock_suite.case("stale_refuse_without_force") def _(): with tempfile.TemporaryDirectory() as td: p = os.path.join(td, "state", "lock") h1 = LockHandle(p, "RUN-A", 111) h1.acquire() with open(p, "r", encoding="utf-8") as fh: import json info = json.load(fh) info["heartbeat_epoch_s"] = time.time() - 400 # stale (>300 s) with open(p, "w", encoding="utf-8") as fh: json.dump(info, fh) h2 = LockHandle(p, "RUN-B", 222) try: h2.acquire() except Exception as e: refused = "force-release" in str(e) else: refused = False return refused, "stale lock refused without --force-release-lock" @lock_suite.case("force_release_human_only") def _(): with tempfile.TemporaryDirectory() as td: p = os.path.join(td, "state", "lock") h1 = LockHandle(p, "RUN-A", 111) h1.acquire() with open(p, "r", encoding="utf-8") as fh: import json info = json.load(fh) info["heartbeat_epoch_s"] = time.time() - 600 with open(p, "w", encoding="utf-8") as fh: json.dump(info, fh) h2 = LockHandle(p, "RUN-B", 222, force_release=True) h2.acquire() h2.release() return (not os.path.exists(p), "force release removed stale lock") # --------------------------------------------------------------------------- cfg_suite = Suite("unit_config_identity") @cfg_suite.case("identity_hash_excludes_operational") def _(): src = "C:\\unused\\src.csv" a = default_config(src, output_root="C:\\unused\\o1", workers_requested=1) b = default_config(src, output_root="C:\\unused\\o2", workers_requested=24) h_a = config_sha256(a) h_b = config_sha256(b) return eq(h_a, h_b, "operational fields excluded from config_sha256") ALL_SUITES = [ts_suite, price_suite, ser_suite, bar_suite, ckpt_suite, lock_suite, cfg_suite] def run(): from tests.harness import run_suites, print_results flat, all_pass = run_suites(ALL_SUITES) print_results(flat) return all_pass if __name__ == "__main__": ok = run() sys.exit(0 if ok else 1)