-
Notifications
You must be signed in to change notification settings - Fork 4
VectrBT CheatSheet
David Brazda edited this page Oct 23, 2024
·
11 revisions
lightweightchart python, simple and fast basic charting excels in large data
Quick helpers to draw fast (supports ohlcv, lines, histograms, markers and n panes). Each pane containing tuples with respective structure:
- ohlv tuple
(series, entries, exits, other_markers) - histogram
[(series, name, colors)] - scaleId
[(series, name, entries, exits, other_markers)]
from lightweight_charts import chart, Panel
# Example usage
pane1 = Panel(
ohlcv=(t1data.data["BAC"],), #(series, entries, exits, other_markers)
histogram=[(order_imbalance_allvolume, "oivol")], # [(series, name, "rgba(53, 94, 59, 0.6)")]
#following attributes corresponds to different priceScaleId and allow to display
# line series on these scale
right=[], # [(series, name, entries, exits, other_markers)]
left=[(sma, "sma", short_signals, short_exits)],
middle1=[],
middle2=[],
)
pane2 = Panel(
ohlcv=(t1data.data["BAC"],),
right=[],
left=[(sma, "sma_below", short_signals, short_exits)],
middle1=[],
middle2=[],
histogram=[(order_imbalance_sma, "oisma")],
)
ch = chart([pane1, pane2], sync=True, title="neco")Detailed control over every aspect of their charts
chart = JupyterChart(width=1000, height=600, inner_width=1, inner_height=0.5, leftScale=True)
chart.topbar.textbox("title","Nadpis")
chart.set(t1data.data["BAC"]) #ohlcv data
#add line
line_vwap = chart.create_line(name="vwap") #picks color automatically, can be overiden by color="blue"
line_vwap.set(t1data.vwap)
#create subchart
chart2 = chart.create_subchart(position='right', width=1, height=0.5, sync=True, leftScale=True) # can by synces (requires same length arrays)
line1 = chart.create_line(name="5minclose")#, color="green")
line1.set(close_realigned)
#histograms
hst = chart2.create_histogram(name="buyvolume", color="rgba(53, 94, 59, 0.6)") #green transparent
hst1 = chart2.create_histogram(name="sellvolume", color="rgba(165, 42, 42, 0.6)") #red transparent
hst.set(t1data.data["BAC"])
hst1.set(t1data.data["BAC"])
#markers
line2.markers_set(short_signals, "entries") #style s "entries", can be implicitly specified by position,shape,color
line2.markers_set(short_exits, "exits") #input can be pdseries or df (one column or col_name parameter)
#others
chart.legend(True)
chart.fit()
chart2.fit()
chart.load() #for Jupyter, otherwise chart.show()Plotly used for various chart types but doesnt like large data.
- MAKE_SUBPLOT Defines layout (if more then 1x1 or secondary y axis are required)
fig = vbt.make_subplots(rows=2, cols=1, shared_xaxes=True,
specs=[[{"secondary_y": True}], [{"secondary_y": False}]],
vertical_spacing=0.02, subplot_titles=("Row 1 title", "Row 2 title"))Then the different sr/df generic accessor are added with ADD_TRACE_KWARGS and TRACE_KWARGS. Other types of plot available in plotting module
#using accessor
close.vbt.plot(fig=fig, add_trace_kwargs=dict(secondary_y=False,row=1, col=1), trace_kwargs=dict(line=dict(color="blue")))
indvolume.vbt.barplot(fig=fig, add_trace_kwargs=dict(secondary_y=False, row=2, col=1))
#using plotting module
vbt.Bar(indvolume, fig=fig, add_trace_kwargs=dict(secondary_y=False, row=2, col=1))- ADD_TRACE_KWARGS - determines positioning withing subplot
add_trace_kwargs=dict(secondary_y=False,row=1, col=1)- TRACE_KWARGS - other styling of trace
trace_kwargs=dict(name="LONGS",
line=dict(color="#ffe476"), #or rgb with opacity rgba(255, 0, 0, 0.5)
marker=dict(color="limegreen"),
fill=None,
connectgaps=True) fig = vbt.make_subplots(rows=2, cols=1, shared_xaxes=True,
specs=[[{"secondary_y": True}], [{"secondary_y": False}]],
vertical_spacing=0.02, subplot_titles=("Price and Indicators", "Volume"))
# Plotting the close price
close.vbt.plot(fig=fig, add_trace_kwargs=dict(secondary_y=False,row=1, col=1), trace_kwargs=dict(line=dict(color="blue")))Load from batch
from v2realbot.tools.loadbatch import load_batch
res, df = load_batch(batch_id="0fb5043a", #0fb5043a bde6d0be
space_resolution_evenly=False,
indicators_columns=["Rsi14"],
main_session_only=True,
verbose = False)
if res < 0:
print("Error" + str(res) + str(df))
df = df["bars"] #or indicators
#df
basic_data = vbt.Data.from_data(vbt.symbol_dict({"BAC": df}), tz_convert=zoneNY)
basic_data = basic_data.transform(lambda df: df.between_time('09:30', '16:00'))Load from Parquet (prepared by prepare_aggregated_data.ipynb)
#list all files is dir directory with parquet extension
dir = DATA_DIR + "/notebooks/"
import os
files = [f for f in os.listdir(dir) if f.endswith(".parquet")]
print('\n'.join(map(str, files)))
file_name = "ohlcv_df-BAC-2023-01-01T09_30_00-2024-05-25T15_30_00-47BCFOPUVWZ-100.parquet"
ohlcv_df = pd.read_parquet(dir+file_name,engine='pyarrow')
#filter ohlcv_df to certain date range (assuming datetime index)
ohlcv_df = ohlcv_df.loc["2024-02-12 10:30":"2024-02-14 12:00"]
#add vwap column to ohlcv_df
#ohlcv_df["hlcc4"] = (ohlcv_df["close"] + ohlcv_df["high"] + ohlcv_df["low"] + ohlcv_df["close"]) / 4
basic_data = vbt.Data.from_data(vbt.symbol_dict({"BAC": ohlcv_df}), tz_convert=zoneNY)
ohlcv_df= None #release memoryt1data = basic_data[['open', 'high', 'low', 'close', 'volume','vwap','buyvolume','sellvolume']].resample("1T")
t1data = t1data.transform(lambda df: df.between_time('09:30', '16:00').dropna()) #main session data only, no nans
t5data = basic_data[['open', 'high', 'low', 'close', 'volume','vwap','buyvolume','sellvolume']].resample("5T")
t5data = t5data.transform(lambda df: df.between_time('09:30', '16:00').dropna())
dailydata = basic_data[['open', 'high', 'low', 'close', 'volume', 'vwap']].resample("D").dropna()
#realign 5min close to 1min so it can be compared with 1min
t5data_close_realigned = t5data.close.vbt.realign_closing("1T").between_time('09:30', '16:00').dropna()
#same with open
t5data.open.vbt.realign_opening("1h")Example of custom feature config Binance Data. Other reduced functions available. (mean, min, max, median, nth ...)
from vectorbtpro.utils.config import merge_dicts, Config, HybridConfig
from vectorbtpro import _typing as tp
from vectorbtpro.generic import nb as generic_nb
_feature_config: tp.ClassVar[Config] = HybridConfig(
{
"buyvolume": dict(
resample_func=lambda self, obj, resampler: obj.vbt.resample_apply(
resampler,
generic_nb.sum_reduce_nb,
)
),
"sellvolume": dict(
resample_func=lambda self, obj, resampler: obj.vbt.resample_apply(
resampler,
generic_nb.sum_reduce_nb,
)
)
}
)
basic_data._feature_config = _feature_configt2dataclose = t2data.close.rename("15MIN - realigned").vbt.realign_closing("1T")
fig = t1data.close.rename("1MIN").vbt.plot()
t2data.close.rename("15MIN").vbt.plot(fig=fig)
t2dataclose.vbt.plot(fig=fig)basic_data.to_parquet(partition_by="day", compression="gzip")
day_data = vbt.ParquetData.pull("BAC", filters=[("group", "==", "2024-05-03")])
vbt.print_dir_tree("BTC-USD")#overeni directory structuret1data_filtered = t1data.xloc["2024-02-12 9:30":"2024-02-12 9:40"]
t5data = t5data.transform(lambda df: df.between_time('09:30', '16:00').dropna())vbt.phelp(vbt.talib(“atr”).run) #parameters it accepts
vbt.phelp(vbt.OLS.run) #indicator OLS
vbt.pdir(pf) - get available properties and methods
vbt.pprint(basic_data) #to get correct shape, info about instanceCreate indicator on several timeframes and realign with source timeframe(ready to compare)
RSI = vbt.talib("RSI").run(
t1data.get("Close"),
skipna=True,
timeframe=["1T", "5T", "15T", "30T", "1D"],
broadcast_kwargs=dict(wrapper_kwargs=dict(freq="1T")) #source frequency as out data have gaps
)
RSI["1T"].plot()
RSI.output_names # (real, )
rsicko = RSI.real.between_time('09:30', '16:00')
rsicko
rsi_df = RSI.real #(or RSI.unpack())bbands = vbt.talib("BBANDS").run(
t1data.get("Close"),
skipna=True,
timeframe=["1T", "5T", "15T", "30T", "1D"],
broadcast_kwargs=dict(wrapper_kwargs=dict(freq="1T")) #source frequency as out data have gaps
)
bandwidth = (bbands.upperband - bbands.lowerband) / bbands.middleband
bandwidth
bbands["5T"].plot() #plotting multiindexpf.orders.readable
pf.entry_trades.readable
pf.exit_trades.readable
pf.trades.readable
pf.positions.readable
pf.trade_history #human readable df expanding trades with metrics
dd = pf.get_drawdowns().records_readable
dd[dd["Status"] == "Active"] #Recovered
pf.metrics #get available metrics and its short names&function
pf.orders.side_buy.count()
pf.orders.stats(group_by=True)pf = vbt.Portfolio.from_signals(
sub_data,
entries=pd.Series([X, O, O, O, O, O, O]),
exits= pd.Series([O, O, O, X, O, O, O]),
price="nextopen", #fill na `close` current baru(deafult) nebo nextopen, lze i jako multiparameter vbt.Param(["close", "nextopen"])
slippage=0.005
)
>>> pf = vbt.PF.from_random_signals(
... vbt.YFData.pull("BTC-USD", start="2021-01", end="2021-02"),
... n=3,
... price=vbt.Param(["close", "nextopen"])
... )
>>> fig = pf.orders["close"].plot(
... buy_trace_kwargs=dict(name="Buy (close)", marker=dict(symbol="triangle-up-open")),
... sell_trace_kwargs=dict(name="Buy (close)", marker=dict(symbol="triangle-down-open"))
... )
>>> pf.orders["nextopen"].plot(
... plot_ohlc=False,
... plot_close=False,
... buy_trace_kwargs=dict(name="Buy (nextopen)"),
... sell_trace_kwargs=dict(name="Sell (nextopen)"),
... fig=fig
... )
>>> fig.show()pf.plot_trade_signals().show() #plot long short entries/exitsPlot Edge ratio pf.trades.plot_running_edge_ratio()
tp_stop = vbt.Param(tp_stop, condition="tp_stop > sl_stop") #conditional hyper parametersbt.parameterized(merge_func="concat")
def sma_crossover_perf(data, fast_window, slow_window):
fast_sma = data.run("sma", fast_window, short_name="fast_sma")
slow_sma = data.run("sma", slow_window, short_name="slow_sma")
entries = fast_sma.real_crossed_above(slow_sma)
exits = fast_sma.real_crossed_below(slow_sma)
pf = vbt.Portfolio.from_signals(
data, entries, exits, direction="both")
return pf.sharpe_ratio
#Let's test a grid of fast_window and slow_window combinations on one year of that data:
perf = sma_crossover_perf(
data["2020":"2020"],
vbt.Param(np.arange(5, 50), condition="x < slow_window"),
vbt.Param(np.arange(5, 50)),
_execute_kwargs=dict(
show_progress=True,
clear_cache=50,
collect_garbage=50
)
)
perf
symbol_wrapper = data.get_symbol_wrapper() #create same index structure
cash_deposits_dca = symbol_wrapper.fill(0).vbt.set(
10,
every="W",
start="2020-01-01"
) #fill it with zeros and every week set value 10
signal.vbt.xloc["04-26-2024":"04-29-2024"].get() #pdseries or df timeindex
signal.vbt.xloc[("BAC", "04-26-2024"):("BAC","04-29-2024")].get() #multiindex
entries.vbt.xloc["04-16-2024"].get() #one day
Plots primary/secondary indicators along with close
# priminds list (same Y as price)
# secinds list (secondary Y napr. rsi)
# close
# voluminds (volume based indicators) list
# ohlcv data
def plot_2y_close(priminds, secinds, close, volumeinds, ohlcv=None):
fig = vbt.make_subplots(rows=2, cols=1, shared_xaxes=True,
specs=[[{"secondary_y": True}], [{"secondary_y": False}]],
vertical_spacing=0.02, subplot_titles=("Price and Indicators", "Volume"))
if ohlcv is not None:
ohlcv.vbt.ohlcv.plot(fig=fig, add_trace_kwargs=dict(row=1, col=1))
# Plotting the close price
close.vbt.plot(fig=fig, add_trace_kwargs=dict(secondary_y=False,row=1, col=1), trace_kwargs=dict(line=dict(color="blue")))
# Plotting primary indicators on the first row
for ind in priminds:
if isinstance(ind, pd.Series):
#if series has no name, make the name same as the variable name
ind = ind.vbt
ind.plot(fig=fig, add_trace_kwargs=dict(secondary_y=False, row=1, col=1))
# Plotting secondary indicators on the first row
for ind in secinds:
#ind = ind.rename(str(ind.name))
if isinstance(ind, pd.Series):
ind = ind.vbt
ind.plot(fig=fig, add_trace_kwargs=dict(secondary_y=True, row=1, col=1), trace_kwargs=dict(line=dict(color="rgba(255, 0, 0, 0.4)")))
for indvolume in volumeinds:
# Plotting the volume on the second row
indvolume.rename(str(indvolume.name)).vbt.barplot(fig=fig, add_trace_kwargs=dict(secondary_y=False, row=2, col=1))
#vbt.Bar(indvolume, fig=fig, add_trace_kwargs=dict(secondary_y=False, row=2, col=1))
return fig
fig = plot_2y_close([], [order_imbalance.rename("order_imbalance_norm")], t1data.close, [t1data.data["BAC"].buyvolume, t1data.data["BAC"].sellvolume, t1data.volume], t1data.data["BAC"])
fig.update_yaxes(range=[32.5,33], secondary_y=False) #update y axis range
fig.update_yaxes(range=[-1,1], secondary_y=True)