An Optimization Routine for the Number of Paths

Models implemented with the class AbstractPathModelDAG assume a fixed number \(k\) of paths. This class provides an automatic method of iterating in an increasing manner over values of \(k\) to find the best one (i.e., when a stopping criterion has been met).

For example, our custom class for the Minimum Flow Decomposition problem could be emulated like in the code below. However, MinFlowDecomp implements also many other optimizations that make it much faster than this generic routine.

mfd_model = fp.NumPathsOptimization(
        model_type = fp.kFlowDecomp,
        stop_on_first_feasible=True,
        G=graph, 
        flow_attr="flow",
        )

If you want to pass additional parameters to the model, you can just add append them, for example:

mfd_model = fp.NumPathsOptimization(
        model_type = fp.kFlowDecomp,
        stop_on_first_feasible=True,
        G=graph, 
        flow_attr="flow",
        subpath_constraints=[[("a", "c"),("c", "t")]], 
        subpath_constraints_coverage=0.5, 
        optimization_options={"optimize_with_greedy": False}
        )

This routine is also used by specialized wrappers that optimize \(k\) automatically for the Minimum Discordant Nodes objective:

• Minimum-Paths Minimum Discordant Nodes
• Minimum-Paths Minimum Discordant Nodes in General Graphs

NumPathsOptimization

NumPathsOptimization(
    model_type,
    stop_on_first_feasible: bool = None,
    stop_on_delta_abs: float = None,
    stop_on_delta_rel: float = None,
    stop_on_infeasible_on_delta: bool = True,
    min_num_paths: int = 1,
    max_num_paths: int = 2
    ** 64,
    time_limit: float = float(
        "inf"
    ),
    **kwargs
)

Bases: AbstractPathModelDAG, AbstractWalkModelDiGraph

This is a generic class to find the “best” number of path/walks for optimization problems implemented using AbstractPathModelDAG or AbstractWalkModelDiGraph, and parameterized by the number of paths/walks considered. The class iterates over a range of k values, creating and solving a model for each path/walk count until one of the stopping conditions is met.

Parameters

• model_type

  The type of the model used for optimization.

• stop_on_first_feasible : bool, optional

  If True, the optimization process stops as soon as a feasible solution is found. Default is None.

• stop_on_delta_abs : float, optional

  The threshold for change (in absolute value) in objective value between iterations to determine stopping the optimization. Default is None.

• stop_on_delta_rel : float, optional

  The relative threshold for change (in absolute value) in objective value between iterations to determine stopping the optimization. This is computed as the difference in objective value between iterations divided by the objective value of the previous iteration. Default is None.

  Pass at least one stopping criterion

  At least one of the stopping criterion must be set: stop_on_first_feasible, stop_on_delta_abs, stop_on_delta_rel.

• stop_on_infeasible_on_delta : bool, optional

  If True (default), and a delta-based stopping mode is enabled (stop_on_delta_abs or stop_on_delta_rel), stop immediately and report infeasible when an iteration becomes infeasible (including the first tried iteration).

• min_num_paths : int, optional

  Minimum number of paths/walks to be considered in the optimization. Default is 1. The class will also call get_lowerbound_k() on the model_type and the provided arguments to get a better lower bound.

• max_num_paths : int, optional

  Maximum number of paths/walks to be computed. Default is 2**64.

• time_limit : float, optional

  Time limit (in seconds) for the optimization process. Default is float("inf").

• **kwargs

  The keyword arguments to be passed to the model.

  Note

  Do not pass the parameter k here, as it will be handled by the internal optimization process.

Raises

• ValueError

  If none of the stopping criteria (stop_on_first_feasible, stop_on_delta_abs, or stop_on_delta_rel) is provided (i.e., all are None).

Source code in flowpaths/numpathsoptimization.py

