Code - NumPy实战（工业数值计算）

第2周 NumPy实战（工业数值计算）详细学习指南

一、知识点讲解

本部分围绕NumPy核心知识点拆解，聚焦工业数值计算常用功能，涵盖数组高效运算、数据筛选与变形、统计分析、数据归一化四大模块，每个知识点配套用法、参数、示例及陷阱提醒。

（一）数组高效运算

核心知识点1：NumPy数组创建与基础运算

「作用说明」：NumPy数组（ndarray）是高效数值计算的核心，相比Python列表，支持向量化运算，可大幅提升工业数据处理速度；基础运算包括元素级运算、广播机制，适用于批量设备数据计算。

核心参数详解

参数	说明	常用可选值及说明	传参示例
shape	指定数组的维度和各维度大小，适用于reshape、array创建时	元组（m,n）：m行n列；如（500,8）表示500台设备，每台8个指标	np.array(data, shape=(500,8))、arr.reshape((500,8))
dtype	指定数组数据类型，影响存储效率和计算精度	float32（单精度浮点）、float64（双精度浮点，默认）、int32	np.zeros((500,8), dtype=np.float32)
axis	指定运算轴（维度），用于聚合运算（如求和、均值）	0（按列运算）、1（按行运算）、None（整体运算）	arr.sum(axis=1)（计算每台设备指标总和）

典型用法示例

import numpy as np

# 1. 创建500台设备8项指标的基础数据（符合工业规律：正态分布，均值100，标准差10）
device_data = np.random.normal(loc=100, scale=10, size=(500, 8), dtype=np.float64)
print("设备数据形状：", device_data.shape)  # 输出 (500,8)

# 2. 元素级运算（替代Python循环，计算每台设备指标与标准值100的偏差）
deviation = device_data - 100  # 向量化运算，无需循环
print("偏差数据前2行：\n", deviation[:2])

# 3. 按行聚合运算（计算每台设备的平均指标）
avg_per_device = device_data.mean(axis=1)
print("前5台设备平均指标：", avg_per_device[:5])

常见陷阱提醒

• 避免混合使用NumPy数组与Python列表运算：列表会触发广播机制，导致运算速度下降，需先将列表转为ndarray。
• 轴参数混淆：axis=0是“垂直方向”按列运算，axis=1是“水平方向”按行运算，工业数据中常按行（单台设备）计算统计值，勿用反。
• 数据类型溢出：工业数据若范围较大（如设备运行时长），避免用int32，建议用int64或float64，防止溢出失真。

（二）数据筛选与变形

核心知识点2：布尔索引筛选与数组变形

「作用说明」：布尔索引可精准筛选异常设备数据，数组变形（reshape、flatten）可调整数据结构，适配后续分析或模型输入，是工业数据清洗的核心操作。

核心参数详解

参数	说明	常用可选值及说明	传参示例
order	reshape时的数组存储顺序，影响数据读取效率	C（行优先，默认）、F（列优先）	arr.reshape((100,40), order='C')
keepdims	筛选后是否保留原数组维度，避免降维导致后续运算错误	True（保留维度）、False（默认，自动降维）	abnormal_data = device_data[mask, keepdims=True]
newaxis	手动增加数组维度，用于维度匹配（广播运算）	无可选值，作为索引使用	avg_per_device[:, np.newaxis]（将一维数组转为二维）

典型用法示例

import numpy as np

# 基于前文设备数据，模拟异常筛选（指标超出[80,120]为异常）
device_data = np.random.normal(loc=100, scale=10, size=(500, 8))

# 1. 布尔索引筛选异常数据（单条件+多条件组合）
# 单条件：任意一项指标>120为异常
abnormal_mask1 = (device_data > 120).any(axis=1)  # 按行判断，是否有异常指标
# 多条件：任意一项指标<80 或 >120为异常（注意用&和|，且加括号）
abnormal_mask2 = (device_data < 80).any(axis=1) | (device_data > 120).any(axis=1)

# 筛选异常设备数据（保留原维度）
abnormal_devices = device_data[abnormal_mask2, keepdims=True]
print(f"异常设备数量：{abnormal_devices.shape[0]}")

# 2. 数组变形：将异常设备数据从（n,1,8）转为（n,8）
abnormal_devices = abnormal_devices.reshape(-1, 8)  # -1表示自动计算行数
print("变形后异常设备数据形状：", abnormal_devices.shape)

# 3. 筛选正常设备数据（取反掩码）
normal_devices = device_data[~abnormal_mask2]

