決済処理データテンプレート

SWIFT

決済処理の属性 (19) 決済処理のアクティビティ (12) 抽出ガイド (3)

決済処理データテンプレート

このテンプレートは、開始から最終settlementまでのSWIFT取引ワークフローをマッピングするための構造的な基盤です。決済ライフサイクルを詳細に把握するために必要な、具体的なデータ項目と業務上のマイルストーンを示します。このガイドを使ってデータを準備することで、隠れた非効率やコンプライアンスリスクを効果的に明らかにできます。

詳細な取引分析に推奨される属性
SWIFTプロセスの主要なアクティビティとマイルストーン
金融データ抽出に関する技術ガイダンス

イベントログを初めて扱いますか。学ぶプロセスマイニングのイベントログを作成する方法.

テンプレートをダウンロード

決済処理の属性

SWIFTの取引ライフサイクルを詳細に分析するため、イベントログに含めることを推奨するデータ項目です。

5 必須 9 推奨 5 任意

	名前	説明
	アクティビティ名 ActivityName	発生したプロセスステップまたはイベントの名称です。
説明この属性は、システムログに記録された具体的なアクションまたはステータス変更を示します。例として、「Payment Request Created」、「Sanctions Screening Passed」、「SWIFT ACK Received」などがあります。プロセスマップの主要なディメンションであり、可視化上のノードを定義します。分析担当者は、決済に対して実行された処理の順序を把握できます。重要な理由プロセスの「何を」を定義し、プロセスフローとバリエーションを可視化できます。入手先取引ログ、監査証跡、またはメッセージタイプ（例：MT103 Sent）から生成します。例決済依頼の作成決済指図を送信SWIFT ACKを受信決済承認済み
	イベントタイムスタンプ EventTimestamp	アクティビティが発生した正確な日時です。
説明この属性は、イベントが発生した正確な時点を記録します。アクティビティを時系列に並べ、リードタイム、サイクルタイム、処理量など、時間に基づくすべての指標を計算するために使用します。正確なタイムスタンプは、「Network Cut Off Time Performance」を分析し、承認経路のボトルネックを特定するうえで欠かせません。重要な理由タイムスタンプは、プロセスマイニングにおける時間分析とパフォーマンス指標の基盤です。入手先システム監査ログ、メッセージ作成タイムスタンプ、またはデータベースの取引時刻です。例 2023-10-25T08:30:15Z2023-10-25T14:45:00Z2023-10-26T09:15:22Z
	ソースシステム SourceSystem	イベントデータの発生元となったシステムの名称です。
説明イベントログを生成したソフトウェアまたはプラットフォームを特定します。決済プロセスでは、基幹銀行システム、決済ゲートウェイ、またはSWIFTインターフェースそのものが該当します。この属性はデータの系譜を確認し、複数の異なる銀行アプリケーションからデータを統合する際の問題をデバッグするのに役立ちます。重要な理由データの発生元に関するコンテキストを提供し、複数システムにまたがるプロセスマイニングに欠かせません。入手先 ETL処理中にハードコードするか、システム識別子から抽出します。例 SWIFT Alliance Access基幹銀行システム支払いエンジン制裁対象者フィルター
	最終データ更新 LastDataUpdate	レコードが最後に抽出または更新された時点のタイムスタンプです。
説明データが最後にプロセスマイニングツールへ読み込まれた時点を示します。データの鮮度を判断し、ダッシュボードに決済業務の最新状態を反映させるために欠かせません。レポート作成パイプラインの遅延を明らかにし、データの信頼性を高めます。重要な理由表示されたデータの最新性と信頼性を利用者が理解できるようにします。入手先 ETL処理の実行タイムスタンプです。例 2023-10-27T12:00:00Z2023-11-01T06:00:00Z
	決済取引ID PaymentTransactionId	エンドツーエンドの決済ケースを表す固有識別子です。
