SniperGold_ML/tests/unit_tests.py

"""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)