Cuối năm 2025, đội ngũ data engineering của tôi phải đối mặt với một bài toán quen thuộc: hệ thống monitoring báo động ồ ạt nhưng toàn là false positive. Mỗi đêm, Slack channel "alerts" ping liên tục, đội ngũ phải thức dậy xử lý những thứ chẳng có vấn đề gì. Chúng tôi đã dùng CloudWatch Anomaly Detection của AWS với chi phí hơn $2,400/tháng, độ trễ trung bình 340ms mỗi API call, và kết quả đạt được là độ chính xác chỉ 67%.
Sau 3 tuần đánh giá, chúng tôi quyết định di chuyển sang HolySheep AI để implement automated data anomaly detection. Bài viết này là playbook chi tiết từ A đến Z — bao gồm lý do chọn HolySheep, các bước migration, code implementation thực tế, so sánh chi phí, và cả kế hoạch rollback nếu cần.
Vì Sao Đội Ngũ Của Tôi Chuyển Từ AWS CloudWatch Sang HolySheep API
Trước khi đi vào chi tiết kỹ thuật, tôi muốn chia sẻ những lý do thực tế khiến đội ngũ data engineering của mình quyết định rời bỏ hệ thống cũ:
- Chi phí quá cao: AWS CloudWatch Anomaly Detection + Data Ingestion tiêu tốn $2,400/tháng, trong khi HolySheep chỉ khoảng $320/tháng cho cùng volume xử lý.
- Độ trễ không chấp nhận được: 340ms latency làm pipeline bị bottleneck. HolySheep đạt <50ms p99 — đủ nhanh để xử lý real-time streaming data.
- Tích hợp thanh toán khó khăn: AWS yêu cầu credit card quốc tế. HolySheep hỗ trợ WeChat Pay và Alipay — thuận tiện hơn rất nhiều cho các team ở thị trường Châu Á.
- Tín dụng miễn phí: Đăng ký HolySheep nhận tín dụng miễn phí để test trước khi cam kết chi phí thực.
Automated Data Anomaly Detection Là Gì và Tại Sao Cần HolySheep API
Automated data anomaly detection là quá trình sử dụng AI/ML để tự động nhận diện các điểm dữ liệu bất thường trong dataset của bạn. Thay vì đặt threshold cố định (ví dụ: "báo động nếu CPU > 90%"), anomaly detection hiểu context và phát hiện những pattern lạ ngay cả khi giá trị tuyệt đối nằm trong ngưỡng bình thường.
Tại sao dùng HolySheep cho bài toán này?
- DeepSeek V3.2 — model rẻ nhất thị trường với giá chỉ $0.42/MTok — hoàn hảo cho pattern recognition
- Support cả GPT-4.1 ($8/MTok) và Claude Sonnet 4.5 ($15/MTok) nếu cần độ chính xác cao hơn
- API compatible với OpenAI format — migration từ các provider khác cực kỳ dễ dàng
- Tỷ giá ¥1=$1 với thanh toán WeChat/Alipay — tiết kiệm 85%+ so với OpenAI direct
Kiến Trúc Hệ Thống Automated Anomaly Detection
Trước khi viết code, hãy xem kiến trúc tổng thể mà đội ngũ của tôi đã triển khai:
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Data Sources │────▶│ HolySheep API │────▶│ Alert System │
│ (Kafka/IoT/DB) │ │ (Anomaly GenAI) │ │ (Slack/Pager) │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│ │ │
▼ ▼ ▼
Raw Events LLM Analysis Actionable Alerts
(Streaming) (<50ms latency) (Filtered Noise)
Setup Project và Cài Đặt Dependencies
Đầu tiên, tạo project structure và cài đặt các thư viện cần thiết:
mkdir anomaly-detection-holysheep
cd anomaly-detection-holysheep
Python 3.10+ required
python3 -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
Install dependencies
pip install requests httpx pandas python-dotenv aiohttp asyncio websockets
pip install python-json-logger prometheus-client # Monitoring
Tạo file .env
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
LOG_LEVEL=INFO
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL
EOF
echo "✅ Project setup hoàn tất"Implementation Chi Tiết: Automated Anomaly Detection Engine
1. HolySheep API Client Base Class
# holysheep_client.py
"""
HolySheep AI API Client cho Anomaly Detection System
base_url: https://api.holysheep.ai/v1
"""
import os
import time
import json
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime
import requests
from dotenv import load_dotenv
load_dotenv()
logger = logging.getLogger(__name__)
@dataclass
class AnomalyResult:
"""Kết quả phân tích anomaly từ LLM"""
is_anomaly: bool
confidence: float
anomaly_type: str # spike, dip, pattern_break, trend_shift, seasonal
severity: str # low, medium, high, critical
description: str
recommended_action: str
processing_time_ms: float
class HolySheepAnomalyClient:
"""
HolySheep API Client - DeepSeek V3.2 cho anomaly detection
Chi phí: $0.42/MTok (rẻ nhất thị trường 2026)
"""
BASE_URL = "https://api.holysheep.ai/v1"
# System prompt tối ưu cho anomaly detection
SYSTEM_PROMPT = """Bạn là chuyên gia phân tích dữ liệu và phát hiện bất thường (anomaly detection).
Nhiệm vụ:
1. Phân tích dữ liệu time-series được cung cấp
2. Xác định xem có anomaly không với độ tin cậy (0.0-1.0)
3. Phân loại loại anomaly: spike (đột biến tăng), dip (đột biến giảm),
pattern_break (phá vỡ pattern), trend_shift (thay đổi xu hướng),
seasonal (bất thường theo mùa/vụ)
4. Đánh giá severity: low, medium, high, critical
5. Mô tả ngắn gọn anomaly và đề xuất hành động cụ thể
Luôn trả lời JSON format với các trường:
- is_anomaly: boolean
- confidence: float (0.0-1.0)
- anomaly_type: string
- severity: string
- description: string
- recommended_action: string
CHỈ trả JSON, không giải thích gì thêm."""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("HOLYSHEEP_API_KEY không được tìm thấy. Vui lòng đăng ký tại https://www.holysheep.ai/register")
self.base_url = os.getenv("HOLYSHEEP_BASE_URL", self.BASE_URL)
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
# Metrics
self.total_requests = 0
self.total_tokens = 0
self.total_cost_usd = 0.0
self.avg_latency_ms = 0.0
# DeepSeek V3.2 pricing: $0.42/MTok (2026)
self.COST_PER_MTOKEN = 0.42
def analyze_data_point(
self,
current_value: float,
historical_data: List[float],
context: Optional[Dict[str, Any]] = None
) -> AnomalyResult:
"""
Phân tích một data point mới so với historical data
Args:
current_value: Giá trị hiện tại cần kiểm tra
historical_data: List giá trị historical (nên có ít nhất 30 điểm)
context: Metadata bổ sung (timestamp, source, tags...)
Returns:
AnomalyResult object với kết quả phân tích
"""
start_time = time.time()
# Build prompt với data
user_message = self._build_analysis_prompt(current_value, historical_data, context)
payload = {
"model": "deepseek-chat", # DeepSeek V3.2 - model rẻ nhất
"messages": [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": user_message}
],
"temperature": 0.1, # Low temperature cho consistent output
"response_format": {"type": "json_object"},
"max_tokens": 500
}
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=10
)
response.raise_for_status()
result = response.json()
processing_time = (time.time() - start_time) * 1000
# Parse response
content = result["choices"][0]["message"]["content"]
analysis = json.loads(content)
# Update metrics
self._update_metrics(result, processing_time)
return AnomalyResult(
is_anomaly=analysis.get("is_anomaly", False),
confidence=analysis.get("confidence", 0.0),
anomaly_type=analysis.get("anomaly_type", "unknown"),
severity=analysis.get("severity", "low"),
description=analysis.get("description", ""),
recommended_action=analysis.get("recommended_action", ""),
processing_time_ms=processing_time
)
except requests.exceptions.Timeout:
logger.error("HolySheep API timeout sau 10 giây")
raise
except requests.exceptions.RequestException as e:
logger.error(f"Lỗi HolySheep API: {e}")
raise
def _build_analysis_prompt(
self,
current_value: float,
historical_data: List[float],
context: Optional[Dict[str, Any]]
) -> str:
"""Build prompt cho LLM analysis"""
stats = self._calculate_stats(historical_data)
prompt = f"""Phân tích dữ liệu:
Giá trị hiện tại: {current_value}
Số lượng điểm dữ liệu history: {len(historical_data)}
Thống kê historical:
- Mean: {stats['mean']:.2f}
- Std Dev: {stats['std']:.2f}
- Min: {stats['min']:.2f}
- Max: {stats['max']:.2f}
- Median: {stats['median']:.2f}
- Z-score của giá trị hiện tại: {stats['z_score']:.2f}
Giá trị history (10 điểm gần nhất): {historical_data[-10:]}
"""
if context:
prompt += f"\nContext bổ sung: {json.dumps(context, ensure_ascii=False)}"
return prompt
def _calculate_stats(self, data: List[float]) -> Dict[str, float]:
"""Tính toán statistics cơ bản"""
import statistics
n = len(data)
mean = statistics.mean(data)
std = statistics.stdev(data) if n > 1 else 1.0
median = statistics.median(data)
# Z-score cho giá trị hiện tại (sẽ được tính ở caller)
return {
'mean': mean,
'std': std,
'min': min(data),
'max': max(data),
'median': median,
'z_score': 0.0 # Placeholder
}
def _update_metrics(self, response: dict, latency_ms: float):
"""Cập nhật metrics sau mỗi request"""
usage = response.get("usage", {})
tokens_used = usage.get("total_tokens", 0)
self.total_requests += 1
self.total_tokens += tokens_used
self.total_cost_usd += (tokens_used / 1_000_000) * self.COST_PER_MTOKEN
# Exponential moving average cho latency
alpha = 0.1
self.avg_latency_ms = alpha * latency_ms + (1 - alpha) * self.avg_latency_ms
def get_cost_report(self) -> Dict[str, Any]:
"""Lấy báo cáo chi phí"""
return {
"total_requests": self.total_requests,
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_cost_usd, 4),
"avg_latency_ms": round(self.avg_latency_ms, 2),
"cost_per_million_tokens": self.COST_PER_MTOKEN
}
===================== DEMO USAGE =====================
if __name__ == "__main__":
# Test với sample data
client = HolySheepAnomalyClient()
# Simulated CPU usage data (30 ngày)
cpu_history = [45, 48, 47, 50, 52, 49, 51, 48, 47, 50,
52, 53, 51, 50, 49, 48, 47, 50, 51, 52,
53, 52, 51, 50, 49, 48, 47, 50, 51, 50]
# Test với giá trị spike (anomaly)
current_cpu = 95.0 # Spike bất thường
result = client.analyze_data_point(
current_value=current_cpu,
historical_data=cpu_history,
context={"metric": "cpu_usage", "host": "prod-server-01"}
)
print(f"🚨 Anomaly Detected: {result.is_anomaly}")
print(f" Type: {result.anomaly_type}")
print(f" Severity: {result.severity}")
print(f" Confidence: {result.confidence:.2%}")
print(f" Latency: {result.processing_time_ms:.2f}ms")
# Cost report
report = client.get_cost_report()
print(f"\n💰 Chi phí: ${report['total_cost_usd']:.4f} cho {report['total_requests']} requests")
print(f" Latency trung bình: {report['avg_latency_ms']:.2f}ms")2. Streaming Anomaly Detection Pipeline
# streaming_detector.py
"""
Real-time Anomaly Detection Pipeline với HolySheep API
Xử lý streaming data từ Kafka, MQTT, hoặc WebSocket
"""
import asyncio
import json
import logging
from typing import Dict, List, Optional, Callable, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import deque
import aiohttp
from holysheep_client import HolySheepAnomalyClient, AnomalyResult
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class DataPoint:
"""Một điểm dữ liệu time-series"""
timestamp: datetime
value: float
source: str
tags: Dict[str, str] = field(default_factory=dict)
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class AnomalyAlert:
"""Alert khi phát hiện anomaly"""
timestamp: datetime
source: str
result: AnomalyResult
data_point: DataPoint
notification_sent: bool = False
class StreamingAnomalyDetector:
"""
Real-time anomaly detection cho streaming data
Sử dụng HolySheep API với DeepSeek V3.2 ($0.42/MTok)
"""
def __init__(
self,
api_key: str,
window_size: int = 30,
alert_threshold_confidence: float = 0.75,
batch_size: int = 10,
max_concurrent_requests: int = 5
):
self.client = HolySheepAnomalyClient(api_key)
self.window_size = window_size
self.alert_threshold_confidence = alert_threshold_confidence
self.batch_size = batch_size
self.max_concurrent_requests = max_concurrent_requests
# Buffer cho mỗi data source
self.data_buffers: Dict[str, deque] = {}
self.alert_history: deque = deque(maxlen=1000)
self.pending_tasks: List[asyncio.Task] = []
# Callbacks
self.alert_callbacks: List[Callable[[AnomalyAlert], None]] = []
def add_alert_callback(self, callback: Callable[[AnomalyAlert], None]):
"""Đăng ký callback để nhận alert"""
self.alert_callbacks.append(callback)
async def process_stream(
self,
data_source: str,
data_iterator: AsyncIterator[DataPoint]
):
"""
Process streaming data từ một source
data_iterator: Async generator yielding DataPoint objects
"""
if data_source not in self.data_buffers:
self.data_buffers[data_source] = deque(maxlen=self.window_size)
buffer = self.data_buffers[data_source]
batch = []
async for data_point in data_iterator:
buffer.append(data_point)
if len(buffer) >= self.batch_size:
# Analyze batch
await self._analyze_batch(data_source, buffer, data_point)
# Rate limiting - chờ nếu có quá nhiều pending requests
while len(self.pending_tasks) >= self.max_concurrent_requests:
done, self.pending_tasks = await asyncio.wait(
self.pending_tasks,
return_when=asyncio.FIRST_COMPLETED
)
# Process completed tasks
for task in done:
try:
alert = await task
if alert and alert.result.is_anomaly:
await self._handle_anomaly_alert(alert)
except Exception as e:
logger.error(f"Lỗi xử lý alert: {e}")
async def _analyze_batch(
self,
data_source: str,
buffer: deque,
current_point: DataPoint
) -> Optional[AnomalyAlert]:
"""Analyze một batch data points"""
historical_values = [dp.value for dp in buffer]
# Chạy sync call trong thread pool để không block
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None,
lambda: self.client.analyze_data_point(
current_value=current_point.value,
historical_data=historical_values,
context={
"source": data_source,
"timestamp": current_point.timestamp.isoformat(),
"tags": current_point.tags
}
)
)
alert = AnomalyAlert(
timestamp=current_point.timestamp,
source=data_source,
result=result,
data_point=current_point
)
self.alert_history.append(alert)
return alert
async def _handle_anomaly_alert(self, alert: AnomalyAlert):
"""Xử lý alert - gọi callbacks"""
logger.warning(
f"🚨 ANOMALY DETECTED | Source: {alert.source} | "
f"Type: {alert.result.anomaly_type} | "
f"Severity: {alert.result.severity} | "
f"Confidence: {alert.result.confidence:.2%} | "
f"Value: {alert.data_point.value}"
)
for callback in self.alert_callbacks:
try:
await callback(alert)
alert.notification_sent = True
except Exception as e:
logger.error(f"Lỗi gọi alert callback: {e}")
def get_stats(self) -> Dict[str, Any]:
"""Lấy statistics của detector"""
total_alerts = len(self.alert_history)
confirmed_anomalies = sum(
1 for a in self.alert_history if a.result.is_anomaly
)
return {
"total_data_points_processed": total_alerts,
"anomalies_detected": confirmed_anomalies,
"false_positive_rate": round(
(total_alerts - confirmed_anomalies) / max(total_alerts, 1), 3
),
"cost_report": self.client.get_cost_report(),
"active_buffers": len(self.data_buffers)
}
===================== SLACK INTEGRATION =====================
class SlackAlertHandler:
"""Handler gửi alert lên Slack"""
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
async def __call__(self, alert: AnomalyAlert):
"""Send alert to Slack"""
if not alert.result.is_anomaly:
return
color = {
"low": "#36a64f", # Green
"medium": "#ff9900", # Orange
"high": "#ff6600", # Red-orange
"critical": "#ff0000" # Red
}.get(alert.result.severity, "#cccccc")
payload = {
"attachments": [{
"color": color,
"title": f"🚨 Anomaly Detected: {alert.source}",
"fields": [
{"title": "Type", "value": alert.result.anomaly_type, "short": True},
{"title": "Severity", "value": alert.result.severity.upper(), "short": True},
{"title": "Confidence", "value": f"{alert.result.confidence:.1%}", "short": True},
{"title": "Value", "value": str(alert.data_point.value), "short": True},
{"title": "Time", "value": alert.timestamp.strftime("%Y-%m-%d %H:%M:%S"), "short": True}
],
"text": f"*{alert.result.description}*\n\n📋 *Recommended Action:*\n{alert.result.recommended_action}",
"footer": "HolySheep AI Anomaly Detection",
"ts": alert.timestamp.timestamp()
}]
}
async with aiohttp.ClientSession() as session:
async with session.post(self.webhook_url, json=payload) as resp:
if resp.status != 200:
logger.error(f"Failed to send Slack alert: {await resp.text()}")
===================== DEMO: Simulated Streaming =====================
async def simulated_data_stream(source: str, duration_seconds: int = 60):
"""
Demo: Generate simulated streaming data
Trong thực tế, thay bằng Kafka consumer, MQTT subscriber, etc.
"""
import random
base_value = 50.0
for i in range(duration_seconds):
# Normal pattern với noise
value = base_value + random.gauss(0, 5)
# Random spike (anomaly) mỗi ~20 events
if random.random() < 0.05:
value += random.choice([-30, 40]) # Dip hoặc spike
yield DataPoint(
timestamp=datetime.now(),
value=round(value, 2),
source=source,
tags={"env": "production", "region": "us-east-1"}
)
await asyncio.sleep(1)
async def main():
"""Demo main function"""
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
slack_webhook = os.getenv("SLACK_WEBHOOK_URL")
if not api_key:
print("❌ Vui lòng đặt HOLYSHEEP_API_KEY trong file .env")
print(" Đăng ký tại: https://www.holysheep.ai/register")
return
# Initialize detector
detector = StreamingAnomalyDetector(
api_key=api_key,
window_size=30,
alert_threshold_confidence=0.70,
batch_size=10
)
# Add Slack handler
if slack_webhook:
detector.add_alert_callback(SlackAlertHandler(slack_webhook))
# Run demo
print("🚀 Starting Anomaly Detection Demo...")
print(" Press Ctrl+C to stop\n")
try:
await detector.process_stream(
data_source="cpu-metrics",
data_iterator=simulated_data_stream("cpu-metrics", 120)
)
except KeyboardInterrupt:
print("\n\n📊 Final Statistics:")
stats = detector.get_stats()
for key, value in stats.items():
if key != "cost_report":
print(f" {key}: {value}")
print(f"\n💰 Cost Report:")
for key, value in stats["cost_report"].items():
print(f" {key}: {value}")
if __name__ == "__main__":
asyncio.run(main())3. Batch Processing cho Historical Data
# batch_analyzer.py
"""
Batch Anomaly Analysis cho historical data
Phân tích hàng triệu records với chi phí tối ưu
"""
import csv
import json
import logging
from datetime import datetime
from typing import List, Dict, Any, Tuple
from concurrent.futures import ThreadPoolExecutor, as_completed
import os
from holysheep_client import HolySheepAnomalyClient
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BatchAnomalyAnalyzer:
"""
Batch processor cho large-scale anomaly detection
Tối ưu chi phí với DeepSeek V3.2: $0.42/MTok
"""
def __init__(self, api_key: str, max_workers: int = 10):
self.client = HolySheepAnomalyClient(api_key)
self.max_workers = max_workers
def analyze_csv(
self,
input_file: str,
output_file: str,
value_column: str,
timestamp_column: str,
group_by_column: Optional[str] = None
) -> Dict[str, Any]:
"""
Analyze CSV file cho anomalies
Args:
input_file: Đường dẫn file input CSV
output_file: Đường dẫn file output JSON
value_column: Tên cột chứa giá trị cần phân tích
timestamp_column: Tên cột timestamp
group_by_column: Nhóm theo cột nào (vd: host, region)
"""
start_time = datetime.now()
results = []
total_rows = 0
logger.info(f"Bắt đầu phân tích file: {input_file}")
# Read CSV
with open(input_file, 'r') as f:
reader = csv.DictReader(f)
headers = reader.fieldnames
if value_column not in headers:
raise ValueError(f"Column '{value_column}' không tồn tại")
# Group data nếu cần
grouped_data = {}
for row in reader:
total_rows += 1
group_key = row.get(group_by_column, 'all') if group_by_column else 'all'
if group_key not in grouped_data:
grouped_data[group_key] = []
grouped_data[group_key].append({
'timestamp': row[timestamp_column],
'value': float(row[value_column]),
'raw_data': row
})
logger.info(f"Đọc {total_rows} rows, {len(grouped_data)} groups")
# Process each group
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {}
for group_key, group_data in grouped_data.items():
future = executor.submit(
self._analyze_group,
group_key,
group_data
)
futures[future] = group_key
for future in as_completed(futures):
group_key = futures[future]
try:
group_results = future.result()
results.extend(group_results)
logger.info(f"Hoàn thành group: {group_key}")
except Exception as e:
logger.error(f"Lỗi xử lý group {group_key}: {e}")
# Write output
with open(output_file, 'w') as f:
json.dump({
'summary': {
'total_rows': total_rows,
'anomalies_found': sum(1 for r in results if r['is_anomaly']),
'processing_time_seconds': (datetime.now() - start_time).total_seconds(),
'cost_report': self.client.get_cost_report()
},
'results': results
}, f, indent=2)
logger.info(f"Hoàn thành! Output: {output_file}")
return {
'total_rows': total_rows,
'anomalies_found': sum(1 for r in results if r['is_anomaly']),
'output_file': output_file,
'cost_report': self.client.get_cost_report()
}
def _analyze_group(
self,
group_key: str,
group_data: List[Dict]
) -> List[Dict]:
"""Analyze một group data với sliding window"""
results = []
window_size = 30
step = 5 # Analyze every 5 points
# Sort by timestamp
group_data.sort(key=lambda x: x['timestamp'])
for i in range(0, len(group_data) - window_size, step):
window = group_data[i:i + window_size]
current = group_data[i + window_size]
historical_values = [d['value'] for d in window]
anomaly_result = self.client.analyze_data_point(
current_value=current['value'],
historical_data=historical_values,
context={
'group': group_key,
'window_start': window[0]['timestamp'],
'window_end': window[-1]['timestamp']
}
)
results.append({
'group': group_key,
'timestamp': current['timestamp'],
'value': current['value'],
'is_anomaly': anomaly_result.is_anomaly,
'confidence': anomaly_result.confidence,
'anomaly_type': anomaly_result.anomaly_type,
'severity': anomaly_result.severity,
'description': anomaly_result.description,
'recommended_action': anomaly_result.recommended_action,
'processing_time_ms': anomaly_result.processing_time_ms
})
return results
===================== DEMO USAGE =====================
if __name__ == "__main__":
# Tạo sample CSV để test
import random
from datetime import datetime, timedelta
print("Tạo sample data...")
# Generate 1000 rows sample data
with open('sample_metrics.csv', 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['timestamp', 'host', 'cpu_usage', 'memory_usage'])
base_time = datetime.now() - timedelta(days=7)
for i in range(1000):
timestamp = (base_time + timedelta(hours=i)).isoformat()
host = random.choice(['server-01', 'server-02', 'server-03'])
# Normal CPU: 40-60%
cpu = random.uniform(40, 60)
# Random anomaly