数据科学家高频面试真题精讲

y = f(x) + ε，其中 E[ε] = 0, Var(ε) = σ²

E[(y - ŷ)²] = Bias² + Variance + Irreducible Error

其中：
- Bias² = (E[ŷ] - f(x))²
- Variance = E[(ŷ - E[ŷ])²]
- Irreducible Error = σ²

总误差 = 偏差² + 方差 + 不可约误差

随着模型复杂度增加：
- 偏差 ↓（模型更能拟合真实函数）
- 方差 ↑（模型更容易过拟合）

z^(l) = W^(l)a^(l-1) + b^(l)
a^(l) = σ(z^(l))

其中：
- l: 层数
- W^(l): 第l层权重矩阵
- b^(l): 第l层偏置向量
- σ: 激活函数

L = 1/2 ||a^(L) - y||²

δ^(L) = ∂L/∂z^(L) = (a^(L) - y) ⊙ σ'(z^(L))

δ^(l) = ((W^(l+1))ᵀδ^(l+1)) ⊙ σ'(z^(l))

∂L/∂W^(l) = δ^(l)(a^(l-1))ᵀ
∂L/∂b^(l) = δ^(l)

∂L/∂W^(1) = δ^(1)(a^(0))ᵀ

其中 δ^(1) 通过链式法则计算：
δ^(1) = (W^(2))ᵀ(W^(3))ᵀ...(W^(L))ᵀδ^(L) ⊙ ∏σ'(z^(l))

如果 ||W^(l)|| < 1 且 |σ'(z^(l))| < 1，
则 δ^(1) ≈ ∏||W^(l)|| × ∏|σ'(z^(l))| → 0

σ(x) = 1/(1 + e^(-x))
σ'(x) = σ(x)(1 - σ(x)) ≤ 0.25

# 对于L层网络，梯度衰减系数：
gradient_decay = (0.25)^L

# 例如10层网络：(0.25)^10 ≈ 9.5×10^(-7)

# 如果权重过小（如标准正态分布N(0,1)）:
# 对于n_in个输入的层，每层输出方差约为 n_in × Var(w)
# 如果Var(w) < 1/n_in，信号会逐层衰减

# Xavier初始化：
Var(w) = 2/(n_in + n_out)

# He初始化（用于ReLU）：
Var(w) = 2/n_in

# ReLU: f(x) = max(0, x)
# 优势：正区间梯度为1，避免饱和
# 问题：负区间梯度为0（Dead ReLU）

# Leaky ReLU: f(x) = max(0.01x, x)
# ELU: f(x) = x if x>0 else α(e^x - 1)
# Swish: f(x) = x × sigmoid(x)

# 标准连接：H(x) = F(x)
# 残差连接：H(x) = F(x) + x

# 梯度流优势：
∂H/∂x = ∂F/∂x + 1

# 即使 ∂F/∂x → 0，梯度仍能通过恒等映射传播

# 标准化输入：x̂ = (x - μ)/σ
# 重参数化：y = γx̂ + β

# 优势：
1. 减少内部协变量偏移
2. 允许使用更大学习率
3. 减少对初始化的依赖
4. 有轻微正则化效果

# 防止梯度爆炸
if ||g|| > threshold:
    g = g × threshold / ||g||

def backward_pass(network, y_true, y_pred):
    """反向传播实现"""
    gradients = {}

    # 输出层误差
    delta = (y_pred - y_true) * sigmoid_derivative(network[-1]['z'])

    # 从输出层向输入层反向传播
    for i in reversed(range(len(network))):
        layer = network[i]

        # 计算权重和偏置梯度
        if i == 0:
            gradients[f'W{i}'] = np.dot(delta, X.T)
        else:
            gradients[f'W{i}'] = np.dot(delta, network[i-1]['a'].T)
        gradients[f'b{i}'] = np.sum(delta, axis=1, keepdims=True)

        # 计算下一层的误差（如果不是输入层）
        if i > 0:
            delta = np.dot(layer['W'].T, delta) * sigmoid_derivative(network[i-1]['z'])

    return gradients

def sigmoid_derivative(z):
    """Sigmoid导数"""
    s = 1 / (1 + np.exp(-z))
    return s * (1 - s)

# 给定编码器隐状态 h₁, h₂, ..., hₙ 和解码器状态 s_t

# 1. 计算注意力分数
e_{ti} = a(s_{t-1}, h_i) = v_a^T tanh(W_a s_{t-1} + U_a h_i)

# 2. 归一化得到注意力权重
α_{ti} = softmax(e_{ti}) = exp(e_{ti}) / Σⱼ exp(e_{tj})

# 3. 计算上下文向量
c_t = Σᵢ α_{ti} h_i

# 4. 生成输出
s_t = f(s_{t-1}, y_{t-1}, c_t)

# 输入序列 X = [x₁, x₂, ..., xₙ]

# 1. 线性变换得到 Q, K, V
Q = XW_Q  # Query矩阵
K = XW_K  # Key矩阵
V = XW_V  # Value矩阵

# 2. 计算注意力分数
Attention(Q,K,V) = softmax(QK^T/√d_k)V

