I 'HATE' study

0. Toolbox

7 min read

  • Caution

    • 주의! 원서에 있는 코드랑 다른 부분이 존재합니다!

      • 순서변경

        • 편의를 위해 아래와 같이 순서를 재배치함

            def crossover_operation()
  def mutation_operation()
  def crossover_blend()
  def mutation_random_deviation()
  def selection_rank_with_elite()
            def selection_rank_with_elite()
  def crossover_operation()
  def crossover_blend()        
  def mutation_operation()        
  def mutation_random_deviation()

      • 코드변경 - 기존 코드에서는 선택된 엘리트 개체도 순위 선택에 포함시킴

        • 엘리트 선택을 먼저 하기 위해 코드를 위로 올림 
            selected = sorted_individuals[0:elite_size]

      • 코드추가

        • 엘리트 선택이 된 개체를 제거하기 위한 코드 추가 
            del sorted_individuals[0:elite_size]

  • Toolbox

    import random
from multiprocessing.pool import Pool
import copy
import random
 
 
# 병렬 처리를 위한 프로세스풀 생성하는 함수
def create_pool():
    return Pool(processes=4)
 
 
# 유전자의 값이 허용된 범위를 벗어나지 않도록 제약을 거는 함수
def constraints(x):
    if x > 10:
        return 10
    elif x < -10:
        return -10
    return x
 
 
# 순위 선택 기반의 엘리트 선택
def selection_rank_with_elite(individuals, elite_size=0):
    # 적합도 기준 내림차순 정렬
    sorted_individuals = sorted(individuals, key=lambda ind: ind.fitness, reverse=True)
 
    # 엘리트 선택: 상위 elite_size개의 개체를 먼저 선택
    selected = sorted_individuals[0:elite_size]
    
    # 엘리트 개체를 제외한 나머지 개체들로 리스트 갱신
    del sorted_individuals[0:elite_size]
 
    # 순위 간격 계산
    rank_distance = 1 / len(individuals)
 
    # 순위 점수 계산
    ranks = [(1 - i * rank_distance) for i in range(len(individuals))]
    ranks_sum = sum(ranks)
 
    # 나머지 개체는 확률 기반으로 선택
    for i in range(len(sorted_individuals) - elite_size):
        shave = random.random() * ranks_sum  # 무작위 선택 임계값
        rank_sum = 0
 
        for i in range(len(sorted_individuals)):
            rank_sum += ranks[i]
            if rank_sum > shave:
                selected.append(sorted_individuals[i])
                break
 
    return selected  # 선택된 개체 리스트 반환
 
 
# 개체 집단에 대해 교차 연산을 병렬로 수행하는 함수 ( 짝을 지어주겠다 )
def crossover_operation(population, method, prob):
    pool = create_pool()     # 병렬처리를 위한 프로세스 풀 생성
    crossed_offspring = []   # 최종 교차 결과가 저장될 리스트
    to_cross = []            # 교차 대상이 되는 개체 쌍 저장
    result = []              # 병렬 결과 저장용
 
    # 개체들을 2개씩 묶어 반복
    for ind1, ind2 in zip(population[::2], population[1::2]):
        # 주어진 확률에 따라 교차 수행 여부 결정
        if random.random() < prob:                    # prob의 확률로 교차할지 결정
            to_cross.extend([ind1, ind2])             # 교차 수행 예정
        else:
            crossed_offspring.extend([ind1, ind2])    # 교차 없이 그대로 유지
 
    # 실제 교차 연산 수행 (병렬 처리)
    for i in range(0, len(to_cross), 2):
        result.append(
            pool.apply_async(method, args=(to_cross[i], to_cross[i + 1]))
        )
 
    # 병렬 처리 종료 및 결과 수집
    pool.close()        # 새로운 작업을 추가 하지 못하게 닫아버림
    pool.join()         # 현재 실행중인 작업이 모두 끝날 때까지 대기    
 
    # 결과 수집
    for r in result:
        crossed_offspring.extend(r.get())
    # 최종 교차 결과 반환
    return crossed_offspring
 
 
# 두 부모 유전자를 섞어 자식 유전자 두 개를 생성하는 함수 ( 지어진 짝을 가지고 교차를 하겠다 )
def crossover_blend(p1, p2, alpha):
    # 부모 유전자의 복사본 생성
    c1 = copy.deepcopy(p1)   #deepcopy는 원본이 변경되지 않도록 하기 위해 사용
    c2 = copy.deepcopy(p2)
 
    # 각 유전자 위치별로 섞기
    for i in range(len(p1)):
        # 교차 범위 하한(Lower bound)과 상한(Upper bound) 계산
        l = min(c1[i], c2[i]) - alpha * abs(c2[i] - c1[i])
        u = max(c1[i], c2[i]) + alpha * abs(c2[i] - c1[i])
 
        # 새로운 유전자 값을 범위 내에서 무작위로 선택
        c1[i] = round(l + random.random() * (u - l), 2)
        c2[i] = round(l + random.random() * (u - l), 2)
 
    # 자식 유전자 리스트 반환
    return [c1, c2]
 
 
# 개체 집단에 대해 돌연변이 연산을 병렬로 수행하는 함수 ( 전체 개체중 prob%의 개체만 돌연변이를 시키겠다 )
def mutation_operation(population, method, prob):
    pool = create_pool()       # 병렬처리 풀 생성
    mutated_offspring = []     # 최종 결과 저장용
    to_mutate = []             # 돌연변이 대상
    result = []                # 병렬 결과
 
    # 각 개체에 대해 돌연변이 여부 판단
    for ind in population:
        if random.random() < prob:           # prob의 확률로 돌연변할지 결정
            to_mutate.append(ind)            # 돌연변이 대상
        else:
            mutated_offspring.append(ind)    # 그대로 유지
 
    # 돌연변이 수행 (병렬 처리)
    for ind in to_mutate:
        result.append(pool.apply_async(method, args=(ind,)))
 
    # 병렬 처리 종료 및 결과 수집
    pool.close()        # 새로운 작업을 추가 하지 못하게 닫아버림
    pool.join()         # 현재 실행중인 작업이 모두 끝날 때까지 대기
 
    # 결과 수집
    for r in result:
        mutated_offspring.append(r.get())
    # 최종 돌연변이이 결과 반환
    return mutated_offspring
 
 
# 한 개체의 유전자에 확률적으로 돌연변이를 가하는 함수 ( p%의 확률로 돌연변이를 실행하겠다 )
def mutation_random_deviation(ind, mu, sigma, p):
    # 부모 유전자의 복사본 생성
    mut = copy.deepcopy(ind)
 
    for i in range(len(mut)):
        # 확률적으로 돌연변이 적용
        if random.random() < p:
            # 정규분포에서 값을 더함
            mut[i] = mut[i] + random.gauss(mu, sigma)  #큰 변화보다는 작은 탐색 위주로 동작한다 → 국지적 최적화에 유리
            
    return mut  # 변경된 유전자 반환

New Posts