Building a Transaction Monitoring System for CrossFi Using Python: A Practical Guide

JohnJohn
7 min read

Introduction

The CrossFi ecosystem is growing rapidly, and with growth comes the need for reliable tools to monitor blockchain activity. In this article, we’ll build a transaction monitoring system using Python that taps into CrossFi’s official XFI-Scan API. This guide will walk you through how to query transaction data, process the JSON response, analyze key fields, and even set up basic alerting mechanisms—all with real-world data from the CrossFi network.

API Reference:
Base URL: https://test.xfiscan.com/
Transaction query endpoint: GET /api/1.0/txs
For additional details, please refer to the official XFI-Scan API docs: CrossFi XFI-Scan API

Overview

In this guide, we will cover the following topics:

  1. Understanding CrossFi Transactions:
    Learn about the structure of a CrossFi transaction, including important fields like txhash, addresses, auth_info, and more.

  2. Querying Transactions via the XFI-Scan API:
    Write Python code to query transactions using the official API endpoint and parameters such as limit, page, and sort.

  3. Processing Transaction Data with Pandas:
    Convert JSON responses into a Pandas DataFrame, extract key fields, and structure the data for analysis.

  4. Visualizing Transaction Data:
    Use Matplotlib to create visualizations, including transaction fee trends and overall transaction volumes.

  5. Automating Alerts for Abnormal Activity:
    Set up a basic alerting system that notifies you when transactions meet certain criteria (e.g., high fees or unusual activity).

  6. Building a Unified Dashboard:
    Integrate these components into a cohesive script that serves as your transaction monitoring system.

By the end of this article, you’ll have a fully functional system to monitor CrossFi transactions in near real-time.

Understanding CrossFi Transaction Structure

Before diving into the code, it’s essential to understand the data you’ll be working with. Here’s a sample transaction JSON output from the XFI-Scan API:

{
  "txhash": "fe828e6bb3ae72f9e4be6c082eb78805af0088c0e0e5630846776351cfd82b85",
  "addresses": [
    "0x8465de71a0c24c358b73a3869b04469706341a07",
    "mx1s3jauudqcfxrtzmn5wrfkpzxjurrgxs8sg7923",
    "0x06512a7eb230a8856cebe0b1d6077322fd24c01e",
    "mx1qegj5l4jxz5g2m8tuzcavpmnyt7jfsq7sfsafw"
  ],
  "auth_info": {
    "signer_infos": [
      {
        "public_key": {
          "key": "AopEGZgoDwPpernTw9Waw2InMhPdvdHuSXj+npkm/3FP",
          "@type": "/ethermint.crypto.v1.ethsecp256k1.PubKey"
        },
        "mode_info": {
          "single": {
            "mode": "SIGN_MODE_DIRECT"
          }
        },
        "sequence": "244988"
      }
    ],
    "fee": {
      "amount": [
        {
          "denom": "mpx",
          "amount": "1561680000000000000"
        }
      ],
      "gas_limit": "156168",
      "payer": "",
      "granter": ""
    }
  },
  "body": {
    "messages": [
      {
        "inputs": [
          {
            "address": "mx1s3jauudqcfxrtzmn5wrfkpzxjurrgxs8sg7923",
            "coins": [
              {
                "denom": "mpx",
                "amount": "1000000000000000000"
              },
              {
                "denom": "xfi",
                "amount": "10000000000000000"
              }
            ]
          }
        ],
        "outputs": [
          {
            "address": "mx1qegj5l4jxz5g2m8tuzcavpmnyt7jfsq7sfsafw",
            "coins": [
              {
                "denom": "mpx",
                "amount": "1000000000000000000"
              }
            ]
          },
          {
            "address": "mx1qegj5l4jxz5g2m8tuzcavpmnyt7jfsq7sfsafw",
            "coins": [
              {
                "denom": "xfi",
                "amount": "10000000000000000"
              }
            ]
          }
        ],
        "@type": "/cosmos.bank.v1beta1.MsgMultiSend"
      }
    ],
    "memo": "",
    "timeout_height": "0",
    "extension_options": [],
    "non_critical_extension_options": []
  },
  "code": 0,
  "codespace": "",
  "gas_used": "129095",
  "gas_wanted": "156168",
  "height": 7665269,
  "info": "",
  "isEVM": false,
  "logs": [ /* ... logs omitted for brevity ... */ ],
  "signatures": [
    "6rVCozu8V9USEqoefS+fOEQpMAKulStU7xCKZRd81yMfEOmuoe3D8Mu9urxDsYlRpXT7IlXZCQ2hsPUDct2v9gE="
  ],
  "timestamp": "2025-02-23T10:50:29.194Z",
  "xds": []
}

