SniperGold_ML/tests/property_tests.py

"""Property / invariant tests (spec 18.3 + session requirement 3).

At minimum: deterministic serialization, deterministic aggregation, worker
scheduling does not change logical output, no gaps, no overlaps, monotonic
source ordering, checkpoint monotonicity, resume equivalence, content-ID
stability, no silent malformed-row transformation, no empty-bar fabrication.
"""

import json
import os
import sys
import tempfile

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

from engine.chunkmap import build_chunk_map  # noqa: E402
from engine.config import default_config  # noqa: E402
from engine.parse import parse_chunk, ALL_MALFORMED_CLASSES  # noqa: E402
from engine.verify.vparse import v_parse_chunk  # noqa: E402
from engine.canonical import tick_chunk_content_id  # noqa: E402
from engine.verify.vinvariants import independent_ticks_content_id, \
    independent_bars_content_id  # noqa: E402
from engine.aggregate import Aggregator  # noqa: E402
from engine.verify.vaggregate import v_aggregate, v_aggregate_eof  # noqa: E402
from engine.checkpoint import write_checkpoint, load_checkpoint  # noqa: E402
from engine.storage import (  # noqa: E402
    ticks_chunk_path, bar_part_path, read_ticks_chunk, read_bar_part,
    checkpoint_path, cert_path,
)
from engine.certify import certify_source  # noqa: E402
from engine.dispatcher import (  # noqa: E402
    init_run, IngestRunner, resume_checks, finalize_completed, new_run_id,
)
from engine.journal import append_commit, read_journal, seq_after  # noqa: E402
from engine.util import sha256_bytes  # noqa: E402
from tests.golden.common import G01_ROWS  # noqa: E402
from tests.golden.run_golden import mini_setup, mini_run  # noqa: E402

CERT_NOW = 4_102_444_800_000


