📊 Data & Analytics Crypto

La blockchain est une base de données publique et transparente. Cependant, elle est **optimisée pour l'ÉCRITURE (Write), pas pour la LECTURE (Read)**.

L'architecture de l'EVM est une **machine à états (State Machine)**. Son objectif est de calculer (et de se mettre d'accord par consensus) sur l'**"État Actuel" (World State)**. Un nœud Ethereum est conçu pour répondre rapidement à la question : "Quel est le solde de l'adresse X *maintenant* ?" (eth_getBalance).

// Solidity
event Transfer(address indexed from, address indexed to, uint256 amount);
...
emit Transfer(msg.sender, to, amount);

**TheGraph** est un **protocole d'indexation décentralisé** (et un réseau). Son objectif est de fournir le service ETL (décrit en 1.1) de manière décentralisée, et de fournir aux développeurs de **DApps** une **API temps-réel (GraphQL)** pour récupérer ces données historiques.

Analogie : C'est le **"Google" de la Blockchain**. Google indexe le "chaos" du Web (HTML) dans une base de données structurée. TheGraph indexe le "chaos" de la Blockchain (Logs/Events) dans une base de données structurée.

// Requête GraphQL pour un Subgraph
query {
  user(id: "0x123...") {
    id
    balance
    transactions(orderBy: timestamp, orderDirection: desc, first: 10) {
      id
      amount
      timestamp
    }
  }
}
// Renvoie l'utilisateur, son solde, ET ses 10 dernières tx
// ... le tout en 1 seul appel réseau.

Construire un "Subgraph" (le module d'indexation) consiste à définir trois fichiers fondamentaux. C'est le "manuel d'instructions" que le nœud Indexer de TheGraph va suivre pour réaliser l'ETL.

specVersion: 0.0.5
schema:
  file: ./schema.graphql
dataSources:
  - kind: ethereum/contract
    name: MyToken
    network: mainnet
    source:
      # L'adresse du contrat à surveiller
      address: "0x6B175474E89094C44Da98b954EedeAC495271d0F"
      # L'ABI (pour que TheGraph puisse décoder)
      abi: MyToken
      # Par où commencer l'indexation (bloc de déploiement)
      startBlock: 6627917
    mapping:
      kind: ethereum/events
      apiVersion: 0.0.7
      language: wasm/assemblyscript
      entities: # Les types de données du Schéma
        - User
        - Transfer
      abis: # Le nom de l'ABI à utiliser
        - name: MyToken
          file: ./abis/MyToken.json
      # Le lien crucial : QUEL Event écouter, et QUELLE fonction appeler
      eventHandlers:
        - event: Transfer(indexed address,indexed address,uint256)
          handler: handleTransfer # Appelle la fonction handleTransfer()
      file: ./src/mapping.ts # Le fichier contenant la logique

# Définition de l'entité "User" (table User)
type User @entity {
  # L'ID unique (ex: l'adresse 0x...)
  id: ID!
  
  # Le solde, stocké en BigInt
  balance: BigInt! @derivedFrom(field: "user")
  
  # La liste des transactions (relation 1-N)
  transactions: [Transaction!]! @derivedFrom(field: "user")
}

# Définition de l'entité "Transaction" (table Transaction)
type Transaction @entity {
  # L'ID unique (ex: le hash de la tx)
  id: ID!
  
  from: User!
  to: User!
  amount: BigInt!
  timestamp: BigInt!
}

import { Transfer as TransferEvent } from "../generated/MyToken/MyToken"
import { User, Transaction } from "../generated/schema"
import { BigInt } from "@graphprotocol/graph-ts"

// La fonction "handler" appelée par le YAML
export function handleTransfer(event: TransferEvent): void {
  
  // =======================================================
  // ÉTAPE T (Transformer)
  // =======================================================
  
  // 1. Charger ou créer l'expéditeur (User "from")
  let from = User.load(event.params.from.toHex())
  if (from == null) {
    from = new User(event.params.from.toHex())
    from.balance = BigInt.fromI32(0)
  }

  // 2. Charger ou créer le destinataire (User "to")
  let to = User.load(event.params.to.toHex())
  if (to == null) {
    to = new User(event.params.to.toHex())
    to.balance = BigInt.fromI32(0)
  }

  // 3. Mettre à jour leurs soldes
  from.balance = from.balance.minus(event.params.amount)
  to.balance = to.balance.plus(event.params.amount)

  // 4. Créer l'entité Transaction
  let tx = new Transaction(event.transaction.hash.toHex())
  tx.from = from.id
  tx.to = to.id
  tx.amount = event.params.amount
  tx.timestamp = event.block.timestamp

  // =======================================================
  // ÉTAPE L (Load)
  // =======================================================
  // Sauvegarder les entités modifiées dans la BDD de TheGraph
  from.save()
  to.save()
  tx.save()
}

**Dune** (anciennement Dune Analytics) est une **plateforme d'analyse de données (SaaS) centralisée**. Son objectif n'est *pas* de fournir des APIs temps-réel aux DApps, mais de fournir aux **analystes, data scientists et au grand public** une interface **SQL** simple pour requêter, agréger et visualiser les données on-chain.

Analogie : C'est le **"Tableau", "Looker Studio" ou "BigQuery"** du Web3. C'est un outil d'**Analyse (BI)**, pas une API de production.

SELECT * FROM ethereum.logs WHERE contract_address = '0x...'

-- Dune utilise SQL (Spark SQL / SQL:2011)
-- On requête la table "décodée" par la communauté

SELECT
    tx_hash,
    block_time,
    "from" AS user_address,
    "to" AS contract_address,
    amount0_unadjusted / 1e18 AS amount0, -- (Décodage des décimales)
    amount1_unadjusted / 1e6 AS amount1,
    (amount0 * p0.price) + (amount1 * p1.price) AS usd_volume -- (Jointure avec les prix)
    
FROM uniswap_v3_ethereum.Swap AS t -- Table décodée
JOIN prices.usd AS p0 ON p0.contract_address = t.token0_address
JOIN prices.usd AS p1 ON p1.contract_address = t.token1_address

WHERE t.block_time > NOW() - INTERVAL '1' DAY

ORDER BY usd_volume DESC
LIMIT 10

TheGraph et Dune résolvent tous deux le "problème de l'ETL", mais pour des **utilisateurs radicalement différents** et avec des **philosophies radicalement différentes**.

Avant d'utiliser les API de haut niveau (Dune/Graph), on peut utiliser Python (avec la bibliothèque `web3.py`) pour interagir **directement** avec un nœud Ethereum (ex: Infura, Alchemy) via JSON-RPC. C'est le "niveau 0" de l'analyse, mais il est essentiel pour des tâches spécifiques.

# Installation
pip install web3

from web3 import Web3

# 1. Définir le Provider (endpoint du nœud RPC)
INFURA_URL = "https://mainnet.infura.io/v3/YOUR_API_KEY"
w3 = Web3(Web3.HTTPProvider(INFURA_URL))

# 2. Vérifier la connexion
if not w3.is_connected():
    print("Erreur de connexion")
    exit()

# 3. Appels RPC simples (Lecture de l'"État Actuel")
latest_block = w3.eth.get_block('latest')
print(f"Dernier Bloc: {latest_block.number}")

# 4. Utiliser l'ABI pour interagir avec un contrat (Lecture)
DAI_ADDRESS = "0x6B175474E89094C44Da98b954EedeAC495271d0F"
DAI_ABI = '[{"constant":true,"inputs":[],"name":"name","outputs":[{"name":"","type":"string"}],"payable":false,"stateMutability":"view","type":"function"}, ...]' # (ABI JSON minifié)

dai_contract = w3.eth.contract(address=DAI_ADDRESS, abi=DAI_ABI)

# 5. Appeler une fonction "view" (similaire à .call())
dai_name = dai_contract.functions.name().call()
print(f"Nom du token: {dai_name}")

from web3 import Web3

# (Setup w3 et dai_contract...)

# 1. Définir le filtre pour l'événement "Transfer"
# On veut les logs 'Transfer' du contrat DAI, du bloc 18000000 au 18000001
event_filter = dai_contract.events.Transfer.create_filter(
    fromBlock=18000000,
    toBlock=18000001
)

# 2. Récupérer les logs bruts (eth_getLogs)
raw_logs = event_filter.get_all_entries()

print(f"Trouvé {len(raw_logs)} logs de transfert.")

# 3. Les logs sont bruts. On doit les transformer.
for log in raw_logs:
    # web3.py peut décoder le 'data blob' en utilisant l'ABI
    processed_log = dai_contract.events.Transfer().process_log(log)
    
    # Maintenant, c'est propre !
    sender = processed_log.args['from']
    receiver = processed_log.args['to']
    amount = processed_log.args['value']
    
    # Convertir 'amount' (qui a 18 décimales)
    human_amount = w3.from_wei(amount, 'ether')
    
    print(f"Transfert de {human_amount} DAI de {sender} à {receiver}")

C'est la méthode "Data Scientist". Au lieu de réinventer l'ETL, on utilise Python pour interroger la base de données déjà prête de Dune (via leur API payante), et on charge les résultats dans **Pandas** (la bibliothèque reine de l'analyse de données en Python) pour une analyse plus poussée.

# Installation
pip install dune-client pandas

import pandas as pd
from dune_client.client import DuneClient
from dune_client.query import Query, QueryParameter
import os

# 1. Authentification (via variable d'environnement)
dune = DuneClient(os.environ["DUNE_API_KEY"])

# 2. Définir la requête (ID d'une requête déjà sauvegardée sur Dune)
# (Ex: Une requête qui trouve les 10 plus gros trades sur Uniswap)
MY_QUERY_ID = 123456

# (Optionnel) Définir des paramètres si la requête SQL les utilise
# params = [
#     QueryParameter.text_type(name="user_address", value="0x..."),
# ]

# 3. Exécuter la requête et attendre les résultats (bloquant)
print("Exécution de la requête Dune (ID: 123456)...")
results = dune.get_latest_result(MY_QUERY_ID)

# 4. Les résultats sont dans 'rows'
# (C'est une liste de dictionnaires JSON)
raw_data = results.result.rows
print(f"Reçu {len(raw_data)} lignes.")

# 5. Charger dans Pandas (La partie la plus importante)
df = pd.DataFrame(raw_data)

# 6. Analyse de Données avec Pandas
if not df.empty:
    # Convertir les colonnes (qui sont des strings)
    df['usd_volume'] = pd.to_numeric(df['usd_volume'])
    df['block_time'] = pd.to_datetime(df['block_time'])

    print("\n--- DataFrame Pandas ---")
    print(df.head())

    print("\n--- Analyse Simple ---")
    print(f"Volume USD Total des 10 trades: {df['usd_volume'].sum()}")
    print(f"Volume USD Moyen: {df['usd_volume'].mean()}")

De même, on peut utiliser Python pour interroger les **Subgraphs de TheGraph**. C'est utile pour des tâches temps-réel légères ou pour agréger des données de DApps. L'API de TheGraph est **GraphQL** (pas SQL), nous devons donc utiliser des requêtes HTTP POST.

# Installation
pip install requests pandas
# (ou une lib GQL : pip install gql)

import requests
import pandas as pd
import json

# 1. Endpoint du Subgraph (Exemple : Uniswap V3)
THEGRAPH_URL = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"

# 2. Requête GraphQL (en string)
# On veut les 5 plus gros pools de liquidité (par volume en USD)
graphql_query = """
{
  pools(orderBy: volumeUSD, orderDirection: desc, first: 5) {
    id
    token0 {
      symbol
    }
    token1 {
      symbol
    }
    feeTier
    totalValueLockedUSD
    volumeUSD
  }
}
"""

# 3. Envoyer la requête HTTP POST
try:
    response = requests.post(THEGRAPH_URL, json={'query': graphql_query})
    response.raise_for_status() # Lève une erreur si le statut n'est pas 200

    # 4. Récupérer le JSON
    data = response.json()
    
    # 5. Charger dans Pandas
    # Les données sont dans data['data']['pools']
    pools_data = data.get('data', {}).get('pools', [])
    
    df = pd.DataFrame(pools_data)

    # 6. Nettoyer les données (le JSON est imbriqué)
    # Aplatir les dictionnaires 'token0' et 'token1'
    df['token0_symbol'] = df['token0'].apply(lambda x: x['symbol'])
    df['token1_symbol'] = df['token1'].apply(lambda x: x['symbol'])
    df = df.drop(columns=['token0', 'token1'])

    print("--- Top 5 Pools Uniswap V3 (via TheGraph) ---")
    print(df)

except requests.exceptions.RequestException as e:
    print(f"Erreur lors de l'appel à TheGraph: {e}")