[ML] Ornstein Uhlenbeck Process

程式語言：Python

Package：

numpy

namedtuple

簡介：Ornstein Uhlenbeck Process 常用於 DDPG

$$ \begin{align*} X_{n+1} &= X_n + \theta (\mu - X_n)\Delta t + \sigma \Delta W_n \\ W_{t_{n+1}}-W_{t_n} &= \Delta W_n \sim N(0,\Delta t) = \sqrt{\Delta t} \space N(0,1) \\ \\ \mathbb{E}(X_t)&=X_0e^{-\theta t}+\mu(1-e^{-\theta t})\\ cov(X_s,X_t)&=\frac{\sigma^2}{2\theta}(e^{-\theta|t-s|}-e^{-\theta|t+s|})\\ \end{align*} $$

• 隨機且連續的過程，但長時間來看會徘徊於 \(\mu\)
• 參數意義
• \(\theta\)：拉回到 \(mu\) 的力道，值越大越會在 \(\mu\) 旁徘徊，也就是漂移越小
  
  
• \(\mu\)：中心平均值
  
  
• \(\Delta t\)：變化的時間間隔
  
  
• \(\sigma\)：隨機的強度，值越大隨機範圍越大
  
  
• DDPG 可用的參數，習慣範圍落在 [-1, 1] 之間，再乘上動作最大值
  
  

程式碼

import numpy as np
from collections import namedtuple


class OrnsteinUhlenbeckNoise:
    """
    A Ornstein Uhlenbeck action noise, this is designed to aproximate brownian motion with friction.

    Based on http://math.stackexchange.com/questions/1287634/implementing-ornstein-uhlenbeck-in-matlab

    :param dim: (tuple) the dimension of the noise
    :param mu: (float) the mean of the noise
    :param theta: (float) the rate of mean reversion, affect converge
    :param sigma: (float) the scale of the noise, affect random
    :param dt: (float) the timestep for the noise
    """

    def __init__(self, dim, mu=0, theta=0.15, sigma=0.2, dt=1.0):
        self.dim = dim
        self.mu = mu
        self.theta = theta
        self.sigma = sigma
        self.dt = dt

        self.X = np.ones(self.dim) * self.mu

    def reset(self):
        self.X = np.ones(self.dim) * self.mu

    def __call__(self):
        drift = self.theta * (self.mu - self.X) * self.dt
        random = self.sigma * self._delta_wiener()

        self.X = self.X + drift + random

        return self.X

    def _delta_wiener(self):
        return np.sqrt(self.dt) * np.random.randn(self.dim)


if __name__ == "__main__":
    Parameter = namedtuple("Parameter", ["theta", "mu", "dt", "sigma"])
    pars = [
        # theta
        Parameter(1, 0, 1, 1),
        Parameter(1e-2, 0, 1, 1),
        # mu
        Parameter(1, 0, 1, 1),
        Parameter(1, 10, 1, 1),
        # dt
        Parameter(1, 0, 1, 1),
        Parameter(1, 0, 1e-2, 1),
        # sigma
        Parameter(1, 0, 1, 1),
        Parameter(1, 0, 1, 1e-2),
        # DDPG
        Parameter(0.15, 0, 1, 0.2),
        Parameter(0.15, 0, 1e-2, 0.3),
    ]
    t = np.linspace(0, 10000, 10000)
    ys = []
    for par in pars:
        noise = OrnsteinUhlenbeckNoise(
            1, mu=par.mu, dt=par.dt, theta=par.theta, sigma=par.sigma
        )
        ys.append([noise() for _ in t])

    import matplotlib.pyplot as plt

    fig, ax = plt.subplots(nrows=len(ys), sharex=True)
    fig.suptitle(
        "$X_{n+1} = X_n + \\theta (\mu - X_n)\Delta t + \sigma \Delta W_n$", fontsize=16
    )
    for i, y in enumerate(ys):
        ax[i].set_title(
            f"$\\theta={pars[i].theta:.2f}, \mu={pars[i].mu:.2f}, \Delta t={pars[i].dt:.2f}, \sigma={pars[i].sigma:.2f}$"
        )
        ax[i].plot(t, y)

    fig.subplots_adjust(top=0.85, hspace=0.9)
    plt.show()

參考

Implementing Ornstein–Uhlenbeck in Matlab
Wiki Ornstein–Uhlenbeck process