其中 d_k 是缩放因子，防止梯度消失

# 将Q、K、V分成h个头
MultiHead(Q,K,V) = Concat(head₁, head₂, ..., headₕ)W_O

其中 headᵢ = Attention(QW_Q^i, KW_K^i, VW_V^i)

# 1. 多头自注意力
x' = MultiHeadAttention(x, x, x)
x = LayerNorm(x + x')  # 残差连接 + 层标准化

# 2. 前馈网络
x'' = FFN(x') = max(0, xW₁ + b₁)W₂ + b₂
x = LayerNorm(x' + x'')

# 正弦位置编码
PE(pos, 2i) = sin(pos / 10000^(2i/d_model))
PE(pos, 2i+1) = cos(pos / 10000^(2i/d_model))

# 优势：相对位置信息，支持任意长度序列

RNN: 串行计算
h_t = f(h_{t-1}, x_t)  # 必须等待 h_{t-1} 计算完成

Transformer: 并行计算
所有位置的注意力可以同时计算
计算复杂度：RNN O(n), Transformer O(1)

RNN: 信息传递路径长度 O(n)
- 梯度消失/爆炸问题
- 信息衰减严重

Transformer: 任意位置直接连接
- 路径长度 O(1)
- 直接建模全局依赖关系

# RNN 的表达限制
- 固定的递归结构
- 信息瓶颈在隐状态维度

# Transformer 的优势
- 灵活的注意力模式
- 多头机制捕获不同类型的关系
- 更强的特征提取能力

import torch
import torch.nn as nn
import math

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads):
        super().__init__()
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads

        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

    def scaled_dot_product_attention(self, Q, K, V, mask=None):
        # 计算注意力分数
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)

        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)

        # Softmax归一化
        attention_weights = torch.softmax(scores, dim=-1)

        # 加权求和
        output = torch.matmul(attention_weights, V)
        return output, attention_weights

    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)

        # 线性变换并重塑为多头
        Q = self.W_q(query).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        K = self.W_k(key).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        V = self.W_v(value).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)

        # 多头注意力
        attention_output, attention_weights = self.scaled_dot_product_attention(Q, K, V, mask)

        # 拼接多头输出
        attention_output = attention_output.transpose(1, 2).contiguous().view(
            batch_size, -1, self.d_model)

        # 最终线性变换
        output = self.W_o(attention_output)
        return output, attention_weights

import numpy as np
from collections import defaultdict
import math

class NaiveBayesClassifier:
    def __init__(self, alpha=1.0):
        """
        朴素贝叶斯分类器

        Parameters:
        alpha: 拉普拉斯平滑参数
        """
        self.alpha = alpha
        self.class_priors = {}  # P(Y=c)
        self.feature_probs = {}  # P(X_i=x|Y=c)
        self.classes = None
        self.feature_values = {}  # 记录每个特征的所有可能值

    def fit(self, X, y):
        """
        训练朴素贝叶斯模型

        Parameters:
        X: 特征矩阵 (n_samples, n_features)
        y: 目标变量 (n_samples,)
        """
        n_samples, n_features = X.shape
        self.classes = np.unique(y)

        # 记录每个特征的所有可能值（用于拉普拉斯平滑）
        for j in range(n_features):
            self.feature_values[j] = np.unique(X[:, j])

        # 计算类别先验概率 P(Y=c)
        class_counts = np.bincount(y)
        for i, class_label in enumerate(self.classes):
            self.class_priors[class_label] = class_counts[i] / n_samples

        # 计算条件概率 P(X_j=x|Y=c)
        self.feature_probs = {}
        for class_label in self.classes:
            self.feature_probs[class_label] = {}
            class_mask = (y == class_label)
            class_samples = X[class_mask]

            for j in range(n_features):
                self.feature_probs[class_label][j] = {}
                feature_values = self.feature_values[j]

                # 计算每个特征值的条件概率（拉普拉斯平滑）
                for value in feature_values:
                    count = np.sum(class_samples[:, j] == value)
                    # 拉普拉斯平滑: (count + alpha) / (class_size + alpha * |V|)
                    prob = (count + self.alpha) / (len(class_samples) + self.alpha * len(feature_values))
                    self.feature_probs[class_label][j][value] = prob

    def predict_proba(self, X):
        """
        预测类别概率

        Returns:
        概率矩阵 (n_samples, n_classes)
        """
        n_samples = X.shape[0]
        n_classes = len(self.classes)
        probas = np.zeros((n_samples, n_classes))

        for i, sample in enumerate(X):
            for j, class_label in enumerate(self.classes):
                # 计算 P(Y=c|X) ∝ P(Y=c) * ∏P(X_i|Y=c)
                log_prob = math.log(self.class_priors[class_label])

                for k, feature_value in enumerate(sample):
                    if feature_value in self.feature_probs[class_label][k]:
                        prob = self.feature_probs[class_label][k][feature_value]
                    else:
                        # 未见过的特征值，使用拉普拉斯平滑
                        prob = self.alpha / (sum(self.class_priors[class_label] * len(X)) +
                                           self.alpha * len(self.feature_values[k]))

                    log_prob += math.log(prob)

                probas[i, j] = log_prob

        # 转换为概率（避免数值下溢）
        probas = np.exp(probas - np.max(probas, axis=1, keepdims=True))
        probas = probas / np.sum(probas, axis=1, keepdims=True)

        return probas

    def predict(self, X):
        """预测类别"""
        probas = self.predict_proba(X)
        return self.classes[np.argmax(probas, axis=1)]

    def score(self, X, y):
        """计算准确率"""
        predictions = self.predict(X)
        return np.mean(predictions == y)

