Metrics API

Metrics for measuring multi-agent system health.

SoftMetrics

Core metrics computed from soft probabilistic labels.

swarm.metrics.soft_metrics.SoftMetrics

Computes soft (probabilistic) metrics for interaction quality.

Key metrics: - Toxicity rate: E[1-p | accepted] - Conditional loss: E[π_a | accepted] - E[π_a] - Spread: Quality filtering effectiveness - Quality gap: E[p | accepted] - E[p | rejected]

Source code in swarm/metrics/soft_metrics.py

class SoftMetrics:
    """
    Computes soft (probabilistic) metrics for interaction quality.

    Key metrics:
    - Toxicity rate: E[1-p | accepted]
    - Conditional loss: E[π_a | accepted] - E[π_a]
    - Spread: Quality filtering effectiveness
    - Quality gap: E[p | accepted] - E[p | rejected]
    """

    def __init__(self, payoff_engine: Optional[SoftPayoffEngine] = None):
        """
        Initialize metrics calculator.

        Args:
            payoff_engine: Engine for payoff calculations (default: SoftPayoffEngine())
        """
        self.payoff_engine = payoff_engine or SoftPayoffEngine()

    def toxicity_rate(self, interactions: List[SoftInteraction]) -> float:
        """
        Compute toxicity rate: E[1-p | accepted]

        This measures the expected fraction of harmful outcomes
        among accepted interactions.

        Args:
            interactions: List of interactions

        Returns:
            Toxicity rate in [0, 1], or 0.0 if no accepted interactions
        """
        accepted = [i for i in interactions if i.accepted]
        if not accepted:
            return 0.0

        return sum(1 - i.p for i in accepted) / len(accepted)

    def toxicity_rate_all(self, interactions: List[SoftInteraction]) -> float:
        """
        Compute unconditional toxicity rate: E[1-p]

        Args:
            interactions: List of interactions

        Returns:
            Toxicity rate in [0, 1]
        """
        if not interactions:
            return 0.0

        return sum(1 - i.p for i in interactions) / len(interactions)

    def conditional_loss_initiator(
        self, interactions: List[SoftInteraction]
    ) -> float:
        """
        Compute conditional loss for initiator: E[π_a | accepted] - E[π_a]

        Negative values indicate adverse selection (accepted interactions
        are worse than average for the initiator).

        Args:
            interactions: List of interactions

        Returns:
            Conditional loss (negative = adverse selection)
        """
        if not interactions:
            return 0.0

        accepted = [i for i in interactions if i.accepted]
        if not accepted:
            return 0.0

        # Average payoff for accepted
        avg_accepted = sum(
            self.payoff_engine.payoff_initiator(i) for i in accepted
        ) / len(accepted)

        # Average payoff overall
        avg_all = sum(
            self.payoff_engine.payoff_initiator(i) for i in interactions
        ) / len(interactions)

        return avg_accepted - avg_all

    def conditional_loss_counterparty(
        self, interactions: List[SoftInteraction]
    ) -> float:
        """
        Compute conditional loss for counterparty: E[π_b | accepted] - E[π_b]

        Args:
            interactions: List of interactions

        Returns:
            Conditional loss (negative = adverse selection)
        """
        if not interactions:
            return 0.0

        accepted = [i for i in interactions if i.accepted]
        if not accepted:
            return 0.0

        avg_accepted = sum(
            self.payoff_engine.payoff_counterparty(i) for i in accepted
        ) / len(accepted)

        avg_all = sum(
            self.payoff_engine.payoff_counterparty(i) for i in interactions
        ) / len(interactions)

        return avg_accepted - avg_all

    def spread(self, interactions: List[SoftInteraction]) -> float:
        """
        Compute spread: (s_plus + s_minus) * (E[p] - E[p | accepted])

        Positive spread indicates the market is filtering out high-quality
        interactions (adverse selection in the quality sense).

        Args:
            interactions: List of interactions

        Returns:
            Spread value
        """
        if not interactions:
            return 0.0

        accepted = [i for i in interactions if i.accepted]
        if not accepted:
            return 0.0

        avg_p_all = sum(i.p for i in interactions) / len(interactions)
        avg_p_accepted = sum(i.p for i in accepted) / len(accepted)

        scale = (
            self.payoff_engine.config.s_plus +
            self.payoff_engine.config.s_minus
        )

        return scale * (avg_p_all - avg_p_accepted)

    def quality_gap(self, interactions: List[SoftInteraction]) -> float:
        """
        Compute quality gap: E[p | accepted] - E[p | rejected]

        Negative quality gap indicates adverse selection (accepted
        interactions have lower quality than rejected ones).

        Args:
            interactions: List of interactions

        Returns:
            Quality gap (negative = adverse selection)
        """
        accepted = [i for i in interactions if i.accepted]
        rejected = [i for i in interactions if not i.accepted]

        if not accepted or not rejected:
            return 0.0

        avg_p_accepted = sum(i.p for i in accepted) / len(accepted)
        avg_p_rejected = sum(i.p for i in rejected) / len(rejected)

        return avg_p_accepted - avg_p_rejected

    def participation_by_quality(
        self,
        interactions: List[SoftInteraction],
        threshold: float = 0.5,
    ) -> dict:
        """
        Compute acceptance rates for high/low quality interactions.

        Args:
            interactions: List of interactions
            threshold: Quality threshold (default 0.5)

        Returns:
            Dictionary with acceptance rates:
            - high_quality_acceptance: P(accepted | p >= threshold)
            - low_quality_acceptance: P(accepted | p < threshold)
            - high_quality_count: Number of high quality interactions
            - low_quality_count: Number of low quality interactions
        """
        high_quality = [i for i in interactions if i.p >= threshold]
        low_quality = [i for i in interactions if i.p < threshold]

        high_accepted = sum(1 for i in high_quality if i.accepted)
        low_accepted = sum(1 for i in low_quality if i.accepted)

        return {
            "high_quality_acceptance": (
                high_accepted / len(high_quality) if high_quality else 0.0
            ),
            "low_quality_acceptance": (
                low_accepted / len(low_quality) if low_quality else 0.0
            ),
            "high_quality_count": len(high_quality),
            "low_quality_count": len(low_quality),
        }

    def flag_uncertain(
        self,
        interactions: List[SoftInteraction],
        band: float = 0.2,
    ) -> List[SoftInteraction]:
        """
        Flag interactions with uncertain labels (p near 0.5).

        Args:
            interactions: List of interactions
            band: Width of uncertainty band around 0.5

        Returns:
            List of uncertain interactions
        """
        return [i for i in interactions if i.is_uncertain(band)]

    def uncertain_fraction(
        self,
        interactions: List[SoftInteraction],
        band: float = 0.2,
    ) -> float:
        """
        Compute fraction of interactions with uncertain labels.

        Args:
            interactions: List of interactions
            band: Width of uncertainty band around 0.5

        Returns:
            Fraction in [0, 1]
        """
        if not interactions:
            return 0.0

        uncertain = self.flag_uncertain(interactions, band)
        return len(uncertain) / len(interactions)

    def average_quality(
        self,
        interactions: List[SoftInteraction],
        accepted_only: bool = False,
    ) -> float:
        """
        Compute average quality E[p].

        Args:
            interactions: List of interactions
            accepted_only: If True, only consider accepted interactions

        Returns:
            Average p value
        """
        if accepted_only:
            interactions = [i for i in interactions if i.accepted]

        if not interactions:
            return 0.0

        return sum(i.p for i in interactions) / len(interactions)

    def quality_distribution(
        self,
        interactions: List[SoftInteraction],
        bins: int = 10,
    ) -> List[Tuple[float, float, int]]:
        """
        Compute quality distribution histogram.

        Args:
            interactions: List of interactions
            bins: Number of bins

        Returns:
            List of (bin_start, bin_end, count) tuples
        """
        if not interactions:
            return []

        bin_width = 1.0 / bins
        result = []

        for i in range(bins):
            bin_start = i * bin_width
            bin_end = (i + 1) * bin_width

            count = sum(
                1 for interaction in interactions
                if bin_start <= interaction.p < bin_end
                or (i == bins - 1 and interaction.p == 1.0)
            )

            result.append((bin_start, bin_end, count))

        return result

    def welfare_metrics(
        self, interactions: List[SoftInteraction]
    ) -> dict:
        """
        Compute aggregate welfare metrics.

        Args:
            interactions: List of interactions

        Returns:
            Dictionary with welfare metrics
        """
        if not interactions:
            return {
                "total_welfare": 0.0,
                "total_social_surplus": 0.0,
                "avg_initiator_payoff": 0.0,
                "avg_counterparty_payoff": 0.0,
            }

        accepted = [i for i in interactions if i.accepted]

        total_welfare = sum(
            self.payoff_engine.total_welfare(i) for i in accepted
        )
        total_social = sum(
            self.payoff_engine.social_surplus(i) for i in accepted
        )
        avg_init = (
            sum(self.payoff_engine.payoff_initiator(i) for i in accepted)
            / len(accepted) if accepted else 0.0
        )
        avg_counter = (
            sum(self.payoff_engine.payoff_counterparty(i) for i in accepted)
            / len(accepted) if accepted else 0.0
        )

        return {
            "total_welfare": total_welfare,
            "total_social_surplus": total_social,
            "avg_initiator_payoff": avg_init,
            "avg_counterparty_payoff": avg_counter,
        }

    # =========================================================================
    # Calibration Metrics
    # =========================================================================

    def calibration_error(
        self, interactions: List[SoftInteraction]
    ) -> Optional[float]:
        """
        Compute calibration error: E[p] - empirical_positive_rate.

        Requires ground_truth to be set on interactions.
        A well-calibrated model has calibration error near 0.

        Args:
            interactions: List of interactions with ground_truth set

        Returns:
            Calibration error, or None if no ground truth available
        """
        with_truth = [i for i in interactions if i.ground_truth is not None]
        if not with_truth:
            return None

        # E[p]
        avg_p = sum(i.p for i in with_truth) / len(with_truth)

        # Empirical positive rate: fraction where ground_truth = +1
        # ground_truth is +1 or -1, so we convert to 0/1
        positive_count = sum(1 for i in with_truth if i.ground_truth == 1)
        empirical_rate = positive_count / len(with_truth)

        return avg_p - empirical_rate

    def brier_score(
        self, interactions: List[SoftInteraction]
    ) -> Optional[float]:
        """
        Compute Brier score: E[(p - v)^2] where v = (ground_truth + 1) / 2.

        The Brier score is a proper scoring rule for probabilistic predictions.
        - 0 is perfect prediction
        - 0.25 is equivalent to always predicting p=0.5

        Args:
            interactions: List of interactions with ground_truth set

        Returns:
            Brier score in [0, 1], or None if no ground truth available
        """
        with_truth = [i for i in interactions if i.ground_truth is not None]
        if not with_truth:
            return None

        total = 0.0
        for i in with_truth:
            # Convert ground_truth from {-1, +1} to {0, 1}
            gt = i.ground_truth if i.ground_truth is not None else 0
            v = (gt + 1) / 2
            total += (i.p - v) ** 2

        return total / len(with_truth)

    def expected_calibration_error(
        self,
        interactions: List[SoftInteraction],
        bins: int = 10,
    ) -> Optional[float]:
        """
        Compute Expected Calibration Error (ECE).

        ECE is the weighted average of |E[p|bin] - accuracy(bin)| across bins.
        A perfectly calibrated model has ECE = 0.

        Args:
            interactions: List of interactions with ground_truth set
            bins: Number of probability bins

        Returns:
            ECE value, or None if no ground truth available
        """
        curve = self.calibration_curve(interactions, bins)
        if not curve:
            return None

        total_count = sum(count for _, _, count in curve)
        if total_count == 0:
            return None

        ece = 0.0
        for mean_predicted, fraction_positive, count in curve:
            if count > 0:
                ece += (count / total_count) * abs(mean_predicted - fraction_positive)

        return ece

    def calibration_curve(
        self,
        interactions: List[SoftInteraction],
        bins: int = 10,
    ) -> List[Tuple[float, float, int]]:
        """
        Compute calibration curve data.

        For each bin of predicted probabilities, compute the fraction of
        actually positive outcomes.

        Args:
            interactions: List of interactions with ground_truth set
            bins: Number of probability bins

        Returns:
            List of (mean_predicted, fraction_positive, count) per bin.
            Returns empty list if no ground truth available.
        """
        with_truth = [i for i in interactions if i.ground_truth is not None]
        if not with_truth:
            return []

        bin_width = 1.0 / bins
        result = []

        for b in range(bins):
            bin_start = b * bin_width
            bin_end = (b + 1) * bin_width

            # Get interactions in this bin
            in_bin = [
                i for i in with_truth
                if bin_start <= i.p < bin_end
                or (b == bins - 1 and i.p == 1.0)
            ]

            if not in_bin:
                # Empty bin - use midpoint as predicted, 0.0 as accuracy
                result.append((bin_start + bin_width / 2, 0.0, 0))
            else:
                mean_predicted = sum(i.p for i in in_bin) / len(in_bin)
                positive_count = sum(1 for i in in_bin if i.ground_truth == 1)
                fraction_positive = positive_count / len(in_bin)
                result.append((mean_predicted, fraction_positive, len(in_bin)))

        return result

    # =========================================================================
    # Information-Theoretic Metrics
    # =========================================================================

    def log_loss(
        self,
        interactions: List[SoftInteraction],
        eps: float = 1e-15,
    ) -> Optional[float]:
        """
        Compute log loss (cross-entropy): -E[v*log(p) + (1-v)*log(1-p)].

        Args:
            interactions: List of interactions with ground_truth set
            eps: Small value to avoid log(0)

        Returns:
            Log loss (lower is better), or None if no ground truth available
        """
        with_truth = [i for i in interactions if i.ground_truth is not None]
        if not with_truth:
            return None

        total = 0.0
        for i in with_truth:
            # Convert ground_truth from {-1, +1} to {0, 1}
            gt = i.ground_truth if i.ground_truth is not None else 0
            v = (gt + 1) / 2
            # Clamp p to avoid log(0)
            p_clamped = max(eps, min(1 - eps, i.p))

            total -= v * math.log(p_clamped) + (1 - v) * math.log(1 - p_clamped)

        return total / len(with_truth)

    def discrimination_auc(
        self, interactions: List[SoftInteraction]
    ) -> Optional[float]:
        """
        Compute Area Under ROC Curve (AUC) for discrimination.

        AUC measures the model's ability to rank positive cases higher
        than negative cases.
        - AUC = 0.5: random guessing
        - AUC = 1.0: perfect discrimination

        Args:
            interactions: List of interactions with ground_truth set

        Returns:
            AUC value in [0, 1], or None if insufficient data
        """
        with_truth = [i for i in interactions if i.ground_truth is not None]
        if not with_truth:
            return None

        positives = [i for i in with_truth if i.ground_truth == 1]
        negatives = [i for i in with_truth if i.ground_truth == -1]

        if not positives or not negatives:
            return None

        # Wilcoxon-Mann-Whitney statistic
        concordant: float = 0
        total_pairs = len(positives) * len(negatives)

        for pos in positives:
            for neg in negatives:
                if pos.p > neg.p:
                    concordant += 1
                elif pos.p == neg.p:
                    concordant += 0.5

        return concordant / total_pairs

    # =========================================================================
    # Variance / Uncertainty Metrics
    # =========================================================================

    def quality_variance(
        self,
        interactions: List[SoftInteraction],
        accepted_only: bool = False,
    ) -> float:
        """
        Compute variance of quality: Var[p].

        Args:
            interactions: List of interactions
            accepted_only: If True, only consider accepted interactions

        Returns:
            Variance of p
        """
        if accepted_only:
            interactions = [i for i in interactions if i.accepted]

        if len(interactions) < 2:
            return 0.0

        mean_p = sum(i.p for i in interactions) / len(interactions)
        variance = sum((i.p - mean_p) ** 2 for i in interactions) / len(interactions)

        return variance

    def quality_std(
        self,
        interactions: List[SoftInteraction],
        accepted_only: bool = False,
    ) -> float:
        """
        Compute standard deviation of quality: Std[p].

        Args:
            interactions: List of interactions
            accepted_only: If True, only consider accepted interactions

        Returns:
            Standard deviation of p
        """
        return math.sqrt(self.quality_variance(interactions, accepted_only))

    def payoff_variance_initiator(
        self, interactions: List[SoftInteraction]
    ) -> float:
        """
        Compute variance of initiator payoffs: Var[π_a].

        Measures risk/dispersion in initiator outcomes.

        Args:
            interactions: List of interactions

        Returns:
            Variance of initiator payoffs
        """
        if len(interactions) < 2:
            return 0.0

        payoffs = [self.payoff_engine.payoff_initiator(i) for i in interactions]
        mean_payoff = sum(payoffs) / len(payoffs)
        variance = sum((p - mean_payoff) ** 2 for p in payoffs) / len(payoffs)

        return variance

    def payoff_variance_counterparty(
        self, interactions: List[SoftInteraction]
    ) -> float:
        """
        Compute variance of counterparty payoffs: Var[π_b].

        Measures risk/dispersion in counterparty outcomes.

        Args:
            interactions: List of interactions

        Returns:
            Variance of counterparty payoffs
        """
        if len(interactions) < 2:
            return 0.0

        payoffs = [self.payoff_engine.payoff_counterparty(i) for i in interactions]
        mean_payoff = sum(payoffs) / len(payoffs)
        variance = sum((p - mean_payoff) ** 2 for p in payoffs) / len(payoffs)

        return variance

    def coefficient_of_variation(
        self, interactions: List[SoftInteraction]
    ) -> dict:
        """
        Compute coefficient of variation (CV = std/mean) for key metrics.

        CV is a standardized measure of dispersion. Higher CV indicates
        more variability relative to the mean.

        Args:
            interactions: List of interactions

        Returns:
            Dictionary with CV for p, π_a, and π_b
        """
        if not interactions:
            return {
                "cv_p": 0.0,
                "cv_payoff_initiator": 0.0,
                "cv_payoff_counterparty": 0.0,
            }

        # CV for p
        mean_p = self.average_quality(interactions)
        std_p = self.quality_std(interactions)
        cv_p = std_p / mean_p if mean_p != 0 else 0.0

        # CV for initiator payoffs
        payoffs_a = [self.payoff_engine.payoff_initiator(i) for i in interactions]
        mean_a = sum(payoffs_a) / len(payoffs_a)
        std_a = math.sqrt(self.payoff_variance_initiator(interactions))
        cv_a = abs(std_a / mean_a) if mean_a != 0 else 0.0

        # CV for counterparty payoffs
        payoffs_b = [self.payoff_engine.payoff_counterparty(i) for i in interactions]
        mean_b = sum(payoffs_b) / len(payoffs_b)
        std_b = math.sqrt(self.payoff_variance_counterparty(interactions))
        cv_b = abs(std_b / mean_b) if mean_b != 0 else 0.0

        return {
            "cv_p": cv_p,
            "cv_payoff_initiator": cv_a,
            "cv_payoff_counterparty": cv_b,
        }

