TimeSeries One-Class SVM MCP
作者：庞力铖
邮箱：3522236586@qq.com
GitHub：https://github.com/Plc912/TimeSeries-One-Class-SVM-MCP.git
基于 One-Class SVM 的时间序列异常检测工具，使用 fastMCP 框架封装为标准 MCP 工具，通过 SSE（Server-Sent Events）协议提供流式异常检测结果。
功能特性
- ✅ 基于 One-Class SVM 的时间序列异常检测
- ✅ 支持 CSV 和 TXT 格式数据文件
- ✅ SSE 流式返回检测结果
- ✅ 支持自然语言 prompt 指定检测参数
- ✅ 滑动窗口特征工程
- ✅ 模型保存与加载
- ✅ 大数据集流式处理优化
安装依赖
1. 确保 Python 版本 >= 3.11
python --version2. 安装项目依赖
pip install -r requirements.txt启动服务
直接运行主程序
python main.py服务启动后，你将看到：
启动 TimeSeries One-Class SVM MCP 服务
使用 fastMCP 框架
服务地址: http://127.0.0.1:3005
传输协议: SSE (Server-Sent Events)fastMCP 服务器将在 http://127.0.0.1:3005 启动，使用 SSE 传输协议。
数据格式要求
数据文件需要满足以下格式：
- 第一列：时间戳（可解析为 datetime 格式）
- 第二列：数值（float 类型）
CSV 格式示例
timestamp,value
2024-01-01 00:00:00,100.5
2024-01-01 01:00:00,102.3
2024-01-01 02:00:00,98.7
...TXT 格式示例（制表符分隔）
timestamp value
2024-01-01 00:00:00 100.5
2024-01-01 01:00:00 102.3
2024-01-01 02:00:00 98.7
...数据集anomalies_c41b7217-1e8b-42f1-8097-7290f0e2ad77.csv是检测出的异常数据，由于token限制只能导出为csv格式的数据集；
客户端调用示例
Python 客户端示例
1. 发送检测请求
import requests
import json
import sseclient
# 1. 发送检测请求
detect_url = "http://127.0.0.1:3005/mcp/detect"
# 方式一：使用 JSON 格式参数
request_data = {
"command": "detect",
"file_path": "data.csv", # 数据文件路径（相对于服务运行目录或绝对路径）
"nu": 0.05,
"window_size": 48,
"threshold_quantile": 0.01,
"kernel": "rbf",
"gamma": "scale"
}
response = requests.post(detect_url, json=request_data)
result = response.json()
task_id = result["task_id"]
print(f"任务 ID: {task_id}")
# 方式二：使用自然语言 prompt
request_data = {
"prompt": "检测 data.csv 文件中的异常，nu=0.05，窗口大小=48，阈值分位数=0.01"
}
response = requests.post(detect_url, json=request_data)
result = response.json()
task_id = result["task_id"]2. 接收 SSE 流式结果
# 2. 接收 SSE 流式结果
# fastMCP 使用 SSE 协议流式返回结果
sse_url = f"http://127.0.0.1:3005/sse/{task_id}"
# 使用 sseclient 库接收 SSE 流
messages = sseclient.SSEClient(sse_url)
for msg in messages:
if msg.event == "start":
print("开始检测...")
elif msg.event == "anomaly_result":
data = json.loads(msg.data)
print(f"时间戳: {data['timestamp']}, "
f"值: {data['value']}, "
f"异常分数: {data['score']:.4f}, "
f"标签: {'异常' if data['label'] == 1 else '正常'}")
elif msg.event == "complete":
print("检测完成！")
break
elif msg.event == "error":
error_data = json.loads(msg.data)
print(f"错误: {error_data['error']}")
break使用 curl 测试
1. 发送检测请求
curl -X POST http://127.0.0.1:3005/mcp/detect \
-H "Content-Type: application/json" \
-d '{
"file_path": "data.csv",
"nu": 0.05,
"window_size": 48,
"threshold_quantile": 0.01
}'2. 接收 SSE 流（使用任务 ID）
curl -N http://127.0.0.1:3005/sse/{task_id}MCP 配置与自然语言 Prompt
MCP 工具配置
MCP 工具配置信息存储在 mcp_config.json 中，包含：
- 工具名称和版本
- 服务端口和协议
- API 端点信息
- 参数说明
标准 JSON 格式
也可以直接使用 JSON 格式指定所有参数：
{
"command": "detect",
"file_path": "data.csv",
"nu": 0.05,
"window_size": 48,
"threshold_quantile": 0.01,
"kernel": "rbf",
"gamma": "scale"
}参数说明
1. nu 参数（异常值比例上界）
范围
- 有效范围:
0 = 2 - 推荐范围:
10 - 200 - 默认值:
48 - 类型:
int
说明
window_size用于构建时间序列特征，表示滑动窗口包含的数据点数量- 窗口大小影响特征提取的粒度和模型对时间模式的捕获能力
- 窗口太小：可能无法捕获长期趋势和周期性模式
- 窗口太大：可能过度平滑，丢失短期异常信号
推荐值（根据数据频率）
- 小时级数据（每小时一个点）:
- 短期模式: 12 - 24（半天到一天） - 中期模式: 48 - 168（2天到一周） - 长期模式: 168 - 720（一周到一个月）
- 分钟级数据（每分钟一个点）:
- 短期模式: 30 - 60（半小时到一小时） - 中期模式: 60 - 1440（1小时到一天） - 长期模式: 1440 - 10080（一天到一周）
- 日级数据（每天一个点）:
- 短期模式: 7 - 14（一周到两周） - 中期模式: 30 - 90（一个月到三个月） - 长期模式: 90 - 365（三个月到一年）
注意事项
- 窗口大小不能超过数据点总数
- 建议窗口大小不超过数据点总数的 1/10
- 对于小数据集（= 2
|48|10 - 200` |
| threshold_quantile | float | 0 <= threshold_quantile <= 1 | 0.01 | 0.001 - 0.2 |
检测结果格式
每条检测结果包含以下字段：
{
"timestamp": "2024-01-01T00:00:00",
"value": 100.5,
"score": -0.0234,
"label": 0
}- timestamp: 时间戳（ISO 格式）
- value: 原始数值
- score: 异常分数（decision_function），分数越小越可能是异常
- label: 异常标签（0=正常，1=异常）
模型保存与加载
模型会自动保存到 models/ 目录，文件名格式为：
model_nu{nu}_kernel{kernel}.pkl可以通过 AnomalyDetector 类的 save_model() 和 load_model() 方法进行模型的保存和加载。
性能优化
- 滑动窗口特征生成使用生成器，减少内存占用
- SSE 流式返回，客户端可以实时接收结果