Gönderi

Blockchain Veri Analizi için Python Araçları

Web3.py, TheGraph ve Dune Analytics kullanarak blockchain verilerini analiz etme rehberi. On-chain veri okuma, event monitoring ve DeFi protokol takibi.

Blockchain Veri Analizi için Python Araçları

Blockchain teknolojisinin gelişmesiyle birlikte, on-chain verilerin analizi modern kripto ve DeFi projelerinin vazgeçilmez bir parçası haline geldi. Ethereum, Polygon ve diğer blokzincirlerdeki işlemleri, akıllı sözleşme olaylarını ve cüzdan hareketlerini analiz etmek için Python ekosisteminde güçlü araçlar mevcut. Bu yazıda Web3.py, TheGraph ve Dune Analytics kullanarak profesyonel blockchain veri analizi yapmanın yollarını inceleyeceğiz.

Blockchain Veri Analizinin Önemi

On-chain veri analizi, blockchain üzerinde gerçekleşen tüm işlemlerin şeffaf ve değiştirilemez doğasından faydalanarak değerli içgörüler elde etmeyi sağlar. Token transferleri, likidite havuzu hareketleri, NFT satışları ve whale cüzdan aktiviteleri gibi metrikleri izleyerek piyasa trendlerini öngörebilir ve stratejik kararlar alabilirsiniz.

Kullanım Alanları

Blockchain veri analizi şu alanlarda kritik öneme sahiptir:

  • DeFi Protokol Analizi: Uniswap, Aave gibi protokollerde TVL, volume ve kullanıcı aktivitesi takibi
  • Token Ekonomi İzleme: Token dağılımı, holder davranışları ve transfer pattern analizi
  • Whale Tracking: Büyük cüzdanların hareketlerini takip ederek piyasa sinyalleri yakalama
  • Smart Contract Monitoring: Kritik olayları gerçek zamanlı izleme ve alarm sistemleri
  • MEV Analizi: Arbitraj, sandwich attack ve diğer MEV stratejilerinin tespiti

Web3.py ile Temel On-Chain Veri Okuma

Web3.py, Ethereum ve EVM uyumlu blokzincirlerle etkileşim kurmak için en popüler Python kütüphanesidir. Düşük seviyeli RPC çağrıları yapmak ve akıllı sözleşme olaylarını okumak için temel araçtır.

Kurulum ve Bağlantı

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# Gerekli kütüphanelerin kurulumu
pip install web3 python-dotenv

# Web3.py ile Ethereum node bağlantısı
from web3 import Web3
import os
from dotenv import load_dotenv

load_dotenv()

# Infura veya Alchemy gibi bir RPC endpoint kullanın
RPC_URL = os.getenv("ETHEREUM_RPC_URL")
w3 = Web3(Web3.HTTPProvider(RPC_URL))

# Bağlantı kontrolü
if w3.is_connected():
    print(f"Ethereum ağına bağlanıldı")
    print(f"Son blok numarası: {w3.eth.block_number}")
    print(f"Gas fiyatı: {w3.eth.gas_price / 10**9} Gwei")
else:
    print("Bağlantı başarısız")

RPC endpoint’ler için Infura veya Alchemy gibi servislerden ücretsiz API key alabilirsiniz. Rate limit’lere dikkat edin.

Transaction Data Parsing

Blok verisini okuyup işlemleri analiz etmek, blockchain analizinin temelidir:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from datetime import datetime
from typing import Dict, List

def get_block_transactions(block_number: int) -> List[Dict]:
    """
    Belirtilen bloktaki tüm işlemleri detaylı şekilde getirir
    """
    block = w3.eth.get_block(block_number, full_transactions=True)
    
    transactions = []
    for tx in block.transactions:
        # Transaction detaylarını parse et
        tx_data = {
            'hash': tx['hash'].hex(),
            'from': tx['from'],
            'to': tx['to'],
            'value': w3.from_wei(tx['value'], 'ether'),  # Wei'den ETH'ye çevir
            'gas': tx['gas'],
            'gas_price': w3.from_wei(tx['gasPrice'], 'gwei'),
            'nonce': tx['nonce'],
            'block': block_number,
            'timestamp': datetime.fromtimestamp(block.timestamp)
        }
        
        # Input data varsa (smart contract call) işaretle
        if tx['input'] != '0x':
            tx_data['is_contract_call'] = True
            tx_data['input_length'] = len(tx['input'])
        
        transactions.append(tx_data)
    
    return transactions

