我将如何在 Python 中生成带有事件的随机数据系列?
How would I generate a random data series in Python with events?
我正在尝试生成用于异常检测的随机数据序列(或时间序列),事件跨越几个连续的数据点。它们可以是值 above/below 某个阈值,或者具有不同已知概率的异常类型。
例如在 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]
但我想知道是否有更简单的方法来根据一组概率生成一系列带有事件的数字。
一种不那么冗长的方法是在旅途中生成事件列表。
例如,设置异常发生的概率(比如 5%)。那么,
events = []
for i in range(60):
if random() <= 0.95:
events.append(1)
else:
events.extend([choice(a = [2, 3, 4], p = [0.5, 0.3, 0.2])] * randint(8))
这完全取决于您如何为流程建模(您要模拟的基础流程)。您可以阅读更多关于一些常用模型的信息 on Wikipedia.
最简单
在下文中,我们使用一个非常简单的模型(与您的略有不同):每个事件都有一个概率(如您的问题)和一个独立于事件本身的随机持续时间。 1
("normal") 是一个与任何其他事件一样的事件(与您的示例代码不同)。我们可以改变它,但现在这是您能想到的最简单的模型之一。
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。其中最简单的是马尔可夫链。这是一个超级简单(但效率不高)的版本:
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])
我正在尝试生成用于异常检测的随机数据序列(或时间序列),事件跨越几个连续的数据点。它们可以是值 above/below 某个阈值,或者具有不同已知概率的异常类型。
例如在 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]
但我想知道是否有更简单的方法来根据一组概率生成一系列带有事件的数字。
一种不那么冗长的方法是在旅途中生成事件列表。
例如,设置异常发生的概率(比如 5%)。那么,
events = []
for i in range(60):
if random() <= 0.95:
events.append(1)
else:
events.extend([choice(a = [2, 3, 4], p = [0.5, 0.3, 0.2])] * randint(8))
这完全取决于您如何为流程建模(您要模拟的基础流程)。您可以阅读更多关于一些常用模型的信息 on Wikipedia.
最简单
在下文中,我们使用一个非常简单的模型(与您的略有不同):每个事件都有一个概率(如您的问题)和一个独立于事件本身的随机持续时间。 1
("normal") 是一个与任何其他事件一样的事件(与您的示例代码不同)。我们可以改变它,但现在这是您能想到的最简单的模型之一。
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。其中最简单的是马尔可夫链。这是一个超级简单(但效率不高)的版本:
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])