def __init__(
    self,
    model_type,
    stop_on_first_feasible: bool = None,
    stop_on_delta_abs: float = None,
    stop_on_delta_rel: float = None,
    stop_on_infeasible_on_delta: bool = True,
    min_num_paths: int = 1,
    max_num_paths: int = 2**64,
    time_limit: float = float("inf"),
    **kwargs,
):
    """
    This is a generic class to find the "best" number of path/walks for optimization
    problems implemented using `AbstractPathModelDAG` or
    `AbstractWalkModelDiGraph`, and parameterized by the number of
    paths/walks considered.
    The class iterates over a range of `k` values, creating and
    solving a model for each path/walk count until one of the stopping conditions is met.

    Parameters
    ----------

    - `model_type`

        The type of the model used for optimization.

    - `stop_on_first_feasible : bool`, optional

        If True, the optimization process stops as soon as a feasible solution is found.
        Default is None.

    - `stop_on_delta_abs` : float, optional

        The threshold for change (in absolute value) in objective value between iterations to determine stopping the optimization.
        Default is `None`.

    - `stop_on_delta_rel` : float, optional

        The relative threshold for change (in absolute value) in objective value between iterations to determine stopping the optimization. 
        This is computed as the difference in objective value between iterations divided by the objective value of the previous iteration.
        Default is `None`.

        !!! warning "Pass at least one stopping criterion"
            At least one of the stopping criterion must be set: `stop_on_first_feasible`, `stop_on_delta_abs`, `stop_on_delta_rel`.

    - `stop_on_infeasible_on_delta : bool`, optional

        If True (default), and a delta-based stopping mode is enabled
        (`stop_on_delta_abs` or `stop_on_delta_rel`), stop immediately and
        report infeasible when an iteration becomes infeasible (including
        the first tried iteration).

    - `min_num_paths : int`, optional

        Minimum number of paths/walks to be considered in the optimization.
        Default is 1. The class will also call `get_lowerbound_k()` on the
        `model_type` and the provided arguments to get a better lower bound.

    - `max_num_paths : int`, optional

        Maximum number of paths/walks to be computed.
        Default is `2**64`.

    - `time_limit : float`, optional

        Time limit (in seconds) for the optimization process.
        Default is `float("inf")`.

    - `**kwargs`

        The keyword arguments to be passed to the model. 

        !!! warning "Note"

            Do not pass the parameter `k` here, as it will be handled by the internal optimization process.

    Raises
    ------

    - `ValueError`

        If none of the stopping criteria (`stop_on_first_feasible`, `stop_on_delta_abs`, or
        `stop_on_delta_rel`) is provided (i.e., all are `None`).
    """

    self.model_type = model_type
    self.stop_on_first_feasible = stop_on_first_feasible
    self.stop_on_delta_abs = stop_on_delta_abs
    self.stop_on_delta_rel = stop_on_delta_rel
    self.stop_on_infeasible_on_delta = stop_on_infeasible_on_delta
    self.min_num_paths = min_num_paths
    self.max_num_paths = max_num_paths
    self.time_limit = time_limit
    self.solve_time_start = None
    self.kwargs = kwargs

    # We allow only one of the stopping criteria to be set
    if (stop_on_first_feasible is None) and (stop_on_delta_abs is None) and (stop_on_delta_rel is None):
        raise ValueError(
            "At least one of the stopping criteria must be set: stop_on_first_feasible, stop_on_delta_abs, stop_on_delta_rel"
        )

    if 'k' in self.kwargs:
        raise ValueError("Do not pass the parameter `k` in the keyword arguments of NumPathsOptimization. This will be iterated over internally to find the best number of paths/walks according to the stopping criteria.")

    self.lowerbound_k = None
    self._solution = None
    self._incumbent_solution = None
    self._incumbent_objective_value = None
    self.solve_statistics = None
    self._is_solved = False

    utils.logger.info(f"{__name__}: created NumPathsOptimization with model_type = {model_type}")

solve_time_elapsed property

solve_time_elapsed

Returns the elapsed time since the start of the solve process.

Returns

• float

  The elapsed time in seconds.

get_objective_value

get_objective_value()

Returns the objective value of the model, if it is solved. Otherwise, raises an exception.

Source code in flowpaths/numpathsoptimization.py

def get_objective_value(self):
    """
    Returns the objective value of the model, if it is solved. Otherwise, raises an exception.
    """

    self.check_is_solved()

    return self.model.get_objective_value()

get_solution

get_solution(
    remove_empty_paths=False,
)

Retrieves the solution from the wrapped model.

Parameters

• remove_empty_paths: bool, optional

  If True, remove empty paths/walks (fewer than two nodes) and filter aligned list fields consistently. Default is False.

Returns

• solution: dict

  The solution obtained from the wrapped model.

Raises

• exception If model is not solved.

Source code in flowpaths/numpathsoptimization.py

def get_solution(self, remove_empty_paths=False):
    """
    Retrieves the solution from the wrapped model.

    Parameters
    ----------
    - `remove_empty_paths: bool`, optional

        If True, remove empty paths/walks (fewer than two nodes)
        and filter aligned list fields consistently. Default is False.

    Returns
    -------

    - `solution: dict`

        The solution obtained from the wrapped model.

    Raises
    -------

    - `exception` If model is not solved.
    """

    self.check_is_solved()
    if self._solution is not None:
        return self._remove_empty_sequences(self._solution) if remove_empty_paths else self._solution

    self._solution = self._get_unfiltered_solution_from_model(self.model)

    return self._remove_empty_sequences(self._solution) if remove_empty_paths else self._solution

