Regresión lineal univariada en Python

Los datos univariados son el tipo de datos en los que el resultado depende solo de una variable. Por ejemplo, el conjunto de datos de puntos en una línea se puede considerar como un dato univariante donde la abscisa se puede considerar como una característica de entrada y la ordenada se puede considerar como una salida/resultado.

Por ejemplo:
Para la línea Y = 2X + 3 ;
La característica de entrada será X e Y será el resultado.

Concepto:
para la regresión lineal univariante, solo hay un vector de características de entrada. La línea de regresión tendrá la forma de:

Y = b0 + b1 * X
Donde,
b0 y b1 son los coeficientes de regresión.

Por tanto, se intenta predecir los coeficientes de regresión b0 y b1 entrenando un modelo.

Funciones de utilidad

1. Predecir

   def predict(x, b0, b1):     """Predicts the value of prediction based on         current value of regression coefficients when input is x"""     # Y = b0 + b1 * X     return b0 + b1 * x

2. Función de costo: la función
   de costo calcula el porcentaje de error con el valor actual de los coeficientes de regresión. Define cuantitativamente hasta qué punto el modelo se compara con los coeficientes de regresión reales que tienen la tasa de error más baja.

   def cost(x, y, b0, b1):     # y is a list of expected value     errors = []     for x, y in zip(x, y):         prediction = predict(x, b0, b1)         expected = y         difference = prediction-expected         errors.append(difference)     # Now, we have errors for all the observations,            # for some input, the value of error might be positive      # and for some input might be negative,      # and if we directly add them up,      # the values might cancel out leading to wrong output."            # Hence, we use concept of mean squared error.     # in mse, we return mean of square of all the errors.     mse = sum([e * e for e in errors])/len(errors)     return mse

3. Derivado de costos
   Después de cada iteración, el costo se actualiza en proporción al error. La naturaleza del error es muy sensible a los datos. Por sensible a los datos me refiero a que el valor del error cambia muy rápido porque teníamos una función de error cuadrada. Por lo tanto, para hacerlo más tolerante a valores altos de errores, derivamos la función de error.
   Las matemáticas son las siguientes:

   

   Código:

   def cost_derivative(x, y, b0, b1, i):     return sum([                   2*(predict(xi, b0, b1)-yi)*1                    if i == 0                    else 2*(predict(xi, b0, b1)-yi)*xi                    for xi, yi in zip(x, y)            ])/len(x)

4. Coeficientes de actualización:
   en cada iteración (época), los valores del coeficiente de regresión se actualizan en un valor específico frente al error de la iteración anterior. Esta actualización es muy importante y es el quid de las aplicaciones de aprendizaje automático que escribe.
   La actualización de los coeficientes con una actualización exacta de un coeficiente se realiza penalizando su valor con una fracción de error que causaron sus valores anteriores.
   Esta fracción se denomina tasa de aprendizaje. Esto define qué tan rápido nuestro modelo llega al punto de convergencia (punto donde el error es idealmente 0).

   

   La función de Python para la misma es la siguiente:

   def update_coeff(x, y, b0, b1, i, alpha):     bi -= alpha * cost_derivative(x, y, b0, b1, i)     return bi

5. Detener iteraciones:
   esta es la función que se utiliza para especificar cuándo deben detenerse las iteraciones.
   Según el usuario, el algoritmo stop_iteration generalmente devuelve verdadero en las siguientes condiciones:
   1. Iteración máxima: el modelo se entrena para un número específico de iteraciones.
   2. Valor del error: dependiendo del valor del error anterior, el algoritmo decide si continuar o detenerse.
   3. Precisión: dependiendo de la última precisión del modelo, si es mayor que la precisión mencionada, el algoritmo devuelve True,
   4. Híbrido: Esto se usa con más frecuencia. Esto combina más de una de las condiciones mencionadas anteriormente junto con una opción de descanso excepcional. El descanso excepcional es una condición en la que el entrenamiento continúa hasta que sucede algo malo. Algo malo podría incluir el desbordamiento de resultados, las limitaciones de tiempo excedidas, etc.

Teniendo todas las funciones de utilidad definidas, veamos el pseudocódigo seguido de su implementación:

Código:

x, y is the given data.
(b0, b1) <-- (0, 0)
i = 0
while True:
    if stop_iteration(i):
        break
    else:
        b0 = update_coeff(x, y, b0, b1, 0, alpha)
        b1 = update_coeff(x, y, b0, b1, 1, alpha)

Implementación final de Oop:

class LinearRegressor:
    def __init__(self, x, y, alpha = 0.01, b0 = 0, b1 = 0):
        """ 
            x: input feature
            y: result / target
            alpha: learning rate, default is 0.01
            b0, b1: linear regression coefficient.
        """
        self.i = 0
        self.x = x
        self.y = y
        self.alpha = alpha
        self.b0 = b0
        self.b1 = b1
        if len(x) != len(y):
            raise TypeError("x and y should have same number of rows.")
  
    def predict(model, x):
        """Predicts the value of prediction based on 
           current value of regression coefficients when input is x"""
        # Y = b0 + b1 * X
        return model.b0 + model.b1 * x
  
    def cost_derivative(model, i):
        x, y, b0, b1 = model.x, model.y, model.b0, model.b1
        predict = model.predict
        return sum([
            2 * (predict(xi) - yi) * 1
            if i == 0
            else (predict(xi) - yi) * xi
            for xi, yi in zip(x, y)
        ]) / len(x)
  
    def update_coeff(model, i):
        cost_derivative = model.cost_derivative
        if i == 0:
            model.b0 -= model.alpha * cost_derivative(i)
        elif i == 1:
            model.b1 -= model.alpha * cost_derivative(i)
  
    def stop_iteration(model, max_epochs = 1000):
        model.i += 1
        if model.i == max_epochs:
            return True
        else:
            return False
  
    def fit(model):
        update_coeff = model.update_coeff
        model.i = 0
        while True:
            if model.stop_iteration():
                break
            else:
                update_coeff(0)
                update_coeff(1)
  
  
if __name__ == '__main__':
    linearRegressor = LinearRegressor(
        x =[i for i in range(12)],
        y =[2 * i + 3 for i in range(12)],
        alpha = 0.03
    )
    linearRegressor.fit()
    print(linearRegressor.predict(12))
  
    # expects 2 * 12 + 3 = 27