NSGA-II Optimization Reference

from pymoo.algorithms.moo.nsga2 import NSGA2

NSGA2(
    pop_size=100,
    sampling=FloatRandomSampling(),
    selection=TournamentSelection(func_comp=binary_tournament),
    crossover=SBX(prob=0.9, eta=15),
    mutation=PM(prob=None, eta=20),
    survival=RankAndCrowding(),
    output=MultiObjectiveOutput(),
    eliminate_duplicates=True,
    n_offsprings=None,
    repair=None,
    mating=None,
    # --- Inherited from Algorithm base ---
    termination=None,
    seed=None,
    verbose=False,
    callback=None,
    save_history=False,
    return_least_infeasible=False,
    display=None,
    evaluator=None,
    archive=None,
    **kwargs
)

algorithm = NSGA2(pop_size=200)

import numpy as np
from pymoo.operators.sampling.rnd import FloatRandomSampling, BinaryRandomSampling
from pymoo.operators.sampling.lhs import LHS

# Option 1: Random float sampling (default)
NSGA2(sampling=FloatRandomSampling())

# Option 2: Latin Hypercube Sampling for better initial coverage
NSGA2(sampling=LHS())

# Option 3: Binary sampling (for binary variable problems)
NSGA2(sampling=BinaryRandomSampling())

# Option 4: Warm-start with a custom numpy array
X_init = np.random.rand(100, 10)  # 100 individuals, 10 variables
NSGA2(sampling=X_init)

from pymoo.operators.selection.tournament import TournamentSelection
from pymoo.operators.selection.rnd import RandomSelection
from pymoo.algorithms.moo.nsga2 import binary_tournament

# Default: binary tournament
NSGA2(selection=TournamentSelection(func_comp=binary_tournament))

# Alternative: purely random selection (removes selection pressure)
NSGA2(selection=RandomSelection())

from pymoo.operators.crossover.sbx import SBX
from pymoo.operators.crossover.pntx import TwoPointCrossover, SinglePointCrossover
from pymoo.operators.crossover.ux import UniformCrossover

# Default: SBX for continuous variables
NSGA2(crossover=SBX(prob=0.9, eta=15))

# Two-point crossover (for binary problems)
NSGA2(crossover=TwoPointCrossover())

# Uniform crossover
NSGA2(crossover=UniformCrossover(prob=0.5))

from pymoo.operators.mutation.pm import PM
from pymoo.operators.mutation.bitflip import BitflipMutation

# Default: polynomial mutation for real variables
NSGA2(mutation=PM(prob=None, eta=20))

# Bit-flip mutation for binary variables
NSGA2(mutation=BitflipMutation())

from pymoo.operators.survival.rank_and_crowding import RankAndCrowding

NSGA2(survival=RankAndCrowding())

from pymoo.util.display.multi import MultiObjectiveOutput

NSGA2(output=MultiObjectiveOutput())

# Default: generate pop_size offspring each generation (full generational)
NSGA2(pop_size=100, n_offsprings=None)

# Steady-state: generate 10 offspring per generation
NSGA2(pop_size=100, n_offsprings=10)

from pymoo.core.duplicate import ElementwiseDuplicateElimination
import numpy as np

# Default: eliminate duplicates
NSGA2(eliminate_duplicates=True)

# Disable
NSGA2(eliminate_duplicates=False)

# Custom: consider two solutions equal if within 1e-4
class MyDuplicateElimination(ElementwiseDuplicateElimination):
    def is_equal(self, a, b):
        return np.allclose(a.X, b.X, atol=1e-4)

NSGA2(eliminate_duplicates=MyDuplicateElimination())

from pymoo.core.repair import Repair
import numpy as np

class NormalizeRepair(Repair):
    """Repair operator that ensures all variables sum to 1.0."""
    def _do(self, problem, X, **kwargs):
        # X shape: (n_individuals, n_var)
        row_sums = X.sum(axis=1, keepdims=True)
        row_sums[row_sums == 0] = 1.0
        return X / row_sums

