Python NumPy - 实战 计算欧氏距离 衡量数据相似度

👋 大家好，欢迎来到我的技术博客！ 📚 在这里，我会分享学习笔记、实战经验与技术思考，力求用简单的方式讲清楚复杂的问题。 🎯 本文将围绕NumPy这个话题展开，希望能为你带来一些启发或实用的参考。 🌱 无论你是刚入门的新手，还是正在进阶的开发者，希望你都能有所收获！

文章目录

• Python NumPy实战：计算欧氏距离衡量数据相似度 📊
• • 什么是欧氏距离？📏
  • NumPy基础：高效数值计算的核心 🔧
  • 多种实现方式对比 🔄
  • • 方法一：手动实现基础版本
    • 方法二：使用NumPy内置函数
    • 方法三：优化版本支持批量计算
  • 性能优化策略 ⚡
  • 实际应用场景 💼
  • • 推荐系统中的用户相似度计算
    • 图像处理中的颜色相似度分析
  • 数据预处理的重要性 📈
  • 高级相似度度量方法 🔍
  • 机器学习集成应用 🤖
  • 异常检测应用 🚨
  • 聚类分析应用 📊
  • 实时数据流处理 ⏱️
  • 性能基准测试 📊
  • 最佳实践建议 💡
  • 总结与展望 🎯

Python NumPy实战：计算欧氏距离衡量数据相似度 📊

在数据分析和机器学习的世界中，衡量数据点之间的相似性是一个核心问题。无论是推荐系统、聚类分析还是异常检测，我们都需要一种有效的方法来量化数据之间的"距离"。欧氏距离作为最直观、最常用的相似度度量方法之一，在实际应用中发挥着重要作用。

什么是欧氏距离？📏

欧氏距离（Euclidean Distance）是两点之间直线距离的度量，它来源于古希腊数学家欧几里得的几何学理论。在二维空间中，这就是我们熟悉的勾股定理；在多维空间中，它是各维度差值平方和的平方根。

对于两个n维向量A(x₁, x₂, …, xₙ)和B(y₁, y₂, …, yₙ)，欧氏距离的计算公式为：

d(A,B) = √[(x₁-y₁)² + (x₂-y₂)² + … + (xₙ-yₙ)²]

这个简单的公式背后蕴含着强大的数据处理能力，让我们能够在高维空间中准确地衡量数据点的相似程度。

NumPy基础：高效数值计算的核心 🔧

NumPy（Numerical Python）是Python科学计算的基础库，提供了高性能的多维数组对象和相关工具。在计算欧氏距离时，NumPy的优势主要体现在以下几个方面：

• 向量化操作：避免显式循环，提高计算效率
• 广播机制：简化不同形状数组间的运算
• 内存优化：连续内存存储，减少内存开销
• 丰富的函数库：提供各种数学运算函数

import numpy as np

# 创建示例数据
point_a = np.array([1, 2, 3])
point_b = np.array([4, 5, 6])

print("点A:", point_a)
print("点B:", point_b)

多种实现方式对比 🔄

方法一：手动实现基础版本

最直接的方式是按照定义逐步计算：

import numpy as np

def euclidean_distance_basic(point1, point2):
"""
基础版本：手动计算欧氏距离
"""
# 确保输入是numpy数组
p1 = np.array(point1)
p2 = np.array(point2)

# 检查维度是否一致
if p1.shape != p2.shape:
raise ValueError("两个点的维度必须相同")

# 计算差值
diff = p1 – p2

# 计算平方和
squared_sum = np.sum(diff ** 2)

# 开平方根得到欧氏距离
distance = np.sqrt(squared_sum)

return distance

# 测试基础版本
a = [1, 2, 3]
b = [4, 5, 6]
distance = euclidean_distance_basic(a, b)
print(f"基础版本计算结果: {distance:.4f}")

方法二：使用NumPy内置函数

NumPy提供了专门用于计算欧氏距离的函数，更加简洁高效：

import numpy as np

def euclidean_distance_numpy(point1, point2):
"""
使用NumPy内置函数计算欧氏距离
"""
p1 = np.array(point1)
p2 = np.array(point2)

# 使用linalg.norm函数计算欧氏范数
distance = np.linalg.norm(p1 – p2)

return distance

# 测试NumPy版本
a = [1, 2, 3]
b = [4, 5, 6]
distance = euclidean_distance_numpy(a, b)
print(f"NumPy版本计算结果: {distance:.4f}")

方法三：优化版本支持批量计算

在实际应用中，我们经常需要计算多个点对之间的距离：

import numpy as np

def euclidean_distances_batch(points1, points2):
"""
批量计算欧氏距离
支持一对多或多对多的距离计算
"""
points1 = np.atleast_2d(points1)
points2 = np.atleast_2d(points2)

# 使用广播机制计算所有点对之间的距离
distances = np.sqrt(np.sum((points1[:, np.newaxis] – points2[np.newaxis, :]) ** 2, axis=2))

return distances

# 测试批量计算
points_a = [[1, 2], [3, 4]]
points_b = [[5, 6], [7, 8], [9, 10]]
distances = euclidean_distances_batch(points_a, points_b)
print("批量计算结果:")
print(distances)

性能优化策略 ⚡

为了获得最佳性能，我们需要考虑以下优化策略：

import numpy as np
import time

def performance_comparison():
"""
性能比较测试
"""
# 生成大量随机数据进行测试
size = 10000
points1 = np.random.rand(size, 100)
points2 = np.random.rand(1000, 100)

# 测试不同方法的性能
methods = {
'基础循环': lambda: basic_loop_version(points1[:100], points2[:100]),
'NumPy向量化': lambda: numpy_vectorized_version(points1, points2),
'优化版本': lambda: optimized_version(points1, points2)
}

for name, method in methods.items():
start_time = time.time()
result = method()
end_time = time.time()
print(f"{name}: {end_time – start_time:.4f}秒")

def basic_loop_version(points1, points2):
"""基础循环版本"""
distances = []
for p1 in points1:
row = []
for p2 in points2:
dist = np.sqrt(np.sum((p1 – p2) ** 2))
row.append(dist)
distances.append(row)
return np.array(distances)

def numpy_vectorized_version(points1, points2):
"""NumPy向量化版本"""
return np.sqrt(np.sum((points1[:, np.newaxis] – points2[np.newaxis, :]) ** 2, axis=2))

def optimized_version(points1, points2):
"""优化版本"""
# 预先计算平方和
sum1 = np.sum(points1 ** 2, axis=1, keepdims=True)
sum2 = np.sum(points2 ** 2, axis=1)

# 使用矩阵乘法优化计算
dot_product = np.dot(points1, points2.T)

# 利用恒等式: ||a-b||² = ||a||² + ||b||² – 2<a,b>
distances_squared = sum1 + sum2 – 2 * dot_product

# 确保没有负数（由于浮点误差可能产生极小负数）
distances_squared = np.maximum(distances_squared, 0)

