API reference¶

Region analysis¶

Analyze a protein region for aggregation-prone clusters.

This is the main analysis function that orchestrates the entire aggregation detection pipeline: 1. Apply all (or selected) rules to the region 2. Collect identified clusters 3. Merge overlapping clusters using Union-Find 4. Identify multi-rule convergence hotspots 5. Build comprehensive results structure

Parameters:

Name	Type	Description	Default
`sequence`	`str`	Full protein sequence (1-letter codes)	required
`start`	`int`	Region start position (1-indexed, inclusive)	required
`stop`	`int`	Region end position (1-indexed, inclusive)	required
`registry`	`RuleRegistry`	RuleRegistry with evaluators to use. If None, uses DEFAULT_REGISTRY.	`None`
`selected_rules`	`Optional[List[str]]`	Optional list of rule names to apply. If None, all registered rules are used.	`None`

Returns:

Type	Description
`Dict[str, Any]`	Dictionary with comprehensive analysis results:
`Dict[str, Any]`	{ 'region': (start, stop), 'sequence': region_sequence, 'length': region_length, 'rules': {rule_name: rule_results, ...}, 'merged_clusters': [cluster_dicts, ...], 'multi_rule_clusters': [cluster_dicts, ...], 'aggregation_hotspots': [positions, ...], 'all_cluster_positions': [positions, ...]
`Dict[str, Any]`	}

Example

seq = "ALVIFQNWYG" results = analyze_region(seq, 1, 10) results['region'] (1, 10) len(results['merged_clusters']) 2

Source code in src/aggressor/analysis/regions.py

def analyze_region(
        sequence: str,
        start: int,
        stop: int,
        registry: RuleRegistry = None,
        selected_rules: Optional[List[str]] = None
) -> Dict[str, Any]:
    """
    Analyze a protein region for aggregation-prone clusters.

    This is the main analysis function that orchestrates the entire
    aggregation detection pipeline:
    1. Apply all (or selected) rules to the region
    2. Collect identified clusters
    3. Merge overlapping clusters using Union-Find
    4. Identify multi-rule convergence hotspots
    5. Build comprehensive results structure

    Args:
        sequence: Full protein sequence (1-letter codes)
        start: Region start position (1-indexed, inclusive)
        stop: Region end position (1-indexed, inclusive)
        registry: RuleRegistry with evaluators to use.
                 If None, uses DEFAULT_REGISTRY.
        selected_rules: Optional list of rule names to apply.
                       If None, all registered rules are used.

    Returns:
        Dictionary with comprehensive analysis results:
        {
            'region': (start, stop),
            'sequence': region_sequence,
            'length': region_length,
            'rules': {rule_name: rule_results, ...},
            'merged_clusters': [cluster_dicts, ...],
            'multi_rule_clusters': [cluster_dicts, ...],
            'aggregation_hotspots': [positions, ...],
            'all_cluster_positions': [positions, ...]
        }

    Example:
        >>> seq = "ALVIFQNWYG"
        >>> results = analyze_region(seq, 1, 10)
        >>> results['region']
        (1, 10)
        >>> len(results['merged_clusters'])
        2
    """
    if registry is None:
        registry = DEFAULT_REGISTRY

    region_seq = sequence[start - 1:stop]

    logger.debug(f"Analyzing region {start}:{stop} ({len(region_seq)} residues)")
    logger.debug(f"Region sequence: {region_seq}")

    # Step 1: Evaluate all applicable rules
    rule_results = registry.evaluate_region(
        sequence, start, stop, selected_rules
    )

    # Step 2: Collect all clusters across rules
    all_clusters: List[Cluster] = []
    for rule_name, clusters in rule_results.items():
        all_clusters.extend(clusters)

    logger.debug(f"Total raw clusters found: {len(all_clusters)}")

    # Step 3: Merge overlapping clusters using Union-Find
    merged_clusters = merge_overlapping_clusters_unionfind(
        all_clusters, sequence, gap_tolerance=MAX_GAP_FOR_MERGING
    )

    logger.debug(f"Clusters after merging: {len(merged_clusters)}")

    # Step 4: Identify multi-rule convergence clusters
    multi_rule_clusters = [
        c for c in merged_clusters
        if isinstance(c, MultiRuleCluster)
    ]

    if multi_rule_clusters:
        logger.debug(
            f"Multi-rule convergence clusters: {len(multi_rule_clusters)}"
        )
        for mrc in multi_rule_clusters:
            logger.debug(
                f"  Rules: {mrc.rules}, "
                f"Score: {mrc.combined_aggregation_score}"
            )

    # Step 5: Collect all hotspot positions
    hotspot_positions: Set[int] = set()
    for cluster in merged_clusters:
        hotspot_positions.update(cluster.positions)

    # Step 6: Build comprehensive results structure
    results = {
        'region': (start, stop),
        'sequence': region_seq,
        'length': len(region_seq),
        'rules': {},
        'merged_clusters': [c.to_dict() for c in merged_clusters],
        'multi_rule_clusters': [c.to_dict() for c in multi_rule_clusters],
        'aggregation_hotspots': sorted(hotspot_positions),
        'all_cluster_positions': sorted(hotspot_positions),
    }

    # Step 7: Add per-rule details with backward-compatible fields
    for rule_name, clusters in rule_results.items():
        evaluator = registry.get(rule_name)

        rule_entry = {
            'description': evaluator.description,
            'qualifying_clusters': [c.to_dict() for c in clusters],
            'condition_met': len(clusters) > 0,
            'aggregation_score': evaluator.aggregation_score,
            # Backward compatibility fields
            'matching_positions': [],
            'matching_residues': [],
            'clusters': [],
        }

        # Populate backward-compatible fields
        for cluster in clusters:
            rule_entry['matching_positions'].extend(cluster.positions)
            rule_entry['matching_residues'].extend(cluster.residues)
            rule_entry['clusters'].append(list(cluster.positions))

        # Remove duplicates while preserving order
        rule_entry['matching_positions'] = list(
            dict.fromkeys(rule_entry['matching_positions'])
        )
        rule_entry['matching_residues'] = list(
            dict.fromkeys(rule_entry['matching_residues'])
        )

        # Special handling for hydrophobic_and_aromatic rule
        if rule_name == 'hydrophobic_and_aromatic' and clusters:
            rule_entry['special_clusters'] = [
                c.to_dict() for c in clusters
            ]

        results['rules'][rule_name] = rule_entry

    return results

