> Mejora continua de procesos > Procesamiento de pagos > SWIFT > Template de datos

Su Template de datos de procesamiento de pagos

SWIFT

Atributos del procesamiento de pagos (19) Actividades del procesamiento de pagos (12) Guías de extracción (3)

Su Template de datos de procesamiento de pagos

Este Template proporciona la base estructural para mapear sus Workflow de transacciones SWIFT desde el inicio hasta la liquidación final. Detalla los componentes de datos y los hitos operativos específicos necesarios para obtener una visibilidad completa del ciclo de vida de sus pagos. Al utilizar esta guía, puede preparar sus datos de forma eficaz para descubrir ineficiencias ocultas y reducir los riesgos de cumplimiento.

Atributos recomendados para un análisis profundo de las transacciones
Actividades e hitos clave del proceso SWIFT
Orientación técnica para extraer datos financieros

¿Es nuevo en los registros de eventos? Aprenda a crear un registro de eventos de Process Mining.

Descargar Template

Atributos del procesamiento de pagos

Estos son los campos de datos recomendados para incluir en su registro de eventos y realizar un análisis completo del ciclo de vida de sus transacciones SWIFT.

5 Obligatorio 9 Recomendado 5 Opcional

	Nombre	Descripción
	ID de la transacción de pago PaymentTransactionId	El identificador único que representa el caso de pago de principio a fin.
Descripción Este Atributo actúa como clave central del caso para Process Mining. Agrupa todas las actividades relacionadas con una única instrucción de pago, desde la solicitud inicial hasta la validación, la transmisión SWIFT y la liquidación final. En contextos SWIFT, suele asignarse al Transaction Reference Number (TRN) o al Unique End-to-end Transaction Reference (UETR) para garantizar la continuidad entre los sistemas bancarios. Se utiliza para reconstruir la ruta de la instancia del proceso y es esencial para todos los análisis de variantes y cálculos de tiempos de ciclo. Por qué es importante La identificación única es la base de Process Mining, ya que permite unir eventos dispersos en una visión coherente del proceso. Dónde obtenerlo Campo 20 de SWIFT (TRN) o campo 121 (UETR) en los encabezados de los mensajes. Ejemplos TRN-20231025-883954392-882-101UETR-9982-1123-5521PAY-US-EU-9912
	Marca de tiempo del evento EventTimestamp	La fecha y hora exactas en las que se produjo la actividad.
Descripción Este Atributo registra el momento exacto en que tuvo lugar un evento. Se utiliza para ordenar cronológicamente las actividades y calcular todas las métricas basadas en la duración, como los plazos, los tiempos de ciclo y el rendimiento. Las marcas de tiempo precisas son fundamentales para analizar el «Network Cut Off Time Performance» e identificar cuellos de botella en la cadena de aprobación. Por qué es importante Las marcas de tiempo son la base de todos los análisis temporales y las métricas de rendimiento de Process Mining. Dónde obtenerlo Registros de auditoría del sistema, marcas de tiempo de creación de mensajes o tiempos de transacción de la base de datos. Ejemplos 2023-10-25T08:30:15Z2023-10-25T14:45:00Z2023-10-26T09:15:22Z
	Nombre de la actividad ActivityName	El nombre del paso del proceso o del evento que se produjo.
Descripción Este Atributo describe la acción específica o el cambio de estado registrado en el registro del sistema. Algunos ejemplos son «Payment Request Created», «Sanctions Screening Passed» o «SWIFT ACK Received». Es la dimensión principal de los mapas de procesos, ya que define los nodos de la visualización y permite a los analistas comprender la secuencia de operaciones realizadas sobre un pago. Por qué es importante Define el «qué» del proceso, lo que permite visualizar el flujo y las variaciones del proceso. Dónde obtenerlo Registros de transacciones, registros de auditoría o datos derivados de los tipos de mensaje, por ejemplo, MT103 Sent. Ejemplos Solicitud de pago creadaInstrucción de pago enviadaACK de SWIFT recibidoPago aprobado
	Sistema de origen SourceSystem	El nombre del sistema del que proceden los datos del evento.
Descripción Identifica el software o la plataforma que generó el registro de eventos. En un proceso de pagos, puede tratarse del sistema bancario central, del gateway de pagos o directamente de la interfaz SWIFT. Este Atributo ayuda a verificar el linaje de los datos y a depurar problemas al combinar datos de varias aplicaciones bancarias independientes. Por qué es importante Proporciona contexto sobre el origen de los datos, algo esencial para Process Mining en varios sistemas. Dónde obtenerlo Codificado durante el proceso ETL o extraído de los identificadores del sistema. Ejemplos SWIFT Alliance AccessSistema bancario centralMotor de pagosFiltro de sanciones
	Última actualización de datos LastDataUpdate	La marca de tiempo en la que el registro se extrajo o actualizó por última vez.
Descripción Indica cuándo se cargaron por última vez los datos en la herramienta de Process Mining. Es fundamental para determinar la actualidad de los datos y garantizar que los Dashboards reflejen el estado más reciente de las operaciones de pago. Ayuda a que los usuarios confíen en los datos al proporcionar transparencia sobre la latencia de la canalización de informes. Por qué es importante Garantiza que los usuarios comprendan la actualidad y fiabilidad de los datos presentados. Dónde obtenerlo Marca de tiempo de ejecución del proceso ETL. Ejemplos 2023-10-27T12:00:00Z2023-11-01T06:00:00Z
	BIC del beneficiario BeneficiaryBic	El Bank Identifier Code de la institución receptora.
Descripción Identifica el banco de destino del pago. Es una dimensión esencial para el Dashboard «Beneficiary Rejection Root Cause». Al agrupar los fallos por BIC del beneficiario, el banco puede identificar contrapartes concretas que rechazan instrucciones con frecuencia debido a problemas de calidad de datos o requisitos de formato específicos. Por qué es importante Esencial para analizar el rendimiento de las contrapartes y los patrones de rechazo. Dónde obtenerlo SWIFT Field 57A (Account With Institution) o 58A (Beneficiary Institution). Ejemplos CITIUS33BARCGB22DRESDEFFHANDSEXX
	Código NAK de SWIFT SwiftNakCode	El código de error que devuelve la red SWIFT cuando se rechaza un mensaje.
Descripción Contiene el código de error específico, por ejemplo, T26 o T13, incluido en un Negative Acknowledgment (NAK) de la red SWIFT. Este atributo alimenta el Dashboard «SWIFT Error and Rework Analysis», que permite categorizar los fallos técnicos para priorizar las correcciones del sistema. Por qué es importante Proporciona las causas técnicas de los rechazos a nivel de red. Dónde obtenerlo MT 015/019 System Messages o Field 451 en ACK/NAK. Ejemplos T26H01T13G02
	Divisa de la transacción TransactionCurrency	El código ISO de 3 letras que indica la divisa del pago.
Descripción Identifica la divisa en la que está denominado el pago, por ejemplo, USD, EUR o GBP. Es fundamental para el Dashboard «Network Cut Off Time Performance», ya que los plazos límite dependen de la divisa. También permite analizar la «Currency Conversion Efficiency» al identificar pares de divisas diferentes. Por qué es importante Determina las reglas de enrutamiento, los plazos límite y los corredores de liquidación. Dónde obtenerlo Campo 32A de SWIFT MT (divisa) o elemento Ccy de ISO 20022. Ejemplos USDEURGBPJPY
	Es STP IsStp	Indicador que señala si el pago no requirió intervención manual.
Descripción Indicador booleano cuyo valor es True cuando el caso no contiene actividades «Error Identified», «Payment Rejected» ni «Modification» manuales. Calcula directamente el KPI «Straight-Through Processing Rate». Es la métrica principal del éxito de la automatización en el procesamiento de pagos. Por qué es importante La medida principal de la eficiencia del proceso y del estado de la automatización. Dónde obtenerlo Se calcula a partir de la ausencia de actividades negativas o manuales específicas en el caso. Ejemplos truefalse
	Fecha valor ValueDate	La fecha en la que los fondos deben estar disponibles para el beneficiario.
Descripción Representa la fecha de liquidación indicada en el mensaje de pago. Comparar esta fecha con la marca de tiempo real de «Payment Settled» ayuda en el Dashboard «Settlement and Reconciliation Aging». Indica la urgencia del pago y se utiliza para medir el cumplimiento de los acuerdos de nivel de servicio relativos a la disponibilidad de los fondos. Por qué es importante Esencial para gestionar la liquidez y medir la puntualidad de la liquidación. Dónde obtenerlo SWIFT MT Field 32A (subcampo Date) o ISO 20022 IntrBkSttlmDt. Ejemplos 2023-10-262023-11-01
	Importe de la transacción TransactionAmount	El valor monetario de la instrucción de pago.
Descripción Representa el importe principal que se transfiere. Estos datos se extraen de campos SWIFT específicos, por ejemplo, el campo 32A de MT103. Es esencial para el Dashboard «High Value Transfer Approval Cycles», ya que permite segmentar los pagos por importe y analizar los retrasos de aprobación de las transacciones de alto riesgo y alto valor. Por qué es importante Permite analizar el impacto financiero y segmentar por tamaño de transacción. Dónde obtenerlo Campo 32A de SWIFT MT (importe) o elemento IntrBkSttlmAmt de ISO 20022. Ejemplos 15000.001250.501000000.0045.00
	Tipo de mensaje SWIFT SwiftMessageType	El tipo de mensaje SWIFT utilizado, por ejemplo, MT103 o pacs.008.
