import json
import os
import random
from io import StringIO
import dash_bootstrap_components as dbc
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import talib
import yfinance as yf
from dash import Dash, dcc, html, dash_table, Input, Output, State, no_update, callback_context
from flask_caching import Cache
from talib import abstract
from talib._ta_lib import CandleSettingType, RangeType, _ta_set_candle_settings
from vectorbt import settings
from vectorbt.portfolio.base import Portfolio
from vectorbt.portfolio.enums import Direction, DirectionConflictMode
from vectorbt.utils.colors import adjust_opacity
from vectorbt.utils.config import merge_dicts
USE_CACHING = (
os.environ.get(
"USE_CACHING",
"True",
)
== "True"
)
HOST = os.environ.get(
"HOST",
"127.0.0.1",
)
PORT = int(
os.environ.get(
"PORT",
8050,
)
)
DEBUG = (
os.environ.get(
"DEBUG",
"True",
)
== "True"
)
GITHUB_LINK = os.environ.get(
"GITHUB_LINK",
"https://github.com/polakowo/vectorbt/tree/master/apps/candlestick-patterns",
)
app = Dash(
__name__,
title="VectorBT: Candlestick Patterns",
meta_tags=[
{
"name": "viewport",
"content": "width=device-width, initial-scale=1, maximum-scale=1.0, user-scalable=no",
}
],
external_stylesheets=[dbc.themes.BOOTSTRAP],
)
CACHE_CONFIG = {
"CACHE_TYPE": "filesystem" if USE_CACHING else "null",
"CACHE_DIR": "data",
"CACHE_DEFAULT_TIMEOUT": 0,
"CACHE_THRESHOLD": 50,
}
cache = Cache()
cache.init_app(app.server, config=CACHE_CONFIG)
periods = ["1d", "5d", "1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max"]
intervals = ["1m", "2m", "5m", "15m", "30m", "60m", "90m", "1d", "5d", "1wk", "1mo", "3mo"]
patterns = talib.get_function_groups()["Pattern Recognition"]
stats_table_columns = ["Metric", "Buy & Hold", "Random (Median)", "Strategy", "Z-Score"]
directions = Direction._fields
conflict_modes = DirectionConflictMode._fields
plot_types = ["OHLC", "Candlestick"]
color_schema = settings["plotting"]["color_schema"]
bgcolor = "#ffffff"
dark_bgcolor = "#f6f7fb"
fontcolor = "#0f172a"
dark_fontcolor = "#475569"
gridcolor = "#e5e7eb"
active_color = "#2563EB"
loadcolor = "#2563EB"
bordercolor = "#e5e7eb"
bordercolor_strong = "#d1d5db"
data_path = "data/data.h5"
default_metric = "Total Return [%]"
default_symbol = "BTC-USD"
default_period = "1y"
default_interval = "1d"
default_date_range = [0, 1]
default_fees = 0.1
default_fixed_fees = 0.0
default_slippage = 5.0
default_yf_options = ["auto_adjust"]
default_exit_n_random = default_entry_n_random = 5
default_prob_options = ["mimic_strategy"]
default_entry_prob = 0.1
default_exit_prob = 0.1
default_entry_patterns = [
"CDLHAMMER",
"CDLINVERTEDHAMMER",
"CDLPIERCING",
"CDLMORNINGSTAR",
"CDL3WHITESOLDIERS",
]
default_exit_options = []
default_exit_patterns = [
"CDLHANGINGMAN",
"CDLSHOOTINGSTAR",
"CDLEVENINGSTAR",
"CDL3BLACKCROWS",
"CDLDARKCLOUDCOVER",
]
default_candle_settings = pd.DataFrame(
{
"SettingType": [
"BodyLong",
"BodyVeryLong",
"BodyShort",
"BodyDoji",
"ShadowLong",
"ShadowVeryLong",
"ShadowShort",
"ShadowVeryShort",
"Near",
"Far",
"Equal",
],
"RangeType": [
"RealBody",
"RealBody",
"RealBody",
"HighLow",
"RealBody",
"RealBody",
"Shadows",
"HighLow",
"HighLow",
"HighLow",
"HighLow",
],
"AvgPeriod": [10, 10, 10, 10, 0, 0, 10, 10, 5, 5, 5],
"Factor": [1.0, 3.0, 1.0, 0.1, 1.0, 2.0, 1.0, 0.1, 0.2, 0.6, 0.05],
}
)
default_entry_dates = []
default_exit_dates = []
default_direction = directions[0]
default_conflict_mode = conflict_modes[0]
default_sim_options = ["allow_accumulate"]
default_n_random_strat = 50
default_stats_options = ["incl_open"]
default_layout = dict(
template="plotly_white",
autosize=True,
margin=dict(b=40, t=20),
font=dict(color=fontcolor),
plot_bgcolor=bgcolor,
paper_bgcolor=bgcolor,
legend=dict(font=dict(size=10), orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
)
default_subplots = ["orders", "trade_pnl", "cum_returns"]
default_plot_type = "OHLC"
def section_summary(text: str):
return html.Summary(className="section-title", children=text)
app.layout = html.Div(
className="app-shell",
children=[
html.Div(
className="app-banner",
children=[
html.H6("VectorBT: Candlestick Patterns"),
html.A(
"View on GitHub",
href=GITHUB_LINK,
target="_blank",
className="btn-link",
),
],
),
dbc.Row(
className="g-3",
children=[
dbc.Col(
lg=8,
sm=12,
children=[
html.Div(
className="pretty-container",
children=[
html.Div(
className="card-header",
children=[html.H6("OHLCV and signals")],
),
dbc.Row(
className="g-2",
children=[
dbc.Col(
lg=4,
sm=12,
children=[
html.Label("Select plot type:"),
dcc.Dropdown(
id="plot_type_dropdown",
options=[{"value": i, "label": i} for i in plot_types],
value=default_plot_type,
className="dropdown-control",
),
],
)
],
),
dcc.Loading(
id="ohlcv_loading",
type="default",
color=loadcolor,
children=[dcc.Graph(id="ohlcv_graph", figure={"layout": default_layout})],
),
html.Small("Hint: Use Box and Lasso Select to filter signals"),
],
),
dbc.Row(
className="g-3",
children=[
dbc.Col(
children=[
html.Div(
className="pretty-container",
children=[
html.Div(
className="card-header",
children=[html.H6("Portfolio")],
),
dbc.Row(
className="g-2",
children=[
dbc.Col(
lg=6,
sm=12,
children=[
html.Label("Select subplots:"),
dcc.Dropdown(
id="subplot_dropdown",
options=[
{"value": k, "label": v["title"]}
for k, v in Portfolio.subplots.items()
],
multi=True,
value=default_subplots,
className="dropdown-control",
),
],
)
],
),
dcc.Loading(
id="portfolio_loading",
type="default",
color=loadcolor,
children=[
dcc.Graph(
id="portfolio_graph",
figure={"layout": default_layout},
)
],
),
],
),
]
)
],
),
dbc.Row(
className="g-3",
children=[
dbc.Col(
children=[
html.Div(
className="pretty-container",
children=[
html.Div(
className="card-header",
children=[html.H6("Stats")],
),
dcc.Loading(
id="stats_loading",
type="default",
color=loadcolor,
children=[
dash_table.DataTable(
id="stats_table",
columns=[{"name": c, "id": c} for c in stats_table_columns],
style_data_conditional=[
{
"if": {"column_id": stats_table_columns[1]},
"fontWeight": "bold",
"borderLeft": "1px solid " + bordercolor,
},
{
"if": {"column_id": stats_table_columns[2]},
"fontWeight": "bold",
},
{
"if": {"column_id": stats_table_columns[3]},
"fontWeight": "bold",
},
{
"if": {"column_id": stats_table_columns[4]},
"fontWeight": "bold",
},
{
"if": {"state": "selected"},
"backgroundColor": dark_bgcolor,
"color": active_color,
"border": "1px solid " + active_color,
},
{
"if": {"state": "active"},
"backgroundColor": dark_bgcolor,
"color": active_color,
"border": "1px solid " + active_color,
},
],
style_header={
"border": "none",
"backgroundColor": bgcolor,
"fontWeight": "bold",
"padding": "0px 5px",
},
style_data={
"border": "none",
"backgroundColor": bgcolor,
"color": dark_fontcolor,
"paddingRight": "10px",
},
style_table={"overflowX": "auto"},
style_as_list_view=False,
editable=False,
),
],
),
],
),
]
)
],
),
dbc.Row(
className="g-3",
children=[
dbc.Col(
children=[
html.Div(
className="pretty-container",
children=[
html.Div(
className="card-header",
children=[html.H6("Metric stats")],
),
dcc.Loading(
id="metric_stats_loading",
type="default",
color=loadcolor,
children=[
html.Label("Metric:"),
dbc.Row(
className="g-2",
children=[
dbc.Col(
lg=4,
sm=12,
children=[
dcc.Dropdown(
id="metric_dropdown",
className="dropdown-control",
),
],
),
dbc.Col(
lg=8,
sm=12,
children=[
dcc.Graph(
id="metric_graph",
figure={"layout": default_layout},
)
],
),
],
),
],
),
],
),
]
)
],
),
],
),
dbc.Col(
lg=4,
sm=12,
children=[
html.Div(
className="pretty-container",
children=[
html.Div(
className="card-header",
children=[html.H6("Settings")],
),
html.Button("Reset", id="reset_button", className="btn-neutral"),
html.Details(
open=True,
children=[
section_summary("Data"),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
html.Label("Yahoo! Finance symbol:"),
dcc.Input(
id="symbol_input",
className="input-control",
type="text",
value=default_symbol,
placeholder="Enter symbol...",
debounce=True,
),
]
),
dbc.Col(),
],
),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
html.Label("Period:"),
dcc.Dropdown(
id="period_dropdown",
options=[{"value": i, "label": i} for i in periods],
value=default_period,
className="dropdown-control",
),
]
),
dbc.Col(
children=[
html.Label("Interval:"),
dcc.Dropdown(
id="interval_dropdown",
options=[{"value": i, "label": i} for i in intervals],
value=default_interval,
className="dropdown-control",
),
]
),
],
),
html.Label("Filter period:"),
dcc.RangeSlider(
id="date_slider",
min=0.0,
max=1.0,
value=default_date_range,
allowCross=False,
tooltip={"placement": "bottom"},
),
dcc.Checklist(
id="yf_checklist",
options=[
{"label": "Adjust all OHLC automatically", "value": "auto_adjust"},
{
"label": "Use back-adjusted data to mimic true historical prices",
"value": "back_adjust",
},
],
value=default_yf_options,
className="checklist",
),
],
),
html.Details(
open=True,
children=[
section_summary("Entry patterns"),
html.Div(
id="entry_settings",
children=[
html.Div(
className="button-row",
children=[
html.Button(
"All", id="entry_all_button", className="btn-neutral"
),
html.Button(
"Random", id="entry_random_button", className="btn-neutral"
),
html.Button(
"Clear", id="entry_clear_button", className="btn-neutral"
),
],
),
html.Label("Number of random patterns:", className="field-label"),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
dcc.Input(
id="entry_n_random_input",
className="input-control",
value=default_entry_n_random,
placeholder="Enter number...",
debounce=True,
type="number",
min=1,
max=len(patterns),
step=1,
),
]
),
dbc.Col(),
],
),
html.Label("Select patterns:"),
dcc.Dropdown(
id="entry_pattern_dropdown",
options=[{"value": i, "label": i} for i in patterns],
multi=True,
value=default_entry_patterns,
className="dropdown-control",
),
],
),
],
),
html.Details(
open=True,
children=[
section_summary("Exit patterns"),
dcc.Checklist(
id="exit_checklist",
options=[{"label": "Same as entry patterns", "value": "same_as_entry"}],
value=default_exit_options,
className="checklist",
),
html.Div(
id="exit_settings",
hidden="same_as_entry" in default_exit_options,
children=[
html.Div(
className="button-row",
children=[
html.Button(
"All", id="exit_all_button", className="btn-neutral"
),
html.Button(
"Random", id="exit_random_button", className="btn-neutral"
),
html.Button(
"Clear", id="exit_clear_button", className="btn-neutral"
),
],
),
html.Label("Number of random patterns:", className="field-label"),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
dcc.Input(
id="exit_n_random_input",
className="input-control",
value=default_exit_n_random,
placeholder="Enter number...",
debounce=True,
type="number",
min=1,
max=len(patterns),
step=1,
),
]
),
dbc.Col(),
],
),
html.Label("Select patterns:"),
dcc.Dropdown(
id="exit_pattern_dropdown",
options=[{"value": i, "label": i} for i in patterns],
multi=True,
value=default_exit_patterns,
className="dropdown-control",
),
],
),
],
),
html.Details(
children=[
section_summary("Candle settings"),
dash_table.DataTable(
id="candle_settings_table",
columns=[
{
"name": c,
"id": c,
"editable": i in (2, 3),
"type": "numeric" if i in (2, 3) else "any",
}
for i, c in enumerate(default_candle_settings.columns)
],
data=default_candle_settings.to_dict("records"),
style_data_conditional=[
{
"if": {"column_editable": True},
"backgroundColor": dark_bgcolor,
"border": "1px solid " + bordercolor_strong,
},
{
"if": {"state": "selected"},
"backgroundColor": dark_bgcolor,
"color": active_color,
"border": "1px solid " + active_color,
},
{
"if": {"state": "active"},
"backgroundColor": dark_bgcolor,
"color": active_color,
"border": "1px solid " + active_color,
},
],
style_header={
"border": "none",
"backgroundColor": bgcolor,
"fontWeight": "bold",
"padding": "0px 5px",
},
style_data={
"border": "none",
"backgroundColor": bgcolor,
"color": dark_fontcolor,
},
style_table={"overflowX": "auto"},
style_as_list_view=False,
editable=True,
),
],
),
html.Details(
open=False,
children=[
section_summary("Custom pattern"),
html.Div(
id="custom_settings",
children=[
html.Label("Select entry dates:"),
dcc.Dropdown(
id="custom_entry_dropdown",
options=[],
multi=True,
value=default_entry_dates,
className="dropdown-control",
),
html.Label("Select exit dates:"),
dcc.Dropdown(
id="custom_exit_dropdown",
options=[],
multi=True,
value=default_exit_dates,
className="dropdown-control",
),
],
),
],
),
html.Details(
open=True,
children=[
section_summary("Simulation settings"),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
html.Label("Fees (in %):"),
dcc.Input(
id="fees_input",
className="input-control",
type="number",
value=default_fees,
placeholder="Enter fees...",
debounce=True,
min=0,
max=100,
),
]
),
dbc.Col(
children=[
html.Label("Fixed fees:"),
dcc.Input(
id="fixed_fees_input",
className="input-control",
type="number",
value=default_fixed_fees,
placeholder="Enter fixed fees...",
debounce=True,
min=0,
),
]
),
],
),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
html.Label("Slippage (in % of H-O):"),
dcc.Input(
id="slippage_input",
className="input-control",
type="number",
value=default_slippage,
placeholder="Enter slippage...",
debounce=True,
min=0,
max=100,
),
]
),
dbc.Col(),
],
),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
html.Label("Direction:"),
dcc.Dropdown(
id="direction_dropdown",
options=[{"value": i, "label": i} for i in directions],
value=default_direction,
className="dropdown-control",
),
]
),
dbc.Col(
children=[
html.Label("Conflict Mode:"),
dcc.Dropdown(
id="conflict_mode_dropdown",
options=[{"value": i, "label": i} for i in conflict_modes],
value=default_conflict_mode,
className="dropdown-control",
),
]
),
],
),
dcc.Checklist(
id="sim_checklist",
options=[
{"label": "Allow signal accumulation", "value": "allow_accumulate"}
],
value=default_sim_options,
className="checklist",
),
html.Label("Number of random strategies to test against:"),
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
dcc.Input(
id="n_random_strat_input",
className="input-control",
value=default_n_random_strat,
placeholder="Enter number...",
debounce=True,
type="number",
min=10,
max=1000,
step=1,
),
]
),
dbc.Col(),
],
),
dcc.Checklist(
id="prob_checklist",
options=[
{"label": "Mimic strategy by shuffling", "value": "mimic_strategy"}
],
value=default_prob_options,
className="checklist",
),
html.Div(
id="prob_settings",
hidden="mimic_strategy" in default_prob_options,
children=[
dbc.Row(
className="g-2",
children=[
dbc.Col(
children=[
html.Label("Entry probability (in %):"),
dcc.Input(
id="entry_prob_input",
className="input-control",
value=default_entry_prob,
placeholder="Enter number...",
debounce=True,
type="number",
min=0,
max=100,
),
]
),
dbc.Col(
children=[
html.Label("Exit probability (in %):"),
dcc.Input(
id="exit_prob_input",
className="input-control",
value=default_exit_prob,
placeholder="Enter number...",
debounce=True,
type="number",
min=0,
max=100,
),
]
),
],
),
],
),
dcc.Checklist(
id="stats_checklist",
options=[
{"label": "Include open trades in stats", "value": "incl_open"},
{
"label": "Use positions instead of trades in stats",
"value": "use_positions",
},
],
value=default_stats_options,
className="checklist",
),
],
),
],
),
],
),
],
),
html.Div(id="data_signal", style={"display": "none"}),
html.Div(id="index_signal", style={"display": "none"}),
html.Div(id="candle_settings_signal", style={"display": "none"}),
html.Div(id="stats_signal", style={"display": "none"}),
html.Div(id="window_width", style={"display": "none"}),
dcc.Location(id="url"),
],
)
app.clientside_callback(
"""
function(href) {
return window.innerWidth;
}
""",
Output("window_width", "children"),
[
Input("url", "href"),
],
)
@cache.memoize()
def fetch_data(symbol, period, interval, auto_adjust, back_adjust):
"""Fetch OHLCV data from Yahoo! Finance."""
df = yf.Ticker(symbol).history(
period=period,
interval=interval,
actions=False,
auto_adjust=auto_adjust,
back_adjust=back_adjust,
)
if df is None or df.empty:
raise ValueError("Empty data from yfinance")
return df
@app.callback(
[
Output("data_signal", "children"),
Output("index_signal", "children"),
],
[
Input("symbol_input", "value"),
Input("period_dropdown", "value"),
Input("interval_dropdown", "value"),
Input("yf_checklist", "value"),
],
)
def update_data(symbol, period, interval, yf_options):
"""Store data into a hidden DIV to avoid repeatedly calling Yahoo's API."""
auto_adjust = "auto_adjust" in yf_options
back_adjust = "back_adjust" in yf_options
df = fetch_data(symbol, period, interval, auto_adjust, back_adjust)
index_as_str = pd.to_datetime(df.index).astype(str).tolist()
return df.to_json(date_format="iso", orient="split"), index_as_str
@app.callback(
[
Output("date_slider", "min"),
Output("date_slider", "max"),
Output("date_slider", "value"),
],
[
Input("index_signal", "children"),
],
)
def update_date_slider(date_list):
"""Once index (dates) has changed, reset the date slider."""
return 0, len(date_list) - 1, [0, len(date_list) - 1]
@app.callback(
[
Output("custom_entry_dropdown", "options"),
Output("custom_exit_dropdown", "options"),
],
[
Input("index_signal", "children"),
Input("date_slider", "value"),
],
)
def update_custom_options(date_list, date_range):
"""Once dates have changed, update entry/exit dates in custom pattern section.
If selected dates cannot be found in new dates, they will be automatically removed."""
filtered_dates = np.asarray(date_list)[date_range[0] : date_range[1] + 1].tolist()
custom_options = [{"value": i, "label": i} for i in filtered_dates]
return custom_options, custom_options
@app.callback(
Output("entry_pattern_dropdown", "value"),
[
Input("entry_all_button", "n_clicks"),
Input("entry_random_button", "n_clicks"),
Input("entry_clear_button", "n_clicks"),
Input("reset_button", "n_clicks"),
],
[
State("entry_n_random_input", "value"),
],
)
def select_entry_patterns(_1, _2, _3, _4, n_random):
"""Select all/random entry patterns or clear."""
ctx = callback_context
if ctx.triggered:
button_id = ctx.triggered[0]["prop_id"].split(".")[0]
if button_id == "entry_all_button":
return patterns
elif button_id == "entry_random_button":
return random.sample(patterns, n_random)
elif button_id == "entry_clear_button":
return []
elif button_id == "reset_button":
return default_entry_patterns
return no_update
@app.callback(
[
Output("exit_settings", "hidden"),
Output("exit_n_random_input", "value"),
Output("exit_pattern_dropdown", "value"),
],
[
Input("exit_checklist", "value"),
Input("exit_all_button", "n_clicks"),
Input("exit_random_button", "n_clicks"),
Input("exit_clear_button", "n_clicks"),
Input("reset_button", "n_clicks"),
Input("entry_n_random_input", "value"),
Input("entry_pattern_dropdown", "value"),
],
[State("exit_n_random_input", "value")],
)
def select_exit_patterns(exit_options, _1, _2, _3, _4, entry_n_random, entry_patterns, exit_n_random):
"""Select all/random exit patterns, clear, or configure the same way as entry patterns."""
ctx = callback_context
same_as_entry = "same_as_entry" in exit_options
if ctx.triggered:
control_id = ctx.triggered[0]["prop_id"].split(".")[0]
if control_id == "exit_checklist":
return same_as_entry, entry_n_random, entry_patterns
elif control_id == "exit_all_button":
return same_as_entry, exit_n_random, patterns
elif control_id == "exit_random_button":
return same_as_entry, exit_n_random, random.sample(patterns, exit_n_random)
elif control_id == "exit_clear_button":
return same_as_entry, exit_n_random, []
elif control_id == "reset_button":
default_same_as_entry = "same_as_entry" in default_exit_options
return default_same_as_entry, default_exit_n_random, default_exit_patterns
elif control_id in ("entry_n_random_input", "entry_pattern_dropdown"):
if same_as_entry:
return same_as_entry, entry_n_random, entry_patterns
return no_update
@app.callback(
Output("candle_settings_signal", "children"),
[
Input("candle_settings_table", "data"),
],
)
def set_candle_settings(data):
"""Update candle settings in TA-Lib."""
for d in data:
AvgPeriod = d["AvgPeriod"]
if isinstance(AvgPeriod, float) and float.is_integer(AvgPeriod):
AvgPeriod = int(AvgPeriod)
Factor = float(d["Factor"])
_ta_set_candle_settings(
getattr(CandleSettingType, d["SettingType"]),
getattr(RangeType, d["RangeType"]),
AvgPeriod,
Factor,
)
@app.callback(
[
Output("ohlcv_graph", "figure"),
Output("prob_settings", "hidden"),
Output("entry_prob_input", "value"),
Output("exit_prob_input", "value"),
],
[
Input("window_width", "children"),
Input("plot_type_dropdown", "value"),
Input("data_signal", "children"),
Input("date_slider", "value"),
Input("entry_pattern_dropdown", "value"),
Input("exit_pattern_dropdown", "value"),
Input("candle_settings_signal", "children"),
Input("custom_entry_dropdown", "value"),
Input("custom_exit_dropdown", "value"),
Input("prob_checklist", "value"),
Input("reset_button", "n_clicks"),
],
[State("entry_prob_input", "value"), State("exit_prob_input", "value")],
)
def update_ohlcv(
window_width,
plot_type,
df_json,
date_range,
entry_patterns,
exit_patterns,
_1,
entry_dates,
exit_dates,
prob_options,
_2,
entry_prob,
exit_prob,
):
"""Update OHLCV graph.
Also update probability settings, as they also depend upon conversion of patterns into signals."""
df = pd.read_json(StringIO(df_json), orient="split")
df = df.iloc[date_range[0] : date_range[1] + 1]
talib_inputs = {
"open": df["Open"].values,
"high": df["High"].values,
"low": df["Low"].values,
"close": df["Close"].values,
"volume": df["Volume"].values,
}
entry_patterns = list((entry_patterns or [])) + ["CUSTOM"]
exit_patterns = list((exit_patterns or [])) + ["CUSTOM"]
all_patterns = list(set(entry_patterns + exit_patterns))
signal_df = pd.DataFrame.vbt.empty(
(len(df.index), len(all_patterns)), fill_value=0.0, index=df.index, columns=all_patterns
)
for pattern in all_patterns:
if pattern != "CUSTOM":
signal_df[pattern] = abstract.Function(pattern)(talib_inputs)
signal_df.loc[entry_dates, "CUSTOM"] += 100.0
signal_df.loc[exit_dates, "CUSTOM"] += -100.0
entry_signal_df = signal_df[entry_patterns]
exit_signal_df = signal_df[exit_patterns]
entry_df = entry_signal_df[(entry_signal_df > 0).any(axis=1)]
entry_patterns = []
for row_i, row in entry_df.iterrows():
entry_patterns.append("<br>".join(row.index[row != 0]))
entry_patterns = np.asarray(entry_patterns)
exit_df = exit_signal_df[(exit_signal_df < 0).any(axis=1)]
exit_patterns = []
for row_i, row in exit_df.iterrows():
exit_patterns.append("<br>".join(row.index[row != 0]))
exit_patterns = np.asarray(exit_patterns)
highest_high = df["High"].max()
lowest_low = df["Low"].min()
distance = (highest_high - lowest_low) / 5
entry_y = df.loc[entry_df.index, "Low"] - distance
entry_y.index = pd.to_datetime(entry_y.index)
exit_y = df.loc[exit_df.index, "High"] + distance
exit_y.index = pd.to_datetime(exit_y.index)
entry_signals = pd.Series.vbt.empty_like(entry_y, True)
exit_signals = pd.Series.vbt.empty_like(exit_y, True)
height = int(9 / 21 * 2 / 3 * window_width)
fig = df.vbt.ohlcv.plot(
plot_type=plot_type,
**merge_dicts(
default_layout,
dict(
width=None,
height=max(500, height),
margin=dict(r=40),
hovermode="closest",
xaxis2=dict(title="Date"),
yaxis2=dict(title="Volume"),
yaxis=dict(
title="Price",
),
),
),
)
entry_signals.vbt.signals.plot_as_entry_markers(
y=entry_y,
trace_kwargs=dict(
customdata=entry_patterns[:, None],
hovertemplate="%{x}<br>%{customdata[0]}",
name="Bullish signal",
),
add_trace_kwargs=dict(row=1, col=1),
fig=fig,
)
exit_signals.vbt.signals.plot_as_exit_markers(
y=exit_y,
trace_kwargs=dict(
customdata=exit_patterns[:, None],
hovertemplate="%{x}<br>%{customdata[0]}",
name="Bearish signal",
),
add_trace_kwargs=dict(row=1, col=1),
fig=fig,
)
fig.update_xaxes(gridcolor=gridcolor)
fig.update_yaxes(gridcolor=gridcolor, zerolinecolor=gridcolor)
figure = dict(data=fig.data, layout=fig.layout)
mimic_strategy = "mimic_strategy" in prob_options
ctx = callback_context
if ctx.triggered:
control_id = ctx.triggered[0]["prop_id"].split(".")[0]
if control_id == "reset_button":
mimic_strategy = "mimic_strategy" in default_prob_options
entry_prob = default_entry_prob
exit_prob = default_exit_prob
if mimic_strategy:
entry_prob = np.round(len(entry_df.index) / len(df.index) * 100, 4)
exit_prob = np.round(len(exit_df.index) / len(df.index) * 100, 4)
return figure, mimic_strategy, entry_prob, exit_prob
def simulate_portfolio(
df,
interval,
date_range,
selected_data,
entry_patterns,
exit_patterns,
entry_dates,
exit_dates,
fees,
fixed_fees,
slippage,
direction,
conflict_mode,
sim_options,
n_random_strat,
prob_options,
entry_prob,
exit_prob,
):
"""Simulate portfolio of the main strategy, buy & hold strategy, and a bunch of random strategies."""
df = df.iloc[date_range[0] : date_range[1] + 1]
talib_inputs = {
"open": df["Open"].values,
"high": df["High"].values,
"low": df["Low"].values,
"close": df["Close"].values,
"volume": df["Volume"].values,
}
entry_patterns = list((entry_patterns or [])) + ["CUSTOM"]
exit_patterns = list((exit_patterns or [])) + ["CUSTOM"]
all_patterns = list(set(entry_patterns + exit_patterns))
entry_i = [all_patterns.index(p) for p in entry_patterns]
exit_i = [all_patterns.index(p) for p in exit_patterns]
signals = np.full((len(df.index), len(all_patterns)), 0.0, dtype=np.float64)
for i, pattern in enumerate(all_patterns):
if pattern != "CUSTOM":
signals[:, i] = abstract.Function(pattern)(talib_inputs)
signals[np.flatnonzero(df.index.isin(entry_dates)), all_patterns.index("CUSTOM")] += 100.0
signals[np.flatnonzero(df.index.isin(exit_dates)), all_patterns.index("CUSTOM")] += -100.0
signals /= 100.0
if selected_data is not None:
new_signals = np.full_like(signals, 0.0)
for point in selected_data["points"]:
if "customdata" in point:
point_patterns = point["customdata"][0].split("<br>")
pi = df.index.get_loc(point["x"])
for p in point_patterns:
pc = all_patterns.index(p)
new_signals[pi, pc] = signals[pi, pc]
signals = new_signals
def _generate_size(signals):
entry_signals = signals[:, entry_i]
exit_signals = signals[:, exit_i]
return np.where(entry_signals > 0, entry_signals, 0).sum(axis=1) + np.where(
exit_signals < 0, exit_signals, 0
).sum(axis=1)
main_size = np.empty((len(df.index),), dtype=np.float64)
main_size[0] = 0
main_size[1:] = _generate_size(signals)[:-1]
hold_size = np.full_like(main_size, 0.0)
hold_size[0] = np.inf
def _shuffle_along_axis(a, axis):
idx = np.random.rand(*a.shape).argsort(axis=axis)
return np.take_along_axis(a, idx, axis=axis)
rand_size = np.empty((len(df.index), n_random_strat), dtype=np.float64)
rand_size[0] = 0
if "mimic_strategy" in prob_options:
for i in range(n_random_strat):
rand_signals = _shuffle_along_axis(signals, 0)
rand_size[1:, i] = _generate_size(rand_signals)[:-1]
else:
entry_signals = pd.DataFrame.vbt.signals.generate_random(
(rand_size.shape[0] - 1, rand_size.shape[1]), prob=entry_prob / 100
).values
exit_signals = pd.DataFrame.vbt.signals.generate_random(
(rand_size.shape[0] - 1, rand_size.shape[1]), prob=exit_prob / 100
).values
rand_size[1:, :] = np.where(entry_signals, 1.0, 0.0) - np.where(exit_signals, 1.0, 0.0)
def _simulate_portfolio(size, init_cash="autoalign"):
return Portfolio.from_signals(
close=df["Close"],
entries=size > 0,
exits=size < 0,
price=df["Open"],
size=np.abs(size),
direction=direction,
upon_dir_conflict=conflict_mode,
accumulate="allow_accumulate" in sim_options,
init_cash=init_cash,
fees=float(fees) / 100,
fixed_fees=float(fixed_fees),
slippage=(float(slippage) / 100) * (df["High"] / df["Open"] - 1),
freq=interval,
)
aligned_portfolio = _simulate_portfolio(np.hstack((main_size[:, None], rand_size)))
aligned_portfolio = aligned_portfolio.replace(init_cash=aligned_portfolio.init_cash)
main_portfolio = aligned_portfolio.iloc[0]
rand_portfolio = aligned_portfolio.iloc[1:]
hold_portfolio = _simulate_portfolio(hold_size, init_cash=main_portfolio.init_cash)
return main_portfolio, hold_portfolio, rand_portfolio
@app.callback(
[
Output("portfolio_graph", "figure"),
Output("stats_table", "data"),
Output("stats_signal", "children"),
Output("metric_dropdown", "options"),
Output("metric_dropdown", "value"),
],
[
Input("window_width", "children"),
Input("subplot_dropdown", "value"),
Input("data_signal", "children"),
Input("symbol_input", "value"),
Input("interval_dropdown", "value"),
Input("date_slider", "value"),
Input("ohlcv_graph", "selectedData"),
Input("entry_pattern_dropdown", "value"),
Input("exit_pattern_dropdown", "value"),
Input("candle_settings_signal", "children"),
Input("custom_entry_dropdown", "value"),
Input("custom_exit_dropdown", "value"),
Input("fees_input", "value"),
Input("fixed_fees_input", "value"),
Input("slippage_input", "value"),
Input("direction_dropdown", "value"),
Input("conflict_mode_dropdown", "value"),
Input("sim_checklist", "value"),
Input("n_random_strat_input", "value"),
Input("prob_checklist", "value"),
Input("entry_prob_input", "value"),
Input("exit_prob_input", "value"),
Input("stats_checklist", "value"),
Input("reset_button", "n_clicks"),
],
[State("metric_dropdown", "value")],
)
def update_stats(
window_width,
subplots,
df_json,
symbol,
interval,
date_range,
selected_data,
entry_patterns,
exit_patterns,
_1,
entry_dates,
exit_dates,
fees,
fixed_fees,
slippage,
direction,
conflict_mode,
sim_options,
n_random_strat,
prob_options,
entry_prob,
exit_prob,
stats_options,
_2,
curr_metric,
):
"""Final stage where we calculate key performance metrics and compare strategies."""
df = pd.read_json(StringIO(df_json), orient="split")
main_portfolio, hold_portfolio, rand_portfolio = simulate_portfolio(
df,
interval,
date_range,
selected_data,
entry_patterns,
exit_patterns,
entry_dates,
exit_dates,
fees,
fixed_fees,
slippage,
direction,
conflict_mode,
sim_options,
n_random_strat,
prob_options,
entry_prob,
exit_prob,
)
subplot_settings = dict()
if "cum_returns" in subplots:
subplot_settings["cum_returns"] = dict(
benchmark_kwargs=dict(
trace_kwargs=dict(line=dict(color=adjust_opacity(color_schema["yellow"], 0.5)), name=symbol)
)
)
height = int(6 / 21 * 2 / 3 * window_width)
fig = main_portfolio.plot(
subplots=subplots,
subplot_settings=subplot_settings,
**merge_dicts(
default_layout,
dict(width=None, height=len(subplots) * max(300, height) if len(subplots) > 1 else max(350, height)),
),
)
fig.update_traces(xaxis="x" if len(subplots) == 1 else "x" + str(len(subplots)))
fig.update_xaxes(gridcolor=gridcolor)
fig.update_yaxes(gridcolor=gridcolor, zerolinecolor=gridcolor)
def _chop_microseconds(delta):
return delta - pd.Timedelta(microseconds=delta.microseconds, nanoseconds=delta.nanoseconds)
def _metric_to_str(x):
if isinstance(x, float):
return "%.2f" % x
if isinstance(x, pd.Timedelta):
return str(_chop_microseconds(x))
return str(x)
incl_open = "incl_open" in stats_options
use_positions = "use_positions" in stats_options
main_stats = main_portfolio.stats(settings=dict(incl_open=incl_open, use_positions=use_positions))
hold_stats = hold_portfolio.stats(settings=dict(incl_open=True, use_positions=use_positions))
rand_stats = rand_portfolio.stats(settings=dict(incl_open=incl_open, use_positions=use_positions), agg_func=None)
rand_stats_median = rand_stats.iloc[:, 3:].median(axis=0)
rand_stats_mean = rand_stats.iloc[:, 3:].mean(axis=0)
rand_stats_std = rand_stats.iloc[:, 3:].std(axis=0, ddof=0)
stats_mean_diff = main_stats.iloc[3:] - rand_stats_mean
def _to_float(x):
if pd.isnull(x):
return np.nan
if isinstance(x, float):
if np.allclose(x, 0):
return 0.0
if isinstance(x, pd.Timedelta):
return float(x.total_seconds())
return float(x)
z = stats_mean_diff.apply(_to_float) / rand_stats_std.apply(_to_float)
table_data = pd.DataFrame(columns=stats_table_columns)
table_data.iloc[:, 0] = main_stats.index
table_data.iloc[:, 1] = hold_stats.apply(_metric_to_str).values
table_data.iloc[:3, 2] = table_data.iloc[:3, 1]
table_data.iloc[3:, 2] = rand_stats_median.apply(_metric_to_str).values
table_data.iloc[:, 3] = main_stats.apply(_metric_to_str).values
table_data.iloc[3:, 4] = z.apply(_metric_to_str).values
metric = curr_metric
ctx = callback_context
if ctx.triggered:
control_id = ctx.triggered[0]["prop_id"].split(".")[0]
if control_id == "reset_button":
metric = default_metric
if metric is None:
metric = default_metric
return (
dict(data=fig.data, layout=fig.layout),
table_data.to_dict("records"),
json.dumps(
{
"main": {m: [_to_float(main_stats[m])] for m in main_stats.index[3:]},
"hold": {m: [_to_float(hold_stats[m])] for m in main_stats.index[3:]},
"rand": {m: rand_stats[m].apply(_to_float).values.tolist() for m in main_stats.index[3:]},
}
),
[{"value": i, "label": i} for i in main_stats.index[3:]],
metric,
)
@app.callback(
Output("metric_graph", "figure"),
[
Input("window_width", "children"),
Input("stats_signal", "children"),
Input("metric_dropdown", "value"),
],
)
def update_metric_stats(window_width, stats_json, metric):
"""Once a new metric has been selected, plot its distribution."""
stats_dict = json.loads(stats_json)
height = int(9 / 21 * 2 / 3 * 2 / 3 * window_width)
return dict(
data=[
go.Box(
x=stats_dict["rand"][metric],
quartilemethod="linear",
jitter=0.3,
pointpos=1.8,
boxpoints="all",
boxmean="sd",
hoveron="points",
hovertemplate="%{x}<br>Random",
name="",
marker=dict(
color=color_schema["blue"],
opacity=0.5,
size=8,
),
),
go.Box(
x=stats_dict["hold"][metric],
quartilemethod="linear",
boxpoints="all",
jitter=0,
pointpos=1.8,
hoveron="points",
hovertemplate="%{x}<br>Buy & Hold",
fillcolor="rgba(0,0,0,0)",
line=dict(color="rgba(0,0,0,0)"),
name="",
marker=dict(
color=color_schema["orange"],
size=8,
),
),
go.Box(
x=stats_dict["main"][metric],
quartilemethod="linear",
boxpoints="all",
jitter=0,
pointpos=1.8,
hoveron="points",
hovertemplate="%{x}<br>Strategy",
fillcolor="rgba(0,0,0,0)",
line=dict(color="rgba(0,0,0,0)"),
name="",
marker=dict(
color=color_schema["green"],
size=8,
),
),
],
layout=merge_dicts(
default_layout,
dict(
height=max(350, height),
showlegend=False,
margin=dict(l=60, r=20, t=40, b=20),
hovermode="closest",
xaxis=dict(gridcolor=gridcolor, title=metric, side="top"),
yaxis=dict(gridcolor=gridcolor),
),
),
)
@app.callback(
[
Output("symbol_input", "value"),
Output("period_dropdown", "value"),
Output("interval_dropdown", "value"),
Output("yf_checklist", "value"),
Output("entry_n_random_input", "value"),
Output("exit_checklist", "value"),
Output("candle_settings_table", "data"),
Output("custom_entry_dropdown", "value"),
Output("custom_exit_dropdown", "value"),
Output("fees_input", "value"),
Output("fixed_fees_input", "value"),
Output("slippage_input", "value"),
Output("conflict_mode_dropdown", "value"),
Output("direction_dropdown", "value"),
Output("sim_checklist", "value"),
Output("n_random_strat_input", "value"),
Output("prob_checklist", "value"),
Output("stats_checklist", "value"),
],
[
Input("reset_button", "n_clicks"),
],
prevent_initial_call=True,
)
def reset_settings(_):
"""Reset most settings. Other settings are reset in their callbacks."""
return (
default_symbol,
default_period,
default_interval,
default_yf_options,
default_entry_n_random,
default_exit_options,
default_candle_settings.to_dict("records"),
default_entry_dates,
default_exit_dates,
default_fees,
default_fixed_fees,
default_slippage,
default_conflict_mode,
default_direction,
default_sim_options,
default_n_random_strat,
default_prob_options,
default_stats_options,
)
if __name__ == "__main__":
app.run(host=HOST, port=PORT, debug=DEBUG)
