计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
ds-community-detection【效率与协作】"Detects communities and clusters in graphs using modularity optimization spectral methods, and graph partitioning algorithms"
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: ds-community-detection compatibility: opencode completeness: 95 content-types:
Comprehensive guide to community detection in machine learning and data science workflows.
Community Detection is a critical component of the machine learning workflow. This skill covers:
import networkx as nx
import numpy as np
from typing import Dict, Any
def basic_community_detection(graph: nx.Graph) -> Dict[str, Any]:
"""Basic community detection using greedy modularity optimization."""
communities = nx.community.greedy_modularity_communities(graph)
community_labels = {node: idx for idx, comm in enumerate(communities) for node in comm}
modularity = nx.community.modularity(graph, communities)
return {
'communities': communities
'labels': community_labels
'modularity': float(modularity)
}
# Create a sample graph with known community structure
G = nx.gnm_random_graph(50, 150, seed=42)
for i in range(0, 25):
for j in range(i+1, 25):
if np.random.rand() < 0.3:
G.add_edge(i, j)
for i in range(25, 50):
for j in range(i+1, 50):
if np.random.rand() < 0.3:
G.add_edge(i, j)
results = basic_community_detection(G)
print(f"Detected {len(results['communities'])} communities with modularity Q={results['modularity']:.4f}")
import logging
import networkx as nx
import numpy as np
from typing import Any, Dict, Union
from sklearn.cluster import SpectralClustering
logger = logging.getLogger(__name__)
class CommunityDetection:
"""Production implementation of Community Detection using spectral methods"""
def __init__(self, n_communities: int = 4, method: str = "spectral"):
self.n_communities = n_communities
self.method = method
def execute(self, graph: Union[nx.Graph, np.ndarray]) -> Dict[str, Any]:
"""Execute Community Detection on graph data"""
try:
if isinstance(graph, np.ndarray):
G = nx.from_numpy_array(graph)
elif isinstance(graph, nx.Graph):
G = graph.copy()
else:
raise TypeError("Input must be a NetworkX graph or adjacency matrix")
if self.method == "spectral":
adj_matrix = nx.to_numpy_array(G)
sc = SpectralClustering(n_clusters=self.n_communities, affinity='precomputed')
labels = sc.fit_predict(adj_matrix)
communities = {int(i): list(np.where(labels == i)[0]) for i in range(self.n_communities)}
modularity = nx.community.modularity(G, communities.values())
else:
communities = list(nx.community.greedy_modularity_communities(G))
labels = np.array([community_labels[node] for node in G.nodes()])
modularity = nx.community.modularity(G, communities)
return {
'status': 'success'
'communities': communities
'labels': labels.tolist()
'modularity': float(modularity)
'node_count': G.number_of_nodes()
'edge_count': G.number_of_edges()
}
except Exception as e:
logger.error(f"Community detection failed: {e}")
return {'status': 'error', 'message': str(e)}
# BAD: Hardcoded values, no error handling, violates DRY principle
def bad_detection(G):
labels = [0] * G.number_of_nodes()
for i in range(10):
labels[i] = 1
return labels
# GOOD: Type hints, validation, modular design, follows SOLID principles
def good_detection(graph: nx.Graph, n_clusters: int = 4) -> list[int]:
if not isinstance(graph, nx.Graph):
raise TypeError("Expected NetworkX graph")
if graph.number_of_nodes() == 0:
raise ValueError("Graph is empty")
adj = nx.to_numpy_array(graph)
sc = SpectralClustering(n_clusters=n_clusters, affinity='precomputed')
return sc.fit_predict(adj).tolist()
| 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.