return np.sqrt(distances_squared)

# 运行性能测试
performance_comparison()

实际应用场景 💼

推荐系统中的用户相似度计算

import numpy as np

class UserSimilarityCalculator:
"""
用户相似度计算器
用于推荐系统中计算用户之间的相似度
"""

def __init__(self, user_features):
"""
初始化用户特征矩阵

Args:
user_features: 用户特征矩阵，每行代表一个用户，每列代表一个特征
"""
self.user_features = np.array(user_features)
self.n_users = self.user_features.shape[0]

def calculate_similarity_matrix(self):
"""
计算所有用户两两之间的相似度矩阵
"""
# 标准化用户特征（均值为0，标准差为1）
normalized_features = (self.user_features – np.mean(self.user_features, axis=0)) / np.std(self.user_features, axis=0)

# 计算欧氏距离矩阵
similarity_matrix = np.zeros((self.n_users, self.n_users))

for i in range(self.n_users):
for j in range(i, self.n_users):
if i == j:
similarity_matrix[i, j] = 0 # 自己与自己的距离为0
else:
distance = np.linalg.norm(normalized_features[i] – normalized_features[j])
# 转换为相似度分数（距离越小相似度越高）
similarity = 1 / (1 + distance)
similarity_matrix[i, j] = similarity
similarity_matrix[j, i] = similarity

return similarity_matrix

def find_similar_users(self, user_id, top_k=5):
"""
查找与指定用户最相似的其他用户

Args:
user_id: 目标用户的ID
top_k: 返回最相似的K个用户

Returns:
相似用户列表及其相似度分数
"""
similarity_matrix = self.calculate_similarity_matrix()
similarities = similarity_matrix[user_id]

# 获取除自己外的所有用户索引
other_users = np.arange(self.n_users)
other_users = other_users[other_users != user_id]

# 按相似度排序
sorted_indices = np.argsort(similarities[other_users])[::–1][:top_k]
similar_users = other_users[sorted_indices]
similarity_scores = similarities[similar_users]

return list(zip(similar_users, similarity_scores))

# 示例：电影评分推荐系统
np.random.seed(42)
n_users = 100
n_movies = 50

# 生成模拟用户特征（用户对不同类型电影的偏好）
user_features = np.random.rand(n_users, n_movies) * 5 # 评分范围0-5

# 创建相似度计算器
calculator = UserSimilarityCalculator(user_features)

# 查找与用户0最相似的5个用户
similar_users = calculator.find_similar_users(0, top_k=5)
print("与用户0最相似的用户:")
for user_id, similarity in similar_users:
print(f"用户{user_id}: 相似度 {similarity:.4f}")

图像处理中的颜色相似度分析

import numpy as np

class ColorSimilarityAnalyzer:
"""
颜色相似度分析器
用于分析图像中像素颜色的相似性
"""

def __init__(self, image_data):
"""
初始化图像数据

Args:
image_data: 图像数据，形状为(height, width, 3)表示RGB三个通道
"""
self.image_data = np.array(image_data)
self.height, self.width, self.channels = self.image_data.shape

def extract_dominant_colors(self, n_colors=5):
"""
提取图像中的主要颜色

Args:
n_colors: 提取的主要颜色数量

Returns:
主要颜色列表
"""
# 将图像重塑为像素列表
pixels = self.image_data.reshape(–1, self.channels)

# 随机采样部分像素以提高效率
sample_size = min(10000, len(pixels))
sampled_pixels = pixels[np.random.choice(len(pixels), sample_size, replace=False)]

# 使用K-means聚类提取主要颜色
from sklearn.cluster import KMeans
kmeans = KMeans(n_clusters=n_colors, random_state=42, n_init=10)
kmeans.fit(sampled_pixels)

return kmeans.cluster_centers_

def color_similarity_matrix(self, colors=None):
"""
计算颜色之间的相似度矩阵

Args:
colors: 颜色列表，如果为None则自动提取主要颜色

Returns:
颜色相似度矩阵
"""
if colors is None:
colors = self.extract_dominant_colors()

colors = np.array(colors)
n_colors = len(colors)

# 计算颜色之间的欧氏距离
similarity_matrix = np.zeros((n_colors, n_colors))

for i in range(n_colors):
for j in range(n_colors):
distance = np.linalg.norm(colors[i] – colors[j])
# 转换为相似度（0-1范围）
similarity_matrix[i, j] = 1 – (distance / (255 * np.sqrt(3))) # RGB最大距离

return similarity_matrix, colors

def find_closest_colors(self, target_color, colors=None, top_k=3):
"""
查找与目标颜色最接近的颜色

Args:
target_color: 目标颜色(R,G,B)
colors: 参考颜色列表
top_k: 返回最接近的K个颜色

Returns:
最接近的颜色列表及其距离
"""
if colors is None:
colors = self.extract_dominant_colors()

colors = np.array(colors)
target_color = np.array(target_color)

# 计算与每个颜色的距离
distances = np.linalg.norm(colors – target_color, axis=1)

# 排序并返回前K个
sorted_indices = np.argsort(distances)[:top_k]
closest_colors = colors[sorted_indices]
closest_distances = distances[sorted_indices]

return list(zip(closest_colors, closest_distances))

# 示例：创建虚拟图像数据
np.random.seed(42)
image_height, image_width = 100, 100
virtual_image = np.random.randint(0, 256, (image_height, image_width, 3))

# 创建颜色分析器
analyzer = ColorSimilarityAnalyzer(virtual_image)

# 提取主要颜色
dominant_colors = analyzer.extract_dominant_colors(n_colors=5)
print("提取的主要颜色:")
for i, color in enumerate(dominant_colors):
print(f"颜色{i+1}: RGB({int(color[0])}, {int(color[1])}, {int(color[2])})")

# 计算颜色相似度矩阵
similarity_matrix, colors = analyzer.color_similarity_matrix(dominant_colors)
print("\\n颜色相似度矩阵:")
print(similarity_matrix)

# 查找与特定颜色最接近的颜色
target_color = [255, 0, 0] # 红色
closest_colors = analyzer.find_closest_colors(target_color, dominant_colors, top_k=3)
print(f"\\n与红色最接近的颜色:")
for color, distance in closest_colors:
print(f"RGB({int(color[0])}, {int(color[1])}, {int(color[2])}): 距离 {distance:.2f}")

数据预处理的重要性 📈

在计算欧氏距离之前，适当的数据预处理可以显著提高相似度计算的准确性：

import numpy as np
from sklearn.preprocessing import StandardScaler, MinMaxScaler

class DataPreprocessor:
"""
数据预处理器
用于准备相似度计算所需的数据
"""

def __init__(self):
self.scalers = {}

def standardize_features(self, data, feature_names=None):
"""
标准化特征（Z-score标准化）
使每个特征的均值为0，标准差为1

Args:
data: 输入数据
feature_names: 特征名称列表

Returns:
标准化后的数据
"""
data = np.array(data)

