tools

The Graph：Web3数据分析工具

The Graph成为了导航去中心化金融（DeFi）复杂世界的首选工具。它允许我分析协议活动、代币流动和智能合约交互，将混乱转化为清晰。

一键发币： SUI | SOL | BNB | ETH | BASE | ARB | OP | POLYGON | AVAX | FTM | OK

几年前，当我开始从事数据分析师的工作时，我发现自己在探索区块链交易、智能合约和代币流动，试图理解它们是如何结合在一起的。Web3的承诺——去中心化、透明性和无限的可能性——无疑是引人注目的，但将原始数据转化为可操作的见解却感觉像是一项艰巨的挑战。这感觉就像在解码一种新的语言。

就在这时，我发现了像The Graph这样的工具。它彻底改变了我对区块链数据的处理方式。特别是，The Graph成为了导航去中心化金融（DeFi）复杂世界的首选工具。它允许我分析协议活动、代币流动和智能合约交互，将混乱转化为清晰。结合Python在数据操作和可视化的多功能性，这些工具开启了揭示见解并理解区块链生态系统的新可能性。

在这篇文章中，我将向你展示The Graph的索引功能和Python的数据处理能力如何使Web3数据分析不仅易于入手，而且令人兴奋。无论你是开发者、数据分析师，还是仅仅对区块链感兴趣，本指南都将帮助你解锁区块链数据的潜力，并将其转化为有意义的见解。

1、The Graph

要深入研究Web3数据分析，了解使这一切成为可能的工具至关重要。The Graph是一个去中心化的协议，简化了查询和索引区块链数据的过程。可以将其视为一个专为区块链设计的搜索引擎：它允许您从Uniswap、Aave或Compound等协议中查询数据。

例如，计算Uniswap V2池中的 总锁定价值（TVL）需要聚合来自多个来源的数据：

池交易：你需要处理由池发出的每个交换、铸造和销毁事件。
代币价格：你需要实时获取池中代币的价格，这些通常来自外部来源，如Chainlink。

如果没有The Graph，这个过程将涉及：

扫描每一块包含相关事件的区块。
存储和处理TB级别的原始数据。
构建和维护一个集中式服务器来处理查询。

这种方法资源密集、缓慢且容易出错。

相反，The Graph使用子图对区块链数据进行索引——自定义API，提取、处理并存储区块链数据，以便通过GraphQL轻松查询。

去中心化应用程序（dApp）在智能合约上发起交易，例如在Uniswap上交换代币或将资金存入池中。
当智能合约处理交易时，它会发出一个或多个事件。这些事件包含有关交易的关键数据，例如代币转移、流动性变化或用户操作。
Graph节点持续扫描网络以查找新块。它寻找与定义的子图匹配的事件（例如，与特定DeFi协议相关的事件）。
当节点找到相关事件时，它运行您提供的映射处理器。这些映射将原始事件数据转换为结构化的实体（例如，代币转移、用户余额），并将其存储在The Graph的数据库中。

通过The Graph的数据库对区块链数据进行索引和结构化后，查询它变得无缝且高效。通过利用子图，我们可以轻松地从去中心化协议中提取有意义的见解，而无需手动处理原始区块链数据。在下一节中，我们将探讨如何与The Graph交互并检索区块链数据。

2、从The Graph查询数据

在本节中，我们将演示如何查询USDC/WETH流动性池的数据——这是Uniswap V2中总锁定价值（TVL）最高和交易量最大的池之一。我们需要在Python中使用requests库发送API请求并检索数据。The Graph提供了一个免费层级API，允许用户查询区块链数据。

import requests  

# 查询URL  
api_key = "your_api_key_here"  
url = (  
    f"https://gateway.thegraph.com/api/{api_key}/"  
    f"subgraphs/id/EYCKATKGBKLWvSfwvBjzfCBmGwYNdVkduYXVivCsLRFu"  
)  

# ETH/USD 池地址  
poolID = "0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc" # 大小写敏感  

