Generating Pareto front representations using Iterative Pareto Representer¶

Iterative Pareto representer is an algorithm to generate Pareto front representations for multiobjective problems with access to exact solvers. To achieve this, the algorithm smartly selects reference points iteratively to generate a scalarized problem such that the found solution is far from previously found solutions. This algorithm is particularly well-suited for problems with an irregular Pareto front.

In this notebook, we will demonstrate the use of the Iterative Pareto Representer algorithm to generate Pareto front representations for a simple optimization problem with three objectives. The problem is from the forestry domain and more information about the problem can be found in the documentation.

We first import the necessary libraries.

In [1]:

import numpy as np
import plotly.express as ex
import polars as pl
from IPython.display import clear_output

from desdeo.problem.testproblems import forest_problem
from desdeo.tools import GurobipySolver, payoff_table_method
from desdeo.tools.GenerateReferencePoints import generate_points
from desdeo.tools.iterative_pareto_representer import _EvaluatedPoint, choose_reference_point
from desdeo.tools.scalarization import add_asf_diff

import numpy as np import plotly.express as ex import polars as pl from IPython.display import clear_output from desdeo.problem.testproblems import forest_problem from desdeo.tools import GurobipySolver, payoff_table_method from desdeo.tools.GenerateReferencePoints import generate_points from desdeo.tools.iterative_pareto_representer import _EvaluatedPoint, choose_reference_point from desdeo.tools.scalarization import add_asf_diff

We first create a wrapper around the problem to make the rest of the code more readable. The initializer of the ProblemWrapper simply initializes the problem, and evaluates the ideal and nadir points. These points are used to normalize the objectives in the scalarized problem. The solve method solves the scalarized problem and archives the found solutions. It returns all archived solutions. The information from this archive is used by the iterative Pareto representer to select reference points for the next iteration.

In [2]:

class ProblemWrapper:
    def __init__(self, holding: int = 5):
        """Initialize the problem wrapper for a UTOPIA problem.

        Args:
            holding (int, optional): The holding to use for the problem.md. Defaults to 5.
        """
        problem = forest_problem(
            simulation_results="../../tests/data/alternatives_290124.csv",
            treatment_key="../../tests/data/alternatives_key_290124.csv",
            holding=holding,  # Holdings 3, 5, 6 seems to have a decent spread
            comparing=True,
        )

        self.ideal, self.nadir = payoff_table_method(problem=problem)
        clear_output()
        self.problem = problem.update_ideal_and_nadir(new_ideal=self.ideal, new_nadir=self.nadir)
        self.evaluated_points: list[_EvaluatedPoint] = []

    def solve(self, scaled_refp: np.ndarray) -> list[_EvaluatedPoint]:
        refp = {
            obj: val * (self.nadir[obj] - self.ideal[obj]) + self.ideal[obj]
            for obj, val in zip(self.ideal.keys(), scaled_refp)
        }
        scaled_problem, target = add_asf_diff(self.problem, "asf", refp)
        solver = GurobipySolver(scaled_problem, options={"OutputFlag": 0})
        results = solver.solve(target)
        objs = results.optimal_objectives
        scaled_objs = {obj: (objs[obj] - self.ideal[obj]) / (self.nadir[obj] - self.ideal[obj]) for obj in objs.keys()}
        self.evaluated_points.append(
            _EvaluatedPoint(
                reference_point=dict(zip(self.ideal.keys(), scaled_refp)), targets=scaled_objs, objectives=objs
            )
        )
        return self.evaluated_points

