我见过太多企业的报销审核流程财务坐在那里像“大家来找茬”一样盯着一张张电子发票寻找“连号发票”、“节假日消费”、“超标准住宿”等违规痕迹。今天我们将利用智能会计中的“内部控制”与“实质性测试”理念用 Python SQL 构建一套报销异常票据批量筛查系统实现从“人工肉眼扫描”到“机器毫秒级狙击”的跨越。一、 实际应用场景描述场景某集团公司每月处理上千份员工报销单差旅费、招待费、交通费等。审计或财务经理突然发话“查一下最近有没有人在周末公款吃喝有没有连号出租车票招待费有没有超标”老做法1. 财务导出 Excel。2. 筛选“日期”列肉眼找周六周日。3. 筛选“发票号码”列排序找连号。4. 按部门/职级筛选对照制度看金额是否超标。结果 耗时3小时眼睛酸痛还不一定查得全。二、 引入痛点 (The Pain Points)我们要颠覆的“逐单翻看”模式存在三大死穴1. 效率极低千张票据人工逐条核对耗时以“小时”计。2. 漏网之鱼人眼容易疲劳连号发票如出租车票 1001, 1002, 1003很难被发现。3. 标准不一A 会计觉得 500 元餐费超标B 会计觉得能过缺乏刚性规则。三、 核心逻辑讲解 (The Algorithm)我们将采用“规则引擎 多维过滤” (Rule‑Based Filtering) 模型。核心思想将财务制度转化为 SQL WHERE 子句。把“违规”定义为满足特定布尔条件Boolean Condition的数据集合。筛查维度Rules1. 时间异常Weekday IN (Saturday, Sunday) 且费用类型非出差补贴。2. 连号异常Invoice_Number 与前后行差值 1。3. 金额超标Amount Policy_Limit (根据职级/城市映射)。4. 无票异常Invoice_Code IS NULL OR Invoice_Code 。逻辑流程1. 从数据库/CSV 读取报销明细。2. 数据预处理解析发票号、日期。3. 依次应用“规则过滤器”。4. 汇总所有异常生成 “整改清单” (Action Item List)。四、 代码模块化实现 (Python SQL Logic)虽然我们用 Python 实现但核心逻辑完全复刻 SQL 的SELECT ... WHERE 思维。1. 项目结构expense_audit_system/├── main.py├── modules/│ ├── __init__.py│ ├── data_loader.py│ ├── rule_engine.py│ └── reporter.py├── sample_expenses.csv└── README.md2. 核心代码modules/data_loader.py模块功能加载报销数据模拟从SQL数据库或CSV文件读取数据import pandas as pddef load_expense_data(source: str) - pd.DataFrame:加载报销明细数据参数:source (str): CSV文件路径或数据库连接字符串(此处简化为文件路径)返回:pd.DataFrame: 报销数据DataFrametry:df pd.read_csv(source)# 关键数据清洗df[Expense_Date] pd.to_datetime(df[Expense_Date])df[Amount] df[Amount].astype(float)df[Invoice_Number] df[Invoice_Number].astype(str)# 添加辅助列星期几df[Weekday] df[Expense_Date].dt.day_name()print(f✅ 成功加载 {len(df)} 条报销记录)return dfexcept FileNotFoundError:raise FileNotFoundError(f错误未找到数据源文件 {source})modules/rule_engine.py模块功能报销异常规则引擎这里是整个系统的核心所有的财务制度都在这里转化为代码import pandas as pdimport numpy as npclass ExpenseAuditEngine:报销审计引擎类def __init__(self, policy_config: dict None):初始化审计规则模拟公司财务制度self.policy policy_config or {max_meal_per_day: 150.0, # 每日餐标上限max_hotel_night: 400.0, # 每晚住宿上限allowed_weekend_types: [Travel_Allowance] # 周末允许的费用类型}print(⚖️ 审计规则引擎已加载)def check_missing_invoice(self, df: pd.DataFrame) - pd.DataFrame:规则1检查无票报销condition df[Invoice_Number].isnull() | (df[Invoice_Number] )return df.loc[condition].copy()def check_weekend_expense(self, df: pd.DataFrame) - pd.DataFrame:规则2检查周末非合规费用weekend_days [Saturday, Sunday]condition ((df[Weekday].isin(weekend_days)) (~df[Expense_Type].isin(self.policy[allowed_weekend_types])))return df.loc[condition].copy()def check_consecutive_invoices(self, df: pd.DataFrame) - pd.DataFrame:规则3检查连号发票 (如出租车票)逻辑对同一员工、同类型的发票排序检查号码是否连续anomalies []# 只检查有发票号的票据valid_df df.dropna(subset[Invoice_Number]).copy()# 尝试将发票号转为数字处理纯数字发票号valid_df[Invoice_Num_Int] pd.to_numeric(valid_df[Invoice_Number], errorscoerce)# 按员工、费用类型、发票号排序sorted_df valid_df.sort_values(by[Employee_ID, Expense_Type, Invoice_Num_Int])# 计算相邻行的差值sorted_df[Num_Diff] sorted_df.groupby(Employee_ID)[Invoice_Num_Int].diff()# 差值为1即为连号consecutive_mask sorted_df[Num_Diff] 1anomalies sorted_df.loc[consecutive_mask]return anomalies.copy()def check_policy_overlimit(self, df: pd.DataFrame) - pd.DataFrame:规则4检查超标费用condition False# 餐费超标meal_over ((df[Expense_Type] Meal) (df[Amount] self.policy[max_meal_per_day]))# 住宿超标hotel_over ((df[Expense_Type] Hotel) (df[Amount] self.policy[max_hotel_night]))condition meal_over | hotel_overreturn df.loc[condition].copy()def run_all_checks(self, df: pd.DataFrame) - dict:运行所有审计规则并返回结果results {missing_invoice: self.check_missing_invoice(df),weekend_expense: self.check_weekend_expense(df),consecutive_invoices: self.check_consecutive_invoices(df),policy_overlimit: self.check_policy_overlimit(df)}return resultsmain.py主执行程序报销异常票据批量筛查系统from modules.data_loader import load_expense_datafrom modules.rule_engine import ExpenseAuditEnginefrom modules.reporter import generate_audit_reportdef main():print( 启动报销异常批量筛查系统...\n)DATA_SOURCE sample_expenses.csv# 1. 加载数据expense_df load_expense_data(DATA_SOURCE)# 2. 初始化审计引擎audit_engine ExpenseAuditEngine()# 3. 执行筛查audit_results audit_engine.run_all_checks(expense_df)# 4. 生成报告generate_audit_report(audit_results)print(\n✅ 审计筛查完成请查看生成的整改清单。)if __name__ __main__:main()modules/reporter.py模块功能审计报告生成import pandas as pddef generate_audit_report(results: dict):生成综合审计报告print(\n *60)print( 报销异常整改清单 (Action Items))print(*60)for rule_name, result_df in results.items():if not result_df.empty:print(f\n [异常类型]: {rule_name.replace(_, ).title()})print(f 发现 {len(result_df)} 条异常记录:)# 打印关键列print(result_df[[Employee_ID, Expense_Type, Amount, Expense_Date, Invoice_Number]].head())# 导出到Excelwith pd.ExcelWriter(expense_audit_report.xlsx) as writer:for sheet_name, df in results.items():if not df.empty:df.to_excel(writer, sheet_namesheet_name[:31], indexFalse)print(\n 详细报告已导出至 expense_audit_report.xlsx)3. 示例数据sample_expenses.csvExpense_ID,Employee_ID,Expense_Type,Amount,Expense_Date,Invoice_NumberEXP001,E1001,Taxi,45.00,2026-04-10,1001EXP002,E1001,Taxi,35.00,2026-04-10,1002EXP003,E1002,Meal,280.00,2026-04-12,INV-MEAL-99EXP004,E1003,Hotel,680.00,2026-04-13,HTL-20260413EXP005,E1002,Meal,180.00,2026-04-11,EXP006,E1001,Meal,200.00,2026-04-12,INV-MEAL-101五、 README 文件与使用说明Expense-Audit-System简介本系统通过 Python 实现报销单据的自动化异常筛查支持无票检查、连号发票检测、超标核查等功能替代低效的人工逐单翻阅模式。使用前准备1. 安装 Python 3.82. 安装依赖pip install pandas openpyxl3. 准备sample_expenses.csv字段需包含Employee_ID, Expense_Type, Amount, Expense_Date, Invoice_Number。运行方式在终端执行python main.py输出结果程序将生成expense_audit_report.xlsx每个 Sheet 对应一种异常类型。六、 核心知识点卡片 (Knowledge Cards)知识点 说明规则引擎 (Rule Engine) 将模糊的财务制度转化为刚性的if/where 条件判断。Pandas 向量化运算 替代for 循环实现对千行数据的毫秒级过滤 (df.loc[condition])。差分运算 (Diff) 通过groupby().diff() 实现连号票据的高效识别。内部控制 (Internal Control) 程序本身就是一种“预防性控制”在报销入账前拦截风险。七、 总结作为全栈工程师我一直坚信“好的系统应该让违规者无处遁形让守规者畅通无阻。”这套报销异常筛查系统的价值在于1. 秒级响应几千张票据几秒钟出结果效率提升百倍。2. 零容忍标准机器不讲情面没有“下不为例”严格执行财务制度。3. 审计留痕所有的异常都被记录在案形成完整的整改清单。利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