# GraphQL 查询  
query = f"""  
{{  
  pairHourDatas(  
    first: 1  
    orderBy: hourStartUnix  
    orderDirection: desc  
    where: {{pair_: {{id: "{poolID}"}}}}  
  ) {{  
    reserveUSD  
    hourlyVolumeUSD  
    hourStartUnix  
  }}  
}}  
"""  

# 发送POST请求  
response = requests.post(  
    url,  
    json={"query": query},  # 将查询作为JSON发送到请求体中  
    headers={"Content-Type": "application/json"}  # 设置内容类型  
)  

# 检查请求是否成功  
if response.status_code == 200:  
    # 打印响应数据  
    print(response.json())  
else:  
    print(f"错误: {response.status_code}")  
    print(response.text)

{  
   "data": {  
      "pairHourDatas": [  
         {  
            "hourStartUnix": 1738764000, # 2025年2月5日星期三14:00:00  
            "hourlyVolumeUSD": "106629.7551700879597672008337821572",  
            "reserveUSD": "43066639.73815987788929369271457476"  
         }  
      ]  
   }  
}

The Graph API 每次查询限制为1000行，这意味着在这种情况下，我们只能一次检索过去1000小时的数据。要访问更早的数据，我们需要分页查询。

此外，应实现错误处理以管理超时、速率限制或网络错误等问题，确保在API交互期间应用程序保持稳定。

def fetchUniswapV2Data(pool_id, n_hours, api_key):  
    url = (  
        f"https://gateway.thegraph.com/api/{api_key}/"  
        f"subgraphs/id/EYCKATKGBKLWvSfwvBjzfCBmGwYNdVkduYXVivCsLRFu"  
    )  
    all_data = []  
    last_timestamp = None  
    remaining_hours = n_hours  
    batch_size = 1000  

    while remaining_hours > 0:  
        fetch_count = min(batch_size, remaining_hours)  
        where_clause = f'pair_: {{id: "{pool_id}"}}'  
        if last_timestamp:  
            where_clause += f', hourStartUnix_lt: {last_timestamp}'  

        query = f"""  
        {{  
          pairHourDatas(  
            first: {fetch_count}  
            orderBy: hourStartUnix  
            orderDirection: desc  
            where: {{{where_clause}}}  
          ) {{  
            reserveUSD  
            hourlyVolumeUSD  
            hourStartUnix  
          }}  
        }}  
        """  

        try:  
            response = requests.post(url, json={"query": query},  
                                     headers={"Content-Type": "application/json"})  

            if response.status_code == 200:  
                data = response.json().get("data", {}).get("pairHourDatas", [])  
                if not data:  
                    break  
                all_data.extend(data)  
                last_timestamp = data[-1]["hourStartUnix"]  
                remaining_hours -= len(data)  
            else:  
                print(f"错误: {response.status_code}")  
                print(response.text)  
                break  
        except requests.exceptions.RequestException as e:  
            print(f"请求失败: {e}")  
            break  

    return all_data

import requests  

# API密钥  
api_key = "your_api_key_here"  

# ETH/USD 池地址  
poolID = "0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc" # 大小写敏感  

# 获取最近3000小时的数据  
n_hours = 5   

result = fetchUniswapV2Data(poolID, 5, api_key)

[  
   {  
      "hourStartUnix": 1738767600, # 2025年2月5日星期三15:00:00  
      "hourlyVolumeUSD": "247216.4302563188235059872851417612",  
      "reserveUSD": "42744075.8906440481697076541020688"  
   },  
   {  
      "hourStartUnix": 1738764000, # 2025年2月5日星期三14:00:00  
      "hourlyVolumeUSD": "207258.0978585019661633977200846229",  
      "reserveUSD": "42883598.18480552341379416668872143"  
   },  
   {  
      "hourStartUnix": 1738760400, # 2025年2月5日星期三13:00:00  
      "hourlyVolumeUSD": "81368.1526878978402360147216202579",  
      "reserveUSD": "43096349.54637477091496254022761866"  
   },  
   {  
      "hourStartUnix": 1738756800, # 2025年2月5日星期三12:00:00  
      "hourlyVolumeUSD": "105545.9968445220489314712746966698",  
      "reserveUSD": "84497199.48512941620683936820457894"  
   },  
   {  
      "hourStartUnix": 1738753200, # 2025年2月5日星期三11:00:00  
      "hourlyVolumeUSD": "407343.6367039131634435166720976953",  
      "reserveUSD": "84580636.21755707303330542581266617"  
   }  
]

