Prototipos de descenso de gradiente en el aprendizaje automático

Aprendiendo

El aprendizaje supervisado es una categoría de aprendizaje automático que utiliza conjuntos de datos etiquetados para entrenar algoritmos para predecir los resultados y reconocer los patrones.

A diferencia del aprendizaje no supervisado, los algoritmos de aprendizaje supervisados ​​reciben capacitación etiquetada para aprender la relación entre la entrada y las salidas.

Requisito previo: Álgebra lineal

Supongamos que tenemos un problema de regresión donde el modelo necesita predecir valores continuos tomando n número de características de entrada (XI).

El valor de predicción se define como una función llamada hipótesis (h):

dónde:

• θi: parámetro i-th correspondiente a cada función de entrada (x_i),
• ϵ (Epsilon): error gaussiano (ϵ ~ n (0, σ²)))

Como la hipótesis de una sola entrada genera un valor escalar (Hθ (x) ∈R), se puede denotar como el producto DOT del Transposición del vector de parámetros (θt) y el Vector de características para esa entrada (x):

Descenso de gradiente de lotes

Descenso de gradiente es un algoritmo de optimización iterativo utilizado para encontrar mínimos locales de una función. En cada paso, se mueve en la dirección opuesta a la dirección del descenso más empinado para reducir progresivamente el valor de la función, simplemente, continúe cuesta abajo.

Ahora, recuerde que tenemos N parámetros que afectan la predicción. Entonces, necesitamos conocer la contribución específica del parámetro individual (θi) correspondiente a los datos de entrenamiento (xi)) a la función.

Supongamos que establecemos el tamaño de cada paso como una tasa de aprendizaje (α), y encuentre una curva de costo (j), entonces el parámetro se deduce en cada paso de tal manera que:

(α: Tasa de aprendizaje, J (θ): COST Función, ∂/∂θi: derivada parcial de la función de costo con respecto a θi)

Gradiente

El gradiente representa la pendiente de la función de costo.

Teniendo en cuenta los parámetros restantes y sus derivadas parciales correspondientes de la función de costo (j), el gradiente de la función de costo en θ para n parámetros se define como:

El gradiente es una notación matriz de derivadas parciales de la función de costo con respecto a todos los parámetros (θ0 a θn).

Dado que la velocidad de aprendizaje es un escalar (α∈R), la regla de actualización del algoritmo de descenso de gradiente se expresa en la notación de la matriz:

Como consecuencia, El parámetro (θ) reside en el espacio dimensional (n+1).

Geográficamente, va cuesta abajo a un paso correspondiente a la tasa de aprendizaje hasta alcanzar la convergencia.

Cálculo

El objetivo de la regresión lineal es minimizar la brecha (MSE) entre los valores predichos y los valores reales dados en el conjunto de datos de capacitación.

Función de costo (función objetivo)

Esta brecha (MSE) se define como una brecha promedio de todos los ejemplos de entrenamiento:

dónde

• Jθ: función de costo (o función de pérdida),
• Hθ: Predicción del modelo,
• X: I_TH INTERTACIÓN DE INTURA,
• Y: I_th Valor objetivo, y
• M: El número de ejemplos de capacitación.

El gradiente se calcula tomando Derivación parcial de la función de costo con respecto a cada parámetro:

Porque tenemos parámetros n+1 (incluido un término de intercepción θ0) y m ejemplos de entrenamiento, formaremos un vector de gradiente usando notación de matriz:

En la notación de la matriz, donde x representa la matriz de diseño, incluido el término de intercepción y θ es el vector de parámetros, el gradiente ∇θ J (θ) viene dado por:

El Regla LMS (cuadrados menos medios) es un algoritmo iterativo que ajusta continuamente los parámetros del modelo en función del error entre sus predicciones y los valores objetivo reales de los ejemplos de entrenamiento.

Regla mínima de cuadrados mínimos (LMS)

En cada uno época de descenso de gradiente, cada parámetro θi se actualiza restando una fracción del error promedio en todos los ejemplos de entrenamiento:

Este proceso permite que el algoritmo encuentre iterativamente el Parámetros óptimos que minimizan la función de costo.

(Nota: θi es un parámetro asociado con la función de entrada XI, y el objetivo del algoritmo es encontrar su valor óptimo, no que ya sea un parámetro óptimo).

