2. 数据源准备：如何获取并清洗DEX的Swap事件日志与交易对数据

做链上数据分析，第一步不是写代码，而是搞清楚数据从哪里来。

我刚开始接触DEX数据时，犯过一个低级错误——直接拿RPC节点拉原始日志，结果被限流到怀疑人生。后来才明白，数据源的选择和清洗，直接决定了分析结果的可靠性。

这一章，我们就来聊聊怎么拿到Uniswap和PancakeSwap的Swap数据，并且把它洗干净。

2.1 数据源的选择：节点 vs 索引器

获取链上数据，主要有两条路：

• 直接调用RPC节点：比如Infura、Alchemy、QuickNode。优点是实时性高，缺点是慢、贵、容易被限流。
• 使用区块链索引器：比如The Graph、Dune Analytics、BigQuery。优点是查询快、数据已解析，缺点是可能有几分钟延迟。

我个人习惯是：做实时监控用RPC，做历史分析用索引器。你想想看，如果我要分析过去三个月洗盘行为，用RPC拉几百万条日志，节点直接把我拉黑。

2.2 从RPC节点获取Swap事件日志

假设你非要自己拉原始日志，那得先理解Swap事件的结构。

Uniswap V2的Swap事件长这样：

event Swap(
    address indexed sender,
    uint256 amount0In,
    uint256 amount1In,
    uint256 amount0Out,
    uint256 amount1Out,
    address indexed to
);

V3版本更复杂，多了tick和liquidity信息。但核心逻辑不变：记录谁、在哪个池子、换了多少。

用Python获取的代码示例：

from web3 import Web3
import json

# 连接节点
w3 = Web3(Web3.HTTPProvider('https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY'))

# Uniswap V2 工厂合约地址
factory_address = '0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f'

# Swap事件的ABI
swap_abi = json.dumps([{
    "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"
}])

# 获取过去1000个区块的Swap事件
from_block = w3.eth.block_number - 1000
to_block = w3.eth.block_number

# 这里需要遍历所有交易对，实际项目中会先获取所有Pair地址
# 然后逐个监听
print(f"从区块 {from_block} 到 {to_block} 获取Swap事件...")

2.3 使用The Graph索引器获取数据

说实话，自己解析日志太累了。我推荐用The Graph，数据已经帮你解析好了。

Uniswap V2的Subgraph查询示例：

{
  swaps(
    first: 1000,
    orderBy: timestamp,
    orderDirection: desc,
    where: {
      timestamp_gt: 1700000000
    }
  ) {
    id
    timestamp
    transaction {
      id
    }
    pair {
      token0 {
        symbol
        id
      }
      token1 {
        symbol
        id
      }
    }
    amount0In
    amount0Out
    amount1In
    amount1Out
    sender
    to
  }
}

用Python请求：

import requests
import pandas as pd

url = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v2"

query = """
{
  swaps(first: 1000, orderBy: timestamp, orderDirection: desc) {
    id
    timestamp
    amount0In
    amount0Out
    amount1In
    amount1Out
    sender
    to
    pair {
      token0 { symbol id }
      token1 { symbol id }
    }
  }
}
"""

response = requests.post(url, json={'query': query})
data = response.json()['data']['swaps']

# 转成DataFrame
df = pd.DataFrame(data)
print(f"获取到 {len(df)} 条Swap记录")
print(df.head())

2.4 数据清洗：从原始日志到可用数据

原始数据拿到手，你会发现一堆问题。嗯，这里要注意几个坑。

2.4.1 时间戳转换

链上时间戳是Unix格式，得转成人类可读的时间：

df['datetime'] = pd.to_datetime(df['timestamp'], unit='s')
print(df[['datetime', 'amount0In', 'amount0Out']].head())

2.4.2 金额转换与精度处理

这是最容易出错的地方。链上金额是原始值，需要除以10的decimals次方。