Mutation engine¶

Generate mutated sequences with point mutations, rule-based mutations, and insertions.

Mutation types generated: - Direct point mutations at specified positions - Rule-based mutations in aggregation-prone regions - Insertions at specified positions - Edge positions receive additional gatekeeping mutations

All mutations are classified by structural context: BETA_CORE, GATEKEEPER, BOUNDARY, FLANKING, DIRECT, INSERTION

Parameters:

Name	Type	Description	Default
`sequence`	`str`	Original protein sequence	required
`positions`	`List[int]`	List of 1-indexed positions for direct mutations	required
`mutations`	`List[str]`	Amino acid codes to mutate to	required
`regions`	`List[str]`	List of region strings (e.g., ["10:50", "60:100"])	`None`
`selected_rules`	`List[str]`	Optional subset of rules to apply	`None`
`insertion_positions`	`List[int]`	Positions for insertions (before this position)	`None`
`insertion_aas`	`List[str]`	Amino acids to insert	`None`
`gatekeeping_aas`	`List[str]`	Gatekeeping amino acids for edge positions	`None`
`verbose`	`bool`	Enable debug logging	`False`

Returns:

Type	Description
`List[Tuple[str, str]]`	Tuple of:
`List[Dict]`	List of (description, sequence) tuples sorted by aggregation score
`Tuple[List[Tuple[str, str]], List[Dict]]`	List of region analysis dictionaries

Raises:

Type	Description
`ValueError`	If positions are out of range or insertion args mismatch

Source code in src/aggressor/mutagenesis/engine.py

def mutate_sequence(
        sequence: str,
        positions: List[int],
        mutations: List[str],
        regions: List[str] = None,
        selected_rules: List[str] = None,
        insertion_positions: List[int] = None,
        insertion_aas: List[str] = None,
        gatekeeping_aas: List[str] = None,
        verbose: bool = False,
        max_gatekeepers_per_apr: Optional[int] = None,
) -> Tuple[List[Tuple[str, str]], List[Dict]]:
    """
    Generate mutated sequences with point mutations, rule-based mutations, and insertions.

    Mutation types generated:
    - Direct point mutations at specified positions
    - Rule-based mutations in aggregation-prone regions
    - Insertions at specified positions
    - Edge positions receive additional gatekeeping mutations

    All mutations are classified by structural context:
    BETA_CORE, GATEKEEPER, BOUNDARY, FLANKING, DIRECT, INSERTION

    Args:
        sequence: Original protein sequence
        positions: List of 1-indexed positions for direct mutations
        mutations: Amino acid codes to mutate to
        regions: List of region strings (e.g., ["10:50", "60:100"])
        selected_rules: Optional subset of rules to apply
        insertion_positions: Positions for insertions (before this position)
        insertion_aas: Amino acids to insert
        gatekeeping_aas: Gatekeeping amino acids for edge positions
        verbose: Enable debug logging

    Returns:
        Tuple of:
        - List of (description, sequence) tuples sorted by aggregation score
        - List of region analysis dictionaries

    Raises:
        ValueError: If positions are out of range or insertion args mismatch
    """
    all_results = []
    region_analyses = []
    seq_len = len(sequence)

    if gatekeeping_aas is None:
        gatekeeping_aas = GATEKEEPING_AAS

    # Validate positions
    for pos in positions:
        if pos < 1 or pos > seq_len:
            raise ValueError(
                f"Position {pos} is out of range "
                f"(sequence length: {seq_len})"
            )

    # ============ RULE-BASED MUTATIONS ============

    if regions:
        for region_str in regions:
            start, stop = _parse_region(region_str, seq_len)
            analysis = analyze_region(
                sequence, start, stop, selected_rules=selected_rules
            )
            region_analyses.append(analysis)

            if verbose:
                logger.debug(f"Analyzing region {start}:{stop}")
                logger.debug(f"Sequence: {analysis['sequence']}")
                logger.debug(f"Length: {analysis['length']} residues")

                for rule_name, rule_data in analysis['rules'].items():
                    if rule_data['condition_met']:
                        logger.debug(
                            f"  {rule_name}: "
                            f"{len(rule_data['qualifying_clusters'])} clusters"
                        )

            # Generate rule-based mutations
            rule_mutations = apply_rule_mutations(
                sequence, analysis, mutations,
                selected_rules, gatekeeping_aas,
                max_gatekeepers_per_apr=max_gatekeepers_per_apr,
            )
            all_results.extend(rule_mutations)

            if verbose:
                logger.debug(
                    f"Generated {len(rule_mutations)} rule-based mutations "
                    f"for region {start}:{stop}"
                )

    # ============ DIRECT POINT MUTATIONS ============

    for pos in positions:
        original_aa = sequence[pos - 1]

        for new_aa in mutations:
            if new_aa == original_aa:
                continue

            mutated_seq = create_mutated_sequence(sequence, pos, new_aa)
            description = (
                f"{original_aa}{pos}{new_aa} | "
                f"Direct mutation | DIRECT (agg_score=0)"
            )
            all_results.append(
                (description, mutated_seq, 0, MutationType.DIRECT)
            )

    # ============ INSERTIONS ============

    if insertion_positions and insertion_aas:
        if len(insertion_positions) != len(insertion_aas):
            raise ValueError(
                "insertion_positions and insertion_aas "
                "must have the same length"
            )

        for ins_pos, ins_aa in zip(insertion_positions, insertion_aas):
            if ins_pos < 1 or ins_pos > seq_len + 1:
                raise ValueError(
                    f"Insertion position {ins_pos} is out of range "
                    f"(1-{seq_len + 1})"
                )

            mutated_seq = (
                    sequence[:ins_pos - 1] + ins_aa + sequence[ins_pos - 1:]
            )
            description = (
                f"Insertion: {ins_aa} inserted before position {ins_pos} | "
                f"INSERTION (agg_score=0)"
            )
            all_results.append(
                (description, mutated_seq, 0, MutationType.INSERTION)
            )

    # ============ SORT AND FORMAT ============

    # Sort by aggregation score (descending), with a deterministic
    # secondary key (description) so ties never depend on hash ordering.
    all_results.sort(key=lambda x: (-x[2], x[0]))

    # Convert to expected format (description, sequence)
    sorted_results = [(desc, seq) for desc, seq, _, _ in all_results]

    return sorted_results, region_analyses

