Similarity Algorithms in Neo4j

from graphdatascience.session import DbmsConnectionInfo, SessionMemory

db_connection = DbmsConnectionInfo(
    uri=os.environ["NEO4J_URI"],
    username=os.environ["NEO4J_USERNAME"],
    password=os.environ["NEO4J_PASSWORD"],
)

gds = sessions.get_or_create(
    session_name="Similarity",
    memory=SessionMemory.m_2GB,
    db_connection=db_connection,
)

gds.verify_connectivity()

import csv
from decimal import Decimal
from itertools import islice
from pathlib import Path

# dataset paths
DATA_DIR = Path("act-mooc_data/act-mooc")
DATASET_ID = "act-mooc"

# Schema constraints to ensure data integrity
CONSTRAINTS = [
    "CREATE CONSTRAINT mooc_user_identity IF NOT EXISTS FOR (u:MoocUser) REQUIRE (u.datasetId, u.userId) IS UNIQUE",
    "CREATE CONSTRAINT mooc_activity_identity IF NOT EXISTS FOR (t:MoocActivity) REQUIRE (t.datasetId, t.targetId) IS UNIQUE",
    "CREATE CONSTRAINT mooc_action_identity IF NOT EXISTS FOR (a:MoocAction) REQUIRE (a.datasetId, a.actionId) IS UNIQUE",
    "CREATE RANGE INDEX mooc_action_time IF NOT EXISTS FOR (a:MoocAction) ON (a.datasetId, a.timestampSeconds)",
]

def get_action_rows(limit=150000):
    """Parses the 3 TSV files into a unified dictionary format."""
    a_path = DATA_DIR / "mooc_actions.tsv"
    f_path = DATA_DIR / "mooc_action_features.tsv"
    l_path = DATA_DIR / "mooc_action_labels.tsv"
    with open(a_path) as ah, open(f_path) as fh, open(l_path) as lh:
        zipped = zip(
            csv.DictReader(ah, delimiter="\t"),
            csv.DictReader(fh, delimiter="\t"),
            csv.DictReader(lh, delimiter="\t"),
        )
        for i, (a_row, f_row, l_row) in enumerate(islice(zipped, limit)):
            yield {
                "actionId": i,
                "userId": int(a_row["USERID"]),
                "targetId": int(a_row["TARGETID"]),
                "timestampSeconds": int(Decimal(a_row["TIMESTAMP"])),
                "featureVector": [float(f_row[f"FEATURE{j}"]) for j in range(4)],
                "dropoutAfterAction": bool(int(l_row["LABEL"])),
                "sourceLabelActionId": int(l_row["ACTIONID"]),
            }


def chunked(rows, size=5000):
    batch = []
    for r in rows:
        batch.append(r)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch

IMPORT_CYPHER = """
UNWIND $rows AS row
MERGE (u:MoocUser {datasetId: $datasetId, userId: row.userId})
MERGE (t:MoocActivity {datasetId: $datasetId, targetId: row.targetId})
MERGE (a:MoocAction {datasetId: $datasetId, actionId: row.actionId})
SET a.timestampSeconds = row.timestampSeconds,
    a.featureVector = row.featureVector,
    a.dropoutAfterAction = row.dropoutAfterAction
MERGE (u)-[:PERFORMED]->(a)
MERGE (a)-[:ON_ACTIVITY]->(t)
"""

print("Applying schema constraints...")
for cmd in CONSTRAINTS:
    gds.run_cypher(cmd)

print("Starting batch import...")
total = 0
for batch in chunked(get_action_rows()):
    gds.run_cypher(IMPORT_CYPHER, params={"datasetId": DATASET_ID, "rows": batch})
    total += len(batch)
    if total % 25000 == 0:
        print(f"Imported {total} actions...")

print(f"\nSuccess! Total actions imported: {total}")

# Make sure the imported data is fully propagated across the cluster
# before the remote projections that read it
gds.set_bookmarks(gds.last_bookmarks())

REMOTE_PROJECT_CYPHER = """
MATCH (u:MoocUser {datasetId: $datasetId})-[:PERFORMED]->(a:MoocAction)-[:ON_ACTIVITY]->(t:MoocActivity {datasetId: $datasetId})
RETURN gds.graph.project.remote(
  u,
  t,
  {
    sourceNodeLabels: labels(u),
    targetNodeLabels: labels(t),
    relationshipType: 'ACTED_ON'
  }
)
""".strip()

