📖 技能介紹

name: data-cleaning description: 通過處理缺失值、去除重複、糾正型別、解決離群值並執行校驗模式，清洗與預處理資料集。 license: MIT metadata: author: awesome-ai-agent-skills version: 1.0.0

資料清洗

本技能讓 AI Agent 系統地將原始資料集清洗和預處理為可供分析的形式。Agent 處理缺失值、重複記錄、資料型別不匹配、格式不一致、離群值處理，以及歸一化。它還可以執行校驗模式（validation schema）以確保持續的資料質量。主要工具鏈是 pandas，並藉助 pyjanitor 和 great_expectations 做高階校驗。

工作流

攝取並探查原始資料。 載入資料集並立即生成質量報告：每列空值計數、識別重複行、對照預期模式檢查資料型別，以及標記含混合型別的列。此探查驅動後續每一個清洗決策。
處理缺失值。 根據資料型別和缺失模式，按列應用策略。對缺失少於 5% 的數值列，使用中位數插補。對類別列，使用眾數或專門的"未知"類別。對缺失超過 40% 的列，標記為可能刪除，並在丟棄前諮詢使用者。
去除重複並解決衝突。 識別精確重複和近似重複（例如，僅空白或大小寫不同的行）。對精確重複，保留首次出現的記錄。對近似重複，應用帶可配置相似度閾值的模糊匹配，並按時效性或完整性合併衝突值。
糾正資料型別並標準化格式。 將列強制轉換為其預期型別——將日期字串解析為 datetime 物件，將數值字串轉換為浮點數，並將類別值歸一化為規範形式。標準化電話號碼、郵政編碼和貨幣表示等格式。
檢測並處理離群值。 對對稱分佈使用 IQR 法（1.5 倍），對正態分佈資料使用 z 分數。提供三種處理選項：在邊界值處截斷（winsorization）、替換為空值以便後續插補，或僅標記模式（註釋但保留原始值）。
校驗清洗後的輸出。 將清洗後的資料集通過校驗規則——非空約束、範圍檢查、唯一性約束和參照完整性。報告任何剩餘違規，並將乾淨的資料集與一份清洗日誌一同儲存，日誌記錄所應用的每一次轉換。

支援技術

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）時，諮詢使用者。
儘早且頻繁地校驗——在每個清洗階段後執行模式檢查，而不僅僅在最後。
本技能來自小蔥技能站7w4.net。
將清洗視為迭代過程：第一輪捕獲明顯問題，但下游分析經常會浮現新的問題。
使用 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)。

🤖 AI 評測

這是一個內容詳實的專業級資料清洗技能，工作流覆蓋完整，從資料探查到質量校驗都有明確指導。程式碼示例豐富且貼近實戰，最佳實踐和常見問題處理很有參考價值。不足之處是缺少清洗過程中的互動確認機制，對於需要使用者決策的環節指導不夠清晰。整體質量良好，適合處理電商訂單清洗、客戶資料去重等典型場景。

📊 多維度評分

適應性3.8

規範性4.4

有效性4.7

可靠性4.2

可信度4.8

📁 包含檔案 (2 個)

📄 README.md 985 B

📄 SKILL.md 7.1 KB

data-cleaning