上市公司寻租支出数据清洗:从zip解压到面板回归的Python实践

上市公司寻租支出数据清洗:从zip解压到面板回归的Python实践 简介面向经济学、管理学研究者及政策分析人员提供2010—2023年上市公司寻租支出面板数据涵盖多行业、多规模企业观测值可用于分析寻租行为的时间演变、行业分布、企业规模差异及反腐政策影响等课题。资源共9个文件核心包括Stata格式数据文件、Excel结果表、Stata处理代码、变量说明文档、数据来源说明及参考文献PDF压缩包约18.68MB。代码与原始数据营业收入、管理费用明细齐备变量定义清晰可直接运行或基于变量定义进行扩展建模有效节省数据收集与清洗时间。当前已有42人学习下载适合开展企业非生产性支出、政商关系与公司财务等领域实证研究也可用于课程教学与科研复现。1. 上市公司寻租支出数据这个 zip 里装的不只是 Excel管理费用的明细科目长期是会计研究里“隔皮猜瓜”的重灾区。上市公司把业务招待费、差旅费、会议费打散进费用明细不同审计机构披露粒度还不一样想做企业寻租支出方面的实证分析往往要先跟年报附注搏斗好几周。现在手边这份 2010 到 2023 年的寻租支出数据压缩包把大部分 A 股公司的费用科目摘出来打包进了 zip理论上解压后就能直接做面板回归。但真正开工之后问题往往出在“解压后”这三个字zip 传输截断、表头错位、多年度文件混排、金额单位不统一每一步都可能让回归结果静默失真。按数据工程师处理这类数据包的通行流程可以把全部工作拆成五步先验包、再解压、统一读表、算寻租指标、最后处理密码保护和交付检查每一步都有可以直接抄走的命令或 Python 片段新手能照着跑熟手可以跳过开头直接看参数强调的部分。2. 解开 zip 并摸清文件底细从压缩包到可读的年度表2.1 先做完整性校验再解压识别损坏 zip 与压缩包炸弹先别急着双击解压。网盘和邮件传输过程中zip 经常被截断或篡改压缩包内的 crc32 校验和在真正解压前就能告诉你包能不能正常用。损坏 zip 的形态一般分三种尾部截断、中央目录损坏、单个成员流损坏。尾部截断时unzip 或 Python zipfile 打开就报 BadZipFile中央目录损坏但文件数据完好的情况更隐蔽某个文件能解开但内容对不上。处理方法是先用testzip()挨个成员解压并校验 crc32找到的第一个坏成员就是定位线索。另一个容易被忽略的问题是“压缩包炸弹”。从公开渠道下载数据包时可能遇到一个只有几百 KB 的成员解压出几个 GB 的 Excel这在公司财报明细数据里不常见但遇到一次就够折腾磁盘。用infolist()看每个成员的file_size与compress_size的比值压缩率超过 200 的成员要单独确认是否值得完整解压还是只读取特定工作表。import zipfile from pathlib import Path zip_path Path(上市公司-企业寻租支出数据2010-2023年.zip) with zipfile.ZipFile(zip_path) as zf: report [] for info in zf.infolist(): ratio info.file_size / info.compress_size if info.compress_size else 0 report.append((info.filename, info.file_size, info.compress_size, round(ratio, 1))) # 按原始文件大小降序看哪些是真正的大头 for name, raw, comp, ratio in sorted(report, keylambda x: x[1], reverseTrue)[:10]: print(f{name:45} raw{raw/1024/1024:8.1f}MB comp{comp/1024/1024:8.1f}MB ratio{ratio}) bad zf.testzip() print(首个损坏成员:, bad if bad else 无)逻辑说明infolist()读取的是 zip 中央目录区不会解压全部内容。每个ZipInfo对象自带file_size和compress_size两个字节数属性ratio大于 200 就值得人工确认。testzip()返回第一个校验失败的成员文件名全部正常时返回None。参数说明如果report里的文件名出现乱码多是创建方用非 UTF-8 编码写入了中文成员名这属于元数据问题不影响成员内容读取。外部工具提示 “error read zip archive” 时先按上述代码判断是截断还是 crc 不一致再决定重新传输还是用 zip 修复工具重建尾部。常见的故障模式照下面这张表排查能省掉大半无头绪的尝试。常见症状典型原因处理方式打开即报 BadZipFile尾部截断重新下载或修复中央目录testzip 返回某一年文件名单个成员 crc 不一致单独重传该文件不重下全部解压后磁盘爆掉压缩炸弹先看 ratio按需读取文件名中文乱码非 UTF-8 字符集写入只取日期文件名手动校正杀毒软件拦截宏或高压缩率先隔离再校验完整性2.2 用 Python zipfile 按年目录解压并检查文件清单校验通过后正确的落盘方式是按年度分目录而不是把所有文件平铺进同一层。平铺的问题在于 2010 到 2023 年间同名文件太多比如不同年份都叫“明细表.xlsx”一解压就互相覆盖。按年份分目录既解决覆盖问题也为后面批量读取提供清晰路径。下面的脚本用文件名前四位数字推断报告期年份并将单个成员写入对应年度目录。import zipfile, re from pathlib import Path out_root Path(rent_seek_data) out_root.mkdir(exist_okTrue) year_pat re.compile(r^(19|20)\d{2}) with zipfile.ZipFile(zip_path) as zf: for info in zf.infolist(): if info.is_dir(): continue src_name Path(info.filename).name m year_pat.match(src_name) if not m: print(f跳过无年份文件: {info.filename}) continue year m.group(0) target_dir out_root / year target_dir.mkdir(exist_okTrue) # 用成员名做精确解压避免把内部目录结构连带建出来 with zf.open(info) as src, open(target_dir / src_name, wb) as dst: dst.write(src.read()) print(f{year} - {target_dir / src_name})逻辑说明zf.open(info)以只读方式打开单个成员写入目标路径等价于手动提取且避免extract()在部分 Python 版本下因内部路径包含特殊字符产生的意外行为。正则只匹配以 19 或 20 开头的四位数字因此2010_寻租支出.xlsx能命中而rent_expense_2010.xlsx不会命中遇到后一种命名规则要改用year_pat.search。参数说明src.read()会把整个成员读进内存。对超大 Excel 文件比如单个 500MB建议换成shutil.copyfileobj分块复制避免内存峰值过高。这里保留简单写法覆盖大多数不超过百兆的数据包。解压后顺手打印一遍文件清单确认 14 个年度都齐了再往下走。2.3 Excel 与 CSV 混排时的编码与表头问题真正读文件时还有一个坑数据包内部经常 Excel 与 CSV 混排CSV 多为 GBK 编码Excel 第一行又可能不是列名。建议写一个通用 loader在函数内部完成编码探测与表头定位后续所有年度文件都用同一个函数读保持行为一致避免某个年度文件读到一半才发现编码不对。import pandas as pd def load_year_file(path: Path) - pd.DataFrame: if path.suffix.lower() .csv: with open(path, rb) as f: head f.read(512) encoding utf-8-sig if b\xef\xbb\xbf in head else gbk df pd.read_csv(path, encodingencoding, dtype{证券代码: str}) elif path.suffix.lower() in (.xlsx, .xls): # 跳过说明行假设真正表头在第 2 行 df pd.read_excel(path, header1, dtype{证券代码: str}) else: raise ValueError(f未知后缀: {path.suffix}) df.columns [str(c).strip().replace(\n, ).replace( , ) for c in df.columns] return df逻辑说明CSV 分支先读前 512 字节判断 BOM存在 BOM 就用utf-8-sig否则按中文数据最常用的 GBK 解码。Excel 分支用header1跳过第一行说明性文字因为原始文件第一行经常是“单位万元”或“数据来源”。列名清洗统一去掉空格和换行防止某年出现“科目名称 ”而另一年是“科目名称”导致后续 merge 静默失败。参数说明dtype{证券代码: str}是为了保住前导零但前提是文件里真的有“证券代码”这一列。如果某年列名是“stkcd”这个 dtype 不会生效需要在 3.2 节的列名映射后单独再转一次。如果某年 Excel 没有说明第一行header1会把第一行数据当作列名读出来少一行加载后检查首条记录是否为正常数据即可识别。3. 读取寻租支出明细并统一多年度表结构3.1 把年度表批量读成 DataFrame保留年份信息2010 到 2023 年一共 14 个年度每年一个 Excel 是理想情况。现实里常见的是某年拆成多个文件比如“2015_寻租支出_1.xlsx”和“2015_寻租支出_2.xlsx”某年又可能少了一个季度文件。批量读取时依赖 2.2 节落盘的年度目录结构给每个 DataFrame 打上year列最后concat纵向拼接。以下代码遍历年度目录调用 2.3 节的load_year_file。out_root Path(rent_seek_data) frames [] for year_dir in sorted(out_root.iterdir(), keylambda x: x.name): if not year_dir.is_dir() or not year_dir.name.isdigit(): continue files ( list(year_dir.glob(*.xlsx)) list(year_dir.glob(*.xls)) list(year_dir.glob(*.csv)) ) for f in files: df load_year_file(f) df[year] int(year_dir.name) frames.append(df) if frames: raw pd.concat(frames, ignore_indexTrue) print(raw.shape) print(raw[year].value_counts().sort_index())逻辑说明sorted让目录按名称排序但依赖目录名格式统一所以 2.2 节落盘时就要固定命名规则。year列取目录名而不是文件名避免同一目录下两个文件年份信息不一致。concat时ignore_indexTrue重新生成全量索引方便后续按行定位和过滤。参数说明value_counts输出能立刻发现年度缺失。如果 2013 年只有 100 家而其他年份有 3000 家多半是文件命名不符合正则或目录里没有对应文件。检查完年度覆盖再继续不要在已经有洞的表上做回归。某年多个文件存在时concat后同一stkcd同一end_date可能出现多行记录这是正常现象4.2 节会把它们聚合成一行。3.2 列名映射与字段类型统一跨年报出来的表列名不一致是最常见的情况。早期年度用“证券代码”和“费用金额”后期改成“股票代码”和“当期发生额”有的文件还多出“序号”“备注”列影响groupby和后续拼接。统一列名的做法不是逐个 if 判断而是维护一张映射表用rename一次性完成。建议先把所有年份的实际列名打印出来再补映射表避免漏改。COLUMN_MAP { 证券代码: stkcd, 股票代码: stkcd, 代码: stkcd, 截止日期: end_date, 报告期: end_date, 会计期间: end_date, 科目名称: item_name, 费用科目: item_name, 明细科目: item_name, 金额: amount, 费用金额: amount, 本期发生额: amount, 行业代码: industry, 行业: industry, } def normalize_columns(df: pd.DataFrame) - pd.DataFrame: df df.rename(columnsCOLUMN_MAP) keep [c for c in [stkcd, end_date, item_name, amount, industry] if c in df.columns] return df[keep] raw_norm normalize_columns(raw) print(raw_norm.head()) print(raw_norm.dtypes)逻辑说明rename遇到不存在的列名不会报错所以映射表可以覆盖所有年份。标准 schema 有五列其中industry不一定每年都有keep列表里用 if 判断缺失列不保留。输出dtypes用来确认stkcd是object字符串而不是整数类型。参数说明COLUMN_MAP的键不要写带空格的变体比如“证券 代码”一定先调用 2.3 节的列名清洗再rename。如果某年列名是“股票简称”而不是“股票代码”映射表里没有它就会被自动丢弃这正是设计目标缺失字段会在后续合并阶段以 NaN 暴露出来不会悄悄改错。映射前列名映射后字段字段类型说明证券代码 / 股票代码 / 代码stkcdobject必须保留前导零截止日期 / 报告期 / 会计期间end_datedatetime64统一 to_datetime科目名称 / 费用科目 / 明细科目item_nameobject参与关键词匹配金额 / 费用金额 / 本期发生额amountfloat64单位需统一行业代码 / 行业industryobject用于行业调整3.3 空值、负数与极大值的处理策略统一结构之后下一个要处理的是脏值。寻租支出明细里常见的三种情况科目名称为空、金额为负数、金额大得离谱。科目为空的行在匹配关键词时会直接漏掉先保留原始行并打上missing_item标记。金额为负可能是红字冲销也可能是录入错误先按原值保留汇总后再看公司年度合计是否异常。极大值多半来自单位混用按stkcd分组把单个公司年度合计明显偏离分位数的记录列出来人工核对。raw_norm[amount] pd.to_numeric(raw_norm[amount], errorscoerce) item_null raw_norm[item_name].isna() amount_neg raw_norm[amount] 0 print(科目为空行数:, item_null.sum()) print(金额为负行数:, amount_neg.sum()) # 按公司年度合计取极端值便于人工复核 tmp raw_norm.groupby([stkcd, year], as_indexFalse)[amount].sum() extreme tmp[tmp[amount] tmp[amount].quantile(0.995)] print(extreme.sort_values(amount, ascendingFalse).head(10))逻辑说明pd.to_numeric带errorscoerce会把字符串数字转成 float转不动的变成 NaN后续sum操作不会被字符串干扰。负数不直接删除因为冲销是正常记账行为先统计再人工判断。分组后按 99.5 分位筛极端值这个阈值根据样本规模调整。参数说明extreme里出现的公司如果总金额比其他公司高一两个数量级多半能在年报口径里找到原因比如合并报表与母公司报表混用。找到原因后在最终面板表里要么保留并注释要么按合并口径替换。不要用缩尾处理代替“弄清楚为什么这个值这么大”缩尾只能用在回归前不能用来掩盖数据口径错误。4. 计算寻租支出指标并与财务数据交叉验证4.1 管理费用明细里的“寻租相关科目”识别寻租支出没有独立会计科目。会计研究里比较常用的做法是从管理费用明细中找业务招待费、差旅费、办公费和会议费等非生产性支出来近似代表但不同公司差异很大有的把业务招待费全放在销售费用有的放管理费用还有的塞进“其他”项不单独披露。关键词匹配要兼顾准确和召回先做宽松匹配再根据科目名分布做二次筛选比一开始就用精确等值匹配要稳。KEYWORDS [招待, 差旅, 办公, 会议, 业务费用] def is_rent_related(item): if not isinstance(item, str): return False return any(k in item for k in KEYWORDS) raw_norm[rent_flag] raw_norm[item_name].map(is_rent_related) rent_df raw_norm[raw_norm[rent_flag]].copy() print(rent_df[item_name].value_counts().head(20))逻辑说明子串匹配可以覆盖“业务招待费”“业务招待费用”“差旅费-国内差旅”等变体。value_counts输出直接反映每个科目名的频次如果“办公费”这类宽口径科目占比太高就单独做个宽版本指标。实际分析时可以维护三个版本窄口径只含招待和差旅中口径加办公和会议宽口径再加“业务费用”回归里分别作为因变量做稳健性检验。参数说明KEYWORDS里的“业务费用”容易误伤“业务手续费”使用前务必看一遍value_counts结果把干扰词删掉。如果研究问题是高管在职消费那“办公费”就不该删它恰恰是测量维度的一部分保留还是剔除完全取决于具体假设不要照搬通用模板。4.2 用证券代码与报告期做纵向合并对齐 2010 到 2023 年识别出寻租相关科目后下一步是聚合到“公司-年度”粒度。相同stkcd和end_date可能对应多条费用明细需要用groupby求和。同时检查每个公司年度内匹配到的科目数n_items科目数过少的记录说明匹配遗漏可能是科目名里混了英文或缩写。rent_total ( rent_df.groupby([stkcd, end_date, year], as_indexFalse) .agg(total_rent(amount, sum), n_items(item_name, nunique)) ) check rent_total.groupby(year).agg( firms(stkcd, nunique), avg_items(n_items, mean) ) print(check)逻辑说明groupby的三个键里stkcd和end_date定义唯一观测year保留年度维度。n_items用nunique而不是count因为同一科目在同一年可能出现两行重复行不影响识别质量。check表里如果某年avg_items从 4 掉到 1.5说明该年关键词覆盖失效需要回到 4.1 检查该年度文件里的科目名写法是否变化。参数说明如果存在同一公司同一报告期既有年报又有中报的记录end_date会分成 12-31 和 06-30 两类聚合结果就会变成两行。做年度面板时先筛选end_date.dt.month 12或者显式区分年报与中报再进回归否则样本量会翻倍而自己还不知道。4.3 指标计算费用率、同比增速与行业调整有了total_rent后就可以计算常用的 3 类指标先合并外部财务数据再逐项计算顺序颠倒会导致增长率的组内排序错误。指标公式用途rent_intensitytotal_rent / total_assets规模调整后的寻租强度rent_salestotal_rent / revenue收入占比受行业利润率影响rent_growthtotal_rent 同比增速观察寻租行为变化fin pd.read_excel(financial_basics.xlsx, dtype{stkcd: str}) fin[end_date] pd.to_datetime(fin[end_date]) merged rent_total.merge( fin[[stkcd, end_date, total_assets, revenue]], on[stkcd, end_date], howleft ) merged[rent_intensity] merged[total_rent] / merged[total_assets] merged[rent_sales] merged[total_rent] / merged[revenue] merged[rent_growth] merged.groupby(stkcd)[total_rent].pct_change() print(merged.groupby(year)[[rent_intensity, rent_sales]].median())逻辑说明pct_change按stkcd分组后对total_rent做环比某一年缺失会导致前后两年都算错所以计算前先按stkcd和year排序。合并顺序上先 merge 再算指标分母缺失的行会返回 NaN至少能知道哪些样本掉了。最后按年取中位数用来观察 2010 到 2023 年间寻租强度有没有系统波动。参数说明如果financial_basics.xlsx里的total_assets单位是万元而total_rent是元比率会差一万倍换算一致后再算比率。行业调整的常见做法是计算同年度同行业的rent_intensity中位数再减掉该中位数可以消除行业属性差异回归里加了行业固定效应后这个做减法步骤可以被吸收两个结果互相作为稳健性检验。5. 密码保护 zip 的恢复方案与面板数据交付前检查5.1 先分清 ZipCrypto 与 AES 加密加密 zip 在数据交换里很常见。先用一段小代码确认加密算法因为 Python 自带的zipfile只支持 ZipCrypto如果是 AES 加密换pyzipper或 7-Zip 手动处理不要浪费时间调zipfile。import zipfile, pyzipper with zipfile.ZipFile(zip_path) as zf: info zf.infolist()[0] print(加密标志:, info.flag_bits 0x1) with pyzipper.AESZipFile(zip_path) as zf: try: zf.pwd b1234 zf.read(info.filename) print(AES 解密成功) except Exception as e: print(AES 解密失败:, e)逻辑说明flag_bits的第 0 位标记成员是否加密但看不出算法类型。pyzipper.AESZipFile兼容 AES 与 ZipCrypto 双读直接用来验证密码比zipfile兼容性更好。5.2 密码范围可控时做掩码扫描如果密码有规律比如六位日期或连续数字不需要上分布式爆破写个小脚本几分钟跑完。先确定密码生成规则再生成候选列表逐一对第一个成员进行 read 试探。关键点是先用小字典试探不要一开始就铺开到 8 位全穷举。from itertools import product import zipfile def check_password(zf_path, member, pwd): with zipfile.ZipFile(zf_path) as zf: try: zf.read(member, pwdpwd) return True except RuntimeError: return False for pwd in (f{y:04d}{m:02d}{d:02d} for y in range(2010, 2024) for m in range(1, 13) for d in range(1, 32)): if check_password(zip_path, info.filename, pwd.encode()): print(密码:, pwd) break参数说明这里用生成器而不是列表避免把十几万条候选一次性装进内存。每次尝试都会打开并读取一个成员实际更快的做法是复用同一个ZipFile对象把 read 放进 try 里循环候选但那样会涉及文件句柄持续占用示例保留简单写法足够覆盖 5 万条以内的候选集。5.3 交付前检查指标算完后最后四道检查按顺序跑一遍年度覆盖核对 2010 到 2023 年每年观测数确认没有年份断层主键去重同一stkcd与end_date不应出现两行用duplicated确认单位一致性检查total_rent的最大值和分位数是否随年份突变指标复现手工加总 5 家公司明细与rent_total对比差异小于 1% 才放行。dup merged.duplicated(subset[stkcd, end_date], keepFalse) print(重复主键记录数:, dup.sum()) sample merged[merged[stkcd].isin(rent_df[stkcd].dropna().unique()[:5])] print(sample[[stkcd, end_date, total_rent, rent_intensity]].head(10))提示检查顺序不能颠倒先看年度覆盖再去重主键最后做指标复现。重复主键或单位突变一旦出现回到对应章节定位原因不要为了赶进度直接删行删行会在回归结果里留下无法解释的缺口。本文还有配套的精品资源点击获取