Compare and Learn

graph = Digraph(comment="Toes", format="png")
graph.attr(rankdir="LR")
# input layer
graph.node("a", "Toes")

# output layer
graph.node("b", "Win")

# edge
graph.edge("a", "b", label="Weight")
graph.render("graphs/toes_model.dot")
graph

def toes_model(toes: int, weight: float=0.1) -> float:
    """Predicts if the team will win based on the number of toes

    Return:
     predction: probability of winning
    """
    return toes * weight

def print_error(predicted: float, actual: int, label: str="Toes Model",
                separator: str="\n") -> float:
    """Prints the (mean) squared error

    Args:

     predicted: what the model predicted
     actual: whether the team won or not
     label: something to identify the model
     separator: How to separate the output
    Returns:
     mse: the error
    """
    error = (actual - predicted)**2
    print(separator.join([label,
                          "Predicted: {:.2f}".format(predicted),
                          "Actual: {}".format(actual),
                          "MSE: {:.4f}".format(error)]))
    return error

toes = 8.5
actual = 1
predicted = toes_model(toes)
error_original = print_error(predicted, actual)

weight_change = 0.01
weight = 0.1
knob_turned_up = toes_model(toes, weight + weight_change)
knob_turned_down = toes_model(toes, weight - weight_change)

error_up = print_error(knob_turned_up, actual, "Turned Up")
print()
error_down = print_error(knob_turned_down, actual, "Turned Down")

if error_original > error_up or error_original > error_down:
    change_direction = -1 if error_down < error_original else 1
    weight_updated = weight + change_direction * weight_change
    prediction = toes_model(toes, weight_updated)
    error_update = print_error(prediction, actual, "Updated Model")
    assert error_update < error_original
else:
    print("Model didn't improve.")

weight = 0.5
input_value = 0.5
actual = 0.8
weight_change = 0.001

prediction = toes_model(input_value, weight)
print_error(prediction, actual, "Step    1 Weight: {}".format(weight), "\t")
errors = []
weights = []
optimal_step = False
tolerance = 0.1**8
for step in range(1, 2001):
    prediction = toes_model(input_value, weight)
    error = (print_error(prediction, actual,
                         "Step {:4} Weight: {:.2f}".format(step, weight), "\t")
             if not step % 100 else (prediction - actual)**2)
    errors.append(error)
    if not optimal_step and error < tolerance:
        optimal_step = step - 1
    up_prediction = toes_model(input_value, weight + weight_change)
    down_prediction = toes_model(input_value, weight - weight_change)
    up_error = (up_prediction - actual)**2
    down_error = (down_prediction - actual)**2
    direction = -1 if down_error < up_error else 1
    weight += direction * weight_change
    weights.append(weight)

print("Optimum Reached at Step {}".format(optimal_step))

figure, axe = pyplot.subplots(figsize=FIGURE_SIZE)
axe.set_title("Hot and Cold Mean Squared Error")
axe.set_xlabel("Training Repetition")
axe.set_ylabel("MSE")
lines = axe.plot(range(len(errors)), errors)

figure, axe = pyplot.subplots(figsize=FIGURE_SIZE)
axe.set_title("Hot and Cold Mean Squared Error vs Weights")
axe.set_xlabel("Weight")
axe.set_ylabel("MSE")
lines = axe.plot(weights, errors, ".")

print(prediction/input_value)

weight = 0.5
input_value = 0.5
actual = 0.8

prediction = toes_model(input_value, weight)
print_error(prediction, actual, "Step    1 Weight: {}".format(weight), "\t")
errors = []
weights = []
optimal_step = False
tolerance = 0.1**8
optimal_count = 0
print_every = 5
stop_after = 30
for step in range(1, 2001):
    prediction = toes_model(input_value, weight)
    difference = (prediction - actual) * input_value
    weight -= difference
    weights.append(weight)    
    error = (print_error(prediction, actual,
                         "Step {:4} Weight: {:.2f}".format(step, weight), "\t")
             if not step % print_every else (prediction - actual)**2)
    errors.append(error)
    if not optimal_step and error < tolerance:
        optimal_step = step - 1
    if error < tolerance:
        optimal_count += 1
    if optimal_count >= stop_after:
        break
print("Optimum Reached at Step {}".format(optimal_step))

figure, axe = pyplot.subplots(figsize=FIGURE_SIZE)
axe.set_title("Gradient Descent Mean Squared Error")
axe.set_xlabel("Training Repetition")
axe.set_ylabel("MSE")
lines = axe.plot(range(len(errors)), errors)

figure, axe = pyplot.subplots(figsize=FIGURE_SIZE)
axe.set_title("Gradient Descent Mean Squared Error vs Weights")
axe.set_xlabel("Weight")
axe.set_ylabel("MSE")
lines = axe.plot(weights, errors, "o")

