Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # My Utility : auxiliars functions
- #my_utility
- import pandas as pd
- import numpy as np
- import random as rd
- import math
- #Initialize Swarm
- #Nh = nodos ocultos
- #Np = numero de particulas
- #d = largo del enjambre
- def iniSwarm(Nh, Np, d):
- X = np.zeros((Np,(Nh*d)), dtype=float)
- for i in range(Np):
- wh = randW(Nh,d)
- a = np.reshape(wh, (1, Nh*d))
- X[i] = a
- return(X)
- # Initialize random weights
- def randW(next_, prev_):
- w = np.random.random((next_, prev_))
- r = np.sqrt(6 / ( next_ + prev_ ))
- w = ( w * 2 * r ) - r
- return(w)
- # MLP-fitness
- #x =
- #y =
- #X =
- #C =
- #param[0] Max. Iterations -> 200
- #param[1] Number of nodes -> 10
- #param[2] Number of particle -> 20
- #param[3] Penalidad C of the Pinv -> 100
- def mlp_Fitness(x, y, X, C):
- nh = int(X.shape[1]/x.shape[0])
- W2 = np.zeros((X.shape[0], nh), dtype=float) #matriz de peso para la capa de salida
- MSE = np.zeros(X.shape[0] , dtype=float)
- for i in range(X.shape[0]):
- p = X[i]
- W1 = np.reshape(p,(x.shape[1],nh))
- print(W1.shape,x.shape)
- H = act_function(x, W1)
- W2[i] = w2_pinv(H, y, C)
- #ze = np.matmul(W2[i], H)
- #MSE[i] = math.sqrt(MSE(y, ze))
- return(W2, MSE)
- # Update Particle fitness
- #param[2] = num_p = Número de particulas
- #X = Enjambre
- #P['Pos'] = Best particle position
- #P['Fit'] = Best particle fitness
- #P['gBest'] = Best global solution
- #P['gFit'] = Best global Fitness
- #w2Best = Mejor peso
- #MSE = Error cuadratico medio
- def upd_pFitness(X, P, new_pfitness, new_beta):
- for i in range(X.shape[0]):
- if (P['Fit'][i] > new_pfitness[i]):
- P['Fit'][i] = new_pfitness[i]
- P['Pos'][i][:] = X[i, :]
- new_gfitness = min(P['Fit'])
- idx = np.argmin(P['gFit'])
- if (new_gfitness < P['gFit']):
- P['gFit'] = new_gfitness
- P['gBest'] = P['Pos'][idx][:]
- w2Best = new_beta[idx][:]
- return(P, w2Best)
- '''# Update Velocity of PSO
- def upd_veloc():
- #complete
- return(...)
- '''
- # Update Swarm of QPSO
- def upd_swarm(P, iTer, alpha, X):
- avg_t = 0
- mBest = P['Pos'].mean(axis = 0)
- for i in range(X.shape[0]):
- for j in range(X.shape[1]):
- phi = rd.random()
- u = rd.random()
- P['Pos'][i][j] = ( phi * P['Pos'][i][j] + (1 - phi) * P['gBest'][j])
- z = alpha[iTer] * math.abs( mBest[j] - X[i][j] ) * math.log( 1 / u )
- avg_t += z
- if rd.random() > 0.5:
- X[i][j] = P['Pos'][i][j] + z
- else:
- X[i][j] = P['Pos'][i][j] - z
- avg_t = avg_t / ( X.shape[0] * X.shape[1] )
- return(P, X)
- # Update W2 via P-inverse
- def w2_pinv(H, y, C):
- #print(f"H {H} y {y} C {C}")
- yh = np.matmul(np.transpose(y), np.transpose(H))
- hh = np.matmul(H, np.transpose(H))
- hh += np.eye(hh.shape[0]) / C
- w2 = np.matmul(np.transpose(yh), np.linalg.pinv(hh))
- return(w2)
- # Feed-forward of ANN
- def forward(xv,w1,w2):
- param = load_config()
- w1 = w1.reshape((param[1],xv.shape[0]+1))
- H = act_function(xv, w1)
- zv = np.matmul(w2,H)
- for number in range(len(zv)):
- if zv[number] < 0:
- zv[number] = -1
- else:
- zv[number] = 1
- return(zv)
- # activation function
- def act_function(x_n, w_j):
- z = np.matmul(w_j,x_n)
- print(z.shape)
- for number in z:
- number = np.exp(-1*(number*number))
- print(z.shape)
- return z
- def distance_e(x,w):
- d = 0
- w = w.transpose()
- print(x.shape, w.shape)
- d = np.where(x.shape[0] == w.shape[0], ((x-w)**2)**(1/2), 0)
- print(f"d = {d}")
- #d = ( (x[0] – w[0])**2 + (x[1] – w[1])**2 )**(1/2)
- return d
- # Measure
- def metrica(y_expected,y_get):
- #se llama con ut.metrica(yv,zv)
- confusion_matrix(y_expected,y_get)
- vp = 0
- fp = 0
- fn = 0
- vn = 0
- for i in range(len(y_expected)):
- if y_expected[i] == 1 and y_get[i] == 1:
- vp += 1
- if y_expected[i] == -1 and y_get[i] == -1:
- vn += 1
- if y_expected[i] == 1 and y_get[i] == -1:
- fn += 1
- if y_expected[i] == -1 and y_get[i] == 1:
- fp += 1
- accuracy_normal = vp/(vp+fp)
- recall_normal = vp/(vp+fn)
- f_score_normal = 2*(accuracy_normal*recall_normal/(accuracy_normal+recall_normal))
- accuracy_attack = vn/(vn+fn)
- recall_attack = vn/(vn + fp)
- f_score_attack = 2*(accuracy_attack*recall_attack/(accuracy_attack + recall_attack))
- pd.DataFrame(accuracy_normal, f_score_normal, f_score_attack).to_csv("fscore_ann.csv")
- #Confusion matrix
- def confusion_matrix(a,b):
- pd.DataFrame(a, b).to_csv("cmatrix_ann.csv")
- # Norm of the data
- def norm_data(df_data = pd.DataFrame()):
- a = 0.01
- b = 0.99
- df_data = pd.DataFrame(df_data)
- # extraer datos para normalizar
- X = df_data.loc[:, df_data.columns != 41]
- Y = df_data.loc[:, df_data.columns == 41]
- #normalizar los datos mediante la fórmula
- X_normalized = ( X - X.min() ) / ( X.max() - X.min() )
- X_normalized *= ( b - a ) + a
- X_normalized = X_normalized.fillna(0)
- #Exportar datos
- X_normalized[41] = Y
- return(X_normalized)
- #
- #------------------------------------------------------------------------
- # LOAD-SAVE
- #-----------------------------------------------------------------------
- # Configuration of the ANN
- def load_config():
- param = np.genfromtxt("config_ann.csv",delimiter=',',dtype=None)
- par=[]
- par.append(np.int16(param[0])) # Max. Iterations
- par.append(np.int16(param[1])) # Number of nodes
- par.append(np.int16(param[2])) # Number of particle
- par.append(np.float(param[3])) # Penalidad C of the Pinv.
- return(par)
- # Load data
- def load_data(fname):
- x = pd.read_csv(fname, header = None)
- #x = np.array(x)
- #down
- df_data = norm_data(x)
- x = df_data.loc[:, df_data.columns != 41]
- y = df_data.loc[:, df_data.columns == 41]
- return(x.to_numpy(), y.to_numpy())
- # save ANN's weights in numpy format and cost in csv format
- def save_w(w1,w2,cost):
- #complete
- np.savez("pesos_ann", w1 = w1, w2 = w2)
- pd.DataFrame(cost).to_csv("cost_ann.csv") #SUJETO A MODIFICACIONES
- #load AN's weight in numpy format
- def load_w():
- #complete}
- open("pesos_ann.npz", "w").close()
- data = np.load("pesos_ann.npz")
- data_weight = {data}
- #fomatear los datos
- data_weight = [data[key] for key in data]
- w1 = data_weight[0]
- w2 = data_weight[1]
- return(w1,w2)
Advertisement
Add Comment
Please, Sign In to add comment