Apply mutations based on aggregation rules with biological classification.

Algorithm: 1. Filter rules if specific ones selected 2. Collect all qualifying clusters 3. Merge overlapping clusters 4. For each position in merged clusters: - Determine if edge position - Apply standard + gatekeeping mutations as appropriate - Classify mutation type (GATEKEEPER, BETA_CORE, BOUNDARY, FLANKING)

Gatekeeper placement

By default every APR boundary position receives the gatekeeping amino acids in addition to the standard mutation set. This reflects the biological reality that the number of effective gatekeeper slots is a structural property of the APR (its two flanks), not a free parameter. When max_gatekeepers_per_apr is set, the boundary slots are ranked by predicted aggregation-propensity reduction (Tartaglia scale) and only the top-N are gatekept — an experimental design budget control, not a change to the underlying biology.

Parameters:

Name	Type	Description	Default
`sequence`	`str`	Full protein sequence	required
`region_analysis`	`Dict`	Analysis results from analyze_region()	required
`mutations`	`List[str]`	Standard mutation amino acids	required
`selected_rules`	`List[str]`	Optional subset of rules to apply	`None`
`gatekeeping_aas`	`List[str]`	Gatekeeping amino acids for edge positions	`None`
`gatekeeper_distance`	`int`	Distance threshold for gatekeeper classification	`GATEKEEPER_DISTANCE`
`max_gatekeepers_per_apr`	`Optional[int]`	Optional cap on gatekept boundary slots per APR	`None`

Returns:

Type	Description
`List[Tuple[str, str, int, MutationType]]`	List of (description, sequence, agg_score, mutation_type) tuples

Source code in src/aggressor/mutagenesis/engine.py

