> 持续流程改进 > 支付处理 > 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
	最后数据更新时间 LastDataUpdate	记录最后一次提取或刷新的时间戳。
说明表示数据最后一次加载到流程挖掘工具的时间。这对于判断数据新鲜度、确保仪表板反映付款运营最新状态至关重要。它还通过展示报告管道的延迟，帮助用户建立对数据的信任。为什么重要确保用户了解所呈现数据的时效性和可靠性。获取位置 ETL流程执行时间戳。示例 2023-10-27T12:00:00Z2023-11-01T06:00:00Z
	支付交易ID PaymentTransactionId	代表端到端付款案例的唯一标识符。
说明该属性是流程挖掘的核心案例键。它将单笔付款指令相关的所有活动关联起来，涵盖初始请求、验证、SWIFT传输和最终结算。在SWIFT场景中，它通常映射到Transaction Reference Number（TRN）或Unique End-to-end Transaction Reference（UETR），以确保跨银行系统的连续性。它用于重建流程实例路径，是所有变体分析和周期时间计算的基础。为什么重要唯一标识是流程挖掘的基础，可将分散的事件串联成连贯的流程视图。获取位置消息头中的SWIFT Block 3、Field 20（TRN）或Field 121（UETR）。示例 TRN-20231025-883954392-882-101UETR-9982-1123-5521PAY-US-EU-9912
	活动名称 ActivityName	已发生的流程步骤或事件名称。
说明该属性描述系统日志中记录的具体操作或状态变化。例如“Payment Request Created”“Sanctions Screening Passed”或“SWIFT ACK Received”。它是流程地图的主要维度，定义可视化中的节点，并帮助分析人员了解付款执行的操作顺序。为什么重要它定义流程中的“内容”，从而实现流程顺序和流程变体的可视化。获取位置交易日志、审计追踪，或根据消息类型推导，例如MT103 Sent。示例创建付款请求已发送付款指令已收到SWIFT ACK付款已批准
	源系统 SourceSystem	事件数据来源系统的名称。
说明用于识别生成事件日志的软件或平台。在付款流程中，它可能是核心银行系统、付款网关或SWIFT接口本身。从多个不同银行应用合并数据时，该属性有助于验证数据血缘并排查问题。为什么重要提供数据来源上下文，是多系统流程挖掘不可或缺的信息。获取位置在ETL过程中硬编码，或从系统标识符中提取。示例 SWIFT Alliance Access核心银行系统支付引擎制裁筛查器
	SWIFT NAK代码 SwiftNakCode	消息被拒绝时由SWIFT网络返回的错误代码。
说明包含SWIFT网络在Negative Acknowledgment（NAK）中提供的具体错误代码（例如T26、T13）。该属性用于“SWIFT Error and Rework Analysis”仪表板，帮助分类技术故障并确定系统修复优先级。为什么重要提供网络层拒绝的技术根因。获取位置 MT 015/019 System Messages，或ACK/NAK中的Field 451。示例 T26H01T13G02
	SWIFT消息类型 SwiftMessageType	所使用的SWIFT消息类型，例如MT103、pacs.008。
说明表示支付指令的具体格式。常见类型包括客户转账使用的MT103，以及pacs.008等ISO 20022对应格式。用于“Payment Journey Variant Analysis”，比较传统MT格式与新ISO标准的处理效率。为什么重要区分支付流和处理标准（传统格式与ISO 20022）。获取位置 SWIFT Block 2（Application Header）中的Message Type字段。示例 MT103MT202pacs.008MT101
	UETR UniqueEndToEndReference	用于在SWIFT gpi中跟踪付款的Unique End-to-end Transaction Reference。
说明 UETR是一个36字符字符串，为付款在整个SWIFT网络中提供单一且不可变的参考标识。与内部ID不同，UETR会跨银行持续保留。该属性对于实现端到端可视性，以及将内部日志与外部SWIFT gpi状态更新关联起来至关重要。为什么重要跨不同机构跟踪跨境付款的行业标准。获取位置 SWIFT Block 3、Field 121。示例 b8c3f4a0-5d2a-4e1b-9c3d-1a2b3c4d5e6f123e4567-e89b-12d3-a456-426614174000
	交易货币 TransactionCurrency	表示付款币种的3字母ISO代码。