__init__(payoff_engine=None)

Initialize metrics calculator.

Parameters:

Name	Type	Description	Default
payoff_engine	Optional[SoftPayoffEngine]	Engine for payoff calculations (default: SoftPayoffEngine())	None

Source code in swarm/metrics/soft_metrics.py

def __init__(self, payoff_engine: Optional[SoftPayoffEngine] = None):
    """
    Initialize metrics calculator.

    Args:
        payoff_engine: Engine for payoff calculations (default: SoftPayoffEngine())
    """
    self.payoff_engine = payoff_engine or SoftPayoffEngine()

average_quality(interactions, accepted_only=False)

Compute average quality E[p].

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required
accepted_only	bool	If True, only consider accepted interactions	False

Returns:

Type	Description
float	Average p value

Source code in swarm/metrics/soft_metrics.py

def average_quality(
    self,
    interactions: List[SoftInteraction],
    accepted_only: bool = False,
) -> float:
    """
    Compute average quality E[p].

    Args:
        interactions: List of interactions
        accepted_only: If True, only consider accepted interactions

    Returns:
        Average p value
    """
    if accepted_only:
        interactions = [i for i in interactions if i.accepted]

    if not interactions:
        return 0.0

    return sum(i.p for i in interactions) / len(interactions)

brier_score(interactions)

