Effects API

See effect annotation for result access and the effect catalog for individual consequence classes.

Effects

varcode.MutationEffect(variant)

Bases: Serializable

Base class for mutation effects.

Source code in varcode/effects/effect_classes.py

def __init__(self, variant):
    self.variant = variant

short_description: str property

A short but human-readable description of the effect. Defaults to class name for most of the non-coding effects, but is more informative for coding ones.

orf property

The selected reference ORF, or None for ordinary annotation.

reference_transcript property

Original annotated RNA, including when an extra coding frame is used.

original_protein_sequence: str | None property

Pre-somatic protein: the assembled patient baseline when supplied, otherwise the reference protein. None means unavailable.

mutant_protein_sequence: str | None property writable

Protein from the assembled transcript, when one is available.

baseline_transcript: MutantTranscript | None = None class-attribute instance-attribute

The actual pre-somatic transcript used for a composed comparison.

None means no baseline model was retained. A model whose sequence is None instead records an explicitly unavailable product. The model uses the same MutantTranscript representation as the post-somatic product.

__reduce__()

Pickle and copy through to_dict(), plus the attributes set after construction.

Source code in varcode/effects/effect_classes.py

def __reduce__(self):
    """Pickle and copy through to_dict(), plus the attributes set after construction."""
    reduced = super().__reduce__()
    attached = {k: self.__dict__[k] for k in self._attached if k in self.__dict__}
    return reduced + (attached,) if attached else reduced

__lt__(other)

Effects are ordered by their associated variants, which have comparison implement in terms of their chromosomal locations.

Source code in varcode/effects/effect_classes.py

def __lt__(self, other):
    """
    Effects are ordered by their associated variants, which have
    comparison implement in terms of their chromosomal locations.
    """
    return self.variant < other.variant

format_short_description(*, deletion_notation: Literal['sequence', 'range'] = 'sequence') -> str

Render the short description with a protein deletion display choice.

PARAMETER	DESCRIPTION
deletion_notation	"sequence" (default) includes every deleted residue and its one-based start, e.g. p.RSP3del. "range" uses the first and last residue positions, e.g. p.R3_P5del. Other consequence and splice-mechanism labels retain their existing notation. TYPE: (sequence, range) DEFAULT: "sequence"

RETURNS	DESCRIPTION
str	A display label. This does not change the effect or its archive.

Source code in varcode/effects/effect_classes.py

def format_short_description(
        self, *, deletion_notation: Literal["sequence", "range"] = "sequence",
) -> str:
    """Render the short description with a protein deletion display choice.

    Parameters
    ----------
    deletion_notation : {"sequence", "range"}
        ``"sequence"`` (default) includes every deleted residue and its
        one-based start, e.g. ``p.RSP3del``. ``"range"`` uses the first
        and last residue positions, e.g. ``p.R3_P5del``. Other consequence
        and splice-mechanism labels retain their existing notation.

    Returns
    -------
    str
        A display label. This does not change the effect or its archive.
    """
    validate_deletion_notation(deletion_notation)
    if deletion_notation == "sequence":
        return self.short_description
    return self._format_short_description(deletion_notation)

varcode.NonsilentCodingMutation(variant, transcript, aa_mutation_start_offset, aa_mutation_end_offset, aa_ref)

Bases: CodingMutation

All coding mutations other than silent codon substitutions

variant : Variant

transcript : Transcript

aa_mutation_start_offset : int Offset of first modified amino acid in protein (starting from 0)

aa_mutation_end_offset : int Offset after last mutated amino acid (half-open coordinates)

aa_ref : str Amino acid string of what used to be at aa_mutation_start_offset in the wildtype (unmutated) protein.

Source code in varcode/effects/effect_classes.py

def __init__(
        self,
        variant,
        transcript,
        aa_mutation_start_offset,
        aa_mutation_end_offset,
        aa_ref):
    """
    variant : Variant

    transcript : Transcript

    aa_mutation_start_offset : int
        Offset of first modified amino acid in protein (starting from 0)

    aa_mutation_end_offset : int
        Offset after last mutated amino acid (half-open coordinates)

    aa_ref : str
        Amino acid string of what used to be at aa_mutation_start_offset
        in the wildtype (unmutated) protein.
    """
    CodingMutation.__init__(
        self,
        variant=variant,
        transcript=transcript)
    self.aa_mutation_start_offset = aa_mutation_start_offset
    self.aa_mutation_end_offset = aa_mutation_end_offset
    self.aa_ref = bio_seq_to_str(aa_ref)

varcode.MultiOutcomeEffect(variant)

Bases: MutationEffect

Marker base class for effects that represent a set of plausible outcomes rather than a single deterministic effect.

Subclasses must expose:

• :attr:candidates — tuple of :class:~varcode.effect_candidates.EffectCandidate objects in producer order. Each entry pairs an inner :class:MutationEffect (concrete or placeholder) with its provenance — source (producer) and evidence dict. The :attr:effects helper unwraps to the inner Effects when callers don't need provenance.
• :attr:priority_class — effect class whose priority this set adopts (read by :func:varcode.effects.effect_priority).