is_valid_solution

is_valid_solution() -> bool

Checks if the solution is valid, by calling the is_valid_solution() method of the model.

Source code in flowpaths/numpathsoptimization.py

def is_valid_solution(self) -> bool:
    """
    Checks if the solution is valid, by calling the `is_valid_solution()` method of the model.
    """
    return self.model.is_valid_solution()

solve

solve() -> bool

Attempts to solve the optimization problem by iterating over a range of k values, creating and solving a model for each count until one of the stopping conditions is met. The method iterates from the maximum between the minimum allowed paths and a lower bound (via get_lowerbound_k() of the model) to the maximum allowed paths. For each iteration:

• Creates a model instance with the current number of paths/walks (k).
• Solves the model, and checks if it has been successfully solved.

• Applies various stopping criteria including:

  • stop_on_first_feasible: stops at the first feasible solution.
  • stop_on_delta_abs: stops if the absolute change in the objective value between iterations is less than or equal to the stop_on_delta_abs value.
  • stop_on_delta_rel: stops if the relative change in the objective value between iterations is less than or equal to the stop_on_delta_rel value.

• Stops if the elapsed time exceeds the designated time limit.

Upon termination, the method sets the overall solve status:

• If no feasible solution was found, the status is marked as infeasible.
• If the process did not exceed the time limit but no other stopping condition was met, the status is marked as unbounded.
• If any of the stopping criteria (feasible or delta conditions) were satisfied, the status is set as solved.

If a valid solution is found, it stores the solution, updates solve statistics and the model, marks the problem as solved, and returns True. Otherwise, it returns False.

Returns

• bool

  True if an optimal solution is found and the problem is marked as solved, False otherwise.

Source code in flowpaths/numpathsoptimization.py