Descripción Indica el formato específico de la instrucción de pago. Entre los tipos habituales se incluyen MT103 para transferencias de clientes y equivalentes de ISO 20022, como pacs.008. Se utiliza en «Payment Journey Variant Analysis» para comparar la eficiencia del procesamiento de los formatos MT heredados con la de los nuevos estándares ISO. Por qué es importante Diferencia los flujos de pago y los estándares de procesamiento (heredado frente a ISO 20022). Dónde obtenerlo SWIFT Block 2 (Application Header), campo Message Type. Ejemplos MT103MT202pacs.008MT101
	UETR UniqueEndToEndReference	La Unique End-to-end Transaction Reference para el seguimiento en SWIFT gpi.
Descripción El UETR es una cadena de 36 caracteres que proporciona una referencia única e inmutable para un pago en toda la red SWIFT. A diferencia de los identificadores internos, el UETR se mantiene entre bancos. Este Atributo es fundamental para obtener visibilidad de principio a fin y correlacionar los registros internos con las actualizaciones externas de estado de SWIFT gpi. Por qué es importante El estándar de referencia para realizar el seguimiento de pagos transfronterizos entre distintas entidades. Dónde obtenerlo Bloque 3 de SWIFT, campo 121. Ejemplos b8c3f4a0-5d2a-4e1b-9c3d-1a2b3c4d5e6f123e4567-e89b-12d3-a456-426614174000
	Usuario de procesamiento ProcessingUser	El usuario o agente del sistema que realizó la actividad.
Descripción Identifica a la persona o al bot automatizado responsable de la actividad, como un responsable de cumplimiento que aprueba una coincidencia de sanciones o un operador que corrige un error de formato. Este atributo alimenta el panel «Validation Error Heatmap», lo que permite a los responsables identificar necesidades de formación o usuarios concretos asociados con altas tasas de retrabajo. Por qué es importante Permite analizar los recursos e identificar cuellos de botella manuales. Dónde obtenerlo Registros de auditoría del sistema, columna «User ID» en las tablas de transacciones. Ejemplos SYSTEMJ.DoeCompliance_Bot_01M.Smith
	Cumplió la hora límite MetCutOffTime	Indicador que señala si la instrucción se envió antes de la hora límite de la red.
Descripción Indicador booleano calculado al comparar la hora de «Payment Instruction Sent» con la hora límite de la red SWIFT específica para cada divisa. Permite calcular el KPI «SWIFT Cut-off Adherence Rate». No cumplir las horas límite provoca retrasos en la liquidación y afecta a las posiciones de liquidez. Por qué es importante KPI operativo crucial para la gestión de tesorería y liquidez. Dónde obtenerlo Se calcula comparando la marca de tiempo con una tabla de referencia estática de horas límite. Ejemplos truefalse
	Es transfronterizo IsCrossBorder	Indicador que señala si el pago involucra a distintos países.
Descripción Atributo booleano cuyo valor es True cuando el país del ordenante difiere del país del beneficiario. Permite calcular el KPI «Multi-Currency Handling Latency» al separar los flujos nacionales de los internacionales. Los pagos transfronterizos suelen tener mayor complejidad, costes y tiempos de procesamiento. Por qué es importante Segmentación fundamental para analizar la complejidad de los pagos. Dónde obtenerlo Comparación del código de país del Sender BIC con el código de país del BIC del beneficiario. Ejemplos truefalse
	Motivo del rechazo RejectionReason	Descripción textual que explica por qué se rechazó un pago.
Descripción Contiene la descripción narrativa o del código cuando falla un pago, normalmente en SWIFT Field 72 (Sender to Receiver Information) o en mensajes de devolución. Permite realizar el análisis «Beneficiary Rejection Root Cause». Permite a los analistas aplicar minería de textos para encontrar temas comunes en los rechazos, como «Invalid Account» o «Beneficiary Name Mismatch». Por qué es importante Proporciona contexto cualitativo sobre los fallos del proceso. Dónde obtenerlo SWIFT Field 72 o 79 en mensajes de devolución (MT103 Return). Ejemplos Cuenta del beneficiario cerradaIBAN no válidoFallo de cumplimiento normativoIdentificador bancario desconocido
	País de origen OriginatingCountry	El código de país de la entidad que inicia el pago.
Descripción Indica la jurisdicción en la que se originó la solicitud de pago. Es fundamental para el Dashboard «Sanctions Screening Lead Time», ya que los pagos procedentes de jurisdicciones de alto riesgo suelen someterse a controles de cumplimiento más rigurosos y prolongados. Permite analizar geográficamente los volúmenes de pagos y los retrasos de procesamiento. Por qué es importante Dimensión clave para analizar el riesgo de cumplimiento y el rendimiento regional. Dónde obtenerlo Se obtiene del Sender BIC (caracteres 5 y 6 del código de país). Ejemplos USGBDEFR
	Prioridad de la instrucción InstructionPriority	La marca de prioridad que indica la urgencia del pago.
Descripción Se obtiene de las cabeceras del mensaje, por ejemplo, «Normal» frente a «Urgent». Permite segmentar los datos en el Dashboard «High Value Transfer Approval Cycles», ya que los pagos urgentes suelen requerir flujos de aprobación acelerados. Comprender la distribución de prioridades ayuda a planificar los recursos en los periodos de mayor actividad. Por qué es importante Ayuda a distinguir los requisitos de SLA entre pagos estándar y pagos acelerados. Dónde obtenerlo SWIFT Block 2, campo Message Priority, por ejemplo, N para Normal y U para Urgent. Ejemplos NormalUrgenteSistema

Obligatorio Recomendado Opcional

Actividades del procesamiento de pagos

Estos son los pasos clave y los hitos del proceso que debe capturar en su registro de eventos para descubrir el proceso con precisión e identificar cuellos de botella.

6 Recomendado 6 Opcional

	Actividad	Descripción
	Filtrado de sanciones superado	La marca de tiempo en la que la instrucción de pago supera correctamente el proceso de filtrado de las listas OFAC y de sanciones. Se registra explícitamente a partir de los registros del sistema de cumplimiento o se infiere cuando se libera el estado «Screening Hold».
Por qué es importante Es fundamental para el Dashboard «Sanctions Screening Lead Time». Los retrasos en esta etapa representan cuellos de botella de cumplimiento, no ineficiencias operativas. Dónde obtenerlo Registros del Compliance Filtering System o indicadores de estado de cumplimiento del Payment Hub. Recopilar Registrado cuando se actualiza el estado del filtrado Tipo de evento explicit
	Instrucción de pago enviada	Transmisión del mensaje formateado, MT103 o ISO 20022 pacs.008, al SWIFT Gateway. Se registra explícitamente cuando se genera la carga útil del mensaje y se entrega a la interfaz de red.
Por qué es importante Se utiliza para calcular la «SWIFT Cut-off Adherence Rate». Marca la transición del procesamiento interno al procesamiento en la red. Dónde obtenerlo Registros de SWIFT Alliance Access (SAA) o registros de transmisión del Gateway. Recopilar Registrado cuando se genera el mensaje de salida Tipo de evento explicit
	Pago aprobado	Acción de autorización final realizada por un responsable designado o una regla automatizada para transferencias de alto valor. Se registra explícitamente cuando el indicador del Workflow de aprobación se establece en true.
Por qué es importante Es clave para el Dashboard «High Value Transfer Approval Cycles». Identifica retrasos en las aprobaciones manuales de grandes movimientos de liquidez. Dónde obtenerlo Registro de auditoría del motor del Workflow de aprobación. Recopilar Registrado cuando se ejecuta la acción de aprobación Tipo de evento explicit
	Pago conciliado	Conciliación de la transacción de pago con el extracto del libro contable de la cuenta Nostro/Vostro. Se infiere cuando el sistema de conciliación vincula el ID de la transacción con una línea del extracto.
Por qué es importante Cierra el ciclo financiero. Los retrasos en esta etapa afectan a la visibilidad del efectivo y a la auditabilidad. Dónde obtenerlo Registros del sistema de conciliación o estado «Matched» en el Core Banking System. Recopilar Comparar el estado de conciliación Tipo de evento inferred
	Pago liquidado	Confirmación de que los fondos se han abonado en la cuenta del beneficiario, normalmente el estado gpi ACSC. Se registra explícitamente mediante actualizaciones del Tracker o mensajes de confirmación.
Por qué es importante Representa el final funcional correcto de la cadena de pago. Es clave para calcular la «Settlement to Reconciliation Gap». Dónde obtenerlo SWIFT gpi Tracker, estado ACSC, o confirmaciones MT900/910. Recopilar Registrado al confirmar la liquidación Tipo de evento explicit
	Solicitud de pago creada	Generación inicial de la instrucción de pago dentro del sistema bancario interno o ERP. Se registra explícitamente cuando se asigna por primera vez una referencia de transacción única, TRN o UETR, a la orden de pago.