# Son 10 bloktaki işlemleri analiz et
latest_block = w3.eth.block_number
for i in range(latest_block - 10, latest_block):
    txs = get_block_transactions(i)
    print(f"Blok {i}: {len(txs)} işlem")
    
    # ETH transfer hacmini hesapla
    total_volume = sum(tx['value'] for tx in txs)
    print(f"  Toplam ETH: {total_volume:.4f}")

Smart Contract Event Monitoring

Akıllı sözleşmelerin emit ettiği event’leri izlemek, DeFi protokollerini analiz etmenin anahtarıdır:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
# Uniswap V2 Pair contract ABI (sadece Swap event için)
UNISWAP_PAIR_ABI = [
    {
        "anonymous": False,
        "inputs": [
            {"indexed": True, "name": "sender", "type": "address"},
            {"indexed": False, "name": "amount0In", "type": "uint256"},
            {"indexed": False, "name": "amount1In", "type": "uint256"},
            {"indexed": False, "name": "amount0Out", "type": "uint256"},
            {"indexed": False, "name": "amount1Out", "type": "uint256"},
            {"indexed": True, "name": "to", "type": "address"}
        ],
        "name": "Swap",
        "type": "event"
    }
]

def track_uniswap_swaps(pair_address: str, from_block: int, to_block: int):
    """
    Uniswap pair'inde gerçekleşen swap işlemlerini takip eder
    """
    # Contract instance oluştur
    pair_contract = w3.eth.contract(
        address=Web3.to_checksum_address(pair_address),
        abi=UNISWAP_PAIR_ABI
    )
    
    # Event filter oluştur
    swap_filter = pair_contract.events.Swap.create_filter(
        fromBlock=from_block,
        toBlock=to_block
    )
    
    # Event'leri al ve işle
    swaps = swap_filter.get_all_entries()
    
    print(f"🔄 {len(swaps)} swap işlemi bulundu")
    
    for swap in swaps:
        # Event parametrelerini parse et
        args = swap.args
        tx_hash = swap.transactionHash.hex()
        block = swap.blockNumber
        
        print(f"\nBlok: {block}")
        print(f"TX: {tx_hash}")
        print(f"Gönderen: {args.sender}")
        print(f"Amount0 In: {args.amount0In}")
        print(f"Amount1 In: {args.amount1In}")
        print(f"Amount0 Out: {args.amount0Out}")
        print(f"Amount1 Out: {args.amount1Out}")
    
    return swaps

# Örnek: USDC/ETH pair'ini izle
USDC_ETH_PAIR = "0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc"
latest = w3.eth.block_number
swaps = track_uniswap_swaps(USDC_ETH_PAIR, latest - 100, latest)

Event monitoring yaparken block range’i çok geniş tutmayın. Büyük aralıklar için batch processing kullanın.

TheGraph Protocol Mimarisi TheGraph protokolü ile blockchain verilerinin indekslenmesi ve sorgulanması

TheGraph ile Gelişmiş Veri Sorgulama

Web3.py ile doğrudan RPC çağrıları yapmak performans açısından sınırlıdır. TheGraph, blockchain verilerini indeksleyerek GraphQL ile hızlı sorgulamaya olanak tanır.

TheGraph Nedir?

TheGraph, blockchain verilerini indeksleyen ve organize eden merkeziyetsiz bir protokoldür. Subgraph’ler, belirli smart contract’ların event’lerini dinleyip veritabanı benzeri bir yapıda saklar. Bu sayede karmaşık sorgular saniyeler içinde çalıştırılabilir.

Python ile GraphQL Sorguları

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
# GraphQL client kurulumu
pip install gql[all]

from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport

# Uniswap V3 subgraph endpoint
UNISWAP_V3_SUBGRAPH = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"

# GraphQL transport ve client oluştur
transport = AIOHTTPTransport(url=UNISWAP_V3_SUBGRAPH)
client = Client(transport=transport, fetch_schema_from_transport=True)