algorithm = NSGA2(
    pop_size=100,
    repair=NormalizeRepair()
)

from pymoo.core.mating import Mating
from pymoo.operators.crossover.sbx import SBX
from pymoo.operators.mutation.pm import PM
from pymoo.operators.selection.tournament import TournamentSelection
from pymoo.algorithms.moo.nsga2 import binary_tournament

custom_mating = Mating(
    selection=TournamentSelection(func_comp=binary_tournament),
    crossover=SBX(prob=0.9, eta=15),
    mutation=PM(prob=None, eta=20),
    repair=None,
    eliminate_duplicates=True,
    n_max_iterations=100   # max tries to fill offspring pool without duplicates
)

algorithm = NSGA2(pop_size=100, mating=custom_mating)

from pymoo.operators.crossover.sbx import SBX

# High exploration (eta=5)
crossover = SBX(prob=0.9, eta=5)

# Conservative exploitation (eta=30), lower crossover rate
crossover = SBX(prob=0.7, eta=30)

# Cross every variable each time (prob_var=1.0)
crossover = SBX(prob=0.9, eta=15, prob_var=1.0)

from pymoo.operators.mutation.pm import PM

# Default: auto probability, eta=20
mutation = PM(prob=None, eta=20)

# Explicit mutation rate with higher spread
mutation = PM(prob=0.1, eta=10)

# Conservative mutation (very small perturbations)
mutation = PM(prob=0.05, eta=30)

from pymoo.operators.selection.tournament import TournamentSelection
from pymoo.algorithms.moo.nsga2 import binary_tournament

# Default
selection = TournamentSelection(func_comp=binary_tournament, pressure=2)

# Stronger selection pressure (3-way tournament)
selection = TournamentSelection(func_comp=binary_tournament, pressure=3)

from pymoo.operators.survival.rank_and_crowding import RankAndCrowding

# Default NSGA-II crowding distance
NSGA2(survival=RankAndCrowding(crowding_func='cd'))

# Pruning crowding distance (generally better in practice)
NSGA2(survival=RankAndCrowding(crowding_func='pcd'))

# Entropy-based (useful for 3-objective problems)
NSGA2(survival=RankAndCrowding(crowding_func='ce'))

from pymoo.core.callback import Callback

class ConvergenceLogger(Callback):
    def __init__(self):
        super().__init__()
        self.data["n_nds"] = []
        self.data["n_evals"] = []

    def notify(self, algorithm):
        self.data["n_nds"].append(len(algorithm.opt))
        self.data["n_evals"].append(algorithm.evaluator.n_eval)

algorithm = NSGA2(
    pop_size=100,
    seed=42,
    verbose=True,
    save_history=True,
    callback=ConvergenceLogger()
)

from pymoo.optimize import minimize

res = minimize(
    problem,
    algorithm,
    termination=None,
    seed=None,
    verbose=False,
    display=None,
    callback=None,
    return_least_infeasible=False,
    save_history=False,
    copy_algorithm=True,
    copy_termination=True,
)

from pymoo.termination.default import DefaultMultiObjectiveTermination

termination = DefaultMultiObjectiveTermination(
    xtol=0.0005,
    cvtol=1e-8,
    ftol=0.005,
    n_skip=5,
    period=50,
    n_max_gen=1000,
    n_max_evals=100000,
)

from pymoo.termination.default import DefaultMultiObjectiveTermination
from pymoo.optimize import minimize

# Stricter convergence, larger window, higher fallback
termination = DefaultMultiObjectiveTermination(
    xtol=1e-8,
    cvtol=1e-8,
    ftol=1e-6,
    period=100,
    n_skip=5,
    n_max_gen=5000,
    n_max_evals=500000,
)

res = minimize(problem, algorithm, termination, seed=1, verbose=True)