Por qué es importante Establece la hora de inicio de todo el ciclo de vida del pago. Es esencial para calcular los tiempos de procesamiento de principio a fin e identificar retrasos previos a la intervención de la red SWIFT. Dónde obtenerlo Marca de tiempo de creación de la tabla de transacciones en el Core Banking System (CBS) o Payment Hub. Recopilar Registrado al crear el registro de la transacción Tipo de evento explicit
	ACK de SWIFT recibido	Recepción de un acuse técnico, ACK, de la red SWIFT que confirma que el mensaje ha sido aceptado para su procesamiento. Se registra explícitamente a partir de los registros de la interfaz de red.
Por qué es importante Confirma que el mensaje ha entrado correctamente en la red global. Permite diferenciar los fallos internos de la propagación real en la red. Dónde obtenerlo Registros del SWIFT Gateway, buscando señales ACK y NACK. Recopilar Registrado al recibir el ACK de la red Tipo de evento explicit
	Datos del pago validados	Finalización correcta de las comprobaciones de sintaxis y formato, por ejemplo, del formato IBAN y de la validez del BIC, en la instrucción de pago. A menudo se infiere cuando el estado de la transacción cambia de «Draft» a «Validated» o «Ready for Auth».
Por qué es importante Las tasas elevadas de fallo en esta etapa indican una mala calidad de los datos en origen. Mide el KPI «Validation First-Pass Yield». Dónde obtenerlo Registros de estado del Payment Engine o tablas históricas de validación. Recopilar Comparar el campo de estado antes y después Tipo de evento inferred
	Error de pago identificado	Evento en el que una transacción se marca para reparación debido a fallos de validación, NACK o mensajes de rechazo. Se infiere cuando la transacción entra en una cola «Repair», «Correction» o «Exception».
Por qué es importante Realiza el seguimiento de la «Error Resolution Rework Rate». Una frecuencia elevada en esta etapa perjudica los índices de Straight-Through Processing (STP). Dónde obtenerlo Columnas de estado del Payment Hub o registros de colas de excepciones. Recopilar Comparar el campo de estado con Error/Repair Tipo de evento inferred
	Error de pago resuelto	Modificación y reenvío correctos de una transacción que anteriormente se encontraba en estado de error. Se infiere cuando una transacción pasa de una cola «Repair» a un estado «Processing» o «Ready».
Por qué es importante Mide la eficiencia del equipo de retrabajo manual. Las duraciones prolongadas en esta etapa indican carencias de formación o códigos de error complejos. Dónde obtenerlo Registros de estado del Payment Hub que muestran la salida de las colas de excepciones. Recopilar Comparar la salida del campo de estado de Repair Tipo de evento inferred
	Pago rechazado	Recepción de un mensaje de rechazo, como una devolución MT103 o el estado gpi RJCT, procedente del banco beneficiario o intermediario. Se registra explícitamente a partir de los mensajes SWIFT entrantes.
Por qué es importante Es fundamental para el Dashboard «Beneficiary Rejection Root Cause». Identifica bloqueos externos, como cuentas cerradas o datos de enrutamiento incorrectos. Dónde obtenerlo Mensajes SWIFT entrantes, MT103 RET o pacs.004, o estado gpi RJCT. Recopilar Registrado al recibir el mensaje de rechazo Tipo de evento explicit
	Pago transferido	Actualización de estado intermedia, a menudo el estado gpi ACSP, que indica que un banco intermediario está procesando el pago. Se registra mediante actualizaciones del SWIFT gpi Tracker o mensajes de estado.
Por qué es importante Proporciona visibilidad sobre la «caja negra» de la banca corresponsal. Es esencial para analizar el recorrido total de principio a fin. Dónde obtenerlo Fuentes de datos del SWIFT gpi Tracker o mensajes MT199/trck. Recopilar Registrado al recibir la actualización del Tracker Tipo de evento explicit

Recomendado Opcional

Guías de extracción

Cómo obtener sus datos de SWIFT

Pasos

Inicie sesión en SWIFT Alliance WebPlatform o en la interfaz gráfica de Alliance Access con una cuenta de administrador que tenga acceso a la configuración de archivado.
Vaya al menú System Management, seleccione Archiving y, a continuación, elija Automated Archiving Configuration.
Configure un nuevo perfil de Daily Export. Establezca Target Directory en una ruta de red segura a la que puedan acceder sus scripts de extracción.
Especifique Content Scope: seleccione «Messages» y asegúrese de marcar «Include History» e «Include Interventions». Esto es fundamental para capturar eventos de auditoría, como aprobaciones y validaciones.
Seleccione Output Format. Elija XML, concretamente «SAA XML v2» o el esquema más reciente disponible, ya que conserva mejor que CSV la relación jerárquica entre la cabecera del mensaje y los eventos históricos.
Configure los filtros de Date Range para extraer el periodo histórico específico que requiere la iniciativa de Process Mining o para ejecutar la extracción diariamente con datos T-1.
Ejecute manualmente el trabajo Archive Export para la carga inicial o espere a que finalice el trabajo por lotes programado. Compruebe que los archivos .xml se hayan generado en el directorio de destino.
Como la exportación SAA es un archivo XML jerárquico y ProcessMind requiere un registro de eventos CSV plano, debe ejecutar un script de transformación. Utilice el script de Python proporcionado en la sección Query para analizar el XML.
El script recorre cada elemento de mensaje. Extrae la Unique End-to-end Transaction Reference (UETR) o el Transaction Reference Number (TRN) para utilizarlo como Case ID.
A continuación, el script recorre los nodos secundarios «History» e «Intervention» de cada mensaje para generar filas de actividad. Asigna los códigos de estado internos de SAA, como «NetworkAck» y «Authorisation», a los nombres de actividad estandarizados.
Ejecute el script de Python con sus archivos XML exportados para generar el archivo plano «events.csv».
Valide el archivo CSV resultante para asegurarse de que columnas como EventTimestamp estén en formato ISO 8601 y, después, cargue el archivo en ProcessMind.

Configuración

Formato del archivo: XML (preferido) o CSV (archivo plano). Se recomienda encarecidamente XML para capturar los nodos de historial anidados.
Incluir texto del mensaje: Sí (obligatorio para analizar el importe, la divisa y los datos del beneficiario).
Incluir historial/intervenciones: Sí (obligatorio para ver aprobaciones, validaciones y errores).
Intervalo de fechas: configure una ventana móvil, por ejemplo, los últimos 30 días, para cargas incrementales, o fechas fijas para el historial inicial.
Tipos de mensaje: filtre por MT103, MT202, pacs.008, pacs.009 y pacs.004/002 (para devoluciones/estados).
Direcciones: incluya tanto Input (emisión) como Output (recepción) para capturar rechazos o confirmaciones entrantes.
Seguridad: asegúrese de que la carpeta de exportación tenga permisos estrictos de lectura y escritura, ya que estos archivos contienen datos de pagos reales.

a Consulta de ejemplo python

import xml.etree.ElementTree as ET
import pandas as pd
import glob
import os

class=class="hl-string">"hl-comment"># CONFIGURATION
class=class="hl-string">"hl-comment"># Update this path to your SAA XML Export directory
input_folder = class="hl-string">"./saa_exports/"
output_file = class="hl-string">"./swift_process_mining_log.csv"

columns = [
    class="hl-string">"PaymentTransactionId", class="hl-string">"ActivityName", class="hl-string">"EventTimestamp", class="hl-string">"SourceSystem", 
    class="hl-string">"LastDataUpdate", class="hl-string">"TransactionAmount", class="hl-string">"TransactionCurrency", class="hl-string">"UniqueEndToEndReference",
    class="hl-string">"SwiftMessageType", class="hl-string">"ProcessingUser", class="hl-string">"BeneficiaryBic", class="hl-string">"SwiftNakCode", class="hl-string">"ValueDate", class="hl-string">"IsStp"
]

data_rows = []

class=class="hl-string">"hl-comment"># Helper to determine Activity Name based on SAA History Events
def map_activity(history_item_name, status, nature, text_content):
    class=class="hl-string">"hl-comment"># 1. Payment Request Created
    if history_item_name == class="hl-string">"Creation" or history_item_name == class="hl-string">"Reception":
        return class="hl-string">"Payment Request Created"
    
    class=class="hl-string">"hl-comment"># 2. Payment Details Validated
    if class="hl-string">"Validation" in history_item_name and status == class="hl-string">"Success":
        return class="hl-string">"Payment Details Validated"
    
    class=class="hl-string">"hl-comment"># 3. Sanctions Screening Passed
    if class="hl-string">"Filter" in history_item_name and status == class="hl-string">"Passed":
        return class="hl-string">"Sanctions Screening Passed"
    
    class=class="hl-string">"hl-comment"># 4. Payment Approved
    if history_item_name == class="hl-string">"Intervention" and nature == class="hl-string">"Authorization":
        return class="hl-string">"Payment Approved"
    
    class=class="hl-string">"hl-comment"># 5. Payment Instruction Sent
    if history_item_name == class="hl-string">"Transmission" or status == class="hl-string">"Transmitted":
        return class="hl-string">"Payment Instruction Sent"
    
    class=class="hl-string">"hl-comment"># 6. SWIFT ACK Received
    if history_item_name == class="hl-string">"NetworkReport" and class="hl-string">"ACK" in text_content:
        return class="hl-string">"SWIFT ACK Received"

    class=class="hl-string">"hl-comment"># 7. Payment Transferred (gpi status)
    if class="hl-string">"TrackerUpdate" in history_item_name and class="hl-string">"ACSP" in text_content:
        return class="hl-string">"Payment Transferred"

    class=class="hl-string">"hl-comment"># 8. Payment Error Identified
    if nature == class="hl-string">"NAK" or status == class="hl-string">"Failed" or nature == class="hl-string">"RoutingFailure":
        return class="hl-string">"Payment Error Identified"

    class=class="hl-string">"hl-comment"># 9. Payment Error Resolved
    if history_item_name == class="hl-string">"Modification" and nature == class="hl-string">"Repair":
        return class="hl-string">"Payment Error Resolved"
    
    class=class="hl-string">"hl-comment"># 10. Payment Rejected
    if class="hl-string">"TrackerUpdate" in history_item_name and class="hl-string">"RJCT" in text_content:
        return class="hl-string">"Payment Rejected"

    class=class="hl-string">"hl-comment"># 11. Payment Settled
    if class="hl-string">"TrackerUpdate" in history_item_name and class="hl-string">"ACSC" in text_content:
        return class="hl-string">"Payment Settled"

    class=class="hl-string">"hl-comment"># 12. Payment Reconciled
    class=class="hl-string">"hl-comment"># Often external, but if SAA receives a Reconciliation update:
    if history_item_name == class="hl-string">"ReconciliationMatch":
        return class="hl-string">"Payment Reconciled"
        
    return None

