OpenClaw 数据清洗自动化：从脏数据到干净数据的一键处理

数据分析师 80% 的时间花在数据清洗上。OpenClaw 能自动化处理缺失值、异常值、格式不一致等常见问题，让你专注于数据分析本身。

数据清洗的痛点

问题	传统方式	OpenClaw 方式
缺失值处理	手动筛选填充	自动检测+智能填充
异常值检测	人工肉眼排查	统计方法自动识别
格式不统一	逐行修改	批量标准化
重复数据	手动去重	智能匹配去重
数据验证	抽检抽查	全量自动验证

---

技能配置

创建数据清洗技能 skills/data-cleaner/SKILL.md：

---
name: 数据清洗助手
description: 自动化数据清洗，处理缺失值、异常值、格式标准化、去重、验证
triggers:
  - 当用户说"清洗数据"时
  - 当用户说"处理数据"时
  - 当用户上传 CSV/Excel 文件时
permissions:
  - file_read
  - file_write
  - execute
---

# 数据清洗助手

## 清洗流程

### 1. 数据探查
- 自动识别数据类型
- 统计缺失值比例
- 检测异常值分布
- 分析数据分布

### 2. 缺失值处理
- 数值型：均值/中位数/众数填充
- 分类型：众数或"未知"填充
- 时间型：前向/后向填充
- 可配置删除阈值

### 3. 异常值处理
- IQR 方法检测
- Z-Score 方法检测
- 截断或删除处理

### 4. 格式标准化
- 日期格式统一
- 电话号码格式化
- 文本大小写统一
- 空白字符清理

### 5. 去重处理
- 完全重复删除
- 近似重复检测
- 主键去重

### 6. 数据验证
- 类型验证
- 范围验证
- 格式验证
- 业务规则验证

---

实战案例

案例 1：CSV 数据清洗

场景：清洗一份销售数据 CSV 文件

输入数据问题：

order_id,customer_name,amount,date,phone
001,张三,1500,2024-01-15,13812345678
002,李四,,2024/01/16,139-1234-5679
003,王五,2000,2024-01-17,
004,,2500,2024-01-18,13712345678
001,张三,1500,2024-01-15,13812345678  # 重复
005,赵六,999999,2024-01-19,1381234567   # 异常金额，电话位数错误

清洗脚本：

# data_cleaner.py
import pandas as pd
import numpy as np
from datetime import datetime

def clean_sales_data(input_path, output_path):
    """销售数据清洗"""
    
    # 1. 读取数据
    df = pd.read_csv(input_path)
    print(f"原始数据: {len(df)} 行")
    
    # 2. 去除完全重复
    df = df.drop_duplicates()
    print(f"去重后: {len(df)} 行")
    
    # 3. 处理缺失值
    # 金额用中位数填充
    df['amount'] = df['amount'].fillna(df['amount'].median())
    # 姓名标记为未知
    df['customer_name'] = df['customer_name'].fillna('未知客户')
    # 电话标记为缺失
    df['phone'] = df['phone'].fillna('缺失')
    
    # 4. 处理异常值
    # 金额超过 10000 视为异常，用中位数替换
    amount_median = df[df['amount'] < 10000]['amount'].median()
    df.loc[df['amount'] > 10000, 'amount'] = amount_median
    
    # 5. 格式标准化
    # 日期统一格式
    df['date'] = pd.to_datetime(df['date'], errors='coerce')
    df['date'] = df['date'].dt.strftime('%Y-%m-%d')
    
    # 电话号码标准化（只保留数字）
    df['phone'] = df['phone'].astype(str).str.replace(r'\D', '', regex=True)
    # 补全区号（假设都是大陆号码）
    df['phone'] = df['phone'].apply(lambda x: x if len(x) == 11 else '无效')
    
    # 6. 数据验证
    valid_mask = (
        (df['order_id'].notna()) &
        (df['amount'] > 0) &
        (df['date'].notna())
    )
    invalid_count = len(df) - valid_mask.sum()
    df = df[valid_mask]
    print(f"验证后: {len(df)} 行，移除 {invalid_count} 行无效数据")
    
    # 7. 保存清洗结果
    df.to_csv(output_path, index=False)
    print(f"清洗完成，保存到: {output_path}")
    
    # 8. 生成清洗报告
    report = generate_report(df, invalid_count)
    return report

