# How Flowsint Traces Cryptocurrency Transactions: Etherscan to Neo4j Pipeline

> Flowsint traces crypto transactions from Etherscan to Neo4j. Discover how wallet addresses become nodes and transactions become relationships in a graph database.

- Repository: [reconurge/flowsint](https://github.com/reconurge/flowsint)
- Tags: how-to-guide
- Published: 2026-06-05

---

**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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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:

```python
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`:

```python
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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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`](https://github.com/reconurge/flowsint/blob/main/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.