# Effectively disable ftol (rely only on xtol and n_max_gen):
termination = DefaultMultiObjectiveTermination(
    ftol=1.0,       # Very large value, never triggered
    n_max_gen=500,  # Rely on generation cap instead
)

res = minimize(problem, algorithm, ('n_gen', 200), seed=1, save_history=True)

# Pareto front results
print(res.X)          # shape (n_nds, n_var)
print(res.F)          # shape (n_nds, n_obj)
print(res.CV)         # shape (n_nds, 1)

# Full population attributes
X_all   = res.pop.get("X")        # all individuals' variables
F_all   = res.pop.get("F")        # all individuals' objectives
ranks   = res.pop.get("rank")     # Pareto rank (0 = first front)
cd      = res.pop.get("crowding") # crowding distance

# Algorithm state
print(f"Generations run:  {res.algorithm.n_gen}")
print(f"Evaluations used: {res.algorithm.evaluator.n_eval}")
print(f"Wall time:        {res.exec_time:.2f}s")
print(f"Seed used:        {res.seed}")

# Convergence history (requires save_history=True)
for gen_snap in res.history:
    n_nds  = len(gen_snap.opt)
    n_eval = gen_snap.evaluator.n_eval
    print(f"Gen {gen_snap.n_gen}: {n_nds} non-dominated, {n_eval} evals")

from pymoo.operators.sampling.lhs import LHS
NSGA2(sampling=LHS())    # Better initial coverage than pure random

from pymoo.indicators.igd import IGD
from pymoo.indicators.igd_plus import IGDPlus
from pymoo.indicators.gd import GD
from pymoo.indicators.hv import HV
import numpy as np

res = minimize(problem, algorithm, ('n_gen', 200), seed=1, save_history=True)

# --- Pareto-front based indicators (need the true Pareto front) ---
pf = problem.pareto_front()       # Only available for test problems

igd      = IGD(pf)
igd_plus = IGDPlus(pf)
gd       = GD(pf)

print("IGD:  ", igd.do(res.F))        # lower is better
print("IGD+: ", igd_plus.do(res.F))   # lower is better
print("GD:   ", gd.do(res.F))         # lower is better

# --- Hypervolume (needs a reference point, not the Pareto front) ---
# ref_point must be strictly dominated by all Pareto front points
ref_point = np.array([1.1, 1.1])
hv = HV(ref_point=ref_point)
print("HV:   ", hv.do(res.F))         # higher is better

# --- Automatic reference point from found solutions ---
max_F = res.F.max(axis=0)
ref_point_auto = max_F * 1.1
hv_auto = HV(ref_point=ref_point_auto)
print("HV (auto ref):", hv_auto.do(res.F))

# --- Convergence curve over generations ---
n_evals = []
hv_over_time = []

for gen_snap in res.history:
    opt_F = gen_snap.opt.get("F")
    n_evals.append(gen_snap.evaluator.n_eval)
    hv_over_time.append(hv.do(opt_F))

import matplotlib.pyplot as plt
plt.plot(n_evals, hv_over_time, '-o', markersize=3)
plt.xlabel("Function Evaluations")
plt.ylabel("Hypervolume")
plt.title("NSGA-II Convergence")
plt.show()

import numpy as np
from pymoo.core.problem import Problem

class VectorizedProblem(Problem):
    def __init__(self):
        super().__init__(n_var=10, n_obj=2, xl=0.0, xu=1.0)

    def _evaluate(self, X, out, *args, **kwargs):
        # X shape: (pop_size, n_var) - entire population at once
        f1 = np.sum(X ** 2, axis=1)
        f2 = np.sum((X - 1) ** 2, axis=1)
        out["F"] = np.column_stack([f1, f2])

from multiprocessing.pool import Pool
from pymoo.core.problem import ElementwiseProblem
from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.optimize import minimize
from pymoo.problems.parallelization import StarMapParallelization