Compute Brier score: E[(p - v)^2] where v = (ground_truth + 1) / 2.

The Brier score is a proper scoring rule for probabilistic predictions. - 0 is perfect prediction - 0.25 is equivalent to always predicting p=0.5

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions with ground_truth set	required

Returns:

Type	Description
Optional[float]	Brier score in [0, 1], or None if no ground truth available

Source code in swarm/metrics/soft_metrics.py

def brier_score(
    self, interactions: List[SoftInteraction]
) -> Optional[float]:
    """
    Compute Brier score: E[(p - v)^2] where v = (ground_truth + 1) / 2.

    The Brier score is a proper scoring rule for probabilistic predictions.
    - 0 is perfect prediction
    - 0.25 is equivalent to always predicting p=0.5

    Args:
        interactions: List of interactions with ground_truth set

    Returns:
        Brier score in [0, 1], or None if no ground truth available
    """
    with_truth = [i for i in interactions if i.ground_truth is not None]
    if not with_truth:
        return None

    total = 0.0
    for i in with_truth:
        # Convert ground_truth from {-1, +1} to {0, 1}
        gt = i.ground_truth if i.ground_truth is not None else 0
        v = (gt + 1) / 2
        total += (i.p - v) ** 2

    return total / len(with_truth)

calibration_curve(interactions, bins=10)