说明用于识别付款计价币种，例如USD、EUR、GBP。该属性对于“网络截止时间绩效”仪表板至关重要，因为截止时间因币种而异。它还支持“货币转换效率”分析，用于识别跨币种货币对。为什么重要决定路由规则、截止时间和结算通道。获取位置 SWIFT MT Field 32A（币种）或ISO 20022元素Ccy。示例 USDEURGBPJPY
	交易金额 TransactionAmount	付款指令的货币价值。
说明表示正在转移的本金金额。该数据从特定SWIFT字段中提取，例如MT103中的Field 32A。它对于“大额转账审批周期”仪表板至关重要，可按金额对付款进行分组，分析高风险、大额交易的审批延迟。为什么重要支持财务影响分析和按交易规模分组。获取位置 SWIFT MT Field 32A（金额）或ISO 20022元素IntrBkSttlmAmt。示例 15000.001250.501000000.0045.00
	处理用户 ProcessingUser	执行该活动的用户或系统代理。
说明标识负责执行活动的个人或自动化机器人，例如批准制裁命中的合规官员，或修复格式错误的操作员。该属性用于“Validation Error Heatmap”，帮助管理人员识别培训需求，或找出返工率较高的特定用户。为什么重要支持资源分析和人工瓶颈识别。获取位置系统审计日志、交易表中的“User ID”列。示例 SYSTEMJ.Doe合规机器人_01M.Smith
	收款方BIC BeneficiaryBic	收款机构的银行识别码。
说明标识支付的目标银行，是“Beneficiary Rejection Root Cause”仪表板的重要分析维度。按收款方BIC对失败情况分组后，银行可以识别经常因数据质量问题或特定格式要求而拒绝指令的交易对手。为什么重要对于分析交易对手表现和拒绝模式至关重要。获取位置 SWIFT Field 57A（Account With Institution）或58A（Beneficiary Institution）。示例 CITIUS33BARCGB22DRESDEFFHANDSEXX
	是否STP IsStp	表示支付是否无需人工干预的标记。
说明布尔指标。当案例中不包含“Error Identified”、“Payment Rejected”或人工“Modification”活动时为True。该指标直接用于计算“Straight-Through Processing Rate”KPI。这是衡量支付处理自动化成效的主要指标。为什么重要衡量流程效率和自动化运行状况的主要指标。获取位置根据案例中是否缺少特定负面或人工活动计算得出。示例 truefalse
	起息日 ValueDate	资金应可供收款方使用的日期。
说明表示支付消息中指定的结算日期。将该日期与实际“Payment Settled”时间戳进行比较，可支持“Settlement and Reconciliation Aging”仪表板分析。该属性反映支付紧迫程度，并用于衡量资金可用性是否符合服务级别协议。为什么重要对于流动性管理和结算及时性衡量至关重要。获取位置 SWIFT MT Field 32A（Date子字段）或ISO 20022 IntrBkSttlmDt。示例 2023-10-262023-11-01
	发起国家 OriginatingCountry	发起支付实体所在国家的代码。
说明表示付款请求发起的司法管辖区。这对于“Sanctions Screening Lead Time”仪表板至关重要，因为来自高风险司法管辖区的付款通常需要经过更严格、耗时更长的一致性检查。它支持对付款量和处理延迟进行地理分析。为什么重要合规风险分析和区域表现分析的关键维度。获取位置根据Sender BIC得出（国家代码为第5至第6位字符）。示例 USGBDEFR
	拒绝原因 RejectionReason	说明支付被拒原因的文本描述。
说明包含支付失败时的叙述或代码说明，通常位于SWIFT Field 72（Sender to Receiver Information）或退回消息中。该属性支持“Beneficiary Rejection Root Cause”分析。分析人员可以通过文本挖掘发现拒绝中的常见主题（例如“Invalid Account”“Beneficiary Name Mismatch”）。为什么重要为流程失败提供定性背景。获取位置退回消息（MT103 Return）中的SWIFT Field 72或79。示例收款人账户已关闭IBAN无效监管合规失败未知银行标识
	指令优先级 InstructionPriority	表示支付紧迫程度的优先级标记。
