<jupyter_start><jupyter_text>1.1 - Univariate Gaussianization<jupyter_code>!pip install "git+https://github.com/IPL-UV/rbig.git"
import sys, os
from pyprojroot import here

# spyder up to find the root
root = here(project_files=[".here"])

# append to path
sys.path.append(str(root))

%load_ext autoreload
%autoreload 2<jupyter_output><empty_output><jupyter_text>Import Packages<jupyter_code>import numpy as np

from sklearn.datasets import make_moons

# Packages
import numpy as np
import pytorch_lightning as pl


# Plot
import matplotlib.pyplot as plt
import corner

import wandb
pl.seed_everything(1234)

%load_ext autoreload
%autoreload 2<jupyter_output>Global seed set to 1234<jupyter_text>Logging<jupyter_code># TODO<jupyter_output><empty_output><jupyter_text>Data<jupyter_code>ntrain = 10_000
ntest = 10_000
noise = 0.1
random_state = 123
train_data = make_moons(n_samples=ntrain, noise=noise, random_state=random_state)[0]
test_data = make_moons(n_samples=ntest, noise=noise, random_state=random_state*10)[0]<jupyter_output><empty_output><jupyter_text>Plot Data<jupyter_code>fig = plt.figure(figsize=(7, 7))

corner.corner(train_data, color="blue", fig=fig)

plt.show()<jupyter_output><empty_output><jupyter_text>Flow Transformation  Univariate Gaussianization  Uniformization<jupyter_code>X_1 = train_data[:, 0][:, None]
X_2 = train_data[:, 1][:, None]

fig = corner.corner(X_1)
from rbig._src.uniform import MarginalHistogramUniformization
hist_uni_bijector = MarginalHistogramUniformization(X_1)
from typing import NamedTuple, Callable, Union
from scipy import stats


class UnivariateHistogram:
    def __init__(self, X: np.ndarray, bins: Union[int,str]=10, alpha: float=1e-10, bound_ext: float=0.1):
        
        diff = X.max() - X.min()
        lower_bound = X.min() - bound_ext * diff
        upper_bound = X.max() + bound_ext * diff

        # create histogram 
        hist = np.histogram(X, bins=bins, range=(lower_bound, upper_bound))

        # create histogram object
        estimator = stats.rv_histogram(hist)

        # add some regularization
        estimator._hpdf += alpha
        
        self.estimator = estimator
        
    def forward(self, X):
        
        return self.estimator.cdf(X)
    
    def inverse(self, X):
        
        return self.estimator.ppf(X)
    
    def gradient(self, X):
        
        return self.estimator.pdf(X)
hist_uni_bijector = UnivariateHistogram(X_1, bins="scott")<jupyter_output><empty_output><jupyter_text>Forward Transformation<jupyter_code>Xu_1 = hist_uni_bijector.forward(X_1)
fig = corner.corner(Xu_1, color="red")<jupyter_output><empty_output><jupyter_text>Inverse Transformation<jupyter_code>X_1_approx = hist_uni_bijector.inverse(Xu_1)
fig = corner.corner(X_1_approx, color="green")<jupyter_output><empty_output><jupyter_text>Det Jacobian<jupyter_code>Xldj_1 = hist_uni_bijector.gradient(X_1)
fig = corner.corner(Xldj_1)<jupyter_output><empty_output><jupyter_text>Density Estimation<jupyter_code>from scipy import stats

base_dist = stats.uniform()
# latent probability
pz = base_dist.pdf(Xu_1)

# det jacobian
det_jacobian = hist_uni_bijector.gradient(X_1)

# total prob
p_x = pz.sum(axis=1) * det_jacobian
fig, ax = plt.subplots()
ax.hist(p_x, bins=25)
plt.show()<jupyter_output><empty_output><jupyter_text>Sampling<jupyter_code># generate samples in uniform domain
pz_samples = base_dist.rvs(size=10_000)[:, None]