if feature_names is None:
feature_names = [f'feature_{i}' for i in range(data.shape[1])]

scaler = StandardScaler()
standardized_data = scaler.fit_transform(data)

# 保存缩放器以便后续使用
self.scalers['standard'] = scaler

return standardized_data

def normalize_features(self, data, feature_names=None):
"""
归一化特征（Min-Max归一化）
将每个特征缩放到0-1范围内

Args:
data: 输入数据
feature_names: 特征名称列表

Returns:
归一化后的数据
"""
data = np.array(data)

if feature_names is None:
feature_names = [f'feature_{i}' for i in range(data.shape[1])]

scaler = MinMaxScaler()
normalized_data = scaler.fit_transform(data)

# 保存缩放器
self.scalers['normalize'] = scaler

return normalized_data

def handle_missing_values(self, data, strategy='mean'):
"""
处理缺失值

Args:
data: 包含缺失值的数据
strategy: 处理策略 ('mean', 'median', 'drop')

Returns:
处理后的数据
"""
data = np.array(data, dtype=float)

if strategy == 'mean':
# 用均值填充
col_means = np.nanmean(data, axis=0)
nan_indices = np.where(np.isnan(data))
data[nan_indices] = np.take(col_means, nan_indices[1])
elif strategy == 'median':
# 用中位数填充
col_medians = np.nanmedian(data, axis=0)
nan_indices = np.where(np.isnan(data))
data[nan_indices] = np.take(col_medians, nan_indices[1])
elif strategy == 'drop':
# 删除包含缺失值的行
data = data[~np.isnan(data).any(axis=1)]

return data

# 示例数据预处理
np.random.seed(42)
raw_data = np.random.randn(100, 5) * 10 + 50 # 均值50，标准差10

# 添加一些缺失值
missing_indices = np.random.choice(raw_data.size, 20, replace=False)
raw_data.flat[missing_indices] = np.nan

# 创建预处理器
preprocessor = DataPreprocessor()

# 处理缺失值
cleaned_data = preprocessor.handle_missing_values(raw_data, strategy='mean')
print("原始数据统计:")
print(f"均值: {np.mean(raw_data[~np.isnan(raw_data)])}")
print(f"标准差: {np.std(raw_data[~np.isnan(raw_data)])}")

# 标准化数据
standardized_data = preprocessor.standardize_features(cleaned_data)
print("\\n标准化后数据统计:")
print(f"均值: {np.mean(standardized_data)}")
print(f"标准差: {np.std(standardized_data)}")

# 归一化数据
normalized_data = preprocessor.normalize_features(cleaned_data)
print("\\n归一化后数据统计:")
print(f"最小值: {np.min(normalized_data)}")
print(f"最大值: {np.max(normalized_data)}")

高级相似度度量方法 🔍

虽然欧氏距离是最常用的相似度度量方法，但在某些情况下，其他方法可能更适合：

import numpy as np
from scipy.spatial.distance import cosine, cityblock

class AdvancedDistanceMetrics:
"""
高级距离度量类
提供多种相似度计算方法
"""

@staticmethod
def euclidean_distance(point1, point2):
"""欧氏距离"""
return np.linalg.norm(np.array(point1) – np.array(point2))

@staticmethod
def manhattan_distance(point1, point2):
"""曼哈顿距离（L1距离）"""
return np.sum(np.abs(np.array(point1) – np.array(point2)))

@staticmethod
def cosine_similarity(point1, point2):
"""余弦相似度"""
p1 = np.array(point1)
p2 = np.array(point2)

# 计算余弦距离，然后转换为相似度
cos_dist = cosine(p1, p2)
return 1 – cos_dist

@staticmethod
def chebyshev_distance(point1, point2):
"""切比雪夫距离（L∞距离）"""
return np.max(np.abs(np.array(point1) – np.array(point2)))

@staticmethod
def minkowski_distance(point1, point2, p=3):
"""闵可夫斯基距离（Lp距离）"""
diff = np.abs(np.array(point1) – np.array(point2))
return np.sum(diff ** p) ** (1/p)

# 比较不同距离度量方法
metrics = AdvancedDistanceMetrics()

# 创建测试数据
point_a = [1, 2, 3, 4, 5]
point_b = [2, 3, 4, 5, 6]
point_c = [5, 4, 3, 2, 1]

points = [point_a, point_b, point_c]
point_names = ['A', 'B', 'C']

print("不同距离度量方法比较:")
print("=" * 50)

# 计算所有点对之间的距离
for i in range(len(points)):
for j in range(i+1, len(points)):
p1_name = point_names[i]
p2_name = point_names[j]
p1 = points[i]
p2 = points[j]

print(f"\\n点{p1_name} vs 点{p2_name}:")
print(f" 欧氏距离: {metrics.euclidean_distance(p1, p2):.4f}")
print(f" 曼哈顿距离: {metrics.manhattan_distance(p1, p2):.4f}")
print(f" 余弦相似度: {metrics.cosine_similarity(p1, p2):.4f}")
print(f" 切比雪夫距离: {metrics.chebyshev_distance(p1, p2):.4f}")
print(f" 闵可夫斯基距离(p=3): {metrics.minkowski_distance(p1, p2, 3):.4f}")

# 可视化不同距离的影响
import matplotlib.pyplot as plt

def visualize_distance_effects():
"""
可视化不同距离度量的效果
"""
# 创建二维数据点
points_2d = np.array([
[0, 0], # 原点
[3, 4], # 欧氏距离为5的点
[3, 0], # 曼哈顿距离为3的点
[0, 4], # 曼哈顿距离为4的点
[5, 5], # 对角线上的点
])

origin = points_2d[0]

print("\\n二维空间中不同距离度量的可视化:")
print("-" * 40)

for i, point in enumerate(points_2d[1:], 1):
euclidean = metrics.euclidean_distance(origin, point)
manhattan = metrics.manhattan_distance(origin, point)
cosine_sim = metrics.cosine_similarity(origin, point)

print(f"点{i}: ({point[0]}, {point[1]})")
print(f" 欧氏距离: {euclidean:.4f}")
print(f" 曼哈顿距离: {manhattan:.4f}")
print(f" 余弦相似度: {cosine_sim:.4f}")
print()

visualize_distance_effects()

机器学习集成应用 🤖

将欧氏距离融入机器学习流程中，可以构建更智能的应用程序：

import numpy as np
from collections import Counter
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

class EuclideanKNNClassifier:
"""
基于欧氏距离的K近邻分类器
"""

def __init__(self, k=3):
"""
初始化分类器

Args:
k: 近邻数量
"""
self.k = k
self.X_train = None
self.y_train = None

def fit(self, X, y):
"""
训练模型（实际上只是存储训练数据）

Args:
X: 训练特征
y: 训练标签
"""
self.X_train = np.array(X)
self.y_train = np.array(y)