说明根据消息头信息得出（例如“Normal”与“Urgent”）。这支持在“High Value Transfer Approval Cycles”仪表板中进行分组分析，因为紧急支付通常需要加急审批流程。了解优先级分布有助于规划高峰期资源。为什么重要帮助区分标准支付与加急支付的SLA要求。获取位置 SWIFT Block 2中的Message Priority字段（例如N表示Normal，U表示Urgent）。示例正常紧急系统
	是否在截止时间前完成 MetCutOffTime	表示指令是否在网络截止时间前发送的标记。
说明通过比较“Payment Instruction Sent”时间与按币种区分的SWIFT网络截止时间计算得出。该属性支持“SWIFT Cut-off Adherence Rate”KPI。未能在截止时间前完成会导致结算延迟，进而影响流动性状况。为什么重要资金管理和流动性管理的重要运营KPI。获取位置通过比较时间戳与截止时间静态参考表计算得出。示例 truefalse
	是否跨境 IsCrossBorder	表示支付是否涉及不同国家的标记。
说明布尔属性。当Sender Country与Beneficiary Country不同时为True。该属性支持“Multi-Currency Handling Latency”KPI，将境内支付流与国际支付流分开分析。跨境支付通常具有更高的复杂度、成本和处理时长。为什么重要支付复杂度分析的基础分组维度。获取位置比较Sender BIC国家代码与Beneficiary BIC国家代码。示例 truefalse

必需建议可选

支付处理活动

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

6 建议 6 可选

	活动	说明
	付款已对账	将付款交易与Nostro/Vostro账户账单进行匹配。当对账系统将交易ID关联到账单行项目时，可推断该事件发生。
为什么重要完成财务闭环。此处延迟会影响现金可视性和可审计性。获取位置对账系统日志或核心银行系统中的“Matched”状态。采集比较对账状态事件类型 inferred
	付款已批准	指定人员或自动规则针对大额转账执行的最终授权操作。审批工作流标记设为true时，会明确记录这一事件。
为什么重要对于“大额转账审批周期”仪表板至关重要。用于识别大额流动性转移中的人工签批延迟。获取位置审批工作流引擎的审计追踪。采集执行审批操作时记录事件类型 explicit
	付款已结算	确认资金已记入收款人账户，通常对应gpi状态ACSC。可通过Tracker更新或确认消息明确获取。
为什么重要表示付款链路在业务上的成功结束。对于计算“结算到对账差距”至关重要。获取位置 SWIFT gpi Tracker（ACSC状态）或MT900/910确认消息。采集确认结算时记录事件类型 explicit
	创建付款请求	在内部银行系统或ERP中首次生成付款指令。付款订单首次分配唯一交易参考号（TRN或UETR）时，会明确记录这一事件。
为什么重要确定整个付款生命周期的起始时间。对于计算端到端处理时长，以及识别SWIFT网络介入前的上游延迟至关重要。获取位置核心银行系统（CBS）或Payment Hub中的交易表创建时间戳。采集创建交易记录时记录事件类型 explicit
	已发送付款指令	将格式化消息（MT103或ISO 20022 pacs.008）发送至SWIFT Gateway。当生成消息载荷并将其交给网络接口时，会明确记录这一事件。
为什么重要用于计算“SWIFT截止时间遵循率”。标志着流程从内部处理转入网络处理。获取位置 SWIFT Alliance Access（SAA）日志或网关发送日志。采集生成消息输出时记录事件类型 explicit
	通过制裁筛查	付款指令成功通过OFAC/制裁名单筛查的时间戳。可从合规系统日志中明确获取，也可通过“Screening Hold”状态解除来推断。
为什么重要对于“制裁筛查周期时间”仪表板至关重要。此处的延迟代表合规瓶颈，而非运营效率问题。获取位置 Compliance Filtering System日志或Payment Hub合规状态标记。采集筛查状态更新时记录事件类型 explicit
	付款已拒绝	收到收款行或中间银行返回的拒付消息（MT103 Return或gpi RJCT状态）。可从收到的SWIFT消息中明确获取。