Compute calibration curve data.

For each bin of predicted probabilities, compute the fraction of actually positive outcomes.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions with ground_truth set	required
bins	int	Number of probability bins	10

Returns:

Type	Description
List[Tuple[float, float, int]]	List of (mean_predicted, fraction_positive, count) per bin.
List[Tuple[float, float, int]]	Returns empty list if no ground truth available.

Source code in swarm/metrics/soft_metrics.py

def calibration_curve(
    self,
    interactions: List[SoftInteraction],
    bins: int = 10,
) -> List[Tuple[float, float, int]]:
    """
    Compute calibration curve data.

    For each bin of predicted probabilities, compute the fraction of
    actually positive outcomes.

    Args:
        interactions: List of interactions with ground_truth set
        bins: Number of probability bins

    Returns:
        List of (mean_predicted, fraction_positive, count) per bin.
        Returns empty list if no ground truth available.
    """
    with_truth = [i for i in interactions if i.ground_truth is not None]
    if not with_truth:
        return []

    bin_width = 1.0 / bins
    result = []

    for b in range(bins):
        bin_start = b * bin_width
        bin_end = (b + 1) * bin_width

        # Get interactions in this bin
        in_bin = [
            i for i in with_truth
            if bin_start <= i.p < bin_end
            or (b == bins - 1 and i.p == 1.0)
        ]

        if not in_bin:
            # Empty bin - use midpoint as predicted, 0.0 as accuracy
            result.append((bin_start + bin_width / 2, 0.0, 0))
        else:
            mean_predicted = sum(i.p for i in in_bin) / len(in_bin)
            positive_count = sum(1 for i in in_bin if i.ground_truth == 1)
            fraction_positive = positive_count / len(in_bin)
            result.append((mean_predicted, fraction_positive, len(in_bin)))

    return result

calibration_error(interactions)

Compute calibration error: E[p] - empirical_positive_rate.

Requires ground_truth to be set on interactions. A well-calibrated model has calibration error near 0.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions with ground_truth set	required

Returns:

Type	Description
Optional[float]	Calibration error, or None if no ground truth available

Source code in swarm/metrics/soft_metrics.py

