15  CSMAR 数据处理

CSMAR (China Stock Market & Accounting Research) 是一个提供中国上市公司数据的数据库。本文将介绍如何使用 CSMAR 数据库获取上市公司基本信息和财务资料,并经过合并、清洗等处理,得到清洁数据。

15.1 从 CAMAR 数据库下载 Excel 数据文件

.csv 优于 .xlsx

我此前对 Python 处理数据的机制了解不足,导致我最初下载的都是 .xlsx 格式的数据。虽然这种 Excel 格式的数据可以使用 pandas 读取,但非常耗时 (一份包含 8w 行观察值的资产负债表大概需要 50s)。相比之下,.csv 格式的数据读取速度更快,通常只需要几秒钟。

因此,大家从 CSMAR 数据库下载数据时,建议选择 .csv 格式,而不是 .xlsx 格式。

15.1.1 数据下载页面

主要有两种查询方案:

  • 单表查询:直接查询某一张表格的数据。
  • 多表查询:可以跨表查询,形成一个新的表格。

两种模式下的查询流程大致相同:选择子库 >> 选择时间范围 >> 选择代码范围 >> 选择字段范围 >> 下载数据。

建议尽量选择 .csv.txt 格式下载数据,因为这两种格式的数据读取速度更快。

下载后的文件通常为 .zip 格式,解压后会得到两份文件: - FileName.csv: 数据文件 - FileName[DES][xlsx].txt: 变量说明文件

数据库说明书:每个子库的右上角都会显示「数据库说明书」,点击后可下载 PDF 格式的说明书,里面包含了该子库的所有表格、字段的详细信息。

# 文件路径设定{tag}
# Note:进行后续分析之前,请先执行本 Cell 中的代码,以确保所有路径正确设置。

import os
import sys
import warnings
warnings.filterwarnings("ignore")

# 打开 .ipynb 文件的路径记为当前工作目录
#path = os.getcwd()
path = r'D:\Github\ds_data\data\CSMAR'    # 替换为你的实际路径
os.chdir(path)  # 切换工作目录
print('Working Directory:', os.getcwd())


# Folders 
zip_folder = os.path.join(path, 'data_raw_zip')       # 原始压缩包存放目录
extract_folder = os.path.join(path, 'data_raw')       # 解压后的数据存放目录
data_clean_folder = os.path.join(path, 'data_clean')  # 清洗后的数据存放目录
Working Directory: d:\Github\ds_data\data\CSMAR
# 列示 'data_raw_zip' 文件夹中的所有文件
for item in os.listdir(zip_folder):
    print(item)
CSMAR常用变量-2000-2024.zip
上市公司基本信息变更表2000-2024.zip
上市公司基本信息年度表.zip
利润表-现金流量表-2000-2010.zip
利润表-现金流量表-2011-2024.zip
资产负债表-2000-2010.zip
资产负债表-2011-2024.zip

15.2 文件夹结构与处理流程说明

本项目的数据处理涉及以下文件夹:

  • data_raw_zip:存放从 CSMAR 下载的原始数据(.zip 压缩包),每个压缩包解压后会生成同名文件夹。
  • data_raw:存放解压后的原始数据文件。
  • data_clean:存放清洗后的数据文件。

15.2.1 处理流程

  1. 列出 data_raw_zip 文件夹中的所有压缩文件。
  2. 将每个压缩包解压到 data_raw 文件夹下,解压后以压缩包同名文件夹存放。
  3. 为便于后续批量处理,对解压后的文件进行统一重命名。
  4. 编写函数,实现指定文件夹下文件的批量读取。

15.3 解压 zip 文件到指定文件夹

提示词
  • 将当前路径下的 ‘data_raw_zip’ 文件夹中的所有 zip 文件解压到 ‘data_raw’ 文件夹中。
  • 如果 ‘data_raw’ 文件夹不存在,则创建该文件夹。
  • 每个 zip 文件解压后生成一个文件夹,文件夹名称与 zip 文件名相同,若有同名文件则覆盖之。
import zipfile
import shutil

# 创建 'data_raw' 文件夹(如果不存在)
if not os.path.exists(extract_folder):
    os.makedirs(extract_folder)

# 遍历 'data_raw_zip' 文件夹中的所有 zip 文件
for item in os.listdir(zip_folder):
    if item.endswith('.zip'):
        zip_path = os.path.join(zip_folder, item)
        folder_name = os.path.splitext(item)[0]
        target_dir = os.path.join(extract_folder, folder_name)
        # 如果目标文件夹已存在,则先删除
        if os.path.exists(target_dir):
            shutil.rmtree(target_dir)
        # 解压 zip 文件到目标文件夹
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            zip_ref.extractall(target_dir)
        print(f"已解压: {zip_path} -> {target_dir}")
已解压: D:\Github\ds_data\data\CSMAR\data_raw_zip\CSMAR常用变量-2000-2024.zip -> D:\Github\ds_data\data\CSMAR\data_raw\CSMAR常用变量-2000-2024
已解压: D:\Github\ds_data\data\CSMAR\data_raw_zip\上市公司基本信息变更表2000-2024.zip -> D:\Github\ds_data\data\CSMAR\data_raw\上市公司基本信息变更表2000-2024
已解压: D:\Github\ds_data\data\CSMAR\data_raw_zip\上市公司基本信息年度表.zip -> D:\Github\ds_data\data\CSMAR\data_raw\上市公司基本信息年度表
已解压: D:\Github\ds_data\data\CSMAR\data_raw_zip\利润表-现金流量表-2000-2010.zip -> D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2000-2010
已解压: D:\Github\ds_data\data\CSMAR\data_raw_zip\利润表-现金流量表-2011-2024.zip -> D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2011-2024
已解压: D:\Github\ds_data\data\CSMAR\data_raw_zip\资产负债表-2000-2010.zip -> D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2000-2010
已解压: D:\Github\ds_data\data\CSMAR\data_raw_zip\资产负债表-2011-2024.zip -> D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2011-2024

15.4 更改部分文件夹中的文件名

15.4.1 问题说明

data_raw 文件夹中,有些子文件夹中的文件名是由 CSMAR 自动生成的,没有实际意义。为了便于后续处理,需要将这些文件名更改为更有意义的名称。

此处以 data_raw/资产负债表-?? 文件夹为例,可以看出,资产负债表-2000-2010资产负债表-2011-2024 这两个文件夹中包含的文件名完全相同,这会导致随后纵向合并时难以区分数据来源。

# 提示词:列出 `data_raw` 文件夹中包含关键词 {'资产负债表'} 的子文件夹的 file tree。

keywords = '资产负债表'

def print_keyword_file_tree(root, keyword, indent=""):
    for item in os.listdir(root):
        item_path = os.path.join(root, item)
        if os.path.isdir(item_path) and keyword in item:
            print(indent + "|-- " + item)
            for subitem in os.listdir(item_path):
                print(indent + "    |-- " + subitem)

print_keyword_file_tree(extract_folder, keywords)
|-- 资产负债表-2000-2010
    |-- 版权声明.pdf
    |-- 跨表查询_沪深京股票(年频).xlsx
    |-- 跨表查询_沪深京股票(年频)[DES][.xlsx].txt
|-- 资产负债表-2011-2024
    |-- 版权声明.pdf
    |-- 跨表查询_沪深京股票(年频).xlsx
    |-- 跨表查询_沪深京股票(年频)[DES][.xlsx].txt

查看后,发现子文件夹 利润表-现金流量表-??CSMAR常用变量-2000-2024 中的文件名也存在类似问题。

处理思路是:用文件夹名称作为该文件夹下的文件的文件名。因此,我们可以使用如下提示词生成处理代码:

提示词
  1. 目的:更改 ‘data_raw’ 文件夹中部分子文件夹中的文件的名称。
  2. 子文件夹名称为 ‘sub_folder_name’,其内部包含的文件记为 {Files}
  3. 如果 ‘sub_folder_name’ 中包含关键词 {‘常用变量’, ‘资产负债表’, ‘利润表’},则把 {Files} 中的 ‘.xlsx’ 和 ‘.txt’ 文件的名称改为 ‘folder_name.xlsx’ 和 ‘folder_name_DES.txt’。
# 增加关键词 '常用变量'
keywords = ['常用变量', '资产负债表', '利润表']