class=class="hl-string">"hl-comment"># PROCESS XML FILES
for xml_file in glob.glob(os.path.join(input_folder, class="hl-string">"*.xml")):
    tree = ET.parse(xml_file)
    root = tree.getroot()

    class=class="hl-string">"hl-comment"># Iterate through Messages
    for message in root.findall(class="hl-string">".//Message"):
        class=class="hl-string">"hl-comment"># Header Extraction
        trn = message.findtext(class="hl-string">"TRN")
        uetr = message.findtext(class="hl-string">"UETR")
        case_id = uetr if uetr else trn
        
        class=class="hl-string">"hl-comment"># Skip if no identifier found
        if not case_id: 
            continue

        msg_type = message.findtext(class="hl-string">"Type")
        amount = message.findtext(class="hl-string">"Amount")
        currency = message.findtext(class="hl-string">"Currency")
        val_date = message.findtext(class="hl-string">"ValueDate")
        ben_bic = message.findtext(class="hl-string">"BeneficiaryBIC")
        
        class=class="hl-string">"hl-comment"># Loop through History/Audit Trail for Activities
        for history in message.findall(class="hl-string">".//History/Item"):
            evt_name = history.findtext(class="hl-string">"Name")
            evt_status = history.findtext(class="hl-string">"Status")
            evt_nature = history.findtext(class="hl-string">"Nature")
            evt_text = history.findtext(class="hl-string">"Text") or class="hl-string">""
            evt_user = history.findtext(class="hl-string">"Operator")
            evt_time = history.findtext(class="hl-string">"Time") class=class="hl-string">"hl-comment"># format: YYYYMMDDHHMMSS
            
            activity = map_activity(evt_name, evt_status, evt_nature, evt_text)
            
            if activity:
                class=class="hl-string">"hl-comment"># Determine NAK Code if applicable
                nak_code = None
                if activity == class="hl-string">"Payment Error Identified" and class="hl-string">"NAK" in evt_text:
                    nak_code = evt_text.split(class="hl-string">"Code:")[-1].strip()
                
                class=class="hl-string">"hl-comment"># Determine STP
                is_stp = class="hl-string">"True" if evt_user == class="hl-string">"System" or evt_user == class="hl-string">"Auto" else class="hl-string">"False"

                data_rows.append({
                    class="hl-string">"PaymentTransactionId": case_id,
                    class="hl-string">"ActivityName": activity,
                    class="hl-string">"EventTimestamp": pd.to_datetime(evt_time, format=class="hl-string">'%Y%m%d%H%M%S'),
                    class="hl-string">"SourceSystem": class="hl-string">"SWIFT Alliance Access",
                    class="hl-string">"LastDataUpdate": pd.Timestamp.now(),
                    class="hl-string">"TransactionAmount": amount,
                    class="hl-string">"TransactionCurrency": currency,
                    class="hl-string">"UniqueEndToEndReference": uetr,
                    class="hl-string">"SwiftMessageType": msg_type,
                    class="hl-string">"ProcessingUser": evt_user,
                    class="hl-string">"BeneficiaryBic": ben_bic,
                    class="hl-string">"SwiftNakCode": nak_code,
                    class="hl-string">"ValueDate": val_date,
                    class="hl-string">"IsStp": is_stp
                })

class=class="hl-string">"hl-comment"># OUTPUT GENERATION
df = pd.DataFrame(data_rows, columns=columns)

class=class="hl-string">"hl-comment"># Remove duplicate timestamp/activity pairs for cleaner log
df.drop_duplicates(subset=[class="hl-string">"PaymentTransactionId", class="hl-string">"ActivityName", class="hl-string">"EventTimestamp"], inplace=True)

df.to_csv(output_file, index=False, quoting=1)
print(fclass="hl-string">"Extraction Complete. Processed {len(df)} events into {output_file}")

Pasos

Prepare el entorno: asegúrese de que Python 3.8 o posterior esté instalado en el servidor de extracción. Instale las bibliotecas necesarias mediante pip (pandas, numpy). Cree un directorio de trabajo específico con una subcarpeta llamada raw_swift_files y coloque allí todos los archivos diarios .out, .txt o .fin.
Comprenda la estructura de los archivos: los Messages SWIFT FIN sin procesar tienen una estructura basada en bloques ({1: Header} {2: Application} {4: Body}). Identifique los delimitadores. Los archivos estándar utilizan llaves. Asegúrese de que sus archivos no estén incluidos en envoltorios propietarios, como encabezados RJE, o elimínelos antes del procesamiento.
Configure las rutas del script: abra el script de Python proporcionado. Actualice la variable INPUT_DIRECTORY para que apunte a la carpeta raw_swift_files. Actualice la ruta de OUTPUT_FILE para el CSV resultante.
Defina los patrones de expresiones regulares: el script utiliza expresiones regulares para aislar los bloques. El bloque 1 contiene el BIC del remitente y el número de secuencia. El bloque 2 contiene el tipo de Message, por ejemplo, I103 u O103, y el BIC del receptor. El bloque 4 contiene las etiquetas de datos empresariales, por ejemplo, :20: TRN y :32A: Value/Amount.
Asigne los tipos de Message a las actividades: el script contiene un diccionario de asignaciones. MT103 implica «Payment Instruction Sent». MT012 (System ACK) implica «SWIFT ACK Received». MT900/910 implica «Payment Settled». MT199/299 con códigos de estado específicos, como ACSP o RJCT, se asigna a «Payment Transferred» o «Payment Rejected».
Extraiga los atributos: el analizador recorre etiquetas específicas. :20: se asigna a Payment Transaction ID. :121: se asigna a UETR. :32A: se analiza para obtener Date, Currency y Amount.
Gestione los estados internos: para actividades como «Payment Request Created» o «Payment Approved», que suelen ser eventos internos del ERP y no aparecen en los archivos SWIFT, el script los deriva de la marca de tiempo de creación del Message en el bloque 1 si no hay registros explícitos, o busca códigos de texto específicos en el campo :72: (Sender to Receiver Info) si su organización registra allí las aprobaciones.
Ejecute la extracción: ejecute el script desde la línea de comandos: python swift_parser.py. El script recorre todos los archivos, analiza los bloques SWIFT válidos y omite las líneas con formato incorrecto.
Valide después del procesamiento: abra el CSV generado. Compruebe el formato de EventTimestamp, para el que se recomienda ISO 8601. Asegúrese de que PaymentTransactionId esté informado en todas las filas para permitir la agrupación de casos.
Transforme los datos: si ValueDate aparece como YYMMDD, el script lo convierte a YYYY-MM-DD. Asegúrese de estandarizar los separadores decimales de TransactionAmount, ya sean puntos o comas.
Realice la revisión final: filtre por un UETR específico para ver el flujo completo, desde Instruction Sent hasta ACK y Settlement. Verifique que los NAK (MT019) aparezcan como «Payment Error Identified».
Cargue los datos en ProcessMind: importe el CSV. Asigne PaymentTransactionId a Case ID, ActivityName a Activity y EventTimestamp a Timestamp.

Configuración

Input Directory: ruta a la carpeta que contiene los archivos planos FIN/MT sin procesar. Asegúrese de conceder permisos de lectura.
Análisis del formato de fecha: las fechas SWIFT sin procesar tienen el formato YYMMDD. El script gestiona la suposición del siglo, 20xx.
Filtros de tipo de Message: el script se centra en la categoría 1 (Payments), la categoría 9 (Statements) y los Messages de sistema habituales (012, 019). Ajuste la lista RELEVANT_MT_TYPES si necesita la categoría 2 (Financial Institution Transfers).
Codificación: el valor predeterminado es 'utf-8'. Cámbielo a 'latin-1' o 'ascii' si los archivos contienen caracteres de codificaciones antiguas.
Procesamiento por lotes: para conjuntos de datos de más de 1 GB, el script utiliza un enfoque basado en generadores para evitar desbordamientos de memoria y procesa los archivos uno por uno.
Ajuste de expresiones regulares: la expresión regular estricta presupone el formato FIN estándar {1:...}{2:...}{4:...}. Si los archivos utilizan $ u otros delimitadores, actualice la variable BLOCK_PATTERN.

