企业官网建设流程全解析

数控连续加工尺寸漂移预测——用时序模型捕捉"刀具在偷偷磨钝"

一、实际应用场景（真实痛点）

时间：周五凌晨 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 加载尺寸时序 CSV，numpy 做滑动窗口，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.md

4.2 核心源码

<details><summary></summary>

"""数控加工尺寸时序数据模型 (dataclass)。"""

from dataclasses import dataclass, field

from typing import List, Optional, Dict

import numpy as np

@dataclass

class DimensionMeasurement:

"""单件尺寸测量记录。"""

piece_no: int

timestamp: str = ""

measured_size_mm: float = 0.0

nominal_size_mm: float = 0.0

spindle_speed_rpm: float = 0.0

feed_rate_mm_min: float = 0.0

cut_depth_mm: float = 0.0

@property

def deviation_mm(self) -> float:

"""尺寸偏差 = 实测 - 理论。"""

return self.measured_size_mm - self.nominal_size_mm

@dataclass

class DriftComponent:

"""漂移分量。"""

trend: np.ndarray = field(default_factory=lambda: np.array([])) # 线性趋势

thermal: np.ndarray = field(default_factory=lambda: np.array([])) # 热变形

step_change: np.ndarray = field(default_factory=lambda: np.array([])) # 阶跃

residual: np.ndarray = field(default_factory=lambda: np.array([])) # 残差

@property

def total(self) -> np.ndarray:

return self.trend + self.thermal + self.step_change + self.residual

@dataclass

class DriftPrediction:

"""漂移预测结果。"""

wear_rate_per_piece: float = 0.0 # 每件的磨损量 (mm/件)

predicted_deviations: np.ndarray = field(

default_factory=lambda: np.array([]))

exceed_piece_no: int = -1 # 预计超差件号 (-1=不超差)

exceed_probability: float = 0.0 # 超差概率

recommended_change_piece: int = -1 # 建议换刀件号

attribution: Dict[str, float] = field(default_factory=dict) # 归因

</details>

<details><summary></summary>

"""数控尺寸时序数据加载器 (pandas)。"""

import pandas as pd

import numpy as np

from pathlib import Path

from typing import List, Tuple

from .models import DimensionMeasurement

class DataLoader:

"""CSV 数据加载与预处理。"""

def load_csv(self, csv_path: str

) -> Tuple[np.ndarray, np.ndarray, List[DimensionMeasurement]]:

"""

加载 CSV。

返回: piece_nos, deviations, measurements

"""