説明この属性は、プロセスマイニングにおける中心的なケースキーです。1件の決済指図に関連するすべてのアクティビティを、最初の依頼からvalidation、SWIFT送信、最終settlementまでまとめます。SWIFTのコンテキストでは、銀行システム間で連続性を確保するため、Transaction Reference Number（TRN）またはUnique End-to-end Transaction Reference（UETR）に対応付けることが一般的です。プロセスインスタンスの経路を再構築するために使用し、すべてのバリアント分析とサイクルタイム計算に欠かせません。重要な理由固有識別はプロセスマイニングの基盤です。分散したイベントをつなぎ合わせ、一貫したプロセスビューを作成できます。入手先メッセージヘッダー内のSWIFT Field 20（TRN）またはField 121（UETR）です。例 TRN-20231025-883954392-882-101UETR-9982-1123-5521PAY-US-EU-9912
	STPかどうか IsStp	決済に手作業が必要なかったかどうかを示すフラグ。
説明ケースに「Error Identified」、「Payment Rejected」、または手作業による「Modification」アクティビティが含まれない場合にTrueとなるブール指標です。「Straight-Through Processing Rate」KPIを直接算出します。決済処理における自動化の成否を測る主要指標です。重要な理由プロセス効率と自動化の健全性を測る主要指標です。入手先ケース内に特定のネガティブまたは手作業のアクティビティがないことに基づいて算出します。例 truefalse
	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（アプリケーションヘッダー）のメッセージタイプフィールド。例 MT103MT202pacs.008MT101
	UETR UniqueEndToEndReference	SWIFT gpiで追跡するためのUnique End-to-end Transaction Referenceです。
説明 UETRは36文字の文字列で、SWIFTネットワーク全体を通じて決済を一意に識別する、変更されない参照番号です。社内IDとは異なり、銀行をまたいで維持されます。エンドツーエンドの可視性を確保し、社内ログと外部のSWIFT gpiステータス更新を関連付けるうえで重要な属性です。重要な理由異なる金融機関をまたぐ国際決済の追跡における標準的な識別子です。入手先 SWIFT Block 3、Field 121です。例 b8c3f4a0-5d2a-4e1b-9c3d-1a2b3c4d5e6f123e4567-e89b-12d3-a456-426614174000
	処理担当ユーザー ProcessingUser	アクティビティを実行したユーザーまたはシステムエージェント。
説明コンプライアンス担当者による制裁検知の承認や、オペレーターによる形式エラーの修正など、アクティビティを実行した個人または自動ボットを特定します。この属性は「Validation Error Heatmap」を支え、管理者がトレーニングの必要性や、再処理率の高い特定ユーザーを把握できるようにします。重要な理由リソース分析と手作業によるボトルネックの特定に役立ちます。入手先システム監査ログ、取引テーブルの「User ID」列。例 SYSTEMJ.DoeCompliance_Bot_01M.Smith
	取引通貨 TransactionCurrency	決済通貨を示す3文字のISOコードです。
説明決済の通貨（例：USD、EUR、GBP）を示します。「Network Cut Off Time Performance」ダッシュボードに欠かせません。カットオフ時刻は通貨ごとに異なるためです。また、通貨間の組み合わせを特定し、「Currency Conversion Efficiency」分析にも役立ちます。重要な理由ルーティングルール、カットオフ時刻、settlement経路を決定します。入手先 SWIFT MT Field 32A（Currency）またはISO 20022要素Ccyです。例 USDEURGBPJPY
	取引金額 TransactionAmount	決済指図の金銭的価値です。
説明送金される元本額を示します。このデータは、SWIFTの特定フィールド（例：MT103のField 32A）から抽出します。「High Value Transfer Approval Cycles」ダッシュボードに欠かせません。決済を金額別に分類し、リスクと金額が大きい取引の承認遅延を分析できます。重要な理由財務的な影響の分析と、取引規模によるセグメント化が可能になります。入手先 SWIFT MT Field 32A（Amount）またはISO 20022要素IntrBkSttlmAmtです。例 15000.001250.501000000.0045.00
	受取人BIC BeneficiaryBic	受取金融機関のBank Identifier Code。
