> 継続的なプロセス改善 > 支払い処理 > SWIFT > データテンプレート

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

SWIFT

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

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

このテンプレートは、SWIFT取引の開始から最終決済までのワークフローをマッピングするための基盤となります。決済プロセスを詳しく把握するために必要なデータ項目と業務上の節目を示しています。このガイドに沿ってデータを準備すると、見えにくい非効率を特定し、コンプライアンスリスクを抑えやすくなります。

取引を詳しく分析するための推奨属性
SWIFTプロセスの主なアクティビティと節目
金融データの抽出に関する技術ガイド

イベントログを初めて作成する方はこちら：プロセスマイニング用イベントログの作成方法.

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

決済処理の属性

SWIFT取引の全体を詳しく分析するには、イベントログに次のデータ項目を含めることをおすすめします。

5 必須 9 推奨 5 任意

	名前	説明
	アクティビティ名 ActivityName	発生したプロセスの工程またはイベントの名前です。
説明この属性は、システムログに記録された具体的な操作やステータスの変化を示します。例として、「支払い依頼を作成」「制裁審査に合格」「SWIFTのACKを受信」などがあります。プロセスマップの主要な分析軸となり、可視化におけるノードを定義します。分析担当者は、支払いに対して行われた操作の順序を把握できます。重要な理由プロセスで「何が」行われたかを定義し、プロセスの流れやバリエーションを可視化します。入手先取引ログ、監査証跡、またはメッセージ種別（例：MT103送信）から取得。例支払い依頼を作成支払い指図を送信SWIFTのACKを受信支払いを承認
	イベントのタイムスタンプ EventTimestamp	アクティビティが発生した正確な日時です。
説明この属性は、イベントが発生した正確な時刻を記録します。アクティビティを時系列に並べ、リードタイム、サイクルタイム、処理量など、時間に基づくすべての指標を算出するために使います。正確なタイムスタンプは、「ネットワークの締め切り時刻の遵守状況」を分析し、承認経路のボトルネックを特定するうえで欠かせません。重要な理由タイムスタンプは、プロセスマイニングにおける時間分析とパフォーマンス指標の基礎です。入手先システム監査ログ、メッセージ作成時のタイムスタンプ、またはデータベースの取引時刻。例 2023-10-25T08:30:15Z2023-10-25T14:45:00Z2023-10-26T09:15:22Z
	ソースシステム SourceSystem	イベントデータの発生元となるシステムの名前です。
説明イベントログを生成したソフトウェアまたはプラットフォームを特定します。支払いプロセスでは、コアバンキングシステム、支払いゲートウェイ、またはSWIFTインターフェースが該当します。この属性はデータの来歴を確認する際に役立ちます。また、複数の異なる銀行アプリケーションのデータを統合する際に問題を調査する手がかりになります。重要な理由データの発生元を把握でき、複数システムにまたがるプロセスマイニングに欠かせないコンテキストを提供します。入手先 ETL処理時に固定値を設定するか、システム識別子から抽出。例 SWIFT Alliance AccessコアバンキングシステムPayment Engine制裁フィルター
	最終データ更新 LastDataUpdate	レコードを最後に抽出または更新した時刻です。
説明データがプロセスマイニングツールに最後に読み込まれた時刻を示します。データの鮮度を確認し、ダッシュボードに支払い業務の最新状況が反映されているかを確かめるうえで重要です。レポート作成のデータ反映にかかる時間を明らかにすることで、ユーザーがデータを信頼できるようにします。重要な理由提示されるデータの最新性と信頼性を、ユーザーが把握できるようにします。入手先 ETL処理の実行タイムスタンプ。例 2023-10-27T12:00:00Z2023-11-01T06:00:00Z
	支払い取引ID PaymentTransactionId	支払い案件を最初から最後まで識別する固有のIDです。
