Workshop: snapshots and market-data reconstruction

"""Original DR market-data recovery and valuation-input exercise. Python 3.13.
python3 run.py --output evidence.json
Controlled fictional data only. Not a CME/FIX implementation, matching engine,
valuation policy, regulated NAV, real approval workflow or trading instruction.
Checkpoint, sequence, invalidation, freshness and rounding rules are local.
"""
import argparse
from copy import deepcopy
from decimal import Decimal, ROUND_HALF_EVEN
import hashlib
import json
from pathlib import Path
import sys

checks=[]
def check(name,actual,expected):
 assert actual==expected,(name,actual,expected)
 checks.append(dict(name=name,actual=actual,expected=expected,passed=True))
def reject(fn):
 try:fn
 except ValueError as e:return str(e)
 raise AssertionError('Expected rejection')
def validate_entry(e):
 if e['side']not in('bid','ask'):raise ValueError('unknown side')
 if type(e['price'])is not int or e['price']<=0:raise ValueError('invalid local price')
 if type(e['quantity'])is not int or e['quantity']<=0:raise ValueError('invalid local quantity')
def snapshot(scope,checkpoint,entries):
 if type(checkpoint)is not int or checkpoint<0:raise ValueError('invalid checkpoint')
 rows={}
 for e in entries:
 if e['id']in rows:raise ValueError('duplicate entry identity')
 validate_entry(e);rows[e['id']]=deepcopy(e)
 return dict(scope=list(scope),seq=checkpoint,entries=rows,usable=True,failure=None)
def apply_delta(book,event):
 if not book['usable']:raise ValueError('snapshot required')
 if event['scope']!=book['scope']:raise ValueError('scope mismatch')
 if event['seq']!=book['seq']+1:raise ValueError('sequence discontinuity')
 out=deepcopy(book)
 for change in event['changes']:
 k=change['entry']['id'];kind=change['action']
 if kind=='add':
 if k in out['entries']:raise ValueError('entry already exists')
 validate_entry(change['entry']);out['entries'][k]=deepcopy(change['entry'])
 elif kind=='change':
 if k not in out['entries']:raise ValueError('entry absent')
 validate_entry(change['entry']);out['entries'][k]=deepcopy(change['entry'])
 elif kind=='delete':
 if k not in out['entries']:raise ValueError('entry absent')
 del out['entries'][k]
 else:raise ValueError('unknown action')
 out['seq']=event['seq'];return out
def observe(book,event):
 try:return apply_delta(book,event)
 except ValueError as e:return {**deepcopy(book),'usable':False,'failure':str(e)}
def restore(book,new_snapshot,buffer):
 if new_snapshot['scope']!=book['scope']:raise ValueError('scope mismatch')
 if new_snapshot['seq']<book['seq']:raise ValueError('older checkpoint')
 out=deepcopy(new_snapshot)
 # Local checkpoint covers all events <= checkpoint, unlike a vendor wire rule.
 checkpoint=out['seq']
 for e in buffer:
 if e['scope']!=out['scope']:raise ValueError('scope mismatch')
 if e['seq']<=checkpoint:continue
 out=apply_delta(out,e)
 return out
def best(book,side):
 if not book['usable']:raise ValueError('book unavailable')
 prices=[e['price']for e in book['entries'].valuesif e['side']==side]
 return (max(prices)if side=='bid'else min(prices))if prices else None
def decimal_text(value):
 if not isinstance(value,str):raise ValueError('decimal text required')
 try:d=Decimal(value)
 except Exception:raise ValueError('invalid decimal')
 if not d.is_finite:raise ValueError('non finite decimal')
 return d