def _calculate_distances(self, X_test):
"""
计算测试样本与所有训练样本的距离

Args:
X_test: 测试样本

Returns:
距离矩阵
"""
X_test = np.array(X_test)

# 使用广播机制计算所有距离
distances = np.sqrt(np.sum((X_test[:, np.newaxis] – self.X_train[np.newaxis, :]) ** 2, axis=2))

return distances

def predict(self, X_test):
"""
预测测试样本的类别

Args:
X_test: 测试样本

Returns:
预测标签
"""
distances = self._calculate_distances(X_test)
predictions = []

for i in range(len(X_test)):
# 获取最近的k个邻居的索引
nearest_neighbors_idx = np.argsort(distances[i])[:self.k]

# 获取这些邻居的标签
nearest_labels = self.y_train[nearest_neighbors_idx]

# 投票决定最终标签
most_common_label = Counter(nearest_labels).most_common(1)[0][0]
predictions.append(most_common_label)

return np.array(predictions)

def predict_proba(self, X_test):
"""
预测测试样本属于各个类别的概率

Args:
X_test: 测试样本

Returns:
类别概率分布
"""
distances = self._calculate_distances(X_test)
probabilities = []

# 获取所有唯一标签
unique_labels = np.unique(self.y_train)

for i in range(len(X_test)):
# 获取最近的k个邻居的索引
nearest_neighbors_idx = np.argsort(distances[i])[:self.k]

# 获取这些邻居的标签
nearest_labels = self.y_train[nearest_neighbors_idx]

# 计算每个类别的概率
label_counts = Counter(nearest_labels)
probs = []

for label in unique_labels:
prob = label_counts.get(label, 0) / self.k
probs.append(prob)

probabilities.append(probs)

return np.array(probabilities), unique_labels

# 创建模拟数据集进行测试
np.random.seed(42)
n_samples = 1000
n_features = 4