class MyExpensiveProblem(ElementwiseProblem):
    def __init__(self, **kwargs):
        super().__init__(n_var=5, n_obj=2, xl=0.0, xu=1.0, **kwargs)

    def _evaluate(self, x, out, *args, **kwargs):
        import time
        time.sleep(0.01)  # Simulate expensive evaluation
        out["F"] = [float(sum(x**2)), float(sum((x-1)**2))]

# Run with 4 parallel workers
n_proccess = 4
pool = Pool(n_proccess)
runner = StarMapParallelization(pool.starmap)

problem = MyExpensiveProblem(elementwise_runner=runner)
algorithm = NSGA2(pop_size=100)
res = minimize(problem, algorithm, ('n_gen', 50), seed=1)
pool.close()

from joblib import Parallel, delayed
import numpy as np
from pymoo.core.problem import Problem
from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.optimize import minimize

class JoblibProblem(Problem):
    def __init__(self, n_jobs=-1):
        super().__init__(n_var=10, n_obj=2, xl=0.0, xu=1.0)
        self.n_jobs = n_jobs

    def _evaluate(self, X, out, *args, **kwargs):
        def eval_single(x):
            return [float(np.sum(x ** 2)), float(np.sum((x - 1) ** 2))]

        F = Parallel(n_jobs=self.n_jobs)(
            delayed(eval_single)(X[i]) for i in range(len(X))
        )
        out["F"] = np.array(F)

problem = JoblibProblem(n_jobs=8)
algorithm = NSGA2(pop_size=100)
res = minimize(problem, algorithm, ('n_gen', 200), seed=1)

from pymoo.core.evaluator import Evaluator
from pymoo.algorithms.moo.nsga2 import NSGA2

class AsyncEvaluator(Evaluator):
    def __init__(self, submit_fn, gather_fn):
        super().__init__()
        self.submit_fn = submit_fn
        self.gather_fn = gather_fn

    def _eval(self, problem, pop, **kwargs):
        # Submit all evaluations asynchronously
        futures = [self.submit_fn(problem._evaluate_single, ind.X) for ind in pop]
        results = self.gather_fn(futures)
        for ind, res in zip(pop, results):
            ind.F = res
        return pop

# Usage with Dask (example)
# from dask.distributed import Client
# client = Client()
# evaluator = AsyncEvaluator(client.submit, client.gather)
# algorithm = NSGA2(pop_size=100, evaluator=evaluator)

from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.problems import get_problem
from pymoo.optimize import minimize
from pymoo.visualization.scatter import Scatter

problem = get_problem("zdt1")
algorithm = NSGA2(pop_size=100)

res = minimize(problem, algorithm, ('n_gen', 200), seed=1, verbose=True)

Scatter().add(problem.pareto_front(), plot_type="line", color="black") \
         .add(res.F, facecolor="none", edgecolor="red").show()

from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.operators.crossover.sbx import SBX
from pymoo.operators.mutation.pm import PM
from pymoo.operators.sampling.lhs import LHS
from pymoo.operators.survival.rank_and_crowding import RankAndCrowding
from pymoo.optimize import minimize

algorithm = NSGA2(
    pop_size=150,
    n_offsprings=30,
    sampling=LHS(),
    crossover=SBX(prob=0.9, eta=20),
    mutation=PM(prob=None, eta=15),
    survival=RankAndCrowding(crowding_func='pcd'),
    eliminate_duplicates=True,
)

res = minimize(problem, algorithm, ('n_gen', 300), seed=42, verbose=True)

from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.operators.crossover.pntx import TwoPointCrossover
from pymoo.operators.mutation.bitflip import BitflipMutation
from pymoo.operators.sampling.rnd import BinaryRandomSampling
from pymoo.problems import get_problem
from pymoo.optimize import minimize

problem = get_problem("zdt5")

algorithm = NSGA2(
    pop_size=100,
    sampling=BinaryRandomSampling(),
    crossover=TwoPointCrossover(),
    mutation=BitflipMutation(),
    eliminate_duplicates=True
)

