freqtrade_origin/freqtrade/plot/plotting.py

import logging
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

from freqtrade.configuration import TimeRange
from freqtrade.constants import Config
from freqtrade.data.btanalysis import (
    analyze_trade_parallelism,
    extract_trades_of_period,
    load_trades,
)
from freqtrade.data.converter import trim_dataframe
from freqtrade.data.dataprovider import DataProvider
from freqtrade.data.history import get_timerange, load_data
from freqtrade.data.metrics import (
    calculate_max_drawdown,
    calculate_underwater,
    combine_dataframes_with_mean,
    create_cum_profit,
)
from freqtrade.enums import CandleType
from freqtrade.exceptions import OperationalException
from freqtrade.exchange import timeframe_to_prev_date, timeframe_to_seconds
from freqtrade.misc import pair_to_filename
from freqtrade.plugins.pairlist.pairlist_helpers import expand_pairlist
from freqtrade.resolvers import ExchangeResolver, StrategyResolver
from freqtrade.strategy import IStrategy
from freqtrade.strategy.strategy_wrapper import strategy_safe_wrapper


logger = logging.getLogger(__name__)


try:
    import plotly.graph_objects as go
    from plotly.offline import plot
    from plotly.subplots import make_subplots
except ImportError:
    logger.exception("Module plotly not found \n Please install using `pip3 install plotly`")
    exit(1)


def init_plotscript(config, markets: list, startup_candles: int = 0):
    """
    Initialize objects needed for plotting
    :return: Dict with candle (OHLCV) data, trades and pairs
    """

    if "pairs" in config:
        pairs = expand_pairlist(config["pairs"], markets)
    else:
        pairs = expand_pairlist(config["exchange"]["pair_whitelist"], markets)

    # Set timerange to use
    timerange = TimeRange.parse_timerange(config.get("timerange"))

    data = load_data(
        datadir=config.get("datadir"),
        pairs=pairs,
        timeframe=config["timeframe"],
        timerange=timerange,
        startup_candles=startup_candles,
        data_format=config["dataformat_ohlcv"],
        candle_type=config.get("candle_type_def", CandleType.SPOT),
    )

    if startup_candles and data:
        min_date, max_date = get_timerange(data)
        logger.info(f"Loading data from {min_date} to {max_date}")
        timerange.adjust_start_if_necessary(
            timeframe_to_seconds(config["timeframe"]), startup_candles, min_date
        )

    no_trades = False
    filename = config.get("exportfilename")
    if config.get("no_trades", False):
        no_trades = True
    elif config["trade_source"] == "file":
        if not filename.is_dir() and not filename.is_file():
            logger.warning("Backtest file is missing skipping trades.")
            no_trades = True
    try:
        trades = load_trades(
            config["trade_source"],
            db_url=config.get("db_url"),
            exportfilename=filename,
            no_trades=no_trades,
            strategy=config.get("strategy"),
        )
    except ValueError as e:
        raise OperationalException(e) from e
    if not trades.empty:
        trades = trim_dataframe(trades, timerange, df_date_col="open_date")

    return {
        "ohlcv": data,
        "trades": trades,
        "pairs": pairs,
        "timerange": timerange,
    }


def add_indicators(fig, row, indicators: dict[str, dict], data: pd.DataFrame) -> make_subplots:
    """
    Generate all the indicators selected by the user for a specific row, based on the configuration
    :param fig: Plot figure to append to
    :param row: row number for this plot
    :param indicators: Dict of Indicators with configuration options.
                       Dict key must correspond to dataframe column.
    :param data: candlestick DataFrame
    """
    plot_kinds = {
        "scatter": go.Scatter,
        "bar": go.Bar,
    }
    for indicator, conf in indicators.items():
        logger.debug(f"indicator {indicator} with config {conf}")
        if indicator in data:
            kwargs = {"x": data["date"], "y": data[indicator].values, "name": indicator}

            plot_type = conf.get("type", "scatter")
            color = conf.get("color")
            if plot_type == "bar":
                kwargs.update(
                    {
                        "marker_color": color or "DarkSlateGrey",
                        "marker_line_color": color or "DarkSlateGrey",
                    }
                )
            else:
                if color:
                    kwargs.update({"line": {"color": color}})
                kwargs["mode"] = "lines"
                if plot_type != "scatter":
                    logger.warning(
                        f"Indicator {indicator} has unknown plot trace kind {plot_type}"
                        f', assuming "scatter".'
                    )

            kwargs.update(conf.get("plotly", {}))
            trace = plot_kinds[plot_type](**kwargs)
            fig.add_trace(trace, row, 1)
        else:
            logger.info(
                'Indicator "%s" ignored. Reason: This indicator is not found ' "in your strategy.",
                indicator,
            )

    return fig