# 生成两类数据
class_0_data = np.random.randn(n_samples//2, n_features) + np.array([2, 2, 2, 2])
class_1_data = np.random.randn(n_samples//2, n_features) + np.array([–2, –2, –2, –2])

# 合并数据
X = np.vstack([class_0_data, class_1_data])
y = np.hstack([np.zeros(n_samples//2), np.ones(n_samples//2)])

# 分割训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 创建和训练分类器
knn_classifier = EuclideanKNNClassifier(k=5)
knn_classifier.fit(X_train, y_train)

# 进行预测
predictions = knn_classifier.predict(X_test)
probabilities, labels = knn_classifier.predict_proba(X_test)

# 计算准确率
accuracy = accuracy_score(y_test, predictions)
print(f"KNN分类器准确率: {accuracy:.4f}")

# 显示部分预测结果
print("\\n部分预测结果:")
print("真实标签 -> 预测标签 -> 概率分布")
for i in range(min(10, len(X_test))):
true_label = y_test[i]
pred_label = predictions[i]
probs = probabilities[i]
print(f"{true_label} -> {pred_label} -> {[f'{p:.3f}' for p in probs]}")

# 不同K值的影响分析
k_values = [1, 3, 5, 7, 9, 11]
accuracies = []

for k in k_values:
knn = EuclideanKNNClassifier(k=k)
knn.fit(X_train, y_train)
preds = knn.predict(X_test)
acc = accuracy_score(y_test, preds)
accuracies.append(acc)
print(f"K={k}: 准确率={acc:.4f}")

# 可视化K值对准确率的影响
import matplotlib.pyplot as plt

plt.figure(figsize=(10, 6))
plt.plot(k_values, accuracies, 'bo-', linewidth=2, markersize=8)
plt.xlabel('K值')
plt.ylabel('准确率')
plt.title('K值对KNN分类器性能的影响')
plt.grid(True, alpha=0.3)
plt.show()

异常检测应用 🚨

利用欧氏距离进行异常检测是一种简单而有效的方法：

import numpy as np
from scipy import stats

class EuclideanAnomalyDetector:
"""
基于欧氏距离的异常检测器
"""

def __init__(self, method='threshold', threshold_percentile=95):
"""
初始化异常检测器

Args:
method: 检测方法 ('threshold', 'zscore', 'isolation')
threshold_percentile: 阈值百分位数
"""
self.method = method
self.threshold_percentile = threshold_percentile
self.normal_data = None
self.threshold = None
self.centroid = None

def fit(self, normal_data):
"""
在正常数据上训练检测器

Args:
normal_data: 正常数据样本
"""
self.normal_data = np.array(normal_data)

# 计算数据中心点（质心）
self.centroid = np.mean(self.normal_data, axis=0)

# 根据选择的方法设置阈值
if self.method == 'threshold':
distances_to_centroid = np.linalg.norm(self.normal_data – self.centroid, axis=1)
self.threshold = np.percentile(distances_to_centroid, self.threshold_percentile)
elif self.method == 'zscore':
distances_to_centroid = np.linalg.norm(self.normal_data – self.centroid, axis=1)
mean_dist = np.mean(distances_to_centroid)
std_dist = np.std(distances_to_centroid)
self.threshold = mean_dist + 2 * std_dist # 2倍标准差

def detect_anomalies(self, test_data):
"""
检测异常数据点

Args:
test_data: 待检测的数据

Returns:
异常标记数组和距离数组
"""
if self.centroid is None:
raise ValueError("请先调用fit方法训练检测器")

test_data = np.array(test_data)

# 计算每个点到中心点的距离
distances = np.linalg.norm(test_data – self.centroid, axis=1)

# 根据阈值判断是否为异常
anomalies = distances > self.threshold

return anomalies, distances

def get_anomaly_scores(self, test_data):
"""
获取异常得分（距离越大得分越高）

Args:
test_data: 待检测的数据

Returns:
异常得分数组
"""
_, distances = self.detect_anomalies(test_data)

# 将距离转换为得分（可以根据需要调整转换方式）
max_distance = np.max(distances) if np.max(distances) > 0 else 1
scores = distances / max_distance

return scores

# 创建正常数据和异常数据进行测试
np.random.seed(42)
n_normal = 1000
n_anomalies = 50
n_features = 3

# 生成正常的三维数据（球形分布）
normal_data = np.random.randn(n_normal, n_features)

# 生成异常数据（远离正常数据区域）
anomaly_data = np.random.randn(n_anomalies, n_features) * 2 + np.array([10, 10, 10])

# 合并测试数据
test_data = np.vstack([normal_data[:200], anomaly_data])
true_labels = np.hstack([np.zeros(200), np.ones(n_anomalies)])

# 创建异常检测器
detector = EuclideanAnomalyDetector(method='threshold', threshold_percentile=95)
detector.fit(normal_data)

# 检测异常
anomalies, distances = detector.detect_anomalies(test_data)
anomaly_scores = detector.get_anomaly_scores(test_data)

# 评估检测效果
tp = np.sum(anomalies[–n_anomalies:]) # 真正例
fp = np.sum(anomalies[:–n_anomalies]) # 假正例
tn = np.sum(~anomalies[:–n_anomalies]) # 真负例
fn = np.sum(~anomalies[–n_anomalies:]) # 假负例

precision = tp / (tp + fp) if (tp + fp) > 0 else 0
recall = tp / (tp + fn) if (tp + fn) > 0 else 0
f1_score = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

print("异常检测结果:")
print(f"真正例(TP): {tp}")
print(f"假正例(FP): {fp}")
print(f"真负例(TN): {tn}")
print(f"假负例(FN): {fn}")
print(f"精确率: {precision:.4f}")
print(f"召回率: {recall:.4f}")
print(f"F1分数: {f1_score:.4f}")
print(f"检测阈值: {detector.threshold:.4f}")

# 显示部分检测结果
print("\\n部分检测结果:")
print("样本编号 | 距离 | 异常得分 | 是否异常 | 真实标签")
print("-" * 50)
for i in range(min(15, len(test_data))):
is_anomaly = "✓" if anomalies[i] else "✗"
true_label = "异常" if true_labels[i] == 1 else "正常"
print(f"{i+1:8d} | {distances[i]:.3f} | {anomaly_scores[i]:.3f} | {is_anomaly:6s} | {true_label}")

# 比较不同方法的效果
methods = ['threshold', 'zscore']
results = {}

for method in methods:
detector = EuclideanAnomalyDetector(method=method)
detector.fit(normal_data)
anomalies, _ = detector.detect_anomalies(test_data)

tp = np.sum(anomalies[–n_anomalies:])
fp = np.sum(anomalies[:–n_anomalies])
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
recall = tp / n_anomalies

results[method] = {'precision': precision, 'recall': recall}

print("\\n不同方法效果比较:")
print("-" * 30)
for method, metrics in results.items():
print(f"{method}: 精确率={metrics['precision']:.4f}, 召回率={metrics['recall']:.4f}")

聚类分析应用 📊

欧氏距离在聚类算法中扮演着重要角色：

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs

class EuclideanKMeans:
"""
基于欧氏距离的K-means聚类算法
"""

def __init__(self, k=3, max_iters=100, tol=1e-4):
"""
初始化K-means算法

Args:
k: 聚类数量
max_iters: 最大迭代次数
tol: 收敛容忍度
"""
self.k = k
self.max_iters = max_iters
self.tol = tol
self.centroids = None
self.labels = None

def fit(self, X):
"""
训练K-means模型

Args:
X: 输入数据
"""
X = np.array(X)
n_samples, n_features = X.shape

# 随机初始化聚类中心
self.centroids = X[np.random.choice(n_samples, self.k, replace=False)]

for iteration in range(self.max_iters):
# 分配每个点到最近的聚类中心
distances = np.sqrt(np.sum((X[:, np.newaxis] – self.centroids[np.newaxis, :]) ** 2, axis=2))
self.labels = np.argmin(distances, axis=1)

# 更新聚类中心
new_centroids = np.array([X[self.labels == i].mean(axis=0) for i in range(self.k)])

# 检查收敛
if np.all(np.abs(new_centroids – self.centroids) < self.tol):
print(f"算法在第{iteration+1}次迭代后收敛")
break

self.centroids = new_centroids

def predict(self, X):
"""
预测新数据点的聚类标签

Args:
X: 新数据点

Returns:
聚类标签
"""
X = np.array(X)
distances = np.sqrt(np.sum((X[:, np.newaxis] – self.centroids[np.newaxis, :]) ** 2, axis=2))
return np.argmin(distances, axis=1)

def get_cluster_info(self):
"""
获取聚类信息

Returns:
聚类中心和每个聚类的统计信息
"""
info = {}
for i in range(self.k):
cluster_points = self.labels == i
info[f'cluster_{i}'] = {
'center': self.centroids[i],
'size': np.sum(cluster_points),
'avg_distance': np.mean(np.linalg.norm(self.centroids[i] – self.centroids[self.labels], axis=1)[cluster_points])
}
return info

# 生成测试数据
np.random.seed(42)
X, y_true = make_blobs(n_samples=300, centers=4, cluster_std=0.60, random_state=0)

# 应用自定义K-means算法
kmeans_custom = EuclideanKMeans(k=4)
kmeans_custom.fit(X)

# 获取聚类结果
labels_custom = kmeans_custom.labels
centroids_custom = kmeans_custom.centroids

# 显示聚类信息
cluster_info = kmeans_custom.get_cluster_info()
print("聚类信息:")
print("=" * 40)
for cluster_name, info in cluster_info.items():
print(f"{cluster_name}:")
print(f" 中心点: [{info['center'][0]:.2f}, {info['center'][1]:.2f}]")
print(f" 点数量: {info['size']}")
print(f" 平均距离: {info['avg_distance']:.2f}")
print()

# 可视化聚类结果
plt.figure(figsize=(12, 5))

# 原始数据
plt.subplot(1, 2, 1)
plt.scatter(X[:, 0], X[:, 1], c=y_true, cmap='viridis', alpha=0.6)
plt.title('原始数据的真实聚类')
plt.xlabel('特征1')
plt.ylabel('特征2')

# K-means聚类结果
plt.subplot(1, 2, 2)
colors = ['red', 'blue', 'green', 'orange']
for i in range(4):
cluster_points = X[labels_custom == i]
plt.scatter(cluster_points[:, 0], cluster_points[:, 1],
c=colors[i], alpha=0.6, label=f'聚类{i}')
plt.scatter(centroids_custom[:, 0], centroids_custom[:, 1],
c='black', marker='x', s=200, linewidths=3, label='聚类中心')
plt.title('K-means聚类结果')
plt.xlabel('特征1')
plt.ylabel('特征2')
plt.legend()

plt.tight_layout()
plt.show()

# 计算聚类质量指标
def calculate_wcss(X, labels, centroids):
"""计算簇内平方和(WCSS)"""
wcss = 0
for i in range(len(centroids)):
cluster_points = X[labels == i]
wcss += np.sum((cluster_points – centroids[i]) ** 2)
return wcss

def calculate_silhouette_score(X, labels):
"""计算轮廓系数"""
n_samples = len(X)
silhouette_scores = []

for i in range(n_samples):
# 计算样本到同簇其他点的平均距离
same_cluster = labels == labels[i]
same_cluster[i] = False # 排除自身

if np.sum(same_cluster) > 0:
a = np.mean(np.linalg.norm(X[i] – X[same_cluster], axis=1))
else:
a = 0

# 计算样本到其他簇的平均距离
b_values = []
for cluster_id in np.unique(labels):
if cluster_id != labels[i]:
other_cluster = labels == cluster_id
b = np.mean(np.linalg.norm(X[i] – X[other_cluster], axis=1))
b_values.append(b)

b = min(b_values) if b_values else 0

# 计算轮廓系数
if a == 0 and b == 0:
silhouette = 0
else:
silhouette = (b – a) / max(a, b)

silhouette_scores.append(silhouette)

return np.mean(silhouette_scores)

# 计算评估指标
wcss = calculate_wcss(X, labels_custom, centroids_custom)
silhouette = calculate_silhouette_score(X, labels_custom)

print(f"簇内平方和(WCSS): {wcss:.2f}")
print(f"轮廓系数: {silhouette:.4f}")

# 确定最优K值
k_range = range(1, 11)
wcss_values = []

for k in k_range:
kmeans_temp = EuclideanKMeans(k=k)
kmeans_temp.fit(X)
wcss_temp = calculate_wcss(X, kmeans_temp.labels, kmeans_temp.centroids)
wcss_values.append(wcss_temp)

# 绘制肘部法则图
plt.figure(figsize=(10, 6))
plt.plot(k_range, wcss_values, 'bo-', linewidth=2, markersize=8)
plt.xlabel('聚类数量(K)')
plt.ylabel('簇内平方和(WCSS)')
plt.title('肘部法则确定最优K值')
plt.grid(True, alpha=0.3)
plt.show()

# 找到肘部点
from kneed import KneeLocator
kl = KneeLocator(k_range, wcss_values, curve="convex", direction="decreasing")
optimal_k = kl.elbow
print(f"通过肘部法则确定的最优K值: {optimal_k}")

#mermaid-svg-DHJNmt9yhEK8aEJg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-DHJNmt9yhEK8aEJg .error-icon{fill:#552222;}#mermaid-svg-DHJNmt9yhEK8aEJg .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-DHJNmt9yhEK8aEJg .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-DHJNmt9yhEK8aEJg .marker{fill:#333333;stroke:#333333;}#mermaid-svg-DHJNmt9yhEK8aEJg .marker.cross{stroke:#333333;}#mermaid-svg-DHJNmt9yhEK8aEJg svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-DHJNmt9yhEK8aEJg p{margin:0;}#mermaid-svg-DHJNmt9yhEK8aEJg .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-DHJNmt9yhEK8aEJg .cluster-label text{fill:#333;}#mermaid-svg-DHJNmt9yhEK8aEJg .cluster-label span{color:#333;}#mermaid-svg-DHJNmt9yhEK8aEJg .cluster-label span p{background-color:transparent;}#mermaid-svg-DHJNmt9yhEK8aEJg .label text,#mermaid-svg-DHJNmt9yhEK8aEJg span{fill:#333;color:#333;}#mermaid-svg-DHJNmt9yhEK8aEJg .node rect,#mermaid-svg-DHJNmt9yhEK8aEJg .node circle,#mermaid-svg-DHJNmt9yhEK8aEJg .node ellipse,#mermaid-svg-DHJNmt9yhEK8aEJg .node polygon,#mermaid-svg-DHJNmt9yhEK8aEJg .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-DHJNmt9yhEK8aEJg .rough-node .label text,#mermaid-svg-DHJNmt9yhEK8aEJg .node .label text,#mermaid-svg-DHJNmt9yhEK8aEJg .image-shape .label,#mermaid-svg-DHJNmt9yhEK8aEJg .icon-shape .label{text-anchor:middle;}#mermaid-svg-DHJNmt9yhEK8aEJg .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-DHJNmt9yhEK8aEJg .rough-node .label,#mermaid-svg-DHJNmt9yhEK8aEJg .node .label,#mermaid-svg-DHJNmt9yhEK8aEJg .image-shape .label,#mermaid-svg-DHJNmt9yhEK8aEJg .icon-shape .label{text-align:center;}#mermaid-svg-DHJNmt9yhEK8aEJg .node.clickable{cursor:pointer;}#mermaid-svg-DHJNmt9yhEK8aEJg .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-DHJNmt9yhEK8aEJg .arrowheadPath{fill:#333333;}#mermaid-svg-DHJNmt9yhEK8aEJg .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-DHJNmt9yhEK8aEJg .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-DHJNmt9yhEK8aEJg .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-DHJNmt9yhEK8aEJg .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-DHJNmt9yhEK8aEJg .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-DHJNmt9yhEK8aEJg .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-DHJNmt9yhEK8aEJg .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-DHJNmt9yhEK8aEJg .cluster text{fill:#333;}#mermaid-svg-DHJNmt9yhEK8aEJg .cluster span{color:#333;}#mermaid-svg-DHJNmt9yhEK8aEJg div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-DHJNmt9yhEK8aEJg .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-DHJNmt9yhEK8aEJg rect.text{fill:none;stroke-width:0;}#mermaid-svg-DHJNmt9yhEK8aEJg .icon-shape,#mermaid-svg-DHJNmt9yhEK8aEJg .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-DHJNmt9yhEK8aEJg .icon-shape p,#mermaid-svg-DHJNmt9yhEK8aEJg .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-DHJNmt9yhEK8aEJg .icon-shape .label rect,#mermaid-svg-DHJNmt9yhEK8aEJg .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-DHJNmt9yhEK8aEJg .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-DHJNmt9yhEK8aEJg .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-DHJNmt9yhEK8aEJg :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

是

否

开始

加载数据

数据预处理

计算欧氏距离矩阵

距离计算完成?

相似度分析

优化计算过程

聚类分析

异常检测

推荐系统

结果评估

输出结果

结束

实时数据流处理 ⏱️

在实时应用中，我们需要能够快速处理连续到达的数据流：

import numpy as np
from collections import deque
import threading
import time

class RealTimeDistanceProcessor:
"""
实时欧氏距离处理器
适用于流式数据处理场景
"""

def __init__(self, window_size=100, reference_point=None):
"""
初始化实时处理器

Args:
window_size: 滑动窗口大小
reference_point: 参考点，用于计算相对距离
"""
self.window_size = window_size
self.reference_point = np.array(reference_point) if reference_point is not None else None
self.data_window = deque(maxlen=window_size)
self.distances_history = deque(maxlen=window_size)
self.lock = threading.Lock()

def add_data_point(self, point):
"""
添加新的数据点

Args:
point: 新的数据点
"""
with self.lock:
point = np.array(point)
self.data_window.append(point)

# 如果设置了参考点，计算到参考点的距离
if self.reference_point is not None:
distance = np.linalg.norm(point – self.reference_point)
self.distances_history.append(distance)

def get_recent_distances(self, n_last=10):
"""
获取最近n个距离值

Args:
n_last: 返回最近的n个值

Returns:
最近的距离值列表
"""
with self.lock:
return list(self.distances_history)[–n_last:]

def get_statistics(self):
"""
获取当前窗口的统计信息

Returns:
统计信息字典
"""
with self.lock:
if len(self.distances_history) == 0:
return {"count": 0}

distances = np.array(list(self.distances_history))
return {
"count": len(distances),
"mean": np.mean(distances),
"std": np.std(distances),
"min": np.min(distances),
"max": np.max(distances),
"median": np.median(distances)
}

def detect_anomaly(self, threshold_multiplier=2.0):
"""
基于统计方法检测异常值

Args:
threshold_multiplier: 阈值倍数（基于标准差）

Returns:
是否检测到异常及异常值
"""
with self.lock:
if len(self.distances_history) < 10:
return False, None

distances = np.array(list(self.distances_history))
mean_dist = np.mean(distances)
std_dist = np.std(distances)

# 使用3σ原则检测异常
upper_threshold = mean_dist + threshold_multiplier * std_dist
lower_threshold = max(0, mean_dist – threshold_multiplier * std_dist)

latest_distance = distances[–1]
is_anomaly = latest_distance > upper_threshold or latest_distance < lower_threshold

return is_anomaly, latest_distance

def set_reference_point(self, point):
"""
设置或更新参考点

Args:
point: 新的参考点
"""
with self.lock:
self.reference_point = np.array(point)
# 重新计算所有历史数据的距离
self.distances_history.clear()
for point in self.data_window:
distance = np.linalg.norm(point – self.reference_point)
self.distances_history.append(distance)

# 模拟实时数据流处理
def simulate_realtime_processing():
"""
模拟实时数据处理场景
"""
print("开始实时数据流处理演示…")

# 创建处理器
processor = RealTimeDistanceProcessor(window_size=50, reference_point=[0, 0, 0])

# 模拟正常数据流
np.random.seed(42)
for i in range(100):
# 生成接近原点的正常数据
if i < 80:
new_point = np.random.randn(3) * 0.5
else:
# 模拟异常数据（远离原点）
new_point = np.random.randn(3) * 3 + np.array([5, 5, 5])

processor.add_data_point(new_point)

# 定期输出统计信息
if (i + 1) % 20 == 0:
stats = processor.get_statistics()
is_anomaly, latest_dist = processor.detect_anomaly()

print(f"步骤 {i+1}:")
print(f" 数据点: [{new_point[0]:.2f}, {new_point[1]:.2f}, {new_point[2]:.2f}]")
print(f" 当前统计: 均值={stats['mean']:.2f}, 标准差={stats['std']:.2f}")
print(f" 最新距离: {latest_dist:.2f}" if latest_dist else " 最新距离: N/A")
print(f" 异常检测: {'发现异常!' if is_anomaly else '正常'}")
print()

time.sleep(0.1) # 模拟实时延迟

# 运行实时处理演示
simulate_realtime_processing()

# 多线程并发处理示例
class ConcurrentDistanceProcessor:
"""
支持并发处理的欧氏距离处理器
"""

def __init__(self, n_workers=4):
"""
初始化并发处理器

Args:
n_workers: 工作线程数量
"""
self.n_workers = n_workers
self.processors = [RealTimeDistanceProcessor() for _ in range(n_workers)]
self.current_worker = 0

def distribute_data(self, data_points):
"""
将数据点分发给不同的工作线程

Args:
data_points: 数据点列表
"""
for i, point in enumerate(data_points):
worker_id = i % self.n_workers
self.processors[worker_id].add_data_point(point)

def get_overall_statistics(self):
"""
获取所有工作线程的整体统计信息

Returns:
整体统计信息
"""
all_distances = []
for processor in self.processors:
with processor.lock:
all_distances.extend(processor.distances_history)

if not all_distances:
return {"count": 0}

distances = np.array(all_distances)
return {
"total_count": len(distances),
"overall_mean": np.mean(distances),
"overall_std": np.std(distances),
"overall_min": np.min(distances),
"overall_max": np.max(distances)
}

# 并发处理示例
print("\\n并发处理示例:")
concurrent_processor = ConcurrentDistanceProcessor(n_workers=3)

# 生成大量数据点
large_dataset = np.random.randn(1000, 4) # 1000个4维数据点

# 分发数据进行处理
start_time = time.time()
concurrent_processor.distribute_data(large_dataset)
processing_time = time.time() – start_time

# 获取整体统计信息
overall_stats = concurrent_processor.get_overall_statistics()

print(f"处理1000个数据点耗时: {processing_time:.4f}秒")
print("整体统计信息:")
for key, value in overall_stats.items():
if isinstance(value, float):
print(f" {key}: {value:.4f}")
else:
print(f" {key}: {value}")

性能基准测试 📊

为了更好地理解不同实现方式的性能差异，我们进行详细的基准测试：

import numpy as np
import time
import pandas as pd
from scipy.spatial.distance import cdist

def benchmark_euclidean_distance_methods():
"""
欧氏距离计算方法性能基准测试
"""
print("🚀 开始欧氏距离计算方法性能基准测试")
print("=" * 60)

# 测试不同数据规模
test_sizes = [
(100, 10), # 小规模：100个10维点
(1000, 50), # 中等规模：1000个50维点
(5000, 100), # 大规模：5000个100维点
]

results = []

for n_points, n_dims in test_sizes:
print(f"\\n📊 测试规模: {n_points}个{n_dims}维点")
print("-" * 40)

# 生成测试数据
np.random.seed(42)
points1 = np.random.rand(n_points, n_dims)
points2 = np.random.rand(n_points, n_dims)

# 方法1: 基础循环实现
def method1_basic_loop():
distances = np.zeros((n_points, n_points))
for i in range(n_points):
for j in range(n_points):
distances[i, j] = np.sqrt(np.sum((points1[i] – points2[j]) ** 2))
return distances

# 方法2: NumPy向量化实现
def method2_numpy_vectorized():
return np.sqrt(np.sum((points1[:, np.newaxis] – points2[np.newaxis, :]) ** 2, axis=2))

# 方法3: 优化的矩阵运算实现
def method3_optimized_matrix():
sum1 = np.sum(points1 ** 2, axis=1, keepdims=True)
sum2 = np.sum(points2 ** 2, axis=1)
dot_product = np.dot(points1, points2.T)
distances_squared = sum1 + sum2 – 2 * dot_product
distances_squared = np.maximum(distances_squared, 0) # 防止负数
return np.sqrt(distances_squared)

# 方法4: 使用SciPy库
def method4_scipy():
return cdist(points1, points2, metric='euclidean')

# 方法5: 使用NumPy linalg.norm
def method5_linalg_norm():
distances = np.zeros((n_points, n_points))
for i in range(n_points):
distances[i] = np.linalg.norm(points1[i] – points2, axis=1)
return distances

# 测试所有方法
methods = [
("基础循环", method1_basic_loop),
("NumPy向量化", method2_numpy_vectorized),
("优化矩阵运算", method3_optimized_matrix),
("SciPy库", method4_scipy),
("NumPy linalg.norm", method5_linalg_norm)
]

size_results = []

for method_name, method_func in methods:
try:
# 预热运行
_ = method_func()

# 实际测试
start_time = time.perf_counter()
result = method_func()
end_time = time.perf_counter()

execution_time = end_time – start_time
memory_usage = result.nbytes / (1024 * 1024) # MB

size_results.append({
'Method': method_name,
'Size': f"{n_points}×{n_dims}",
'Time (s)': execution_time,
'Memory (MB)': memory_usage,
'Speed Rank': 0 # 后续计算
})

print(f" {method_name:15s}: {execution_time:.6f}s ({memory_usage:.2f}MB)")

except Exception as e:
print(f" {method_name:15s}: 错误 – {str(e)}")
size_results.append({
'Method': method_name,
'Size': f"{n_points}×{n_dims}",
'Time (s)': float('inf'),
'Memory (MB)': 0,
'Speed Rank': 999
})

# 计算速度排名
valid_times = [r['Time (s)'] for r in size_results if r['Time (s)'] != float('inf')]
if valid_times:
min_time = min(valid_times)
for result in size_results:
if result['Time (s)'] != float('inf'):
result['Speed Rank'] = result['Time (s)'] / min_time

results.extend(size_results)

# 创建结果DataFrame
df = pd.DataFrame(results)

print("\\n🏆 性能排名总结:")
print("=" * 60)

# 按数据规模分组显示最佳方法
for size in df['Size'].unique():
size_data = df[df['Size'] == size]
best_method = size_data.loc[size_data['Time (s)'].idxmin()]
print(f"规模 {size}: 最佳方法是 '{best_method['Method']}' "
f"(耗时: {best_method['Time (s)']:.6f}s)")

return df

# 运行基准测试
benchmark_results = benchmark_euclidean_distance_methods()

# 创建性能对比图表数据
def create_performance_summary_chart():
"""
创建性能对比摘要图表
"""
summary_data = []

for size in benchmark_results['Size'].unique():
size_data = benchmark_results[benchmark_results['Size'] == size]
for _, row in size_data.iterrows():
summary_data.append({
'Size': row['Size'],
'Method': row['Method'],
'Normalized_Time': row['Speed Rank'],
'Execution_Time': row['Time (s)']
})

summary_df = pd.DataFrame(summary_data)
return summary_df

performance_summary = create_performance_summary_chart()

print("\\n📈 性能对比摘要:")
print(performance_summary.pivot(index='Method', columns='Size', values='Normalized_Time'))

# 内存效率分析
print("\\n💾 内存效率分析:")
memory_efficiency = benchmark_results.groupby('Method')['Memory (MB)'].mean()
print(memory_efficiency.sort_values())

最佳实践建议 💡

基于以上分析和测试，总结出以下最佳实践：

class EuclideanDistanceBestPractices:
"""
欧氏距离计算最佳实践指南
"""

@staticmethod
def choose_right_method(data_size, dimensionality, real_time=False):
"""
根据数据特征选择合适的计算方法

Args:
data_size: 数据规模
dimensionality: 数据维度
real_time: 是否实时处理

Returns:
推荐方法和说明
"""
recommendations = []

if data_size < 1000 and dimensionality < 50:
recommendations.append({
'method': 'NumPy向量化',
'reason': '小规模数据，向量化实现简单高效'
})
elif data_size >= 1000 or dimensionality >= 50:
recommendations.append({
'method': '优化矩阵运算',
'reason': '大规模数据，矩阵运算充分利用硬件加速'
})
elif real_time:
recommendations.append({
'method': '预计算缓存',
'reason': '实时场景下预先计算并缓存常用距离'
})

if dimensionality > 1000:
recommendations.append({
'method': '降维预处理',
'reason': '超高维数据建议先进行PCA等降维处理'
})

return recommendations

@staticmethod
def preprocessing_tips():
"""
数据预处理建议
"""
tips = [
"✅ 标准化特征：确保各维度具有相同的量纲",
"✅ 处理缺失值：使用均值填充或删除含缺失值的样本",
"✅ 特征选择：移除无关或冗余特征以提高效率",
"✅ 数据类型优化：使用float32而非float64可节省内存",
"✅ 批量处理：大数据集分批处理避免内存溢出"
]
return tips

@staticmethod
def optimization_strategies():
"""
性能优化策略
"""
strategies = [
"⚡ 使用NumPy内置函数而非手工实现",
"⚡ 充分利用广播机制减少显式循环",
"⚡ 预先分配结果数组避免动态扩展",
"⚡ 使用适当的数据类型（float32 vs float64）",
"⚡ 考虑使用并行计算处理独立任务",
"⚡ 缓存重复计算的结果",
"⚡ 在必要时使用近似算法（如LSH）"
]
return strategies

# 展示最佳实践建议
practices = EuclideanDistanceBestPractices()

print("🎯 欧氏距离计算最佳实践指南")
print("=" * 50)

# 方法选择建议
print("\\n📋 方法选择建议:")
recommendations = practices.choose_right_method(5000, 100, False)
for rec in recommendations:
print(f" • {rec['method']}: {rec['reason']}")

# 预处理建议
print("\\n🔧 数据预处理建议:")
for tip in practices.preprocessing_tips():
print(f" {tip}")

# 优化策略
print("\\n🚀 性能优化策略:")
for strategy in practices.optimization_strategies():
print(f" {strategy}")

# 实际应用中的注意事项
print("\\n⚠️ 实际应用注意事项:")
considerations = [
"• 数值稳定性：注意浮点运算精度问题",
"• 内存管理：监控内存使用避免溢出",
"• 算法复杂度：理解时间复杂度O(n²)的含义",
"• 结果验证：检查计算结果的合理性",
"• 错误处理：妥善处理边界情况和异常输入"
]
for consideration in considerations:
print(f" {consideration}")

总结与展望 🎯

欧氏距离作为最基础也是最重要的相似度度量方法，在数据分析和机器学习领域有着广泛的应用。通过本文的深入探讨，我们可以看到：

随着技术的发展，虽然出现了更多复杂的相似度度量方法，但欧氏距离凭借其直观性和高效性，仍然是许多应用的首选方案。未来，我们可以期待：

• 更高效的硬件加速支持
• 与深度学习框架的更好集成
• 针对特定领域的专业化改进
• 更智能的自适应算法选择

无论你是初学者还是经验丰富的数据科学家，掌握欧氏距离的计算和应用都将为你的工作带来巨大帮助。记住，最好的算法不是最复杂的，而是最适合你具体需求的那个！🌟

想要了解更多关于相似度度量和机器学习的内容，可以参考 scikit-learn官方文档 和 NumPy官方教程。

🙌 感谢你读到这里！ 🔍 技术之路没有捷径，但每一次阅读、思考和实践，都在悄悄拉近你与目标的距离。 💡 如果本文对你有帮助，不妨 👍 点赞、📌 收藏、📤 分享 给更多需要的朋友！ 💬 欢迎在评论区留下你的想法、疑问或建议，我会一一回复，我们一起交流、共同成长 🌿 🔔 关注我，不错过下一篇干货！我们下期再见！✨