# 高斯朴素贝叶斯（连续特征）
class GaussianNaiveBayes:
    def __init__(self):
        self.class_priors = {}
        self.feature_means = {}  # μ_{c,i}
        self.feature_vars = {}   # σ²_{c,i}
        self.classes = None

    def fit(self, X, y):
        self.classes = np.unique(y)
        n_features = X.shape[1]

        # 计算先验概率
        class_counts = np.bincount(y)
        for i, class_label in enumerate(self.classes):
            self.class_priors[class_label] = class_counts[i] / len(y)

        # 计算每个类别下每个特征的统计量
        for class_label in self.classes:
            class_mask = (y == class_label)
            class_data = X[class_mask]

            self.feature_means[class_label] = np.mean(class_data, axis=0)
            self.feature_vars[class_label] = np.var(class_data, axis=0, ddof=1)

    def _gaussian_pdf(self, x, mean, var):
        """高斯概率密度函数"""
        return (1 / np.sqrt(2 * np.pi * var)) * np.exp(-0.5 * ((x - mean) ** 2) / var)

    def predict_proba(self, X):
        n_samples = X.shape[0]
        n_classes = len(self.classes)
        probas = np.zeros((n_samples, n_classes))

        for i, sample in enumerate(X):
            for j, class_label in enumerate(self.classes):
                # log P(Y=c|X) = log P(Y=c) + Σ log P(X_i|Y=c)
                log_prob = math.log(self.class_priors[class_label])

                for k in range(len(sample)):
                    mean = self.feature_means[class_label][k]
                    var = self.feature_vars[class_label][k]
                    pdf = self._gaussian_pdf(sample[k], mean, var)
                    log_prob += math.log(pdf + 1e-10)  # 避免log(0)

                probas[i, j] = log_prob

        # 归一化
        probas = np.exp(probas - np.max(probas, axis=1, keepdims=True))
        probas = probas / np.sum(probas, axis=1, keepdims=True)
        return probas

    def predict(self, X):
        probas = self.predict_proba(X)
        return self.classes[np.argmax(probas, axis=1)]

# 使用示例和测试
def test_naive_bayes():
    """测试朴素贝叶斯实现"""
    # 生成测试数据
    np.random.seed(42)
    X = np.random.randint(0, 3, (1000, 4))  # 离散特征
    y = np.random.randint(0, 2, 1000)       # 二分类

    # 训练模型
    nb = NaiveBayesClassifier(alpha=1.0)
    nb.fit(X, y)

    # 测试预测
    predictions = nb.predict(X[:10])
    probabilities = nb.predict_proba(X[:10])
    accuracy = nb.score(X, y)

    print(f"Predictions: {predictions}")
    print(f"Probabilities:\n{probabilities}")
    print(f"Accuracy: {accuracy:.3f}")

    # 测试高斯朴素贝叶斯
    X_continuous = np.random.randn(1000, 4)
    gnb = GaussianNaiveBayes()
    gnb.fit(X_continuous, y)
    accuracy_gaussian = gnb.score(X_continuous, y)
    print(f"Gaussian NB Accuracy: {accuracy_gaussian:.3f}")

if __name__ == "__main__":
    test_naive_bayes()

P(X₁, X₂, ..., Xₙ|Y=c) = ∏ᵢ P(Xᵢ|Y=c)

问题：
- 现实中特征往往相关（如身高体重、词汇共现）
- 违反假设会导致某些特征被重复计算

解决方案：
- 特征选择去除强相关特征
- 使用更复杂的模型（如贝叶斯网络）
- 特征工程降低相关性

训练集和测试集的特征分布相同

问题：
- 数据分布漂移
- 时序数据的非平稳性

解决方案：
- 定期重训练模型
- 在线学习更新参数
- 领域适应技术

P(Xᵢ|Y=c) ~ N(μc,i, σ²c,i)

问题：
- 特征分布可能非正态
- 单峰假设过于简单

解决方案：
- 特征变换（如Box-Cox）
- 使用混合高斯模型
- 非参数方法（如核密度估计）

正样本（点击/购买）: 通常 < 5%
负样本（未交互）: > 95%

特殊性：
- 负样本不等于不感兴趣（可能是未曝光）
- 用户行为稀疏性极高
- 不同用户的活跃度差异巨大

# 准确率在极度不平衡数据上会误导
# 例如：99%负样本的数据，全预测为负也有99%准确率

accuracy = (TP + TN) / (TP + FP + TN + FN)
# 在推荐系统中意义不大

