数据清洗与存储：Pandas处理链上数据、清洗空值与异常值、数据入库到SQLite/MySQL

链上数据拿到手，第一件事是什么？

不是分析，是清洗。我见过太多人，数据还没洗干净就急着跑模型，结果跑出来的结论全是错的。说白了，链上数据比传统互联网数据脏得多——空值、异常值、重复记录、格式不统一，这些问题几乎每张表都有。

为什么链上数据这么脏？

你想想看，区块链上的数据是去中心化产生的。每个节点、每个钱包、每个合约都可能产生不同格式的数据。比如一个转账事件，有的合约会记录from和to，有的只记录sender和recipient。还有那些失败的交易，数据字段经常是空的。

我个人习惯，拿到数据后先做三件事：

• 看结构——用df.info()快速了解字段类型和缺失情况
• 看分布——用df.describe()检查数值范围是否合理
• 看样本——用df.head(10)随机抽几行，肉眼扫一遍

Pandas处理链上数据的常用操作

我在项目中遇到过一张交易表，里面有500万行数据，光gas_price字段就有30%是空的。嗯，这里要注意，空值处理不能一刀切。

1. 空值处理

import pandas as pd

# 读取链上数据
df = pd.read_csv('onchain_transactions.csv')

# 查看空值比例
null_ratio = df.isnull().sum() / len(df)
print(null_ratio)

# 对于数值型字段，用中位数填充（抗异常值）
df['gas_price'].fillna(df['gas_price'].median(), inplace=True)

# 对于类别型字段，用众数填充
df['token_symbol'].fillna(df['token_symbol'].mode()[0], inplace=True)

# 如果某字段空值超过70%，直接删除该字段
if null_ratio['some_field'] > 0.7:
    df.drop('some_field', axis=1, inplace=True)

2. 异常值检测

链上数据最怕什么？闪电贷攻击、机器人刷量、异常大额转账。这些都会让数据分布变得很奇怪。

# 用IQR方法检测异常值
Q1 = df['amount'].quantile(0.25)
Q3 = df['amount'].quantile(0.75)
IQR = Q3 - Q1

lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 标记异常值
df['is_outlier'] = (df['amount'] < lower_bound) | (df['amount'] > upper_bound)

# 查看异常值占比
print(f"异常值占比: {df['is_outlier'].mean():.2%}")

我曾经遇到过一个案例，某条链上的转账数据，amount字段出现了负数。你想想看，转账金额怎么可能为负？后来查出来是某个合约的bug，把手续费算反了。这种异常值必须剔除。

数据入库：从Pandas到SQLite/MySQL

数据洗干净了，接下来就是存起来。我个人习惯用SQLite做本地开发，用MySQL做生产环境。为什么？SQLite轻量，不需要装服务端，适合单机调试；MySQL支持并发，适合团队协作。

1. 写入SQLite

import sqlite3

# 连接SQLite数据库（文件型）
conn = sqlite3.connect('onchain_data.db')

# 将DataFrame写入表
df.to_sql('transactions', conn, if_exists='replace', index=False)

# 验证写入结果
result = pd.read_sql('SELECT COUNT(*) FROM transactions', conn)
print(f"写入行数: {result.iloc[0, 0]}")

conn.close()

2. 写入MySQL

from sqlalchemy import create_engine

# 连接MySQL（需要提前安装pymysql）
engine = create_engine('mysql+pymysql://user:password@localhost:3306/onchain_db')

# 分批写入，避免内存溢出
chunk_size = 10000
for i in range(0, len(df), chunk_size):
    chunk = df.iloc[i:i+chunk_size]
    chunk.to_sql('transactions', engine, if_exists='append', index=False)
    print(f"已写入 {i+len(chunk)} 行")

实战：清洗一条完整的链上交易数据

我们拿以太坊的转账记录来演示。假设原始数据长这样：

字段名	类型	常见问题
block_number	整数	偶尔出现负数（数据源错误）
from_address	字符串	大小写不统一，有的带0x前缀有的不带
to_address	字符串	合约地址可能为空
value	浮点数	单位不统一（wei/ether）
gas_used	整数	失败交易为0

# 完整清洗流程
def clean_onchain_data(df):
    # 1. 统一地址格式（小写+0x前缀）
    df['from_address'] = df['from_address'].str.lower().str.strip()
    df['to_address'] = df['to_address'].str.lower().str.strip()
    
    # 2. 统一金额单位（全部转为ether）
    df['value_ether'] = df['value'] / 1e18
    
    # 3. 处理空地址
    df['to_address'].fillna('0x0000000000000000000000000000000000000000', inplace=True)
    
    # 4. 过滤异常区块号
    df = df[df['block_number'] > 0]
    
    # 5. 处理gas_used为0的情况（失败交易）
    df['is_success'] = df['gas_used'] > 0
    
    return df

# 执行清洗
df_clean = clean_onchain_data(df)
print(f"清洗前: {len(df)} 行, 清洗后: {len(df_clean)} 行")

知识体系图

下面这张图总结了本章的核心流程：

这张图把整个流程串起来了。从原始数据开始，经过空值处理、异常值检测、格式统一，最后入库。同时别忘了保留原始数据做备份。

---

公众号:蓝海数据掘金营，微信deep3321