def calibration_error(
    self, interactions: List[SoftInteraction]
) -> Optional[float]:
    """
    Compute calibration error: E[p] - empirical_positive_rate.

    Requires ground_truth to be set on interactions.
    A well-calibrated model has calibration error near 0.

    Args:
        interactions: List of interactions with ground_truth set

    Returns:
        Calibration error, or None if no ground truth available
    """
    with_truth = [i for i in interactions if i.ground_truth is not None]
    if not with_truth:
        return None

    # E[p]
    avg_p = sum(i.p for i in with_truth) / len(with_truth)

    # Empirical positive rate: fraction where ground_truth = +1
    # ground_truth is +1 or -1, so we convert to 0/1
    positive_count = sum(1 for i in with_truth if i.ground_truth == 1)
    empirical_rate = positive_count / len(with_truth)

    return avg_p - empirical_rate

coefficient_of_variation(interactions)

Compute coefficient of variation (CV = std/mean) for key metrics.

CV is a standardized measure of dispersion. Higher CV indicates more variability relative to the mean.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
dict	Dictionary with CV for p, π_a, and π_b

Source code in swarm/metrics/soft_metrics.py

def coefficient_of_variation(
    self, interactions: List[SoftInteraction]
) -> dict:
    """
    Compute coefficient of variation (CV = std/mean) for key metrics.

    CV is a standardized measure of dispersion. Higher CV indicates
    more variability relative to the mean.

    Args:
        interactions: List of interactions

    Returns:
        Dictionary with CV for p, π_a, and π_b
    """
    if not interactions:
        return {
            "cv_p": 0.0,
            "cv_payoff_initiator": 0.0,
            "cv_payoff_counterparty": 0.0,
        }

    # CV for p
    mean_p = self.average_quality(interactions)
    std_p = self.quality_std(interactions)
    cv_p = std_p / mean_p if mean_p != 0 else 0.0

    # CV for initiator payoffs
    payoffs_a = [self.payoff_engine.payoff_initiator(i) for i in interactions]
    mean_a = sum(payoffs_a) / len(payoffs_a)
    std_a = math.sqrt(self.payoff_variance_initiator(interactions))
    cv_a = abs(std_a / mean_a) if mean_a != 0 else 0.0

    # CV for counterparty payoffs
    payoffs_b = [self.payoff_engine.payoff_counterparty(i) for i in interactions]
    mean_b = sum(payoffs_b) / len(payoffs_b)
    std_b = math.sqrt(self.payoff_variance_counterparty(interactions))
    cv_b = abs(std_b / mean_b) if mean_b != 0 else 0.0

    return {
        "cv_p": cv_p,
        "cv_payoff_initiator": cv_a,
        "cv_payoff_counterparty": cv_b,
    }

conditional_loss_counterparty(interactions)

Compute conditional loss for counterparty: E[π_b | accepted] - E[π_b]

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Conditional loss (negative = adverse selection)

Source code in swarm/metrics/soft_metrics.py

def conditional_loss_counterparty(
    self, interactions: List[SoftInteraction]
) -> float:
    """
    Compute conditional loss for counterparty: E[π_b | accepted] - E[π_b]

    Args:
        interactions: List of interactions

    Returns:
        Conditional loss (negative = adverse selection)
    """
    if not interactions:
        return 0.0

    accepted = [i for i in interactions if i.accepted]
    if not accepted:
        return 0.0

    avg_accepted = sum(
        self.payoff_engine.payoff_counterparty(i) for i in accepted
    ) / len(accepted)

    avg_all = sum(
        self.payoff_engine.payoff_counterparty(i) for i in interactions
    ) / len(interactions)

    return avg_accepted - avg_all

conditional_loss_initiator(interactions)

Compute conditional loss for initiator: E[π_a | accepted] - E[π_a]

Negative values indicate adverse selection (accepted interactions are worse than average for the initiator).

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Conditional loss (negative = adverse selection)

Source code in swarm/metrics/soft_metrics.py

def conditional_loss_initiator(
    self, interactions: List[SoftInteraction]
) -> float:
    """
    Compute conditional loss for initiator: E[π_a | accepted] - E[π_a]

    Negative values indicate adverse selection (accepted interactions
    are worse than average for the initiator).

    Args:
        interactions: List of interactions

    Returns:
        Conditional loss (negative = adverse selection)
    """
    if not interactions:
        return 0.0

    accepted = [i for i in interactions if i.accepted]
    if not accepted:
        return 0.0

    # Average payoff for accepted
    avg_accepted = sum(
        self.payoff_engine.payoff_initiator(i) for i in accepted
    ) / len(accepted)

    # Average payoff overall
    avg_all = sum(
        self.payoff_engine.payoff_initiator(i) for i in interactions
    ) / len(interactions)

    return avg_accepted - avg_all

discrimination_auc(interactions)

Compute Area Under ROC Curve (AUC) for discrimination.

AUC measures the model's ability to rank positive cases higher than negative cases. - AUC = 0.5: random guessing - AUC = 1.0: perfect discrimination

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions with ground_truth set	required

Returns:

Type	Description
Optional[float]	AUC value in [0, 1], or None if insufficient data

Source code in swarm/metrics/soft_metrics.py

def discrimination_auc(
    self, interactions: List[SoftInteraction]
) -> Optional[float]:
    """
    Compute Area Under ROC Curve (AUC) for discrimination.

    AUC measures the model's ability to rank positive cases higher
    than negative cases.
    - AUC = 0.5: random guessing
    - AUC = 1.0: perfect discrimination

    Args:
        interactions: List of interactions with ground_truth set

    Returns:
        AUC value in [0, 1], or None if insufficient data
    """
    with_truth = [i for i in interactions if i.ground_truth is not None]
    if not with_truth:
        return None

    positives = [i for i in with_truth if i.ground_truth == 1]
    negatives = [i for i in with_truth if i.ground_truth == -1]

    if not positives or not negatives:
        return None

    # Wilcoxon-Mann-Whitney statistic
    concordant: float = 0
    total_pairs = len(positives) * len(negatives)

    for pos in positives:
        for neg in negatives:
            if pos.p > neg.p:
                concordant += 1
            elif pos.p == neg.p:
                concordant += 0.5

    return concordant / total_pairs

