1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
class SPCControlChart:
"""统计过程控制图"""
# 控制图系数表 (A2, D3, D4)
CONSTANTS = {
2: (1.880, 0, 3.267),
3: (1.023, 0, 2.574),
4: (0.729, 0, 2.282),
5: (0.577, 0, 2.114),
6: (0.483, 0, 2.004),
7: (0.419, 0.076, 1.924),
8: (0.373, 0.136, 1.864),
9: (0.337, 0.184, 1.816),
10: (0.308, 0.223, 1.777),
}
def __init__(self, subgroup_size: int = 5):
self.n = subgroup_size
self.A2, self.D3, self.D4 = self.CONSTANTS.get(
subgroup_size, (0.308, 0.223, 1.777)
)
def xbar_r_chart(self, data: pl.DataFrame,
value_col: str,
subgroup_col: str) -> dict:
"""
X-bar / R 控制图
data: 包含子组标识和度量值的数据框
"""
# 计算每个子组的均值和极差
subgroups = (
data.group_by(subgroup_col)
.agg([
pl.col(value_col).mean().alias("x_bar"),
(pl.col(value_col).max() - pl.col(value_col).min()).alias("R"),
])
.sort(subgroup_col)
)
x_bar_bar = subgroups["x_bar"].mean() # 总均值
r_bar = subgroups["R"].mean() # 平均极差
# X-bar 图控制限
ucl_x = x_bar_bar + self.A2 * r_bar
lcl_x = x_bar_bar - self.A2 * r_bar
cl_x = x_bar_bar
# R 图控制限
ucl_r = self.D4 * r_bar
lcl_r = self.D3 * r_bar
cl_r = r_bar
# 判异规则检查
violations = self._detect_violations(subgroups["x_bar"].to_numpy(),
cl_x, ucl_x, lcl_x)
return {
"x_bar_chart": {
"center_line": round(cl_x, 3),
"ucl": round(ucl_x, 3),
"lcl": round(lcl_x, 3),
"subgroup_means": subgroups["x_bar"].to_list(),
},
"r_chart": {
"center_line": round(cl_r, 3),
"ucl": round(ucl_r, 3),
"lcl": round(lcl_r, 3),
"subgroup_ranges": subgroups["R"].to_list(),
},
"violations": violations,
"process_capable": len(violations) == 0,
}
def _detect_violations(self, values: np.ndarray,
cl: float, ucl: float, lcl: float) -> list:
"""
SPC 判异规则（Western Electric Rules）
"""
violations = []
n = len(values)
for i in range(n):
# 规则1: 点出控制限
if values[i] > ucl or values[i] < lcl:
violations.append({
"rule": "超出控制限",
"index": i,
"value": float(values[i]),
})
# 规则2: 连续9点在中心线同一侧
for i in range(8, n):
window = values[i-8:i+1]
if all(v > cl for v in window) or all(v < cl for v in window):
violations.append({
"rule": "连续9点在中心线同侧",
"index": i,
"value": float(values[i]),
})
# 规则3: 连续6点递增或递减
for i in range(5, n):
window = values[i-5:i+1]
diffs = np.diff(window)
if all(d > 0 for d in diffs) or all(d < 0 for d in diffs):
violations.append({
"rule": "连续6点单调变化",
"index": i,
"value": float(values[i]),
})
# 规则4: 连续14点交替上下
for i in range(13, n):
window = values[i-13:i+1]
diffs = np.diff(window)
signs = np.sign(diffs)
alternating = all(
signs[j] * signs[j+1] < 0 for j in range(len(signs)-1)
)
if alternating:
violations.append({
"rule": "连续14点交替变化",
"index": i,
"value": float(values[i]),
})
return violations
|