Key Fields to Note:

  • txhash: Unique identifier for the transaction.

  • addresses: Array of addresses involved in the transaction.

  • auth_info: Contains fee details and signer information.

  • body: Includes the actual transaction messages, such as inputs and outputs.

  • gas_used and gas_wanted: Metrics related to transaction processing.

  • height: The block height at which the transaction was included.

  • timestamp: The time the transaction was recorded.

Understanding these fields will help you extract and analyze the data effectively.

Step 1: Querying the XFI-Scan API

Let’s start by building a Python module to query transactions. Create a file named xfi_scan_api.py:

import requests

# Base URL for the XFI-Scan API from the official docs.
BASE_URL = "https://test.xfiscan.com"

def get_transactions(limit=10, page=1, sort="-height"):
    """
    Fetches recent transactions from the XFI-Scan API.

    Parameters:
      - limit (number): Number of transactions to return.
      - page (number): Pagination page number.
      - sort (string): Sorting order (e.g., '-height' for descending by block height).

    Returns:
      A list of transaction dictionaries.
    """
    endpoint = f"{BASE_URL}/api/1.0/txs"
    params = {
        "limit": limit,
        "page": page,
        "sort": sort
    }
    try:
        response = requests.get(endpoint, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        return data.get("data", [])
    except requests.RequestException as e:
        print("Error fetching transactions:", e)
        return []

def search_transactions(query, limit=10, page=1):
    """
    Searches for transactions based on a query.

    Parameters:
      - query (string): The transaction hash to search for.
      - limit (number): Number of transactions to return.
      - page (number): Pagination page number.

    Returns:
      A list of matching transaction dictionaries.
    """
    endpoint = f"{BASE_URL}/api/1.0/txs"
    params = {
        "txhash": query,
        "limit": limit,
        "page": page
    }
    try:
        response = requests.get(endpoint, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        return data.get("data", [])
    except requests.RequestException as e:
        print("Error searching transactions:", e)
        return []

if __name__ == "__main__":
    # Test fetching transactions
    txs = get_transactions(limit=5)
    print("Recent Transaction Hashes:")
    for tx in txs:
        print(tx.get("txhash", "No txhash provided"))

Explanation:
This module defines two functions:

  • get_transactions: Retrieves recent transactions using query parameters such as limit, page, and sort.

  • search_transactions: Searches transactions by a specific query (using the txhash parameter).

Run this module to test if it fetches data properly.

Step 2: Processing Transaction Data with Pandas

Next, create process_data.py to convert the JSON data into a structured DataFrame:

import pandas as pd
from xfi_scan_api import get_transactions

def process_transactions(transactions):
    """
    Converts a list of transactions into a Pandas DataFrame.
    Extracts key fields: Tx Hash, Timestamp, Fee, Sender, Receiver, and Block Height.
    """
    if not transactions:
        return pd.DataFrame()

    processed = []
    for tx in transactions:
        # Extract fee information from auth_info
        fee_info = tx.get("auth_info", {}).get("fee", {}).get("amount", [])
        fee_amount = fee_info[0]["amount"] if fee_info else "N/A"

        # Extract sender and receiver from the first message in body
        messages = tx.get("body", {}).get("messages", [])
        sender = messages[0].get("inputs", [{}])[0].get("address", "N/A") if messages else "N/A"
        receiver = messages[0].get("outputs", [{}])[0].get("address", "N/A") if messages else "N/A"

        processed.append({
            "Tx Hash": tx.get("txhash", "N/A"),
            "Timestamp": tx.get("timestamp", "N/A"),
            "Fee": fee_amount,
            "Sender": sender,
            "Receiver": receiver,
            "Block Height": tx.get("height", "N/A")
        })

    df = pd.DataFrame(processed)
    return df

if __name__ == "__main__":
    txs = get_transactions(limit=5)
    df_txs = process_transactions(txs)
    print("Transactions DataFrame:")
    print(df_txs)

Explanation:
This script processes the JSON data:

  • It extracts fields such as transaction hash, timestamp, fee, sender, receiver, and block height.

  • The output is a Pandas DataFrame, which simplifies further analysis.

Run process_data.py to verify that the DataFrame is structured correctly.

Step 3: Visualizing Transaction Data with Matplotlib

Now, create visualize_data.py to visualize key metrics:

import matplotlib.pyplot as plt
import pandas as pd
from process_data import process_transactions
from xfi_scan_api import get_transactions

def plot_transaction_fees(df):
    """
    Plots transaction fees over the transactions.
    """
    try:
        df["Fee"] = pd.to_numeric(df["Fee"], errors="coerce")
    except Exception as e:
        print("Error converting fee data:", e)

    plt.figure(figsize=(10, 6))
    plt.plot(df["Tx Hash"], df["Fee"], marker="o", linestyle="--", color="green")
    plt.title("Transaction Fees Over Recent Transactions")
    plt.xlabel("Transaction Hash")
    plt.ylabel("Fee")
    plt.xticks(rotation=45, ha="right")
    plt.tight_layout()
    plt.show()

def plot_transaction_volume(df):
    """
    Plots the total number of transactions as a bar chart.
    """
    volume = len(df)
    plt.figure(figsize=(5, 5))
    plt.bar(["Transaction Volume"], [volume], color="blue")
    plt.title("Number of Recent Transactions")
    plt.ylabel("Count")
    plt.show()

if __name__ == "__main__":
    txs = get_transactions(limit=10)
    df_txs = process_transactions(txs)
    plot_transaction_fees(df_txs)
    plot_transaction_volume(df_txs)

Explanation:

  • plot_transaction_fees visualizes the fees for each transaction.

  • plot_transaction_volume shows a simple bar chart with the total transaction count.

  • Run visualize_data.py to open the plots.

Step 4: Building the Unified Dashboard

Finally, create dashboard.py to integrate everything into one script:

from xfi_scan_api import get_transactions, search_transactions
from process_data import process_transactions
from visualize_data import plot_transaction_fees, plot_transaction_volume

def main():
    print("Fetching recent transactions from the XFI-Scan API...")
    transactions = get_transactions(limit=10)

    # Process transactions into a DataFrame.
    df_transactions = process_transactions(transactions)

    if df_transactions.empty:
        print("No transaction data available.")
    else:
        print("Transactions Data:")
        print(df_transactions)

        # Visualize transaction fees and transaction volume.
        plot_transaction_fees(df_transactions)
        plot_transaction_volume(df_transactions)

    # Optional: Allow the user to search for specific transactions.
    query = input("Enter a transaction hash to search for (or press Enter to skip): ")
    if query:
        searched_txs = search_transactions(query)
        df_search = process_transactions(searched_txs)
        print("Search Results:")
        print(df_search)
        plot_transaction_fees(df_search)
        plot_transaction_volume(df_search)

if __name__ == "__main__":
    main()

Explanation:

  • dashboard.py serves as the central entry point.

  • It fetches recent transactions, processes them, displays the DataFrame, and then visualizes the data.

  • Optionally, it allows users to search for transactions by hash and visualize those results.

Conclusion

In this comprehensive guide, we built a Python-based transaction monitoring system for the CrossFi ecosystem using the official XFI-Scan API. We covered:

  • API Integration:
    We defined functions in xfi_scan_api.py to query transactions using the endpoint GET /api/1.0/txs from https://test.xfiscan.com/.

  • Data Processing:
    Using process_data.py, we converted JSON responses into a Pandas DataFrame, extracting key fields such as transaction hash, fee, sender, receiver, block height, and timestamp.

  • Visualization:
    In visualize_data.py, we visualized transaction fees and overall transaction volume using Matplotlib, turning raw data into actionable insights.

  • Dashboard Assembly:
    Finally, dashboard.py ties all components together, offering a unified view of recent transactions along with an optional search feature.

This project demonstrates how to harness CrossFi’s XFI-Scan API for real-time blockchain analytics. With further enhancements—such as adding real-time updates, integrating additional endpoints (e.g., for validator performance), or building a web interface using Flask or Django—you can extend this foundation into a robust monitoring and alerting system tailored to your specific needs.

By following these steps, you now have a practical, accurate transaction monitoring system built entirely in Python. Happy coding, and enjoy exploring the rich data available on the CrossFi network!

0
Subscribe to my newsletter

Read articles from John directly inside your inbox. Subscribe to the newsletter, and don't miss out.

Web3Blockchaincrossfiautomation

Written by

John
John