> 持续流程改进 > 付款处理 > SWIFT > 数据模板

支付处理数据模板

SWIFT

← 返回改进指南

支付处理属性(19) 支付处理活动(12) 数据提取指南(3)

支付处理数据模板

此模板为梳理SWIFT交易工作流提供基础，涵盖从发起到最终结算的全过程。模板列出了全面了解支付生命周期所需的具体数据项和运营里程碑。按照本指南准备数据，您可以发现隐藏的低效环节并降低合规风险。

用于深入分析交易的建议属性
SWIFT流程的关键活动与里程碑
财务数据提取技术指南

刚接触事件日志？了解如何创建流程挖掘事件日志.

下载模板

支付处理属性

建议在事件日志中纳入以下数据字段，以全面分析SWIFT交易生命周期。

5 必填 9 建议填写 5 选填

	名称	说明
	事件时间戳 EventTimestamp	活动发生的准确日期和时间。
说明此属性记录事件发生的准确时刻，用于按时间顺序排列活动，并计算各类时长指标，例如前置时间、周期时间和吞吐量。准确的时间戳对于分析“网络截止时间表现”和发现审批链中的瓶颈至关重要。重要性时间戳是流程挖掘中所有时间分析和绩效指标的基础。获取途径系统审计日志、报文创建时间戳或数据库交易时间。示例 2023-10-25T08:30:15Z2023-10-25T14:45:00Z2023-10-26T09:15:22Z
	付款交易ID PaymentTransactionId	用于标识端到端付款案例的唯一标识符。
说明此属性是流程挖掘中的核心案例标识，可将单笔付款指令相关的所有活动归为一组，涵盖初始申请、校验、SWIFT传输直至最终结算。在SWIFT场景中，通常映射为交易参考号（TRN）或端到端唯一交易参考号（UETR），以确保数据在各银行系统间保持连续。该标识用于还原流程实例路径，是分析流程变体和计算周期时间的基础。重要性唯一标识是流程挖掘的基础，可将分散的事件串联起来，形成连贯的流程视图。获取途径报文头中的SWIFT字段20（TRN）或字段121（UETR）。示例 TRN-20231025-883954392-882-101UETR-9982-1123-5521PAY-US-EU-9912
	数据最近更新时间 LastDataUpdate	记录最近一次提取或刷新的时间戳。
说明表示数据最近一次载入流程挖掘工具的时间。这对于判断数据是否为最新，以及确保仪表板反映当前付款运营情况至关重要。通过透明展示报表管道的延迟情况，帮助用户信任所呈现的数据。重要性帮助用户了解所呈现数据的时效性和可靠性。获取途径 ETL流程执行时间戳。示例 2023-10-27T12:00:00Z2023-11-01T06:00:00Z
	活动名称 ActivityName	发生的流程步骤或事件名称。
说明此属性描述系统日志中记录的具体操作或状态变更，例如“创建付款申请”“通过制裁筛查”或“收到SWIFT确认”。这是流程图的主要维度，用于定义可视化图中的节点，帮助分析人员了解付款操作的先后顺序。重要性它定义流程中的“做什么”，使流程走向和变体得以可视化。获取途径交易日志、审计轨迹，或根据报文类型推导（例如已发送MT103）。示例创建付款申请发送付款指令收到SWIFT确认批准付款
	源系统 SourceSystem	事件数据的来源系统名称。
说明用于识别生成事件日志的软件或平台。在付款流程中，来源可能是核心银行系统、付款网关或SWIFT接口。当合并多个不同银行应用的数据时，此属性有助于验证数据来源并排查问题。重要性提供数据来源背景，是开展多系统流程挖掘的必要信息。获取途径在ETL过程中硬编码，或从系统标识符中提取。示例 SWIFT Alliance Access核心银行系统付款引擎制裁筛查系统
	SWIFT NAK代码 SwiftNakCode	消息被拒绝时，SWIFT网络返回的错误代码。