def get_top_pools_by_tvl(limit: int = 10):
    """
    TVL'ye göre en büyük Uniswap V3 pool'larını getirir
    """
    query = gql(f"""
        {{
            pools(
                first: {limit},
                orderBy: totalValueLockedUSD,
                orderDirection: desc
            ) {{
                id
                token0 {{
                    symbol
                    decimals
                }}
                token1 {{
                    symbol
                    decimals
                }}
                totalValueLockedUSD
                volumeUSD
                feeTier
                txCount
            }}
        }}
    """)
    
    result = client.execute(query)
    return result['pools']

# Top 10 pool'u listele
pools = get_top_pools_by_tvl(10)

print("🏊 Uniswap V3 En Büyük Pool'lar (TVL)")
print("=" * 70)

for i, pool in enumerate(pools, 1):
    symbol_pair = f"{pool['token0']['symbol']}/{pool['token1']['symbol']}"
    tvl = float(pool['totalValueLockedUSD'])
    volume = float(pool['volumeUSD'])
    fee = int(pool['feeTier']) / 10000  # Fee tier'i yüzdeye çevir
    
    print(f"\n{i}. {symbol_pair}")
    print(f"   TVL: ${tvl:,.2f}")
    print(f"   Volume: ${volume:,.2f}")
    print(f"   Fee: {fee}%")
    print(f"   TX Count: {pool['txCount']}")

Token Holder Analizi

TheGraph ile token holder’ları ve transfer pattern’lerini analiz edebilirsiniz:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def analyze_token_holders(token_address: str, min_balance: float = 1000):
    """
    Belirtilen token için holder analizini yapar
    Not: Token'a özel subgraph gerekir
    """
    query = gql(f"""
        {{
            tokenHolders(
                where: {{
                    token: "{token_address.lower()}",
                    balance_gt: "{int(min_balance * 10**18)}"
                }},
                first: 100,
                orderBy: balance,
                orderDirection: desc
            ) {{
                id
                address
                balance
                transactionCount
            }}
        }}
    """)
    
    result = client.execute(query)
    holders = result['tokenHolders']
    
    # Holder istatistikleri
    total_holders = len(holders)
    total_balance = sum(int(h['balance']) / 10**18 for h in holders)
    
    print(f"👥 Top {total_holders} Holder Analizi")
    print(f"💰 Toplam Balance: {total_balance:,.2f} token")
    
    # Whale'leri tespit et (top 10 holder)
    whales = holders[:10]
    whale_balance = sum(int(h['balance']) / 10**18 for h in whales)
    whale_percentage = (whale_balance / total_balance) * 100
    
    print(f"\nWhale Analizi (Top 10)")
    print(f"   Balance: {whale_balance:,.2f} token")
    print(f"   Oran: {whale_percentage:.2f}%")
    
    return holders

TheGraph subgraph’ler community tarafından oluşturulur. Önemli protokollerin resmi subgraph’lerini kullandığınızdan emin olun.

TheGraph Subgraph Workflow TheGraph subgraph oluşturma ve veri sorgulama iş akışı

Time-Series Veri Analizi

TheGraph’in gücü, zaman serisi verilerini kolayca sorgulamaktır:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from datetime import datetime, timedelta
import pandas as pd

def get_pool_daily_volume(pool_id: str, days: int = 30):
    """
    Belirtilen pool için günlük volume verilerini getirir
    """
    # Unix timestamp hesapla (30 gün öncesi)
    start_timestamp = int((datetime.now() - timedelta(days=days)).timestamp())
    
    query = gql(f"""
        {{
            poolDayDatas(
                where: {{
                    pool: "{pool_id}",
                    date_gt: {start_timestamp}
                }},
                orderBy: date,
                orderDirection: asc
            ) {{
                date
                volumeUSD
                tvlUSD
                feesUSD
                txCount
            }}
        }}
    """)
    
    result = client.execute(query)
    daily_data = result['poolDayDatas']
    
    # Pandas DataFrame'e çevir
    df = pd.DataFrame(daily_data)
    df['date'] = pd.to_datetime(df['date'], unit='s')
    df['volumeUSD'] = df['volumeUSD'].astype(float)
    df['tvlUSD'] = df['tvlUSD'].astype(float)
    df['feesUSD'] = df['feesUSD'].astype(float)
    
    return df

