python的运筹学工业场景模拟第三十五篇:解析运输业务表，清洗异常运费，构建源点—目的地单位运费矩阵，输出运输问题模型输入矩阵。

运输业务表解析与单位运费矩阵构建用 Python 打通运输问题模型的成本神经某食品饮料集团在全国有4个生产基地、12个区域配送中心RDC。每月用线性规划做生产基地→RDC的调拨优化模型是经典的运输问题。但运费数据来自三方物流TMS系统导出的Excel——里面充斥着异常记录有同一路线录了3次不同价格合同价/竞价价/紧急价混在一起、有零运费系统默认、有负数退款冲销、有文字描述面议。计划员手工整理运费矩阵要花2天而且上个月因为把面议当0处理模型选了一条实际要8000元的路线导致单月运费多花了23万元。后来我写了个Python脚本自动解析业务表、清洗异常、构建干净的单位运费矩阵 c_{ij} 直接输出给PuLP——现在每月跑一次5分钟出结果运费零异常。—— 参考北京理工大学《运筹学》第7章运输与分配问题一、实际应用场景描述在快消、食品饮料、化工分销、汽车零部件等行业从工厂到配送中心的调拨优化是一个经典场景。运筹学模型非常成熟——运输问题Transportation Problem但模型输入的质量决定了输出方案的好坏。运输问题需要三个核心输入1. 供应向量 \mathbf{a} 每个源点工厂的可发运量2. 需求向量 \mathbf{b} 每个目的地RDC的需求量3. 单位运费矩阵 \mathbf{C} [c_{ij}] 从源点 i 到目的地 j 的单位运输成本第3个——单位运费矩阵——是数据质量最差的一个。┌──────────────────────────────────────────────────────────────┐│ 运输业务表解析 · 单位运费矩阵构建系统 ││ ││ 【输入数据源典型TMS/物流报表】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 运输费用_2025_08.xlsx │││ │ ├── 运费明细! (源点, 目的地, 货物类型, 运量吨, │││ │ │ 运费总额, 单价, 计价方式, 日期, 合同号) │││ │ ├── 路线主数据! (源点, 目的地, 标准距离km, 车型) │││ │ └── 合同价格! (源点, 目的地, 合同单价, 生效日期, 有效期)││ └─────────────────────────────────────────────────────────┘││ ││ 【数据质量问题真实情况】 ││ ⚠ 上海仓→北京RDC: 3条记录分别是4200、3800、5500元/车 ││ 合同价/竞价价/紧急价混在一起 ││ ⚠ 广州仓→成都RDC: 单价面议未签合同临时线路 ││ ⚠ 武汉仓→西安RDC: 运费总额0系统默认实际未录入 ││ ⚠ 天津仓→沈阳RDC: 单价-200退款冲销录入为负数 ││ ⚠ 南京仓→杭州RDC: 运量0.5吨运费1500元→单价3000元/吨 ││ 实际应该是整车价不是吨价 ││ ││ 【本程序处理流程】 ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐││ │ Excel读取│──►│ 异常清洗 │──►│ 单价计算 │──►│ 运费矩阵 │││ │ (pandas) │ │ (零/负/ │ │ (总额/ │ │ C[i][j] │││ │ │ │ 重复/ │ │ 运量) │ │ 构建输出 │││ │ │ │ 文本) │ │ 或取合同 │ │ │││ └──────────┘ └──────────┘ └──────────┘ └──────────┘││ ││ 【输出结果】 ││ • 清洗后的运费明细表 ││ • 源点×目的地的单位运费矩阵 C (DataFrame) ││ • 缺失路线标记需人工补充或估算 ││ • 可直接用于PuLP运输问题建模的c[i][j]字典 │└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某食品饮料集团物流计划员原话我们集团有4个生产基地上海、广州、武汉、天津负责给12个区域配送中心供货。每月初我要用运输问题模型做调拨优化——目标是总运费最小。模型本身很成熟北理工《运筹学》第7章就是讲这个。但单位运费矩阵 c_{ij} 的准备工作太痛苦了1. 我从TMS系统导出上个月的运费明细有300多行。2. 我要从中提取每条路线源点→目的地的平均单价作为本月模型的 c_{ij} 。3. 问题来了- 上海→北京这条线上个月跑了3趟第1趟是合同价4200元/车第2趟是临时竞价3800元/车第3趟是紧急插单5500元/车。我该用哪个价格 手工取了个平均值4500但实际本月合同价已经降到了4000。- 广州→成都上个月没跑过新开的RDC单价写的是面议。我手工填了个8000但模型算出来选了这条线实际一问物流商要12000。- 武汉→西安运费总额是0系统里这条记录是系统自动生成的占位符没录实际运费。我当0处理了模型以为这条线免费拼命往这条线塞货——结果实际运费要6000元/车当月多花了23万。- 天津→沈阳有一条负数记录-200是退款冲销。我SUMIF的时候把它加进去了单价被拉低了。这个手工整理工作我每月要花2天16小时。后来我学了Python写了个脚本自动读TMS明细→自动清洗零/负/文本异常→优先用合同价格→缺失路线标记告警→输出干净的 c_{ij} 矩阵。现在每月跑一次5分钟出结果而且面议和零运费再也不会被当成0了。2.2 人工处理 vs Python自动化量化对比指标 人工Excel处理 Python自动化本方案 改善效果矩阵准备时间 16 小时/月 5 分钟/月 -99.5%零运费误用 1次武汉→西安多花23万 0次 消除缺失路线遗漏 广州→成都填错8000 vs 实际12000 标记告警强制人工确认 可控价格混淆 合同价/竞价混用 优先合同价次选加权平均 准确年化价值 - 约 276 万元避免零运费缺失手工低效 综合关键发现运输问题的目标函数是 \min \sum c_{ij} x_{ij} 。 c_{ij} 是模型的成本神经——如果某条路线的 c_{ij} 被错误地设为0模型会无限偏好这条路线输出完全荒谬的调拨方案。本程序的核心价值就是确保 c_{ij} 矩阵中没有任何虚假的零或虚假的低价。2.3 核心矛盾运输成本数据的核心矛盾是实际执行价格的多样性与模型所需标准单价的唯一性之间的冲突。同一条路线上个月可能跑了合同价、竞价价、紧急价三种价格。模型只需要一个 c_{ij} 。选哪个 选低了→模型偏好→实际执行时价格更高→成本超支。选高了→模型回避→实际可能能拿到更低价格→运力浪费。本程序采用合同价优先、次选加权平均的策略并标记所有非合同价来源让计划员知道哪些价格需要确认。三、核心逻辑讲解大白话版3.1 用大白话解释运费矩阵构建想象你在帮公司安排快递发货场景- 你有3个仓库上海、广州、武汉要给5个门店发货。- 你需要决定每个仓库各给哪些门店发多少货使总运费最低。- 你需要一张表从每个仓库到每个门店每件商品运费多少钱。但现实中的运费记录是这样的- 上海→北京上次发了3次分别花了4200、3800、5500元。你该填哪个- 广州→成都从来没发过记录写的是面议。你该填多少- 武汉→西安系统里写着0元。你敢填0吗 如果填0模型会说哇这条线免费全从武汉发西安——实际上要6000元。你的聪明做法1. 先找合同——合同上写的价格最靠谱优先用合同价。2. 如果没有合同价看历史记录把最近几次的价格取平均加权平均运量大的权重高。3. 如果是面议或0——不要猜标记为缺失让老板决定。4. 如果是负数——那是退款记录直接忽略。工业现场版- 仓库 源点工厂- 门店 目的地RDC- 合同价 标准单价最可靠- 历史均价 次优选择- 面议/0 缺失必须人工介入- 你的聪明做法 本程序大白话总结- 输入乱七八糟的运费明细表有零、负、文本、重复- 处理清洗异常 → 优先合同价 → 次选加权平均 → 标记缺失- 输出干净的 c_{ij} 矩阵直接作为LP的目标函数系数3.2 运筹学模型北理工《运筹学》标准建模运输问题的标准模型北理工《运筹学》§7.1\min \sum_{i \in I} \sum_{j \in J} c_{ij} x_{ij}\text{s.t.} \quad \sum_{j \in J} x_{ij} \le s_i \quad \forall i \in I \quad \text{(供应约束)}\sum_{i \in I} x_{ij} \ge d_j \quad \forall j \in J \quad \text{(需求约束)}x_{ij} \ge 0c_{ij} 的确定方法本程序核心逻辑c_{ij} \begin{cases} \text{ContractPrice}_{ij} \text{if 合同价存在且有效} \\ \dfrac{\sum_k (\text{运量}_k \times \text{运费}_k)}{\sum_k \text{运量}_k} \text{else if 有历史记录} \\ \text{MISSING} \text{otherwise} \end{cases}参考北理工《运筹学》- 第7章运输与分配问题§7.1 运输问题及其数学模型- 第7章§7.2 表上作业法需要单位运费矩阵作为输入3.3 如何映射到代码中数学模型/概念 Python 代码源点集合 Isources: List[str]目的地集合 Jdestinations: List[str]运费矩阵 c_{ij}Dict[Tuple[str,str], float] 或pd.DataFrame合同价优先if contract_price exists: use it加权平均df.groupby([源点,目的地]).apply(weighted_avg)异常清洗df[df[单价] 0] 文本处理缺失标记if pd.isna(c_ij): mark_missingLP目标函数prob lpSum(c[i][j] * x[i][j])四、OOP 代码实现精简可运行4.1 项目结构transportation_cost_matrix/├── transportation_cost_matrix.py # 核心代码单文件~280行├── sample_transport_data.xlsx # 示例数据自动生成├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary运输业务表解析 · 单位运费矩阵构建器参考: 北京理工大学《运筹学》第7章运输与分配问题功能:1. 读取运输费用Excel模拟含零/负/文本/重复等异常2. 清洗异常运费记录负数/零/文本处理3. 构建源点→目的地的单位运费矩阵 c[i][j]4. 策略: 合同价优先 → 历史加权平均 → 标记缺失5. 输出可直接用于PuLP运输问题建模的c矩阵运行:pip install pandas openpyxl pulppython transportation_cost_matrix.pyimport warningsfrom dataclasses import dataclass, fieldfrom pathlib import Pathfrom typing import Dict, List, Optional, Tupleimport numpy as npimport pandas as pdimport pulpwarnings.filterwarnings(ignore)# ─── 数据模型 ────────────────────────────────────────────────────────────dataclassclass RouteCost:路线成本合并后source: strdestination: strunit_cost: float 0.0 # 最终单位运费 c_ijsource_type: str # 价格来源: 合同价/加权平均/缺失sample_count: int 0 # 历史记录条数total_volume: float 0.0 # 总运量warnings: List[str] field(default_factorylist)# ─── 数据读取器 ──────────────────────────────────────────────────────────class TransportDataReader:读取运输业务Exceldef __init__(self, file_path: str sample_transport_data.xlsx):self.file_path Path(file_path)def read_all(self) - Dict[str, pd.DataFrame]:if self.file_path.exists():xl pd.ExcelFile(self.file_path)return {name: pd.read_excel(xl, sheet_namename)for name in xl.sheet_names}print( ⚠️ 未找到数据文件使用内置模拟数据含典型异常)return self._generate_mock_data()def _generate_mock_data(self) - Dict[str, pd.DataFrame]:生成含典型异常的运输费用模拟数据# 运费明细含异常freight pd.DataFrame({源点: [上海仓, 上海仓, 上海仓, 上海仓,广州仓, 广州仓, 广州仓,武汉仓, 武汉仓, 武汉仓,天津仓, 天津仓, 天津仓,南京仓, 南京仓],目的地: [北京RDC, 北京RDC, 北京RDC, 济南RDC,成都RDC, 重庆RDC, 贵阳RDC,西安RDC, 郑州RDC, 长沙RDC,沈阳RDC, 大连RDC, 哈尔滨RDC,杭州RDC, 合肥RDC],货物类型: [常温, 常温, 常温, 常温,冷藏, 常温, 常温,常温, 常温, 常温,常温, 常温, 常温,常温, 常温],运量吨: [25.0, 22.0, 18.0, 15.0,8.0, 12.0, 6.0,20.0, 18.0, 14.0,16.0, 10.0, 8.0,5.0, 6.0],运费总额: [4200.0, 3800.0, 5500.0, 1800.0,面议, 3200.0, 1500.0,0.0, 2400.0, 1900.0,-200.0, 1700.0, 1300.0,1500.0, 1100.0],单价: [168.0, 172.7, 305.6, 120.0,面议, 266.7, 250.0,0.0, 133.3, 135.7,-12.5, 170.0, 162.5,300.0, 183.3],计价方式: [元/吨, 元/吨, 元/吨, 元/吨,面议, 元/吨, 元/吨,元/吨, 元/吨, 元/吨,元/吨, 元/吨, 元/吨,元/车, 元/车],合同号: [C2025-01, SPOT-8821, URGENT-001, None,None, C2025-03, None,None, C2025-02, C2025-02,C2025-04, C2025-04, None,None, None],})# 合同价格表contracts pd.DataFrame({源点: [上海仓, 武汉仓, 武汉仓, 天津仓, 天津仓],目的地: [北京RDC, 郑州RDC, 长沙RDC, 沈阳RDC, 大连RDC],合同单价: [160.0, 130.0, 140.0, 175.0, 165.0],计价单位: [元/吨, 元/吨, 元/吨, 元/吨, 元/吨],生效日期: [2025-01-01] * 5,有效期至: [2025-12-31] * 5,})# 供应与需求简化supply pd.DataFrame({源点: [上海仓, 广州仓, 武汉仓, 天津仓, 南京仓],供应量: [500, 300, 400, 350, 200],})demand pd.DataFrame({目的地: [北京RDC, 济南RDC, 成都RDC, 重庆RDC, 贵阳RDC,西安RDC, 郑州RDC, 长沙RDC, 沈阳RDC, 大连RDC,哈尔滨RDC, 杭州RDC, 合肥RDC],需求量: [200, 100, 80, 120, 60,150, 180, 90, 130, 70,50, 110, 80],})return {运费明细: freight,合同价格: contracts,供应量: supply,需求量: demand,}# ─── 运费清洗与矩阵构建器 ────────────────────────────────────────────────class TransportCostMatrixBuilder:构建单位运费矩阵 c[i][j]参考: 北理工《运筹学》§7.1 运输问题及其数学模型def __init__(self):self.routes: Dict[Tuple[str, str], RouteCost] {}self.missing_routes: List[Tuple[str, str]] []self.cleaner_report: List[str] []def build(self, raw_data: Dict[str, pd.DataFrame]) - pd.DataFrame:主流程freight_df raw_data.get(运费明细, pd.DataFrame()).copy()contract_df raw_data.get(合同价格, pd.DataFrame())# 1. 清洗运费明细freight_df self._clean_freight(freight_df)# 2. 加载合同价contract_prices self._load_contracts(contract_df)# 3. 获取所有路线all_sources set(freight_df[源点].unique()) | set(contract_df[源点].unique())all_dests set(freight_df[目的地].unique()) | set(contract_df[目的地].unique())# 4. 逐路线确定单价for src in all_sources:for dst in all_dests:route_key (src, dst)contract_price contract_prices.get(route_key)if contract_price is not None:# 合同价优先self.routes[route_key] RouteCost(sourcesrc, destinationdst,unit_costcontract_price,source_type合同价)else:# 从历史记录计算加权平均route_data freight_df[(freight_df[源点] src) (freight_df[目的地] dst)]if len(route_data) 0:weighted_avg self._weighted_average(route_data)if weighted_avg 0:rc RouteCost(sourcesrc, destinationdst,unit_costweighted_avg,source_type加权平均,sample_countlen(route_data),total_volumeroute_data[运量吨].sum(),)if any(route_data[合同号].notna()):rc.warnings.append(含有合同价记录但不在合同表中)self.routes[route_key] rcelse:self.missing_routes.append(route_key)self.cleaner_report.append(f路线 {src}→{dst}: 历史记录单价为0或无效)else:self.missing_routes.append(route_key)self.cleaner_report.append(f路线 {src}→{dst}: 无历史记录无合同价 → 缺失!)return self._to_dataframe()def _clean_freight(self, df: pd.DataFrame) - pd.DataFrame:清洗运费明细original_len len(df)# 处理负数neg_mask df[运费总额].apply(lambda x: isinstance(x, (int, float)) and x 0)neg_count neg_mask.sum()if neg_count 0:self.cleaner_report.append(f删除 {neg_count} 条负数运费记录)df df[~neg_mask].copy()# 处理文本型运费面议等text_mask df[运费总额].apply(lambda x: isinstance(x, str))text_count text_mask.sum()if text_count 0:self.cleaner_report.append(f标记 {text_count} 条文本运费为缺失面议/协商)# 将文本型单价设为NaNdf.loc[text_mask, 单价] np.nandf.loc[text_mask, 运费总额] np.nan# 处理零运费zero_mask (df[运费总额] 0) | (df[单价] 0)zero_count zero_mask.sum()if zero_count 0:self.cleaner_report.append(f标记 {zero_count} 条零运费为缺失)df.loc[zero_mask, 单价] np.nan# 重新计算单价从运费总额/运量valid_mask df[运费总额].notna() df[运量吨] 0df.loc[valid_mask, 单价] (df.loc[valid_mask, 运费总额] / df.loc[valid_mask, 运量吨])self.cleaner_report.append(f清洗完成: {original_len} → {len(df)} 条有效记录)return dfdef _load_contracts(self, df: pd.DataFrame) - Dict[Tuple[str, str], float]:加载合同价格contracts {}for _, row in df.iterrows():key (row[源点], row[目的地])contracts[key] float(row[合同单价])return contractsdef _weighted_average(self, route_data: pd.DataFrame) - float:计算加权平均单价valid route_data[route_data[单价].notna() (route_data[单价] 0)]if len(valid) 0:return 0.0total_volume valid[运量吨].sum()if total_volume 0:return valid[单价].mean()weighted_sum (valid[单价] * valid[运量吨]).sum()return weighted_sum / total_volumedef _to_dataframe(self) - pd.DataFrame:转换为DataFramerows []for (src, dst), rc in self.routes.items():rows.append({源点: src,目的地: dst,单位运费: rc.unit_cost,来源: rc.source_type,样本数: rc.sample_count,总运量: rc.total_volume,警告: ; .join(rc.warnings) if rc.warnings else ,})return pd.DataFrame(rows)def get_cost_matrix_for_pulp(self) - Dict[Tuple[str, str], float]:输出PuLP可用的c[i][j]字典return {(src, dst): rc.unit_costfor (src, dst), rc in self.routes.items()}def get_missing_routes(self) - List[Tuple[str, str]]:返回缺失路线列表return self.missing_routes# ─── 报告生成器 ───────────────────────────────────────────────────────────class TransportCostReport:运费矩阵分析报告staticmethoddef print_report(matrix_df: pd.DataFrame,missing_routes: List[Tuple[str, str]],cleaner_report: List[str],):print(f\n {*68})print(f 单位运费矩阵报告 · 运输问题模型输入)print(f {*68})if cleaner_report:print(f\n 数据清洗记录:)for r in cleaner_report:print(f ✓ {r})print(f\n 单位运费矩阵 c[i][j] (元/吨):)# 透视表pivot matrix_df.pivot(index源点, columns目的地, values单位运费)print(f\n{pivot.to_string(float_format%.1f)})if missing_routes:print(f\n 缺失路线需人工补充:)for src, dst in missing_routes:print(f ⚠️ {src} → {dst})print(f\n 价格来源分布:)if 来源 in matrix_df.columns:src_counts matrix_df[来源].value_counts()for src_type, count in src_counts.items():print(f {src_type}: {count} 条路线)# ─── 主流程 ──────────────────────────────────────────────────────────────class TransportCostPipeline:运输成本矩阵构建管道def __init__(self, excel_path: str sample_transport_data.xlsx):self.reader TransportDataReader(excel_path)self.builder TransportCostMatrixBuilder()def run(self, verbose: bool True) - Dict[Tuple[str, str], float]:if verbose:print( 步骤1: 读取运输业务数据...)raw self.reader.read_all()for name, df in raw.items():print(f ✓ {name}: {df.shape[0]}行 × {df.shape[1]}列)if verbose:print(\n 步骤2: 清洗异常运费零/负/文本...)print( 步骤3: 构建单位运费矩阵合同价优先→加权平均...)matrix_df self.builder.build(raw)if verbose:print(\n 步骤4: 生成分析报告...)TransportCostReport.print_report(matrix_df,self.builder.get_missing_routes(),self.builder.cleaner_report,)# 导出矩阵matrix_df.to_csv(transport_cost_matrix.csv, indexFalse, encodingutf-8-sig)print(f\n 运费矩阵已导出: transport_cost_matrix.csv)# 演示PuLP用法c_matrix self.builder.get_cost_matrix_for_pulp()print(f\n PuLP目标函数代码示例:)print(f prob lpSum(c_matrix[(i,j)] * x[i][j] for (i,j) in routes))return self.builder.get_cost_matrix_for_pulp()# ─── 演示 ──────────────────────────────────────────────────────────────def demo():print( * 70)print( 运输业务表解析 · 单位运费矩阵构建器)print( 参考: 北京理工大学《运筹学》第7章运输与分配问题)print( * 70)print(\n 场景: 食品饮料集团生产基地→RDC调拨优化)print( 痛点: 运费数据异常→c_ij矩阵错误→模型选错路线→成本失控)print( 方案: Python自动清洗合同价优先→干净的c矩阵→LP模型\n)pipeline TransportCostPipeline(sample_transport_data.xlsx)c_matrix pipeline.run(verboseTrue)# 构建完整运输问题LP示例print(f\n ️ 步骤5: 构建完整运输问题PuLP模型示例...)raw pipeline.reader.read_all()supply_df raw.get(供应量, pd.DataFrame())demand_df raw.get(需求量, pd.DataFrame())prob pulp.LpProblem(Transport_Optimization, pulp.LpMinimize)sources supply_df[源点].tolist()dests demand_df[目的地].tolist()# 决策变量x {}for (src, dst), cost in c_matrix.items():x[(src, dst)] pulp.LpVariable(fx_{src}_{dst}, lowBound0, catContinuous)# 目标函数prob pulp.lpSum(cost * x[(src, dst)] for (src, dst), cost in c_matrix.items())# 供应约束for _, row in supply_df.iterrows():src row[源点]cap row[供应量]incoming pulp.lpSum(x.get((src, d), 0) for d in dests)prob incoming cap, fSupply_{src}# 需求约束for _, row in demand_df利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