# -*- coding: utf-8 -*- """INTEGRITY AUDIT — machine_annotations_f7_v2.csv (P3-S.1 pre-freeze). Verifikasi (brief P3-S.1 §2-5): A. Case set integrity : 60 case; case_id/decision_timestamp/symbol/data-window v1 == v2 B. Machine counts : machine YES = 45, NO = 15; semua NO = age > SEQ_WINDOW (stale), BUKAN krn timestamp mismatch / missing data / parser / N/A / TF C. V1->V2 diff : changed 22, unchanged 38, removed 22, added 0; seluruh perubahan dijelaskan OLEH event expiration/lifecycle saja (tidak ada perubahan reference/threshold/pivot/ATR/TF/rejection/close-back) D. Field contract : kolom wajib ada; nilai berasal dari implementation (N/A utk tdk applicable) Output: output/integrity_audit_f7_v2.json + ringkasan konsol. Usage: python integrity_audit.py """ import os import sys import csv import json import hashlib import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) OUT = os.path.join(HERE, "output") REQUIRED_FIELDS = [ "case_id", "decision_timestamp", "symbol", "decision_tf", "machine_decision", "primary_primitive", "active_primitives", "f7_legacy", "f7_corrected", "f7_state_age", "f7_expired", "f10", "f11", "reference_type", "reference_level", "sweep_price", "excess", "close_price", "rejection", "timeframe", "machine_reason", "annotation_version", ] def sha256_file(p): h = hashlib.sha256() with open(p, "rb") as f: for blk in iter(lambda: f.read(1 << 20), b""): h.update(blk) return h.hexdigest() def load(p): rows = {} with open(p, encoding="utf-8-sig") as f: for r in csv.DictReader(f): rows[r["case_id"]] = r return rows def main(): rep = {"generated_utc": __import__("datetime").datetime.now( __import__("datetime").timezone.utc).isoformat()} v1 = load(os.path.join(OUT, "machine_annotations_f7_v1_LEGACY.csv")) v2 = load(os.path.join(OUT, "machine_annotations_f7_v2.csv")) meta = json.load(open(os.path.join(OUT, "cases_meta.json"), encoding="utf-8")) meta_by_id = {c["case_id"]: c for c in meta["cases"]} # ============ A. CASE SET INTEGRITY ============ a = {} a["v1_count"] = len(v1) a["v2_count"] = len(v2) a["ids_equal"] = sorted(v1.keys()) == sorted(v2.keys()) == sorted(meta_by_id.keys()) ts_diff = [cid for cid in v1 if v1[cid]["decision_timestamp"] != v2[cid]["decision_timestamp"]] sym_diff = [cid for cid in v1 if v1[cid]["symbol"] != v2[cid]["symbol"]] a["timestamp_diff_cases"] = ts_diff a["symbol_diff_cases"] = sym_diff a["decision_tf_v2_all_M15"] = all(v2[c]["decision_tf"] == "M15" for c in v2) # data window: sama krn sampling tidak dijalankan ulang (cases_meta identik dgn v1) a["case_set_hash_meta"] = sha256_file(os.path.join(OUT, "cases_meta.json")) a["note"] = "case set = cases_meta.json (tidak di-resample); v1/v2 pakai case yg sama" # ============ B. MACHINE COUNTS + NO TABLE ============ yes = [c for c in v2 if v2[c]["machine_decision"] == "YES"] no = [c for c in v2 if v2[c]["machine_decision"] == "NO"] b = {"machine_YES": len(yes), "machine_NO": len(no)} no_rows = [] for cid in no: r = v2[cid] age = r["f7_state_age"] try: age_i = int(age) expected = "STALE_EXPECTED" if age_i > 40 else "NOT_STALE" except (TypeError, ValueError): age_i, expected = None, "NON_NUMERIC" no_rows.append({ "case_id": cid, "legacy_decision": v1[cid]["machine_decision"] if cid in v1 else "?", "v2_decision": "NO", "reason": r["machine_reason"], "event_age": age, "expected_age_status": expected, "f10": r["f10"], "f11": r["f11"], "primary": r["primary_primitive"], }) b["no_table"] = no_rows b["all_NO_stale"] = all(x["expected_age_status"] == "STALE_EXPECTED" for x in no_rows) b["no_cases_with_f1011"] = [x["case_id"] for x in no_rows if int(x["f10"]) == 1 or int(x["f11"]) == 1] # ============ C. V1 -> V2 DIFF REVIEW ============ cmp = json.load(open(os.path.join(OUT, "f7_v1_vs_v2_comparison.json"), encoding="utf-8")) c_ = {} c_["summary"] = cmp["summary"] # verify summary numbers c_["summary_matches"] = ( cmp["summary"]["cases_changed"] == 22 and cmp["summary"]["cases_unchanged"] == 38 and cmp["summary"]["events_removed"] == 22 and cmp["summary"]["events_added"] == 0 and cmp["summary"]["f10_f11_unchanged"] is True) # per-case: pastikan perubahan HANYA lifecycle (f10/f11/close_price/timeframe/rejection-utk-EQ tak berubah) anomalies = [] for pc in cmp["per_case"]: cid = pc["case_id"] if not pc["changed"]: continue r2, r1 = v2[cid], v1[cid] # field yang WAJIB tetap sama utk kasus berubah (lifecycle-only): # - f10/f11 harus sama (0 di kedua utk pure grab; sama utk EQ cases) # - close_price sama # - timeframe sama # - rejection EQ (utk kasus yg primary EQ di v2) tidak berubah semantik if int(r2["f10"]) != int(r1.get("parity_f10", r1.get("f10", 0))): anomalies.append({"case_id": cid, "field": "f10", "v1": r1.get("parity_f10"), "v2": r2["f10"]}) if int(r2["f11"]) != int(r1.get("parity_f11", r1.get("f11", 0))): anomalies.append({"case_id": cid, "field": "f11", "v1": r1.get("parity_f11"), "v2": r2["f11"]}) if r2["close_price"] != r1["close_price"]: anomalies.append({"case_id": cid, "field": "close_price", "v1": r1["close_price"], "v2": r2["close_price"]}) if r2["timeframe"] != r1["timeframe"]: anomalies.append({"case_id": cid, "field": "timeframe", "v1": r1["timeframe"], "v2": r2["timeframe"]}) c_["anomalies_outside_lifecycle"] = anomalies c_["all_changes_lifecycle_only"] = (len(anomalies) == 0) # ============ D. FIELD CONTRACT ============ d = {"required_fields_present": {}, "missing": []} header = list(v2[next(iter(v2))].keys()) for f in REQUIRED_FIELDS: d["required_fields_present"][f] = f in header if f not in header: d["missing"].append(f) # N/A usage: pastikan tidak ada nilai buatan d["na_used"] = sum(1 for c in v2 for fld in ("reference_type", "reference_level", "sweep_price", "excess", "rejection") if v2[c][fld] == "N/A") d["field_source_note"] = ("reference_type/level: dari primitif kode (Swing len=5 / EQH/EQL); " "sweep_price: ekstrem bar crossing; excess: dari harga kode; " "rejection: kondisi kode (f7 close-back YES; EQ N/A); " "event_age: (r - sweep_bar) dari kode") # hashes d["hashes"] = { "machine_annotation_v2_sha": sha256_file(os.path.join(OUT, "machine_annotations_f7_v2.csv")), "machine_annotation_v1_legacy_sha": sha256_file(os.path.join(OUT, "machine_annotations_f7_v1_LEGACY.csv")), "case_set_meta_sha": sha256_file(os.path.join(OUT, "cases_meta.json")), "source_commit": cmp["provenance"]["source_commit"], } rep.update({"A_case_set": a, "B_counts": b, "C_diff": c_, "D_field": d}) out = os.path.join(OUT, "integrity_audit_f7_v2.json") with open(out, "w", encoding="utf-8") as f: json.dump(rep, f, indent=2, default=str) print(json.dumps({"A": a, "B": {k: v for k, v in b.items() if k != "no_table"}, "C": c_, "D": {k: v for k, v in d.items() if k != "hashes"}, "hashes": d["hashes"]}, indent=2, default=str)) print(f"\n[saved] {out}") if __name__ == "__main__": main()