Skip to content

Instantly share code, notes, and snippets.

@Stfort52
Created August 17, 2023 07:57
Show Gist options
  • Select an option

  • Save Stfort52/319b96cd1c61d77a6cac7ca32c3c38a1 to your computer and use it in GitHub Desktop.

Select an option

Save Stfort52/319b96cd1c61d77a6cac7ca32c3c38a1 to your computer and use it in GitHub Desktop.
Train Celltypist with your data and perform 5-fold Cross-Validation
Display the source blob
Display the rendered blob
Raw
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Train Celltypist and perform 5-fold CV"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from collections import Counter\n",
"from pathlib import Path\n",
"\n",
"import celltypist as ct\n",
"import matplotlib.pyplot as plt\n",
"import numpy as np\n",
"import pandas as pd\n",
"import seaborn as sns\n",
"import scanpy as sc\n",
"from sklearn.metrics import confusion_matrix, f1_score\n",
"from sklearn.model_selection import StratifiedKFold"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"LOOM_PATH = Path('path/to/your/loom')\n",
"data = sc.read_loom(LOOM_PATH)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"TARGET_LABEL = 'Annotation'\n",
"# TARGET_LABEL = 'Subtype'"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"data.X = data.layers['counts']"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"before performing data splits, make sure to drop rare cell types!"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"celltype_counter = Counter(data.obs[TARGET_LABEL])\n",
"total_cells = sum(celltype_counter.values())\n",
"cells_to_drop = [k for k,v in celltype_counter.items() if v<=(0.005*total_cells)]\n",
"\n",
"def if_not_rare_celltype(example):\n",
" return example[\"cell_type\"] not in cells_to_drop"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"data = data[~data.obs[TARGET_LABEL].isin(cells_to_drop),]"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from typing import Optional\n",
"\n",
"\n",
"def normalize(data:sc.AnnData, inplace = False, target_sum = 1e4) -> Optional[sc.AnnData]:\n",
" if inplace:\n",
" sc.pp.normalize_total(data, target_sum=target_sum)\n",
" sc.pp.log1p(data)\n",
" else:\n",
" data = sc.pp.normalize_total(data, target_sum=target_sum, copy=True)\n",
" sc.pp.log1p(data)\n",
" return data\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"def eval_loop(\n",
" data: sc.AnnData,\n",
" splits: tuple[np.ndarray[int], np.ndarray[int]],\n",
" label: str = TARGET_LABEL,\n",
" nprocs: int = -1,\n",
" use_SGD: bool = False,\n",
" majority_voting: bool = False,\n",
") -> pd.DataFrame:\n",
" train, test = splits\n",
" model = ct.train(\n",
" normalize(data[train,]), labels=label, n_jobs=nprocs, use_SGD=use_SGD\n",
" )\n",
" pred = ct.annotate(\n",
" normalize(data[test,]), model=model, majority_voting=majority_voting\n",
" )\n",
" results = pd.concat([pred.predicted_labels, data[test,].obs[TARGET_LABEL]], axis=1)\n",
" return results\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"crossValidater = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)\n",
"\n",
"predictions = []\n",
"results = []\n",
"\n",
"for i, splits in enumerate(crossValidater.split(data.obs[TARGET_LABEL], data.obs[TARGET_LABEL])):\n",
" print(f\"Fold {i+1}\")\n",
" prediction = eval_loop(data, splits, TARGET_LABEL, 32)\n",
" accuracy = prediction.iloc[:, 0].eq(prediction.iloc[:, 1]).mean()\n",
" print(f\"Accuracy: {accuracy}\")\n",
" predictions.append(prediction)\n",
" results.append(accuracy)\n",
"\n",
"print(results)\n",
"print(f\"Mean accuracy: {np.mean(results)}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"all_predictions = pd.concat(predictions)\n",
"print('Macro F1:', f1_score(all_predictions[TARGET_LABEL], all_predictions.predicted_labels, average='macro'))\n",
"all_predictions"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"labels = list(set(all_predictions[TARGET_LABEL]))\n",
"cfm = confusion_matrix(all_predictions[TARGET_LABEL], all_predictions.predicted_labels, labels=labels)\n",
"cfm = cfm.astype('float') / cfm.sum(axis=1)[:, np.newaxis]\n",
"cfm = cfm.round(2)\n",
"plt.figure(figsize=(12,10))\n",
"sns.heatmap(cfm, annot=True, cmap='Blues', fmt='g', xticklabels=labels, yticklabels=labels)\n",
"sns.set_theme(font_scale=1.3)\n",
"plt.tick_params(axis='both', top=False, bottom=False, left=False, right=False, labelleft=True, labelbottom=True)\n",
"plt.xlabel(\"Predicted\")\n",
"plt.ylabel(\"True\")"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "adiFormer",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.12"
},
"orig_nbformat": 4
},
"nbformat": 4,
"nbformat_minor": 2
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment