Polynomial regression

Polynomial regression is a simple 1D regression. The samples are of the form $\xi = (x,y) \in \mathbb{R}\times\mathbb{R}$ and the sought predictor is of the form $f(x) = \sum_{i=0}^d a_i x^i$ where $(a_0,..,a_d)$ are the $d+1$ coefficients to lean.

In the following example, we seek to learn a polynomial fitting the function

$f^\star(x) = \frac{10}{e^{x}+e^{-x}} + x$

from $n=100$ samples uniformly drawn from $[-2,2]$ and corrupted by a Gaussian noise with zero mean and variance $0.1$ .

from typing import Iterable
import numpy as np
import matplotlib.pyplot as plt
import torch as pt
import torch.nn as nn
from torch.utils.data import TensorDataset, DataLoader
from tqdm import tqdm

from skwdro.torch import robustify
from skwdro.solvers.oracle_torch import DualLoss

Problem setup

n = 100                                     # Number of observations

var = pt.tensor(0.1)                        # Variance of the noise

def f_star(x):                              # Generating function
    return 10/(pt.exp(x)+pt.exp(-x)) + x

xi = pt.rand(n)*4.0 - 2.0                   # x_i's are uniformly drawn from (-2,2]
xi = pt.sort(xi)[0]                         # we sort them for easier plotting
yi = f_star(xi) + pt.sqrt(var)*pt.randn(n)  # y_i's are f(x_i) + noise

dataset = DataLoader(TensorDataset(xi.unsqueeze(-1), yi.unsqueeze(-1)), batch_size=n//2, shuffle=True)

degree = 4                                  # Degree of the regression

device = "cuda" if pt.cuda.is_available() else "cpu"

Polynomial model

class PolynomialModel(nn.Module):
    def __init__(self, degree : int) -> None:
        super().__init__()
        self._degree = degree
        self.linear = nn.Linear(self._degree, 1)

    def forward(self, x):
        return self.linear(self._polynomial_features(x))

    def _polynomial_features(self, x):
        return pt.cat([x ** i for i in range(1, self._degree + 1)],dim=-1)

model = PolynomialModel(degree).to(device)  # Our polynomial regression model
loss  = nn.MSELoss(reduction='none')        # Our error will be measure in quadratic loss

Training loop

def train(dual_loss: DualLoss, dataset: Iterable[tuple[pt.Tensor, pt.Tensor]], epochs: int=10):

    lbfgs = pt.optim.LBFGS(dual_loss.parameters())   # LBFGS is used to optimize thanks to the nature of the problem

    def closure():          # Closure for the LBFGS solver
        lbfgs.zero_grad()
        loss = dual_loss(xi, xi_label, reset_sampler=True).mean()
        loss.backward()
        return loss

    pbar = tqdm(range(epochs))

    for _ in pbar:
        # Every now and then, try to rectify the dual parameter (e.g. once per epoch).
        dual_loss.get_initial_guess_at_dual(*next(iter(dataset))) # *

        # Main train loop
        inpbar = tqdm(dataset, leave=False)
        for xi, xi_label in inpbar:
            lbfgs.step(closure)

        pbar.set_postfix({"lambda": f"{dual_loss.lam.item():.2f}"})

    return dual_loss.primal_loss.transform

Training

radius = pt.tensor(0.001)   # Robustness radius

dual_loss = robustify(
            loss,
            model,
            radius,
            xi.unsqueeze(-1),
            yi.unsqueeze(-1)
        ) # Replaces the loss of the model by the dual WDRO loss

model1 = train(dual_loss, dataset, epochs=5) # type: ignore

model1.eval()  # type: ignore

  0%|          | 0/5 [00:00<?, ?it/s]

  0%|          | 0/2 [00:00<?, ?it/s]

 50%|█████     | 1/2 [00:00<00:00,  9.88it/s]

  0%|          | 0/5 [00:00<?, ?it/s, lambda=4469.09]
 20%|██        | 1/5 [00:00<00:00,  6.69it/s, lambda=4469.09]

  0%|          | 0/2 [00:00<?, ?it/s]

 20%|██        | 1/5 [00:00<00:00,  6.69it/s, lambda=292.16]

  0%|          | 0/2 [00:00<?, ?it/s]

 20%|██        | 1/5 [00:00<00:00,  6.69it/s, lambda=293.42]
 60%|██████    | 3/5 [00:00<00:00, 10.32it/s, lambda=293.42]

  0%|          | 0/2 [00:00<?, ?it/s]

100%|██████████| 2/2 [00:00<00:00, 13.94it/s]

 60%|██████    | 3/5 [00:00<00:00, 10.32it/s, lambda=286.42]

  0%|          | 0/2 [00:00<?, ?it/s]

100%|██████████| 2/2 [00:00<00:00, 16.96it/s]

 60%|██████    | 3/5 [00:00<00:00, 10.32it/s, lambda=293.42]
100%|██████████| 5/5 [00:00<00:00,  8.61it/s, lambda=293.42]
100%|██████████| 5/5 [00:00<00:00,  8.69it/s, lambda=293.42]

PolynomialModel(
  (linear): Linear(in_features=4, out_features=1, bias=True)
)

Results

We plot the obtained polynomial and print the coefficients

fig, ax = plt.subplots()
xtrial = pt.linspace(-2.1,2.1,100)
ax.scatter(xi.cpu(), yi.cpu(), c='g', label='train data')
ax.plot(xtrial, f_star(xtrial), 'k', label='generating function')

pred1 = model1(xtrial[:,None,None]).detach().cpu().squeeze() # type: ignore
ax.plot(xtrial,pred1, 'r', label='WDRO prediction')

fig.legend()
plt.show()


coeffs = model1.linear.weight.tolist()[0] # type: ignore
biais  = model1.linear.bias.tolist()[0] # type: ignore
polyString = "Polynomial regressor (degree {:d}, radius {:3.2e}):\n {:3.2f} ".format(degree,radius.float(),biais)
for i,a in enumerate(coeffs):
    if a>=0.0:
        polyString += "+ {:3.2f}x**{:d} ".format(a,i+1)
    else:
        polyString += "- {:3.2f}x**{:d} ".format(abs(a),i+1)

print(polyString)

Polynomial regressor (degree 4, radius 1.00e-03):
 4.76 + 1.03x**1 - 1.64x**2 - 0.00x**3 + 0.20x**4

Different degree and radius, with a more compact formulation.

radius2 = pt.tensor(1e-6)   # Robustness radius
degree2 = 7

model2 = train(robustify(
            nn.MSELoss(reduction='none'),
            PolynomialModel(degree2).to(device),
            radius2,
            xi.unsqueeze(-1),
            yi.unsqueeze(-1)
        ), dataset, epochs=5) # type: ignore

  0%|          | 0/5 [00:00<?, ?it/s]

  0%|          | 0/2 [00:00<?, ?it/s]

 50%|█████     | 1/2 [00:00<00:00,  9.36it/s]

100%|██████████| 2/2 [00:00<00:00,  8.66it/s]

  0%|          | 0/5 [00:00<?, ?it/s, lambda=4040500.25]
 20%|██        | 1/5 [00:00<00:00,  4.31it/s, lambda=4040500.25]

  0%|          | 0/2 [00:00<?, ?it/s]

 50%|█████     | 1/2 [00:00<00:00,  7.51it/s]

100%|██████████| 2/2 [00:00<00:00,  7.11it/s]

 20%|██        | 1/5 [00:00<00:00,  4.31it/s, lambda=343521.34]
 40%|████      | 2/5 [00:00<00:00,  3.82it/s, lambda=343521.34]

  0%|          | 0/2 [00:00<?, ?it/s]

 50%|█████     | 1/2 [00:00<00:00,  6.19it/s]

100%|██████████| 2/2 [00:00<00:00,  6.06it/s]

 40%|████      | 2/5 [00:00<00:00,  3.82it/s, lambda=295055.28]
 60%|██████    | 3/5 [00:00<00:00,  3.40it/s, lambda=295055.28]

  0%|          | 0/2 [00:00<?, ?it/s]

 50%|█████     | 1/2 [00:00<00:00,  5.94it/s]

100%|██████████| 2/2 [00:00<00:00,  5.98it/s]

 60%|██████    | 3/5 [00:01<00:00,  3.40it/s, lambda=296239.22]
 80%|████████  | 4/5 [00:01<00:00,  3.21it/s, lambda=296239.22]

  0%|          | 0/2 [00:00<?, ?it/s]

 50%|█████     | 1/2 [00:00<00:00,  5.99it/s]

100%|██████████| 2/2 [00:00<00:00,  6.01it/s]

 80%|████████  | 4/5 [00:01<00:00,  3.21it/s, lambda=295597.41]
100%|██████████| 5/5 [00:01<00:00,  3.12it/s, lambda=295597.41]
100%|██████████| 5/5 [00:01<00:00,  3.28it/s, lambda=295597.41]

fig, ax = plt.subplots()
xtrial = pt.linspace(-2.1,2.1,100)
ax.scatter(xi.cpu(), yi.cpu(), c='g', label='train data')
ax.plot(xtrial, f_star(xtrial), 'k', label='generating function')

pred2 = model2(xtrial[:,None,None]).detach().cpu().squeeze() # type: ignore
ax.plot(xtrial,pred2, 'r', label='WDRO prediction')

fig.legend()
plt.show()


coeffs = model2.linear.weight.tolist()[0] # type: ignore
biais  = model2.linear.bias.tolist()[0] # type: ignore
polyString = "Polynomial regressor (degree {:d}, radius {:3.2e}):\n {:3.2f} ".format(degree2,radius2.float(),biais)
for i,a in enumerate(coeffs):
    if a>=0.0:
        polyString += "+ {:3.2f}x**{:d} ".format(a,i+1)
    else:
        polyString += "- {:3.2f}x**{:d} ".format(abs(a),i+1)

print(polyString)

Polynomial regressor (degree 7, radius 1.00e-06):
 4.91 + 0.93x**1 - 2.31x**2 + 0.29x**3 + 0.71x**4 - 0.21x**5 - 0.10x**6 + 0.04x**7

Total running time of the script: (0 minutes 2.238 seconds)

Gallery generated by Sphinx-Gallery