def add_profit(fig, row, data: pd.DataFrame, column: str, name: str) -> make_subplots:
    """
    Add profit-plot
    :param fig: Plot figure to append to
    :param row: row number for this plot
    :param data: candlestick DataFrame
    :param column: Column to use for plot
    :param name: Name to use
    :return: fig with added profit plot
    """
    profit = go.Scatter(
        x=data.index,
        y=data[column],
        name=name,
    )
    fig.add_trace(profit, row, 1)

    return fig


def add_max_drawdown(
    fig, row, trades: pd.DataFrame, df_comb: pd.DataFrame, timeframe: str, starting_balance: float
) -> make_subplots:
    """
    Add scatter points indicating max drawdown
    """
    try:
        drawdown = calculate_max_drawdown(trades, starting_balance=starting_balance)

        drawdown = go.Scatter(
            x=[drawdown.high_date, drawdown.low_date],
            y=[
                df_comb.loc[timeframe_to_prev_date(timeframe, drawdown.high_date), "cum_profit"],
                df_comb.loc[timeframe_to_prev_date(timeframe, drawdown.low_date), "cum_profit"],
            ],
            mode="markers",
            name=f"Max drawdown {drawdown.relative_account_drawdown:.2%}",
            text=f"Max drawdown {drawdown.relative_account_drawdown:.2%}",
            marker=dict(symbol="square-open", size=9, line=dict(width=2), color="green"),
        )
        fig.add_trace(drawdown, row, 1)
    except ValueError:
        logger.warning("No trades found - not plotting max drawdown.")
    return fig


def add_underwater(fig, row, trades: pd.DataFrame, starting_balance: float) -> make_subplots:
    """
    Add underwater plots
    """
    try:
        underwater = calculate_underwater(
            trades, value_col="profit_abs", starting_balance=starting_balance
        )

        underwater_plot = go.Scatter(
            x=underwater["date"],
            y=underwater["drawdown"],
            name="Underwater Plot",
            fill="tozeroy",
            fillcolor="#cc362b",
            line={"color": "#cc362b"},
        )

        underwater_plot_relative = go.Scatter(
            x=underwater["date"],
            y=(-underwater["drawdown_relative"]),
            name="Underwater Plot (%)",
            fill="tozeroy",
            fillcolor="green",
            line={"color": "green"},
        )

        fig.add_trace(underwater_plot, row, 1)
        fig.add_trace(underwater_plot_relative, row + 1, 1)
    except ValueError:
        logger.warning("No trades found - not plotting underwater plot")
    return fig


def add_parallelism(fig, row, trades: pd.DataFrame, timeframe: str) -> make_subplots:
    """
    Add Chart showing trade parallelism
    """
    try:
        result = analyze_trade_parallelism(trades, timeframe)

        drawdown = go.Scatter(
            x=result.index,
            y=result["open_trades"],
            name="Parallel trades",
            fill="tozeroy",
            fillcolor="#242222",
            line={"color": "#242222"},
        )
        fig.add_trace(drawdown, row, 1)
    except ValueError:
        logger.warning("No trades found - not plotting Parallelism.")
    return fig


