存货收发存核算（先进先出法）- Pandas 实现

一、生成测试 Excel 数据（两年数据，每月约 5 万行）

核心思路

def generate_inventory_excel(target_dir, years=2):
"""
生成存货收发存测试Excel数据
:param target_dir: 目标保存目录
:param years: 生成数据的年数
"""
# 1. 定义基础配置数据
material_codes = [f'MAT{str(i).zfill(6)}' for i in range(1, 200)] # 200种物料
warehouse_codes = [f'WH{str(i).zfill(3)}' for i in range(1, 50)] # 50个仓库
business_types = [
'采购入库', '销售出库',
'品种调整入库', '品种调整出库',
'调拨入库', '调拨出库',
'盘盈', '盘亏'
]
# 业务类型对应收发标识（入库=1，出库=-1，用于后续数量计算）
biz_type_flag = {
'采购入库': 1, '销售出库': -1,
'品种调整入库': 1, '品种调整出库': -1,
'调拨入库': 1, '调拨出库': -1,
'盘盈': 1, '盘亏': -1
}

# 2. 计算时间范围（从当前日期往前推2年，按自然月拆分）
end_date = datetime.now().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
start_date = (end_date – timedelta(days=365*years)).replace(day=1)

# 3. 逐月生成数据
current_month = start_date
while current_month < end_date:
# 获取当前月份的起始和结束日期
next_month = (current_month + timedelta(days=32)).replace(day=1)
month_days = (next_month – timedelta(days=1)).day
month_str = current_month.strftime('%Y%m')

# 4. 生成当月数据（约5万行）
row_count = 50000
data = {
'单据日期': [current_month.replace(day=random.randint(1, month_days)) for _ in range(row_count)],
'物料编码': [random.choice(material_codes) for _ in range(row_count)],
'仓库编码': [random.choice(warehouse_codes) for _ in range(row_count)],
'业务类型': [random.choice(business_types) for _ in range(row_count)],
'数量': np.random.randint(1, 100, size=row_count), # 基础数量（1-99）
'单价': np.round(np.random.uniform(10, 1000, size=row_count), 2), # 基础单价（10-1000，保留2位小数）
'批次号': [f'BAT{current_month.strftime("%Y%m")}{str(i).zfill(6)}' for i in range(row_count)] # 按月份生成批次号
}

# 5. 处理出入库数量（出库转为负数）
df = pd.DataFrame(data)
df['数量'] = df.apply(lambda x: x['数量'] * biz_type_flag[x['业务类型']], axis=1)

# 6. 计算金额（数量*单价，保留2位小数）
df['金额'] = np.round(df['数量'] * df['单价'], 2)

# 7. 排序（按单据日期、物料编码、批次号）
df = df.sort_values(by=['单据日期', '物料编码', '批次号']).reset_index(drop=True)

# 8. 保存为Excel文件（按月份命名，便于月度滚动）
excel_path = os.path.join(target_dir, f'存货收发存流水_{month_str}.xlsx')
df.to_excel(excel_path, index=False, engine='openpyxl')
print(f'已生成当月数据：{excel_path}')

# 9. 切换到下一个月
current_month = next_month

# 调用函数生成2年测试数据
generate_inventory_excel(target_dir, years=2)

二、先进先出法核算实现（支持月度滚动计算）

先进先出法（FIFO）核算

先进先出法（FIFO）的核心是按入库时间先后顺序匹配出库，优先扣减最早入库批次的库存，滚动计算的核心是「上月期末有效结存批次（带数量 / 金额 / 单价）= 本月期初可用批次」，关键要点：
核算维度：按「物料编码 + 仓库编码」分组，每组内单独执行 FIFO 逻辑（不同物料 / 仓库互不干扰）。
入库处理：直接追加批次信息（数量 / 金额 / 单价）到当前结存列表，按入库时间排序。
出库处理：按批次入库先后顺序逐批扣减，先扣完最早批次，再扣次早批次，直至满足出库数量。
期末结存：剩余未扣减的批次构成月末结存，作为下月期初数据，保持批次的时间顺序。
核心公式：
批次单价 = 批次金额 / 批次数量
发出金额 = 扣减数量 × 对应批次单价
期末结存金额 = 各剩余批次金额之和
期末结存数量 = 各剩余批次数量之和