# 检查并重命名文件
for subfolder in os.listdir(extract_folder):
    subfolder_path = os.path.join(extract_folder, subfolder)
    if os.path.isdir(subfolder_path) and any(kw in subfolder for kw in keywords):
        for file in os.listdir(subfolder_path):
            file_path = os.path.join(subfolder_path, file)
            if file.endswith('.xlsx'):
                new_name = f"{subfolder}.xlsx"
                new_path = os.path.join(subfolder_path, new_name)
                os.rename(file_path, new_path)
                print(f"重命名: {file_path} -> {new_path}")
            elif file.endswith('.txt'):
                new_name = f"{subfolder}_DES.txt"
                new_path = os.path.join(subfolder_path, new_name)
                os.rename(file_path, new_path)
                print(f"重命名: {file_path} -> {new_path}")
重命名: D:\Github\ds_data\data\CSMAR\data_raw\CSMAR常用变量-2000-2024\常用变量查询(年度).xlsx -> D:\Github\ds_data\data\CSMAR\data_raw\CSMAR常用变量-2000-2024\CSMAR常用变量-2000-2024.xlsx
重命名: D:\Github\ds_data\data\CSMAR\data_raw\CSMAR常用变量-2000-2024\常用变量查询(年度)[DES][xlsx].txt -> D:\Github\ds_data\data\CSMAR\data_raw\CSMAR常用变量-2000-2024\CSMAR常用变量-2000-2024_DES.txt
重命名: D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2000-2010\跨表查询_沪深京股票(年频).xlsx -> D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2000-2010\利润表-现金流量表-2000-2010.xlsx
重命名: D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2000-2010\跨表查询_沪深京股票(年频)[DES][.xlsx].txt -> D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2000-2010\利润表-现金流量表-2000-2010_DES.txt
重命名: D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2011-2024\跨表查询_沪深京股票(年频).xlsx -> D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2011-2024\利润表-现金流量表-2011-2024.xlsx
重命名: D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2011-2024\跨表查询_沪深京股票(年频)[DES][.xlsx].txt -> D:\Github\ds_data\data\CSMAR\data_raw\利润表-现金流量表-2011-2024\利润表-现金流量表-2011-2024_DES.txt
重命名: D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2000-2010\跨表查询_沪深京股票(年频).xlsx -> D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2000-2010\资产负债表-2000-2010.xlsx
重命名: D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2000-2010\跨表查询_沪深京股票(年频)[DES][.xlsx].txt -> D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2000-2010\资产负债表-2000-2010_DES.txt
重命名: D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2011-2024\跨表查询_沪深京股票(年频).xlsx -> D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2011-2024\资产负债表-2011-2024.xlsx
重命名: D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2011-2024\跨表查询_沪深京股票(年频)[DES][.xlsx].txt -> D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2011-2024\资产负债表-2011-2024_DES.txt

15.4.2 查看处理后的文件名

# 提示词:列出 `data_raw` 文件夹中包含关键词 {'资产负债表'} 的子文件夹的 file tree。

keyword = '资产负债表'

def print_keyword_file_tree(root, keyword, indent=""):
    for item in os.listdir(root):
        item_path = os.path.join(root, item)
        if os.path.isdir(item_path) and keyword in item:
            print(indent + "|-- " + item)
            for subitem in os.listdir(item_path):
                print(indent + "    |-- " + subitem)

print_keyword_file_tree(extract_folder, keyword)
|-- 资产负债表-2000-2010
    |-- 版权声明.pdf
    |-- 资产负债表-2000-2010.xlsx
    |-- 资产负债表-2000-2010_DES.txt
|-- 资产负债表-2011-2024
    |-- 版权声明.pdf
    |-- 资产负债表-2011-2024.xlsx
    |-- 资产负债表-2011-2024_DES.txt

15.5 列示 ‘data_raw’ 文件夹中的 file tree

提示词
  • 列示 ‘data_raw’ 文件夹中的文件树结构。
  • 只列示文件夹名称和文件名称,不需要显示文件内容。
  • 如果文件夹中有子文件夹,则显示子文件夹名称。
def print_file_tree(root, indent=""):
    for item in os.listdir(root):
        item_path = os.path.join(root, item)
        print(indent + "|-- " + item)
        if os.path.isdir(item_path):
            print_file_tree(item_path, indent + "    ")

print_file_tree(extract_folder)
|-- CSMAR常用变量-2000-2024
    |-- CSMAR常用变量-2000-2024.xlsx
    |-- CSMAR常用变量-2000-2024_DES.txt
    |-- 版权声明.pdf
|-- 上市公司基本信息变更表2000-2024
    |-- STK_LISTEDCOINFOCHG.xlsx
    |-- STK_LISTEDCOINFOCHG[DES][xlsx].txt
    |-- 版权声明.pdf
|-- 上市公司基本信息年度表
    |-- STK_LISTEDCOINFOANL.xlsx
    |-- STK_LISTEDCOINFOANL[DES][xlsx].txt
    |-- 上市公司基本信息 数据库说明书.pdf
|-- 利润表-现金流量表-2000-2010
    |-- 利润表-现金流量表-2000-2010.xlsx
    |-- 利润表-现金流量表-2000-2010_DES.txt
    |-- 版权声明.pdf
|-- 利润表-现金流量表-2011-2024
    |-- 利润表-现金流量表-2011-2024.xlsx
    |-- 利润表-现金流量表-2011-2024_DES.txt
    |-- 版权声明.pdf
|-- 资产负债表-2000-2010
    |-- 版权声明.pdf
    |-- 资产负债表-2000-2010.xlsx
    |-- 资产负债表-2000-2010_DES.txt
|-- 资产负债表-2011-2024
    |-- 版权声明.pdf
    |-- 资产负债表-2011-2024.xlsx
    |-- 资产负债表-2011-2024_DES.txt

15.6 定义函数以批量读取指定文件夹下的文件

15.6.1 文件内容分析

观察后发现,每个文件夹下都包含两个文件:

  • *.xlsx:包含数据的 Excel 文件。
  • *DES*.txt:包含数据描述的文本文件。

我们可以查看这两份文件的内容和结构,然后再决定读取方案。这里仍以 [资产负债表-2011-2024] 文件夹为例进行说明。

资产负债表-2011-2024.xlsx 是数据文件,存储结构如下:前 6 列和前 9 行数据如下:

可以看出,从第 5 行开始是具体的观察值,而此前的四行则是表头信息:

  • 第一行是变量名
  • 第二行是变量的中文简称 (可以用作变量中文标签)
  • 第三行是变量的单位
  • 第四行是变量的其他说明信息。如第 5-6 列显示的是报表类型。

资产负债表-2011-2024_DES.txt 是变量描述文件,主要包含变量的中文名称、单位、数据来源等信息。其原始内容如下:

-------------------- 资产负债表-2011-2024_DES.txt ---------begin--
本信息:
code[证券代码] - 
stknme[证券简称] - 
listingDate[上市日期] - 
EndDate[时间] - 

数据库名称:财务报表--> 表名称:资产负债表(FS_Combas)
A001101000[货币资金] - 公司库存现金、银行结算户存款、……等的合计数。1990年起使用
A001107000[交易性金融资产] - 交易性金融资产是……的债券投资、股票投资……。2007年起使用。
……
-------------------- 资产负债表-2011-2024_DES.txt ---------over--

可以看出,资产负债表-2011-2024_DES.txt 文件中每行的格式为:

varname[中文简称] - 变量说明

  • varname 是变量的英文名称。
  • 中文简称 是变量的中文名称,可以作为变量标签。
  • 变量说明 是对该变量的详细描述,包括变量的含义、计算方法、单位、数据来源等信息。

对比这两份文件,有如下几种处理方案:

方案 1:简化版

如果只需要 varname中文简称 (后者可以作为前者的变量标签),则处理过程为:

input资产负债表-2011-2024.xlsx

  • 提取 资产负债表-2011-2024.xlsx 文件中第 5 行以后的所有数据,定义为数据框 df
  • 提取 资产负债表-2011-2024.xlsx 前两行数据,定义为一个字典 dict - {varname: 中文简称}
  • 将该字典附加到数据框 df 中,作为它的一个属性。

outputdf 数据框 + dict 字典。

这种处理方式的好处是只需要数据文件 资产负债表-2011-2024.xlsx,缺陷是使用数据时,如果想知道每个变量详细信息,就需要翻阅 CSMAR 提供的 PDF 说明书,或 资产负债表-2011-2024_DES.txt 文件。

方案 2:完整信息版

此版本的思路是把变量的中文名和变量说明信息都提取出来,制作成两个字典,整合到数据框中。其好处是,我们可以随时在 Python 内部查看变量的中文名称和说明信息,而不需要翻阅 PDF 或其他文件。

处理过程如下:

input资产负债表-2011-2024.xlsx + 资产负债表-2011-2024_DES.txt

  • 提取 资产负债表-2011-2024.xlsx 文件中第 5 行以后的所有数据,定义为数据框 df
  • 资产负债表-2011-2024_DES.txt 文件中提取信息,定义两个字典:
    • 字典 1:{varname: 中文简称},用于将英文变量名与其中文简称对应起来。
    • 字典 2:{varname: 变量说明},用于将英文变量名与其详细说明对应起来。
  • 将这两个字典附加到数据框 df 中,作为它的两个属性。

outputdf 数据框 + dict1 字典 + dict2 字典。

方案 3:简化版-中文变量名

如果已经对变量名有了充分了解,且只需要 中文简称,则处理过程为:

input资产负债表-2011-2024.xlsx

  • 读入 资产负债表-2011-2024.xlsx 文件中第 2 行以后的所有数据,定义为数据框 df
  • 删除数据框 df 中的第 2-3 行;

上述方案的对比:

  • 方案 1:只保留英文变量名和中文简称,适用于对 CSMAR 数据库比较熟悉的用户。
  • 方案 2:保留英文变量名、中文简称和变量说明,适用于需要详细了解变量含义的用户。
  • 方案 3:只保留中文简称,变量含义直观明了。缺陷是,有些变量的中文简称中包含了 ( 等特殊字符,需要额外转换。因此,不太推荐这种方式。

下面,我们将实现方案 1 的处理过程。

15.6.2 方案 1:简化版处理代码

处理思路:

从 CSMAR 下载下来的每个 .zip 文件解压后,都会生成一个同名文件夹 (记为 'folder_path')。该文件夹下包含一个 .xlsx 格式的数据文件和一个 .txt 格式的描述文件。因此,我们只需要指定 'folder_path',便可以让 Python 根据文件后缀自动识别并读取这两个文件。对于 方案 1,我们只需要读取 .xlsx 文件,并将该文件的第二行作为变量 (第一行) 的中文标签即可。

提示词
  • 目的:读取指定文件夹下的 Excel 文件,并将其第二行作为变量的中文标签。

  • input:folder = ‘资产负债表-2011-2024’

    • folder_path = ‘{path}/data_raw/{folder}’
    • {path} 已经在第一个 cell (‘文件路径设定{tag}’) 中定义
  • 处理思路:

    • 检查该文件夹下以 .xlsx 结尾的文件的个数,如果大于 1 个,则报错。
    • 读入该文件夹下以 .xlsx 结尾的文件。完整文件名为:{filename}.xlsx
    • 存入数据框 df_filename
      • 打印 forder_path 和 df_filename 的名称
    • 删除 Excel 表格第 3 行和第 4 行
    • 做一个字典:dict_filename,键为英文变量名,值为中文变量名。
      • key: 英文变量名,从 Excel 第 1 行提取。
      • value: 中文变量名,从 Excel 第 2 行提取。
    • dict_filename 附加到 df_filename 数据框中,作为它的一个属性。
    • 删除第 2 行
  • 显示处理后的数据框 df_filename.head() 的前 5 列和字典 dict_filename 的全部 {key: value}。

import pandas as pd

# 构造文件夹路径
folder = '资产负债表-2011-2024'
folder_path = os.path.join(path, 'data_raw', folder)

# 检查 .xlsx 文件数量,排除临时文件(如~$开头的文件)
xlsx_files = [f for f in os.listdir(folder_path) if f.endswith('.xlsx') and not f.startswith('~$')]
if len(xlsx_files) != 1:
    raise ValueError(f"{folder_path} 下存在 {len(xlsx_files)} 个有效 .xlsx 文件,请检查!")
xlsx_file = xlsx_files[0]
xlsx_path = os.path.join(folder_path, xlsx_file)

print(f"folder_path: {folder_path}")
print(f"df_filename: {xlsx_file}")
print(' ')

# 读取前4行
header = pd.read_excel(xlsx_path, nrows=4, header=None)

# 删除第3、4行(索引2、3),再删除第2行(索引1),只保留第0行
header_clean = header.drop([2, 3]).reset_index(drop=True)
header_clean = header_clean.drop([1]).reset_index(drop=True)

# 提取英文变量名和中文变量名
en_names = header.iloc[0]
cn_names = header.iloc[1]
dict_filename = dict(zip(en_names, cn_names))

# 读取数据,跳过前4行
df_filename = pd.read_excel(xlsx_path, skiprows=4, header=None)
df_filename.columns = en_names

# 附加字典为属性
df_filename.varname_cn = dict_filename

# 显示前5列
print(df_filename.iloc[:, :5].head())
# 显示字典的前10个键值对
for k, v in list(dict_filename.items())[:10]:
    print(f"{k}: {v}")
folder_path: d:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2011-2024
df_filename: 资产负债表-2011-2024.xlsx
 
0  code  stknme listingDate  EndDate  FS_Combas-A001101000
0     1    平安银行  1991-04-03     2011          0.000000e+00
1     2     万科A  1991-01-29     2011          3.423951e+10
2     3  PT 金田A  1991-07-03     2011                   NaN
3     4  *ST 国华  1991-01-14     2011          5.712837e+07
4     5   ST 星源  1990-12-10     2011          1.629275e+07
code: 证券代码
stknme: 证券简称
listingDate: 上市日期
EndDate: 时间
FS_Combas-A001101000: 货币资金
FS_Combas-A001107000: 交易性金融资产
FS_Combas-A001109000: 短期投资净额
FS_Combas-A001123000: 存货净额
FS_Combas-A001100000: 流动资产合计
FS_Combas-A001212000: 固定资产净额

15.6.3 优化代码

上述代码可以一次性顺利执行,但耗时为 53s,太慢。于是,我把上述提示词和代码发给了 ChatGPT,请它优化代码。它给出的建议是:pandas 读取 .xlsx 格式本身就比较慢,尤其是数据量大或格式复杂时,速度瓶颈主要在于底层解析 Excel 文件的过程。此外,上述代码读取了两次 Excel 文件,第一次读取是为了获取变量名和中文简称,第二次读取是为了获取数据,这样会导致重复的 I/O 操作,进一步降低速度。

应对方法是,先使用 xlsx2csv 包将 .xlsx 文件快速转换为 .csv 格式,然后再用 pandasread_csv 读取数据,这样可以极大提升读取速度。

优化后的代码只需 1-2 秒即可完成 (ChatGPT 提示词):

提示词

目标:定义一个函数 read_files_in_folder(folder_path, Fname=None),用于批量读取指定文件夹下的 .xlsx 文件,并返回数据框和变量名-中文名字典。

要求:

  1. 参数说明:
    • folder_path:字符串,指定要读取的文件夹路径。
    • Fname:字符串,可选,指定数据框和字典的名称前缀。若为 None,则自动取 .xlsx 文件名(不含扩展名),并将特殊字符替换为下划线。
  2. 处理流程:
    • 检查文件夹下以 .xlsx 结尾且不以 ~$ 开头的文件,若数量不为 1,则报错。
    • 将 .xlsx 文件转换为 .csv 文件(如已存在则跳过),转换时编码为 gbk。
    • 读取 .csv 文件前 4 行,提取英文变量名和中文变量名。
    • 对英文变量名,若为 ‘A-B’ 结构,则只保留 B 部分。
    • 构建 {英文变量名: 中文变量名} 的字典。
    • 跳过前 4 行读取正文数据,列名用处理后的英文变量名。
    • 返回一个 dict,包含数据框(键名为 df_{Fname})和变量名字典(键名为 dic_{Fname})。
  3. 代码需包含必要的 import,且不重复导入已在 notebook 其他 cell 导入的模块。
  4. 代码块必须完整、可直接运行。
# {tag}: 'fcn_read_files_in_folder'
#%pip install xlsx2csv

import os
import pandas as pd
from xlsx2csv import Xlsx2csv

folder = '资产负债表-2011-2024'
folder_path = os.path.join(path, 'data_raw', folder)
# ========================================

# 检查文件夹下 .xlsx 文件
xlsx_files = [f for f in os.listdir(folder_path) if f.endswith('.xlsx') and not f.startswith('~$')]
if len(xlsx_files) != 1:
    raise ValueError(f"{folder_path} 下存在 {len(xlsx_files)} 个有效 .xlsx 文件,请检查!")
xlsx_file = xlsx_files[0]
xlsx_path = os.path.join(folder_path, xlsx_file)

print('\n', '-'*20, '文件存储路径和文件名:', '-'*20)
print(f"folder_path: {folder_path}")
print(f"df_filename: {xlsx_file}\n")

# 自动生成 CSV 文件名
csv_path = xlsx_path.replace('.xlsx', '.csv')
# 如已存在则跳过,否则转换
if not os.path.exists(csv_path):
    Xlsx2csv(xlsx_path, outputencoding="gbk").convert(csv_path)

# 读取前4行
header = pd.read_csv(csv_path, nrows=4, header=None)

# 提取变量名
en_names = header.iloc[0].tolist()
cn_names = header.iloc[1].tolist()
dict_filename = dict(zip(en_names, cn_names))

# 跳过前4行读取正文
df_filename = pd.read_csv(
    csv_path, skiprows=4, header=None, names=en_names, dtype=str
)

# 附加字典为属性
setattr(df_filename, 'varname_cn', dict_filename)

# 显示前5列
print(df_filename.iloc[:, :5].head())
print('\n\n', '-'*20, '变量名字典:', '-'*20)
for k, v in list(dict_filename.items())[:10]:
    print(f"{k}: {v}")
print('\n\n', '-'*20, '数据框维度:', '-'*20)
print(df_filename.shape)

 -------------------- 文件存储路径和文件名: --------------------
folder_path: D:\Github\ds_data\data\CSMAR\data_raw\资产负债表-2011-2024
df_filename: 资产负债表-2011-2024.xlsx

     code  stknme listingDate EndDate FS_Combas-A001101000
0  000001    平安银行  1991-04-03    2011                    0
1  000002     万科A  1991-01-29    2011   34239514295.080002
2  000003  PT 金田A  1991-07-03    2011                  NaN
3  000004  *ST 国华  1991-01-14    2011      57128374.050000
4  000005   ST 星源  1990-12-10    2011      16292748.160000


 -------------------- 变量名字典: --------------------
code: 证券代码
stknme: 证券简称
listingDate: 上市日期
EndDate: 时间
FS_Combas-A001101000: 货币资金
FS_Combas-A001107000: 交易性金融资产
FS_Combas-A001109000: 短期投资净额
FS_Combas-A001123000: 存货净额
FS_Combas-A001100000: 流动资产合计
FS_Combas-A001212000: 固定资产净额


 -------------------- 数据框维度: --------------------
(81662, 32)

15.7 定义函数以批量读取指定文件夹下的文件

几个要点:

  • 这个提示词还需修改:目前的提示词不具有独立性,应该与上一个 Cell 的提示词合并。
  • 变量名的处理:在 ‘利润表-现金流量表xxx.csv’ 和 ‘资产负债表xxx.csv’ 文件中,变量名的格式为 ‘A-B’,我们需要删除 ‘A-’ 部分,只保留 ‘C’ 部分。
    • FS_Combas-A001101000: 货币资金 变为 A001101000: 货币资金
    • FS_Comins-B003000000: 基本每股收益 变为 B003000000: 基本每股收益
    • FS_Comscfd-C001021000: 支付的各项税费 变为 C001021000: 支付的各项税费
提示词
  • 以 {tag}: ‘read_xlsx_example’ Cell 中的代码为基础,定义一个函数 read_files_in_folder(folder_path, Fname),用于批量读取指定文件夹下的所有文件。
    • 函数参数 folder_path 是一个字符串,表示要读取的文件夹路径。
    • Fname 是一个字符串,表示文件夹的名称,用于生成数据框和字典的名称。如果用户不指定,则默认使用 .xlsx 文件的名称,特殊字符采用 _ 替换,确保符合 Python 命名规则。
    • 如果变量名为 A-B 结构,则删除 ‘A-’ 部分,只保留 ‘B’ 部分。
  • 函数返回:一个数据框,名称为 df_{Fname};一个字典,用于存储 {变量名: 变量中文简称},名称为 dic_{Fname}
import os
import pandas as pd
import re
from xlsx2csv import Xlsx2csv

def read_files_in_folder(folder_path, Fname=None):
    """
    批量读取指定文件夹下的 .xlsx 文件,返回数据框和变量名-中文名字典。
    参数:
        folder_path: 文件夹路径
        Fname: 文件夹名称(可选),用于生成数据框和字典的名称。若为 None,则自动取 .xlsx 文件名(不含扩展名),并将特殊字符替换为下划线。
    返回:
        df_{Fname}: 数据框
        dic_{Fname}: {英文变量名: 中文变量名} 字典
    """
    # 查找 .xlsx 文件
    xlsx_files = [f for f in os.listdir(folder_path) if f.endswith('.xlsx') and not f.startswith('~$')]
    if len(xlsx_files) != 1:
        raise ValueError(f"{folder_path} 下存在 {len(xlsx_files)} 个有效 .xlsx 文件,请检查!")
    xlsx_file = xlsx_files[0]
    xlsx_path = os.path.join(folder_path, xlsx_file)

    # 自动生成 CSV 文件名
    csv_path = xlsx_path.replace('.xlsx', '.csv')
    if not os.path.exists(csv_path):
        Xlsx2csv(xlsx_path, outputencoding="gbk").convert(csv_path)  # 若使用 'utf-8' encoding,可能会导致中文乱码问题

    # 读取前4行
    header = pd.read_csv(csv_path, nrows=4, header=None)
    en_names = header.iloc[0].tolist()
    cn_names = header.iloc[1].tolist()

    # 变量名处理:如果为 'A-B' 结构,则只保留 B 部分
    def clean_varname(name):
        if isinstance(name, str) and '-' in name:
            parts = name.split('-')
            if len(parts) >= 2:
                return parts[-1]
        return name

    en_names_clean = [clean_varname(n) for n in en_names]
    dic = dict(zip(en_names_clean, cn_names))

    # 跳过前4行读取正文
    df = pd.read_csv(csv_path, skiprows=4, header=None, names=en_names_clean, dtype=str)

    # 生成 Fname
    if Fname is None:
        Fname = os.path.splitext(xlsx_file)[0]
    Fname = re.sub(r'\W+', '_', Fname)

    # 返回带有指定名称的数据框和字典
    return {f'df_{Fname}': df, f'dic_{Fname}': dic}

15.8 批量处理

接下来,我们就可以用上面定义的函数来批量处理 data_raw 文件夹中的所有子文件夹了。

提示词

用 {tag}: ‘fcn_read_files_in_folder’ Cell 中定义的 read_files_in_folder() 函数批量读取 data_raw 文件夹中的所有子文件夹下的文件,并将结果存储到 data_clean 文件夹中。主要步骤如下:

  • 如果 data_clean 文件夹不存在,则创建该文件夹;如果已经存在,则清空该文件夹中的所有内容。
  • 遍历 data_raw 文件夹中的所有子文件夹。
    • 对于每个子文件夹,调用 read_files_in_folder 函数,读取其中的 .xlsx 文件和 .txt 文件。
    • 将读取到的数据框和字典保存到 data_clean 文件夹中,文件名为 {子文件夹名称}.pkl
import shutil
import pickle

# 1. 如果 data_clean_folder 不存在则创建,已存在则清空
if os.path.exists(data_clean_folder):
    shutil.rmtree(data_clean_folder)
os.makedirs(data_clean_folder)

# 2. 遍历 data_raw 文件夹下所有子文件夹
for subfolder in os.listdir(extract_folder):
    subfolder_path = os.path.join(extract_folder, subfolder)
    if os.path.isdir(subfolder_path):
        try:
            # 读取数据和变量名字典
            result = read_files_in_folder(subfolder_path, Fname=subfolder)
            # 统一保存为 {'df': df, 'varname_cn': dic}
            df_key = [k for k in result if k.startswith('df_')][0]
            dic_key = [k for k in result if k.startswith('dic_')][0]
            save_dict = {'df': result[df_key], 'varname_cn': result[dic_key]}
            # 保存为 pkl 文件
            save_path = os.path.join(data_clean_folder, f"{subfolder}.pkl")
            with open(save_path, 'wb') as f:
                pickle.dump(save_dict, f)
            print(f"已保存: {save_path}")
        except Exception as e:
            print(f"处理 {subfolder} 时出错: {e}")
处理 CSMAR常用变量-2000-2024 时出错: 'utf-8' codec can't decode byte 0xb9 in position 457: invalid start byte
处理 上市公司基本信息变更表2000-2024 时出错: 'gbk' codec can't encode character '\xa0' in position 93: illegal multibyte sequence
处理 上市公司基本信息年度表 时出错: 'gbk' codec can't encode character '\ufeff' in position 188: illegal multibyte sequence
处理 利润表-现金流量表-2000-2010 时出错: 'utf-8' codec can't decode byte 0xb4 in position 728: invalid start byte
处理 利润表-现金流量表-2011-2024 时出错: 'utf-8' codec can't decode byte 0xb4 in position 728: invalid start byte
处理 资产负债表-2000-2010 时出错: 'utf-8' codec can't decode byte 0xb4 in position 624: invalid start byte
处理 资产负债表-2011-2024 时出错: 'utf-8' codec can't decode byte 0xb4 in position 624: invalid start byte

15.8.1 查看导入结果

提示词

列出 data_clean 文件夹中的所有文件,以及每个文件的对应字典的前五个 {key: value}

import os
import pickle

# 列出 data_clean 文件夹中每个文件的如下信息:
#   - 文件名
#   - 数据框的维度以及 '前五行+前五列'
#   - 字典的前五个 {key: value} 对

print("\n>>>> 当前 data_clean 文件夹中的文件:")
for item in os.listdir(data_clean_folder):
    if item.endswith('.pkl'):
        print('\n','-'*10, item, '-'*10)
        with open(os.path.join(data_clean_folder, item), 'rb') as f:
            data = pickle.load(f)
            if isinstance(data, dict) and 'df' in data and 'varname_cn' in data:
                df = data['df']
                var_dict = data['varname_cn']
                print(f"数据框 shape: {df.shape}")
                print("前五行前五列:")
                print(df.iloc[:5, :5])
                print("前五个变量名-中文名:")
                for k, v in list(var_dict.items())[:5]:
                    print(f"  {k}: {v}")

>>>> 当前 data_clean 文件夹中的文件:

 ---------- CSMAR常用变量-2000-2024.pkl ----------
数据框 shape: (61455, 33)
前五行前五列:
    Stkcd accper stknme AnaAttention       Audittyp
0  000001   2001   平安银行          NaN        标准无保留意见
1  000001   2002   平安银行     1.098612        标准无保留意见
2  000001   2003   平安银行     1.386294        标准无保留意见
3  000001   2004   平安银行     1.791759  带有解释性说明的无保留意见
4  000001   2005   平安银行     1.791759        标准无保留意见
前五个变量名-中文名:
  Stkcd: 股票代码
  accper: 会计年度
  stknme: 股票简称
  AnaAttention: 分析师关注度
  Audittyp: 审计意见

 ---------- 上市公司基本信息变更表2000-2024.pkl ----------
数据框 shape: (160275, 8)
前五行前五列:
   Symbol AnnouncementDate ImplementDate ChangedItem    SecurityID
0  000001              NaN    1991-04-03        所属省份  201000000001
1  000001              NaN    1991-04-03        注册地址  201000000001
2  000001              NaN    1991-04-03        公司全称  201000000001
3  000001              NaN    1991-04-03        证券简称  201000000001
4  000001              NaN    1991-04-03      公司经营性质  201000000001
前五个变量名-中文名:
  Symbol: 股票代码
  AnnouncementDate: 公告日期
  ImplementDate: 实施日期
  ChangedItem: 变更属性
  SecurityID: 证券ID

 ---------- 上市公司基本信息年度表.pkl ----------
数据框 shape: (64170, 40)
前五行前五列:
   Symbol ShortName     EndDate ListedCoID    SecurityID
0  000001      深发展A  2001-12-31     101704  201000000001
1  000001      深发展A  2002-12-31     101704  201000000001
2  000001      深发展A  2003-12-31     101704  201000000001
3  000001      深发展A  2004-12-31     101704  201000000001
4  000001      深发展A  2005-12-31     101704  201000000001
前五个变量名-中文名:
  Symbol: 股票代码
  ShortName: 股票简称
  EndDate: 统计截止日期
  ListedCoID: 上市公司ID
  SecurityID: 证券ID

 ---------- 利润表-现金流量表-2000-2010.pkl ----------
数据框 shape: (64163, 36)
前五行前五列:
     code  stknme listingDate EndDate         B001101000
0  000001    平安银行  1991-04-03    2000                NaN
1  000002     万科A  1991-01-29    2000  3783668674.180000
2  000003  PT 金田A  1991-07-03    2000   464723527.060000
3  000004  *ST 国华  1991-01-14    2000   131006632.910000
4  000005   ST 星源  1990-12-10    2000   145947499.350000
前五个变量名-中文名:
  code: 证券代码
  stknme: 证券简称
  listingDate: 上市日期
  EndDate: 时间
  B001101000: 营业收入

 ---------- 利润表-现金流量表-2011-2024.pkl ----------
数据框 shape: (81662, 36)
前五行前五列:
     code  stknme listingDate EndDate          B001101000
0  000001    平安银行  1991-04-03    2011                 NaN
1  000002     万科A  1991-01-29    2011  71782749800.679993
2  000003  PT 金田A  1991-07-03    2011                 NaN
3  000004  *ST 国华  1991-01-14    2011     74503718.530000
4  000005   ST 星源  1990-12-10    2011     63534839.010000
前五个变量名-中文名:
  code: 证券代码
  stknme: 证券简称
  listingDate: 上市日期
  EndDate: 时间
  B001101000: 营业收入

 ---------- 资产负债表-2000-2010.pkl ----------
数据框 shape: (64163, 32)
前五行前五列:
     code  stknme listingDate EndDate        A001101000
0  000001    平安银行  1991-04-03    2000               NaN
1  000002     万科A  1991-01-29    2000  995745160.050000
2  000003  PT 金田A  1991-07-03    2000   58018167.850000
3  000004  *ST 国华  1991-01-14    2000   64780229.730000
4  000005   ST 星源  1990-12-10    2000   29118049.740000
前五个变量名-中文名:
  code: 证券代码
  stknme: 证券简称
  listingDate: 上市日期
  EndDate: 时间
  A001101000: 货币资金

 ---------- 资产负债表-2011-2024.pkl ----------
数据框 shape: (81662, 32)
前五行前五列:
     code  stknme listingDate EndDate          A001101000
0  000001    平安银行  1991-04-03    2011                   0
1  000002     万科A  1991-01-29    2011  34239514295.080002
2  000003  PT 金田A  1991-07-03    2011                 NaN
3  000004  *ST 国华  1991-01-14    2011     57128374.050000
4  000005   ST 星源  1990-12-10    2011     16292748.160000
前五个变量名-中文名:
  code: 证券代码
  stknme: 证券简称
  listingDate: 上市日期
  EndDate: 时间
  A001101000: 货币资金

简要分析:

考虑到随后要将这些文件合并为一个数据框,我们需要重点关注如下几点:

  • 上市公司基本信息变更表2000-2024.pkl 暂时用不到,不予处理。
  • 哪些文件需要纵向合并 (append)?
    • 利润表-现金流量表-xxx.pkl资产负债表-xxx.pkl 这两类文件需要纵向合并,分别存入数据框 df_profitdf_asset 中。
  • 哪些文件需要横向合并 (merge)?
    • df_profitdf_asset 横向合并为一个数据框 df_financial
    • 进而与 CSMAR常用变量-2000-2024.pkl上市公司基本信息年度表.pkl 横向合并,得到最终的数据框 df_final
  • 横向合并时,需要保证两份数据中有相同的变量名 (keys)。由于我们要合并的数据都是 ‘firm-year’ 格式的面板数据,因此需要保证每个数据框中都有 ‘code’ 和 ‘year’ 这两个变量。
    • 我们需要为 CSMAR常用变量-2000-2024.pkl上市公司基本信息年度表.pkl 这两份文件添加 ‘code’ 和 ‘year’ 变量。二者分别对应这三个数据文件中的 SymbolEndDate 变量。
    • 对于 利润表-现金流量表-xxx.pkl资产负债表-xxx 文件,需要将 EndDate 变量转换为 year 变量,以便后续合并。

15.8.2 查看处理后的单个文件

提示词
  • 列示 ‘CSMAR常用变量-2000-2024.pkl’ 中的所有对象;
  • 空一行+分隔线;
  • 列示 ‘CSMAR常用变量-2000-2024.pkl’ 的 shape;
  • 空一行+分隔线;
  • 列示 ‘CSMAR常用变量-2000-2024.pkl’ 数据框中的 ‘前五行+前五列’;
  • 空一行+分隔线;
  • 列示字典中的所有 {变量名:中文简称}。
import pickle

# 载入数据
pkl_path = os.path.join(data_clean_folder, 'CSMAR常用变量-2000-2024.pkl')
with open(pkl_path, 'rb') as f:
    data = pickle.load(f)

# 列示所有对象
print("对象列表:", list(data.keys()))

print('\n' + '-'*40 + '\n')

# 列示 shape
df_csmar = data['df']
print("数据框 shape:", df_csmar.shape)

print('\n' + '-'*40 + '\n')

# 列示前五行前五列
print(df_csmar.iloc[:5, :5])

print('\n' + '-'*40 + '\n')

# 列示字典中的所有 {变量名:中文简称}
varname_cn_csmar = data['varname_cn']
for k, v in varname_cn_csmar.items():
    print(f"{k}: {v}")
对象列表: ['df', 'varname_cn']

----------------------------------------

数据框 shape: (61455, 33)

----------------------------------------

    Stkcd accper stknme AnaAttention       Audittyp
0  000001   2001   平安银行          NaN        标准无保留意见
1  000001   2002   平安银行     1.098612        标准无保留意见
2  000001   2003   平安银行     1.386294        标准无保留意见
3  000001   2004   平安银行     1.791759  带有解释性说明的无保留意见
4  000001   2005   平安银行     1.791759        标准无保留意见

----------------------------------------

Stkcd: 股票代码
accper: 会计年度
stknme: 股票简称
AnaAttention: 分析师关注度
Audittyp: 审计意见
InternationalBig4: 审计师是否来自国际四大
Ysmvosd: 年个股流通市值
Ysmvttl: 年个股总市值
Yretwd: 考虑现金红利再投资的年个股回报率
PropertyRightsNature: 产权性质
Seperation: 两权分离度
ActualControllerNatureID: 实际控制人性质编码
OwnershipProportion: 实际控制人拥有上市公司所有权比例
ControlProportion: 实际控制人拥有上市公司控制权比例
Shrcr1: 股权集中度1
Shrhfd5: 股权集中度9
Shrz: 股权集中度5
FundHoldProportion: 基金持股比例
QFIIHoldProportion: 合格境外投资者持股比例
BrokerHoldProportion: 券商持股比例
BankHoldProportion: 银行持股比例
NonFinanceHoldProportion: 非金融类上市公司持股比例
InsInvestorProp: 机构投资者持股比例
StaffNumber: 员工人数
ConcurrentPosition: 两职合一
Boardsize2: 董事会规模A
ExecutivesNumber: 高管人数
IndDirector: 独立董事人数
SumSalary: 管理层薪酬
TOP3SumSalary: 高管前三名薪酬总额
Ynshrtrd: 年个股交易股数
DirectorHoldshares: 董事会持股数量
ManageHoldshares: 高级管理人员持股数量

15.9 合并数据

15.9.1 纵向合并

提示词

== 纵向合并不同年度的同名文件

  • 纵向合并 data_clean 文件夹中所有以 ‘资产负债表’ 开头的文件。
    • EndDate 列重命名为 year 列。
    • 合并后的数据框命名为 df_assets,并保存到 data_clean 文件夹中,文件名为 df_assets.pkl
  • 纵向合并 data_clean 文件夹中所有以 ‘利润表’ 开头的文件。
    • EndDate 列重命名为 year 列。
    • 合并后的数据框命名为 df_profit,并保存到 data_clean 文件夹中,文件名为 df_profit.pkl
  • 上述合并完成后,打印合并后的数据框的 shape;展示 ‘前三行 + 前五列’。
import os
import pickle
import pandas as pd

# 合并资产负债表
assets_dfs = []
for item in os.listdir(data_clean_folder):
    if item.startswith('资产负债表') and item.endswith('.pkl'):
        with open(os.path.join(data_clean_folder, item), 'rb') as f:
            data = pickle.load(f)
            df = data['df'] if isinstance(data, dict) and 'df' in data else data
            if 'EndDate' in df.columns:
                df = df.rename(columns={'EndDate': 'year'})
            assets_dfs.append(df)
if assets_dfs:
    df_assets = pd.concat(assets_dfs, axis=0, ignore_index=True)
    with open(os.path.join(data_clean_folder, 'df_assets.pkl'), 'wb') as f:
        pickle.dump(df_assets, f)
    print("df_assets shape:", df_assets.shape)
    print(df_assets.iloc[:3, :5])

print('\n' + '-'*40 + '\n')

# 合并利润表
profit_dfs = []
for item in os.listdir(data_clean_folder):
    if item.startswith('利润表') and item.endswith('.pkl'):
        with open(os.path.join(data_clean_folder, item), 'rb') as f:
            data = pickle.load(f)
            df = data['df'] if isinstance(data, dict) and 'df' in data else data
            if 'EndDate' in df.columns:
                df = df.rename(columns={'EndDate': 'year'})
            profit_dfs.append(df)
if profit_dfs:
    df_profit = pd.concat(profit_dfs, axis=0, ignore_index=True)
    with open(os.path.join(data_clean_folder, 'df_profit.pkl'), 'wb') as f:
        pickle.dump(df_profit, f)
    print("df_profit shape:", df_profit.shape)
    print(df_profit.iloc[:3, :5])
df_assets shape: (145825, 32)
     code  stknme listingDate  year        A001101000
0  000001    平安银行  1991-04-03  2000               NaN
1  000002     万科A  1991-01-29  2000  995745160.050000
2  000003  PT 金田A  1991-07-03  2000   58018167.850000

----------------------------------------

df_profit shape: (145825, 36)
     code  stknme listingDate  year         B001101000
0  000001    平安银行  1991-04-03  2000                NaN
1  000002     万科A  1991-01-29  2000  3783668674.180000
2  000003  PT 金田A  1991-07-03  2000   464723527.060000

15.9.2 横向合并

提示词

== 横向合并不同来源的数据

  1. 一些出现两次以上的功能,可以预先定义函数,确保代码结构清晰
  2. 资产负债表与利润表的合并
  • 读入 data_clean 文件夹中的 df_assets.pkl
  • 按照关键词 ‘{code, year}’ 与 df_profit.pkl 数据框横向合并;
    • 如果有同名变量,则自动忽略。
  • 合并后的数据存入数据框 df_financial;
  • 数据框的前两列变量为 ‘code, year’
  1. 继续与 上市公司基本信息年度表, CSMAR常用变量-2000-2024.pkl 数据框横向合并。
  • 合并前,先将 上市公司基本信息年度表.pkl 中的 Symbol 列重命名为 codeEndDate 列重命名为 year
  • 按照关键词 ‘{code, year}’ 与 df_financial 数据框横向合并;
    • 如果有同名变量,则自动忽略。
  1. 输出合并后的数据文件
  • 合并后的数据存入数据框 df_final
    • 第一列为 ‘code’,第二列为 ‘year’,是两个独立的变量。
    • 保存到 data_clean 文件夹中,文件名为 df_final.pkl
  • 最终数据框 df_final 的前两列变量为 ‘code, year’,并展示其 shape 和 ‘前三行 + 前五列’。
  1. 合并数据字典。将上述文件的字典合并为一个大字典 dict_CSMAR
  • 同名变量的中文简称和说明信息取先进入的文件的。
  1. 打印数据字典尺寸,以及前十行信息,格式:{变量名: 中文简称};
  2. 每组打印结果之间空一行,添加标题文字和分割线
import os
import pickle
import pandas as pd

# 工具函数:按 code, year 横向合并,自动忽略同名变量
def merge_on_code_year(left, right, suffix='_right'):
    # 只保留右表中不与左表重复的列
    overlap = set(left.columns) & set(right.columns)
    overlap -= {'code', 'year'}
    right_use = right.drop(columns=list(overlap), errors='ignore')
    merged = pd.merge(left, right_use, on=['code', 'year'], how='left', suffixes=('', suffix))
    return merged

# 1. 资产负债表与利润表的合并
with open(os.path.join(data_clean_folder, 'df_assets.pkl'), 'rb') as f:
    df_assets = pickle.load(f)
with open(os.path.join(data_clean_folder, 'df_profit.pkl'), 'rb') as f:
    df_profit = pickle.load(f)

# 确保 'code' 和 'year' 列为字符串类型
df_assets['code'] = df_assets['code'].astype(str)
df_assets['year'] = df_assets['year'].astype(str)
df_profit['code'] = df_profit['code'].astype(str)
df_profit['year'] = df_profit['year'].astype(str)

df_financial = merge_on_code_year(df_assets, df_profit, suffix='_profit')

# 2. 继续与 “上市公司基本信息年度表” 和 “CSMAR常用变量-2000-2024.pkl” 合并
with open(os.path.join(data_clean_folder, '上市公司基本信息年度表.pkl'), 'rb') as f:
    data_basic = pickle.load(f)
df_basic = data_basic['df'] if isinstance(data_basic, dict) and 'df' in data_basic else data_basic
df_basic = df_basic.rename(columns={'Symbol': 'code', 'EndDate': 'year'})
df_basic['code'] = df_basic['code'].astype(str)
df_basic['year'] = df_basic['year'].astype(str)

with open(os.path.join(data_clean_folder, 'CSMAR常用变量-2000-2024.pkl'), 'rb') as f:
    data_csmar = pickle.load(f)
df_csmar = data_csmar['df'] if isinstance(data_csmar, dict) and 'df' in data_csmar else data_csmar
df_csmar['code'] = df_csmar['Stkcd'].astype(str)
df_csmar['year'] = df_csmar['accper'].astype(str)

df_financial = merge_on_code_year(df_financial, df_basic, suffix='_basic')
df_final = merge_on_code_year(df_financial, df_csmar, suffix='_csmar')

# 调整前两列顺序
cols = list(df_final.columns)
if 'code' in cols and 'year' in cols:
    cols.remove('code')
    cols.remove('year')
    df_final = df_final[['code', 'year'] + cols]

# 保存
with open(os.path.join(data_clean_folder, 'df_final.pkl'), 'wb') as f:
    pickle.dump(df_final, f)

print("==== 合并后数据框 df_final ====")
print("df_final shape:", df_final.shape)
print(df_final.iloc[:3, :5])
print('-'*40)

# 4. 合并数据字典
dicts = []
# 资产负债表
with open(os.path.join(data_clean_folder, '资产负债表-2011-2024.pkl'), 'rb') as f:
    d = pickle.load(f)
    if isinstance(d, dict) and 'varname_cn' in d:
        dicts.append(d['varname_cn'])
# 利润表
for fname in ['利润表-现金流量表-2000-2010.pkl', '利润表-现金流量表-2011-2024.pkl']:
    pkl_path = os.path.join(data_clean_folder, fname)
    if os.path.exists(pkl_path):
        with open(pkl_path, 'rb') as f:
            d = pickle.load(f)
            if isinstance(d, dict) and 'varname_cn' in d:
                dicts.append(d['varname_cn'])
# 基本信息
if isinstance(data_basic, dict) and 'varname_cn' in data_basic:
    dicts.append(data_basic['varname_cn'])
# CSMAR常用变量
if isinstance(data_csmar, dict) and 'varname_cn' in data_csmar:
    dicts.append(data_csmar['varname_cn'])

dict_CSMAR = {}
for d in dicts:
    dict_CSMAR.update(d)

print("==== 合并后数据字典 dict_CSMAR ====")
print(f"字典总变量数: {len(dict_CSMAR)}")
for i, (k, v) in enumerate(dict_CSMAR.items()):
    print(f"{k}: {v}")
    if i >= 200:
        break
print('-'*40)
---------------------------------------------------------------------------
FileNotFoundError                         Traceback (most recent call last)
Cell In[58], line 15
     12     return merged
     14 # 1. 资产负债表与利润表的合并
---> 15 with open(os.path.join(data_clean_folder, 'df_assets.pkl'), 'rb') as f:
     16     df_assets = pickle.load(f)
     17 with open(os.path.join(data_clean_folder, 'df_profit.pkl'), 'rb') as f:

File c:\ProgramData\anaconda3\Lib\site-packages\IPython\core\interactiveshell.py:324, in _modified_open(file, *args, **kwargs)
    317 if file in {0, 1, 2}:
    318     raise ValueError(
    319         f"IPython won't let you open fd={file} by default "
    320         "as it is likely to crash IPython. If you know what you are doing, "
    321         "you can use builtins' open."
    322     )
--> 324 return io_open(file, *args, **kwargs)

FileNotFoundError: [Errno 2] No such file or directory: 'D:\\Github\\ds_data\\data\\CSMAR\\data_clean\\df_assets.pkl'
import pandas as pd
import numpy as np

# 只对数值型变量做统计
numeric_cols = df_final.select_dtypes(include=[np.number]).columns

# 统计量
desc = df_final[numeric_cols].agg(['count', 'mean', 'std', 'min', 'max']).T
desc = desc.rename(columns={'count': 'N', 'mean': 'Mean', 'std': 'SD', 'min': 'Min', 'max': 'Max'})

# 缺失值统计
missing = df_final.isnull().sum()
missing = missing[missing > 0].sort_values(ascending=False)

print("==== 数值型变量基本统计量 (N, Mean, SD, Min, Max) ====")
display(desc)

print("\n==== 各变量缺失值数量 (只显示有缺失的变量) ====")
display(missing)
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
Cell In[57], line 8
      5 numeric_cols = df_final.select_dtypes(include=[np.number]).columns
      7 # 统计量
----> 8 desc = df_final[numeric_cols].agg(['count', 'mean', 'std', 'min', 'max']).T
      9 desc = desc.rename(columns={'count': 'N', 'mean': 'Mean', 'std': 'SD', 'min': 'Min', 'max': 'Max'})
     11 # 缺失值统计

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\frame.py:10149, in DataFrame.aggregate(self, func, axis, *args, **kwargs)
  10146 axis = self._get_axis_number(axis)
  10148 op = frame_apply(self, func=func, axis=axis, args=args, kwargs=kwargs)
> 10149 result = op.agg()
  10150 result = reconstruct_and_relabel_result(result, func, **kwargs)
  10151 return result

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\apply.py:928, in FrameApply.agg(self)
    926 result = None
    927 try:
--> 928     result = super().agg()
    929 finally:
    930     self.obj = obj

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\apply.py:193, in Apply.agg(self)
    190     return self.agg_dict_like()
    191 elif is_list_like(func):
    192     # we require a list, but not a 'str'
--> 193     return self.agg_list_like()
    195 if callable(func):
    196     f = com.get_cython_func(func)

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\apply.py:326, in Apply.agg_list_like(self)
    318 def agg_list_like(self) -> DataFrame | Series:
    319     """
    320     Compute aggregation in the case of a list-like argument.
    321 
   (...)
    324     Result of aggregation.
    325     """
--> 326     return self.agg_or_apply_list_like(op_name="agg")

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\apply.py:745, in NDFrameApply.agg_or_apply_list_like(self, op_name)
    742     raise NotImplementedError("axis other than 0 is not supported")
    744 keys, results = self.compute_list_like(op_name, obj, kwargs)
--> 745 result = self.wrap_results_list_like(keys, results)
    746 return result

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\apply.py:402, in Apply.wrap_results_list_like(self, keys, results)
    399 obj = self.obj
    401 try:
--> 402     return concat(results, keys=keys, axis=1, sort=False)
    403 except TypeError as err:
    404     # we are concatting non-NDFrame objects,
    405     # e.g. a list of scalars
    406     from pandas import Series

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\reshape\concat.py:382, in concat(objs, axis, join, ignore_index, keys, levels, names, verify_integrity, sort, copy)
    379 elif copy and using_copy_on_write():
    380     copy = False
--> 382 op = _Concatenator(
    383     objs,
    384     axis=axis,
    385     ignore_index=ignore_index,
    386     join=join,
    387     keys=keys,
    388     levels=levels,
    389     names=names,
    390     verify_integrity=verify_integrity,
    391     copy=copy,
    392     sort=sort,
    393 )
    395 return op.get_result()

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\reshape\concat.py:445, in _Concatenator.__init__(self, objs, axis, join, keys, levels, names, ignore_index, verify_integrity, copy, sort)
    442 self.verify_integrity = verify_integrity
    443 self.copy = copy
--> 445 objs, keys = self._clean_keys_and_objs(objs, keys)
    447 # figure out what our result ndim is going to be
    448 ndims = self._get_ndims(objs)

File c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\reshape\concat.py:507, in _Concatenator._clean_keys_and_objs(self, objs, keys)
    504     objs_list = list(objs)
    506 if len(objs_list) == 0:
--> 507     raise ValueError("No objects to concatenate")
    509 if keys is None:
    510     objs_list = list(com.not_none(*objs_list))

ValueError: No objects to concatenate
df_final.summary()
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
~\AppData\Local\Temp\ipykernel_53792\2207892642.py in ?()
----> 1 df_final.summary()

c:\ProgramData\anaconda3\Lib\site-packages\pandas\core\generic.py in ?(self, name)
   6295             and name not in self._accessors
   6296             and self._info_axis._can_hold_identifiers_and_holds_name(name)
   6297         ):
   6298             return self[name]