Ecuación normal

Para encontrar el parámetro óptimo (θ*) que minimiza la función de costo, podemos usar el ecuación normal.

Este método ofrece una solución analítica para la regresión lineal, lo que nos permite calcular directamente el valor θ que minimiza la función de costo.

A diferencia de las técnicas de optimización iterativa, la ecuación normal encuentra esto óptimo al resolver directamente el punto donde el gradiente es cero, asegurando la convergencia inmediata:

Por eso:

Esto se basa en la suposición de que la matriz de diseño X es invertiblelo que implica que todas sus características de entrada (de x_0 a x_n) son linealmente independiente.

Si X no es invertible, necesitaremos ajustar las funciones de entrada para garantizar su independencia mutua.

Simulación

En realidad, repetimos el proceso hasta que la convergencia establezca:

• Función de costos y su gradiente
• Tasa de aprendizaje
• Tolerancia (umbral de costo mín. Para detener la iteración)
• Número máximo de iteraciones
• Punto de partida

Lote por tasa de aprendizaje

El siguiente fragmento de codificación demuestra que el proceso de descenso de gradiente encuentra mínimos locales de una función de costo cuadrático por tasas de aprendizaje (0.1, 0.3, 0.8 y 0.9):

def cost_func(x):
    return x**2 - 4 * x + 1

def gradient(x):
    return 2*x - 4

def gradient_descent(gradient, start, learn_rate, max_iter, tol):
    x = start
    steps = [start] # records learning steps

    for _ in range(max_iter):
        diff = learn_rate * gradient(x)
        if np.abs(diff) < tol:
            break
        x = x - diff
        steps.append(x)

    return x, steps

x_values = np.linspace(-4, 11, 400)
y_values = cost_func(x_values)
initial_x = 9
iterations = 100
tolerance = 1e-6
learning_rates = [0.1, 0.3, 0.8, 0.9]

def gradient_descent_curve(ax, learning_rate):
    final_x, history = gradient_descent(gradient, initial_x, learning_rate, iterations, tolerance)

    ax.plot(x_values, y_values, label=f'Cost function: $J(x) = x^2 - 4x + 1$', lw=1, color='black')

    ax.scatter(history, [cost_func(x) for x in history], color='pink', zorder=5, label='Steps')
    ax.plot(history, [cost_func(x) for x in history], 'r--', lw=1, zorder=5)

    ax.annotate('Start', xy=(history[0], cost_func(history[0])), xytext=(history[0], cost_func(history[0]) + 10),
                arrowprops=dict(facecolor='black', shrink=0.05), ha='center')
    ax.annotate('End', xy=(final_x, cost_func(final_x)), xytext=(final_x, cost_func(final_x) + 10),
                arrowprops=dict(facecolor='black', shrink=0.05), ha='center')
    
    ax.set_title(f'Learning Rate: {learning_rate}')
    ax.set_xlabel('Input feature: x')
    ax.set_ylabel('Cost: J')
    ax.grid(True, alpha=0.5, ls='--', color='grey')
    ax.legend()

fig, axs = plt.subplots(1, 4, figsize=(30, 5))
fig.suptitle('Gradient Descent Steps by Learning Rate')

for ax, lr in zip(axs.flatten(), learning_rates):
    gradient_descent_curve(ax=ax, learning_rate=lr)

Predecir la transacción de la tarjeta de crédito

Usemos un conjunto de datos de muestra en Kaggle para predecir la transacción de la tarjeta de crédito usando regresión lineal con GD por lotes.

1. Preprocesamiento de datos

a) Base DataFrame

Primero, fusionaremos estos cuatro archivos del conjunto de datos de muestra utilizando IDS como clave, mientras desinfectan los datos sin procesar:

• transacción (CSV)
• Usuario (CSV)
• Tarjeta de crédito (CSV)
• Train_fraud_labels (JSON)

# load transaction data
trx_df = pd.read_csv(f'{dir}/transactions_data.csv')

# sanitize the dataset 
trx_df = trx_df[trx_df['errors'].isna()]
trx_df = trx_df.drop(columns=['merchant_city','merchant_state', 'date', 'mcc', 'errors'], axis='columns')
trx_df['amount'] = trx_df['amount'].apply(sanitize_df)