説明決済の送金先銀行を特定します。「Beneficiary Rejection Root Cause」ダッシュボードにおける重要な分析軸です。 Beneficiary BICごとに失敗をグループ化すると、データ品質の問題や特定の形式要件によって指示を頻繁に拒否する取引先を特定できます。重要な理由取引先のパフォーマンスと拒否パターンの分析に不可欠です。入手先 SWIFT Field 57A（Account With Institution）または58A（Beneficiary Institution）。例 CITIUS33BARCGB22DRESDEFFHANDSEXX
	受取日 ValueDate	受取人が資金を利用できるようになる日付。
説明決済メッセージで指定された決済日を示します。この日付と実際の「Payment Settled」タイムスタンプを比較することで、「Settlement and Reconciliation Aging」ダッシュボードで分析できます。決済の緊急度を示し、資金利用可能時期に関するサービスレベル合意の遵守状況を測定するために使用します。重要な理由流動性管理と決済の適時性の測定に不可欠です。入手先 SWIFT MT Field 32A（日付サブフィールド）またはISO 20022 IntrBkSttlmDt。例 2023-10-262023-11-01
	国際送金かどうか IsCrossBorder	決済に異なる国が関係するかどうかを示すフラグ。
説明送金元国と受取人国が異なる場合にTrueとなるブール属性です。「Multi-Currency Handling Latency」KPIで、国内フローと国際フローを区別するために使用します。国際送金は通常、複雑さ、コスト、処理時間が大きくなります。重要な理由決済の複雑さを分析するための基本的なセグメントです。入手先送金元BICの国コードと受取人BICの国コードを比較します。例 truefalse
	拒否理由 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のメッセージ優先度フィールド（例：NはNormal、UはUrgent）。例通常緊急システム
	締切時刻内に送信 MetCutOffTime	指示がネットワークの締切時刻より前に送信されたかどうかを示すフラグ。
説明「Payment Instruction Sent」の時刻と、通貨ごとに定められたSWIFTネットワークの締切時刻を比較して算出するブールフラグです。「SWIFT Cut-off Adherence Rate」KPIを支えます。締切時刻に間に合わないと決済が遅れ、流動性ポジションに影響します。重要な理由資金管理と流動性管理における重要な業務KPIです。入手先タイムスタンプと締切時刻の静的な参照テーブルを比較して算出します。例 truefalse
	送金元国 OriginatingCountry	決済を開始した組織の国コード。
説明決済依頼が発生した法域を示します。「Sanctions Screening Lead Time」ダッシュボードにおいて重要です。高リスク法域からの決済では、より厳格で時間のかかるコンプライアンス確認が行われることが多いためです。決済量と処理遅延を地理的に分析できます。重要な理由コンプライアンスリスク分析と地域別パフォーマンスにおける主要な分析軸です。入手先送金元BICから取得します（国コードの5～6文字目）。例 USGBDEFR

必須推奨任意

決済処理のアクティビティ

正確なプロセスディスカバリーとボトルネックの特定に向けて、イベントログに記録すべき主要なプロセス手順と節目です。

6 推奨 6 任意

	アクティビティ	説明
	制裁審査に合格	決済指図がOFAC／制裁リストの審査を正常に通過した時刻です。コンプライアンスシステムのログから明示的に取得するか、「Screening Hold」ステータスが解除された時点から推定します。
重要な理由「Sanctions Screening Lead Time」ダッシュボードに欠かせません。ここでの遅延は、業務運用上の非効率ではなく、コンプライアンス上のボトルネックを示します。入手先 Compliance Filtering SystemのログまたはPayment Hubのコンプライアンスステータスフラグです。取得審査ステータスの更新時に記録イベントタイプ explicit
	決済を照合	決済取引とNostro／Vostro口座の台帳明細を突合することです。照合システムが取引IDを明細行に関連付けた時点から推定します。
