157 lines
5.3 KiB
Python
157 lines
5.3 KiB
Python
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 cumulative_rsi
|
|
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': 'Cumulative RSI', 'function': cumulative_rsi.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'},
|
|
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) |