Visualisation

API documentation for tradeexecutor.state.visualisation.Visualisation Python class in Trading Strategy framework.

class Visualisation

Bases: object

Strategy visualisation helper.

This object is returned from the strategy execution cycle. It allows you to plot values, add debug messages, etc. It is not used in any trading, but can help and visualize trade backtesting and execution.

• See plot_indicator() for how to draw plots inside decide_trades()

• See get_series() how to extract data for charting as pd.Series.

__init__(messages=<factory>, calculations=<factory>, plots=<factory>, pair_ids=<factory>, discardable_data=<factory>)

Parameters:

• messages (Dict[int, List[str]]) –

• calculations (Dict[int, dict]) –

• plots (Dict[str, Plot]) –

• pair_ids (list[int]) –

• discardable_data (Dict[str, Any]) –

Return type:

None

Methods

__init__([messages, calculations, plots, ...])
add_calculations(timestamp, cycle_calculations)	Update strategy cycle calculations diagnostics.
add_message(timestamp, content)	Write a debug message.
from_dict(kvs, *[, infer_missing])
from_json(s, *[, parse_float, parse_int, ...])
get_messages_tail(count[, filter_func])	Get N latest messages.
get_series(name)	Get plot data for charts and tables.
get_timestamp_range([plot_name])	Get the time range for which we have data.
get_total_points()	Get number of data points stored in all plots.
plot_indicator(timestamp, name, kind, value)	Add a value to the output data and diagram.
schema(*[, infer_missing, only, exclude, ...])
set_discardable_data(key, value)	Set data for which we only keep the last set.
set_visualised_pairs(pairs)	Set the trading pair for this plot.
to_dict([encode_json])
to_json(*[, skipkeys, ensure_ascii, ...])

Attributes

messages	Messages for each strategy cycle.
calculations	Extra calculation diagnostics for each strategy cycle.
plots	Name -> Plot value mappings
pair_ids	Trading pair IDs to visualise instead of the default trading pairs from strategy universe.
discardable_data	Data for which we only keep the last value.

messages: Dict[int, List[str]]

Messages for each strategy cycle.

Because we cannot use datetime.datetime directly as a key in JSON, we use UNIX timestamp here to keep our state easily serialisable.

calculations: Dict[int, dict]

Extra calculation diagnostics for each strategy cycle.

Cycle -> dict of values mappings.

Currently used to record the alpha model state when doing doing portfolio construction modelling.

Because we cannot use datetime.datetime directly as a key in JSON, we use UNIX timestamp here to keep our state easily serialisable.

plots: Dict[str, Plot]

Name -> Plot value mappings

pair_ids: list[int]

Trading pair IDs to visualise instead of the default trading pairs from strategy universe.

discardable_data: Dict[str, Any]

Data for which we only keep the last value.

Most useful in unit test/backtest debugging.

See set_discardable_data().

set_visualised_pairs(pairs)

Set the trading pair for this plot.

Parameters:

pairs (list[tradeexecutor.state.identifier.TradingPairIdentifier]) – Use these trading pairs to plot candle charts instead of the default trading pairs from strategy universe.

Return type:

None

add_message(timestamp, content)

Write a debug message.

• Each message is associated to a different timepoint

• Multiple messages per timepoint are possible

• The first message is in strategy thinking Discord/Telegram logging output for each cycle

To display in backtesting notebook:

For example see get_messages_tail().

Parameters:

• timestamp (datetime) – The current strategy cycle timestamp

• content (str) – The contents of the message

get_messages_tail(count, filter_func=None)

Get N latest messages.

• If there are multiple messages per timepoint get only one.

Example:

# Find rebalance event
def find_rebalance_log_messages(msg: str):
    return "Rebalanced: 👍" in msg

messages = state.visualisation.get_messages_tail(10, filter_func=find_rebalance_log_messages)

table = pd.Series(
    data=list(messages.values()),
    index=list(messages.keys()),
)

df = table.to_frame()

display(df.style.set_properties(**{
    'text-align': 'left',
    'white-space': 'pre-wrap',
}))

Parameters:

• count (int) – How many last messages to return.

• filter_func (Callable) –

  Match messages against this filter.

  Matches the first message for a timepoint.

Return type:

dict[datetime.datetime, str]

add_calculations(timestamp, cycle_calculations)

Update strategy cycle calculations diagnostics.

