stock-operation/user_events/analyze/DetectAnomalies.py

# -*- coding: utf-8 -*-
"""
@author: zsc
@time: 2024/11/18
@desc:  行为分析
"""
from collections import defaultdict


# 异常检测模块
class AnomalyDetector:
    def __init__(self, data):
        self.data = data

    def detect(self):
        # 简单示例：检测行为次数异常高的用户
        user_behavior_count = defaultdict(int)
        for user_actions in self.data:
            user = user_actions['user']
            actions = user_actions['actions']
            user_behavior_count[user] += len(actions)

        anomalies = []
        for user, count in user_behavior_count.items():
            if count > 20:  # 假设行为次数超过20为异常
                anomalies.append(user)

        return anomalies