#!/usr/bin/env python3 """Synthetic historical simulation only. No trade or payment actions.""" import argparse import copy import hashlib import json import sqlite3 from pathlib import Path def _validate(value, schema, path='input'): from datetime import date, datetime, timedelta if value is None: if schema.get('nullable'):return raise ValueError(path+': null is not allowed') kind=schema['kind'] if kind=='object': if type(value) is not dict:raise ValueError(path+': object required') for key,child in schema['fields'].items(): if key not in value:raise ValueError(path+'.'+key+': required') _validate(value[key],child,path+'.'+key) elif kind=='list': if type(value) is not list or len(value)schema['maximum']:raise ValueError(path+': above maximum') elif kind=='boolean': if type(value) is not bool:raise ValueError(path+': boolean required') elif kind in ['string','date','utc_timestamp']: if type(value) is not str or not value.strip():raise ValueError(path+': nonblank string required') if kind=='date':date.fromisoformat(value) if kind=='utc_timestamp': parsed=datetime.fromisoformat(value) if parsed.tzinfo is None or parsed.utcoffset()!=timedelta(0):raise ValueError(path+': explicit UTC timestamp required') else:raise ValueError(path+': unsupported schema type') _INPUT_SCHEMA = {'kind': 'object', 'fields': {'decision_cutoff': {'kind': 'utc_timestamp'}, 'loan_sale_gain_cents': {'kind': 'integer'}, 'operating_cost_cents': {'kind': 'integer', 'minimum': 0}, 'actual_combined_contribution_cents': {'kind': 'integer'}, 'notional_cents': {'kind': 'integer', 'minimum': 1}, 'open_price_bps': {'kind': 'integer', 'minimum': 1}, 'risk_limit_cents_per_bp': {'kind': 'integer', 'minimum': 0}, 'candidates': {'kind': 'list', 'item': {'kind': 'object', 'fields': {'id': {'kind': 'string'}, 'quote_observed_at': {'kind': 'utc_timestamp'}, 'close_price_bps': {'kind': 'integer', 'minimum': 1}, 'fees_cents': {'kind': 'integer', 'minimum': 0}, 'carry_cents': {'kind': 'integer', 'minimum': 0}, 'executable': {'kind': 'boolean'}, 'residual_dv01_cents_per_bp': {'kind': 'integer'}}}}}} def _calculate(data): from datetime import datetime from decimal import Decimal, ROUND_HALF_UP cutoff=datetime.fromisoformat(data['decision_cutoff']) if cutoff.tzinfo is None:raise ValueError('Timezone required') rows=[] for c in data['candidates']: observed=datetime.fromisoformat(c['quote_observed_at']) if observed.tzinfo is None:raise ValueError('Timezone required') if observed>cutoff: rows.append({'candidate':c['id'],'status':'excluded_future_information'});continue if c['executable'] is not True: rows.append({'candidate':c['id'],'status':'excluded_no_executable_source'});continue gross=int((Decimal(data['notional_cents'])*Decimal(data['open_price_bps']-c['close_price_bps'])/Decimal(10000)).quantize(Decimal(1),rounding=ROUND_HALF_UP)) combined=data['loan_sale_gain_cents']+gross-c['fees_cents']-data['operating_cost_cents']-c['carry_cents'] status='infeasible_risk_limit' if abs(c['residual_dv01_cents_per_bp'])>data['risk_limit_cents_per_bp'] else 'hypothetical_feasible' rows.append({'candidate':c['id'],'status':status,'hypothetical_contribution_cents':combined,'difference_from_actual_cents':combined-data['actual_combined_contribution_cents']}) return {'label':'hypothetical_fixed_population_and_cost_assumptions','candidates':rows,'realized_incremental_gain_claimed_cents':0,'external_actions':0} def run(data): try: _validate(data, _INPUT_SCHEMA) return _calculate(data) except (KeyError, TypeError, ValueError, ArithmeticError): return {'status':'hold', 'reason':'invalid_input', 'external_actions':0} 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()