-> 6299         return object.__getattribute__(self, name)

AttributeError: 'DataFrame' object has no attribute 'summary'

15.9.3 输出最终数据文件

提示词

== 输出 csv 和 txt 文件到 [data_final] 文件夹

  1. df_final 数据框保存为 CSMAR_final.csv 文件,存储在 [data_final] 文件夹中。
  2. dict_CSMAR 字典保存为 CSMAR_var_label.txt 文件,存储在 [data_final] 文件夹中。每行格式为:变量名: 中文简称
import os

# 创建 data_final 文件夹(如果不存在)
data_final_folder = os.path.join(path, 'data_final')
if not os.path.exists(data_final_folder):
    os.makedirs(data_final_folder)

# 1. 保存 df_final 为 CSV
csv_path = os.path.join(data_final_folder, 'CSMAR_final.csv')
df_final.to_csv(csv_path, index=False, encoding='utf-8-sig')
print(f"已保存数据文件: {csv_path}")

# 2. 保存 dict_CSMAR 为 txt
txt_path = os.path.join(data_final_folder, 'CSMAR_var_label.txt')
with open(txt_path, 'w', encoding='utf-8') as f:
    for k, v in dict_CSMAR.items():
        f.write(f"{k}: {v}\n")
print(f"已保存变量标签文件: {txt_path}")
已保存数据文件: d:\Github\ds_data\data\CSMAR\data_final\CSMAR_final.csv
已保存变量标签文件: d:\Github\ds_data\data\CSMAR\data_final\CSMAR_var_label.txt