重要な理由財務処理を完結させます。ここでの遅延は、資金状況の把握と監査可能性に影響します。入手先照合システムのログまたはCore Banking Systemにおける「Matched」ステータスです。取得照合ステータスを比較イベントタイプ inferred
	決済依頼の作成	社内銀行システムまたはERPで決済指図が最初に生成されることです。決済注文に固有の取引参照番号（TRNまたはUETR）が初めて割り当てられた時点を、明示的に記録します。
重要な理由決済ライフサイクル全体の開始時刻を確定します。エンドツーエンドの処理時間を計算し、SWIFTネットワークが関与する前の上流工程の遅延を特定するために欠かせません。入手先 Core Banking System（CBS）またはPayment Hubにおける取引テーブルの作成タイムスタンプです。取得取引レコードの作成時に記録イベントタイプ explicit
	決済完了	受取人の口座に資金が入金されたことの確認です（多くの場合、gpiステータスACSC）。Trackerの更新または確認メッセージから明示的に取得します。
重要な理由決済チェーンが機能上正常に完了したことを示します。「Settlement to Reconciliation Gap」の計算における主要項目です。入手先 SWIFT gpi Tracker（ACSCステータス）またはMT900／910の確認メッセージです。取得 settlementの確認時に記録イベントタイプ explicit
	決済承認済み	高額送金について、指定された担当者または自動ルールが行う最終承認アクションです。承認ワークフローのフラグがtrueに設定された時点を、明示的に記録します。
重要な理由「High Value Transfer Approval Cycles」ダッシュボードの主要項目です。大規模な資金移動における手動承認の遅延を特定します。入手先承認ワークフローエンジンの監査証跡です。取得承認アクションの実行時に記録イベントタイプ explicit
	決済指図を送信	形式化されたメッセージ（MT103またはISO 20022 pacs.008）をSWIFTゲートウェイへ送信することです。メッセージペイロードが生成され、ネットワークインターフェースへ引き渡された時点を、明示的に記録します。
重要な理由「SWIFT Cut-off Adherence Rate」の計算に使用します。社内処理からネットワーク処理へ移行したことを示します。入手先 SWIFT Alliance Access（SAA）のログまたはゲートウェイ送信ログです。取得メッセージ出力時に記録イベントタイプ explicit
	SWIFT ACKを受信	メッセージが処理対象として受け付けられたことを示す、SWIFTネットワークからの技術的なAcknowledgment（ACK）を受信することです。ネットワークインターフェースのログから明示的に取得します。
重要な理由メッセージがグローバルネットワークに正常に入ったことを確認します。社内の失敗と、実際のネットワーク伝送を区別できます。入手先 SWIFTゲートウェイのログ（ACKとNACKのシグナルを確認）です。取得ネットワークACKの受信時に記録イベントタイプ explicit
	決済エラーを特定	validationの失敗、NACK、または却下メッセージにより、取引が修復対象としてフラグ付けされるイベントです。取引が「Repair」、「Correction」、または「Exception」キューに入った時点から推定します。
重要な理由「Error Resolution Rework Rate」を追跡します。ここでの発生頻度が高いと、ストレートスルー処理（STP）率が大きく低下します。入手先 Payment Hubのステータス列または例外キューのログです。取得ステータスフィールドをError／Repairと比較イベントタイプ inferred
	決済エラーを解消	以前エラー状態だった取引を正常に修正し、再送信することです。取引が「Repair」キューから「Processing」または「Ready」状態に移った時点から推定します。
