与えられた確率に応じて個体を選択するルーレット選択を実装したい。

Question

前提

遺伝的プログラミングGPで選択を行う際にルーレット選択を行いたいです。
添付しているソースコードのpopulation変数には、1000個の個体(数式)が格納されています。
populationに格納されているそれぞれの個体はTreeクラスによって作られたインスタンスで、tree.fitnessという変数を持ちます。tree.fitnessは各個体の適応度を表しています。

コード中のmain関数でselectという関数を用いて、選択という操作を行っています。
このコードは見た感じトーナメント選択という手法をとっています。
トーナメント選択とは、populationの中からランダムにM個(指定した分)選び、その中でtree.fitnessが最も小さいものを新たにpopulationに格納、といった操作を行っています。
この部分をルーレット選択で実装したいのですが、うまくできず困っています。

ルーレット選択はpopulationに格納されている各個体の適応度tree.fitnessに注目し、
個体i(population[i])の選ばれる確率 = 個体iの適応度 / 全個体の適応度の和
とし、各個体が選択される確率を与えます。その後、与えられた確率に応じて個体を選択し、新たにpopulationに格納、という手順をとります。

実現したいこと

コード中のトーナメント選択をルーレット選択に変えて実装したい。

コードの補足説明

〇main関数
N: 個体数（population）に格納される数
M: トーナメント選択数
initial_population: 初期個体（数式。Treeインスタンス）をランダムに1000個生成し、populationに格納。
各個体はfitnessという変数をもっており、適応度を表す。

実際に今回使う関数はこれくらいだと思うので、ほかの部分は無視していただいて結構です。
わからない箇所あれば補足いたします。
よろしくお願いいたします。

補足情報（FW/ツールのバージョンなど）

参考

該当のソースコード

import operator
import math
import random
import numpy as np
from copy import deepcopy
import matplotlib.pyplot as plt


class Tree(list):
    def __init__(self,content):
        list.__init__(self,content)
        self.fitness = None
    @property
    def height(self):
        stack = [0]
        max_depth = 0
        for elem in self:
            depth = stack.pop()
            max_depth = max(max_depth, depth)
            stack.extend([depth+1] * elem[2])
        return max_depth
    def searchSubtree(self, begin):
        end = begin + 1
        total = self[begin][2]
        while total > 0:
            total += self[end][2] - 1
            end += 1
        return slice(begin, end)
    def __str__(self):
        start = 0
        str_arr = [""]
        for node in self:
            for i in range(len(str_arr)):
                if str_arr[i] == "":
                    name,func,arity = node
                    if arity != 0:
                        tmp_arr = [name,"(",""]
                        for x in range(arity-1):
                            tmp_arr += [",",""]
                        tmp_arr.append(")")
                        str_arr = str_arr[:i] + tmp_arr + str_arr[i+1:]
                    else:
                        str_arr = str_arr[:i] + [name] + str_arr[i+1:]
                    break
        return "".join(str_arr)

def makePrimeNumbers(len_numbers):
    primes = [2, 3]
    n = 5
    while len(primes) < len_numbers:
        isprime = True
        for i in range(1, len(primes)):
            if primes[i] ** 2 > n:
                break
            if n % primes[i] == 0:
                isprime = False
                break
        if isprime:
            primes.append(n)
        n += 2
    return primes

def protectedDiv(left, right):
    with np.errstate(divide='ignore',invalid='ignore'):
        y = np.divide(left, right)
        if isinstance(y, np.ndarray):
            y[np.isinf(y)] = 1
            y[np.isnan(y)] = 1
        elif np.isinf(y) or np.isnan(y):
            y = 1
    return y

def protectedPower(left,right):
    with np.errstate(over='ignore',invalid='ignore'):
        y = np.power(left, right)
        if isinstance(y, np.ndarray):
            y[np.isinf(y)] = 1
            y[np.isnan(y)] = 1
        elif np.isinf(y) or np.isnan(y):
            y = 1
    return y

def makeNodeSet():
    nodeSet = []
    leafSet = []
    nodeSet.append(("add",np.add,2))
    nodeSet.append(("sub",np.subtract,2))
    nodeSet.append(("mul",np.multiply,2))
    nodeSet.append(("div",protectedDiv,2))
    #nodeSet.append(("pow",np.power,2))
    nodeSet.append(("sin",np.sin,1))
    nodeSet.append(("cos",np.cos,1))
    nodeSet.append(("tan",np.tan,1))
    # nodeSet.append(("log",np.log,1))
    # nodeSet.append(("exp",np.exp,1))
    # nodeSet.append(("sqrt",np.sqrt,1))

    leafSet.append(("-1",1,0))
    leafSet.append(("0",0,0))
    leafSet.append(("1",1,0))
    leafSet.append(("0.5",0.5,0))
    leafSet.append(("0.1",0.1,0))
    leafSet.append(("pi",np.pi,0))
    leafSet.append(("e",np.e,0))
    leafSet.append(("x",None,0))

    return nodeSet,leafSet

