Utility maximizer problem

boilerplate.py

"""The problem is to maximize utility/subjective value given a number of
participants and a quantity of things of different categories.

The original phrasing of the problem is as follows:
Math concept tangentially related to both social choice theory and logistics:
given N people and N unique things, and further given subjective value/utility
ratings from each person for each thing, devise an algorithm that distributes the
things to the people while maximizing total utility. And a related problem is
generalizing it to N people, K unique categories of things, and a function M
mapping category to integer count of the things of that category available.

Utility ratings per person would be a function mapping unique thing/category to
an integer in some arbitrary range, say 0 to +10.

Utility of a distribution/apportionment strategy is defined by running the
assigned thing/category through the person's utility function, then multiplying
by the number of things of that category assigned to that person.

The brute force approach is to generate all permutations and then score them, so
the goal is to get to an optimal solution in some small, finite number of tries.
"""

from dataclasses import dataclass, field
from hashlib import sha256
from sys import argv
from typing import Protocol, runtime_checkable
from uuid import uuid4


@dataclass
class Category:
    name: str = field()
    details: str = field(default_factory=str)

    def __hash__(self) -> int:
        """Makes Category hashable for inclusion in sets and as dict keys."""
        return hash((self.name, self.details))

@dataclass
class Thing:
    category: Category = field()
    uuid: bytes = field(default_factory=uuid4)

    def __hash__(self) -> int:
        """Makes the Thing hashable for inclusion in sets and as dict keys."""
        return hash((self.category, self.uuid))

class PersonProtocol(Protocol):
    def utility_of(self, thing: Thing) -> int|float:
        """Returns the person's utility rating for a given thing."""
        ...

    def __hash__(self) -> int:
        """Makes the person hashable for inclusion in sets and as dict keys.
            Should be different for each different person object.
        """
        ...

@dataclass
class Person:
    mapping: dict = field(default_factory=dict)
    uuid: bytes = field(default_factory=uuid4)

    def add_rating(self, thing: Thing, rating: int|float):
        """Helper method to add a rating for a Thing to this Person."""
        self.mapping[thing] = rating

    def utility_of(self, thing: Thing) -> int|float:
        """Returns the person's utility rating for a given thing."""
        return self.mapping.get(thing, 0)

    def __hash__(self) -> int:
        """Makes the person hashable for inclusion in sets and as dict keys.
            Should be different for each different person object.
        """
        return hash(self.uuid)

@dataclass
class Allocation:
    person: Person = field()
    thing: Thing = field()

@dataclass
class Solution:
    result: list[Allocation] = field()

    def score(self) -> int:
        """Calculates the total utility/subjective value of the solution."""
        total = 0
        for allocation in self.result:
            total += allocation.person.utility_of(allocation.thing)
        return total

    def __lt__(self, other) -> bool:
        """Magic method for comparing with < symbol."""
        if not isinstance(other, Solution):
            raise TypeError('incomparable types')
        return self.score() < other.score()

    def __gt__(self, other) -> bool:
        """Magic method for comparing with > symbol."""
        if not isinstance(other, Solution):
            raise TypeError('incomparable types')
        return self.score() > other.score()

class UtilityOptimizer(Protocol):
    def optimize(
            self, persons: list[PersonProtocol], things: dict[Thing, int],
            **options
        ) -> Solution:
        """Runs the optimization algorithm and returns a Solution assigning
            things to persons. Accepts a list of persons and a dict mapping
            things to quantities.
        """
        ...

@dataclass
class Randomizer:
    seed: bytes = field()
    calls: int = field(default=0)

    def next_state(self) -> bytes:
        state = sha256(self.seed + self.calls.to_bytes(4, 'big')).digest()
        self.calls += 1
        return state

    def bytes(self, count: int) -> bytes:
        state = self.next_state()
        while len(state) < count:
            state += self.next_state()
        return state[:count]

    def int(self, maximum: int = 2**256-1) -> int:
        return int.from_bytes(self.next_state(), 'big') % maximum

    def str(self, length: int = 16) -> str:
        n_bytes = length // 2 + 1
        return self.bytes(n_bytes).hex()[:length]

def setup(
    n_persons: int = 100, n_things: int = 100,
    quantity_min: int = 1, quantity_max: int = 1,
    rating_min: int = 0, rating_max: int = 10,
    seed: bytes = b'deterministic testing',
    **_
) -> tuple[list[Person], dict[Thing, int]]:
    """Creates a deterministic, randomized optimization problem setup
        comprised of a list of Persons and a dict mapping Things to 
        quantities. Use with `persons, things = setup()`. A different
        seed can be supplied to get a different randomized setup with
        otherwise identical parameters.
    """
    random = Randomizer(seed)
    categories = [Category(random.str()) for _ in range(n_things)]
    persons = []
    things = {}

    for cat in categories:
        t = Thing(cat)
        things[t] = max(quantity_min, random.int(quantity_max))

    while len(persons) < n_persons:
        p = Person()
        for t in things:
            p.add_rating(t, max(rating_min, random.int(rating_max)))
        persons.append(p)

    return (persons, things)


