Skip to content

gdm

desdeo.gdm.gdmtools

This module contains tools for group decision making such as manipulating set of preferences.

agg_aspbounds 

agg_aspbounds(
    po_list: list[dict[str, float]], problem: Problem
) -> tuple[dict[str, float], dict[str, float]]

Aggregate a set of preferences into shared aspiration levels and bounds.

For each objective, the aggregated aspiration level is the most optimistic value across the given preferences (the maximum for objectives to be maximized, the minimum for objectives to be minimized), while the aggregated bound is the least optimistic value.

Parameters:

Name Type Description Default
po_list list[dict[str, float]]

a list of preferences, where each preference maps objective symbols to values.

 required
problem Problem

the problem the preferences relate to. Used to determine the objective symbols and whether each is maximized.

 required

Returns:

Type Description
tuple[dict[str, float], dict[str, float]]

tuple[dict[str, float], dict[str, float]]: the aggregated aspiration levels and the aggregated bounds, each mapping objective symbols to values.
Source code in desdeo/gdm/gdmtools.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def agg_aspbounds(po_list: list[dict[str, float]], problem: Problem) -> tuple[dict[str, float], dict[str, float]]:
    """Aggregate a set of preferences into shared aspiration levels and bounds.

    For each objective, the aggregated aspiration level is the most optimistic
    value across the given preferences (the maximum for objectives to be
    maximized, the minimum for objectives to be minimized), while the aggregated
    bound is the least optimistic value.

    Args:
        po_list (list[dict[str, float]]): a list of preferences, where each
            preference maps objective symbols to values.
        problem (Problem): the problem the preferences relate to. Used to
            determine the objective symbols and whether each is maximized.

    Returns:
        tuple[dict[str, float], dict[str, float]]: the aggregated aspiration
            levels and the aggregated bounds, each mapping objective symbols to
            values.
    """
    agg_aspirations = {}
    agg_bounds = {}

    for obj in problem.objectives:
        if obj.maximize:
            agg_aspirations.update({obj.symbol: max(s[obj.symbol] for s in po_list)})
            agg_bounds.update({obj.symbol: min(s[obj.symbol] for s in po_list)})
        else:
            agg_aspirations.update({obj.symbol: min(s[obj.symbol] for s in po_list)})
            agg_bounds.update({obj.symbol: max(s[obj.symbol] for s in po_list)})

    return agg_aspirations, agg_bounds

dict_of_rps_to_list_of_rps 

dict_of_rps_to_list_of_rps(
    reference_points: dict[str, dict[str, float]],
) -> list[dict[str, float]]

Convert a dict of preferences keyed by decision maker into an ordered list of preferences.

Source code in desdeo/gdm/gdmtools.py 
7
8
9
def dict_of_rps_to_list_of_rps(reference_points: dict[str, dict[str, float]]) -> list[dict[str, float]]:
    """Convert a dict of preferences keyed by decision maker into an ordered list of preferences."""
    return list(reference_points.values())

list_of_rps_to_dict_of_rps 

list_of_rps_to_dict_of_rps(
    reference_points: list[dict[str, float]],
) -> dict[str, dict[str, float]]

Convert an ordered list of preferences into a dict keyed by decision maker (DM1, DM2, ...).

Source code in desdeo/gdm/gdmtools.py 
12
13
14
def list_of_rps_to_dict_of_rps(reference_points: list[dict[str, float]]) -> dict[str, dict[str, float]]:
    """Convert an ordered list of preferences into a dict keyed by decision maker (``DM1``, ``DM2``, ...)."""
    return {f"DM{i + 1}": rp for i, rp in enumerate(reference_points)}

scale_delta 

scale_delta(problem: Problem, d: float) -> dict[str, float]

Scale a step size into a per-objective delta based on the objective ranges.

For each objective, the delta has magnitude equal to the fraction d of the objective's range, i.e. the distance between the ideal and nadir points.

Parameters:

Name Type Description Default
problem Problem

the problem whose ideal and nadir points define the objective ranges.

 required
d float

the fraction of each objective's range to use as the delta.

 required

Returns:

Type Description
dict[str, float]

dict[str, float]: a mapping from objective symbols to the scaled delta.
Source code in desdeo/gdm/gdmtools.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def scale_delta(problem: Problem, d: float) -> dict[str, float]:
    """Scale a step size into a per-objective delta based on the objective ranges.

    For each objective, the delta has magnitude equal to the fraction `d` of the
    objective's range, i.e. the distance between the ideal and nadir points.

    Args:
        problem (Problem): the problem whose ideal and nadir points define the
            objective ranges.
        d (float): the fraction of each objective's range to use as the delta.

    Returns:
        dict[str, float]: a mapping from objective symbols to the scaled delta.
    """
    delta = {}
    ideal = problem.get_ideal_point()
    nadir = problem.get_nadir_point()

    for obj in problem.objectives:
        if obj.maximize:
            delta.update({obj.symbol: d * (ideal[obj.symbol] - nadir[obj.symbol])})
        else:
            delta.update({obj.symbol: d * (nadir[obj.symbol] - ideal[obj.symbol])})
    return delta