expected_calibration_error(interactions, bins=10)

Compute Expected Calibration Error (ECE).

ECE is the weighted average of |E[p|bin] - accuracy(bin)| across bins. A perfectly calibrated model has ECE = 0.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions with ground_truth set	required
bins	int	Number of probability bins	10

Returns:

Type	Description
Optional[float]	ECE value, or None if no ground truth available

Source code in swarm/metrics/soft_metrics.py

def expected_calibration_error(
    self,
    interactions: List[SoftInteraction],
    bins: int = 10,
) -> Optional[float]:
    """
    Compute Expected Calibration Error (ECE).

    ECE is the weighted average of |E[p|bin] - accuracy(bin)| across bins.
    A perfectly calibrated model has ECE = 0.

    Args:
        interactions: List of interactions with ground_truth set
        bins: Number of probability bins

    Returns:
        ECE value, or None if no ground truth available
    """
    curve = self.calibration_curve(interactions, bins)
    if not curve:
        return None

    total_count = sum(count for _, _, count in curve)
    if total_count == 0:
        return None

    ece = 0.0
    for mean_predicted, fraction_positive, count in curve:
        if count > 0:
            ece += (count / total_count) * abs(mean_predicted - fraction_positive)

    return ece

flag_uncertain(interactions, band=0.2)

Flag interactions with uncertain labels (p near 0.5).

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required
band	float	Width of uncertainty band around 0.5	0.2

Returns:

Type	Description
List[SoftInteraction]	List of uncertain interactions

Source code in swarm/metrics/soft_metrics.py

def flag_uncertain(
    self,
    interactions: List[SoftInteraction],
    band: float = 0.2,
) -> List[SoftInteraction]:
    """
    Flag interactions with uncertain labels (p near 0.5).

    Args:
        interactions: List of interactions
        band: Width of uncertainty band around 0.5

    Returns:
        List of uncertain interactions
    """
    return [i for i in interactions if i.is_uncertain(band)]

log_loss(interactions, eps=1e-15)

Compute log loss (cross-entropy): -E[vlog(p) + (1-v)log(1-p)].

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions with ground_truth set	required
eps	float	Small value to avoid log(0)	1e-15

Returns:

Type	Description
Optional[float]	Log loss (lower is better), or None if no ground truth available

Source code in swarm/metrics/soft_metrics.py

def log_loss(
    self,
    interactions: List[SoftInteraction],
    eps: float = 1e-15,
) -> Optional[float]:
    """
    Compute log loss (cross-entropy): -E[v*log(p) + (1-v)*log(1-p)].

    Args:
        interactions: List of interactions with ground_truth set
        eps: Small value to avoid log(0)

    Returns:
        Log loss (lower is better), or None if no ground truth available
    """
    with_truth = [i for i in interactions if i.ground_truth is not None]
    if not with_truth:
        return None

    total = 0.0
    for i in with_truth:
        # Convert ground_truth from {-1, +1} to {0, 1}
        gt = i.ground_truth if i.ground_truth is not None else 0
        v = (gt + 1) / 2
        # Clamp p to avoid log(0)
        p_clamped = max(eps, min(1 - eps, i.p))

        total -= v * math.log(p_clamped) + (1 - v) * math.log(1 - p_clamped)

    return total / len(with_truth)

participation_by_quality(interactions, threshold=0.5)

Compute acceptance rates for high/low quality interactions.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required
threshold	float	Quality threshold (default 0.5)	0.5

Returns:

Type	Description
dict	Dictionary with acceptance rates:
dict	high_quality_acceptance: P(accepted | p >= threshold)
dict	low_quality_acceptance: P(accepted | p < threshold)
dict	high_quality_count: Number of high quality interactions
dict	low_quality_count: Number of low quality interactions

Source code in swarm/metrics/soft_metrics.py

def participation_by_quality(
    self,
    interactions: List[SoftInteraction],
    threshold: float = 0.5,
) -> dict:
    """
    Compute acceptance rates for high/low quality interactions.

    Args:
        interactions: List of interactions
        threshold: Quality threshold (default 0.5)

    Returns:
        Dictionary with acceptance rates:
        - high_quality_acceptance: P(accepted | p >= threshold)
        - low_quality_acceptance: P(accepted | p < threshold)
        - high_quality_count: Number of high quality interactions
        - low_quality_count: Number of low quality interactions
    """
    high_quality = [i for i in interactions if i.p >= threshold]
    low_quality = [i for i in interactions if i.p < threshold]

    high_accepted = sum(1 for i in high_quality if i.accepted)
    low_accepted = sum(1 for i in low_quality if i.accepted)

    return {
        "high_quality_acceptance": (
            high_accepted / len(high_quality) if high_quality else 0.0
        ),
        "low_quality_acceptance": (
            low_accepted / len(low_quality) if low_quality else 0.0
        ),
        "high_quality_count": len(high_quality),
        "low_quality_count": len(low_quality),
    }

payoff_variance_counterparty(interactions)

Compute variance of counterparty payoffs: Var[π_b].

Measures risk/dispersion in counterparty outcomes.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Variance of counterparty payoffs

Source code in swarm/metrics/soft_metrics.py

def payoff_variance_counterparty(
    self, interactions: List[SoftInteraction]
) -> float:
    """
    Compute variance of counterparty payoffs: Var[π_b].

    Measures risk/dispersion in counterparty outcomes.

    Args:
        interactions: List of interactions

    Returns:
        Variance of counterparty payoffs
    """
    if len(interactions) < 2:
        return 0.0

    payoffs = [self.payoff_engine.payoff_counterparty(i) for i in interactions]
    mean_payoff = sum(payoffs) / len(payoffs)
    variance = sum((p - mean_payoff) ** 2 for p in payoffs) / len(payoffs)

    return variance

payoff_variance_initiator(interactions)

Compute variance of initiator payoffs: Var[π_a].

Measures risk/dispersion in initiator outcomes.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Variance of initiator payoffs

Source code in swarm/metrics/soft_metrics.py