# merge the dataframe with fraud transaction flag.
with open(f'{dir}/train_fraud_labels.json', 'r') as fp:
    fraud_labels_json = json.load(fp=fp)

fraud_labels_dict = fraud_labels_json.get('target', {})
fraud_labels_series = pd.Series(fraud_labels_dict, name='is_fraud')
fraud_labels_series.index = fraud_labels_series.index.astype(int)

merged_df = pd.merge(trx_df, fraud_labels_series, left_on='id', right_index=True, how='left')
merged_df.fillna({'is_fraud': 'No'}, inplace=True)
merged_df['is_fraud'] = merged_df['is_fraud'].map({'Yes': 1, 'No': 0})
merged_df = merged_df.dropna()

# load card data
card_df = pd.read_csv(f'{dir}/cards_data.csv')
card_df = card_df.replace('nan', np.nan).dropna()
card_df = card_df[card_df['card_on_dark_web'] == 'No']
card_df = card_df.drop(columns=['acct_open_date', 'card_number', 'expires', 'cvv', 'card_on_dark_web'], axis='columns')
card_df['credit_limit'] = card_df['credit_limit'].apply(sanitize_df)

# load user data
user_df = pd.read_csv(f'{dir}/users_data.csv')
user_df = user_df.drop(columns=['birth_year', 'birth_month', 'address', 'latitude', 'longitude'], axis='columns')
user_df = user_df.replace('nan', np.nan).dropna()
user_df['per_capita_income'] = user_df['per_capita_income'].apply(sanitize_df)
user_df['yearly_income'] = user_df['yearly_income'].apply(sanitize_df)
user_df['total_debt'] = user_df['total_debt'].apply(sanitize_df)

# merge transaction and card data
merged_df = pd.merge(left=merged_df, right=card_df, left_on='card_id', right_on='id', how='inner')
merged_df = pd.merge(left=merged_df, right=user_df, left_on='client_id_x', right_on='id', how='inner')
merged_df = merged_df.drop(columns=['id_x', 'client_id_x', 'card_id', 'merchant_id', 'id_y', 'client_id_y', 'id'], axis='columns')
merged_df = merged_df.dropna()

# finalize the dataframe
categorical_cols = merged_df.select_dtypes(include=['object']).columns
df = merged_df.copy()
df = pd.get_dummies(df, columns=categorical_cols, dummy_na=False, dtype=float)
df = df.dropna()
print('Base data frame: \n', df.head(n=3))

b) preprocesamiento
Desde la base de datos base, elegiremos funciones de entrada adecuadas con:
valores continuos y una relación aparentemente lineal con la cantidad de transacción.

df = df[df['is_fraud'] == 0]
df = df[['amount', 'per_capita_income', 'yearly_income', 'credit_limit', 'credit_score', 'current_age']]

Entonces, filtraremos atípicos más allá 3 desviaciones estándar lejos de la media:

def filter_outliers(df, column, std_threshold) -> pd.DataFrame:
    mean = df[column].mean()
    std = df[column].std()
    upper_bound = mean + std_threshold * std
    lower_bound = mean - std_threshold * std
    filtered_df = df[(df[column] <= upper_bound) | (df[column] >= lower_bound)]
    return filtered_df

df = df.replace(to_replace='NaN', value=0)
df = filter_outliers(df=df, column='amount', std_threshold=3)
df = filter_outliers(df=df, column='per_capita_income', std_threshold=3)
df = filter_outliers(df=df, column='credit_limit', std_threshold=3)

Finalmente, tomaremos el logaritmo del valor objetivo amount Para mitigar la distribución sesgada:

df['amount'] = df['amount'] + 1
df['amount_log'] = np.log(df['amount'])
df = df.drop(columns=['amount'], axis='columns')
df = df.dropna()

*Se agregó uno a cantidad para evitar el infinito negativo en cantidad_log columna.

Final DataFrame:

c) Transformador
Ahora, podemos dividir y transformar el marco de datos final en conjuntos de datos de tren/prueba:

categorical_features = X.select_dtypes(include=['object']).columns.tolist()
categorical_transformer = Pipeline(steps=[('imputer', SimpleImputer(strategy='most_frequent')),('onehot', OneHotEncoder(handle_unknown='ignore'))])