def plot_trades(fig, trades: pd.DataFrame) -> make_subplots:
    """
    Add trades to "fig"
    """
    # Trades can be empty
    if trades is not None and len(trades) > 0:
        # Create description for exit summarizing the trade
        trades["desc"] = trades.apply(
            lambda row: f"{row['profit_ratio']:.2%}, "
            + (f"{row['enter_tag']}, " if row["enter_tag"] is not None else "")
            + f"{row['exit_reason']}, "
            + f"{row['trade_duration']} min",
            axis=1,
        )
        trade_entries = go.Scatter(
            x=trades["open_date"],
            y=trades["open_rate"],
            mode="markers",
            name="Trade entry",
            text=trades["desc"],
            marker=dict(symbol="circle-open", size=11, line=dict(width=2), color="cyan"),
        )

        trade_exits = go.Scatter(
            x=trades.loc[trades["profit_ratio"] > 0, "close_date"],
            y=trades.loc[trades["profit_ratio"] > 0, "close_rate"],
            text=trades.loc[trades["profit_ratio"] > 0, "desc"],
            mode="markers",
            name="Exit - Profit",
            marker=dict(symbol="square-open", size=11, line=dict(width=2), color="green"),
        )
        trade_exits_loss = go.Scatter(
            x=trades.loc[trades["profit_ratio"] <= 0, "close_date"],
            y=trades.loc[trades["profit_ratio"] <= 0, "close_rate"],
            text=trades.loc[trades["profit_ratio"] <= 0, "desc"],
            mode="markers",
            name="Exit - Loss",
            marker=dict(symbol="square-open", size=11, line=dict(width=2), color="red"),
        )
        fig.add_trace(trade_entries, 1, 1)
        fig.add_trace(trade_exits, 1, 1)
        fig.add_trace(trade_exits_loss, 1, 1)
    else:
        logger.warning("No trades found.")
    return fig


def create_plotconfig(
    indicators1: list[str], indicators2: list[str], plot_config: dict[str, dict]
) -> dict[str, dict]:
    """
    Combines indicators 1 and indicators 2 into plot_config if necessary
    :param indicators1: List containing Main plot indicators
    :param indicators2: List containing Sub plot indicators
    :param plot_config: Dict of Dicts containing advanced plot configuration
    :return: plot_config - eventually with indicators 1 and 2
    """

    if plot_config:
        if indicators1:
            plot_config["main_plot"] = {ind: {} for ind in indicators1}
        if indicators2:
            plot_config["subplots"] = {"Other": {ind: {} for ind in indicators2}}

    if not plot_config:
        # If no indicators and no plot-config given, use defaults.
        if not indicators1:
            indicators1 = ["sma", "ema3", "ema5"]
        if not indicators2:
            indicators2 = ["macd", "macdsignal"]

        # Create subplot configuration if plot_config is not available.
        plot_config = {
            "main_plot": {ind: {} for ind in indicators1},
            "subplots": {"Other": {ind: {} for ind in indicators2}},
        }
    if "main_plot" not in plot_config:
        plot_config["main_plot"] = {}

    if "subplots" not in plot_config:
        plot_config["subplots"] = {}
    return plot_config


def plot_area(
    fig,
    row: int,
    data: pd.DataFrame,
    indicator_a: str,
    indicator_b: str,
    label: str = "",
    fill_color: str = "rgba(0,176,246,0.2)",
) -> make_subplots:
    """Creates a plot for the area between two traces and adds it to fig.
    :param fig: Plot figure to append to
    :param row: row number for this plot
    :param data: candlestick DataFrame
    :param indicator_a: indicator name as populated in strategy
    :param indicator_b: indicator name as populated in strategy
    :param label: label for the filled area
    :param fill_color: color to be used for the filled area
    :return: fig with added  filled_traces plot
    """
    if indicator_a in data and indicator_b in data:
        # make lines invisible to get the area plotted, only.
        line = {"color": "rgba(255,255,255,0)"}
        # TODO: Figure out why scattergl causes problems plotly/plotly.js#2284
        trace_a = go.Scatter(x=data.date, y=data[indicator_a], showlegend=False, line=line)
        trace_b = go.Scatter(
            x=data.date,
            y=data[indicator_b],
            name=label,
            fill="tonexty",
            fillcolor=fill_color,
            line=line,
        )
        fig.add_trace(trace_a, row, 1)
        fig.add_trace(trace_b, row, 1)
    return fig