说明包含SWIFT网络在否定确认（NAK）中提供的具体错误代码（如T26、T13）。该属性支持“SWIFT错误与返工分析”仪表板，可对技术故障分类并优先处理系统修复。重要性提供网络层拒绝的技术根因。获取途径 MT 015/019系统消息，或ACK/NAK中的字段451。示例 T26H01T13G02
	SWIFT消息类型 SwiftMessageType	所用SWIFT消息的类型（如MT103、pacs.008）。
说明表示付款指令的具体格式。常见类型包括客户转账使用的MT103，以及对应的ISO 20022格式（如pacs.008）。可用于“付款路径变体分析”，比较传统MT格式与新版ISO标准的处理效率。重要性区分付款流转方式和处理标准（传统格式与ISO 20022）。获取途径 SWIFT第2区块（应用程序头），消息类型字段。示例 MT103MT202pacs.008MT101
	UETR UniqueEndToEndReference	用于在SWIFT gpi中跟踪交易的端到端唯一交易参考编号。
说明 UETR是一个36字符的字符串，为付款提供贯穿整个SWIFT网络的唯一且不可变的参考编号。与内部ID不同，UETR在不同银行之间保持不变。该属性对于端到端可视性至关重要，也可用于将内部日志与外部SWIFT gpi状态更新关联起来。重要性跨机构跟踪跨境付款的行业标准。获取途径 SWIFT第3区块，字段121。示例 b8c3f4a0-5d2a-4e1b-9c3d-1a2b3c4d5e6f123e4567-e89b-12d3-a456-426614174000
	交易币种 TransactionCurrency	表示付款币种的3位ISO代码。
说明标识付款所使用的币种（如USD、EUR、GBP）。这对“网络截止时间表现”仪表板至关重要，因为截止时间因币种而异。该属性还可用于“货币转换效率”分析，识别不同币种之间的货币对。重要性决定路由规则、截止时间和结算通道。获取途径 SWIFT MT字段32A（币种）或ISO 20022元素Ccy。示例 USDEURGBPJPY
	交易金额 TransactionAmount	付款指令的金额。
说明表示转账本金金额。该数据从特定SWIFT字段中提取，例如MT103中的字段32A。这是“大额转账审批周期”仪表板的关键数据，可按金额细分付款，分析高风险、大额交易的审批延误。重要性支持财务影响分析，并可按交易金额进行细分。获取途径 SWIFT MT字段32A（金额）或ISO 20022元素IntrBkSttlmAmt。示例 15000.001250.501000000.0045.00
	受益人BIC BeneficiaryBic	收款机构的银行识别代码。
说明标识付款的目标银行，是“受益人拒绝原因”仪表板的重要分析维度。按受益人BIC对失败情况分组，银行便可识别经常因数据质量问题或特定格式要求而拒绝指令的交易对手。重要性分析交易对手表现和拒绝模式的关键属性。获取途径 SWIFT字段57A（账户开户机构）或58A（受益人机构）。示例 CITIUS33BARCGB22DRESDEFFHANDSEXX
	处理用户 ProcessingUser	执行该活动的用户或系统代理。
说明标识负责该活动的人员或自动化机器人，例如批准制裁命中项的合规专员，或修正格式错误的操作员。该属性可用于生成“验证错误热力图”，帮助管理者识别培训需求，或找出返工率较高的特定用户。重要性支持资源分析并识别人工处理瓶颈。获取途径系统审计日志、交易表中的“用户ID”列。示例系统J.DoeCompliance_Bot_01M.Smith
	是否直通处理 IsStp	表示付款是否无需人工干预的标记。
说明布尔指标。如果案例中没有“发现错误”“付款被拒绝”或人工“修改”活动，值为True。该指标可直接用于计算“直通处理率”KPI。这是衡量付款处理自动化成效的主要指标。重要性衡量流程效率和自动化运行状况的主要指标。获取途径根据案例中是否存在特定的负面或人工活动计算得出。示例 truefalse
	起息日 ValueDate	资金应可供受益人使用的日期。
