app/classifier.py

"""Supervised LOF/GOF/NEUTRAL classifier.

Pipeline: mean-pooled ESM-2 embedding of the wild-type sequence, concatenated
with the mean-pooled embedding of the mutated sequence, fed to a
gradient-boosted 3-class classifier. The wild-type embedding is cached per
sequence, so scoring many variants in one protein only pays for the alt
embedding each time. Honest direction (GOF vs LOF) AUROC is ~0.76 on held-out
genes (gene-disjoint CV); the naive gene-mixed 0.99 is mostly protein-identity
leakage, so treat the direction call as a weak prior.
"""
from __future__ import annotations

from functools import lru_cache

import joblib
import numpy as np
import torch

from . import cache, config

torch.set_num_threads(config.TORCH_NUM_THREADS)
AA20 = set("ACDEFGHIKLMNPQRSTVWY")


@lru_cache(maxsize=1)
def _classifier():
    return joblib.load(config.CLASSIFIER_PATH)


@lru_cache(maxsize=1)
def _esm():
    from transformers import AutoTokenizer, EsmModel
    tok = AutoTokenizer.from_pretrained(config.ESM_MODEL_NAME)
    model = EsmModel.from_pretrained(config.ESM_MODEL_NAME).eval()
    return tok, model


def _mean_embedding(sequence: str) -> np.ndarray:
    cached = cache.get_embedding(sequence)
    if cached is not None:
        return cached
    tok, model = _esm()
    inputs = tok(sequence, return_tensors="pt", truncation=True,
                 max_length=config.MAX_SEQUENCE_LENGTH, padding=False)
    with torch.no_grad():
        out = model(**inputs, output_hidden_states=True)
        emb = out.hidden_states[-1].mean(dim=1).squeeze(0).cpu().numpy()
    emb = emb.astype(np.float32)
    cache.set_embedding(sequence, emb)
    return emb


def _apply(sequence: str, pos: int, alt: str) -> str:
    chars = list(sequence)
    chars[pos - 1] = alt
    return "".join(chars)


def classify(sequence: str, pos: int, alt: str) -> dict:
    """Return {label, probabilities} for a single missense substitution."""
    ref_emb = _mean_embedding(sequence)
    alt_emb = _mean_embedding(_apply(sequence, pos, alt))
    features = np.concatenate([ref_emb, alt_emb])
    proba = _classifier().predict_proba([features])[0]
    idx = int(np.argmax(proba))
    return {
        "label": config.LABELS[idx],
        "probabilities": {config.LABELS[i]: float(proba[i])
                          for i in range(len(config.LABELS))},
    }