swing-trading-dashboard/dashboard.py

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 high_volume_days
from strategies import internal_bar_strength
from strategies import internal_bar_strength_band
from strategies import large_moves_down
from strategies import lower_lows
from strategies import short_term_lows
from strategies import tps
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': 'High Volume Days', 'function': high_volume_days.signals},
    {'strategy': 'Internal Bar Strength', 'function': internal_bar_strength.signals},
    {'strategy': 'Internal Bar Strength Band', 'function': internal_bar_strength_band.signals},
    {'strategy': 'Large Moves Down', 'function': large_moves_down.signals},
    {'strategy': 'Lower Lows', 'function': lower_lows.signals},
    {'strategy': 'Short-Term Lows', 'function': short_term_lows.signals},
    {'strategy': 'TPS', 'function': tps.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)