def randomTreeMake(nodeSet,leafSet,_min,_max):
    height = random.randint(_min,_max)
    items = []
    stack = [0]
    while len(stack) != 0:
        depth = stack.pop()
        if depth == height:
            node = random.choice(leafSet)
        else:
            node = random.choice(nodeSet)
            stack.extend([depth+1]*node[2])
        items.append(node)
    tree = Tree(items)
    return tree

def get_y(tree,nodeSet,leafSet,x):
    stack = []
    for node in tree[::-1]:
        name,func,arity = node
        if name == "x":
            output = x
        else:
            if arity == 0:
                output = func*np.ones(x.shape[0])
            if arity == 1:
                output = func(stack.pop())
            if arity == 2:
                output = func(stack.pop(),stack.pop())
        stack.append(output)
    return stack.pop()

def mate(tree1,tree2,limit,limMode):
    save1 = deepcopy(tree1)
    save2 = deepcopy(tree2)
    while True:
        index1 = random.randint(1,len(tree1)-1)
        index2 = random.randint(1,len(tree2)-1)
        slice1 = tree1.searchSubtree(index1)
        slice2 = tree2.searchSubtree(index2)
        tree1[slice1],tree2[slice2] = tree2[slice2],tree1[slice1]
        if limMode == "LENGTH":
            if len(tree1) <= limit and len(tree2) <= limit:
                return tree1,tree2
        elif limMode == "HEIGHT":
            if tree1.height <= limit and tree2.height <= limit:
                return tree1,tree2
        else:
            return tree1,tree2
        tree1,tree2 = deepcopy(save1),deepcopy(save2)

def mutate(tree,NodeSet,leafSet,limit,limMode):
    save = deepcopy(tree)
    while True:
        index = random.randrange(len(tree))
        slice_ = tree.searchSubtree(index)
        tree[slice_] = randomTreeMake(NodeSet,leafSet,1,2)
        if limMode == "LENGTH":
            if len(tree) <= limit:
                return tree
        elif limMode == "HEIGHT":
            if tree.height <= limit:
                return tree
        else:
            return tree
        tree = deepcopy(save)

def initial_population(N,nodeSet,leafSet,_min,_max):
    population = []
    for n in range(N):
        tree = randomTreeMake(nodeSet,leafSet,_min,_max)
        population.append(tree)
    return population

def evaluate(population,nodeSet,leafSet,x,y):
    for tree in population:
        if tree.fitness == None:
            _y = get_y(tree,nodeSet,leafSet,x)
            fitness = np.sum(np.abs(_y - y))
            tree.fitness = fitness

def select(population,M):
    new_population = []
    for i in range(len(population)):
        sub_population = random.sample(population,M)
        best = min(sub_population,key=lambda tree:tree.fitness)
        new_population.append(deepcopy(best))
    return new_population

def crossover(population,Pcx):
    random.shuffle(population)
    for i in range(int(len(population)/2)):
        if random.random() < Pcx:
            childA = population[2*i]
            childB = population[2*i+1]
            childA,childB = mate(childA,childB,17,"HEIGHT")
            population[2*i] = deepcopy(childA)
            population[2*i+1] = deepcopy(childB)
            population[2*i].fitness = None
            population[2*i+1].fitness = None
    return population

def mutation(population,Pmut,nodeSet,leafSet):
    random.shuffle(population)
    for i in range(len(population)):
        if random.random() < Pmut:
            child = population[i]
            child = mutate(child,nodeSet,leafSet,17,"HEIGHT")
            population[i] = deepcopy(child)
            population[i].fitness = None

    return population

def printLog(population,g):
    key=lambda tree:tree.fitness
    worst = max(population,key=key)
    best = min(population,key=key)
    ave = sum([tree.fitness for tree in population])/len(population)
    key=lambda tree:len(tree)
    max_node = len(max(population,key=key))
    min_node = len(min(population,key=key))
    ave_node = sum([len(tree) for tree in population])/len(population)
    print("===========",g,"genelation===========")
    print("worst fitness : ",worst.fitness,"  max nodes : ",max_node)
    print("ave   fitness : ",ave          ,"  ave nodes : ",ave_node)
    print("best  fitness : ",best.fitness, "  min nodes : ",min_node)

def makeGraph(tree,x,y,nodeSet,leafSet):
    _y = get_y(tree,nodeSet,leafSet,x)
    plt.plot(x,y,label="true prime number")
    plt.plot(x,_y,label="gp prime number")
    plt.legend(loc="upper left")
    plt.savefig("test.pdf")