# 精确率：推荐的物品中用户真正感兴趣的比例
precision = TP / (TP + FP)

# 召回率：用户感兴趣的物品中被推荐的比例
recall = TP / (TP + FN)

# F1-Score: 平衡精确率和召回率
f1_score = 2 * precision * recall / (precision + recall)

# PR曲线下面积：更适合不平衡数据
from sklearn.metrics import average_precision_score
ap_score = average_precision_score(y_true, y_scores)

# NDCG: 考虑位置信息的排序质量
def ndcg_at_k(y_true, y_scores, k=10):
    """
    计算NDCG@K
    """
    # 按预测分数排序
    order = np.argsort(y_scores)[::-1]
    y_true_sorted = np.take(y_true, order[:k])

    # 计算DCG
    dcg = y_true_sorted[0]
    for i in range(1, len(y_true_sorted)):
        dcg += y_true_sorted[i] / np.log2(i + 2)

    # 计算IDCG（理想情况）
    y_true_ideal = np.sort(y_true)[::-1][:k]
    idcg = y_true_ideal[0]
    for i in range(1, len(y_true_ideal)):
        idcg += y_true_ideal[i] / np.log2(i + 2)

    return dcg / idcg if idcg > 0 else 0

# MAP: 平均精确率
def mean_average_precision(y_true, y_scores):
    """计算MAP"""
    order = np.argsort(y_scores)[::-1]
    y_true_sorted = np.take(y_true, order)

    precisions = []
    relevant_count = 0

    for i, relevant in enumerate(y_true_sorted):
        if relevant:
            relevant_count += 1
            precision = relevant_count / (i + 1)
            precisions.append(precision)

    return np.mean(precisions) if precisions else 0

# CTR (Click Through Rate)
ctr = clicks / impressions

# CVR (Conversion Rate)
cvr = conversions / clicks

# 用户满意度指标
user_satisfaction = {
    'session_length': avg_session_time,
    'return_rate': returning_users / total_users,
    'diversity': len(unique_categories_clicked) / total_recommendations
}

class NegativeSampler:
    def __init__(self, strategy='random', ratio=4):
        self.strategy = strategy
        self.ratio = ratio  # 负正样本比例

    def random_sampling(self, user_id, positive_items, all_items):
        """随机负采样"""
        candidate_items = all_items - set(positive_items)
        neg_samples = np.random.choice(
            list(candidate_items),
            size=len(positive_items) * self.ratio,
            replace=False
        )
        return neg_samples

    def popularity_based_sampling(self, user_id, positive_items, item_popularity):
        """基于流行度的负采样"""
        # 按流行度加权采样，热门物品更容易被选为负样本
        candidate_items = list(set(item_popularity.keys()) - set(positive_items))
        probs = [item_popularity[item] for item in candidate_items]
        probs = np.array(probs) / np.sum(probs)

        neg_samples = np.random.choice(
            candidate_items,
            size=len(positive_items) * self.ratio,
            p=probs,
            replace=False
        )
        return neg_samples

    def hard_negative_sampling(self, user_id, positive_items, model, all_items):
        """困难负样本挖掘"""
        candidate_items = all_items - set(positive_items)

        # 预测候选物品的分数
        scores = model.predict(user_id, list(candidate_items))

        # 选择分数较高的作为困难负样本
        hard_negatives = np.argsort(scores)[-len(positive_items) * self.ratio:]
        return [candidate_items[i] for i in hard_negatives]

def data_augmentation_for_recommender(user_item_matrix):
    """推荐系统数据增强"""

    # 1. 用户行为序列增强
    def sequence_augmentation(user_sequence):
        # 随机mask某些交互
        masked_seq = user_sequence.copy()
        mask_ratio = 0.1
        mask_indices = np.random.choice(
            len(masked_seq),
            int(len(masked_seq) * mask_ratio),
            replace=False
        )
        for idx in mask_indices:
            masked_seq[idx] = 0  # mask token
        return masked_seq

    # 2. 相似用户行为迁移
    def user_behavior_transfer(target_user, similar_users, similarity_threshold=0.8):
        # 将相似用户的行为以一定概率迁移给目标用户
        augmented_behaviors = []
        for similar_user in similar_users:
            if similarity_score(target_user, similar_user) > similarity_threshold:
                # 以较低概率采用相似用户的行为
                transfer_prob = 0.1
                for item in similar_user.interactions:
                    if np.random.random() < transfer_prob:
                        augmented_behaviors.append(item)
        return augmented_behaviors

    return augmented_data

