Data Cleaning

此技能使AI代理能够系统地将原始数据集清洗并预处理为分析就绪的形式。该代理处理缺失值、重复记录、数据类型不匹配、格式不一致、异常值处理和标准化。它还可以强制执行验证模式以确保持续的数据质量。主要工具链是pandas，同时使用pyjanitor和great_expectations进行高级验证。

Workflow

导入并分析原始数据。加载数据集并立即生成质量报告：统计每列中的空值数量、识别重复行、检查数据类型是否符合预期模式，并标记混合类型的列。此分析驱动后续所有清洗决策。
处理缺失值。根据数据类型和缺失模式为每列应用策略。对于缺失率低于5%的数值列，使用中位数插补；对于分类列，使用众数或专门的“未知”类别；对于缺失率超过40%的列，标记以供潜在删除，并在删除前咨询用户。
去除重复项并解决冲突。识别完全重复和近似重复（例如仅在空格或大小写上不同的行）。对于完全重复项，保留第一次出现的记录；对于近似重复项，使用可配置相似度阈值进行模糊匹配，并通过时效性或完整性合并冲突值。
修正数据类型并标准化格式。强制转换列到其预期类型——将日期字符串解析为datetime对象、将数值字符串转换为浮点数，并将分类值标准化为规范形式。标准化电话号码、邮政编码和货币表示等格式。
检测并处理异常值。对对称分布使用IQR方法（1.5倍），对正态分布数据使用z-score。提供三种处理选项：边界值截断（winsorization）、替换为null以供后续插补、仅标记模式（标注但保留原始值）。
验证清洗后的输出。运行清洗后的数据集通过验证规则——非空约束、范围检查、唯一性约束和参照完整性。报告任何剩余违规情况，并保存清洗后的数据集及记录所有转换操作的清洗日志。

支持的技术

pandas — 核心数据操作和类型强制转换
pyjanitor — 清洗操作的方法链便利工具
great_expectations — 模式验证和数据质量检查
fuzzywuzzy — 用于近似重复检测的模糊字符串匹配
numpy — 异常值检测的数值运算

Usage

向代理提供原始数据集的文件路径，并可选择提供一个模式定义，指定预期列类型、有效范围和唯一性约束。代理将生成清洗后的文件和转换日志。

Examples

示例 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 而不是崩溃。

Edge Cases

完全为空的列。如果某列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)。