def add_areas(fig, row: int, data: pd.DataFrame, indicators) -> make_subplots:
    """Adds all area plots (specified in plot_config) to fig.
    :param fig: Plot figure to append to
    :param row: row number for this plot
    :param data: candlestick DataFrame
    :param indicators: dict with indicators. ie.: plot_config['main_plot'] or
                            plot_config['subplots'][subplot_label]
    :return: fig with added  filled_traces plot
    """
    for indicator, ind_conf in indicators.items():
        if "fill_to" in ind_conf:
            indicator_b = ind_conf["fill_to"]
            if indicator in data and indicator_b in data:
                label = ind_conf.get("fill_label", f"{indicator}<>{indicator_b}")
                fill_color = ind_conf.get("fill_color", "rgba(0,176,246,0.2)")
                fig = plot_area(
                    fig, row, data, indicator, indicator_b, label=label, fill_color=fill_color
                )
            elif indicator not in data:
                logger.info(
                    'Indicator "%s" ignored. Reason: This indicator is not '
                    "found in your strategy.",
                    indicator,
                )
            elif indicator_b not in data:
                logger.info(
                    'fill_to: "%s" ignored. Reason: This indicator is not ' "in your strategy.",
                    indicator_b,
                )
    return fig


def create_scatter(data, column_name, color, direction) -> go.Scatter | None:
    if column_name in data.columns:
        df_short = data[data[column_name] == 1]
        if len(df_short) > 0:
            shorts = go.Scatter(
                x=df_short.date,
                y=df_short.close,
                mode="markers",
                name=column_name,
                marker=dict(
                    symbol=f"triangle-{direction}-dot",
                    size=9,
                    line=dict(width=1),
                    color=color,
                ),
            )
            return shorts
        else:
            logger.warning(f"No {column_name}-signals found.")

    return None


def generate_candlestick_graph(
    pair: str,
    data: pd.DataFrame,
    trades: pd.DataFrame | None = None,
    *,
    indicators1: list[str] | None = None,
    indicators2: list[str] | None = None,
    plot_config: dict[str, dict] | None = None,
) -> go.Figure:
    """
    Generate the graph from the data generated by Backtesting or from DB
    Volume will always be plotted in row2, so Row 1 and 3 are to our disposal for custom indicators
    :param pair: Pair to Display on the graph
    :param data: OHLCV DataFrame containing indicators and entry/exit signals
    :param trades: All trades created
    :param indicators1: List containing Main plot indicators
    :param indicators2: List containing Sub plot indicators
    :param plot_config: Dict of Dicts containing advanced plot configuration
    :return: Plotly figure
    """
    plot_config = create_plotconfig(
        indicators1 or [],
        indicators2 or [],
        plot_config or {},
    )
    rows = 2 + len(plot_config["subplots"])
    row_widths = [1 for _ in plot_config["subplots"]]
    # Define the graph
    fig = make_subplots(
        rows=rows,
        cols=1,
        shared_xaxes=True,
        row_width=row_widths + [1, 4],
        vertical_spacing=0.0001,
    )
    fig["layout"].update(title=pair)
    fig["layout"]["yaxis1"].update(title="Price")
    fig["layout"]["yaxis2"].update(title="Volume")
    for i, name in enumerate(plot_config["subplots"]):
        fig["layout"][f"yaxis{3 + i}"].update(title=name)
    fig["layout"]["xaxis"]["rangeslider"].update(visible=False)
    fig.update_layout(modebar_add=["v1hovermode", "toggleSpikeLines"])

    # Common information
    candles = go.Candlestick(
        x=data.date, open=data.open, high=data.high, low=data.low, close=data.close, name="Price"
    )
    fig.add_trace(candles, 1, 1)

    longs = create_scatter(data, "enter_long", "green", "up")
    exit_longs = create_scatter(data, "exit_long", "red", "down")
    shorts = create_scatter(data, "enter_short", "blue", "down")
    exit_shorts = create_scatter(data, "exit_short", "violet", "up")

    for scatter in [longs, exit_longs, shorts, exit_shorts]:
        if scatter:
            fig.add_trace(scatter, 1, 1)

    # Add Bollinger Bands
    fig = plot_area(fig, 1, data, "bb_lowerband", "bb_upperband", label="Bollinger Band")
    # prevent bb_lower and bb_upper from plotting
    try:
        del plot_config["main_plot"]["bb_lowerband"]
        del plot_config["main_plot"]["bb_upperband"]
    except KeyError:
        pass
    # main plot goes to row 1
    fig = add_indicators(fig=fig, row=1, indicators=plot_config["main_plot"], data=data)
    fig = add_areas(fig, 1, data, plot_config["main_plot"])
    fig = plot_trades(fig, trades)
    # sub plot: Volume goes to row 2
    volume = go.Bar(
        x=data["date"],
        y=data["volume"],
        name="Volume",
        marker_color="DarkSlateGrey",
        marker_line_color="DarkSlateGrey",
    )
    fig.add_trace(volume, 2, 1)
    # add each sub plot to a separate row
    for i, label in enumerate(plot_config["subplots"]):
        sub_config = plot_config["subplots"][label]
        row = 3 + i
        fig = add_indicators(fig=fig, row=row, indicators=sub_config, data=data)
        # fill area between indicators ( 'fill_to': 'other_indicator')
        fig = add_areas(fig, row, data, sub_config)

    return fig