説明この属性は、プロセスマイニングにおける中心的な案件キーです。1件の支払い指図に関連するすべてのアクティビティを、最初の依頼から検証、SWIFT送信、最終決済までまとめます。SWIFTでは、銀行システム間で一貫して追跡できるよう、取引参照番号（TRN）または一意のエンドツーエンド取引参照番号（UETR）に対応付けることがよくあります。プロセスインスタンスの経路を再構成するために使われ、バリアント分析やサイクルタイムの算出に欠かせません。重要な理由固有の識別情報はプロセスマイニングの基盤です。散在するイベントをつなぎ合わせ、プロセス全体を把握できます。入手先メッセージヘッダーのSWIFTフィールド20（TRN）またはフィールド121（UETR）。例 TRN-20231025-883954392-882-101UETR-9982-1123-5521PAY-US-EU-9912
	STP処理 IsStp	支払いに手作業での介入が不要だったかを示すフラグです。
説明ケースに「エラー特定」「支払い拒否」、または手作業による「変更」のアクティビティが含まれない場合にTrueとなる真偽値です。「ストレートスルー処理率」KPIを直接算出します。支払い処理の自動化の成果を測る主要指標です。重要な理由プロセス効率と自動化の状況を測る主要な指標です。入手先ケースに特定の否定的なアクティビティや手作業のアクティビティがないことを基に算出します。例 truefalse
	SWIFT NAKコード SwiftNakCode	メッセージが拒否された場合に、SWIFTネットワークから返されるエラーコードです。
説明 SWIFTネットワークから否定応答（NAK）とともに返される具体的なエラーコード（例：T26、T13）を含みます。「SWIFTエラーと手戻り分析」ダッシュボードで使用し、技術的な障害を分類して、システム修正の優先順位付けに役立てます。重要な理由ネットワークレベルで拒否された原因を技術面から特定できます。入手先 MT 015／019システムメッセージ、またはACK／NAKのフィールド451です。例 T26H01T13G02
	SWIFTメッセージタイプ SwiftMessageType	使用されたSWIFTメッセージの種類です（例：MT103、pacs.008）。
説明支払い指図の形式を示します。一般的な形式には、顧客送金に使われるMT103や、これに相当するISO 20022のpacs.008などがあります。「支払い経路のバリアント分析」では、従来のMT形式と新しいISO規格の処理効率を比較するために使用します。重要な理由支払いの流れと処理規格（従来形式とISO 20022）を区別します。入手先 SWIFTブロック2（アプリケーションヘッダー）のメッセージタイプフィールドです。例 MT103MT202pacs.008MT101
	UETR UniqueEndToEndReference	SWIFT gpiで追跡するための、一意のエンドツーエンド取引参照番号です。
説明 UETRは36文字の文字列で、SWIFTネットワーク全体を通じて支払いを一意に識別する、変更されない参照番号です。内部IDとは異なり、銀行間でも同じUETRが維持されます。エンドツーエンドの可視性を確保し、内部ログと外部のSWIFT gpiステータス更新を関連付けるうえで重要な属性です。重要な理由異なる金融機関をまたぐ国際送金の追跡における標準的な識別子です。入手先 SWIFTブロック3、フィールド121です。例 b8c3f4a0-5d2a-4e1b-9c3d-1a2b3c4d5e6f123e4567-e89b-12d3-a456-426614174000
	処理ユーザー ProcessingUser	アクティビティを実行したユーザーまたはシステムエージェントです。
説明アクティビティを担当した個人または自動ボットを特定します。たとえば、制裁対象との照合結果を承認したコンプライアンス担当者や、書式エラーを修正したオペレーターなどです。この属性は「検証エラーヒートマップ」に使用され、管理者は研修が必要な領域や、手戻り率の高い特定のユーザーを把握できます。重要な理由リソース分析や、手作業によるボトルネックの特定に役立ちます。入手先システム監査ログ、取引テーブルの「ユーザーID」列です。例システムJ.DoeCompliance_Bot_01M.Smith
	取引通貨 TransactionCurrency	支払い通貨を示す3文字のISOコードです。
