tessl/pypi-gerrychain
Use Markov chain Monte Carlo to analyze districting plans and gerrymanders
—
Pending
proposal-algorithms.mddocs/
Proposal Algorithms
Generate new districting plans through various algorithms including ReCom (recombination), random flips, and tree-based partitioning. Proposal functions are the core mechanism for exploring the space of possible redistricting plans.
Capabilities
ReCom (Recombination) Algorithm
The primary proposal method that creates new plans by merging adjacent districts and repartitioning them using spanning trees.
def recom(
partition: Partition,
pop_col: str,
pop_target: Union[int, float],
epsilon: float,
node_repeats: int = 1,
region_surcharge: Optional[Dict] = None,
method: Callable = bipartition_tree
) -> Partition:
"""
Generate new partition using ReCom (recombination) algorithm.
The algorithm:
1. Selects a random edge that crosses district boundaries
2. Merges the two districts connected by that edge
3. Generates a random spanning tree of the merged area
4. Cuts the tree to create two balanced districts
Parameters:
- partition (Partition): Current partition to propose from
- pop_col (str): Column name for population data
- pop_target (Union[int, float]): Target population per district
- epsilon (float): Population balance tolerance (e.g., 0.05 for 5%)
- node_repeats (int): Number of spanning tree attempts per node
- region_surcharge (Optional[Dict]): Additional constraints by region
- method (Callable): Tree bipartition method to use
Returns:
Partition: New partition with recombined districts
"""Usage example:
from gerrychain.proposals import recom
# Use in Markov chain
chain = MarkovChain(
proposal=recom,
constraints=constraints,
accept=always_accept,
initial_state=partition,
total_steps=1000
)
# Single proposal
new_partition = recom(current_partition)ReCom Class Implementation
The ReCom algorithm is also available as a class for more advanced usage:
class ReCom:
def __init__(
self,
pop_col: str,
ideal_pop: Union[int, float],
epsilon: float,
node_repeats: int = 1
) -> None:
"""
ReCom proposal class for reusable configuration.
Parameters:
- pop_col (str): Population column name
- ideal_pop (Union[int, float]): Ideal population per district
- epsilon (float): Population balance tolerance
- node_repeats (int): Number of spanning tree attempts per node
Returns:
None
"""
def __call__(self, partition: Partition) -> Partition:
"""
Apply ReCom proposal to partition.
Parameters:
- partition (Partition): Current partition
Returns:
Partition: New partition from ReCom
"""
def reversible_recom(
partition: Partition,
pop_col: str,
pop_target: Union[int, float],
epsilon: float,
node_repeats: int = 1,
method: Callable = bipartition_tree
) -> Partition:
"""
Reversible ReCom proposal maintaining detailed balance.
Parameters:
- partition (Partition): Current partition
- pop_col (str): Population column name
- pop_target (Union[int, float]): Target population per district
- epsilon (float): Population balance tolerance
- node_repeats (int): Number of spanning tree attempts per node
- method (Callable): Tree bipartition method to use
Returns:
Partition: New partition with reversible ReCom
"""
def spectral_recom(
partition: Partition,
pop_col: str,
pop_target: Union[int, float],
epsilon: float,
node_repeats: int = 1
) -> Partition:
"""
ReCom proposal using spectral clustering for tree cuts.
Parameters:
- partition (Partition): Current partition
- pop_col (str): Population column name
- pop_target (Union[int, float]): Target population per district
- epsilon (float): Population balance tolerance
- node_repeats (int): Number of spanning tree attempts per node
Returns:
Partition: New partition using spectral ReCom
"""Random Flip Proposals
Simple proposals that flip individual units or small groups between adjacent districts.
def propose_random_flip(partition: Partition) -> Partition:
"""
Propose flipping a single random unit to an adjacent district.
Parameters:
- partition (Partition): Current partition
Returns:
Partition: New partition with one unit flipped
"""
def propose_several_flips(partition: Partition, k: int = 2) -> Partition:
"""
Propose flipping several units simultaneously.
Parameters:
- partition (Partition): Current partition
- k (int): Number of units to flip
Returns:
Partition: New partition with k units flipped
"""
def propose_chunk_flip(partition: Partition, k: int = 3) -> Partition:
"""
Propose flipping a connected chunk of k units to adjacent district.
Parameters:
- partition (Partition): Current partition
- k (int): Size of chunk to flip
Returns:
Partition: New partition with chunk flipped
"""
def slow_reversible_propose(partition: Partition) -> Partition:
"""
Reversible proposal method with detailed balance properties.
Parameters:
- partition (Partition): Current partition
Returns:
Partition: New partition maintaining detailed balance
"""Tree-Based Partitioning
Low-level tree algorithms used by ReCom and other proposal methods.
def recursive_tree_part(
graph: Graph,
parts: List[Set[NodeId]],
pop_target: float,
pop_col: str = "population",
epsilon: float = 0.05
) -> Dict[NodeId, int]:
"""
Recursively partition graph into balanced parts using spanning trees.
Parameters:
- graph (Graph): Graph to partition
- parts (List[Set[NodeId]]): Target partition structure
- pop_target (float): Target population per part
- pop_col (str): Population attribute name
- epsilon (float): Population balance tolerance
Returns:
Dict[NodeId, int]: Assignment of nodes to parts
"""
def bipartition_tree(
graph: Graph,
pop_col: str = "population",
pop_target: float = None,
epsilon: float = 0.05
) -> Dict[NodeId, int]:
"""
Partition graph into two balanced parts using spanning tree.
Parameters:
- graph (Graph): Graph to partition
- pop_col (str): Population attribute name
- pop_target (float, optional): Target population per part
- epsilon (float): Population balance tolerance
Returns:
Dict[NodeId, int]: Assignment of nodes to two parts (0 or 1)
"""
def bipartition_tree_random(
graph: Graph,
pop_col: str = "population",
epsilon: float = 0.05,
node_repeats: int = 1
) -> Dict[NodeId, int]:
"""
Random bipartition using spanning tree with multiple attempts.
Parameters:
- graph (Graph): Graph to partition
- pop_col (str): Population attribute name
- epsilon (float): Population balance tolerance
- node_repeats (int): Number of random attempts per node
Returns:
Dict[NodeId, int]: Assignment of nodes to two parts
"""Spanning Tree Operations
Generate and manipulate spanning trees for tree-based partitioning algorithms.
def random_spanning_tree(graph: Graph) -> Graph:
"""
Generate a random spanning tree using Wilson's algorithm.
Parameters:
- graph (Graph): Input graph
Returns:
Graph: Random spanning tree as NetworkX graph
"""
def uniform_spanning_tree(graph: Graph) -> Graph:
"""
Generate uniformly random spanning tree.
Parameters:
- graph (Graph): Input graph
Returns:
Graph: Uniformly random spanning tree
"""
def wilson_algorithm(graph: Graph, root: NodeId = None) -> Graph:
"""
Wilson's algorithm for uniform random spanning trees.
Parameters:
- graph (Graph): Input graph
- root (NodeId, optional): Root node for tree
Returns:
Graph: Random spanning tree generated by Wilson's algorithm
"""
def find_balanced_edge_cuts(
tree: Graph,
parts: int,
pop_col: str = "population",
epsilon: float = 0.05
) -> List[Tuple[NodeId, NodeId]]:
"""
Find edge cuts that create balanced partitions of spanning tree.
Parameters:
- tree (Graph): Spanning tree
- parts (int): Number of parts to create
- pop_col (str): Population attribute name
- epsilon (float): Population balance tolerance
Returns:
List[Tuple[NodeId, NodeId]]: Edges to cut for balanced partition
"""
def predecessors(graph: Graph, root: NodeId) -> Dict[NodeId, NodeId]:
"""
Find predecessor relationships in tree rooted at given node.
Parameters:
- graph (Graph): Tree graph
- root (NodeId): Root node
Returns:
Dict[NodeId, NodeId]: Predecessor mapping for each node
"""
def successors(graph: Graph, root: NodeId) -> Dict[NodeId, List[NodeId]]:
"""
Find successor relationships in tree rooted at given node.
Parameters:
- graph (Graph): Tree graph
- root (NodeId): Root node
Returns:
Dict[NodeId, List[NodeId]]: Successor mapping for each node
"""
def wilson_algorithm(graph: Graph, root: NodeId = None) -> Graph:
"""
Wilson's algorithm for uniform random spanning trees.
Parameters:
- graph (Graph): Input graph
- root (NodeId, optional): Root node for tree
Returns:
Graph: Random spanning tree generated by Wilson's algorithm
"""
def get_seed_chunks(
graph: Graph,
num_chunks: int,
pop_target: float,
pop_col: str = "population",
epsilon: float = 0.05,
node_repeats: int = 1
) -> List[Set[NodeId]]:
"""
Generate seed districts for recursive partitioning.
Parameters:
- graph (Graph): Graph to partition
- num_chunks (int): Number of seed chunks to create
- pop_target (float): Target population per chunk
- pop_col (str): Population attribute name
- epsilon (float): Population balance tolerance
- node_repeats (int): Number of attempts per node
Returns:
List[Set[NodeId]]: List of seed district node sets
"""Advanced Proposal Patterns
Examples of how to create custom proposal functions and use advanced features:
from gerrychain.proposals import recom, propose_random_flip
import random
# Custom proposal combining multiple methods
def mixed_proposal(partition):
"""Custom proposal that uses ReCom 80% of time, random flip 20%."""
if random.random() < 0.8:
return recom(partition)
else:
return propose_random_flip(partition)
# Proposal with custom parameters
def chunked_recom(partition, chunk_size=5):
"""ReCom followed by chunk flip for fine-tuning."""
step1 = recom(partition)
return propose_chunk_flip(step1, k=chunk_size)
# Use in analysis
chain = MarkovChain(
proposal=mixed_proposal, # Custom proposal function
constraints=constraints,
accept=always_accept,
initial_state=partition,
total_steps=10000
)
# Sequential proposals for multi-step changes
def multi_step_proposal(partition, steps=3):
"""Apply multiple proposal steps."""
current = partition
for _ in range(steps):
current = propose_random_flip(current)
return currentTypes
ProposalFunction = Callable[[Partition], Partition]
NodeId = Union[int, str]
DistrictId = intInstall with Tessl CLI
npx tessl i tessl/pypi-gerrychain