weighter

weighter

Pattern weighting and priority calculation.

This module implements the pattern weighting system designed in Movement III: - Combined recency + effectiveness weighting (CV 0.80) - 10% monthly decay without confirmation - Effectiveness threshold enforcement (0.3) - Uncertainty classification (epistemic vs aleatoric)

The weighter calculates priority scores for patterns, determining which patterns are most relevant for application in future executions.

Attributes

DEFAULT_DECAY_RATE_PER_MONTH module-attribute

DEFAULT_DECAY_RATE_PER_MONTH = 0.1

Fraction of priority lost per month (10%).

DEFAULT_EFFECTIVENESS_THRESHOLD module-attribute

DEFAULT_EFFECTIVENESS_THRESHOLD = 0.3

Below this effectiveness score, patterns are deprecated.

DEFAULT_EPISTEMIC_THRESHOLD module-attribute

DEFAULT_EPISTEMIC_THRESHOLD = 0.4

Variance threshold for classifying uncertainty as epistemic vs aleatoric.

DEFAULT_MIN_APPLICATIONS module-attribute

DEFAULT_MIN_APPLICATIONS = 3

Minimum applications before using actual effectiveness rate.

DEFAULT_FREQUENCY_BASE module-attribute

DEFAULT_FREQUENCY_BASE = 100

Log base for frequency normalization factor.

DEFAULT_FREQUENCY_FLOOR module-attribute

DEFAULT_FREQUENCY_FLOOR = 0.6

Minimum frequency factor to prevent single-occurrence patterns from being crushed below query thresholds. Without this, occ=1 produces frequency ≈ 0.15, yielding priority ≈ 0.075 — invisible at exploitation_threshold=0.3. The floor of 0.6 ensures single-occurrence patterns land at priority ≈ effectiveness × 0.6, keeping them queryable. Must match FREQUENCY_FACTOR_FLOOR in patterns_crud.py.

Classes

PatternWeighter

PatternWeighter(*, decay_rate_per_month=DEFAULT_DECAY_RATE_PER_MONTH, effectiveness_threshold=DEFAULT_EFFECTIVENESS_THRESHOLD, epistemic_threshold=DEFAULT_EPISTEMIC_THRESHOLD, min_applications_for_effectiveness=DEFAULT_MIN_APPLICATIONS, frequency_normalization_base=DEFAULT_FREQUENCY_BASE)

Calculates priority scores for patterns.

Implements the combined recency + effectiveness weighting algorithm as specified in the Movement III design document.

The priority formula is

priority = ( effectiveness_score × recency_factor × frequency_factor × (1 - variance) )

Where

• effectiveness_score = successes / (successes + failures + 1)
• recency_factor = (1 - decay_rate) ^ months_since_last_confirmed
• frequency_factor = min(1.0, log(occurrence_count + 1) / log(100))
• variance = std_dev(pattern_application_outcomes)

Initialize the pattern weighter.

Parameters:

Name	Type	Description	Default
decay_rate_per_month	float	Fraction of priority lost per month.	DEFAULT_DECAY_RATE_PER_MONTH
effectiveness_threshold	float	Below this, patterns are deprecated.	DEFAULT_EFFECTIVENESS_THRESHOLD
epistemic_threshold	float	Variance threshold for learnable patterns.	DEFAULT_EPISTEMIC_THRESHOLD
min_applications_for_effectiveness	int	Min applications before using actual rate.	DEFAULT_MIN_APPLICATIONS
frequency_normalization_base	int	Log base for frequency factor.	DEFAULT_FREQUENCY_BASE

Source code in src/marianne/learning/weighter.py

