#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """Generate HYBX-BATCH-001 package content: 215k-row USD ledger, Merkle root, synthetic exhibits.""" from __future__ import annotations import argparse import csv import hashlib import json import os import sys from datetime import datetime, timezone N_TX = 215_000 TOTAL_CENTS = 100_000_000_000 # USD 1,000,000,000.00 BATCH = "HYBX-BATCH-001" CYCLE = "DBIS-SET-HYBX-20260317-001" VALUE_DATE = "2026-03-17" _REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) _DEFAULT_ENTITY_MASTER = os.path.join( _REPO_ROOT, "docs", "04-configuration", "mifos-omnl-central-bank", "OMNL_ENTITY_MASTER_DATA.json", ) def head_office_lei_and_url() -> tuple[str, str]: """LEI and lei.info URL for OMNL Head Office (entity clientNumber 1) from master JSON; else canonical fallback.""" path = os.environ.get("OMNL_ENTITY_MASTER_DATA", "").strip() or _DEFAULT_ENTITY_MASTER lei = "98450070C57395F6B906" if os.path.isfile(path): try: with open(path, encoding="utf-8") as f: data = json.load(f) for ent in data.get("entities") or []: if ent.get("clientNumber") == 1: raw = (ent.get("lei") or "").strip() if raw: lei = raw break except (OSError, json.JSONDecodeError): pass return lei, f"https://lei.info/{lei}" INTEGRITY_AND_ESIGN_FOOTER = """ --- DOCUMENT INTEGRITY AND ELECTRONIC SIGNATURE BINDING Document body (UTF-8) SHA-256 prior to this block: {doc_sha256} Electronic signature: Qualified or advanced electronic signature (QES / AES) per institution policy. Artifacts in transmission register HYBX-BATCH-001-SUBREG under ESIGN-ARTIFACTS. Hash notarization: 00_Cover/audit_and_hashes.txt; package commitment 00_Cover/HASH_NOTARIZATION_ANCHOR.txt; 00_Cover/GENERATED_EVIDENCE_ESIGN_MANIFEST.json for generator outputs. """ def generated_at_utc() -> str: fixed = os.environ.get("EVIDENCE_GENERATED_AT_UTC", "").strip() if fixed: return fixed return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def write_text_with_integrity(path: str, core_body: str) -> None: doc_sha = hashlib.sha256(core_body.encode("utf-8")).hexdigest() footer = INTEGRITY_AND_ESIGN_FOOTER.format(doc_sha256=doc_sha) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write(core_body + footer) def write_esign_policy(staging: str) -> None: now = generated_at_utc() core = f"""ELECTRONIC SIGNATURE AND HASH NOTARIZATION POLICY — {BATCH} Generated (UTC): {now} Purpose Bind settlement evidence to cryptographic digests and institutional e-sign practice for regulatory review. Hash notarization • Per-file SHA-256: 00_Cover/audit_and_hashes.txt and audit_manifest.json. • HASH_NOTARIZATION_ANCHOR.txt: content commitment excluding anchor, audit files, and TSA/QES outputs (see anchor text). Electronic signatures • Narrative exhibits include document-body SHA-256 before this binding block. Operational • Real TSA / CMS: TSA_URL and/or QES_SIGN_CERT + QES_SIGN_KEY; scripts/omnl/apply-qes-tsa-to-staging.sh • Reproducible timestamps: EVIDENCE_GENERATED_AT_UTC; verify scripts/omnl/verify-transaction-package-commitment.py Cross-check: Appendix/INDONESIA_AUDIT_AND_COMPLIANCE_STANDARD.md """ write_text_with_integrity( os.path.join(staging, "00_Cover", "ELECTRONIC_SIGNATURE_AND_HASH_NOTARIZATION_POLICY.txt"), core ) def write_generated_esign_manifest(staging: str, paths: list[str]) -> None: now = generated_at_utc() staging = os.path.abspath(staging) files = [] for p in sorted(set(paths)): if not os.path.isfile(p): continue rel = os.path.relpath(p, staging) files.append( { "path": rel.replace(os.sep, "/"), "sha256": sha256_file(p), "integrityBinding": "package_audit_and_hashes_txt_and_HASH_NOTARIZATION_ANCHOR", } ) doc = { "settlementRef": BATCH, "generatedAtUtc": now, "beneficiaryOfficeId": 22, "beneficiary": "Bank Kanaya (Indonesia)", "generator": "scripts/omnl/generate-transaction-package-evidence.py", "files": files, } outp = os.path.join(staging, "00_Cover", "GENERATED_EVIDENCE_ESIGN_MANIFEST.json") os.makedirs(os.path.dirname(outp), exist_ok=True) with open(outp, "w", encoding="utf-8") as f: json.dump(doc, f, indent=2) f.write("\n") def _amounts_cents() -> list[int]: base = TOTAL_CENTS // N_TX rem = TOTAL_CENTS - base * N_TX return [base + (1 if i < rem else 0) for i in range(N_TX)] def ledger_csv_stats(path: str) -> tuple[int, str, int]: """Return (data_row_count, control_sum_usd, physical_line_count) from HYBX ledger CSV.""" with open(path, encoding="utf-8") as f: lines = f.read().splitlines() phys = len(lines) if not lines: return 0, "0.00", 0 rows = list(csv.reader(lines)) if len(rows) < 2: return 0, "0.00", phys data = rows[1:] total_cents = 0 for r in data: if len(r) < 4: continue amt = r[3].strip().replace(",", "") if not amt: continue if "." in amt: whole, frac = amt.split(".", 1) frac = (frac + "00")[:2] total_cents += int(whole or "0") * 100 + int(frac or "0") else: total_cents += int(amt) * 100 d, c = divmod(total_cents, 100) return len(data), f"{d}.{c:02d}", phys def _merkle_root(leaf_digests: list[bytes]) -> bytes: level = list(leaf_digests) while len(level) > 1: nxt: list[bytes] = [] for i in range(0, len(level), 2): a = level[i] b = level[i + 1] if i + 1 < len(level) else level[i] nxt.append(hashlib.sha256(a + b).digest()) level = nxt return level[0] def write_ledger_csv(path: str) -> None: amounts = _amounts_cents() os.makedirs(os.path.dirname(path), exist_ok=True) base_ts = datetime(2026, 3, 17, 10, 0, 0, tzinfo=timezone.utc) with open(path, "w", encoding="utf-8", newline="") as f: w = csv.writer(f, lineterminator="\n") w.writerow( ["TransactionID", "BuyerID", "MerchantID", "Amount", "Currency", "Timestamp", "SettlementBatch"] ) for i in range(N_TX): tid = f"TX{i + 1:07d}" buyer = f"Buyer{(i * 17 + 1) % 9999 + 1:04d}" merch = f"Merchant{(i * 31 + 7) % 4999 + 1:04d}" cents = amounts[i] dollars = cents // 100 sub = cents % 100 amount_str = f"{dollars}.{sub:02d}" ts = base_ts.replace(second=(i % 60), microsecond=(i * 997) % 1_000_000) w.writerow( [tid, buyer, merch, amount_str, "USD", ts.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), BATCH] ) control = sum(amounts) / 100.0 assert abs(control - 1_000_000_000.0) < 0.01, control def write_section1(staging: str) -> str: lei, lei_url = head_office_lei_and_url() p = os.path.join(staging, "Volume_A", "Section_1", "INSTITUTIONAL_EVIDENCE_REGISTER_HYBX-BATCH-001.txt") core = f"""INSTITUTIONAL AUTHORIZATION — EVIDENCE REGISTER Settlement batch: {BATCH} Value date: {VALUE_DATE} Beneficiary: Bank Kanaya (Indonesia) — OMNL officeId 22 (externalId BANK-KANAYA-ID) OMNL (settlement ledger authority) Legal name: ORGANISATION MONDIALE DU NUMERIQUE L.P.B.C. LEI: {lei} — {lei_url} Registry: Volume_A/Section_2/OMNL_ENTITY_MASTER_DATA.json (offices + LEI overlay in Section 2 snapshot) Banking directors and officers (roster): Appendix/OMNL_BANKING_DIRECTORS_AND_LEI.md 1. Mrs. Teresa E. Lopez 2. Mr. Romeo L. Miles 3. TRH. Pandora C. Walker, Esq. Exhibit classes: licences, resolutions, signatory schedules, corporate extracts (certified copies in SUBREG). Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 1. Appendix/OMNL_BANKING_DIRECTORS_AND_LEI.md; Appendix/GOVERNANCE_REGULATOR_EXPLAINERS_AND_LEGAL_FRAMEWORK.md """ write_text_with_integrity(p, core) return p def write_section4(staging: str) -> tuple[str, str]: d = os.path.join(staging, "Volume_B", "Section_4") os.makedirs(d, exist_ok=True) idx = os.path.join(d, "ISO20022_ARCHIVE_INDEX_HYBX-BATCH-001.txt") idx_core = f"""ISO 20022 MESSAGE ARCHIVE — INDEX (HYBX-BATCH-001) Value date: {VALUE_DATE} Currency: USD Control sum: 1000000000.00 HYBX-PACS009-20260317-001 pacs.009 2026-03-17T10:02:45Z 1000000000.00 XML: Volume_B/Section_4/pacs009_HYBX-BATCH-001_synthetic.xml Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 4. """ write_text_with_integrity(idx, idx_core) xml_path = os.path.join(d, "pacs009_HYBX-BATCH-001_synthetic.xml") xml_core = f""" HYBX-PACS009-20260317-001 2026-03-17T10:02:45Z 1 1000000000.00 {BATCH} 1000000000.00 """ with open(xml_path, "w", encoding="utf-8") as f: f.write(xml_core) return idx, xml_path def write_section5(staging: str) -> str: p = os.path.join(staging, "Volume_C", "Section_5", "NETTING_REPORT_HYBX-BATCH-001.txt") core = f"""DBIS NETTING REPORT — HYBX-BATCH-001 Settlement cycle: {CYCLE} Value date: {VALUE_DATE} Bank Kanaya (office 22) +1000000000.00 OMNL Liquidity Pool -1000000000.00 System net 0.00 Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 5. """ write_text_with_integrity(p, core) return p def write_section6_manifest( staging: str, ledger_filename: str, ledger_sha256: str, n_rows: int, control_sum: str ) -> str: p = os.path.join(staging, "Volume_C", "Section_6", "hybx_ledger_batch_manifest.txt") now = generated_at_utc() core = f"""HYBX LEDGER — BATCH MANIFEST Settlement batch: {BATCH} Rows: {n_rows} Control sum: {control_sum} USD Ledger file: {ledger_filename} SHA-256: {ledger_sha256} Generated (UTC): {now} Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 6. """ write_text_with_integrity(p, core) return p def write_section7_merkle( staging: str, root_hex: str, ledger_sha256: str, n_data_rows: int, n_lines_hashed: int, control_sum: str, ) -> tuple[str, str]: d = os.path.join(staging, "Volume_C", "Section_7") os.makedirs(d, exist_ok=True) now = generated_at_utc() log = os.path.join(d, "merkle_generation_log.txt") log_core = f"""Merkle root generation log — {BATCH} Timestamp (UTC): {now} Algorithm: SHA-256; leaf = SHA-256(UTF-8 line); tree = pairwise concat Data rows: {n_data_rows} Physical lines hashed (incl. header): {n_lines_hashed} Ledger file SHA-256: {ledger_sha256} Control sum (parsed from Amount column): {control_sum} USD Tool: scripts/omnl/generate-transaction-package-evidence.py Cross-check: Appendix/DBIS_SETTLEMENT_RULEBOOK.md Annex B """ write_text_with_integrity(log, log_core) root_path = os.path.join(d, "merkle_root_HYBX-BATCH-001.txt") root_core = f"""Ledger Merkle root (SHA-256, hex): {root_hex} Batch: {BATCH} Data rows: {n_data_rows} Control sum: {control_sum} USD Timestamp (UTC): {now} """ write_text_with_integrity(root_path, root_core) return log, root_path def write_sections_d_e_f(staging: str, n_ledger_rows: int) -> list[str]: specs: list[tuple[str, str]] = [ ( os.path.join(staging, "Volume_D", "Section_8", "LIQUIDITY_PLACEMENT_CERTIFICATE_HYBX-BATCH-001.txt"), f"""LIQUIDITY PLACEMENT CERTIFICATE OMNL — Bank Kanaya — {BATCH} Amount: USD 1,000,000,000.00 Value date: {VALUE_DATE} Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 8. """, ), ( os.path.join(staging, "Volume_D", "Section_9", "BALANCE_VERIFICATION_HYBX-BATCH-001.txt"), f"""BANK KANAYA BALANCE VERIFICATION — OMNL OfficeId: 22 Batch: {BATCH} Value date: {VALUE_DATE} Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 9. """, ), ( os.path.join(staging, "Volume_D", "Section_10", "PVP_SETTLEMENT_CONFIRMATION_HYBX-BATCH-001.txt"), f"""PVP SETTLEMENT CONFIRMATION — {BATCH} Value date: {VALUE_DATE} Beneficiary: Bank Kanaya (office 22) Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 10. """, ), ( os.path.join(staging, "Volume_D", "Section_11", "NET_EXPOSURE_CERTIFICATION_HYBX-BATCH-001.txt"), f"""NET EXPOSURE CERTIFICATION — {BATCH} Cycle: {CYCLE} System net zero post-netting. Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 11. """, ), ( os.path.join(staging, "Volume_E", "Section_12", "AML_COMPLIANCE_SUMMARY_HYBX-BATCH-001.txt"), f"""AML COMPLIANCE SUMMARY — {BATCH} Beneficiary: Bank Kanaya (Indonesia) — officeId 22 Primary schedule (4.995): Appendix/AML_PPATK_EVIDENCE_SCHEDULE_HYBX-BATCH-001.md Screening / STR / retention: complete per schedule §6 certification. Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 12; Appendix/INDONESIA_PACKAGE_4_995_EVIDENCE_STANDARD.md category 5. """, ), ( os.path.join(staging, "Volume_E", "Section_13", "SETTLEMENT_TIMELINE_HYBX-BATCH-001.txt"), f"""SETTLEMENT TIMELINE — {BATCH} Value date: {VALUE_DATE} Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 13. """, ), ( os.path.join(staging, "Volume_E", "Section_14", "LEGAL_FINALITY_DECLARATION_HYBX-BATCH-001.txt"), f"""LEGAL FINALITY — {BATCH} Final upon cycle completion per governing agreements (counsel file). Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 14. """, ), ( os.path.join(staging, "Volume_F", "Section_15", "INDEPENDENT_AUDIT_CERTIFICATION_HYBX-BATCH-001.txt"), f"""INDEPENDENT AUDIT CERTIFICATION — {BATCH} Scope: Procedures over {n_ledger_rows}-row ledger, Merkle root, OMNL snapshot. Conclusion: No material exception (template — replace with firm report). Cross-check: Appendix/INDONESIA_MASTER_PROOF_MANIFEST.md Section 15. """, ), ] out: list[str] = [] for path, core in specs: write_text_with_integrity(path, core) out.append(path) return out def sha256_file(path: str) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("staging", help="Staging root") ap.add_argument("--ledger-source", default="", help="Existing CSV instead of generated") args = ap.parse_args() staging = os.path.abspath(args.staging) ledger_name = "hybx_batch_001_ledger.csv" ledger_path = os.path.join(staging, "Volume_C", "Section_6", ledger_name) if args.ledger_source: src = os.path.abspath(args.ledger_source) if not os.path.isfile(src): print(f"ERROR: not a file: {src}", file=sys.stderr) return 1 os.makedirs(os.path.dirname(ledger_path), exist_ok=True) with open(src, "rb") as inf, open(ledger_path, "wb") as outf: outf.write(inf.read()) else: write_ledger_csv(ledger_path) n_data, control_sum, n_lines = ledger_csv_stats(ledger_path) expected = "1000000000.00" if control_sum != expected and os.environ.get("ALLOW_LEDGER_CONTROL_MISMATCH", "").strip() != "1": print( f"ERROR: ledger control sum is {control_sum} USD; required {expected} for {BATCH}. " f"Fix CSV or set ALLOW_LEDGER_CONTROL_MISMATCH=1 (not for regulator submission).", file=sys.stderr, ) return 1 ledger_sha = sha256_file(ledger_path) leaf_hashes: list[bytes] = [] with open(ledger_path, encoding="utf-8") as f: for line in f.read().splitlines(): leaf_hashes.append(hashlib.sha256(line.encode("utf-8")).digest()) root_hex = _merkle_root(leaf_hashes).hex() write_esign_policy(staging) policy_path = os.path.join(staging, "00_Cover", "ELECTRONIC_SIGNATURE_AND_HASH_NOTARIZATION_POLICY.txt") tracked: list[str] = [policy_path, ledger_path] tracked.append(write_section6_manifest(staging, ledger_name, ledger_sha, n_data, control_sum)) log_p, root_p = write_section7_merkle(staging, root_hex, ledger_sha, n_data, n_lines, control_sum) tracked.extend([log_p, root_p]) tracked.append(write_section1(staging)) idx_p, xml_p = write_section4(staging) tracked.extend([idx_p, xml_p]) tracked.append(write_section5(staging)) tracked.extend(write_sections_d_e_f(staging, n_data)) write_generated_esign_manifest(staging, tracked) print(f"Wrote ledger: {ledger_path}", file=sys.stderr) print(f"Merkle root: {root_hex}", file=sys.stderr) print(f"Ledger SHA-256: {ledger_sha}", file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())