mutation

lab4/gp/ga.py

@@ -17,7 +17,7 @@ type FitnessFn = Callable[[Chromosome], float]
 
 type InitializePopulationFn = Callable[[int], Population]
 type CrossoverFn = Callable[[Chromosome, Chromosome], tuple[Chromosome, Chromosome]]
-type MutationFn = Callable[[Chromosome, int], Chromosome]
+type MutationFn = Callable[[Chromosome], Chromosome]
 type SelectionFn = Callable[[Population, Fitnesses], Population]
 
 
@@ -132,7 +132,7 @@ def mutation(
     next_population = []
     for chrom in population:
         next_population.append(
-            mutation_fn(chrom, gen_num) if np.random.random() <= pm else chrom
+            mutation_fn(chrom) if np.random.random() <= pm else chrom
         )
     return next_population
 

lab4/gp/mutations.py

@@ -1,12 +1,22 @@
 import random
+from abc import ABC, abstractmethod
+from typing import Sequence
 
 from .chromosome import Chromosome
 
 
-def shrink_mutation(chromosome: Chromosome) -> Chromosome:
-    """Усекающая мутация. Заменяет случайно выбранную операцию на случайный терминал."""
+class BaseMutation(ABC):
+    @abstractmethod
+    def mutate(self, chromosome: Chromosome) -> Chromosome: ...
+    def __call__(self, chromosome: Chromosome) -> Chromosome:
         chromosome = chromosome.copy()
+        return self.mutate(chromosome)
 
+
+class ShrinkMutation(BaseMutation):
+    """Усекающая мутация. Заменяет случайно выбранную операцию на случайный терминал."""
+
+    def mutate(self, chromosome: Chromosome) -> Chromosome:
         operation_nodes = [n for n in chromosome.root.list_nodes() if n.value.arity > 0]
 
         if not operation_nodes:
@@ -19,13 +29,16 @@ def shrink_mutation(chromosome: Chromosome) -> Chromosome:
         return chromosome
 
 
-def grow_mutation(chromosome: Chromosome, max_depth: int) -> Chromosome:
+class GrowMutation(BaseMutation):
     """Растущая мутация. Заменяет случайно выбранный узел на случайное поддерево."""
-    chromosome = chromosome.copy()
 
+    def __init__(self, max_depth: int):
+        self.max_depth = max_depth
+
+    def mutate(self, chromosome: Chromosome) -> Chromosome:
         target_node = random.choice(chromosome.root.list_nodes())
 
-    max_subtree_depth = max_depth - target_node.get_level() + 1
+        max_subtree_depth = self.max_depth - target_node.get_level() + 1
 
         subtree = Chromosome.grow_init(
             chromosome.terminals, chromosome.operations, max_subtree_depth
@@ -39,7 +52,7 @@ def grow_mutation(chromosome: Chromosome, max_depth: int) -> Chromosome:
         return chromosome
 
 
-def node_replacement_mutation(chromosome: Chromosome) -> Chromosome:
+class NodeReplacementMutation(BaseMutation):
     """Мутация замены операции (Node Replacement Mutation).
 
     Выбирает случайный узел и заменяет его
@@ -47,8 +60,8 @@ def node_replacement_mutation(chromosome: Chromosome) -> Chromosome:
 
     Если подходящей альтернативы нет — возвращает копию без изменений.
     """
-    chromosome = chromosome.copy()
 
+    def mutate(self, chromosome: Chromosome) -> Chromosome:
         target_node = random.choice(chromosome.root.list_nodes())
         current_arity = target_node.value.arity
 
@@ -67,16 +80,15 @@ def node_replacement_mutation(chromosome: Chromosome) -> Chromosome:
         return chromosome
 
 
-def hoist_mutation(chromosome: Chromosome) -> Chromosome:
+class HoistMutation(BaseMutation):
+    def mutate(self, chromosome: Chromosome) -> Chromosome:
         """Hoist-мутация (анти-bloat).
 
-    Выбирает случайное поддерево, затем внутри него — случайное поддерево меньшей глубины,
-    и заменяет исходное поддерево на это внутреннее.
+        Выбирает случайное поддерево, затем внутри него — случайное поддерево меньшей
+        глубины, и заменяет исходное поддерево на это внутреннее.
 
         В результате дерево становится короче, сохраняя часть структуры.
         """
-    chromosome = chromosome.copy()
-
         operation_nodes = [n for n in chromosome.root.list_nodes() if n.value.arity > 0]
         if not operation_nodes:
             return chromosome
@@ -92,3 +104,28 @@ def hoist_mutation(chromosome: Chromosome) -> Chromosome:
             chromosome.root = inner_subtree
 
         return chromosome
+
+
+class CombinedMutation(BaseMutation):
+    """Комбинированная мутация.
+
+    Принимает список (или словарь) мутаций и случайно выбирает одну из них
+    для применения. Можно задать веса вероятностей.
+    """
+
+    def __init__(
+        self, mutations: Sequence[BaseMutation], probs: Sequence[float] | None = None
+    ):
+        if probs is not None:
+            assert abs(sum(probs) - 1.0) < 1e-8, (
+                "Сумма вероятностей должна быть равна 1"
+            )
+            assert len(probs) == len(mutations), (
+                "Число вероятностей должно совпадать с числом мутаций"
+            )
+        self.mutations = mutations
+        self.probs = probs
+
+    def mutate(self, chromosome: Chromosome) -> Chromosome:
+        mutation = random.choices(self.mutations, weights=self.probs, k=1)[0]
+        return mutation(chromosome)

lab4/main.py

@@ -15,10 +15,11 @@ from gp.fitness import (
 )
 from gp.ga import GARunConfig, genetic_algorithm
 from gp.mutations import (
-    grow_mutation,
-    hoist_mutation,
-    node_replacement_mutation,
-    shrink_mutation,
+    CombinedMutation,
+    GrowMutation,
+    HoistMutation,
+    NodeReplacementMutation,
+    ShrinkMutation,
 )
 from gp.ops import ADD, COS, DIV, EXP, MUL, NEG, POW, SIN, SQUARE, SUB
 from gp.population import ramped_initialization
@@ -36,7 +37,6 @@ X = np.random.uniform(-5.536, 5.536, size=(TEST_POINTS, NUM_VARS))
 # axes = [np.linspace(-5.536, 5.536, TEST_POINTS) for _ in range(NUM_VARS)]
 # X = np.array(np.meshgrid(*axes)).T.reshape(-1, NUM_VARS)
 operations = [SQUARE, SIN, COS, EXP, ADD, SUB, MUL, DIV, POW]
-# operations = [SQUARE, ADD, SUB, MUL]
 terminals = [Var(f"x{i}") for i in range(1, NUM_VARS + 1)]
 
 
@@ -53,36 +53,16 @@ def target_function(x: NDArray[np.float64]) -> NDArray[np.float64]:
     return np.sum(prefix_sums, axis=1)
 
 
-# fitness_function = MSEFitness(target_function, lambda: X)
-# fitness_function = HuberFitness(target_function, lambda: X, delta=0.5)
-# fitness_function = PenalizedFitness(
-#     target_function, lambda: X, base_fitness=fitness, lambda_=0.1
-# )
-# fitness_function = NRMSEFitness(target_function, lambda: X)
 fitness_function = RMSEFitness(target_function, lambda: X)
-
-# fitness_function = PenalizedFitness(
-#     target_function, lambda: X, base_fitness=fitness_function, lambda_=0.0001
-# )
-
-
-def adaptive_mutation(
-    chromosome: Chromosome,
-    generation: int,
-    max_generations: int,
-    max_depth: int,
-) -> Chromosome:
-    r = random.random()
-
-    if r < 0.4:
-        return grow_mutation(chromosome, max_depth=max_depth)
-    elif r < 0.7:
-        return node_replacement_mutation(chromosome)
-    elif r < 0.85:
-        return hoist_mutation(chromosome)
-
-    return shrink_mutation(chromosome)
-
+combined_mutation = CombinedMutation(
+    mutations=[
+        GrowMutation(max_depth=MAX_DEPTH),
+        NodeReplacementMutation(),
+        HoistMutation(),
+        ShrinkMutation(),
+    ],
+    probs=[0.4, 0.3, 0.15, 0.15],
+)
 
 init_population = ramped_initialization(
     20, [i for i in range(MAX_DEPTH - 9, MAX_DEPTH + 1)], terminals, operations
@@ -93,9 +73,7 @@ print("Population size:", len(init_population))
 config = GARunConfig(
     fitness_func=fitness_function,
     crossover_fn=lambda p1, p2: crossover_subtree(p1, p2, max_depth=MAX_DEPTH),
-    mutation_fn=lambda chrom, gen_num: adaptive_mutation(
-        chrom, gen_num, MAX_GENERATIONS, MAX_DEPTH
-    ),
+    mutation_fn=combined_mutation,
     # selection_fn=roulette_selection,
     selection_fn=lambda p, f: tournament_selection(p, f, k=3),
     init_population=init_population,