a Consulta de ejemplo python

import os
import re
import glob
import pandas as pd
from datetime import datetime

class=class="hl-string">"hl-comment"># === CONFIGURATION ===
INPUT_DIRECTORY = class="hl-string">"./raw_swift_files/"
OUTPUT_FILE = class="hl-string">"./extracted_swift_event_log.csv"

class=class="hl-string">"hl-comment"># === REGEX PATTERNS ===
class=class="hl-string">"hl-comment"># Matches standard SWIFT block structure: {1:...}{2:...}{3:...}{4:...}
BLOCK_PATTERN = re.compile(rclass="hl-string">'(\{1:.*?\})(\{2:.*?\})(?:\{3:.*?\})?(\{4:.*?\})', re.DOTALL)

class=class="hl-string">"hl-comment"># Field extractors
TAG_20_TRN = re.compile(rclass="hl-string">':20:([^\r\n]+)')
TAG_121_UETR = re.compile(rclass="hl-string">':121:([a-fA-F0-9\-]+)')
TAG_32A_VAL = re.compile(rclass="hl-string">':32A:(\d{6})([A-Z]{3})([\d,]+)')
TAG_72_INFO = re.compile(rclass="hl-string">':72:([^\r\n]+)')
TAG_57A_BIC = re.compile(rclass="hl-string">':57[A-Z]?:(?:[/]{0,2}[A-Z0-9]+)?\r?\n?([A-Z0-9]{8,11})')

class=class="hl-string">"hl-comment"># Header parsers
BLOCK_1_DATE = re.compile(rclass="hl-string">'\{1:[F|A|L]01[A-Z0-9]{12}(\d{4})(\d{6})')
BLOCK_2_TYPE = re.compile(rclass="hl-string">'\{2:([IO])(\d{3})')

class=class="hl-string">"hl-comment"># === ACTIVITY MAPPING LOGIC ===
def map_message_to_activity(io_direction, mt_type, block4_text):
    class=class="hl-string">"hl-comment"># Outgoing Payments (MT103, MT202)
    if io_direction == class="hl-string">'I' and mt_type in [class="hl-string">'103', class="hl-string">'202']:
        return class="hl-string">"Payment Instruction Sent"
    
    class=class="hl-string">"hl-comment"># Incoming Payments (MT103 received from another bank)
    if io_direction == class="hl-string">'O' and mt_type in [class="hl-string">'103', class="hl-string">'202']:
        return class="hl-string">"Payment Transferred"

    class=class="hl-string">"hl-comment"># System Acknowledgment (MT012)
    if mt_type == class="hl-string">'012':
        return class="hl-string">"SWIFT ACK Received"

    class=class="hl-string">"hl-comment"># System NAK / Abort (MT019)
    if mt_type == class="hl-string">'019':
        return class="hl-string">"Payment Error Identified"

    class=class="hl-string">"hl-comment"># Statements / Confirmation (MT900 Debit, MT910 Credit, MT950)
    if mt_type in [class="hl-string">'900', class="hl-string">'910', class="hl-string">'910', class="hl-string">'950']:
        return class="hl-string">"Payment Settled"
    
    class=class="hl-string">"hl-comment"># Returns (MT103 RETN)
    if mt_type == class="hl-string">'103' and class="hl-string">':23E:RETN' in block4_text:
        return class="hl-string">"Payment Rejected"

    class=class="hl-string">"hl-comment"># GPI Status Updates (MT199)
    if mt_type == class="hl-string">'199':
        if class="hl-string">'/RJCT/' in block4_text:
            return class="hl-string">"Payment Rejected"
        if class="hl-string">'/ACSP/' in block4_text:
            return class="hl-string">"Payment Transferred"
        if class="hl-string">'/ACSC/' in block4_text:
            return class="hl-string">"Payment Settled"
            
    return class="hl-string">"Message Processed"  class=class="hl-string">"hl-comment"># Fallback

def parse_swift_date(yymmdd):
    class=class="hl-string">"hl-comment"># Assumes 20xx for years
    try:
        return datetime.strptime(yymmdd, class="hl-string">'%y%m%d').strftime(class="hl-string">'%Y-%m-%d')
    except ValueError:
        return None

class=class="hl-string">"hl-comment"># === MAIN EXTRACTION ===
def process_files():
    records = []
    files = glob.glob(os.path.join(INPUT_DIRECTORY, class="hl-string">"*"))
    
    print(fclass="hl-string">"Found {len(files)} files to process.")

    for filepath in files:
        try:
            with open(filepath, class="hl-string">'r', encoding=class="hl-string">'utf-8', errors=class="hl-string">'ignore') as f:
                content = f.read()
                
            class=class="hl-string">"hl-comment"># Find all full messages in file
            messages = BLOCK_PATTERN.findall(content)
            
            for b1, b2, b4 in messages:
                class=class="hl-string">"hl-comment"># -- Extract Core IDs --
                trn_match = TAG_20_TRN.search(b4)
                uetr_match = TAG_121_UETR.search(b4)
                
                class=class="hl-string">"hl-comment"># Primary Key preference: UETR > TRN > Generated ID
                uetr = uetr_match.group(1) if uetr_match else None
                trn = trn_match.group(1) if trn_match else class="hl-string">"UNKNOWN"
                case_id = uetr if uetr else trn
                
                if case_id == class="hl-string">"UNKNOWN":
                    continue class=class="hl-string">"hl-comment"># Skip structural noise

                class=class="hl-string">"hl-comment"># -- Extract Metadata --
                type_match = BLOCK_2_TYPE.search(b2)
                io_dir = type_match.group(1) if type_match else class="hl-string">"I"
                mt_type = type_match.group(2) if type_match else class="hl-string">"999"
                
                class=class="hl-string">"hl-comment"># -- Determine Timestamp --
                class=class="hl-string">"hl-comment"># Attempt to find timestamp in Block 1 (Session header)
                class=class="hl-string">"hl-comment"># Format often: {1:F01BANKBEBBXXXX1234123456}
                class=class="hl-string">"hl-comment"># Using current time as fallback if header parsing fails
                timestamp = datetime.now().isoformat()
                date_match = BLOCK_1_DATE.search(b1)
                if date_match:
                    class=class="hl-string">"hl-comment"># YYMMDD + HHMM
                    class=class="hl-string">"hl-comment"># Note: Sequence number often occupies this space, this regex is generic
                    class=class="hl-string">"hl-comment"># For simplicity, we assume file creation time or construct from header
                    pass 
                    
                class=class="hl-string">"hl-comment"># -- Extract Financials --
                amount = None
                currency = None
                val_date = None
                amt_match = TAG_32A_VAL.search(b4)
                if amt_match:
                    val_date_raw, currency, amt_str = amt_match.groups()
                    val_date = parse_swift_date(val_date_raw)
                    amount = float(amt_str.replace(class="hl-string">',', class="hl-string">'.'))

                class=class="hl-string">"hl-comment"># -- Extract Beneficiary --
                ben_bic = None
                ben_match = TAG_57A_BIC.search(b4)
                if ben_match:
                    ben_bic = ben_match.group(1)

                class=class="hl-string">"hl-comment"># -- Map Activity --
                activity = map_message_to_activity(io_dir, mt_type, b4)

                class=class="hl-string">"hl-comment"># -- Create Record --
                record = {
                    class="hl-string">"PaymentTransactionId": case_id,
                    class="hl-string">"ActivityName": activity,
                    class="hl-string">"EventTimestamp": datetime.now().strftime(class="hl-string">"%Y-%m-%d %H:%M:%S"), class=class="hl-string">"hl-comment"># Replace with actual extracted time
                    class="hl-string">"SourceSystem": class="hl-string">"SWIFT_Alliance_Access",
                    class="hl-string">"TransactionAmount": amount,
                    class="hl-string">"TransactionCurrency": currency,
                    class="hl-string">"UniqueEndToEndReference": uetr,
                    class="hl-string">"SwiftMessageType": fclass="hl-string">"MT{mt_type}",
                    class="hl-string">"BeneficiaryBic": ben_bic,
                    class="hl-string">"ValueDate": val_date,
                    class="hl-string">"LastDataUpdate": datetime.now().isoformat()
                }
                
                records.append(record)

                class=class="hl-string">"hl-comment"># -- Derived Logic for class="hl-string">'Created' --
                class=class="hl-string">"hl-comment"># If this is the very first outgoing MT103, we can imply creation
                class=class="hl-string">"hl-comment"># In a real scenario, this logic happens after sorting all events
                if activity == class="hl-string">"Payment Instruction Sent":
                    creation_record = record.copy()
                    creation_record[class="hl-string">"ActivityName"] = class="hl-string">"Payment Request Created"
                    class=class="hl-string">"hl-comment"># Offset time slightly backwards to simulate flow
                    creation_record[class="hl-string">"EventTimestamp"] = class="hl-string">"[Derive from Log Creation Time]"
                    records.append(creation_record)

        except Exception as e:
            print(fclass="hl-string">"Error processing file {filepath}: {e}")

    class=class="hl-string">"hl-comment"># -- Convert to DataFrame --
    df = pd.DataFrame(records)
    
    class=class="hl-string">"hl-comment"># -- Final Cleanup --
    class=class="hl-string">"hl-comment"># Sort by ID and Time to ensure logical order if needed
    if not df.empty:
        df.sort_values(by=[class="hl-string">'PaymentTransactionId', class="hl-string">'EventTimestamp'], inplace=True)
        df.to_csv(OUTPUT_FILE, index=False)
        print(fclass="hl-string">"Extraction complete. Data saved to {OUTPUT_FILE}")
    else:
        print(class="hl-string">"No valid SWIFT messages found.")