def rows_source(n=60, base_ms=1_672_900_000_000):
    """Deterministic synthetic tick stream (valid prices around 100.0..100.3).
    Prices are rendered from integer micro-unit values (no float formatting)."""
    import datetime
    lines = []
    ts = base_ms
    for i in range(n):
        p = 100000000 + (i % 30) * 10000
        ask = p + 100
        dt = datetime.datetime.fromtimestamp(ts / 1000, datetime.timezone.utc)
        stamp = dt.strftime("%Y.%m.%d %H:%M:%S.") + ("%03d" % (ts % 1000))
        bid_txt = "%d.%06d" % (p // 1000000, p % 1000000)
        ask_txt = "%d.%06d" % (ask // 1000000, ask % 1000000)
        lines.append("%s,%s,%s" % (stamp, bid_txt, ask_txt))
        ts += 60_000 // 4
    return ("\n".join(lines) + "\n").encode("utf-8")


def run():
    cfg = default_config("C:\\unused\\src.csv", output_root="C:\\unused\\out",
                         source_tz_offset_minutes=0)
    results = []

    # 1. deterministic serialization / content-ID stability (producer==verifier)
    data = rows_source(40)
    ids_prod, ids_ver = [], []
    for _ in range(3):
        recs, _, _, _ = parse_chunk(data, chunk_index=0, byte_start=0,
                                    global_line_start=0, cfg=cfg,
                                    expect_header=True,
                                    certify_time_ms=CERT_NOW)
        vrecs, _, _, _ = v_parse_chunk(data, chunk_index=0, byte_start=0,
                                        global_line_start=0, cfg=cfg,
                                        expect_header=True,
                                        certify_time_ms=CERT_NOW)
        ids_prod.append(tick_chunk_content_id([(r[0], r[1], r[2], r[4])
                                               for r in recs]))
        ids_ver.append(independent_ticks_content_id([(r[0], r[1], r[2], r[4])
                                                     for r in vrecs]))
        nn = len(recs)
    results.append(("deterministic_serialization",
                    all(x == ids_prod[0] for x in ids_prod)
                    and all(x == ids_ver[0] for x in ids_ver), ""))
    results.append(("content_id_stability_producer_verifier",
                    ids_prod[0] == ids_ver[0] and nn > 0, ""))

    # 2. deterministic aggregation: producer vs independent verifier
    recs, _, _, _ = parse_chunk(data, chunk_index=0, byte_start=0,
                                global_line_start=0, cfg=cfg,
                                expect_header=True, certify_time_ms=CERT_NOW)
    agg = Aggregator(cfg)
    for r in recs:
        agg.consume(r)
    prod_bars = {tf: agg.finish_eof(tf) for tf in cfg["timeframes"]}
    vbars, vcarry, vcounts = v_aggregate([r for r in recs], None, cfg)
    veof, _, veof_counts = v_aggregate_eof(cfg, vcarry, vcounts)
    ok = True
    for tf in cfg["timeframes"]:
        if vbars[tf] + veof[tf] != prod_bars[tf]:
            ok = False
    agg2 = Aggregator(cfg)
    for r in recs:
        agg2.consume(r)
    prod2 = {tf: agg2.finish_eof(tf) for tf in cfg["timeframes"]}
    ok2 = all(prod_bars[tf] == prod2[tf] for tf in cfg["timeframes"])
    results.append(("deterministic_aggregation", ok, ""))
    results.append(("aggregation_producer_verifier_equal", ok2, ""))

    # 3. no gaps / no overlaps in chunkmap on a mixed-length file
    with tempfile.TemporaryDirectory() as td:
        src = os.path.join(td, "mix.txt")
        lines = [b"2023.01.01 00:00:%02d.00%d,100.0%d,100.0%d"
                 % (i % 60, i % 10, i % 9, i % 9 + 1) for i in range(200)]
        data_mix = b"\n".join(lines) + b"\n"
        with open(src, "wb") as fh:
            fh.write(data_mix)
        cm = build_chunk_map(src, 500, 5_000_000)
        gap_ok = cm["chunks"][0]["byte_start"] == 0 and \
            cm["chunks"][-1]["byte_end"] == cm["total_bytes"]
        for i in range(1, len(cm["chunks"])):
            gap_ok = gap_ok and \
                cm["chunks"][i]["byte_start"] == cm["chunks"][i - 1]["byte_end"]
        results.append(("no_gaps_no_overlaps", gap_ok and len(cm["chunks"]) > 1, ""))

    # 4. monotonic source ordering
    mono = True
    prev = None
    for r in recs:
        if prev is not None and r[0] < prev:
            mono = False
        prev = r[0]
    results.append(("monotonic_source_order", mono, ""))

    # 5. checkpoint monotonicity + journal seq monotonicity
    with tempfile.TemporaryDirectory() as td:
        jp = os.path.join(td, "commits.jsonl")
        prev_hash = sha256_bytes(b"")
        seqs = []
        for i in range(5):
            prev_hash = append_commit(jp, i, i, "c%d" % i, "s%d" % i,
                                      "t%d" % i, prev_hash)
            seqs.append(i)
        recs_j, tail = read_journal(jp)
        ok_seq = [r["seq"] for r in recs_j] == seqs
        # checkpoint writes with increasing next_chunk
        che1 = {"n": 0, "next_chunk": 1}
        che2 = {"n": 1, "next_chunk": 2}
        cp = os.path.join(td, "checkpoint.json")
        write_checkpoint(cp, che1)
        write_checkpoint(cp, che2)
        loaded = load_checkpoint(cp)
        results.append(("journal_seq_monotonic", ok_seq, ""))
        results.append(("checkpoint_next_chunk_monotonic",
                        loaded["next_chunk"] == 2, ""))

    # 6. resume equivalence: full run vs interrupted/resumed run
    data6 = rows_source(120)
    with tempfile.TemporaryDirectory() as tdA, tempfile.TemporaryDirectory() as tdB:
        cfgA, certA, cmA, ridA = mini_setup(data6, tdA, chunk_bytes=700,
                                            workload_bytes=1400)
        stA = mini_run(cfgA, certA, cmA, tdA, ridA)
        idsA = _file_ids(cmA, tdA, cfgA)
        cfgB, certB, cmB, ridB = mini_setup(data6, tdB, chunk_bytes=700,
                                            workload_bytes=1400)
        stB1 = mini_run(cfgB, certB, cmB, tdB, ridB, pause_after=1)
        _c, reason = resume_checks(cfgB, tdB, cfgB["source_path"], cmB, certB)
        if reason is None:
            rr = IngestRunner(cfgB, tdB, cfgB["source_path"], cmB["source_id"],
                              cmB, certB, run_id=ridB)
            stB2 = rr.run()
        else:
            stB2 = None
        idsB = _file_ids(cmA, tdA, cfgA)
        idsB2 = _file_ids(cmB, tdB, cfgB)
        results.append(("resume_equivalence",
                        stB2 == "CHUNK_COMMITTED" and idsA == idsB2,
                        "stB2=%r" % stB2))

    # 7. no silent malformed-row transformation
    bad = G01_ROWS + ["2023.13.01 00:00:00.000,100.0,101.0",
                      "x,1,2", "2023.01.01 00:00:00.000,200,100",
                      "2023.01.01 00:00:00.000,100.000000,100.000100"]
    data_bad = (b"\n".join(x.encode() for x in bad) + b"\n")
    brecs, _m, ctr, _f = parse_chunk(data_bad, chunk_index=0, byte_start=0,
                                     global_line_start=0, cfg=cfg,
                                     expect_header=True,
                                     certify_time_ms=CERT_NOW)
    n_mal = sum(ctr.values())
    every_valid = all(0 < r[1] <= r[2] for r in brecs)
    results.append(("no_silent_malformed_transformation",
                    n_mal == 4 and every_valid and len(brecs) == 4,
                    "mal=%d rows=%d" % (n_mal, len(brecs))))

    # 8. no empty-bar fabrication (property scale)
    agg3 = Aggregator(cfg)
    agg3.consume((0, 100000000, 100000100, 100, 1, 1, 0))
    agg3.consume((91 * 86400000, 100000000, 100000100, 100, 1, 2, 1))
    rows_m1 = agg3.finish_eof("M1")
    results.append(("no_empty_bar_fabrication",
                    len(rows_m1) == 2 and [r[1] for r in rows_m1] == [0, 91 * 1440],
                    [str(r[1]) for r in rows_m1]))

    return results


def _file_ids(cm, out_root, cfg):
    ids = {}
    for c in cm["chunks"]:
        rows = read_ticks_chunk(ticks_chunk_path(out_root, c["index"]))
        ids["ticks/%d" % c["index"]] = independent_ticks_content_id(
            [(r[0], r[1], r[2], r[4]) for r in rows])
    for wl in cm["workloads"]:
        for tf in cfg["timeframes"]:
            p = bar_part_path(out_root, tf, wl["index"])
            if os.path.exists(p):
                ids["bars/%s/%d" % (tf, wl["index"])] = \
                    independent_bars_content_id(read_bar_part(p))
    return ids


if __name__ == "__main__":
    res = run()
    for name, ok, detail in res:
        print("%-40s %s %s" % (name, "PASS" if ok else "FAIL", detail))
    sys.exit(0 if all(r[1] for r in res) else 1)