class ImbalancedLoss:
    def __init__(self, loss_type='focal', alpha=0.25, gamma=2.0):
        self.loss_type = loss_type
        self.alpha = alpha
        self.gamma = gamma

    def focal_loss(self, y_true, y_pred):
        """
        Focal Loss: 解决类别不平衡和困难样本问题
        FL(p_t) = -α_t * (1-p_t)^γ * log(p_t)
        """
        epsilon = 1e-8
        y_pred = np.clip(y_pred, epsilon, 1 - epsilon)

        # 计算pt
        p_t = np.where(y_true == 1, y_pred, 1 - y_pred)

        # 计算alpha_t
        alpha_t = np.where(y_true == 1, self.alpha, 1 - self.alpha)

        # 计算focal loss
        focal_loss = -alpha_t * np.power(1 - p_t, self.gamma) * np.log(p_t)
        return np.mean(focal_loss)

    def weighted_bce_loss(self, y_true, y_pred, pos_weight=10):
        """
        加权二元交叉熵损失
        """
        epsilon = 1e-8
        y_pred = np.clip(y_pred, epsilon, 1 - epsilon)

        loss = -(y_true * pos_weight * np.log(y_pred) +
                (1 - y_true) * np.log(1 - y_pred))
        return np.mean(loss)

    def ghm_loss(self, y_true, y_pred, bins=10, alpha=0.75):
        """
        Gradient Harmonizing Mechanism Loss
        根据梯度密度重新加权样本
        """
        # 计算梯度模长
        gradient = np.abs(y_pred - y_true)

        # 构建梯度直方图
        hist, bin_edges = np.histogram(gradient, bins=bins)

        # 计算梯度密度
        gradient_density = np.zeros_like(gradient)
        for i in range(len(bin_edges) - 1):
            mask = (gradient >= bin_edges[i]) & (gradient < bin_edges[i + 1])
            gradient_density[mask] = hist[i]

        # 计算权重
        N = len(y_true)
        weights = N / (gradient_density + 1e-8)
        weights = np.power(weights, alpha)

        # 加权BCE损失
        epsilon = 1e-8
        y_pred = np.clip(y_pred, epsilon, 1 - epsilon)
        loss = -(y_true * np.log(y_pred) + (1 - y_true) * np.log(1 - y_pred))

        return np.mean(weights * loss)

class ImbalancedRecommender:
    def __init__(self, embedding_dim=64, use_class_weights=True):
        self.embedding_dim = embedding_dim
        self.use_class_weights = use_class_weights

    def build_model_with_class_weights(self, pos_weight):
        """
        构建带类别权重的推荐模型
        """
        import tensorflow as tf

        # 用户和物品嵌入
        user_input = tf.keras.Input(shape=(), name='user_id')
        item_input = tf.keras.Input(shape=(), name='item_id')

        user_embedding = tf.keras.layers.Embedding(
            self.n_users, self.embedding_dim)(user_input)
        item_embedding = tf.keras.layers.Embedding(
            self.n_items, self.embedding_dim)(item_input)

        # 特征交互
        dot_product = tf.keras.layers.Dot(axes=1)([user_embedding, item_embedding])

        # 预测层
        output = tf.keras.layers.Dense(1, activation='sigmoid')(dot_product)

        model = tf.keras.Model(inputs=[user_input, item_input], outputs=output)

        # 使用加权损失函数
        model.compile(
            optimizer='adam',
            loss=tf.keras.losses.BinaryCrossentropy(),
            weighted_metrics=['accuracy']
        )

        return model

    def ensemble_with_cost_sensitive_learning(self, models, cost_matrix):
        """
        集成学习 + 代价敏感学习
        """
        def cost_sensitive_predict(predictions, cost_matrix):
            """
            基于代价矩阵的预测
            cost_matrix: [[C(0|0), C(1|0)], [C(0|1), C(1|1)]]
            """
            expected_costs = []
            for pred in predictions:
                cost_pred_0 = pred * cost_matrix[0][1] + (1-pred) * cost_matrix[0][0]
                cost_pred_1 = pred * cost_matrix[1][1] + (1-pred) * cost_matrix[1][0]
                expected_costs.append([cost_pred_0, cost_pred_1])

            # 选择期望代价最小的类别
            return np.argmin(expected_costs, axis=1)

        # 集成多个模型的预测
        ensemble_predictions = []
        for model in models:
            pred = model.predict(X_test)
            ensemble_predictions.append(pred)

        # 平均预测结果
        avg_predictions = np.mean(ensemble_predictions, axis=0)

        # 基于代价矩阵做最终决策
        final_predictions = cost_sensitive_predict(avg_predictions, cost_matrix)

        return final_predictions

class OnlineImbalancedOptimizer:
    def __init__(self, initial_threshold=0.5, adaptation_rate=0.01):
        self.threshold = initial_threshold
        self.adaptation_rate = adaptation_rate
        self.performance_history = []

    def adaptive_threshold_tuning(self, y_true, y_pred, target_metric='f1'):
        """
        自适应阈值调优
        """
        best_threshold = 0.5
        best_score = 0

        # 在不同阈值下评估性能
        for threshold in np.arange(0.1, 0.9, 0.05):
            y_pred_binary = (y_pred >= threshold).astype(int)

            if target_metric == 'f1':
                score = f1_score(y_true, y_pred_binary)
            elif target_metric == 'precision':
                score = precision_score(y_true, y_pred_binary)
            elif target_metric == 'recall':
                score = recall_score(y_true, y_pred_binary)

            if score > best_score:
                best_score = score
                best_threshold = threshold

        # 平滑更新阈值
        self.threshold = (1 - self.adaptation_rate) * self.threshold + \
                        self.adaptation_rate * best_threshold

        return best_threshold, best_score

    def online_hard_example_mining(self, model, user_item_pairs, window_size=1000):
        """
        在线困难样本挖掘
        """
        hard_examples = []

        for i in range(0, len(user_item_pairs), window_size):
            batch = user_item_pairs[i:i+window_size]
            predictions = model.predict(batch)

            # 找出预测困难的样本（预测概率接近0.5的）
            uncertainty = np.abs(predictions - 0.5)
            hard_indices = np.argsort(uncertainty)[:int(len(batch) * 0.1)]

            hard_examples.extend([batch[idx] for idx in hard_indices])

        return hard_examples