重要な理由手動で手戻り対応を行うチームの効率を測定します。ここで時間が長い場合、トレーニング不足や複雑なエラーコードが考えられます。入手先例外キューからの退出を示すPayment Hubのステータスログです。取得ステータスフィールドのRepairからの退出を比較イベントタイプ inferred
	決済を却下	受取人銀行または中継銀行から却下メッセージ（MT103 ReturnまたはgpiステータスRJCT）を受信することです。受信したSWIFTメッセージから明示的に取得します。
重要な理由「Beneficiary Rejection Root Cause」ダッシュボードに欠かせません。口座閉鎖や誤ったルーティングデータなど、外部要因による障害を特定します。入手先受信SWIFTメッセージ（MT103 RET、pacs.004）またはgpiステータスRJCTです。取得却下メッセージの受信時に記録イベントタイプ explicit
	決済を送金	決済が中継銀行で処理されていることを示す中間ステータス更新（多くの場合、gpiステータスACSP）です。SWIFT gpi Trackerの更新またはステータスメッセージから取得します。
重要な理由コルレス銀行業務という「ブラックボックス」を可視化します。エンドツーエンドの全行程を分析するために欠かせません。入手先 SWIFT gpi TrackerのデータフィードまたはMT199／trckメッセージです。取得 Tracker更新の受信時に記録イベントタイプ explicit
	決済詳細のvalidation完了	決済指図に対する構文および形式チェック（例：IBAN形式、BICの有効性）が正常に完了したことです。通常は、取引ステータスが「Draft」から「Validated」または「Ready for Auth」に変わった時点から推定します。
重要な理由ここで失敗率が高い場合、入力元のデータ品質に問題があることを示します。「Validation First-Pass Yield」KPIを測定します。入手先 Payment Engineのステータスログまたはvalidation履歴テーブルです。取得前後のステータスフィールドを比較イベントタイプ inferred

推奨任意

抽出ガイド

SWIFTからデータを取得する方法

ステップ

Archiving設定へのアクセス権を持つ管理者アカウントで、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（Emission）とOutput（Reception）の両方を含め、受信した拒否や確認を取得します。
セキュリティ：これらのファイルには実際の支払データが含まれるため、エクスポートフォルダーに厳格な読み取り／書き込み権限が設定されていることを確認します。

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ヘッダーなど）でラップされていないことを確認し、ラップされている場合は処理前に取り除きます。
スクリプトのパスを設定する：提供されたPythonスクリプトを開きます。INPUT_DIRECTORY変数をraw_swift_filesフォルダーに設定します。生成するCSVのOUTPUT_FILEパスも更新します。
正規表現パターンを定義する：スクリプトでは、ブロックを分離するために正規表現を使用します。Block 1には送信者BICとSequence Numberが含まれます。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」にマッピングします。
属性を抽出する：Parserは特定のタグを順に処理します。:20:はPayment Transaction IDにマッピングします。:121:はUETRにマッピングします。:32A:はDate、Currency、Amountに分解します。
内部状態を処理する：「Payment Request Created」や「Payment Approved」など、通常はSWIFTファイルに含まれない内部ERPイベントについては、明示的なログがない場合、Block 1のMessage Creation Timestampから導出します。または、組織が承認情報を記録している場合は、フィールド: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検索機能、社内の支払いシステムにおけるイベントソースへのアクセスが利用できることを確認します。正確なAPIベースURL、認証方式、レスポンススキーマ、レート制限、ページネーションのルールを、SWIFTサービスのドキュメントで確認します。エンドポイントのパスは、契約しているSWIFTサービスと導入環境によって異なるため、推測で設定しないでください。
必要な接続文字列、認証情報、日付範囲、会社固有のフィルターをスクリプトに設定します。初回抽出では、3～6か月の日付範囲を使用します。社内ソースのデータは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へ問い合わせます。リクエストURL、認証ヘッダー、リクエストパラメーター、ページネーション、レスポンスフィールドは、ご利用の環境向けに提供されたAPIドキュメントに従って設定します。監査に備えて、生のレスポンスを保存します。
gpi trackerの該当するステータスを明示的なイベント行に変換します。スクリプトでは、ACSPをPayment Transferred、RJCTをPayment Rejected、ACSCをPayment Settledにマッピングします。契約しているレスポンスで異なるステータス名やタイムスタンプが使われている場合は、文書化されたレスポンススキーマに基づいてステータスマッピングを更新します。APIが対応するステータスの発生を返さない場合は、イベントを作成しないでください。
すべてのタイムスタンプを一貫したタイムゾーン、できればUTCに統一します。監査上必要な場合は、元のタイムスタンプも保持します。金額を数値に変換し、取引通貨を保持し、Boolean形式のSTP値を統一します。ソースに属性がない場合は、NULL値を保持します。
PaymentTransactionId、ActivityName、EventTimestamp、利用可能な中で最も具体的なソースイベント識別子を使ってイベントを重複排除します。修復や再送の繰り返しなど、正当な複数回の発生は保持します。ただし、業務ルールでアクティビティごとに1回の発生のみが明示的に求められている場合を除きます。
すべての出力行にPaymentTransactionId、ActivityName、EventTimestamp、SourceSystem、LastDataUpdateが含まれていることを検証します。ActivityNameに文書化されたアクティビティ値のみが含まれ、ソースデータに存在する場合は12種類すべてのアクティビティが表現されていることを確認します。
生成したデータセットをUTF-8 CSVまたはProcessMindがサポートする別のイベントログ形式でエクスポートします。1イベントにつき1行とし、必須列と推奨列を含めます。PaymentTransactionIdをケース識別子、ActivityNameをアクティビティ、EventTimestampをイベントタイムスタンプにマッピングします。追加のイベント導出を期待せず、ファイルをProcessMindにアップロードします。