def payoff_variance_initiator(
    self, interactions: List[SoftInteraction]
) -> float:
    """
    Compute variance of initiator payoffs: Var[π_a].

    Measures risk/dispersion in initiator outcomes.

    Args:
        interactions: List of interactions

    Returns:
        Variance of initiator payoffs
    """
    if len(interactions) < 2:
        return 0.0

    payoffs = [self.payoff_engine.payoff_initiator(i) for i in interactions]
    mean_payoff = sum(payoffs) / len(payoffs)
    variance = sum((p - mean_payoff) ** 2 for p in payoffs) / len(payoffs)

    return variance

quality_distribution(interactions, bins=10)

Compute quality distribution histogram.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required
bins	int	Number of bins	10

Returns:

Type	Description
List[Tuple[float, float, int]]	List of (bin_start, bin_end, count) tuples

Source code in swarm/metrics/soft_metrics.py

def quality_distribution(
    self,
    interactions: List[SoftInteraction],
    bins: int = 10,
) -> List[Tuple[float, float, int]]:
    """
    Compute quality distribution histogram.

    Args:
        interactions: List of interactions
        bins: Number of bins

    Returns:
        List of (bin_start, bin_end, count) tuples
    """
    if not interactions:
        return []

    bin_width = 1.0 / bins
    result = []

    for i in range(bins):
        bin_start = i * bin_width
        bin_end = (i + 1) * bin_width

        count = sum(
            1 for interaction in interactions
            if bin_start <= interaction.p < bin_end
            or (i == bins - 1 and interaction.p == 1.0)
        )

        result.append((bin_start, bin_end, count))

    return result

quality_gap(interactions)

Compute quality gap: E[p | accepted] - E[p | rejected]

Negative quality gap indicates adverse selection (accepted interactions have lower quality than rejected ones).

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Quality gap (negative = adverse selection)

Source code in swarm/metrics/soft_metrics.py

def quality_gap(self, interactions: List[SoftInteraction]) -> float:
    """
    Compute quality gap: E[p | accepted] - E[p | rejected]

    Negative quality gap indicates adverse selection (accepted
    interactions have lower quality than rejected ones).

    Args:
        interactions: List of interactions

    Returns:
        Quality gap (negative = adverse selection)
    """
    accepted = [i for i in interactions if i.accepted]
    rejected = [i for i in interactions if not i.accepted]

    if not accepted or not rejected:
        return 0.0

    avg_p_accepted = sum(i.p for i in accepted) / len(accepted)
    avg_p_rejected = sum(i.p for i in rejected) / len(rejected)

    return avg_p_accepted - avg_p_rejected

quality_std(interactions, accepted_only=False)

Compute standard deviation of quality: Std[p].

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required
accepted_only	bool	If True, only consider accepted interactions	False

Returns:

Type	Description
float	Standard deviation of p

Source code in swarm/metrics/soft_metrics.py

def quality_std(
    self,
    interactions: List[SoftInteraction],
    accepted_only: bool = False,
) -> float:
    """
    Compute standard deviation of quality: Std[p].

    Args:
        interactions: List of interactions
        accepted_only: If True, only consider accepted interactions

    Returns:
        Standard deviation of p
    """
    return math.sqrt(self.quality_variance(interactions, accepted_only))

quality_variance(interactions, accepted_only=False)

Compute variance of quality: Var[p].

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required
accepted_only	bool	If True, only consider accepted interactions	False

Returns:

Type	Description
float	Variance of p

Source code in swarm/metrics/soft_metrics.py

def quality_variance(
    self,
    interactions: List[SoftInteraction],
    accepted_only: bool = False,
) -> float:
    """
    Compute variance of quality: Var[p].

    Args:
        interactions: List of interactions
        accepted_only: If True, only consider accepted interactions

    Returns:
        Variance of p
    """
    if accepted_only:
        interactions = [i for i in interactions if i.accepted]

    if len(interactions) < 2:
        return 0.0

    mean_p = sum(i.p for i in interactions) / len(interactions)
    variance = sum((i.p - mean_p) ** 2 for i in interactions) / len(interactions)

    return variance

spread(interactions)

Compute spread: (s_plus + s_minus) * (E[p] - E[p | accepted])

Positive spread indicates the market is filtering out high-quality interactions (adverse selection in the quality sense).

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Spread value

Source code in swarm/metrics/soft_metrics.py

def spread(self, interactions: List[SoftInteraction]) -> float:
    """
    Compute spread: (s_plus + s_minus) * (E[p] - E[p | accepted])

    Positive spread indicates the market is filtering out high-quality
    interactions (adverse selection in the quality sense).

    Args:
        interactions: List of interactions

    Returns:
        Spread value
    """
    if not interactions:
        return 0.0

    accepted = [i for i in interactions if i.accepted]
    if not accepted:
        return 0.0

    avg_p_all = sum(i.p for i in interactions) / len(interactions)
    avg_p_accepted = sum(i.p for i in accepted) / len(accepted)

    scale = (
        self.payoff_engine.config.s_plus +
        self.payoff_engine.config.s_minus
    )

    return scale * (avg_p_all - avg_p_accepted)

toxicity_rate(interactions)

Compute toxicity rate: E[1-p | accepted]

This measures the expected fraction of harmful outcomes among accepted interactions.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Toxicity rate in [0, 1], or 0.0 if no accepted interactions

Source code in swarm/metrics/soft_metrics.py

def toxicity_rate(self, interactions: List[SoftInteraction]) -> float:
    """
    Compute toxicity rate: E[1-p | accepted]

    This measures the expected fraction of harmful outcomes
    among accepted interactions.

    Args:
        interactions: List of interactions

    Returns:
        Toxicity rate in [0, 1], or 0.0 if no accepted interactions
    """
    accepted = [i for i in interactions if i.accepted]
    if not accepted:
        return 0.0

    return sum(1 - i.p for i in accepted) / len(accepted)

toxicity_rate_all(interactions)

Compute unconditional toxicity rate: E[1-p]

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
float	Toxicity rate in [0, 1]

Source code in swarm/metrics/soft_metrics.py