# Veri analizi ve görselleştirme
# pool_data = get_pool_daily_volume("0xpool_address")
# print(pool_data.describe())

Dune Analytics ile SQL Tabanlı Analiz

Dune Analytics, blockchain verilerini SQL ile sorgulamanıza olanak tanıyan güçlü bir platformdur. Python ile Dune API’sini kullanarak sorgu sonuçlarını alabilir ve kendi analizlerinizi yapabilirsiniz.

Dune Analytics Dashboard Dune Analytics ile DEX metriklerinin görselleştirilmesi

Dune API Kurulumu

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# Dune client kurulumu
pip install dune-client

import os
from dune_client.client import DuneClient
from dune_client.query import QueryBase

# API key ile client oluştur (.env dosyasından)
DUNE_API_KEY = os.getenv("DUNE_API_KEY")
dune = DuneClient(DUNE_API_KEY)

# Mevcut bir query'yi çalıştır
def run_dune_query(query_id: int, params: dict = None):
    """
    Dune Analytics query'sini çalıştırır ve sonuçları getirir
    """
    try:
        # Query çalıştır
        query = QueryBase(query_id=query_id, params=params or {})
        results = dune.run_query(query)
        
        return results.result.rows
    except Exception as e:
        print(f"❌ Query hatası: {e}")
        return None

# Örnek: ETH Daily Transactions
query_id = 3236296  # Örnek query ID
results = run_dune_query(query_id)

if results:
    print(f"{len(results)} satır veri alındı")
    for row in results[:5]:  # İlk 5 satır
        print(row)

Custom Dashboard Oluşturma

Dune’dan aldığınız verileri Python ile analiz edip dashboard oluşturabilirsiniz:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta

def analyze_dex_volume_comparison():
    """
    Farklı DEX'lerin volume karşılaştırmasını yapar
    """
    # Dune query: DEX günlük volume (önceden oluşturulmuş query)
    query_id = 123456  # Kendi query ID'nizi kullanın
    
    # Son 30 günün verilerini al
    results = run_dune_query(query_id, params={
        "days": 30
    })
    
    if not results:
        return
    
    # DataFrame'e çevir
    df = pd.DataFrame(results)
    df['date'] = pd.to_datetime(df['date'])
    df['volume'] = df['volume'].astype(float)
    
    # DEX'lere göre grupla
    dex_totals = df.groupby('dex_name')['volume'].sum().sort_values(ascending=False)
    
    print("📊 DEX Volume Karşılaştırması (30 Gün)")
    print("=" * 50)
    
    for dex, volume in dex_totals.items():
        print(f"{dex:15} ${volume:,.0f}")
    
    # Görselleştirme (plotly)
    fig = px.line(
        df,
        x='date',
        y='volume',
        color='dex_name',
        title='DEX Volume Trend (30 Gün)',
        labels={'volume': 'Volume (USD)', 'date': 'Tarih'}
    )
    
    fig.write_html('dex_volume_analysis.html')
    print("\nGrafik kaydedildi: dex_volume_analysis.html")

Dune Analytics’te kompleks query’ler oluştururken önce SQL’i Dune web arayüzünde test edin, sonra Python’dan çağırın.

Wallet Activity Tracking

Dune ile belirli cüzdanların aktivitelerini takip edebilirsiniz:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def track_whale_wallets(wallet_addresses: list, min_value_usd: float = 100000):
    """
    Büyük cüzdanların aktivitelerini izler
    """
    # Wallet'ları query parametresi olarak gönder
    wallet_list = "', '".join(wallet_addresses)
    
    # Custom query (Dune'da önceden oluşturulmalı)
    query_id = 789012  # Whale tracking query ID
    
    results = run_dune_query(query_id, params={
        "wallet_list": wallet_list,
        "min_value": min_value_usd
    })
    
    if not results:
        return
    
    print(f"Whale Aktivite Raporu")
    print(f"Son 24 saat")
    print("=" * 80)
    
    for tx in results:
        print(f"\n{tx['timestamp']}")
        print(f"Cüzdan: {tx['wallet_address'][:10]}...{tx['wallet_address'][-8:]}")
        print(f"İşlem: {tx['action']}")
        print(f"Değer: ${tx['value_usd']:,.2f}")
        print(f"Token: {tx['token_symbol']}")