if __name__ == class="hl-string">"__main__":
    process_files()

Pasos

Confirme que estén disponibles su suscripción a la API de SWIFT gpi Tracker, las credenciales de API, la capacidad de búsqueda por UETR y el acceso a la fuente de eventos del sistema interno de pagos. Obtenga en la documentación de su servicio SWIFT la URL base exacta de la API, el método de autenticación, el esquema de respuesta, los límites de solicitudes y las reglas de paginación. No dé por supuesta una ruta de endpoint, ya que depende del servicio SWIFT contratado y de su implementación.
Configure el script con la cadena de conexión, las credenciales, el intervalo de fechas y los filtros específicos de su empresa. Para la extracción inicial, utilice un intervalo de tres a seis meses. Limite los datos de la fuente interna a las transacciones de pago que tengan un TRN o UETR y conserve el nombre del sistema de origen para cada evento.
Extraiga los eventos internos del ciclo de vida para Payment Request Created, Payment Details Validated, Sanctions Screening Passed, Payment Approved, Payment Instruction Sent, SWIFT ACK Received, Payment Error Identified, Payment Error Resolved, Payment Rejected, Payment Settled y Payment Reconciled. Cada evento de origen debe convertirse en una fila independiente del registro de eventos. No dependa de ProcessMind para inferir actividades.
Extraiga los UETR de los registros internos de pagos y consulte la API con licencia de SWIFT gpi Tracker para cada UETR. Utilice la documentación de la API proporcionada para su entorno a fin de configurar la URL de solicitud, los encabezados de autenticación, los parámetros, la paginación y los campos de respuesta. Almacene la respuesta sin procesar para garantizar la auditabilidad.
Convierta los estados aplicables de gpi tracker en filas de eventos explícitas. El script asigna ACSP a Payment Transferred, RJCT a Payment Rejected y ACSC a Payment Settled. Si la respuesta contratada utiliza nombres de estado o marcas de tiempo diferentes, actualice la asignación según el esquema de respuesta documentado. No cree un evento cuando la API no proporcione una ocurrencia de estado correspondiente.
Normalice todas las marcas de tiempo a una zona horaria coherente, preferiblemente UTC, y conserve la marca de tiempo original cuando sea necesario para fines de auditoría. Convierta los valores monetarios a valores numéricos, conserve la moneda de la transacción, normalice los valores booleanos de STP y mantenga los valores NULL cuando la fuente no proporcione un atributo.
Elimine los duplicados utilizando PaymentTransactionId, ActivityName, EventTimestamp y el identificador de evento de origen más específico disponible. Conserve las ocurrencias legítimas múltiples, como ciclos repetidos de reparación y reenvío, salvo que la regla de negocio exija explícitamente una sola ocurrencia por actividad.
Valide que cada fila de salida contenga PaymentTransactionId, ActivityName, EventTimestamp, SourceSystem y LastDataUpdate. Confirme que ActivityName solo contenga los valores de actividad documentados y que las 12 actividades estén representadas cuando aparezcan en los datos de origen.
Exporte el conjunto de datos resultante como CSV UTF-8 u otro formato de registro de eventos compatible con ProcessMind. Utilice una fila por evento, incluya las columnas obligatorias y recomendadas, y asigne PaymentTransactionId como identificador del caso, ActivityName como actividad y EventTimestamp como marca de tiempo del evento. Cargue el archivo en ProcessMind sin esperar una derivación adicional de eventos.

Configuración

Endpoint y esquema de la API: configure la URL base exacta de la API de SWIFT gpi Tracker, la ruta del recurso, los parámetros de solicitud, los encabezados de autenticación, la paginación y los nombres de los campos de respuesta a partir de la documentación de SWIFT contratada para su entorno. El script no inventa deliberadamente ningún endpoint.
Credenciales y autorización: utilice el mecanismo aprobado de credenciales de la API de SWIFT, los requisitos de certificados o tokens, el método de almacenamiento de secretos y las autorizaciones del sistema de origen interno. No incluya secretos de producción directamente en el código fuente.
Intervalo de fechas: comience con tres a seis meses. Utilice la marca de tiempo del evento o la fecha de creación del pago que corresponda a su objetivo de reporting. Para volúmenes grandes, procese intervalos más pequeños y registre puntos de control para los UETR completados.
Filtros de la fuente interna: configure los filtros de empresa, entidad legal, sucursal, producto de pago, tipo de documento, tipo de Message, moneda y estado según su sistema. Utilice marcadores como [Configure based on your system] hasta confirmar el esquema de origen.
Asignación de actividades: mantenga asignaciones explícitas para los estados y las colas internas. Algunos ejemplos son Draft a Validated o Ready for Auth para Payment Details Validated, la liberación de Screening Hold para Sanctions Screening Passed, la entrada en la cola Repair o Exception para Payment Error Identified y el retorno a Processing o Ready for Payment para Payment Error Resolved.
Asignación de estados de gpi: asigne los estados documentados del tracker a actividades explícitas. Esta implementación asigna ACSP a Payment Transferred, RJCT a Payment Rejected y ACSC a Payment Settled. Amplíe la asignación únicamente cuando la documentación de la API contratada confirme el significado de los estados.
Política de marcas de tiempo: normalice las marcas de tiempo a UTC siempre que sea posible. Defina la prioridad entre la hora del evento de origen, la hora del evento del tracker, la hora de recepción del Message y la hora de extracción. LastDataUpdate debe representar la hora de extracción o actualización, no la hora del evento de negocio.
Límites de solicitudes y reintentos: configure el tiempo de espera, el número de reintentos, el retroceso exponencial, la gestión de estados HTTP y la paginación. Respete los límites del servicio SWIFT y evite las solicitudes paralelas sin control.
Rendimiento: almacene en caché las respuestas por UETR, procese los UETR por lotes, escriba puntos de control y utilice la extracción incremental cuando sea compatible. Evite consultar repetidamente el tracker para el mismo UETR y el mismo intervalo de fechas.
Calidad de los datos: exija PaymentTransactionId, ActivityName, EventTimestamp, SourceSystem y LastDataUpdate. Permita valores NULL para los atributos recomendados cuando la fuente no los proporcione, pero informe sobre sus tasas de cobertura.
Auditabilidad: conserve la respuesta sin procesar de la API, la marca de tiempo de la solicitud, el estado de la respuesta, el UETR y el identificador de la ejecución de extracción en una ubicación de auditoría aprobada. Aplique las normas de conservación y seguridad exigidas para los datos de pagos.
Requisitos previos: confirme el acceso a los registros internos del ciclo de vida de pagos, Cumplimiento, la pasarela SWIFT, conciliación y las interfaces de Messages, además de la licencia de SWIFT gpi Tracker y los certificados o permisos de red necesarios.

a Consulta de ejemplo python

import csv
import json
import os
import time
import uuid
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, Iterable, List, Optional

import requests

class=class="hl-string">"hl-comment"># Replace only the following placeholders with environment-specific values.
SWIFT_API_BASE_URL = class="hl-string">"[SWIFT gpi Tracker API base URL]"
SWIFT_API_PATH = class="hl-string">"[SWIFT gpi Tracker API resource path]"
SWIFT_API_TOKEN = class="hl-string">"[SWIFT API token or credential reference]"
START_DATE = class="hl-string">"[YYYY-MM-DD]"
END_DATE = class="hl-string">"[YYYY-MM-DD]"
COMPANY_FILTER = class="hl-string">"[Configure based on your system]"
INTERNAL_EVENTS_FILE = class="hl-string">"[path to internal payment lifecycle events CSV or JSON]"
OUTPUT_FILE = class="hl-string">"swift_gpi_event_log.csv"
RAW_RESPONSE_FILE = class="hl-string">"swift_gpi_raw_responses.jsonl"
REQUEST_TIMEOUT_SECONDS = 60
MAX_RETRIES = 4
PAGE_SIZE = 100

REQUIRED_COLUMNS = [
    class="hl-string">"PaymentTransactionId",
    class="hl-string">"ActivityName",
    class="hl-string">"EventTimestamp",
    class="hl-string">"SourceSystem",
    class="hl-string">"LastDataUpdate",
]

RECOMMENDED_COLUMNS = [
    class="hl-string">"TransactionAmount",
    class="hl-string">"TransactionCurrency",
    class="hl-string">"UniqueEndToEndReference",
    class="hl-string">"SwiftMessageType",
    class="hl-string">"ProcessingUser",
    class="hl-string">"BeneficiaryBic",
    class="hl-string">"SwiftNakCode",
    class="hl-string">"ValueDate",
    class="hl-string">"IsStp",
]

