import argparse import numpy as np import pandas as pd import torch from sklearn.model_selection import KFold from DeepTraCDR_model import DeepTraCDR, Optimizer from utils import evaluate_auc from data_sampler import RandomSampler from data_loader import load_data def parse_arguments(): """ Parses command-line arguments for the DeepTraCDR model. Returns: Parsed arguments containing model and training configurations. """ parser = argparse.ArgumentParser(description="DeepTraCDR: Graph-based Cell-Drug Interaction Prediction") parser.add_argument('-device', type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="Device to run the model on (cuda:0 or cpu)") parser.add_argument('-data', type=str, default='ccle', help="Dataset to use (default: ccle)") parser.add_argument('--wd', type=float, default=1e-4, help="Weight decay for optimizer") parser.add_argument('--layer_size', nargs='+', type=int, default=[512], help="Layer sizes for the model") parser.add_argument('--gamma', type=float, default=15, help="Gamma parameter for decoder") parser.add_argument('--epochs', type=int, default=1000, help="Number of training epochs") parser.add_argument('--test_freq', type=int, default=50, help="Frequency of evaluation during training") parser.add_argument('--lr', type=float, default=0.0005, help="Learning rate for optimizer") return parser.parse_args() def main(): """Main function to execute the DeepTraCDR training and evaluation pipeline.""" args = parse_arguments() # Load dataset full_adj, drug_fingerprints, exprs, null_mask, pos_num, args = load_data(args) # Log data shapes for debugging print(f"Original adj_mat shape: {full_adj.shape}") print("\n--- Data Shapes ---") print(f"Expression data shape: {exprs.shape}") print(f"Null mask shape: {null_mask.shape}") # Convert adjacency matrix to torch tensor if necessary if isinstance(full_adj, np.ndarray): full_adj = torch.from_numpy(full_adj).float() print(f"Converted adj_mat shape: {full_adj.shape}") # Initialize k-fold cross-validation k = 5 n_kfolds = 5 all_metrics = { 'auc': [], 'auprc': [], 'precision': [], 'recall': [], 'f1_score': [] } # Perform k-fold cross-validation for n_kfold in range(n_kfolds): kfold = KFold(n_splits=k, shuffle=True, random_state=n_kfold) for fold, (train_idx, test_idx) in enumerate(kfold.split(np.arange(pos_num))): # Initialize data sampler sampler = RandomSampler(full_adj, train_idx, test_idx, null_mask) # Initialize model model = DeepTraCDR( adj_mat=full_adj, cell_exprs=exprs, drug_finger=drug_fingerprints, layer_size=args.layer_size, gamma=args.gamma, device=args.device ) # Initialize optimizer opt = Optimizer( model=model, train_data=sampler.train_data, test_data=sampler.test_data, test_mask=sampler.test_mask, train_mask=sampler.train_mask, adj_matrix=full_adj, evaluate_fun=evaluate_auc, lr=args.lr, wd=args.wd, epochs=args.epochs, test_freq=args.test_freq, device=args.device ) # Train model and collect metrics true, pred, best_auc, best_auprc, best_precision, best_recall, best_f1 = opt.train() # Store metrics all_metrics['auc'].append(best_auc) all_metrics['auprc'].append(best_auprc) all_metrics['precision'].append(best_precision) all_metrics['recall'].append(best_recall) all_metrics['f1_score'].append(best_f1) print(f"Fold {n_kfold * k + fold + 1}: AUC={best_auc:.4f}, AUPRC={best_auprc:.4f}, " f"Precision={best_precision:.4f}, Recall={best_recall:.4f}, F1-Score={best_f1:.4f}") # Compute and display final metrics print("\nFinal Average Metrics:") for metric, values in all_metrics.items(): mean = np.mean(values) std = np.std(values) print(f"{metric.upper()}: {mean:.4f} ± {std:.4f}") if __name__ == "__main__": torch.set_float32_matmul_precision('high') main()