常见陷阱提醒

• 布尔索引与逻辑运算符：用&表示“且”、|表示“或”，不可用and/or；每个条件需加括号，否则会因运算优先级出错。
• reshape维度匹配：变形后总元素数需与原数组一致，如（500,8）可转为（100,40），但不可转为（100,39），否则报错。
• 降维风险：筛选后数组默认降维（如从二维变一维），若后续需按行运算，需用keepdims=True保留维度。

（三）统计分析

核心知识点3：工业数据常用统计方法

「作用说明」：NumPy统计函数可快速计算设备数据的均值、标准差、异常率等指标，为工业设备状态评估提供依据，运算速度远超Python循环。

核心参数详解

参数	说明	常用可选值及说明	传参示例
ddof	标准差、方差计算中自由度修正，影响结果精度	0（总体标准差，默认）、1（样本标准差）	np.std(device_data, ddof=1, axis=1)
percentile	分位数计算，用于异常阈值确定（如95分位数）	0~100的整数，代表分位比例	np.percentile(device_data, 95, axis=0)（按列算95分位数）
out	指定结果存储数组，节省内存（大规模数据适用）	已创建的空ndarray	result = np.empty((500,))；np.mean(device_data, axis=1, out=result)

典型用法示例

import numpy as np

device_data = np.random.normal(loc=100, scale=10, size=(500, 8))
abnormal_mask = (device_data < 80).any(axis=1) | (device_data > 120).any(axis=1)

# 1. 计算异常率（异常设备数/总设备数）
abnormal_count = np.sum(abnormal_mask)
abnormal_rate = abnormal_count / len(device_data)
print(f"设备异常率：{abnormal_rate:.2%}")

# 2. 按列统计（各指标的均值、标准差、95分位数）
index_mean = np.mean(device_data, axis=0)
index_std = np.std(device_data, axis=0, ddof=1)  # 样本标准差
index_95 = np.percentile(device_data, 95, axis=0)

# 合并统计结果（每行对应一个指标的统计值）
stats_result = np.vstack([index_mean, index_std, index_95])
print("各指标统计结果（均值/标准差/95分位数）：\n", stats_result[:,:3])  # 展示前3个指标

# 3. 计算每台设备的指标波动系数（标准差/均值）
device_std = np.std(device_data, axis=1, ddof=1)
device_avg = np.mean(device_data, axis=1)
fluctuation_coeff = device_std / device_avg
print("前5台设备波动系数：", fluctuation_coeff[:5])

常见陷阱提醒

• 自由度ddof误用：工业数据若为抽样数据（如部分设备），需用ddof=1计算样本标准差；若为全量数据（500台全量），用默认ddof=0即可。
• 分位数轴参数错误：按指标（列）计算分位数时用axis=0，按设备（行）计算用axis=1，勿混淆导致阈值计算错误。
• 浮点数精度问题：统计结果若需保留小数，建议用np.round()处理，避免因浮点数误差影响异常判断。

（四）数据归一化

核心知识点4：Min-Max归一化（工业常用）

「作用说明」：将设备数据映射到[0,1]区间，消除指标量纲影响（如温度、压力单位不同），使数据可直接输入大模型或机器学习模型，且保证数据无失真。

核心参数详解

参数	说明	常用可选值及说明	传参示例
axis	指定归一化轴，按列（指标）归一化符合工业习惯	0（按列归一化，默认）、1（按行归一化）	np.min(device_data, axis=0)（计算各指标最小值）
keepdims	归一化时保留维度，确保广播运算正常	True（推荐）、False（默认）	min_val = np.min(device_data, axis=0, keepdims=True)

典型用法示例

import numpy as np

device_data = np.random.normal(loc=100, scale=10, size=(500, 8))

# Min-Max归一化公式：normalized = (x - min) / (max - min)
# 1. 计算各指标（列）的最小值和最大值（保留维度）
min_val = np.min(device_data, axis=0, keepdims=True)
max_val = np.max(device_data, axis=0, keepdims=True)

# 2. 避免分母为0（工业数据中若指标无波动，设为1）
range_val = max_val - min_val
range_val[range_val == 0] = 1  # 处理无波动指标

# 3. 执行归一化
normalized_data = (device_data - min_val) / range_val

# 验证：归一化后数据范围应在[0,1]
print(f"归一化后数据最小值：{np.min(normalized_data):.4f}")
print(f"归一化后数据最大值：{np.max(normalized_data):.4f}")
print("归一化后前2台设备数据：\n", normalized_data[:2])