ALL_ACTIVITIES = {
    class="hl-string">"Payment Request Created",
    class="hl-string">"Payment Details Validated",
    class="hl-string">"Sanctions Screening Passed",
    class="hl-string">"Payment Approved",
    class="hl-string">"Payment Instruction Sent",
    class="hl-string">"SWIFT ACK Received",
    class="hl-string">"Payment Transferred",
    class="hl-string">"Payment Error Identified",
    class="hl-string">"Payment Error Resolved",
    class="hl-string">"Payment Rejected",
    class="hl-string">"Payment Settled",
    class="hl-string">"Payment Reconciled",
}


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def first_value(record: Dict[str, Any], names: Iterable[str], default: Any = None) -> Any:
    for name in names:
        if name in record and record[name] not in (None, class="hl-string">""):
            return record[name]
    return default


def normalize_timestamp(value: Any) -> Optional[str]:
    if value in (None, class="hl-string">""):
        return None
    text = str(value).strip()
    if text.endswith(class="hl-string">"Z"):
        text = text[:-1] + class="hl-string">"+00:00"
    try:
        parsed = datetime.fromisoformat(text)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed.astimezone(timezone.utc).isoformat()
    except ValueError:
        return str(value)


def normalize_amount(value: Any) -> Optional[str]:
    if value in (None, class="hl-string">""):
        return None
    try:
        return str(Decimal(str(value)))
    except (InvalidOperation, ValueError):
        return str(value)


def normalize_bool(value: Any) -> Optional[bool]:
    if value in (None, class="hl-string">""):
        return None
    if isinstance(value, bool):
        return value
    return str(value).strip().lower() in {class="hl-string">"true", class="hl-string">"1", class="hl-string">"yes", class="hl-string">"y"}


def load_records(path: str) -> List[Dict[str, Any]]:
    if path.lower().endswith(class="hl-string">".json"):
        with open(path, class="hl-string">"r", encoding=class="hl-string">"utf-8") as handle:
            data = json.load(handle)
        if isinstance(data, list):
            return data
        if isinstance(data, dict):
            for key in (class="hl-string">"records", class="hl-string">"events", class="hl-string">"payments", class="hl-string">"data"):
                if isinstance(data.get(key), list):
                    return data[key]
        raise ValueError(class="hl-string">"The JSON input must contain a list of records.")
    with open(path, class="hl-string">"r", encoding=class="hl-string">"utf-8-sig", newline=class="hl-string">"") as handle:
        return list(csv.DictReader(handle))


def base_event(payment: Dict[str, Any], activity: str, timestamp: Any, source_system: str, run_time: str) -> Optional[Dict[str, Any]]:
    payment_id = first_value(payment, [class="hl-string">"PaymentTransactionId", class="hl-string">"payment_transaction_id", class="hl-string">"transaction_id", class="hl-string">"TRN", class="hl-string">"trn", class="hl-string">"UETR", class="hl-string">"uetr"])
    event_time = normalize_timestamp(timestamp)
    if not payment_id or not event_time:
        return None
    return {
        class="hl-string">"PaymentTransactionId": str(payment_id),
        class="hl-string">"ActivityName": activity,
        class="hl-string">"EventTimestamp": event_time,
        class="hl-string">"SourceSystem": source_system,
        class="hl-string">"LastDataUpdate": run_time,
        class="hl-string">"TransactionAmount": normalize_amount(first_value(payment, [class="hl-string">"TransactionAmount", class="hl-string">"transaction_amount", class="hl-string">"amount"])),
        class="hl-string">"TransactionCurrency": first_value(payment, [class="hl-string">"TransactionCurrency", class="hl-string">"transaction_currency", class="hl-string">"currency"]),
        class="hl-string">"UniqueEndToEndReference": first_value(payment, [class="hl-string">"UniqueEndToEndReference", class="hl-string">"UETR", class="hl-string">"uetr", class="hl-string">"uetr_value"]),
        class="hl-string">"SwiftMessageType": first_value(payment, [class="hl-string">"SwiftMessageType", class="hl-string">"swift_message_type", class="hl-string">"message_type"]),
        class="hl-string">"ProcessingUser": first_value(payment, [class="hl-string">"ProcessingUser", class="hl-string">"processing_user", class="hl-string">"user", class="hl-string">"performed_by"]),
        class="hl-string">"BeneficiaryBic": first_value(payment, [class="hl-string">"BeneficiaryBic", class="hl-string">"beneficiary_bic", class="hl-string">"beneficiaryBIC"]),
        class="hl-string">"SwiftNakCode": first_value(payment, [class="hl-string">"SwiftNakCode", class="hl-string">"swift_nak_code", class="hl-string">"nak_code"]),
        class="hl-string">"ValueDate": first_value(payment, [class="hl-string">"ValueDate", class="hl-string">"value_date"]),
        class="hl-string">"IsStp": normalize_bool(first_value(payment, [class="hl-string">"IsStp", class="hl-string">"is_stp", class="hl-string">"stp"])),
    }


def add_internal_event(events: List[Dict[str, Any]], payment: Dict[str, Any], activity: str, timestamp_names: List[str], source_system: str, run_time: str) -> None:
    timestamp = first_value(payment, timestamp_names)
    event = base_event(payment, activity, timestamp, source_system, run_time)
    if event:
        events.append(event)


