计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
ds-clustering【效率与协作】"Implements clustering algorithms including K-means, hierarchical clustering DBSCAN, Gaussian mixture models, and spectral clustering"
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: ds-clustering compatibility: opencode completeness: 95 content-types:
Comprehensive guide to clustering in machine learning and data science workflows.
Clustering is a critical component of the machine learning workflow. This skill covers:
import pandas as pd
import numpy as np
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
from sklearn.datasets import make_blobs
from sklearn.preprocessing import StandardScaler
from typing import Dict, Any
def basic_clustering_workflow(
n_samples: int = 300,
n_clusters: int = 3
) -> Dict[str, Any]:
"""
Demonstrates a basic clustering workflow with validation, scaling,
and evaluation metrics. Follows PEP 8 and scikit-learn API standards.
"""
if n_samples < n_clusters:
raise ValueError("Number of samples must exceed number of clusters")
X, y_true = make_blobs(n_samples=n_samples, centers=n_clusters, random_state=42)
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
cluster_labels = kmeans.fit_predict(X_scaled)
silhouette_avg = silhouette_score(X_scaled, cluster_labels)
inertia = kmeans.inertia_
return {
"labels": cluster_labels
"silhouette_score": float(silhouette_avg)
"inertia": float(inertia)
"centers": kmeans.cluster_centers_
}
import logging
import pandas as pd
import numpy as np
from typing import Any, Dict, Literal
from sklearn.cluster import KMeans, DBSCAN
from sklearn.preprocessing import StandardScaler
import warnings
logger = logging.getLogger(__name__)
class ProductionClustering:
"""Production-ready clustering implementation with validation and metrics."""
def __init__(self, algorithm: Literal["kmeans", "dbscan"] = "kmeans", **kwargs: Any):
self.algorithm = algorithm
self.kwargs = kwargs
self.scaler = StandardScaler()
self.model = None
def _validate_input(self, data: pd.DataFrame) -> pd.DataFrame:
if data.empty:
raise ValueError("Input DataFrame cannot be empty")
if not np.isfinite(data.values).all():
raise ValueError("Input data contains non-finite values")
return data.dropna()
def execute(self, data: pd.DataFrame) -> Dict[str, Any]:
try:
clean_data = self._validate_input(data)
X_scaled = self.scaler.fit_transform(clean_data)
if self.algorithm == "kmeans":
self.model = KMeans(random_state=42, n_init=10, **self.kwargs)
elif self.algorithm == "dbscan":
self.model = DBSCAN(**self.kwargs)
else:
raise ValueError(f"Unsupported algorithm: {self.algorithm}")
labels = self.model.fit_predict(X_scaled)
result = {
"status": "success"
"labels": labels
"n_clusters": len(np.unique(labels)) if self.algorithm == "kmeans" else "variable"
"inertia": getattr(self.model, 'inertia_', None)
"metadata": {"rows_processed": len(clean_data), "algorithm": self.algorithm}
}
logger.info("Clustering completed successfully.")
return result
except Exception as e:
logger.error(f"Clustering failed: {str(e)}")
return {"status": "failed", "error": str(e)}
# BAD: Hardcoded values, no validation, ignores scaling, bypasses error handling
def bad_clustering(data: pd.DataFrame) -> np.ndarray:
model = KMeans(n_clusters=3)
return model.fit_predict(data)
# GOOD: Validates input, scales data, checks convergence, returns structured results
def good_clustering(data: pd.DataFrame, n_clusters: int = 3) -> Dict[str, Any]:
if data.empty or data.shape[0] < n_clusters:
raise ValueError("Insufficient data for clustering")
scaler = StandardScaler()
X_scaled = scaler.fit_transform(data)
model = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
labels = model.fit_predict(X_scaled)
return {"labels": labels, "inertia": model.inertia_, "scaler": scaler}
| 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.