random elections and spatial generation

src/votekit/ballot_generator.py

@@ -8,12 +8,13 @@
 import pickle
 import random
 import warnings
-from typing import Optional, Union, Tuple
+from typing import Optional, Union, Tuple, Callable, Dict, Any
 import apportionment.methods as apportion  # type: ignore
 
 from .ballot import Ballot
 from .pref_profile import PreferenceProfile
 from .pref_interval import combine_preference_intervals, PreferenceInterval
+from votekit.metrics import euclidean_dist
 
 
 def sample_cohesion_ballot_types(
@@ -1948,3 +1949,244 @@ def generate_profile(
         # else return the combined profiles
         else:
             return pp
+
+
+class DSpatial(BallotGenerator):
+    """
+    Spatial model for ballot generation. In some metric space determined
+    by an input distance function, randomly sample each voter's and
+    each candidate's positions from input voter and candidate distributions.
+    Using generate_profile() outputs a ranked profile which is consistent
+    with the sampled positions (respects distances).
+
+    Args:
+        candidates (list[str]): List of candidate strings.
+        voter_dist (Callable[..., np.ndarray], optional): Distribution to sample a single
+            voter's position from, defaults to uniform distribution.
+        voter_params: (Optional[Dict[str, Any]], optional): Parameters to be passed to
+            voter_dist, defaults to None, which creates the unif(0,1) distribution in 2 dimensions.
+        candidate_dist: (Callable[..., np.ndarray], optional): Distribution to sample a
+            single candidate's position from, defaults to uniform distribution.
+        candidate_params: (Optional[Dict[str, Any]], optional): Parameters to be passed
+            to candidate_dist, defaults to None, which creates the unif(0,1)
+            distribution in 2 dimensions.
+        distance: (Callable[[np.ndarray, np.ndarray], float]], optional):
+            Computes distance between a voter and a candidate,
+            defaults to euclidean distance.
+    Attributes:
+        candidates (list[str]): List of candidate strings.
+        voter_dist (Callable[..., np.ndarray], optional): Distribution to sample a single
+            voter's position from, defaults to uniform distribution.
+        voter_params: (Optional[Dict[str, Any]], optional): Parameters to be passed to
+            voter_dist, defaults to None, which creates the unif(0,1) distribution in 2 dimensions.
+        candidate_dist: (Callable[..., np.ndarray], optional): Distribution to sample a
+            single candidate's position from, defaults to uniform distribution.
+        candidate_params: (Optional[Dict[str, Any]], optional): Parameters to be passed
+            to candidate_dist, defaults to None, which creates the unif(0,1)
+            distribution in 2 dimensions.
+        distance: (Callable[[np.ndarray, np.ndarray], float]], optional):
+            Computes distance between a voter and a candidate,
+            defaults to euclidean distance.
+    """
+
+    def __init__(
+        self,
+        candidates: list[str],
+        voter_dist: Callable[..., np.ndarray] = np.random.uniform,
+        voter_params: Optional[Dict[str, Any]] = None,
+        candidate_dist: Callable[..., np.ndarray] = np.random.uniform,
+        candidate_params: Optional[Dict[str, Any]] = None,
+        distance: Callable[[np.ndarray, np.ndarray], float] = euclidean_dist,
+    ):
+        super().__init__(candidates=candidates)
+        self.voter_dist = voter_dist
+        self.candidate_dist = candidate_dist
+
+        if voter_params is None:
+            # default params used for np.random.uniform
+            self.voter_params = {"low": 0.0, "high": 1.0, "size": 2.0}
+        else:
+            self.voter_params = voter_params
+
+        if candidate_params is None:
+            # default params used for np.random.uniform
+            self.candidate_params = {"low": 0.0, "high": 1.0, "size": 2.0}
+        else:
+            self.candidate_params = candidate_params
+
+        self.distance = distance
+
+    def generate_profile(
+        self, number_of_ballots: int, by_bloc: bool = False, seed: Optional[int] = None
+    ) -> Union[PreferenceProfile, Tuple]:
+        """
+        Args:
+            number_of_ballots (int): The number of ballots to generate.
+            by_bloc (bool): Dummy variable from parent class.
+            seed (int, optional): Seed for random generation, defaults to None
+            which resets the random seed.
+
+        Returns:
+            (Union[PreferenceProfile, Tuple]):
+            preference profile (Preference Profile),
+            candidate positions (dict[str, np.ndarray),
+            voter positions (np.ndarray):
+                A tuple containing the preference profile object,
+                a dictionary with each candidate's position in the metric
+                space, and a matrix where each row is a single voter's position
+                in the metric space.
+        """
+
+        np.random.seed(seed)
+        candidate_position_dict = {
+            c: self.candidate_dist(**self.candidate_params) for c in self.candidates
+        }
+        voter_positions = np.array(
+            [self.voter_dist(**self.voter_params) for v in range(number_of_ballots)]
+        )
+
+        ballot_pool = []
+        for v in voter_positions:
+            distance_dict = {
+                i: self.distance(v, c) for i, c, in candidate_position_dict.items()
+            }
+            candidate_order = sorted(distance_dict, key=distance_dict.__getitem__)
+            ballot_pool.append(candidate_order)
+
+        # reset the seed
+        np.random.seed(None)
+
+        return (
+            self.ballot_pool_to_profile(ballot_pool, self.candidates),
+            candidate_position_dict,
+            voter_positions,
+        )
+
+
+class Clustered_DSpatial(BallotGenerator):
+    """
+    Clustered spatial model for ballot generation. In some metric space
+    determined by an input distance function, randomly sample
+    each candidate's positions from input candidate distribution. Then
+    sample voters's positions from a distribution centered around each
+    of the candidate's positions. Using generate_profile()
+    outputs a ranked profile which is consistent
+    with the sampled positions (respects distances).
+
+    Args:
+        candidates (list[str]): List of candidate strings.
+        voter_dist (Callable[..., np.ndarray], optional): Distribution to sample a single
+            voter's position from, defaults to normal(0,1) distribution.
+        voter_params: (Optional[dict[str, Any]], optional): Parameters to be passed to
+            voter_dist, defaults to None, which creates the unif(0,1) distribution in 2 dimensions.
+        candidate_dist: (Callable[..., np.ndarray], optional): Distribution to sample a
+            single candidate's position from, defaults to uniform distribution.
+        candidate_params: (Optional[Dict[str, float]], optional): Parameters to be passed
+            to candidate_dist, defaults None which creates the unif(0,1) 
+            distribution in 2 dimensions.
+        distance: (Callable[[np.ndarray, np.ndarray], float]], optional):
+            Computes distance between a voter and a candidate,
+            defaults to euclidean distance.
+    Attributes:
+        candidates (list[str]): List of candidate strings.
+        voter_dist (Callable[..., np.ndarray], optional): Distribution to sample a single
+            voter's position from, defaults to uniform distribution.
+        voter_params: (Optional[dict[str, Any]], optional): Parameters to be passed to
+            voter_dist, defaults to None, which creates the unif(0,1) distribution in 2 dimensions.
+        candidate_dist: (Callable[..., np.ndarray], optional): Distribution to sample a
+            single candidate's position from, defaults to uniform distribution.
+        candidate_params: (Optional[Dict[str, float]], optional): Parameters to be passed
+            to candidate_dist, defaults None which creates the unif(0,1) 
+            distribution in 2 dimensions.
+        distance: (Callable[[np.ndarray, np.ndarray], float]], optional):
+            Computes distance between a voter and a candidate,
+            defaults to euclidean distance.
+    """
+
+    def __init__(
+        self,
+        candidates: list[str],
+        voter_dist: Callable[..., np.ndarray] = np.random.normal,
+        voter_params: Optional[Dict[str, Any]] = None,
+        candidate_dist: Callable[..., np.ndarray] = np.random.uniform,
+        candidate_params: Optional[Dict[str, Any]] = None,
+        distance: Callable[[np.ndarray, np.ndarray], float] = euclidean_dist,
+    ):
+        super().__init__(candidates=candidates)
+        self.candidate_dist = candidate_dist
+        self.voter_dist = voter_dist
+
+        if voter_params is None:
+            # default params used for np.random.normal
+            self.voter_params = {"loc": 0, "std": np.array(1.0), "size": np.array(2.0)}
+        else:
+            # Trying to ensure that the input distribution's mean is parameterized.
+            # This allows me to center it on the candidates themselves later
+            try:
+                voter_params['loc'] = 0
+                self.voter_dist(**voter_params)
+            except TypeError:
+                raise ValueError("Invalid parameters for the voter distribution, make sure\
+                        that the input distribution has a 'loc' parameter specifying the mean.")
+
+            self.voter_params = voter_params
+
+        if candidate_params is None:
+            # default params used for np.random.uniform
+            self.candidate_params = {"low": 0.0, "high": 1.0, "size": 2.0}
+        else:
+            self.candidate_params = candidate_params
+
+        self.distance = distance
+
+    def generate_profile_with_dict(
+        self, number_of_ballots: dict[str,int], by_bloc: bool = False, seed: Optional[int] = None
+    ) -> Union[PreferenceProfile, Tuple]:
+        """
+        Args:
+            number_of_ballots (dict[str, int]): The number of voters attributed
+                        to each candidate {candidate string: # voters}
+            by_bloc (bool): Dummy variable from parent class.
+            seed (int, optional): Seed for random generation, defaults to None
+                which resets the random seed.
+
+        Returns:
+            (Union[PreferenceProfile, Tuple]):
+            preference profile (Preference Profile),
+            candidate positions (dict[str, np.ndarray),
+            voter positions (np.ndarray):
+                A tuple containing the preference profile object,
+                a dictionary with each candidate's position in the metric
+                space, and a matrix where each row is a single voter's position
+                in the metric space.
+        """
+
+        np.random.seed(seed)
+        candidate_position_dict = {
+            c: self.candidate_dist(**self.candidate_params) for c in self.candidates
+        }
+        voter_positions = []
+        for c in self.candidates:
+            for v in range(number_of_ballots[c]):
+                self.voter_params['loc'] = candidate_position_dict[c]
+                voter_positions.append(self.voter_dist(**self.voter_params))
+
+        voter_positions_array = np.vstack(voter_positions)
+
+        ballot_pool = []
+        for v_position in voter_positions:
+            distance_dict = {
+                c: self.distance(v_position, c_position)
+                for c, c_position, in candidate_position_dict.items()
+            }
+            candidate_order = sorted(distance_dict, key=distance_dict.__getitem__)
+            ballot_pool.append(candidate_order)
+
+        # reset the seed
+        np.random.seed(None)
+
+        return (
+            self.ballot_pool_to_profile(ballot_pool, self.candidates),
+            candidate_position_dict,
+            voter_positions_array,
+        )

src/votekit/elections/__init__.py

@@ -13,6 +13,9 @@
     IRV,
     HighestScore,
     Cumulative,
+    RandomDictator,
+    BoostedRandomDictator,
+    PluralityVeto,
 )
 
 from .transfers import fractional_transfer, random_transfer  # noqa