説明支払いに使用される通貨を示します（例：USD、EUR、GBP）。通貨ごとに締切時刻が異なるため、「ネットワーク締切時刻のパフォーマンス」ダッシュボードで重要な属性です。また、異なる通貨間の組み合わせを特定し、「通貨換算効率」の分析にも役立ちます。重要な理由ルーティングルール、締切時刻、決済経路の判定に使用します。入手先 SWIFT MTのフィールド32A（通貨）、またはISO 20022の要素Ccyです。例 USDEURGBPJPY
	取引金額 TransactionAmount	支払い指図の金額です。
説明送金される元本の金額を示します。SWIFTの特定フィールド（例：MT103のフィールド32A）から抽出します。「高額送金の承認サイクル」ダッシュボードに欠かせない項目です。支払いを金額別に分類し、リスクと金額の高い取引における承認の遅れを分析できます。重要な理由取引金額に応じた財務影響の分析と分類に役立ちます。入手先 SWIFT MTフィールド32A（金額）またはISO 20022要素IntrBkSttlmAmt。例 15000.001250.501000000.0045.00
	受取人銀行BIC BeneficiaryBic	受取金融機関の銀行識別コードです。
説明支払い先の銀行を特定します。「受取人による拒否の根本原因」ダッシュボードで重要な分析軸です。受取人銀行BICごとに失敗を分類すると、データ品質の問題や特定の書式要件を理由に指図を頻繁に拒否する取引先を特定できます。重要な理由取引先のパフォーマンスや拒否の傾向を分析するうえで欠かせません。入手先 SWIFTフィールド57A（口座保有機関）または58A（受取人金融機関）です。例 CITIUS33BARCGB22DRESDEFFHANDSEXX
	決済日 ValueDate	受取人が資金を利用できるようになる日付です。
説明支払いメッセージで指示された決済日を示します。この日付と実際の「支払い決済済み」タイムスタンプを比較すると、「決済と照合の経過時間」ダッシュボードで分析できます。支払いの緊急度を示すほか、資金の利用可能時期に関するサービスレベル合意の遵守状況を測定するために使用します。重要な理由流動性管理と決済の適時性の測定に欠かせません。入手先 SWIFT MTのフィールド32A（日付サブフィールド）、またはISO 20022のIntrBkSttlmDtです。例 2023-10-262023-11-01
	国境を越える支払い IsCrossBorder	支払いに複数の国が関係するかどうかを示すフラグです。
説明送金元の国と受取人の国が異なる場合にTrueとなる真偽値属性です。国内送金と国際送金を分け、「複数通貨処理の遅延」KPIを分析できます。国際送金は一般に、処理が複雑で、コストや処理時間も増加します。重要な理由支払いの複雑さを分析するための基本的な分類軸です。入手先送金元BICの国コードと受取人BICの国コードを比較します。例 truefalse
	拒否理由 RejectionReason	支払いが拒否された理由を説明するテキストです。
説明支払いが失敗した際の説明文またはコードの説明を含みます。通常、SWIFTフィールド72（送信者から受信者への情報）や返却メッセージに記録され、「受取人による拒否の根本原因」の分析に使用します。テキストマイニングにより、拒否理由に共通する傾向（例：「口座番号が無効」「受取人名が一致しない」）を見つけられます。重要な理由プロセスの失敗に関する定性的な背景情報を提供します。入手先 SWIFTフィールド72または返却メッセージ（MT103 Return）のフィールド79です。例受取人口座が閉鎖済みIBANが無効規制コンプライアンス違反銀行識別情報が不明
	指図の優先度 InstructionPriority	支払いの緊急度を示す優先度フラグです。
説明メッセージヘッダー（例：「通常」または「緊急」）から取得します。緊急の支払いでは承認を迅速に進める必要があることが多いため、「高額送金の承認サイクル」ダッシュボードで分類に使用できます。優先度の分布を把握すると、繁忙期のリソース計画に役立ちます。重要な理由通常の支払いと迅速処理の支払いで、SLA要件を区別できます。入手先 SWIFTブロック2のメッセージ優先度フィールドです（例：通常はN、緊急はU）。例通常緊急システム
	締切時刻に間に合ったか MetCutOffTime	指図がネットワークの締切時刻より前に送信されたかを示すフラグです。