def apply_rule_mutations(
        sequence: str,
        region_analysis: Dict,
        mutations: List[str],
        selected_rules: List[str] = None,
        gatekeeping_aas: List[str] = None,
        gatekeeper_distance: int = GATEKEEPER_DISTANCE,
        max_gatekeepers_per_apr: Optional[int] = None,
) -> List[Tuple[str, str, int, MutationType]]:
    """
    Apply mutations based on aggregation rules with biological classification.

    Algorithm:
    1. Filter rules if specific ones selected
    2. Collect all qualifying clusters
    3. Merge overlapping clusters
    4. For each position in merged clusters:
       - Determine if edge position
       - Apply standard + gatekeeping mutations as appropriate
       - Classify mutation type (GATEKEEPER, BETA_CORE, BOUNDARY, FLANKING)

    Gatekeeper placement:
        By default every APR boundary position receives the gatekeeping
        amino acids in addition to the standard mutation set. This reflects
        the biological reality that the number of effective gatekeeper slots
        is a *structural* property of the APR (its two flanks), not a free
        parameter. When ``max_gatekeepers_per_apr`` is set, the boundary
        slots are ranked by predicted aggregation-propensity reduction
        (Tartaglia scale) and only the top-N are gatekept — an experimental
        design budget control, not a change to the underlying biology.

    Args:
        sequence: Full protein sequence
        region_analysis: Analysis results from analyze_region()
        mutations: Standard mutation amino acids
        selected_rules: Optional subset of rules to apply
        gatekeeping_aas: Gatekeeping amino acids for edge positions
        gatekeeper_distance: Distance threshold for gatekeeper classification
        max_gatekeepers_per_apr: Optional cap on gatekept boundary slots per APR

    Returns:
        List of (description, sequence, agg_score, mutation_type) tuples
    """
    results = []

    if gatekeeping_aas is None:
        gatekeeping_aas = GATEKEEPING_AAS

    # Combine user-specified and canonical gatekeepers
    gatekeeper_aa_set = (
            set(aa.upper() for aa in gatekeeping_aas) |
            CANONICAL_GATEKEEPER_AAS
    )

    # Filter rules if specific ones are selected
    if selected_rules:
        rules_to_apply = {
            k: v for k, v in region_analysis['rules'].items()
            if k in selected_rules
        }
    else:
        rules_to_apply = region_analysis['rules']

    region_start, region_end = region_analysis['region']

    # Collect all clusters from all rules
    cluster_objects = []
    for rule_name, rule_data in rules_to_apply.items():
        if not rule_data['condition_met']:
            continue

        for cluster_dict in rule_data['qualifying_clusters']:
            cluster_objects.append(create_cluster(
                positions=cluster_dict['positions'],
                sequence=sequence,
                rule_name=cluster_dict.get('rule_name', rule_name),
                aggregation_score=cluster_dict.get(
                    'aggregation_score',
                    rule_data['aggregation_score']
                )
            ))

    # Merge overlapping clusters
    merged_clusters = merge_overlapping_clusters_unionfind(
        cluster_objects, sequence, gap_tolerance=MAX_GAP_FOR_MERGING
    )

    # Track mutated positions to avoid duplicates
    mutated_positions: Set[int] = set()

    # Apply mutations to each merged cluster
    for cluster in merged_clusters:
        positions_to_mutate = list(cluster.positions)

        # Get cluster boundaries
        cluster_min = min(positions_to_mutate)
        cluster_max = max(positions_to_mutate)

        # Decide which boundary slots receive gatekeeping AAs. With no cap
        # this is exactly {cluster_min, cluster_max}; with a cap it is the
        # propensity-ranked subset chosen by the gatekeepers module.
        gatekept_positions = select_gatekeeper_positions(
            cluster_positions=positions_to_mutate,
            sequence=sequence,
            gatekeeping_aas=gatekeeper_aa_set,
            max_gatekeepers=max_gatekeepers_per_apr,
        )

        # Get aggregation score and rule description
        if isinstance(cluster, MultiRuleCluster):
            agg_score = cluster.combined_aggregation_score
            rule_desc = f"MERGED RULES {'+'.join(cluster.rules)}"
        else:
            agg_score = cluster.aggregation_score
            rule_desc = f"Rule '{cluster.rule_name}'"

        for pos in positions_to_mutate:
            if pos in mutated_positions:
                continue

            mutated_positions.add(pos)
            original_aa = sequence[pos - 1]

            # Get applicable mutations (order-preserving dedup keeps output
            # deterministic; set() previously made emission order vary per run)
            if pos in gatekept_positions:
                all_mutations = _dedup_preserve_order(mutations + gatekeeping_aas)
            else:
                all_mutations = list(mutations)

            for new_aa in all_mutations:
                if new_aa == original_aa:
                    continue

                mutated_seq = create_mutated_sequence(sequence, pos, new_aa)

                # Classify mutation type
                mutation_type = classify_mutation_type(
                    position=pos,
                    new_aa=new_aa,
                    cluster_positions=positions_to_mutate,
                    cluster_min=cluster_min,
                    cluster_max=cluster_max,
                    gatekeeper_aas=gatekeeper_aa_set
                )

                # Build description with type label
                type_label = mutation_type.name
                description = (
                    f"{original_aa}{pos}{new_aa} | {rule_desc} | "
                    f"{type_label} (agg_score={agg_score})"
                )

                results.append(
                    (description, mutated_seq, agg_score, mutation_type)
                )

    return results

Gatekeepers¶

Gatekeeper-slot detection and optimal selection.

Biological rationale¶

Aggregation-prone regions (APRs) form the cross-beta core of amyloid fibrils. Natural proteins suppress this by enriching gatekeeper residues at the APR flanks (Rousseau et al., J Mol Biol 2006; Beerten et al., FEBS Lett 2012; Reumers/Maurer-Stroh and colleagues on sequence-encoded aggregation safeguards). Two mechanisms dominate:

Charged residues (D, E, K, R) impose electrostatic repulsion between molecules stacking in-register in the fibril, and raise local solubility.
Proline is a beta-strand breaker: its backbone cannot donate an amide hydrogen bond and its phi is conformationally restricted, so it truncates beta-sheet propagation.

A key consequence for design: the number of effective gatekeeper slots is a structural property of the APR — essentially its two boundaries (and the immediately flanking positions) — not a free integer the user picks. This module therefore makes the automatic choice the primary mode:

Detect whether a gatekeeper substitution at a boundary is even useful (skip positions already occupied by a gatekeeper residue — "is it possible / needed").
When the caller supplies a budget cap, rank candidate slots by the predicted reduction in intrinsic aggregation propensity (Tartaglia et al., J Mol Biol 2008 scale) and keep the highest-value ones.

Public API¶

select_gatekeeper_positions(...) -> Set[int] # boundary slots to gatekeep flanking_gatekeeper_positions(...) -> List[int] # conservative out-of-APR flanks rank_gatekeeper_slots(...) -> List[(pos, gain)] best_gatekeeper_for_position(...) -> Optional[str]

`best_gatekeeper_for_position(position, sequence, gatekeeping_aas)` ¶

Choose the single most propensity-lowering gatekeeper AA for a position.

Returns the gatekeeper amino acid that yields the lowest intrinsic aggregation propensity at this site, or None if the original residue is already a gatekeeper (substitution would not help).

Example

best_gatekeeper_for_position(1, "I", {"P", "K", "R", "D", "E"}) 'K'

Source code in src/aggressor/analysis/gatekeepers.py

def best_gatekeeper_for_position(
        position: int,
        sequence: str,
        gatekeeping_aas: Iterable[str],
) -> Optional[str]:
    """
    Choose the single most propensity-lowering gatekeeper AA for a position.

    Returns the gatekeeper amino acid that yields the lowest intrinsic
    aggregation propensity at this site, or None if the original residue is
    already a gatekeeper (substitution would not help).

    Example:
        >>> best_gatekeeper_for_position(1, "I", {"P", "K", "R", "D", "E"})
        'K'
    """
    original = sequence[position - 1].upper()
    if _is_gatekeeper_residue(original, gatekeeping_aas):
        return None
    candidates = [g.upper() for g in gatekeeping_aas if g.upper() != original]
    if not candidates:
        return None
    # Lowest propensity wins (most aggregation-suppressing)
    return min(candidates, key=lambda g: AGGREGATION_PROPENSITY.get(g, 0.0))

`rank_gatekeeper_slots(candidate_positions, sequence, gatekeeping_aas)` ¶

Rank candidate gatekeeper positions by predicted propensity reduction.

The "gain" at a position is (propensity of the original residue) minus (propensity of the best gatekeeper substitution). High-propensity boundary residues (I, V, L, F) yield the largest gains and are therefore the most valuable gatekeeper slots.

Positions whose residue is already a gatekeeper are excluded.

Returns:

Type	Description
`List[Tuple[int, float]]`	List of (position, gain) sorted by gain descending, then position
`List[Tuple[int, float]]`	ascending (deterministic tie-break).

Source code in src/aggressor/analysis/gatekeepers.py

def rank_gatekeeper_slots(
        candidate_positions: List[int],
        sequence: str,
        gatekeeping_aas: Iterable[str],
) -> List[Tuple[int, float]]:
    """
    Rank candidate gatekeeper positions by predicted propensity reduction.

    The "gain" at a position is (propensity of the original residue) minus
    (propensity of the best gatekeeper substitution). High-propensity
    boundary residues (I, V, L, F) yield the largest gains and are therefore
    the most valuable gatekeeper slots.

    Positions whose residue is already a gatekeeper are excluded.

    Returns:
        List of (position, gain) sorted by gain descending, then position
        ascending (deterministic tie-break).
    """
    scored: List[Tuple[int, float]] = []
    for pos in candidate_positions:
        original = sequence[pos - 1].upper()
        gk = best_gatekeeper_for_position(pos, sequence, gatekeeping_aas)
        if gk is None:
            continue
        gain = AGGREGATION_PROPENSITY.get(original, 0.0) - AGGREGATION_PROPENSITY.get(gk, 0.0)
        scored.append((pos, gain))
    scored.sort(key=lambda t: (-t[1], t[0]))
    return scored

`select_gatekeeper_positions(cluster_positions, sequence, gatekeeping_aas=CANONICAL_GATEKEEPER_AAS, max_gatekeepers=None)` ¶

Decide which APR boundary positions should receive gatekeeping AAs.

Default (max_gatekeepers is None): both cluster boundaries — i.e. {min, max} — exactly the automatic behaviour. This is intentional: the two flanks are the biologically meaningful gatekeeper slots, so the count is detected from APR geometry, not requested by the user.

With a cap: the boundary slots are ranked by predicted propensity reduction and only the top max_gatekeepers are returned. Slots whose residue is already a gatekeeper are skipped, so the cap is applied to useful slots only.

Parameters:

Name	Type	Description	Default
`cluster_positions`	`List[int]`	All positions in the merged APR cluster.	required
`sequence`	`str`	Full protein sequence (for residue lookup).	required
`gatekeeping_aas`	`Iterable[str]`	Allowed gatekeeper amino acids.	`CANONICAL_GATEKEEPER_AAS`
`max_gatekeepers`	`Optional[int]`	Optional budget cap on gatekept slots per APR.	`None`

Returns:

Type	Description
`Set[int]`	Set of 1-indexed positions that should receive gatekeeping AAs.

Example

sorted(select_gatekeeper_positions([5, 6, 7], "AAAAIFLAA")) [5, 7]

Source code in src/aggressor/analysis/gatekeepers.py

