连续生产批次参数波动与次品率相关性分析系统 —— 基于 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 dataclass
from typing import List, Dict, Optional, Tuple
from enum import Enum
import numpy as np
import csv
from pathlib import Path
from collections import defaultdict
class BatchResult(Enum):
"""批次质检结果"""
PASS = "合格"
FAIL = "次品"
PENDING = "待检"
@dataclass(frozen=True)
class BatchRecord:
"""单条批次时序记录 —— 值对象"""
batch_id: str # 批次号
timestamp: float # 时间戳 (s, 从批次开始计)
parameter: str # 参数名称 (如 "temperature")
value: float # 参数值
unit: str = "" # 单位
@dataclass(frozen=True)
class QualityLabel:
"""批次质检标签 —— 值对象"""
batch_id: str
result: BatchResult
defect_type: str = "" # 缺陷类型 (如 "尺寸超差")
severity: str = "minor" # minor/major/critical
@dataclass(frozen=True)
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(frozen=True)
class CorrelationConfig:
"""相关性分析配置"""
method: str = "pearson" # pearson / spearman / both
alpha: 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,unit
BATCH_001,0,temperature,178.5,℃
BATCH_001,60,temperature,179.2,℃
...
CSV 格式 (quality_data):
batch_id,result,defect_type,severity
BATCH_001,PASS,,
BATCH_002,FAIL,尺寸超差,major
"""
def __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', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
record = BatchRecord(
batch_id=row['batch_id'],
timestamp=float(row['timestamp']),
parameter=row['parameter'],
value=float(row['value']),
unit=row.get('unit', '')
)
self.process_records.append(record)
return self.process_records
def load_quality_data(self, file_path: str) -> List[QualityLabel]:
"""加载质检结果数据"""
self.quality_labels.clear()
with open(file_path, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
label = QualityLabel(
batch_id=row['batch_id'],
result=BatchResult(row['result']),
defect_type=row.get('defect_type', ''),
severity=row.get('severity', 'minor')
)
self.quality_labels.append(label)
return self.quality_labels
4.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(key=lambda 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] = features
return all_features
def 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.0
return features
4.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.0
r = numerator / denominator
# t 统计量
if abs(r) >= 1.0:
return r, 0.0
t = r * np.sqrt((n - 2) / (1 - r**2))
# 近似 p 值 (双侧检验)
from scipy import stats
p = 2 * (1 - stats.t.cdf(abs(t), n - 2)) if n > 2 else 1.0
return 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)) + 1
rank_y = np.argsort(np.argsort(y)) + 1
# Pearson on ranks
r, p = self.pearson(rank_x.astype(float), rank_y.astype(float))
return r, p
def 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.alpha
results.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.alpha
results.append({
'feature_index': i,
'method': 'spearman',
'coefficient': round(rho, 4),
'p_value': round(p, 4),
'significant': significant,
'strength': self._interpret(rho)
})
return results
def _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, key=lambda x: abs(x['coefficient']), reverse=True):
fname = feature_names[r['feature_index']] if r['feature_index'] < len(feature_names) else f"F{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, 合格=0
label = 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 report
4.9 完整演示
def demo():
"""完整演示"""
print("=" * 65)
print(" 连续生产批次参数波动与次品率相关性分析系统 v1.0")
print(" 基于哈尔滨工程大学《工业过程控制》课程理论")
print("=" * 65)
import tempfile
import os
# 生成模拟工艺数据
np.random.seed(42)
process_lines = ["batch_id,timestamp,parameter,value,unit"]
parameters = ["temperature", "pressure"]
n_batches = 30
for b in range(1, n_batches + 1):
batch_id = f"BATCH_{b:03d}"
# 温度: 次品批次有更大的波动
if b % 5 == 0: # 每5批出一个次品
base_temp = 180.0
noise_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.02
process_lines.append(f"{batch_id},{t},temperature,{temp:.2f},℃")
# 压力: 次品批次有更大的斜率漂移
if b % 5 == 0:
slope = 0.5 # 漂移
else:
slope = 0.05
for 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 = f"BATCH_{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_config=FeatureConfig(),
correlation_config=CorrelationConfig(method='both', alpha=0.05)
)
report = system.run(process_csv, quality_csv)
print(report)
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！