def solve(self) -> bool:
    """
    Attempts to solve the optimization problem by iterating over a range of `k` values, creating and
    solving a model for each count until one of the stopping conditions is met.
    The method iterates from the maximum between the minimum allowed paths and a lower bound (via
    `get_lowerbound_k()` of the model) to the maximum allowed paths. For each iteration:

    - Creates a model instance with the current number of paths/walks (`k`).
    - Solves the model, and checks if it has been successfully solved.
    - Applies various stopping criteria including:

        - `stop_on_first_feasible`: stops at the first feasible solution.
        - `stop_on_delta_abs`: stops if the absolute change in the objective value between iterations is
            less than or equal to the `stop_on_delta_abs` value.
        - `stop_on_delta_rel`: stops if the relative change in the objective value between iterations is
            less than or equal to the `stop_on_delta_rel` value.

    - Stops if the elapsed time exceeds the designated time limit.

    Upon termination, the method sets the overall solve status:

    - If no feasible solution was found, the status is marked as infeasible.
    - If the process did not exceed the time limit but no other stopping condition was met, the status
        is marked as unbounded.
    - If any of the stopping criteria (feasible or delta conditions) were satisfied, the status is set as solved.

    If a valid solution is found, it stores the solution, updates solve statistics and the model,
    marks the problem as solved, and returns `True`. Otherwise, it returns `False`.

    Returns
    -------

    - `bool`

        `True` if an optimal solution is found and the problem is marked as solved, `False` otherwise.

    """

    self.solve_time_start = time.perf_counter()
    previous_solution_objective_value = None
    solve_status = None
    found_feasible = False
    previous_feasible_model = None
    selected_model = None
    self._incumbent_solution = None
    self._incumbent_objective_value = None

    for k in range(max(self.min_num_paths,self.get_lowerbound_k()), self.max_num_paths+1):
        remaining_time = self.time_limit - self.solve_time_elapsed
        if remaining_time <= 0:
            solve_status = NumPathsOptimization.timeout_status_name
            utils.logger.info(
                f"{__name__}: model id = {id(self)}, no remaining time before iteration k = {k}, stopping with timeout"
            )
            break

        model_kwargs = copy.deepcopy(self.kwargs)
        # Enforce a global wall-clock budget by capping each per-k model with
        # the remaining time (without overriding a stricter caller-provided cap).
        if self.time_limit != float("inf"):
            if "solver_options" in model_kwargs and isinstance(model_kwargs["solver_options"], dict):
                solver_time_limit = model_kwargs["solver_options"].get("time_limit", float("inf"))
                model_kwargs["solver_options"]["time_limit"] = min(solver_time_limit, remaining_time)
            if "time_limit" in model_kwargs:
                model_time_limit = model_kwargs.get("time_limit", float("inf"))
                model_kwargs["time_limit"] = min(model_time_limit, remaining_time)

        # Create the model
        utils.logger.info(f"{__name__}: model id = {id(self)}, iteration with k = {k}")
        model = self.model_type(**model_kwargs, k=k)
        model.solve()

        model_status = None
        if hasattr(model, "solver") and hasattr(model.solver, "get_model_status"):
            model_status = model.solver.get_model_status()
        elif hasattr(model, "solve_statistics") and isinstance(model.solve_statistics, dict):
            model_status = model.solve_statistics.get("model_status")

        if model.is_solved():
            found_feasible = True
            current_solution_objective_value = model.get_objective_value()
            utils.logger.info(f"{__name__}: model id = {id(self)}, iteration with k = {k}, current_solution_objective_value = {current_solution_objective_value}")
            if self.stop_on_first_feasible:
                solve_status = NumPathsOptimization.solved_status_name
                selected_model = model
                break
            if previous_solution_objective_value is not None:
                if self.stop_on_delta_abs is not None:
                    if abs(previous_solution_objective_value - current_solution_objective_value) <= self.stop_on_delta_abs:
                        solve_status = NumPathsOptimization.solved_status_name
                        selected_model = previous_feasible_model
                        break
                if self.stop_on_delta_rel is not None:
                    abs_delta = abs(previous_solution_objective_value - current_solution_objective_value)
                    if previous_solution_objective_value == 0:
                        if abs_delta == 0:
                            solve_status = NumPathsOptimization.solved_status_name
                            selected_model = previous_feasible_model
                            break
                    elif abs_delta / abs(previous_solution_objective_value) <= self.stop_on_delta_rel:
                        solve_status = NumPathsOptimization.solved_status_name
                        selected_model = previous_feasible_model
                        break

            previous_feasible_model = model
            previous_solution_objective_value = current_solution_objective_value

            # Preserve paper-like timeout semantics: when current k hits the
            # time limit but has an incumbent, return best-so-far.
            if model_status == "kTimeLimit":
                solve_status = NumPathsOptimization.timeout_status_name
                selected_model = model
                utils.logger.info(
                    f"{__name__}: model id = {id(self)}, iteration with k = {k}, solver reported time limit with feasible incumbent"
                )
                break
        else:
            utils.logger.info(f"{__name__}: model id = {id(self)}, iteration with k = {k}, model is not solved")
            if model_status == "kTimeLimit":
                solve_status = NumPathsOptimization.timeout_status_name
                if hasattr(model, "has_incumbent_solution") and model.has_incumbent_solution():
                    if hasattr(model, "get_incumbent_solution_paths"):
                        self._incumbent_solution = {
                            "paths": model.get_incumbent_solution_paths(),
                        }
                    elif hasattr(model, "get_incumbent_solution_walks"):
                        self._incumbent_solution = {
                            "walks": model.get_incumbent_solution_walks(),
                        }
                    if hasattr(model, "get_incumbent_objective_value"):
                        self._incumbent_objective_value = model.get_incumbent_objective_value()
                utils.logger.info(
                    f"{__name__}: model id = {id(self)}, iteration with k = {k}, solver reported time limit"
                )
                break
            if (
                self.stop_on_infeasible_on_delta
                and (self.stop_on_delta_abs is not None or self.stop_on_delta_rel is not None)
                and model_status == sw.SolverWrapper.infeasible_status
            ):
                solve_status = NumPathsOptimization.infeasible_status_name
                utils.logger.info(
                    f"{__name__}: model id = {id(self)}, iteration with k = {k}, stopping on infeasible model while running delta-based stopping"
                )
                break
        if self.solve_time_elapsed >= self.time_limit:
            solve_status = NumPathsOptimization.timeout_status_name
            utils.logger.info(f"{__name__}: model id = {id(self)}, iteration with k = {k}, time out")
            break

    if solve_status != NumPathsOptimization.timeout_status_name:
        if not found_feasible:
            solve_status = NumPathsOptimization.infeasible_status_name
        elif solve_status is None:
            solve_status = NumPathsOptimization.unbounded_status_name

    self.solve_statistics = {
            "solve_status": solve_status,
            "solve_time": self.solve_time_elapsed,
        }

    if solve_status == NumPathsOptimization.solved_status_name:
        if selected_model is None:
            selected_model = previous_feasible_model if previous_feasible_model is not None else model
        # Store the unfiltered solution when the wrapped model supports it,
        # so callers can decide whether to remove empty paths/walks.
        self._solution = self._get_unfiltered_solution_from_model(selected_model)
        self.set_solved()
        self.solve_statistics.update(selected_model.solve_statistics)
        self.model = selected_model
        return True

    return False