df = 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(f"CSV 缺少列: {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_no=int(row["piece_no"]),

timestamp=str(row.get("timestamp", "")),

measured_size_mm=float(row["measured_size_mm"]),

nominal_size_mm=float(row.get("nominal_size_mm",

row["measured_size_mm"])),

spindle_speed_rpm=float(row.get("spindle_speed_rpm", 0)),

feed_rate_mm_min=float(row.get("feed_rate_mm_min", 0)),

cut_depth_mm=float(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, measurements

def 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_ok=True)

records = []

for i, pn in enumerate(piece_nos):

nominal = 50.0

measured = nominal + deviations[i]

records.append({

"piece_no": pn,

"timestamp": f"2026-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", index=False)

return piece_nos, deviations

</details>

<details><summary></summary>

"""尺寸漂移趋势分析与时序预测 (scipy + sklearn)。"""

import numpy as np

from typing import Dict, Tuple

from scipy import stats

from sklearn.svm import SVR

from sklearn.preprocessing import StandardScaler

from sklearn.pipeline import Pipeline

from sklearn.metrics import mean_absolute_error

from .models import DriftComponent, DriftPrediction

class TrendAnalyzer:

"""

尺寸漂移分析:

1. 线性回归 → 磨损速率

2. 指数平滑 → 热变形分量

3. SVR 时序预测 → 外推

4. ACF 自相关 → 时序特征

"""

def __init__(self, tolerance_mm: float = 0.030):

self.tolerance_mm = tolerance_mm

self.wear_rate_: float = 0.0

self.intercept_: float = 0.0

self.svr_pipeline_ = None

def 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_ = slope

self.intercept_ = intercept

comp.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.3

smoothed = [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.thermal

comp.step_change = np.zeros(n)

diff = np.abs(np.diff(residual_after_thermal))

threshold = np.std(diff) * 2.5

for 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 comp

def 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 = 3

X = []

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(kernel='rbf', C=100, gamma=0.1, epsilon=0.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_mm

if 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_attr

for k, v in prediction.attribution.items()

}

return prediction

def acf(self, deviations: np.ndarray,

max_lag: int = 20) -> np.ndarray:

"""自相关函数。"""

n = len(deviations)

mean = np.mean(deviations)

c0 = np.sum((deviations - mean) ** 2) / n

acf_vals = np.zeros(max_lag + 1)

acf_vals[0] = 1.0

for lag in range(1, max_lag + 1):

if lag >= n:

break

num = np.sum((deviations[:-lag] - mean) *

(deviations[lag:] - mean)) / (n - lag)

acf_vals[lag] = num / (c0 + 1e-8)

return acf_vals

</details>

<details><summary></summary>

"""异常/阶跃检测 (CUSUM + 滑动 t-test)。"""

import numpy as np

from typing import List, Tuple

class AnomalyDetector:

"""

检测尺寸时序中的均值漂移和异常点。

"""

def __init__(self, cusum_threshold: float = 5.0,

t_test_window: int = 10):

self.cusum_threshold = cusum_threshold

self.t_test_window = t_test_window

def cusum(self, deviations: np.ndarray) -> List[int]:

"""

CUSUM 累积和控制图。

返回检测到的漂移起始点索引列表。

"""

mean = np.mean(deviations)

std = np.std(deviations, ddof=1)

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_threshold

drift_starts = []

in_drift = False

for i, val in enumerate(cusum_pos):

if val > threshold and not in_drift:

drift_starts.append(i)

in_drift = True

elif val <= threshold:

in_drift = False

return drift_starts

def 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_window

n = 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

# Welch's t-test

mean1, mean2 = np.mean(before), np.mean(after)

var1 = np.var(before, ddof=1)

var2 = np.var(after, ddof=1)

denom = np.sqrt(var1/window_size + var2/window_size)

if denom == 0:

continue

t_stat = (mean1 - mean2) / denom

df = (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 stats

p_value = 2 * (1 - stats.t.cdf(abs(t_stat), df))

if p_value < 0.01:

change_points.append((i, p_value))

return change_points

</details>

<details><summary></summary>

"""尺寸漂移归因网络 (networkx)。"""

import networkx as nx

import numpy as np

from typing import Dict

class AttributionNetworkBuilder:

"""

构建误差源归因网络:

节点 = 误差源 + 尺寸偏差

边权重 = 方差贡献比例

"""

def __init__(self, attribution: Dict[str, float]):

self.attribution = attribution

self.graph_ = nx.DiGraph()

def build(self) -> nx.DiGraph:

"""构建网络。"""

self.graph_.clear()

# 添加源节点

for source, weight in self.attribution.items():

self.graph_.add_node(source, node_type="source",

weight=weight)

# 添加目标节点

self.graph_.add_node("尺寸漂移", node_type="target")

# 添加边

for source, weight in self.attribution.items():

self.graph_.add_edge(source, "尺寸漂移", weight=weight)

return self.graph_

</details>

<details><summary></summary>

"""尺寸漂移可视化 (matplotlib + networkx)。"""

import numpy as np

import matplotlib.pyplot as plt

import networkx as nx

from pathlib import Path

from typing import Dict

plt.rcParams["font.sans-serif"] = ["SimHei", "WenQuanYi Micro Hei", "DejaVu Sans"]

plt.rcParams["axes.unicode_minus"] = False

class DriftVisualizer:

def __init__(self, results_dir: str = "results"):

self.results_dir = Path(results_dir)

self.results_dir.mkdir(exist_ok=True)

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', markersize=4,

linewidth=1.5, label="实际偏差", alpha=0.8)

# 趋势线

if trend_line is not None:

ax.plot(piece_nos, trend_line, 'r--', linewidth=2,

label="磨损趋势线")

# 预测

if predicted_nos is not None and predicted_devs is not None:

ax.plot(predicted_nos, predicted_devs, 'g--o', markersize=4,

linewidth=1.5, label="预测偏差", alpha=0.7)

# 公差带

ax.axhline(y=tolerance_mm, color='orange', linestyle='-',

linewidth=1, alpha=0.7, label=f"+公差 {tolerance_mm}mm")

ax.axhline(y=-tolerance_mm, color='orange', linestyle='-',

linewidth=1, alpha=0.7, label=f"-公差 {tolerance_mm}mm")

ax.axhline(y=0, color='black', linewidth=0.5)

# 超差标注

if exceed_piece > 0:

ax.axvline(x=exceed_piece, color='red', linestyle=':',

linewidth=2, alpha=0.7,

label=f"预计超差: 第{exceed_piece}件")

ax.set_xlabel("件号", fontsize=12)

ax.set_ylabel("尺寸偏差 (mm)", fontsize=12)

ax.set_title("数控连续加工尺寸漂移趋势与预测",

fontsize=14, fontweight='bold')

ax.legend(fontsize=10)

ax.grid(alpha=0.3)

plt.tight_layout()

plt.savefig(self.results_dir / "drift_curve.png",

dpi=150, bbox_inches="tight")

plt.close()

def residual_hist(self, residuals: np.ndarray):

"""残差分布直方图。"""

fig, ax = plt.subplots(figsize=(7, 5))

ax.hist(residuals, bins=20

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！