说明表示付款消息中指定的结算日期。将该日期与实际“付款已结算”时间戳进行比较，有助于“结算与对账账龄”仪表板分析。该属性反映付款的紧急程度，也可用于衡量资金可用性是否符合服务级别协议。重要性对流动性管理和结算时效衡量至关重要。获取途径 SWIFT MT字段32A（日期子字段）或ISO 20022 IntrBkSttlmDt。示例 2023-10-262023-11-01
	发起国家 OriginatingCountry	发起付款实体所在国家的代码。
说明表示付款请求发起地所属的司法管辖区。这对“制裁筛查处理时长”仪表板至关重要，因为来自高风险司法管辖区的付款通常需要经过更严格、耗时更长的一致性检查。该属性还可用于分析付款量和处理延迟的地理分布。重要性合规风险分析和区域表现分析的关键维度。获取途径根据发送方BIC推导（第5至第6位为国家代码）。示例 USGBDEFR
	拒绝原因 RejectionReason	说明付款被拒绝原因的文本描述。
说明付款失败时记录的文字说明或代码描述，通常见于SWIFT字段72（发送方至接收方信息）或退回消息。该属性支持“受益人拒绝原因”分析。分析人员可通过文本挖掘找出拒绝原因中的常见主题（如“账户无效”“受益人名称不匹配”）。重要性为流程失败提供定性背景信息。获取途径退回消息（MT103退回）中的SWIFT字段72或79。示例受益人账户已关闭IBAN无效未通过监管一致性检查未知银行识别代码
	指令优先级 InstructionPriority	表示付款紧急程度的优先级标记。
说明根据消息头信息确定（如“普通”或“紧急”）。这可用于“高价值转账审批周期”仪表板中的细分分析，因为紧急付款通常需要加快审批。了解优先级分布有助于规划高峰时段的资源。重要性有助于区分标准付款与加急付款的SLA要求。获取途径 SWIFT第2区块，消息优先级字段（如N表示普通，U表示紧急）。示例普通紧急系统
	是否赶上截止时间 MetCutOffTime	表示指令是否在网络截止时间前发送的标记。
说明通过比较“付款指令已发送”时间与币种对应的SWIFT网络截止时间计算得出。该属性支持“SWIFT截止时间遵守率”KPI。未赶上截止时间会导致结算延迟，影响流动性状况。重要性资金管理和流动性管理的重要运营KPI。获取途径将时间戳与截止时间静态参考表进行比较后计算得出。示例 truefalse
	是否跨境 IsCrossBorder	表示付款是否涉及不同国家的标记。
说明布尔属性。当发送方国家与受益人国家不同时，值为True。该属性可用于“多币种处理延迟”KPI，区分境内与国际付款流转。跨境付款通常更复杂、成本更高，处理时间也更长。重要性分析付款复杂度的基础细分维度。获取途径比较发送方BIC国家代码与受益人BIC国家代码。示例 truefalse

必填建议填写选填

支付处理活动

建议在事件日志中记录以下关键流程步骤和里程碑，以准确发现流程并识别瓶颈。

6 建议填写 6 选填

	活动	说明
	付款已对账	将付款交易与Nostro/Vostro账户账簿对账单进行匹配。对账系统将交易ID关联至对账单明细时，可据此推断。
重要性完成财务闭环。此处延误会影响现金可视性和审计能力。获取途径对账系统日志或核心银行系统中的“已匹配”状态。采集比较对账状态事件类型 inferred
	付款已结算	确认资金已记入收款人账户（通常为gpi状态ACSC）。可通过跟踪器更新或确认报文明确获取。