15.9.4 收尾:删除无用文件和过程文件

Note:这一步是可选的,主要是为了清理不必要的文件,以节省存储空间和提高数据处理效率。
建议:确认所有数据处理和合并工作完成后,再执行此步骤。

有些过程文件已经不需要了:

  • data_raw 文件夹中的 .xlsx 文件已经转换为 .csv 文件,因此可以删除。
  • data_clean 文件夹中的一些文件也不需要了,如 利润表-现金流量表-xxx.pkl资产负债表-xxx.pkl,因为它们已经被合并为 df_financial.pkl
提示词
  • 删除 data_raw 文件夹中所有子文件夹中的 .xlsx 文件。
  • 删除 data_clean 文件夹中如下文件:
    • 利润表-现金流量表- 开头的文件;
    • 资产负债表- 开头的文件;
import os

# 1. 删除 data_raw 文件夹中所有子文件夹中的 .xlsx 文件
for subfolder in os.listdir(extract_folder):
    subfolder_path = os.path.join(extract_folder, subfolder)
    if os.path.isdir(subfolder_path):
        for fname in os.listdir(subfolder_path):
            if fname.endswith('.xlsx'):
                file_path = os.path.join(subfolder_path, fname)
                os.remove(file_path)
                print(f"已删除: {file_path}")

