1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
| # Python 实现 - ML 异常检测
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
class MLBotDetector:
"""基于机器学习的机器人检测"""
def __init__(self):
self.model = None
self.scaler = StandardScaler()
def extract_features(self, user_data: Dict) -> np.ndarray:
"""提取特征向量"""
features = [
# 请求频率特征
user_data.get('requests_per_minute', 0),
user_data.get('requests_per_hour', 0),
user_data.get('unique_endpoints', 0),
# 时间模式特征
user_data.get('timing_mean', 0),
user_data.get('timing_std', 0),
user_data.get('timing_min', 0),
user_data.get('timing_max', 0),
# 会话特征
user_data.get('session_duration', 0),
user_data.get('page_views', 0),
user_data.get('bounce_rate', 0),
# 设备特征
1 if user_data.get('has_touch', False) else 0,
1 if user_data.get('has_sensors', False) else 0,
user_data.get('screen_width', 0),
user_data.get('screen_height', 0),
# 行为特征
user_data.get('mouse_movements', 0),
user_data.get('keyboard_events', 0),
user_data.get('scroll_events', 0)
]
return np.array(features).reshape(1, -1)
def train(self, training_data: List[Dict]):
"""训练模型"""
X = np.vstack([
self.extract_features(d) for d in training_data
])
# 标准化
X_scaled = self.scaler.fit_transform(X)
# 使用隔离森林进行异常检测
self.model = IsolationForest(
n_estimators=100,
contamination=0.1, # 预期异常比例
random_state=42
)
self.model.fit(X_scaled)
def predict(self, user_data: Dict) -> Dict:
"""预测是否为机器人"""
if self.model is None:
raise ValueError("模型未训练")
X = self.extract_features(user_data)
X_scaled = self.scaler.transform(X)
# -1 表示异常，1 表示正常
prediction = self.model.predict(X_scaled)[0]
score = self.model.decision_function(X_scaled)[0]
return {
'is_bot': prediction == -1,
'confidence': abs(score),
'anomaly_score': -score # 越高越异常
}
def save_model(self, path: str):
"""保存模型"""
joblib.dump({
'model': self.model,
'scaler': self.scaler
}, path)
def load_model(self, path: str):
"""加载模型"""
data = joblib.load(path)
self.model = data['model']
self.scaler = data['scaler']
class DeepLearningBotDetector:
"""基于深度学习的机器人检测"""
def __init__(self):
self.model = None
def build_model(self, input_dim: int):
"""构建神经网络模型"""
import tensorflow as tf
self.model = tf.keras.Sequential([
tf.keras.layers.Dense(128, activation='relu', input_shape=(input_dim,)),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Dense(64, activation='relu'),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Dense(32, activation='relu'),
tf.keras.layers.Dense(1, activation='sigmoid')
])
self.model.compile(
optimizer='adam',
loss='binary_crossentropy',
metrics=['accuracy', 'AUC']
)
def train(self, X_train: np.ndarray, y_train: np.ndarray,
X_val: np.ndarray = None, y_val: np.ndarray = None):
"""训练模型"""
callbacks = [
tf.keras.callbacks.EarlyStopping(
patience=5,
restore_best_weights=True
),
tf.keras.callbacks.ReduceLROnPlateau(
factor=0.5,
patience=3
)
]
validation_data = None
if X_val is not None and y_val is not None:
validation_data = (X_val, y_val)
self.model.fit(
X_train, y_train,
epochs=100,
batch_size=32,
validation_data=validation_data,
callbacks=callbacks
)
def predict(self, X: np.ndarray) -> np.ndarray:
"""预测"""
return self.model.predict(X)
|
支付宝
微信