重要性标志着付款链路在业务层面成功完成，是计算“结算至对账差距”的关键数据。获取途径 SWIFT gpi Tracker（ACSC状态）或MT900/910确认报文。采集确认结算时记录事件类型 explicit
	创建付款申请	在内部银行系统或ERP中首次生成付款指令。系统首次为付款订单分配唯一交易参考号（TRN或UETR）时，即可明确记录此事件。
重要性确定整个付款生命周期的开始时间，是计算端到端处理时长、发现SWIFT网络介入前上游延误的关键。获取途径核心银行系统（CBS）或付款中心中的交易表创建时间戳。采集创建交易记录时记录事件类型 explicit
	发送付款指令	将格式化报文（MT103或ISO 20022 pacs.008）发送至SWIFT Gateway。生成报文内容并将其交给网络接口时，即可明确记录此事件。
重要性用于计算“SWIFT截止时间遵守率”，标志着处理流程从内部处理转入网络处理。获取途径 SWIFT Alliance Access（SAA）日志或Gateway传输日志。采集生成报文输出时记录事件类型 explicit
	批准付款	指定人员或自动规则对大额转账执行的最终授权操作。审批工作流标记设为true时，即可明确记录此事件。
重要性这是“大额转账审批周期”仪表板的关键数据，可用于发现大额流动资金转移中的人工签批延误。获取途径审批工作流引擎的审计轨迹。采集执行审批操作时记录事件类型 explicit
	通过制裁筛查	付款指令成功通过OFAC/制裁名单筛查的时间戳。可从合规系统日志中直接获取，也可根据“筛查暂停”状态解除的时间推断。
重要性这是“制裁筛查前置时间”仪表板的关键数据。此处延误反映的是合规瓶颈，而非运营效率低下。获取途径合规筛查系统日志或付款中心的合规状态标记。采集筛查状态更新时记录事件类型 explicit
	付款已转出	中间状态更新（通常为gpi状态ACSP），表示中间银行正在处理付款。可通过SWIFT gpi Tracker更新或状态报文获取。
重要性让代理行处理这一“黑箱”环节变得可见，是分析端到端完整流程的关键。获取途径 SWIFT gpi Tracker数据流或MT199/trck报文。采集收到跟踪器更新时记录事件类型 explicit
	付款被拒	收到收款银行或中间银行发来的拒付报文（MT103退汇或gpi RJCT状态）。可从收到的SWIFT报文中明确获取。
重要性这是“收款方拒付根因”仪表板的关键数据，可用于发现账户已关闭或路由数据错误等外部阻碍。获取途径收到的SWIFT报文（MT103 RET、pacs.004）或gpi状态RJCT。采集收到拒付报文时记录事件类型 explicit
	发现付款错误	交易因校验失败、NACK或拒付报文而被标记为待修复的事件。交易进入“修复”“更正”或“异常”队列时，可据此推断。
重要性用于跟踪“错误解决返工率”。此类情况频繁发生会拉低直通处理（STP）比例。获取途径付款中心状态列或异常队列日志。采集检查状态字段是否为Error/Repair 事件类型 inferred
	收到SWIFT确认	收到SWIFT网络发出的技术确认（ACK），表示报文已获接收并进入处理流程。可从网络接口日志中明确获取。
重要性确认报文已成功进入全球网络，便于区分内部故障与实际网络传输问题。获取途径 SWIFT Gateway日志（查看ACK与NACK信号）。采集收到网络ACK时记录事件类型 explicit
	解决付款错误	修改并重新提交此前出错的交易，且操作成功。交易从“修复”队列返回“处理中”或“就绪”状态时，可据此推断。
重要性用于衡量人工返工团队的效率。此处耗时过长，可能说明培训不足或错误代码复杂。获取途径显示交易已退出异常队列的付款中心状态日志。采集检查状态字段是否已退出Repair 事件类型 inferred
	验证付款详情	成功完成付款指令的语法和格式检查，例如IBAN格式、BIC有效性。通常可根据交易状态从“草稿”变为“已验证”或“待授权”来推断。
