如何使用pandasql快速计算字段标准差与COV变异系数

需要在处理pandas DataFrame时，通过类SQL逻辑计算QUANTITY字段的总体标准差，进而得到变异系数COV = 总体标准差/均值，结合ADI指标完成需求模式分类。但pandasql底层基于SQLite，原生不支持STD/STDEV/STDEVP等标准差聚合函数，原编写的SQL查询无法直接运行。

原查询代码如下：

mysql = lambda q: sqldf(q, globals())
classified_df = mysql("""with data as (
    select
        CP_REF,
        count(*) * 1.0 /
          nullif(count(case when QUANTITY > 0 then 1 end), 0) as ADI,
          stdevp(QUANTITY) / nullif(avg(QUANTITY), 0) as COV
    from df
    where parent is not null
    group by CP_REF
)
select
    CP_REF, ADI, COV,
    case
        when ADI <  1.32 and COV <  0.49 then 'Smooth'
        when ADI >= 1.32 and COV <  0.49 then 'Intermittent'
        when ADI <  1.32 and COV >= 0.49 then 'Erratic'
        when ADI >= 1.32 and COV >= 0.49 then 'Lumpy'
        else 'Smooth'
    end as DEMAND
from data;""")

核心需要实现的逻辑是stdevp(QUANTITY) / nullif(avg(QUANTITY), 0) as COV，方案不限制必须使用pandasql，只要支持类SQL方式处理DataFrame、满足计算逻辑即可。

方案1：替换为DuckDB实现类SQL查询（最小改动） #

DuckDB是面向分析场景的嵌入式数据库，原生支持pandas DataFrame直接查询，兼容标准SQL语法，内置所有常用统计聚合函数，性能远高于pandasql，不需要修改原有CTE查询结构，仅需替换标准差函数名即可运行。

如果未安装DuckDB，先执行安装命令：
pip install duckdb

查询代码如下：

import duckdb

classified_df = duckdb.sql("""
with data as (
    select
        CP_REF,
        count(*) * 1.0 /
          nullif(count(case when QUANTITY > 0 then 1 end), 0) as ADI,
          -- stddev_pop对应原stdevp总体标准差，样本标准差替换为stddev_samp即可
          stddev_pop(QUANTITY) / nullif(avg(QUANTITY), 0) as COV
    from df
    where parent is not null
    group by CP_REF
)
select
    CP_REF, ADI, COV,
    case
        when ADI <  1.32 and COV <  0.49 then 'Smooth'
        when ADI >= 1.32 and COV <  0.49 then 'Intermittent'
        when ADI <  1.32 and COV >= 0.49 then 'Erratic'
        when ADI >= 1.32 and COV >= 0.49 then 'Lumpy'
        else 'Smooth'
    end as DEMAND
from data
""").df()

代码中所有逻辑和原SQL完全对齐，空值处理、分类规则不需要做任何调整。

方案2：pandas原生实现（无额外依赖） #

如果不想引入新的第三方库，可以直接用pandas分组聚合实现相同逻辑，计算效率更高：

import pandas as pd
import numpy as np

# 先过滤parent非空的数据
filter_df = df[df["parent"].notna()]

# 分组计算ADI和COV指标
agg_df = filter_df.groupby("CP_REF").apply(
    lambda x: pd.Series({
        "ADI": len(x) / (x["QUANTITY"] > 0).sum() if (x["QUANTITY"] > 0).sum() != 0 else np.nan,
        "COV": x["QUANTITY"].std(ddof=0) / x["QUANTITY"].mean() if x["QUANTITY"].mean() != 0 else np.nan
    })
).reset_index()

# 按规则匹配需求类型
cond_list = [
    (agg_df["ADI"] < 1.32) & (agg_df["COV"] < 0.49),
    (agg_df["ADI"] >= 1.32) & (agg_df["COV"] < 0.49),
    (agg_df["ADI"] < 1.32) & (agg_df["COV"] >= 0.49),
    (agg_df["ADI"] >= 1.32) & (agg_df["COV"] >= 0.49)
]
label_list = ["Smooth", "Intermittent", "Erratic", "Lumpy"]
agg_df["DEMAND"] = np.select(cond_list, label_list, default="Smooth")

classified_df = agg_df

注意std(ddof=0)是总体标准差计算，完全对齐原SQL中stdevp的逻辑，如果需要样本标准差改为ddof=1即可。

内容的提问来源于stack exchange，提问作者Ravaal