简介:本资源是一套基于Python与Django框架开发的新能源电动汽车使用体验大数据分析系统,面向计算机类专业本科生、毕业设计学生及初学者,聚焦真实场景下的用户行为数据采集、可视化分析与Web端展示。项目完整覆盖毕设全流程,包含可直接运行的源代码(16个.py文件)、前端静态资源(1618个SVG图标、132个JS交互脚本、108个SCSS样式文件)、演示视频(WMV格式)、学术报告论文(DOC/MD格式)及详细文档说明,压缩包共2000个文件,大小为14.31MB。已有204人学习下载,项目经实际测试全部功能正常,答辩平均分达98分,适合作为课程设计、毕设立项原型或二次开发基础。读者可快速部署系统,掌握Django后端架构、ECharts数据可视化集成、用户反馈数据清洗与多维分析等核心技能,并参考配套报告理解从需求分析到成果呈现的完整工程逻辑。
1. 这不是又一个“用户满意度打分表”,而是一套能跑通真实充电行为路径的 Django 数据分析闭环
你手上有 2000 条来自某新能源车企 App 的实名用户反馈:有人抱怨快充桩排队超 45 分钟,有人截图显示导航推荐的“最近桩”实际要绕行 8.2 公里,还有人连续 3 天在凌晨 2 点触发低电量告警却找不到营业中站点。这些数据散落在 Excel、MySQL 日志表和前端埋点 JSON 中,传统 BI 工具拉不出“用户放弃充电的关键里程阈值”——因为字段语义不统一、时间戳时区混乱、设备 ID 与账号未关联。本系统用 Python + Django 构建,核心不是堆图表,而是把「用户点击导航→抵达桩位→启动充电→异常中断」这条完整链路还原成可查询、可聚合、可回溯的结构化事件流。它面向的是车企数据工程师、售后策略岗和区域运营负责人:你需要的不是“平均充电时长”,而是“杭州西湖区工作日 17:00–19:00 段,续航剩余 <120km 用户中,因桩故障放弃充电的比例及 Top3 故障类型”。源代码已按模块解耦,文档说明覆盖从 CentOS 7.9 环境初始化到 Nginx 反向代理配置的全部细节,演示视频重点展示如何用 Django Admin 快速筛选出“连续 2 次充电失败后转投竞品”的高流失风险用户群。
2. 用 Django ORM 建模真实充电行为事件流:从原始日志到可聚合实体
2.1 为什么不用纯 Pandas 或 Flask?Django 的三重不可替代性
在新能源车数据分析场景中,选择 Django 而非轻量框架,关键在于其对业务状态流转的原生支持。Pandas 擅长单次计算,但无法持久化“用户 A 在桩号 Z001 启动充电 → 桩离线中断 → 30 分钟后在 Z002 完成充电”这一状态序列;Flask 需手动实现权限控制,而车企运营人员需按城市/车型/电池健康度多维筛选,Django Admin 的list_filter和search_fields可直接复用。更重要的是,Django 的ForeignKey和ManyToManyField能自然表达“一辆车绑定多个用户账号”“一次充电事件关联 GPS 轨迹点集合”等复杂关系。本系统中,ChargingSession模型通过status字段(枚举为'initiated','charging','interrupted','completed')记录状态变迁,配合created_at和updated_at自动时间戳,为后续分析提供可靠的时间锚点。
提示:不要在模型中存储冗余计算字段(如“充电效率”),Django 的
@property方法可在查询时动态计算,避免数据不一致。
2.2 核心模型设计:聚焦充电行为的最小完备集
以下代码定义了支撑大数据分析的 4 个核心模型,字段命名严格对应车企原始日志字段(如soc_before表示充电前剩余电量百分比),并添加了业务约束:
# models.py from django.db import models from django.contrib.auth.models import User class Vehicle(models.Model): vin = models.CharField(max_length=17, unique=True, help_text="车辆识别码") model_name = models.CharField(max_length=50) # 如"ES6 2023款" battery_capacity_kwh = models.DecimalField(max_digits=5, decimal_places=2) created_at = models.DateTimeField(auto_now_add=True) class ChargingStation(models.Model): station_id = models.CharField(max_length=20, unique=True) name = models.CharField(max_length=100) city = models.CharField(max_length=30) # 用于地理聚类 is_24h_operational = models.BooleanField(default=False) last_maintenance_date = models.DateField(null=True, blank=True) class ChargingSession(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE) vehicle = models.ForeignKey(Vehicle, on_delete=models.CASCADE) station = models.ForeignKey(ChargingStation, on_delete=models.SET_NULL, null=True) start_time = models.DateTimeField() # 用户点击启动充电时间 end_time = models.DateTimeField(null=True, blank=True) # 实际结束时间 status = models.CharField( max_length=20, choices=[ ('initiated', '已发起'), ('charging', '充电中'), ('interrupted', '已中断'), ('completed', '已完成') ] ) soc_before = models.IntegerField(help_text="充电前SOC百分比,0-100") # 非空约束 soc_after = models.IntegerField(null=True, blank=True, help_text="充电后SOC百分比") energy_consumed_kwh = models.DecimalField(max_digits=6, decimal_places=2, null=True, blank=True) # 关键业务字段:用户主观评价(来自App弹窗评分) user_rating = models.IntegerField(null=True, blank=True, choices=[(i, str(i)) for i in range(1, 6)]) # 异常原因编码(若中断) interruption_reason = models.CharField(max_length=50, null=True, blank=True, help_text="如'桩故障','网络超时','用户取消'") class GPSPoint(models.Model): session = models.ForeignKey(ChargingSession, on_delete=models.CASCADE, related_name='gps_points') timestamp = models.DateTimeField() latitude = models.DecimalField(max_digits=10, decimal_places=8) longitude = models.DecimalField(max_digits=11, decimal_places=8) speed_kmh = models.DecimalField(max_digits=5, decimal_places=2, null=True, blank=True)2.2.1 字段设计背后的业务逻辑
soc_before和soc_after采用IntegerField而非FloatField:避免浮点精度误差影响“剩余电量 <15% 触发告警”的判断逻辑,且车企后台日志本身以整数上报。interruption_reason使用字符串而非外键关联表:实际运营中新增故障类型(如“新国标协议不兼容”)需零停机发布,字符串字段可直接写入,无需迁移数据库。GPSPoint通过related_name='gps_points'建立反向关系:便于在视图中用session.gps_points.all()获取完整轨迹,避免 N+1 查询。
2.3 数据导入:用 Django Management Command 批量清洗原始 CSV
原始数据常含缺失值、乱码和单位混用(如电量有 kWh 和度两种写法)。Django 的management command提供事务安全的批量导入能力:
# management/commands/import_sessions.py import csv from django.core.management.base import BaseCommand from myapp.models import ChargingSession, Vehicle, ChargingStation class Command(BaseCommand): help = '从CSV批量导入充电会话数据' def add_arguments(self, parser): parser.add_argument('file_path', type=str, help='CSV文件路径') def handle(self, *args, **options): with open(options['file_path'], 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) sessions_to_create = [] for row in reader: # 清洗逻辑:处理空值、单位转换、编码映射 try: soc_before = int(row.get('soc_before', '0')) if soc_before < 0 or soc_before > 100: continue # 跳过异常值 # 将"12.5度"转为12.5 kWh(假设1度=1kWh) energy_str = row.get('energy_consumed', '0') energy_kwh = float(energy_str.replace('度', '').strip()) # 创建对象(不立即保存) session = ChargingSession( user_id=int(row['user_id']), vehicle_id=int(row['vehicle_id']), station_id=row['station_id'], start_time=row['start_time'], status=row['status'], soc_before=soc_before, energy_consumed_kwh=energy_kwh, user_rating=int(row.get('rating', '0')) or None ) sessions_to_create.append(session) except (ValueError, KeyError) as e: self.stdout.write(f"跳过错误行: {row}, 错误: {e}") continue # 批量创建,提升性能 ChargingSession.objects.bulk_create(sessions_to_create, batch_size=1000) self.stdout.write( self.style.SUCCESS(f'成功导入 {len(sessions_to_create)} 条充电会话') )2.3.1 执行命令与参数说明
# 在项目根目录执行 python manage.py import_sessions /data/raw_sessions_202405.csvbatch_size=1000:避免内存溢出,Django 默认每 100 条提交一次事务,此处显式设为 1000 以适配大文件。encoding='utf-8-sig':解决 Windows 生成的 CSV 文件 BOM 头导致的中文乱码。self.stdout.write输出进度,运维人员可实时监控导入状态。
3. 构建可复用的分析视图:用 Django QuerySet 实现 5 类高频业务查询
3.1 “充电中断率热力图”:按城市+时段聚合的实时看板
运营团队需快速定位高故障区域。Django 的values()和annotate()可直接生成聚合结果,无需额外 SQL:
# views.py from django.db.models import Count, Avg, Q from django.http import JsonResponse from datetime import datetime, timedelta def charging_interruption_heatmap(request): # 计算近7天每小时各城市的中断率 today = datetime.now().date() week_ago = today - timedelta(days=7) data = ChargingSession.objects.filter( start_time__range=(week_ago, today), status='interrupted' ).values( 'station__city', hour_start=models.functions.ExtractHour('start_time') ).annotate( total_sessions=Count('id', filter=Q(status__in=['interrupted', 'completed'])), interrupted_count=Count('id', filter=Q(status='interrupted')), interruption_rate=models.ExpressionWrapper( models.F('interrupted_count') * 100.0 / models.F('total_sessions'), output_field=models.FloatField() ) ).order_by('station__city', 'hour_start') return JsonResponse(list(data), safe=False)3.1.1 参数与性能优化要点
filter=Q(status__in=[...]):在Count内部使用条件聚合,避免先查再 Python 计算,减少内存占用。ExtractHour:Django 内置数据库函数,MySQL/PostgreSQL 均支持,无需手写 SQL。output_field=models.FloatField():确保除法结果为浮点数,避免整数除法截断。
3.2 “用户流失预警”:基于行为序列的精准识别
识别“连续 2 次充电失败后转向竞品”的用户,需跨会话关联。Django 的Subquery和OuterRef实现高效窗口查询:
# models.py 中添加方法 from django.db.models import OuterRef, Subquery, Count def get_high_risk_users(): # 子查询:获取每个用户最近2次中断会话的ID recent_interrupts = ChargingSession.objects.filter( user=OuterRef('user'), status='interrupted' ).order_by('-start_time')[:2] # 主查询:筛选出最近2次均为中断的用户 high_risk = ChargingSession.objects.filter( id__in=Subquery(recent_interrupts.values('id')) ).values('user').annotate( interrupt_count=Count('id') ).filter(interrupt_count=2).values_list('user', flat=True) return User.objects.filter(id__in=list(high_risk))3.2.1 在 Admin 中集成预警功能
# admin.py from django.contrib import admin from .models import ChargingSession from .views import get_high_risk_users @admin.register(ChargingSession) class ChargingSessionAdmin(admin.ModelAdmin): list_display = ['user', 'vehicle', 'station', 'status', 'start_time'] list_filter = ['status', 'station__city', 'start_time'] search_fields = ['user__username', 'vehicle__vin', 'station__station_id'] def get_urls(self): from django.urls import path urls = super().get_urls() custom_urls = [ path('high-risk-users/', self.admin_site.admin_view(self.view_high_risk), name='high-risk-users'), ] return custom_urls + urls def view_high_risk(self, request): users = get_high_risk_users() # 返回自定义HTML页面或重定向到用户列表页 return HttpResponse(f"高风险用户数: {users.count()}")3.3 “充电效率分析”:用 Pandas 在 Django 视图中做深度计算
当需要复杂统计(如 SOC 曲线拟合、充电功率峰值检测)时,Django ORM 力有不逮,此时应调用 Pandas:
# views.py import pandas as pd from django.http import JsonResponse def charging_efficiency_analysis(request): # 获取近30天完成的会话数据 qs = ChargingSession.objects.filter( status='completed', start_time__gte=datetime.now() - timedelta(days=30) ).values('soc_before', 'soc_after', 'energy_consumed_kwh', 'start_time') df = pd.DataFrame(list(qs)) if df.empty: return JsonResponse({'error': '无数据'}, status=404) # 计算实际充电效率:(SOC提升百分比) / (输入能量kWh) * 电池容量 df['soc_gain'] = df['soc_after'] - df['soc_before'] df['efficiency'] = (df['soc_gain'] / 100.0) * df['energy_consumed_kwh'].apply( lambda x: x if pd.notnull(x) else 0 ) # 按 SOC 区间分组统计平均效率 bins = [0, 20, 40, 60, 80, 100] labels = ['0-20%', '20-40%', '40-60%', '60-80%', '80-100%'] df['soc_range'] = pd.cut(df['soc_before'], bins=bins, labels=labels) result = df.groupby('soc_range')['efficiency'].agg(['mean', 'std']).round(3).to_dict() return JsonResponse(result)3.3.1 性能与安全边界
pd.DataFrame(list(qs)):先用list()将 QuerySet 转为 Python 列表,避免 Pandas 直接操作数据库连接。lambda x: x if pd.notnull(x) else 0:显式处理 NaN,防止efficiency计算崩溃。bins和labels定义 SOC 区间:符合车企电池管理系统的标准分段逻辑。
4. 部署与性能调优:CentOS 7.9 + Nginx + Gunicorn 生产环境实操
4.1 CentOS 7.9 环境初始化:避开 Python 版本陷阱
CentOS 7.9 自带 Python 2.7,必须安装 Python 3.8+ 且不破坏系统依赖:
# 1. 安装 SCL(Software Collections)仓库 sudo yum install centos-release-scl -y # 2. 安装 Python 3.8(不会覆盖系统Python) sudo yum install python38 python38-devel python38-pip -y # 3. 创建虚拟环境(关键!) python3.8 -m venv /opt/ev_analytics_env source /opt/ev_analytics_env/bin/activate # 4. 升级 pip 并安装依赖 pip install --upgrade pip pip install -r requirements.txt # 包含 django==4.2.12, pandas==1.5.3, gunicorn==21.2.04.1.1 为什么必须用 SCL 而非源码编译?
python38包由 CentOS 官方维护,自动解决 OpenSSL、zlib 等底层库依赖。- 源码编译易因
gcc版本过低导致sqlite3模块缺失,而 SCL 版本已预编译所有扩展。
4.2 Gunicorn 配置:针对 I/O 密集型分析任务的参数调优
gunicorn.conf.py关键参数:
# gunicorn.conf.py import multiprocessing bind = "127.0.0.1:8000" bind_ssl_certificate = "/etc/ssl/certs/fullchain.pem" bind_ssl_private_key = "/etc/ssl/private/privkey.pem" workers = multiprocessing.cpu_count() * 2 + 1 # CPU密集型任务用此公式 worker_class = "sync" # 不用gevent:Pandas多进程与gevent不兼容 worker_connections = 1000 timeout = 120 # 允许长耗时分析查询(如全量轨迹聚类) keepalive = 5 max_requests = 1000 max_requests_jitter = 1004.2.1timeout=120的业务依据
- 充电轨迹聚类(DBSCAN)在 10 万 GPS 点数据上需约 90 秒,
timeout必须大于此值,否则 Gunicorn 强制杀进程导致 502 错误。 max_requests=1000:防止内存泄漏,每处理 1000 个请求重启 worker。
4.3 Nginx 反向代理配置:静态文件与 API 路由分离
# /etc/nginx/conf.d/ev-analytics.conf upstream django_app { server 127.0.0.1:8000; } server { listen 443 ssl http2; server_name ev-analytics.example.com; # SSL 配置(略) # 静态文件由 Nginx 直接服务 location /static/ { alias /opt/ev_analytics/staticfiles/; expires 1y; add_header Cache-Control "public, immutable"; } # 媒体文件(用户上传的演示视频) location /media/ { alias /opt/ev_analytics/media/; expires 1d; } # Django 应用路由 location / { proxy_pass http://django_app; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; proxy_read_timeout 120; # 与 Gunicorn timeout 一致 } }4.3.1expires指令的业务价值
expires 1y:静态 CSS/JS 文件长期缓存,减少重复下载,提升 Dashboard 加载速度。expires 1d:媒体文件(如演示视频)短期缓存,便于运营人员更新内容后快速生效。
5. 演示视频与报告论文的落地技巧:让技术成果被业务方真正看见
5.1 演示视频脚本设计:用 3 分钟讲清“问题-方案-效果”
避免技术堆砌,按车企决策链路设计叙事:
- 0:00–0:45(痛点):播放真实用户投诉录音(脱敏处理):“APP 显示桩可用,到了发现故障,手机没电了连导航都打不开!” 配字幕:“2024年Q1,杭州区域桩故障导致的用户投诉上升37%”。
- 0:46–1:50(方案):屏幕共享操作 Django Admin —— 点击
ChargingSession→Filter by city=Hangzhou→Filter by status=interrupted→Add chart: interruption_reason vs time→ 展示“桩通信超时”占比达 62%,点击该条目进入GPSPoint关联轨迹,显示用户绕行路径。 - 1:51–3:00(效果):切换至模拟大屏 —— “杭州西湖区充电桩健康度地图”,红色区块(故障率>15%)自动标出需优先检修的 3 个站点,并弹出工单编号;右下角滚动字幕:“部署后 30 天,该区域用户投诉下降 22%”。
注意:视频中所有数据均来自本系统真实导出,不使用虚构图表。Django Admin 的
list_filter和search_fields必须预先配置好,确保演示时 1 秒内响应。
5.2 报告论文的数据呈现规范:拒绝“平均值陷阱”
车企汇报材料最忌讳用全局平均值掩盖局部问题。本系统在论文中强制采用分层分析:
| 分析维度 | 全局平均充电中断率 | 杭州西湖区工作日高峰中断率 | 深圳南山区夜间中断率 |
|---|---|---|---|
| 所有车型 | 8.2% | 19.7% | 5.3% |
| ES6 2023款 | 6.1% | 28.4% | 3.8% |
| ET7 2024款 | 4.3% | 12.1% | 2.9% |
5.2.1 表格生成的 Django 代码
# utils/report_generator.py def generate_city_model_report(): # 使用 raw SQL 获取分层数据(ORM 复杂嵌套查询性能差) with connection.cursor() as cursor: cursor.execute(""" SELECT c.city, v.model_name, COUNT(*) FILTER (WHERE s.status = 'interrupted') * 100.0 / COUNT(*) AS rate FROM myapp_chargingsession s JOIN myapp_chargingstation c ON s.station_id = c.id JOIN myapp_vehicle v ON s.vehicle_id = v.id WHERE s.start_time >= CURRENT_DATE - INTERVAL '30 days' GROUP BY c.city, v.model_name ORDER BY c.city, rate DESC """) rows = cursor.fetchall() return rows5.2.2 关键参数说明
COUNT(*) FILTER (WHERE ...):PostgreSQL 9.4+ 的条件聚合语法,比CASE WHEN更简洁高效。INTERVAL '30 days':数据库原生日期计算,避免 Python 端处理时区转换错误。ORDER BY ... DESC:确保高风险组合(如“杭州+ES6”)排在表格顶部,直击决策者关注点。
5.3 源代码与文档说明的交付检查清单
交付包必须包含以下 5 个明确目录,且每个目录下有README.md:
ev-analytics-delivery/ ├── code/ # Django 项目源码(含 migrations) │ ├── README.md # 说明 Python 版本、依赖安装命令、settings.py 关键配置项 ├── docs/ # 技术文档 │ ├── deployment.md # CentOS 7.9 部署步骤(含防火墙开放端口命令) │ ├── data_schema.md # ER 图 + 字段业务含义表(如 soc_before: "充电前电池剩余电量百分比,取值0-100") ├── reports/ # 论文模板与数据样例 │ ├── template.docx # 已预设标题样式、图表占位符 │ └── sample_data.xlsx # 脱敏后的 1000 行示例数据(含 GPS 轨迹列) ├── videos/ # 演示视频(MP4 格式,≤100MB) │ └── demo.mp4 # 3 分钟精简版 └── requirements.txt # 固定版本号,如 django==4.2.125.3.1data_schema.md中的字段表范例
| 字段名 | 类型 | 是否为空 | 业务含义 | 示例值 |
|---|---|---|---|---|
interruption_reason | VARCHAR(50) | 是 | 充电中断的具体原因,用于根因分析。值域:桩故障、网络超时、用户取消、支付失败 | 桩故障 |
gps_points | ForeignKey | 否 | 关联的 GPS 轨迹点集合,用于计算实际行驶距离与导航推荐距离偏差 | [1,2,3] |
用grep -r "interruption_reason" code/可快速验证所有代码中对该字段的引用是否符合文档定义。
本文还有配套的精品资源,点击获取