Downstream consumers filter for multi-outcome results with isinstance(effect, MultiOutcomeEffect), so new wrappers (RNA evidence #259, germline-aware #268, SV-at-breakpoint) implement the same protocol uniformly (#382).

External integrations (RNA evidence, SpliceAI scoring, etc.) attach extra candidates post-hoc via the _extra_candidates slot — subclasses that override :attr:candidates must include those extras in their returned tuple. The :meth:_combine_with_extra_candidates helper does the right thing.

Picking the candidate

Two orthogonal "best candidate" notions are available; pick the one that matches your question:

• Most likely: the first candidate after producer ordering. Producers preserve their own deterministic order. :attr:most_likely_candidate returns the wrapped :class:EffectCandidate (provenance + inner effect); :attr:most_likely_effect returns just the inner :class:MutationEffect. Always equal to candidates[0] / effects[0].

• Highest priority: top by varcode's effect-priority ordering (see :func:~varcode.effects.effect_priority) — the most protein-disruptive candidate regardless of producer order. :attr:highest_priority_candidate and :attr:highest_priority_effect are the analogous accessors. Use this for clinical / functional filtering ("flag if any candidate is at least a frameshift"), since a disruptive candidate sitting behind a less-disruptive primary candidate should still light up.

The two coincide when producer order and priority ranking agree, which is common but not guaranteed. Pick consciously.

Source code in varcode/effects/effect_classes.py

def __init__(self, variant):
    self.variant = variant

effects property

Tuple of inner :class:MutationEffect objects, in :attr:candidates order. Convenience for callers that don't need per-candidate provenance — equivalent to tuple(c.effect for c in self.candidates).

most_likely_candidate property

The first :class:EffectCandidate in producer order. Pairs the inner effect with its source / evidence provenance.

For just the inner :class:MutationEffect, use :attr:most_likely_effect. For the most protein-disruptive candidate (independent of producer order), use :attr:highest_priority_candidate.

most_likely_effect property

The :class:MutationEffect of :attr:most_likely_candidate. Equivalent to most_likely_candidate.effect / effects[0] — given here so callers that don't need provenance don't have to reach through the wrapper.

highest_priority_candidate property

The :class:EffectCandidate whose inner effect has the highest :func:~varcode.effects.effect_priority (most protein-disruptive). Pure priority ranking — producer order deliberately doesn't factor in, so a frameshift sitting behind a less-disruptive primary candidate still surfaces here.

Ties on priority resolve to the first matching entry of :attr:candidates, preserving the subclass's candidate order.

Behavior is deterministic.

highest_priority_effect property

The inner :class:MutationEffect of :attr:highest_priority_candidate. Use when you want the worst-case effect for clinical / functional filtering and don't need provenance.

varcode.EffectCollection(effects, distinct=False, sort_key=None, sources=None, annotator=None, annotator_version=None, annotated_at=None)

Bases: Collection

Collection of MutationEffect objects and helpers for grouping or filtering them.

PARAMETER	DESCRIPTION
effects	Collection of any class which is compatible with the sort key TYPE: list
distinct	Only keep distinct entries or allow duplicates. TYPE: bool DEFAULT: False
sort_key	Function which maps each element to a sorting criterion. If None (the default), effects are sorted by priority with the most severe effects first. Pass an explicit sort_key to override this behaviour, or False to disable sorting. TYPE: fn DEFAULT: None
sources	Set of files from which this collection was generated. TYPE: set DEFAULT: None
annotator	Name of the :class:EffectAnnotator that produced these effects. Populated automatically by :func:predict_variant_effects; None for collections built by hand. See openvax/varcode#271. TYPE: str or None DEFAULT: None
annotator_version	Version string of the annotator (typically the varcode version for built-in annotators). None when annotator is None. TYPE: str or None DEFAULT: None
annotated_at	ISO-8601 UTC timestamp recording when the annotation ran. Populated by :func:predict_variant_effects. TYPE: str or None DEFAULT: None

Source code in varcode/effects/effect_collection.py

def __init__(
        self,
        effects,
        distinct=False,
        sort_key=None,
        sources=None,
        annotator=None,
        annotator_version=None,
        annotated_at=None):
    """
    Parameters
    ----------
    effects : list
        Collection of any class which  is compatible with the sort key


    distinct : bool
        Only keep distinct entries or allow duplicates.

    sort_key : fn
        Function which maps each element to a sorting criterion.
        If None (the default), effects are sorted by priority with
        the most severe effects first. Pass an explicit sort_key to
        override this behaviour, or `False` to disable sorting.

    sources : set
        Set of files from which this collection was generated.

    annotator : str or None
        Name of the :class:`EffectAnnotator` that produced these
        effects. Populated automatically by
        :func:`predict_variant_effects`; ``None`` for collections
        built by hand. See openvax/varcode#271.

    annotator_version : str or None
        Version string of the annotator (typically the varcode
        version for built-in annotators). ``None`` when
        ``annotator`` is None.

    annotated_at : str or None
        ISO-8601 UTC timestamp recording when the annotation ran.
        Populated by :func:`predict_variant_effects`.
    """
    if sort_key is None:
        sort_key = _default_effect_sort_key
    elif sort_key is False:
        sort_key = None
    if sources is None:
        sources = set()
    Collection.__init__(
        self,
        elements=effects,
        distinct=distinct,
        sort_key=sort_key,
        sources=sources)
    # Keep self.effects in sync with the Collection's post-sort
    # elements so that iterating and reading `.effects` produce
    # the same order.  See openvax/varcode#220.
    self.effects = self.elements
    self.annotator = annotator
    self.annotator_version = annotator_version
    self.annotated_at = annotated_at

orf_annotations property

Detailed native annotations for the extra ORF effects in this collection.

to_proteome(*, sample_id=None, run_id=None)

Export ordinary and ncORF outcomes to ORFdata's shared Proteome.

All supplied alternatives and omitted outcomes are retained. Requires the optional orf dependency. See :func:varcode.to_proteome for the source identity, reference and offline provenance contracts.

Source code in varcode/effects/effect_collection.py

def to_proteome(self, *, sample_id=None, run_id=None):
    """Export ordinary and ncORF outcomes to ORFdata's shared Proteome.

    All supplied alternatives and omitted outcomes are retained. Requires
    the optional ``orf`` dependency. See :func:`varcode.to_proteome` for
    the source identity, reference and offline provenance contracts.
    """
    from ..proteome import to_proteome
    return to_proteome(self, sample_id=sample_id, run_id=run_id)

gene_counts()

Returns number of elements overlapping each gene name. Expects the derived class (VariantCollection or EffectCollection) to have an implementation of groupby_gene_name.

Source code in varcode/effects/effect_collection.py

def gene_counts(self):
    """
    Returns number of elements overlapping each gene name. Expects the
    derived class (VariantCollection or EffectCollection) to have
    an implementation of groupby_gene_name.
    """
    return {
        gene_name: len(group)
        for (gene_name, group)
        in self.groupby_gene_name().items()
    }

filter_by_transcript_expression(transcript_expression_dict, min_expression_value=0.0)

Filters effects to those which have an associated transcript whose expression value in the transcript_expression_dict argument is greater than min_expression_value.

PARAMETER	DESCRIPTION
transcript_expression_dict	Dictionary mapping Ensembl transcript IDs to expression estimates (either FPKM or TPM) TYPE: dict
min_expression_value	Threshold above which we'll keep an effect in the result collection TYPE: float DEFAULT: 0.0

Source code in varcode/effects/effect_collection.py

def filter_by_transcript_expression(
        self,
        transcript_expression_dict,
        min_expression_value=0.0):
    """
    Filters effects to those which have an associated transcript whose
    expression value in the transcript_expression_dict argument is greater
    than min_expression_value.

    Parameters
    ----------
    transcript_expression_dict : dict
        Dictionary mapping Ensembl transcript IDs to expression estimates
        (either FPKM or TPM)

    min_expression_value : float
        Threshold above which we'll keep an effect in the result collection
    """
    return self.filter_above_threshold(
        key_fn=lambda effect: effect.transcript_id,
        value_dict=transcript_expression_dict,
        threshold=min_expression_value)

filter_by_gene_expression(gene_expression_dict, min_expression_value=0.0)

Filters effects to those which have an associated gene whose expression value in the gene_expression_dict argument is greater than min_expression_value.

PARAMETER	DESCRIPTION
gene_expression_dict	Dictionary mapping Ensembl gene IDs to expression estimates (either FPKM or TPM) TYPE: dict
min_expression_value	Threshold above which we'll keep an effect in the result collection TYPE: float DEFAULT: 0.0

Source code in varcode/effects/effect_collection.py

def filter_by_gene_expression(
        self,
        gene_expression_dict,
        min_expression_value=0.0):
    """
    Filters effects to those which have an associated gene whose
    expression value in the gene_expression_dict argument is greater
    than min_expression_value.

    Parameters
    ----------
    gene_expression_dict : dict
        Dictionary mapping Ensembl gene IDs to expression estimates
        (either FPKM or TPM)

    min_expression_value : float
        Threshold above which we'll keep an effect in the result collection
    """
    return self.filter_above_threshold(
        key_fn=lambda effect: effect.gene_id,
        value_dict=gene_expression_dict,
        threshold=min_expression_value)

filter_by_effect_priority(min_priority_class)

Create a new EffectCollection containing only effects whose priority falls below the given class.

Source code in varcode/effects/effect_collection.py

def filter_by_effect_priority(self, min_priority_class):
    """
    Create a new EffectCollection containing only effects whose priority
    falls below the given class.
    """
    min_priority = transcript_effect_priority_dict[min_priority_class]
    return self.filter(
        lambda effect: effect_priority(effect) >= min_priority)

drop_silent_and_noncoding(keep_unresolved: bool = True) -> EffectCollection

Keep effects with a protein-changing candidate or unresolved SV outcome.

PARAMETER	DESCRIPTION
keep_unresolved	Keep effects whose protein-change status is None (default True). False requires a positive prediction of protein change. Candidate sets are retained intact if any alternative qualifies; their order and provenance are unchanged. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
EffectCollection	Filtered collection retaining complete qualifying outcome sets.

Source code in varcode/effects/effect_collection.py

def drop_silent_and_noncoding(self, keep_unresolved: bool = True) -> EffectCollection:
    """
    Keep effects with a protein-changing candidate or unresolved SV outcome.

    Parameters
    ----------
    keep_unresolved : bool
        Keep effects whose protein-change status is None (default True).
        False requires a positive prediction of protein change. Candidate
        sets are retained intact if any alternative qualifies; their order
        and provenance are unchanged.

    Returns
    -------
    EffectCollection
        Filtered collection retaining complete qualifying outcome sets.
    """
    from .sequence_change import modification_status

    def retain(effect):
        status = modification_status(effect, "modifies_protein_sequence")
        return status is True or (keep_unresolved and status is None)

    return self.filter(retain)

detailed_string()

Create a long string with all transcript effects for each mutation, grouped by gene (if a mutation affects multiple genes).

Source code in varcode/effects/effect_collection.py

def detailed_string(self):
    """
    Create a long string with all transcript effects for each mutation,
    grouped by gene (if a mutation affects multiple genes).
    """
    lines = []
    # TODO: annoying to always write `groupby_result.items()`,
    # consider makings a GroupBy class which iterates over pairs
    # and also common helper methods like `map_values`.
    for variant, variant_effects in self.groupby_variant().items():
        lines.append("\n%s" % variant)

        gene_effects_groups = variant_effects.groupby_gene_id()
        for (gene_id, gene_effects) in gene_effects_groups.items():
            if gene_id:
                gene_name = variant.ensembl.gene_name_of_gene_id(gene_id)
                lines.append("  Gene: %s (%s)" % (gene_name, gene_id))
            # place transcript effects with more significant impact
            # on top (e.g. FrameShift should go before NoncodingTranscript)
            for effect in sorted(
                    gene_effects,
                    key=effect_priority,
                    reverse=True):
                lines.append("  -- %s" % effect)

        # if we only printed one effect for this gene then
        # it's redundant to print it again as the highest priority effect
        if len(variant_effects) > 1:
            best = variant_effects.top_priority_effect()
            lines.append("  Highest Priority Effect: %s" % best)
    return "\n".join(lines)

top_priority_effect() -> MutationEffect

Select one effect using Varcode's severity and transcript heuristics.

RETURNS	DESCRIPTION
MutationEffect	Selected effect, retaining complete splice outcome sets. This is a ranking summary, not evidence that the selected transcript or outcome is expressed. See :func:varcode.top_priority_effect.

RAISES	DESCRIPTION
ValueError	The collection is empty. This method does not synthesize an Intergenic effect for an empty or filtered collection.

Source code in varcode/effects/effect_collection.py

def top_priority_effect(self) -> MutationEffect:
    """Select one effect using Varcode's severity and transcript heuristics.

    Returns
    -------
    MutationEffect
        Selected effect, retaining complete splice outcome sets. This is
        a ranking summary, not evidence that the selected transcript or
        outcome is expressed. See :func:`varcode.top_priority_effect`.

    Raises
    ------
    ValueError
        The collection is empty. This method does not synthesize an
        Intergenic effect for an empty or filtered collection.
    """
    return top_priority_effect(self.elements)

top_priority_effect_per_variant()

Highest priority effect for each unique variant

Source code in varcode/effects/effect_collection.py

def top_priority_effect_per_variant(self):
    """Highest priority effect for each unique variant"""
    return OrderedDict(
        (variant, top_priority_effect(variant_effects))
        for (variant, variant_effects)
        in self.groupby_variant().items())

top_priority_effect_per_transcript_id()

Highest priority effect for each unique transcript ID

Source code in varcode/effects/effect_collection.py

def top_priority_effect_per_transcript_id(self):
    """Highest priority effect for each unique transcript ID"""
    return OrderedDict(
        (transcript_id, top_priority_effect(variant_effects))
        for (transcript_id, variant_effects)
        in self.groupby_transcript_id().items())

top_priority_effect_per_gene_id()

Highest priority effect for each unique gene ID

Source code in varcode/effects/effect_collection.py

def top_priority_effect_per_gene_id(self):
    """Highest priority effect for each unique gene ID"""
    return OrderedDict(
        (gene_id, top_priority_effect(variant_effects))
        for (gene_id, variant_effects)
        in self.groupby_gene_id().items())

effect_expression(expression_levels)

PARAMETER	DESCRIPTION
expression_levels	Dictionary mapping transcript IDs to length-normalized expression levels (either FPKM or TPM) TYPE: dict

RETURNS	DESCRIPTION
OrderedDict	Mapping from each transcript effect to an expression quantity. Effects that don't have an associated transcript (e.g. Intergenic) are excluded.

Source code in varcode/effects/effect_collection.py

def effect_expression(self, expression_levels):
    """
    Parameters
    ----------
    expression_levels : dict
        Dictionary mapping transcript IDs to length-normalized expression
        levels (either FPKM or TPM)

    Returns
    -------
    OrderedDict
        Mapping from each transcript effect to an expression quantity.
        Effects that don't have an associated transcript (e.g. Intergenic)
        are excluded.
    """
    return OrderedDict(
        (effect, expression_levels.get(effect.transcript.id, 0.0))
        for effect in self
        if effect.transcript is not None)

top_expression_effect(expression_levels)

Return effect whose transcript has the highest expression level. If none of the effects are expressed or have associated transcripts, then return None. In case of ties, add lexicographical sorting by effect priority and transcript length.

Source code in varcode/effects/effect_collection.py

def top_expression_effect(self, expression_levels):
    """
    Return effect whose transcript has the highest expression level.
    If none of the effects are expressed or have associated transcripts,
    then return None. In case of ties, add lexicographical sorting by
    effect priority and transcript length.
    """
    effect_expression_dict = self.effect_expression(expression_levels)

    if len(effect_expression_dict) == 0:
        return None

    def key_fn(effect_fpkm_pair):
        """
        Sort effects primarily by their expression level
        and secondarily by the priority logic used in
        `top_priority_effect`.
        """
        (effect, fpkm) = effect_fpkm_pair
        return (fpkm, multi_gene_effect_sort_key(effect))

    return max(effect_expression_dict.items(), key=key_fn)[0]

to_dataframe(include_archive=False, *, deletion_notation='sequence')

Build a summary; optionally retain full graphs in effect_json.

deletion_notation="range" renders protein deletion labels using endpoint positions instead of the default full deleted sequence. See :meth:MutationEffect.format_short_description.

Source code in varcode/effects/effect_collection.py

def to_dataframe(self, include_archive=False, *, deletion_notation="sequence"):
    """Build a summary; optionally retain full graphs in ``effect_json``.

    ``deletion_notation="range"`` renders protein deletion labels using
    endpoint positions instead of the default full deleted sequence.
    See :meth:`MutationEffect.format_short_description`.
    """
    from ..splice_outcomes import SpliceOutcomeSet
    from .effect_classes import StartLoss

    validate_deletion_notation(deletion_notation)
    structural_columns = STRUCTURAL_VARIANT_COLUMNS if any(
        getattr(e.variant, "is_structural", False) for e in self) else ()
    orf_columns = ("orf_id", "orf_source", "orf_unresolved_reason") if any(
        e.orf_annotation is not None for e in self) else ()
    splice_columns = ("splice_signal",) if any(
        isinstance(e, SpliceOutcomeSet) for e in self) else ()
    start_loss_columns = (
        "ref_codon", "alt_codon", "ref_codon_offset", "alt_codon_offset",
        "codon_change_status", "alt_codon_amino_acid",
    ) if any(isinstance(e, StartLoss) for e in self) else ()
    # list of properties to extract from Variant objects if they're
    # not None
    variant_properties = [
        "contig",
        "start",
        "ref",
        "alt",
        "is_snv",
        "is_transversion",
        "is_transition"
    ] + list(structural_columns)

    def row_from_effect(effect):
        row: OrderedDict[str, object] = OrderedDict()

        row['variant'] = str(effect.variant.short_description)
        for field_name in variant_properties:
            # if effect.variant is None then this column value will be None
            row[field_name] = getattr(effect.variant, field_name, None)
        row['gene_id'] = effect.gene_id
        row['gene_name'] = effect.gene_name
        row['transcript_id'] = effect.transcript_id
        row['transcript_name'] = effect.transcript_name

        row['effect_type'] = effect.__class__.__name__
        row['effect'] = effect.format_short_description(deletion_notation=deletion_notation)
        if isinstance(effect, StartLoss):
            for field_name in start_loss_columns:
                row[field_name] = getattr(effect, field_name)
        if isinstance(effect, SpliceOutcomeSet):
            signal = effect.disrupted_signal_class
            row['splice_signal'] = signal.__name__ if signal is not None else None
        annotation = effect.orf_annotation
        if annotation is not None:
            row['orf_id'] = annotation.orf_id
            row['orf_source'] = annotation.record.source
            row['orf_unresolved_reason'] = annotation.unresolved_reason
        return row
    # Always emit the same column set even for empty collections so
    # CSV round-trip doesn't have to special-case len == 0.
    frame = pd.DataFrame.from_records(
        [row_from_effect(effect) for effect in self],
        columns=(self._DATAFRAME_COLUMNS + structural_columns + orf_columns
                 + splice_columns + start_loss_columns),
    )
    if include_archive and (self.orf_annotations or start_loss_columns):
        frame["effect_json"] = [effect.to_json() for effect in self]
    return frame

to_csv(path, include_header=True, *, deletion_notation='sequence')

Write this collection to CSV.

PARAMETER	DESCRIPTION
path	Output path. TYPE: str
include_header	If True (default), prepend # key=value metadata lines with varcode version and reference genome so the file can be read back via from_csv without supplying a genome explicitly. Pass include_header=False for fast consumers that don't tolerate comment lines. TYPE: bool DEFAULT: True
deletion_notation	Display pure protein deletions as p.RSP3del (default) or p.R3_P5del. This does not change the archived effect; loading the CSV restores effects with their default display notation. TYPE: ('sequence', 'range') DEFAULT: "sequence"

Source code in varcode/effects/effect_collection.py

def to_csv(self, path, include_header=True, *, deletion_notation="sequence"):
    """Write this collection to CSV.

    Parameters
    ----------
    path : str
        Output path.

    include_header : bool
        If True (default), prepend ``# key=value`` metadata lines
        with varcode version and reference genome so the file can
        be read back via ``from_csv`` without supplying a genome
        explicitly. Pass ``include_header=False`` for fast
        consumers that don't tolerate comment lines.

    deletion_notation : {"sequence", "range"}
        Display pure protein deletions as ``p.RSP3del`` (default) or
        ``p.R3_P5del``. This does not change the archived effect; loading
        the CSV restores effects with their default display notation.
    """
    df = self.to_dataframe(include_archive=True, deletion_notation=deletion_notation)
    if not include_header:
        df.to_csv(path, index=False)
        return

    metadata = OrderedDict()
    metadata["varcode_version"] = _varcode_version
    metadata["reference_name"] = self._serialized_reference_name()
    # Annotator provenance. Each field is skipped
    # when None by write_metadata_header.
    metadata["annotator"] = self.annotator
    metadata["annotator_version"] = self.annotator_version
    metadata["annotated_at"] = self.annotated_at
    with open(path, "w") as f:
        write_metadata_header(f, metadata)
        df.to_csv(f, index=False)

from_csv(path, genome=None) classmethod

Rebuild an EffectCollection from a CSV previously written by EffectCollection.to_csv().

ORF and start-loss exports include an effect_json column. Those rows restore the saved graphs without reannotation; edited identity columns that disagree with their archives raise.

Ordinary summary CSV records (contig, start, ref, alt, transcript_id) but not enough per-effect state to reconstruct effects byte-for-byte. This method takes the pragmatic semantic round-trip path: rebuild each Variant, re-annotate against the recorded transcript, and emit the resulting effect. The resulting collection should match the original whenever annotation is deterministic for a given (variant, transcript) pair.

Prefer from_json for byte-for-byte round-trip or for larger collections (≳10k effects); per-row re-annotation makes CSV loading significantly slower than JSON. Emits a warning when the CSV header reports a different major varcode version than the one currently installed — annotation logic can change across major versions and the reconstructed effects may differ from the ones that were written.

PARAMETER	DESCRIPTION
path	Path to the CSV file. Lines starting with '#' are treated as comments and parsed as key=value metadata. TYPE: str
genome	Reference genome to associate with the loaded variants and to look up transcripts by ID. If None, the reference is read from the CSV's metadata header (# reference_name=...). If neither is available, raises ValueError. TYPE: pyensembl.Genome, str, int, or None DEFAULT: None

RETURNS	DESCRIPTION
EffectCollection

Source code in varcode/effects/effect_collection.py

@classmethod
def from_csv(cls, path, genome=None):
    """Rebuild an EffectCollection from a CSV previously written by
    ``EffectCollection.to_csv()``.

    ORF and start-loss exports include an ``effect_json`` column. Those rows
    restore the saved graphs without reannotation; edited identity columns
    that disagree with their archives raise.

    Ordinary summary CSV records (contig, start, ref, alt,
    transcript_id) but not enough per-effect state to reconstruct
    effects byte-for-byte. This method takes the pragmatic semantic
    round-trip path: rebuild each Variant, re-annotate against the
    recorded transcript, and emit the resulting effect. The
    resulting collection should match the original whenever
    annotation is deterministic for a given (variant, transcript)
    pair.

    **Prefer ``from_json`` for byte-for-byte round-trip** or for
    larger collections (≳10k effects); per-row re-annotation makes
    CSV loading significantly slower than JSON. Emits a warning
    when the CSV header reports a different *major* varcode version
    than the one currently installed — annotation logic can change
    across major versions and the reconstructed effects may differ
    from the ones that were written.

    Parameters
    ----------
    path : str
        Path to the CSV file. Lines starting with '#' are treated as
        comments and parsed as ``key=value`` metadata.

    genome : pyensembl.Genome, str, int, or None
        Reference genome to associate with the loaded variants and
        to look up transcripts by ID. If ``None``, the reference is
        read from the CSV's metadata header
        (``# reference_name=...``). If neither is available, raises
        ``ValueError``.

    Returns
    -------
    EffectCollection
    """
    # Import here to avoid a circular import at module load time.
    import warnings
    from ..variant import Variant

    header = read_metadata_header(path)
    df = pd.read_csv(
        path, comment="#", dtype={"chr": str, "contig": str})
    if "effect_json" in df:
        from .effect_classes import MutationEffect
        if df["effect_json"].isna().any():
            raise ValueError("Incomplete effect_json archive column; every row must retain its effect")
        effects = []
        for payload in df["effect_json"]:
            effect = MutationEffect.from_json(payload)
            if not isinstance(effect, MutationEffect):
                raise ValueError("effect_json row does not contain a mutation effect")
            effects.append(effect)
        result = cls(effects, sort_key=False,
                     annotator=header.get("annotator"), annotator_version=header.get("annotator_version"),
                     annotated_at=header.get("annotated_at"))
        summary = result.to_dataframe()
        # Refuse a stale payload after spreadsheet edits to identifying
        # columns. Empty alleles and missing transcript IDs normalize alike.
        for column in ("contig", "start", "ref", "alt", "transcript_id", "effect_type", "orf_id", "orf_source",
                       "ref_codon", "alt_codon", "codon_change_status", "alt_codon_amino_acid"):
            if column in df and column in summary:
                original = df[column].fillna("").astype(str).tolist()
                restored = summary[column].fillna("").astype(str).tolist()
                if original != restored:
                    raise ValueError("CSV %s disagrees with its archived effect_json" % column)
        return result
    warn_on_version_drift(header, _varcode_version, path)
    if genome is None:
        genome = header.get("reference_name")
    if genome is None:
        raise ValueError(
            "from_csv needs a reference genome: pass the `genome` "
            "argument explicitly, or write the CSV with "
            "`to_csv(include_header=True)` so `# reference_name=...` "
            "is recorded in the header. Neither was found at %s." % path)

    # Accept either "contig" or "chr" as the contig column so
    # CSVs are interchangeable between VariantCollection and
    # EffectCollection (openvax/varcode#274). Declaring both in
    # dtype is a no-op for whichever column is absent.
    reject_structural_csv(df)
    if "orf_id" in df and df["orf_id"].notna().any():
        raise ValueError("Extra ORF CSV rows require their reference coding models; "
                         "canonical transcript re-annotation would lose the ORF. "
                         "Use EffectCollection.to_csv() or to_json() for full exports.")
    contig_col = resolve_contig_column(df.columns)
    if contig_col is None:
        raise ValueError(
            "CSV at %s is missing a contig column: expected one of %s."
            % (path, list(CONTIG_COLUMN_ALIASES)))
    required = {"start", "ref", "alt", "transcript_id", "effect_type"}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(
            "CSV at %s is missing required columns: %s" % (
                path, sorted(missing)))

    # Extract columns by name and coerce types up front; zip is
    # robust against column reordering and extra columns, unlike
    # itertuples which depends on attribute access to
    # valid-identifier column names in fixed positions.
    contigs = df[contig_col].astype(str)
    starts = df["start"].astype(int)
    refs = df["ref"].fillna("")
    alts = df["alt"].fillna("")
    transcript_ids = df["transcript_id"]
    effect_types = df["effect_type"]

    effects = []
    resolved_genome = None
    for contig, start, ref, alt, transcript_id, effect_type in zip(
            contigs, starts, refs, alts, transcript_ids, effect_types):
        variant = Variant(
            contig=contig,
            start=start,
            ref=ref,
            alt=alt,
            genome=genome,
        )
        # Cache the resolved pyensembl.Genome from the first variant
        # so subsequent transcript lookups don't pay the inference cost.
        if resolved_genome is None:
            resolved_genome = variant.ensembl
        if pd.isna(transcript_id) or transcript_id == "":
            # Row is for an intergenic / intragenic effect — no
            # transcript context. Rebuild by running the full effect
            # set on the variant and taking the non-transcript match.
            effects_for_variant = variant.effects()
            matched = False
            for e in effects_for_variant:
                if e.transcript is None and e.__class__.__name__ == effect_type:
                    effects.append(e)
                    matched = True
                    break
            if not matched:
                warnings.warn(
                    "Could not recover effect of type %r for variant %s "
                    "with no transcript_id in %s; row dropped from "
                    "reconstructed collection." % (
                        effect_type, variant, path)
                )
            continue
        transcript = resolved_genome.transcript_by_id(str(transcript_id))
        effects.append(variant.effect_on_transcript(transcript))
    return cls(
        effects=effects,
        annotator=header.get("annotator"),
        annotator_version=header.get("annotator_version"),
        annotated_at=header.get("annotated_at"),
    )

varcode.EffectCandidate(effect: Any, source: str = 'varcode', evidence: Mapping[str, Any] = dict()) dataclass

Bases: DataclassSerializable

One plausible effect for a variant, paired with provenance.

PARAMETER	DESCRIPTION
effect	The effect this candidate represents. Guaranteed to be a :class:~varcode.effects.MutationEffect instance. For outcomes whose protein math isn't yet resolved (e.g. a splice-mechanism Effect built without a genomic_sequence provider), the inner Effect still satisfies the interface — aa_ref / aa_alt / mutant_protein_sequence are simply None, and consumers can read candidate.effect.short_description uniformly across SV, splice, and point-variant candidates. TYPE: MutationEffect
source	Name of the tool or annotator that produced this candidate. Defaults to "varcode" for built-in classifications. External integrations set their own opaque string. Varcode does not interpret this field beyond exact-string equality. TYPE: str DEFAULT: 'varcode'
evidence	Open-ended provenance dict. Shape is source-specific; the convention is that keys match the source's native field names. Consumers that need a particular shape should type-check at the call site rather than rely on a rigid schema here. TYPE: Mapping[str, Any] DEFAULT: dict()

short_description: str property

Convenience passthrough to self.effect.short_description. Lets callers build tables without unpacking candidate.effect.short_description everywhere.

varcode.candidates_from_effects(effects: Tuple[Any, ...], source: str = 'varcode') -> Tuple[EffectCandidate, ...]

Wrap a tuple of :class:MutationEffect instances as :class:EffectCandidate objects with a shared source string.

Convenience for the common case where a producer has a tuple of Effects and wants to lift it into the wrapped form without setting evidence fields. Callers with producer-specific metadata construct :class:EffectCandidate directly.

Source code in varcode/effect_candidates.py

def candidates_from_effects(
        effects: Tuple[Any, ...],
        source: str = "varcode") -> Tuple[EffectCandidate, ...]:
    """Wrap a tuple of :class:`MutationEffect` instances as
    :class:`EffectCandidate` objects with a shared ``source`` string.

    Convenience for the common case where a producer has a tuple of
    Effects and wants to lift it into the wrapped form without
    setting evidence fields. Callers with producer-specific metadata
    construct :class:`EffectCandidate` directly.
    """
    return tuple(EffectCandidate(effect=c, source=source) for c in effects)

Splice candidates

See splice variants for the distinction between signal disruption and candidate consequences. The scores below are sequence heuristics, not probabilities that a candidate is expressed.

varcode.SpliceCandidateRNAEvidence(candidate: EffectCandidate, rna_evidence: Tuple[EffectCandidate, ...] = tuple()) dataclass

Bases: DataclassSerializable

RNA observations supporting one visible splice candidate.

varcode.enumerate_cryptic_exon_candidates(contig: str, sequence: str, sequence_start: int, variant=None, donor_threshold: float = 0.7, acceptor_threshold: float = 0.7, min_intron_length: int = 30, max_intron_length: int = 10000, score_fn: Optional[ScoreFn] = None) -> List[CrypticExonCandidate]

Scan sequence for candidate cryptic exons — (acceptor, donor) pairs that could form a new internal exon on the rearranged allele.

PARAMETER	DESCRIPTION
contig	Chromosome / contig the sequence lives on (for reporting; not used for scoring). TYPE: str
sequence	The genomic sequence to scan — typically the region around an SV breakpoint, or a long-read-assembled rearranged allele. TYPE: str
sequence_start	1-based genomic start position of sequence[0]. Candidate intervals are reported relative to this origin. TYPE: int
variant	Source variant for provenance on returned candidates. None is allowed for pure-sequence scans (e.g. long-read assemblies not tied to a single SV). TYPE: Variant or None DEFAULT: None
donor_threshold	Minimum score to accept a site. Defaults (0.7) are chosen to be liberal — motif-based scoring has high false-positive rates, so the point is to enumerate candidates for downstream scoring, not to be the final caller. TYPE: float DEFAULT: 0.7
acceptor_threshold	Minimum score to accept a site. Defaults (0.7) are chosen to be liberal — motif-based scoring has high false-positive rates, so the point is to enumerate candidates for downstream scoring, not to be the final caller. TYPE: float DEFAULT: 0.7
min_intron_length	A donor following an acceptor must leave an intron in this length range for the candidate to be reported. Narrow defaults keep the output tractable; callers that want everything pass min=0, max=None. TYPE: int DEFAULT: 30
max_intron_length	A donor following an acceptor must leave an intron in this length range for the candidate to be reported. Narrow defaults keep the output tractable; callers that want everything pass min=0, max=None. TYPE: int DEFAULT: 30
score_fn	Override the default motif scorer. Signature (window_sequence, kind) -> score in [0, 1], with kind one of "donor", "acceptor". Plug in a SpliceAI / Pangolin wrapper here. TYPE: callable or None DEFAULT: None

RETURNS	DESCRIPTION
list of :class:`CrypticExonCandidate`	One entry per accepted (acceptor, donor) pair forming a plausible internal exon. Empty list when nothing scores above threshold.

Source code in varcode/cryptic_exons.py

def enumerate_candidates(
        contig: str,
        sequence: str,
        sequence_start: int,
        variant=None,
        donor_threshold: float = 0.7,
        acceptor_threshold: float = 0.7,
        min_intron_length: int = 30,
        max_intron_length: int = 10_000,
        score_fn: Optional[ScoreFn] = None,
) -> List[CrypticExonCandidate]:
    """Scan ``sequence`` for candidate cryptic exons — (acceptor,
    donor) pairs that could form a new internal exon on the
    rearranged allele.

    Parameters
    ----------
    contig : str
        Chromosome / contig the sequence lives on (for reporting;
        not used for scoring).
    sequence : str
        The genomic sequence to scan — typically the region around
        an SV breakpoint, or a long-read-assembled rearranged
        allele.
    sequence_start : int
        1-based genomic start position of ``sequence[0]``. Candidate
        intervals are reported relative to this origin.
    variant : Variant or None
        Source variant for provenance on returned candidates. ``None``
        is allowed for pure-sequence scans (e.g. long-read
        assemblies not tied to a single SV).
    donor_threshold, acceptor_threshold : float
        Minimum score to accept a site. Defaults (0.7) are chosen to
        be liberal — motif-based scoring has high false-positive
        rates, so the point is to enumerate candidates for
        downstream scoring, not to be the final caller.
    min_intron_length, max_intron_length : int
        A donor following an acceptor must leave an intron in this
        length range for the candidate to be reported. Narrow
        defaults keep the output tractable; callers that want
        everything pass ``min=0, max=None``.
    score_fn : callable or None
        Override the default motif scorer. Signature
        ``(window_sequence, kind) -> score in [0, 1]``, with
        ``kind`` one of ``"donor"``, ``"acceptor"``. Plug in a
        SpliceAI / Pangolin wrapper here.

    Returns
    -------
    list of :class:`CrypticExonCandidate`
        One entry per accepted (acceptor, donor) pair forming a
        plausible internal exon. Empty list when nothing scores
        above threshold.
    """
    scorer = score_fn if score_fn is not None else _default_score
    seq = sequence.upper()

    # First pass: collect all candidate donor and acceptor positions.
    donor_positions: List[Tuple[int, float]] = []
    acceptor_positions: List[Tuple[int, float]] = []

    # Donor: the window is 9 bp centered on the exon/intron boundary.
    # Position p means the boundary is after exon position p+2
    # (0-indexed exonic anchor).
    for i in range(len(seq) - DONOR_WINDOW + 1):
        window = seq[i:i + DONOR_WINDOW]
        score = scorer(window, "donor")
        if score >= donor_threshold:
            # Report the intron-start position (index of first intron
            # base within the local sequence) so acceptor / donor
            # distances are comparable.
            intron_start = i + 3  # after the 3 exonic bases
            donor_positions.append((intron_start, score))

    # Acceptor: 4-bp window, intronic AG at positions 1/2, exonic
    # base at position 3.
    for i in range(len(seq) - ACCEPTOR_WINDOW + 1):
        window = seq[i:i + ACCEPTOR_WINDOW]
        score = scorer(window, "acceptor")
        if score >= acceptor_threshold:
            # Report the exon-start position (index of first exonic
            # base within the local sequence).
            exon_start = i + 3
            acceptor_positions.append((exon_start, score))

    # Second pass: pair acceptors with downstream donors to produce
    # cryptic internal exon candidates.
    candidates: List[CrypticExonCandidate] = []
    max_len = max_intron_length if max_intron_length is not None else len(seq)
    for exon_start_local, acc_score in acceptor_positions:
        for intron_start_local, don_score in donor_positions:
            if intron_start_local <= exon_start_local:
                continue
            exon_length = intron_start_local - exon_start_local
            if exon_length < min_intron_length:
                continue
            if exon_length > max_len:
                continue
            candidates.append(CrypticExonCandidate(
                variant=variant,
                contig=contig,
                interval_start=sequence_start + exon_start_local,
                interval_end=sequence_start + intron_start_local,
                donor_score=don_score,
                acceptor_score=acc_score,
            ))
    # Sort by summed motif score, best first — downstream consumers
    # typically only care about the top N.
    candidates.sort(
        key=lambda c: (c.donor_score or 0) + (c.acceptor_score or 0),
        reverse=True)
    return candidates

varcode.enumerate_from_structural_variant(variant, flank: int = 500, score_fn: Optional[ScoreFn] = None, **kwargs) -> List[CrypticExonCandidate]

Convenience wrapper: enumerate candidates in the genomic region around an SV breakpoint.

Reads flank bp either side of each of the variant's breakpoints (see :attr:StructuralVariant.breakpoints, which covers both ends of every junction the variant creates). If the variant carries an alt_assembly (long-read resolution), scans that instead of the reference.

Returns the combined, sorted candidate list.

Source code in varcode/cryptic_exons.py

def enumerate_from_structural_variant(
        variant,
        flank: int = 500,
        score_fn: Optional[ScoreFn] = None,
        **kwargs) -> List[CrypticExonCandidate]:
    """Convenience wrapper: enumerate candidates in the genomic
    region around an SV breakpoint.

    Reads ``flank`` bp either side of each of the variant's
    breakpoints (see :attr:`StructuralVariant.breakpoints`, which
    covers both ends of every junction the variant creates). If the
    variant carries an ``alt_assembly`` (long-read resolution), scans
    that instead of the reference.

    Returns the combined, sorted candidate list.
    """
    candidates: List[CrypticExonCandidate] = []
    genome = getattr(variant, "genome", None) or getattr(
        variant, "ensembl", None)

    # If the SV carries a long-read-assembled allele, prefer it.
    if getattr(variant, "alt_assembly", None):
        contig = getattr(variant, "contig", "alt_assembly")
        candidates.extend(enumerate_candidates(
            contig=contig,
            sequence=variant.alt_assembly,
            sequence_start=getattr(variant, "start", 0),
            variant=variant,
            score_fn=score_fn,
            **kwargs))
        return candidates

    # Otherwise, scan reference flanking regions around breakpoints.
    breakpoints = list(getattr(variant, "breakpoints", ()) or ())
    if not breakpoints:
        breakpoints = [(variant.contig, variant.start)]

    if genome is None:
        return candidates
    for contig, start, end in _scan_windows(breakpoints, flank):
        ref_seq = reference_range(genome, contig, start, end)
        if not ref_seq:
            # Warn once per genome — firing per-breakpoint creates noise
            # in pipelines that annotate many SVs without a FASTA. The
            # user gets one actionable signal per genome; subsequent
            # uncovered breakpoints are silent.
            _warn_missing_reference_once(genome)
            continue
        candidates.extend(enumerate_candidates(
            contig=contig,
            sequence=ref_seq,
            sequence_start=start,
            variant=variant,
            score_fn=score_fn,
            **kwargs))
    candidates.sort(
        key=lambda c: (c.donor_score or 0) + (c.acceptor_score or 0),
        reverse=True)
    return candidates

varcode.score_donor(window: str) -> float

Score a 9-bp window as a candidate 5' splice site (donor). Returns a ratio in [0, 1]. Anchor mismatches (GT at +1/+2) force the score to 0, since biologically a donor without GT is not a donor under the canonical (U2-type) spliceosome.

Source code in varcode/cryptic_exons.py

def score_donor(window: str) -> float:
    """Score a 9-bp window as a candidate 5' splice site (donor).
    Returns a ratio in ``[0, 1]``. Anchor mismatches (GT at +1/+2)
    force the score to 0, since biologically a donor without GT is
    not a donor under the canonical (U2-type) spliceosome.
    """
    if len(window) != DONOR_WINDOW:
        return 0.0
    w = window.upper()
    # Anchor check: both GT positions must match.
    for pos in _DONOR_ANCHOR_POSITIONS:
        if w[pos] not in _DONOR_CONSENSUS[pos]:
            return 0.0
    matches = sum(
        1 for i, base in enumerate(w)
        if base in _DONOR_CONSENSUS[i])
    return matches / DONOR_WINDOW

varcode.score_acceptor(window: str) -> float

Score a 4-bp window as a candidate 3' splice site (acceptor). Returns a ratio in [0, 1]. Anchor mismatches (AG at -2/-1) force the score to 0.

Source code in varcode/cryptic_exons.py

def score_acceptor(window: str) -> float:
    """Score a 4-bp window as a candidate 3' splice site (acceptor).
    Returns a ratio in ``[0, 1]``. Anchor mismatches (AG at -2/-1)
    force the score to 0.
    """
    if len(window) != ACCEPTOR_WINDOW:
        return 0.0
    w = window.upper()
    for pos in _ACCEPTOR_ANCHOR_POSITIONS:
        if w[pos] not in _ACCEPTOR_CONSENSUS[pos]:
            return 0.0
    matches = sum(
        1 for i, base in enumerate(w)
        if base in _ACCEPTOR_CONSENSUS[i])
    return matches / ACCEPTOR_WINDOW

Protein collections

See protein collections and FASTA for varcode.to_proteome(), offline provenance, alternatives and reference composition.

Priority ordering

varcode.effect_priority(effect)

Returns the integer priority for a given transcript effect.

Effects may opt out of class-based priority lookup by exposing a priority_class attribute — used by wrapper classes like :class:varcode.splice_outcomes.SpliceOutcomeSet to delegate to the wrapped effect's class.

Shared transcript comparisons rank by their most severe retained candidate. Selecting an effect never drops its other hypotheses.

Source code in varcode/effects/effect_ordering.py

def effect_priority(effect):
    """
    Returns the integer priority for a given transcript effect.

    Effects may opt out of class-based priority lookup by exposing a
    ``priority_class`` attribute — used by wrapper classes like
    :class:`varcode.splice_outcomes.SpliceOutcomeSet` to delegate to
    the wrapped effect's class.

    Shared transcript comparisons rank by their most severe retained
    candidate. Selecting an effect never drops its other hypotheses.
    """
    candidates = getattr(effect, "realized_candidates", None)
    if candidates:
        return max(_classification_priority(candidate.effect) for candidate in candidates)
    return _classification_priority(effect)

varcode.top_priority_effect(effects)

Given a collection of variant transcript effects, return the top priority object, retaining complete splice outcome sets. Raw ExonicSpliceSite effects select their higher-priority consequence. In cases where multiple transcripts give rise to multiple effects, use a variety of filtering and sorting heuristics to pick the canonical transcript.

Source code in varcode/effects/effect_ordering.py

def top_priority_effect(effects):
    """
    Given a collection of variant transcript effects,
    return the top priority object, retaining complete splice outcome sets.
    Raw ExonicSpliceSite effects select their higher-priority consequence.
    In cases where multiple transcripts give rise to multiple
    effects, use a variety of filtering and sorting heuristics to pick
    the canonical transcript.
    """
    if len(effects) == 0:
        raise ValueError("List of effects cannot be empty")

    effects = map(
        select_between_exonic_splice_site_and_alternate_effect,
        effects)

    effects_grouped_by_gene = apply_groupby(
        effects, fn=gene_id_of_associated_transcript, skip_none=False)

    if None in effects_grouped_by_gene:
        effects_without_genes = effects_grouped_by_gene.pop(None)
    else:
        effects_without_genes = []

    # if we had any effects associated with genes then choose one of those
    if len(effects_grouped_by_gene) > 0:
        effects_with_genes = [
            top_priority_effect_for_single_gene(gene_effects)
            for gene_effects in effects_grouped_by_gene.values()
        ]
        return max(effects_with_genes, key=multi_gene_effect_sort_key)
    else:
        # if all effects were without genes then choose the best among those
        assert len(effects_without_genes) > 0
        return max(effects_without_genes, key=multi_gene_effect_sort_key)