def __init__(
    self,
    *,
    decay_rate_per_month: float = DEFAULT_DECAY_RATE_PER_MONTH,
    effectiveness_threshold: float = DEFAULT_EFFECTIVENESS_THRESHOLD,
    epistemic_threshold: float = DEFAULT_EPISTEMIC_THRESHOLD,
    min_applications_for_effectiveness: int = DEFAULT_MIN_APPLICATIONS,
    frequency_normalization_base: int = DEFAULT_FREQUENCY_BASE,
) -> None:
    """Initialize the pattern weighter.

    Args:
        decay_rate_per_month: Fraction of priority lost per month.
        effectiveness_threshold: Below this, patterns are deprecated.
        epistemic_threshold: Variance threshold for learnable patterns.
        min_applications_for_effectiveness: Min applications before using actual rate.
        frequency_normalization_base: Log base for frequency factor.
    """
    self.decay_rate_per_month = decay_rate_per_month
    self.effectiveness_threshold = effectiveness_threshold
    self.epistemic_threshold = epistemic_threshold
    self.min_applications_for_effectiveness = min_applications_for_effectiveness
    self.frequency_normalization_base = frequency_normalization_base
    self.frequency_floor = DEFAULT_FREQUENCY_FLOOR

Functions

calculate_priority

calculate_priority(occurrence_count, led_to_success_count, led_to_failure_count, last_confirmed, variance=0.0, now=None)

Calculate the priority score for a pattern.

Parameters:

Name	Type	Description	Default
occurrence_count	int	How many times this pattern was observed.	required
led_to_success_count	int	Times pattern led to success when applied.	required
led_to_failure_count	int	Times pattern led to failure when applied.	required
last_confirmed	datetime	When this pattern was last confirmed effective.	required
variance	float	Standard deviation of pattern application outcomes.	0.0
now	datetime | None	Current time for recency calculation (defaults to now).	None

Returns:

Type	Description
float	Priority score from 0.0 to 1.0.

Source code in src/marianne/learning/weighter.py

def calculate_priority(
    self,
    occurrence_count: int,
    led_to_success_count: int,
    led_to_failure_count: int,
    last_confirmed: datetime,
    variance: float = 0.0,
    now: datetime | None = None,
) -> float:
    """Calculate the priority score for a pattern.

    Args:
        occurrence_count: How many times this pattern was observed.
        led_to_success_count: Times pattern led to success when applied.
        led_to_failure_count: Times pattern led to failure when applied.
        last_confirmed: When this pattern was last confirmed effective.
        variance: Standard deviation of pattern application outcomes.
        now: Current time for recency calculation (defaults to now).

    Returns:
        Priority score from 0.0 to 1.0.
    """
    now = now or datetime.now()

    # Calculate effectiveness score
    effectiveness = self.calculate_effectiveness(
        led_to_success_count,
        led_to_failure_count,
    )

    # Calculate recency factor with exponential decay
    recency = self.calculate_recency_factor(last_confirmed, now)

    # Calculate frequency factor
    frequency = self.calculate_frequency_factor(occurrence_count)

    # Calculate variance penalty
    variance_factor = 1.0 - min(1.0, variance)

    # Combined priority score
    priority = effectiveness * recency * frequency * variance_factor

    return max(0.0, min(1.0, priority))

calculate_effectiveness

calculate_effectiveness(success_count, failure_count)

Calculate effectiveness score from success/failure counts.

Uses Laplace smoothing (add-one) to handle cold start: effectiveness = (successes + 0.5) / (successes + failures + 1)

This gives a slightly optimistic prior (0.5) for patterns with no application history.

Parameters:

Name	Type	Description	Default
success_count	int	Number of successful outcomes after application.	required
failure_count	int	Number of failed outcomes after application.	required

Returns:

Type	Description
float	Effectiveness score from 0.0 to 1.0.

Source code in src/marianne/learning/weighter.py

def calculate_effectiveness(
    self,
    success_count: int,
    failure_count: int,
) -> float:
    """Calculate effectiveness score from success/failure counts.

    Uses Laplace smoothing (add-one) to handle cold start:
    effectiveness = (successes + 0.5) / (successes + failures + 1)

    This gives a slightly optimistic prior (0.5) for patterns with
    no application history.

    Args:
        success_count: Number of successful outcomes after application.
        failure_count: Number of failed outcomes after application.

    Returns:
        Effectiveness score from 0.0 to 1.0.
    """
    total = success_count + failure_count
    if total < self.min_applications_for_effectiveness:
        # Not enough data, return neutral prior
        return 0.5

    # Laplace smoothing
    return (success_count + 0.5) / (total + 1)

calculate_recency_factor

calculate_recency_factor(last_confirmed, now=None)

Calculate recency factor with exponential decay.

