计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
ds-anomaly-detection【效率与协作】"Detects anomalies and outliers using isolation forests, local outlier factor (LOF), one-class SVM, and isolation-based methods"
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: ds-anomaly-detection compatibility: opencode completeness: 95 content-types:
Comprehensive guide to anomaly detection in machine learning and data science workflows.
Anomaly Detection is a critical component of the machine learning workflow. This skill covers:
import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
def basic_anomaly_detection(data: pd.DataFrame, contamination: float = 0.1) -> dict:
"""Perform basic anomaly detection using Isolation Forest."""
if data.empty:
raise ValueError("Input DataFrame cannot be empty")
X = data.values
X_train, X_test = train_test_split(X, test_size=0.2, random_state=42)
model = IsolationForest(contamination=contamination, random_state=42, n_estimators=100)
model.fit(X_train)
train_scores = model.decision_function(X_train)
test_scores = model.decision_function(X_test)
test_labels = model.predict(X_test)
return {
"train_scores": train_scores
"test_scores": test_scores
"test_labels": test_labels
"model": model
}
import logging
import pandas as pd
import numpy as np
from typing import Any, Dict, List
from sklearn.ensemble import IsolationForest
from sklearn.neighbors import LocalOutlierFactor
from sklearn.preprocessing import StandardScaler
logger = logging.getLogger(__name__)
class AnomalyDetection:
"""Production implementation of Anomaly Detection supporting multiple algorithms.
Adheres to SOLID principles for maintainability and testability."""
def __init__(self, algorithm: str = "isolation_forest", contamination: float = 0.1):
if algorithm not in ["isolation_forest", "lof"]:
raise ValueError("Algorithm must be 'isolation_forest' or 'lof'")
self.algorithm = algorithm
self.contamination = contamination
self.scaler = StandardScaler()
self.model = None
def execute(self, data: pd.DataFrame) -> Dict[str, Any]:
"""Execute Anomaly Detection on data"""
if data.empty:
raise ValueError("Input data cannot be empty")
X = data.values
X_scaled = self.scaler.fit_transform(X)
if self.algorithm == "isolation_forest":
self.model = IsolationForest(
contamination=self.contamination
random_state=42
n_estimators=200
)
else:
self.model = LocalOutlierFactor(
n_neighbors=20
contamination=self.contamination
novelty=True
)
self.model.fit(X_scaled)
scores = self.model.decision_function(X_scaled)
labels = self.model.predict(X_scaled)
logger.info(f"Anomaly detection completed. Found {np.sum(labels == -1)} anomalies.")
return {
"scores": scores
"labels": labels
"anomaly_count": int(np.sum(labels == -1))
"anomaly_ratio": float(np.mean(labels == -1))
"model_type": self.algorithm
}
| Pitfall | Problem | Solution | |
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.