4. NumPy 入门：数组创建、数组运算、索引与切片、常用统计函数

NumPy（Numerical Python）是 Python 数据科学生态系统的核心库，提供了高性能的多维数组对象和丰富的数学运算工具。在实时监控大屏场景中，NumPy 负责底层数据的批量计算、统计聚合和预处理，是连接原始数据与可视化图表的桥梁。

4.1 NumPy 数组创建

NumPy 的核心数据结构是 ndarray（N 维数组）。创建数组是使用 NumPy 的第一步，以下是几种最常用的创建方式：

import numpy as np

# 从列表创建一维数组
arr1 = np.array([10, 20, 30, 40, 50])
print("一维数组:", arr1)

# 创建二维数组（矩阵）
arr2 = np.array([[1, 2, 3], [4, 5, 6]])
print("二维数组:\n", arr2)

# 创建全零数组，用于初始化监控指标缓存
zero_cache = np.zeros((3, 4))
print("全零缓存矩阵:\n", zero_cache)

# 创建等差数列，用于生成时间轴
time_axis = np.arange(0, 60, 5)  # 0到55秒，步长5秒
print("时间轴(秒):", time_axis)

# 创建随机数组，模拟传感器数据
sensor_data = np.random.randn(100)  # 100个标准正态分布随机数
print("模拟传感器数据(前5个):", sensor_data[:5])

4.2 数组运算

NumPy 最强大的特性之一是矢量化运算——无需编写循环，即可对整个数组执行元素级运算。这在大屏实时计算中能显著提升性能。

4.2.1 算术运算

import numpy as np

a = np.array([10, 20, 30, 40])
b = np.array([1, 2, 3, 4])

# 元素级运算
print("加法:", a + b)        # [11 22 33 44]
print("减法:", a - b)        # [9 18 27 36]
print("乘法:", a * b)        # [10 40 90 160]
print("除法:", a / b)        # [10. 10. 10. 10.]
print("幂运算:", a ** 2)     # [100 400 900 1600]

# 标量与数组运算
print("标量加法:", a + 100)  # [110 120 130 140]
print("标量乘法:", a * 2)    # [20 40 60 80]

4.2.2 广播机制

广播（Broadcasting）允许不同形状的数组进行运算，NumPy 会自动扩展较小数组的维度。这在处理监控数据时非常实用，例如将一组阈值与多维数据进行比较。

# 广播示例：二维数组与一维数组相加
matrix = np.array([[1, 2, 3],
                   [4, 5, 6],
                   [7, 8, 9]])
row_vector = np.array([10, 20, 30])

result = matrix + row_vector
print("广播加法结果:\n", result)
# [[11 22 33]
#  [14 25 36]
#  [17 28 39]]

4.3 索引与切片

NumPy 的索引与切片语法与 Python 列表类似，但功能更强大，支持多维索引和布尔索引。在监控大屏中，切片用于提取特定时间窗口的数据，布尔索引用于条件过滤。

4.3.1 基本索引与切片

import numpy as np

arr = np.array([10, 20, 30, 40, 50, 60])

# 基本索引
print("索引第3个元素:", arr[2])       # 30

# 切片 [start:stop:step]
print("切片 1~4:", arr[1:4])          # [20 30 40]
print("步长为2:", arr[::2])           # [10 30 50]
print("反向切片:", arr[::-1])         # [60 50 40 30 20 10]

# 二维数组切片
matrix = np.array([[1, 2, 3, 4],
                   [5, 6, 7, 8],
                   [9, 10, 11, 12]])

print("第2行:", matrix[1, :])         # [5 6 7 8]
print("第3列:", matrix[:, 2])         # [3 7 11]
print("子矩阵(前2行, 第2~3列):\n", matrix[:2, 1:3])
# [[2 3]
#  [6 7]]

4.3.2 布尔索引

布尔索引是 NumPy 最灵活的筛选方式，通过条件表达式生成布尔数组，再作为索引使用。