重要性此处失败率偏高，说明源头数据质量较差。用于衡量“首次校验通过率”KPI。获取途径付款引擎状态日志或校验历史表。采集比较操作前后的状态字段事件类型 inferred

建议填写选填

数据提取指南

如何从SWIFT获取数据

步骤

使用具有归档配置权限的管理员账户，登录SWIFT Alliance WebPlatform或Alliance Access GUI。
打开“System Management”菜单，选择“Archiving”，然后选择“Automated Archiving Configuration”。
新建每日导出配置文件。将目标目录设为提取脚本可访问的安全网络路径。
指定内容范围：选择“Messages”，并勾选“Include History”和“Include Interventions”。这对于捕获审批、验证等审计轨迹事件至关重要。
选择输出格式。选择XML（具体为“SAA XML v2”或最新可用架构），因为与CSV相比，它更能保留消息标头与历史事件之间的层级关系。
设置日期范围筛选条件，提取流程挖掘所需的历史数据；也可以设为每日运行，以获取T-1数据。
首次加载时手动执行归档导出任务，或等待计划批处理任务完成。确认目标目录中已生成.xml文件。
SAA导出文件为层级XML，而ProcessMind需要扁平CSV事件日志，因此必须运行转换脚本。使用“Query”部分提供的Python脚本解析XML。
脚本会遍历每个消息元素，提取唯一端到端交易参考号（UETR）或交易参考号（TRN）作为案例ID。
随后，脚本会遍历每条消息的“History”和“Intervention”子节点，生成活动行，并将SAA内部状态代码（如“NetworkAck”和“Authorisation”）映射为标准活动名称。
使用Python脚本处理导出的XML文件，生成扁平的“events.csv”文件。
验证CSV输出，确保EventTimestamp等列采用ISO 8601格式，然后将文件上传至ProcessMind。

配置

归档格式：XML（首选）或CSV（扁平文件）。强烈建议使用XML，以保留嵌套的历史记录节点。
包含消息正文：是（解析金额、币种和受益人信息所必需）。
包含历史记录/干预记录：是（查看审批、验证和错误所必需）。
日期范围：增量加载时设置滚动时间范围（如最近30天）；首次加载历史数据时设置固定日期。
消息类型：筛选MT103、MT202、pacs.008、pacs.009和pacs.004/002（退回/状态）。
消息方向：同时包含输入（发送）和输出（接收），以捕获收到的拒绝或确认消息。
安全：严格限制导出文件夹的读写权限，因为其中包含实时支付数据。

查询示例 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}")

步骤

准备环境：确保提取服务器已安装Python 3.8或以上版本。使用pip安装所需库（pandas、numpy）。创建专用工作目录，并在其中新建名为raw_swift_files的子文件夹，将每日.out、.txt或.fin文件放入其中。
了解文件结构：原始SWIFT FIN消息采用分块结构（{1: Header} {2: Application} {4: Body}）。确认分隔符。标准文件使用花括号。确保文件未封装在专有信封格式（如RJE标头）中；如有，请在处理前移除。
配置脚本路径：打开提供的Python脚本。将INPUT_DIRECTORY变量设为raw_swift_files文件夹的路径，并设置生成CSV文件的OUTPUT_FILE路径。
定义正则表达式模式：脚本使用正则表达式提取各个数据块。数据块1包含发送方BIC和序列号；数据块2包含消息类型（如I103、O103）和接收方BIC；数据块4包含业务数据标签（如：20：TRN、：32A：价值日期/金额）。
将消息类型映射为活动：脚本包含映射字典。MT103表示“付款指令已发送”；MT012（系统ACK）表示“已收到SWIFT ACK”；MT900/910表示“付款已结算”。带有特定状态代码（ACSP、RJCT）的MT199/299映射为“付款已转出”或“付款已拒绝”。
提取属性：解析器会遍历指定标签。:20:映射为Payment Transaction ID；:121:映射为UETR；:32A:解析为日期、币种和金额。
处理内部状态：对于“付款请求已创建”或“付款已批准”等通常不会记录在SWIFT文件中的内部ERP事件，如果没有明确日志，脚本会从数据块1中的消息创建时间戳推导；也可以在:72:字段（发件人至收件人信息）中查找特定文本代码，前提是贵组织在此记录审批信息。
执行提取：在命令行运行脚本：python swift_parser.py。脚本会遍历所有文件，解析有效的SWIFT数据块并跳过格式错误的行。
后处理验证：打开生成的CSV。检查EventTimestamp格式（建议使用ISO 8601）。确保每行都填写了PaymentTransactionId，以便按案例分组。
数据转换：如果“ValueDate”以YYMMDD格式输出，脚本会将其转换为YYYY-MM-DD。确保“TransactionAmount”中的小数分隔符统一（使用点号或逗号）。
最终检查：按特定UETR筛选，查看从指令已发送→ACK→结算的端到端流程。确认NAK（MT019）显示为“已发现付款错误”。
上传至ProcessMind：导入CSV。将PaymentTransactionId映射为案例ID，将ActivityName映射为活动，将EventTimestamp映射为时间戳。