desdeo.gdm.voting_rules

This module contains voting rules for group decision making such as majority rule.

consensus_rule 

consensus_rule(
    votes: dict[str, int], min_votes: int
) -> list[int]

Choose all options that have at least min_votes votes.

Parameters:

Name Type Description Default
votes dict[str, int]

A dictionary mapping voter IDs to their votes.

 required
min_votes int

The minimum number of votes required for an option to be selected.

 required
Source code in desdeo/gdm/voting_rules.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def consensus_rule(votes: dict[str, int], min_votes: int) -> list[int]:
    """Choose all options that have at least min_votes votes.

    Args:
        votes (dict[str, int]): A dictionary mapping voter IDs to their votes.
        min_votes (int): The minimum number of votes required for an option to be selected.

    """
    if min_votes <= 0:
        raise ValueError("min_votes must be greater than 0.")
    if min_votes > len(votes):
        raise ValueError("min_votes cannot be greater than the number of voters.")
    counts = Counter(votes.values())
    return [vote for vote, c in counts.items() if c >= min_votes]

majority_rule 

majority_rule(votes: dict[str, int]) -> int | None

Choose the option that has more than half of the votes.

Parameters:

Name Type Description Default
votes dict[str, int]

A dictionary mapping voter IDs to their votes.

 required

Returns:

Type Description
int | None

int | None: The option that has more than half of the votes, or None if no such option exists.
Source code in desdeo/gdm/voting_rules.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
def majority_rule(votes: dict[str, int]) -> int | None:
    """Choose the option that has more than half of the votes.

    Args:
        votes (dict[str, int]): A dictionary mapping voter IDs to their votes.

    Returns:
        int | None: The option that has more than half of the votes, or None if no such option exists.
    """
    counts = Counter(votes.values())
    all_votes = sum(counts.values())
    for vote, c in counts.items():
        if c > all_votes // 2:
            return vote
    return None

plurality_rule 

plurality_rule(votes: dict[str, int]) -> list[int]

Choose the option that has the most votes.

Parameters:

Name Type Description Default
votes dict[str, int]

A dictionary mapping voter IDs to their votes.

 required

Returns:

Type Description
list[int]

list[int]: A list of options that have the most votes (in case of a tie).
Source code in desdeo/gdm/voting_rules.py 
23
24
25
26
27
28
29
30
31
32
33
34
def plurality_rule(votes: dict[str, int]) -> list[int]:
    """Choose the option that has the most votes.

    Args:
        votes (dict[str, int]): A dictionary mapping voter IDs to their votes.

    Returns:
        list[int]: A list of options that have the most votes (in case of a tie).
    """
    counts = Counter(votes.values())
    max_votes = max(counts.values())
    return [vote for vote, c in counts.items() if c == max_votes]

desdeo.gdm.score_bands

Implements a interactive SCORE bands based GDM.

SCOREBandsGDMConfig

Bases: BaseModel

Configuration for the SCORE bands based GDM.

Source code in desdeo/gdm/score_bands.py 
10
11
12
13
14
15
16
17
18
19
20
class SCOREBandsGDMConfig(BaseModel):
    """Configuration for the SCORE bands based GDM."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    score_bands_config: SCOREBandsConfig = Field(default_factory=SCOREBandsConfig)
    """Configuration for the SCORE bands method."""
    minimum_votes: int = Field(default=1, gt=0)
    """Minimum number of votes required to select a cluster."""
    from_iteration: int | None
    """The iteration number from which to consider the clusters. Set to None if method is initializing."""
from_iteration instance-attribute 
from_iteration: int | None

The iteration number from which to consider the clusters. Set to None if method is initializing.

minimum_votes class-attribute instance-attribute 
minimum_votes: int = Field(default=1, gt=0)

Minimum number of votes required to select a cluster.

score_bands_config class-attribute instance-attribute 
score_bands_config: SCOREBandsConfig = Field(
    default_factory=SCOREBandsConfig
)

Configuration for the SCORE bands method.

SCOREBandsGDMResult

Bases: BaseModel

Result of the SCORE bands based GDM.

Source code in desdeo/gdm/score_bands.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class SCOREBandsGDMResult(BaseModel):
    """Result of the SCORE bands based GDM."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    votes: dict[str, int] | None = Field(default=None)
    """The votes given by the decision makers."""
    score_bands_result: SCOREBandsResult
    """Result of the SCORE bands method."""
    relevant_ids: list[int]
    """IDs of the relevant solutions in the current iteration. Assumes that data is not modified between iterations."""
    # If the data keeps changing, we need to store the actual data instead of just the IDs.
    iteration: int
    previous_iteration: int | None
    """The previous iteration number, if any."""