説明「支払い指図の送信」時刻と、通貨ごとに定められたSWIFTネットワークの締切時刻を比較して算出する真偽値フラグです。「SWIFT締切遵守率」KPIの分析に使用します。締切時刻に間に合わないと決済が遅れ、流動性に影響します。重要な理由財務部門の資金管理と流動性管理に欠かせない業務KPIです。入手先タイムスタンプを、締切時刻の固定参照テーブルと比較して算出します。例 truefalse
	送金元の国 OriginatingCountry	支払いを開始した組織の国コードです。
説明支払い依頼が発生した法域を示します。高リスクの法域からの支払いは、厳格で時間のかかるコンプライアンスチェックの対象となることが多いため、「制裁スクリーニングのリードタイム」ダッシュボードで重要な属性です。支払い件数や処理遅延を地域別に分析できます。重要な理由コンプライアンスリスクの分析や地域別のパフォーマンスを把握するための重要な分析軸です。入手先送金元BICの国コード（5～6文字目）から取得します。例 USGBDEFR

必須推奨任意

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

プロセスを正確に把握し、ボトルネックを特定するために、イベントログに記録する主な工程と節目を示します。

6 推奨 6 任意

	アクティビティ	説明
	制裁審査に合格	支払い指図がOFAC／制裁リストの審査を通過した時刻です。コンプライアンスシステムのログから明示的に取得するか、「審査保留」ステータスが解除された時点から推定します。
重要な理由「制裁審査のリードタイム」ダッシュボードに欠かせない項目です。この段階の遅延は、業務効率の問題ではなく、コンプライアンス対応のボトルネックを示します。入手先コンプライアンスフィルタリングシステムのログ、またはPayment Hubのコンプライアンスステータスフラグ。記録審査ステータスの更新時に記録イベントタイプ explicit
	支払いを承認	高額送金について、指定された担当者または自動ルールが最終承認を行った状態です。承認ワークフローのフラグがtrueに設定された時点を明示的に記録します。
重要な理由「高額送金の承認サイクル」ダッシュボードに必要な項目です。多額の資金移動における手動承認の遅れを特定します。入手先承認ワークフローエンジンの監査証跡。記録承認操作の実行時に記録イベントタイプ explicit
	支払いを決済	受取人の口座に資金が入金されたことを確認した状態です（多くの場合、gpiステータスACSC）。Trackerの更新情報または確認メッセージから明示的に取得します。
重要な理由支払い処理の実質的な完了を示します。「決済から照合までの時間差」の算出に必要です。入手先 SWIFT gpi Tracker（ACSCステータス）またはMT900／910の確認メッセージ。記録決済の確認時に記録イベントタイプ explicit
	支払いを照合	支払い取引をNostro／Vostro口座の元帳明細と照合する工程です。照合システムが取引IDを明細項目に関連付けた時点から推定します。
重要な理由財務処理を完了させます。この段階の遅延は、資金状況の把握や監査可能性に影響します。入手先照合システムのログ、またはコアバンキングシステムの「照合済み」ステータス。記録照合ステータスを比較イベントタイプ inferred
	支払い依頼を作成	社内の銀行システムまたはERPで支払い指図を作成する最初の工程です。支払い指図に固有の取引参照番号（TRNまたはUETR）が初めて割り当てられた時点を、明示的に記録します。
重要な理由支払いライフサイクル全体の開始時刻を定めます。処理全体にかかる時間の算出や、SWIFTネットワークに接続する前の上流工程での遅延特定に欠かせません。入手先コアバンキングシステム（CBS）またはPayment Hubにおける取引テーブルの作成タイムスタンプ。記録取引レコードの作成時に記録イベントタイプ explicit
	支払い指図を送信	書式設定済みのメッセージ（MT103またはISO 20022 pacs.008）をSWIFTゲートウェイに送信する工程です。メッセージのペイロードが生成され、ネットワークインターフェースに引き渡された時点を明示的に記録します。
