别再只用KNN了！用Python手写LOF算法，实战识别信用卡欺诈与异常用户

用Python手写LOF算法：实战信用卡欺诈检测与参数调优全指南

在金融风控领域，识别异常交易如同大海捞针——传统方法如KNN往往力不从心。当欺诈行为伪装成正常交易，或正常用户突然改变消费模式时，基于全局距离的方法容易误判。这正是局部离群因子(LOF)算法的用武之地，它能敏锐捕捉局部密度变化，发现那些"在正常人群中显得不正常"的数据点。

1. 为什么LOF比KNN更适合金融风控

1.1 密度不均数据的检测困境

金融数据通常呈现不均匀分布特征：

• 同一用户在不同时段的交易金额可能相差数个数量级
• 高端客户与普通用户的消费模式密度截然不同
• 欺诈行为往往模仿正常交易模式，仅在细微处存在差异

传统KNN的三大局限：

1. 对全局距离敏感，无法适应不同区域的密度变化
2. 在高方差数据集中容易产生大量误报
3. 难以区分真正的异常与正常但罕见的行为模式

1.2 LOF的局部密度比较优势

LOF算法通过计算相对密度而非绝对距离，解决了上述痛点：

# KNN与LOF的核心区别示意 def knn_score(point, k): distances = [euclidean(point, x) for x in data] return sorted(distances)[k] # 返回第k近的距离 def lof_score(point, k): lrd_point = local_reachability_density(point, k) lrd_neighbors = [local_reachability_density(x, k) for x in get_neighbors(point, k)] return sum(lrd_neighbors)/(k * lrd_point) # 密度比值

典型业务场景中的表现对比：

2. 从零实现LOF算法

2.1 核心数学概念实现

LOF依赖的几个关键计算步骤：

import numpy as np from collections import defaultdict def k_distance(p, data, k): """计算第k距离及邻域""" distances = [np.linalg.norm(np.array(p)-np.array(x)) for x in data] sorted_dist = sorted(zip(distances, data), key=lambda x: x[0]) k_dist = sorted_dist[k][0] if k < len(sorted_dist) else sorted_dist[-1][0] neighbors = [x[1] for x in sorted_dist[:k+1] if x[1] is not p] return k_dist, neighbors def reachability_distance(p, o, data, k): """计算可达距离""" k_dist_o, _ = k_distance(o, data, k) dist_p_o = np.linalg.norm(np.array(p)-np.array(o)) return max(k_dist_o, dist_p_o)

2.2 完整LOF类实现

封装成可复用的Python类：

class LOFDetector: def __init__(self, k=20): self.k = k self._distance_cache = {} def _cached_distance(self, a, b): """带缓存的距离计算""" key = tuple(sorted((tuple(a), tuple(b)))) if key not in self._distance_cache: self._distance_cache[key] = np.linalg.norm(np.array(a)-np.array(b)) return self._distance_cache[key] def fit_predict(self, data): scores = [] for i, point in enumerate(data): # 计算局部可达密度 k_dist, neighbors = k_distance(point, data, self.k) lrd = len(neighbors) / sum( reachability_distance(point, n, data, self.k) for n in neighbors ) # 计算LOF分数 neighbor_lrds = [] for n in neighbors: n_k_dist, n_neighbors = k_distance(n, data, self.k) n_lrd = len(n_neighbors) / sum( reachability_distance(n, nn, data, self.k) for nn in n_neighbors ) neighbor_lrds.append(n_lrd) lof_score = sum(neighbor_lrds) / (len(neighbors) * lrd) scores.append((i, point, lof_score)) return sorted(scores, key=lambda x: x[2], reverse=True)

3. 信用卡欺诈检测实战

3.1 数据预处理关键步骤

使用Kaggle信用卡数据集时的特殊处理：

import pandas as pd from sklearn.preprocessing import RobustScaler def preprocess_credit_data(df): # 处理类别型特征 df = pd.get_dummies(df, columns=['merchant_category']) # 对金额进行鲁棒缩放 scaler = RobustScaler() df['amount_scaled'] = scaler.fit_transform(df[['amount']]) # 构造时间特征 df['hour'] = df['transaction_time'].dt.hour df['day_of_week'] = df['transaction_time'].dt.dayofweek # 选择最终特征 features = ['amount_scaled', 'hour', 'day_of_week'] + \ [c for c in df.columns if 'merchant_category_' in c] return df[features].values

3.2 参数k的选择策略

k值对结果的影响及选择方法：

网格搜索确定最优k值：

from sklearn.metrics import precision_at_k def find_optimal_k(data, labels, k_candidates): best_k = k_candidates[0] best_score = 0 for k in k_candidates: detector = LOFDetector(k=k) scores = detector.fit_predict(data) ordered_labels = [labels[i] for i, _, _ in scores] score = precision_at_k(ordered_labels, 100) # 考察前100个预测 if score > best_score: best_score = score best_k = k return best_k

4. 结果分析与业务解释

4.1 可视化技术

使用Pyplot进行多维数据展示：

import matplotlib.pyplot as plt from mpl_toolkits.mplot3d import Axes3D def plot_lof_results(data, scores, top_n=50): fig = plt.figure(figsize=(15, 10)) # 3D散点图 ax1 = fig.add_subplot(121, projection='3d') x, y, z = data[:,0], data[:,1], data[:,2] ax1.scatter(x, y, z, c='b', alpha=0.1) outliers = [scores[i][1] for i in range(top_n)] ox, oy, oz = zip(*outliers) ax1.scatter(ox, oy, oz, c='r', marker='x', s=100) # LOF分数分布 ax2 = fig.add_subplot(122) all_scores = [s[2] for s in scores] ax2.hist(all_scores, bins=50, alpha=0.7) ax2.axvline(x=np.mean(all_scores)+2*np.std(all_scores), color='r') plt.show()

4.2 业务规则融合

将LOF结果与实际业务规则结合：

def business_rules_validation(transaction, lof_score): rules = [ (transaction['amount'] > 10000 and lof_score > 1.5), (transaction['foreign'] and lof_score > 1.2), (transaction['hour'] in [2,3,4] and lof_score > 1.3), (lof_score > 2.0) # 极高LOF分数直接触发 ] return any(rules)

4.3 性能优化技巧

处理大规模数据时的加速方案：

from numba import jit import numpy as np @jit(nopython=True) def fast_euclidean(a, b): """使用numba加速的距离计算""" return np.sqrt(np.sum((a - b)**2)) class OptimizedLOF(LOFDetector): def __init__(self, k=20): super().__init__(k) self._distance_cache = {} def _cached_distance(self, a, b): key = (tuple(a), tuple(b)) if tuple(a) < tuple(b) else (tuple(b), tuple(a)) if key not in self._distance_cache: self._distance_cache[key] = fast_euclidean(np.array(a), np.array(b)) return self._distance_cache[key] def batch_predict(self, data, batch_size=1000): """分批处理大数据集""" scores = [] for i in range(0, len(data), batch_size): batch = data[i:i+batch_size] scores.extend(self.fit_predict(batch)) return sorted(scores, key=lambda x: x[2], reverse=True)

在实际项目中，LOF算法与业务场景的结合往往需要多次迭代。一个有效的实践方案是先用历史数据确定基准阈值，再通过A/B测试验证不同参数组合的效果。记住，没有放之四海皆准的最优参数，只有最适合当前业务场景的调参策略。