APRENDER. PRATICAR. EVOLUIR.Definir o âmbito do estado
O laboratório identifica cada livro por fornecedor, canal, instrumento e época local. B1 num canal não é automaticamente B1 noutro. As sequências só têm significado dentro do contrato que as define. Antes do ensaio, escrever qual conjunto de entradas o snapshot cobre e que eventos estão incluídos no checkpoint. O código usa campos Python inventados pela DR, sem descodificar mensagens FIX ou CME. Os preços são inteiros positivos numa unidade local e as quantidades também são positivas. Esta restrição facilita observar alterações de estado; não representa todos os instrumentos, tipos de cotação ou valores possíveis nos mercados.
Aplicar operações sem publicar estado parcial
A base começa em 100 com B1 bid e A1 ask. O evento 101 altera B1; o 102 elimina A1. Sem outra entrada ask, a consulta devolve ausência, não preço zero nem o valor antigo com data renovada. Cada evento prepara uma cópia e valida as operações antes de devolver o candidato. Se uma operação posterior tentar eliminar um ID inexistente, nenhuma das alterações desse evento é publicada. O estado anterior continua disponível para diagnóstico, mas a marca utilizável passa a falsa. O código trabalha em memória e não demonstra transações distribuídas ou durabilidade após falha de energia.
Tornar a lacuna visível aos consumidores
Depois de aceitar 102, receber 104 revela uma descontinuidade. O modelo conserva a sequência 102 e bloqueia a consulta do melhor preço. Um heartbeat recente não remove essa condição. A política local exige novo snapshot para voltar a utilizar o livro; receber entretanto 103 não reabre automaticamente o gate. Uma sequência repetida também provoca indisponibilidade neste exercício conservador. Não generalizar este comportamento a todos os protocolos: alguns contratos definem tratamento próprio para duplicados e retransmissões. O PM deve pedir os critérios reais de invalidação e recuperação, além do resultado esperado quando o fornecedor, canal ou época não coincide.
Construir um candidato a partir do checkpoint
O novo snapshot em 103 contém apenas A2. A instalação substitui o conjunto anterior, retirando B1. O buffer local ignora eventos até 103 inclusive e começa a aplicar em 104. Se o primeiro evento posterior for 105, a recuperação do candidato falha; não preencher o intervalo com uma atualização vazia inventada. Esta fronteira inclusiva é uma regra original da oficina. A orientação CME consultada usa relações específicas entre sequência de pacote, instrumento e TransactTime, bem como iterações de recuperação. O campo local checkpoint não é uma implementação do campo 369. Antes de integrar um fornecedor, qualificar o contrato completo e as diferenças.
Rever versões e separar recuperação de história
Reconstruir o estado atual não prova que se recuperou cada evento perdido. Também não demonstra execução ou liquidação de uma ordem. A orientação de recuperação permite distinguir instrumentos já recuperados dos que ainda aguardam tratamento. O relatório deve conservar esse âmbito. Um aviso CME de agosto de 2026 acrescenta outra dependência de versão: no contexto indicado, as estatísticas diárias não devem ser limpas após channel reset. Um runbook antigo precisa de revisão contra os avisos aplicáveis. O modelo não implementa essas estatísticas. O documento FIX de janeiro de 2007 localizado na pesquisa não foi usado como especificação atual do exercício.
Executar o ensaio e explicar o resultado
Reservar dez minutos para prever o livro, vinte para executar e introduzir falhas, vinte para comparar candidatos e dez para comunicar os limites. Copiar o programa completo abaixo e executar python3 run.py --output evidence.json. Usa a biblioteca padrão de Python e produz 45 verificações, incluindo as entradas de avaliação da próxima aula. Guardar hash, versão do intérprete, previsões e resultados. Ensaiar explicitamente uma entrada ausente, um ID repetido e uma lacuna no buffer. A execução não liga a uma bolsa, não usa feeds licenciados e não envia ordens. A aceitação real exige evidência dos componentes e serviços contratados.
"""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'].values()if 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()
Caso: um snapshot correto é fundido com dados antigos e conserva uma cotação retirada. A sequência parece recente, mas a comparação das identidades revela o erro.
Armadilhas comuns
Fusão de snapshot completo; sequência recente como prova; heartbeat como atualização de preços; algoritmo local apresentado como conformidade com o fornecedor.
Tópicos relacionados: Dados de referência e de mercado · Sessões FIX e estado das ordens · Operação, mudança e preparação T+1
Só publicar um candidato cujo âmbito, base e continuidade foram demonstrados, preservando o estado de falha e as dependências ainda por qualificar.
Referência: CME MDP 3.0 MBP and MBOFD market recovery · DR financial markets technology professional assessment2026.10