重要な理由「SWIFT締め切り時刻の遵守率」の算出に使用します。社内処理からネットワーク処理への移行を示します。入手先 SWIFT Alliance Access（SAA）のログまたはゲートウェイの送信ログ。記録メッセージの出力時に記録イベントタイプ explicit
	SWIFTのACKを受信	メッセージが処理対象として受理されたことを示す、SWIFTネットワークからの技術的な確認応答（ACK）を受信した状態です。ネットワークインターフェースのログから明示的に取得します。
重要な理由メッセージがグローバルネットワークに正常に到達したことを確認します。社内での失敗と、実際のネットワーク送信を区別できます。入手先 SWIFTゲートウェイのログ（ACKとNACKの信号を確認）。記録ネットワークACKの受信時に記録イベントタイプ explicit
	支払いエラーを特定	検証失敗、NACK、拒否メッセージなどにより、取引が修正対象としてフラグ付けされたイベントです。取引が「修正」「訂正」「例外」の待ち行列に入った時点から推定します。
重要な理由「エラー解消の手戻り率」を追跡します。この段階の発生頻度が高いと、ストレートスルー処理（STP）の比率が大きく低下します。入手先 Payment Hubのステータス列または例外処理の待ち行列ログ。記録ステータス項目が「エラー」または「修正」になったかを比較イベントタイプ inferred
	支払いエラーを解消	エラー状態だった取引を正常に修正し、再送信した状態です。取引が「修正」の待ち行列から「処理中」または「準備完了」に戻った時点から推定します。
重要な理由手作業による修正チームの効率を測定します。この段階に時間がかかる場合、トレーニング不足や複雑なエラーコードが原因である可能性があります。入手先例外処理の待ち行列からの退出を示すPayment Hubのステータスログ。記録ステータス項目が「修正」から退出したかを比較イベントタイプ inferred
	支払いを拒否	受取銀行または中継銀行から拒否メッセージ（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
	支払い情報を検証	支払い指図の構文と形式（例：IBAN形式、BICの有効性）のチェックが正常に完了した状態です。取引ステータスが「下書き」から「検証済み」または「承認待ち」に変わった時点から推定できることがよくあります。
重要な理由この段階で失敗率が高い場合、発生源のデータ品質に問題があることを示します。「検証の初回合格率」KPIを測定します。入手先 Payment Engineのステータスログまたは検証履歴テーブル。記録変更前後のステータス項目を比較イベントタイプ inferred

推奨任意

データ抽出ガイド

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

ステップ

アーカイブ設定にアクセスできる管理者アカウントで、SWIFT Alliance WebPlatformまたはAlliance Access GUIにログインします。
［System Management］メニューから［Archiving］を選択し、［Automated Archiving Configuration］を開きます。
新しい日次エクスポートプロファイルを設定します。抽出スクリプトからアクセスできる安全なネットワークパスを［Target Directory］に指定します。
コンテンツの範囲を指定します。［Messages］を選択し、［Include History］と［Include Interventions］にチェックを入れます。承認や検証などの監査証跡イベントを取得するために必要です。
出力形式を選択します。メッセージヘッダーと履歴イベントの階層関係をCSVより適切に保持できるXML（「SAA XML v2」または利用可能な最新スキーマ）を選びます。
プロセスマイニングに必要な過去の期間を日付範囲フィルターで指定するか、T-1データを毎日取得するよう設定します。
初回データの取得ではアーカイブエクスポートジョブを手動で実行するか、スケジュール済みのバッチジョブが完了するまで待ちます。出力先に.xmlファイルが生成されたことを確認します。
SAAのエクスポートデータは階層構造のXMLファイルですが、ProcessMindではフラットなCSVイベントログが必要です。変換スクリプトを実行してください。クエリセクションにあるPythonスクリプトでXMLを解析します。
スクリプトは各メッセージ要素を順に処理し、一意のエンドツーエンド取引参照番号（UETR）または取引参照番号（TRN）を抽出して、ケース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（返金／ステータス）を対象にします。
方向：受信時の拒否や確認を取得するため、入力（送信）と出力（受信）の両方を含めます。
セキュリティ：これらのファイルには実際の決済データが含まれるため、エクスポートフォルダーの読み取り／書き込み権限を厳格に設定します。