配置

输入目录：存放原始FIN/MT扁平文件的文件夹路径。确保已授予读取权限。
日期格式解析：SWIFT原始日期采用YYMMDD格式。脚本会按20xx处理世纪部分。
消息类型筛选：脚本重点处理类别1（支付）、类别9（对账单）和常见系统消息（012、019）。如需类别2（金融机构间转账），请调整RELEVANT_MT_TYPES列表。
编码：默认使用“utf-8”。如果文件包含旧版编码字符，请改用“latin-1”或“ascii”。
批量处理：对于大于1GB的数据集，脚本采用生成器方式逐个处理文件，避免内存溢出。
正则表达式调整：严格的正则表达式假设文件采用标准FIN格式{1:...}{2:...}{4:...}。如果文件使用$或其他分隔符，请更新BLOCK_PATTERN变量。

查询示例 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()

步骤

确认您已订阅SWIFT gpi Tracker API，具备API凭据和UETR查询能力，并可访问内部支付系统事件源。请从SWIFT服务文档中获取准确的API基础URL、身份验证方式、响应架构、速率限制和分页规则。不要假设端点路径，因为它取决于获许可的SWIFT服务和部署方式。
使用所需的连接字符串、凭据、日期范围和公司专属筛选条件配置脚本。首次提取建议使用三至六个月的日期范围。将内部源数据限制为具有TRN或UETR的支付交易，并为每个事件保留源系统名称。
提取以下内部生命周期事件：已创建支付请求、已验证支付详情、制裁筛查已通过、支付已批准、已发送支付指令、已收到SWIFT ACK、已发现支付错误、支付错误已解决、支付已拒绝、支付已结算和支付已对账。每个源事件都必须转换为单独的事件日志行。不要依赖ProcessMind推断活动。
从内部支付记录中提取UETR，并针对每个UETR查询获许可的SWIFT gpi Tracker API。根据您所在环境的API文档配置请求URL、身份验证标头、请求参数、分页和响应字段。保存原始响应，以便完整审计。
将适用的gpi Tracker状态转换为明确的事件行。脚本会将ACSP映射为“支付已转出”，RJCT映射为“支付已拒绝”，ACSC映射为“支付已结算”。如果获许可的响应使用其他状态名称或时间戳，请依据已记录的响应架构更新状态映射。API未提供相应状态记录时，不要创建事件。
将所有时间戳统一到同一时区，最好使用UTC；如审计需要，同时保留原始时间戳。将金额转换为数值，保留交易币种，统一STP布尔值的格式；源数据未提供属性时保留NULL。
使用PaymentTransactionId、ActivityName、EventTimestamp和最具体的可用源事件标识去重。除非业务规则明确要求每项活动只保留一次，否则应保留多次修复和重新提交等合理的重复事件。
验证每行输出都包含PaymentTransactionId、ActivityName、EventTimestamp、SourceSystem和LastDataUpdate。确认ActivityName仅包含文档规定的活动值；如果源数据中存在相应记录，确保12项活动均有体现。
将生成的数据集导出为UTF-8 CSV或其他ProcessMind支持的事件日志格式。每行对应一个事件，包含必需和建议列，并将PaymentTransactionId映射为案例标识符、ActivityName映射为活动、EventTimestamp映射为事件时间戳。将文件上传至ProcessMind，无需期待系统进一步推导事件。