class OneNode:
    """Implements a single-node network

    Args:
     weight: the starting weight for the edge from the input to the output
     input_value: the input to the node
     actual: the actual output we are trying to predict

     training_steps: how many times to train the model
     tolerance: how close to zero we need our error to be
     print_every: how often to print training status
     stop_after: how many times to keep going after the optimal was found
    """
    def __init__(self, weight: float,
                 input_value: float,
                 actual: float,
                 training_steps: int=200,
                 tolerance: float=0.1**8,
                 print_every: int=5, stop_after: int=30) -> None:
        self.original_weight = weight
        self.weight = weight
        self.input_value = input_value
        self.actual = actual
        self.training_steps = training_steps
        self.tolerance = tolerance
        self.print_every = print_every
        self.stop_after = stop_after
        self._errors = None
        self._weights = None
        self._predictions = None
        return

    @property
    def errors(self) -> list:
        """list of MSE values"""
        if self._errors is None:
            self._errors = []
        return self._errors

    @property
    def weights(self) -> list:
        """List of weights built during training"""
        if self._weights is None:
            self._weights = [self.weight]
        return self._weights

    @property
    def predictions(self) -> list:
        """List of predictions made"""
        if self._predictions is None:
            self._predictions = []
        return self._predictions

    @property
    def prediction(self) -> float:
        """The current model's prediction"""
        return self.weight * self.input_value

    def mean_squared_error(self) -> float:
        """The mean squared error for the prediction"""
        try:
            self.predictions.append(self.prediction)
            return (self.prediction - self.actual)**2
        except OverflowError as error:
            print(error)
            print("prediction: {}".format(self.prediction))
            print("actual: {}".format(self.actual))
        return

    def print_error(self,
                    step: int,
                    separator: str="\t",
                    force_print: bool=False,
                    store_error: bool=True) -> float:
        """Prints the (mean) squared error

       Args:
        step: what step this is
        separator: How to separate the output
        force_print: ignore the step count and print anyway
        store_error: whether to add to the errors
       Returns:
        mse: the error
       """
        error = self.mean_squared_error()
        if store_error:
            self.errors.append(error)
        if force_print or not step % self.print_every:
            print("Input: {} Actual Output: {}".format(self.input_value,
                                                       self.actual))
            print(separator.join(["Step: {}".format(step),
                                  "Weight: {}".format(self.weight),
                                  "Predicted: {:.2f}".format(self.prediction),
                                  "MSE: {:.4f}".format(error)]))
        return error

    def adjust_weight(self) -> None:
        """Takes the gradient descent step"""
        scaled_derivative = (self.prediction - self.actual) * self.input_value
        self.weight -= scaled_derivative
        self.weights.append(self.weight)
        return

    def train(self):
        """Trains our model on the values

       """
        self.reset()
        prediction = self.weight * self.input_value
        optimal_step = optimal_count = 0
        error = self.print_error(1,
                                 force_print=True,
                                 store_error=False)
        for step in range(1, self.training_steps + 1):
            error = self.print_error(step)
            if error is None:
                return

            self.adjust_weight()
            if not optimal_step and error < self.tolerance:
                optimal_step = step - 1
            if error < tolerance:
                optimal_count += 1
            if optimal_count >= self.stop_after:
                break
        print("Optimum Reached at Step {}".format(optimal_step))
        return

    def plot_errors_over_time(self, scale="linear") -> None:
        """Plots the error as it is trained

       Args:
        scale: y-axis scale
       """
        figure, axe = pyplot.subplots(figsize=FIGURE_SIZE)
        axe.set_title("Gradient Descent Mean Squared Error")
        axe.set_xlabel("Training Repetition")
        axe.set_ylabel("MSE")
        axe.set_yscale(scale)
        lines = axe.plot(range(len(self.errors)), self.errors)
        return

    def plot_errors_vs_weights(self, style="o") -> None:
        """Plots errors given weights"""
        figure, axe = pyplot.subplots(figsize=FIGURE_SIZE)
        axe.set_title("Gradient Descent Mean Squared Error vs Weights")
        axe.set_xlabel("Weight")
        axe.set_ylabel("MSE")
        lines = axe.plot(self.weights, self.errors, style)
        return

    def plot_weight_distribution(self) -> None:
        """Plots the distribution of the weights"""
        figure, axe = pyplot.subplots(figsize=FIGURE_SIZE)
        axe.set_title("Distribution Of Weights")
        lines = seaborn.distplot(self.weights)
        return


    def reset(self):
        """Resets the properties"""
        self._errors = None
        self._weights = None
        self._predictions = None
        self.weight = self.original_weight
        return

class AlphaNode(OneNode):
    """Incorporates weight update reduction

    Args:
     alpha: amount to weight the update

     weight: the starting weight for the edge from the input to the output
     input_value: the input to the node
     actual: the actual output we are trying to predict

     training_steps: how many times to train the model
     tolerance: how close to zero we need our error to be
     print_every: how often to print training status
     stop_after: how many times to keep going after the optimal was found
    """
    def __init__(self, alpha: float, *args, **kwargs) -> None:
        self.alpha = alpha
        __class__
        super().__init__(*args, **kwargs)
        return

    def adjust_weight(self) -> None:
        """Takes the gradient descent step with alpha weight"""
        scaled_derivative = (self.prediction - self.actual) * self.input_value
        self.weight -= self.alpha * scaled_derivative
        self.weights.append(self.weight)
        return

alpha_network = AlphaNode(alpha=0.1, weight=0.5, input_value=5,
                          actual=0.8, print_every=30)
alpha_network.train()

alpha_network.alpha = 0.01
alpha_network.train()