クエリの例 python

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

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

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

data_rows = []

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

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

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

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

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

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

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

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

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

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

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

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

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

ステップ

環境の準備：抽出サーバーにPython 3.8以降がインストールされていることを確認します。pipで必要なライブラリ（pandas、numpy）をインストールします。専用の作業ディレクトリを作成し、その中にraw_swift_filesというサブフォルダーを用意して、日次の.out、.txt、.finファイルをすべて配置します。
ファイル構造の確認：SWIFT FINの生メッセージはブロック構造（{1: Header} {2: Application} {4: Body}）です。区切り文字を確認してください。標準ファイルでは波括弧が使われます。ファイルが独自形式のエンベロープ（RJEヘッダーなど）で囲まれていないことを確認し、囲まれている場合は処理前に取り除きます。
スクリプトのパス設定：用意されているPythonスクリプトを開きます。INPUT_DIRECTORY変数をraw_swift_filesフォルダーのパスに変更します。出力CSVの保存先に合わせてOUTPUT_FILEのパスも更新します。
正規表現パターンの定義：スクリプトは正規表現を使ってブロックを抽出します。ブロック1には送信者BICとシーケンス番号、ブロック2にはメッセージタイプ（例：I103、O103）と受信者BIC、ブロック4には業務データのタグ（例：:20: TRN、:32A: Value／Amount）が含まれます。
メッセージタイプとアクティビティの対応付け：スクリプトには対応関係を定義する辞書があります。MT103は「決済指図の送信」、MT012（システムACK）は「SWIFT ACKの受信」、MT900／910は「決済完了」に対応します。特定のステータスコード（ACSP、RJCT）を含むMT199／299は、「決済の送金」または「決済の拒否」に対応します。
属性の抽出：パーサーは特定のタグを順に処理します。:20:はPayment Transaction IDに、:121:はUETRに対応付けます。:32A:は日付、通貨、金額に分解します。
内部ステータスの処理：「決済依頼の作成」や「決済の承認」など、通常はSWIFTファイルに含まれない社内ERPのイベントについては、明示的なログがない場合、ブロック1のメッセージ作成タイムスタンプから推定します。または、組織で承認を記録している場合は、フィールド:72:（送信者から受信者への情報）にある特定のテキストコードを確認します。
抽出の実行：コマンドラインからpython swift_parser.pyを実行します。スクリプトはすべてのファイルを処理し、有効なSWIFTブロックを解析して、形式に合わない行をスキップします。
処理後の検証：生成されたCSVを開きます。EventTimestampが推奨形式のISO 8601になっていることを確認します。ケースをグループ化できるよう、すべての行にPaymentTransactionIdが入力されていることも確認します。
データ変換：「ValueDate」がYYMMDD形式の場合、スクリプトでYYYY-MM-DD形式に変換します。「TransactionAmount」の小数点記号（ピリオドまたはカンマ）が統一されていることを確認します。
最終確認：特定のUETRで絞り込み、指図の送信からACK、決済完了までの流れを確認します。NAK（MT019）が「決済エラーの検出」として記録されていることを確認します。
ProcessMindへのアップロード：CSVをインポートします。PaymentTransactionIdをケースIDに、ActivityNameをアクティビティに、EventTimestampをタイムスタンプに対応付けます。

設定