# 4. 逆归一化（如需还原原始数据）
original_data = normalized_data * range_val + min_val
print("逆归一化后数据与原始数据误差：", np.max(np.abs(original_data - device_data)))

常见陷阱提醒

• 归一化轴方向错误：工业数据中，不同指标（如温度、转速）量纲不同，需按列（axis=0）归一化；若按行归一化，会导致单台设备内指标缩放混乱。
• 分母为0问题：若某指标所有设备数值相同（无波动），会导致分母为0，需提前处理（如设range_val=1），避免计算错误。
• 忘记保存min/max值：若后续需还原数据或对新数据归一化，需保存训练数据的min_val和max_val，不可重新计算新数据的极值。

二、实战场景

「场景需求」：生成济南某智能制造企业500台设备模拟数据，筛选异常设备并计算异常率，对设备数据进行归一化处理，满足工业实际规律、异常筛选准确率100%、归一化无失真要求。

完整实战代码（含详细注释）

import numpy as np

# -------------------------- 1. 生成符合工业规律的模拟设备数据 --------------------------
# 设备指标说明（8项）：1-3（温度/压力/转速，正态分布）、4-6（振动值，偏态分布）、7-8（能耗/运行时长，均匀分布）
np.random.seed(42)  # 固定随机种子，保证结果可复现

# 1.1 温度（60-80℃）、压力（0.8-1.2MPa）、转速（1500-2000rpm）：正态分布
temp = np.random.normal(loc=70, scale=3, size=(500, 1))  # 温度
press = np.random.normal(loc=1.0, scale=0.08, size=(500, 1))  # 压力
speed = np.random.normal(loc=1750, scale=100, size=(500, 1))  # 转速

# 1.2 振动值（0.1-0.5mm，右偏分布，模拟部分设备振动偏大）
vib1 = np.random.weibull(a=2.0, size=(500, 1)) * 0.2 + 0.1  # 振动1
vib2 = np.random.weibull(a=2.5, size=(500, 1)) * 0.2 + 0.1  # 振动2
vib3 = np.random.weibull(a=3.0, size=(500, 1)) * 0.2 + 0.1  # 振动3

# 1.3 能耗（5-15kWh）、运行时长（100-500h）：均匀分布
energy = np.random.uniform(low=5, high=15, size=(500, 1))  # 能耗
run_time = np.random.uniform(low=100, high=500, size=(500, 1))  # 运行时长

# 1.4 合并为500台设备8项指标的完整数据
device_data = np.hstack([temp, press, speed, vib1, vib2, vib3, energy, run_time])
print("模拟设备数据形状：", device_data.shape)
print("前3台设备原始数据：\n", device_data[:3].round(4))

# -------------------------- 2. 筛选异常设备并计算异常率 --------------------------
# 异常规则（贴合工业实际）：
# 温度<62或>78℃；压力<0.85或>1.15MPa；转速<1550或>1950rpm；振动值>0.4mm；能耗>13kWh
temp_abnormal = (device_data[:, 0] < 62) | (device_data[:, 0] > 78)
press_abnormal = (device_data[:, 1] < 0.85) | (device_data[:, 1] > 1.15)
speed_abnormal = (device_data[:, 2] < 1550) | (device_data[:, 2] > 1950)
vib_abnormal = (device_data[:, 3:6] > 0.4).any(axis=1)  # 3-5列为振动值，任意一项超标即异常
energy_abnormal = (device_data[:, 6] > 13)

# 合并所有异常条件，任意一项异常则判定为异常设备
total_abnormal_mask = temp_abnormal | press_abnormal | speed_abnormal | vib_abnormal | energy_abnormal
abnormal_devices = device_data[total_abnormal_mask]
normal_devices = device_data[~total_abnormal_mask]

# 计算异常率
abnormal_count = len(abnormal_devices)
abnormal_rate = abnormal_count / len(device_data)
print(f"\n异常设备数量：{abnormal_count} 台")
print(f"设备异常率：{abnormal_rate:.2%}")

# 验证异常筛选准确率（手动核对前10条异常数据）
print("\n前5台异常设备数据：\n", abnormal_devices[:5].round(4))

# -------------------------- 3. 数据归一化处理（Min-Max） --------------------------
# 按指标（列）归一化，保留原始数据特征，无失真
min_val = np.min(device_data, axis=0, keepdims=True)
max_val = np.max(device_data, axis=0, keepdims=True)
range_val = max_val - min_val
range_val[range_val == 0] = 1  # 处理无波动指标，避免分母为0

