大模型应用：从交易行为到实时反欺诈：向量数据库驱动的智能风控实践.33

{
  "user_reg_days": 180,
  "history_trade_count": 56,
  "trade_amount": 299.9,
  "trade_hour": 20,
  "is_common_device": 1,
  "ip_city": "北京"
}

import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
import chromadb
from chromadb.config import Settings
import os
import logging

# ===================== 1. 配置详细日志（控制台+文件） =====================
# 日志配置：输出级别、格式、存储路径
log_dir = "./logs"
if not os.path.exists(log_dir):
    os.makedirs(log_dir)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.StreamHandler(),  # 控制台输出
        logging.FileHandler(f"{log_dir}/fraud_detection.log", encoding="utf-8")  # 文件存储
    ]
)
logger = logging.getLogger("交易反欺诈系统")

# ===================== 2. 初始化ChromaDB（适配新版） =====================
persist_directory = "./chroma_persist"
if not os.path.exists(persist_directory):
    os.makedirs(persist_directory)
    logger.info(f"创建ChromaDB持久化目录：{persist_directory}")

# 初始化持久化客户端
client = chromadb.PersistentClient(
    path=persist_directory,
    settings=Settings(
        allow_reset=True,
        anonymized_telemetry=False
    )
)
client.reset()
logger.info("重置ChromaDB数据库（测试环境）")

# ===================== 3. 定义核心配置（便于维护） =====================
# 风控阈值配置
FRAUD_THRESHOLD = 0.7  # 相似度阈值：低于此值判定为异常
TOP_K = 3  # 检索相似向量数量
USER_ID = "U001"  # 目标用户ID
NORMAL_TRADE_NUM = 4  # 正常交易样本数

# 输出核心配置日志
logger.info("="*50)
logger.info("【核心风控配置】")
logger.info(f"异常交易判定阈值：余弦相似度 < {FRAUD_THRESHOLD}")
logger.info(f"检索相似向量数量：Top-{TOP_K}")
logger.info(f"目标分析用户：{USER_ID}")
logger.info(f"正常交易样本数量：{NORMAL_TRADE_NUM}")
logger.info("="*50)

# ===================== 4. 数据预处理与向量化 =====================
# 模拟交易数据
data = pd.DataFrame({
    "user_id": ["U001", "U001", "U001", "U001", "U001", "U001"],
    "user_reg_days": [180, 180, 180, 180, 180, 180],
    "history_trade_count": [56, 57, 58, 59, 60, 61],
    "trade_amount": [299.9, 199.0, 399.5, 499.0, 10.0, 9999.0],
    "trade_hour": [20, 19, 21, 20, 3, 23],
    "is_common_device": [1, 1, 1, 1, 0, 0]
})
logger.info(f"加载交易数据：共{len(data)}条，其中前{NORMAL_TRADE_NUM}条为正常交易，后{len(data)-NORMAL_TRADE_NUM}条为待检测交易")

# 特征预处理
scaler = MinMaxScaler(feature_range=(0, 1))
feature_cols = ["user_reg_days", "history_trade_count", "trade_amount", "trade_hour", "is_common_device"]
data_scaled = scaler.fit_transform(data[feature_cols])
vectors = data_scaled.tolist()
metadatas = [{"user_id": row["user_id"], "trade_id": f"T{idx}"} for idx, row in data.iterrows()]
ids = [f"T{idx}" for idx in range(len(vectors))]

# 输出向量维度日志
logger.info(f"特征列：{feature_cols}")
logger.info(f"向量维度：{len(vectors[0])}维")
logger.info(f"示例向量（T0）：{vectors[0]}")

# ===================== 5. 正常交易向量入库 =====================
collection_name = f"user_{USER_ID}_normal_trades"
# 删除旧集合
try:
    client.delete_collection(name=collection_name)
    logger.info(f"删除已存在的集合：{collection_name}")
except Exception as e:
    logger.warning(f"集合{collection_name}不存在，无需删除：{e}")

# 创建新集合
collection = client.create_collection(
    name=collection_name,
    metadata={"hnsw:space": "cosine"}  # 余弦相似度空间
)
logger.info(f"创建用户{USER_ID}的正常交易向量集合：{collection_name}")

