Shapley Value Strategy#

class rizemind.strategies.contribution.shapley.shapley_value_strategy.ShapleyValueStrategy(strategy: ~flwr.server.strategy.strategy.Strategy, swarm: ~rizemind.strategies.contribution.shapley.shapley_value_strategy.SupportsShapleyValueStrategy, coalition_to_score_fn: ~collections.abc.Callable[[~rizemind.strategies.contribution.shapley.trainer_set.TrainerSetAggregate], float] | None = None, aggregate_coalition_metrics_fn: ~collections.abc.Callable[[list[~rizemind.strategies.contribution.shapley.trainer_set.TrainerSetAggregate]], dict[str, bool | bytes | float | int | str]] | None = None, shapley_sampling_strat: ~rizemind.strategies.contribution.sampling.sets_sampling_strat.SetsSamplingStrategy = <rizemind.strategies.contribution.sampling.all_sets.AllSets object>, contribution_calculator: ~rizemind.strategies.contribution.calculators.calculator.ContributionCalculator = <rizemind.strategies.contribution.calculators.shapley_value.ShapleyValueCalculator object>)[source]

Bases: Strategy

Federated learning strategy using Shapley values for contribution calculation.

This strategy extends the Flower Strategy to incorporate Shapley value-based contribution calculation. It creates coalitions of trainers, aggregates their models, and evaluates each coalition’s performance to compute fair contribution scores using the Shapley value method.

strategy

The underlying federated learning strategy for aggregation.

Type:: flwr.server.strategy.strategy.Strategy

swarm

The swarm manager handling reward distribution and round progression.

Type:: rizemind.strategies.contribution.shapley.shapley_value_strategy.SupportsShapleyValueStrategy

coalition_to_score_fn

Optional function to compute a score from a coalition aggregate.

Type:: collections.abc.Callable[[rizemind.strategies.contribution.shapley.trainer_set.TrainerSetAggregate], float] | None

last_round_parameters

Parameters from the previous round.

Type:: flwr.common.typing.Parameters | None

aggregate_coalition_metrics

Optional function to aggregate metrics across coalitions.

Type:: collections.abc.Callable[[list[rizemind.strategies.contribution.shapley.trainer_set.TrainerSetAggregate]], dict[str, bool | bytes | float | int | str]] | None

sets_sampling_strat

Strategy for sampling trainer subsets/coalitions.

Type:: rizemind.strategies.contribution.sampling.sets_sampling_strat.SetsSamplingStrategy

set_aggregates

Store for coalitions.

Type:: rizemind.strategies.contribution.shapley.trainer_set.TrainerSetAggregateStore

contribution_calculator

Calculator for computing Shapley value contributions.

Type:: rizemind.strategies.contribution.calculators.calculator.ContributionCalculator

aggregate_coalition_metrics: Callable[[list[TrainerSetAggregate]], dict[str, bool | bytes | float | int | str]] | None = None

Aggregate client training results and form coalitions.

Creates coalitions from client fit results and delegates parameter aggregation to the underlying strategy.

Parameters:

server_round – The current server round number.
results – List of tuples containing client proxies and their fit results.
failures – List of any failed client results.

Returns:

A tuple containing the aggregated parameters (or None) and a dictionary of metrics.

close_round(round_id: int) → tuple[float, dict[str, bool | bytes | float | int | str]][source]

Finalize the current round by computing contributions and distributing rewards.

Computes trainer contributions, normalizes scores, distributes rewards, and prepares for the next round.

Parameters:: round_id – The current round identifier.
Returns:: A tuple containing the best coalition loss and aggregated metrics.

coalition_to_score_fn: Callable[[TrainerSetAggregate], float] | None = None

compute_contributions(round_id: int, coalitions: list[TrainerSetAggregate] | None) → list[PlayerScore][source]

Compute Shapley value contribution score for each trainer.

Uses the contribution calculator to determine each trainer’s contribution to the overall model performance based on coalition evaluations.

Parameters:

round_id – The current round identifier.
coalitions – Optional list of coalitions to compute contributions from.
None (If)
coalitions. (uses all available)

