TriRNSC Algorithm for Gene Triclustering
TriRNSC Algorithm for Gene Triclustering
import json
import pandas as pd
import networkx as nx
import [Link] as plt
import seaborn as sns
from [Link] import pearsonr
from mpl_toolkits.mplot3d import Axes3D
from [Link] import Normalize
import [Link] as cm
from collections import defaultdict
import random
from [Link] import PCA
import gseapy as gp
import os
class TriRNSC:
"""
Implementation of the TriRNSC algorithm for triclustering gene expression data.
Parameters:
-----------
threshold : float
Correlation threshold for constructing the gene co-expression network.
tabu_length_ratio : float
Ratio of the number of vertices to determine the tabu list length.
naive_stopping_tolerance : int
Number of moves without improvement before switching to scaled cost.
scaled_stopping_tolerance : int
Number of moves without improvement before terminating an experiment.
diversification_frequency : int
Period for shuffling diversification.
diversification_length : int
Number of random moves in a diversification phase.
max_experiments : int
Maximum number of experiments to run.
"""
[Link] = threshold
self.tabu_length_ratio = tabu_length_ratio
self.naive_stopping_tolerance = naive_stopping_tolerance
self.scaled_stopping_tolerance = scaled_stopping_tolerance
self.diversification_frequency = diversification_frequency
self.diversification_length = diversification_length
self.max_experiments = max_experiments
[Link] = None
self.gene_names = None
self.condition_names = None
self.time_points = None
[Link] = None
self.adjacency_list = None
self.best_triclusters = None
self.correlation_matrix = None
Parameters:
-----------
data_matrix : [Link]
3D array with dimensions (genes, conditions, time_points).
gene_names : list, optional
List of gene names.
condition_names : list, optional
List of condition names.
time_points : list, optional
List of time points.
"""
[Link] = data_matrix
def construct_gene_co_expression_network(self):
"""
Construct a gene co-expression network (GCN) based on Pearson correlation.
Returns:
--------
[Link]
The gene co-expression network.
"""
G = [Link]()
G.add_nodes_from(range(n_genes))
edges_added = []
for i in range(n_genes):
for j in range(i + 1, n_genes):
gene_i_profile = [Link][i].reshape(-1)
gene_j_profile = [Link][j].reshape(-1)
correlation_matrix[i, j] = corr
correlation_matrix[j, i] = corr
[Link] = G
self.correlation_matrix = correlation_matrix
self.adjacency_list = {node: list([Link](node)) for node in [Link]()}
[Link]['expression_data'] = [Link]
if G.number_of_edges() == 0:
print(f"WARNING: No edges in the network. The correlation threshold
({[Link]}) may be too high.")
print("Try lowering the threshold or check your data.")
Parameters:
-----------
n_clusters : int, optional
Number of clusters to initialize. If None, a random number is chosen.
Returns:
--------
dict
A dictionary mapping cluster IDs to sets of nodes.
"""
nodes = list([Link]())
if n_clusters is None:
n_clusters = [Link](2, int([Link](len(nodes))))
Parameters:
-----------
clusters : dict
A dictionary mapping cluster IDs to sets of nodes.
Returns:
--------
float
The naive cost value.
"""
cost = 0
cluster = clusters[cluster_id]
return cost / 2
Parameters:
-----------
clusters : dict
A dictionary mapping cluster IDs to sets of nodes.
Returns:
--------
float
The scaled cost value.
"""
n = len([Link]())
cost = 0
neighborhood = set(self.adjacency_list[node]).union(cluster)
if not neighborhood:
continue
return cost
Parameters:
-----------
node : int
The node to find.
clusters : dict
A dictionary mapping cluster IDs to sets of nodes.
Returns:
--------
int or None
The cluster ID or None if the node is not in any cluster.
"""
Parameters:
-----------
node : int
The node to move.
from_cluster_id : int
The source cluster ID.
to_cluster_id : int
The destination cluster ID.
clusters : dict
A dictionary mapping cluster IDs to sets of nodes.
Returns:
--------
dict
The updated clusters.
"""
new_clusters[from_cluster_id].remove(node)
new_clusters[to_cluster_id].add(node)
return new_clusters
Parameters:
-----------
initial_clusters : dict
Initial clustering.
Returns:
--------
dict
The best clusters found during the naive phase.
"""
current_clusters = initial_clusters
best_clusters = current_clusters
best_cost = self.naive_cost(current_clusters)
no_improvement_count = 0
iteration = 0
if iteration % self.diversification_frequency == 0:
for _ in range(self.diversification_length):
node = [Link](list([Link]()))
node = [Link](list([Link]()))
possible_clusters = list(current_clusters.keys())
possible_clusters = [cid for cid in possible_clusters if cid !=
from_cluster_id]
if not possible_clusters:
continue # no other cluster to move to
to_cluster_id = [Link](possible_clusters)
current_clusters = new_clusters
tabu_list.append((node, to_cluster_id))
if len(tabu_list) > tabu_length:
tabu_list.pop(0)
if iteration % 1000 == 0:
print(f"Naive phase iteration {iteration}, current cost:
{new_cost:.4f}, best cost: {best_cost:.4f}")
Parameters:
-----------
initial_clusters : dict
Initial clustering from the naive phase.
Returns:
--------
dict
The best clusters found during the scaled phase.
"""
current_clusters = initial_clusters
best_clusters = current_clusters
best_cost = self.scaled_cost(current_clusters)
no_improvement_count = 0
iteration = 0
if iteration % self.diversification_frequency == 0:
for _ in range(self.diversification_length):
node = [Link](list([Link]()))
node = [Link](list([Link]()))
possible_clusters = list(current_clusters.keys())
possible_clusters = [cid for cid in possible_clusters if cid !=
from_cluster_id]
if not possible_clusters:
continue # no other cluster to move to
to_cluster_id = [Link](possible_clusters)
if (node, to_cluster_id) in tabu_list:
continue
current_clusters = new_clusters
tabu_list.append((node, to_cluster_id))
if len(tabu_list) > tabu_length:
tabu_list.pop(0)
if iteration % 1000 == 0:
print(f"Scaled phase iteration {iteration}, current cost:
{new_cost:.4f}, best cost: {best_cost:.4f}")
def run_rnsc(self):
"""
Run the RNSC algorithm to find triclusters.
Returns:
--------
dict
The best triclusters found.
"""
if [Link] is None:
self.construct_gene_co_expression_network()
best_overall_clusters = None
best_overall_cost = float('inf')
random_clusters = self.initialize_random_clusters(n_clusters=2)
naive_clusters = self.run_naive_phase(random_clusters)
scaled_clusters = self.run_scaled_phase(naive_clusters)
current_cost = self.scaled_cost(scaled_clusters)
self.best_triclusters = best_overall_clusters
return self.best_triclusters
def get_tricluster_info(self):
"""
Get information about the found triclusters.
Returns:
--------
list of dict
List of triclusters with gene names.
"""
if self.best_triclusters is None:
print("No triclusters found. Run the RNSC algorithm first.")
return None
tricluster_info = []
msr = self.calculate_msr(submatrix)
tricluster_info.append({
'id': cluster_id,
'genes': genes,
'gene_indices': list(nodes),
'size': len(nodes),
'msr': msr,
'volume': volume,
'tqi': msr / volume if volume > 0 else float('inf')
})
return tricluster_info
Parameters:
-----------
submatrix : [Link]
3D submatrix of gene expression values for a tricluster.
Returns:
--------
float
The MSR value.
"""
if [Link] == 0:
return float('inf')
mgct = [Link](submatrix)
r = 0
for i in range(n_genes):
for j in range(n_conditions):
for k in range(n_time_points):
r_ijk = (submatrix[i, j, k] + mgct - mg[i, 0, 0] - mc[0, j, 0]
- mt[0, 0, k])**2
r += r_ijk
return msr
Parameters:
-----------
cluster_labels : list or dict, optional
Cluster assignments for genes. Can be a list of labels or a dictionary
mapping node indices to cluster IDs.
"""
if self.correlation_matrix is None:
if [Link] is None:
self.construct_gene_co_expression_network()
else:
print("Correlation matrix not found. Recomputing...")
n_genes = [Link][0]
self.correlation_matrix = [Link]((n_genes, n_genes))
for i in range(n_genes):
for j in range(i+1, n_genes):
gene_i_profile = [Link][i].reshape(-1)
gene_j_profile = [Link][j].reshape(-1)
corr, _ = pearsonr(gene_i_profile, gene_j_profile)
self.correlation_matrix[i, j] = corr
self.correlation_matrix[j, i] = corr
[Link](figsize=(10, 8))
boundaries = []
last_cluster = cluster_labels[sorted_indices[0]]
for i, idx in enumerate(sorted_indices):
if cluster_labels[idx] != last_cluster:
[Link](i - 0.5)
last_cluster = cluster_labels[idx]
plt.tight_layout()
[Link]()
Parameters:
-----------
cluster_labels : list or dict, optional
Cluster assignments for genes. If None, uses best_triclusters.
condition_idx : int
Index of condition to display.
"""
if [Link] is None:
print("No data loaded. Load data first.")
return
if cluster_labels is None:
if self.best_triclusters is None:
print("No clusters available. Run RNSC first or provide cluster
labels.")
return
if isinstance(cluster_labels, dict):
labels = [Link]([Link][0], dtype=int)
for cluster_id, nodes in cluster_labels.items():
for node in nodes:
labels[node] = cluster_id
cluster_labels = labels
n_clusters = len([Link](cluster_labels))
time_points = range([Link][2])
[Link](figsize=(15, 4 * n_clusters))
if len(genes_in_cluster) > 0:
mean_expr = [Link]([Link][genes_in_cluster, condition_idx, :],
axis=0)
[Link](time_points, mean_expr, 'k-', linewidth=2.5,
label=f'Mean (n={len(genes_in_cluster)})')
plt.tight_layout()
[Link](f'Gene Expression Patterns by Cluster (Condition:
{self.condition_names[condition_idx]})',
fontsize=16, y=1.02)
[Link]()
Parameters:
-----------
cluster_labels : list or dict, optional
Cluster assignments for genes. If None, uses best_triclusters.
pos : dict, optional
Node positions. If None, spring layout is used.
node_size_factor : float
Scaling factor for node sizes
edge_alpha : float
Transparency for edges
"""
if [Link] is None:
self.construct_gene_co_expression_network()
if len([Link]()) == 0:
print("The graph has no edges to visualize. Try lowering the
correlation threshold.")
return
[Link](figsize=(12, 10))
if pos is None:
pos = nx.spring_layout([Link], seed=42)
if isinstance(cluster_labels, dict):
labels = [Link](len([Link]()), dtype=int)
for cluster_id, nodes in cluster_labels.items():
for node in nodes:
labels[node] = cluster_id
cluster_labels = labels
unique_clusters = [Link](cluster_labels)
for cluster in unique_clusters:
[Link]([0], [0], 'o', color=[Link].tab10(cluster / max(9,
max(unique_clusters))),
label=f'Cluster {cluster}', markersize=10, alpha=0)
[Link](loc='upper right')
else:
node_degrees = dict([Link]())
node_colors = [node_degrees[n] for n in [Link]()]
nx.draw_networkx_nodes([Link], pos, node_color=node_colors,
node_size=node_sizes, alpha=0.8,
cmap=[Link])
Parameters:
-----------
cluster_labels : list or dict, optional
Cluster assignments for genes. If None, uses best_triclusters.
edge_threshold : float, optional
Minimum edge weight to display. If None, uses [Link].
use_pca : bool
Whether to use PCA for node positioning.
"""
if [Link] is None:
self.construct_gene_co_expression_network()
if len([Link]()) == 0:
print("The graph has no edges to visualize. Try lowering the
correlation threshold.")
return
if edge_threshold is None:
edge_threshold = [Link]
pca = PCA(n_components=3)
positions = pca.fit_transform(flattened_data)
if isinstance(cluster_labels, dict):
labels = [Link](len([Link]()), dtype=int)
for cluster_id, nodes in cluster_labels.items():
for node in nodes:
labels[node] = cluster_id
cluster_labels = labels
scatter = [Link](
node_xyz[:, 0], node_xyz[:, 1], node_xyz[:, 2],
c=node_colors, cmap=color_map, norm=color_norm,
s=node_sizes, alpha=0.8, edgecolors='w', linewidth=0.5
)
edge_color = [Link](weight)
line_width = weight * 3
ax.set_xlabel('X', fontweight='bold')
ax.set_ylabel('Y', fontweight='bold')
ax.set_zlabel('Z', fontweight='bold')
ax.set_title('Gene Co-expression Network (3D)', fontsize=16)
if cluster_labels is not None:
unique_clusters = [Link](cluster_labels)
legend_elements = []
for cluster in unique_clusters:
color = color_map(color_norm(cluster))
legend_elements.append(
plt.Line2D([0], [0], marker='o', color='w',
markerfacecolor=color,
markersize=10, label=f'Cluster {cluster}')
)
[Link](handles=legend_elements, loc='upper right')
ax.view_init(elev=20, azim=30)
plt.tight_layout()
[Link]()
if self.best_triclusters is None:
print("No triclusters found. Run the RNSC algorithm first.")
return
[Link](save_dir, exist_ok=True)
try:
import gseapy as gp
except ImportError:
print("ERROR: gseapy package is not installed. Install it using: pip
install gseapy")
return
successful_clusters = 0
try:
organisms = ['Human', 'Yeast', 'Mouse']
success = False
if not success:
print(f"Could not perform GO enrichment for cluster
{cluster_id} with any organism")
with
open(f"{save_dir}/cluster_{cluster_id}_go_enrichment_failed.txt", 'w') as f:
[Link](f"GO enrichment failed for cluster {cluster_id}\n")
[Link](f"Genes in cluster: {', '.join(gene_list)}\n")
continue
# Save results
enr.res2d.to_csv(f"{save_dir}/cluster_{cluster_id}_go_enrichment.csv", index=False)
print(f"Saved GO enrichment results to CSV for cluster
{cluster_id}")
if not top_terms.empty:
[Link](figsize=(10, 6))
[Link](top_terms['Term'].[Link](0, 50) + '...',
-np.log10(top_terms['Adjusted P-value']),
color='skyblue')
[Link]('-log10(Adjusted P-value)')
[Link]('GO Terms')
[Link](f'Top GO Terms (Cluster {cluster_id})')
[Link]().invert_yaxis()
plt.tight_layout()
[Link](f"{save_dir}/cluster_{cluster_id}_go_plot.png")
[Link]()
print(f"Created GO plot for cluster {cluster_id}")
successful_clusters += 1
except Exception as e:
print(f"Error creating GO plot for cluster {cluster_id}:
{str(e)}")
except Exception as e:
print(f"GO Analysis failed for cluster {cluster_id}: {str(e)}")
with open(f"{save_dir}/cluster_{cluster_id}_go_error.log", 'w') as
f:
[Link](f"Error: {str(e)}\n")
[Link](f"Genes: {', '.join(gene_list)}\n")
[Link](save_dir, exist_ok=True)
successful_clusters = 0
try:
organisms = ['Human', 'Yeast', 'Mouse']
gene_set_options = ['KEGG_2021_Human', 'KEGG_2019_Human',
'KEGG_2019_Mouse']
success = False
if success:
break
if not success:
print(f"Could not perform KEGG enrichment for cluster
{cluster_id} with any organism/gene set")
with
open(f"{save_dir}/cluster_{cluster_id}_kegg_enrichment_failed.txt", 'w') as f:
[Link](f"KEGG enrichment failed for cluster {cluster_id}\
n")
[Link](f"Genes in cluster: {', '.join(gene_list)}\n")
continue
enr.res2d.to_csv(f"{save_dir}/cluster_{cluster_id}_kegg_enrichment.csv",
index=False)
print(f"Saved KEGG enrichment results to CSV for cluster
{cluster_id}")
try:
top_paths = enr.res2d.sort_values('Adjusted P-
value').head(top_n)
if not top_paths.empty:
[Link](figsize=(10, 6))
[Link](top_paths['Term'].[Link](0, 50) + '...',
-np.log10(top_paths['Adjusted P-value']),
color='lightgreen')
[Link]('-log10(Adjusted P-value)')
[Link]('KEGG Pathways')
[Link](f'Top KEGG Pathways (Cluster {cluster_id})')
[Link]().invert_yaxis()
plt.tight_layout()
[Link](f"{save_dir}/cluster_{cluster_id}_kegg_plot.png")
[Link]()
print(f"Created KEGG plot for cluster {cluster_id}")
successful_clusters += 1
except Exception as e:
print(f"Error creating KEGG plot for cluster {cluster_id}:
{str(e)}")
except Exception as e:
print(f"KEGG Analysis failed for cluster {cluster_id}: {str(e)}")
with open(f"{save_dir}/cluster_{cluster_id}_kegg_error.log", 'w')
as f:
[Link](f"Error: {str(e)}\n")
[Link](f"Genes: {', '.join(gene_list)}\n")
gene_names = df['Gene'].tolist()
print(f"Found {len(gene_names)} genes.")
if not [Link]:
print(f"First expression value (raw):
{df['Expression_values'].iloc[0][:100]}...")
data = []
if idx == 0:
print(f"Type of expression_values:
{type(expression_values)}")
if isinstance(expression_values, str):
try:
expression_values = [Link](expression_values)
if idx == 0:
print("Successfully parsed as JSON")
except [Link]:
try:
expression_values =
ast.literal_eval(expression_values)
if idx == 0:
print("Successfully parsed as Python literal")
except (SyntaxError, ValueError) as e:
print(f"Error parsing expression values for gene
{row['Gene']}: {str(e)}")
print(f"Raw value: {expression_values[:100]}...")
expression_values = []
[Link](expression_values)
if idx == 0:
print(f"Processed first gene. Data structure shape:
{[Link](expression_values).shape if expression_values else 'empty'}")
except Exception as e:
print(f"Error processing row {idx}: {str(e)}")
try:
data = [Link](data, dtype=float)
print(f"Converted data to numpy array with shape: {[Link]}")
except Exception as e:
print(f"Error converting to numpy array: {str(e)}")
print(f"Data structure: {[type(item) for item in data[:3]]}")
return None, None, None
try:
print("Clustering genes...")
true_clusters = cluster_genes(data, n_clusters)
print(f"Generated {len(true_clusters)} cluster labels")
except Exception as e:
print(f"Error during clustering: {str(e)}")
true_clusters = [i % n_clusters for i in range(len(gene_names))]
except Exception as e:
print(f"Unexpected error loading data from {filename}:")
traceback.print_exc()
return None, None, None
Parameters:
-----------
data : [Link]
Gene expression data with shape (n_genes, n_conditions, n_time_points)
n_clusters : int
Number of clusters to form
Returns:
--------
[Link]
Cluster labels for each gene
"""
n_genes = [Link][0]
flat_data = [Link](n_genes, -1)
return cluster_labels
def main():
# Parameters
n_genes = 10
n_conditions = 5
n_time_points = 8
n_clusters = 2
correlation_threshold = 0.7
# Loading dataset
print("Loading input dataset...")
data, gene_names, true_clusters = load_dataset(
filename="dataset_yeast.csv",
n_genes=n_genes,
n_conditions=n_conditions,
n_time_points=n_time_points,
n_clusters=n_clusters
)
# Initialize TriRNSC
trirnsc = TriRNSC(
threshold=correlation_threshold,
tabu_length_ratio=0.02,
naive_stopping_tolerance=15,
scaled_stopping_tolerance=15,
diversification_frequency=50,
diversification_length=10,
max_experiments=3
)
trirnsc.load_data(data, gene_names)
print("\nVisualizing true expression patterns...")
trirnsc.visualize_gene_expression_patterns(true_clusters)
tricluster_info = trirnsc.get_tricluster_info()
print("\nTop 3 triclusters:")
for i, tc in enumerate(tricluster_info[:3]):
print(f"Tricluster {tc['id']}: {tc['size']} genes, MSR={tc['msr']:.6f},
TQI={tc['tqi']:.6e}")
trirnsc.perform_go_analysis(top_n=10)
trirnsc.perform_kegg_analysis(top_n=10)
if __name__ == "__main__":
main()