python - 我将如何在 Python 中生成带有事件的随机数据系列？

Question

我正在尝试生成用于异常检测的随机数据序列(或时间序列)，事件跨越几个连续的数据点。它们可以是高于/低于某个阈值的值，或者具有不同已知概率的异常类型。

例如在 1 为正常且事件类型在 [2, 3, 4] 内的情况下:11112221113333111111112211111

我查看了 np.random 和 random 方法，但找不到任何生成这些事件的方法。我当前的解决方案是选择随机点，向它们添加随机持续时间以生成事件开始和结束位置，用随机事件类型标记每个事件，然后加入回数据集，例如:

import numpy as np
num_events = np.random.randint(1, 10)
number_series = [1]*60
first_pos = 0 
event_starts = sorted([first_pos + i for i in np.random.randint(50, size = num_events)])
event_ends = [sum(i) for i in list(zip(event_starts, np.random.randint(8, size = num_events)))]
for c in list(zip(event_starts, event_ends)):
    rand_event_type  = np.random.choice(a = [2, 3, 4], p = [0.5, 0.3, 0.2])
    number_series[c[0]:c[1]] = [rand_event_type]*len(number_series[c[0]:c[1]])
print(number_series)

[1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 1, 1, 1, 1, 3, 3, 3, 3, 3, 3, 3, 1, 1, 1, 1, 3, 3, 4, 4, 4, 4, 4, 4, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 4, 4, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]

但我想知道是否有一种更简单的方法可以根据一组概率生成一系列带有事件的数字。

Answer 1

这完全取决于您如何为流程建模(您要模拟的基础流程)。您可以阅读更多关于一些常用型号的信息 on Wikipedia .

最简单

在下文中，我们使用一个非常简单的模型(与您的模型略有不同):每个事件都有一个概率(如您的问题)和一个独立于事件本身的随机持续时间。 1(“正常”)是一个与任何其他事件一样的事件(与您的示例代码不同)。我们可以改变它，但现在这是您能想到的最简单的模型之一。

def gen_events(n):
    events = np.random.choice(a=[1, 2, 3, 4], p=[0.6, 0.2, 0.12, 0.08], size=n)
    durations = np.random.randint(1, 8, size=n)
    return np.repeat(events, durations)

np.random.seed(0)  # repeatable example
number_series = gen_events(10)  # for example

>>> number_series
array([1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2, 2, 1, 1, 1, 1, 1, 1, 1,
       1, 2, 2, 1, 1, 1, 1, 1, 1, 3, 4, 4, 1, 1, 1, 1, 1])

请注意，这是非常快的:

%timeit gen_events(1_000_000)
# 44.9 ms ± 138 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

马尔可夫链

另一个模型(更容易参数化，实现起来有点复杂)是 Markov model .其中最简单的是马尔可夫链。这是一个 super 简单(但效率不高)的版本:

def markov_chain(P, n, initial_state=0):
    m = P.shape[0]
    ix = np.arange(m)
    s = np.empty(n, dtype=int)
    s[0] = initial_state
    for i in range(1, n):
        s[i] = np.random.choice(ix, p=P[s[i-1]])
    return s

上面，P是一个转移矩阵，其中每个单元格P[i,j]是从状态i转移到的概率状态 j。这是一个示例应用程序:

P = np.array([
    [.7, .1, .12, .08],  # from 0 to others
    [.3, .6, .05, .05],
    [.3, 0, .65, .05],
    [.4, 0, .05, .55],
])

np.random.seed(0)
n = 100
s = markov_chain(P, n) + 1
>>> s
array([1, 1, 2, 2, 2, 2, 2, 2, 2, 4, 1, 2, 2, 2, 3, 1, 1, 1, 3, 3, 3, 4,
       4, 4, 4, 1, 1, 1, 4, 4, 3, 1, 2, 2, 2, 1, 1, 1, 1, 4, 4, 1, 1, 1,
       1, 1, 1, 1, 1, 1, 1, 1, 1, 4, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
       1, 3, 1, 3, 1, 4, 4, 4, 4, 4, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
       1, 1, 4, 1, 1, 1, 2, 1, 1, 1, 1, 3])

请注意，每个事件的一元概率称为 pi，对应于 lim_{k ->\infty} P**k 的任何行:

>>> pd.Series(markov_chain(P, 1000, 0)).value_counts(normalize=True).sort_index()
0    0.530
1    0.135
2    0.209
3    0.126

>>> np.linalg.matrix_power(P, 40)[0]
array([0.52188552, 0.13047138, 0.21632997, 0.13131313])