add test trend indicators

page_symbol_details.py

@@ -4,6 +4,7 @@ from plotly.subplots import make_subplots
 from streamlit_plotly_events import plotly_events
 import plotly.graph_objects as go
 
+import calculator
 import data_retriever
 import trends
 import ui
@@ -34,7 +35,7 @@ def run():
         # max time period
         st.text_input('No. of years look-back:', value=1, key="years_back")
         years_back = int(st.session_state.years_back)
-        weeks_back = years_back*12*4
+        weeks_back = years_back * 12 * 4
 
         symbol_prices = data_retriever.get_current_stock_data(symbol, weeks_back)
         if not any(symbol_prices):
@@ -55,17 +56,38 @@ def run():
     # indicator_datasets = indicators.create_indicators(dates, symbol_prices)
     # ui.create_indicators(fig1, indicator_datasets)
     # trend lines
-    sigma_multiplier = st.slider(label='min trend deviation', min_value=1.0, max_value=10.0, value=3.0, step=0.1)
-    sigma_dates, sigma_values = trends.trend_line(symbol_prices_backtest.index, np.array(symbol_prices_backtest['Close']), min_trend_size=21, sigma_multiplier=sigma_multiplier)
-    ui.create_markers(candleFigure, sigma_dates, sigma_values, f"{sigma_multiplier} Deviations from regression line", 'Deviations')
-
-    # vwap
-    vwap = trends.vwap(symbol_prices_backtest)
-    ui.create_line(candleFigure, backtest_dates, vwap, "Volume Weighted Average Price (VWAP)", "VWAP")
-
-    # bollinger bands
-    bollinger_dates, bollinger_low, bollinger_high = trends.bollinger_bands(backtest_dates, symbol_prices_backtest)
-    ui.create_fill_area(candleFigure, bollinger_dates, bollinger_low, bollinger_high, "Bollinger Bands")
+    # sigma_multiplier = st.slider(label='min trend deviation', min_value=1.0, max_value=10.0, value=3.0, step=0.1)
+    # sigma_dates, sigma_values = trends.trend_line(symbol_prices_backtest.index, np.array(symbol_prices_backtest['Close']), min_trend_size=21, sigma_multiplier=sigma_multiplier)
+    # ui.create_markers(candleFigure, sigma_dates, sigma_values, f"{sigma_multiplier} Deviations from regression line", 'Deviations')
+
+    # # vwap
+    # vwap = trends.vwap(symbol_prices_backtest)
+    # ui.create_line(candleFigure, backtest_dates, vwap, "Volume Weighted Average Price (VWAP)", "VWAP")
+
+    # # bollinger bands
+    # bollinger_dates, bollinger_low, bollinger_high = trends.bollinger_bands(backtest_dates, symbol_prices_backtest)
+    # ui.create_fill_area(candleFigure, bollinger_dates, bollinger_low, bollinger_high, "Bollinger Bands")
+
+    normalization_factor = 0.20*(max(symbol_prices['Close']) - min(symbol_prices['Close']))
+
+    # trend deviation area
+    period = 7
+    trend_deviation = trends.trend_deviation(symbol_prices['Close'][:], period=period)
+    trend_deviation_norm = calculator.normalize(trend_deviation, low=0.0, high=normalization_factor)
+    price_offset_low = [price - vol for price, vol in zip(symbol_prices['Close'][period:], trend_deviation_norm)]
+    price_offset_high = [price + vol for price, vol in zip(symbol_prices['Close'][period:], trend_deviation_norm)]
+    ui.create_fill_area(candleFigure, dates[period:], price_offset_low, price_offset_high, "Trend Deviation", color="rgba(0,50,50,0.3)")
+
+    # trade volume area
+    volume_norm = calculator.normalize(symbol_prices['Volume'], low=0.0, high=normalization_factor)
+    log_volume_offset_low = [price - vol for price, vol in
+                             zip(symbol_prices['Close'], volume_norm)]
+    log_volume_offset_high = [price + vol for price, vol in
+                              zip(symbol_prices['Close'], volume_norm)]
+    ui.create_fill_area(candleFigure, dates, log_volume_offset_low, log_volume_offset_high, "Volume Bands", color="rgba(100,20,0,0.3)")
+
+    sma_dates, deltas_sma = trends.sma(dates, symbol_prices['Close'], period)
+    ui.create_line(candleFigure, sma_dates, deltas_sma, f"Close price {period} day SMA")
 
     # plot all
     candleFigure.update_layout(title="Symbol Ticker",
@@ -75,17 +97,26 @@ def run():
     selected_points = plotly_events(candleFigure, click_event=True, hover_event=False, select_event=False, key='3')
 
     if st.checkbox('Tests'):
-        testFigure = make_subplots(rows=1, cols=1)
-        ui.create_line(testFigure, dates, symbol_prices['Volume'], "Volume")
-        ui.add_mouse_indicator(testFigure, selected_points, min=np.min(symbol_prices['Volume']),
-                               max=np.max(symbol_prices['Volume']))
-        test_id = st.plotly_chart(testFigure, use_container_width=True, key='test')
-
-        fft_steps = st.slider(label='FFT Steps', min_value=8, max_value=128, value=16, step=8)
-        specFig1 = ui.create_spectrogram(dates, symbol_prices['Close'], sampling_frequency=1, num_points_fft=fft_steps,
-                                         overlap_percent=0.0, title="Close Price Spectrogram")
-        specFig2 = ui.create_spectrogram(dates, symbol_prices['Volume'], sampling_frequency=1, num_points_fft=fft_steps,
-                                         overlap_percent=0.0, title="Volume Spectrogram")
+        # testFigure = make_subplots(rows=1, cols=1)
+        # ui.create_line(testFigure, dates, symbol_prices['Volume'], "Volume")
+        # ui.add_mouse_indicator(testFigure, selected_points, min=np.min(symbol_prices['Volume']),
+        #                        max=np.max(symbol_prices['Volume']))
+        # testFigure.update_layout(title="Volume")
+        # test_id = st.plotly_chart(testFigure, use_container_width=True, key='test')
+
+        deltas = [cur - symbol_prices['Close'][max(0, idx - 1)] for idx, cur in enumerate(symbol_prices['Close'])]
+        testFigure2 = make_subplots(rows=1, cols=1)
+        ui.create_line(testFigure2, dates, deltas, "Deltas")
+        ui.add_mouse_indicator(testFigure2, selected_points, min=np.min(deltas),
+                               max=np.max(deltas))
+        testFigure2.update_layout(title="Deltas")
+
+        ui.create_heatmap(dates, deltas, title="Deltas heatmap")
+
+        # sma
+        sma_dates, deltas_sma = trends.sma(dates, deltas)
+        ui.create_line(testFigure2, sma_dates, deltas_sma, "Deltas SMA")
+        test_id = st.plotly_chart(testFigure2, use_container_width=True, key='test2')
 
     if st.checkbox('Sector Trends'):
         # plot the trend of the market as a candlestick graph.

trends.py

@@ -1,6 +1,7 @@
 from collections import defaultdict
 
 import numpy as np
+import scipy as sp
 
 import calculator
 import data_retriever
@@ -116,11 +117,19 @@ def vwap(symbol_price_data):
     return vwap_values
 
 
+def sma(dates, data_list, n_days=20):
+    res_dates, res_values = [], []
+    res_dates = dates[n_days-1:]
+    for index in range(len(data_list) - n_days + 1):
+        res_values.append(np.mean(data_list[index:index + n_days]))
+    return res_dates, res_values
+
+
 def bollinger_bands(dates, symbol_price_data, n_days_deviation=20, deviation_multiplier=2):
     bollinger_dates, bollinger_high, bollinger_low = [], [], []
     closes = symbol_price_data['Close']
 
-    for index in range(0, len(closes) - n_days_deviation):
+    for index in range(len(closes) - n_days_deviation):
         bollinger_dates.append(dates[index + n_days_deviation])
 
         closes_subset = closes[index:index + n_days_deviation]
@@ -133,3 +142,28 @@ def bollinger_bands(dates, symbol_price_data, n_days_deviation=20, deviation_mul
 
 def support_lines():
     a = 1
+
+
+# the volatility is the average deviation from the current trend.
+# this is measured by doing linear regression over each period, then calculating the deviation of values from that line.
+def trend_deviation(prices, period=32):
+    np_dates = np.array([ts.timestamp() for ts in prices.index])
+    np_values = np.array(prices)
+
+    value_count = len(prices)
+
+    avg_deviations = []
+    for index in range(value_count-period):
+        np_dates_subset = np_dates[index:index+period+1]
+        np_values_subset = np_values[index:index+period+1]
+        m, c = calculator.linear_regression_line(np_dates_subset, np_values_subset)
+        predicted_values = m * np_dates_subset + c
+
+        deviation_sum = 0
+        for value, predicted_value in zip(np_values_subset, predicted_values):
+            deviation = abs(value - predicted_value)
+            deviation_sum += deviation
+        avg_deviation = deviation_sum/period
+        avg_deviations.append(avg_deviation)
+
+    return avg_deviations

ui.py

@@ -102,7 +102,7 @@ def create_fill_area(fig, dates, y_low, y_high, title, color='rgba(0,100,80,0.2)
     fig.add_trace(fill_area)
 
 
-def create_spectrogram(dates, data_list, sampling_frequency=1, num_points_fft=128, overlap_percent=50.0, title="title"):
+def create_spectrogram(dates, data_list, sampling_frequency=1, num_points_fft=128, overlap_percent=50.0, title="title", log_scale=True):
     data_list = calculator.normalize(data_list, 1, -1)
 
     # Spectrogram
@@ -111,10 +111,15 @@ def create_spectrogram(dates, data_list, sampling_frequency=1, num_points_fft=12
 
     dates_subset = [dates[int(bin)] for bin in bins]
 
+    if log_scale:
+        z = 10 * np.log10(pxx)
+    else:
+        z = pxx
+
     trace = [go.Heatmap(
         x=dates_subset,
         y=freqs,
-        z=10 * np.log10(pxx),
+        z=z,
         colorscale='Jet',
     )]
     layout = go.Layout(
@@ -123,10 +128,49 @@ def create_spectrogram(dates, data_list, sampling_frequency=1, num_points_fft=12
         xaxis=dict(title='Time'),  # y-axis label
     )
     fig = go.Figure(data=trace, layout=layout)
+    fig.update_layout(title=title)
     st.plotly_chart(fig, use_container_width=True)
 
     return fig
 
+
+def create_heatmap(dates, data_list, bin_count=10, time_steps=20, title='title'):
+    time_width = int(len(dates)/time_steps)
+    dates_subset = [dates[index*time_width] for index in range(time_steps)]
+    min_val = np.min(data_list)
+    max_val = np.max(data_list)
+    delta = (max_val-min_val)
+    min_val -= 0.2*delta
+    max_val += 0.2*delta
+    delta = (max_val-min_val)
+
+    bin_width = delta/(bin_count + 1)
+    bins = np.arange(min_val, max_val, bin_width)
+
+    values = np.empty(shape=(time_steps, bin_count))
+    for time_index in range(time_steps):
+        data_subset = data_list[time_index*time_width:time_index*time_width+time_width]
+        counts, res_bins = np.histogram(data_subset, bins=bins)
+        values[time_index:] = counts
+
+    trace = [go.Heatmap(
+        x=dates_subset,
+        y=bins,
+        z=values.transpose(),
+        colorscale='Jet',
+    )]
+    layout = go.Layout(
+        title=title,
+        yaxis=dict(title='Values'),  # x-axis label
+        xaxis=dict(title='Time'),  # y-axis label
+    )
+    fig = go.Figure(data=trace, layout=layout)
+    fig.update_layout(title=title)
+    st.plotly_chart(fig, use_container_width=True)
+
+    return fig
+
+
 def add_mouse_indicator(fig, selected_points, min, max):
     if any(selected_points):
         fig.add_shape(