def fingerprint(inputs):return hashlib.sha256(json.dumps(inputs,sort_keys=True,separators=(',',':')).encode).hexdigest
def valuation(inputs,now,max_age):
 # Bounded positive positions and EUR prices only. No FX, fees or share classes.
 total=Decimal('0')
 for instrument,quantity in inputs['positions'].items:
 if instrument not in inputs['prices']:raise ValueError('missing price')
 p=inputs['prices'][instrument]
 if p['currency']!='EUR':raise ValueError('currency mismatch')
 if p['unit']not in('EUR','cent'):raise ValueError('unknown price unit')
 age=now-p['observed']
 if age<0:raise ValueError('future price timestamp')
 if age>max_age:raise ValueError('stale price')
 qty=decimal_text(quantity);price=decimal_text(p['value'])
 if qty<0 or price<0:raise ValueError('outside positive fixture scope')
 total+=qty*price/(100 if p['unit']=='cent'else 1)
 assets=total+decimal_text(inputs['cash']);net=assets-decimal_text(inputs['liabilities']);units=decimal_text(inputs['units'])
 if units<=0:raise ValueError('positive units required')
 per_unit=net/units
 return dict(assets=str(assets),net=str(net),raw=str(per_unit),display=str(per_unit.quantize(Decimal('0.01'),rounding=ROUND_HALF_EVEN)),inputHash=fingerprint(inputs))
def approved(result,record):return record.get('inputHash')==result['inputHash'] and record.get('decision')=='approved'