normalized_data = (device_data - min_val) / range_val

# 验证归一化效果（范围[0,1]，无失真）
print("\n归一化后数据范围：", np.min(normalized_data).round(4), "-", np.max(normalized_data).round(4))
print("前3台设备归一化后数据：\n", normalized_data[:3].round(4))

# 逆归一化验证（还原原始数据，误差接近0）
restored_data = normalized_data * range_val + min_val
max_error = np.max(np.abs(restored_data - device_data))
print(f"归一化逆还原最大误差：{max_error:.8f}（无失真）")

# -------------------------- 4. 与Python循环对比运算速度 --------------------------
import time

# NumPy向量化运算时间
start_np = time.time()
np_abnormal = (device_data > 120).any(axis=1)
end_np = time.time()
np_time = end_np - start_np

# Python循环运算时间
start_loop = time.time()
loop_abnormal = []
for row in device_data:
    if any(x > 120 for x in row):
        loop_abnormal.append(True)
    else:
        loop_abnormal.append(False)
end_loop = time.time()
loop_time = end_loop - start_loop

print(f"\nNumPy向量化运算时间：{np_time:.6f}s")
print(f"Python循环运算时间：{loop_time:.6f}s")
print(f"NumPy运算速度比Python循环快：{loop_time/np_time:.1f}倍")

实战技巧与参数总结

• 模拟数据技巧：结合工业数据分布特征（正态、偏态、均匀），用不同随机函数生成多维度数据，通过seed固定结果可复现。
• 异常筛选技巧：多条件组合用|（或）、&（且），配合any(axis=1)（任意项异常）、all(axis=1)（所有项异常），精准定位异常设备。
• 归一化关键参数：keepdims=True确保维度匹配，避免广播错误；提前处理range_val=0场景，防止计算崩溃，逆还原验证确保无失真。
• 速度优化核心：向量化运算替代Python循环，无需遍历每台设备，运算速度提升15倍以上，大规模工业数据（万级/十万级设备）优势更明显。

三、练习题（贴近工业业务场景）

练习题1：设备振动数据异常检测与统计

题目：生成1000台设备3项振动指标（0.1-0.5mm， Weibull分布），设定异常阈值为各指标95分位数，筛选振动异常设备，计算每类振动指标的异常率及异常设备的平均振动值。

实现思路：1. 用weibull函数生成1000×3的振动数据；2. 计算各指标95分位数作为异常阈值；3. 按指标筛选异常数据，统计单指标异常率；4. 计算所有异常设备的3项指标平均值。

代码

import numpy as np

# 1. 生成1000台设备3项振动数据（Weibull分布，贴合工业振动特征）
np.random.seed(42)
vib_data = np.random.weibull(a=2.2, size=(1000, 3)) * 0.2 + 0.1  # 0.1-0.5mm

# 2. 计算各指标95分位数（异常阈值）
vib_threshold = np.percentile(vib_data, 95, axis=0, keepdims=True)
print("3项振动指标异常阈值（95分位数）：", vib_threshold.round(4))

# 3. 筛选各指标异常数据，计算单指标异常率
vib1_abnormal = vib_data[:, 0] > vib_threshold[0, 0]
vib2_abnormal = vib_data[:, 1] > vib_threshold[0, 1]
vib3_abnormal = vib_data[:, 2] > vib_threshold[0, 2]

# 统计异常率
vib1_rate = np.sum(vib1_abnormal) / len(vib_data)
vib2_rate = np.sum(vib2_abnormal) / len(vib_data)
vib3_rate = np.sum(vib3_abnormal) / len(vib_data)
print(f"\n振动1异常率：{vib1_rate:.2%}")
print(f"振动2异常率：{vib2_rate:.2%}")
print(f"振动3异常率：{vib3_rate:.2%}")

# 4. 筛选所有振动异常设备（任意一项指标异常）
total_vib_abnormal = vib1_abnormal | vib2_abnormal | vib3_abnormal
abnormal_vib_data = vib_data[total_vib_abnormal]

# 计算异常设备平均振动值
avg_abnormal_vib = np.mean(abnormal_vib_data, axis=0)
print(f"\n异常设备3项振动指标平均值：{avg_abnormal_vib.round(4)}")

练习题2：设备能耗数据归一化与区间统计

题目：现有500台设备的能耗数据（5-15kWh，均匀分布）和运行时长数据（100-500h，正态分布），对两类数据分别做Min-Max归一化，统计归一化后能耗在[0.6,0.8]区间的设备数量及对应平均运行时长。