graph_name = "user-activity-similarity"

g_sim, project_result = gds.graph.project.cypher(
    graph_name,
    REMOTE_PROJECT_CYPHER,
    query_parameters={"datasetId": "act-mooc"},
    overwrite=True,
)

print(f"Graph '{graph_name}' projected successfully.")
display(project_result)

plt.figure(figsize=(10, 6))
plt.hist(similarity_results["jaccard_score"], bins=30, color="skyblue", edgecolor="black")
plt.title("Distribution of Jaccard Similarity Scores between Users")
plt.xlabel("Jaccard Similarity Score")
plt.ylabel("Frequency")
plt.grid(axis="y", linestyle="--", alpha=0.7)
plt.show()

mutate_results = gds.node_similarity.mutate(
    g_sim,
    mutate_relationship_type="SIMILAR_TO",
    mutate_property="score",
    top_k=5,
    similarity_cutoff=0.5,
)

print("Mutate Mode Results (Relationships added to in-memory graph):")
display(mutate_results)

print(f"Nodes compared:        {mutate_results.nodes_compared}")
print(f"Relationships created: {mutate_results.relationships_written}")
print(f"Compute time (ms):     {mutate_results.compute_millis}")
print("\nSimilarity distribution:")
display(pd.Series(mutate_results.similarity_distribution))

write_results = gds.graph.relationships.write(
    g_sim,
    "SIMILAR_TO",
    ["score"],
)

print("Write Mode Results (Persisted mutated relationships back to Neo4j):")
display(write_results)

g_sim, project_result = gds.graph.project.cypher(
    "user-activity-similarity",
    REMOTE_PROJECT_CYPHER,
    query_parameters={"datasetId": "act-mooc"},
    overwrite=True,
)

jaccard_df = gds.node_similarity.stream(
    g_sim,
    similarity_metric="JACCARD",
    top_k=5,
    similarity_cutoff=0.1,
)

print("Top Pairs using Jaccard Similarity (Default):")
display(jaccard_df.head(5))

g_sim, project_result = gds.graph.project.cypher(
    "user-activity-similarity",
    REMOTE_PROJECT_CYPHER,
    query_parameters={"datasetId": "act-mooc"},
    overwrite=True,
)

overlap_df = gds.node_similarity.stream(
    g_sim,
    similarity_metric="OVERLAP",
    top_k=5,
    similarity_cutoff=0.1,
)

print("Top Pairs using Overlap Coefficient:")
display(overlap_df.head(5))

weighted_project_query = """
MATCH (u:MoocUser {datasetId: $datasetId})-[:PERFORMED]->(a:MoocAction)-[:ON_ACTIVITY]->(t:MoocActivity {datasetId: $datasetId})
WITH u, t, count(a) as interaction_count
RETURN gds.graph.project.remote(
  u,
  t,
  {
    sourceNodeLabels: labels(u),
    targetNodeLabels: labels(t),
    relationshipType: 'ACTED_ON',
    relationshipProperties: { weight: toFloat(interaction_count) }
  }
)
""".strip()

g_weighted, _ = gds.graph.project.cypher(
    "weighted-sim",
    weighted_project_query,
    query_parameters={"datasetId": "act-mooc"},
    overwrite=True,
)

cosine_df = gds.node_similarity.stream(
    g_weighted,
    similarity_metric="COSINE",
    relationship_weight_property="weight",
    top_k=5,
    similarity_cutoff=0.1,
)

print("Top Pairs using Cosine Similarity (Weighted):")
display(cosine_df.head(5))

similarity_comp_query = """
MATCH (u:MoocUser {datasetId: 'act-mooc'})
WHERE u.userId IN [0, 1]
MATCH (u)-[:PERFORMED]->(a:MoocAction)
WITH u.userId AS userId,
     [avg(a.featureVector[0]), avg(a.featureVector[1]), avg(a.featureVector[2]), avg(a.featureVector[3])] AS vector
WITH max(CASE WHEN userId = 0 THEN vector END) AS v1,
     max(CASE WHEN userId = 1 THEN vector END) AS v2
RETURN
    gds.similarity.jaccard(v1, v2) AS jaccard,
    gds.similarity.overlap(v1, v2) AS overlap,
    gds.similarity.cosine(v1, v2) AS cosine,
    gds.similarity.pearson(v1, v2) AS pearson
"""

sim_comparison = gds.run_cypher(similarity_comp_query)
display(sim_comparison)

