refactor(portfolio): extract common MOEA/D logic to base class and fix linting

Author: MichalRedm

src/portfolio/moead.py

@@ -36,7 +36,6 @@ def generate_pareto_front(
         """
         MOEA/D with decomposition, stable normalisation, and optional adaptive bounds.
         """
-        num_assets = len(kwargs["expected_returns"])
         num_objectives = len(self.problem.objectives)
 
         # ── 1. Weight vectors & neighbourhood ────────────────────────────────
@@ -47,44 +46,10 @@ def generate_pareto_front(
         # ── 2. Initialize Normalisation ──────────────────────────────────────
         self._init_normalization(kwargs)
 
-        def repair(w):
-            if repair_method == "euclidean":
-                return self._repair(w)
-            elif repair_method == "simple":
-                return self._repair_simple(w)
-            else:
-                return w
-
-        # ── 3. Build candidate pool and assign best to each weight vector ─────
-        n_extra = max(2 * num_points, 3 * num_assets)
-        candidates = []
-        for j in range(num_assets):
-            w = np.zeros(num_assets)
-            w[j] = 1.0
-            candidates.append(w)
-        candidates.append(np.ones(num_assets) / num_assets)
-        candidates.extend(np.random.dirichlet(np.ones(num_assets), n_extra))
-        cand_arr = np.array(candidates)
-
-        # Evaluate and normalise candidates
-        cand_f = np.array(
-            [list(self.problem.evaluate_all(w, **kwargs).values()) for w in cand_arr]
+        # ── 3. Initial Population ─────────────────────────────────────────────
+        population, f_phys, f_norm = self._initial_sampling(
+            weight_vectors, reference_type, **kwargs
         )
-        cand_fn = np.array([self._normalise(f) for f in cand_f])
-
-        # Initial assignment
-        population = np.empty((num_points, num_assets))
-        f_phys = np.empty((num_points, num_objectives))
-        f_norm = np.empty((num_points, num_objectives))
-
-        for k, wv in enumerate(weight_vectors):
-            g_vals = np.array(
-                [self._get_scalar_value(fn, wv, reference_type) for fn in cand_fn]
-            )
-            best = int(np.argmin(g_vals))
-            population[k] = cand_arr[best]
-            f_phys[k] = cand_f[best]
-            f_norm[k] = cand_fn[best]
 
         # ── 4. Evolution ──────────────────────────────────────────────────────
         history = []
@@ -97,57 +62,35 @@ def repair(w):
                 nb_idx = neighbors[i]
                 p1_idx, p2_idx = np.random.choice(nb_idx, 2, replace=False)
 
-                # Crossover
-                if crossover_operator == "sbx":
-                    offspring = self._sbx_crossover(
-                        population[p1_idx], population[p2_idx]
-                    )
-                    offspring = repair(offspring)
-                elif crossover_operator == "simplex":
-                    offspring = self._simplex_crossover(
-                        population[p1_idx], population[p2_idx]
-                    )
-                elif crossover_operator == "ldc":
-                    offspring = self._ldc_crossover(
-                        population[p1_idx], population[p2_idx]
-                    )
-                elif crossover_operator == "none":
-                    offspring = population[p1_idx].copy()
-                else:
-                    raise ValueError(f"Unknown crossover: {crossover_operator}")
-
-                # Mutation
-                if mutation_operator == "polynomial":
-                    offspring = self._polynomial_mutation(offspring)
-                    offspring = repair(offspring)
-                elif mutation_operator == "simplex":
-                    offspring = self._simplex_mutation(offspring, eta=mutation_eta)
-                elif mutation_operator == "none":
-                    pass
-                else:
-                    raise ValueError(f"Unknown mutation: {mutation_operator}")
+                # Generate offspring using shared operators
+                offspring = self._apply_genetic_operators(
+                    population[p1_idx],
+                    population[p2_idx],
+                    crossover_operator,
+                    mutation_operator,
+                    mutation_eta,
+                    repair_method,
+                )
 
                 # Evaluate offspring
                 off_f = np.array(
                     list(self.problem.evaluate_all(offspring, **kwargs).values())
                 )
                 off_fn = self._normalise(off_f)
 
-                # Scalar scores and neighborhood update
-                count = 0
-                for j in nb_idx:
-                    if count >= nr:
-                        break
-
-                    wv_j = weight_vectors[j]
-                    curr_g = self._get_scalar_value(f_norm[j], wv_j, reference_type)
-                    off_g = self._get_scalar_value(off_fn, wv_j, reference_type)
-
-                    if off_g < curr_g:
-                        population[j] = offspring
-                        f_phys[j] = off_f
-                        f_norm[j] = off_fn
-                        count += 1
+                # Update neighborhood using shared logic
+                self._update_neighborhood(
+                    offspring,
+                    off_f,
+                    off_fn,
+                    nb_idx,
+                    nr,
+                    weight_vectors,
+                    population,
+                    f_phys,
+                    f_norm,
+                    reference_type,
+                )
 
             # ── 5. Adaptive Normalisation ─────────────────────────────────────
             if adaptive != "none":

src/portfolio/moead_awa.py

@@ -27,7 +27,6 @@ def generate_pareto_front(
         """
         MOEA/D-AWA with optional adaptive bounds and weight adjustment.
         """
-        num_assets = len(kwargs["expected_returns"])
         num_objectives = len(self.problem.objectives)
 
         # ── 1. Weight vectors & neighbourhood ────────────────────────────────
@@ -38,42 +37,10 @@ def generate_pareto_front(
         # ── 2. Initialize Normalisation ──────────────────────────────────────
         self._init_normalization(kwargs)
 
-        def repair(w):
-            if repair_method == "euclidean":
-                return self._repair(w)
-            elif repair_method == "simple":
-                return self._repair_simple(w)
-            else:
-                return w
-
         # ── 3. Initial Population ─────────────────────────────────────────────
-        candidates = []
-        for j in range(num_assets):
-            w = np.zeros(num_assets)
-            w[j] = 1.0
-            candidates.append(w)
-        candidates.append(np.ones(num_assets) / num_assets)
-        candidates.extend(
-            np.random.dirichlet(np.ones(num_assets), max(100, 2 * num_points))
-        )
-        cand_arr = np.array(candidates)
-        cand_f = np.array(
-            [list(self.problem.evaluate_all(w, **kwargs).values()) for w in cand_arr]
+        population, f_phys, f_norm = self._initial_sampling(
+            weight_vectors, reference_type, **kwargs
         )
-        cand_fn = np.array([self._normalise(f) for f in cand_f])
-
-        population = np.empty((num_points, num_assets))
-        f_phys = np.empty((num_points, num_objectives))
-        f_norm = np.empty((num_points, num_objectives))
-
-        for k, wv in enumerate(weight_vectors):
-            g_vals = np.array(
-                [self._get_scalar_value(fn, wv, reference_type) for fn in cand_fn]
-            )
-            best = int(np.argmin(g_vals))
-            population[k] = cand_arr[best]
-            f_phys[k] = cand_f[best]
-            f_norm[k] = cand_fn[best]
 
         # Elite Population (EP) for AWA
         ep_weights = population.copy()
@@ -91,48 +58,35 @@ def repair(w):
                 nb_idx = neighbors[i]
                 p1_idx, p2_idx = np.random.choice(nb_idx, 2, replace=False)
 
-                # Generate offspring
-                if crossover_operator == "sbx":
-                    offspring = self._sbx_crossover(
-                        population[p1_idx], population[p2_idx]
-                    )
-                    offspring = repair(offspring)
-                elif crossover_operator == "simplex":
-                    offspring = self._simplex_crossover(
-                        population[p1_idx], population[p2_idx]
-                    )
-                elif crossover_operator == "ldc":
-                    offspring = self._ldc_crossover(
-                        population[p1_idx], population[p2_idx]
-                    )
-                else:
-                    offspring = population[p1_idx].copy()
-
-                if mutation_operator == "polynomial":
-                    offspring = self._polynomial_mutation(offspring)
-                    offspring = repair(offspring)
-                elif mutation_operator == "simplex":
-                    offspring = self._simplex_mutation(offspring, eta=mutation_eta)
+                # Generate offspring using shared operators
+                offspring = self._apply_genetic_operators(
+                    population[p1_idx],
+                    population[p2_idx],
+                    crossover_operator,
+                    mutation_operator,
+                    mutation_eta,
+                    repair_method,
+                )
 
+                # Evaluate offspring
                 off_f = np.array(
                     list(self.problem.evaluate_all(offspring, **kwargs).values())
                 )
                 off_fn = self._normalise(off_f)
 
-                # Update neighbors
-                count = 0
-                for j in nb_idx:
-                    if count >= nr:
-                        break
-                    wv_j = weight_vectors[j]
-                    curr_g = self._get_scalar_value(f_norm[j], wv_j, reference_type)
-                    off_g = self._get_scalar_value(off_fn, wv_j, reference_type)
-
-                    if off_g < curr_g:
-                        population[j] = offspring
-                        f_phys[j] = off_f
-                        f_norm[j] = off_fn
-                        count += 1
+                # Update neighbors using shared logic
+                self._update_neighborhood(
+                    offspring,
+                    off_f,
+                    off_fn,
+                    nb_idx,
+                    nr,
+                    weight_vectors,
+                    population,
+                    f_phys,
+                    f_norm,
+                    reference_type,
+                )
 
                 # ── 5. Update Elite Population (EP) ───────────────────────────
                 # Check if offspring is non-dominated by current EP
@@ -290,20 +244,3 @@ def repair(w):
         if kwargs.get("record_history", False):
             return metrics, population, weight_vectors, history
         return metrics, population, weight_vectors
-
-    def _get_non_dominated_indices(self, objectives):
-        num_solutions = len(objectives)
-        if num_solutions <= 1:
-            return np.ones(num_solutions, dtype=bool)
-
-        is_efficient = np.ones(num_solutions, dtype=bool)
-        for i in range(num_solutions):
-            if is_efficient[i]:
-                c = objectives[i]
-                # A solution j is dominated by c if all(c <= j) and any(c < j)
-                # Set is_efficient[j] to False if c dominates j
-                is_dominated = np.all(c <= objectives, axis=1) & np.any(
-                    c < objectives, axis=1
-                )
-                is_efficient[is_dominated] = False
-        return is_efficient