forked from chiki2bum2/SniperGold_ML
Chiki Bum
0d5ca3c197
Implements the frozen P3_DATA_ENGINE_V1_SPEC (SHA 84bf0f217ffba51197112a6bbacbcc297058e04b5ca47f0459028fe33e0321e5). Components: engine/ producer (certify, chunkmap, parse, canonical, worker, dispatcher, aggregate, storage, journal, checkpoint, lock, evidence, manifest, run_complete, dataset_builder, cli) + engine/verify independent verifier (vparse, vaggregate, vinvariants, vcompare); headless CLI sniper-data; golden corpus G01-G17; unit/property/adversarial/mutation/legacy-diff/CLI suites. Qualification verdict: QUALIFIED (all 9 mandatory gates pass; independent verifier accepted). Spec, governance record, and legacy checkpoint untouched. G-14 NOT AUTHORIZED honored; no real-data processing, no pilot, no workload 46, no chunk 760 access.
457 lines
No EOL
16 KiB
Python
457 lines
No EOL
16 KiB
Python
"""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) |