similarity_comp_query_2 = """
MATCH (u:MoocUser {datasetId: 'act-mooc'})
WHERE u.userId IN [2, 3]
MATCH (u)-[:PERFORMED]->(a:MoocAction)
WITH u.userId AS userId,
     [avg(a.featureVector[0]), avg(a.featureVector[1]), avg(a.featureVector[2]), avg(a.featureVector[3])] AS vector
WITH max(CASE WHEN userId = 2 THEN vector END) AS v1,
     max(CASE WHEN userId = 3 THEN vector END) AS v2
RETURN
    gds.similarity.jaccard(v1, v2) AS jaccard,
    gds.similarity.overlap(v1, v2) AS overlap,
    gds.similarity.cosine(v1, v2) AS cosine,
    gds.similarity.pearson(v1, v2) AS pearson
"""

sim_comparison_2 = gds.run_cypher(similarity_comp_query_2)
display(sim_comparison_2)

print("Calculating average feature vectors for users...")

gds.run_cypher("""
MATCH (u:MoocUser {datasetId: 'act-mooc'})-[:PERFORMED]->(a:MoocAction)
WITH u,
     avg(a.featureVector[0]) as f1,
     avg(a.featureVector[1]) as f2,
     avg(a.featureVector[2]) as f3,
     avg(a.featureVector[3]) as f4
SET u.featureVector = [f1, f2, f3, f4]
""")

# Make sure the `featureVector` writes are fully propagated across the cluster
# before the remote KNN projection that reads them
gds.set_bookmarks(gds.last_bookmarks())

knn_graph_name = "user-features-knn"

REMOTE_KNN_PROJECT = """
MATCH (u:MoocUser {datasetId: $datasetId})
RETURN gds.graph.project.remote(
  u,
  null,
  {
    sourceNodeLabels: labels(u),
    sourceNodeProperties: { featureVector: u.featureVector }
  }
)
""".strip()

g_knn, _ = gds.graph.project.cypher(
    knn_graph_name,
    REMOTE_KNN_PROJECT,
    query_parameters={"datasetId": "act-mooc"},
    overwrite=True,
)

print("Running KNN algorithm...")

knn_results = gds.knn.stream(
    g_knn,
    node_properties=["featureVector"],
    top_k=3,
    sample_rate=1.0,
    random_seed=42,
)

knn_results = knn_results.rename(columns={"node1": "user_a", "node2": "user_b", "similarity": "score"})

print("Top KNN Similarity results (Feature-based similarity):")
display(knn_results.sort_values("score", ascending=False).head(10))

top_5_knn = knn_results.sort_values("score", ascending=False).head(5)

print("Top 5 Most Similar User Pairs (KNN):")
for index, row in top_5_knn.iterrows():
    print(f"- User {int(row['user_a'])} and User {int(row['user_b'])} with a similarity score of {row['score']:.4f}")

display(top_5_knn)

# Prepare Jaccard results for comparison
jaccard_compare = jaccard_df.rename(columns={"node1": "user_a", "node2": "user_b", "similarity": "jaccard_score"})

# Convert IDs to integer to ensure matching types across datasets
jaccard_compare["user_a"] = jaccard_compare["user_a"].astype(int)
jaccard_compare["user_b"] = jaccard_compare["user_b"].astype(int)
knn_results["user_a"] = knn_results["user_a"].astype(int)
knn_results["user_b"] = knn_results["user_b"].astype(int)

# Inner merge on matching user pairs
comparison_df = pd.merge(knn_results, jaccard_compare, on=["user_a", "user_b"], how="inner")

print(f"Found {len(comparison_df)} pairs that appear in both similarity results.")
display(comparison_df.sort_values(by=["score", "jaccard_score"], ascending=False).head(10))

# 1. Delete all relationships first, in batches
while True:
    deleted = gds.run_cypher(
        """
        MATCH ()-[r]->()
        WITH r LIMIT $batch_size
        DELETE r
        RETURN count(*) AS deleted
        """,
        params={"batch_size": 10000},
    )["deleted"].item()
    if deleted == 0:
        break

# 2. Delete all nodes in batches
while True:
    deleted = gds.run_cypher(
        """
        MATCH (n)
        WITH n LIMIT $batch_size
        DETACH DELETE n
        RETURN count(*) AS deleted
        """,
        params={"batch_size": 5000},
    )["deleted"].item()
    if deleted == 0:
        break

gds.delete()