Formula: recency = (1 - decay_rate) ^ months_since_last_confirmed

With default 10% decay, a pattern loses: - 10% after 1 month - 19% after 2 months - 27% after 3 months - 35% after 4 months - etc.

Parameters:

Name	Type	Description	Default
last_confirmed	datetime	When the pattern was last confirmed effective.	required
now	datetime | None	Current time (defaults to now).	None

Returns:

Type	Description
float	Recency factor from 0.0 to 1.0.

Source code in src/marianne/learning/weighter.py

def calculate_recency_factor(
    self,
    last_confirmed: datetime,
    now: datetime | None = None,
) -> float:
    """Calculate recency factor with exponential decay.

    Formula: recency = (1 - decay_rate) ^ months_since_last_confirmed

    With default 10% decay, a pattern loses:
    - 10% after 1 month
    - 19% after 2 months
    - 27% after 3 months
    - 35% after 4 months
    - etc.

    Args:
        last_confirmed: When the pattern was last confirmed effective.
        now: Current time (defaults to now).

    Returns:
        Recency factor from 0.0 to 1.0.
    """
    now = now or datetime.now()
    delta = now - last_confirmed
    months = delta.days / 30.0

    # Exponential decay
    decay_base = 1.0 - self.decay_rate_per_month
    recency: float = decay_base ** months

    return float(max(0.0, min(1.0, recency)))

calculate_frequency_factor

calculate_frequency_factor(occurrence_count)

Calculate frequency factor from occurrence count.

Formula: frequency = max(floor, min(1.0, log(count + 1) / log(100)))

This gives diminishing returns for high counts: - 1 occurrence: max(0.6, ~0.15) = 0.6 (floored) - 10 occurrences: max(0.6, ~0.52) = 0.6 (floored) - 50 occurrences: max(0.6, ~0.85) = 0.85 - 100+ occurrences: 1.0

The floor prevents single-occurrence patterns from being crushed below the exploitation query threshold. See issue #101 and the matching FREQUENCY_FACTOR_FLOOR in patterns_crud.py.

Parameters:

Name	Type	Description	Default
occurrence_count	int	How many times the pattern was observed.	required

Returns:

Type	Description
float	Frequency factor from frequency_floor to 1.0.

Source code in src/marianne/learning/weighter.py

def calculate_frequency_factor(self, occurrence_count: int) -> float:
    """Calculate frequency factor from occurrence count.

    Formula: frequency = max(floor, min(1.0, log(count + 1) / log(100)))

    This gives diminishing returns for high counts:
    - 1 occurrence: max(0.6, ~0.15) = 0.6 (floored)
    - 10 occurrences: max(0.6, ~0.52) = 0.6 (floored)
    - 50 occurrences: max(0.6, ~0.85) = 0.85
    - 100+ occurrences: 1.0

    The floor prevents single-occurrence patterns from being crushed
    below the exploitation query threshold.  See issue #101 and the
    matching FREQUENCY_FACTOR_FLOOR in patterns_crud.py.

    Args:
        occurrence_count: How many times the pattern was observed.

    Returns:
        Frequency factor from ``frequency_floor`` to 1.0.
    """
    if occurrence_count <= 0:
        return self.frequency_floor

    base = self.frequency_normalization_base
    log_count = math.log(occurrence_count + 1)
    log_base = math.log(base)

    raw = min(1.0, log_count / log_base)
    return max(self.frequency_floor, raw)

is_deprecated

is_deprecated(led_to_success_count, led_to_failure_count)

Check if a pattern should be deprecated due to low effectiveness.

A pattern is deprecated if: 1. It has enough application data (>= min_applications) 2. Its effectiveness is below the threshold

Parameters:

Name	Type	Description	Default
led_to_success_count	int	Times pattern led to success.	required
led_to_failure_count	int	Times pattern led to failure.	required

Returns:

Type	Description
bool	True if pattern should be deprecated.

Source code in src/marianne/learning/weighter.py