if __name__ == '__main__':
    options = {}
    largv = len(argv)
    for i in range(largv):
        if argv[i] == '--seed' and largv > i+1:
            options['seed'] = argv[i+1].encode('utf-8')
        if argv[i] == '--rounds' and largv > i+1:
            options['rounds'] = int(argv[i+1])
        if argv[i] == '--n_persons' and largv > i+1:
            options['n_persons'] = int(argv[i+1])
        if argv[i] == '--n_things' and largv > i+1:
            options['n_things'] = int(argv[i+1])
        if argv[i] == '--quantity_min' and largv > i+1:
            options['quantity_min'] = int(argv[i+1])
        if argv[i] == '--quantity_max' and largv > i+1:
            options['quantity_max'] = int(argv[i+1])
        if argv[i] == '--rating_min' and largv > i+1:
            options['rating_min'] = int(argv[i+1])
        if argv[i] == '--rating_max' and largv > i+1:
            options['rating_max'] = int(argv[i+1])

    if 'n_persons' in options and 'n_things' in options:
        if options['n_things'] < options['n_persons']:
            print('Error: n_things must be >= n_persons')
            exit(1)
    elif 'n_persons' in options:
        if options['n_persons'] > 100:
            print('Error: n_persons must be <= n_things (default 100)')
            exit(1)
    elif 'n_things' in options:
        if options['n_things'] < 100:
            print('Error: n_things must be >= n_persons (default 100)')
            exit(1)

    if 'quantity_min' in options and 'quantity_max' in options:
        if options['quantity_min'] > options['quantity_max']:
            print('Error: quantity_min must be <= quantity_max')
            exit(1)
    if 'quantity_min' in options and options['quantity_min'] < 1:
        print('Error: quantity_min must be >= 1')
        exit(1)

    persons, things = setup(**options)
    solution = SomeOptimizer.optimize(persons, things, **options)
    ...

catallaxy_optimizer.py

"""
My solution is an algorithmic instance of catallaxy. The form that catallaxy
takes in reality involves subsets and imperfect sorting, and I suspect that
this can be similarly approximated with a bit of elaboration.
"""

class CatallaxyOpitmizer:
    def optimize(
            self, persons: list[PersonProtocol], things: dict[Thing, int],
            **options
        ) -> Solution:
        """Runs the optimization algorithm and returns a Solution assigning
            things to persons. Accepts a list of persons and a dict mapping
            things to quantities.
        """
        allocations = []
        rounds = options.get('rounds', 100)
        random = Randomizer(options.get('seed', b'too cool for the Austrian school'))

        pidx, plen = 0, len(persons)
        for thing, amount in things.items():
            for i in range(amount):
                allocations.append(Allocation(persons[pidx], thing))
                pidx = (pidx + 1 ) % plen
        initial_solution = Solution([Allocation(a.person, a.thing) for a in allocations])

        for _ in range(rounds):
            allocations.sort(key=lambda a: a.person.utility_of(a.thing))
            for i in range(len(allocations)):
                i2 = random.int(plen-1)
                while i2 == i:
                    i2 = random.int(plen-1)
                p1, t1 = allocations[i].person, allocations[i].thing
                p2, t2 = allocations[i2].person, allocations[i2].thing
                current_utility = (p1.utility_of(t1), p2.utility_of(t2))
                swap_utility = (p1.utility_of(t2), p2.utility_of(t1))
                if swap_utility[0] >= current_utility[0] and swap_utility[1] >= current_utility[1]:
                    allocations[i].person = p2
                    allocations[i2].person = p1
        final_solution = Solution(allocations)

        print(f"{initial_solution.score()=}")
        print(f"{final_solution.score()=}")
        return final_solution

catallaxy_optimizer_results_corrected.txt

$ for i in 1 2 4 8 16 32 64 128 256 512 1024; do echo "Catallaxy: $i rounds"; python catallaxy_optimizer.py --rounds $i; echo; done
Catallaxy: 1 rounds
initial_solution.score()=450
final_solution.score()=607

Catallaxy: 2 rounds
initial_solution.score()=450
final_solution.score()=679

Catallaxy: 4 rounds
initial_solution.score()=450
final_solution.score()=745

Catallaxy: 8 rounds
initial_solution.score()=450
final_solution.score()=786

Catallaxy: 16 rounds
initial_solution.score()=450
final_solution.score()=817

Catallaxy: 32 rounds
initial_solution.score()=450
final_solution.score()=850

Catallaxy: 64 rounds
initial_solution.score()=450
final_solution.score()=874

Catallaxy: 128 rounds
initial_solution.score()=450
final_solution.score()=889

Catallaxy: 256 rounds
initial_solution.score()=450
final_solution.score()=897

Catallaxy: 512 rounds
initial_solution.score()=450
final_solution.score()=899

Catallaxy: 1024 rounds
initial_solution.score()=450
final_solution.score()=900

catallaxy_optimizer_results_error.txt

# I initially had an error in the algorithm that maximized total utility instead
# of requiring improved-or-even utility for both participants of the swap. After
# making a few unrelated updates, I restored that error for an additional test.

for i in 1 2 4 8 16 32 64 128 256 512 1024; do echo "Catallaxy: $i rounds"; python catallaxy_optimizer.py --rounds $i; echo; done
Catallaxy: 1 rounds
initial_solution.score()=450
final_solution.score()=644

Catallaxy: 2 rounds
initial_solution.score()=450
final_solution.score()=712

Catallaxy: 4 rounds
initial_solution.score()=450
final_solution.score()=756

Catallaxy: 8 rounds
initial_solution.score()=450
final_solution.score()=794

Catallaxy: 16 rounds
initial_solution.score()=450
final_solution.score()=834

Catallaxy: 32 rounds
initial_solution.score()=450
final_solution.score()=855

Catallaxy: 64 rounds
initial_solution.score()=450
final_solution.score()=875

Catallaxy: 128 rounds
initial_solution.score()=450
final_solution.score()=882

Catallaxy: 256 rounds
initial_solution.score()=450
final_solution.score()=890

Catallaxy: 512 rounds
initial_solution.score()=450
final_solution.score()=890

Catallaxy: 1024 rounds
initial_solution.score()=450
final_solution.score()=890