Inhomogeneous Poisson Process

Application: software canary testing when all processes share a common multiplicative time-varying effect.

Consider points are observed from one of \(i \in \{1, 2\}\) Poisson point processes with intensity functions \(\lambda_i(t) = \rho_i \exp(\delta_i) \lambda(t)\), with \(\rho = [0.8, 0.2]\) and \(\delta = [1.5, 2]\). The probability that the next point comes from process \(i\) is

\[ \theta_i = \frac{\rho_i \exp(\delta_i)}{\sum_{j=1}^d \rho_j \exp(\delta_j)}. \]

Therefore, the next point comes from a random process, distributed as \(\mathrm{Multinomial}(1, \mathbf{\theta})\), with \(\mathbf{\theta} \approx [0.7, 0.3]\).

import numpy as np

rho = np.array([0.8, 0.2])
delta = np.array([1.5, 2])
theta = rho * np.exp(delta) / np.sum(rho * np.exp(delta))
size = 1000
np.random.seed(1)
xs = np.random.multinomial(1, theta, size=size)
print(xs)

[[1 0]
 [0 1]
 [1 0]
 ...
 [1 0]
 [1 0]
 [0 1]]

We can test the hypothesis

\[ \begin{align} H_0: \delta_1 - \delta_0 = 0 \quad (\mathbf{\theta} = \mathbf{\rho}) \\ H_1: \delta_1 - \delta_0 \neq 0 \quad (\mathbf{\theta} \neq \mathbf{\rho}) \end{align} \]

using a Multinomial test with \(\mathbf{\theta}_0 = \mathbf{\rho}\). To estimate a \((1 - \alpha)\) confidence sequence for \(\delta_1 - \delta_0\), we may set weights \([-1, 1]\):

from savvi.multinomial import InhomogeneousPoissonProcess

alpha = 0.05
weights = np.array([[-1, 1]])
ipp = InhomogeneousPoissonProcess(alpha, rho, weights)

AttributeError: _ARRAY_API not found

(CVXPY) Oct 24 05:34:42 PM: Encountered unexpected exception importing solver ECOS:
ImportError('numpy.core.multiarray failed to import')
(CVXPY) Oct 24 05:34:42 PM: Encountered unexpected exception importing solver SCS:
ImportError('numpy.core.multiarray failed to import')
(CVXPY) Oct 24 05:34:42 PM: Encountered unexpected exception importing solver ECOS_BB:
ImportError('numpy.core.multiarray failed to import')
(CVXPY) Oct 24 05:34:42 PM: Encountered unexpected exception importing solver OSQP:
ImportError('numpy.core.multiarray failed to import')

AttributeError: _ARRAY_API not found

AttributeError: _ARRAY_API not found

AttributeError: _ARRAY_API not found

For each new unit sample \(n\), we run the test. If \(p_n < \alpha\), we have the option to stop running:

sequence = ipp.batch(xs)
optional_stop = next(s for s in sequence if s.p_value <= alpha)
optional_stop

Parameter	Estimate	CI Lower	CI Upper
\(- \delta_0 + \delta_1\)	0.5162	0.0046	0.7549

Sample size: 210, P-value: 0.0456

%config InlineBackend.figure_formats = ["svg"]

import matplotlib.pyplot as plt
from savvi.utils import plot

contrasts = ipp.weights @ delta
_, ax1, _ = plot(sequence, contrasts)
ax1.set_ylim(-1, 2)
plt.show()