import dash_bootstrap_components as dbc 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 from plotting import CandlestickChart, figure_with_subplots from daily_data import get_daily_data from strategies import advances from strategies import cumulative_rsi from strategies import declines from strategies import double_5s from strategies import down_days_in_a_row from strategies import end_of_month from strategies import internal_bar_strength from strategies import lower_lows from strategies import short_term_lows from strategies import trin from strategies import trin_thrusts from strategies import turnaround from strategies import two_period_rsi from strategies import vix_reversal_1 from strategies import vix_reversal_2 from strategies import vix_reversal_3 from strategies import vix_rsi 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] signal_functions: List[Dict[str, SignalFunction]] = [ {'strategy': '2-Period RSI', 'function': two_period_rsi.signals}, {'strategy': 'Advances', 'function': advances.signals}, {'strategy': 'Cumulative RSI', 'function': cumulative_rsi.signals}, {'strategy': 'Declines', 'function': declines.signals}, {'strategy': 'Double 5\'s', 'function': double_5s.signals}, {'strategy': 'Down Days in a Row', 'function': down_days_in_a_row.signals}, {'strategy': 'End of Month', 'function': end_of_month.signals}, {'strategy': 'Internal Bar Strength', 'function': internal_bar_strength.signals}, {'strategy': 'Lower Lows', 'function': lower_lows.signals}, {'strategy': 'Short-Term Lows', 'function': short_term_lows.signals}, {'strategy': 'TRIN', 'function': trin.signals}, {'strategy': 'TRIN Thrusts', 'function': trin_thrusts.signals}, {'strategy': 'Turnaround', 'function': turnaround.signals}, {'strategy': 'VIX Reversal 1', 'function': vix_reversal_1.signals}, {'strategy': 'VIX Reversal 2', 'function': vix_reversal_2.signals}, {'strategy': 'VIX Reversal 3', 'function': vix_reversal_3.signals}, {'strategy': 'VIX RSI', 'function': vix_rsi.signals} ] def calculate_signals(data: DataFrame, days: int = 12) -> DataFrame: signal_data = [] for signal_info in signal_functions: signal_dict = {'Strategy': signal_info['strategy']} signals = signal_info['function'](data).tail(days) dates = [datetime.strptime(str(date), '%Y-%m-%d').strftime('%m/%d') for date in data.tail(days)['Date']] for date, signal in zip(dates, signals): signal_dict[date] = signal signal_data.append(signal_dict) return DataFrame(signal_data) def load_chart(data: DataFrame) -> dict: chart_data = data.tail(180) candlestick_chart = CandlestickChart( x = chart_data['Date'], opens = chart_data['Open'], highs = chart_data['High'], lows = chart_data['Low'], closes = chart_data['Close'] ) return figure_with_subplots([[candlestick_chart]]) @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 = figure_with_subplots([[CandlestickChart( x = Series(datetime.today()), opens = Series(), highs = Series(), lows = Series(), closes = Series(), title = '' )]]) ), html.Div( AgGrid( id = 'signal-table', columnDefs = [{'field': 'Strategy', 'flex': 3}], rowData = [], defaultColDef = {'flex': 1, 'sortable': False, 'resizable': False}, dashGridOptions = { 'domLayout': 'autoHeight', 'pagination': True, 'paginationPageSizeSelector': False, 'paginationPageSize': 10, }, style = {'height': None} ), className = 'dbc dbc-ag-grid' ) ], style = {'maxWidth': '1200px', 'margin': '0 auto'}, 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)