def toxicity_rate_all(self, interactions: List[SoftInteraction]) -> float:
    """
    Compute unconditional toxicity rate: E[1-p]

    Args:
        interactions: List of interactions

    Returns:
        Toxicity rate in [0, 1]
    """
    if not interactions:
        return 0.0

    return sum(1 - i.p for i in interactions) / len(interactions)

uncertain_fraction(interactions, band=0.2)

Compute fraction of interactions with uncertain labels.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required
band	float	Width of uncertainty band around 0.5	0.2

Returns:

Type	Description
float	Fraction in [0, 1]

Source code in swarm/metrics/soft_metrics.py

def uncertain_fraction(
    self,
    interactions: List[SoftInteraction],
    band: float = 0.2,
) -> float:
    """
    Compute fraction of interactions with uncertain labels.

    Args:
        interactions: List of interactions
        band: Width of uncertainty band around 0.5

    Returns:
        Fraction in [0, 1]
    """
    if not interactions:
        return 0.0

    uncertain = self.flag_uncertain(interactions, band)
    return len(uncertain) / len(interactions)

welfare_metrics(interactions)

Compute aggregate welfare metrics.

Parameters:

Name	Type	Description	Default
interactions	List[SoftInteraction]	List of interactions	required

Returns:

Type	Description
dict	Dictionary with welfare metrics

Source code in swarm/metrics/soft_metrics.py

def welfare_metrics(
    self, interactions: List[SoftInteraction]
) -> dict:
    """
    Compute aggregate welfare metrics.

    Args:
        interactions: List of interactions

    Returns:
        Dictionary with welfare metrics
    """
    if not interactions:
        return {
            "total_welfare": 0.0,
            "total_social_surplus": 0.0,
            "avg_initiator_payoff": 0.0,
            "avg_counterparty_payoff": 0.0,
        }

    accepted = [i for i in interactions if i.accepted]

    total_welfare = sum(
        self.payoff_engine.total_welfare(i) for i in accepted
    )
    total_social = sum(
        self.payoff_engine.social_surplus(i) for i in accepted
    )
    avg_init = (
        sum(self.payoff_engine.payoff_initiator(i) for i in accepted)
        / len(accepted) if accepted else 0.0
    )
    avg_counter = (
        sum(self.payoff_engine.payoff_counterparty(i) for i in accepted)
        / len(accepted) if accepted else 0.0
    )

    return {
        "total_welfare": total_welfare,
        "total_social_surplus": total_social,
        "avg_initiator_payoff": avg_init,
        "avg_counterparty_payoff": avg_counter,
    }

Usage

from swarm.metrics.soft_metrics import SoftMetrics

metrics = SoftMetrics()

# Compute individual metrics
toxicity = metrics.toxicity_rate(interactions)
quality_gap = metrics.quality_gap(interactions)
conditional_loss = metrics.conditional_loss(interactions, payoff_engine)

MetricsReporter

Dual reporting of soft and hard metrics.

from swarm.metrics.reporters import MetricsReporter

reporter = MetricsReporter(threshold=0.5)

# Generate report
report = reporter.format_report(interactions, verbose=True)
print(report)

# Get structured data
data = reporter.compute_all(interactions)
print(data['soft']['toxicity_rate'])
print(data['hard']['true_positive_rate'])

Report Format

=== SWARM Metrics Report ===
Interactions: 100 (70 accepted, 30 rejected)

Soft Metrics:
  Toxicity Rate:    0.287
  Quality Gap:      0.142
  Conditional Loss: -0.051

Hard Metrics (threshold=0.5):
  Accept Rate:      0.700
  True Positive:    0.821
  False Positive:   0.179

Incoherence Metrics

Measure decision variance across replays.

from swarm.metrics.incoherence import IncoherenceMetrics, DecisionRecord

incoherence = IncoherenceMetrics()

# Record decisions across replays
for replay in replays:
    record = DecisionRecord(
        decision_id=decision_id,
        replay_id=replay.id,
        decision=replay.decision,
        outcome=replay.outcome,
    )
    incoherence.record(record)

# Compute incoherence index
I = incoherence.compute_index()
print(f"Incoherence Index: {I:.3f}")

Incoherence Components

Component	Formula	Meaning
D	Var[decision]	Decision variance
E	E[error]	Expected error
I	D / E	Incoherence index

Collusion Metrics

Detect coordinated behavior.

from swarm.metrics.collusion import CollusionMetrics

collusion = CollusionMetrics()

# Analyze pair-level patterns
pair_scores = collusion.pair_analysis(interactions)

# Analyze group-level patterns
group_scores = collusion.group_analysis(interactions, group_size=3)

# Get suspicious pairs
suspicious = collusion.get_suspicious_pairs(threshold=0.8)

Security Metrics

Track security-related signals.

from swarm.metrics.security import SecurityMetrics

security = SecurityMetrics()

# Compute security scores
attack_rate = security.attack_detection_rate(interactions)
evasion_rate = security.governance_evasion_rate(interactions, governance)
damage = security.total_externality(interactions)

Capability Metrics

Track emergent capabilities.

from swarm.metrics.capabilities import CapabilityMetrics

capabilities = CapabilityMetrics()

# Compute capability scores
task_completion = capabilities.task_completion_rate(interactions)
collaboration_success = capabilities.collaboration_success_rate(interactions)
composite_capability = capabilities.composite_task_capability(interactions)

Custom Metrics

Create custom metrics:

from swarm.metrics.base import BaseMetric

class CustomMetric(BaseMetric):
    def compute(self, interactions: list) -> float:
        # Custom computation
        accepted = [i for i in interactions if i.accepted]
        return sum(i.p for i in accepted) / len(accepted) if accepted else 0.0

# Use in reporter
reporter = MetricsReporter(
    extra_metrics={'custom': CustomMetric()}
)

Aggregation

Aggregate metrics across epochs or runs.

from swarm.analysis.aggregation import MetricsAggregator

aggregator = MetricsAggregator()

for epoch_metrics in all_epochs:
    aggregator.add(epoch_metrics)

summary = aggregator.summary()
print(f"Mean toxicity: {summary['toxicity_mean']:.3f}")
print(f"Std toxicity: {summary['toxicity_std']:.3f}")