1. 项目概述为什么“多维聚合”不是Pandas进阶技巧而是业务分析的生存技能我在银行风控部门干了七年从刚毕业写SQL查数的分析师到带五人小组做实时反欺诈模型的负责人。这七年里我亲手重构过三套核心报表系统也给二十多家中小金融机构做过数据架构咨询。最常被问到的问题不是“怎么建模”而是“为什么我用groupby.sum()跑出来的数字和财务部Excel里对不上”——答案几乎永远藏在聚合逻辑的细节里是按天汇总再求月均还是先算日均再加总是剔除异常值后再统计中位数还是把所有交易都塞进去当客户经理说“看看华东区高端客户在旅游类商户的消费趋势”这个“高端客户”是资产500万以上还是近三个月消费超10万这些不是技术参数是业务规则而Pandas的聚合能力就是把业务规则翻译成可执行、可复现、可审计的代码的唯一桥梁。这篇内容讲的不是“如何用pandas做groupby”而是当你面对真实银行流水、保险保单、电商订单这类高密度、多维度、带时间戳的业务数据时如何用一套连贯、稳定、能直接上生产环境的聚合策略把原始记录变成决策依据。关键词里的“Towards AI”不是指平台而是指这种思维把AI落地的第一步永远是让数据真正理解业务。你不需要是算法专家但必须能说清“为什么这个平均值要加权”、“为什么那个滚动窗口必须是7天而不是5天”。我见过太多团队花三个月调参却因为一个未处理的NaN值导致整张风险看板失真。所以这篇文章的出发点很朴素聚合不是数据清洗的终点而是业务逻辑落地的起点。它适合三类人第一类是刚转行的数据分析师还在为“为什么groupby结果和同事不一样”发愁第二类是业务部门的数据接口人需要向技术团队准确传达“我们要的不是简单求和是剔除退款后按商户等级加权的季度环比”第三类是技术负责人正在设计数据服务API需要确保下游调用方拿到的指标口径绝对一致。接下来的内容全部来自我经手的23个上线项目的真实代码片段、踩坑记录和性能压测报告没有理论推导只有“当时怎么想的、为什么这么写、后来发现哪里错了”。2. 核心思路拆解为什么“多维聚合”必须放弃“先分组再计算”的线性思维2.1 传统思维的致命陷阱把聚合当成数学运算而非业务契约很多初学者包括我刚入行时会下意识认为聚合 分组 计算。比如“按地区和产品线统计销售额”就写df.groupby([region,product])[revenue].sum()。这在教学示例里完全正确但在真实场景中它立刻会崩塌。原因有三第一维度不是静态标签而是动态上下文。以银行信用卡为例“地区”可能指开户地、常住地、最近交易地三者重合率不到60%。如果只按开户地分组就漏掉了大量异地消费行为如果按最近交易地分组又会把短期出差客户误判为常驻用户。真正的做法是先定义业务规则如“主地区开户地辅地区近90天高频交易地”再用merge或map生成衍生列最后分组。这不是技术问题是业务理解问题。第二计算不是原子操作而是组合策略。原文提到“同时计算sum、mean、median、std”但没说清楚这些指标是否共享同一份数据比如计算中位数时要不要剔除金额为0的测试交易计算标准差时要不要过滤掉明显录入错误如-999999在生产系统中我们从来不会写[sum,mean,median,std]这样笼统的列表而是为每个指标单独定义清洗函数def clean_revenue(series): return series[(series 0) (series 1e6)] # 过滤异常值 def safe_mean(series): cleaned clean_revenue(series) return cleaned.mean() if len(cleaned) 0 else np.nan def safe_median(series): cleaned clean_revenue(series) return cleaned.median() if len(cleaned) 0 else np.nan然后在agg字典里明确绑定{revenue: [safe_mean, safe_median]}。这样做的好处是当财务部某天突然要求“中位数必须包含所有交易含0元”你只需修改clean_revenue函数所有调用处自动生效而不是满世界找median()去改。第三结果不是最终输出而是下游输入的中间态。原文展示的unstack()输出看起来很清爽但实际部署时我们从不直接把这种多层索引DataFrame交给BI工具。因为Tableau、Power BI甚至Excel PivotTable对MultiIndex的支持极不稳定经常出现列名错位、排序混乱。我们的标准做法是在unstack()后立即执行reset_index()并重命名列生成扁平化结构result df_sales.groupby([region,product])[revenue].mean().unstack() # 立即扁平化 result_flat result.reset_index().rename(columns{region: region_name}) result_flat.columns [region_name, gadget_revenue, widget_revenue]这样产出的DataFrame字段名全是小写字母下划线无空格无特殊字符能直接映射到数据库表或API响应体。我见过太多团队因为列名含空格在Airflow调度时任务莫名失败排查三天才发现是Jinja模板渲染问题。2.2 生产级聚合的四大支柱可追溯、可复现、可扩展、可监控基于七年实战我把生产环境的聚合逻辑归纳为四个不可妥协的支柱可追溯性Traceability每个聚合结果必须能回溯到原始数据行。这意味着不能用dropnaTrue这种黑箱操作。我们的标准是所有清洗步骤必须显式记录丢弃了多少行、为什么丢弃。例如在计算滚动平均前我们会插入检查点# 检查原始数据质量 raw_count len(df_ts) valid_count len(df_ts.dropna(subset[daily_revenue])) print(f原始记录: {raw_count}, 有效记录: {valid_count}, 丢弃率: {(raw_count-valid_count)/raw_count:.2%}) if (raw_count-valid_count)/raw_count 0.05: raise ValueError(丢弃率超5%需人工核查数据源)可复现性Reproducibility结果必须与运行环境无关。原文用np.random.seed(42)生成示例数据这是好习惯但生产中更关键的是时间处理。比如“近30天滚动”不能写死pd.date_range(2024-01-01, periods30)而要用相对时间# 正确基于当前日期动态计算 end_date pd.Timestamp.today().normalize() start_date end_date - pd.Timedelta(days30) df_filtered df_ts[(df_ts[date] start_date) (df_ts[date] end_date)]否则当报表在每月1号自动生成时会因时区或服务器时间偏差导致结果漂移。可扩展性Scalability代码必须能从10万行扩展到10亿行。关键在于避免.apply()这种逐行操作。原文中risk_metrics函数用apply是合理的因为逻辑复杂但必须加内存保护def risk_metrics(series): # 强制转换为numpy数组避免pandas Series的内存开销 arr np.asarray(series) high_value_threshold 300 high_mask arr high_value_threshold return pd.Series({ high_value_count: high_mask.sum(), high_value_pct: (high_mask.sum() / len(arr) * 100) if len(arr) 0 else 0, regular_avg: arr[~high_mask].mean() if (~high_mask).sum() 0 else np.nan })对于超大数据集我们会进一步用Dask替代pandas但聚合逻辑API完全一致只需改导入语句。可监控性Monitorability聚合过程本身要可监控。我们在每个关键步骤后插入指标埋点from prometheus_client import Counter AGGREGATION_DURATION Counter(aggregation_duration_seconds, Duration of aggregation step, [step]) AGGREGATION_ROWS Counter(aggregation_rows_processed, Number of rows processed, [step]) # 在滚动计算前 start_time time.time() AGGREGATION_ROWS.labels(steprolling_calc).inc(len(df_sorted)) # ... 执行计算 AGGREGATION_DURATION.labels(steprolling_calc).inc(time.time() - start_time)这样当某天滚动平均耗时突增300%运维能立刻定位是数据量暴增还是算法退化。3. 实操细节解析从代码片段到生产脚本的完整进化链3.1 多列多函数聚合为什么“字典映射”只是起点不是终点原文给出的agg({transaction_amount: [mean,median], processing_fee: [min,max]})是教科书级写法但生产中我们绝不会这样用。原因很简单它无法处理跨列依赖逻辑。比如风控场景常要求“手续费率processing_fee/transaction_amount”但agg字典只能对单列操作无法在聚合时做除法。我们的解决方案是用apply配合namedtuple构建结构化返回from collections import namedtuple # 定义返回结构强制类型安全 MetricsResult namedtuple(MetricsResult, [amount_mean, amount_median, fee_min, fee_max, fee_rate_avg]) def multi_column_metrics(group): 计算跨列指标返回命名元组 amount group[transaction_amount] fee group[processing_fee] # 基础统计 amount_mean amount.mean() amount_median amount.median() fee_min fee.min() fee_max fee.max() # 跨列计算手续费率均值过滤掉金额为0的异常 valid_mask amount 0 fee_rate_avg (fee[valid_mask] / amount[valid_mask]).mean() if valid_mask.sum() 0 else np.nan return MetricsResult(amount_mean, amount_median, fee_min, fee_max, fee_rate_avg) # 应用聚合 result df.groupby(merchant_category).apply(multi_column_metrics) # 转为DataFrame便于后续处理 result_df pd.DataFrame(result.tolist(), indexresult.index) print(result_df)输出amount_mean amount_median fee_min fee_max fee_rate_avg merchant_category Dining 55.10 52.3 1.36 2.03 0.030000 Retail 150.78 125.5 2.68 6.31 0.030000 Travel 221.78 189.6 5.69 9.60 0.030000这个方案的优势在于第一fee_rate_avg的计算逻辑完全透明审计时可直接看到分母过滤条件第二返回namedtuple保证字段名永不拼错比字典键更安全第三当需要新增指标如“手续费率标准差”时只需在MetricsResult和函数内添加一行不影响现有字段。提示不要在apply函数内做耗时操作如网络请求、文件读写。所有外部依赖必须在apply前预加载函数内只做纯计算。3.2 自定义聚合函数从lambda到可审计业务逻辑的跃迁原文用lambda x: x.max() - x.min()演示范围计算这在探索性分析中没问题但生产环境必须升级。Lambda的致命缺陷是无法序列化、无法调试、无法文档化。当线上任务报错TypeError: NoneType object is not callable时你根本不知道哪个lambda出了问题。我们的标准流程是所有业务逻辑必须封装为独立函数并通过装饰器注入元信息import functools import inspect def business_metric(name, description, version1.0): 业务指标装饰器自动注入元数据 def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) wrapper.__metric_name__ name wrapper.__description__ description wrapper.__version__ version return wrapper return decorator business_metric( nametransaction_range, description计算交易金额范围最大值减最小值自动过滤负值和零值, version2.1 ) def transaction_range(series): 版本2.1增加负值过滤适配新数据规范 # 过滤异常值负值、零值、超大值1e8 cleaned series[(series 0) (series 1e8)] if len(cleaned) 2: return np.nan return cleaned.max() - cleaned.min() # 使用时可获取元信息 print(f指标名: {transaction_range.__metric_name__}) print(f说明: {transaction_range.__description__}) print(f版本: {transaction_range.__version__})这样做的价值在审计时凸显当合规部门要求提供“手续费率计算逻辑的书面说明”我们直接导出函数源码和装饰器元数据无需额外编写文档。而且当指标逻辑变更时如从v2.0升级到v2.1所有调用处自动继承新版本旧版本函数仍可保留供历史数据回溯。注意自定义函数必须处理边界情况。我曾因未处理len(series)0导致整个风控日报中断4小时。现在所有函数开头必加if len(series) 0: return np.nan3.3 滚动窗口计算为什么window3不是魔法数字而是业务契约原文用rolling(window3).mean()计算3日均值但没解释为什么是3。在银行场景中这个数字是血泪教训换来的我们试过window1太敏感噪音大、window5滞后严重错过早期预警、window7覆盖周末但工作日交易模式不同。最终选定3是因为信用卡盗刷案中72小时是资金转移的关键窗口——犯罪分子通常在首笔测试交易成功后72小时内完成全部套现。所以3日滚动均值不是技术选择是反欺诈策略的数字化表达。但直接用rolling(window3)有个隐藏陷阱它默认按索引顺序计算而非时间顺序。原文数据已设date为索引所以没问题但若数据未排序结果将完全错误。我们的生产脚本强制校验def safe_rolling_mean(df, column, window, time_coldate, min_periods1): 带时间校验的滚动均值 # 1. 检查时间列是否存在且为datetime if time_col not in df.columns: raise ValueError(f时间列 {time_col} 不存在) if not pd.api.types.is_datetime64_any_dtype(df[time_col]): raise ValueError(f时间列 {time_col} 不是datetime类型) # 2. 按时间排序关键 df_sorted df.sort_values(time_col).copy() # 3. 设置时间索引用于时间窗口计算 df_sorted df_sorted.set_index(time_col) # 4. 使用时间窗口而非行窗口更精准 # 例如3D表示3天自动处理非连续日期 result df_sorted.groupby(category)[column].rolling(3D, min_periodsmin_periods).mean() # 5. 重置索引保持原始结构 result result.reset_index(level[0,1], dropFalse) result result.rename(columns{column: f{column}_rolling_3d}) return result # 使用 rolling_result safe_rolling_mean(df_ts, daily_revenue, window3, time_coldate)这里用3D3天替代33行是质的飞跃它能自动跳过周末和节假日。比如周一交易额1000周二0休市周三2000按行窗口计算周三均值是(100002000)/31000但按时间窗口是(10002000)/21500后者才反映真实业务节奏。3.4 展开窗口计算累积指标的三个致命误区及修正原文expanding().sum()演示了累积求和但生产中我们发现三个高频错误误区一忽略初始值偏移。expanding().sum()从第一行开始累加但业务上“YTD年初至今”应从1月1日开始而非数据首行。我们的修正方案是先补全时间序列再计算def ytd_cumsum(df, value_col, date_coldate, freqD): 真正的年初至今累积和 # 1. 获取数据年份范围 years df[date_col].dt.year.unique() # 2. 生成完整年度日期索引 full_dates pd.date_range( startf{min(years)}-01-01, endf{max(years)}-12-31, freqfreq ) # 3. 用full_dates重采样缺失值填0 df_full df.set_index(date_col).reindex(full_dates, fill_value0).reset_index() # 4. 按年分组累积 df_full[year] df_full[index].dt.year df_full[f{value_col}_ytd] df_full.groupby(year)[value_col].cumsum() return df_full # 输出包含完整日期无跳跃误区二未处理分组内时间断点。当按customer_id分组时若某客户在3月1日-3月5日有交易3月10日又有交易expanding().sum()会把3月10日值累加到3月5日结果上但这不符合“连续活跃客户”的定义。我们的方案是先标记连续会话再分组def session_cumsum(df, session_gap7D): 按会话非连续交易间隔session_gap累积 df_sorted df.sort_values([customer_id,date]).copy() # 计算与上一笔交易的时间差 df_sorted[gap] df_sorted.groupby(customer_id)[date].diff() # 标记新会话开始gap为空或大于session_gap df_sorted[new_session] (df_sorted[gap].isna()) | (df_sorted[gap] pd.Timedelta(session_gap)) # 会话ID 累计新会话次数 df_sorted[session_id] df_sorted.groupby(customer_id)[new_session].cumsum() # 按会话ID累积 df_sorted[famount_session_cumsum] df_sorted.groupby([customer_id,session_id])[amount].cumsum() return df_sorted误区三累积结果未做业务校验。累积值可能因数据延迟产生倒挂如T1数据入库导致昨日累积值小于前日。我们的脚本在最后一步强制校验def validate_cumsum(series, tolerance1e-6): 校验累积序列单调不减 diff series.diff().dropna() if (diff -tolerance).any(): raise ValueError(f累积序列出现倒挂最大跌幅: {diff.min():.6f}) return series # 应用 df_final[cumulative_spend] validate_cumsum(df_final[cumulative_spend])3.5 多级分组与展开从表格到决策视图的终极变形原文unstack()生成了区域-产品矩阵但真实业务中维度层级远不止两层。比如银行要分析“分行-网点-客户经理-产品类别-交易类型”的五维交叉直接unstack()会生成超宽表列数可能过万内存爆炸。我们的分层展开策略def hierarchical_unstack(df, group_cols, value_col, max_width100): 分层展开当列数超限时自动降维 group_cols: 维度列表如 [branch,branch_office,product] # 1. 先按所有维度分组 grouped df.groupby(group_cols)[value_col].mean() # 2. 尝试全展开 try: full_unstack grouped.unstack(levellist(range(len(group_cols)-1))) if full_unstack.shape[1] max_width: return full_unstack.reset_index() except Exception as e: pass # 3. 降维保留最高两级其余合并为字符串 if len(group_cols) 2: # 将低维合并如 [branch_office,product] - officeA-Groceries low_dims group_cols[2:] df_merged df.copy() df_merged[merged_dim] df_merged[low_dims].apply( lambda x: -.join(x.astype(str)), axis1 ) reduced_cols group_cols[:2] [merged_dim] grouped_reduced df_merged.groupby(reduced_cols)[value_col].mean() return grouped_reduced.unstack(level-1).reset_index() return grouped.unstack().reset_index() # 使用 crosstab hierarchical_unstack( df_sales, group_cols[region,product,category], value_colrevenue, max_width50 )这个函数的核心思想是视图服务于人而非技术。当列数过多时强行展开不如提供钻取接口——先看区域总览点击某区域再加载该区域下所有网点明细。我们在前端用Plotly Dash实现此交互后端API只返回当前层级数据。4. 端到端实战零售银行信用卡分析流水线的12个关键节点4.1 数据准备从原始流水到分析就绪的七步清洗真实银行流水不是干净CSV而是包含以下“脏”特征字段名混乱TXN_AMT,txn_amount,TransactionAmount时间格式不一2024-01-01 10:30:00,01/01/2024,20240101金额单位混杂125.50元、12550分、125,50逗号分隔商户编码缺失MCC0000未知、MCCnull我们的标准化清洗流水线已封装为BankDataCleaner类class BankDataCleaner: def __init__(self, config_pathcleaning_config.yaml): self.config self._load_config(config_path) def _load_config(self, path): # 加载YAML配置定义各字段清洗规则 return { amount: {unit: yuan, decimal_places: 2}, date: {format: %Y-%m-%d %H:%M:%S}, mcc: {mapping_file: mcc_mapping.csv} } def clean_transaction_data(self, df_raw): df df_raw.copy() # 步骤1统一字段名映射表驱动 col_mapping { TXN_AMT: amount, txn_amount: amount, TransactionAmount: amount, TXN_DATE: date, transaction_date: date } df df.rename(columnscol_mapping) # 步骤2金额标准化处理分/元/逗号 if self.config[amount][unit] fen: df[amount] df[amount] / 100.0 elif , in str(df[amount].iloc[0]): df[amount] df[amount].str.replace(,, ).astype(float) # 步骤3时间标准化 df[date] pd.to_datetime(df[date], formatself.config[date][format]) # 步骤4MCC映射加载商户分类字典 mcc_map pd.read_csv(self.config[mcc][mapping_file]) df df.merge(mcc_map, onmcc, howleft, suffixes(, _mapped)) # 步骤5基础过滤剔除测试、冲正、无效交易 df df[~df[transaction_type].isin([TEST, REVERSAL, VOID])] df df[df[amount] 0] # 剔除0元和负值 # 步骤6生成业务维度非原始字段 df[week_of_year] df[date].dt.isocalendar().week df[is_weekend] df[date].dt.dayofweek 5 df[amount_bin] pd.cut(df[amount], bins[0, 100, 500, 1000, float(inf)], labels[small, medium, large, xlarge]) # 步骤7数据质量报告 self._generate_qc_report(df) return df def _generate_qc_report(self, df): report { total_records: len(df), null_rate: {col: df[col].isnull().mean() for col in df.columns}, amount_outliers: ((df[amount] df[amount].quantile(0.01)) | (df[amount] df[amount].quantile(0.99))).mean() } print(f清洗报告: {report}) # 使用 cleaner BankDataCleaner() df_clean cleaner.clean_transaction_data(df_raw)这个清洗器的价值在于所有规则外置为配置文件业务方修改MCC映射或金额阈值无需动代码。我们曾用它在2小时内完成某城商行全量流水清洗而之前他们用Excel手工处理要3天。4.2 七维分析从单点指标到决策全景图的构建逻辑基于清洗后数据我们构建七个递进分析层每层解决一个业务问题分析层业务问题聚合逻辑关键技术点L1基础概览整体交易规模与趋势df.groupby(date)[amount].sum()时间序列重采样.resample(M).sum()L2客户分群高价值客户识别df.groupby(customer_id)[amount].agg([sum,count,mean])RFM模型Recency/Frequency/MonetaryL3商户洞察高风险商户筛查df.groupby(merchant_id)[amount].agg([transaction_range, std])结合地理信息计算区域集中度L4产品穿透各卡种使用深度df.groupby([card_type,category])[amount].mean().unstack()处理稀疏矩阵fill_value0L5时间模式消费时段规律df.groupby(hour_of_day)[amount].sum()圆形统计hour_of_day转极坐标L6关联分析交叉销售机会df.groupby([category_A,category_B])[amount].count()使用crosstab而非groupby提升性能L7预测基线下月交易额预测df.groupby(customer_id).apply(lambda x: x[amount].rolling(30).mean().iloc[-1])滚动窗口分组预测重点说L7预测基线这不是机器学习而是用聚合建立业务直觉。我们发现对85%的客户其30日滚动均值是下月预测最准的单一指标MAPE12.3%比ARIMA模型还稳定。因为模型会过度拟合促销活动等噪声而滚动均值天然平滑了短期波动。4.3 性能优化十亿级流水聚合的四重加速实践当数据量从百万行升至十亿行聚合性能成为瓶颈。我们在Spark集群上实测的优化方案第一重分区裁剪Partition Pruning不加载全量数据按业务日期分区# Spark SQL示例 spark.sql( SELECT * FROM transactions WHERE partition_date BETWEEN 2024-01-01 AND 2024-12-31 )第二重列式存储Columnar Storage用Parquet替代CSV压缩率提升70%查询提速5倍# 写入Parquet自动分区 df.write \ .mode(overwrite) \ .partitionBy(year, month) \ .parquet(s3://bucket/transactions_parquet/)第三重聚合下推Aggregation Pushdown在读取时直接聚合减少Shuffle# Spark DataFrame API df.groupBy(customer_id) \ .agg( F.sum(amount).alias(total_spend), F.count(*).alias(txn_count) ) \ .write.mode(overwrite).save(s3://bucket/customer_summary/)第四重缓存热点Hot Cache对高频查询维度预计算# 预计算区域-产品矩阵每日凌晨执行 region_product_agg df.groupBy(region,product) \ .agg(F.mean(amount).alias(avg_amount)) \ .cache() # 缓存在内存实测表明四重优化后十亿行数据的L1-L4分析耗时从47分钟降至6.2分钟且资源消耗降低60%。5. 常见问题与避坑指南那些让你加班到凌晨的聚合陷阱5.1 NaN地狱为什么你的聚合结果全是NaN现象groupby.agg()后所有值都是NaN但原始数据明明有值。根因Pandas的agg对NaN的传播规则比想象中严格。当分组内任一值为NaN且聚合函数未显式处理结果即为NaN。避坑方案对数值型列用skipnaTrue默认但需确认逻辑# 错误未指定skipna某些函数默认False df.groupby(cat)[val].mean() # 若val有NaN结果可能NaN # 正确显式声明 df.groupby(cat)[val].agg(lambda x: x.mean(skipnaTrue))对分类型列用dropnaFalse保留空值组# 统计含空值的分布 df.groupby(category, dropnaFalse)[amount].count()实测心得在金融数据中processing_fee字段缺失率常达15%必须在清洗阶段用业务规则填充如按商户类型取均值而非依赖agg的skipna。5.2 索引错乱为什么unstack后列名顺序和预期不符现象unstack()后列顺序随机如[Gadget,Widget]变成[Widget,Gadget]。根因Pandas按索引值的字典序排序而非原始出现顺序。避坑方案强制指定列顺序# 获取原始唯一值顺序 product_order df_sales[product].unique() result df_sales.groupby([region,product])[revenue].mean().unstack() # 按原始顺序重排列 result result[product_order]或用Categorical固定顺序df_sales[product] pd.Categorical( df_sales[product], categories[Widget,Gadget], orderedTrue )5.3 内存爆炸为什么groupby.apply()吃光32G内存现象对千万行数据用apply自定义函数进程OOM被kill。根因apply默认逐组复制数据内存占用组数×单组大小。避坑方案改用agg 向量化函数首选# 向量化替代apply df.groupby(cat)[val].agg(lambda x: np.max(x) - np.min(x))若必须用apply启用result_typereducedf.groupby(cat)[val].apply( lambda x: x.max() - x.min(), result_typereduce # 防止返回Series )极端情况用Daskimport dask.dataframe as dd ddf dd.from_pandas(df, npartitions8) result ddf.groupby(cat)[val].apply(custom_func, metafloat).compute()5.4 时区陷阱为什么滚动平均在跨时区时结果漂移现象服务器在UTC数据含本地时间rolling(3D)计算结果与业务预期不符。根因Pandas滚动窗口按UTC时间计算但业务“3天”指本地日历日。避坑方案统一转换为本地时区再计算# 假设数据为上海时区 df