def generate_profit_graph(
    pairs: str,
    data: dict[str, pd.DataFrame],
    trades: pd.DataFrame,
    timeframe: str,
    stake_currency: str,
    starting_balance: float,
) -> go.Figure:
    # Combine close-values for all pairs, rename columns to "pair"
    try:
        df_comb = combine_dataframes_with_mean(data, "close")
    except ValueError:
        raise OperationalException(
            "No data found. Please make sure that data is available for "
            "the timerange and pairs selected."
        )

    # Trim trades to available OHLCV data
    trades = extract_trades_of_period(df_comb, trades, date_index=True)
    if len(trades) == 0:
        raise OperationalException("No trades found in selected timerange.")

    # Add combined cumulative profit
    df_comb = create_cum_profit(df_comb, trades, "cum_profit", timeframe)

    # Plot the pairs average close prices, and total profit growth
    avgclose = go.Scatter(
        x=df_comb.index,
        y=df_comb["mean"],
        name="Avg close price",
    )

    fig = make_subplots(
        rows=6,
        cols=1,
        shared_xaxes=True,
        row_heights=[1, 1, 1, 0.5, 0.75, 0.75],
        vertical_spacing=0.05,
        subplot_titles=[
            "AVG Close Price",
            "Combined Profit",
            "Profit per pair",
            "Parallelism",
            "Underwater",
            "Relative Drawdown",
        ],
    )
    fig["layout"].update(title="Freqtrade Profit plot")
    fig["layout"]["yaxis1"].update(title="Price")
    fig["layout"]["yaxis2"].update(title=f"Profit {stake_currency}")
    fig["layout"]["yaxis3"].update(title=f"Profit {stake_currency}")
    fig["layout"]["yaxis4"].update(title="Trade count")
    fig["layout"]["yaxis5"].update(title="Underwater Plot")
    fig["layout"]["yaxis6"].update(title="Underwater Plot Relative (%)", tickformat=",.2%")
    fig["layout"]["xaxis"]["rangeslider"].update(visible=False)
    fig.update_layout(modebar_add=["v1hovermode", "toggleSpikeLines"])

    fig.add_trace(avgclose, 1, 1)
    fig = add_profit(fig, 2, df_comb, "cum_profit", "Profit")
    fig = add_max_drawdown(fig, 2, trades, df_comb, timeframe, starting_balance)
    fig = add_parallelism(fig, 4, trades, timeframe)
    # Two rows consumed
    fig = add_underwater(fig, 5, trades, starting_balance)

    for pair in pairs:
        profit_col = f"cum_profit_{pair}"
        try:
            df_comb = create_cum_profit(
                df_comb, trades[trades["pair"] == pair], profit_col, timeframe
            )
            fig = add_profit(fig, 3, df_comb, profit_col, f"Profit {pair}")
        except ValueError:
            pass
    return fig


