impact of a turning point (pivot)
def impact(s, peaks, troughs, down_threshold=0., up_threshold=0.):
'''Routine to calculate the directional impact of peaks and troughs,
given a sequence (pandas Series) and lists of indices of peaks and troughs.
Returns indices and magnitudes of impacts > threshold. Threshold may be set
independently for upward and downward impacts'''
impacts_p = []
impacts_t = []
impacts_p_idx = []
impacts_t_idx = []
for p in peaks:
#print ('p = ', p, s[p], s[p:].max())
if s[p] == s[p:].max() :
#print ('global max = ', s[p], min(s[p:]) / s[p])
impact_p = (1. - min(s[p:]) / s[p])
else :
next_index = p + next(x[0] for x in enumerate(s[p + 1:])if x[1] > s[p])
impact_p = (1. - min(s[p : next_index + 1]) / s[p])
# only include peak if the downward threshold is exceeded
if impact_p > down_threshold:
impacts_p.append(impact_p)
impacts_p_idx.append(p)
#print (p, impact_p)
for t in troughs:
#print ('t = ', t, s[t], s[t:].min())
if s[t] == s[t:].min() :
#print ('global min = ', s[t], max(s[t:]) / s[t])
impact_t = (max(s[t:]) / s[t] - 1.)
else :
next_index = t + next(x[0] for x in enumerate(s[t + 1:])if x[1] < s[t])
impact_t = (max(s[t : next_index + 1]) / s[t] - 1.)
# only include trough if the upward threshold is exceeded
if impact_t > up_threshold:
impacts_t.append(impact_t)
impacts_t_idx.append(t)
#print (t, impact_t)
return impacts_p_idx, impacts_p, impacts_t_idx, impacts_t
Definition 2: (Impact of a turning point) The upward impact of a trough $t$ is the ratio max{$x_t$,...,$x_n$} / $x_t$, where $n$ is the first index greater than $t$, such that $x_n < x_t$. That is, if the sequence increases after the trough $t$ to some maximal value, $xmax$, and then decreases below $x_t$, the impact is the ratio $xmax$ / $x_t$. If $x_t$ is the global minimum of the sequence, then the numerator is taken as the global maximum appearing after time $t$. The downward impact of a peak is defined conversely.