numerical_features = X.select_dtypes(include=['int64', 'float64']).columns.tolist()
numerical_transformer = Pipeline(steps=[('imputer', SimpleImputer(strategy='mean')), ('scaler', StandardScaler())])

preprocessor = ColumnTransformer(
    transformers=[
        ('num', numerical_transformer, numerical_features),
        ('cat', categorical_transformer, categorical_features)
    ]
)


X_train_processed = preprocessor.fit_transform(X_train)
X_test_processed = preprocessor.transform(X_test)

2. Definición de lotes GD Regresor

class BatchGradientDescentLinearRegressor:
    def __init__(self, learning_rate=0.01, n_iterations=1000, l2_penalty=0.01, tol=1e-4, patience=10):
        self.learning_rate = learning_rate
        self.n_iterations = n_iterations
        self.l2_penalty = l2_penalty
        self.tol = tol
        self.patience = patience
        self.weights = None
        self.bias = None
        self.history = {'loss': [], 'grad_norm': [], 'weight':[], 'bias': [], 'val_loss': []}
        self.best_weights = None
        self.best_bias = None
        self.best_val_loss = float('inf')
        self.epochs_no_improve = 0

    def _mse_loss(self, y_true, y_pred, weights):
        m = len(y_true)
        loss = (1 / (2 * m)) * np.sum((y_pred - y_true)**2)
        l2_term = (self.l2_penalty / (2 * m)) * np.sum(weights**2)
        return loss + l2_term

    def fit(self, X_train, y_train, X_val=None, y_val=None):
        n_samples, n_features = X_train.shape
        self.weights = np.zeros(n_features)
        self.bias = 0

        for i in range(self.n_iterations):
            y_pred = np.dot(X_train, self.weights) + self.bias
        
            dw = (1 / n_samples) * np.dot(X_train.T, (y_pred - y_train)) + (self.l2_penalty / n_samples) * self.weights
            db = (1 / n_samples) * np.sum(y_pred - y_train)

            loss = self._mse_loss(y_train, y_pred, self.weights)
            gradient = np.concatenate([dw, [db]])
            grad_norm = np.linalg.norm(gradient)

            # update history
            self.history['weight'].append(self.weights[0])
            self.history['loss'].append(loss)
            self.history['grad_norm'].append(grad_norm)
            self.history['bias'].append(self.bias)

            # descent
            self.weights -= self.learning_rate * dw
            self.bias -= self.learning_rate * db

            if X_val is not None and y_val is not None:
                val_y_pred = np.dot(X_val, self.weights) + self.bias
                val_loss = self._mse_loss(y_val, val_y_pred, self.weights)
                self.history['val_loss'].append(val_loss)

                if val_loss < self.best_val_loss - self.tol:
                    self.best_val_loss = val_loss
                    self.best_weights = self.weights.copy()
                    self.best_bias = self.bias
                    self.epochs_no_improve = 0
                else:
                    self.epochs_no_improve += 1
                    if self.epochs_no_improve >= self.patience:
                        print(f"Early stopping at iteration {i+1} (validation loss did not improve for {self.patience} epochs)")
                        self.weights = self.best_weights
                        self.bias = self.best_bias
                        break

            if (i + 1) % 100 == 0:
                print(f"Iteration {i+1}/{self.n_iterations}, Loss: {loss:.4f}", end="")
                if X_val is not None:
                    print(f", Validation Loss: {val_loss:.4f}")
                else:
                    pass

    def predict(self, X_test):
        return np.dot(X_test, self.weights) + self.bias

3. Predicción y evaluación

model = BatchGradientDescentLinearRegressor(learning_rate=0.001, n_iterations=10000, l2_penalty=0, tol=1e-5, patience=5)
model.fit(X_train_processed, y_train.values)
y_pred = model.predict(X_test_processed)

Producción:
De las cinco características de entrada, per_capita_income mostró la correlación más alta con la cantidad de transacción:

(Izquierda: peso por características de entrada (abajo: más transacción), derecha: función de costo (aprendizaje_rate = 0.001, i = 10,000, m = 50,000, n = 5))

Error al cuadrado medio (MSE): 1.5752
R-cuadrado: 0.0206
Error absoluto medio (MAE): 1.0472

