大家好我是Java1234_小锋老师分享一套锋哥原创的基于Spark实时金融交易风险监控与预测系统(Python版本pyspark可视化大屏KafkaFastAPIVue3)项目介绍随着移动支付、电子商务和互联网金融业务的快速发展金融交易呈现出高并发、多渠道、实时性强等特征传统基于批处理的风控模式难以满足秒级甚至分钟级风险识别的需要。针对上述问题本文设计并实现了一套基于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?pwd1234提取码: 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, dtypefloat) y_pred np.array(y_pred, dtypefloat) 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], outputColfeatures ) train_df assembler.transform(train_df) test_df assembler.transform(test_df) lr LinearRegression(featuresColfeatures, labelColrisk_score, maxIter100) 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_timep[window_time], true_scoreDecimal(str(p[true_score])), pred_scoreDecimal(str(p[pred_score])), )) db.add(ErrorMetric( rmseDecimal(str(error[rmse])), maeDecimal(str(error[mae])), mapeDecimal(str(error[mape])), )) db.commit() print(f[Spark ML] 已保存 {len(predictions)} 条预测结果) finally: db.close()template div classpage-container div classpage-card div classpage-title交易风险预测分析/div div classerror-cards div classerror-carddiv classmetric-labelRMSE (均方根误差)/divdiv classmetric-value{{ errorMetric.rmse }}/div/div div classerror-carddiv classmetric-labelMAE (平均绝对误差)/divdiv classmetric-value{{ errorMetric.mae }}/div/div div classerror-carddiv classmetric-labelMAPE (平均绝对百分比误差 %)/divdiv classmetric-value{{ errorMetric.mape }}%/div/div /div div refcompareRef classpred-chart/div div refresidualRef classpred-chart pred-residual/div el-table :datatableData stripe border el-table-column propwindow_time label时间窗口 min-width170 template #default{ row }{{ formatWindowTime(row.window_time) }}/template /el-table-column el-table-column proptrue_score label真实风险分 min-width120 template #default{ row }span stylecolor:#1677ff;font-weight:600{{ row.true_score }}/span/template /el-table-column el-table-column proppred_score label预测风险分 min-width120 template #default{ row }span stylecolor:#52c41a;font-weight:600{{ row.pred_score }}/span/template /el-table-column el-table-column label误差 min-width100 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 propcreate_time label生成时间 min-width170 template #default{ row }{{ formatDateTime(row.create_time) }}/template /el-table-column /el-table el-pagination stylemargin-top:16px;justify-content:flex-end v-model:current-pagepage v-model:page-sizesize :totaltotal layouttotal, prev, pager, next changeloadTable / /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