3、数据清洗数据清洗与可视化

在从The Graph API获取数据后，下一步是对其进行清理和格式化以供分析。API响应返回的数值为字符串类型。我们首先将响应转换为Pandas的DataFrame，然后将相关列转换为适当的数据类型，如float64或int64，以确保准确的计算和可视化。此外，我们将Unix时间戳转换为人类可读的日期时间格式，以便更轻松地分析和解释数据。

# 将结果转换为pandas DataFrame  
df = pd.DataFrame(result)  

# 将列转换为float64或int64  
df = df.apply(pd.to_numeric)  

# 创建新列，包含转换后的Unix时间  
df['date'] = pd.to_datetime(df['hourStartUnix'], unit='s')

|   hourStartUnix |   hourlyVolumeUSD |   reserveUSD | date                |  
|----------------:|------------------:|-------------:|:--------------------|  
|      1738767600 |          247216   |  4.27441e+07 | 2025-02-05 15:00:00 |  
|      1738764000 |          207258   |  4.28836e+07 | 2025-02-05 14:00:00 |  
|      1738760400 |           81368.2 |  4.30963e+07 | 2025-02-05 13:00:00 |  
|      1738756800 |          105546   |  8.44972e+07 | 2025-02-05 12:00:00 |  
|      1738753200 |          407344   |  8.45806e+07 | 2025-02-05 11:00:00 |

在清理数据后，我们可以轻松地操作它并使用描述性统计来探索其特征。这有助于我们理解关键指标，如总锁定价值（TVL）和交易量及其分布和总体趋势。通过计算描述性统计，我们可以深入了解样本的集中趋势、变异性以及异常值。例如，交易量的大标准差表明了活跃期的存在，这可能对应于TVL的急剧下降和上升。更仔细地检查这些事件可能会揭示流动性移动的模式以及影响池子的外部市场因素。

# 描述性统计  
descStats = df[["reserveUSD", "hourlyVolumeUSD"]].describe()

|       |     reserveUSD |   hourlyVolumeUSD |  
|:------|---------------:|------------------:|  
| count |           5000 |              5000 |  
| mean  |    7.91768e+07 |            302822 |  
| std   |    1.79741e+07 |        1.9486e+06 |  
| min   |    4.27482e+07 |           2567.23 |  
| 25%   |    7.13574e+07 |           82372.5 |  
| 50%   |    8.67742e+07 |            157175 |  
| 75%   |    9.05591e+07 |            293179 |  
| max   |    1.04645e+08 |       1.23407e+08 |

在清理和分析数据之后，下一步是可视化数据以揭示模式、趋势和见解。可视化是理解和传达大数据集发现的强大工具。

import plotly.graph_objects as go  

x = df['date'].tolist()  
y = df['reserveUSD'].tolist()  

fig = go.Figure(  
    data=go.Scatter(  
        x=x,   
        y=y,   
    )  
)  

fig.update_layout(  
    xaxis_title='日期',  
    yaxis_title='TVL（以美元计）',  
    title_x=0.5,  
    width=1400,    
    height=750  
)  

fig.show()

Uniswap V2 USDC/WETH池总锁定价值（TVL）（最近5000小时）

TVL显示波动，有稳定期、急剧下降和突然恢复的情况。值得注意的是，在2024年12月和2025年1月期间TVL有显著下降，这可能表明流动性迁移、大规模提款或主要市场事件，如协议被攻击或监管变化。同样，2025年2月的急剧恢复表明流动性突然流入，可能是由于投资者信心恢复、套利机会或战略基金重新配置。

与其手动跟踪TVL数据，不如使用链上分析工具如The Graph来高效获取和分析这些信息。