import os
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# ===================== 核心配置项（与之前流水数据保持一致） =====================
# 1. 数据目录（必须和生成流水的目录一致，否则无法读取文件）
target_dir = r'E:\\BI模型\\Python练习数据\\收发存练习'
# 2. 核算月份范围（格式：YYYYMM，对应之前生成的2年数据，示例：202401-202512）
start_month = '202401'
end_month = '202512'
# 3. 财务数据精度（保留2位小数，符合会计核算要求）
decimal_places = 2
# ==============================================================================

def ensure_directory_exists(dir_path):
"""确保目标目录存在，不存在则自动创建"""
if not os.path.exists(dir_path):
os.makedirs(dir_path, exist_ok=True)

def generate_month_list(start_month, end_month):
"""生成起始月份到结束月份的连续月份列表，用于批量滚动核算"""
# 转换为日期对象
start_date = datetime.strptime(start_month, '%Y%m')
end_date = datetime.strptime(end_month, '%Y%m')
month_list = []

current_date = start_date
while current_date <= end_date:
month_list.append(current_date.strftime('%Y%m'))
# 切换到下一个月（避免月末天数问题，用+32天再取1号）
current_date = (current_date + timedelta(days=32)).replace(day=1)

return month_list

def fifo_costing_single(month_str, data_dir):
"""
单月先进先出法（FIFO）单独核算（核心函数）
:param month_str: 待核算月份，格式YYYYMM
:param data_dir: 流水数据与结果保存目录
:return: 当月FIFO收发存汇总表、月末有效结存表（用于下月滚动）
"""
# 1. 校验当月流水文件是否存在
monthly_flow_file = os.path.join(data_dir, f'存货收发存流水_{month_str}.xlsx')
if not os.path.exists(monthly_flow_file):
raise FileNotFoundError(f"当月收发存流水文件缺失，无法核算：{monthly_flow_file}")

# 2. 读取当月流水数据，数据清洗与预处理
df_flow = pd.read_excel(monthly_flow_file, engine='openpyxl')
# 数据类型转换与空值填充（避免计算异常）
df_flow['单据日期'] = pd.to_datetime(df_flow['单据日期'], errors='coerce')
df_flow['数量'] = pd.to_numeric(df_flow['数量'], errors='coerce').fillna(0)
df_flow['金额'] = pd.to_numeric(df_flow['金额'], errors='coerce').fillna(0)
# 确保核心关键字段非空，删除无效记录
df_flow = df_flow.dropna(subset=['物料编码', '仓库编码', '批次号', '单据日期']).reset_index(drop=True)
# 按物料、仓库、单据日期排序（保证入库顺序正确，FIFO核心前提）
df_flow = df_flow.sort_values(by=['物料编码', '仓库编码', '单据日期', '批次号']).reset_index(drop=True)

# 3. 读取上月FIFO结存数据（实现月度滚动：上月期末 = 本月期初）
last_month_date = datetime.strptime(month_str, '%Y%m') – timedelta(days=1)
last_month_str = last_month_date.strftime('%Y%m')
last_month_balance_file = os.path.join(data_dir, f'FIFO_存货结存_{last_month_str}.xlsx')

if os.path.exists(last_month_balance_file):
# 上月有结存数据，读取并作为本月期初批次
df_begin_balance = pd.read_excel(last_month_balance_file, engine='openpyxl')
# 数据类型校验
df_begin_balance['结存数量'] = pd.to_numeric(df_begin_balance['结存数量'], errors='coerce').fillna(0)
df_begin_balance['结存金额'] = pd.to_numeric(df_begin_balance['结存金额'], errors='coerce').fillna(0)
df_begin_balance['结存单价'] = pd.to_numeric(df_begin_balance['结存单价'], errors='coerce').fillna(0)
else:
# 首次核算（无上月结存），初始化期初结存为空表
df_begin_balance = pd.DataFrame(
columns=['物料编码', '仓库编码', '批次号', '结存数量', '结存金额', '结存单价']
)

# 4. 按「物料编码+仓库编码」分组，逐组执行FIFO核心核算
summary_result_list = [] # 存储各组收发存汇总结果
end_balance_list = [] # 存储各组月末有效结存

for (material, warehouse), group_df in df_flow.groupby(['物料编码', '仓库编码']):
# 4.1 提取该物料-仓库的期初结存批次，按批次号排序（保持入库先后顺序）
group_begin_balance = df_begin_balance[
(df_begin_balance['物料编码'] == material) &
(df_begin_balance['仓库编码'] == warehouse)
].sort_values(by='批次号').reset_index(drop=True)