設定

APIエンドポイントとスキーマ：ご利用の環境向けに契約しているSWIFTドキュメントを参照し、SWIFT gpi Tracker APIの正確なベースURL、リソースパス、リクエストパラメーター、認証ヘッダー、ページネーション、レスポンスフィールド名を設定します。スクリプトでは、エンドポイントを意図的に推測していません。
認証情報と認可：承認済みのSWIFT API認証情報の仕組み、証明書またはトークンの要件、秘密情報の保存方法、内部ソースシステムの認可を使用します。本番環境の秘密情報をソースコードに直接記載しないでください。
日付範囲：まず3～6か月を対象にします。レポートの目的に応じて、イベントタイムスタンプまたは支払い作成日を使用します。データ量が多い場合は、より短い期間に分けて処理し、完了したUETRをチェックポイントに記録します。
内部ソースのフィルター：システムに応じて、会社、法人、支店、支払い商品、文書タイプ、メッセージタイプ、通貨、ステータスの各フィルターを設定します。ソーススキーマが確定するまでは、[Configure based on your system]などのプレースホルダーを使用します。
アクティビティマッピング：内部ステータスとキューの明示的なマッピングを管理します。例として、Payment Details ValidatedにはDraftからValidatedまたはReady for Authへの遷移、Sanctions Screening PassedにはScreening Holdの解除、Payment Error IdentifiedにはRepairまたはExceptionキューへの登録、Payment Error ResolvedにはProcessingまたはReady forへの復帰をマッピングします。
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 Parser

ステップ

設定

a サンプルクエリ python

SWIFT gpi API Connector

ステップ

設定

a サンプルクエリ python

準備はできましたか。

SWIFT決済処理を今すぐ最適化

決済処理データテンプレート

決済処理データテンプレート

決済処理の属性

決済処理のアクティビティ

抽出ガイド

SWIFT Alliance Accessアーカイブエクスポート

ステップ

設定

a サンプルクエリ python

生のSWIFT MTファイル用Python Parser

ステップ

設定

a サンプルクエリ python

SWIFT gpi API Connector

ステップ

設定

a サンプルクエリ python

準備はできましたか。

SWIFT決済処理を今すぐ最適化

プライバシーを大切にしています