How Flowsint Traces Cryptocurrency Transactions: Etherscan to Neo4j Pipeline

Flowsint traces cryptocurrency transactions by pulling on-chain data from the Etherscan API and transforming wallet addresses into nodes and TRANSACTION relationships inside a Neo4j graph via the GraphService abstraction.

The open-source reconurge/flowsint repository provides a modular intelligence framework for blockchain analysis. Understanding how Flowsint traces cryptocurrency transactions reveals a pipeline that bridges external block explorers with graph-based storage, enabling investigators to follow fund flows across multiple hops.

The CryptoWalletAddressToTransactions Enricher

The core component for cryptocurrency transaction tracing is the CryptoWalletAddressToTransactions enricher, defined in flowsint-enrichers/src/flowsint_enrichers/crypto/to_transactions.py. It consumes a CryptoWallet input model, which is a Pydantic schema representing a blockchain address located in flowsint-types/src/flowsint_types/wallet.py.

Inside the enricher, the _get_transactions method constructs an HTTP request to the Etherscan API (or any compatible endpoint) and parses the JSON response into a list of CryptoWalletTransaction objects. This step isolates the raw blockchain API logic from the graph storage layer.

How Flowsint Converts Transactions into Graph Relationships

After fetching the transaction list, the enricher’s postprocess method iterates over every CryptoWalletTransaction and writes graph structures through the GraphService abstraction found in flowsint-core/src/flowsint_core/core/graph/service.py.

The flow follows three concrete steps:

• GraphService.create_node inserts both the source and target wallet addresses as cryptowallet nodes.
• GraphService.query executes a Cypher statement that creates a TRANSACTION relationship between the two wallet nodes.
• Rich metadata — including value, timestamp, gas, gas_price, and block_number — is stored as properties on that relationship, enabling later filtering and aggregation.

The GraphService also provides log_graph_message, which forwards diagnostic events to the configured logger, keeping the tracing pipeline observable.

Under the hood, GraphService delegates persistence to a concrete Neo4jGraphRepository implementation, defined in flowsint-core/src/flowsint_core/core/graph/repository.py. This design isolates the enricher from database specifics while still supporting batching, custom Cypher, and node creation.

Code Examples for Cryptocurrency Transaction Tracing

Running the Async Enricher

The following example initializes the enricher, scans a wallet, writes the results to the graph, and runs a follow-up Cypher query:

import asyncio
from flowsint_enrichers.crypto.to_transactions import CryptoWalletAddressToTransactions
from flowsint_core.core.graph.service import create_graph_service

async def trace_wallet(address: str, sketch_id: str):
    # Initialise the enricher

    enricher = CryptoWalletAddressToTransactions(
        sketch_id=sketch_id,
        params={"ETHERSCAN_API_KEY": "<your‑key>"}
    )

    # Provide a CryptoWallet instance as input

    from flowsint_types.wallet import CryptoWallet
    result = await enricher.scan([CryptoWallet(address=address)])

    # Post‑process writes the graph

    enricher.postprocess(result, [CryptoWallet(address=address)])

    # Example query: all outgoing transactions from the wallet

    graph = create_graph_service(sketch_id)
    query = """
        MATCH (w:cryptowallet {wallet: $addr})-[:TRANSACTION]->(t:cryptowallet)
        RETURN w.wallet AS source, t.wallet AS target, r.value AS value
    """
    records = graph.query(query, {"addr": address})
    return records

# Usage

if __name__ == "__main__":
    asyncio.run(trace_wallet("0x1234…abcd", "demo‑sketch"))

Manual Graph Insertion with Cypher

You can also bypass the enricher and insert transactions directly using GraphService:

from flowsint_core.core.graph.service import create_graph_service
from flowsint_types.wallet import CryptoWallet, CryptoWalletTransaction

# 1️⃣ Create nodes

svc = create_graph_service("demo-sketch")
svc.create_node(CryptoWallet(address="0xSource"))
svc.create_node(CryptoWallet(address="0xTarget"))

# 2️⃣ Create a transaction edge

tx = CryptoWalletTransaction(
    source=CryptoWallet(address="0xSource"),
    target=CryptoWallet(address="0xTarget"),
    hash="0xTxHash",
    value=1.23,
    timestamp="1650000000",
    block_number=1234567,
    gas=21000,
    gas_price=50000000000,
)
svc.query(
    """
    MATCH (src:cryptowallet {wallet: $src})
    MATCH (tgt:cryptowallet {wallet: $tgt})
    MERGE (src)-[r:TRANSACTION {hash: $hash}]->(tgt)
    SET r += $props
    """,
    {
        "src": tx.source.address,
        "tgt": tx.target.address,
        "hash": tx.hash,
        "props": {
            "value": tx.value,
            "timestamp": tx.timestamp,
            "block_number": tx.block_number,
            "gas": tx.gas,
            "gas_price": tx.gas_price,
        },
    },
)

Key Files in the Tracing Pipeline

• CryptoWalletAddressToTransactions — flowsint-enrichers/src/flowsint_enrichers/crypto/to_transactions.py: The enricher that resolves raw transactions via the Etherscan API.
• CryptoWallet & CryptoWalletTransaction — flowsint-types/src/flowsint_types/wallet.py: Pydantic models that represent wallet addresses and on-chain transaction records.
• GraphService — flowsint-core/src/flowsint_core/core/graph/service.py: High-level graph API that provides create_node, query, and log_graph_message methods.
• Neo4jGraphRepository — flowsint-core/src/flowsint_core/core/graph/repository.py: The concrete repository implementation accessed through create_graph_service.
• CryptoWalletAddressToNFTs — flowsint-enrichers/src/flowsint_enrichers/crypto/to_nfts.py: A related enricher that demonstrates the same graph ingestion pattern for NFT assets.

Summary

• Flowsint traces cryptocurrency transactions through the CryptoWalletAddressToTransactions enricher, which queries the Etherscan API.
• The GraphService abstraction translates wallets into nodes and transactions into TRANSACTION relationships inside Neo4j.
• Every relationship stores granular metadata — value, timestamp, gas, block number, and gas price — for downstream graph analytics.
• Developers can use the built-in async enricher or perform manual insertion via create_graph_service and custom Cypher queries.

Frequently Asked Questions

How does Flowsint trace cryptocurrency transactions from a wallet address?

It routes the address through the CryptoWalletAddressToTransactions enricher, which calls the Etherscan API to fetch normal transactions. The enricher then passes the parsed CryptoWalletTransaction objects to GraphService, creating wallet nodes and TRANSACTION edges in Neo4j.

What transaction metadata does Flowsint store in the graph?

Flowsint stores value, timestamp, block_number, gas, and gas_price as properties on each TRANSACTION relationship. This allows Cypher queries to filter, sort, and aggregate fund movements directly inside the graph database.

Can I trace transactions without running the enricher?

Yes. You can use create_graph_service to obtain a GraphService instance, manually call create_node for each wallet, and then run a Cypher MERGE query to create TRANSACTION relationships with arbitrary properties, as shown in the manual insertion example.

What graph database backend does Flowsint use?

The default implementation is Neo4j, accessed through the Neo4jGraphRepository class in flowsint-core/src/flowsint_core/core/graph/repository.py. The GraphService isolates the rest of the codebase from the underlying database, so other graph stores could be adapted by implementing the same repository interface.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}