# 4.2 初始化当前结存（期初批次作为初始可用库存）
current_balance = group_begin_balance.copy() if not group_begin_balance.empty else pd.DataFrame(
columns=['批次号', '结存数量', '结存金额', '结存单价']
)

# 4.3 拆分当月收发记录（入库=正数，出库=负数）
in_df = group_df[group_df['数量'] > 0].reset_index(drop=True) # 入库记录
out_df = group_df[group_df['数量'] < 0].reset_index(drop=True) # 出库记录

# 4.4 处理入库：直接追加批次到当前结存，计算批次单价
for _, in_row in in_df.iterrows():
batch_quantity = in_row['数量']
batch_amount = in_row['金额']
batch_price = np.round(batch_amount / batch_quantity, decimal_places) if batch_quantity > 0 else 0.0

# 构造新批次记录
new_batch = pd.DataFrame({
'批次号': [in_row['批次号']],
'结存数量': [batch_quantity],
'结存金额': [batch_amount],
'结存单价': [batch_price]
})

# 追加到当前结存，保持顺序
current_balance = pd.concat([current_balance, new_batch], ignore_index=True)

# 4.5 处理出库：按先进先出顺序逐批扣减，核心FIFO逻辑
for _, out_row in out_df.iterrows():
out_quantity_need = abs(out_row['数量']) # 需出库的总数量（转为正数）
remaining_out = out_quantity_need # 剩余未扣减的出库数量

# 逐批扣减，直至剩余出库数量为0或结存为空
if not current_balance.empty:
for idx, balance_row in current_balance.iterrows():
if remaining_out <= 0:
break

batch_available = balance_row['结存数量']
batch_price = balance_row['结存单价']

if batch_available > remaining_out:
# 情况1：当前批次数量足够，扣减部分数量和金额
current_balance.loc[idx, '结存数量'] = np.round(
batch_available – remaining_out, decimal_places
)
current_balance.loc[idx, '结存金额'] = np.round(
current_balance.loc[idx, '结存数量'] * batch_price, decimal_places
)
remaining_out = 0
else:
# 情况2：当前批次数量不足，全额扣减，删除该批次
remaining_out -= batch_available
current_balance.drop(idx, inplace=True)

# 重置索引，避免后续迭代异常
current_balance = current_balance.reset_index(drop=True)

# 4.6 汇总该物料-仓库的收发存数据
# 期初数据汇总
begin_quantity = np.round(group_begin_balance['结存数量'].sum(), decimal_places)
begin_amount = np.round(group_begin_balance['结存金额'].sum(), decimal_places)
# 当月收发数据汇总
total_in_quantity = np.round(in_df['数量'].sum(), decimal_places) if not in_df.empty else 0.0
total_in_amount = np.round(in_df['金额'].sum(), decimal_places) if not in_df.empty else 0.0
total_out_quantity = np.round(abs(out_df['数量'].sum()), decimal_places) if not out_df.empty else 0.0
# 发出金额：通过FIFO扣减计算（期初+入库-期末，避免直接求和误差）
end_quantity = np.round(current_balance['结存数量'].sum(), decimal_places)
end_amount = np.round(current_balance['结存金额'].sum(), decimal_places)
total_out_amount = np.round(begin_amount + total_in_amount – end_amount, decimal_places)

# 4.7 存入汇总结果列表
summary_result_list.append({
'物料编码': material,
'仓库编码': warehouse,
'期初数量': begin_quantity,
'期初金额': begin_amount,
'本月收入数量': total_in_quantity,
'本月收入金额': total_in_amount,
'本月发出数量': total_out_quantity,
'本月发出金额': total_out_amount,
'期末数量': end_quantity,
'期末金额': end_amount
})

# 4.8 存入月末结存列表（用于下月滚动，添加物料/仓库标识）
if not current_balance.empty:
current_balance['物料编码'] = material
current_balance['仓库编码'] = warehouse
end_balance_list.append(current_balance)