Complejidad del tiempo: Entrenamiento: O (N²M + N³) + Predicción: O (N)
Complejidad espacial: O (nm)
(M: Tamaño de ejemplo de entrenamiento, N: tamaño de la característica de entrada, suponiendo m >>> n)

Descenso de gradiente estocástico

Usos de GD por lotes todo el conjunto de datos de capacitación Para calcular el gradiente en cada paso de iteración (época), que es computacionalmente costoso, especialmente cuando tenemos millones de conjuntos de datos.

Descenso de gradiente estocástico (SGD) por otro lado,

1. Por lo general, baraja los datos de entrenamiento al comienzo de cada época,
2. seleccionar al azar a soltero ejemplo de entrenamiento en cada iteración,
3. calcula el gradiente usando el ejemplo y
4. actualiza los pesos y el sesgo del modelo Después de procesar cada ejemplo de capacitación individual.

Esto da como resultado muchas actualizaciones de peso por época (igual al número de muestras de entrenamiento), muchas actualizaciones rápidas y computacionalmente baratas basadas en puntos de datos individuales, Permitir que se itera a través de grandes conjuntos de datos mucho más rápido.

Simulación

Similar a Batch GD, definiremos la clase SGD y ejecutaremos la predicción:

class StochasticGradientDescentLinearRegressor:
    def __init__(self, learning_rate=0.01, n_iterations=100, l2_penalty=0.01, random_state=None):
        self.learning_rate = learning_rate
        self.n_iterations = n_iterations
        self.l2_penalty = l2_penalty
        self.random_state = random_state
        self._rng = np.random.default_rng(seed=random_state)
        self.weights_history = []
        self.bias_history = []
        self.loss_history = []
        self.weights = None
        self.bias = None

    def _mse_loss_single(self, y_true, y_pred):
        return 0.5 * (y_pred - y_true)**2

    def fit(self, X, y):
        n_samples, n_features = X.shape
        self.weights = self._rng.random(n_features)
        self.bias = 0.0

        for epoch in range(self.n_iterations):
            permutation = self._rng.permutation(n_samples)
            X_shuffled = X[permutation]
            y_shuffled = y[permutation]

            epoch_loss = 0
            for i in range(n_samples):
                xi = X_shuffled[i]
                yi = y_shuffled[i]

                y_pred = np.dot(xi, self.weights) + self.bias
                dw = xi * (y_pred - yi) + self.l2_penalty * self.weights
                db = y_pred - yi

                self.weights -= self.learning_rate * dw
                self.bias -= self.learning_rate * db
                epoch_loss += self._mse_loss_single(yi, y_pred)

                if n_features >= 2:
                    self.weights_history.append(self.weights[:2].copy())
                elif n_features == 1:
                    self.weights_history.append(np.array([self.weights[0], 0]))
                self.bias_history.append(self.bias)
                self.loss_history.append(self._mse_loss_single(yi, y_pred) + (self.l2_penalty / (2 * n_samples)) * (np.sum(self.weights**2) + self.bias**2)) # Approx L2

            print(f"Epoch {epoch+1}/{self.n_iterations}, Loss: {epoch_loss/n_samples:.4f}")

    def predict(self, X):
        return np.dot(X, self.weights) + self.bias

model = StochasticGradientDescentLinearRegressor(learning_rate=0.001, n_iterations=200, random_state=42)
model.fit(X=X_train_processed, y=y_train.values)
y_pred = model.predict(X_test_processed)

Producción:

Izquierda: peso por características de entrada, derecha: función de costo (aprendizaje_rate= 0.001, i = 200, m = 50,000, n = 5)

SGD introducido aleatoriedad en el proceso de optimización (Fig. Derecha).

Este “ruido” puede ayudar al algoritmo Saltar de mínimos locales o puntos de silla y potencialmente encontrar mejores regiones del espacio de parámetros.

Resultados:
Error cuadrado medio (MSE): 1.5808
R-cuadrado: 0.0172
Error absoluto medio (MAE): 1.0475

Complejidad del tiempo: Entrenamiento: O (N²M + N³) + Predicción: O (N)
Complejidad espacial: O (n)
(M: Tamaño de ejemplo de entrenamiento, N: tamaño de la característica de entrada, suponiendo m >>> n)