#!/usr/bin/env python3 """Synthetic Secondary Marketing calculation and local replay receipts. No live connectors, messages, prices, orders, commitments or payments. """ import argparse import copy import hashlib import json import sqlite3 from pathlib import Path from datetime import datetime def _object(value, name): if type(value) is not dict: raise ValueError(name+":expected_object") def _array(value, name, nonempty=False): if type(value) is not list or (nonempty and not value): raise ValueError(name+":expected_"+("nonempty_" if nonempty else "")+"array") def _text(value, name): if type(value) is not str or not value.strip() or value != value.strip(): raise ValueError(name+":expected_trimmed_nonblank_string") def _integer(value, name, minimum=None, maximum=None): if type(value) is not int: raise ValueError(name+":expected_integer") if minimum is not None and value < minimum: raise ValueError(name+":below_minimum") if maximum is not None and value > maximum: raise ValueError(name+":above_maximum") def _boolean(value, name): if type(value) is not bool: raise ValueError(name+":expected_boolean") def _utc(value, name): _text(value, name) try: parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ") except ValueError: raise ValueError(name+":invalid_utc_timestamp") from None if parsed.strftime("%Y-%m-%dT%H:%M:%SZ") != value: raise ValueError(name+":noncanonical_utc_timestamp") def _date(value, name): _text(value, name) try: parsed = datetime.strptime(value, "%Y-%m-%d") except ValueError: raise ValueError(name+":invalid_date") from None if parsed.strftime("%Y-%m-%d") != value: raise ValueError(name+":noncanonical_date") def _enum(value, allowed, name): _text(value, name) if value not in allowed: raise ValueError(name+":unsupported_value") def _strings(values, name, nonempty=True): _array(values, name, nonempty) for value in values: _text(value, name) if len(values) != len(set(values)): raise ValueError(name+":duplicate_identifier") def _validate(data): _object(data,"data") _utc(data["as_of"],"as_of") _text(data["channel"],"channel") _text(data["product"],"product") _integer(data["principal_cents"],"principal_cents",1) _integer(data["adjustment_price_bps"],"adjustment_price_bps") _integer(data["margin_price_bps"],"margin_price_bps") q = data["quote"] _object(q,"quote") _text(q["quote_id"],"quote.quote_id") _text(q["product"],"quote.product") _integer(q["base_price_bps"],"quote.base_price_bps",1) _utc(q["effective_at"],"quote.effective_at") _utc(q["expires_at"],"quote.expires_at") if q["effective_at"] >= q["expires_at"]: raise ValueError("quote:invalid_effective_interval") _strings(q["eligible_channels"],"quote.eligible_channels") if q["base_price_bps"]+data["adjustment_price_bps"]-data["margin_price_bps"] <= 0: raise ValueError("candidate_price:nonpositive") def _calculate(data): q = data["quote"] errors = [] if not (q["effective_at"] <= data["as_of"] < q["expires_at"]): errors.append("quote_not_current") if data["channel"] not in q["eligible_channels"]: errors.append("channel_not_eligible") if q["product"] != data["product"]: errors.append("product_mismatch") fields = [q["base_price_bps"], data["adjustment_price_bps"], data["margin_price_bps"], data["principal_cents"]] if any(type(v) is not int for v in fields) or data["principal_cents"] <= 0: errors.append("invalid_numeric_type_or_principal") if errors: return {"status":"hold", "candidate_price_bps":None, "illustrative_price_amount_cents":None, "errors":sorted(errors)} price = q["base_price_bps"]+data["adjustment_price_bps"]-data["margin_price_bps"] # Integers represent 1/100 of a price point; 10000 is par. # Require exact cents for this fixture; production rounding is an approved rule. value, remainder = divmod(data["principal_cents"]*price,10000) if remainder: errors.append("rounding_rule_required") return {"status":"hold" if errors else "ready_for_review", "candidate_price_bps":price, "illustrative_price_amount_cents":value if not errors else None, "errors":errors} def run(data): try: _validate(data) return _calculate(data) except KeyError as error: return {"status":"hold", "errors":["missing_field:"+str(error.args[0])]} except (ValueError, TypeError, ArithmeticError) as error: return {"status":"hold", "errors":["invalid_input:"+str(error)]} def patched_input(bundle, case_number): data = copy.deepcopy(bundle["input"]) if not case_number: return data, bundle["event_id"] case = bundle["changes"][case_number - 1] if "replace_input" in case: data = copy.deepcopy(case["replace_input"]) else: path = case.get("path", case.get("append_path", case.get("remove_path"))) if path is None: return data, bundle["event_id"] + ":case-" + str(case_number) target = data for key in path[:-1]: target = target[key] if "remove_path" in case: del target[path[-1]] elif "append_path" in case: target[path[-1]].append(copy.deepcopy(case["append_value"])) else: target[path[-1]] = copy.deepcopy(case["value"]) return data, bundle["event_id"] + ":case-" + str(case_number) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--fixture", required=True, type=Path) parser.add_argument("--db", required=True, type=Path) parser.add_argument("--case", type=int, default=0) args = parser.parse_args() bundle = json.loads(args.fixture.read_text(encoding="utf-8")) if not 0 <= args.case <= len(bundle["changes"]): parser.error("--case must be 0 through " + str(len(bundle["changes"]))) data, event_id = patched_input(bundle, args.case) process_id, policy = bundle["process_id"], bundle["policy_version"] if any(not isinstance(v, str) or not v.strip() for v in [process_id, policy, event_id]): parser.error("Process, policy and event IDs must be nonblank strings") raw = json.dumps(data, sort_keys=True, separators=(",", ":"), allow_nan=False) input_hash = hashlib.sha256(raw.encode()).hexdigest() code_hash = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() with sqlite3.connect(args.db) as db: db.execute("""CREATE TABLE IF NOT EXISTS receipts ( process TEXT, policy TEXT, event_id TEXT, input_hash TEXT, code_hash TEXT, input_json TEXT, original_result TEXT, status TEXT, PRIMARY KEY(process, policy, event_id))""") db.execute("""CREATE TABLE IF NOT EXISTS conflicts ( process TEXT, policy TEXT, event_id TEXT, input_hash TEXT, code_hash TEXT, input_json TEXT, PRIMARY KEY(process, policy, event_id, input_hash, code_hash))""") db.commit() db.execute("BEGIN IMMEDIATE") key = (process_id, policy, event_id) old = db.execute("""SELECT input_hash,code_hash,original_result,status FROM receipts WHERE process=? AND policy=? AND event_id=?""", key).fetchone() if old and (old[0] != input_hash or old[1] != code_hash): result = {"status": "hold", "errors": ["conflicting_replay"]} action = "conflict" db.execute("INSERT OR IGNORE INTO conflicts VALUES (?,?,?,?,?,?)", (*key, input_hash, code_hash, raw)) db.execute("UPDATE receipts SET status='conflicted' WHERE process=? AND policy=? AND event_id=?", key) elif old: action = "reused" result = ({"status": "hold", "errors": ["conflicting_replay"]} if old[3] == "conflicted" else json.loads(old[2])) else: action = "new" try: result = run(data) encoded = json.dumps(result, sort_keys=True, allow_nan=False) except (KeyError, ValueError, TypeError, ArithmeticError, IndexError) as error: result = {"status": "hold", "errors": ["invalid_input:" + type(error).__name__]} encoded = json.dumps(result, sort_keys=True) db.execute("INSERT INTO receipts VALUES (?,?,?,?,?,?,?,?)", (*key, input_hash, code_hash, raw, encoded, "stored")) db.commit() count = db.execute("SELECT COUNT(*) FROM receipts WHERE process=? AND policy=?", (process_id, policy)).fetchone()[0] print(json.dumps({"process_id": process_id, "event_id": event_id, "receipt": action, "unique_receipts": count, "external_actions": 0, "result": result}, indent=2, allow_nan=False)) if __name__ == "__main__": main()