Add support for live intraday data / signals and support caching requests for them server-side

dashboard.py

@@ -1,9 +1,10 @@
 import dash_bootstrap_components as dbc
 
-from dash import Dash, dcc, html
+from dash import Dash, dcc, html, Input, Output
 from dash_ag_grid import AgGrid
 from dash_bootstrap_templates import load_figure_template
 from datetime import datetime
+from flask_caching import Cache
 from pandas import DataFrame, Series
 from typing import Callable, List, Dict
 
@@ -26,6 +27,11 @@ from stylesheets import grid_stylesheet, theme_stylesheet
 app = Dash(__name__, external_stylesheets = [theme_stylesheet, grid_stylesheet])
 app.title = 'Swing Trading Dashboard'
 
+cache = Cache(app.server, config = {
+    'CACHE_TYPE': 'SimpleCache',
+    'CACHE_DEFAULT_TIMEOUT': 60 # Seconds.
+})
+
 load_figure_template('lux_dark')
 
 SignalFunction = Callable[[DataFrame], Series]
@@ -44,9 +50,7 @@ signal_functions: List[Dict[str, SignalFunction]] = [
     {'strategy': 'VIX RSI', 'function': vix_rsi.signals}
 ]
 
-data = get_daily_data('SPY')
-
-def calculate_signals(days: int = 12) -> DataFrame:
+def calculate_signals(data: DataFrame, days: int = 12) -> DataFrame:
     signal_data = []
 
     for signal_info in signal_functions:
@@ -61,7 +65,7 @@ def calculate_signals(days: int = 12) -> DataFrame:
     
     return DataFrame(signal_data)
 
-def load_chart() -> dict:
+def load_chart(data: DataFrame) -> dict:
     chart_data = data.tail(180)
     candlestick_chart = CandlestickChart(
         x = chart_data['Date'],
@@ -73,22 +77,45 @@ def load_chart() -> dict:
 
     return figure_with_subplots([[candlestick_chart]])
 
-signal_data = calculate_signals()
+@cache.cached(timeout = 60, key_prefix = 'get_daily_data')
+def get_cached_data():
+    return get_daily_data('SPY')
+
+@cache.cached(timeout = 60, key_prefix = 'calculate_signals')
+def get_cached_signals():
+    data = get_cached_data()
+    return calculate_signals(data)
+
+@cache.cached(timeout = 60, key_prefix = 'load_chart')
+def get_cached_chart():
+    data = get_cached_data()
+    return load_chart(data)
 
 app.layout = dbc.Container(
     [
+        dcc.Interval(
+            id = 'refresh-interval',
+            interval = 60 * 1000, # Milliseconds.
+            n_intervals = 0
+        ),
         dcc.Graph(
             id = 'candlestick-chart',
             config = {'displayModeBar': False},
-            figure = load_chart(),
+            figure = figure_with_subplots([[CandlestickChart(
+                x = Series(datetime.today()),
+                opens = Series(),
+                highs = Series(),
+                lows = Series(),
+                closes = Series(),
+                title = ''
+        )]])
+
         ),
         html.Div(
             AgGrid(
-                columnDefs = [
-                    {'field': col, 'flex': 3} if col == 'Strategy' else {'field': col, 'flex': 1, 'cellRenderer': 'SignalRenderer'}
-                    for col in signal_data.columns
-                ],
-                rowData = signal_data.to_dict(orient = 'records'),
+                id = 'signal-table',
+                columnDefs = [{'field': 'Strategy', 'flex': 3}],
+                rowData = [],
                 defaultColDef = {'flex': 1, 'sortable': False, 'resizable': False},
                 dashGridOptions = {'domLayout': 'autoHeight'},
                 style = {'height': None}
@@ -100,5 +127,24 @@ app.layout = dbc.Container(
     fluid = True
 )
 
+@app.callback(
+    [
+        Output('candlestick-chart', 'figure'),
+        Output('signal-table', 'columnDefs'),
+        Output('signal-table', 'rowData')
+    ],
+    Input('refresh-interval', 'n_intervals')
+)
+def update_dashboard(_):
+    signal_data = get_cached_signals()
+    chart_figure = get_cached_chart()
+
+    column_defs = [
+        {'field': col, 'flex': 3} if col == 'Strategy' else {'field': col, 'flex': 1, 'cellRenderer': 'SignalRenderer'}
+        for col in signal_data.columns
+    ]
+
+    return chart_figure, column_defs, signal_data.to_dict(orient = 'records')
+
 if __name__ == '__main__':
     app.run_server(debug = False)