企业官网建设流程全解析

大家好，我是Java1234_小锋老师，分享一套锋哥原创的基于Spark实时金融交易风险监控与预测系统(Python版本+pyspark+可视化大屏+Kafka+FastAPI+Vue3)

项目介绍

随着移动支付、电子商务和互联网金融业务的快速发展，金融交易呈现出高并发、多渠道、实时性强等特征，传统基于批处理的风控模式难以满足秒级甚至分钟级风险识别的需要。针对上述问题，本文设计并实现了一套基于Spark的实时金融交易风险监控与预测系统。系统以后端Python语言和FastAPI框架构建RESTful接口服务，以前端Vue3与Element Plus实现管理后台和数据可视化大屏；在数据处理链路中引入Kafka作为交易事件缓冲与解耦中间件，利用Spark Streaming对交易流进行窗口聚合统计，并基于Spark ML对风险评分序列开展预测，输出RMSE、MAE、MAPE等误差指标。系统围绕管理员认证、个人中心、账户与交易流水管理、风险预警、实时统计和预测分析等功能完成开发，数据库采用MySQL，库名为db_finance_risk，表名统一以t_开头。测试结果表明，系统能够稳定完成交易数据接入、实时统计、预警处置与风险趋势预测，具有较好的工程完整性和本科毕业设计实践价值。

源码下载

链接: https://pan.baidu.com/s/18LvykgWUL-0drjP9aZqVXg?pwd=1234
提取码: 1234

系统展示

核心代码

