数控连续加工尺寸漂移预测——用时序模型捕捉刀具在偷偷磨钝一、实际应用场景真实痛点时间周五凌晨 2:00夜班车间。地点某轴承套圈厂数控车削车间8 台数控车床沿墙一字排开切削液泵嗡嗡作响。质检员小方揉着眼睛坐在三坐标检测室里面前摊着一叠刚打出来的尺寸报告。又超差了。小方把第 47 件的检测单拍在桌上内孔直径前 20 件都在 φ50.012~50.018 之间稳稳的。第 21 件开始慢慢变大第 30 件到 φ50.025第 40 件 φ50.031第 47 件 φ50.038——图纸要求 φ50(0/0.03)第 40 件以后全超。刀具寿命不是设了 50 件换刀吗夜班班长问。设了但这是第二批料。小方翻记录第一批 50 件用的是正常棒料尺寸一直很稳。换第二批料后前 20 件也稳后面就开始漂。我怀疑是第二批料硬度偏高刀具磨损加速了——但问题是我怎么知道从第几件开始尺寸会超差难道每 10 件就全检一次你之前怎么处理的我问。首件检 每 10 件抽检 1 件。小方说但这批料从第 21 件开始漂移抽检碰巧第 20、30、40 件抽到了发现了趋势。如果抽检落在 22、25、28 呢就漏了。而且——就算我发现尺寸在变大我也只能停机换刀。能不能提前预测按当前磨损速度第几件会超差这样我可以在超差前主动换刀而不是等抽检发现再停机。你缺的不是抽检是趋势预测。我说你需要一个程序读取数控系统连续加工的尺寸检测时序数据件号、测量尺寸、时间戳建立时序模型预测后续 10~20 件的尺寸漂移趋势给出预计第 N 件超差的预警。这样你在第 25 件时就知道第 38 件要超提前安排换刀零报废。对小方一拍桌子最好还能区分是刀具磨损在主导还是热变形、还是料硬度波动因为如果是料的问题换刀没用得调刀补。这活儿我干过。我说pandas 加载尺寸时序 CSVnumpy 做滑动窗口scipy 做线性回归拟合磨损趋势线 指数平滑捕捉热变形分量scikit-learn 的 SVR支持向量回归做非线性时序预测用自相关性分析ACF判断时序特征matplotlib 画尺寸漂移曲线趋势预测带残差分析图networkx 建误差源→尺寸漂移归因网络。纯 Python不依赖任何商业 SPC 软件。二、痛点分析映射到滨州职业学院课程模型《先进制造技术》课程模块 本篇痛点对应数控加工与 CAD/CAM 技术 加工精度控制、刀具磨损与尺寸漂移、热变形补偿先进制造技术基础 机械制造工艺学工艺系统误差刀具磨损、热变形、力变形智能制造与数字孪生 数据驱动质量预测、虚拟量仪、数字孪生体状态估计柔性制造系统 FMS 与先进生产管理 统计过程控制SPC、预防性维护触发先进制造新模式 预测性质量控制、自适应加工一句话总结本篇解决一个给定数控连续加工尺寸时序数据 → 分解趋势/季节/残差分量 → 建立时序预测模型 → 外推后续尺寸漂移 → 超差预警 归因分析的时序预测问题映射到课程就是数控加工精度控制中的刀具磨损与热变形漂移建模。三、核心逻辑讲解大白话3.1 问题本质把尺寸漂移想成温水煮青蛙零件尺寸在连续加工中慢慢变化就像你泡在热水里感觉不到温度上升——等发现烫了已经晚了。尺寸漂移的元凶通常有三个- 刀具磨损主因每切一件刀尖就磨掉一丁点。积累到一定程度切深变浅孔就变大。这是线性漂移——匀速变大。- 热变形机床刚开机时冷加工一会儿热了主轴膨胀、导轨变形尺寸偏移。这是指数曲线——开始快、后来慢、最终稳定。- 材料硬度波动换料时硬度变了切削力变化导致弹性变形不同。这是阶跃变化——突然跳一下。我们的目标把这三个分量从尺寸数据里拆出来分别建模然后外推预测按当前趋势第几件会超差。3.2 业务逻辑 → 代码映射输入: 数控加工尺寸时序 CSV字段: 件号, 时间戳, 测量尺寸(mm), 理论尺寸(mm),主轴转速(rpm), 进给速度(mm/min), 切削深度(mm)│▼ models.py (dataclass)数据模型:DimensionMeasurement(单件尺寸测量)DriftComponent(漂移分量: 趋势/热变形/残差)DriftPrediction(预测结果)│▼ data_loader.py (pandas)数据加载器:读 CSV → 计算偏差量 → 时间索引化│▼ trend_analyzer.py趋势分析器 (scipy sklearn):1. 线性回归拟合刀具磨损趋势 (斜率 磨损速率)2. 指数平滑分解热变形分量3. SVR 时序预测 (RBF 核, 滑动窗口)4. 自相关分析 (ACF) 判断时序依赖性│▼ anomaly_detector.py异常/阶跃检测:1. CUSUM 累积和控制图 (检测均值漂移)2. 滑动 t-test (检测批次切换点)│▼ network_builder.py (networkx)归因网络:节点 误差源(刀具磨损/热变形/材料/其他)边 贡献比例│▼ visualizer.py (matplotlib)可视化:1. 尺寸漂移曲线 趋势线 预测带2. 残差分布直方图3. 热变形指数衰减曲线4. 超差预警标注│▼ 输出磨损速率 预计超差件号 归因比例 换刀建议3.3 为什么抽检 控制图不够用维度 传统 SPC 控制图 时序预测模型检测方式 事后发现已超控制限 事前预警预测 N 件后超差趋势判断 肉眼看走势 模型量化磨损速率多分量分解 无法分离 趋势热变形阶跃分别建模预警能力 已经超了 第 38 件会超现在换刀归因分析 无 区分刀具/热/材料四、OOP 代码实现4.1 项目结构cnc_drift_predictor/├── cnc_drift_predictor/│ ├── __init__.py│ ├── models.py # 数据模型│ ├── data_loader.py # CSV加载│ ├── trend_analyzer.py # 趋势分析与时序预测│ ├── anomaly_detector.py # 异常/阶跃检测│ ├── network_builder.py # 归因网络│ ├── visualizer.py # 可视化│ └── report_generator.py # 分析报告├── tests/│ ├── __init__.py│ └── test_drift.py├── sample_data/│ └── cnc_dimensions.csv├── results/│ ├── drift_curve.png│ ├── residual_hist.png│ ├── thermal_curve.png│ ├── attribution_network.png│ └── drift_report.txt├── run_drift.py└── README.md4.2 核心源码detailssummary/summary数控加工尺寸时序数据模型 (dataclass)。from dataclasses import dataclass, fieldfrom typing import List, Optional, Dictimport numpy as npdataclassclass DimensionMeasurement:单件尺寸测量记录。piece_no: inttimestamp: str measured_size_mm: float 0.0nominal_size_mm: float 0.0spindle_speed_rpm: float 0.0feed_rate_mm_min: float 0.0cut_depth_mm: float 0.0propertydef deviation_mm(self) - float:尺寸偏差 实测 - 理论。return self.measured_size_mm - self.nominal_size_mmdataclassclass DriftComponent:漂移分量。trend: np.ndarray field(default_factorylambda: np.array([])) # 线性趋势thermal: np.ndarray field(default_factorylambda: np.array([])) # 热变形step_change: np.ndarray field(default_factorylambda: np.array([])) # 阶跃residual: np.ndarray field(default_factorylambda: np.array([])) # 残差propertydef total(self) - np.ndarray:return self.trend self.thermal self.step_change self.residualdataclassclass DriftPrediction:漂移预测结果。wear_rate_per_piece: float 0.0 # 每件的磨损量 (mm/件)predicted_deviations: np.ndarray field(default_factorylambda: np.array([]))exceed_piece_no: int -1 # 预计超差件号 (-1不超差)exceed_probability: float 0.0 # 超差概率recommended_change_piece: int -1 # 建议换刀件号attribution: Dict[str, float] field(default_factorydict) # 归因/detailsdetailssummary/summary数控尺寸时序数据加载器 (pandas)。import pandas as pdimport numpy as npfrom pathlib import Pathfrom typing import List, Tuplefrom .models import DimensionMeasurementclass DataLoader:CSV 数据加载与预处理。def load_csv(self, csv_path: str) - Tuple[np.ndarray, np.ndarray, List[DimensionMeasurement]]:加载 CSV。返回: piece_nos, deviations, measurementsdf pd.read_csv(csv_path)required [piece_no, measured_size_mm]missing [c for c in required if c not in df.columns]if missing:raise ValueError(fCSV 缺少列: {missing})# 填充理论尺寸if nominal_size_mm not in df.columns:df[nominal_size_mm] df[measured_size_mm].median()df df.sort_values(piece_no)measurements []for _, row in df.iterrows():m DimensionMeasurement(piece_noint(row[piece_no]),timestampstr(row.get(timestamp, )),measured_size_mmfloat(row[measured_size_mm]),nominal_size_mmfloat(row.get(nominal_size_mm,row[measured_size_mm])),spindle_speed_rpmfloat(row.get(spindle_speed_rpm, 0)),feed_rate_mm_minfloat(row.get(feed_rate_mm_min, 0)),cut_depth_mmfloat(row.get(cut_depth_mm, 0)),)measurements.append(m)piece_nos np.array([m.piece_no for m in measurements])deviations np.array([m.deviation_mm for m in measurements])return piece_nos, deviations, measurementsdef generate_synthetic_data(self, n_pieces: int 60,change_point: int 25) - Tuple[np.ndarray, np.ndarray]:生成合成尺寸漂移数据 (教学用)。物理模型:偏差 刀具磨损(线性) 热变形(指数衰减) 材料阶跃 噪声np.random.seed(42)piece_nos np.arange(1, n_pieces 1)# 1. 刀具磨损 (线性, 约 0.0008 mm/件)wear np.linspace(0, 0.045, n_pieces)# 2. 热变形 (指数, 前10件影响大)thermal 0.012 * np.exp(-np.arange(n_pieces) / 8)# 3. 材料阶跃 (第 change_point 件后硬度偏高 → 尺寸偏大)material_step np.zeros(n_pieces)material_step[change_point - 1:] 0.015# 4. 噪声noise np.random.normal(0, 0.003, n_pieces)# 总偏差deviations wear thermal material_step noise# 保存sample_dir Path(sample_data)sample_dir.mkdir(exist_okTrue)records []for i, pn in enumerate(piece_nos):nominal 50.0measured nominal deviations[i]records.append({piece_no: pn,timestamp: f2026-10-09T{8 i//10:02d}:{i%60:02d}:00,measured_size_mm: measured,nominal_size_mm: nominal,spindle_speed_rpm: 2000 np.random.randint(-100, 100),feed_rate_mm_min: 150 np.random.randint(-10, 10),cut_depth_mm: 1.5,})df pd.DataFrame(records)df.to_csv(sample_dir / cnc_dimensions.csv, indexFalse)return piece_nos, deviations/detailsdetailssummary/summary尺寸漂移趋势分析与时序预测 (scipy sklearn)。import numpy as npfrom typing import Dict, Tuplefrom scipy import statsfrom sklearn.svm import SVRfrom sklearn.preprocessing import StandardScalerfrom sklearn.pipeline import Pipelinefrom sklearn.metrics import mean_absolute_errorfrom .models import DriftComponent, DriftPredictionclass TrendAnalyzer:尺寸漂移分析:1. 线性回归 → 磨损速率2. 指数平滑 → 热变形分量3. SVR 时序预测 → 外推4. ACF 自相关 → 时序特征def __init__(self, tolerance_mm: float 0.030):self.tolerance_mm tolerance_mmself.wear_rate_: float 0.0self.intercept_: float 0.0self.svr_pipeline_ Nonedef decompose(self, piece_nos: np.ndarray,deviations: np.ndarray) - DriftComponent:分解漂移分量。n len(deviations)comp DriftComponent()# 1. 线性趋势 (刀具磨损)slope, intercept, _, _, _ stats.linregress(piece_nos, deviations)self.wear_rate_ slopeself.intercept_ interceptcomp.trend slope * piece_nos intercept - np.mean(slope * piece_nos intercept)# 2. 残差 (去除趋势后)detrended deviations - (slope * piece_nos intercept)# 3. 热变形 (指数平滑残差的前部)comp.thermal np.zeros(n)alpha 0.3smoothed [detrended[0]]for i in range(1, n):smoothed.append(alpha * detrended[i] (1 - alpha) * smoothed[i-1])comp.thermal np.array(smoothed) * 0.4 # 取部分作为热变形# 4. 阶跃检测 (剩余残差的大跳变)residual_after_thermal detrended - comp.thermalcomp.step_change np.zeros(n)diff np.abs(np.diff(residual_after_thermal))threshold np.std(diff) * 2.5for i in range(1, len(diff)):if diff[i] threshold:comp.step_change[i:] residual_after_thermal[i] - residual_after_thermal[i-1]# 5. 最终残差comp.residual deviations - (comp.trend comp.thermal comp.step_change)return compdef predict_drift(self, piece_nos: np.ndarray,deviations: np.ndarray,forecast_pieces: int 20) - DriftPrediction:预测后续尺寸漂移。n len(deviations)prediction DriftPrediction()# 磨损速率prediction.wear_rate_per_piece self.wear_rate_# SVR 时序预测# 构造特征: [件号, 偏差_滞后1, 偏差_滞后2, 偏差_滞后3]max_lag 3X []y []for i in range(max_lag, n):features [piece_nos[i]]for lag in range(1, max_lag 1):features.append(deviations[i - lag])X.append(features)y.append(deviations[i])X np.array(X)y np.array(y)self.svr_pipeline_ Pipeline([(scaler, StandardScaler()),(svr, SVR(kernelrbf, C100, gamma0.1, epsilon0.001))])if len(X) 5:self.svr_pipeline_.fit(X, y)# 外推预测last_deviations list(deviations[-max_lag:])forecast_nos np.arange(n 1, n forecast_pieces 1)forecast_devs []for fn in forecast_nos:features [fn] last_deviations[-max_lag:]pred self.svr_pipeline_.predict([features])[0]forecast_devs.append(pred)last_deviations.append(pred)prediction.predicted_deviations np.array(forecast_devs)# 超差判断all_deviations np.concatenate([deviations, forecast_devs])all_nos np.concatenate([piece_nos, forecast_nos])exceed_mask np.abs(all_deviations) self.tolerance_mmif np.any(exceed_mask):first_exceed_idx np.where(exceed_mask)[0][0]prediction.exceed_piece_no int(all_nos[first_exceed_idx])# 超差概率 (基于预测偏差与公差的距离)margin self.tolerance_mm - abs(forecast_devs[0])prediction.exceed_probability max(0, min(1,1 - margin / self.tolerance_mm * 0.5))# 建议换刀件号 (在超差前 3 件)if prediction.exceed_piece_no 0:prediction.recommended_change_piece max(int(piece_nos[-1]),prediction.exceed_piece_no - 3)# 归因分析total_var np.var(deviations)if total_var 0:comp self.decompose(piece_nos, deviations)prediction.attribution {刀具磨损: min(1.0, np.var(comp.trend) / total_var),热变形: min(1.0, np.var(comp.thermal) / total_var),材料/阶跃: min(1.0, np.var(comp.step_change) / total_var),随机噪声: min(1.0, np.var(comp.residual) / total_var),}# 归一化total_attr sum(prediction.attribution.values())if total_attr 0:prediction.attribution {k: v / total_attrfor k, v in prediction.attribution.items()}return predictiondef acf(self, deviations: np.ndarray,max_lag: int 20) - np.ndarray:自相关函数。n len(deviations)mean np.mean(deviations)c0 np.sum((deviations - mean) ** 2) / nacf_vals np.zeros(max_lag 1)acf_vals[0] 1.0for lag in range(1, max_lag 1):if lag n:breaknum np.sum((deviations[:-lag] - mean) *(deviations[lag:] - mean)) / (n - lag)acf_vals[lag] num / (c0 1e-8)return acf_vals/detailsdetailssummary/summary异常/阶跃检测 (CUSUM 滑动 t-test)。import numpy as npfrom typing import List, Tupleclass AnomalyDetector:检测尺寸时序中的均值漂移和异常点。def __init__(self, cusum_threshold: float 5.0,t_test_window: int 10):self.cusum_threshold cusum_thresholdself.t_test_window t_test_windowdef cusum(self, deviations: np.ndarray) - List[int]:CUSUM 累积和控制图。返回检测到的漂移起始点索引列表。mean np.mean(deviations)std np.std(deviations, ddof1)if std 0:return []# 标准化z (deviations - mean) / (std 1e-8)# CUSUM (单方向, 检测正向漂移)cusum_pos np.zeros(len(z))for i in range(1, len(z)):cusum_pos[i] max(0, cusum_pos[i-1] z[i] - 0.5)# 检测超过阈值的位置threshold self.cusum_thresholddrift_starts []in_drift Falsefor i, val in enumerate(cusum_pos):if val threshold and not in_drift:drift_starts.append(i)in_drift Trueelif val threshold:in_drift Falsereturn drift_startsdef sliding_t_test(self, deviations: np.ndarray,window_size: int None) - List[Tuple[int, float]]:滑动 t-test 检测均值变化点。返回: [(变化点索引, p-value), ...]if window_size is None:window_size self.t_test_windown len(deviations)change_points []for i in range(window_size, n - window_size):before deviations[i - window_size:i]after deviations[i:i window_size]if len(before) 3 or len(after) 3:continue# Welchs t-testmean1, mean2 np.mean(before), np.mean(after)var1 np.var(before, ddof1)var2 np.var(after, ddof1)denom np.sqrt(var1/window_size var2/window_size)if denom 0:continuet_stat (mean1 - mean2) / denomdf (var1/window_size var2/window_size)**2 / ((var1/window_size)**2/(window_size-1) (var2/window_size)**2/(window_size-1) 1e-8)# 近似 p-value (双尾)from scipy import statsp_value 2 * (1 - stats.t.cdf(abs(t_stat), df))if p_value 0.01:change_points.append((i, p_value))return change_points/detailsdetailssummary/summary尺寸漂移归因网络 (networkx)。import networkx as nximport numpy as npfrom typing import Dictclass AttributionNetworkBuilder:构建误差源归因网络:节点 误差源 尺寸偏差边权重 方差贡献比例def __init__(self, attribution: Dict[str, float]):self.attribution attributionself.graph_ nx.DiGraph()def build(self) - nx.DiGraph:构建网络。self.graph_.clear()# 添加源节点for source, weight in self.attribution.items():self.graph_.add_node(source, node_typesource,weightweight)# 添加目标节点self.graph_.add_node(尺寸漂移, node_typetarget)# 添加边for source, weight in self.attribution.items():self.graph_.add_edge(source, 尺寸漂移, weightweight)return self.graph_/detailsdetailssummary/summary尺寸漂移可视化 (matplotlib networkx)。import numpy as npimport matplotlib.pyplot as pltimport networkx as nxfrom pathlib import Pathfrom typing import Dictplt.rcParams[font.sans-serif] [SimHei, WenQuanYi Micro Hei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass DriftVisualizer:def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)def drift_curve(self, piece_nos: np.ndarray,deviations: np.ndarray,trend_line: np.ndarray None,predicted_nos: np.ndarray None,predicted_devs: np.ndarray None,tolerance_mm: float 0.030,exceed_piece: int -1):尺寸漂移曲线 趋势 预测 超差标注。fig, ax plt.subplots(figsize(12, 6))# 实际偏差ax.plot(piece_nos, deviations, b-o, markersize4,linewidth1.5, label实际偏差, alpha0.8)# 趋势线if trend_line is not None:ax.plot(piece_nos, trend_line, r--, linewidth2,label磨损趋势线)# 预测if predicted_nos is not None and predicted_devs is not None:ax.plot(predicted_nos, predicted_devs, g--o, markersize4,linewidth1.5, label预测偏差, alpha0.7)# 公差带ax.axhline(ytolerance_mm, colororange, linestyle-,linewidth1, alpha0.7, labelf公差 {tolerance_mm}mm)ax.axhline(y-tolerance_mm, colororange, linestyle-,linewidth1, alpha0.7, labelf-公差 {tolerance_mm}mm)ax.axhline(y0, colorblack, linewidth0.5)# 超差标注if exceed_piece 0:ax.axvline(xexceed_piece, colorred, linestyle:,linewidth2, alpha0.7,labelf预计超差: 第{exceed_piece}件)ax.set_xlabel(件号, fontsize12)ax.set_ylabel(尺寸偏差 (mm), fontsize12)ax.set_title(数控连续加工尺寸漂移趋势与预测,fontsize14, fontweightbold)ax.legend(fontsize10)ax.grid(alpha0.3)plt.tight_layout()plt.savefig(self.results_dir / drift_curve.png,dpi150, bbox_inchestight)plt.close()def residual_hist(self, residuals: np.ndarray):残差分布直方图。fig, ax plt.subplots(figsize(7, 5))ax.hist(residuals, bins20利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
数控连续加工尺寸漂移预测——用时序模型捕捉刀具在偷偷磨钝一、实际应用场景真实痛点时间周五凌晨 2:00夜班车间。地点某轴承套圈厂数控车削车间8 台数控车床沿墙一字排开切削液泵嗡嗡作响。质检员小方揉着眼睛坐在三坐标检测室里面前摊着一叠刚打出来的尺寸报告。又超差了。小方把第 47 件的检测单拍在桌上内孔直径前 20 件都在 φ50.012~50.018 之间稳稳的。第 21 件开始慢慢变大第 30 件到 φ50.025第 40 件 φ50.031第 47 件 φ50.038——图纸要求 φ50(0/0.03)第 40 件以后全超。刀具寿命不是设了 50 件换刀吗夜班班长问。设了但这是第二批料。小方翻记录第一批 50 件用的是正常棒料尺寸一直很稳。换第二批料后前 20 件也稳后面就开始漂。我怀疑是第二批料硬度偏高刀具磨损加速了——但问题是我怎么知道从第几件开始尺寸会超差难道每 10 件就全检一次你之前怎么处理的我问。首件检 每 10 件抽检 1 件。小方说但这批料从第 21 件开始漂移抽检碰巧第 20、30、40 件抽到了发现了趋势。如果抽检落在 22、25、28 呢就漏了。而且——就算我发现尺寸在变大我也只能停机换刀。能不能提前预测按当前磨损速度第几件会超差这样我可以在超差前主动换刀而不是等抽检发现再停机。你缺的不是抽检是趋势预测。我说你需要一个程序读取数控系统连续加工的尺寸检测时序数据件号、测量尺寸、时间戳建立时序模型预测后续 10~20 件的尺寸漂移趋势给出预计第 N 件超差的预警。这样你在第 25 件时就知道第 38 件要超提前安排换刀零报废。对小方一拍桌子最好还能区分是刀具磨损在主导还是热变形、还是料硬度波动因为如果是料的问题换刀没用得调刀补。这活儿我干过。我说pandas 加载尺寸时序 CSVnumpy 做滑动窗口scipy 做线性回归拟合磨损趋势线 指数平滑捕捉热变形分量scikit-learn 的 SVR支持向量回归做非线性时序预测用自相关性分析ACF判断时序特征matplotlib 画尺寸漂移曲线趋势预测带残差分析图networkx 建误差源→尺寸漂移归因网络。纯 Python不依赖任何商业 SPC 软件。二、痛点分析映射到滨州职业学院课程模型《先进制造技术》课程模块 本篇痛点对应数控加工与 CAD/CAM 技术 加工精度控制、刀具磨损与尺寸漂移、热变形补偿先进制造技术基础 机械制造工艺学工艺系统误差刀具磨损、热变形、力变形智能制造与数字孪生 数据驱动质量预测、虚拟量仪、数字孪生体状态估计柔性制造系统 FMS 与先进生产管理 统计过程控制SPC、预防性维护触发先进制造新模式 预测性质量控制、自适应加工一句话总结本篇解决一个给定数控连续加工尺寸时序数据 → 分解趋势/季节/残差分量 → 建立时序预测模型 → 外推后续尺寸漂移 → 超差预警 归因分析的时序预测问题映射到课程就是数控加工精度控制中的刀具磨损与热变形漂移建模。三、核心逻辑讲解大白话3.1 问题本质把尺寸漂移想成温水煮青蛙零件尺寸在连续加工中慢慢变化就像你泡在热水里感觉不到温度上升——等发现烫了已经晚了。尺寸漂移的元凶通常有三个- 刀具磨损主因每切一件刀尖就磨掉一丁点。积累到一定程度切深变浅孔就变大。这是线性漂移——匀速变大。- 热变形机床刚开机时冷加工一会儿热了主轴膨胀、导轨变形尺寸偏移。这是指数曲线——开始快、后来慢、最终稳定。- 材料硬度波动换料时硬度变了切削力变化导致弹性变形不同。这是阶跃变化——突然跳一下。我们的目标把这三个分量从尺寸数据里拆出来分别建模然后外推预测按当前趋势第几件会超差。3.2 业务逻辑 → 代码映射输入: 数控加工尺寸时序 CSV字段: 件号, 时间戳, 测量尺寸(mm), 理论尺寸(mm),主轴转速(rpm), 进给速度(mm/min), 切削深度(mm)│▼ models.py (dataclass)数据模型:DimensionMeasurement(单件尺寸测量)DriftComponent(漂移分量: 趋势/热变形/残差)DriftPrediction(预测结果)│▼ data_loader.py (pandas)数据加载器:读 CSV → 计算偏差量 → 时间索引化│▼ trend_analyzer.py趋势分析器 (scipy sklearn):1. 线性回归拟合刀具磨损趋势 (斜率 磨损速率)2. 指数平滑分解热变形分量3. SVR 时序预测 (RBF 核, 滑动窗口)4. 自相关分析 (ACF) 判断时序依赖性│▼ anomaly_detector.py异常/阶跃检测:1. CUSUM 累积和控制图 (检测均值漂移)2. 滑动 t-test (检测批次切换点)│▼ network_builder.py (networkx)归因网络:节点 误差源(刀具磨损/热变形/材料/其他)边 贡献比例│▼ visualizer.py (matplotlib)可视化:1. 尺寸漂移曲线 趋势线 预测带2. 残差分布直方图3. 热变形指数衰减曲线4. 超差预警标注│▼ 输出磨损速率 预计超差件号 归因比例 换刀建议3.3 为什么抽检 控制图不够用维度 传统 SPC 控制图 时序预测模型检测方式 事后发现已超控制限 事前预警预测 N 件后超差趋势判断 肉眼看走势 模型量化磨损速率多分量分解 无法分离 趋势热变形阶跃分别建模预警能力 已经超了 第 38 件会超现在换刀归因分析 无 区分刀具/热/材料四、OOP 代码实现4.1 项目结构cnc_drift_predictor/├── cnc_drift_predictor/│ ├── __init__.py│ ├── models.py # 数据模型│ ├── data_loader.py # CSV加载│ ├── trend_analyzer.py # 趋势分析与时序预测│ ├── anomaly_detector.py # 异常/阶跃检测│ ├── network_builder.py # 归因网络│ ├── visualizer.py # 可视化│ └── report_generator.py # 分析报告├── tests/│ ├── __init__.py│ └── test_drift.py├── sample_data/│ └── cnc_dimensions.csv├── results/│ ├── drift_curve.png│ ├── residual_hist.png│ ├── thermal_curve.png│ ├── attribution_network.png│ └── drift_report.txt├── run_drift.py└── README.md4.2 核心源码detailssummary/summary数控加工尺寸时序数据模型 (dataclass)。from dataclasses import dataclass, fieldfrom typing import List, Optional, Dictimport numpy as npdataclassclass DimensionMeasurement:单件尺寸测量记录。piece_no: inttimestamp: str measured_size_mm: float 0.0nominal_size_mm: float 0.0spindle_speed_rpm: float 0.0feed_rate_mm_min: float 0.0cut_depth_mm: float 0.0propertydef deviation_mm(self) - float:尺寸偏差 实测 - 理论。return self.measured_size_mm - self.nominal_size_mmdataclassclass DriftComponent:漂移分量。trend: np.ndarray field(default_factorylambda: np.array([])) # 线性趋势thermal: np.ndarray field(default_factorylambda: np.array([])) # 热变形step_change: np.ndarray field(default_factorylambda: np.array([])) # 阶跃residual: np.ndarray field(default_factorylambda: np.array([])) # 残差propertydef total(self) - np.ndarray:return self.trend self.thermal self.step_change self.residualdataclassclass DriftPrediction:漂移预测结果。wear_rate_per_piece: float 0.0 # 每件的磨损量 (mm/件)predicted_deviations: np.ndarray field(default_factorylambda: np.array([]))exceed_piece_no: int -1 # 预计超差件号 (-1不超差)exceed_probability: float 0.0 # 超差概率recommended_change_piece: int -1 # 建议换刀件号attribution: Dict[str, float] field(default_factorydict) # 归因/detailsdetailssummary/summary数控尺寸时序数据加载器 (pandas)。import pandas as pdimport numpy as npfrom pathlib import Pathfrom typing import List, Tuplefrom .models import DimensionMeasurementclass DataLoader:CSV 数据加载与预处理。def load_csv(self, csv_path: str) - Tuple[np.ndarray, np.ndarray, List[DimensionMeasurement]]:加载 CSV。返回: piece_nos, deviations, measurementsdf pd.read_csv(csv_path)required [piece_no, measured_size_mm]missing [c for c in required if c not in df.columns]if missing:raise ValueError(fCSV 缺少列: {missing})# 填充理论尺寸if nominal_size_mm not in df.columns:df[nominal_size_mm] df[measured_size_mm].median()df df.sort_values(piece_no)measurements []for _, row in df.iterrows():m DimensionMeasurement(piece_noint(row[piece_no]),timestampstr(row.get(timestamp, )),measured_size_mmfloat(row[measured_size_mm]),nominal_size_mmfloat(row.get(nominal_size_mm,row[measured_size_mm])),spindle_speed_rpmfloat(row.get(spindle_speed_rpm, 0)),feed_rate_mm_minfloat(row.get(feed_rate_mm_min, 0)),cut_depth_mmfloat(row.get(cut_depth_mm, 0)),)measurements.append(m)piece_nos np.array([m.piece_no for m in measurements])deviations np.array([m.deviation_mm for m in measurements])return piece_nos, deviations, measurementsdef generate_synthetic_data(self, n_pieces: int 60,change_point: int 25) - Tuple[np.ndarray, np.ndarray]:生成合成尺寸漂移数据 (教学用)。物理模型:偏差 刀具磨损(线性) 热变形(指数衰减) 材料阶跃 噪声np.random.seed(42)piece_nos np.arange(1, n_pieces 1)# 1. 刀具磨损 (线性, 约 0.0008 mm/件)wear np.linspace(0, 0.045, n_pieces)# 2. 热变形 (指数, 前10件影响大)thermal 0.012 * np.exp(-np.arange(n_pieces) / 8)# 3. 材料阶跃 (第 change_point 件后硬度偏高 → 尺寸偏大)material_step np.zeros(n_pieces)material_step[change_point - 1:] 0.015# 4. 噪声noise np.random.normal(0, 0.003, n_pieces)# 总偏差deviations wear thermal material_step noise# 保存sample_dir Path(sample_data)sample_dir.mkdir(exist_okTrue)records []for i, pn in enumerate(piece_nos):nominal 50.0measured nominal deviations[i]records.append({piece_no: pn,timestamp: f2026-10-09T{8 i//10:02d}:{i%60:02d}:00,measured_size_mm: measured,nominal_size_mm: nominal,spindle_speed_rpm: 2000 np.random.randint(-100, 100),feed_rate_mm_min: 150 np.random.randint(-10, 10),cut_depth_mm: 1.5,})df pd.DataFrame(records)df.to_csv(sample_dir / cnc_dimensions.csv, indexFalse)return piece_nos, deviations/detailsdetailssummary/summary尺寸漂移趋势分析与时序预测 (scipy sklearn)。import numpy as npfrom typing import Dict, Tuplefrom scipy import statsfrom sklearn.svm import SVRfrom sklearn.preprocessing import StandardScalerfrom sklearn.pipeline import Pipelinefrom sklearn.metrics import mean_absolute_errorfrom .models import DriftComponent, DriftPredictionclass TrendAnalyzer:尺寸漂移分析:1. 线性回归 → 磨损速率2. 指数平滑 → 热变形分量3. SVR 时序预测 → 外推4. ACF 自相关 → 时序特征def __init__(self, tolerance_mm: float 0.030):self.tolerance_mm tolerance_mmself.wear_rate_: float 0.0self.intercept_: float 0.0self.svr_pipeline_ Nonedef decompose(self, piece_nos: np.ndarray,deviations: np.ndarray) - DriftComponent:分解漂移分量。n len(deviations)comp DriftComponent()# 1. 线性趋势 (刀具磨损)slope, intercept, _, _, _ stats.linregress(piece_nos, deviations)self.wear_rate_ slopeself.intercept_ interceptcomp.trend slope * piece_nos intercept - np.mean(slope * piece_nos intercept)# 2. 残差 (去除趋势后)detrended deviations - (slope * piece_nos intercept)# 3. 热变形 (指数平滑残差的前部)comp.thermal np.zeros(n)alpha 0.3smoothed [detrended[0]]for i in range(1, n):smoothed.append(alpha * detrended[i] (1 - alpha) * smoothed[i-1])comp.thermal np.array(smoothed) * 0.4 # 取部分作为热变形# 4. 阶跃检测 (剩余残差的大跳变)residual_after_thermal detrended - comp.thermalcomp.step_change np.zeros(n)diff np.abs(np.diff(residual_after_thermal))threshold np.std(diff) * 2.5for i in range(1, len(diff)):if diff[i] threshold:comp.step_change[i:] residual_after_thermal[i] - residual_after_thermal[i-1]# 5. 最终残差comp.residual deviations - (comp.trend comp.thermal comp.step_change)return compdef predict_drift(self, piece_nos: np.ndarray,deviations: np.ndarray,forecast_pieces: int 20) - DriftPrediction:预测后续尺寸漂移。n len(deviations)prediction DriftPrediction()# 磨损速率prediction.wear_rate_per_piece self.wear_rate_# SVR 时序预测# 构造特征: [件号, 偏差_滞后1, 偏差_滞后2, 偏差_滞后3]max_lag 3X []y []for i in range(max_lag, n):features [piece_nos[i]]for lag in range(1, max_lag 1):features.append(deviations[i - lag])X.append(features)y.append(deviations[i])X np.array(X)y np.array(y)self.svr_pipeline_ Pipeline([(scaler, StandardScaler()),(svr, SVR(kernelrbf, C100, gamma0.1, epsilon0.001))])if len(X) 5:self.svr_pipeline_.fit(X, y)# 外推预测last_deviations list(deviations[-max_lag:])forecast_nos np.arange(n 1, n forecast_pieces 1)forecast_devs []for fn in forecast_nos:features [fn] last_deviations[-max_lag:]pred self.svr_pipeline_.predict([features])[0]forecast_devs.append(pred)last_deviations.append(pred)prediction.predicted_deviations np.array(forecast_devs)# 超差判断all_deviations np.concatenate([deviations, forecast_devs])all_nos np.concatenate([piece_nos, forecast_nos])exceed_mask np.abs(all_deviations) self.tolerance_mmif np.any(exceed_mask):first_exceed_idx np.where(exceed_mask)[0][0]prediction.exceed_piece_no int(all_nos[first_exceed_idx])# 超差概率 (基于预测偏差与公差的距离)margin self.tolerance_mm - abs(forecast_devs[0])prediction.exceed_probability max(0, min(1,1 - margin / self.tolerance_mm * 0.5))# 建议换刀件号 (在超差前 3 件)if prediction.exceed_piece_no 0:prediction.recommended_change_piece max(int(piece_nos[-1]),prediction.exceed_piece_no - 3)# 归因分析total_var np.var(deviations)if total_var 0:comp self.decompose(piece_nos, deviations)prediction.attribution {刀具磨损: min(1.0, np.var(comp.trend) / total_var),热变形: min(1.0, np.var(comp.thermal) / total_var),材料/阶跃: min(1.0, np.var(comp.step_change) / total_var),随机噪声: min(1.0, np.var(comp.residual) / total_var),}# 归一化total_attr sum(prediction.attribution.values())if total_attr 0:prediction.attribution {k: v / total_attrfor k, v in prediction.attribution.items()}return predictiondef acf(self, deviations: np.ndarray,max_lag: int 20) - np.ndarray:自相关函数。n len(deviations)mean np.mean(deviations)c0 np.sum((deviations - mean) ** 2) / nacf_vals np.zeros(max_lag 1)acf_vals[0] 1.0for lag in range(1, max_lag 1):if lag n:breaknum np.sum((deviations[:-lag] - mean) *(deviations[lag:] - mean)) / (n - lag)acf_vals[lag] num / (c0 1e-8)return acf_vals/detailsdetailssummary/summary异常/阶跃检测 (CUSUM 滑动 t-test)。import numpy as npfrom typing import List, Tupleclass AnomalyDetector:检测尺寸时序中的均值漂移和异常点。def __init__(self, cusum_threshold: float 5.0,t_test_window: int 10):self.cusum_threshold cusum_thresholdself.t_test_window t_test_windowdef cusum(self, deviations: np.ndarray) - List[int]:CUSUM 累积和控制图。返回检测到的漂移起始点索引列表。mean np.mean(deviations)std np.std(deviations, ddof1)if std 0:return []# 标准化z (deviations - mean) / (std 1e-8)# CUSUM (单方向, 检测正向漂移)cusum_pos np.zeros(len(z))for i in range(1, len(z)):cusum_pos[i] max(0, cusum_pos[i-1] z[i] - 0.5)# 检测超过阈值的位置threshold self.cusum_thresholddrift_starts []in_drift Falsefor i, val in enumerate(cusum_pos):if val threshold and not in_drift:drift_starts.append(i)in_drift Trueelif val threshold:in_drift Falsereturn drift_startsdef sliding_t_test(self, deviations: np.ndarray,window_size: int None) - List[Tuple[int, float]]:滑动 t-test 检测均值变化点。返回: [(变化点索引, p-value), ...]if window_size is None:window_size self.t_test_windown len(deviations)change_points []for i in range(window_size, n - window_size):before deviations[i - window_size:i]after deviations[i:i window_size]if len(before) 3 or len(after) 3:continue# Welchs t-testmean1, mean2 np.mean(before), np.mean(after)var1 np.var(before, ddof1)var2 np.var(after, ddof1)denom np.sqrt(var1/window_size var2/window_size)if denom 0:continuet_stat (mean1 - mean2) / denomdf (var1/window_size var2/window_size)**2 / ((var1/window_size)**2/(window_size-1) (var2/window_size)**2/(window_size-1) 1e-8)# 近似 p-value (双尾)from scipy import statsp_value 2 * (1 - stats.t.cdf(abs(t_stat), df))if p_value 0.01:change_points.append((i, p_value))return change_points/detailsdetailssummary/summary尺寸漂移归因网络 (networkx)。import networkx as nximport numpy as npfrom typing import Dictclass AttributionNetworkBuilder:构建误差源归因网络:节点 误差源 尺寸偏差边权重 方差贡献比例def __init__(self, attribution: Dict[str, float]):self.attribution attributionself.graph_ nx.DiGraph()def build(self) - nx.DiGraph:构建网络。self.graph_.clear()# 添加源节点for source, weight in self.attribution.items():self.graph_.add_node(source, node_typesource,weightweight)# 添加目标节点self.graph_.add_node(尺寸漂移, node_typetarget)# 添加边for source, weight in self.attribution.items():self.graph_.add_edge(source, 尺寸漂移, weightweight)return self.graph_/detailsdetailssummary/summary尺寸漂移可视化 (matplotlib networkx)。import numpy as npimport matplotlib.pyplot as pltimport networkx as nxfrom pathlib import Pathfrom typing import Dictplt.rcParams[font.sans-serif] [SimHei, WenQuanYi Micro Hei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass DriftVisualizer:def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)def drift_curve(self, piece_nos: np.ndarray,deviations: np.ndarray,trend_line: np.ndarray None,predicted_nos: np.ndarray None,predicted_devs: np.ndarray None,tolerance_mm: float 0.030,exceed_piece: int -1):尺寸漂移曲线 趋势 预测 超差标注。fig, ax plt.subplots(figsize(12, 6))# 实际偏差ax.plot(piece_nos, deviations, b-o, markersize4,linewidth1.5, label实际偏差, alpha0.8)# 趋势线if trend_line is not None:ax.plot(piece_nos, trend_line, r--, linewidth2,label磨损趋势线)# 预测if predicted_nos is not None and predicted_devs is not None:ax.plot(predicted_nos, predicted_devs, g--o, markersize4,linewidth1.5, label预测偏差, alpha0.7)# 公差带ax.axhline(ytolerance_mm, colororange, linestyle-,linewidth1, alpha0.7, labelf公差 {tolerance_mm}mm)ax.axhline(y-tolerance_mm, colororange, linestyle-,linewidth1, alpha0.7, labelf-公差 {tolerance_mm}mm)ax.axhline(y0, colorblack, linewidth0.5)# 超差标注if exceed_piece 0:ax.axvline(xexceed_piece, colorred, linestyle:,linewidth2, alpha0.7,labelf预计超差: 第{exceed_piece}件)ax.set_xlabel(件号, fontsize12)ax.set_ylabel(尺寸偏差 (mm), fontsize12)ax.set_title(数控连续加工尺寸漂移趋势与预测,fontsize14, fontweightbold)ax.legend(fontsize10)ax.grid(alpha0.3)plt.tight_layout()plt.savefig(self.results_dir / drift_curve.png,dpi150, bbox_inchestight)plt.close()def residual_hist(self, residuals: np.ndarray):残差分布直方图。fig, ax plt.subplots(figsize(7, 5))ax.hist(residuals, bins20利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
