python的工业过程控制场景模拟第四十三篇:解析连续生产批次数据，分析工艺参数波动与次品率相关性。

连续生产批次参数波动与次品率相关性分析系统 —— 基于 OOP 的数据驱动实战车间里最让人头疼的不是设备坏了而是设备明明在转、参数看起来也没超限但次品率就是忽高忽低。老师傅说最近炉温不太稳可 DCS 趋势图上温度一直在设定值±2℃以内晃——到底哪里不稳答案是波动的模式比波动的数值更重要。—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸一、实际应用场景描述在连续生产的流程工业中化工聚合、发酵、热处理、烧结等批次与批次之间的质量差异往往不是由单一时刻的超标造成的而是由整个批次过程中参数的波动特性决定的。┌──────────────────────────────────────────────┐│ 连续生产批次分析全景 ││ ││ 批次001: 温度曲线 ──→ 标准差σ1.2 → 合格 ││ 批次002: 温度曲线 ──→ 标准差σ4.8 → 次品 ││ 批次003: 温度曲线 ──→ 标准差σ1.5 → 合格 ││ 批次004: 温度曲线 ──→ 爬升慢30min → 次品 ││ ││ 问题: 所有批次的均值都在 180±2℃ ││ 但次品批次的波动形态完全不同 │└──────────────────────────────────────────────┘常见的隐形质量问题波动类型 表现 对产品的影响周期性振荡 温度每 15 分钟一个小波峰 结晶粒度不均匀缓慢漂移 8 小时内温度逐渐升高 3℃ 反应转化率偏低爬升过慢 升温阶段比标准多花了 20 分钟 副反应增多突刺噪声 偶发的 ±5℃ 尖峰 局部过热碳化方差偏大 均值正常但 σ 持续偏高 批次一致性差哈尔滨工程大学《工业过程控制》课程在第十一章工业过程数据分析与故障诊断中指出过程能力指数 Cp 和 Cpk 是衡量过程稳定性的核心指标。仅仅监控过程变量是否在限内是不够的还需要分析其分布特性和变化趋势。统计过程控制SPC的核心思想是通过对过程波动的分层分析区分偶然因素和异常因素。二、引入痛点2.1 现场的真实困境场景 现场发生了什么 根因均值正常但次品多 温度一直在 180℃怎么还出次品 只看均值忽略方差和波动模式数据散落各处 工艺数据在 DCS质检数据在 LIMS对不上 缺乏跨系统集成相关性凭感觉 好像炉温不稳的时候次品就多 没有量化证据支撑审计被问住 你怎么证明这个工艺参数跟次品有关 缺少统计显著性检验调参盲目 改了 PID 参数不知道对良率有没有帮助 没有前后对比分析2.2 核心矛盾传统的控制图只回答有没有超限但连续生产中大多数次品批次根本没有超限——它们只是波动得不太正常。你需要的是从整条批次曲线中提取特征均值、方差、斜率、振荡频率然后把这些特征和次品记录做相关性分析。这才是数据驱动的工艺优化。2.3 我们要解决什么用一段 Python 程序构建一个连续生产批次参数波动与次品率相关性分析系统实现1. 批次数据解析 —— 从 CSV 加载多批次的时序工艺参数2. 波动特征提取 —— 均值、标准差、极差、斜率、振荡频率3. 相关性分析 —— Pearson/Spearman 相关系数 p 值检验4. 次品率关联 —— 批次特征与合格/次品标签的映射5. 可视化输出 —— 散点图矩阵 相关系数热力图6. 面向对象设计 —— 分层清晰可扩展三、核心逻辑讲解3.1 理论基础统计过程控制与相关性分析本工具基于哈工程《工业过程控制》第十一章过程数据分析① 批次特征提取对每一个批次的时间序列 X \{x_1, x_2, ..., x_n\} 提取以下特征特征 公式 物理意义均值 \mu \frac{1}{n}\sum x_i 批次的平均工艺水平标准差 \sigma \sqrt{\frac{1}{n}\sum(x_i-\mu)^2} 波动剧烈程度极差 R x_{max}-x_{min} 极端偏差斜率 k 线性回归系数 漂移趋势变异系数 CV \sigma/\mu 相对波动振荡频率 f 过零率/FFT 主频 周期性扰动② 相关性度量Pearson 相关系数r \frac{\sum(X_i-\bar{X})(Y_i-\bar{Y})}{\sqrt{\sum(X_i-\bar{X})^2\sum(Y_i-\bar{Y})^2}}Spearman 秩相关系数对非线性单调关系更鲁棒\rho 1 - \frac{6\sum d_i^2}{n(n^2-1)}③ 统计显著性检验H_0 : 相关系数为 0无关H_1 : 相关系数不为 0有关t r\sqrt{\frac{n-2}{1-r^2}} \sim t(n-2)p 0.05 → 拒绝原假设认为存在显著相关性。3.2 分析流程架构┌──────────────────────────────┐│ 数据加载 清洗 ││ 工艺参数 CSV 质检结果 CSV │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ① 批次分组 ││ 按 batch_id 聚合时序数据 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ② 特征工程 ││ 每个批次 → 特征向量 ││ [μ, σ, R, k, CV, f] │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ③ 标签对齐 ││ 批次特征 ↔ 质检结果 JOIN │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ④ 相关性计算 ││ Pearson Spearman p值 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ⑤ 可视化 报告 ││ 热力图 散点图 结论 │└──────────────────────────────┘四、代码讲解面向对象设计4.1 类结构总览类名 职责 设计模式BatchRecord 单条批次时序记录dataclass 值对象QualityLabel 批次质检标签dataclass 值对象FeatureConfig 特征提取配置值对象 值对象CorrelationConfig 相关性分析配置值对象 值对象DataLoader CSV 数据加载与合并 封装BatchGrouper 批次分组器 策略模式FeatureExtractor 波动特征提取器 封装CorrelationAnalyzer 相关性分析引擎 策略模式ReportGenerator 分析报告生成器 模板方法BatchAnalysisSystem 系统编排器聚合根 聚合根4.2 数据模型层from dataclasses import dataclassfrom typing import List, Dict, Optional, Tuplefrom enum import Enumimport numpy as npimport csvfrom pathlib import Pathfrom collections import defaultdictclass BatchResult(Enum):批次质检结果PASS 合格FAIL 次品PENDING 待检dataclass(frozenTrue)class BatchRecord:单条批次时序记录 —— 值对象batch_id: str # 批次号timestamp: float # 时间戳 (s, 从批次开始计)parameter: str # 参数名称 (如 temperature)value: float # 参数值unit: str # 单位dataclass(frozenTrue)class QualityLabel:批次质检标签 —— 值对象batch_id: strresult: BatchResultdefect_type: str # 缺陷类型 (如 尺寸超差)severity: str minor # minor/major/criticaldataclass(frozenTrue)class FeatureConfig:特征提取配置features: List[str] None # 要提取的特征列表fft_enabled: bool False # 是否启用 FFT 频域特征def __post_init__(self):if self.features is None:object.__setattr__(self, features, [mean, std, range, slope, cv])dataclass(frozenTrue)class CorrelationConfig:相关性分析配置method: str pearson # pearson / spearman / bothalpha: float 0.05 # 显著性水平min_samples: int 10 # 最少批次数量4.3 数据加载器class DataLoader:批次数据加载器支持两个 CSV 文件:1. process_data.csv: 工艺参数时序数据2. quality_data.csv: 质检结果标签CSV 格式 (process_data):batch_id,timestamp,parameter,value,unitBATCH_001,0,temperature,178.5,℃BATCH_001,60,temperature,179.2,℃...CSV 格式 (quality_data):batch_id,result,defect_type,severityBATCH_001,PASS,,BATCH_002,FAIL,尺寸超差,majordef __init__(self):self.process_records: List[BatchRecord] []self.quality_labels: List[QualityLabel] []def load_process_data(self, file_path: str) - List[BatchRecord]:加载工艺参数数据self.process_records.clear()with open(file_path, r, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:record BatchRecord(batch_idrow[batch_id],timestampfloat(row[timestamp]),parameterrow[parameter],valuefloat(row[value]),unitrow.get(unit, ))self.process_records.append(record)return self.process_recordsdef load_quality_data(self, file_path: str) - List[QualityLabel]:加载质检结果数据self.quality_labels.clear()with open(file_path, r, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:label QualityLabel(batch_idrow[batch_id],resultBatchResult(row[result]),defect_typerow.get(defect_type, ),severityrow.get(severity, minor))self.quality_labels.append(label)return self.quality_labels4.4 批次分组器class BatchGrouper:批次分组器 —— 按 batch_id 和 parameter 聚合时序数据输出结构:{BATCH_001: {temperature: [(t1, v1), (t2, v2), ...],pressure: [(t1, v1), (t2, v2), ...]},...}def group(self, records: List[BatchRecord]) - Dict[str, Dict[str, List[Tuple[float, float]]]]:按批次和参数分组Returns:嵌套字典: batch_id → parameter → [(timestamp, value), ...]grouped defaultdict(lambda: defaultdict(list))for r in records:grouped[r.batch_id][r.parameter].append((r.timestamp, r.value))# 每个参数组内按时间排序for batch_id in grouped:for param in grouped[batch_id]:grouped[batch_id][param].sort(keylambda x: x[0])return dict(grouped)4.5 波动特征提取器核心算法class FeatureExtractor:波动特征提取器对每个批次的每个参数时间序列提取以下特征:- mean: 均值- std: 标准差- range: 极差- slope: 线性回归斜率 (漂移趋势)- cv: 变异系数- oscillation: 振荡指数 (相邻点变化绝对值之和 / 均值)def __init__(self, config: FeatureConfig None):self.config config or FeatureConfig()def extract_all(self, grouped_data: Dict[str, Dict[str, List[Tuple[float, float]]]]) - Dict[str, Dict[str, Dict[str, float]]]:对所有批次的所有参数提取特征Returns:{batch_id: {parameter: {feature_name: value}}}all_features {}for batch_id, params in grouped_data.items():all_features[batch_id] {}for param, time_series in params.items():values np.array([v for _, v in time_series])features self.extract_single(values)all_features[batch_id][param] featuresreturn all_featuresdef extract_single(self, values: np.ndarray) - Dict[str, float]:对单条时间序列提取特征Args:values: 数值数组Returns:特征字典if len(values) 2:return {mean: 0.0, std: 0.0, range: 0.0, slope: 0.0, cv: 0.0, oscillation: 0.0}features {}# 均值features[mean] float(np.mean(values))# 标准差features[std] float(np.std(values))# 极差features[range] float(np.max(values) - np.min(values))# 斜率 (线性回归)n len(values)x np.arange(n)if n 1:features[slope] float(np.polyfit(x, values, 1)[0])else:features[slope] 0.0# 变异系数mean_val features[mean]features[cv] float(features[std] / abs(mean_val) * 100.0) if mean_val ! 0 else 0.0# 振荡指数 (Oscillation Index)# 相邻点变化绝对值之和 / (均值 × n)diff_sum float(np.sum(np.abs(np.diff(values))))features[oscillation] diff_sum / (abs(mean_val) * n) if mean_val ! 0 else 0.0return features4.6 相关性分析引擎class CorrelationAnalyzer:相关性分析引擎功能:1. Pearson 相关系数2. Spearman 秩相关系数3. 显著性检验 (t-test)4. 多参数多特征的相关性矩阵def __init__(self, config: CorrelationConfig None):self.config config or CorrelationConfig()def pearson(self, x: np.ndarray, y: np.ndarray) - Tuple[float, float]:Pearson 相关系数及 p 值Returns:(r, p_value)n len(x)if n 3:return 0.0, 1.0# 相关系数x_mean, y_mean np.mean(x), np.mean(y)numerator np.sum((x - x_mean) * (y - y_mean))denominator np.sqrt(np.sum((x - x_mean)**2) * np.sum((y - y_mean)**2))if denominator 0:return 0.0, 1.0r numerator / denominator# t 统计量if abs(r) 1.0:return r, 0.0t r * np.sqrt((n - 2) / (1 - r**2))# 近似 p 值 (双侧检验)from scipy import statsp 2 * (1 - stats.t.cdf(abs(t), n - 2)) if n 2 else 1.0return float(r), float(p)def spearman(self, x: np.ndarray, y: np.ndarray) - Tuple[float, float]:Spearman 秩相关系数Returns:(rho, p_value)n len(x)if n 3:return 0.0, 1.0# 排名rank_x np.argsort(np.argsort(x)) 1rank_y np.argsort(np.argsort(y)) 1# Pearson on ranksr, p self.pearson(rank_x.astype(float), rank_y.astype(float))return r, pdef analyze_feature_correlations(self, feature_matrix: np.ndarray,target: np.ndarray) - List[Dict]:分析所有特征与目标变量的相关性Args:feature_matrix: (n_samples, n_features)target: (n_samples,) 目标变量 (如次品1, 合格0)Returns:相关性结果列表results []n_features feature_matrix.shape[1]for i in range(n_features):x feature_matrix[:, i]if self.config.method in (pearson, both):r, p self.pearson(x, target)significant p self.config.alpharesults.append({feature_index: i,method: pearson,coefficient: round(r, 4),p_value: round(p, 4),significant: significant,strength: self._interpret(r)})if self.config.method in (spearman, both):rho, p self.spearman(x, target)significant p self.config.alpharesults.append({feature_index: i,method: spearman,coefficient: round(rho, 4),p_value: round(p, 4),significant: significant,strength: self._interpret(rho)})return resultsdef _interpret(self, r: float) - str:解释相关性强弱abs_r abs(r)if abs_r 0.8:return 极强相关elif abs_r 0.6:return 强相关elif abs_r 0.4:return 中等相关elif abs_r 0.2:return 弱相关else:return 极弱/无相关4.7 分析报告生成器class ReportGenerator:分析报告生成器def generate(self, correlation_results: List[Dict],feature_names: List[str],summary_stats: Dict None) - str:生成文本报告lines [ * 65, 连续生产批次参数波动与次品率相关性分析报告, * 65,,]# 显著相关特征significant [r for r in correlation_results if r[significant]]lines.append(f 共分析 {len(correlation_results)} 个特征-目标组合)lines.append(f 其中 {len(significant)} 个呈现显著相关性 (p 0.05))lines.append()if significant:lines.append(- * 65)lines.append( 显著相关特征排行:)lines.append(- * 65)lines.append(f {特征:20} {方法:10} {系数:10} {p值:10} {强度})lines.append(f {-*55})for r in sorted(significant, keylambda x: abs(x[coefficient]), reverseTrue):fname feature_names[r[feature_index]] if r[feature_index] len(feature_names) else fF{r[feature_index]}lines.append(f {fname:20} {r[method]:10} {r[coefficient]:10.4f} {r[p_value]:10.4f} {r[strength]})else:lines.append( ⚠️ 未发现显著相关性可能需要:)lines.append( - 增加批次样本数量)lines.append( - 检查特征工程是否合理)lines.append( - 验证质检标签的准确性)lines.append()lines.append( * 65)return \n.join(lines)4.8 系统编排器class BatchAnalysisSystem:连续生产批次分析系统 —— 聚合根串联: 加载 → 分组 → 特征提取 → 相关性分析 → 报告def __init__(self, feature_config: FeatureConfig None,correlation_config: CorrelationConfig None):self.feature_config feature_config or FeatureConfig()self.correlation_config correlation_config or CorrelationConfig()self.loader DataLoader()self.grouper BatchGrouper()self.extractor FeatureExtractor(self.feature_config)self.analyzer CorrelationAnalyzer(self.correlation_config)self.reporter ReportGenerator()def run(self, process_csv: str, quality_csv: str) - str:执行完整的分析流程Args:process_csv: 工艺参数数据路径quality_csv: 质检结果数据路径Returns:分析报告文本# ① 加载self.loader.load_process_data(process_csv)quality_labels self.loader.load_quality_data(quality_csv)if len(self.loader.process_records) 0:return 错误: 未加载到工艺数据# ② 分组grouped self.grouper.group(self.loader.process_records)# ③ 特征提取features self.extractor.extract_all(grouped)# ④ 构建特征矩阵和目标向量batch_ids sorted(features.keys())feature_names []feature_matrix []target []# 收集所有参数名all_params set()for bid in batch_ids:all_params.update(features[bid].keys())# 展平特征: 每个批次一行每个(参数特征)一列for param in sorted(all_params):for feat_name in self.feature_config.features:feature_names.append(f{param}_{feat_name})# 构建矩阵for bid in batch_ids:row []for param in sorted(all_params):param_features features[bid].get(param, {})for feat_name in self.feature_config.features:row.append(param_features.get(feat_name, 0.0))feature_matrix.append(row)# 目标: 次品1, 合格0label next((l for l in quality_labels if l.batch_id bid), None)target.append(1.0 if label and label.result BatchResult.FAIL else 0.0)X np.array(feature_matrix)y np.array(target)if len(X) self.correlation_config.min_samples:return f错误: 有效批次数量 ({len(X)}) 少于最小要求 ({self.correlation_config.min_samples})# ⑤ 相关性分析results self.analyzer.analyze_feature_correlations(X, y)# ⑥ 生成报告report self.reporter.generate(results, feature_names)return report4.9 完整演示def demo():完整演示print( * 65)print( 连续生产批次参数波动与次品率相关性分析系统 v1.0)print( 基于哈尔滨工程大学《工业过程控制》课程理论)print( * 65)import tempfileimport os# 生成模拟工艺数据np.random.seed(42)process_lines [batch_id,timestamp,parameter,value,unit]parameters [temperature, pressure]n_batches 30for b in range(1, n_batches 1):batch_id fBATCH_{b:03d}# 温度: 次品批次有更大的波动if b % 5 0: # 每5批出一个次品base_temp 180.0noise_scale 4.0 # 高波动else:noise_scale 1.0 # 正常波动for t in range(0, 120, 10): # 12个时间点temp base_temp np.random.normal(0, noise_scale) t * 0.02process_lines.append(f{batch_id},{t},temperature,{temp:.2f},℃)# 压力: 次品批次有更大的斜率漂移if b % 5 0:slope 0.5 # 漂移else:slope 0.05for t in range(0, 120, 10):press 200.0 slope * t np.random.normal(0, 2)process_lines.append(f{batch_id},{t},pressure,{press:.2f},kPa)# 生成质检数据quality_lines [batch_id,result,defect_type,severity]for b in range(1, n_batches 1):batch_id fBATCH_{b:03d}if b % 5 0:quality_lines.append(f{batch_id},FAIL,性能不达标,major)else:quality_lines.append(f{batch_id},PASS,,)# 写入临时文件tmpdir tempfile.mkdtemp()process_csv os.path.join(tmpdir, process.csv)quality_csv os.path.join(tmpdir, quality.csv)with open(process_csv, w) as f:f.write(\n.join(process_lines))with open(quality_csv, w) as f:f.write(\n.join(quality_lines))# 运行分析system BatchAnalysisSystem(feature_configFeatureConfig(),correlation_configCorrelationConfig(methodboth, alpha0.05))report system.run(process_csv, quality_csv)print(report)利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