res = minimize(problem, algorithm, ('n_gen', 500), seed=1, verbose=False)

import numpy as np
from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.core.problem import ElementwiseProblem
from pymoo.core.repair import Repair
from pymoo.optimize import minimize

class PortfolioProblem(ElementwiseProblem):
    """Two-objective portfolio: maximize return, minimize risk. Weights must sum to 1."""
    def __init__(self):
        super().__init__(n_var=5, n_obj=2, xl=0.0, xu=1.0)

    def _evaluate(self, x, out, *args, **kwargs):
        out["F"] = [-np.dot(x, [0.10, 0.20, 0.15, 0.12, 0.18]),
                     np.dot(x, [0.05, 0.08, 0.07, 0.06, 0.09])]

class SumToOneRepair(Repair):
    def _do(self, problem, X, **kwargs):
        row_sums = X.sum(axis=1, keepdims=True)
        row_sums[row_sums == 0] = 1.0
        return X / row_sums

algorithm = NSGA2(
    pop_size=100,
    repair=SumToOneRepair(),
    eliminate_duplicates=True,
    return_least_infeasible=True,
)

res = minimize(PortfolioProblem(), algorithm, ('n_gen', 200), seed=1, verbose=True)
print("Optimal weights:\n", res.X)
print("Return / Risk tradeoffs:\n", res.F)

from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.core.callback import Callback
from pymoo.indicators.hv import HV
from pymoo.indicators.igd_plus import IGDPlus
from pymoo.problems import get_problem
from pymoo.optimize import minimize
import numpy as np
import matplotlib.pyplot as plt

class HVCallback(Callback):
    """Track hypervolume every generation."""
    def __init__(self, ref_point):
        super().__init__()
        self.hv_indicator = HV(ref_point=ref_point)
        self.data["hv"] = []
        self.data["n_eval"] = []

    def notify(self, algorithm):
        F = algorithm.opt.get("F")
        if F is not None and len(F) > 0:
            self.data["hv"].append(self.hv_indicator.do(F))
            self.data["n_eval"].append(algorithm.evaluator.n_eval)

problem = get_problem("zdt1")
ref_point = np.array([1.1, 1.1])
callback = HVCallback(ref_point=ref_point)

algorithm = NSGA2(pop_size=100, eliminate_duplicates=True, callback=callback)

res = minimize(problem, algorithm, ('n_gen', 200), seed=1, verbose=True, save_history=True)

# Plot convergence
plt.plot(callback.data["n_eval"], callback.data["hv"], '-o', markersize=3)
plt.xlabel("Function Evaluations")
plt.ylabel("Hypervolume")
plt.title("NSGA-II Convergence on ZDT1")
plt.show()

# Final performance
igd_plus = IGDPlus(problem.pareto_front())
hv       = HV(ref_point=ref_point)
print(f"IGD+:        {igd_plus.do(res.F):.6f}")
print(f"Hypervolume: {hv.do(res.F):.6f}")
print(f"Exec time:   {res.exec_time:.2f}s")
print(f"Total evals: {res.algorithm.evaluator.n_eval}")

from pymoo.algorithms.moo.nsga2 import NSGA2
from pymoo.termination.default import DefaultMultiObjectiveTermination
from pymoo.problems import get_problem
from pymoo.optimize import minimize

problem = get_problem("zdt3")

termination = DefaultMultiObjectiveTermination(
    xtol=1e-8,
    cvtol=1e-8,
    ftol=0.001,
    period=30,
    n_skip=5,
    n_max_gen=2000,
    n_max_evals=200000,
)

algorithm = NSGA2(pop_size=100, eliminate_duplicates=True)

res = minimize(problem, algorithm, termination, seed=1, verbose=True)
print(f"Stopped at generation: {res.algorithm.n_gen}")
print(f"Non-dominated solutions found: {len(res.F)}")