# 入库正常交易向量
normal_vectors = vectors[:NORMAL_TRADE_NUM]
normal_metadatas = metadatas[:NORMAL_TRADE_NUM]
normal_ids = ids[:NORMAL_TRADE_NUM]

try:
    collection.add(
        embeddings=normal_vectors,
        metadatas=normal_metadatas,
        ids=normal_ids
    )
    logger.info(f"正常交易向量入库完成：{len(normal_vectors)}条")
    logger.info(f"入库向量ID：{normal_ids}")
    logger.info(f"集合当前向量总数：{collection.count()}")
except Exception as e:
    logger.error(f"正常向量入库失败：{str(e)}", exc_info=True)

# ===================== 6. 异常交易检测（带详细日志） =====================
logger.info("\n" + "="*50)
logger.info("开始检测异常交易")
logger.info("="*50)

# 待检测向量（异常样本）
test_vectors = vectors[NORMAL_TRADE_NUM:]
test_metadatas = metadatas[NORMAL_TRADE_NUM:]
test_ids = ids[NORMAL_TRADE_NUM:]

for idx, (vector, meta, trade_id) in enumerate(zip(test_vectors, test_metadatas, test_ids)):
    logger.info(f"\n--- 检测交易 {trade_id} ---")
    # 输出待检测交易原始特征
    raw_data = data.iloc[NORMAL_TRADE_NUM + idx]
    logger.info(f"原始特征：{raw_data.to_dict()}")
    logger.info(f"待检测向量：{vector}")
    
    # 检索相似向量
    try:
        results = collection.query(
            query_embeddings=[vector],
            n_results=TOP_K,
            where={"user_id": USER_ID}
        )
        # 解析检索结果
        retrieved_ids = results["ids"][0]
        retrieved_distances = results["distances"][0]
        retrieved_metadatas = results["metadatas"][0]
        
        # 输出检索结果日志
        logger.info(f"检索到Top-{TOP_K}相似正常交易：{retrieved_ids}")
        logger.info(f"余弦距离（越小越相似）：{[round(d, 4) for d in retrieved_distances]}")
        
        # 计算平均相似度（余弦相似度=1-余弦距离）
        avg_distance = np.mean(retrieved_distances) if retrieved_distances else 1.0
        avg_similarity = 1 - avg_distance
        logger.info(f"计算过程：平均余弦距离={round(avg_distance, 4)} → 平均余弦相似度={round(avg_similarity, 4)}")
        
        # 风控决策
        if avg_similarity < FRAUD_THRESHOLD:
            logger.warning(f"【异常交易判定】交易{trade_id}：平均相似度{round(avg_similarity, 4)} < 阈值{FRAUD_THRESHOLD}")
            logger.warning(f"判定原因：该交易行为向量与用户{USER_ID}的正常交易向量相似度过低，偏离正常行为模式")
            print(f"\n交易 {trade_id}：平均相似度 {avg_similarity:.2f} → 🚨 异常交易，触发风控")
        else:
            logger.info(f"【正常交易判定】交易{trade_id}：平均相似度{round(avg_similarity, 4)} ≥ 阈值{FRAUD_THRESHOLD}")
            logger.info(f"判定原因：该交易行为向量与用户{USER_ID}的正常交易向量高度相似，符合正常行为模式")
            print(f"\n交易 {trade_id}：平均相似度 {avg_similarity:.2f} → ✅ 正常交易")
            
    except Exception as e:
        logger.error(f"检测交易{trade_id}失败：{str(e)}", exc_info=True)
        print(f"\n交易 {trade_id}：检测失败 → ❌ 触发风控审核")

logger.info("\n" + "="*50)
logger.info("异常交易检测流程结束")
logger.info("="*50)

import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
import chromadb
from chromadb.config import Settings
import os
import logging
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings("ignore")
import seaborn as sns

img_dir = "./analysis_imgs"
for dir_path in [img_dir]:
    if not os.path.exists(dir_path):
        os.makedirs(dir_path)