class ProblemWrapper: def __init__(self, holding: int = 5): """Initialize the problem wrapper for a UTOPIA problem. Args: holding (int, optional): The holding to use for the problem.md. Defaults to 5. """ problem = forest_problem( simulation_results="../../tests/data/alternatives_290124.csv", treatment_key="../../tests/data/alternatives_key_290124.csv", holding=holding, # Holdings 3, 5, 6 seems to have a decent spread comparing=True, ) self.ideal, self.nadir = payoff_table_method(problem=problem) clear_output() self.problem = problem.update_ideal_and_nadir(new_ideal=self.ideal, new_nadir=self.nadir) self.evaluated_points: list[_EvaluatedPoint] = [] def solve(self, scaled_refp: np.ndarray) -> list[_EvaluatedPoint]: refp = { obj: val * (self.nadir[obj] - self.ideal[obj]) + self.ideal[obj] for obj, val in zip(self.ideal.keys(), scaled_refp) } scaled_problem, target = add_asf_diff(self.problem, "asf", refp) solver = GurobipySolver(scaled_problem, options={"OutputFlag": 0}) results = solver.solve(target) objs = results.optimal_objectives scaled_objs = {obj: (objs[obj] - self.ideal[obj]) / (self.nadir[obj] - self.ideal[obj]) for obj in objs.keys()} self.evaluated_points.append( _EvaluatedPoint( reference_point=dict(zip(self.ideal.keys(), scaled_refp)), targets=scaled_objs, objectives=objs ) ) return self.evaluated_points

The two main functions needed for the Iterative Pareto Representer algorithm are choose_reference_point and generate_points. The choose_reference_point function selects a reference point to create a scalarized problem from a large set of reference points. This larger set of reference points needs to lie on a specific hyperplane in the objective space. The generate_points function generates such reference points.

Here, we generate 10000 reference points and run the iterative Pareto representer algorithm for 100 iterations.

In [3]:

_, refp = generate_points(num_points=10000, num_dims=3)

num_runs = 100
evaluated_points = None
problem = ProblemWrapper()
for i in range(num_runs):
    if (i + 1) % 10 == 0 or i == 0:
        clear_output()
        print(f"Run {i + 1}/{num_runs}")
    reference_point, _ = choose_reference_point(refp, evaluated_points)
    evaluated_points = problem.solve(reference_point)

_, refp = generate_points(num_points=10000, num_dims=3) num_runs = 100 evaluated_points = None problem = ProblemWrapper() for i in range(num_runs): if (i + 1) % 10 == 0 or i == 0: clear_output() print(f"Run {i + 1}/{num_runs}") reference_point, _ = choose_reference_point(refp, evaluated_points) evaluated_points = problem.solve(reference_point)

Run 100/100

Now we extract and visualize the Pareto front.

In [4]:

fig = ex.scatter_3d()

# Add reference points
chosen_refps = pl.DataFrame([point.reference_point for point in evaluated_points])
# rescale reference points
chosen_refps = chosen_refps.with_columns(
    [
        (pl.col(obj) * (problem.nadir[obj] - problem.ideal[obj]) + problem.ideal[obj]).alias(obj)
        for obj in problem.ideal.keys()
    ]
)
fig = fig.add_scatter3d(
    x=chosen_refps["f_1"].to_numpy(),
    y=chosen_refps["f_2"].to_numpy(),
    z=chosen_refps["f_3"].to_numpy(),
    name="Reference Points",
    mode="markers",
    marker_symbol="circle",
    opacity=0.8,
)
# Add front
front = pl.DataFrame([point.objectives for point in evaluated_points])
fig = fig.add_scatter3d(
    x=front["f_1"].to_numpy(),
    y=front["f_2"].to_numpy(),
    z=front["f_3"].to_numpy(),
    mode="markers",
    name="Front",
    marker_symbol="circle",
    opacity=0.9,
)


fig.update_layout(
    scene={
        "xaxis_title": problem.problem.objectives[0].name,
        "yaxis_title": problem.problem.objectives[1].name,
        "zaxis_title": problem.problem.objectives[2].name,
    }
)

fig.update_layout(
    scene=dict(
        xaxis_range=(chosen_refps["f_1"].min(), front["f_1"].max()),
        yaxis_range=(chosen_refps["f_2"].min(), front["f_2"].max()),
        zaxis_range=(chosen_refps["f_3"].min(), front["f_3"].max()),
    )
)
fig.layout.scene.camera.projection.type = "orthographic"
fig.show(renderer="notebook")

fig = ex.scatter_3d() # Add reference points chosen_refps = pl.DataFrame([point.reference_point for point in evaluated_points]) # rescale reference points chosen_refps = chosen_refps.with_columns( [ (pl.col(obj) * (problem.nadir[obj] - problem.ideal[obj]) + problem.ideal[obj]).alias(obj) for obj in problem.ideal.keys() ] ) fig = fig.add_scatter3d( x=chosen_refps["f_1"].to_numpy(), y=chosen_refps["f_2"].to_numpy(), z=chosen_refps["f_3"].to_numpy(), name="Reference Points", mode="markers", marker_symbol="circle", opacity=0.8, ) # Add front front = pl.DataFrame([point.objectives for point in evaluated_points]) fig = fig.add_scatter3d( x=front["f_1"].to_numpy(), y=front["f_2"].to_numpy(), z=front["f_3"].to_numpy(), mode="markers", name="Front", marker_symbol="circle", opacity=0.9, ) fig.update_layout( scene={ "xaxis_title": problem.problem.objectives[0].name, "yaxis_title": problem.problem.objectives[1].name, "zaxis_title": problem.problem.objectives[2].name, } ) fig.update_layout( scene=dict( xaxis_range=(chosen_refps["f_1"].min(), front["f_1"].max()), yaxis_range=(chosen_refps["f_2"].min(), front["f_2"].max()), zaxis_range=(chosen_refps["f_3"].min(), front["f_3"].max()), ) ) fig.layout.scene.camera.projection.type = "orthographic" fig.show(renderer="notebook")