# 2. 删除 data_clean 文件夹中指定前缀的文件
prefixes = ['利润表-现金流量表-', '资产负债表-']
for fname in os.listdir(data_clean_folder):
    if any(fname.startswith(prefix) for prefix in prefixes):
        file_path = os.path.join(data_clean_folder, fname)
        if os.path.isfile(file_path):
            os.remove(file_path)
            print(f"已删除: {file_path}")

15.9.5 呈现项目文档树

要点:

  • 代码执行后会自动将工作目录切换回项目根目录,确保后续操作路径一致。
提示词
  • 列示项目文档树结构。
  • 只列示文件夹名称和文件名称,不需要显示文件内容。
  • 如果文件夹中有子文件夹,则显示子文件夹名称。
  • 处理完后,将工作目录切换到项目根目录。
  • 为此代码块添加标题和合适的注释。
import os

# === 项目文档树结构展示 ===
# 本代码用于递归列示当前项目的文件夹和文件结构,便于快速了解项目目录布局。
# 最后将工作目录切换回项目根目录。

def print_project_tree(root, indent=""):
    for item in os.listdir(root):
        item_path = os.path.join(root, item)
        print(indent + "|-- " + item)
        if os.path.isdir(item_path):
            print_project_tree(item_path, indent + "    ")