入力ディレクトリ：FIN／MTの生フラットファイルが格納されたフォルダーのパスです。読み取り権限が付与されていることを確認します。
日付形式の解析：SWIFTの生データの日付はYYMMDD形式です。スクリプトでは世紀を20xx年とみなします。
メッセージタイプのフィルター：スクリプトはカテゴリ1（決済）、カテゴリ9（明細書）、一般的なシステムメッセージ（012、019）を対象とします。カテゴリ2（金融機関間送金）が必要な場合は、RELEVANT_MT_TYPESリストを変更します。
エンコーディング：既定値は「utf-8」です。ファイルに旧式のエンコーディング文字が含まれる場合は、「latin-1」または「ascii」に変更します。
バッチ処理：データセットが1GBを超える場合、メモリ不足を避けるため、ジェネレーター方式でファイルを1つずつ処理します。
正規表現の調整：厳密な正規表現では標準FIN形式{1:...}{2:...}{4:...}を想定しています。ファイルで「$」など別の区切り文字が使われている場合は、BLOCK_PATTERN変数を更新します。

クエリの例 python

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

ステップ

SWIFT gpi Tracker APIの契約、API認証情報、UETR検索機能、社内決済システムのイベントソースへのアクセスが利用可能であることを確認します。正確なAPIベースURL、認証方式、レスポンススキーマ、レート制限、ページネーションのルールは、SWIFTサービスの文書で確認してください。エンドポイントのパスは契約中のSWIFTサービスや導入環境によって異なるため、推測しないでください。
必要な接続文字列、認証情報、日付範囲、会社固有のフィルターをスクリプトに設定します。初回の抽出では3～6か月分を対象にします。社内データソースからはTRNまたはUETRを持つ決済取引に絞り、各イベントのソースシステム名を保持します。
決済依頼の作成、決済詳細の検証、制裁スクリーニングの通過、決済の承認、決済指図の送信、SWIFT ACKの受信、決済エラーの検出、決済エラーの解決、決済の拒否、決済完了、決済の照合に関する社内ライフサイクルイベントを抽出します。各ソースイベントを個別のイベントログ行に変換してください。ProcessMindがアクティビティを推定することを前提にしないでください。
社内決済記録からUETRを抽出し、UETRごとに契約中のSWIFT gpi Tracker APIを照会します。利用環境向けに提供されたAPI文書に従って、リクエストURL、認証ヘッダー、リクエストパラメーター、ページネーション、レスポンスフィールドを設定します。監査に備えて、生のレスポンスを保存します。
該当するgpi Trackerのステータスを明示的なイベント行に変換します。スクリプトではACSPを「決済の送金」、RJCTを「決済の拒否」、ACSCを「決済完了」に対応付けます。契約中のAPIレスポンスで異なるステータス名やタイムスタンプが使われている場合は、文書化されたレスポンススキーマに従って対応関係を更新します。APIに該当するステータスの発生記録がない場合、イベントを作成しないでください。
監査上必要な場合は元のタイムスタンプを保持しながら、すべてのタイムスタンプを一貫したタイムゾーン（できればUTC）に統一します。金額を数値に変換し、取引通貨を保持して、真偽値のSTP値を正規化します。ソースに属性がない場合はNULL値を保持します。
PaymentTransactionId、ActivityName、EventTimestamp、利用可能な中で最も具体的なソースイベント識別子を使ってイベントの重複を除去します。業務ルールでアクティビティごとに1回のみと定められていない限り、修正と再送の繰り返しなど、正当な複数回の発生は保持します。
すべての出力行にPaymentTransactionId、ActivityName、EventTimestamp、SourceSystem、LastDataUpdateが含まれていることを確認します。ActivityNameには文書化されたアクティビティ値のみが含まれ、ソースデータに存在する場合は12種類すべてが記録されていることを確認します。
生成したデータセットをUTF-8のCSV、またはProcessMindが対応する別のイベントログ形式でエクスポートします。イベントごとに1行とし、必須列と推奨列を含めます。PaymentTransactionIdをケース識別子に、ActivityNameをアクティビティに、EventTimestampをイベントのタイムスタンプに対応付けます。追加のイベント推定を期待せずに、ファイルをProcessMindにアップロードします。

設定