previous_iteration instance-attribute 
previous_iteration: int | None

The previous iteration number, if any.

relevant_ids instance-attribute 
relevant_ids: list[int]

IDs of the relevant solutions in the current iteration. Assumes that data is not modified between iterations.

score_bands_result instance-attribute 
score_bands_result: SCOREBandsResult

Result of the SCORE bands method.

votes class-attribute instance-attribute 
votes: dict[str, int] | None = Field(default=None)

The votes given by the decision makers.

score_bands_gdm 

score_bands_gdm(
    data: DataFrame,
    config: SCOREBandsGDMConfig,
    state: list[SCOREBandsGDMResult],
    votes: dict[str, int] | None = None,
) -> list[SCOREBandsGDMResult]

Run the SCORE bands based interactive GDM.

Parameters:

Name Type Description Default
data DataFrame

The data to run the GDM on.

 required
config SCOREBandsGDMConfig

Configuration for the GDM.

 required
state list[SCOREBandsGDMResult]

List of previous state of the GDM. Empty list if first iteration.

 required
votes dict[str, int] | None

Votes from the decision makers. Defaults to None.

 None

Raises:

Type Description
ValueError

Both state and votes must be provided or neither.

Returns:

Type Description
list[SCOREBandsGDMResult]

list[SCOREBandsGDMResult]: The updated state of the GDM.
Source code in desdeo/gdm/score_bands.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def score_bands_gdm(
    data: pl.DataFrame,
    config: SCOREBandsGDMConfig,
    state: list[SCOREBandsGDMResult],
    votes: dict[str, int] | None = None,
) -> list[SCOREBandsGDMResult]:
    """Run the SCORE bands based interactive GDM.

    Args:
        data (pl.DataFrame): The data to run the GDM on.
        config (SCOREBandsGDMConfig): Configuration for the GDM.
        state (list[SCOREBandsGDMResult]): List of previous state of the GDM. Empty list if first iteration.
        votes (dict[str, int] | None, optional): Votes from the decision makers. Defaults to None.

    Raises:
        ValueError: Both state and votes must be provided or neither.

    Returns:
        list[SCOREBandsGDMResult]: The updated state of the GDM.
    """
    if bool(state) != bool(votes):
        raise ValueError("Both state and votes must be provided or neither.")
    if votes is None:
        # First iteration. No votes yet.
        score_bands_result = score_json(data, config.score_bands_config)
        return [
            SCOREBandsGDMResult(
                score_bands_result=score_bands_result,
                relevant_ids=list(range(len(data))),
                iteration=1,
                previous_iteration=None,
            )
        ]
    if not state:
        raise ValueError("State must be provided if votes are provided.")
    if config.from_iteration is None:
        raise ValueError("from_iteration must be set in the config for subsequent iterations.")

    winning_clusters = consensus_rule(votes, config.minimum_votes)

    index_column_name = "index"
    if index_column_name in data.columns:
        index_column_name = "index_"
    cluster_column_name = "cluster"
    if cluster_column_name in data.columns:
        cluster_column_name = "cluster_"

    current_iteration = state[-1].iteration + 1

    clusters = state[config.from_iteration - 1].score_bands_result.clusters
    relevant_data = (
        data.with_row_index(name=index_column_name)  # Add index column
        .filter(
            pl.col(index_column_name).is_in(state[config.from_iteration - 1].relevant_ids)
        )  # Get the solutions from previous iteration
        .with_columns(pl.Series(cluster_column_name, clusters))  # Add clustering information from last iteration
        .filter(pl.col(cluster_column_name).is_in(winning_clusters))  # Keep only winning clusters
        .drop(cluster_column_name)  # Drop cluster column
    )

    relevant_ids = relevant_data[index_column_name].to_list()
    relevant_data = relevant_data.drop(index_column_name)  # Drop index column

    return [
        *state,
        SCOREBandsGDMResult(
            votes=votes,
            score_bands_result=score_json(relevant_data, config.score_bands_config),
            relevant_ids=relevant_ids,
            iteration=current_iteration,
            previous_iteration=config.from_iteration,
        ),
    ]