schrodinger.application.transforms.pareto module

Transform for ranking elements based on a Pareto minimization.

class schrodinger.application.transforms.pareto.ParetoPoint(element: Any, front: int, intra_front_score: float = 0.0)

Bases: object

Represents an element in a Pareto front.

Parameters:

  • element – The original element that was ranked.

  • front – The 1-based Pareto front that contained the element.

  • intra_front_score – Score for ordering elements within the same Pareto front. Higher is better. Defaults to 0.0 when no intra_front_scorer is used.

element: Any
front: int
intra_front_score: float = 0.0
__init__(element: Any, front: int, intra_front_score: float = 0.0) → None
class schrodinger.application.transforms.pareto.FrontScorer(*args, **kwargs)

Bases: Protocol

Scores elements for ordering within each Pareto front.

__init__(*args, **kwargs)
schrodinger.application.transforms.pareto.pairwise_intra_front_score(vectors: Sequence[Tuple[float, ...]], fronts: Sequence[int]) → List[float]

Score elements within each Pareto front using pairwise comparison.

For each element A in a front, compares A against every other element B in the same front across all objectives. Each objective is min-max normalized using global min/max across ALL elements (not just the front). The score is sum over B != A of sum over objectives of (B_normalized - A_normalized). Higher score indicates a more desirable element within its front.

Parameters:

  • vectors – Objective-value tuples for all ranked elements. Each tuple holds one float per objective, oriented so that lower is better.

  • fronts – Front number (1-based) for each element, aligned with vectors.

Returns:

Scores aligned with vectors. Higher is better within a front. Single-element fronts receive a score of 0.0.

class schrodinger.application.transforms.pareto.ParetoRank(vectorize: Callable[[T], Tuple[float, ...]], first_front_only=False, intra_front_scorer: FrontScorer | None = None)

Bases: PTransform

Transform for ranking elements based on Pareto optimization, specifically minimizing the objective values (vector).

Each element is translated into a tuple of objective values (using the vectorize function). Then, every element is assigned a Pareto front value based on their dominance relative to other vectors.

Pareto dominance definition: A dominates B if A is no worse in all objectives and better in at least one (i.e., A is strictly “better” than B).

Pareto front: The set of non-dominated elements (rank 1). For every element on the Pareto front, there are no other elements that dominate it.

For more information, see this wiki page

Example usage::
>>> DURABLITY_VALUE_MAP = {
...     # since ParetoRank minimizes, 'Good' should be the lowest value
...     'Good': 0,
...     'Average': 1,
...     'Bad': 2
... }
...
>>> @dataclass(frozen=True)
>>> class Element:
...     model: str
...     cost: float
...     durability: str
...
>>> def vectorize(element) -> Tuple[float, ...]:
...     return element.cost, DURABLITY_VALUE_MAP[element.durability]
...
>>> elements = [
...     Element(model='Door1', cost=75.99, durability='Good'),
...     Element(model='Door2', cost=59.99, durability='Average'),
...     Element(model='Door3', cost=59.99, durability='Good'),
...     Element(model='Door4', cost=49.99, durability='Bad'),
... ]
...
>>> with beam.Pipeline() as p:
...     pareto_points = (p
...                      | beam.Create(elements)
...                      | ParetoRank(vectorize)
...                      | beam.LogElements())
ParetoPoint(element=Element(model='Door1', cost=75.99, durability='Good'), front=2)
ParetoPoint(element=Element(model='Door2', cost=59.99, durability='Average'), front=2)
ParetoPoint(element=Element(model='Door3', cost=59.99, durability='Good'), front=1)
ParetoPoint(element=Element(model='Door4', cost=49.99, durability='Bad'), front=1)
Parameters:

  • vectorize – A function that takes an element and returns the values, a tuple of floats, to be minimized.

  • first_front_only – If True, only the ParetoPoints in the first Pareto front will be returned

  • intra_front_scorer – Optional FrontScorer that computes a secondary score for ordering elements within the same front. When None (default), intra_front_score on every ParetoPoint is 0.0.

__init__(vectorize: Callable[[T], Tuple[float, ...]], first_front_only=False, intra_front_scorer: FrontScorer | None = None)
expand(pcoll)