print("=== 项目文档树结构 ===")
print_project_tree(path)

# 切换工作目录到项目根目录
os.chdir(path)
=== 项目文档树结构 ===
|-- 01_read_merge.ipynb
|-- 02_data_clean.ipynb
|-- CSMAR_API.md
|-- CSMAR_firm_basic_infor.ipynb
|-- data_clean
    |-- CSMAR常用变量-2000-2024.pkl
    |-- df_assets.pkl
    |-- df_final.pkl
    |-- df_profit.pkl
    |-- 上市公司基本信息变更表2000-2024.pkl
    |-- 上市公司基本信息年度表.pkl
    |-- 利润表-现金流量表-2000-2010.pkl
    |-- 利润表-现金流量表-2011-2024.pkl
    |-- 资产负债表-2000-2010.pkl
    |-- 资产负债表-2011-2024.pkl
|-- data_final
    |-- CSMAR_final.csv
    |-- CSMAR_var_label.txt
|-- data_raw
    |-- CSMAR常用变量-2000-2024
        |-- CSMAR常用变量-2000-2024.csv
        |-- CSMAR常用变量-2000-2024.xlsx
        |-- CSMAR常用变量-2000-2024_DES.txt
        |-- 版权声明.pdf
    |-- 上市公司基本信息变更表2000-2024
        |-- STK_LISTEDCOINFOCHG.csv
        |-- STK_LISTEDCOINFOCHG.xlsx
        |-- STK_LISTEDCOINFOCHG[DES][xlsx].txt
        |-- 版权声明.pdf
    |-- 上市公司基本信息年度表
        |-- STK_LISTEDCOINFOANL.csv
        |-- STK_LISTEDCOINFOANL.xlsx
        |-- STK_LISTEDCOINFOANL[DES][xlsx].txt
        |-- 上市公司基本信息 数据库说明书.pdf
    |-- 利润表-现金流量表-2000-2010
        |-- 利润表-现金流量表-2000-2010.csv
        |-- 利润表-现金流量表-2000-2010.xlsx
        |-- 利润表-现金流量表-2000-2010_DES.txt
        |-- 版权声明.pdf
    |-- 利润表-现金流量表-2011-2024
        |-- 利润表-现金流量表-2011-2024.csv
        |-- 利润表-现金流量表-2011-2024.xlsx
        |-- 利润表-现金流量表-2011-2024_DES.txt
        |-- 版权声明.pdf
    |-- 资产负债表-2000-2010
        |-- 版权声明.pdf
        |-- 资产负债表-2000-2010.csv
        |-- 资产负债表-2000-2010.xlsx
        |-- 资产负债表-2000-2010_DES.txt
    |-- 资产负债表-2011-2024
        |-- 版权声明.pdf
        |-- 资产负债表-2011-2024.csv
        |-- 资产负债表-2011-2024.xlsx
        |-- 资产负债表-2011-2024_DES.txt