def select_gatekeeper_positions(
        cluster_positions: List[int],
        sequence: str,
        gatekeeping_aas: Iterable[str] = CANONICAL_GATEKEEPER_AAS,
        max_gatekeepers: Optional[int] = None,
) -> Set[int]:
    """
    Decide which APR boundary positions should receive gatekeeping AAs.

    Default (``max_gatekeepers is None``): both cluster boundaries — i.e.
    {min, max} — exactly the automatic behaviour. This is intentional: the
    two flanks are the biologically meaningful gatekeeper slots, so the
    count is detected from APR geometry, not requested by the user.

    With a cap: the boundary slots are ranked by predicted propensity
    reduction and only the top ``max_gatekeepers`` are returned. Slots whose
    residue is already a gatekeeper are skipped, so the cap is applied to
    *useful* slots only.

    Args:
        cluster_positions: All positions in the merged APR cluster.
        sequence: Full protein sequence (for residue lookup).
        gatekeeping_aas: Allowed gatekeeper amino acids.
        max_gatekeepers: Optional budget cap on gatekept slots per APR.

    Returns:
        Set of 1-indexed positions that should receive gatekeeping AAs.

    Example:
        >>> sorted(select_gatekeeper_positions([5, 6, 7], "AAAAIFLAA"))
        [5, 7]
    """
    if not cluster_positions:
        return set()

    cluster_min = min(cluster_positions)
    cluster_max = max(cluster_positions)
    boundary_slots = {cluster_min, cluster_max}

    if max_gatekeepers is None:
        # Automatic mode: geometry decides the count.
        return boundary_slots

    if max_gatekeepers <= 0:
        return set()

    ranked = rank_gatekeeper_slots(sorted(boundary_slots), sequence, gatekeeping_aas)
    chosen = {pos for pos, _ in ranked[:max_gatekeepers]}
    if not chosen:
        logger.debug(
            "No useful gatekeeper slot at APR %s:%s "
            "(boundaries already gatekeeper residues)",
            cluster_min, cluster_max,
        )
    return chosen

`flanking_gatekeeper_positions(cluster_positions, region_start, region_end)` ¶

Return the positions immediately flanking an APR (one N-terminal, one C-terminal), clamped to the analysis region.

This supports the conservative gatekeeper-design strategy in which a charge is introduced just outside the APR core, leaving the cross-beta register intact while walling it off electrostatically — distinct from substituting a terminal core residue. Exposed for downstream/expert use; not part of the default mutation path so that standard output stays backward-compatible.

Source code in src/aggressor/analysis/gatekeepers.py

def flanking_gatekeeper_positions(
        cluster_positions: List[int],
        region_start: int,
        region_end: int,
) -> List[int]:
    """
    Return the positions immediately flanking an APR (one N-terminal, one
    C-terminal), clamped to the analysis region.

    This supports the conservative gatekeeper-design strategy in which a
    charge is introduced *just outside* the APR core, leaving the cross-beta
    register intact while walling it off electrostatically — distinct from
    substituting a terminal core residue. Exposed for downstream/expert use;
    not part of the default mutation path so that standard output stays
    backward-compatible.
    """
    if not cluster_positions:
        return []
    flanks = []
    left = min(cluster_positions) - 1
    right = max(cluster_positions) + 1
    if left >= region_start:
        flanks.append(left)
    if right <= region_end:
        flanks.append(right)
    return flanks

Data models¶

Bases: Enum

Classification of mutations by structural/functional context.

Based on the gatekeeper hypothesis (Rousseau et al., 2006) and aggregation-prone region (APR) architecture.

`from_description(description)` `classmethod` ¶

Parse mutation type from a description string.

Uses keyword matching to determine the mutation type from the formatted description strings generated by the pipeline.

Parameters:

Name	Type	Description	Default
`description`	`str`	Formatted mutation description string	required

Returns:

Type	Description
`MutationType`	Corresponding MutationType enum value

Examples:

>>> MutationType.from_description("A10P | GATEKEEPER (agg_score=3)")
<MutationType.GATEKEEPER: 2>

Source code in src/aggressor/core/models.py

@classmethod
def from_description(cls, description: str) -> 'MutationType':
    """
    Parse mutation type from a description string.

    Uses keyword matching to determine the mutation type from
    the formatted description strings generated by the pipeline.

    Args:
        description: Formatted mutation description string

    Returns:
        Corresponding MutationType enum value

    Examples:
        >>> MutationType.from_description("A10P | GATEKEEPER (agg_score=3)")
        <MutationType.GATEKEEPER: 2>
    """
    if 'GATEKEEPER' in description:
        return cls.GATEKEEPER
    elif 'BOUNDARY' in description:
        return cls.BOUNDARY
    elif 'FLANKING' in description:
        return cls.FLANKING
    elif 'Direct mutation' in description:
        return cls.DIRECT
    elif 'Insertion' in description:
        return cls.INSERTION
    elif 'CORE' in description:
        return cls.BETA_CORE
    elif "Rule '" in description or 'MERGED RULES' in description:
        return cls.BETA_CORE
    return cls.UNKNOWN

`str()` ¶

Human-readable string representation.

Source code in src/aggressor/core/models.py

def __str__(self) -> str:
    """Human-readable string representation."""
    return self.name

Immutable representation of an aggregation-prone cluster identified by a single rule.