"""
实时反作弊系统架构

数据流：
用户行为 → 实时特征提取 → 模型预测 → 风险决策 → 业务动作

核心组件：
1. 数据接入层：Kafka消息队列
2. 特征计算层：Flink实时计算
3. 模型服务层：TensorFlow Serving / TorchServe
4. 决策引擎：规则引擎 + ML模型
5. 存储层：Redis (热数据) + HBase (历史数据)
6. 监控层：实时监控 + 告警
"""

class AntiFraudSystem:
    def __init__(self):
        self.feature_extractor = RealTimeFeatureExtractor()
        self.model_service = ModelService()
        self.rule_engine = RuleEngine()
        self.decision_engine = DecisionEngine()

    def process_event(self, event):
        """处理单个用户事件"""
        try:
            # 1. 特征提取
            features = self.feature_extractor.extract(event)

            # 2. 模型预测
            risk_score = self.model_service.predict(features)

            # 3. 规则检查
            rule_result = self.rule_engine.check(event, features)

            # 4. 综合决策
            decision = self.decision_engine.decide(risk_score, rule_result)

            # 5. 执行动作
            return self.execute_action(event, decision)

        except Exception as e:
            # 降级处理：系统异常时的安全策略
            return self.fallback_decision(event)

    def fallback_decision(self, event):
        """系统异常时的降级决策"""
        # 基于简单规则的快速判断
        if event.get('amount', 0) > 10000:  # 大额交易
            return {'action': 'review', 'confidence': 0.5}
        return {'action': 'pass', 'confidence': 0.8}

class RealTimeFeatureExtractor:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379)
        self.feature_cache = {}

    def extract(self, event):
        """实时特征提取"""
        user_id = event['user_id']
        device_id = event.get('device_id')
        timestamp = event['timestamp']

        features = {}

        # 1. 基础特征
        features.update(self._extract_basic_features(event))

        # 2. 统计特征（时间窗口聚合）
        features.update(self._extract_statistical_features(user_id, timestamp))

        # 3. 设备指纹特征
        features.update(self._extract_device_features(device_id, event))

        # 4. 行为序列特征
        features.update(self._extract_sequence_features(user_id, event))

        # 5. 图特征（关系网络）
        features.update(self._extract_graph_features(user_id, device_id))

        return features

    def _extract_basic_features(self, event):
        """基础特征：直接从事件中提取"""
        return {
            'amount': event.get('amount', 0),
            'hour_of_day': datetime.fromtimestamp(event['timestamp']).hour,
            'day_of_week': datetime.fromtimestamp(event['timestamp']).weekday(),
            'transaction_type': event.get('type', 'unknown'),
            'channel': event.get('channel', 'unknown')
        }

    def _extract_statistical_features(self, user_id, timestamp):
        """统计特征：时间窗口内的聚合统计"""
        features = {}

        # 定义时间窗口
        windows = [300, 1800, 3600, 86400]  # 5min, 30min, 1hour, 1day

        for window in windows:
            window_key = f"user_stats_{user_id}_{window}"

            # 从Redis获取时间窗口内的统计数据
            stats = self._get_window_stats(user_id, timestamp - window, timestamp)

            features.update({
                f'txn_count_{window}s': stats.get('count', 0),
                f'total_amount_{window}s': stats.get('total_amount', 0),
                f'avg_amount_{window}s': stats.get('avg_amount', 0),
                f'unique_merchants_{window}s': stats.get('unique_merchants', 0),
                f'unique_locations_{window}s': stats.get('unique_locations', 0)
            })

        return features

    def _extract_device_features(self, device_id, event):
        """设备指纹特征"""
        if not device_id:
            return {}

        # 设备基础信息
        device_info = {
            'os_type': event.get('os_type', 'unknown'),
            'app_version': event.get('app_version', 'unknown'),
            'network_type': event.get('network_type', 'unknown'),
            'is_rooted': event.get('is_rooted', False),
            'is_emulator': event.get('is_emulator', False)
        }

        # 设备行为统计
        device_stats = self._get_device_stats(device_id)
        device_info.update({
            'device_user_count': device_stats.get('user_count', 1),
            'device_txn_count_24h': device_stats.get('txn_count_24h', 0),
            'device_first_seen_days': device_stats.get('first_seen_days', 0)
        })

        return device_info

    def _extract_sequence_features(self, user_id, event):
        """行为序列特征"""
        # 获取用户近期行为序列
        recent_actions = self._get_user_action_sequence(user_id, limit=50)

        if not recent_actions:
            return {}

        # 计算序列特征
        features = {}

        # 时间间隔特征
        time_intervals = [
            recent_actions[i]['timestamp'] - recent_actions[i-1]['timestamp']
            for i in range(1, len(recent_actions))
        ]

        if time_intervals:
            features.update({
                'avg_time_interval': np.mean(time_intervals),
                'std_time_interval': np.std(time_intervals),
                'min_time_interval': np.min(time_intervals),
                'max_time_interval': np.max(time_intervals)
            })

        # 行为模式特征
        action_types = [action['type'] for action in recent_actions]
        features.update({
            'action_diversity': len(set(action_types)),
            'most_common_action': max(set(action_types), key=action_types.count),
            'action_pattern_score': self._calculate_pattern_score(action_types)
        })

        return features

    def _extract_graph_features(self, user_id, device_id):
        """图特征：基于用户关系网络"""
        features = {}

        # 用户-设备图特征
        if device_id:
            shared_device_users = self._get_shared_device_users(device_id)
            features['shared_device_user_count'] = len(shared_device_users)
            features['shared_device_risk_score'] = self._calculate_shared_device_risk(shared_device_users)

        # 用户-商户图特征
        frequent_merchants = self._get_user_frequent_merchants(user_id)
        features['frequent_merchant_count'] = len(frequent_merchants)
        features['merchant_risk_score'] = self._calculate_merchant_risk(frequent_merchants)

        return features

    def _get_window_stats(self, user_id, start_time, end_time):
        """获取时间窗口内的统计数据"""
        # 实际实现中会查询Redis/数据库
        # 这里返回模拟数据
        return {
            'count': np.random.randint(0, 10),
            'total_amount': np.random.uniform(0, 1000),
            'avg_amount': np.random.uniform(0, 200),
            'unique_merchants': np.random.randint(1, 5),
            'unique_locations': np.random.randint(1, 3)
        }