• Each strategy cycle can dump whatever intermediate calculations state on the visualisation record keeping, so that it can be later pulled up in the analysis.

• Currently this is used to store the alpha model calculations for portfolio construction model.

Note

Using this method may slow down your backtests because serialising cycle_calculations might be slow. Avoid if not needed.

Parameters:

• timestamp (datetime) – The current strategy cycle timestamp

• cycle_calculations (dict) –

  The contents of the calculations.

  Must be JSON serialisable dict.

plot_indicator(timestamp, name, kind, value, colour=None, plot_shape=PlotShape.linear, detached_overlay_name=None, indicator_size=None, recording_time=RecordingTime.decision_making_time, pair=None, label=PlotLabel.axis, height=None)

Add a value to the output data and diagram.

Plots are stored by their name.

Example how to draw a detached RSI indicator and top/bottom indicator line for it:

# Current daily
visualisation.plot_indicator(
    timestamp,
    f"RSI {token}",
    PlotKind.technical_indicator_detached,
    current_rsi_values[pair],
)

# Low (vertical line)
visualisation.plot_indicator(
    timestamp,
    f"RSI {token} low trigger",
    PlotKind.technical_indicator_overlay_on_detached,
    rsi_low,
    detached_overlay_name=f"RSI {token}",
    plot_shape=PlotShape.horizontal_vertical,
)

# High (vertical line)
visualisation.plot_indicator(
    timestamp,
    f"RSI {token} high trigger",
    PlotKind.technical_indicator_overlay_on_detached,
    rsi_high,
    detached_overlay_name=f"RSI {token}",
    plot_shape=PlotShape.horizontal_vertical,
)

Parameters:

• timestamp (Union[datetime, Timestamp]) – The current strategy cycle timestamp

• name (str) – The plot label

• kind (PlotKind) – The plot type

• value (float) – Current value e.g. price as USD

• colour (Optional[str]) – Optional colour

• plot_shape (Optional[PlotShape]) – PlotShape enum value e.g. Plotshape.linear or Plotshape.horizontal_vertical

• detached_overlay_name (Optional[str]) – If this plot is overlayed on top of a detached technical indicator, this is the name of the overlay it should be attached to.

• indicator_size (Optional[float]) – Optional indicator to determine the size of the indicator. For a line, this is the width of the line. For a marker, this is the size of the marker.

• recording_time (Optional[RecordingTime]) – Optional recording time to determine when the plot should be recorded. For example, if you want to record the plot at the decision making time, you can set this to RecordingTime.decision_making_time. Default is RecordingTime.decision_making_time.

• label (PlotLabel) –

  How to render the label for this plot.

  The last set value is effective.

• height (Optional[int]) –

  Currently not supported.

  Plotly does not support setting heights of individual subplots. Instead, you can adjust the overall plotly.Figure size in pixels and then % of subplot height in them.

• pair (Optional[TradingPairIdentifier]) –

__init__(messages=<factory>, calculations=<factory>, plots=<factory>, pair_ids=<factory>, discardable_data=<factory>)

Parameters:

• messages (Dict[int, List[str]]) –

• calculations (Dict[int, dict]) –

• plots (Dict[str, Plot]) –

• pair_ids (list[int]) –

• discardable_data (Dict[str, Any]) –

Return type:

None

get_timestamp_range(plot_name=None)

Get the time range for which we have data.

Parameters:

plot_name (Optional[str]) –

Use range from a specific plot.

If not given use the first plot.

Returns:

UTC started at, ended at.

Return None, None if no data.

Return type:

Tuple[Optional[datetime], Optional[datetime]]

get_total_points()

Get number of data points stored in all plots.

Return type:

int

get_series(name)

Get plot data for charts and tables.

Example from a backtesting notebook:

import plotly.express as px

visualisation = state.visualisation

df = pd.DataFrame({
    "assets_available": visualisation.get_series("assets_available"),
    "assets_chosen": visualisation.get_series("assets_chosen")
})

fig = px.line(df, title='Assets tradeable vs. chosen to the basket')
fig.update_yaxes(title="Number of assets")
fig.update_xaxes(title="Time")
fig.show()

Parameters:

name (str) – Same as in plot_indicator()

Returns:

Pandas series with DateTimeIndex and float values.

Return type:

Series

set_discardable_data(key, value)

Set data for which we only keep the last set.