Using frozen=True enables: - Hashability (can be used in sets, as dict keys) - Thread safety (no mutation possible after creation) - Semantic clarity (clusters are identified, not modified)

Using slots=True reduces memory footprint by ~40% per instance compared to standard dataclass with dict.

Attributes:

Name	Type	Description
`positions`	`Tuple[int, ...]`	1-indexed positions of residues in the cluster (must be sorted)
`residues`	`Tuple[str, ...]`	Amino acid single-letter codes at each position
`rule_name`	`str`	Name of the rule that identified this cluster
`aggregation_score`	`int`	Integer score indicating aggregation propensity weight

Raises:

Type	Description
`ValueError`	If positions and residues have different lengths, if positions are empty, or if positions are not in sorted order

Source code in src/aggressor/core/models.py

@dataclass(frozen=True, slots=True)
class Cluster:
    """
    Immutable representation of an aggregation-prone cluster identified by a single rule.

    Using frozen=True enables:
    - Hashability (can be used in sets, as dict keys)
    - Thread safety (no mutation possible after creation)
    - Semantic clarity (clusters are identified, not modified)

    Using slots=True reduces memory footprint by ~40% per instance compared
    to standard dataclass with __dict__.

    Attributes:
        positions: 1-indexed positions of residues in the cluster (must be sorted)
        residues: Amino acid single-letter codes at each position
        rule_name: Name of the rule that identified this cluster
        aggregation_score: Integer score indicating aggregation propensity weight

    Raises:
        ValueError: If positions and residues have different lengths,
                   if positions are empty, or if positions are not in sorted order
    """
    positions: Tuple[int, ...]
    residues: Tuple[str, ...]
    rule_name: str
    aggregation_score: int

    def __post_init__(self):
        """Validate cluster invariants after initialization."""
        if len(self.positions) != len(self.residues):
            raise ValueError(
                f"Position count ({len(self.positions)}) must match "
                f"residue count ({len(self.residues)})"
            )
        if not self.positions:
            raise ValueError("Cluster must contain at least one position")

        if any(p < 1 for p in self.positions):
            raise ValueError("Positions must be 1-indexed (≥1)")
        if list(self.positions) != sorted(self.positions):
            raise ValueError("Positions must be in sorted order")

    @property
    def size(self) -> int:
        """Number of residues in the cluster."""
        return len(self.positions)

    @property
    def span(self) -> int:
        """Sequence span from first to last position (inclusive)."""
        return max(self.positions) - min(self.positions) + 1

    @property
    def density(self) -> float:
        """
        Cluster density: occupied positions / total span.

        Higher density indicates tighter packing of aggregation-prone
        residues in the primary sequence, which correlates with
        stronger aggregation propensity.
        """
        return self.size / self.span if self.span > 0 else 0.0

    @property
    def motif(self) -> str:
        """The cluster residues concatenated as a sequence motif."""
        return ''.join(self.residues)

    @property
    def mean_propensity(self) -> float:
        """
        Average intrinsic aggregation propensity of cluster residues.

        Uses the Tartaglia et al. (2008) aggregation propensity scale.
        Higher values indicate stronger predicted aggregation tendency.
        """
        propensities = [AGGREGATION_PROPENSITY.get(aa, 0.0) for aa in self.residues]
        return sum(propensities) / len(propensities) if propensities else 0.0

    @property
    def min_position(self) -> int:
        """N-terminal boundary of the cluster."""
        return min(self.positions)

    @property
    def max_position(self) -> int:
        """C-terminal boundary of the cluster."""
        return max(self.positions)

    def overlaps_with(self, other: 'Cluster', gap_tolerance: int = 0) -> bool:
        """
        Check if this cluster overlaps with another cluster.

        Two clusters overlap if their position ranges intersect or
        are within gap_tolerance positions of each other.

        Args:
            other: Another cluster to check for overlap
            gap_tolerance: Maximum allowed gap between clusters 
                          to still consider them overlapping

        Returns:
            True if clusters overlap or are within gap_tolerance

        Examples:
            >>> c1 = Cluster((1,2,3), ('A','L','V'), 'test', 1)
            >>> c2 = Cluster((4,5), ('I','F'), 'test', 1)
            >>> c1.overlaps_with(c2, gap_tolerance=1)
            True
            >>> c1.overlaps_with(c2, gap_tolerance=0)
            False
        """
        self_min, self_max = self.min_position, self.max_position
        other_min, other_max = other.min_position, other.max_position

        return not (
                self_max + gap_tolerance < other_min or
                other_max + gap_tolerance < self_min
        )

    def to_dict(self) -> Dict[str, Any]:
        """
        Convert to dictionary for serialization and backward compatibility.

        Returns:
            Dictionary with all cluster properties
        """
        return {
            'positions': list(self.positions),
            'residues': list(self.residues),
            'size': self.size,
            'span': self.span,
            'rule_name': self.rule_name,
            'aggregation_score': self.aggregation_score,
            'density': self.density,
            'mean_propensity': self.mean_propensity,
        }

`size` `property` ¶

Number of residues in the cluster.

`span` `property` ¶

Sequence span from first to last position (inclusive).

`density` `property` ¶

Cluster density: occupied positions / total span.

Higher density indicates tighter packing of aggregation-prone residues in the primary sequence, which correlates with stronger aggregation propensity.

`motif` `property` ¶

