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
