# Peak valley detection in python

Below is a peak/valley detection algorithm I've designed. The basic premises for this function is that we split our data into chunks, do linear regressions on these chunks, detrend the data based on the line of best fit, points above a certain standard deviation are noted. Chunks are determined with a specified amount of overlap. For example if we set the overlap to be 0.5, chunk one will be indexes 0-100, chunk 2 will be 50-150, and so on.

Because of the ability for overlapping chunks, the same index may be noted in multiple chunks. After this we only look at the indexes that have a counts past the threshold. For example, a threshold of 1 will mean that all indexes gathered are valid. For a count of 2, an index needs to be noted in two separate chunks for it to be valid, etc.

Lastly, we iterate through our data and index, and only carry over the most extreme point from a group of similar consecutive events. For example if there are 3 peaks in a row before a valley is noted. Only the highest peak from this set of 3 will be carried through.

Note: A safe thing to do with this data is to clip the start of the data by window_size, and the end of the data by window_size * 2. This is to ensure the integrity of the data, i.e. that peaks and valleys exist that weren't detected due to lack of context of availability in multiple chunk overlaps.

Now that I've explained how this algorithm works, I am looking to hear some suggestions on optimization. I want this to run faster. Any ideas?

def get_peak_valley(threshold, window_size, overlap, req_angles):
# validate params
window_size = int(round(window_size))
req_angles = int(round(req_angles))
window_step = int(round(window_size * (1 - overlap)))
if window_step == 0:
window_step = 1
if req_angles == 0:
req_angles = 1

# get all points that classify as a peak/valley
ind = 0
peak_inds, valley_inds = [], []

while ind + window_size <= len(close_prices):
flattened = detrend(close_prices[ind:ind + window_size])
std, avg = np.std(flattened), np.mean(flattened)
lower_b = avg - std * threshold
upper_b = avg + std * threshold
for idx, val in enumerate(flattened):
if val < lower_b:
valley_inds.append(idx + ind)
elif val > upper_b:
peak_inds.append(idx + ind)
ind += window_step

# discard points that have counts below the threshold
peak_counts = Counter(peak_inds)
pk_inds = [c for c in peak_counts.keys() if peak_counts[c] >= req_angles]

valley_counts = Counter(valley_inds)
vly_inds = [c for c in valley_counts.keys() if valley_counts[c] >= req_angles]

# initialize iterator to find to best peak/valley for consecutive detections
if len(pk_inds) == 0 or len(vly_inds) == 0:
return pk_inds, vly_inds

if pk_inds[0] < vly_inds[0]:
curr_event = 'peak'
best_price = close_prices[pk_inds[0]]
else:
curr_event = 'valley'
best_price = close_prices[vly_inds[0]]

#iterate through points and only carry forward the index that has the highest or lowest value from the current group
best_ind = 0
new_vly_inds, new_pk_inds = [], []

for x in range(len(close_prices)):
if x in pk_inds and curr_event == 'valley':
new_vly_inds.append(best_ind)
curr_event = 'peak'
best_price = close_prices[x]
best_ind = x
continue
if x in vly_inds and curr_event == 'peak':
new_pk_inds.append(best_ind)
curr_event = 'valley'
best_price = close_prices[x]
best_ind = x
continue

if x in pk_inds and curr_event == 'peak' and close_prices[x] > best_price:
best_price = close_prices[x]
best_ind = x
elif x in vly_inds and curr_event == 'valley' and close_prices[x] < best_price:
best_price = close_prices[x]
best_ind = x

# deal with the final group of events
if curr_event == 'valley':
new_vly_inds.append(best_ind)
if curr_event == 'peak':
new_pk_inds.append(best_ind)

return new_pk_inds, new_vly_inds

• Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers. Commented Oct 7, 2019 at 14:18
• I didn't update my code based on any feedback here. Commented Oct 7, 2019 at 14:39
• Why you update the code really doesn't matter. As soon as answers start coming in, you no longer touch the code.
– Mast
Commented Oct 7, 2019 at 14:46
• Gotcha, didn't realize. Will keep that in mind for the future Commented Oct 7, 2019 at 14:49

Putting aside the matter of optimization for now: I have concerns about the numerical approach here. If this is written with a specific application in mind, and you can make certain assumptions about your data, you might be fine; but if this is to be applied generally there are certain cases that are going to give you a lot of trouble.

I suspect that the data being processed are financial in nature, due to your close_prices variable, in which case the following degenerate case is entirely possible.

What if your data are periodic, and the fundamental spectral component is close to the reciprocal of your window size? In the simplest case, imagine a cosine whose peaks align with the borders of your window. Noise at the borders will generate false peak and valley positives. For this and other reasons, having a fixed window size doesn't lend itself well to accurate analysis.

Numerical statistics is a deep and very complex topic. You'll need to do some reading on this, and will likely find that it's more appropriate for you to apply a library with local minima/maxima search than to roll your own. Read for example this SO answer:

https://stackoverflow.com/a/22640362/313768

and some of the many links it includes.

• This is solved by the use of overlap with the windows no? Commented Oct 7, 2019 at 7:51
• Also in response to the post you linked, I wanted to design an algo that looks at both sides of a peak. For example with moving average you're only concerned about data in the past. What if we have a situation in which the signal abruptly enters a very noisy period. With moving average the beginning of this period would be incorrectly identified as a peak. Commented Oct 7, 2019 at 8:05
• And by letting user set both overlap and required number of angles, I believe this can let some decide if they want a situation like when the signal jumps and plateaus for a while to be considered a peak or not Commented Oct 7, 2019 at 8:08