如何用Pandas聚合CSV周报数据计算周度销售差值与变动率

直接按以下步骤写脚本即可，所有可配置项都抽出来单独放，后续改参数不用动核心逻辑。

核心实现步骤 #

• 基础依赖导入与参数配置
  先把所有需要调整的规则（目标年份、基准周、目标周、待统计商品列表）统一放在配置块里，后续逐周统计只改配置就行，不用动计算逻辑，示例代码：

import pandas as pd

# 可配置项，根据自己的实际报表修改
CONFIG = {
    "csv_path": "你的周度报表存储路径.csv",
    "target_year": 2022,
    "base_compare_week": 16, # 作为对比基准的周次，对应需求里的数据集B
    "current_target_week": 18, # 当前要统计的目标周次，对应需求里的数据集A
    "target_product_list": ["可乐", "薯片", "矿泉水"] # 你需要统计的预设商品名称列表
}

• 数据加载与预过滤
  读入CSV后先筛出指定年份的数据，减少后续计算的数据量，避免无关年份数据干扰结果：

# 读取原始CSV
raw_df = pd.read_csv(CONFIG["csv_path"])
# 预过滤2022年数据
df_2022 = raw_df[raw_df["year"] == CONFIG["target_year"]].reset_index(drop=True)

• 目标商品匹配
  针对item info列做模糊匹配，只要列内容包含预设商品名就算命中，同时提取命中的具体商品名，方便后续按商品维度聚合：

# 构造商品匹配规则
def match_product(item_text):
    item_text = str(item_text)
    for product in CONFIG["target_product_list"]:
        if product in item_text:
            return product
    return None

# 过滤出命中目标商品的记录
df_2022["matched_product"] = df_2022["item info"].apply(match_product)
df_target = df_2022[df_2022["matched_product"].notna()].reset_index(drop=True)

• 分周聚合销售额
  按周次、匹配到的商品名分组，对POS/销售总额列求和，得到每个商品每周的总销售额：

# 注意把pos_sales替换成你CSV里实际的销售额列名
week_agg = df_target.groupby(
    ["week_num", "matched_product"], 
    as_index=False
)["pos_sales"].sum().rename(columns={"pos_sales":"total_sales"})

• 差值与变动率计算
  分别提取目标周、基准周（16周）的聚合结果，按商品关联合并后计算指标：

# 提取数据集A：目标周各商品销售额
set_A = week_agg[week_agg["week_num"] == CONFIG["current_target_week"]][
    ["matched_product", "total_sales"]
].rename(columns={"total_sales":"target_week_sales"})

# 提取数据集B：16周各商品销售额
set_B = week_agg[week_agg["week_num"] == CONFIG["base_compare_week"]][
    ["matched_product", "total_sales"]
].rename(columns={"total_sales":"base_week_sales"})

# 按商品合并两份数据，无销售额的记录补0
final_result = pd.merge(set_A, set_B, on="matched_product", how="outer").fillna(0)

# 计算指标C：两周销售额差值
final_result["sales_diff_C"] = final_result["target_week_sales"] - final_result["base_week_sales"]

# 计算周度变动百分比：仅当差值为正、且基准周销售额不为0时计算，其余情况留空
final_result["week_change_rate"] = final_result.apply(
    lambda row: round(row["sales_diff_C"] / row["base_week_sales"], 4) 
    if row["sales_diff_C"] > 0 and row["base_week_sales"] != 0 
    else None,
    axis=1
)

• 结果导出
  计算完成后直接导出为CSV，可直接复用在周报里：

# 导出时用utf-8-sig编码避免中文乱码
final_result.to_csv("周度销售差值统计结果.csv", index=False, encoding="utf_8_sig")

落地注意事项 #

• 代码里的列名（year、week_num、item info、pos_sales）需要和你自己CSV里的实际列名保持一致，否则会报KeyError
• 如果商品名包含正则特殊字符（比如+、*、(），直接用字符串in判断即可，不用正则，避免匹配出错
• 如果不需要统计某一周无销售记录的商品，把合并时的how="outer"改成how="inner"即可，只会保留两周都有销售数据的商品
• 需要批量统计多个周的差值时，只要写个循环遍历所有目标周次，替换CONFIG里的current_target_week值批量跑就行，不用重复写逻辑
• 变动率如果需要百分比格式展示，导出前加一行代码转换即可：final_result["week_change_rate"] = final_result["week_change_rate"].apply(lambda x: f"{x*100:.2f}%" if pd.notna(x) else "")

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