class FraudDetectionModel:
    def __init__(self, model_type='ensemble'):
        self.model_type = model_type
        self.models = {}
        self.feature_importance = {}

    def build_ensemble_model(self, X_train, y_train):
        """构建集成模型"""
        from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
        from sklearn.linear_model import LogisticRegression
        from lightgbm import LGBMClassifier

        # 基础模型
        models = {
            'rf': RandomForestClassifier(
                n_estimators=100,
                max_depth=10,
                class_weight='balanced',
                random_state=42
            ),
            'gbdt': GradientBoostingClassifier(
                n_estimators=100,
                learning_rate=0.1,
                max_depth=6,
                random_state=42
            ),
            'lr': LogisticRegression(
                class_weight='balanced',
                random_state=42
            ),
            'lgb': LGBMClassifier(
                n_estimators=100,
                learning_rate=0.1,
                num_leaves=31,
                class_weight='balanced',
                random_state=42
            )
        }

        # 训练基础模型
        for name, model in models.items():
            model.fit(X_train, y_train)
            self.models[name] = model

        # 模型融合（Stacking）
        self._build_meta_model(X_train, y_train)

    def _build_meta_model(self, X_train, y_train):
        """构建元模型进行模型融合"""
        from sklearn.model_selection import cross_val_predict

        # 生成元特征
        meta_features = np.zeros((X_train.shape[0], len(self.models)))

        for i, (name, model) in enumerate(self.models.items()):
            # 使用交叉验证生成元特征，避免过拟合
            meta_features[:, i] = cross_val_predict(
                model, X_train, y_train,
                cv=5, method='predict_proba'
            )[:, 1]

        # 训练元模型
        meta_model = LogisticRegression(random_state=42)
        meta_model.fit(meta_features, y_train)
        self.models['meta'] = meta_model

    def predict_proba(self, X):
        """预测概率"""
        if self.model_type == 'ensemble':
            return self._ensemble_predict_proba(X)
        else:
            return self.models[self.model_type].predict_proba(X)[:, 1]

    def _ensemble_predict_proba(self, X):
        """集成预测"""
        # 基础模型预测
        base_predictions = np.zeros((X.shape[0], len(self.models) - 1))

        for i, (name, model) in enumerate(self.models.items()):
            if name != 'meta':
                base_predictions[:, i] = model.predict_proba(X)[:, 1]

        # 元模型预测
        final_predictions = self.models['meta'].predict_proba(base_predictions)[:, 1]

        return final_predictions

    def get_feature_importance(self):
        """获取特征重要性"""
        importance_dict = {}

        for name, model in self.models.items():
            if hasattr(model, 'feature_importances_'):
                importance_dict[name] = model.feature_importances_
            elif hasattr(model, 'coef_'):
                importance_dict[name] = np.abs(model.coef_[0])

        return importance_dict