def generate_report(df, invalid_count):
    """生成清洗报告"""
    report = f"""
# 数据清洗报告

## 清洗统计
- 有效数据: {len(df)} 行
- 移除数据: {invalid_count} 行
- 有效率: {len(df) / (len(df) + invalid_count) * 100:.1f}%

## 数据概览
- 订单数: {df['order_id'].nunique()}
- 客户数: {df['customer_name'].nunique()}
- 金额范围: {df['amount'].min():.0f} - {df['amount'].max():.0f}
- 日期范围: {df['date'].min()} - {df['date'].max()}

## 字段统计
| 字段 | 缺失数 | 缺失率 |
|------|--------|--------|
| order_id | 0 | 0% |
| customer_name | 0 | 0% |
| amount | 0 | 0% |
| date | 0 | 0% |
| phone | {len(df[df['phone'] == '缺失'])} | {len(df[df['phone'] == '缺失']) / len(df) * 100:.1f}% |
"""
    return report

if __name__ == "__main__":
    report = clean_sales_data("input.csv", "output.csv")
    print(report)

运行方式：

用户：帮我清洗销售数据 input.csv

Agent：
1. 分析数据结构和问题
2. 执行清洗脚本
3. 生成清洗报告
4. 保存清洗结果到 output.csv

---

案例 2：Excel 多表合并清洗

场景：合并多个 Excel 文件并清洗

# excel_merger_cleaner.py
import pandas as pd
import glob
import os

def merge_and_clean_excel(input_dir, output_file):
    """合并多个 Excel 并清洗"""
    
    all_data = []
    
    # 1. 读取所有 Excel 文件
    for file in glob.glob(f"{input_dir}/*.xlsx"):
        df = pd.read_excel(file)
        df['source_file'] = os.path.basename(file)
        all_data.append(df)
    
    # 2. 合并数据
    merged = pd.concat(all_data, ignore_index=True)
    print(f"合并数据: {len(merged)} 行")
    
    # 3. 统一列名（处理不同文件的列名差异）
    column_mapping = {
        '姓名': 'name',
        '名字': 'name',
        '客户名': 'name',
        '金额': 'amount',
        '交易额': 'amount',
        '日期': 'date',
        '交易日期': 'date'
    }
    merged.columns = [column_mapping.get(col, col) for col in merged.columns]
    
    # 4. 数据清洗
    # 处理日期格式差异
    merged['date'] = pd.to_datetime(merged['date'], errors='coerce')
    
    # 数值字段清洗
    merged['amount'] = pd.to_numeric(merged['amount'], errors='coerce')
    merged['amount'] = merged['amount'].fillna(0)
    
    # 去重（基于关键字段组合）
    merged = merged.drop_duplicates(subset=['name', 'date', 'amount'])
    
    # 5. 数据验证
    valid_mask = (
        merged['name'].notna() &
        (merged['name'] != '') &
        merged['date'].notna()
    )
    merged = merged[valid_mask]
    
    # 6. 保存结果
    merged.to_excel(output_file, index=False)
    print(f"清洗完成: {len(merged)} 行，保存到 {output_file}")
    
    return merged

# 使用示例
if __name__ == "__main__":
    merge_and_clean_excel("./data/raw/", "./data/cleaned.xlsx")

---

案例 3：JSON/API 数据清洗

场景：清洗 API 返回的嵌套 JSON 数据

# json_cleaner.py
import json
import pandas as pd
from datetime import datetime

def clean_json_api_data(json_file, output_csv):
    """清洗 API 返回的 JSON 数据"""
    
    # 1. 读取 JSON
    with open(json_file, 'r', encoding='utf-8') as f:
        data = json.load(f)
    
    # 2. 扁平化嵌套结构
    records = []
    for item in data['results']:
        record = {
            'id': item.get('id'),
            'name': item.get('profile', {}).get('name'),
            'email': item.get('profile', {}).get('email'),
            'age': item.get('profile', {}).get('age'),
            'city': item.get('location', {}).get('city'),
            'created_at': item.get('metadata', {}).get('created_at'),
            'status': item.get('status'),
            'tags': ','.join(item.get('tags', []))  # 数组转字符串
        }
        records.append(record)
    
    df = pd.DataFrame(records)
    
    # 3. 数据类型转换
    df['created_at'] = pd.to_datetime(df['created_at'], errors='coerce')
    df['age'] = pd.to_numeric(df['age'], errors='coerce')
    
    # 4. 清洗逻辑
    # 邮箱格式验证
    df['email_valid'] = df['email'].str.contains(r'^[\w\.-]+@[\w\.-]+\.\w+$', na=False)
    
    # 年龄范围验证
    df['age_valid'] = (df['age'] >= 0) & (df['age'] <= 120)
    
    # 状态标准化
    status_mapping = {
        'active': '活跃',
        'inactive': '不活跃',
        'pending': '待审核',
        'deleted': '已删除'
    }
    df['status'] = df['status'].map(status_mapping).fillna('未知')
    
    # 5. 标记无效数据
    df['is_valid'] = df['email_valid'] & df['age_valid'] & df['name'].notna()
    
    # 6. 分离有效和无效数据
    valid_df = df[df['is_valid']]
    invalid_df = df[~df['is_valid']]
    
    # 7. 保存结果
    valid_df.to_csv(output_csv, index=False)
    invalid_df.to_csv(output_csv.replace('.csv', '_invalid.csv'), index=False)
    
    print(f"有效数据: {len(valid_df)} 行")
    print(f"无效数据: {len(invalid_df)} 行")
    
    return valid_df, invalid_df

