Real-world applications of graph algorithms including social networks, routing, dependency resolution, recommendation systems, and network analysis
Last Updated: 2025-10-25
Activate when:
Prerequisites: graph-theory-fundamentals.md, graph-data-structures.md, graph-traversal-algorithms.md, shortest-path-algorithms.md
Related Skills: network-flow-algorithms.md, minimum-spanning-tree.md, advanced-graph-algorithms.md
Graph Modeling:
from typing import List, Set, Dict, Tuple
from collections import defaultdict, deque
import heapq
class SocialNetwork:
"""Social network graph with user relationships"""
def __init__(self):
# Directed graph: user_id -> [friend_ids]
self.friends: Dict[int, Set[int]] = defaultdict(set)
# Edge weights: interaction scores
self.interaction_scores: Dict[Tuple[int, int], float] = {}
def add_user(self, user_id: int):
"""Add user to network"""
if user_id not in self.friends:
self.friends[user_id] = set()
def add_friendship(self, user1: int, user2: int,
bidirectional: bool = True):
"""Add friendship (optionally bidirectional for mutual friends)"""
self.friends[user1].add(user2)
if bidirectional:
self.friends[user2].add(user1)
def set_interaction_score(self, user1: int, user2: int, score: float):
"""Set interaction strength between users"""
self.interaction_scores[(user1, user2)] = score
def mutual_friends(self, user1: int, user2: int) -> Set[int]:
"""Find mutual friends between two users"""
return self.friends[user1] & self.friends[user2]
def friend_suggestions(self, user: int, limit: int = 10) -> List[int]:
"""
Suggest friends based on mutual connections
Algorithm: Friends-of-friends with mutual friend count
"""
suggestions = defaultdict(int)
# Count mutual friends for each friend-of-friend
for friend in self.friends[user]:
for fof in self.friends[friend]:
if fof != user and fof not in self.friends[user]:
suggestions[fof] += 1
# Sort by mutual friend count
ranked = sorted(suggestions.items(),
key=lambda x: x[1],
reverse=True)
return [user_id for user_id, _ in ranked[:limit]]
def influence_score(self, user: int) -> float:
"""
Calculate user influence using PageRank-like algorithm
Influence = weighted sum of follower influences
"""
# Simplified PageRank
scores = {u: 1.0 for u in self.friends}
damping = 0.85
iterations = 10
for _ in range(iterations):
new_scores = {}
for u in self.friends:
# Base score
score = (1 - damping)
# Add contribution from followers
for follower in self.friends:
if u in self.friends[follower]:
# Follower points to u
out_degree = len(self.friends[follower])
if out_degree > 0:
score += damping * scores[follower] / out_degree
new_scores[u] = score
scores = new_scores
return scores.get(user, 0.0)
def community_detection_louvain(self) -> Dict[int, int]:
"""
Detect communities using Louvain method
Returns: Dict mapping user_id to community_id
"""
# Simplified community detection
# Full Louvain algorithm is complex - this is basic version
communities = {user: user for user in self.friends}
improved = True
while improved:
improved = False
for user in self.friends:
# Count connections to each community
community_connections = defaultdict(int)
for neighbor in self.friends[user]:
community = communities[neighbor]
community_connections[community] += 1
# Move to community with most connections
if community_connections:
best_community = max(community_connections.items(),
key=lambda x: x[1])[0]
if best_community != communities[user]:
communities[user] = best_community
improved = True
return communities
# Example usage
social = SocialNetwork()
# Add users and friendships
for i in range(10):
social.add_user(i)
social.add_friendship(0, 1)
social.add_friendship(0, 2)
social.add_friendship(1, 3)
social.add_friendship(2, 3)
social.add_friendship(3, 4)
social.add_friendship(4, 5)
social.add_friendship(5, 6)
social.add_friendship(6, 7)
suggestions = social.friend_suggestions(0, limit=3)
print(f"Friend suggestions for user 0: {suggestions}")
communities = social.community_detection_louvain()
print(f"Communities: {communities}")Graph Modeling:
import math
from dataclasses import dataclass
from typing import Optional
@dataclass
class Location:
"""Geographic location"""
id: int
name: str
lat: float # Latitude
lon: float # Longitude
class RouteNavigationSystem:
"""Route planning using graph algorithms"""
def __init__(self):
self.locations: Dict[int, Location] = {}
# Road network: (loc1, loc2) -> (distance, time)
self.roads: Dict[Tuple[int, int], Tuple[float, float]] = {}
def add_location(self, location: Location):
"""Add location to map"""
self.locations[location.id] = location
def add_road(self, loc1: int, loc2: int,
distance_km: float, time_min: float,
bidirectional: bool = True):
"""Add road between locations"""
self.roads[(loc1, loc2)] = (distance_km, time_min)
if bidirectional:
self.roads[(loc2, loc1)] = (distance_km, time_min)
def haversine_distance(self, loc1_id: int, loc2_id: int) -> float:
"""
Calculate great-circle distance between two locations
Used as heuristic for A* algorithm
"""
loc1 = self.locations[loc1_id]
loc2 = self.locations[loc2_id]
# Convert to radians
lat1, lon1 = math.radians(loc1.lat), math.radians(loc1.lon)
lat2, lon2 = math.radians(loc2.lat), math.radians(loc2.lon)
# Haversine formula
dlat = lat2 - lat1
dlon = lon2 - lon1
a = (math.sin(dlat / 2) ** 2 +
math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2)
c = 2 * math.asin(math.sqrt(a))
# Earth radius in km
r = 6371
return r * c
def find_shortest_route(self, start: int, end: int,
optimize_for: str = "distance"
) -> Tuple[List[int], float]:
"""
Find shortest route using A* with geographic heuristic
optimize_for: "distance" or "time"
"""
# A* algorithm with haversine heuristic
g_score = {start: 0.0}
f_score = {start: self.haversine_distance(start, end)}
parents = {start: None}
pq = [(f_score[start], start)]
closed = set()
while pq:
_, current = heapq.heappop(pq)
if current == end:
# Reconstruct path
path = []
node = end
while node is not None:
path.append(node)
node = parents[node]
return list(reversed(path)), g_score[end]
if current in closed:
continue
closed.add(current)
# Explore neighbors
for (u, v), (dist, time) in self.roads.items():
if u != current:
continue
if v in closed:
continue
# Choose cost metric
cost = dist if optimize_for == "distance" else time
tentative_g = g_score[current] + cost
if v not in g_score or tentative_g < g_score[v]:
g_score[v] = tentative_g
h = self.haversine_distance(v, end)
f_score[v] = tentative_g + h
parents[v] = current
heapq.heappush(pq, (f_score[v], v))
return [], math.inf # No route found
def alternative_routes(self, start: int, end: int,
num_alternatives: int = 3
) -> List[Tuple[List[int], float]]:
"""
Find multiple alternative routes using edge penalization
Algorithm: Find shortest path, increase weights on used edges, repeat
"""
routes = []
penalty_factor = 1.5
# Track edge usage penalties
penalties: Dict[Tuple[int, int], float] = defaultdict(lambda: 1.0)
for _ in range(num_alternatives):
# Temporarily apply penalties to roads
original_roads = self.roads.copy()
for (u, v), (dist, time) in list(self.roads.items()):
penalty = penalties[(u, v)]
self.roads[(u, v)] = (dist * penalty, time * penalty)
# Find route with current penalties
route, cost = self.find_shortest_route(start, end)
if not route:
break
routes.append((route, cost))
# Increase penalties on used edges
for i in range(len(route) - 1):
u, v = route[i], route[i + 1]
penalties[(u, v)] *= penalty_factor
# Restore original roads
self.roads = original_roads
return routes
# Example usage
nav = RouteNavigationSystem()
# Add locations
nav.add_location(Location(0, "Downtown", 37.7749, -122.4194))
nav.add_location(Location(1, "Airport", 37.6213, -122.3790))
nav.add_location(Location(2, "Suburbs", 37.8272, -122.2913))
nav.add_location(Location(3, "Harbor", 37.7955, -122.3934))
# Add roads
nav.add_road(0, 1, distance_km=20, time_min=25)
nav.add_road(0, 2, distance_km=15, time_min=20)
nav.add_road(1, 3, distance_km=10, time_min=15)
nav.add_road(2, 3, distance_km=12, time_min=18)
route, distance = nav.find_shortest_route(0, 3, optimize_for="distance")
print(f"Shortest route: {route}, distance: {distance:.2f} km")Graph Modeling:
class DependencyResolver:
"""Build system dependency resolution"""
def __init__(self):
# DAG: task -> [dependencies]
self.dependencies: Dict[str, Set[str]] = defaultdict(set)
self.build_order: List[str] = []
def add_task(self, task: str, dependencies: List[str] = None):
"""Add task with dependencies"""
if dependencies is None:
dependencies = []
self.dependencies[task] = set(dependencies)
def topological_sort(self) -> List[str]:
"""
Compute build order using topological sort
Returns: List of tasks in build order
Raises: ValueError if circular dependency detected
"""
# Kahn's algorithm
in_degree = defaultdict(int)
tasks = set(self.dependencies.keys())
# Add all tasks that are dependencies
for deps in self.dependencies.values():
tasks.update(deps)
# Compute in-degrees
for task in tasks:
in_degree[task] = 0
for deps in self.dependencies.values():
for dep in deps:
in_degree[dep] += 1
# Queue of tasks with no dependencies
queue = deque([task for task in tasks if in_degree[task] == 0])
build_order = []
while queue:
task = queue.popleft()
build_order.append(task)
# Process dependents
for dependent in tasks:
if task in self.dependencies[dependent]:
in_degree[dependent] -= 1
if in_degree[dependent] == 0:
queue.append(dependent)
if len(build_order) != len(tasks):
raise ValueError("Circular dependency detected")
return build_order
def parallel_build_levels(self) -> List[List[str]]:
"""
Compute parallelizable build levels
Returns: List of lists, where each inner list can be built in parallel
"""
in_degree = defaultdict(int)
tasks = set(self.dependencies.keys())
for deps in self.dependencies.values():
tasks.update(deps)
for task in tasks:
in_degree[task] = 0
for deps in self.dependencies.values():
for dep in deps:
in_degree[dep] += 1
levels = []
current_level = [task for task in tasks if in_degree[task] == 0]
while current_level:
levels.append(current_level)
next_level = []
for task in current_level:
# Reduce in-degree of dependents
for dependent in tasks:
if task in self.dependencies[dependent]:
in_degree[dependent] -= 1
if in_degree[dependent] == 0:
next_level.append(dependent)
current_level = next_level
return levels
# Example: Build system
build_system = DependencyResolver()
build_system.add_task("install_deps", [])
build_system.add_task("compile_core", ["install_deps"])
build_system.add_task("compile_utils", ["install_deps"])
build_system.add_task("compile_app", ["compile_core", "compile_utils"])
build_system.add_task("run_tests", ["compile_app"])
build_system.add_task("build_docs", ["compile_app"])
build_system.add_task("package", ["run_tests", "build_docs"])
build_order = build_system.topological_sort()
print(f"Build order: {build_order}")
parallel_levels = build_system.parallel_build_levels()
print(f"Parallel build levels: {parallel_levels}")Graph Modeling:
class CollaborativeFilteringGraph:
"""Recommendation system using graph-based collaborative filtering"""
def __init__(self):
# Bipartite graph: users and items
self.user_items: Dict[int, Set[int]] = defaultdict(set)
self.item_users: Dict[int, Set[int]] = defaultdict(set)
# Ratings: (user, item) -> rating
self.ratings: Dict[Tuple[int, int], float] = {}
def add_rating(self, user: int, item: int, rating: float):
"""Add user rating for item"""
self.user_items[user].add(item)
self.item_users[item].add(user)
self.ratings[(user, item)] = rating
def item_similarity(self, item1: int, item2: int) -> float:
"""
Compute similarity between items using Jaccard index
Similarity = |users who rated both| / |users who rated either|
"""
users1 = self.item_users[item1]
users2 = self.item_users[item2]
if not users1 or not users2:
return 0.0
intersection = len(users1 & users2)
union = len(users1 | users2)
return intersection / union if union > 0 else 0.0
def recommend_items(self, user: int, limit: int = 10) -> List[int]:
"""
Recommend items using collaborative filtering
Algorithm: Find similar users, recommend their highly-rated items
"""
# Find users with similar taste (Jaccard similarity)
similar_users = []
user_items = self.user_items[user]
for other_user in self.user_items:
if other_user == user:
continue
other_items = self.user_items[other_user]
if not other_items:
continue
# Compute Jaccard similarity
intersection = len(user_items & other_items)
union = len(user_items | other_items)
similarity = intersection / union if union > 0 else 0.0
if similarity > 0:
similar_users.append((other_user, similarity))
# Sort by similarity
similar_users.sort(key=lambda x: x[1], reverse=True)
# Collect items from similar users (weighted by similarity)
item_scores = defaultdict(float)
for other_user, similarity in similar_users[:20]: # Top 20 similar users
for item in self.user_items[other_user]:
if item not in user_items: # Not already rated
rating = self.ratings.get((other_user, item), 0.0)
item_scores[item] += similarity * rating
# Sort by score
ranked_items = sorted(item_scores.items(),
key=lambda x: x[1],
reverse=True)
return [item for item, _ in ranked_items[:limit]]
# Example usage
recommender = CollaborativeFilteringGraph()
# Add ratings (user, item, rating)
recommender.add_rating(0, 100, 5.0) # User 0 rates item 100: 5 stars
recommender.add_rating(0, 101, 4.0)
recommender.add_rating(1, 100, 5.0)
recommender.add_rating(1, 102, 3.0)
recommender.add_rating(2, 101, 4.0)
recommender.add_rating(2, 102, 5.0)
recommendations = recommender.recommend_items(0, limit=5)
print(f"Recommendations for user 0: {recommendations}")| Application | Graph Type | Key Algorithms | |-------------|-----------|----------------| | Social Networks | Directed, weighted | PageRank, community detection, BFS | | Route Planning | Weighted, geographic | A*, Dijkstra, alternative paths | | Dependency Resolution | DAG | Topological sort, parallel levels | | Recommendations | Bipartite | Collaborative filtering, similarity | | Network Design | Undirected, weighted | MST, Steiner tree | | Fraud Detection | Directed, temporal | Pattern matching, anomaly detection |
Foundations:
graph-theory-fundamentals.md → Basic graph conceptsgraph-data-structures.md → Efficient representationsgraph-traversal-algorithms.md → BFS, DFS, topological sortshortest-path-algorithms.md → Dijkstra, A*, Floyd-Warshallminimum-spanning-tree.md → Kruskal, Primnetwork-flow-algorithms.md → Max flow, bipartite matchingadvanced-graph-algorithms.md → Coloring, matching, cliquesGraph algorithms power diverse real-world applications:
Key takeaways:
Graph theory provides fundamental algorithms for modern systems and applications.