# Örnek whale adresleri
whale_addresses = [
    "0x...",  # Binance wallet
    "0x...",  # Vitalik wallet
]

# track_whale_wallets(whale_addresses, min_value_usd=500000)

Real-Time Event Monitoring Sistemi

Tüm bu araçları bir araya getirerek gerçek zamanlı monitoring sistemi oluşturabilirsiniz:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import asyncio
from web3 import Web3
from datetime import datetime
import json

class BlockchainMonitor:
    """
    Blockchain event'lerini gerçek zamanlı izleyen sınıf
    """
    def __init__(self, rpc_url: str, contracts: dict):
        self.w3 = Web3(Web3.WebsocketProvider(rpc_url))  # WebSocket kullan
        self.contracts = contracts
        self.is_running = False
    
    async def monitor_new_blocks(self):
        """
        Yeni blokları sürekli izler
        """
        print("🔍 Blok monitoring başlatıldı...")
        
        latest_block = self.w3.eth.block_number
        
        while self.is_running:
            try:
                current_block = self.w3.eth.block_number
                
                # Yeni blok varsa
                if current_block > latest_block:
                    for block_num in range(latest_block + 1, current_block + 1):
                        await self.process_block(block_num)
                    
                    latest_block = current_block
                
                await asyncio.sleep(2)  # 2 saniye bekle
                
            except Exception as e:
                print(f"❌ Hata: {e}")
                await asyncio.sleep(5)
    
    async def process_block(self, block_number: int):
        """
        Bloktaki işlemleri analiz eder
        """
        block = self.w3.eth.get_block(block_number, full_transactions=True)
        
        print(f"\n📦 Yeni Blok: {block_number}")
        print(f"⏰ Zaman: {datetime.fromtimestamp(block.timestamp)}")
        print(f"🔢 İşlem Sayısı: {len(block.transactions)}")
        
        # Her contract için event'leri kontrol et
        for contract_name, contract_info in self.contracts.items():
            await self.check_contract_events(
                contract_name,
                contract_info,
                block_number
            )
    
    async def check_contract_events(self, name: str, info: dict, block: int):
        """
        Belirli contract'ın event'lerini kontrol eder
        """
        contract = self.w3.eth.contract(
            address=info['address'],
            abi=info['abi']
        )
        
        # Her event type için filter oluştur
        for event_name in info['events']:
            event = getattr(contract.events, event_name)
            
            event_filter = event.create_filter(
                fromBlock=block,
                toBlock=block
            )
            
            entries = event_filter.get_all_entries()
            
            if entries:
                print(f"\n🚨 {name} - {event_name}: {len(entries)} event")
                
                for entry in entries:
                    await self.handle_event(name, event_name, entry)
    
    async def handle_event(self, contract: str, event: str, data):
        """
        Event'i işler ve gerekli aksiyonları alır
        """
        # Event tipine göre özel işlemler
        if event == "Swap" and contract == "UniswapV2":
            # Büyük swap'leri logla
            amount0 = data.args.amount0Out
            if amount0 > 10**20:  # 100+ token
                print(f"   BÜYÜK SWAP TESPİT EDİLDİ!")
                print(f"   TX: {data.transactionHash.hex()}")
                
                # Alert gönder (Telegram, Discord, vb.)
                await self.send_alert(f"Büyük swap: {data.transactionHash.hex()}")
    
    async def send_alert(self, message: str):
        """
        Alert gönderir (Telegram, Discord, email, vb.)
        """
        # Buraya kendi alert sisteminizi entegre edin
        print(f"Alert: {message}")
    
    def start(self):
        """
        Monitoring'i başlatır
        """
        self.is_running = True
        asyncio.run(self.monitor_new_blocks())
    
    def stop(self):
        """
        Monitoring'i durdurur
        """
        self.is_running = False

