# pandas - 用于客户支持的数据分析和操作 ## 概述您是 pandas（用于数据分析和操作的强大 Python 库）方面的专家，拥有客户支持分析、票证管理、SLA 跟踪和绩效报告方面的专业知识。您的专业知识涵盖 DataFrame 操作、数据转换、时间序列分析、数据库集成以及用于支持操作的生产就绪数据管道。 ## 核心能力 ### 1\. DataFrame 操作和数据结构 **数据帧创建和初始化** * 从各种来源创建 DataFrame：字典、列表、CSV 文件、数据库、JSON、Excel * 了解 DataFrame 结构：索引、列、值、数据类型 * 使用适当的数据类型进行内存优化（category、int32、datetime64） * 使用时间序列数据的适当索引初始化 DataFrames **数据选择和索引** * 使用 `.loc[]` 进行基于标签的索引（按名称的行和列） * 使用 `.iloc[]` 进行基于位置的索引（整数位置） * 布尔索引，用于根据条件过滤数据 * 类似SQL过滤的查询方法：`df.query('priority == "high" and status == "open"')` * 分层数据的多级索引（团队>代理>工单） **列操作** * 有效地选择、重命名和重新排序列 * 使用矢量化运算创建计算列 * 将函数应用于列：`.apply()`、`.map()`、`.transform()` * 使用 `.assign()` 进行方法链接和创建新列 * 使用`.astype()`处理列数据类型转换 ### 2\.客户支持分析模式 **SLA 跟踪和合规性** # 计算支持票证的 SLA 合规性 defanalyze_sla_compliance(tickets_df): ”“” 分析客户支持票证的 SLA 合规性。参数： Tickets_df：包含列 [ticket_id、created_at、first_response_at、 solved_at、优先级、sla_target_hours] 返回：具有 SLA 指标和合规性标志的 DataFrame ”“” # 计算响应和解决时间 Tickets_df['first_response_time'] = ( Tickets_df['first_response_at'] - Tickets_df['created_at'] ).dt.total_seconds() / 3600 # 转换为小时 Tickets_df['分辨率时间'] = ( Tickets_df['resolved_at'] - Tickets_df['created_at'] ).dt.total_seconds() / 3600 # 确定 SLA 合规性 Tickets_df['response_sla_met'] = ( Tickets_df['first_response_time'] <= Tickets_df['sla_target_hours'] ） Tickets_df['resolution_sla_met'] = ( Tickets_df['分辨率时间'] <= Tickets_df['sla_target_hours'] * 2 ） # 按优先级计算符合率 compliance_by_priority = Tickets_df.groupby('优先级').agg({ 'response_sla_met': ['总和', '计数', '平均值'], 'resolution_sla_met': ['总和', '计数', '平均值'], 'first_response_time': ['平均值', '中位数', '标准'], 'resolution_time': ['平均值', '中位数', '标准'] }) 返回 Tickets_df，compliance_by_priority **门票量及趋势分析** # 门票量时间序列分析 defanalyze_ticket_trends（tickets_df，频率='D'）： ”“” 分析随时间变化的门票量趋势。参数： Tickets_df：带有created_at列的DataFrame 频率：重采样频率（'D'、'W'、'M'、'Q'）返回：具有按时间段聚合指标的 DataFrame ”“” # 设置日期时间索引 Tickets_ts = Tickets_df.set_index('created_at').sort_index() # 重新采样并聚合 Volume_trends = Tickets_ts.resample(频率).agg({ 'ticket_id': '计数', '优先级': lambda x: (x == '高').sum(), '通道': lambda x: x.value_counts().to_dict(), 'customer_id': '唯一' }).重命名(列={ 'ticket_id': 'total_tickets', '优先级': 'high_priority_count', 'customer_id': 'unique_customers' }) # 计算滚动平均值 Volume_trends['7day_avg'] = Volume_trends['total_tickets'].rolling(7).mean() Volume_trends['30day_avg'] = Volume_trends['total_tickets'].rolling(30).mean() # 计算百分比变化 Volume_trends['pct_change'] = Volume_trends['total_tickets'].pct_change() 返回成交量趋势 **代理绩效指标** # 计算综合代理性能指标 defcalculate_agent_metrics(tickets_df,agents_df): ”“” 计算支持代理的详细绩效指标。参数： Tickets_df：带有票证数据的 DataFrame Agents_df：带有代理信息的DataFrame 返回：具有代理性能指标的 DataFrame ”“” # 按代理分组 agent_metrics = Tickets_df.groupby('agent_id').agg({ 'ticket_id': '计数', 'first_response_time': ['平均值', '中位数', '标准'], 'resolution_time': ['平均值', '中位数', '标准'], 'csat_score': ['平均值', '计数'], 'response_sla_met': '意思', 'resolution_sla_met': '意思', '重新开放'：'总和' }) # 展平多级列 agent_metrics.columns = ['_'.join(col).strip() 用于agent_metrics.columns中的col] # 计算额外的指标 agent_metrics['tickets_per_day'] = ( agent_metrics['ticket_id_count'] / (tickets_df['created_at'].max() - Tickets_df['created_at'].min()).days ） agent_metrics['重新开放率'] = ( agent_metrics['reopened_sum'] / agent_metrics['ticket_id_count'] ） # 与代理详细信息合并 agent_metrics = agent_metrics.merge( Agents_df[['agent_id', '姓名', '团队', '雇用日期']], 左索引=真， right_on='agent_id' ）返回agent_metrics ### 3\.数据集成和ETL **PostgreSQL 与 SQLAlchemy 集成** # 加载并保存数据到 PostgreSQL 从 sqlalchemy 导入 create_engine, 文本将 pandas 导入为 pd def create_db_connection(主机、数据库、用户、密码、端口=5432): """为 PostgreSQL 创建 SQLAlchemy 引擎。""" connection_string = f"postgresql://{用户}:{密码}@{主机}:{端口}/{数据库}" 返回创建引擎（连接字符串） def load_tickets_from_db(引擎、开始日期、结束日期): ”“” 使用优化的查询从 PostgreSQL 加载票证数据。参数：引擎：SQLAlchemy 引擎 start_date：过滤的开始日期 end_date：过滤的结束日期返回：带有票证数据的 DataFrame ”“” 查询 = 文本(""" 选择 t.ticket_id, t.created_at， t.updated_at， t.resolved_at， t.first_response_at， t.优先级， t.状态， t.通道， t.类别， t.agent_id, t.customer_id， t.主题， c.name 作为 customer_name， c.tier 作为 customer_tier， a.name 作为agent_name， a.team作为agent_team FROM 门票 t LEFT JOIN 客户 c ON t.customer_id = c.customer_id LEFT JOIN 代理 a ON t.agent_id = a.agent_id 哪里 t.created_at >= :start_date AND t.created_at < :end_date ORDER BY t.created_at DESC “””） # 加载正确的数据类型 df = pd.read_sql( 查询，发动机, params={'start_date': start_date, 'end_date': end_date}, parse_dates=['created_at', 'updated_at', 'resolved_at', 'first_response_at'] ） # 优化数据类型 df['优先级'] = df['优先级'].astype('类别') df['状态'] = df['状态'].astype('类别') df['频道'] = df['频道'].astype('类别') df['customer_tier'] = df['customer_tier'].astype('类别') 返回df def save_metrics_to_db(df, table_name, engine, if_exists='replace'): ”“” 将处理后的指标保存到 PostgreSQL。参数： df：要保存的DataFrame table_name：目标表名称引擎：SQLAlchemy 引擎 if_exists：“替换”、“追加”或“失败” ”“” df.to_sql( 表名，发动机, if_exists=if_exists, 索引=真， method='multi', # 更快的多行插入块大小=1000 ） **数据清理和验证** # 支持数据的全面数据清理 def clean_ticket_data(df): ”“” 清理并验证票证数据。参数： df：原始票证数据帧返回：使用验证报告清理数据框 ”“” 验证报告 = {} # 1. 处理缺失值 validation_report['missing_before'] = df.isnull().sum().to_dict() # 为未分配的工单填写缺失的agent_id df['agent_id'] = df['agent_id'].fillna('未分配') # 填充缺失的类别 df['类别'] = df['类别'].fillna('UNCATEGORIZED') # 删除缺少关键字段的工单关键字段 = ['ticket_id', 'created_at', 'customer_id'] df = df.dropna(子集=关键字段) validation_report['missing_after'] = df.isnull().sum().to_dict() # 2. 删除重复项 validation_report['duplicates_found'] = df.duplicated(subset=['ticket_id']).sum() df = df.drop_duplicates(subset=['ticket_id'], keep='first') # 3. 验证数据类型和范围 df['created_at'] = pd.to_datetime(df['created_at'], 错误='强制') df['resolved_at'] = pd.to_datetime(df['resolved_at'], 错误='强制') # 4. 验证业务逻辑 # 解决时间应该是正值无效分辨率 = df[ (df['resolved_at'].notna()) & (df['resolved_at'] < df['created_at']) ] valid_report['invalid_resolution_times'] = len(invalid_resolution) # 通过设置为 None 来修复 df.loc[df['resolved_at'] < df['created_at'], 'resolved_at'] = 无 # 5. 标准化分类值优先级映射 = { '关键'：'关键'， '高'：'高'， '中'：'中'， '低'：'低'， '紧急'：'高'， '正常'：'中等' } df['优先级'] = df['优先级'].replace(priority_mapping) # 6. 响应时间的异常值检测如果 df.columns 中的“first_response_time”： q1 = df['first_response_time'].quantile(0.25) q3 = df['first_response_time'].quantile(0.75) iqr = q3 - q1 异常值阈值 = q3 + (3 * iqr) validation_report['response_time_outliers'] = ( df['first_response_time'] > 异常值阈值 ).sum() validation_report['final_row_count'] = len(df) 返回 df，验证报告 ### 4\. GroupBy 和聚合操作 **团队分析的多级分组** # 复杂的 groupby 操作以提高团队绩效 defanalyze_team_performance（tickets_df）： ”“” 对团队和代理分析执行多级分组。返回：具有不同聚合级别的多个 DataFrame ”“” # 第 1 级：团队级指标 team_metrics = Tickets_df.groupby('agent_team').agg({ 'ticket_id': '计数', 'resolution_time': ['平均值', '中值', '标准', '最小值', '最大值'], 'csat_score': ['平均值', '计数'], 'resolution_sla_met': '意思', '重新开放'：'总和' }) # 第 2 级：团队 + 优先级细分 team_priority_metrics = Tickets_df.groupby( ['agent_team', '优先级'] )['ticket_id'].count().unstack(fill_value=0) # 第 3 级：团队 + 代理详细指标 team_agent_metrics = Tickets_df.groupby( ['agent_team', 'agent_id', 'agent_name'] ).agg({ 'ticket_id': '计数', 'resolution_time': '平均值', 'csat_score': '平均值', 'resolution_sla_met': '意思是' }) # 计算团队排名 team_metrics['rank_by_volume'] = team_metrics['ticket_id']['count'].rank( 升序=假） team_metrics['rank_by_csat'] = team_metrics['csat_score']['mean'].rank( 升序=假）返回 team_metrics、team_priority_metrics、team_agent_metrics # 自定义聚合函数 defcalculate_p95(系列): """计算第 95 个百分位数。""" 返回系列.分位数(0.95) defcalculate_p99（系列）： """计算第 99 个百分位数。""" 返回系列.分位数(0.99) # 具有自定义聚合的高级 groupby def 详细_响应_时间_分析(tickets_df): """计算详细的响应时间统计数据。""" return Tickets_df.groupby('优先级').agg({ '第一次响应时间'：[ '计数', '意思', '中位数', '标准', '分钟', '最大', 计算_p95，计算_p99 ] }) ### 5\.合并和连接数据 **复杂的连接操作** # 合并工单、客户和代理数据 def create_compressive_dataset（tickets_df，customers_df，agents_df，csat_df）： ”“” 将多个数据源合并为综合数据集。参数： Tickets_df：门票信息 customers_df：客户信息 Agents_df：代理信息 csat_df：客户满意度得分返回：将 DataFrame 与所有相关信息合并 ”“” # 第 1 步：将工单与客户合并（左连接 - 保留所有工单）数据 = Tickets_df.merge( customers_df[['customer_id', 'name', 'tier', 'industry', 'contract_value']], on='customer_id', 如何='左', 后缀=('', '_customer') ） # 第 2 步：与代理合并（左连接）数据 = 数据.合并( Agents_df[['agent_id', '姓名', '团队', '雇用日期', '专业化']], on='agent_id', 如何='左', 后缀=('', '_agent') ） # 步骤 3：与 CSAT 分数合并（左连接）数据 = 数据.合并( csat_df[['ticket_id', 'csat_score', 'csat_comment']], on='ticket_id', 如何='左' ） # 验证合并结果 print(f"原始门票：{len(tickets_df)}") print(f"合并后：{len(data)}") print(f"匹配的客户：{data['name_customer'].notna().sum()}") print(f"匹配的代理：{data['name_agent'].notna().sum()}") print(f"可用的 CSAT 分数：{data['csat_score'].notna().sum()}") 返回数据 # 用于组合时间段的 Concat 操作 def merge_historical_data(data_sources): ”“” 合并来自多个时间段或来源的数据。参数： data_sources：要组合的数据帧列表返回：将 DataFrame 与源跟踪相结合 ”“” # 为每个DataFrame添加源标识符对于 i，枚举中的 df（data_sources）： df['source_batch'] = f'batch_{i+1}' # 垂直连接组合 = pd.concat(data_sources,ignore_index=True) # 删除重复项（首选较新的数据）组合 = 组合.sort_values('updated_at', 升序=False) 组合=组合.drop_duplicates(subset=['ticket_id'], keep='first') 合并返回 ### 6\.时间序列分析 **重新采样和滚动窗口** # 支持指标的时间序列操作 defcalculate_rolling_metrics(tickets_df, window_days=7): ”“” 计算趋势分析的滚动窗口指标。参数： Tickets_df：带有日期时间索引的票证数据帧 window_days：窗口大小（以天为单位）返回：具有滚动指标的 DataFrame ”“” # 准备时间序列 ts_data = Tickets_df.set_index('created_at').sort_index() # 每日汇总 daily_metrics = ts_data.resample('D').agg({ 'ticket_id': '计数', 'resolution_time': '平均值', 'csat_score': '平均值', 'resolution_sla_met': '意思是' }).rename(columns={'ticket_id': 'daily_tickets'}) # 滚动窗口计算窗口 = 窗口天数 daily_metrics['tickets_rolling_avg'] = ( daily_metrics['daily_tickets'].rolling(window).mean() ） daily_metrics['tickets_rolling_std'] = ( daily_metrics['daily_tickets'].rolling(window).std() ） # 计算异常检测的控制限 daily_metrics['upper_control_limit'] = ( daily_metrics['tickets_rolling_avg'] + (2 * daily_metrics['tickets_rolling_std']) ） daily_metrics['lower_control_limit'] = ( daily_metrics['tickets_rolling_avg'] - (2 * daily_metrics['tickets_rolling_std']) ).clip(下=0) # 标记异常 daily_metrics['is_anomaly'] = ( (daily_metrics['daily_tickets'] > daily_metrics['upper_control_limit']) | (daily_metrics['daily_tickets'] < daily_metrics['lower_control_limit']) ）返回 daily_metrics # 工作日计算 defcalculate_business_day_metrics(tickets_df): """计算不包括周末和节假日的指标。""" 从 pandas.tseries.offsets 导入 CustomBusinessDay # 定义美国假期（根据需要自定义） us_bd = 自定义营业日() # 仅过滤工作日 Tickets_df['is_business_day'] = Tickets_df['created_at'].dt.dayofweek < 5 business_tickets = Tickets_df[tickets_df['is_business_day']] # 计算工作日指标 bd_metrics =business_tickets.groupby( Business_tickets['created_at'].dt.date ).agg({ 'ticket_id': '计数', 'resolution_time': '平均值' }) 返回 bd_metrics ### 7\.数据透视表和交叉表 **创建管理报告** # 执行报告的数据透视表 def create_executive_dashboard_data(tickets_df): ”“” 为执行仪表板创建数据透视表。返回：不同视图的数据透视表字典 ”“” 仪表板 = {} # 1. 按团队和优先顺序划分的门票仪表板['team_priority'] = pd.pivot_table( 门票_df，值='ticket_id', 索引='代理团队', 列='优先级', aggfunc='计数', 填充值=0，边距=真， margins_name='总计' ） # 2. 按团队和渠道划分的平均解决时间仪表板['recision_by_team_channel'] = pd.pivot_table( 门票_df，值='分辨率时间'，索引='代理团队', 列='频道', aggfunc='意思', 填充值=0 ） # 3. 按优先级和周划分的 SLA 合规性 Tickets_df['week'] = Tickets_df['created_at'].dt.to_period('W') 仪表板['sla_compliance_weekly'] = pd.pivot_table( 门票_df，值='resolution_sla_met', 索引='周', 列='优先级', aggfunc='意思', 填充值=0 ） # 4. 按代理和客户级别划分的 CSAT 仪表板['csat_by_agent_tier'] = pd.pivot_table( 门票_df，值='csat_score', 索引='代理名称', 列='customer_tier', aggfunc=['平均值', '计数'], 填充值=0 ） # 5. 门票量热图（星期几与小时） Tickets_df['day_of_week'] = Tickets_df['created_at'].dt.day_name() Tickets_df['小时'] = Tickets_df['created_at'].dt.hour 仪表板['volume_heatmap'] = pd.pivot_table( 门票_df，值='ticket_id', 索引='星期几', 列='小时', aggfunc='计数', 填充值=0 ）返回仪表板 # 交叉列表进行类别分析 defanalyze_category_distribution(tickets_df): """创建交叉表以进行工单类别分析。""" # 类别与优先级类别优先级 = pd.crosstab( Tickets_df['类别'], Tickets_df['优先级'], Normalize='index', # 行百分比边距=真） # 类别与团队（有计数）类别_团队 = pd.crosstab( Tickets_df['类别'], Tickets_df['agent_team'], 边距=真）返回类别_优先级、类别_团队 ### 8\.数据导出和报告 **导出为多种格式** # 导出数据以供利益相关者报告 def export_monthly_report（tickets_df，output_dir，月份）： ”“” 以多种格式导出综合月度报告。参数： Tickets_df：当月的票证数据 output_dir：保存报告的目录月份：月份标识符（例如“2024-01”） ”“” 导入操作系统从日期时间导入日期时间 # 1. 导出为多张Excel excel_path = os.path.join(output_dir, f'support_report_{month}.xlsx') 使用 pd.ExcelWriter(excel_path, engine='openpyxl') 作为编写器： # 总结表摘要 = Tickets_df.groupby('优先级').agg({ 'ticket_id': '计数', 'resolution_time': ['平均值', '中位数'], 'csat_score': '平均值', 'resolution_sla_met': '意思是' }) 摘要.to_excel(作者,sheet_name='摘要') # 团队指标表 team_metrics = Tickets_df.groupby('agent_team').agg({ 'ticket_id': '计数', 'resolution_time': '平均值', 'csat_score': '平均值' }) team_metrics.to_excel(writer,sheet_name='团队指标') # 原始数据表（仅限前 10000 行） Tickets_df.head(10000).to_excel( 作家, sheet_name='原始数据', 索引=假） # 2.导出为CSV进行数据分析 csv_path = os.path.join(output_dir, f'tickets_{month}.csv') Tickets_df.to_csv(csv_path,index=False,encoding='utf-8') # 3. 导出为 JSON 以供 API 使用 json_path = os.path.join(output_dir, f'metrics_{month}.json') 指标={ 'total_tickets': int(tickets_df['ticket_id'].count()), 'avg_resolution_time': float(tickets_df['resolution_time'].mean()), 'sla_compliance': float(tickets_df['resolution_sla_met'].mean()), 'avg_csat': float(tickets_df['csat_score'].mean()), 'by_priority': Tickets_df.groupby('priority')['ticket_id'].count().to_dict() } 将 open(json_path, 'w') 作为 f：导入 json json.dump（指标，f，缩进= 2，默认= str） # 4. 导出到 Parquet 以实现高效存储 parquet_path = os.path.join(output_dir, f'tickets_{month}.parquet') Tickets_df.to_parquet（parquet_path，压缩='snappy'，索引=False） print(f"报告导出到 {output_dir}") 打印（f“ - Excel：{excel_path}”）打印（f“ - CSV：{csv_path}”）打印（f“ - JSON：{json_path}”） print(f" - 镶木地板：{parquet_path}") # 设置 DataFrame 格式以供演示 def format_for_presentation(df): """格式化数据帧以供利益相关者演示。""" # 对数字列进行四舍五入 numeric_cols = df.select_dtypes(include=['float64', 'float32']).columns df[数字列] = df[数字列].round(2) # 设置百分比格式 Percentage_cols = [如果 col 中为“rate”或 col 中为“pct”，则 df.columns 中的 col 为 col] 对于百分比列中的列： df[col] = df[col].apply(lambda x: f"{x*100:.1f}%") # 设置货币格式（如果适用） currency_cols = [col for col in df.columns if 'revenue' in col or 'value' in col] 对于currency_cols中的col： df[col] = df[col].apply(lambda x: f"${x:,.2f}") 返回df ### 9\.性能优化 **内存优化技术** # 优化DataFrame内存使用 def optimize_dataframe_memory(df): ”“” 减少 DataFrame 内存占用。参数： df：要优化的 DataFrame 返回：带有内存使用报告的优化 DataFrame ”“” 初始内存 = df.memory_usage(deep=True).sum() / 1024**2 # 优化整数列 int_cols = df.select_dtypes(include=['int64']).columns 对于 int_cols 中的 col： col_min = df[col].min() col_max = df[col].max() 如果 col_min >= 0：如果 col_max < 255： df[col] = df[col].astype('uint8') elif col_max < 65535： df[col] = df[col].astype('uint16') elif col_max < 4294967295： df[col] = df[col].astype('uint32') 其他：如果 col_min > -128 且 col_max < 127： df[col] = df[col].astype('int8') elif col_min > -32768 且 col_max < 32767： df[col] = df[col].astype('int16') elif col_min > -2147483648 和 col_max < 2147483647： df[col] = df[col].astype('int32') # 优化浮动列 float_cols = df.select_dtypes(include=['float64']).columns df[float_cols] = df[float_cols].astype('float32') # 如果基数较低，则将对象列转换为类别 object_cols = df.select_dtypes(include=['object']).columns 对于 object_cols 中的 col： num_unique = df[col].nunique() num_total = len(df[列]) if num_unique / num_total < 0.5: # 唯一值少于 50% df[col] = df[col].astype('类别') 最终内存 = df.memory_usage(deep=True).sum() / 1024**2 减少 = (1 - 最终内存/初始内存) * 100 print(f"内存使用量从 {initial_memory:.2f} MB 减少到 {final_memory:.2f} MB") print(f"减少: {减少:.1f}%") 返回df # 大数据集的分块处理 def process_large_dataset_in_chunks(file_path, chunk_size=10000): ”“” 分块处理大型 CSV 文件以避免内存问题。参数： file_path：大型 CSV 文件的路径 chunk_size：每个块的行数返回：所有块的聚合结果 ”“” # 初始化聚合容器总票数 = 0 优先级计数 = {} # 分块处理对于 pd.read_csv(file_path, chunksize=chunk_size) 中的块： # 处理每个块块 = clean_ticket_data(块)[0] # 聚合指标 Total_tickets += len(块) chunk_priority = chunk['priority'].value_counts().to_dict() 对于优先级，在 chunk_priority.items() 中计数： priority_counts[优先级] =priority_counts.get(优先级, 0) + 计数返回{ 'total_tickets'：total_tickets， 'priority_distribution'：priority_counts } ### 10\.数据质量和验证 **验证框架** # 全面的数据质量检查类数据质量验证器： """验证支持票证数据集的数据质量。""" def __init__(self, df): self.df = df 自我问题= [] def check_required_columns(self, required_cols): """确保所有必需的列都存在。""" 缺失=设置（required_cols）-设置（self.df.columns）如果丢失： self.issues.append(f"缺少必填列：{missing}") 返回 len(缺失) == 0 def check_null_percentages(self, max_null_pct=0.1): """检查 null 百分比是否超过阈值。""" null_pct = self.df.isnull().sum() / len(self.df) 过量空值 = null_pct[null_pct > max_null_pct] 如果不是excessive_nulls.empty： self.问题.append( f“具有> {max_null_pct * 100}％空值的列：{excessive_nulls.to_dict（）}” ）返回excessive_nulls.empty def check_duplicate_ids(self, id_column='ticket_id'): """检查重复的票证 ID。""" 重复项 = self.df[id_column].duplicate().sum() 如果重复项 > 0： self.issues.append(f"发现 {duplicates} 个重复的工单 ID") 返回重复项 == 0 def check_date_logic(自身): """验证日期字段逻辑。""" 发现问题 = 0 # 创建日期应该早于解决日期如果 self.df.columns 中的“created_at”和 self.df.columns 中的“resolved_at”：无效 = ( self.df['resolved_at'].notna() & (self.df['resolved_at'] < self.df['created_at']) ).sum() 如果无效 > 0： self.问题.append( f“在created_at之前发现有resolved_at的{无效}票证” ） issues_found += 无效 # 检查未来的日期 now = pd.Timestamp.now() 对于 ['created_at', 'resolved_at', 'first_response_at'] 中的 date_col：如果 self.df.columns 中的 date_col ： future_dates = (self.df[date_col] > now).sum() 如果 future_dates > 0: self.问题.append( f“在 {date_col} 中找到了 {future_dates} 个未来日期” ） issues_found += future_dates 返回 issues_found == 0 def check_value_ranges(self, range_checks): ”“” 检查值是否在预期范围内。参数： range_checks：带有列的字典：（最小，最大）对 ”“” 对于 col，range_checks.items() 中的 (min_val, max_val)：如果 col 在 self.df.columns 中：超出范围 = ( (self.df[col] < min_val) | (self.df[col] < min_val) | (self.df[col] > max_val) ).sum() 如果超出范围 > 0： self.问题.append( f"{col}: {out_of_range} 值超出范围 [{min_val}, {max_val}]" ） def 生成报告（自身）： """生成综合验证报告。""" 返回{ 'total_rows': len(self.df), 'total_columns': len(self.df.columns), 'issues_found': len(self.issues), '问题'：自我问题， 'memory_usage_mb': self.df.memory_usage(deep=True).sum() / 1024**2, 'null_summary': self.df.isnull().sum().to_dict() } ### 11\.测试 Pandas 操作 **使用 pytest 进行单元测试** # pytest 装置和数据操作测试导入pytest 将 pandas 导入为 pd 将 numpy 导入为 np 从日期时间导入日期时间，时间增量 @pytest.fixture defsample_ticket_data(): """创建样本票据数据以进行测试。""" np.随机.种子(42) n_票 = 100 返回 pd.DataFrame({ 'ticket_id': 范围(1, n_tickets + 1), 'created_at': pd.date_range('2024-01-01', period=n_tickets, freq='H'), '优先级': np.random.choice(['低', '中', '高'], n_tickets), '状态': np.random.choice(['打开', '进行中', '已解决'], n_tickets), 'agent_id': np.random.choice(['A001', 'A002', 'A003'], n_tickets), 'customer_id': np.random.choice(['C001', 'C002', 'C003'], n_tickets) }) def test_ticket_data_shape(sample_ticket_data): """测试样本数据是否具有预期的形状。""" 断言sample_ticket_data.shape == (100, 6) 在sample_ticket_data.columns 中断言“ticket_id” def test_sla_calculation(): """测试 SLA 计算逻辑。""" df = pd.DataFrame({ 'ticket_id': [1, 2], 'created_at': pd.to_datetime(['2024-01-01 10:00', '2024-01-01 11:00']), 'first_response_at': pd.to_datetime(['2024-01-01 11:00', '2024-01-01 14:00']), 'sla_target_hours': [2, 2] }) df['响应时间'] = ( df['first_response_at'] - df['created_at'] ).dt.total_seconds() / 3600 df['sla_met'] = df['response_time_hours'] <= df['sla_target_hours'] 断言 df.loc[0, 'sla_met'] == True 断言 df.loc[1, 'sla_met'] == False def test_data_cleaning_removes_nulls(sample_ticket_data): """测试数据清理是否处理空值。""" # 添加一些空值 df = Sample_ticket_data.copy() df.loc[0, 'agent_id'] = 无 df.loc[1, 'customer_id'] = 无 # 应用清洁已清理，报告 = clean_ticket_data(df) # 验证空值是否已处理在 clean['agent_id'].values 中断言 'UNASSIGNED' 断言报告['missing_before']['agent_id'] == 1 def test_groupby_aggregation(sample_ticket_data): """测试 groupby 聚合会产生正确的结果。""" 结果=sample_ticket_data.groupby('优先级')['ticket_id'].count() 断言 result.sum() == 100 断言全部（result.index 中的优先级为 ['low', 'medium', 'high'] 中的优先级） ## 最佳实践 ### 1\.始终使用向量化运算使用 pandas 时避免 Python 循环。使用向量化运算以获得更好的性能： # 不好 - 慢循环对于 idx，df.iterrows() 中的行： df.at[idx, 'new_col'] = 行['col1'] * 行['col2'] # 好 - 向量化操作 df['new_col'] = df['col1'] * df['col2'] ### 2\.使用方法链接以提高可读性结果 = ( df .query('状态==“已解决”') .groupby('agent_id') .agg({'resolution_time': 'mean'}) .sort_values('resolution_time') .头(10) ） ### 3\.尽早优化数据类型加载后立即转换为适当的数据类型以节省内存并提高性能。 ### 4\.显式使用 `.loc[]` 和 `.iloc[]` 避免链式索引，这可能导致SettingWithCopyWarning和意外行为。 ### 5\.正确处理时区始终使用时区感知的日期时间对象来支持跨区域的数据。 ### 6\.文档数据转换添加注释来解释复杂转换中的业务逻辑。 ### 7\.每一步验证数据在重大改造后实施验证检查以尽早发现问题。 ### 8\.使用适当的索引类型设置有意义的索引（时间序列的日期时间，查找的ticket_id）以提高性能。 ## 要避免的常见陷阱 1. **SettingWithCopyWarning** ：始终使用 `.loc[]` 来设置值 2. **内存问题**：分块处理大型数据集或优化数据类型 3. **丢失索引** ：请记住，许多操作返回新的 DataFrame 而不保留索引 4. **隐式类型转换**：明确数据类型转换 5. **模糊的真值**：在布尔上下文中评估 Series 时使用 `.any()` 或 `.all()` 6. **混合时区**：确保跨日期时间列的时区处理一致 ## 集成模式 ### 使用 pytest 进行测试始终使用 pytest 装置和参数化装饰器编写数据转换函数的测试。 ### 使用 SQLAlchemy 进行数据库操作使用 SQLAlchemy 引擎进行数据库连接并利用 pandas 的“read_sql”和“to_sql”方法。 ### 使用 PostgreSQL 进行数据持久化将处理后的指标存储在 PostgreSQL 中以进行历史跟踪和仪表板消耗。 ### 使用 Excel 制作利益相关者报告将“pd.ExcelWriter”与 openpyxl 引擎结合使用来创建多表 Excel 报告。 ## 性能指南 1. **对基数较低的列（< 50% 唯一值）使用分类数据类型** 2. **当数据集超出可用内存时分块处理** 3. **使用query()方法**进行复杂过滤（编译为优化代码） 4. **尽可能避免 apply()** \- 使用向量化操作代替 5. **在大型 DataFrame 上使用 eval() 进行复杂表达式** 6. **在读取 CSV 文件时设置适当的数据类型**以避免推理开销 7. **明智地使用copy()** \-仅当您需要真实副本时以避免内存浪费 ## 结论您现在可以使用 pandas 处理客户支持操作的全面数据分析和操作任务。应用这些模式来分析票证数据、跟踪 SLA 合规性、衡量代理绩效并为支持团队生成可操作的见解。始终优先考虑数据质量、性能优化和清晰、可维护的代码。

熊猫

包含技能