📚 课程目录第2章数据源接入与清洗

数据源接入与清洗：从原始数据到可用资产

审计报表自动化，第一步不是写代码，而是搞定数据源。我见过太多团队，算法写得漂亮，结果数据源一换，全盘崩溃。说白了，数据源接入与清洗，是整个自动化流程的「地基」。

这一章，我们就来聊聊常见审计数据源怎么接、数据清洗的核心原则是什么、以及如何用 Python Pandas 快速上手。嗯，都是我在项目里踩过坑之后总结出来的经验。

常见审计数据源：三种主流接入方式

审计数据源，说白了就三大类：数据库、Excel 文件、API 接口。每一类都有它的脾气，我们得对症下药。

1. 数据库接入

数据库是审计数据的「大本营」。企业 ERP、财务系统、业务系统，数据基本都躺在数据库里。我个人习惯用 pymysql 或 sqlalchemy 来连接 MySQL，用 psycopg2 连 PostgreSQL。

举个例子，从 MySQL 拉取某月交易流水：

import pandas as pd
from sqlalchemy import create_engine

# 创建数据库连接
engine = create_engine('mysql+pymysql://user:password@host:3306/audit_db')

# 执行查询，直接返回 DataFrame
df = pd.read_sql("SELECT * FROM transaction_log WHERE month = '2024-03'", engine)

# 关闭连接
engine.dispose()

我的经验： 别一次性拉全量数据。审计数据动辄上千万行，全量拉取会把数据库拖垮。我习惯先做「采样查询」——比如 LIMIT 1000 看看字段结构，确认无误后再按时间分片拉取。

2. Excel 文件接入

Excel 是审计人员的老朋友。但 Excel 文件有个问题——格式不统一。有的表头在第一行，有的在第三行；有的合并单元格，有的带空行。

Pandas 读取 Excel 很简单：

import pandas as pd

# 读取 Excel 文件
df = pd.read_excel('审计数据_2024Q1.xlsx', sheet_name='交易明细', header=0)

# 如果表头不在第一行，可以指定行号
# df = pd.read_excel('审计数据_2024Q1.xlsx', header=2)

注意： Excel 文件里的合并单元格，Pandas 读取后会出现 NaN 值。我曾经因为这个原因，漏掉了整整一个月的部门数据。建议在读取前，先让业务方把合并单元格取消掉。

3. API 接口接入

现在很多 SaaS 系统都提供 API 接口。比如钉钉审批流、企业打卡记录、第三方支付平台的对账单。API 返回的数据通常是 JSON 格式，需要做一层「展平」处理。

import requests
import pandas as pd

# 调用 API 获取数据
url = "https://api.audit-system.com/v1/expense-reports"
headers = {"Authorization": "Bearer your_token_here"}
response = requests.get(url, headers=headers)

# 解析 JSON 并转为 DataFrame
data = response.json()
df = pd.json_normalize(data['items'])

print(df.head())

避坑指南： API 接口通常有频率限制和分页。我曾经没处理分页，结果只拿到了前 100 条数据，审计报告直接少报了 80% 的金额。记得写循环，把每一页的数据都拉下来。

数据清洗原则：脏数据是审计的「天敌」

数据源接进来了，但数据本身可能「脏」得不行。空值、重复值、异常值、格式不一致……这些在审计场景里都是大忌。我总结了几条核心原则，你可以直接拿来用。

原则	说明	审计场景举例
完整性	缺失值必须处理，不能留空	交易金额为空，无法计算汇总
唯一性	去除重复记录，保留唯一标识	同一笔报销单出现两次，导致重复支付
一致性	统一数据格式和单位	日期格式有的 yyyy-mm-dd，有的 dd/mm/yyyy
准确性	识别并修正异常值	金额出现负数，或超出合理范围
时效性	数据时间范围必须正确	审计期间外的数据混入，影响统计

你想想看，如果数据清洗没做好，后面生成的报表再漂亮，也是「垃圾进，垃圾出」。审计报告一旦出错，后果很严重。

Python Pandas 基础操作：清洗三板斧

Pandas 是数据清洗的「瑞士军刀」。我平时用得最多的，就是下面这三个操作。掌握了它们，80% 的清洗场景都能搞定。

第一板斧：处理缺失值

缺失值在审计数据里很常见。比如某笔交易的「审批人」字段为空。处理方式有两种：删除或填充。

import pandas as pd

# 查看缺失值情况
print(df.isnull().sum())

# 方式一：删除包含缺失值的行（慎用）
df_clean = df.dropna(subset=['交易金额', '交易日期'])

# 方式二：填充缺失值（推荐）
df['审批人'] = df['审批人'].fillna('未指定')
df['交易金额'] = df['交易金额'].fillna(0)

注意： 审计场景下，不要轻易删除数据。缺失值本身可能就是审计线索——为什么这笔交易没有审批人？我建议先填充占位符，然后在报表里标记出来，留给审计人员判断。

第二板斧：去除重复值

重复数据会导致统计结果虚高。比如同一张发票被录入两次，报销总额就翻倍了。

# 查看重复行
print(df.duplicated().sum())

# 基于指定列去重（保留第一条）
df_unique = df.drop_duplicates(subset=['发票号', '交易日期'])

# 查看去重后的数据量
print(f"去重前: {len(df)} 行, 去重后: {len(df_unique)} 行")

我的经验： 去重时，一定要指定 subset 参数。如果不指定，Pandas 会基于所有列判断重复，但审计数据里有些字段（比如备注）可能不同，导致重复数据没被识别出来。

第三板斧：数据类型转换与格式统一

数据类型不对，后面没法做计算。比如「交易金额」被读成了字符串，求和就会报错。日期格式不统一，也没法按时间做筛选。

# 转换数据类型
df['交易金额'] = pd.to_numeric(df['交易金额'], errors='coerce')
df['交易日期'] = pd.to_datetime(df['交易日期'], format='%Y-%m-%d', errors='coerce')

# 统一字符串格式（去除空格、转大写）
df['部门'] = df['部门'].str.strip().str.upper()

# 筛选指定时间范围的数据
df_filtered = df[(df['交易日期'] >= '2024-01-01') & (df['交易日期'] <= '2024-03-31')]

  核心要点： 数据清洗不是一次性工作。我建议把清洗步骤写成函数，每次接入新数据源时，直接调用。这样既保证一致性，又节省时间。

本章知识体系：数据源接入与清洗全流程

下面这张图，是我自己画的数据源接入与清洗的完整流程。你可以把它当作操作手册来用。

这张图把整个流程串起来了。从数据源接入，到清洗处理，再到输出可用数据。每一步都有对应的 Pandas 操作。你可以在实际项目中，照着这个流程一步步走。

小结：数据清洗是审计自动化的「第一道关」

数据源接入与清洗，听起来基础，但做扎实了，后面的工作会顺畅很多。我个人习惯在项目初期，花 30% 的时间在数据清洗上。别嫌多，这 30% 的时间，能省掉后面 70% 的返工。