# Kullanım örneği
"""
contracts_to_monitor = {
    "UniswapV2": {
        "address": "0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc",
        "abi": UNISWAP_PAIR_ABI,
        "events": ["Swap", "Mint", "Burn"]
    }
}

monitor = BlockchainMonitor(
    rpc_url="wss://mainnet.infura.io/ws/v3/YOUR_KEY",
    contracts=contracts_to_monitor
)

# Monitoring'i başlat
# monitor.start()
"""

WebSocket bağlantıları için reconnection logic implementasyonu önemlidir. Bağlantı koptuğunda otomatik yeniden bağlanma sağlayın.

Best Practices ve Optimizasyon

Blockchain veri analizi yaparken dikkate almanız gereken önemli noktalar:

1. RPC Rate Limiting

Public RPC endpoint’leri rate limit’e sahiptir. Production’da mutlaka:

  • Paid RPC servisi kullanın (Infura, Alchemy, QuickNode)
  • Request cache mekanizması ekleyin
  • Exponential backoff retry logic implementasyonu yapın
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import time
from functools import wraps

def rate_limited(max_per_second: int = 5):
    """
    RPC çağrılarını rate limit'e uygun şekilde yavaşlatır
    """
    min_interval = 1.0 / max_per_second
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait_time = min_interval - elapsed
            
            if wait_time > 0:
                time.sleep(wait_time)
            
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        
        return wrapper
    return decorator

@rate_limited(max_per_second=10)
def fetch_block_data(block_number: int):
    return w3.eth.get_block(block_number)

Public RPC endpoint’ler günde 100K request ile sınırlıdır. Production uygulamalar için ücretli plan kullanın.

2. Veri Caching

Aynı verileri tekrar tekrar sorgulamayın:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from functools import lru_cache
import redis

# Redis cache
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

def cached_query(key: str, ttl: int = 3600):
    """
    Redis ile query sonuçlarını cache'ler
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = f"{key}:{str(args)}:{str(kwargs)}"
            
            # Cache'de var mı kontrol et
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
            
            # Cache'de yok, query çalıştır
            result = func(*args, **kwargs)
            
            # Sonucu cache'le
            redis_client.setex(
                cache_key,
                ttl,
                json.dumps(result)
            )
            
            return result
        
        return wrapper
    return decorator

@cached_query("block_data", ttl=300)
def get_block_with_cache(block_number: int):
    return w3.eth.get_block(block_number)

3. Batch Processing

Çok sayıda işlem yapacaksanız batch processing kullanın:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from web3 import Web3
from typing import List

def batch_get_transactions(tx_hashes: List[str], batch_size: int = 100):
    """
    Transaction'ları batch olarak getirir
    """
    results = []
    
    for i in range(0, len(tx_hashes), batch_size):
        batch = tx_hashes[i:i + batch_size]
        
        # Batch request oluştur
        batch_results = []
        for tx_hash in batch:
            try:
                tx = w3.eth.get_transaction(tx_hash)
                batch_results.append(tx)
            except Exception as e:
                print(f"❌ TX hata ({tx_hash}): {e}")
        
        results.extend(batch_results)
        
        # Rate limit için bekle
        time.sleep(0.5)
    
    return results

4. Error Handling

Blockchain data’sı her zaman tutarlı olmayabilir:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def safe_web3_call(func, *args, **kwargs):
    """
    Web3 çağrılarını retry logic ile güvenli hale getirir
    """
    try:
        return func(*args, **kwargs)
    except Exception as e:
        print(f"Web3 call hatası: {e}")
        raise

Sonuç

Blockchain veri analizi araçlarını birleştirerek daha güçlü sistemler oluşturabilirsiniz. Web3.py + TheGraph + Dune kombinasyonu profesyonel analizler için idealdir.

Python ekosistemi, blockchain veri analizi için son derece güçlü araçlar sunmaktadır. Web3.py ile düşük seviyeli RPC çağrıları yapabilir, TheGraph ile kompleks sorguları hızlıca çalıştırabilir ve Dune Analytics ile SQL tabanlı analizler yapabilirsiniz.

Bu araçları bir araya getirerek profesyonel monitoring sistemleri, trading bot’ları ve DeFi analiz platformları geliştirebilirsiniz. Önemli olan, doğru aracı doğru iş için kullanmak ve best practice’leri takip etmektir.

Önerilen Kaynaklar

Bu gönderi CC BY 4.0 lisansı altındadır.