Returns:

List of player scores containing trainer addresses and their contribution values.

configure_fit(server_round: int, parameters: Parameters, client_manager: ClientManager) → list[tuple[ClientProxy, FitIns]][source]

Configure the next round of training for clients.

Selects the coalition where every trainer participated from the previous round and uses its parameters as the starting point for the next training round.

Parameters:

server_round – The current server round number.
parameters – Current model parameters.
client_manager – Manager handling available clients.

Returns:

List of tuples containing client proxies and their fit instructions.

contribution_calculator: ContributionCalculator

create_coalitions(server_round: int, results: list[tuple[ClientProxy, FitRes]]) → list[TrainerSetAggregate][source]

Create coalitions from client training results.

Samples trainer subsets using the sampling strategy, aggregates parameters for each subset, and stores the resulting coalition aggregates.

Parameters:

server_round – The current server round number.
results – List of tuples containing client proxies and their fit results.

Returns:

List of created coalitions.

Raises:

ValueError – If no aggregate parameters are returned for a trainer set.

evaluate_coalitions() → tuple[float, dict[str, bool | bytes | float | int | str]][source]

Evaluate all coalitions and determine the best performance.

Calculates loss values for all coalitions and optionally aggregates metrics.

Returns:: A tuple containing the minimum coalition loss and aggregated metrics.

get_coalition(id: str) → TrainerSetAggregate[source]

Get a specific coalition by ID.

Parameters:: id – The identifier of the coalition.
Returns:: The requested coalition aggregate.
Raises:: Exception – If the coalition with the given ID is not found.

get_coalition_score(coalition: TrainerSetAggregate) → float[source]

Get the performance score for a coalition.

If no coalition_to_score_fn is provided it defaults to the loss value. The loss value is inversed since higher loss means lower performance.

Parameters:: coalition – The coalition aggregate to score.
Returns:: The performance score of the coalition.
Raises:: Exception – If the coalition has not been evaluated.

get_coalitions() → list[TrainerSetAggregate][source]: Returns all coalitions.

initialize_parameters(client_manager: ClientManager) → Parameters | None[source]

Delegate the initialization of model parameters to the underlying strategy.

Parameters:: client_manager – Manager handling available clients.
Returns:: The initialized model parameters, or None if not applicable.

last_round_parameters: Parameters | None

normalize_contribution_scores(trainers_and_contributions: list[PlayerScore]) → list[PlayerScore][source]

Normalize contribution scores to ensure non-negative values.

Parameters:: trainers_and_contributions – List of player scores to normalize.
Returns:: List of player scores with negative values clamped to zero.

select_aggregate() → TrainerSetAggregate | None[source]

Select the coalition aggregate to use for the next training round.

Selects the coalition with the highest number of members as the base for the next round.

Returns:: The selected coalition aggregate, or None if no coalitions exist.

set_aggregates: TrainerSetAggregateStore

sets_sampling_strat: SetsSamplingStrategy

strategy: Strategy

swarm: SupportsShapleyValueStrategy

class rizemind.strategies.contribution.shapley.shapley_value_strategy.SupportsShapleyValueStrategy(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the interface for swarm management in Shapley value strategies.

This protocol specifies the required methods for managing trainer compensation and round progression in a federated learning swarm using Shapley value-based contribution calculation.

distribute(trainer_scores: list[tuple[ChecksumAddress, float]]) → str[source]

Distribute rewards to trainers based on their contribution scores.

Parameters:

trainer_scores – List of tuples containing trainer addresses and their
scores. (corresponding contribution)

Returns:

Transaction hash or confirmation string of the distribution operation.

next_round(round_id: int, n_trainers: int, model_score: float, total_contributions: float) → str[source]

Advance to the next training round and record round statistics.

Parameters:

round_id – The identifier of the current round.
n_trainers – Number of trainers participating in this round.
model_score – Performance score of the selected model for next round.
total_contributions – Sum of all trainer contribution scores.

Returns:

Transaction hash or confirmation string of the next round operation.