---

数据清洗工作流配置

Agent 配置示例

# .openclaw/agents/data-cleaner.yaml
name: 数据清洗助手
model: deepseek-chat
skills:
  - data-cleaner
  - browser-use  # 用于从网页抓取数据
  - file-manager

memory:
  enabled: true
  retention: 30d

tools:
  - file_read
  - file_write
  - execute
  - python_runner

自动化清洗任务

# .openclaw/tasks/auto_clean.yaml
name: 每日数据清洗
schedule: "0 9 * * *"  # 每天早上 9 点
actions:
  - type: read
    path: "./data/raw/*.csv"
  - type: clean
    steps:
      - remove_duplicates
      - fill_missing_values
      - standardize_formats
      - validate_data
  - type: write
    path: "./data/clean/cleaned_{date}.csv"
  - type: report
    output: "./reports/cleaning_{date}.md"

---

常见清洗规则速查

数据类型	常见问题	清洗方法
日期	格式不统一	pd.to_datetime() 自动解析
金额	包含符号、逗号	str.replace() + pd.to_numeric()
电话	格式混乱、位数不一	正则提取数字 + 长度验证
邮箱	格式错误、大小写	正则验证 + str.lower()
姓名	空格、特殊字符	str.strip() + 正则过滤
地址	格式不完整	分词 + 关键词匹配补全
ID	格式不统一	左补零或右截断统一长度

---

数据质量评分

def calculate_data_quality_score(df):
    """计算数据质量评分"""
    
    scores = {}
    
    # 1. 完整性（缺失值比例）
    completeness = 1 - df.isnull().sum().sum() / (df.shape[0] * df.shape[1])
    scores['完整性'] = completeness * 100
    
    # 2. 唯一性（重复值比例）
    uniqueness = 1 - df.duplicated().sum() / len(df)
    scores['唯一性'] = uniqueness * 100
    
    # 3. 一致性（格式统一度）
    # 检查日期、电话等字段的格式一致性
    consistency_scores = []
    if 'date' in df.columns:
        valid_dates = pd.to_datetime(df['date'], errors='coerce').notna().sum()
        consistency_scores.append(valid_dates / len(df))
    if 'phone' in df.columns:
        valid_phones = df['phone'].astype(str).str.match(r'^\d{11}$').sum()
        consistency_scores.append(valid_phones / len(df))
    scores['一致性'] = (sum(consistency_scores) / len(consistency_scores)) * 100 if consistency_scores else 100
    
    # 4. 准确性（业务规则验证）
    accuracy_scores = []
    if 'amount' in df.columns:
        valid_amounts = (df['amount'] >= 0).sum()
        accuracy_scores.append(valid_amounts / len(df))
    scores['准确性'] = (sum(accuracy_scores) / len(accuracy_scores)) * 100 if accuracy_scores else 100
    
    # 总分
    scores['总分'] = sum(scores.values()) / len(scores)
    
    return scores

# 使用示例
scores = calculate_data_quality_score(df)
for metric, score in scores.items():
    print(f"{metric}: {score:.1f}%")

---

总结

OpenClaw 数据清洗自动化的核心优势：

优势	说明
效率提升	批量处理，10 倍效率提升
质量保证	规则驱动，减少人为遗漏
可复用	清洗规则可保存复用
可追溯	自动生成清洗报告
可扩展	支持自定义清洗规则

下一步建议：

1. 创建你的第一个数据清洗技能
2. 配置自动化清洗任务
3. 建立数据质量监控