def extract_internal_events(payments: List[Dict[str, Any]], run_time: str) -> List[Dict[str, Any]]:
    events: List[Dict[str, Any]] = []
    for payment in payments:
        source_system = str(first_value(payment, [class="hl-string">"SourceSystem", class="hl-string">"source_system"], class="hl-string">"Internal Payment System"))
        add_internal_event(events, payment, class="hl-string">"Payment Request Created", [class="hl-string">"PaymentRequestCreatedTimestamp", class="hl-string">"payment_request_created_timestamp", class="hl-string">"created_at", class="hl-string">"creation_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Details Validated", [class="hl-string">"PaymentDetailsValidatedTimestamp", class="hl-string">"payment_details_validated_timestamp", class="hl-string">"validated_at", class="hl-string">"validation_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Sanctions Screening Passed", [class="hl-string">"SanctionsScreeningPassedTimestamp", class="hl-string">"sanctions_screening_passed_timestamp", class="hl-string">"screening_passed_at", class="hl-string">"screening_release_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Approved", [class="hl-string">"PaymentApprovedTimestamp", class="hl-string">"payment_approved_timestamp", class="hl-string">"approved_at", class="hl-string">"approval_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Instruction Sent", [class="hl-string">"PaymentInstructionSentTimestamp", class="hl-string">"payment_instruction_sent_timestamp", class="hl-string">"instruction_sent_at", class="hl-string">"swift_sent_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"SWIFT ACK Received", [class="hl-string">"SwiftAckReceivedTimestamp", class="hl-string">"swift_ack_received_timestamp", class="hl-string">"ack_received_at", class="hl-string">"swift_ack_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Error Identified", [class="hl-string">"PaymentErrorIdentifiedTimestamp", class="hl-string">"payment_error_identified_timestamp", class="hl-string">"error_identified_at", class="hl-string">"repair_queue_timestamp", class="hl-string">"exception_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Error Resolved", [class="hl-string">"PaymentErrorResolvedTimestamp", class="hl-string">"payment_error_resolved_timestamp", class="hl-string">"error_resolved_at", class="hl-string">"repair_resolved_timestamp", class="hl-string">"resubmission_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Rejected", [class="hl-string">"PaymentRejectedTimestamp", class="hl-string">"payment_rejected_timestamp", class="hl-string">"rejected_at", class="hl-string">"return_received_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Settled", [class="hl-string">"PaymentSettledTimestamp", class="hl-string">"payment_settled_timestamp", class="hl-string">"settled_at", class="hl-string">"beneficiary_credit_timestamp"], source_system, run_time)
        add_internal_event(events, payment, class="hl-string">"Payment Reconciled", [class="hl-string">"PaymentReconciledTimestamp", class="hl-string">"payment_reconciled_timestamp", class="hl-string">"reconciled_at", class="hl-string">"reconciliation_timestamp"], source_system, run_time)
    return events


def request_json(session: requests.Session, url: str, headers: Dict[str, str], params: Dict[str, Any]) -> Dict[str, Any]:
    last_error: Optional[Exception] = None
    for attempt in range(MAX_RETRIES):
        try:
            response = session.get(url, headers=headers, params=params, timeout=REQUEST_TIMEOUT_SECONDS)
            if response.status_code == 429 or response.status_code >= 500:
                response.raise_for_status()
            response.raise_for_status()
            return response.json()
        except (requests.RequestException, ValueError) as error:
            last_error = error
            if attempt == MAX_RETRIES - 1:
                break
            time.sleep(2 ** attempt)
    raise RuntimeError(fclass="hl-string">"SWIFT API request failed: {last_error}")


def get_items(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
    for key in (class="hl-string">"events", class="hl-string">"statuses", class="hl-string">"statusEvents", class="hl-string">"data", class="hl-string">"items", class="hl-string">"results"):
        value = payload.get(key)
        if isinstance(value, list):
            return [item for item in value if isinstance(item, dict)]
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    return []


def fetch_gpi_events(uetr: str, session: requests.Session, raw_handle: Any, run_time: str) -> List[Dict[str, Any]]:
    url = SWIFT_API_BASE_URL.rstrip(class="hl-string">"/") + class="hl-string">"/" + SWIFT_API_PATH.lstrip(class="hl-string">"/")
    headers = {class="hl-string">"Accept": class="hl-string">"application/json", class="hl-string">"Authorization": fclass="hl-string">"Bearer {SWIFT_API_TOKEN}"}
    page = 1
    output: List[Dict[str, Any]] = []
    while True:
        params = {class="hl-string">"uetr": uetr, class="hl-string">"startDate": START_DATE, class="hl-string">"endDate": END_DATE, class="hl-string">"page": page, class="hl-string">"pageSize": PAGE_SIZE}
        payload = request_json(session, url, headers, params)
        raw_handle.write(json.dumps({class="hl-string">"run_time": run_time, class="hl-string">"uetr": uetr, class="hl-string">"request_params": params, class="hl-string">"response": payload}, default=str) + class="hl-string">"\n")
        items = get_items(payload)
        output.extend(items)
        has_next = first_value(payload, [class="hl-string">"hasNext", class="hl-string">"has_next", class="hl-string">"nextPage", class="hl-string">"next_page"])
        if not items or has_next in (False, None, class="hl-string">"false", class="hl-string">"False"):
            break
        if isinstance(has_next, (int, str)) and str(has_next).isdigit():
            page = int(has_next)
        else:
            page += 1
    return output


def tracker_event(payment: Dict[str, Any], tracker: Dict[str, Any], activity: str, run_time: str) -> Optional[Dict[str, Any]]:
    merged = dict(payment)
    merged.update({
        class="hl-string">"UniqueEndToEndReference": first_value(tracker, [class="hl-string">"uetr", class="hl-string">"UETR"], first_value(payment, [class="hl-string">"UniqueEndToEndReference", class="hl-string">"UETR", class="hl-string">"uetr"])),
        class="hl-string">"SwiftMessageType": first_value(tracker, [class="hl-string">"messageType", class="hl-string">"message_type", class="hl-string">"swiftMessageType"], first_value(payment, [class="hl-string">"SwiftMessageType", class="hl-string">"swift_message_type", class="hl-string">"message_type"])),
        class="hl-string">"BeneficiaryBic": first_value(tracker, [class="hl-string">"beneficiaryBic", class="hl-string">"beneficiary_bic", class="hl-string">"beneficiaryBIC"], first_value(payment, [class="hl-string">"BeneficiaryBic", class="hl-string">"beneficiary_bic", class="hl-string">"beneficiaryBIC"])),
        class="hl-string">"TransactionAmount": first_value(tracker, [class="hl-string">"amount", class="hl-string">"transactionAmount", class="hl-string">"transaction_amount"], first_value(payment, [class="hl-string">"TransactionAmount", class="hl-string">"transaction_amount", class="hl-string">"amount"])),
        class="hl-string">"TransactionCurrency": first_value(tracker, [class="hl-string">"currency", class="hl-string">"transactionCurrency", class="hl-string">"transaction_currency"], first_value(payment, [class="hl-string">"TransactionCurrency", class="hl-string">"transaction_currency", class="hl-string">"currency"])),
        class="hl-string">"ValueDate": first_value(tracker, [class="hl-string">"valueDate", class="hl-string">"value_date"], first_value(payment, [class="hl-string">"ValueDate", class="hl-string">"value_date"])),
        class="hl-string">"IsStp": first_value(tracker, [class="hl-string">"isStp", class="hl-string">"is_stp", class="hl-string">"stp"], first_value(payment, [class="hl-string">"IsStp", class="hl-string">"is_stp", class="hl-string">"stp"])),
        class="hl-string">"SwiftNakCode": first_value(tracker, [class="hl-string">"nakCode", class="hl-string">"nak_code", class="hl-string">"swiftNakCode"], first_value(payment, [class="hl-string">"SwiftNakCode", class="hl-string">"swift_nak_code", class="hl-string">"nak_code"])),
    })
    timestamp = first_value(tracker, [class="hl-string">"eventTimestamp", class="hl-string">"event_timestamp", class="hl-string">"timestamp", class="hl-string">"statusTimestamp", class="hl-string">"status_timestamp", class="hl-string">"occurredAt", class="hl-string">"occurred_at"])
    return base_event(merged, activity, timestamp, class="hl-string">"SWIFT gpi Tracker API", run_time)


def extract_tracker_events(payments: List[Dict[str, Any]], run_time: str, raw_handle: Any) -> List[Dict[str, Any]]:
    events: List[Dict[str, Any]] = []
    seen_uetrs = set()
    with requests.Session() as session:
        for payment in payments:
            uetr = first_value(payment, [class="hl-string">"UniqueEndToEndReference", class="hl-string">"UETR", class="hl-string">"uetr", class="hl-string">"uetr_value"])
            if not uetr or str(uetr) in seen_uetrs:
                continue
            seen_uetrs.add(str(uetr))
            for tracker in fetch_gpi_events(str(uetr), session, raw_handle, run_time):
                status = str(first_value(tracker, [class="hl-string">"status", class="hl-string">"statusCode", class="hl-string">"status_code", class="hl-string">"gpiStatus", class="hl-string">"gpi_status"], class="hl-string">"")).upper()
                activity = {class="hl-string">"ACSP": class="hl-string">"Payment Transferred", class="hl-string">"RJCT": class="hl-string">"Payment Rejected", class="hl-string">"ACSC": class="hl-string">"Payment Settled"}.get(status)
                if activity:
                    event = tracker_event(payment, tracker, activity, run_time)
                    if event:
                        events.append(event)
    return events


def deduplicate(events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    result: List[Dict[str, Any]] = []
    seen = set()
    for event in sorted(events, key=lambda row: (row[class="hl-string">"PaymentTransactionId"], row[class="hl-string">"EventTimestamp"], row[class="hl-string">"ActivityName"])):
        key = (event[class="hl-string">"PaymentTransactionId"], event[class="hl-string">"ActivityName"], event[class="hl-string">"EventTimestamp"], event.get(class="hl-string">"UniqueEndToEndReference"))
        if key not in seen:
            seen.add(key)
            result.append(event)
    return result


def write_csv(events: List[Dict[str, Any]], path: str) -> None:
    columns = REQUIRED_COLUMNS + RECOMMENDED_COLUMNS
    with open(path, class="hl-string">"w", encoding=class="hl-string">"utf-8", newline=class="hl-string">"") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns, extrasaction=class="hl-string">"ignore")
        writer.writeheader()
        for event in events:
            writer.writerow({column: event.get(column) for column in columns})


def main() -> None:
    run_time = utc_now()
    payments = load_records(INTERNAL_EVENTS_FILE)
    if COMPANY_FILTER != class="hl-string">"[Configure based on your system]":
        payments = [payment for payment in payments if str(first_value(payment, [class="hl-string">"CompanyCode", class="hl-string">"company_code", class="hl-string">"Company", class="hl-string">"company"], class="hl-string">"")) == COMPANY_FILTER]
    internal_events = extract_internal_events(payments, run_time)
    with open(RAW_RESPONSE_FILE, class="hl-string">"w", encoding=class="hl-string">"utf-8") as raw_handle:
        tracker_events = extract_tracker_events(payments, run_time, raw_handle)
    events = deduplicate(internal_events + tracker_events)
    invalid = [event for event in events if any(event.get(column) in (None, class="hl-string">"") for column in REQUIRED_COLUMNS) or event[class="hl-string">"ActivityName"] not in ALL_ACTIVITIES]
    if invalid:
        raise ValueError(fclass="hl-string">"The extraction produced {len(invalid)} invalid event rows.")
    write_csv(events, OUTPUT_FILE)
    print(json.dumps({class="hl-string">"run_id": str(uuid.uuid4()), class="hl-string">"rows_written": len(events), class="hl-string">"output_file": OUTPUT_FILE, class="hl-string">"activities": sorted({event[class="hl-string">"ActivityName"] for event in events})}, indent=2))


if __name__ == class="hl-string">"__main__":
    main()

¿Listo para comenzar?

Aproveche estos templates para transformar sus registros sin procesar de SWIFT en una herramienta eficaz para la excelencia operativa y una liquidación de pagos más rápida.

Volver a la guía de mejora Descargar Template

Optimice ahora su procesamiento de pagos SWIFT

Alcance un procesamiento directo del 98 % y elimine los errores manuales

Inicie su prueba gratuita

No necesita tarjeta de crédito. Configuración en 5 minutos

Su Template de datos de procesamiento de pagos

Su Template de datos de procesamiento de pagos

Atributos del procesamiento de pagos

Actividades del procesamiento de pagos

Guías de extracción

Exportación del archivo de SWIFT Alliance Access

Pasos

Configuración

a Consulta de ejemplo python

Analizador de Python para archivos SWIFT MT sin procesar

Pasos

Configuración

a Consulta de ejemplo python

Conector de API de SWIFT gpi

Pasos

Configuración

a Consulta de ejemplo python

¿Listo para comenzar?

Optimice ahora su procesamiento de pagos SWIFT

Su Template de datos de procesamiento de pagos

Su Template de datos de procesamiento de pagos

Atributos del procesamiento de pagos

Actividades del procesamiento de pagos

Guías de extracción

Exportación del archivo de SWIFT Alliance Access

Pasos

Configuración

a Consulta de ejemplo python

Analizador de Python para archivos SWIFT MT sin procesar

Pasos

Configuración

a Consulta de ejemplo python

Conector de API de SWIFT gpi

Pasos

Configuración

a Consulta de ejemplo python

¿Listo para comenzar?

Optimice ahora su procesamiento de pagos SWIFT

Valoramos su privacidad