仓储环境—货物品质关联分析系统 —— 基于 OOP 的保质影响评估实战
"仓库里堆着几百万的货，温湿度记录仪每天都在跑，数据也在往服务器传——但你问仓管：这批原料放了一个月，品质有没有受影响？他只能凭经验说'应该没事'。实际上，温湿度对货物的影响往往是滞后的、累积的。今天的高湿可能不会明天就发霉，但连续两周相对湿度 85% 以上，霉变风险就在指数级增长。问题是，没有人把进出库记录和温湿度数据放在一起算过这笔账。"
—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸
一、实际应用场景描述
在食品、药品、化工原料、电子元器件等行业，仓储环境控制直接关系到货物保质期和出库品质。典型的仓储监测架构如下：
┌──────────────────────────────────────────────┐
│ 智能仓储管理系统 │
│ │
│ ┌─────────────┐ ┌──────────────────────┐ │
│ │ WMS 系统 │ │ 环境监测子系统 │ │
│ │ │ │ │ │
│ │ 入库: 批次A │ │ 温度传感器 × N │ │
│ │ 数量: 500箱 │ │ 湿度传感器 × N │ │
│ │ 时间: 3/1 │ │ 位置: 各库区 │ │
│ │ 保质期: 6个月│ │ 采样: 1次/10分钟 │ │
│ │ │ │ │ │
│ │ 出库: 批次A │ │ 数据 → 时序数据库 │ │
│ │ 数量: 200箱 │ └──────────┬───────────┘ │
│ │ 时间: 3/15 │ │ │
│ └─────────────┘ │ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────────────────────────────────┐ │
│ │ 关联分析引擎 │ │
│ │ 批次停留时段 × 环境暴露量 = 品质风险 │ │
│ └──────────────────────────────────────────┘ │
└──────────────────────────────────────────────┘
不同货物的环境敏感性
货物类型 温度敏感度 湿度敏感度 主要影响 典型限值
纸包装食品 中 高 吸潮变质、霉变 RH < 65%
药品原料 高 高 降解、结块 15~25℃, RH 35~65%
电子元器件 低 极高 引脚氧化、受潮爆米花效应 RH < 10%（MSD）
化工粉末 中 高 结块、流动性下降 RH < 60%
冷冻食品 极高 低 解冻变质、冰晶重结晶 -18℃ 以下
哈尔滨工程大学《工业过程控制》课程在第十四章"环境与过程监控"中专门讨论了环境参数的累积效应：
"环境因素对产品质量的影响通常不是瞬时的，而是与时间相关的累积过程。温度偏离的'面积'（积分）、湿度超标的'时长'（持续时间），这些才是决定货物最终品质的关键指标。孤立地看某一时刻的温度或湿度数值意义有限，必须将环境数据与货物在库时间线进行关联分析。"
二、引入痛点
2.1 现场的真实困境
场景 现场发生了什么 根因
出库抽检不合格 "同一批货，A 仓的出库合格，B 仓的不合格" 两仓环境不同，未关联分析
客户投诉 "夏天存的货，秋天卖出去就坏了" 高温高湿季节的累积暴露
保险理赔争议 "保险公司说不赔，说你们仓库温度超标了" 缺乏量化的暴露量证据
货位分配随意 "新来的货随便找个空位放" 没有基于货物敏感度分配货位
数据孤岛 "温湿度在环控系统，进出库在 WMS，各管各的" 系统未打通
2.2 核心矛盾
货物品质衰减是一个"剂量—时间"函数：环境超标越严重、暴露时间越长，品质劣化越厉害。但大多数仓储管理只关心"有没有超温超湿"（布尔值），而不关心"超了多少、超了多久、累积暴露量是多少"（连续量）。这就好比体检只看"血压正不正常"，而不看"偏高了多少 mmHg、持续了几个月"。
2.3 我们要解决什么
用一段 Python 程序，构建一个仓储环境—货物品质关联分析系统，实现：
1. 数据融合 —— 将进出库流水与环境时序数据按时间对齐
2. 批次环境暴露计算 —— 每个货物批次在库期间的温湿度积分
3. 品质风险评估 —— 基于暴露量模型的劣化概率
4. 货位环境画像 —— 各库区的历史环境表现
5. 智能货位推荐 —— 根据货物敏感度匹配最优库区
6. 面向对象设计 —— 分层清晰，可扩展
三、核心逻辑讲解
3.1 理论基础：品质衰减模型
本工具基于哈工程《工业过程控制》第十四章"环境与过程监控"：
① Arrhenius 温度加速模型（适用于温度敏感货物）
k(T) = A \cdot \exp\left(-\frac{E_a}{RT}\right)
其中 k(T) 是温度 T 下的反应速率（品质衰减速度）， E_a 是活化能， R 是气体常数。
工程简化：温度每升高 10℃，反应速率翻倍（Q₁₀ 规则）：
Q_{10} = 2.0 \implies k(T+\Delta T) = k(T) \times 2^{\Delta T/10}
② 湿度累积暴露模型（适用于湿度敏感货物）
H_{exposure} = \int_{t_{in}}^{t_{out}} \max(0, RH(t) - RH_{safe}) \, dt
③ 综合品质风险指数（QRI）
QRI = \alpha \cdot \frac{T_{exposure}}{T_{threshold}} + \beta \cdot \frac{H_{exposure}}{H_{threshold}}
其中 \alpha, \beta 是根据货物特性的权重系数。
④ 风险分级
QRI 范围 风险等级 建议
0 ~ 0.3 低风险 正常出库
0.3 ~ 0.6 中风险 优先出库 / 加强抽检
0.6 ~ 0.8 高风险 限制出库 / 降级使用
0.8 ~ 1.0+ 极高风险 禁止出库 / 报废
3.2 系统数据流
┌──────────────────────────────────────────────┐
│ WMS 进出库流水 CSV │
│ (batch_id, sku, qty, op, timestamp, zone) │
└──────────────┬───────────────────────────────┘
│
┌──────────────▼───────────────┐
│ ① 批次时间线重建 │
│ 入库→在库→出库 分段 │
└──────────────┬───────────────┘
│
┌──────────────▼───────────────┐
│ ② 环境数据对齐 │
│ 按批次在库时段截取温湿度 │
└──────────────┬───────────────┘
│
┌──────────────▼───────────────┐
│ ③ 暴露量计算 │
│ 温度积分 + 湿度积分 │
└──────────────┬───────────────┘
│
┌──────────────▼───────────────┐
│ ④ 品质风险指数计算 │
│ QRI = α·T_exp + β·H_exp │
└──────────────┬───────────────┘
│
┌──────────────▼───────────────┐
│ ⑤ 货位推荐 & 报告 │
│ 环境画像 + 风险评级 │
└──────────────────────────────┘
四、代码讲解（面向对象设计）
4.1 类结构总览
类名 职责 设计模式
"GoodsBatch" 货物批次信息（dataclass） 值对象
"GoodsProfile" 货物环境敏感度档案（值对象） 值对象
"EnvReading" 单条环境记录（dataclass） 值对象
"RiskThresholds" 风险分级阈值（值对象） 值对象
"InventoryEvent" 库存事件（namedtuple） 值对象
"ExposureResult" 暴露量计算结果（dataclass） 值对象
"WMSDataLoader" 进出库数据加载器 封装
"EnvDataLoader" 环境数据加载器 封装
"BatchTimelineBuilder" 批次时间线重建器 策略模式
"ExposureCalculator" 环境暴露量计算器 策略模式
"QualityRiskAssessor" 品质风险评估器 状态模式
"ZoneProfiler" 库区环境画像器 封装
"StorageAdvisor" 智能货位推荐器 策略模式
"ReportGenerator" 分析报告生成器 模板方法
"WarehouseAnalysisSystem" 系统编排器（聚合根） 聚合根
4.2 数据模型层
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, NamedTuple
from enum import Enum, auto
import numpy as np
import csv
from pathlib import Path
from datetime import datetime, timedelta
from collections import defaultdict
class RiskLevel(Enum):
"""品质风险等级"""
LOW = "低风险"
MEDIUM = "中风险"
HIGH = "高风险"
CRITICAL = "极高风险"
class InventoryOp(Enum):
"""库存操作"""
INBOUND = "入库"
OUTBOUND = "出库"
TRANSFER = "移库"
@dataclass(frozen=True)
class GoodsProfile:
"""货物环境敏感度档案 —— 值对象"""
sku: str # 物料编码
name: str # 品名
temp_sensitivity: float = 1.0 # 温度敏感度 (0~2)
humidity_sensitivity: float = 1.0 # 湿度敏感度 (0~2)
temp_safe_min: float = 15.0 # 安全温度下限
temp_safe_max: float = 25.0 # 安全温度上限
humidity_safe_max: float = 65.0 # 安全湿度上限
shelf_life_days: int = 180 # 标准保质期 (天)
q10_factor: float = 2.0 # Q10 温度加速因子
@dataclass(frozen=True)
class GoodsBatch:
"""货物批次 —— 值对象"""
batch_id: str
sku: str
quantity: int
inbound_time: datetime
outbound_time: Optional[datetime] = None
zone: str = "Zone_A"
@dataclass(frozen=True)
class EnvReading:
"""单条环境记录 —— 值对象"""
timestamp: datetime
zone: str
temperature: float # ℃
humidity: float # %RH
sensor_id: str = ""
@dataclass(frozen=True)
class RiskThresholds:
"""风险分级阈值"""
low: float = 0.3
medium: float = 0.6
high: float = 0.8
class InventoryEvent(NamedTuple):
"""库存事件"""
timestamp: datetime
batch_id: str
operation: InventoryOp
quantity: int
zone: str
@dataclass
class ExposureResult:
"""暴露量计算结果"""
batch_id: str
zone: str
duration_hours: float = 0.0
# 温度暴露
temp_mean: float = 0.0
temp_max: float = 0.0
temp_min: float = 0.0
temp_integral_over: float = 0.0 # 超温积分 (℃·h)
# 湿度暴露
humidity_mean: float = 0.0
humidity_max: float = 0.0
humidity_over_hours: float = 0.0 # 超湿时长 (h)
humidity_integral_over: float = 0.0 # 超湿积分 (%RH·h)
# 综合
qri: float = 0.0
risk_level: RiskLevel = RiskLevel.LOW
4.3 数据加载器
class WMSDataLoader:
"""
WMS 进出库数据加载器
CSV 格式:
timestamp,batch_id,sku,operation,quantity,zone
2024-03-01 08:00:00,BATCH-001,SKU-A001,INBOUND,500,Zone_A
2024-03-15 14:00:00,BATCH-001,SKU-A001,OUTBOUND,200,Zone_A
支持:
- 多批次交叉进出
- 自动重建批次在库时段
"""
def __init__(self):
self.events: List[InventoryEvent] = []
def load_csv(self, file_path: str) -> List[InventoryEvent]:
"""加载进出库流水"""
self.events.clear()
with open(file_path, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
ts = datetime.strptime(row['timestamp'], "%Y-%m-%d %H:%M:%S")
op = self._parse_op(row.get('operation', ''))
event = InventoryEvent(
timestamp=ts,
batch_id=row['batch_id'],
operation=op,
quantity=int(row['quantity']),
zone=row.get('zone', 'Zone_A')
)
self.events.append(event)
return self.events
def _parse_op(self, op_str: str) -> InventoryOp:
op_map = {
'INBOUND': InventoryOp.INBOUND,
'OUTBOUND': InventoryOp.OUTBOUND,
'TRANSFER': InventoryOp.TRANSFER
}
return op_map.get(op_str.strip().upper(), InventoryOp.INBOUND)
def build_batch_timeline(self, batch_id: str) -> List[Tuple[datetime, datetime, str]]:
"""
重建指定批次的在库时段列表
Returns:
[(inbound_time, outbound_time_or_now, zone), ...]
"""
batch_events = [e for e in self.events if e.batch_id == batch_id]
batch_events.sort(key=lambda x: x.timestamp)
timeline = []
pending_in = None
for event in batch_events:
if event.operation == InventoryOp.INBOUND:
pending_in = event
elif event.operation == InventoryOp.OUTBOUND and pending_in:
timeline.append((pending_in.timestamp, event.timestamp, event.zone))
pending_in = None
# 如果还有未出库的
if pending_in:
timeline.append((pending_in.timestamp, datetime.now(), pending_in.zone))
return timeline
class EnvDataLoader:
"""
环境数据加载器
CSV 格式:
timestamp,zone,temperature,humidity,sensor_id
2024-03-01 08:00:00,Zone_A,22.5,58.3,SENSOR-01
"""
def __init__(self):
self.readings: List[EnvReading] = []
def load_csv(self, file_path: str) -> List[EnvReading]:
"""加载环境数据"""
self.readings.clear()
with open(file_path, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
ts = datetime.strptime(row['timestamp'], "%Y-%m-%d %H:%M:%S")
reading = EnvReading(
timestamp=ts,
zone=row['zone'],
temperature=float(row['temperature']),
humidity=float(row['humidity']),
sensor_id=row.get('sensor_id', '')
)
self.readings.append(reading)
return self.readings
def get_readings_in_range(self, zone: str, start: datetime,
end: datetime) -> List[EnvReading]:
"""获取指定时段和区域的环境数据"""
return [r for r in self.readings
if r.zone == zone and start <= r.timestamp <= end]
4.4 批次时间线重建器
class BatchTimelineBuilder:
"""
批次时间线重建器
处理复杂场景:
- 分批出库（同一批次多次出库）
- 移库操作
- 退货返库
"""
def __init__(self, wms_loader: WMSDataLoader):
self.wms = wms_loader
def build_sub_batches(self, batch_id: str) -> List[Dict]:
"""
将批次拆分为多个子批次（按出库事件拆分）
例如:
3/1 入库 500箱 → 3/15 出库 200箱 → 3/30 出库 300箱
拆分为:
子批次1: 3/1~3/15, 200箱
子批次2: 3/1~3/30, 300箱
"""
events = [e for e in self.wms.events if e.batch_id == batch_id]
events.sort(key=lambda x: x.timestamp)
sub_batches = []
remaining_qty = None
inbound_event = None
for event in events:
if event.operation == InventoryOp.INBOUND:
inbound_event = event
remaining_qty = event.quantity
elif event.operation == InventoryOp.OUTBOUND and inbound_event:
out_qty = min(event.quantity, remaining_qty)
sub_batches.append({
'sub_id': f"{batch_id}-OUT-{event.timestamp.strftime('%Y%m%d%H%M')}",
'inbound_time': inbound_event.timestamp,
'outbound_time': event.timestamp,
'zone': event.zone,
'quantity': out_qty
})
remaining_qty -= out_qty
# 剩余未出库部分
if remaining_qty and remaining_qty > 0 and inbound_event:
sub_batches.append({
'sub_id': f"{batch_id}-REMAIN",
'inbound_time': inbound_event.timestamp,
'outbound_time': datetime.now(),
'zone': inbound_event.zone,
'quantity': remaining_qty
})
return sub_batches
4.5 环境暴露量计算器（核心算法）
class ExposureCalculator:
"""
环境暴露量计算器 —— 策略模式
核心计算:
1. 温度积分: ∫ max(0, T - T_safe) dt
2. 湿度积分: ∫ max(0, RH - RH_safe) dt
3. 超温/超湿时长统计
"""
def __init__(self, profile: GoodsProfile):
self.profile = profile
def calculate(self, env_readings: List[EnvReading]) -> ExposureResult:
"""
计算环境暴露量
Args:
env_readings: 指定时段的环境数据
Returns:
暴露量结果
"""
if not env_readings:
return ExposureResult(batch_id="", zone="")
temps = np.array([r.temperature for r in env_readings])
humids = np.array([r.humidity for r in env_readings])
timestamps = [r.timestamp for r in env_readings]
# 时间间隔 (小时)
if len(timestamps) >= 2:
dt_hours = (timestamps[1] - timestamps[0]).total_seconds() / 3600.0
else:
dt_hours = 0.0
duration_hours = (timestamps[-1] - timestamps[0]).total_seconds() / 3600.0
# 温度统计
temp_mean = float(np.mean(temps))
temp_max = float(np.max(temps))
temp_min = float(np.min(temps))
# 温度超温积分 (超过上限)
temp_over = np.maximum(0, temps - self.profile.temp_safe_max)
temp_integral_over = float(np.sum(temp_over) * dt_hours)
# 温度低于下限的积分
temp_under = np.maximum(0, self.profile.temp_safe_min - temps)
temp_integral_under = float(np.sum(temp_under) * dt_hours)
# 湿度统计
humidity_mean = float(np.mean(humids))
humidity_max = float(np.max(humids))
# 湿度超湿积分
humidity_over = np.maximum(0, humids - self.profile.humidity_safe_max)
humidity_integral_over = float(np.sum(humidity_over) * dt_hours)
# 超湿时长
humidity_over_hours = float(np.sum(humids > self.profile.humidity_safe_max) * dt_hours)
return ExposureResult(
batch_id="",
zone=env_readings[0].zone if env_readings else "",
duration_hours=round(duration_hours, 2),
temp_mean=round(temp_mean, 2),
temp_max=round(temp_max, 2),
temp_min=round(temp_min, 2),
temp_integral_over=round(temp_integral_over, 2),
humidity_mean=round(humidity_mean, 2),
humidity_max=round(humidity_max, 2),
humidity_over_hours=round(humidity_over_hours, 2),
humidity_integral_over=round(humidity_integral_over, 2)
)
4.6 品质风险评估器
class QualityRiskAssessor:
"""
品质风险评估器 —— 状态模式
综合 QRI = α·(T_integral / T_threshold) + β·(H_integral / H_threshold)
"""
def __init__(self, thresholds: RiskThresholds = None):
self.thresholds = thresholds or RiskThresholds()
def assess(self, exposure: ExposureResult, profile: GoodsProfile) -> Tuple[RiskLevel, float]:
"""
评估品质风险
Args:
exposure: 暴露量计算结果
profile: 货物敏感度档案
Returns:
(风险等级, QRI 值)
"""
# 温度暴露基准: 假设标准条件下允许整个保质期
# T_threshold = (T_safe_max - T_safe_min) × shelf_life_hours × 0.1
shelf_life_hours = profile.shelf_life_days * 24
temp_threshold = (profile.temp_safe_max - profile.temp_safe_min) * shelf_life_hours * 0.1
# 湿度暴露基准
humidity_threshold = (100.0 - profile.humidity_safe_max) * shelf_life_hours * 0.1
# 归一化暴露量
temp_norm = exposure.temp_integral_over / (temp_threshold + 1e-10)
humidity_norm = exposure.humidity_integral_over / (humidity_threshold + 1e-10)
# QRI 计算
qri = (profile.temp_sensitivity * temp_norm +
profile.humidity_sensitivity * humidity_norm) / \
(profile.temp_sensitivity + profile.humidity_sensitivity)
qri = min(1.0, max(0.0, qri))
# 分级
if qri <= self.thresholds.low:
level = RiskLevel.LOW
elif qri <= self.thresholds.medium:
level = RiskLevel.MEDIUM
elif qri <= self.thresholds.high:
level = RiskLevel.HIGH
else:
level = RiskLevel.CRITICAL
return level, round(qri, 4)
4.7 库区环境画像器
class ZoneProfiler:
"""
库区环境画像器
统计各库区的历史环境表现:
- 平均温湿度
- 超标频率
- 稳定性（标准差）
"""
def profile_zones(self, env_readings: List[EnvReading]) -> Dict[str, dict]:
"""
生成各库区的环境画像
Args:
env_readings: 所有环境数据
Returns:
{zone: {stats}}
"""
zones = defaultdict(list)
for r in env_readings:
zones[r.zone].append(r)
profiles = {}
for zone, readings in zones.items():
temps = [r.temperature for r in readings]
humids = [r.humidity for r in readings]
profiles[zone] = {
'reading_count': len(readings),
'temp_mean': round(float(np.mean(temps)), 2),
'temp_std': round(float(np.std(temps)), 2),
'temp_max': round(float(np.max(temps)), 2),
'temp_min': round(float(np.min(temps)), 2),
'humidity_mean': round(float(np.mean(humids)), 2),
'humidity_std': round(float(np.std(humids)), 2),
'humidity_max': round(float(np.max(humids)), 2),
'stability_score': round(1.0 / (1.0 + float(np.std(temps)) + float(np.std(humids)) * 0.1), 3)
}
return profiles
4.8 智能货位推荐器
class StorageAdvisor:
"""
智能货位推荐器
根据货物敏感度和库区环境画像，推荐最优存放位置
"""
def recommend(self, profile: GoodsProfile, zone_profiles: Dict[str, dict]) -> List[dict]:
"""
推荐货位
Args:
profile: 货物敏感度档案
zone_profiles: 库区环境画像
Returns:
推荐列表（按适配度排序）
"""
recommendations = []
for zone, stats in zone_profiles.items():
# 适配度评分 (0~100)
# 温度适配: 离安全区间越近越好
temp_center = (profile.temp_safe_max + profile.temp_safe_min) / 2
temp_deviation = abs(stats['temp_mean'] - temp_center)
temp_score = max(0, 100 - temp_deviation * 10 - stats['temp_std'] * 5)
# 湿度适配: 越低越好
humidity_deviation = max(0, stats['humidity_mean'] - profile.humidity_safe_max)
humidity_score = max(0, 100 - humidity_deviation * 2 - stats['humidity_std'] * 3)
# 综合适配度
suitability = (
profile.temp_sensitivity * temp_score +
profile.humidity_sensitivity * humidity_score
) / (profile.temp_sensitivity + profile.humidity_sensitivity)
recommendations.append({
'zone': zone,
'suitability': round(suitability, 1),
'temp_score': round(temp_score, 1),
'humidity_score': round(humidity_score, 1),
'reason': self._generate_reason(zone, stats, profile)
})
recommendations.sort(key=lambda x: x['suitability'], reverse=True)
return recommendations
def _generate_reason(self, zone: str, stats: dict, profile: GoodsProfile) -> str:
"""生成推荐理由"""
reasons = []
if stats['temp_mean'] <= profile.temp_safe_max
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