为什么重要对于“收款行拒付根因”仪表板至关重要。用于识别账户关闭、路由数据错误等外部阻碍。获取位置收到的SWIFT消息（MT103 RET、pacs.004）或gpi状态RJCT。采集收到拒付消息时记录事件类型 explicit
	付款已转移	表示付款正在由中间银行处理的中间状态更新，通常对应gpi状态ACSP。可通过SWIFT gpi Tracker更新或状态消息获取。
为什么重要帮助了解代理行体系这一“黑箱”。对于分析完整端到端路径至关重要。获取位置 SWIFT gpi Tracker数据流或MT199/trck消息。采集收到Tracker更新时记录事件类型 explicit
	付款错误已解决	此前处于错误状态的交易完成修改并重新提交。当交易从“Repair”队列返回“Processing”或“Ready”状态时，可推断该事件发生。
为什么重要用于衡量人工返工团队的效率。此处耗时较长，通常说明培训不足或错误代码复杂。获取位置显示交易退出例外队列的Payment Hub状态日志。采集比较状态字段退出Repair 事件类型 inferred
	发现付款错误	交易因验证失败、NACK或拒付消息而被标记为需要修复的事件。当交易进入“Repair”“Correction”或“Exception”队列时，可推断该事件发生。
为什么重要用于跟踪“错误解决返工率”。此处频率过高会严重拉低直通式处理（STP）比例。获取位置 Payment Hub状态列或例外队列日志。采集将状态字段与Error/Repair进行比较事件类型 inferred
	完成付款详情验证	付款指令成功完成语法和格式检查，例如IBAN格式、BIC有效性检查。通常可通过交易状态从“Draft”变为“Validated”或“Ready for Auth”来推断。
为什么重要此处失败率较高，说明源头数据质量较差。用于衡量“验证首次通过率”KPI。获取位置 Payment Engine状态日志或验证历史表。采集比较前后状态字段事件类型 inferred
	已收到SWIFT ACK	收到SWIFT网络返回的技术确认（ACK），表明消息已被接受处理。可从网络接口日志中明确获取。
为什么重要确认消息已成功进入全球网络，并区分内部失败与实际网络传播失败。获取位置 SWIFT Gateway日志，查找ACK与NACK信号。采集收到网络ACK时记录事件类型 explicit

建议可选

提取指南

如何从SWIFT获取数据

步骤

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

配置

归档格式：XML（首选）或CSV（扁平文件）。强烈建议使用XML，以捕获嵌套的历史节点。
包含消息正文：是（必需，用于解析Amount、Currency和Beneficiary详细信息）。
包含历史记录/干预记录：是（必需，用于查看审批、验证和错误）。
日期范围：增量加载时配置滚动窗口，例如最近30天；首次加载历史数据时使用固定日期。
消息类型：筛选MT103、MT202、pacs.008、pacs.009和pacs.004/002（用于退回/状态）。
方向：同时包含Input（发送）和Output（接收），以捕获传入的拒绝或确认消息。
安全性：确保导出文件夹具有严格的读写权限，因为这些文件包含实时支付数据。

a 示例查询 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 headers）封装；如已封装，请在处理前移除。
配置脚本路径：打开提供的Python脚本。将INPUT_DIRECTORY变量更新为raw_swift_files文件夹路径，并更新生成CSV文件的OUTPUT_FILE路径。
定义正则表达式模式：脚本使用正则表达式分离各数据块。Block 1包含发送方BIC和序列号。Block 2包含Message Type（例如I103、O103）和接收方BIC。Block 4包含业务数据标签（例如：20：TRN、：32A：Value/Amount）。
将消息类型映射到活动：脚本包含映射字典。MT103表示“Payment Instruction Sent”。MT012（System ACK）表示“SWIFT ACK Received”。MT900/910表示“Payment Settled”。带有特定状态代码（ACSP、RJCT）的MT199/299映射为“Payment Transferred”或“Payment Rejected”。
提取属性：解析器遍历指定标签。：20：映射为Payment Transaction ID；：121：映射为UETR；：32A：解析为日期、货币和金额。
处理内部状态：对于“Payment Request Created”或“Payment Approved”等通常不会出现在SWIFT文件中的内部ERP事件，如果没有明确日志，脚本会根据Block 1中的消息创建时间推导这些事件；如果组织在字段：72：（Sender to Receiver Info）中记录审批信息，则查找其中的特定文本代码。
执行提取：通过命令行运行脚本：python swift_parser.py。脚本遍历所有文件，解析有效的SWIFT数据块，并跳过格式错误的行。
后处理验证：打开生成的CSV。检查EventTimestamp格式，建议使用ISO 8601格式。确保每一行都填充PaymentTransactionId，以便进行案例分组。
数据转换：如果ValueDate以YYMMDD格式输出，脚本会将其转换为YYYY-MM-DD。确保TransactionAmount中的小数分隔符统一使用点号或逗号。
最终检查：按特定UETR筛选，查看从Instruction Sent到ACK再到Settlement的端到端流程。确认NAK（MT019）显示为“Payment Error Identified”。
上传到ProcessMind：导入CSV。将PaymentTransactionId映射为Case ID，将ActivityName映射为Activity，将EventTimestamp映射为Timestamp。

