def dynamic_threshold(df, col, sensitivity=0.2):
    q25 = df[col].quantile(0.25)
    q75 = df[col].quantile(0.75)
    iqr = q75 - q25
    return {
        'lower': q25 - (1.5 + sensitivity) * iqr,
        'upper': q75 + (1.5 + sensitivity) * iqr
    }

from sklearn.preprocessing import RobustScaler
from joblib import Parallel, delayed

def parallel_scale(df_chunk):
    scaler = RobustScaler()
    return pd.DataFrame(scaler.fit_transform(df_chunk),
                        columns=df_chunk.columns)

# 分块处理10GB级数据
scaled_data = Parallel(n_jobs=4)(
    delayed(parallel_scale)(chunk)
    for chunk in np.array_split(big_data, 8)
)

# 保留原始分布信息
orders['amount_scaled'] = orders['amount'].pipe(
    lambda x: (x - x.mean()) / x.std()
).add_prefix('zscore_')

orders['amount_original_ratio'] = orders['amount'] / orders['amount'].max()

class DataQualityMonitor:  
    def __init__(self, df):  
        self.df = df.copy()  
        self.metrics = {}  

    def generate_report(self):  
        self._check_completeness()  
        self._check_consistency()  
        self._check_anomalies()  
        return pd.DataFrame(self.metrics).T  

    def _check_completeness(self):  
        self.metrics['missing_rate'] = self.df.isna().mean()  
        self.metrics['zero_rate'] = (self.df == 0).mean()

    def _check_consistency(self):  
        self.metrics['id_duplicates'] = self.df.duplicated(subset='id').sum()  
        self.metrics['date_range'] = {  
            'start': self.df['date'].min(),  
            'end': self.df['date'].max()  
        }  

    def _check_anomalies(self):  
        for col in ['amount', 'quantity']:  
            stats = self.df[col].describe()  
            self.metrics[f'{col}_anomaly'] = {  
                'outliers': self.df[col].between(  
                    stats['25%'] - 1.5*(stats['75%']-stats['25%']),  
                    stats['75%'] + 1.5*(stats['75%']-stats['25%'])  
                ).sum()  
            }  

# 生成HTML可视化报告  
monitor = DataQualityMonitor(orders)  
report = monitor.generate_report()  
report.style.background_gradient(cmap='viridis').to_html('data_quality_report.html')

| 特性              | 1.x版本               | 2.x最佳实践           |
|-------------------|-----------------------|----------------------|
| 空值处理          | `NaN`                 | 强制使用`pd.NA`      |
| 字符串处理        | `object`类型          | 专用`StringDtype`    |
| 类型推断          | 手动指定              | `convert_dtypes()`   |
| 性能优化          | 单线程处理            | 支持PyArrow引擎      |

# 类型优化矩阵  
dtype_map = {  
    'int64': 'Int32',          # 空值兼容整型  
    'float64': 'Float32',  
    'object': 'string[pyarrow]' # 使用Arrow字符串  
}  

optimized_df = df.astype(dtype_map)  
print(f"内存节省：{(1 - optimized_df.memory_usage().sum() / df.memory_usage().sum()):.1%}")

# 使用Pandas实现漏斗分析  
funnel_steps = ['login', 'browse', 'add_cart', 'checkout']  
funnel_data = (  
    user_events  
    .assign(step=pd.Categorical(user_events['event_type'], categories=funnel_steps))  
    .groupby(['user_id', 'step'])['timestamp'].min()  
    .unstack()  
    .pipe(lambda df: df[funnel_steps])  # 强制顺序  
)  

# 计算转化率  
conversion_rates = {}  
for i in range(len(funnel_steps)-1):  
    rate = funnel_data[funnel_steps[i+1]].notna().sum() / funnel_data[funnel_steps[i]].notna().sum()  
    conversion_rates[f'{funnel_steps[i]}_to_{funnel_steps[i+1]}'] = f"{rate:.1%}"