""" Spark ML 交易风险评分预测模块 """ import numpy as np from decimal import Decimal from config import settings def compute_error_metrics(y_true: list, y_pred: list) -> dict: """计算误差指标：RMSE、MAE、MAPE""" y_true = np.array(y_true, dtype=float) y_pred = np.array(y_pred, dtype=float) rmse = float(np.sqrt(np.mean((y_true - y_pred) ** 2))) mae = float(np.mean(np.abs(y_true - y_pred))) mask = y_true >= 1.0 if mask.any(): mape = float(np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100) else: mape = 0.0 return {"rmse": round(rmse, 4), "mae": round(mae, 4), "mape": round(mape, 4)} def run_spark_prediction(score_series: list = None) -> tuple: """ 使用 Spark ML 进行交易风险评分预测 返回 (predictions, error_metrics) """ try: from pyspark.sql import SparkSession from pyspark.ml.feature import VectorAssembler from pyspark.ml.regression import LinearRegression from pyspark.sql.types import StructType, StructField, DoubleType, IntegerType, StringType import pyspark.sql.functions as F spark = SparkSession.builder \ .appName("FinanceRiskPrediction") \ .master(settings.SPARK_MASTER) \ .config("spark.driver.memory", "2g") \ .getOrCreate() spark.sparkContext.setLogLevel("WARN") if score_series is None: from database import SessionLocal from models.realtime_stat import RealtimeStat db = SessionLocal() stats = db.query(RealtimeStat).order_by(RealtimeStat.window_time.asc()).all() db.close() score_series = [ {"window_time": s.window_time, "risk_score": float(s.risk_score or 0)} for s in stats ] if len(score_series) < 5: spark.stop() return [], {"rmse": 0, "mae": 0, "mape": 0} data = [] for i in range(3, len(score_series)): data.append({ "window_time": score_series[i]["window_time"], "lag1": score_series[i - 1]["risk_score"], "lag2": score_series[i - 2]["risk_score"], "lag3": score_series[i - 3]["risk_score"], "hour": int(str(score_series[i]["window_time"])[11:13]), "risk_score": score_series[i]["risk_score"], }) schema = StructType([ StructField("window_time", StringType()), StructField("lag1", DoubleType()), StructField("lag2", DoubleType()), StructField("lag3", DoubleType()), StructField("hour", IntegerType()), StructField("risk_score", DoubleType()), ]) df = spark.createDataFrame(data, schema) split_idx = max(int(len(data) * 0.6), 1) if len(data) - split_idx < 5: split_idx = max(len(data) - max(5, len(data) // 3), 1) train_df = df.limit(split_idx) test_df = df.filter(F.monotonically_increasing_id() >= split_idx) assembler = VectorAssembler( inputCols=["lag1", "lag2", "lag3", "hour"], outputCol="features" ) train_df = assembler.transform(train_df) test_df = assembler.transform(test_df) lr = LinearRegression(featuresCol="features", labelCol="risk_score", maxIter=100) model = lr.fit(train_df) predictions_raw = model.transform(test_df).collect() predictions = [] y_true, y_pred = [], [] for row in predictions_raw: true_val = float(row["risk_score"]) pred_val = float(row["prediction"]) predictions.append({ "window_time": row["window_time"], "true_score": round(true_val, 2), "pred_score": round(max(min(pred_val, 100), 0), 2), }) y_true.append(true_val) y_pred.append(pred_val) error = compute_error_metrics(y_true, y_pred) spark.stop() return predictions, error except Exception as e: print(f"[Spark ML] 预测失败: {e}") return None, None def save_predictions_to_db(predictions: list, error: dict): """保存预测结果和误差指标到数据库""" from database import SessionLocal from models.prediction import Prediction from models.error_metric import ErrorMetric db = SessionLocal() try: db.query(Prediction).delete() for p in predictions: db.add(Prediction( window_time=p["window_time"], true_score=Decimal(str(p["true_score"])), pred_score=Decimal(str(p["pred_score"])), )) db.add(ErrorMetric( rmse=Decimal(str(error["rmse"])), mae=Decimal(str(error["mae"])), mape=Decimal(str(error["mape"])), )) db.commit() print(f"[Spark ML] 已保存 {len(predictions)} 条预测结果") finally: db.close()

<template> <div class="page-container"> <div class="page-card"> <div class="page-title">交易风险预测分析</div> <div class="error-cards"> <div class="error-card"><div class="metric-label">RMSE (均方根误差)</div><div class="metric-value">{{ errorMetric.rmse }}</div></div> <div class="error-card"><div class="metric-label">MAE (平均绝对误差)</div><div class="metric-value">{{ errorMetric.mae }}</div></div> <div class="error-card"><div class="metric-label">MAPE (平均绝对百分比误差 %)</div><div class="metric-value">{{ errorMetric.mape }}%</div></div> </div> <div ref="compareRef" class="pred-chart"></div> <div ref="residualRef" class="pred-chart pred-residual"></div> <el-table :data="tableData" stripe border> <el-table-column prop="window_time" label="时间窗口" min-width="170"> <template #default="{ row }">{{ formatWindowTime(row.window_time) }}</template> </el-table-column> <el-table-column prop="true_score" label="真实风险分" min-width="120"> <template #default="{ row }"><span style="color:#1677ff;font-weight:600">{{ row.true_score }}</span></template> </el-table-column> <el-table-column prop="pred_score" label="预测风险分" min-width="120"> <template #default="{ row }"><span style="color:#52c41a;font-weight:600">{{ row.pred_score }}</span></template> </el-table-column> <el-table-column label="误差" min-width="100"> <template #default="{ row }"> <span :style="{ color: Math.abs(row.true_score - row.pred_score) > 5 ? '#ff4d4f' : '#909399' }"> {{ (row.true_score - row.pred_score).toFixed(2) }} </span> </template> </el-table-column> <el-table-column prop="create_time" label="生成时间" min-width="170"> <template #default="{ row }">{{ formatDateTime(row.create_time) }}</template> </el-table-column> </el-table> <el-pagination style="margin-top:16px;justify-content:flex-end" v-model:current-page="page" v-model:page-size="size" :total="total" layout="total, prev, pager, next" @change="loadTable" /> </div> </div> </template> <script setup> /** * 预测分析页面：真实 vs 预测对比 + 误差分析 */ import { ref, onMounted, onUnmounted } from 'vue' import * as echarts from 'echarts' import request from '@/utils/request' import { formatDateTime, formatWindowTime } from '@/utils/format' const errorMetric = ref({ rmse: 0, mae: 0, mape: 0 }) const tableData = ref([]) const page = ref(1) const size = ref(10) const total = ref(0) const compareRef = ref(null) const residualRef = ref(null) let charts = [] let pollTimer = null function axisLabel() { return { rotate: 30, interval: 'auto', formatter(val) { const t = formatWindowTime(val); return t.length >= 16 ? `${t.slice(0,10)}\n${t.slice(11)}` : t } } } function initCompareChart(data) { if (!compareRef.value) return const chart = echarts.init(compareRef.value) const labels = data.map(d => formatWindowTime(d.window_time)) chart.setOption({ title: { text: '真实风险分 vs 预测风险分 对比', left: 'center', textStyle: { fontSize: 15 } }, tooltip: { trigger: 'axis' }, legend: { data: ['真实风险分', '预测风险分'], top: 32 }, grid: { left: 20, right: 24, bottom: 28, top: 72, containLabel: true }, xAxis: { type: 'category', data: labels, axisLabel: axisLabel() }, yAxis: { type: 'value', name: '风险评分' }, series: [ { name: '真实风险分', type: 'line', smooth: true, data: data.map(d => Number(d.true_score)), itemStyle: { color: '#1677ff' }, lineStyle: { width: 3 } }, { name: '预测风险分', type: 'line', smooth: true, data: data.map(d => Number(d.pred_score)), itemStyle: { color: '#52c41a' }, lineStyle: { width: 3, type: 'dashed' } }, ], }) charts.push(chart) } function initResidualChart(data) { if (!residualRef.value) return const chart = echarts.init(residualRef.value) const labels = data.map(d => formatWindowTime(d.window_time)) const residuals = data.map(d => Number((Number(d.true_score) - Number(d.pred_score)).toFixed(2))) chart.setOption({ title: { text: '预测残差分析 (真实值 - 预测值)', left: 'center', textStyle: { fontSize: 15 } }, tooltip: { trigger: 'axis' }, grid: { left: 20, right: 24, bottom: 28, top: 56, containLabel: true }, xAxis: { type: 'category', data: labels, axisLabel: axisLabel() }, yAxis: { type: 'value', name: '残差' }, series: [{ type: 'bar', data: residuals.map(v => ({ value: v, itemStyle: { color: v >= 0 ? '#1677ff' : '#ff4d4f' } })), barWidth: 20 }], }) charts.push(chart) } async function loadData() { const [errorRes, compareRes] = await Promise.all([ request.get('/prediction/error'), request.get('/prediction/compare'), ]) errorMetric.value = errorRes.data || { rmse: 0, mae: 0, mape: 0 } const compareData = compareRes.data || [] charts.forEach(c => c.dispose()) charts = [] initCompareChart(compareData) initResidualChart(compareData) } async function loadTable() { const res = await request.get('/prediction/list', { params: { page: page.value, size: size.value } }) tableData.value = res.data.items total.value = res.data.total } onMounted(() => { loadData() loadTable() pollTimer = setInterval(loadData, 8000) window.addEventListener('resize', () => charts.forEach(c => c.resize())) }) onUnmounted(() => { clearInterval(pollTimer) charts.forEach(c => c.dispose()) }) </script>