配置

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

a 示例查询 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的支付交易，并为每个事件保留源系统名称。
提取以下内部生命周期事件：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和Payment Reconciled。每个源事件都必须转换为单独的事件日志行。不要依赖ProcessMind推断活动。
从内部支付记录中提取UETR，并针对每个UETR查询已授权的SWIFT gpi Tracker API。根据您所在环境提供的API文档配置请求URL、身份验证标头、请求参数、分页方式和响应字段。保存原始响应，以便审计。
将适用的gpi tracker状态转换为明确的事件行。脚本将ACSP映射为Payment Transferred，将RJCT映射为Payment Rejected，将ACSC映射为Payment Settled。如果您获得授权的响应使用不同的状态名称或时间戳，请根据文档中的响应架构更新状态映射。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检查点。
内部源筛选条件：根据您的系统配置公司、法人实体、分支机构、支付产品、单据类型、消息类型、货币和状态筛选条件。在确认源架构前，使用“[Configure based on your system]”等占位符。
活动映射：为内部状态和队列维护明确映射。例如，Draft映射为Payment Details Validated，Validated或Ready for Auth也可作为该活动；Screening Hold release映射为Sanctions Screening Passed；Repair或Exception queue entry映射为Payment Error Identified；return to Processing或Ready for映射为Payment Error Resolved。
gpi状态映射：将文档规定的跟踪器状态映射为明确活动。本实现将ACSP映射为Payment Transferred，将RJCT映射为Payment Rejected，将ACSC映射为Payment Settled。只有在授权API文档确认状态含义后，才扩展映射。
时间戳策略：尽可能将时间戳统一为UTC。定义源事件时间、跟踪器事件时间、消息接收时间和提取时间之间的优先级。LastDataUpdate必须表示提取或刷新时间，而不是业务事件时间。
速率限制和重试：配置请求超时、重试次数、指数退避、HTTP状态处理和分页。遵守SWIFT服务限制，避免不受控的并行请求。
性能：按UETR缓存响应，批量处理UETR，写入检查点，并在支持时采用增量提取。避免针对相同UETR和日期范围重复查询跟踪器。
数据质量：要求包含PaymentTransactionId、ActivityName、EventTimestamp、SourceSystem和LastDataUpdate。源数据未提供建议属性时允许NULL值，但应报告其填充率。
可审计性：将原始API响应、请求时间戳、响应状态、UETR和提取运行标识符持久化到获批准的审计位置。遵守支付数据所需的保留和安全规则。
前置条件：确认可以访问内部支付生命周期、合规、SWIFT网关、对账和消息接口日志，并确认已获得SWIFT gpi Tracker API许可证以及所需证书或网络白名单配置。

a 示例查询 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归档导出

步骤

配置

a 示例查询 python

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

步骤

配置

a 示例查询 python

SWIFT gpi API连接器

步骤

配置

a 示例查询 python

准备开始了吗？

立即优化您的SWIFT支付处理

您的支付处理数据模板

您的支付处理数据模板

支付处理属性

支付处理活动

提取指南

SWIFT Alliance Access归档导出

步骤

配置

a 示例查询 python

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

步骤

配置

a 示例查询 python

SWIFT gpi API连接器

步骤

配置

a 示例查询 python

准备开始了吗？

立即优化您的SWIFT支付处理

我们重视您的隐私