def main:
 scope=['DR-VENDOR','DR-CHANNEL','INSTRUMENT-A','EPOCH-1']
 bid=dict(id='B1',side='bid',price=10000,quantity=10);ask=dict(id='A1',side='ask',price=10005,quantity=15)
 base=snapshot(scope,100,[bid,ask]);original=deepcopy(base)
 def event(seq,changes,sc=scope):return dict(scope=list(sc),seq=seq,changes=changes)
 def change(action,entry):return dict(action=action,entry=entry)
 e101=event(101,[change('change',{**bid,'price':10001})]);e102=event(102,[change('delete',{'id':'A1'})])
 one=observe(base,e101);two=observe(one,e102)
 check('increment changes existing quote',best(one,'bid'),10001)
 check('deletion removes old ask',best(two,'ask'),None)
 check('accepted sequence advances',two['seq'],102)
 check('inputs preserved after delta',base==original,True)
 gap=observe(two,event(104,[change('add',ask)]))
 check('gap disables local usability',gap['usable'],False)
 check('gap retains last accepted sequence',gap['seq'],102)
 check('gap does not publish partial entry',list(gap['entries']),['B1'])
 check('unusable book cannot supply best quote',reject(lambda:best(gap,'bid')),'book unavailable')
 check('missing delta alone does not clear local gate',observe(gap,event(103,[]))['failure'],'snapshot required')
 check('duplicate sequence invalidates local book',observe(one,e101)['failure'],'sequence discontinuity')
 check('scope mismatch invalidates local book',observe(base,event(101,[],['DR-VENDOR','OTHER','INSTRUMENT-A','EPOCH-1']))['failure'],'scope mismatch')
 check('change absent entry rejected',observe(base,event(101,[change('change',{**bid,'id':'MISSING'})]))['failure'],'entry absent')
 check('add existing entry rejected',observe(base,event(101,[change('add',bid)]))['failure'],'entry already exists')
 bad=observe(base,event(101,[change('change',{**bid,'price':9999}),change('delete',{'id':'MISSING'})]))
 check('failed multi-change retains original quote',bad['entries']['B1']['price'],10000)
 check('failed multi-change retains sequence',bad['seq'],100)
 fresh=snapshot(scope,103,[{**ask,'id':'A2','price':10006}])
 restored=restore(gap,fresh,[event(103,[]),event(104,[change('add',{**bid,'id':'B2','price':10002})])])
 check('snapshot replaces rather than merges',sorted(restored['entries']),['A2','B2'])
 check('buffer resumes after local checkpoint',restored['seq'],104)
 check('recovered local gate is usable',restored['usable'],True)
 check('recovered best bid',best(restored,'bid'),10002)
 check('old checkpoint rejected',reject(lambda:restore(gap,base,[])),'older checkpoint')
 check('buffer gap prevents recovery candidate',reject(lambda:restore(gap,fresh,[event(105,[])])),'sequence discontinuity')
 check('recovery candidate leaves old book untouched',gap['usable'],False)
 check('duplicate snapshot identity rejected',reject(lambda:snapshot(scope,100,[bid,bid])),'duplicate entry identity')
 check('invalid price rejected',reject(lambda:snapshot(scope,100,[{**bid,'price':0}])),'invalid local price')
 check('invalid quantity rejected',reject(lambda:snapshot(scope,100,[{**bid,'quantity':-1}])),'invalid local quantity')
 inputs=dict(positions={'A':'10','B':'5'},prices={'A':dict(value='10001',unit='cent',currency='EUR',observed=95,revision='A-1'),'B':dict(value='200.00',unit='EUR',currency='EUR',observed=98,revision='B-1')},cash='50.00',liabilities='10.00',units='100')
 result=valuation(inputs,100,5)
 check('assets include normalized prices and cash',result['assets'],'2050.10')
 check('net subtracts liabilities',result['net'],'2040.10')
 check('unrounded per unit value',result['raw'],'20.401')
 check('display rounds only final value',result['display'],'20.40')
 check('freshness exact boundary passes',valuation(inputs,100,5)['display'],'20.40')
 check('one tick beyond freshness held',reject(lambda:valuation(inputs,101,5)),'stale price')
 future=deepcopy(inputs);future['prices']['A']['observed']=101
 check('future price timestamp held',reject(lambda:valuation(future,100,5)),'future price timestamp')
 missing=deepcopy(inputs);del missing['prices']['B']
 check('missing instrument price held',reject(lambda:valuation(missing,100,5)),'missing price')
 wrong=deepcopy(inputs);wrong['prices']['B']['currency']='USD'
 check('currency mismatch held',reject(lambda:valuation(wrong,100,5)),'currency mismatch')
 wrong_unit=deepcopy(inputs);wrong_unit['prices']['A']['unit']='UNKNOWN'
 check('unknown price unit held',reject(lambda:valuation(wrong_unit,100,5)),'unknown price unit')
 check('float input rejected',reject(lambda:decimal_text(0.1)),'decimal text required')
 check('NaN rejected',reject(lambda:decimal_text('NaN')),'non finite decimal')
 check('Infinity rejected',reject(lambda:decimal_text('Infinity')),'non finite decimal')
 zero={**inputs,'units':'0'}
 check('zero outstanding units rejected',reject(lambda:valuation(zero,100,5)),'positive units required')
 approval=dict(inputHash=result['inputHash'],decision='approved')
 check('approval binds current inputs',approved(result,approval),True)
 corrected=deepcopy(inputs);corrected['prices']['A']['value']='10011'corrected['prices']['A']['revision']='A-2'
 updated=valuation(corrected,100,5)
 check('correction changes net',updated['net'],'2041.10')
 check('old approval does not bind corrected inputs',approved(updated,approval),False)
 check('inputs retained for reproducibility',inputs['prices']['A']['revision'],'A-1')
 check('local half even tie',str(Decimal('1.005').quantize(Decimal('0.01'),rounding=ROUND_HALF_EVEN)),'1.00')
 check('multiply before display rounding',str((Decimal('1.005')*2).quantize(Decimal('0.01'),rounding=ROUND_HALF_EVEN)),'2.01')
 output=dict(scope='Original bounded book-recovery and valuation-input fixtures; no CME/FIX implementation, licensed feed, trading, regulated NAV, actual approval or independent acceptance.',python=sys.version.split[0],runnerSha256=hashlib.sha256(Path(__file__).read_bytes).hexdigest,passed=len(checks),checks=checks)
 parser=argparse.ArgumentParser;parser.add_argument('--output',required=True)
 Path(parser.parse_args.output).write_text(json.dumps(output,indent=2)+'\n');print(json.dumps({'passed':len(checks),'scope':output['scope']}))
if __name__=='__main__':main

Related topics: Reference and market data · FIX sessions and order state · Operations, change, and T+1 readiness

Reference: CME MDP 3.0 MBP and MBOFD market recovery · BigSavant financial markets technology professional assessment2026.10