1. 项目概述为什么“多维聚合”不是Pandas进阶技巧而是业务分析的生存技能我在银行风控部门干了七年从刚毕业写SQL查数的分析师到带三个人小团队做反欺诈模型的数据架构师。这七年里我亲手重构过四套核心报表系统也踩过无数个“看起来能跑、上线就崩”的坑。最典型的一次是某次季度经营分析会前夜财务总监临时要求加一个维度“把上季度所有信用卡交易按客户等级金卡/白金卡/黑卡、商户类型餐饮/零售/旅行、交易时间工作日/周末/节假日三个维度交叉统计再算出每个组合的平均单笔金额、中位数、30天滚动均值、以及交易金额极差——明天早会要用。”当时我手里的代码还是用五个独立的groupby().agg()拼起来的跑一次要17分钟中间还因为内存溢出失败两次。那天凌晨三点我一边改代码一边想如果早两年就真正吃透多维聚合的底层逻辑而不是只会抄几个.agg({col: mean})根本不会被这种需求卡住脖子。这就是Part 20讲的“Data Manipulation in Multi-Dimensional Aggregation”——它根本不是教你怎么用pandas的某个函数而是在教你怎么把业务语言精准翻译成数据操作语言。你看到的“按区域和产品线分组求均值”背后是销售总监在看哪个市场该加大资源投入你写的“滚动7天平均交易额”其实是风控系统在实时判断某客户是否突然开始高频大额消费你调用的unstack()本质是把数据库里扁平的交易流水变成老板一眼就能看懂的交叉表格。关键词里反复出现的“Towards AI”不是指AI技术本身而是指一种思维方式用工程化、可复现、可审计的方式把模糊的业务问题拆解成确定的数据操作步骤。这不是给程序员看的API文档这是给每天要和业务方吵架、要对数据结果负责、要为线上模型提供稳定特征的数据从业者写的实战手册。如果你还在用for循环遍历分组、用pd.concat()硬拼结果、或者每次加个新指标就重写整个聚合逻辑——那你不是在用pandas你是在用pandas给自己挖坑。2. 核心思路拆解为什么必须放弃“单指标单聚合”的思维定式2.1 业务问题从来不是单点的而是网状的我们先看一个真实场景某城商行的信用卡中心发现Q3整体交易额环比下降5%但粗看数据又找不到原因。业务方第一反应是“查下餐饮类交易”结果发现餐饮类涨了8%第二反应是“查下高净值客户”结果发现白金卡客户交易额涨了12%。问题来了既然细分领域都在涨为什么总量在跌答案藏在交叉维度里——白金卡客户在餐饮类的交易额确实涨了但他们同时大幅减少了在“线上购物”类别的交易而后者占其总交易额的40%。这个结论只有把“客户等级”和“商户类型”两个维度同时纳入分组再计算各自占比变化才能暴露出来。如果还停留在“先按客户等级分组再按商户类型分组”的串行思维你永远得不到这个洞察。因为串行分组会丢失维度间的关联性你算出白金卡客户餐饮交易额8%但不知道这个8%是来自原有客户的消费升级还是新客涌入带来的增量你算出线上购物交易额-15%但不知道是老客户流失还是全量客户都减少了频次。真正的解法是一次性构建三维分组索引客户等级 × 商户类型 × 时间段然后在这个立方体上做聚合。这就像医生不能只看血压或只看血糖必须同时看血脂、胰岛素抵抗指数、尿酸水平才能判断是代谢综合征还是单纯高血压。2.2 性能瓶颈不在CPU而在数据搬运和逻辑重复很多工程师觉得“多个agg分开写更清晰”这是典型的开发思维而非数据工程思维。我做过一组压测对1000万行交易数据分别测试两种写法方案A分离式df.groupby(cat).amount.mean()→ 生成中间结果Adf.groupby(cat).amount.std()→ 生成中间结果Bdf.groupby(cat).amount.count()→ 生成中间结果C最后pd.concat([A,B,C], axis1)方案B聚合式df.groupby(cat).agg({amount: [mean,std,count]})结果方案A耗时21.3秒方案B仅需6.8秒。差距在哪不是计算慢是三次分组操作导致数据被扫描三次索引重建三次内存拷贝三次。pandas的groupby本质是哈希分桶每次调用都要重新计算hash值、分配内存块、处理空值。而聚合式写法只做一次分桶所有聚合函数共享同一个分组结果。这就像快递员送10个包裹方案A是送完第一个再回站点取第二个方案B是一次性把10个包裹装车按路线规划好。在生产环境这种差异直接决定报表能否在T1凌晨两点前跑完——而凌晨两点是下游BI系统刷新数据的死线。2.3 可维护性陷阱当业务逻辑散落在二十个脚本里最致命的是可维护性。我接手过一个“客户价值分层”模型原始代码由五个人在三年间陆续编写核心逻辑分散在segment_v1.py里用lambda算RFM中的Recencysegment_v2.py里用自定义函数算Monetary的加权均值segment_v3.py里用rolling计算Frequencty的30天滚动计数report_gen.py里手动unstack生成交叉表dashboard_api.py里又用reset_index()把多级索引拍平当业务方要求“把高频小额交易50元单独标记为‘碎片化消费’并计入新指标”时我花了三天时间才定位到所有需要修改的代码位置改完后发现segment_v2.py里的加权逻辑没适配新字段导致全量客户Monetary值集体偏低12%。而如果所有逻辑都封装在统一的聚合管道里新增一个指标只需在agg()字典里加一行is_fragmented: lambda x: (x 50).sum() / len(x)连测试都不用重写——因为聚合框架本身已通过了千万级数据的压力验证。3. 多维聚合的四大支柱从原理到实操的深度解析3.1 多指标聚合不只是语法糖而是数据流的并行化设计原理层面为什么agg({col: [mean,std]})比agg({col: np.mean})快关键在pandas的聚合器复用机制。当你传入列表[mean,std]时pandas会启动一个优化路径对每个分组一次性计算所有需要的中间统计量如sum、sum_sq、count再用这些中间量推导出最终结果mean sum/count,std sqrt((sum_sq - sum²/count)/(count-1))而如果分开调用每次都要重新计算sum和count造成冗余运算。提示这个优化对count、nunique等非数值聚合不生效因为它们无法共享中间状态。所以agg({col: [count,nunique]})仍是两次独立扫描。实操细节如何避免多级列名带来的后续处理灾难看原文输出transaction_amount processing_fee mean median min max Dining 55.10 52.30 1.36 2.03这个多级列结构MultiIndex在后续处理中极易出错。比如你想取所有mean值不能写result[mean]会报KeyError必须写result[(transaction_amount,mean)]。更糟的是当你要导出到Excel时Excel根本不认识MultiIndex会把列名显示为(transaction_amount, mean)这种丑陋格式。我的解决方案是在聚合后立即标准化列名# 原始聚合 result df.groupby(merchant_category).agg({ transaction_amount: [mean,median], processing_fee: [min,max] }) # 标准化列名用下划线连接层级去除括号 result.columns [_.join(col).strip() for col in result.columns.values] result result.reset_index() # 确保merchant_category变回普通列输出变为merchant_category transaction_amount_mean transaction_amount_median processing_fee_min processing_fee_max Dining 55.10 52.30 1.36 2.03这样后续所有操作result[transaction_amount_mean] 100、result.to_excel()、result.to_sql()都完全无感。我在所有生产脚本里强制执行这条规范团队新人入职第一天就要背熟这个转换模板。高阶技巧混合聚合策略应对异构业务逻辑实际业务中不同指标的计算逻辑可能完全不同。比如风控场景avg_transaction需要排除异常值用quantile(0.95)截断fraud_risk_score需要调用外部API计算不能向量化is_high_risk是布尔判断交易额5000且当日频次10这时不能硬塞进一个agg()正确做法是分层聚合# 第一层基础聚合向量化高性能 base_agg df.groupby(customer_id).agg({ amount: [mean, lambda x: x.quantile(0.95)], transaction_time: lambda x: (x.max() - x.min()).total_seconds() / 3600 }).round(2) # 第二层复杂逻辑逐分组处理但范围已极大缩小 def complex_risk_calc(group): high_value_count (group[amount] 5000).sum() recent_freq group[group[transaction_time] pd.Timestamp.now() - pd.Timedelta(24H)].shape[0] return pd.Series({ fraud_risk_score: call_external_risk_api(group[customer_id].iloc[0]), is_high_risk: high_value_count 0 and recent_freq 10 }) # 应用到已分组的数据groupby对象可直接传入apply risk_detail df.groupby(customer_id).apply(complex_risk_calc) final_result pd.concat([base_agg, risk_detail], axis1)这样既保证了80%的计算在向量化层面完成又为20%的特殊逻辑留出空间。关键点在于永远让数据在分组后尽可能早地进入向量化通道只对必要部分降级到Python循环。3.2 自定义聚合函数把业务规则刻进数据DNA为什么lambda不够用看一个血泪教训曾经有个需求计算“客户月度活跃度”定义为“当月有交易天数 / 当月总天数”。我最初写了df.groupby(customer_id).agg({ transaction_date: lambda x: x.dt.day.nunique() / x.dt.days_in_month.iloc[0] })上线后发现所有结果都是NaN。排查三天才发现x.dt.days_in_month返回的是Series而iloc[0]取的是第一个交易日期所在月份的天数——但如果某客户当月只在1号和30号交易x.dt.days_in_month就是[31,31]iloc[0]取31没问题但如果客户跨月交易比如31号和1号x.dt.days_in_month就是[31,30]iloc[0]取31但1号属于30天的月份结果就错了。正确解法是自定义函数必须显式处理时间边界def monthly_active_rate(series): 计算客户在series覆盖月份内的活跃率 if series.empty: return 0.0 month_start series.min().replace(day1) month_end (month_start pd.offsets.MonthEnd(1)) total_days (month_end - month_start).days 1 active_days series.dt.date.nunique() return round(active_days / total_days, 4) # 使用 result df.groupby(customer_id)[transaction_date].agg(monthly_active_rate)命名函数的三大黄金法则函数名即文档weighted_average()比calc_wgt_avg()好transaction_range()比get_diff()好。我在代码审查时如果看到def func1(x): ...直接打回重写。必须带类型提示和docstring不仅说明做什么更要说明为什么这么做。比如def risk_adjusted_return(series: pd.Series) - float: 计算风险调整后收益夏普比率简化版 业务背景银行理财部要求剔除无风险利率影响聚焦超额收益。 计算逻辑(年化收益率 - 2.5%) / 年化波动率其中2.5%为当前1年期国债收益率 注意输入series为日收益率单位为小数0.01表示1% if len(series) 20: # 至少20个交易日才有统计意义 return 0.0 annual_return (1 series.mean()) ** 252 - 1 annual_vol series.std() * np.sqrt(252) return (annual_return - 0.025) / (annual_vol 1e-8) # 防止除零禁止副作用函数内不能修改入参、不能读写文件、不能调用print()。所有依赖必须通过参数注入如risk_free_rate0.025方便单元测试。生产环境必加的防御性编程自定义函数在大数据量下极易崩溃。我在所有自定义聚合函数开头都加三道保险def safe_custom_agg(series): # 保险1空值防护 if series.empty or series.isna().all(): return np.nan # 保险2数据类型校验 if not pd.api.types.is_numeric_dtype(series): raise TypeError(fCustom agg requires numeric input, got {series.dtype}) # 保险3极端值过滤业务允许的阈值 valid_series series[(series -1e6) (series 1e6)] # 排除百万级异常值 if len(valid_series) len(series) * 0.9: # 超过10%被过滤记录告警 logger.warning(fHigh outlier rate in custom agg: {len(series)-len(valid_series)}/{len(series)}) # 正常计算逻辑 return valid_series.mean() * 1.05 # 示例加5%风险溢价3.3 滚动窗口聚合时间序列分析的底层心跳滚动窗口的本质不是“滑动”而是“锚定”很多人误解rolling(window7)是“过去7天”其实pandas的滚动窗口默认是包含当前行的向前7行。比如date2024-01-07的滚动均值计算的是2024-01-01到2024-01-07这7天——这叫“左闭右闭”区间。但业务需求往往是“截至昨日的7天均值”即2024-01-01到2024-01-06。这就需要显式指定closedleft# 正确截至昨日的7天均值不含当天 df[7day_avg_excl_today] df.groupby(customer_id)[amount].rolling( window7, closedleft # 关键只包含左侧数据 ).mean().reset_index(level0, dropTrue)时间窗口 vs 行窗口选错等于分析报废原文示例用的是window3行窗口但在金融场景中必须用时间窗口。原因很简单交易不是均匀发生的。某客户可能周一连刷5笔周二休眠3天周三再刷2笔。用行窗口计算“3笔滚动均值”会把周一的5笔和周三的2笔强行绑定完全扭曲真实消费节奏。正确做法是用on参数指定时间列# 按真实时间滚动必须先排序 df_sorted df.sort_values([customer_id, transaction_time]) df_sorted[7day_amount_avg] df_sorted.groupby(customer_id).apply( lambda g: g.set_index(transaction_time)[amount].rolling( 7D, # 关键7D表示7天不是7行 min_periods3 # 至少3天有数据才计算避免稀疏数据干扰 ).mean() ).reset_index(level0, dropTrue)注意7D是pandas的时间频率字符串支持30min、1H、7D、1M等。min_periods3表示只要在过去7天内有至少3天的交易数据就参与计算否则返回NaN。这个参数极其重要——没有它月初数据会因历史数据不足而全为NaN导致整个月报失效。滚动计算的性能陷阱与绕过方案当数据量超千万行时rolling().mean()会变慢。根本原因是pandas对每个分组都重新构建时间索引。我的优化方案是预计算时间分箱再聚合# 方案用cut()替代rolling() df_sorted[week_start] df_sorted[transaction_time].dt.floor(7D) # 计算每周均值 weekly_avg df_sorted.groupby([customer_id, week_start])[amount].mean().reset_index() # 用shift(-1)实现“下周均值”作为“本周滚动均值” weekly_avg[rolling_7day_avg] weekly_avg.groupby(customer_id)[amount].shift(-1) # 合并回原数据 df_final df_sorted.merge(weekly_avg[[customer_id,week_start,rolling_7day_avg]], on[customer_id,week_start], howleft)实测对5000万行数据此方案比原生rolling()快4.2倍且内存占用降低60%。代价是精度损失它把7天窗口近似为周粒度。但在日报、周报场景中这种精度损失完全可接受且稳定性远超原生方案。3.4 多级分组与Unstack从数据库范式到业务语言的翻译器为什么unstack()不是“转置”而是“维度升维”看原文示例result df_sales.groupby([region,product])[revenue].mean().unstack()输出是product Gadget Widget region North 12000.0 15500.0 South 13750.0 18000.0这里unstack()把product这个分组维度从行索引“提升”为列索引本质上是将二维分组结果region×product映射到二维矩阵rowsregion, colsproduct。这对应着OLAP中的切片slice操作固定region为行product为列revenue为单元格值。但新手常犯的错误是对非数值列unstack()。比如# 错误category是字符串unstack后会报错 df.groupby([region,category])[customer_id].nunique().unstack()因为nunique()返回整数但unstack()要求索引唯一。正确做法是先reset_index()再pivot()# 正确用pivot处理分类变量 pivoted df.groupby([region,category])[customer_id].nunique().reset_index() result pivoted.pivot(indexregion, columnscategory, valuescustomer_id)Unstack后的终极难题缺失值填充策略业务数据天然稀疏。比如某地区根本没有“Travel”类商户unstack()后该单元格就是NaN。填0错0表示“有数据且为0”NaN表示“无数据”。在风控场景中把“无数据”填0会导致模型误判为“零风险”。我的标准处理流程# 步骤1明确缺失语义 # NaN 该region-category组合无交易记录应保留NaN # 0 有记录但交易额为0业务上极少发生 # 步骤2按业务规则填充 result df.groupby([region,category])[revenue].sum().unstack(fill_value0) # fill_value0只填充unstack产生的NaN不影响原始数据中的NaN # 步骤3添加缺失标识列关键 result[has_travel_data] (~result[Travel].isna()).astype(int) result[has_dining_data] (~result[Dining].isna()).astype(int)这样下游既能看到数值又能知道哪些维度是真实缺失。我在所有报表系统中强制要求任何unstack操作后必须生成对应的“存在性标识列”这是数据可信度的基石。高阶技巧多级Unstack构建三维透视表当需要三个维度时如region × product × time_periodunstack()可链式调用# 先按region和product分组再按time_period分组 df[time_period] pd.cut(df[date], binspd.date_range(2024-01-01,2024-04-01,freqM), labels[Q1-Jan,Q1-Feb,Q1-Mar]) # 三级分组 multi_group df.groupby([region,product,time_period])[revenue].sum() # 两级unstack先提time_period为列再提product为列 result multi_group.unstack(time_period).unstack(product) # 输出indexregion, columns(time_period, product)即MultiIndex列此时列名是(Q1-Jan,Gadget)这样的元组。要展平它result.columns [f{t}_{p} for t,p in result.columns] # 得到Q1-Jan_Gadget, Q1-Jan_Widget, Q1-Feb_Gadget...这种结构可直接喂给Tableau或Power BI它们原生支持MultiIndex列。4. 端到端实战银行信用卡分析流水线的七层炼金术4.1 数据准备模拟真实世界的脏乱差原文的模拟数据过于干净。真实信用卡数据有三大毒瘤时间戳漂移同一笔交易在不同系统中时间差达数秒金额精度不一致POS机传0.01核心系统存0.005导致sum不等商户类别错标同一家餐厅在不同月份被标为“Dining”或“Food”我的清洗脚本强制执行def clean_transaction_data(df): # 时间对齐以交易ID为key取各系统时间的中位数 df[transaction_time_clean] df.groupby(transaction_id)[transaction_time].transform(median) # 金额归一化强制保留两位小数消除浮点误差 df[amount_clean] df[amount].round(2) # 商户类别标准化建立映射字典业务方确认 category_map { Food: Dining, Restaurant: Dining, Retail: Retail, Shopping: Retail, Travel: Travel, Airline: Travel } df[category_clean] df[category].map(category_map).fillna(Other) return df # 应用清洗 df_clean clean_transaction_data(df_raw)4.2 七层分析流水线详解分析1客户-品类双维度统计解决“谁在买什么”# 关键用named aggregation避免列名混乱 analysis1 df_clean.groupby([customer_id,category_clean]).agg( avg_amount(amount_clean, mean), median_amount(amount_clean, median), trans_count(amount_clean, count), fee_min(fee, min), fee_max(fee, max), amount_std(amount_clean, std) ).round(2).reset_index() # 添加业务标签高价值客户月均交易额5000 analysis1[is_high_value] (analysis1[avg_amount] * analysis1[trans_count] 5000).astype(int)实操心得永远用named aggregation(col,func)元组代替字典因为前者生成的列名是明确的字符串后者生成MultiIndex列后续处理成本翻倍。分析2交易极差分析解决“消费是否异常”def transaction_range(series): 业务定义极差1000元视为高波动触发人工核查 rng series.max() - series.min() return pd.Series({ range: rng, is_high_volatility: int(rng 1000), volatility_score: rng / (series.mean() 1e-5) # 波动率相对值 }) analysis2 df_clean.groupby(category_clean)[amount_clean].apply(transaction_range) # 展开为宽表 analysis2 pd.json_normalize(analysis2).set_index(category_clean)分析3滚动7日均值解决“消费趋势是否突变”# 按客户时间排序关键 df_sorted df_clean.sort_values([customer_id,transaction_time_clean]) df_sorted df_sorted.set_index(transaction_time_clean) # 滚动计算时间窗口 analysis3 df_sorted.groupby(customer_id)[amount_clean].rolling( 7D, min_periods3, closedleft # 截至昨日 ).mean().reset_index(name7day_avg_amount) # 合并回原数据用于对比 df_with_trend df_sorted.reset_index().merge( analysis3, on[customer_id,transaction_time_clean], howleft )分析4累计消费解决“客户生命周期价值”# 按客户时间排序后用expanding计算 df_sorted[cumulative_spend] df_sorted.groupby(customer_id)[amount_clean].expanding().sum().values # 计算LTV分位数业务指标 ltv_stats df_sorted.groupby(customer_id)[cumulative_spend].last().describe() # 生成分层标签 df_sorted[ltv_tier] pd.qcut( df_sorted.groupby(customer_id)[cumulative_spend].last(), q[0,0.3,0.7,1], labels[Bronze,Silver,Gold] )分析5交叉表解决“品类偏好矩阵”# 用pivot_table替代unstack更灵活 analysis5 pd.pivot_table( df_clean, indexcustomer_id, columnscategory_clean, valuesamount_clean, aggfuncmean, fill_value0 ).round(2) # 添加行/列总计 analysis5[TOTAL] analysis5.sum(axis1) analysis5.loc[GRAND_TOTAL] analysis5.sum(axis0)分析6高管摘要解决“一句话结论”analysis6 df_clean.groupby(customer_id).agg( total_spend(amount_clean, sum), avg_trans(amount_clean, mean), trans_count(amount_clean, count), total_fee(fee, sum), first_trans(transaction_time_clean, min), last_trans(transaction_time_clean, max) ).round(2) # 计算关键比率 analysis6[avg_fee_rate] (analysis6[total_fee] / analysis6[total_spend] * 100).round(2) analysis6[active_days] (analysis6[last_trans] - analysis6[first_trans]).dt.days 1 analysis6[spend_per_active_day] (analysis6[total_spend] / analysis6[active_days]).round(2) # 生成文字结论这才是高管要的 def generate_summary(row): if row[spend_per_active_day] 500: return 高价值活跃客户 elif row[trans_count] 30: return 高频低额客户疑似代付 else: return 常规客户 analysis6[summary] analysis6.apply(generate_summary, axis1)分析7风险分层解决“谁该被重点监控”def risk_segmentation(series): 基于监管要求的三层风险模型 # 监管红线1单笔5000元 high_value_count (series 5000).sum() # 监管红线21小时内5笔 hourly_freq series.groupby(series.index.floor(1H)).size().max() # 综合评分 score (high_value_count * 3 min(hourly_freq, 5) * 2) return pd.Series({ high_value_count: high_value_count, hourly_peak_freq: hourly_freq, risk_score: score, risk_level: HIGH if score 10 else MEDIUM if score 5 else LOW }) analysis7 df_clean.groupby(customer_id)[amount_clean].apply(risk_segmentation)4.3 流水线编排用函数式编程组装分析模块我把所有分析封装成纯函数用functools.reduce()组装from functools import reduce def build_analysis_pipeline(df): 声明式定义分析流水线 steps [ lambda d: clean_transaction_data(d), # 清洗 lambda d: d.assign(customer_tierassign_customer_tier(d)), # 分层 lambda d: d.merge(analysis1(d), on[customer_id,category_clean]), # 双维度 lambda d: d.merge(analysis3(d), on[customer_id,transaction_time_clean]), # 滚动 lambda d: d.merge(analysis4(d), oncustomer_id), # 累计 ] return reduce(lambda acc, step: step(acc), steps, df) # 执行 final_report build_analysis_pipeline(df_raw)这种写法的好处每个步骤可独立测试、可插拔替换、可添加日志埋点。当业务方说“把滚动窗口改成14天”我只需改analysis3()函数整个流水线自动更新。5. 生产环境避坑指南那些文档里不会写的血泪经验5.1 内存爆炸的五大征兆与急救方案征兆根本原因急救方案长期方案MemoryError在groupby().agg()时报出pandas默认用objectdtype存储字符串内存占用是category的5-10倍df[col] df[col].astype(category)在ETL阶段强制转换所有枚举列rolling()耗时超10分钟时间窗口未设min_periodspandas对每行都尝试全量计算rolling(7D, min_periods1)用cut()预分箱替代rolling()unstack()后内存翻倍MultiIndex列未及时reset_index()pandas缓存原始索引result result.reset_index()所有聚合后立即标准化列结构apply()卡死自定义函数中有IO操作如读文件、调API改用swifter库并行化将IO操作移到聚合前用map()预加载concat()后变慢不同DataFrame的dtype不一致如int64 vs int32pd.concat(..., ignore_indexTrue, sortFalse)统一ETL阶段的dtype策略5.2 结果验证的三重校验法生产环境绝不相信单次计算结果。我的标准验证流程总量守恒校验agg_result[total_spend].sum()必须等于raw_df[amount].sum()允许浮点误差0.01%维度完整性校验检查所有分组键的唯一值数量是否符合业务预期expected_regions {North,South,East,West} assert set(analysis1[region]) expected_regions, Region dimension missing业务逻辑穿透校验随机抽样3个客户手工计算其7day_avg_amount与程序结果比对sample_cust analysis3[customer_id].sample(3).tolist() # 手工计算逻辑...5.3 性能调优的黄金参数清单场景参数推荐值说明大数据量groupbysortFalseTrue禁用排序可提速30-50%除非业务要求分组有序字符串分组observedTrueTrue只对实际出现的类别分组跳过未出现的category值滚动窗口min_periodsmax(3, window//2)避免早期数据全为NaN平衡灵敏度与稳定性unstack