""" detect_communities.py --------------------- Loads graph.gpickle, builds two separate customer-customer projections (strong-anchored and weak-only), runs community detection on each, filters candidate clusters, validates against ground truth, and writes candidate_clusters.json. Phase 2 fix: uses two separate connectivity passes instead of one blended weight projection, preventing incidental address-sharing from merging otherwise-unrelated rings into a single blob. Usage: python graph/detect_communities.py [--data-dir DATA_DIR] DATA_DIR defaults to data/synthetic/ """ import argparse import json import os import pickle import sys from collections import defaultdict import community as community_louvain # python-louvain import networkx as nx # --------------------------------------------------------------------------- # Paths (overridable via CLI so the same script serves both datasets) # --------------------------------------------------------------------------- BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) LOUVAIN_THRESHOLD = 12 LOUVAIN_RESOLUTION = 1.0 LOUVAIN_SEED = 42 # --------------------------------------------------------------------------- # Load graph (handles both nx.read_gpickle and pickle.load fallback) # --------------------------------------------------------------------------- def load_graph(path: str) -> nx.Graph: try: return nx.read_gpickle(path) except AttributeError: with open(path, "rb") as f: return pickle.load(f) # --------------------------------------------------------------------------- # 1.2 Build two customer-customer projections # --------------------------------------------------------------------------- # Strong edge formation threshold: a customer-entity-customer edge only counts # as a strong link if the customer has used that shared entity in >= 2 orders. # This filters out accidental single-order pool collisions (which would create # cross-ring contamination) while preserving intentional ring device sharing # (rings place 2-4 orders per member on their shared device). # same_contact edges are always strong regardless of order count. STRONG_EDGE_MIN_ORDERS = 2 def _build_entity_customer_index(G: nx.Graph): """ Returns three items: strong_entity_to_custs: entity_id → list of (customer_id, order_count) for STRONG edges (uses_device / pays_with). weak_entity_to_custs: entity_id → list of customer_ids for WEAK edges. same_contact_pairs: list of (customer_a, customer_b) direct strong edges. """ strong_entity_to_custs: dict[str, list[tuple[str, int]]] = defaultdict(list) weak_entity_to_custs: dict[str, list[str]] = defaultdict(list) same_contact_pairs: list[tuple[str, str]] = [] for u, v, data in G.edges(data=True): u_type = G.nodes[u].get("node_type", "") v_type = G.nodes[v].get("node_type", "") # same_contact is a direct customer-customer strong edge if data.get("relation") == "same_contact": if u_type == "customer" and v_type == "customer": same_contact_pairs.append((u, v)) continue # For customer-entity edges determine which customer and which entity if u_type == "customer" and v_type != "customer": cust, entity = u, v elif v_type == "customer" and u_type != "customer": cust, entity = v, u else: continue strength = data.get("strength", "weak") order_count = data.get("count", 1) if strength == "strong": strong_entity_to_custs[entity].append((cust, order_count)) else: weak_entity_to_custs[entity].append(cust) return strong_entity_to_custs, weak_entity_to_custs, same_contact_pairs def build_strong_subgraph(G: nx.Graph) -> nx.Graph: """ Edge between two customers exists only if they share >= 1 STRONG link (shared device used >= STRONG_EDGE_MIN_ORDERS times, shared payment instrument used >= STRONG_EDGE_MIN_ORDERS times, or same_contact). Edge weight = count of shared strong-link attributes. The order-count threshold eliminates accidental single-order pool collisions that would create false cross-ring bridges in the data. """ strong_entity_to_custs, _, same_contact_pairs = _build_entity_customer_index(G) all_customers = [n for n, d in G.nodes(data=True) if d.get("node_type") == "customer"] SG = nx.Graph() SG.add_nodes_from(all_customers) # Shared entity edges — only for customers meeting the order-count threshold for entity, custs_and_counts in strong_entity_to_custs.items(): eligible = [(c, cnt) for c, cnt in custs_and_counts if cnt >= STRONG_EDGE_MIN_ORDERS] if len(eligible) < 2: continue for i in range(len(eligible)): for j in range(i + 1, len(eligible)): ca, _ = eligible[i] cb, _ = eligible[j] if ca == cb: continue if SG.has_edge(ca, cb): SG[ca][cb]["weight"] += 1 SG[ca][cb]["strong_share_count"] += 1 else: SG.add_edge(ca, cb, weight=1, strong_share_count=1, weak_share_count=0) # same_contact direct edges (no order-count threshold — contact match is always strong) for (ca, cb) in same_contact_pairs: if SG.has_edge(ca, cb): SG[ca][cb]["weight"] += 1 SG[ca][cb]["strong_share_count"] += 1 else: SG.add_edge(ca, cb, weight=1, strong_share_count=1, weak_share_count=0) return SG def build_weak_only_subgraph(G: nx.Graph, strong_subgraph: nx.Graph) -> nx.Graph: """ Edge between two customers exists only if they share >= 1 WEAK link (address) AND they share ZERO strong links (i.e., not already connected in strong_subgraph). Edge weight = count of shared weak-link attributes. """ _, weak_entity_to_custs, _ = _build_entity_customer_index(G) all_customers = [n for n, d in G.nodes(data=True) if d.get("node_type") == "customer"] WG = nx.Graph() WG.add_nodes_from(all_customers) for entity, custs in weak_entity_to_custs.items(): if len(custs) < 2: continue for i in range(len(custs)): for j in range(i + 1, len(custs)): ca, cb = custs[i], custs[j] if ca == cb: continue # Skip if the pair has any strong link if strong_subgraph.has_edge(ca, cb): continue if WG.has_edge(ca, cb): WG[ca][cb]["weight"] += 1 WG[ca][cb]["weak_share_count"] += 1 else: WG.add_edge(ca, cb, weight=1, strong_share_count=0, weak_share_count=1) return WG # --------------------------------------------------------------------------- # 1.3 + 1.4 Community detection on each subgraph # --------------------------------------------------------------------------- def detect_on_subgraph(subgraph: nx.Graph, formation_type: str) -> list[tuple[frozenset, str, str]]: """ Returns list of (frozenset_of_customer_ids, method_string, formation_type). """ connected_nodes = [n for n in subgraph.nodes() if subgraph.degree(n) > 0] if not connected_nodes: return [] CG_connected = subgraph.subgraph(connected_nodes) components = list(nx.connected_components(CG_connected)) communities = [] for comp in components: if len(comp) <= LOUVAIN_THRESHOLD: communities.append((frozenset(comp), "connected_components", formation_type)) else: sub = subgraph.subgraph(comp).copy() partition = community_louvain.best_partition( sub, weight="weight", resolution=LOUVAIN_RESOLUTION, random_state=LOUVAIN_SEED, ) sub_communities: dict[int, list[str]] = defaultdict(list) for node, label in partition.items(): sub_communities[label].append(node) for members in sub_communities.values(): communities.append((frozenset(members), "louvain", formation_type)) return communities # --------------------------------------------------------------------------- # Compute cluster metadata (strong_link_count + weak_link_count) # These are computed from BOTH subgraphs regardless of formation_type (1.7) # --------------------------------------------------------------------------- def compute_cluster_metadata( cluster_nodes: frozenset, strong_subgraph: nx.Graph, weak_only_subgraph: nx.Graph, ) -> dict: """ strong_link_count: total strong-link edges within the cluster weak_link_count: total weak-link edges within the cluster Both are computed regardless of formation_type (per spec 1.7). """ strong = 0 weak = 0 members = list(cluster_nodes) for i in range(len(members)): for j in range(i + 1, len(members)): ca, cb = members[i], members[j] if strong_subgraph.has_edge(ca, cb): strong += strong_subgraph[ca][cb].get("strong_share_count", 0) # weak_only edges: only pairs not in strong_subgraph if weak_only_subgraph.has_edge(ca, cb): weak += weak_only_subgraph[ca][cb].get("weak_share_count", 0) return {"strong_link_count": strong, "weak_link_count": weak} # --------------------------------------------------------------------------- # Write candidate_clusters.json # --------------------------------------------------------------------------- def write_clusters(clusters: list[dict], path: str): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(clusters, f, indent=2, ensure_ascii=False) print(f"Wrote {path} ({len(clusters)} clusters)") # --------------------------------------------------------------------------- # Ground-truth validation (works for both Phase 1 and Phase 2 datasets) # Phase 1 ground truth key: "stress_clusters" # Phase 2 ground truth key: "benign_clusters" (falls back to "stress_clusters") # --------------------------------------------------------------------------- def validate(clusters: list[dict], ground_truth_path: str): with open(ground_truth_path, encoding="utf-8") as f: gt = json.load(f) rings = gt["rings"] # Support both Phase 1 ("stress_clusters") and Phase 2 ("benign_clusters") naming benign_clusters = gt.get("benign_clusters", gt.get("stress_clusters", [])) cand_sets = [set(c["customer_ids"]) for c in clusters] # Ring recovery: each ring should appear in its OWN separate candidate cluster rings_recovered = 0 ring_detail = [] for ring in rings: ring_members = set(ring["customer_ids"]) ring_id = ring.get("ring_id", ring.get("cluster_id", "?")) threshold = len(ring_members) * 0.5 best_overlap = 0 best_cand = None for i, cs in enumerate(cand_sets): overlap = len(ring_members & cs) if overlap > best_overlap: best_overlap = overlap best_cand = clusters[i]["cluster_id"] recovered = best_overlap >= threshold if recovered: rings_recovered += 1 ring_detail.append({ "ring_id": ring_id, "ring_size": len(ring_members), "best_overlap": best_overlap, "recovered": recovered, "matched_cluster": best_cand, }) # Separation check: rings should be in SEPARATE candidates # Count how many rings map to the same best candidate matched_by_cand: dict[str, int] = defaultdict(int) for d in ring_detail: if d["matched_cluster"]: matched_by_cand[d["matched_cluster"]] += 1 rings_sharing_cluster = sum(1 for cnt in matched_by_cand.values() if cnt > 1) # Benign/stress cluster check: should appear as weak_only candidates # with strong_link_count == 0 benign_detail = [] benign_correctly_weak = 0 for sc in benign_clusters: sc_members = set(sc["customer_ids"]) sc_id = sc.get("cluster_id", "?") threshold = len(sc_members) * 0.5 best_overlap = 0 best_cand_id = None best_strong = 0 best_formation = None for cand in clusters: overlap = len(sc_members & set(cand["customer_ids"])) if overlap > best_overlap: best_overlap = overlap best_cand_id = cand["cluster_id"] best_strong = cand["strong_link_count"] best_formation = cand.get("formation_type", "unknown") appears = best_overlap >= threshold correctly_weak = appears and best_strong == 0 and best_formation == "weak_only" if correctly_weak: benign_correctly_weak += 1 benign_detail.append({ "cluster_id": sc_id, "sc_size": len(sc_members), "appears_as_candidate": appears, "best_overlap": best_overlap, "best_cand_id": best_cand_id, "max_strong_in_match": best_strong, "formation_type": best_formation, "correctly_weak": correctly_weak, }) # Print print("\n========== Ground-Truth Validation ==========") print(f"Rings recovered (>=50% members in a candidate): {rings_recovered} / {len(rings)}") # Separation check separate_count = sum(1 for cnt in matched_by_cand.values() if cnt == 1) multi_count = sum(1 for cnt in matched_by_cand.values() if cnt > 1) print(f"Rings in separate candidates: {separate_count} (rings sharing a candidate: {rings_sharing_cluster})") for d in ring_detail: flag = "[OK]" if d["recovered"] else "[--]" share_flag = "" if d["matched_cluster"] and matched_by_cand.get(d["matched_cluster"], 0) > 1: share_flag = " [SHARES WITH OTHER RING]" print(f" {flag} {d['ring_id']} (size={d['ring_size']}) " f"best_overlap={d['best_overlap']} " f"candidate={d['matched_cluster']}{share_flag}") below_filter = [d for d in benign_detail if not d["appears_as_candidate"]] appeared = [d for d in benign_detail if d["appears_as_candidate"]] print(f"\nBenign/stress clusters (size>=3) appearing as candidates: {len(appeared)} / {len(benign_clusters)}") print(f" Correctly as weak_only with strong_link_count=0: {benign_correctly_weak} / {len(appeared)}") for d in benign_detail: if not d["appears_as_candidate"]: print(f" [SKIP] {d['cluster_id']} (size={d['sc_size']}) -- below size>=3 filter or no match") else: flag = "[OK]" if d["correctly_weak"] else "[FP]" print(f" {flag} {d['cluster_id']} (size={d['sc_size']}) " f"formation={d['formation_type']} " f"strong={d['max_strong_in_match']} " f"cand={d['best_cand_id']}") print("=============================================\n") return rings_recovered, ring_detail, benign_detail, matched_by_cand # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main(data_dir: str = None): if data_dir is None: data_dir = os.path.join(BASE_DIR, "data", "synthetic") graph_path = os.path.join(data_dir, "graph.gpickle") clusters_path = os.path.join(data_dir, "candidate_clusters.json") gt_path = os.path.join(data_dir, "ground_truth.json") print(f"Loading graph from {graph_path} ...") G = load_graph(graph_path) print(f" Graph: {G.number_of_nodes()} nodes, {G.number_of_edges()} edges") # 1.2 — two projections print("Building strong subgraph...") SG = build_strong_subgraph(G) strong_edges = SG.number_of_edges() print(f" Strong subgraph: {SG.number_of_nodes()} nodes, {strong_edges} edges") print("Building weak-only subgraph...") WG = build_weak_only_subgraph(G, SG) weak_edges = WG.number_of_edges() print(f" Weak-only subgraph: {WG.number_of_nodes()} nodes, {weak_edges} edges") # 1.3 — strong_anchored candidates print("Detecting strong_anchored communities...") strong_communities = detect_on_subgraph(SG, "strong_anchored") print(f" Raw strong communities (all sizes): {len(strong_communities)}") # 1.4 — weak_only candidates print("Detecting weak_only communities...") weak_communities = detect_on_subgraph(WG, "weak_only") print(f" Raw weak-only communities (all sizes): {len(weak_communities)}") # 1.5 — size >= 3 filter strong_filtered = [(m, method, ft) for m, method, ft in strong_communities if len(m) >= 3] weak_filtered = [(m, method, ft) for m, method, ft in weak_communities if len(m) >= 3] print(f" Strong candidates after size>=3: {len(strong_filtered)}") print(f" Weak-only candidates after size>=3: {len(weak_filtered)}") # 1.6 — combine and build output records all_filtered = strong_filtered + weak_filtered output_clusters = [] for idx, (members, method, formation_type) in enumerate(all_filtered, start=1): meta = compute_cluster_metadata(members, SG, WG) output_clusters.append({ "cluster_id": f"CAND{idx:02d}", "customer_ids": sorted(members), "size": len(members), "strong_link_count": meta["strong_link_count"], "weak_link_count": meta["weak_link_count"], "formation_type": formation_type, "method": method, }) # Write write_clusters(output_clusters, clusters_path) # Print overview print("\nCandidate clusters:") for c in output_clusters: print(f" {c['cluster_id']}: size={c['size']:3d} " f"strong={c['strong_link_count']:3d} " f"weak={c['weak_link_count']:3d} " f"formation={c['formation_type']} " f"method={c['method']}") # Validation if os.path.exists(gt_path): rings_recovered, ring_detail, benign_detail, matched_by_cand = validate( output_clusters, gt_path ) total_rings = len(ring_detail) separate = sum(1 for cnt in matched_by_cand.values() if cnt == 1) print(f"SUMMARY: Rings recovered={rings_recovered}/{total_rings} " f"Rings in separate candidates={separate}/{total_rings} " f"Benign/stress clusters=see above") else: print(f"Ground truth not found at {gt_path}, skipping validation.") return output_clusters if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument( "--data-dir", default=None, help="Path to data directory containing graph.gpickle and ground_truth.json. " "Defaults to data/synthetic/", ) args = parser.parse_args() main(args.data_dir)