forked from stumpy-dev/stumpy
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathchains.py
More file actions
102 lines (80 loc) · 2.37 KB
/
chains.py
File metadata and controls
102 lines (80 loc) · 2.37 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
# STUMPY
# Copyright 2019 TD Ameritrade. Released under the terms of the 3-Clause BSD license.
# STUMPY is a trademark of TD Ameritrade IP Company, Inc. All rights reserved.
from collections import deque
import numpy as np
def atsc(IL, IR, j):
"""
Compute the anchored time series chain (ATSC).
Parameters
----------
IL : ndarray
Left matrix profile indices
IR : ndarray
Right matrix profile indices
j : int
The index value for which to compute the ATSC
Returns
-------
output : ndarray
Anchored time series chain for index, `j`
Notes
-----
DOI: 10.1109/ICDM.2017.79
See Table I
This is the implementation for the anchored time series chains (ATSC).
Unlike the original paper, we've replaced the while-loop with a more stable
for-loop.
"""
C = deque([j])
for i in range(IL.size):
if IR[j] == -1 or IL[IR[j]] != j:
break
else:
j = IR[j]
C.append(j)
return np.array(list(C), dtype=np.int64)
def allc(IL, IR):
"""
Compute the all-chain set (ALLC)
Parameters
----------
IL : ndarray
Left matrix profile indices
IR : ndarray
Right matrix profile indices
Returns
-------
S : list(ndarray)
All-chain set
C : ndarray
Anchored time series chain for the longest chain
Notes
-----
DOI: 10.1109/ICDM.2017.79
See Table II
Unlike the original paper, we've replaced the while-loop with a more stable
for-loop.
This is the implementation for the all-chain set (ALLC) and the unanchored
chain is simply the longest one among the all-chain set. Both the
all-chain set and unanchored chain are returned.
The all-chain set, S, is returned as a list of unique numpy arrays.
"""
L = np.ones(IL.size, dtype=np.int64)
S = set()
for i in range(IL.size):
if L[i] == 1:
j = i
C = deque([j])
for k in range(IL.size):
if IR[j] == -1 or IL[IR[j]] != j:
break
else:
j = IR[j]
L[j] = -1
L[i] = L[i] + 1
C.append(j)
S.update([tuple(C)])
C = atsc(IL, IR, L.argmax())
S = [np.array(s, dtype=np.int64) for s in S]
return S, C