# 5. 整理最终核算结果
# 5.1 完整收发存汇总表
df_final_summary = pd.DataFrame(summary_result_list).reset_index(drop=True)
# 5.2 月末有效结存表（仅保留有结存的批次，用于下月期初）
if end_balance_list:
df_end_balance_valid = pd.concat(end_balance_list, ignore_index=True)
# 筛选有效结存（数量>0），去除冗余
df_end_balance_valid = df_end_balance_valid[df_end_balance_valid['结存数量'] > 0].reset_index(drop=True)
# 调整列顺序，便于阅读
df_end_balance_valid = df_end_balance_valid[['物料编码', '仓库编码', '批次号', '结存数量', '结存金额', '结存单价']]
else:
df_end_balance_valid = pd.DataFrame(
columns=['物料编码', '仓库编码', '批次号', '结存数量', '结存金额', '结存单价']
)

# 6. 保存核算结果到Excel（按月份命名，便于追溯和滚动）
summary_save_path = os.path.join(data_dir, f'FIFO_收发存汇总_{month_str}.xlsx')
balance_save_path = os.path.join(data_dir, f'FIFO_存货结存_{month_str}.xlsx')

df_final_summary.to_excel(summary_save_path, index=False, engine='openpyxl')
df_end_balance_valid.to_excel(balance_save_path, index=False, engine='openpyxl')

# 7. 打印核算完成日志
print(f"✅ {month_str} 先进先出法（FIFO）核算完成")
print(f" 完整汇总表保存路径：{summary_save_path}")
print(f" 月末结存表（下月期初）保存路径：{balance_save_path}")

return df_final_summary, df_end_balance_valid

def batch_monthly_rolling_fifo_costing(data_dir, start_month, end_month):
"""
批量执行先进先出法月度滚动核算（遍历指定月份范围，自动承接上月结存）
"""
# 确保目录存在
ensure_directory_exists(data_dir)
# 生成待核算月份列表
month_calc_list = generate_month_list(start_month, end_month)

# 逐月累加核算
for month in month_calc_list:
try:
fifo_costing_single(month, data_dir)
except Exception as e:
print(f"❌ {month} 先进先出法（FIFO）核算失败，错误信息：{str(e)}")
continue

# ===================== 执行先进先出法批量滚动核算 =====================
if __name__ == "__main__":
batch_monthly_rolling_fifo_costing(target_dir, start_month, end_month)

关键说明

1. 配置项注意事项

• target_dir：必须与之前生成流水数据的目录完全一致，否则会提示流水文件缺失，无法进行核算。
• start_month/end_month：需对应之前生成的 2 年流水数据月份（例如 202401-202512），直接复用即可实现全周期滚动。
• decimal_places：固定为 2 位小数，符合财务核算的金额精度要求，规避浮点运算带来的尾差问题。

2. FIFO 核心实现要点

• 排序前提：流水数据先按「物料编码→仓库编码→单据日期→批次号」排序，保证入库批次的时间顺序正确，这是 FIFO 核算的基础。
• 逐批扣减：出库时遍历当前结存批次，优先扣减最早入库批次，分「批次足够扣减」和「批次不足扣减」两种场景处理，确保逻辑严谨。
• 金额勾稽：本月发出金额通过「期初金额 + 本月收入金额 – 期末金额」倒算，避免直接累加扣减金额带来的精度误差，保证数据勾稽关系正确。

3. 月度滚动的实现核心

• 每月核算完成后，生成「FIFO_存货结存_YYYYMM.xlsx」，记录剩余有效批次的「物料 + 仓库 + 批次 + 数量 + 金额 + 单价」。
• 下月核算时，自动读取上月结存文件，将该批次数据作为期初可用库存，保持批次的时间顺序，实现数据的连续承接。
• 首次核算无上月结存时，自动初始化空结存表，不影响核算流程，确保首次运行即可正常产出结果。

4. 文件命名规则（清晰可追溯）

5. 异常处理与数据兼容

• 若某月份流水文件缺失，会打印错误信息并跳过该月，不影响其他月份的正常核算。
• 自动清洗流水数据：转换日期类型、填充数值空值、删除核心字段无效记录，避免核算过程中出现报错。
• 兼容流水数据的正负值格式：入库为正、出库为负，拆分收发记录时直接按数值符号筛选，无需额外转换。

运行步骤

输出结果说明

• 每个月份生成 2 个 Excel 文件，数据无冗余，符合财务对账和系统对接的要求。
• 汇总表展示完整的收发存勾稽关系，可直接用于财务报表编制；结存表保留批次明细，为下月滚动核算提供准确期初数据。
• 所有金额、数量、单价均保留 2 位小数，符合会计核算准则，无精度误差，可直接用于正式财务工作。