jung-geun/PSO
1
0
Fork 0
mirror of https://github.com/jung-geun/PSO.git synced 2025-12-19 20:44:39 +09:00
Code Issues Packages Projects Releases Wiki Activity

23-10-25

Browse Source 
version 1.0.3
최고 점수 클래스 변수로 변경
log 저장 위치 고정
bean, seeds 데이터셋 추가 실험
This commit is contained in:
jung-geun
2023-10-25 12:53:23 +09:00
parent c8741dcd6d
commit 389027409d
13 changed files with 19828 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
pso/particle.py
View File

@@ -1,5 +1,4 @@
import numpy as np
from tensorflow import keras
@@ -13,11 +12,14 @@ class Particle:
4. 가중치 업데이트
5. 2번으로 돌아가서 반복
"""
g_best_score = [np.inf, 0, np.inf]
g_best_weights = None
count = 0
def __init__(
self,
model: keras.models,
loss,
loss: any = None,
negative: bool = False,
mutation: float = 0,
converge_reset: bool = False,
@@ -50,7 +52,7 @@ class Particle:
print(e)
exit(1)
self.__reset_particle__()
self.__reset_particle()
self.best_weights = self.model.get_weights()
self.before_best = self.model.get_weights()
self.negative = negative
@@ -62,6 +64,7 @@ class Particle:
self.converge_reset_patience = converge_reset_patience
self.converge_reset_monitor = converge_reset_monitor
self.converge_reset_min_delta = converge_reset_min_delta
Particle.count += 1
def __del__(self):
del self.model
@@ -70,6 +73,7 @@ class Particle:
del self.negative
del self.best_score
del self.best_weights
Particle.count -= 1
def _encode(self, weights: list):
"""
@@ -135,35 +139,37 @@ class Particle:
score = self.model.evaluate(x, y, verbose=0)
if renewal == "loss":
if score[0] < self.best_score[0]:
self.best_score[0] = score[0]
self.best_score = score
self.best_weights = self.model.get_weights()
elif renewal == "acc":
if score[1] > self.best_score[1]:
self.best_score[1] = score[1]
self.best_score = score
self.best_weights = self.model.get_weights()
elif renewal == "mse":
if score[2] < self.best_score[2]:
self.best_score[2] = score[2]
self.best_score = score
self.best_weights = self.model.get_weights()
else:
raise ValueError("renewal must be 'acc' or 'loss' or 'mse'")
return score
def __check_converge_reset__(self, score, monitor="loss", patience: int = 10, min_delta: float = 0.0001):
def __check_converge_reset(self, score, monitor: str = None, patience: int = 10, min_delta: float = 0.0001):
"""
early stop을 구현한 함수
Args:
score (float): 현재 점수 [0] - loss, [1] - acc
monitor (str, optional): 감시할 점수. Defaults to "loss".
monitor (str, optional): 감시할 점수. Defaults to acc. | "acc" or "loss" or "mse"
patience (int, optional): early stop을 위한 기다리는 횟수. Defaults to 10.
min_delta (float, optional): early stop을 위한 최소 변화량. Defaults to 0.0001.
"""
if monitor in ["acc", "accuracy"]:
self.score_history.append(score[1])
elif monitor in ["loss"]:
if monitor is None:
monitor = "acc"
if monitor in ["loss"]:
self.score_history.append(score[0])
elif monitor in ["acc", "accuracy"]:
self.score_history.append(score[1])
elif monitor in ["mse"]:
self.score_history.append(score[2])
else:
@@ -176,7 +182,7 @@ class Particle:
return True
return False
def __reset_particle__(self):
def __reset_particle(self):
self.model = keras.models.model_from_json(self.model.to_json())
self.model.compile(
optimizer="adam",
@@ -184,13 +190,13 @@ class Particle:
metrics=["accuracy", "mse"]
)
i_w_, i_s, i_l = self._encode(self.model.get_weights())
i_w_ = np.random.uniform(-0.05, 0.05, len(i_w_))
i_w_ = np.random.uniform(-0.05, 0.1, len(i_w_))
self.velocities = self._decode(i_w_, i_s, i_l)
del i_w_, i_s, i_l
self.score_history = []
def _update_velocity(self, local_rate, global_rate, w, g_best):
def _update_velocity(self, local_rate, global_rate, w):
"""
현재 속도 업데이트
@@ -198,19 +204,20 @@ class Particle:
local_rate (float): 지역 최적해의 영향력
global_rate (float): 전역 최적해의 영향력
w (float): 현재 속도의 영향력 - 관성 | 0.9 ~ 0.4 이 적당
g_best (list): 전역 최적해
"""
encode_w, w_sh, w_len = self._encode(weights=self.model.get_weights())
encode_v, v_sh, v_len = self._encode(weights=self.velocities)
encode_p, p_sh, p_len = self._encode(weights=self.best_weights)
encode_g, g_sh, g_len = self._encode(weights=g_best)
encode_g, g_sh, g_len = self._encode(weights=Particle.g_best_weights)
encode_before, before_sh, before_len = self._encode(
weights=self.before_best
)
r_0 = np.random.rand()
r_1 = np.random.rand()
# 이전 전역 최적해와 현재 전역 최적해가 다르면 관성을 순간적으로 증가 - 값이 바뀔 경우 기존 관성을 특정 기간동안 유지
if not np.array_equal(encode_before, encode_g, equal_nan=True):
# 이전 가중치 중요도의 1.5 배로 관성을 증가
self.before_w = w * 0.5
w = w + self.before_w
else:
@@ -218,13 +225,14 @@ class Particle:
w = w + self.before_w
if self.negative:
# 지역 최적해와 전역 최적해를 음수로 사용하여 전역 탐색을 유도
new_v = (
w * encode_v
+ local_rate * r_0 * (encode_p - encode_w)
+ -1 * global_rate * r_1 * (encode_g - encode_w)
- local_rate * r_0 * (encode_p - encode_w)
- global_rate * r_1 * (encode_g - encode_w)
)
if len(self.score_history) > 10 and max(self.score_history[-10:]) - min(self.score_history[-10:]) < 0.01:
self.__reset_particle__()
self.__reset_particle()
else:
new_v = (
@@ -246,7 +254,7 @@ class Particle:
del encode_before, before_sh, before_len
del r_0, r_1
def _update_velocity_w(self, local_rate, global_rate, w, w_p, w_g, g_best):
def _update_velocity_w(self, local_rate, global_rate, w, w_p, w_g):
"""
현재 속도 업데이트
기본 업데이트의 변형으로 지역 최적해와 전역 최적해를 분산시켜 조기 수렴을 방지
@@ -257,12 +265,11 @@ class Particle:
w (float): 현재 속도의 영향력 - 관성 | 0.9 ~ 0.4 이 적당
w_p (float): 지역 최적해의 분산 정도
w_g (float): 전역 최적해의 분산 정도
g_best (list): 전역 최적해
"""
encode_w, w_sh, w_len = self._encode(weights=self.model.get_weights())
encode_v, v_sh, v_len = self._encode(weights=self.velocities)
encode_p, p_sh, p_len = self._encode(weights=self.best_weights)
encode_g, g_sh, g_len = self._encode(weights=g_best)
encode_g, g_sh, g_len = self._encode(weights=Particle.g_best_weights)
encode_before, before_sh, before_len = self._encode(
weights=self.before_best
)
@@ -275,11 +282,12 @@ class Particle:
else:
self.before_w *= 0.75
w = w + self.before_w
if self.negative:
new_v = (
w * encode_v
+ local_rate * r_0 * (w_p * encode_p - encode_w)
+ -1 * global_rate * r_1 * (w_g * encode_g - encode_w)
- local_rate * r_0 * (w_p * encode_p - encode_w)
- global_rate * r_1 * (w_g * encode_g - encode_w)
)
else:
new_v = (
@@ -289,7 +297,7 @@ class Particle:
)
if np.random.rand() < self.mutation:
m_v = np.random.uniform(-0.1, 0.1, len(encode_v))
m_v = np.random.uniform(-0.05, 0.05, len(encode_v))
new_v = m_v
self.velocities = self._decode(new_v, w_sh, w_len)
@@ -313,7 +321,7 @@ class Particle:
del encode_w, w_sh, w_len
del encode_v, v_sh, v_len
def step(self, x, y, local_rate, global_rate, w, g_best, renewal: str = "acc"):
def step(self, x, y, local_rate, global_rate, w, renewal: str = "acc"):
"""
파티클의 한 스텝을 진행합니다.
@@ -329,19 +337,35 @@ class Particle:
Returns:
list: 현재 파티클의 점수
"""
self._update_velocity(local_rate, global_rate, w, g_best)
self._update_velocity(local_rate, global_rate, w)
self._update_weights()
score = self.get_score(x, y, renewal)
if self.converge_reset and self.__check_converge_reset__(
if self.converge_reset and self.__check_converge_reset(
score, self.converge_reset_monitor, self.converge_reset_patience, self.converge_reset_min_delta):
self.__reset_particle__()
self.__reset_particle()
score = self.get_score(x, y, renewal)
while np.isnan(score[0]) or np.isnan(score[1]) or np.isnan(score[2]) or score[0] == 0 or score[1] == 0 or score[2] == 0 or np.isinf(score[0]) or np.isinf(score[1]) or np.isinf(score[2]) or score[0] > 1000 or score[1] > 1 or score[2] > 1000:
self.__reset_particle()
score = self.get_score(x, y, renewal)
# # score 가 inf 이면 가중치를 초기화
# # score 가 nan 이면 가중치를 초기화
# # score 가 0 이면 가중치를 초기화
# if np.isinf(score[0]) or np.isinf(score[1]) or np.isinf(score[2]) or np.isnan(score[0]) or np.isnan(score[1]) or np.isnan(score[2]) or score[0] == 0 or score[1] == 0 or score[2] == 0:
# self.__reset_particle()
# score = self.get_score(x, y, renewal)
# # score 가 상식적인 범위를 벗어나면 가중치를 초기화
# if score[0] > 1000 or score[1] > 1 or score[2] > 1000:
# self.__reset_particle()
# score = self.get_score(x, y, renewal)
return score
def step_w(
self, x, y, local_rate, global_rate, w, g_best, w_p, w_g, renewal: str = "acc"
self, x, y, local_rate, global_rate, w, w_p, w_g, renewal: str = "acc"
):
"""
파티클의 한 스텝을 진행합니다.
@@ -361,10 +385,32 @@ class Particle:
Returns:
float: 현재 파티클의 점수
"""
self._update_velocity_w(local_rate, global_rate, w, w_p, w_g, g_best)
self._update_velocity_w(local_rate, global_rate, w, w_p, w_g)
self._update_weights()
return self.get_score(x, y, renewal)
score = self.get_score(x, y, renewal)
if self.converge_reset and self.__check_converge_reset(
score, self.converge_reset_monitor, self.converge_reset_patience, self.converge_reset_min_delta):
self.__reset_particle()
score = self.get_score(x, y, renewal)
while np.isnan(score[0]) or np.isnan(score[1]) or np.isnan(score[2]) or score[0] == 0 or score[1] == 0 or score[2] == 0 or np.isinf(score[0]) or np.isinf(score[1]) or np.isinf(score[2]) or score[0] > 1000 or score[1] > 1 or score[2] > 1000:
self.__reset_particle()
score = self.get_score(x, y, renewal)
# # score 가 inf 이면 가중치를 초기화
# # score 가 nan 이면 가중치를 초기화
# # score 가 0 이면 가중치를 초기화
# if np.isinf(score[0]) or np.isinf(score[1]) or np.isinf(score[2]) or np.isnan(score[0]) or np.isnan(score[1]) or np.isnan(score[2]) or score[0] == 0 or score[1] == 0 or score[2] == 0:
# self.__reset_particle()
# score = self.get_score(x, y, renewal)
# # score 가 상식적인 범위를 벗어나면 가중치를 초기화
# if score[0] > 1000 or score[1] > 1 or score[2] > 1000:
# self.__reset_particle()
# score = self.get_score(x, y, renewal)
return score
def get_best_score(self):
"""
@@ -383,3 +429,19 @@ class Particle:
list: 가중치 리스트
"""
return self.best_weights
def set_global_score(self):
"""전역 최고점수를 현재 파티클의 최고점수로 설정합니다
"""
Particle.g_best_score = self.best_score
def set_global_weights(self):
"""전역 최고점수를 받은 가중치를 현재 파티클의 최고점수를 받은 가중치로 설정합니다
"""
Particle.g_best_weights = self.best_weights
def update_global_best(self):
"""현재 파티클의 점수와 가중치를 전역 최고점수와 가중치로 설정합니다
"""
self.set_global_score()
self.set_global_weights()