数据清洗

本技能让 AI Agent 系统地将原始数据集清洗和预处理为可供分析的形式。Agent 处理缺失值、重复记录、数据类型不匹配、格式不一致、离群值处理，以及归一化。它还可以执行校验模式（validation schema）以确保持续的数据质量。主要工具链是 pandas，并借助 pyjanitor 和 great_expectations 做高级校验。

工作流

1. 摄取并探查原始数据。 加载数据集并立即生成质量报告：每列空值计数、识别重复行、对照预期模式检查数据类型，以及标记含混合类型的列。此探查驱动后续每一个清洗决策。

2. 处理缺失值。 根据数据类型和缺失模式，按列应用策略。对缺失少于 5% 的数值列，使用中位数插补。对类别列，使用众数或专门的"未知"类别。对缺失超过 40% 的列，标记为可能删除，并在丢弃前咨询用户。

3. 去除重复并解决冲突。 识别精确重复和近似重复（例如，仅空白或大小写不同的行）。对精确重复，保留首次出现的记录。对近似重复，应用带可配置相似度阈值的模糊匹配，并按时效性或完整性合并冲突值。

4. 纠正数据类型并标准化格式。 将列强制转换为其预期类型——将日期字符串解析为 datetime 对象，将数值字符串转换为浮点数，并将类别值归一化为规范形式。标准化电话号码、邮政编码和货币表示等格式。

5. 检测并处理离群值。 对对称分布使用 IQR 法（1.5 倍），对正态分布数据使用 z 分数。提供三种处理选项：在边界值处截断（winsorization）、替换为空值以便后续插补，或仅标记模式（注释但保留原始值）。

6. 校验清洗后的输出。 将清洗后的数据集通过校验规则——非空约束、范围检查、唯一性约束和参照完整性。报告任何剩余违规，并将干净的数据集与一份清洗日志一同保存，日志记录所应用的每一次转换。

支持技术

• pandas — 核心数据操作与类型强制转换
• pyjanitor — 清洗操作的链式方法便利
• great_expectations — 模式校验与数据质量检查
• fuzzywuzzy — 近似重复检测的模糊字符串匹配
• numpy — 离群值检测的数值运算

使用方式

为 Agent 提供原始数据集的文件路径，以及可选的 schema 定义，指明预期的列类型、有效范围和唯一性约束。Agent 将产出清洗后的文件和转换日志。

示例

示例 1：用 pandas 清洗混乱的 CSV

import pandas as pd
import numpy as np

# Load raw data
df = pd.read_csv("messy_orders.csv")
print(f"Raw shape: {df.shape}")  # (2340, 8)
print(df.isnull().sum())
# order_id         0
# customer_name   12
# email           45
# order_date      18
# amount          23
# status           0
# region          67
# discount         0

# 1. Fix data types — order_date has mixed formats
df["order_date"] = pd.to_datetime(df["order_date"], format="mixed", dayfirst=False)
df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

# 2. Handle missing values
df["customer_name"] = df["customer_name"].fillna("Unknown")
df["email"] = df["email"].fillna("missing@placeholder.com")
df["amount"] = df["amount"].fillna(df["amount"].median())
df["region"] = df["region"].fillna(df["region"].mode()[0])
df["order_date"] = df["order_date"].fillna(method="ffill")

# 3. Remove duplicates
before = len(df)
df = df.drop_duplicates(subset=["order_id"], keep="first")
print(f"Removed {before - len(df)} duplicate orders")  # Removed 34 duplicate orders

# 4. Standardize categorical values
df["status"] = df["status"].str.strip().str.lower().replace({
    "shipped": "shipped", "ship": "shipped",
    "cancelled": "cancelled", "canceled": "cancelled",
    "pending": "pending", "pend": "pending"
})
df["region"] = df["region"].str.strip().str.title()

# 5. Outlier treatment — cap amounts at IQR bounds
Q1 = df["amount"].quantile(0.25)
Q3 = df["amount"].quantile(0.75)
IQR = Q3 - Q1
lower, upper = Q1 - 1.5 * IQR, Q3 + 1.5 * IQR
df["amount"] = df["amount"].clip(lower=lower, upper=upper)

print(f"Clean shape: {df.shape}")  # (2306, 8)
df.to_csv("clean_orders.csv", index=False)

示例 2：带模式强制的数据校验流水线

import great_expectations as gx

context = gx.get_context()

# Define a validation suite
suite = context.add_expectation_suite("orders_validation")

# Add expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0.01, max_value=50000.00
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="status", value_set=["pending", "shipped", "delivered", "cancelled"]
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email", regex=r"^[^@]+@[^@]+.[^@]+$"
    )
)

# Run validation against cleaned data
results = context.run_validation(suite, batch=gx.read_csv("clean_orders.csv"))

print(f"Success: {results.success}")
print(f"Passed: {results.statistics['successful_expectations']}/{results.statistics['evaluated_expectations']}")
# Success: True
# Passed: 5/5

最佳实践

• 在任何转换之前务必保存一份原始数据副本——清洗应当是可复现的，而非破坏性的。
• 记录每次转换及其计数（例如"用中位数 312.45 填充了 amount 列的 23 个空值"），以创建可审计的清洗轨迹。
• 优先选择领域知情插补而非机械默认值；当某列的缺失模式非随机（MNAR）时，咨询用户。
• 尽早且频繁地校验——在每个清洗阶段后运行模式检查，而不仅仅在最后。
• 将清洗视为迭代过程：第一轮捕获明显问题，但下游分析经常会浮现新的问题。
• 使用 errors="coerce" 配合 pd.to_numeric 和 pd.to_datetime，将转换失败以 NaN 形式暴露，而非崩溃。

边缘情况

• 完全为空的列。 如果某列 100% 为空，自动丢弃它并记录警告，而非对零信息尝试插补。
• 重复的列名。 Pandas 静默允许重复列名。加载时检测它们，并在任何操作之前用后缀（_1、_2）重命名。
• 编码问题。 如果 read_csv 抛出 UnicodeDecodeError，先用 encoding="latin-1"，再试 encoding="cp1252"，并记录哪种编码成功。
• 含多种格式的日期列。 当单列同时包含 "2024-01-15"、"01/15/2024" 和 "Jan 15, 2024" 时，使用 pd.to_datetime(col, format="mixed") 并通过抽查验证解析结果。
• 以字符串形式存储、带货币符号的数值列。 在类型强制转换前，剥离 $、€、逗号和空白：df["price"].str.replace(r"[$€,s]", "", regex=True).astype(float)。