配置

API端点和架构：根据您所在环境的SWIFT许可文档，配置准确的SWIFT gpi Tracker API基础URL、资源路径、请求参数、身份验证标头、分页和响应字段名称。脚本不会自行编造端点。
凭据和授权：使用获批准的SWIFT API凭据机制，并遵循证书或令牌要求、机密信息存储方式及内部源系统授权规定。不要将生产环境机密直接写入源代码。
日期范围：从三至六个月开始。根据报告目标，使用事件时间戳或支付创建日期。数据量较大时，缩短处理时间窗口，并为已完成的UETR设置检查点。
内部源筛选条件：根据您的系统配置公司、法律实体、分支机构、支付产品、单据类型、消息类型、币种和状态筛选条件。确认源架构前，请使用[根据您的系统配置]等占位符。
活动映射：为内部状态和队列维护明确的映射。例如，将“草稿”映射为“已验证”或“待授权”，表示支付详情已验证；将“解除筛查暂停”映射为制裁筛查已通过；将“进入修复或异常队列”映射为已发现支付错误；将“返回处理中”或“准备付款”映射为支付错误已解决。
gpi状态映射：将文档记录的Tracker状态映射为明确活动。此实现将ACSP映射为支付已转出、RJCT映射为支付已拒绝、ACSC映射为支付已结算。只有在获许可的API文档确认状态含义后，才扩展映射。
时间戳规则：尽可能将时间戳统一为UTC。明确源事件时间、Tracker事件时间、消息接收时间和提取时间的优先顺序。LastDataUpdate应表示提取或刷新时间，而非业务事件时间。
速率限制和重试：配置请求超时、重试次数、指数退避、HTTP状态处理和分页。遵守SWIFT服务限制，避免不受控的并行请求。
性能：按UETR缓存响应，分批处理UETR，写入检查点，并在支持时进行增量提取。避免针对相同UETR和日期范围重复查询Tracker。
数据质量：要求包含PaymentTransactionId、ActivityName、EventTimestamp、SourceSystem和LastDataUpdate。源数据未提供建议属性时允许NULL，但应报告其填充率。
可审计性：在获批准的审计位置保存API原始响应、请求时间戳、响应状态、UETR和提取运行标识。遵守支付数据所需的保留和安全规则。
前提条件：确认可访问内部支付生命周期、合规、SWIFT网关、对账和消息接口日志，并已获得SWIFT gpi Tracker API许可及所需证书或网络白名单。

查询示例 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()

准备好开始了吗？

借助这些模板，将原始SWIFT日志转化为改善运营、加快支付结算的依据。

返回改进指南

下载模板

立即优化SWIFT支付处理

实现98%的直通式处理，减少人工错误

开始免费试用

无需信用卡，5分钟即可完成设置

支付处理数据模板

支付处理数据模板

支付处理属性

支付处理活动

数据提取指南

SWIFT Alliance Access归档导出

步骤

配置

查询示例 python

用于解析原始SWIFT MT文件的Python解析器

步骤

配置

查询示例 python

SWIFT gpi API连接器

步骤

配置

查询示例 python

准备好开始了吗？

立即优化SWIFT支付处理