17370845950

本文介绍如何基于公司名称的模糊匹配（而非精确相等）合并多个结构不完全一致的 pandas dataframe，并对重复列值自动聚合为元组，同时保留所有原始列。

在实际数据整合场景中，常遇到多个来源表（如财务、人事、工商系统导出）均含“公司名称”字段，但命名存在拼写差异、缩写、空格/标点不一致等问题（如 "A corp" vs "A Corporation"），且各表字段高度异构——部分列共通（如 Value, Currency），部分列独有（如 Leadership, HQ）。此时，传统 pd.merge 或 pd.concat 无法直接满足需求：前者要求精确键匹配，后者仅按行堆叠、不支持语义级对齐。

理想方案需三步闭环：标准化 → 模糊对齐 → 冲突聚合。但原代码存在关键缺陷：它在预收集 all_company_names 后，用 fuzzymatch 在 全局归一化名集合 中反向匹配单个名称，这本质是“静态字典查表”，无法处理跨表间动态相似性（例如 D corp 在表4中出现两次，但全局去重后仅存一个，导致第二次匹配失效）；更严重的是，consolidated_data 使用 effective_name 作为键，却未对同名不同形（如 D corp 和 D corporation）做统一归一化锚定，致使模糊匹配结果未真正驱动行合并逻辑。

更简洁、鲁棒的解法是绕过显式模糊匹配，改用

1. 标准化公司名并添加序号索引：对每张表，先将 'Company Name' 统一小写、去空格（可扩展为正则清洗），再用 groupby('Company Name').cumcount() 为每个重复公司名打序号（如 D corp 出现两次 → (D corp, 0) 和 (D corp, 1)），构成复合索引；
2. 水平拼接（concat axis=1）：以 (Company Name, 序号) 为索引对齐所有表，缺失位置自动填充 NaN；
3. 按公司名分组聚合：对每列应用自定义聚合函数：非空值去重后若多于1个则转为 tuple，否则取唯一值（float('nan') 表示全空）。

import pandas as pd
from typing import List, Tuple, Any

def fuzzy_concat(dfs: List[pd.DataFrame]) -> pd.DataFrame:
    """
    基于公司名称模糊语义合并多个DataFrame。
    注意：本实现假设名称差异可通过标准化（小写+去空格）解决；
    如需强模糊匹配（如编辑距离），可在标准化后增加fuzzywuzzy预处理步骤。
    """
    def clean_name(x):
        return x.astype(str).str.lower().str.strip()

    # 步骤1：为每张表构建 (Company Name, 序号) 复合索引
    indexed_dfs = []
    for df in dfs:
        if 'Company Name' not in df.columns:
            raise ValueError("所有DataFrame必须包含'Company Name'列")
        # 标准化公司名并生成序号
        cleaned_names = clean_name(df['Company Name'])
        idx = pd.MultiIndex.from_arrays(
            [cleaned_names, df.groupby(cleaned_names).cumcount()],
            names=['Company Name', 'Seq']
        )
        indexed_dfs.append(df.set_index(idx))

    # 步骤2：水平拼接，自动对齐复合索引
    combined = pd.concat(indexed_dfs, axis=1)

    # 步骤3：按公司名分组，对每列聚合
    def agg_col(series: pd.Series) -> Any:
        non_null = series.dropna()
        if len(non_null) == 0:
            return float('nan')
        elif len(non_null) == 1:
            return non_null.iloc[0]
        else:
            # 去重后转元组（保留原始类型）
            unique_vals = list(set(non_null))
            return tuple(unique_vals) if len(unique_vals) > 1 else unique_vals[0]

    result = combined.groupby(level='Company Name').agg(agg_col).reset_index()
    return result

# 示例调用
# master_df = fuzzy_concat([df1, df2, df3, df4])

✅ 优势说明：

• 无需外部库：摆脱 fuzzywuzzy 依赖，降低部署复杂度；
• 天然支持重复名：通过 cumcount() 区分同名多记录（如 D corp 的两条数据独立保留）；
• 列完整性保障：concat(axis=1) 确保所有输入列无损进入结果，缺失值自动补 NaN；
• 冲突处理明确：tuple 仅在同公司名下该列存在多个不同非空值时生成，避免误合并。

⚠️ 注意事项：

• 若公司名差异过大（如 "Apple Inc." vs "AAPL"），需在 clean_name 中集成 fuzzywuzzy.process.extractOne 预映射到标准名库；
• 元组内顺序不保证，如需确定性，可改为 tuple(sorted(set(non_null)))（仅适用于可排序类型）；
• 性能敏感场景建议用 dask 或 polars 替代 pandas 处理超大表。

此方法以“标准化+索引对齐”替代“逐行模糊匹配”，既提升鲁棒性，又大幅简化逻辑，是工业级数据融合的推荐实践。