链上数据分析：Dune、Flipside 与数据仓库

Transfer(address indexed from, address indexed to, uint256 value)
topics[0] = keccak256("Transfer(address,address,uint256)")
          = 0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef
topics[1] = from（左补零到 32 字节）
topics[2] = to
data      = value（未索引参数，ABI 编码）

# 用 cast 计算选择器与事件哈希
cast sig "swapExactTokensForTokens(uint256,uint256,address[],address,uint256)"
# 0x38ed1739

cast keccak "Transfer(address,address,uint256)"
# 0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef

from eth_abi import decode
from eth_utils import keccak

def decode_call(abi, data: bytes):
    selector = data[:4]
    for item in abi:
        if item.get("type") != "function":
            continue
        sig = f"{item['name']}({','.join(i['type'] for i in item['inputs'])})"
        if keccak(text=sig)[:4] == selector:
            types = [i["type"] for i in item["inputs"]]
            return item["name"], decode(types, data[4:])
    return None, None

CALL  router.swapExactInput
 ├─ STATICCALL  pool.slot0            # 读价格
 ├─ CALL        pool.swap            # 真正换币
 │   ├─ CALL    token0.transfer
 │   └─ CALL    token1.transfer
 └─ CALL        refund

SELECT
    date_trunc('day', block_time) AS day,
    COUNT(*)                      AS trades,
    SUM(amount_usd)               AS volume_usd
FROM dex.trades
WHERE block_time >= now() - interval '7' day
  AND project = 'uniswap'
  AND blockchain = 'ethereum'
GROUP BY 1
ORDER BY 1;

-- 把重查询物化为视图，定时刷新
CREATE MATERIALIZED VIEW dune.myteam.daily_dex_volume AS
SELECT block_date, project, SUM(amount_usd) AS vol
FROM dex.trades
WHERE block_date >= date '2024-01-01'
GROUP BY 1, 2;

# 执行查询并轮询结果
curl -X POST "https://api.dune.com/api/v1/query/<QUERY_ID>/execute" \
  -H "X-Dune-API-Key: $DUNE_KEY"

# 拉取执行结果
curl "https://api.dune.com/api/v1/execution/<EXECUTION_ID>/results" \
  -H "X-Dune-API-Key: $DUNE_KEY" | jq '.result.rows[:3]'

采集层：Erigon / Reth 全节点 → 提取 blocks/txs/receipts/traces
传输层：Kafka 或直接写对象存储（Parquet）
存储层：ClickHouse（分析） + PostgreSQL（元数据）
计算层：dbt 做建模与测试
调度层：Airflow / Dagster 做增量编排
服务层：Metabase / Superset 做可视化

-- ClickHouse：按块号建分区，便于回滚与剪枝
CREATE TABLE transactions (
    block_number UInt64,
    block_time   DateTime,
    tx_hash      String,
    from_addr    String,
    to_addr      Nullable(String),
    value        UInt256,
    status       UInt8
)
ENGINE = ReplacingMergeTree
PARTITION BY intDiv(block_number, 1000000)   -- 每百万块一个分区
ORDER BY (block_number, tx_hash);

-- 回滚：删掉受影响的尾部块再重放
ALTER TABLE transactions DELETE WHERE block_number >= 20000000;

SELECT
    SUM(amount / pow(10, decimals)) AS amount_normalized
FROM token_transfers t
JOIN tokens tk ON tk.address = t.token_address;

models:
  - name: dex_trades
    columns:
      - name: tx_hash
        tests: [unique, not_null]
      - name: amount_usd
        tests:
          - not_null
    tests:
      - dbt_utils.expression_is_true:
          expression: "block_time >= '2020-01-01'"

-- 反例：全表扫描 + SELECT *
SELECT * FROM ethereum.transactions WHERE to = '0x...';

-- 正例：分区过滤 + 列裁剪 + 覆盖索引思路
SELECT block_time, tx_hash, value
FROM ethereum.transactions
WHERE block_time >= now() - interval '1' day
  AND to = '0x...';