APRENDER. PRATICAR. EVOLUIR.Definir identidade e responsabilidade
Uma migração de mandatos não fica demonstrada pela igualdade do número de clientes. O exercício usa dois credores com UMR-1 para mostrar uma colisão: guardar apenas a UMR perde uma das relações. Conservar credor e referência permite distinguir as autorizações. A orientação pública EPC atribui ao credor a conservação do mandato original e da informação sobre alterações ou cancelamentos. No projeto fictício, identificar quem fornece esses registos, quem os transforma e quem confirma que continuam recuperáveis. O laboratório não contém assinaturas nem verifica validade jurídica; trabalha apenas a coerência dos dados e a evidência que o PM deve pedir.
Conservar a situação aplicável no tempo
A versão 1 é efetiva em 10, a versão 2 altera a conta em 20 e a versão 3 regista revogação em 30. São unidades sintéticas, sem calendário bancário. Em 19, o modelo seleciona a versão 1; exatamente em 20, seleciona a versão 2. Uma decisão nova em 30 encontra revogação, embora continue possível reconstruir o estado de 15. Selecionar sempre a última linha destruiria esta distinção. Quando duas versões têm o mesmo instante efetivo, o exercício recusa escolher pela ordem do ficheiro. A regra de precedência real precisa de ser acordada e demonstrada pelos responsáveis pelos dados.
Ligar confirmação aos dados usados
O fluxo B2B exige atenção à confirmação e verificação dos dados do mandato pelo PSP do devedor. Para ensinar essa dependência, a DR define confirmedVersion: a confirmação deve apontar para a versão avaliada. A conta B pode estar correta e, ainda assim, existir apenas confirmação ligada à versão da conta A. O modelo mantém a decisão suspensa. Este campo e esta política de versões são escolhas didáticas, não campos oficiais de mensagem nem um procedimento universal EPC. Alterar o número à mão não cria evidência. A integração real deve definir os dados confirmados, alterações relevantes e critérios de aceitação aplicáveis.
Investigar a causa com o contexto
Num ensaio de migração, MD02 acompanha dados transmitidos que divergem do mandato guardado. Começar pela propagação da alteração e pelo mapeamento é uma hipótese mais útil do que aumentar recursos de infraestrutura. MD01, por si só, não demonstra que todos os devedores revogaram mandatos; analisar a confirmação e o contexto de cada caso. AC13 num fluxo B2B para conta de consumidor orienta a revisão do instrumento e da conta. Não converter o esquema silenciosamente para fazer passar o ficheiro. Estas decisões usam a orientação de códigos v8.1 consultada; não substituem regras completas, acordos operacionais ou análise jurídica.
Demonstrar a utilização após cutover
O restauro pode carregar a revogação correta na base enquanto um worker conserva a versão ativa em memória. O health check continua verde. Pedir prova do estado carregado e da decisão produzida em cada consumidor afetado. A aula anterior usou snapshots para associar respostas VOP; aqui o problema é preservar e aplicar o estado do mandato ao longo de uma migração. As duas verificações não se substituem. Incluir no plano de cutover um caso antes da alteração, um na fronteira e um após revogação. Se um consumidor continuar desatualizado, conservar a ação e o impacto no relatório de passagem à operação.
Executar o modelo e conservar os limites
Reservar quinze minutos para prever snapshots e decisões, vinte para executar e introduzir colisões, quinze para comparar históricos e dez para explicar o resultado. Copiar o programa completo e executar python3 run.py --output evidence.json. A biblioteca padrão é suficiente. O relatório inclui 40 verificações, também cobrindo a reconciliação da próxima aula. Uma cópia independente impede que a alteração de um snapshot devolvido modifique o histórico original. Isto não demonstra persistência, concorrência ou resistência a falhas de energia. Os dados são fictícios, sem contas reais, chamadas PSP ou execução de cobranças. Guardar previsões, resultados, hash do programa e questões que o exercício não resolve.
"""Original DR collection evidence model. Python 3.13; standard library only.
python3 run.py --output evidence.json
Controlled fictional snapshots and observations. No bank connection, SEPA
message parser, mandate validity decision, payment execution or legal advice.
Integer time and cents, immutable snapshots and version-bound confirmation are
local teaching rules. Only one full debit after a settlement is modeled.
"""
import argparse
from copy import deepcopy
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 snapshot(history,creditor,umr,at):
candidates=[s for s in history.get((creditor,umr),[]) if s['effective']<=at]
if not candidates:raise ValueError('no applicable snapshot')
times=[s['effective'] for s in candidates]
if len(times)!=len(set(times)):raise ValueError('ambiguous effective time')
return deepcopy(max(candidates,key=lambda s:s['effective']))
def eligibility(history,collection):
try:s=snapshot(history,collection['creditor'],collection['umr'],collection['at'])
except ValueError as e:return str(e)
if s['state']!='ACTIVE':return 'inactive mandate'
if s['scheme']!=collection['scheme']:return 'scheme mismatch'
if s['account']!=collection['account']:return 'account mismatch'
if s['scheme']=='B2B' and s['confirmedVersion']!=s['version']:return 'confirmation not bound to version'
return 'LOCAL_CHECKS_PASS'
def reconcile(collections,events):
# Isolated candidate. Any error discards it; inputs remain untouched.
rows={k:dict(amount=v['amount'],currency=v['currency'],state='UNKNOWN',credit=0,debit=0) for k,v in collections.items()}
seen={}
for e in events:
identity=e['event'];signature=tuple(e[k]for k in('collection','kind','amount','currency'))
if identity in seen:
if seen[identity]!=signature:raise ValueError('event identity conflict')
continue
if e['collection'] not in rows:raise ValueError('unknown collection')
row=rows[e['collection']]
if e['amount']!=row['amount'] or type(e['amount']) is not int:raise ValueError('amount mismatch')
if e['currency']!=row['currency']:raise ValueError('currency mismatch')
kind=e['kind']
if kind=='SUBMITTED':
if row['state']!='UNKNOWN':raise ValueError('unexpected submission')
row['state']='PENDING'
elif kind=='REJECTED':
if row['credit'] or row['debit']:raise ValueError('reject after movement')
row['state']='REJECTED'
elif kind=='SETTLED':
if row['credit']:raise ValueError('duplicate economic effect')
if row['state']=='REJECTED':raise ValueError('conflicting outcome')
row['credit']=row['amount'];row['state']='SETTLED'
elif kind=='REFUND_REQUEST':
if not row['credit']:raise ValueError('missing settlement evidence')
if row['debit']:raise ValueError('already reversed')
row['state']='REFUND_PENDING'
elif kind in('RETURN_DEBIT','REFUND_DEBIT'):
if not row['credit']:raise ValueError('missing settlement evidence')
if row['debit']:raise ValueError('duplicate economic effect')
row['debit']=row['amount'];row['state']='RETURNED' if kind=='RETURN_DEBIT' else 'REFUNDED'
else:raise ValueError('unknown observation kind')
seen[identity]=signature
return dict(rows=rows,observations=len(seen),credit=sum(r['credit']for r in rows.values()),debit=sum(r['debit']for r in rows.values()),net=sum(r['credit']-r['debit']for r in rows.values()))
def main():
v1=dict(effective=10,version=1,state='ACTIVE',scheme='B2B',account='ACCOUNT-A',confirmedVersion=1)
v2={**v1,'effective':20,'version':2,'account':'ACCOUNT-B','confirmedVersion':1}
v3={**v2,'effective':30,'version':3,'state':'REVOKED'}
history={('CREDITOR-A','UMR-1'):[v1,v2,v3],('CREDITOR-B','UMR-1'):[{**v1,'account':'ACCOUNT-Z'}]}
before=deepcopy(history)
c=dict(creditor='CREDITOR-A',umr='UMR-1',at=15,scheme='B2B',account='ACCOUNT-A')
check('snapshot before amendment',snapshot(history,'CREDITOR-A','UMR-1',19)['version'],1)
check('snapshot at amendment boundary',snapshot(history,'CREDITOR-A','UMR-1',20)['version'],2)
check('snapshot at revocation boundary',snapshot(history,'CREDITOR-A','UMR-1',30)['state'],'REVOKED')
check('creditor separates equal UMR',snapshot(history,'CREDITOR-B','UMR-1',15)['account'],'ACCOUNT-Z')
check('before first snapshot rejected',reject(lambda:snapshot(history,'CREDITOR-A','UMR-1',9)),'no applicable snapshot')
check('unknown mandate rejected',reject(lambda:snapshot(history,'MISSING','UMR-1',15)),'no applicable snapshot')
copy=snapshot(history,'CREDITOR-A','UMR-1',15);copy['account']='CHANGED'
check('snapshot copy preserves history',history==before,True)
check('original B2B snapshot passes local checks',eligibility(history,c),'LOCAL_CHECKS_PASS')
check('changed account blocks old data',eligibility(history,{**c,'at':20}),'account mismatch')
check('old confirmation does not bind new version',eligibility(history,{**c,'at':20,'account':'ACCOUNT-B'}),'confirmation not bound to version')
confirmed=deepcopy(history);confirmed[('CREDITOR-A','UMR-1')][1]['confirmedVersion']=2
check('version bound confirmation passes locally',eligibility(confirmed,{**c,'at':20,'account':'ACCOUNT-B'}),'LOCAL_CHECKS_PASS')
check('revoked mandate blocks new local check',eligibility(history,{**c,'at':30,'account':'ACCOUNT-B'}),'inactive mandate')
check('historical decision remains reconstructable',eligibility(history,c),'LOCAL_CHECKS_PASS')
check('scheme change is not implicit',eligibility(history,{**c,'scheme':'CORE'}),'scheme mismatch')
ambiguous=deepcopy(history);ambiguous[('CREDITOR-A','UMR-1')].append(dict(v1))
check('same effective time is ambiguous',reject(lambda:snapshot(ambiguous,'CREDITOR-A','UMR-1',15)),'ambiguous effective time')
collections={k:dict(amount=n,currency='EUR') for k,n in [('A',10000),('B',20000),('C',30000),('D',40000)]}
def event(i,c,kind,**change):return dict(event=i,collection=c,kind=kind,**{**collections[c],**change})
events=[event('E1','A','SETTLED'),event('E2','B','SETTLED'),event('E3','C','REJECTED'),event('E4','D','SUBMITTED'),event('E5','A','RETURN_DEBIT'),event('E6','B','REFUND_REQUEST')]
saved=deepcopy(events);out=reconcile(collections,events)
check('credit sum in integer cents',out['credit'],30000)
check('return debit sum in integer cents',out['debit'],10000)
check('net observed amount in integer cents',out['net'],20000)
check('return retains original credit',out['rows']['A']['credit'],10000)
check('refund request keeps money unchanged',out['rows']['B']['debit'],0)
check('refund request remains pending',out['rows']['B']['state'],'REFUND_PENDING')
check('submission does not prove settlement',out['rows']['D']['state'],'PENDING')
check('rejected collection has zero movement',out['rows']['C']['credit']-out['rows']['C']['debit'],0)
check('identical repeated event counted once',reconcile(collections,events+[events[0]])['observations'],6)
check('identical repeat does not change totals',reconcile(collections,events+[events[0]])['net'],20000)
check('conflicting event identity rejected',reject(lambda:reconcile(collections,events+[{**events[0],'amount':9999}])),'event identity conflict')
check('new ID cannot duplicate economic credit',reject(lambda:reconcile(collections,events+[event('E7','B','SETTLED')])),'duplicate economic effect')
check('amount mismatch rejected',reject(lambda:reconcile(collections,[event('E1','A','SETTLED',amount=10001)])),'amount mismatch')
check('currency mismatch rejected',reject(lambda:reconcile(collections,[event('E1','A','SETTLED',currency='USD')])),'currency mismatch')
check('unknown collection rejected',reject(lambda:reconcile(collections,[{**events[0],'collection':'MISSING'}])),'unknown collection')
check('return before settlement held for investigation',reject(lambda:reconcile(collections,[events[4]])),'missing settlement evidence')
check('refund completion updates net',reconcile(collections,events+[event('E7','B','REFUND_DEBIT')])['net'],0)
check('refund completion retains original credit',reconcile(collections,events+[event('E7','B','REFUND_DEBIT')])['rows']['B']['credit'],20000)
check('double debit rejected',reject(lambda:reconcile(collections,events+[event('E7','A','REFUND_DEBIT')])),'duplicate economic effect')
check('reject after settlement rejected',reject(lambda:reconcile(collections,events+[event('E7','B','REJECTED')])),'reject after movement')
check('settlement after reject needs investigation',reject(lambda:reconcile(collections,events+[event('E7','C','SETTLED')])),'conflicting outcome')
check('unknown observation kind rejected',reject(lambda:reconcile(collections,[event('E7','A','RETRY_NOW')])),'unknown observation kind')
check('failed candidates preserve event input',events==saved,True)
check('missing evidence remains unknown',reconcile(collections,[])['rows']['A']['state'],'UNKNOWN')
check('all collections retain identity',sorted(out['rows']),['A','B','C','D'])
output=dict(scope='Original mandate-snapshot and collection-observation fixtures; local eligibility is not authorization, observed net is not a bank ledger, no actual SEPA processing or conformance.',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: a migração preserva o número de clientes, mas funde mandatos de dois credores com a mesma UMR. A reconciliação deve comparar relações e histórico, não apenas totais.
Armadilhas comuns
UMR isolada como chave; última linha como verdade histórica; confirmação antiga como prova de dados novos; base atualizada como prova de todos os consumidores.
Tópicos relacionados: Mandatos, Core e B2B · Dados e verificação do beneficiário · Alterações e prontidão operacional
Uma migração útil conserva o significado do mandato e permite reconstruir a decisão com dados e evidência aplicáveis ao instante em análise.
Referência: EPC SEPA Direct Debit overview · DR Payments and SEPA professional assessment2026.10