I'm trying to get a point which is higher in a range of points, i.e., pivot high, then among a range of pivot high I want to find a significant pivot high. For this I am trying to create a range which is not pre-defined but calculated on every go. It is being calculated by knee plot to identify the best parameters which gives the points above the range and points below the range.
This works fine for a lot of data. If the loop is not able to find the optimal parameters, I'm manually assigning the optimal high and optimal low data. Also there is a range where we can check for the parameter values, and the lower parameter, has a condition that it cannot exceed a certain value.
This is enough of the background and to make sure the code is understood well.
Now I want to include a functionality that plots trend-lines to the plot containing the significant pivot high, significant pivot low and closing prices. The characteristic of the trend line should be such that, I am able to connect significant pivot lows with upward trendline on a price chart. The more significant pivot lows the line touches, the stronger is the trendline. Similar will be the case for downward trendline and the significant pivot low points.
What my code plots currently is something like:
The dotted red lines and the dotted green lines represent the current lines being plotted respectively. The black and blue connecting lines is something that I desire from my code.
I think, I am not able to think of the logic correctly and once, that clears out I can write the algorithm clearly.
Code:
import os
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy.signal import argrelextrema
def calculate_pivot_points(data):
pivot_points = []
resistance_levels = []
support_levels = []
pivot_high_points = []
pivot_low_points = []
for i in range(len(data)):
high = data.loc[i, 'high']
low = data.loc[i, 'low']
close = data.loc[i, 'close']
# Calculate Pivot Point
pivot_point = (high + low + close) / 3
pivot_points.append(pivot_point)
# Calculate Resistance Levels
resistance1 = (2 * pivot_point) - low
resistance2 = pivot_point + (high - low)
resistance3 = high + 2 * (pivot_point - low)
resistance_levels.append({'R1': resistance1, 'R2': resistance2, 'R3': resistance3})
# Calculate Support Levels
support1 = (2 * pivot_point) - high
support2 = pivot_point - (high - low)
support3 = low - 2 * (high - pivot_point)
support_levels.append({'S1': support1, 'S2': support2, 'S3': support3})
# Identify Pivot High Points using swing points
if i > 0 and i < len(data) - 1:
if high > data.loc[i-1, 'high'] and high > data.loc[i+1, 'high']:
pivot_high_points.append({'index': i, 'value': high})
# Identify Pivot Low Points using swing points
if i > 0 and i < len(data) - 1:
if low < data.loc[i-1, 'low'] and low < data.loc[i+1, 'low']:
pivot_low_points.append({'index': i, 'value': low})
return pivot_points, resistance_levels, support_levels, pivot_high_points, pivot_low_points
# Create a list to store all the data frames
data_frames = []
# Specify the folder path containing the CSV files
folder_path = "./data_frames"
# Iterate over each file in the folder
for filename in os.listdir(folder_path):
if filename.endswith(".csv"):
file_path = os.path.join(folder_path, filename)
# Read the data from the CSV file
data = pd.read_csv(file_path)
# Add the data frame to the list
data_frames.append(data)
# Extract the file name without the extension
file_name = os.path.splitext(filename)[0]
# Calculate pivot points and other parameters
pivot_points, resistance_levels, support_levels, pivot_high_points, pivot_low_points = calculate_pivot_points(data)
# Extract closing prices
closing_prices = data['close']
# Define the range of parameter values to test
parameter_range = range(1, 40)
# Calculate scores for different parameter combinations
parameter_scores = []
for high_parameter in parameter_range:
for low_parameter in parameter_range:
if low_parameter <= 8: # Add the condition here
# Determine significant pivot high points using swing points
significant_high_points = []
for point in pivot_high_points:
if point['index'] > 0 and point['index'] < len(data) - 1:
high_range = data.loc[point['index'] - high_parameter: point['index'] + low_parameter, 'high']
if point['value'] == high_range.max():
significant_high_points.append(point)
# Determine significant pivot low points using swing points
significant_low_points = []
for point in pivot_low_points:
if point['index'] > 0 and point['index'] < len(data) - 1:
low_range = data.loc[point['index'] - high_parameter: point['index'] + low_parameter, 'low']
if point['value'] == low_range.min():
significant_low_points.append(point)
# Calculate the score as the difference between high and low point counts
score = len(significant_high_points) - len(significant_low_points)
parameter_scores.append((high_parameter, low_parameter, score))
# Convert the scores to a NumPy array for easier manipulation
scores = np.array(parameter_scores)
# Find the optimal parameter values using the knee point
if len(scores) > 0:
knee_index = argrelextrema(scores[:, 2], np.less)[0][-1]
optimal_high_parameter, optimal_low_parameter, optimal_score = scores[knee_index]
else:
optimal_high_parameter = 16 # Manually assign the value
optimal_low_parameter = 2 # Manually assign the value
print("Optimal high parameter value:", optimal_high_parameter)
print("Optimal low parameter value:", optimal_low_parameter)
# Plot line chart for closing prices
plt.plot(closing_prices, label='Closing Prices')
# Calculate the trendlines for connecting the pivot high points
trendlines_high = []
trendline_points_high = []
for i in range(0, len(significant_high_points) - 1):
point1 = significant_high_points[i]
point2 = significant_high_points[i+1]
slope = (point2['value'] - point1['value']) / (point2['index'] - point1['index'])
if slope > 0:
if not trendline_points_high:
trendline_points_high.append(point1)
trendline_points_high.append(point2)
else:
if len(trendline_points_high) > 1:
trendlines_high.append(trendline_points_high)
trendline_points_high = []
if len(trendline_points_high) > 1:
trendlines_high.append(trendline_points_high)
# Calculate the trendlines for connecting the pivot low points
trendlines_low = []
trendline_points_low = []
for i in range(0, len(significant_low_points) - 1):
point1 = significant_low_points[i]
point2 = significant_low_points[i+1]
slope = (point2['value'] - point1['value']) / (point2['index'] - point1['index'])
if slope < 0:
if not trendline_points_low:
trendline_points_low.append(point1)
trendline_points_low.append(point2)
else:
if len(trendline_points_low) > 1:
trendlines_low.append(trendline_points_low)
trendline_points_low = []
if len(trendline_points_low) > 1:
trendlines_low.append(trendline_points_low)
# Plot the trendlines for positive slope
for trendline_points_high in trendlines_high:
x_values = [point['index'] for point in trendline_points_high]
y_values = [point['value'] for point in trendline_points_high]
plt.plot(x_values, y_values, color='red', linestyle='dashed')
# Plot the significant pivot high points
x_values = [point['index'] for point in significant_high_points]
y_values = [point['value'] for point in significant_high_points]
plt.scatter(x_values, y_values, color='red', label='Significant Pivot High Points')
# Plot the trendlines for positive slope
for trendline_points_low in trendlines_low:
x_values = [point['index'] for point in trendline_points_low]
y_values = [point['value'] for point in trendline_points_low]
plt.plot(x_values, y_values, color='green', linestyle='dashed')
# Plot the significant pivot low points
x_values = [point['index'] for point in significant_low_points]
y_values = [point['value'] for point in significant_low_points]
plt.scatter(x_values, y_values, color='green', label='Significant Pivot Low Points')
# Set chart title and labels
plt.title(f'Closing Prices with Trendlines and Significant Pivot Points ({file_name})')
plt.xlabel('Index')
plt.ylabel('Closing Price')
# Show the chart for the current data frame
plt.legend()
plt.show()
The data can be found at this drive link if you wish to attempt the code yourself: Link
PS: In the current code, I'm just checking if the two points line on the same straight trendline. This is not going to be the case in a lot of time. So instead what I am thinking is we define a range and if firstly the slope between n and n+1th point is > or < 0 then we proceed to the next two points, i.e., n+1 and n+2th point. Here if the difference between the two slopes, i.e., slope between n and n+1th and n+1th and n+2th is within certain range, then we can shift the main slope variable to slope between n and n+2 and similarly run the loop. This will be a great start, but now I'm stuck with the coding part. If someone can help me code this out, that will be very helpful.
import os
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy.signal import argrelextrema
def calculate_pivot_points(data):
pivot_points = []
resistance_levels = []
support_levels = []
pivot_high_points = []
pivot_low_points = []
for i in range(len(data)):
high = data.loc[i, 'high']
low = data.loc[i, 'low']
close = data.loc[i, 'close']
# Calculate Pivot Point
pivot_point = (high + low + close) / 3
pivot_points.append(pivot_point)
# Calculate Resistance Levels
resistance1 = (2 * pivot_point) - low
resistance2 = pivot_point + (high - low)
resistance3 = high + 2 * (pivot_point - low)
resistance_levels.append({'R1': resistance1, 'R2': resistance2, 'R3': resistance3})
# Calculate Support Levels
support1 = (2 * pivot_point) - high
support2 = pivot_point - (high - low)
support3 = low - 2 * (high - pivot_point)
support_levels.append({'S1': support1, 'S2': support2, 'S3': support3})
# Identify Pivot High Points using swing points
if i > 0 and i < len(data) - 1:
if high > data.loc[i-1, 'high'] and high > data.loc[i+1, 'high']:
pivot_high_points.append({'index': i, 'value': high})
# Identify Pivot Low Points using swing points
if i > 0 and i < len(data) - 1:
if low < data.loc[i-1, 'low'] and low < data.loc[i+1, 'low']:
pivot_low_points.append({'index': i, 'value': low})
return pivot_points, resistance_levels, support_levels, pivot_high_points, pivot_low_points
# Create a list to store all the data frames
data_frames = []
# Specify the folder path containing the CSV files
folder_path = "./data_frames"
# Iterate over each file in the folder
for filename in os.listdir(folder_path):
if filename.endswith(".csv"):
file_path = os.path.join(folder_path, filename)
# Read the data from the CSV file
data = pd.read_csv(file_path)
# Add the data frame to the list
data_frames.append(data)
# Extract the file name without the extension
file_name = os.path.splitext(filename)[0]
# Calculate pivot points and other parameters
pivot_points, resistance_levels, support_levels, pivot_high_points, pivot_low_points = calculate_pivot_points(data)
# Extract closing prices
closing_prices = data['close']
# Define the range of parameter values to test
parameter_range = range(1, 40)
# Calculate scores for different parameter combinations
parameter_scores = []
for high_parameter in parameter_range:
for low_parameter in parameter_range:
if low_parameter <= 8: # Add the condition here
# Determine significant pivot high points using swing points
significant_high_points = []
for point in pivot_high_points:
if point['index'] > 0 and point['index'] < len(data) - 1:
high_range = data.loc[point['index'] - high_parameter: point['index'] + low_parameter, 'high']
if point['value'] == high_range.max():
significant_high_points.append(point)
# Determine significant pivot low points using swing points
significant_low_points = []
for point in pivot_low_points:
if point['index'] > 0 and point['index'] < len(data) - 1:
low_range = data.loc[point['index'] - high_parameter: point['index'] + low_parameter, 'low']
if point['value'] == low_range.min():
significant_low_points.append(point)
# Calculate the score as the difference between high and low point counts
score = len(significant_high_points) - len(significant_low_points)
parameter_scores.append((high_parameter, low_parameter, score))
# Convert the scores to a NumPy array for easier manipulation
scores = np.array(parameter_scores)
# Find the optimal parameter values using the knee point
if len(scores) > 0:
knee_index = argrelextrema(scores[:, 2], np.less)[0][-1]
optimal_high_parameter, optimal_low_parameter, optimal_score = scores[knee_index]
else:
optimal_high_parameter = 16 # Manually assign the value
optimal_low_parameter = 2 # Manually assign the value
print("Optimal high parameter value:", optimal_high_parameter)
print("Optimal low parameter value:", optimal_low_parameter)
# Plot line chart for closing prices
plt.plot(closing_prices, label='Closing Prices')
slope_range = 1 # Adjust this range as per your requirement
# Calculate the trendlines for connecting the pivot high points
trendlines_high = []
trendline_points_high = []
for i in range(0, len(significant_high_points) - 2):
point1 = significant_high_points[i]
point2 = significant_high_points[i+1]
slope1 = (point2['value'] - point1['value']) / (point2['index'] - point1['index'])
point3 = significant_high_points[i+1]
point4 = significant_high_points[i+2]
slope2 = (point4['value'] - point3['value']) / (point4['index'] - point3['index'])
slope_difference = abs(slope2 - slope1)
if slope1 < 0:
if not trendline_points_high:
trendline_points_high.append(point1)
if slope_difference <= slope_range:
trendline_points_high.append(point2)
else:
if len(trendline_points_high) > 1:
trendlines_high.append(trendline_points_high)
trendline_points_high = [point2] # Start a new trendline with point2
if len(trendline_points_high) > 1:
trendlines_high.append(trendline_points_high)
# Calculate the trendlines for connecting the pivot low points
trendlines_low = []
trendline_points_low = []
for i in range(0, len(significant_low_points) - 2):
point1 = significant_low_points[i]
point2 = significant_low_points[i+1]
slope1 = (point2['value'] - point1['value']) / (point2['index'] - point1['index'])
point3 = significant_low_points[i+1]
point4 = significant_low_points[i+2]
slope2 = (point4['value'] - point3['value']) / (point4['index'] - point3['index'])
slope_difference = abs(slope2 - slope1)
if slope1 > 0:
if not trendline_points_low:
trendline_points_low.append(point1)
if slope_difference <= slope_range:
trendline_points_low.append(point2)
else:
if len(trendline_points_low) > 1:
trendlines_low.append(trendline_points_low)
trendline_points_low = [point2] # Start a new trendline with point2
if len(trendline_points_low) > 1:
trendlines_low.append(trendline_points_low)
# Plot the trendlines for positive slope
for trendline_points_high in trendlines_high:
x_values = [point['index'] for point in trendline_points_high]
y_values = [point['value'] for point in trendline_points_high]
plt.plot(x_values, y_values, color='red', linestyle='dashed')
# Plot the significant pivot high points
x_values = [point['index'] for point in significant_high_points]
y_values = [point['value'] for point in significant_high_points]
plt.scatter(x_values, y_values, color='red', label='Significant Pivot High Points')
# Plot the trendlines for positive slope
for trendline_points_low in trendlines_low:
x_values = [point['index'] for point in trendline_points_low]
y_values = [point['value'] for point in trendline_points_low]
plt.plot(x_values, y_values, color='green', linestyle='dashed')
# Plot the significant pivot low points
x_values = [point['index'] for point in significant_low_points]
y_values = [point['value'] for point in significant_low_points]
plt.scatter(x_values, y_values, color='green', label='Significant Pivot Low Points')
# Set chart title and labels
plt.title(f'Closing Prices with Trendlines and Significant Pivot Points ({file_name})')
plt.xlabel('Index')
plt.ylabel('Closing Price')
# Show the chart for the current data frame
plt.legend()
plt.show()
This is my new approach as per the logic I just stated, but still the plotting isn't anywhere near what we desire.