def is_deprecated(
    self,
    led_to_success_count: int,
    led_to_failure_count: int,
) -> bool:
    """Check if a pattern should be deprecated due to low effectiveness.

    A pattern is deprecated if:
    1. It has enough application data (>= min_applications)
    2. Its effectiveness is below the threshold

    Args:
        led_to_success_count: Times pattern led to success.
        led_to_failure_count: Times pattern led to failure.

    Returns:
        True if pattern should be deprecated.
    """
    total = led_to_success_count + led_to_failure_count
    if total < self.min_applications_for_effectiveness:
        # Not enough data to deprecate
        return False

    effectiveness = self.calculate_effectiveness(
        led_to_success_count,
        led_to_failure_count,
    )

    return effectiveness < self.effectiveness_threshold

classify_uncertainty

classify_uncertainty(variance)

Classify the uncertainty type of a pattern.

Epistemic uncertainty (variance < threshold): The pattern behavior can be learned with more data. Keep tracking and applying.

Aleatoric uncertainty (variance >= threshold): The pattern has inherently unpredictable outcomes. Deprioritize but don't remove.

Parameters:

Name	Type	Description	Default
variance	float	The variance of pattern application outcomes.	required

Returns:

Type	Description
str	'epistemic' or 'aleatoric'

Source code in src/marianne/learning/weighter.py

def classify_uncertainty(self, variance: float) -> str:
    """Classify the uncertainty type of a pattern.

    Epistemic uncertainty (variance < threshold): The pattern behavior
    can be learned with more data. Keep tracking and applying.

    Aleatoric uncertainty (variance >= threshold): The pattern has
    inherently unpredictable outcomes. Deprioritize but don't remove.

    Args:
        variance: The variance of pattern application outcomes.

    Returns:
        'epistemic' or 'aleatoric'
    """
    if variance < self.epistemic_threshold:
        return "epistemic"
    return "aleatoric"

calculate_variance

calculate_variance(outcomes)

Calculate variance from a list of boolean outcomes.

Used to track consistency of pattern applications.

Parameters:

Name	Type	Description	Default
outcomes	list[bool]	List of True (success) / False (failure) outcomes.	required

Returns:

Type	Description
float	Variance from 0.0 (all same) to 0.25 (50/50 split).

Source code in src/marianne/learning/weighter.py

def calculate_variance(self, outcomes: list[bool]) -> float:
    """Calculate variance from a list of boolean outcomes.

    Used to track consistency of pattern applications.

    Args:
        outcomes: List of True (success) / False (failure) outcomes.

    Returns:
        Variance from 0.0 (all same) to 0.25 (50/50 split).
    """
    if len(outcomes) < 2:
        return 0.0

    mean = sum(1 if o else 0 for o in outcomes) / len(outcomes)
    # Need parentheses around the subtraction to get correct variance
    variance = sum(((1 if o else 0) - mean) ** 2 for o in outcomes) / len(outcomes)

    return variance

Functions

calculate_priority

calculate_priority(occurrence_count, led_to_success_count, led_to_failure_count, last_confirmed, variance=0.0)

Convenience function to calculate priority without creating a weighter.

Parameters:

Name	Type	Description	Default
occurrence_count	int	How many times this pattern was observed.	required
led_to_success_count	int	Times pattern led to success when applied.	required
led_to_failure_count	int	Times pattern led to failure when applied.	required
last_confirmed	datetime	When this pattern was last confirmed effective.	required
variance	float	Standard deviation of pattern application outcomes.	0.0

Returns:

Type	Description
float	Priority score from 0.0 to 1.0.

Source code in src/marianne/learning/weighter.py

def calculate_priority(
    occurrence_count: int,
    led_to_success_count: int,
    led_to_failure_count: int,
    last_confirmed: datetime,
    variance: float = 0.0,
) -> float:
    """Convenience function to calculate priority without creating a weighter.

    Args:
        occurrence_count: How many times this pattern was observed.
        led_to_success_count: Times pattern led to success when applied.
        led_to_failure_count: Times pattern led to failure when applied.
        last_confirmed: When this pattern was last confirmed effective.
        variance: Standard deviation of pattern application outcomes.

    Returns:
        Priority score from 0.0 to 1.0.
    """
    weighter = PatternWeighter()
    return weighter.calculate_priority(
        occurrence_count=occurrence_count,
        led_to_success_count=led_to_success_count,
        led_to_failure_count=led_to_failure_count,
        last_confirmed=last_confirmed,
        variance=variance,
    )