def generate_plot_filename(pair: str, timeframe: str) -> str:
    """
    Generate filenames per pair/timeframe to be used for storing plots
    """
    pair_s = pair_to_filename(pair)
    file_name = "freqtrade-plot-" + pair_s + "-" + timeframe + ".html"

    logger.info("Generate plot file for %s", pair)

    return file_name


def store_plot_file(fig, filename: str, directory: Path, auto_open: bool = False) -> None:
    """
    Generate a plot html file from pre populated fig plotly object
    :param fig: Plotly Figure to plot
    :param filename: Name to store the file as
    :param directory: Directory to store the file in
    :param auto_open: Automatically open files saved
    :return: None
    """
    directory.mkdir(parents=True, exist_ok=True)

    _filename = directory.joinpath(filename)
    plot(fig, filename=str(_filename), auto_open=auto_open)
    logger.info(f"Stored plot as {_filename}")


def load_and_plot_trades(config: Config):
    """
    From configuration provided
    - Initializes plot-script
    - Get candle (OHLCV) data
    - Generate Dafaframes populated with indicators and signals based on configured strategy
    - Load trades executed during the selected period
    - Generate Plotly plot objects
    - Generate plot files
    :return: None
    """
    strategy = StrategyResolver.load_strategy(config)

    exchange = ExchangeResolver.load_exchange(config)
    IStrategy.dp = DataProvider(config, exchange)
    strategy.ft_bot_start()
    strategy_safe_wrapper(strategy.bot_loop_start)(current_time=datetime.now(timezone.utc))
    plot_elements = init_plotscript(config, list(exchange.markets), strategy.startup_candle_count)
    timerange = plot_elements["timerange"]
    trades = plot_elements["trades"]
    pair_counter = 0
    for pair, data in plot_elements["ohlcv"].items():
        pair_counter += 1
        logger.info("analyse pair %s", pair)

        df_analyzed = strategy.analyze_ticker(data, {"pair": pair})
        df_analyzed = trim_dataframe(df_analyzed, timerange)
        if not trades.empty:
            trades_pair = trades.loc[trades["pair"] == pair]
            trades_pair = extract_trades_of_period(df_analyzed, trades_pair)
        else:
            trades_pair = trades

        fig = generate_candlestick_graph(
            pair=pair,
            data=df_analyzed,
            trades=trades_pair,
            indicators1=config.get("indicators1", []),
            indicators2=config.get("indicators2", []),
            plot_config=strategy.plot_config if hasattr(strategy, "plot_config") else {},
        )

        store_plot_file(
            fig,
            filename=generate_plot_filename(pair, config["timeframe"]),
            directory=config["user_data_dir"] / "plot",
        )

    logger.info("End of plotting process. %s plots generated", pair_counter)


def plot_profit(config: Config) -> None:
    """
    Plots the total profit for all pairs.
    Note, the profit calculation isn't realistic.
    But should be somewhat proportional, and therefore useful
    in helping out to find a good algorithm.
    """
    if "timeframe" not in config:
        raise OperationalException("Timeframe must be set in either config or via --timeframe.")

    exchange = ExchangeResolver.load_exchange(config)
    plot_elements = init_plotscript(config, list(exchange.markets))
    trades = plot_elements["trades"]
    # Filter trades to relevant pairs
    # Remove open pairs - we don't know the profit yet so can't calculate profit for these.
    # Also, If only one open pair is left, then the profit-generation would fail.
    trades = trades[
        (trades["pair"].isin(plot_elements["pairs"])) & (~trades["close_date"].isnull())
    ]
    if len(trades) == 0:
        raise OperationalException(
            "No trades found, cannot generate Profit-plot without "
            "trades from either Backtest result or database."
        )

    # Create an average close price of all the pairs that were involved.
    # this could be useful to gauge the overall market trend
    fig = generate_profit_graph(
        plot_elements["pairs"],
        plot_elements["ohlcv"],
        trades,
        config["timeframe"],
        config.get("stake_currency", ""),
        config.get("available_capital", config["dry_run_wallet"]),
    )
    store_plot_file(
        fig,
        filename="freqtrade-profit-plot.html",
        directory=config["user_data_dir"] / "plot",
        auto_open=config.get("plot_auto_open", False),
    )