class DecisionEngine:
    def __init__(self):
        self.rules = self._load_rules()
        self.thresholds = {
            'high_risk': 0.8,
            'medium_risk': 0.5,
            'low_risk': 0.2
        }

    def decide(self, risk_score, rule_result, event_context):
        """综合决策"""
        decision = {
            'action': 'pass',
            'confidence': 0.0,
            'reason': [],
            'risk_level': 'low'
        }

        # 1. 规则决策
        if rule_result['triggered']:
            decision['action'] = rule_result['action']
            decision['reason'].extend(rule_result['reasons'])
            decision['confidence'] = max(decision['confidence'], rule_result['confidence'])

        # 2. 模型决策
        if risk_score >= self.thresholds['high_risk']:
            decision['action'] = 'block'
            decision['risk_level'] = 'high'
            decision['confidence'] = max(decision['confidence'], risk_score)
            decision['reason'].append(f'High model risk score: {risk_score:.3f}')

        elif risk_score >= self.thresholds['medium_risk']:
            if decision['action'] == 'pass':  # 只有在规则未触发时才设置为review
                decision['action'] = 'review'
                decision['risk_level'] = 'medium'
            decision['confidence'] = max(decision['confidence'], risk_score)
            decision['reason'].append(f'Medium model risk score: {risk_score:.3f}')

        # 3. 业务上下文调整
        decision = self._adjust_by_context(decision, event_context)

        return decision

    def _adjust_by_context(self, decision, context):
        """根据业务上下文调整决策"""
        # VIP用户特殊处理
        if context.get('user_level') == 'VIP':
            if decision['action'] == 'block' and decision['confidence'] < 0.9:
                decision['action'] = 'review'
                decision['reason'].append('VIP user protection')

        # 小额交易放宽
        if context.get('amount', 0) < 100:
            if decision['action'] == 'review' and decision['confidence'] < 0.7:
                decision['action'] = 'pass'
                decision['reason'].append('Small amount transaction')

        # 业务高峰期策略调整
        if context.get('is_peak_hour', False):
            # 高峰期适当放宽，避免影响用户体验
            if decision['action'] == 'review' and decision['confidence'] < 0.6:
                decision['action'] = 'pass'
                decision['reason'].append('Peak hour adjustment')

        return decision

    def _load_rules(self):
        """加载规则配置"""
        return [
            {
                'name': 'high_frequency_rule',
                'condition': lambda features: features.get('txn_count_300s', 0) > 10,
                'action': 'block',
                'confidence': 0.9
            },
            {
                'name': 'large_amount_rule',
                'condition': lambda features: features.get('amount', 0) > 50000,
                'action': 'review',
                'confidence': 0.8
            },
            {
                'name': 'suspicious_device_rule',
                'condition': lambda features: features.get('device_user_count', 1) > 5,
                'action': 'review',
                'confidence': 0.7
            }
        ]

class SystemMonitor:
    def __init__(self):
        self.metrics = {
            'throughput': 0,
            'latency': [],
            'accuracy': 0,
            'false_positive_rate': 0,
            'false_negative_rate': 0
        }

    def log_prediction(self, event, prediction, actual_result=None):
        """记录预测结果用于监控"""
        # 记录延迟
        processing_time = time.time() - event['timestamp']
        self.metrics['latency'].append(processing_time)

        # 记录吞吐量
        self.metrics['throughput'] += 1

        # 如果有真实标签，计算准确性指标
        if actual_result is not None:
            self._update_accuracy_metrics(prediction, actual_result)

        # 异常检测
        if processing_time > 1.0:  # 超过1秒认为异常
            self._alert('High latency detected', {
                'processing_time': processing_time,
                'event_id': event.get('event_id')
            })

    def _update_accuracy_metrics(self, prediction, actual):
        """更新准确性指标"""
        # 这里需要实现滑动窗口的准确性计算
        pass

    def generate_report(self):
        """生成监控报告"""
        return {
            'avg_latency': np.mean(self.metrics['latency'][-1000:]),  # 最近1000次
            'p95_latency': np.percentile(self.metrics['latency'][-1000:], 95),
            'throughput': self.metrics['throughput'],
            'accuracy': self.metrics['accuracy'],
            'fpr': self.metrics['false_positive_rate'],
            'fnr': self.metrics['false_negative_rate']
        }

# 在线学习和模型更新
class OnlineLearning:
    def __init__(self, model, learning_rate=0.01):
        self.model = model
        self.learning_rate = learning_rate
        self.feedback_buffer = []

    def collect_feedback(self, event_id, prediction, actual_label):
        """收集反馈数据"""
        self.feedback_buffer.append({
            'event_id': event_id,
            'prediction': prediction,
            'actual': actual_label,
            'timestamp': time.time()
        })

        # 批量更新模型
        if len(self.feedback_buffer) >= 1000:
            self._update_model()

    def _update_model(self):
        """增量更新模型"""
        # 提取特征和标签
        features = []
        labels = []

        for feedback in self.feedback_buffer:
            # 这里需要重新提取特征
            feature = self._reconstruct_features(feedback['event_id'])
            features.append(feature)
            labels.append(feedback['actual'])

        # 增量学习（这里简化处理）
        X = np.array(features)
        y = np.array(labels)

        # 使用SGD进行增量更新
        self.model.partial_fit(X, y)

        # 清空缓冲区
        self.feedback_buffer = []