# propagate them through
X_1_samples = hist_uni_bijector.inverse(pz_samples)
fig, ax = plt.subplots()
ax.hist(X_1, bins=25, color='blue', density=True, label="True")
ax.hist(X_1_samples, bins=25, color='green', density=True, label="Generated")
plt.legend()
plt.tight_layout()
plt.show()<jupyter_output><empty_output><jupyter_text>Inverse CDF<jupyter_code>class InverseGaussCDF:
    def __init__(self, eps: float=1e-5):
        self.eps = eps
        
        # create histogram object
        self.estimator = stats.norm(loc=0, scale=1)
    
    def forward(self, X):
        
        X = np.clip(X, self.eps, 1-self.eps)
        
        Z = self.estimator.ppf(X)
        
        return Z
        
    def inverse(self, Z):
        X = self.estimator.cdf(Z)
        
        return X
    
    def gradient(self, X):
        X = self.inverse(X)
        
        det_jacobian = - self.estimator.pdf(X)
        
        return det_jacobian


def inverse_cdf(eps: float=1e-5):
    
    
    
    def forward(X):
        

        
        return X
    
    def inverse(X):
        
        
        
        
        return X
    
    def gradient(X):
        
        
        
        
        
        return X
    
    
    return Bijector(
        forward=forward, 
        inverse=inverse, 
        gradient=gradient
    )
from rbig._src.invcdf import InverseGaussCDF
invcdf_bijector = InverseGaussCDF(eps=1e-7)<jupyter_output><empty_output><jupyter_text>Forward<jupyter_code>Xg_1 = invcdf_bijector.forward(Xu_1)<jupyter_output><empty_output><jupyter_text>Inverse<jupyter_code>Xu_1_approx = invcdf_bijector.inverse(Xg_1)
fig = corner.corner(Xu_1_approx)
fig, ax = plt.subplots()
ax.hist(Xu_1, bins=25, color='blue', density=True, label="True")
ax.hist(Xu_1_approx, bins=25, color='green', density=True, label="Inverse")
plt.legend()
plt.tight_layout()
plt.show()<jupyter_output><empty_output><jupyter_text>Gradient<jupyter_code>Xg_1_gradient = invcdf_bijector.gradient(Xu_1)
fig = corner.corner(Xg_1_gradient)<jupyter_output><empty_output><jupyter_text>Density Estimation<jupyter_code># 1) uniformization
Xu_1 = hist_uni_bijector.forward(X_1)
det_jacobian_u = hist_uni_bijector.gradient(X_1)

# 2) gaussianization
Xg_1 = hist_uni_bijector.forward(Xu_1)
det_jacobian_g = hist_uni_bijector.gradient(Xu_1)

# latent probability
base_dist = stats.norm(loc=0, scale=1)

pz = base_dist.pdf(Xg_1).sum(axis=1)



# total prob
p_x = pz * det_jacobian_g * det_jacobian_u
fig = corner.corner(p_x)<jupyter_output><empty_output><jupyter_text>Sampling<jupyter_code># 1) samples from latent probability
base_dist = stats.norm(loc=0, scale=1)
pz_samples = base_dist.rvs(size=10_000)[:, None]

# 2) inverse gaussianization
Xu_1_samples = hist_uni_bijector.inverse(pz_samples)

# 3) inverse uniformization
X_1_samples = hist_uni_bijector.inverse(Xu_1_samples)
fig, ax = plt.subplots()
ax.hist(X_1, bins=25, color='blue', density=True, label="True")
ax.hist(X_1_samples, bins=25, color='green', density=True, label="Samples")
plt.legend()
plt.tight_layout()
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11492836234786627, 'fraction_numerical': 0.01956555230318903, 'mean_word_length': 4.378624689312344, 'pattern_counts': {'":': 0, '<': 68, '<?xml version=': 0, '>': 68, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '368951', 'license': 'MIT', 'n_tokens_mistral': 2609, 'n_tokens_neox': 2464, 'n_words': 398, 'path': 'research_notebook/content/tutorials/normalizing_flows/1.0_univariate_gauss.ipynb', 'repo_name': 'jejjohnson/research_notebook', 'size': '6491'}