import pandas as pd
import numpy as np

# 1. 创建一个包含异常值的示例数据集
np.random.seed(42)
# 生成 100 个正常数据，均值 50，标准差 10
data = 50 + 10 * np.random.randn(100)
# 故意插入一些异常值
outliers = np.array([120, -20, 150, -30])
data_with_outliers = np.concatenate([data, outliers])
# 创建一个 DataFrame
df = pd.DataFrame({'value': data_with_outliers})

print("原始数据统计描述：")
print(df['value'].describe())
print("-" * 40)

# ==================== 检测异常值 ====================

# 方法一：Z-score 法
def detect_outliers_zscore(data, threshold=3):
    """
    使用 Z-score 检测异常值
    :param data: pandas Series 或 numpy array
    :param threshold: 阈值，默认为 3
    :return: 布尔掩码，True 表示是异常值
    """
    mean = np.mean(data)
    std = np.std(data)
    z_scores = [(x - mean) / std for x in data]
    return np.abs(z_scores) > threshold

# 方法二：IQR 法
def detect_outliers_iqr(data, factor=1.5):
    """
    使用 IQR 检测异常值
    :param data: pandas Series 或 numpy array
    :param factor: IQR 的倍数，默认为 1.5
    :return: 布尔掩码，True 表示是异常值
    """
    Q1 = np.percentile(data, 25)
    Q3 = np.percentile(data, 75)
    IQR = Q3 - Q1
    lower_bound = Q1 - factor * IQR
    upper_bound = Q3 + factor * IQR
    return (data < lower_bound) | (data > upper_bound)

# 应用检测方法
df['is_outlier_zscore'] = detect_outliers_zscore(df['value'])
df['is_outlier_iqr'] = detect_outliers_iqr(df['value'])

print("使用 Z-score (阈值=3) 检测到的异常值：")
print(df[df['is_outlier_zscore'] == True][['value']])
print("\n使用 IQR (倍数=1.5) 检测到的异常值：")
print(df[df['is_outlier_iqr'] == True][['value']])
print("-" * 40)

# ==================== 处理异常值 ====================
# 选择 IQR 方法的结果进行处理演示
outliers_mask = df['is_outlier_iqr']
print(f"共发现 {outliers_mask.sum()} 个异常值。")

# 策略一：删除
# df_cleaned = df[~outliers_mask].copy() # ~ 表示取反，即保留非异常值

# 策略二：盖帽法（替换为边界值）
Q1 = df['value'].quantile(0.25)
Q3 = df['value'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 将低于下限的值替换为下限，高于上限的值替换为上限
df['value_capped'] = df['value'].clip(lower=lower_bound, upper=upper_bound)

# 策略三：替换为中位数或均值
# median_value = df['value'].median()
# df['value_filled_median'] = df['value'].where(~outliers_mask, median_value)

print("\n原始数据（前5个）：", df['value'].head().values)
print("盖帽处理后数据（前5个）：", df['value_capped'].head().values)
# 查看被处理掉的异常值原值和新值
print("\n被盖帽处理的异常值对比：")
print(df[outliers_mask][['value', 'value_capped']])

# 更新 DataFrame，使用处理后的数据
df_processed = df[['value_capped']].copy()
df_processed.columns = ['value'] # 恢复列名
print("\n处理后数据的统计描述：")
print(df_processed['value'].describe())

Z-score (t=2) 检测到的异常值: [50000, 60000]
IQR (k=1.5) 检测到的异常值: [50000, 60000]
我选择 IQR 法，因为它对极端值不敏感，更稳健。
处理前均值: 15650.0
处理后均值: 6750.0