# 假设token0是USDC（6位小数），token1是WETH（18位小数）
def convert_amount(raw_amount, decimals):
    return float(raw_amount) / (10 ** decimals)

# 实际项目中，需要从合约获取decimals
# 这里简化处理
df['amount0In_adj'] = df['amount0In'].apply(lambda x: convert_amount(x, 6))
df['amount1In_adj'] = df['amount1In'].apply(lambda x: convert_amount(x, 18))

2.4.3 识别Swap方向

Swap事件里，amount0In和amount0Out不会同时非零。判断方向很简单：

• 如果amount0In > 0，说明用户用token0换token1
• 如果amount0Out > 0，说明用户用token1换token0

def get_swap_direction(row):
    if float(row['amount0In']) > 0:
        return f"{row['pair.token0.symbol']} -> {row['pair.token1.symbol']}"
    else:
        return f"{row['pair.token1.symbol']} -> {row['pair.token0.symbol']}"

df['direction'] = df.apply(get_swap_direction, axis=1)

2.5 交易对数据：获取池子信息

光有Swap事件还不够，你得知道每个池子的流动性、价格范围。这些信息来自Pair合约。

关键数据字段：

字段	说明	获取方式
reserve0 / reserve1	池子中两种代币的储备量	Pair合约的getReserves()
totalSupply	LP代币总供应量	Pair合约的totalSupply()
kLast	上次更新时的k值（用于手续费计算）	Pair合约的kLast()
price0CumulativeLast	累计价格（用于TWAP）	Pair合约的price0CumulativeLast()

获取池子储备量的代码：

# 假设已经知道Pair合约地址
pair_address = '0x...'  # 某个交易对地址
pair_contract = w3.eth.contract(address=pair_address, abi=pair_abi)

reserves = pair_contract.functions.getReserves().call()
reserve0 = reserves[0] / (10 ** decimals0)
reserve1 = reserves[1] / (10 ** decimals1)

print(f"池子储备: {reserve0} token0, {reserve1} token1")
print(f"当前价格: {reserve1 / reserve0} token1 per token0")

2.6 数据存储与清洗流程

我一般把清洗后的数据存成Parquet格式，比CSV快10倍以上。清洗流程如下：

1. 从The Graph拉取原始Swap数据
2. 转换时间戳为datetime
3. 根据代币decimals调整金额
4. 识别Swap方向
5. 合并池子储备量数据
6. 计算交易价格（amountOut / amountIn）
7. 过滤异常交易（比如金额为0的）
8. 保存到本地文件

def clean_swap_data(raw_df, token0_decimals, token1_decimals):
    df = raw_df.copy()
    
    # 时间戳转换
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='s')
    
    # 金额转换
    df['amount0_in_adj'] = df['amount0In'].astype(float) / (10 ** token0_decimals)
    df['amount0_out_adj'] = df['amount0Out'].astype(float) / (10 ** token0_decimals)
    df['amount1_in_adj'] = df['amount1In'].astype(float) / (10 ** token1_decimals)
    df['amount1_out_adj'] = df['amount1Out'].astype(float) / (10 ** token1_decimals)
    
    # 计算交易价格
    df['price'] = df.apply(
        lambda x: x['amount1_out_adj'] / x['amount0_in_adj'] 
        if x['amount0_in_adj'] > 0 
        else x['amount1_in_adj'] / x['amount0_out_adj'],
        axis=1
    )
    
    # 过滤异常
    df = df[df['price'] > 0]
    df = df[df['price'] < 1e10]  # 过滤极端值
    
    return df

2.7 本章知识体系

下面这张图，帮你理清数据获取与清洗的完整流程：

说白了，数据准备是整个分析流程的地基。地基没打好，后面再漂亮的模型也是空中楼阁。我见过太多人花80%时间写分析代码，结果数据源选错了，或者精度没调对，最后结论全废。

记住一句话：垃圾进，垃圾出。数据清洗花的时间，永远值得。