# 设置中文字体（避免乱码）
plt.rcParams["font.family"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False  # 解决负号显示问题

# ===================== 2. 初始化ChromaDB =====================
persist_directory = "./chroma_persist"
if not os.path.exists(persist_directory):
    os.makedirs(persist_directory)
    print(f"创建ChromaDB持久化目录：{persist_directory}")

client = chromadb.PersistentClient(
    path=persist_directory,
    settings=Settings(allow_reset=True, anonymized_telemetry=False)
)
client.reset()
print("重置ChromaDB数据库（测试环境）")

# ===================== 3. 核心配置 =====================
FRAUD_THRESHOLD = 0.7  # 相似度阈值
TOP_K = 3  # 检索相似向量数量
USER_ID = "U001"
NORMAL_TRADE_NUM = 4

print("="*50)
print("【核心风控配置】")
print(f"异常交易判定阈值：余弦相似度 < {FRAUD_THRESHOLD}")
print(f"检索相似向量数量：Top-{TOP_K}")
print(f"目标分析用户：{USER_ID}")
print(f"正常交易样本数量：{NORMAL_TRADE_NUM}")
print("="*50)

# ===================== 4. 数据预处理与向量化 =====================
# 模拟交易数据
data = pd.DataFrame({
    "user_id": ["U001", "U001", "U001", "U001", "U001", "U001"],
    "user_reg_days": [180, 180, 180, 180, 180, 180],
    "history_trade_count": [56, 57, 58, 59, 60, 61],
    "trade_amount": [299.9, 199.0, 399.5, 499.0, 10.0, 9999.0],
    "trade_hour": [20, 19, 21, 20, 3, 23],
    "is_common_device": [1, 1, 1, 1, 0, 0]
})
# 标记交易类型（正常/待检测）
data["trade_type"] = ["正常"]*NORMAL_TRADE_NUM + ["待检测"]*(len(data)-NORMAL_TRADE_NUM)

print(f"加载交易数据：共{len(data)}条，其中前{NORMAL_TRADE_NUM}条为正常交易，后{len(data)-NORMAL_TRADE_NUM}条为待检测交易")

# 特征预处理
scaler = MinMaxScaler(feature_range=(0, 1))
feature_cols = ["user_reg_days", "history_trade_count", "trade_amount", "trade_hour", "is_common_device"]
data_scaled = scaler.fit_transform(data[feature_cols])
vectors = data_scaled.tolist()
metadatas = [{"user_id": row["user_id"], "trade_id": f"T{idx}", "trade_type": row["trade_type"]} 
             for idx, row in data.iterrows()]
ids = [f"T{idx}" for idx in range(len(vectors))]

# 输出向量维度日志
print(f"特征列：{feature_cols}")
print(f"向量维度：{len(vectors[0])}维")
print(f"示例向量（T0）：{vectors[0]}")

# ===================== 5. 正常交易向量入库 =====================
collection_name = f"user_{USER_ID}_normal_trades"
try:
    client.delete_collection(name=collection_name)
    print(f"删除已存在的集合：{collection_name}")
except Exception as e:
    print(f"集合{collection_name}不存在，无需删除：{e}")

collection = client.create_collection(
    name=collection_name,
    metadata={"hnsw:space": "cosine"}
)
print(f"创建用户{USER_ID}的正常交易向量集合：{collection_name}")

# 入库正常交易向量
normal_vectors = vectors[:NORMAL_TRADE_NUM]
normal_metadatas = metadatas[:NORMAL_TRADE_NUM]
normal_ids = ids[:NORMAL_TRADE_NUM]

try:
    collection.add(embeddings=normal_vectors, metadatas=normal_metadatas, ids=normal_ids)
    print(f"正常交易向量入库完成：{len(normal_vectors)}条")
    print(f"入库向量ID：{normal_ids}")
    print(f"集合当前向量总数：{collection.count()}")
except Exception as e:
    print(f"正常向量入库失败：{str(e)}", exc_info=True)

# ===================== 6. 可视化分析函数 =====================
def plot_trade_feature_comparison(data, feature_cols, save_path):
    """
    绘制正常/待检测交易特征对比图
    """
    fig, axes = plt.subplots(2, 2, figsize=(12, 8))
    fig.suptitle(f"用户{USER_ID}交易特征对比（正常VS待检测）", fontsize=14, fontweight="bold")
    
    # 选择核心特征绘制
    plot_features = ["trade_amount", "trade_hour", "history_trade_count", "is_common_device"]
    for idx, feat in enumerate(plot_features):
        ax = axes[idx//2, idx%2]
        # 分组统计
        normal_feat = data[data["trade_type"]=="正常"][feat]
        test_feat = data[data["trade_type"]=="待检测"][feat]
        
        # 绘制对比
        x = ["正常交易均值", "待检测交易均值"]
        y = [normal_feat.mean(), test_feat.mean()]
        ax.bar(x, y, color=["#2E8B57", "#DC143C"], alpha=0.7)
        ax.set_title(f"{feat.replace('_', ' ')} 对比", fontsize=12)
        ax.set_ylabel("特征值")
        
        # 添加数值标注
        for i, v in enumerate(y):
            ax.text(i, v+0.02, f"{v:.2f}", ha="center", va="bottom", fontsize=10)
    
    plt.tight_layout()
    plt.savefig(save_path, dpi=300, bbox_inches="tight")
    print(f"特征对比图已保存：{save_path}")
    plt.show()

def plot_similarity_distribution(similarity_data, save_path):
    """
    绘制相似度分布直方图（含阈值线）
    """
    fig, ax = plt.subplots(figsize=(10, 6))
    # 绘制相似度分布
    ax.hist(similarity_data, bins=10, color="#4682B4", alpha=0.7)
    # 添加阈值线
    ax.axvline(FRAUD_THRESHOLD, color="#DC143C", linestyle="--", linewidth=2, label=f"异常阈值：{FRAUD_THRESHOLD}")
    # 标注异常交易
    for idx, sim in enumerate(similarity_data):
        if sim < FRAUD_THRESHOLD:
            ax.scatter(sim, 0.5, color="red", s=50, label=f"异常交易T{NORMAL_TRADE_NUM+idx}" if idx==0 else "")
    
    ax.set_title(f"用户{USER_ID}交易相似度分布", fontsize=14, fontweight="bold")
    ax.set_xlabel("余弦相似度")
    ax.set_ylabel("频次")
    ax.legend()
    ax.grid(axis="y", alpha=0.3)
    
    plt.savefig(save_path, dpi=300, bbox_inches="tight")
    print(f"相似度分布图已保存：{save_path}")
    plt.show()

def plot_vector_space(vectors, labels, save_path):
    """
    向量空间可视化（PCA降维到2D）
    """
    from sklearn.decomposition import PCA
    # PCA降维到2维
    pca = PCA(n_components=2)
    vectors_2d = pca.fit_transform(vectors)
    
    fig, ax = plt.subplots(figsize=(10, 8))
    # 绘制正常交易向量
    normal_2d = vectors_2d[:NORMAL_TRADE_NUM]
    ax.scatter(normal_2d[:,0], normal_2d[:,1], color="#2E8B57", s=100, label="正常交易", alpha=0.8)
    # 绘制待检测交易向量
    test_2d = vectors_2d[NORMAL_TRADE_NUM:]
    ax.scatter(test_2d[:,0], test_2d[:,1], color="#DC143C", s=100, marker="x", label="待检测交易", alpha=0.8)
    
    # 添加标注
    for idx, (x, y) in enumerate(vectors_2d):
        ax.text(x+0.01, y+0.01, f"T{idx}", fontsize=9)
    
    ax.set_title(f"用户{USER_ID}交易向量空间分布（PCA降维）", fontsize=14, fontweight="bold")
    ax.set_xlabel("PCA维度1")
    ax.set_ylabel("PCA维度2")
    ax.legend()
    ax.grid(alpha=0.3)
    
    plt.savefig(save_path, dpi=300, bbox_inches="tight")
    print(f"向量空间分布图已保存：{save_path}")
    plt.show()

# ===================== 7. 异常交易检测（带细化分析+可视化） =====================
print("\n" + "="*50)
print("开始检测异常交易")
print("="*50)

# 待检测向量
test_vectors = vectors[NORMAL_TRADE_NUM:]
test_metadatas = metadatas[NORMAL_TRADE_NUM:]
test_ids = ids[NORMAL_TRADE_NUM:]

# 存储相似度结果（用于可视化）
similarity_results = []

for idx, (vector, meta, trade_id) in enumerate(zip(test_vectors, test_metadatas, test_ids)):
    print(f"\n--- 检测交易 {trade_id} ---")
    # 1. 输出原始特征
    raw_data = data.iloc[NORMAL_TRADE_NUM + idx]
    print(f"原始特征：{raw_data.to_dict()}")
    print(f"待检测向量：{vector}")
    
    # 2. 特征异常度分析
    print("【特征异常度分析】")
    normal_data = data[data["trade_type"]=="正常"]
    for feat in ["trade_amount", "trade_hour", "is_common_device"]:
        normal_mean = normal_data[feat].mean()
        test_val = raw_data[feat]
        # 计算偏离度（相对差值）
        deviation = abs(test_val - normal_mean) / normal_mean if normal_mean !=0 else 1.0
        deviation_pct = deviation * 100
        if deviation_pct > 50:  # 偏离超过50%判定为异常特征
            print(f"  {feat}：正常均值={normal_mean:.2f}，当前值={test_val}，偏离度={deviation_pct:.1f}% → 异常")
        else:
            print(f"  {feat}：正常均值={normal_mean:.2f}，当前值={test_val}，偏离度={deviation_pct:.1f}% → 正常")
    
    # 3. 检索相似向量
    try:
        results = collection.query(
            query_embeddings=[vector],
            n_results=TOP_K,
            where={"user_id": USER_ID}
        )
        retrieved_ids = results["ids"][0]
        retrieved_distances = results["distances"][0]
        retrieved_metadatas = results["metadatas"][0]
        
        # 4. 输出检索结果
        print(f"检索到Top-{TOP_K}相似正常交易：{retrieved_ids}")
        print(f"余弦距离（越小越相似）：{[round(d, 4) for d in retrieved_distances]}")
        
        # 5. 计算相似度
        avg_distance = np.mean(retrieved_distances) if retrieved_distances else 1.0
        avg_similarity = 1 - avg_distance
        similarity_results.append(avg_similarity)
        print(f"计算过程：平均余弦距离={round(avg_distance, 4)} → 平均余弦相似度={round(avg_similarity, 4)}")
        
        # 6. 风控决策
        if avg_similarity < FRAUD_THRESHOLD:
            print(f"【异常交易判定】交易{trade_id}：平均相似度{round(avg_similarity, 4)} < 阈值{FRAUD_THRESHOLD}")
            print(f"判定原因：1. 特征偏离度超过50%；2. 与正常交易向量相似度过低，行为模式异常")
            print(f"\n交易 {trade_id}：平均相似度 {avg_similarity:.2f} → 🚨 异常交易，触发风控")
        else:
            print(f"【正常交易判定】交易{trade_id}：平均相似度{round(avg_similarity, 4)} ≥ 阈值{FRAUD_THRESHOLD}")
            print(f"判定原因：特征偏离度在正常范围，与正常交易向量高度相似")
            print(f"\n交易 {trade_id}：平均相似度 {avg_similarity:.2f} → ✅ 正常交易")
            
    except Exception as e:
        print(f"检测交易{trade_id}失败：{str(e)}", exc_info=True)
        similarity_results.append(0.0)
        print(f"\n交易 {trade_id}：检测失败 → ❌ 触发风控审核")

# ===================== 8. 生成可视化分析图 =====================
print("\n" + "="*50)
print("生成可视化分析图表")
print("="*50)

# 8.1 交易特征对比图
plot_trade_feature_comparison(
    data=data,
    feature_cols=feature_cols,
    save_path=f"{img_dir}/trade_feature_comparison.png"
)

# 8.2 相似度分布直方图
plot_similarity_distribution(
    similarity_data=similarity_results,
    save_path=f"{img_dir}/similarity_distribution.png"
)

# 8.3 向量空间分布图
plot_vector_space(
    vectors=vectors,
    labels=["正常"]*NORMAL_TRADE_NUM + ["待检测"]*(len(vectors)-NORMAL_TRADE_NUM),
    save_path=f"{img_dir}/vector_space_distribution.png"
)

print("\n" + "="*50)
print("异常交易检测流程结束")
print(f"可视化图表已保存至：{img_dir}")
print("="*50)