The cluster residues concatenated as a sequence motif.

`mean_propensity` `property` ¶

Average intrinsic aggregation propensity of cluster residues.

Uses the Tartaglia et al. (2008) aggregation propensity scale. Higher values indicate stronger predicted aggregation tendency.

`min_position` `property` ¶

N-terminal boundary of the cluster.

`max_position` `property` ¶

C-terminal boundary of the cluster.

`__post_init__()` ¶

Validate cluster invariants after initialization.

Source code in src/aggressor/core/models.py

def __post_init__(self):
    """Validate cluster invariants after initialization."""
    if len(self.positions) != len(self.residues):
        raise ValueError(
            f"Position count ({len(self.positions)}) must match "
            f"residue count ({len(self.residues)})"
        )
    if not self.positions:
        raise ValueError("Cluster must contain at least one position")

    if any(p < 1 for p in self.positions):
        raise ValueError("Positions must be 1-indexed (≥1)")
    if list(self.positions) != sorted(self.positions):
        raise ValueError("Positions must be in sorted order")

`overlaps_with(other, gap_tolerance=0)` ¶

Check if this cluster overlaps with another cluster.

Two clusters overlap if their position ranges intersect or are within gap_tolerance positions of each other.

Parameters:

Name	Type	Description	Default
`other`	`Cluster`	Another cluster to check for overlap	required
`gap_tolerance`	`int`	Maximum allowed gap between clusters to still consider them overlapping	`0`

Returns:

Type	Description
`bool`	True if clusters overlap or are within gap_tolerance

Examples:

>>> c1 = Cluster((1,2,3), ('A','L','V'), 'test', 1)
>>> c2 = Cluster((4,5), ('I','F'), 'test', 1)
>>> c1.overlaps_with(c2, gap_tolerance=1)
True
>>> c1.overlaps_with(c2, gap_tolerance=0)
False

Source code in src/aggressor/core/models.py

def overlaps_with(self, other: 'Cluster', gap_tolerance: int = 0) -> bool:
    """
    Check if this cluster overlaps with another cluster.

    Two clusters overlap if their position ranges intersect or
    are within gap_tolerance positions of each other.

    Args:
        other: Another cluster to check for overlap
        gap_tolerance: Maximum allowed gap between clusters 
                      to still consider them overlapping

    Returns:
        True if clusters overlap or are within gap_tolerance

    Examples:
        >>> c1 = Cluster((1,2,3), ('A','L','V'), 'test', 1)
        >>> c2 = Cluster((4,5), ('I','F'), 'test', 1)
        >>> c1.overlaps_with(c2, gap_tolerance=1)
        True
        >>> c1.overlaps_with(c2, gap_tolerance=0)
        False
    """
    self_min, self_max = self.min_position, self.max_position
    other_min, other_max = other.min_position, other.max_position

    return not (
            self_max + gap_tolerance < other_min or
            other_max + gap_tolerance < self_min
    )

`to_dict()` ¶

Convert to dictionary for serialization and backward compatibility.

Returns:

Type	Description
`Dict[str, Any]`	Dictionary with all cluster properties

Source code in src/aggressor/core/models.py

def to_dict(self) -> Dict[str, Any]:
    """
    Convert to dictionary for serialization and backward compatibility.

    Returns:
        Dictionary with all cluster properties
    """
    return {
        'positions': list(self.positions),
        'residues': list(self.residues),
        'size': self.size,
        'span': self.span,
        'rule_name': self.rule_name,
        'aggregation_score': self.aggregation_score,
        'density': self.density,
        'mean_propensity': self.mean_propensity,
    }

API reference¶

Region analysis¶

Mutation engine¶

Gatekeepers¶

Biological rationale¶

Public API¶

best_gatekeeper_for_position(position, sequence, gatekeeping_aas) ¶

rank_gatekeeper_slots(candidate_positions, sequence, gatekeeping_aas) ¶

select_gatekeeper_positions(cluster_positions, sequence, gatekeeping_aas=CANONICAL_GATEKEEPER_AAS, max_gatekeepers=None) ¶

flanking_gatekeeper_positions(cluster_positions, region_start, region_end) ¶

Data models¶

from_description(description) classmethod ¶

__str__() ¶

size property ¶

span property ¶

density property ¶

motif property ¶

mean_propensity property ¶

min_position property ¶

max_position property ¶

__post_init__() ¶

overlaps_with(other, gap_tolerance=0) ¶

to_dict() ¶

`best_gatekeeper_for_position(position, sequence, gatekeeping_aas)` ¶

`rank_gatekeeper_slots(candidate_positions, sequence, gatekeeping_aas)` ¶

`select_gatekeeper_positions(cluster_positions, sequence, gatekeeping_aas=CANONICAL_GATEKEEPER_AAS, max_gatekeepers=None)` ¶

`flanking_gatekeeper_positions(cluster_positions, region_start, region_end)` ¶

`from_description(description)` `classmethod` ¶

`str()` ¶

`size` `property` ¶

`span` `property` ¶

`density` `property` ¶

`motif` `property` ¶

`mean_propensity` `property` ¶

`min_position` `property` ¶

`max_position` `property` ¶

`__post_init__()` ¶

`overlaps_with(other, gap_tolerance=0)` ¶

`to_dict()` ¶