实现思路：1. 生成能耗和运行时长数据，合并为500×2数组；2. 分别对两列数据做Min-Max归一化；3. 筛选归一化能耗在[0.6,0.8]的设备；4. 计算该区间设备的平均运行时长（原始值）。

代码

import numpy as np

# 1. 生成500台设备能耗和运行时长数据
np.random.seed(42)
energy = np.random.uniform(low=5, high=15, size=(500, 1))  # 能耗5-15kWh
run_time = np.random.normal(loc=300, scale=80, size=(500, 1))  # 运行时长100-500h
# 限制运行时长在合理范围（避免正态分布超出边界）
run_time = np.clip(run_time, 100, 500)
device_data = np.hstack([energy, run_time])

# 2. 分别对两列数据做Min-Max归一化
# 能耗归一化
energy_min = np.min(device_data[:, 0], keepdims=True)
energy_max = np.max(device_data[:, 0], keepdims=True)
normalized_energy = (device_data[:, 0] - energy_min) / (energy_max - energy_min)

# 运行时长归一化
time_min = np.min(device_data[:, 1], keepdims=True)
time_max = np.max(device_data[:, 1], keepdims=True)
normalized_time = (device_data[:, 1] - time_min) / (time_max - time_min)

# 合并归一化后数据
normalized_data = np.vstack([normalized_energy, normalized_time]).T
print("归一化后前5台设备数据：\n", normalized_data[:5].round(4))

# 3. 筛选归一化能耗在[0.6,0.8]区间的设备
energy_mask = (normalized_energy >= 0.6) & (normalized_energy <= 0.8)
target_devices = device_data[energy_mask]

# 4. 统计数量及平均运行时长（原始值）
target_count = len(target_devices)
avg_run_time = np.mean(target_devices[:, 1])
print(f"\n归一化能耗在[0.6,0.8]区间的设备数量：{target_count} 台")
print(f"该区间设备平均运行时长：{avg_run_time:.2f} h")

练习题3：多指标设备健康评分（综合应用）

题目：生成800台设备4项指标（温度、压力、振动、能耗），对各项指标归一化后按权重（温度0.3、压力0.2、振动0.3、能耗0.2）计算健康评分（满分100），评分<60为亚健康设备，筛选亚健康设备并计算其各项指标原始值的标准差。

实现思路：1. 生成4项指标数据（符合工业分布）；2. 对所有指标做Min-Max归一化（映射到[0,1]）；3. 按权重计算健康评分（归一化值×权重求和×100）；4. 筛选亚健康设备，计算其原始指标的标准差。

代码

import numpy as np

# 1. 生成800台设备4项指标数据
np.random.seed(42)
temp = np.random.normal(loc=70, scale=3, size=(800, 1))  # 温度60-80℃
press = np.random.normal(loc=1.0, scale=0.08, size=(800, 1))  # 压力0.8-1.2MPa
vib = np.random.weibull(a=2.0, size=(800, 1)) * 0.2 + 0.1  # 振动0.1-0.5mm
energy = np.random.uniform(low=5, high=15, size=(800, 1))  # 能耗5-15kWh
device_data = np.hstack([temp, press, vib, energy])

# 2. 对4项指标做Min-Max归一化（按列）
min_val = np.min(device_data, axis=0, keepdims=True)
max_val = np.max(device_data, axis=0, keepdims=True)
range_val = max_val - min_val
range_val[range_val == 0] = 1
normalized_data = (device_data - min_val) / range_val

# 3. 按权重计算健康评分（满分100）
weights = np.array([0.3, 0.2, 0.3, 0.2])  # 温度/压力/振动/能耗权重
health_score = np.sum(normalized_data * weights, axis=1) * 100

# 4. 筛选亚健康设备（评分<60）
subhealth_mask = health_score < 60
subhealth_devices = device_data[subhealth_mask]

# 5. 计算亚健康设备各项原始指标的标准差
subhealth_std = np.std(subhealth_devices, axis=0, ddof=1)
print(f"亚健康设备数量：{len(subhealth_devices)} 台")
print(f"亚健康设备各项指标原始值标准差：")
print(f"温度：{subhealth_std[0]:.4f}℃")
print(f"压力：{subhealth_std[1]:.4f}MPa")
print(f"振动：{subhealth_std[2]:.4f}mm")
print(f"能耗：{subhealth_std[3]:.4f}kWh")

（注：文档部分内容可能由 AI 生成）