# 布尔索引示例
data = np.array([12, 25, 8, 35, 18, 42, 30])

# 筛选大于20的元素
mask = data > 20
print("布尔掩码:", mask)              # [False  True False  True False  True  True]
print("大于20的数据:", data[mask])    # [25 35 42 30]

# 组合条件：大于15且小于35
filtered = data[(data > 15) & (data < 35)]
print("15~35之间的数据:", filtered)   # [25 18 30]

# 将满足条件的元素替换为0
data[data < 20] = 0
print("替换后:", data)               # [ 0 25  0 35  0 42 30]

4.4 常用统计函数

NumPy 提供了丰富的统计函数，用于描述数据的集中趋势、离散程度和分布特征。这些函数是监控指标计算的核心工具。

import numpy as np

# 模拟监控数据：100个请求的响应时间（毫秒）
response_times = np.random.exponential(scale=50, size=100)

# 核心统计指标
mean_rt = np.mean(response_times)
median_rt = np.median(response_times)
std_rt = np.std(response_times)
min_rt = np.min(response_times)
max_rt = np.max(response_times)
p95_rt = np.percentile(response_times, 95)
p99_rt = np.percentile(response_times, 99)

print(f"平均响应时间: {mean_rt:.2f} ms")
print(f"中位数响应时间: {median_rt:.2f} ms")
print(f"标准差: {std_rt:.2f} ms")
print(f"最小值: {min_rt:.2f} ms")
print(f"最大值: {max_rt:.2f} ms")
print(f"P95 延迟: {p95_rt:.2f} ms")
print(f"P99 延迟: {p99_rt:.2f} ms")

# 沿指定轴统计（适用于多维数据）
# 假设有3个服务节点，每个节点有10个采样点
node_data = np.random.randn(3, 10)  # 3行10列
node_means = np.mean(node_data, axis=1)  # 每个节点的均值
node_stds = np.std(node_data, axis=1)    # 每个节点的标准差
print("各节点均值:", node_means)
print("各节点标准差:", node_stds)

4.5 综合案例：监控指标计算

将本章知识点串联起来，实现一个简单的监控指标计算模块：

import numpy as np

class MonitorMetrics:
    """监控指标计算器"""

    def __init__(self, window_size=60):
        self.window_size = window_size
        self.data_buffer = np.zeros(window_size)

    def update(self, new_value):
        """更新数据缓冲区（滑动窗口）"""
        self.data_buffer = np.roll(self.data_buffer, -1)
        self.data_buffer[-1] = new_value

    def compute_metrics(self):
        """计算当前窗口的统计指标"""
        return {
            'mean': np.mean(self.data_buffer),
            'median': np.median(self.data_buffer),
            'std': np.std(self.data_buffer),
            'min': np.min(self.data_buffer),
            'max': np.max(self.data_buffer),
            'p95': np.percentile(self.data_buffer, 95),
            'p99': np.percentile(self.data_buffer, 99),
            'sum': np.sum(self.data_buffer)
        }

    def detect_anomaly(self, threshold=2.0):
        """基于标准差检测异常值"""
        mean = np.mean(self.data_buffer)
        std = np.std(self.data_buffer)
        z_scores = (self.data_buffer - mean) / std
        anomalies = np.abs(z_scores) > threshold
        return anomalies, self.data_buffer[anomalies]

# 模拟实时数据流
metrics = MonitorMetrics(window_size=30)
for i in range(50):
    # 模拟传感器读数，偶尔出现异常值
    if i % 10 == 5:
        value = 100.0  # 异常值
    else:
        value = np.random.normal(50, 10)
    metrics.update(value)

# 计算并输出指标
result = metrics.compute_metrics()
print("当前窗口统计指标:")
for key, val in result.items():
    print(f"  {key}: {val:.2f}")

# 检测异常
anomaly_mask, anomaly_values = metrics.detect_anomaly(threshold=2.0)
print(f"\n检测到 {len(anomaly_values)} 个异常值:")
print(anomaly_values)