def main():
    N = 1000
    M = 7
    maxG = 500
    Pcx = 0.5
    Pmut = 0.1
    nodeSet,leafSet = makeNodeSet()
    x = np.array(range(1,10))
    y = np.array(makePrimeNumbers(x.shape[0]))

    population = initial_population(N,nodeSet,leafSet,1,2)
    g = 0
    while True:
        evaluate(population,nodeSet,leafSet,x,y)
        printLog(population,g)
        if g == maxG:
            break
        population = select(population,M)  # 変更したい部分
        population = crossover(population,Pcx)
        population = mutation(population,Pmut,nodeSet,leafSet)
        g+=1
    best = min(population,key=lambda tree:tree.fitness)
    print(best)
    makeGraph(best,x,y,nodeSet,leafSet)

main()

Answer 1

重み付き乱数を使用してみてはいかがでしょうか。
https://docs.python.org/ja/3.10/library/random.html#random.choices
random.choices()関数の引数 weights に各要素の重みを指定することで、相対的な重みに基づいて要素が選ばれます。

sub_population = random.choices(population, weights=(p.fitness for p in population), k=M)

fitnessが小さいものを優先するのであれば下記のようになるかと

sub_population = random.choices(population, weights=(1/p.fitness for p in population), k=M)

追記

M個を重複なしで、かつfitness に応じて優先的に選出したい（fitness大ならば選出確率大）という場合、
random.sample の重み付き版である numpy.random.choice(replace=false, p=重みリスト)を使ってみてはいかがでしょうか。
（非復元重み付きランダムサンプリング）

def select(population,M):
    new_population = []
    # probeの総和を1にするためにfitnessの合計値を求めておく
    s = sum(u.fitness for u in population)
    # fitnessから求めた population各個体の選出確率のリスト（重みリスト）
    prob = [p.fitness/s for p in population]
    for i in range(len(population)):
        sub_population = np.random.choice(population,size=M,replace=False,p=prob)
        best = min(sub_population,key=lambda tree:tree.fitness)
        new_population.append(deepcopy(best))
    return new_population

numpyを使わない場合は下記のような形で。

def select(population,M):
    new_population = []
    for i in range(len(population)):
        sub_population = []
        # 選出用populationのインデックスとfitnessのペアを格納したプール
        pool = [[idx, p.fitness] for idx,p in enumerate(population)]

        # M個選出するためにM回繰り返す
        for _ in range(M):
            # 現在のプールの重みリスト
            w = [p[1] for p in pool]
            # 重み付き確率に応じて選出プールから1個選出する
            selected = random.choices(pool, weights=w, k=1)[0]
            sub_population.append(population[selected[0]])
            # 選出された個体の重みをゼロにして選出対象外とする
            selected[1]=0

        best = min(sub_population,key=lambda tree:tree.fitness)
        new_population.append(deepcopy(best))
    return new_population

※いずれにしても計算は重くなります。

Comments

[AI_engineer]

ご回答ありがとうございます。
URL見ましたが、いまいち理解できませんでした...
ルーレット選択では、
（個体の選ばれる確率）＝　（個体の適応度）/ （全個体の適応度の和）
で表されます。
相対的な重みに基づいてとありますが、この重み付き乱数でも上式と同じ確率で選択されるのでしょうか。

[AI_engineer]

書いていただいたコードを試してみましたが、sub_populationに選ばられる個体が毎回同じになってしまいます...

[退会済みユーザー]

はい。
たとえば下記のコードの結果をみていただければ適応度とみなした「重み」配列の確率で選択されていることが
お分かりいただけるかと思います。
import random
numbers = list(range(0,10)) #背番号0～9
w = [1, 3, 5, 7, 9, 9, 7, 5, 3, 1] # index0～9：各背番号0～9に応じた重み　（＝適応度に相当）
samples = random.choices(numbers, k = 10000, weights = w)
print([samples.count(i) for i in numbers])

[退会済みユーザー]

>sub_populationに選ばられる個体が毎回同じになってしまいます...
そもそももともとのコードでは、設定されるfitness が、世代を経るごとに大幅な偏りが出るつくりになっているためではないでしょうか。
「全体個体のfitness総和に対する、各個体のfitnessに応じて選出されるようにする」というのが要件と伺っておりますので、仮にfitness が突出して**大きい**個体が存在する場合、サンプリングの過程でその個体が優先的に選ばれるのは当然です。
現状の**select関数内で**計算毎に使用される、各個体のfitnessが、どのような値であるか、確認されましたでしょうか？
(場合によっては、質問者さんが想定しない値になっている可能性があるかもしれませんが）
（もしかしてfitnessが小さい方を優先的に選ぶ、という要件がありましたでしょうか・であれば回答に追記した形のように逆数にしてみてください。）

（fitnessに応じた確率での選出は無視して）1回あたりのサンプリングではとりあえずM個重複なしで選出したい、というのであれば、コードはまた別の形になります。ただしこれは非復元重み付きサンプリングになりますので必ずしも直感的に期待される分布にはなりません。

[退会済みユーザー]

重複なしで選出する場合のコードを追記しました。

[AI_engineer]

ご回答ありがとうございます。

コードを見直してみると適合度が小さいものほどよいっぽいですね。
コードの方も勉強になりました。
ありがとうございました。