基于 Django 与 Python 的电商用户行为分析系统设计与实现

行为数据收集

# views.py
from django.http import JsonResponse
from .models import UserBehavior

def track_behavior(request):
    if request.method == 'POST':
        data = request.POST
        behavior = UserBehavior.objects.create(
            user=request.user if request.user.is_authenticated else None,
            behavior_type=data.get('type'),
            product_id=data.get('product_id'),
            extra_data={
                'ip': request.META.get('REMOTE_ADDR'),
                'user_agent': request.META.get('HTTP_USER_AGENT'),
                'referrer': request.META.get('HTTP_REFERER')
            }
        )
        return JsonResponse({'status': 'success'})
    return JsonResponse({'status': 'error'}, status=400)

行为分析核心

# analytics.py
from django.db.models import Count, Q
from django.utils import timezone
from datetime import timedelta
from .models import UserBehavior

def get_user_behavior_stats(user_id):
    stats = UserBehavior.objects.filter(user_id=user_id).values(
        'behavior_type'
    ).annotate(count=Count('id'))
    return {item['behavior_type']: item['count'] for item in stats}

def get_hot_products(days=7):
    return UserBehavior.objects.filter(
        behavior_type='PV', 
        timestamp__gte=timezone.now()-timedelta(days=days)
    ).values('product_id').annotate(views=Count('id')).order_by('-views')[:10]

def calculate_conversion_rate():
    pv_count = UserBehavior.objects.filter(behavior_type='PV').count()
    buy_count = UserBehavior.objects.filter(behavior_type='BUY').count()
    return (buy_count / pv_count) * 100 if pv_count else 0

用户画像生成

# profiling.py
from collections import defaultdict
from django.db.models import Count
from .models import UserBehavior, UserProfile

def update_user_profile(user_id):
    behaviors = UserBehavior.objects.filter(user_id=user_id)
    profile, _ = UserProfile.objects.get_or_create(user_id=user_id)
    # 计算购买次数
    profile.purchase_count = behaviors.filter(
        behavior_type='BUY'
    ).count()
    # 分析用户偏好
    category_counts = defaultdict(int)
    for behavior in behaviors.filter(behavior_type__in=['PV', 'CART', 'BUY']):
        category = get_product_category(behavior.product_id) # 假设有此函数
        category_counts[category] += 1
    profile.preferences = {
        'top_categories': sorted(
            category_counts.items(), key=lambda x: x[1], reverse=True
        )[:3]
    }
    profile.save()

数据可视化 API

# api.py
from rest_framework import viewsets
from rest_framework.response import Response
from .models import UserBehavior
from .analytics import get_user_behavior_stats

class BehaviorAnalyticsViewSet(viewsets.ViewSet):
    def list(self, request):
        user_id = request.query_params.get('user_id')
        if user_id:
            stats = get_user_behavior_stats(user_id)
            return Response(stats)
        return Response({})

定时任务

# tasks.py
from celery import shared_task
from django.utils import timezone
from .profiling import update_user_profile

@shared_task
def daily_user_analysis():
    # 更新所有活跃用户画像
    active_users = UserBehavior.objects.filter(
        timestamp__gte=timezone.now()-timedelta(days=1)
    ).values('user').distinct()
    for user in active_users:
        update_user_profile(user['user'])

数据库设计

电商用户行为分析系统的核心数据模型应包含用户、商品、订单和行为日志四大模块。典型设计如下：

用户模型 (User)

• 用户 ID、用户名、密码、邮箱、手机号、注册时间、最后登录时间
• 用户等级、积分、优惠券等扩展属性

商品模型 (Product)

• 商品 ID、名称、分类、价格、库存、销量
• 商品详情、图片、上架时间、评分

订单模型 (Order)

• 订单 ID、用户 ID、商品 ID、数量、总价
• 支付状态、配送地址、创建时间、完成时间

行为日志模型 (UserBehavior)

• 日志 ID、用户 ID、商品 ID、行为类型 (浏览/收藏/加购/购买)
• 行为时间、停留时长、设备信息、IP 地址

关键关联设计

用户与订单的一对多关系：

class Order(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    products = models.ManyToManyField(Product, through='OrderItem')

商品与行为日志的一对多关系：

class UserBehavior(models.Model):
    BEHAVIOR_TYPES = [
        ('view', '浏览'),
        ('fav', '收藏'),
        ('cart', '加购'),
        ('buy', '购买')
    ]
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    product = models.ForeignKey(Product, on_delete=models.CASCADE)
    behavior_type = models.CharField(max_length=10, choices=BEHAVIOR_TYPES)
    timestamp = models.DateTimeField(auto_now_add=True)

系统测试方案

单元测试设计

• 测试用户行为数据采集功能
• 测试数据分析统计功能
• 测试推荐算法准确性

示例测试用例：

from django.test import TestCase
from analytics.models import UserBehavior

class BehaviorTrackingTest(TestCase):
    def test_behavior_recording(self):
        user = User.objects.create(username='test')
        product = Product.objects.create(name='Test Product')
        UserBehavior.objects.create(
            user=user, 
            product=product, 
            behavior_type='view'
        )
        self.assertEqual(UserBehavior.objects.count(), 1)

性能测试要点

• 模拟高并发用户行为数据写入
• 测试大数据量下的查询性能
• 验证定时分析任务的执行效率

集成测试场景

• 用户浏览->加购->购买完整流程验证
• 数据分析报表生成流程测试
• 推荐系统与前端展示集成测试

数据分析功能实现

典型用户行为分析 SQL 示例：

SELECT product_id, COUNT(CASE WHEN behavior_type = 'view' THEN 1 END) as view_count, COUNT(CASE WHEN behavior_type = 'buy' THEN 1 END) as purchase_count, COUNT(CASE WHEN behavior_type = 'buy' THEN 1 END) / NULLIF(COUNT(CASE WHEN behavior_type = 'view' THEN 1 END), 0) as conversion_rate FROM user_behavior GROUP BY product_id ORDER BY conversion_rate DESC

部署优化建议

• 使用 Redis 缓存高频访问的用户行为数据
• 采用 Celery 异步处理耗时分析任务
• 对历史行为数据实施分表存储策略
• 使用 Django Debug Toolbar 监控查询性能