# 数据转换（通用） ## 概述此技能使您能够执行全面的数据转换，包括清理、规范化、整形、过滤和特征工程。与云托管解决方案不同，此技能使用标准 Python 数据操作库（**pandas**、**numpy**、**sklearn**）并在您的环境中**本地**执行，使其与**所有 LLM 提供商**兼容，包括 GPT、Gemini、Claude、DeepSeek 和 Qwen。 ## 何时使用此技能 * 清理和预处理原始数据 * 标准化或缩放数字特征 * 在宽格式和长格式之间重塑数据 * 处理缺失值 * 过滤和子集数据集 * 合并多个数据集 * 从现有功能创建新功能 * 转换数据类型和格式 ## 如何使用 ### 第 1 步：导入所需的库将 pandas 导入为 pd 将 numpy 导入为 np 从 sklearn.preprocessing 导入 StandardScaler、MinMaxScaler、RobustScaler 从 sklearn.preprocessing 导入 LabelEncoder、OneHotEncoder 进口警告 warnings.filterwarnings('忽略') ### 步骤 2：数据清理 # 加载数据 df = pd.read_csv('data.csv') # 检查是否有缺失值 print("每列缺少值：") 打印（df.isnull（）。sum（）） # 删除重复项 df_clean = df.drop_duplicates() print(f"删除了 {len(df) - len(df_clean)} 重复行") # 删除有缺失值的行 df_clean = df_clean.dropna() # 或者填充缺失值 df_clean = df.copy() df_clean['numeric_col'] = df_clean['numeric_col'].fillna(df_clean['numeric_col'].median()) df_clean['categorical_col'] = df_clean['categorical_col'].fillna('未知') # 使用IQR方法去除异常值 def remove_outliers(df, 列, 乘数=1.5): Q1 = df[列].分位数(0.25) Q3 = df[列].分位数(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 乘数 * IQR upper_bound = Q3 + 乘数 * IQR 返回 df[(df[列] >= lower_bound) & (df[column] <= upper_bound)] df_clean = remove_outliers(df_clean, '表达式级别') print(f"✅ 已清理的数据：剩余 {len(df_clean)} 行") ### 步骤 3：标准化和缩放 # 选择数字列 numeric_cols = df.select_dtypes(include=[np.number]).columns # 方法 1：Z 分数归一化 (StandardScaler) 定标器=标准定标器() df_normalized = df.copy() df_normalized[numeric_cols] = scaler.fit_transform(df[numeric_cols]) print("标准化 Z 分数（平均值 = 0，标准差 = 1）"）打印（df_normalized[numeric_cols].describe（）） # 方法 2：最小-最大缩放（0-1 范围） scaler_minmax = MinMaxScaler() df_scaled = df.copy() df_scaled[numeric_cols] = scaler_minmax.fit_transform(df[numeric_cols]) print("\n最小-最大缩放（范围 0-1）") 打印（df_scaled [numeric_cols].describe（）） # 方法 3：稳健缩放（抵抗异常值） scaler_robust = RobustScaler() df_robust = df.copy() df_robust[numeric_cols] = scaler_robust.fit_transform(df[numeric_cols]) print("\n稳健缩放 (中位数=0, IQR=1)") 打印（df_robust[numeric_cols].describe（）） # 方法四：对数变换 df_log = df.copy() df_log['log_表达式'] = np.log1p(df_log['表达式']) # log1p(x) = log(1+x) print("✅ 数据标准化和缩放") ### 步骤 4：数据重塑 # 将宽格式转换为长格式（melt） # 宽格式：列是不同的条件/样本 # 长格式：一列用于变量，一列用于值 df_wide = pd.DataFrame({ '基因': ['GENE1', 'GENE2', 'GENE3'], '样本_A': [10, 20, 15], '样本_B': [12, 18, 14], '样本_C'：[11,22,16] }) df_long = df_wide.melt( id_vars=['基因'], var_name='样本', value_name='表达式' ） print("长格式：") 打印（df_long） # 将长格式转换为宽格式（pivot） df_wide_reconstructed = df_long.pivot( 索引='基因', 列='样本'，值='表达式' ） print("\n宽格式（重构）：") 打印（df_wide_reconstructed） # 带聚合的数据透视表 df_pivot = df_long.pivot_table( 索引='基因', 列='样本'，值='表达式', aggfunc='mean' # 可以使用总和、中位数等。） print("✅ 数据重塑") ### 步骤 5：过滤和子集化 # 按条件过滤行高表达式 = df[df['表达式'] > 100] # 多个条件（AND）过滤 = df[(df['表达式'] > 50) & (df['qvalue'] < 0.05)] # 多个条件（或）过滤 = df[(df['celltype'] == 'T 细胞') | (df['细胞类型'] == 'B细胞')] # 按值列表过滤 selected_genes = ['GENE1', 'GENE2', 'GENE3'] 过滤 = df[df['gene'].isin(selected_genes)] # 按字符串模式过滤 Filtered = df[df['gene'].str.startswith('MT-')] # 线粒体基因 # 选择特定列 selected_cols = df[['基因', 'log2FC', 'pvalue', 'qvalue']] # 按模式选择列 numeric_cols = df.select_dtypes(include=[np.number]) categorical_cols = df.select_dtypes(include=['对象', '类别']) # 随机行样本 df_sample = df.sample(n=1000, random_state=42) # 1000 随机行 df_sample_frac = df.sample(frac=0.1, random_state=42) # 10% 的行 # 前 N 行 top_genes = df.nlargest(10, '表达式') Bottom_genes = df.nsmallest(10, 'pvalue') print(f"✅ 过滤后的数据集：{len(filtered)} 行") ### 步骤 6：合并和连接数据集 # 内连接（仅匹配行）合并= pd.merge(df1, df2, on='基因', how='内部') # 左连接（df1 中的所有行）合并= pd.merge(df1, df2, on='gene', how='left') # 外连接（两者的所有行）合并= pd.merge(df1, df2, on='gene', how='outer') # 连接多列合并= pd.merge(df1, df2, on=['基因', '样本'], how='内部') # 连接不同的列名合并 = pd.merge( df1、df2、 left_on='基因名称', right_on='gene_id', 如何='内在' ） # 垂直连接（堆叠数据帧）组合 = pd.concat([df1, df2], axis=0,ignore_index=True) # 水平连接（并排）组合 = pd.concat([df1, df2], axis=1) print(f"✅ 合并数据集：{len(merged)} 行") ## 高级功能 ### 处理缺失值 # 检查缺失值模式 Missing_summary = pd.DataFrame({ '列': df.columns, 'missing_count': df.isnull().sum(), 'missing_percent': (df.isnull().sum() / len(df) * 100).round(2) }) print("缺失值摘要：") 打印(missing_summary[missing_summary['missing_count'] > 0]) # 策略1：填充统计指标 df_filled = df.copy() df_filled['numeric_col'].fillna(df_filled['numeric_col'].median(), inplace=True) df_filled['categorical_col'].fillna(df_filled['categorical_col'].mode()[0], inplace=True) # 策略2：前向填充（使用之前的值） df_filled = df.fillna(方法='ffill') # 策略 3：插值（对于时间序列） df_filled = df.copy() df_filled['表达式'] = df_filled['表达式'].interpolate(method='线性') # 策略 4：删除缺失值过多的列阈值 = 0.5 # 如果缺失 >50% 则丢弃 df_cleaned = df.dropna(thresh=len(df) * 阈值，轴=1) print("✅ 处理缺失值") ### 特征工程 # 从现有功能创建新功能 # 1. 对连续变量进行分箱 df['表达式类别'] = pd.cut( df['表达式'], bin=[0, 10, 50, 100, np.inf], 标签=['非常低', '低', '中', '高'] ） # 2.创建比率特征 df['gene_to_umi_ratio'] = df['n_genes'] / df['n_counts'] # 3. 创建交互功能 df['交互'] = df['特征1'] * df['特征2'] # 4. 提取日期时间特征 df['日期'] = pd.to_datetime(df['时间戳']) df['年份'] = df['日期'].dt.year df['月份'] = df['日期'].dt.月份 df['day_of_week'] = df['date'].dt.dayofweek # 5. 分类变量的 One-hot 编码 df_encoded = pd.get_dummies(df, columns=['celltype', 'condition'], prefix=['cell', 'cond']) # 6. 标签编码（序数） le = 标签编码器() df['celltype_encoded'] = le.fit_transform(df['celltype']) # 7. 创建多项式特征 df['表达式_平方'] = df['表达式'] ** 2 df['表达式_cubed'] = df['表达式'] ** 3 # 8. 创建滞后特征（时间序列） df['表达式_滞后1'] = df.groupby('基因')['表达式'].shift(1) df['表达_滞后2'] = df.groupby('基因')['表达'].shift(2) print("✅ 创建了新功能") ### 分组和聚合 # 按单列分组并聚合 cluster_stats = df.groupby('cluster').agg({ '表达式': ['平均值', '中位数', '标准', '计数'], 'n_genes': '意思', 'n_counts': '总和' }) print("集群统计信息：") 打印（集群统计） # 按多列分组 stats = df.groupby(['cluster', 'celltype']).agg({ '表达式': '意思', 'qvalue': lambda x: (x < 0.05).sum() # 计数显着性 }) # 应用自定义函数 def custom_stats（组）：返回 pd.Series({ 'mean_expr': group['表达式'].mean(), 'cv': group['expression'].std() / group['expression'].mean(), # 变异系数 'n_cells': len(组) }) cluster_custom = df.groupby('cluster').apply(custom_stats) print("✅ 数据汇总") ### 数据类型转换 # 将列转换为不同类型 df['簇'] = df['簇'].astype(str) df['表达式'] = df['表达式'].astype(float) df['重要'] = df['重要'].astype(bool) # 转换为分类（节省内存） df['celltype'] = df['celltype'].astype('类别') # 解析日期 df['date'] = pd.to_datetime(df['date_string'], format='%Y-%m-%d') # 将数值转换为分类 df['表达式级别'] = pd.cut(df['表达式'], bins=3, labels=['低', '中', '高']) # 字符串操作 df['gene_upper'] = df['gene'].str.upper() df['is_mitochondrial'] = df['gene'].str.startswith('MT-') print("✅ 数据类型转换") ## 常见用例 ### AnnData 到 DataFrame 的转换 # 将 AnnData .obs（单元格元数据）转换为 DataFrame df_cells = adata.obs.copy() # 将 .var（基因元数据）转换为 DataFrame df_genes = adata.var.copy() # 提取表达矩阵到DataFrame # 警告：对于大型数据集来说，这可能会占用大量内存 df_表达式 = pd.DataFrame( adata.X.toarray() if hasattr(adata.X, 'toarray') else adata.X, 索引=adata.obs_names，列=adata.var_names ） # 提取特定层如果在 adata.layers 中“标准化”： df_normalized = pd.DataFrame( adata.layers['标准化'], 索引=adata.obs_names，列=adata.var_names ） print("✅ AnnData 转换为 DataFrames") ### 基因表达矩阵变换 # 转置：基因为行，细胞为列 → 细胞为行，基因为列 df_转置 = df.T # 对数变换基因表达 df_log = np.log1p(df) # log1p(x) = log(1+x)，避免 log(0) # 每个基因的 Z 分数标准化（跨细胞） df_zscore = df.apply(lambda x: (x - x.mean()) / x.std(), axis=1) # 每个细胞的比例（除以文库大小）库大小 = df.sum(轴=1) df_normalized = df.div(library_sizes, axis=0) * 1e6 # CPM 标准化 # 过滤低表达基因 min_cells = 10 # 基因必须在至少10个细胞中表达基因掩码 = (df > 0).sum(axis=0) >= min_cells df_filtered = df.loc[:, 基因掩码] print(f"✅ 过滤到 {df_filtered.shape[1]} 基因") ### 差异表达结果处理 # 假设 deg_df 有列：gene、log2FC、pvalue、qvalue # 添加重要性标签 deg_df['regulation'] = '不显着' deg_df.loc[(deg_df['log2FC'] > 1) & (deg_df['qvalue'] < 0.05), '调节'] = '上调' deg_df.loc[(deg_df['log2FC'] < -1) & (deg_df['qvalue'] < 0.05), '调节'] = '下调' # 按重要性排序 deg_df_sorted = deg_df.sort_values('qvalue') # 上调最多的基因 top_up = deg_df[deg_df['调节'] == '上调'].nlargest(20, 'log2FC') # 下调最多的基因 top_down = deg_df[deg_df['调节'] == '下调'].nsmallest(20, 'log2FC') # 创建汇总表摘要 = deg_df.groupby('监管').agg({ '基因': '计数', 'log2FC': ['平均值', '中位数'], 'qvalue': '最小值' }) print("DEG摘要：") 打印（摘要） # 导出结果 deg_df_sorted.to_csv('deg_results_processed.csv',index=False) print("✅ DEG 结果已处理并保存") ### 批量处理多个文件导入全局 # 查找所有CSV文件 file_paths = glob.glob('data/*.csv') # 读取并组合 dfs = [] 对于 file_paths 中的 file_path： df = pd.read_csv(文件路径) # 添加源文件作为列 df['源文件'] = file_path.split('/')[-1] dfs.append(df) # 合并所有 df_combined = pd.concat(dfs,ignore_index=True) print(f"✅ 已处理 {len(file_paths)} 个文件，总共 {len(df_combined)} 行") ## 最佳实践 1. **首先检查数据** ：始终使用 `df.head()`、`df.info()`、`df.describe()` 来理解数据 2. **修改前复制** ：使用`df.copy()`避免修改原始数据 3. **链操作** ：使用方法链以提高可读性： `df.dropna().drop_duplicates().reset_index(drop=True)` 4. **索引管理** : 过滤后重置索引： `df.reset_index(drop=True)` 5. **内存效率**：对低基数字符串列使用分类数据类型 6. **矢量化**：避免循环；使用向量化运算（numpy、pandas 内置） 7. **文档**：注释复杂的转换 8. **Validation**：每次重大改造后检查数据 ## 故障排除 ### 问题：“SettingWithCopyWarning” **解决方案**：使用 `.copy()` 创建显式副本 df_subset = df[df['表达式'] > 10].copy() df_subset['new_col'] = value # 无警告 ### 问题：“大数据集出现内存错误” **解决方案**：分块处理块大小 = 10000 块=[] 对于 pd.read_csv('large_file.csv', chunksize=chunk_size) 中的块： # 处理块已处理=块[块['表达式']> 0] chunks.append（已处理） df = pd.concat(块,ignore_index=True) ### 问题：“合并时出现关键错误” **解决方案**：检查列名称和存在 print("df1 中的列：", df1.columns.tolist()) print("df2 中的列：", df2.columns.tolist()) # 如果名称不同，则使用 left_on/right_on 合并 = pd.merge(df1, df2, left_on='gene_name', right_on='gene_id') ### 问题：“合并时数据类型不匹配” **解决方案**：确保类型一致 df1['基因'] = df1['基因'].astype(str) df2['基因'] = df2['基因'].astype(str) 合并= pd.merge(df1, df2, on='基因') ### 问题：“索引对齐错误” **解决方案**：重置索引或指定 `ignore_index=True` df_combined = pd.concat([df1, df2],ignore_index=True) ## 关键 API 参考 - DataFrame 与系列属性 ### 重要提示：`.dtype` 与 `.dtypes` \- 常见陷阱！ **正确用法：** # 对于 DataFrame - 使用 .dtypes（复数）获取所有列类型 df.dtypes # 返回以列名作为索引、dtypes 作为值的系列 # 对于单列（系列） - 使用 .dtype (SINGULAR) df['column_name'].dtype # 返回单个 dtype 对象 # 检查特定列类型如果 df['表达式'].dtype == 'float64': print("表达式为 float64") # 检查所有列类型 print(df.dtypes) # 显示每列的 dtype **错误 - 请勿使用：** # 错了！ DataFrame 没有 .dtype （单数） # df.dtype # AttributeError: 'DataFrame' 对象没有属性 'dtype' # 错了！这会失败 # if df.dtype == 'float64': # 错误！ ### DataFrame 类型检查方法 # 获取所有列的数据类型 df.dtypes # 获取详细信息，包括数据类型 df.info() # 检查列是否为数字 pd.api.types.is_numeric_dtype(df['列']) # 检查列是否是分类的 pd.api.types.is_categorical_dtype(df['列']) # 按数据类型选择列 numeric_cols = df.select_dtypes(include=['number']) string_cols = df.select_dtypes(include=['object', 'string']) ### Series 与 DataFrame - 主要区别属性/方法 |系列|数据框 ---|---|--- `.dtype` | ✅ 返回单一数据类型 | ❌ 属性错误 `.dtypes` | ❌ 属性错误 | ✅ 返回一系列 dtypes `.shape` | `(n,)` 元组 | `(n, m)` 元组 `.values` |一维数组 |二维阵列 ## 技术说明 * **库** ：使用 `pandas` (1.x+)、`numpy`、`scikit-learn` （广泛支持） * **执行** ：在代理的沙箱中本地运行 * **兼容性**：适用于所有 LLM 提供商（GPT、Gemini、Claude、DeepSeek、Qwen 等） * **性能** ：Pandas 使用 C 后端进行了优化；对于 <1M 行，大多数操作都很快 * **内存** ：Pandas DataFrame 将数据存储在内存中；对非常大的文件使用分块 * **精度** ：数值运算默认使用float64（可以使用float32来节省内存） ## 参考文献 * pandas 文档： * pandas 用户指南： * scikit-learn 预处理： * pandas备忘单：

数据转换

包含技能