|-- data_raw_zip
    |-- CSMAR常用变量-2000-2024.zip
    |-- 上市公司基本信息变更表2000-2024.zip
    |-- 上市公司基本信息年度表.zip
    |-- 利润表-现金流量表-2000-2010.zip
    |-- 利润表-现金流量表-2011-2024.zip
    |-- 资产负债表-2000-2010.zip
    |-- 资产负债表-2011-2024.zip
|-- functions
    |-- CSMAR_var_label.py

15.9.6 保留哪些文件夹?

在本地运行完毕后,项目根目录下会包含如下文件夹:

  • [1] data_raw_zip 文件夹:务必保留,里面存放了从 CSMAR 下载的原始数据压缩包。
  • [2] data_raw 文件夹:可以删除,里面存放了解压后的 .xlsx 和 .txt 文档
  • [3] data_clean 文件夹:酌情删除,里面存放了清洗后的 .pkl 文件。
  • [4] data_final 文件夹:建议保留,里面存放了最终的 .csv.txt 文件。

可复现文档分享方案:

一旦确定上述代码无误,在分享时,只需保留 01_read_merge.ipynb 文件和 [data_raw_zip] 文件夹即可。

  • 完整复现文档:
    • 01_read_merge.ipynb 文件:包含了从数据读取到合并的完整代码。
    • [data_raw_zip] 文件夹:包含了从 CSMAR 下载的原始数据压缩包。
  • 直接使用最终文档 + 了解处理过程:
    • data_final 文件夹:包含了最终的 .csv.txt 文件,便于直接使用。
    • 01_read_merge.ipynb 文件:包含了从数据读取到合并的完整代码。
## 下一步?

接下来,我们会编写:

-  `02_data_clean.ipynb` 文件,对 `data_final` 文件夹中的数据做进一步处理,包括:缺失值、文字变量转换为数值变量、生成新的变量、处理离群值等。
-  `03_data_analysis.ipynb` 文件,对 `data_final` 文件夹中的数据进行分析,包括:描述性统计、相关性分析、回归分析等。