APIエンドポイントとスキーマ：契約中のSWIFT文書に基づき、正確なSWIFT gpi Tracker APIのベースURL、リソースパス、リクエストパラメーター、認証ヘッダー、ページネーション、レスポンスフィールド名を設定します。スクリプトでエンドポイントを推測することはありません。
認証情報と認可：承認済みのSWIFT API認証情報の仕組み、証明書またはトークンの要件、シークレットの保管方法、社内ソースシステムの認可を使用します。本番環境のシークレットをソースコードに直接記述しないでください。
日付範囲：まず3～6か月分を対象にします。レポートの目的に応じて、イベントのタイムスタンプまたは決済作成日を使います。データ量が多い場合は期間を短く区切り、処理済みのUETRを記録します。
社内データソースのフィルター：システムに応じて、会社、法人、支店、決済商品、文書タイプ、メッセージタイプ、通貨、ステータスのフィルターを設定します。ソーススキーマを確認するまでは、[Configure based on your system]などのプレースホルダーを使用します。
アクティビティの対応付け：社内ステータスとキューの対応関係を明示的に管理します。例として、決済詳細の検証には「下書き」から「検証済み」または「承認待ち」への変更、制裁スクリーニングの通過にはスクリーニング保留の解除、決済エラーの検出には修正または例外キューへの登録、決済エラーの解決には処理中または決済準備完了への復帰を対応付けます。
gpiステータスの対応付け：文書化されたTrackerのステータスを明示的なアクティビティに対応付けます。この実装では、ACSPを「決済の送金」、RJCTを「決済の拒否」、ACSCを「決済完了」に対応付けます。契約中のAPI文書でステータスの意味が確認できる場合に限り、対応関係を追加します。
タイムスタンプの方針：可能な場合はタイムスタンプをUTCに統一します。ソースイベント時刻、Trackerイベント時刻、メッセージ受信時刻、抽出時刻の優先順位を定めます。LastDataUpdateには業務イベントの時刻ではなく、抽出または更新の時刻を設定します。
レート制限と再試行：リクエストのタイムアウト、再試行回数、指数バックオフ、HTTPステータスの処理、ページネーションを設定します。SWIFTサービスの制限を守り、制御されていない並列リクエストは避けてください。
パフォーマンス：UETRごとにレスポンスをキャッシュし、UETRをまとめて処理して、チェックポイントを保存します。対応している場合は増分抽出を行います。同じUETRと日付範囲を繰り返し照会しないでください。
データ品質：PaymentTransactionId、ActivityName、EventTimestamp、SourceSystem、LastDataUpdateを必須とします。ソースに推奨属性がない場合はNULL値を許容しますが、各属性の入力率を報告します。
監査対応：生のAPIレスポンス、リクエスト時刻、レスポンスステータス、UETR、抽出実行IDを、承認済みの監査用保管先に保存します。決済データに適用される保持期間とセキュリティ規則に従ってください。
前提条件：社内の決済ライフサイクル、コンプライアンス、SWIFTゲートウェイ、照合、メッセージインターフェースの各ログにアクセスできることに加え、SWIFT gpi Tracker APIのライセンスと必要な証明書またはネットワーク許可リストを確認します。

クエリの例 python

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

import requests

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

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

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

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


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


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


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


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


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


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


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


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


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


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


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


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


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


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


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


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


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


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

準備はできましたか？

これらのテンプレートを使って、SWIFTの生ログをオペレーショナルエクセレンスと決済の迅速化に役立つデータに変えましょう。

改善ガイドに戻る

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

SWIFT決済処理を今すぐ改善

ストレートスルー処理率98％を達成し、手作業によるエラーをなくしましょう

無料トライアルを開始しましょう

クレジットカード不要。5分で設定できます

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

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

決済処理の属性

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

データ抽出ガイド

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

ステップ

設定

クエリの例 python

SWIFT MT生ファイル用Pythonパーサー

ステップ

設定

クエリの例 python

SWIFT gpi APIコネクター

ステップ

設定

クエリの例 python

準備はできましたか？

SWIFT決済処理を今すぐ改善