freqtrade_origin/scripts/plot_dataframe.py

#!/usr/bin/env python3
"""
Script to display when the bot will buy a specific pair

Mandatory Cli parameters:
-p / --pair: pair to examine

Optional Cli parameters
-s / --strategy: strategy to use
-d / --datadir: path to pair backtest data
--timerange: specify what timerange of data to use.
-l / --live: Live, to download the latest ticker for the pair
-db / --db-url: Show trades stored in database
"""
import logging
import os
import sys
from argparse import Namespace
from typing import Dict, List, Any
from sqlalchemy import create_engine
from plotly import tools
from plotly.offline import plot
import plotly.graph_objs as go
from freqtrade.arguments import Arguments
from freqtrade.analyze import Analyze
from freqtrade.optimize.backtesting import setup_configuration
from freqtrade import exchange
import freqtrade.optimize as optimize
from freqtrade import persistence
from freqtrade.persistence import Trade

logger = logging.getLogger(__name__)
_CONF: Dict[str, Any] = {}


# Update the global variable TO_DISPLAY to select which indicator you want to display
TO_DISPLAY = {
    'sma': False,  # On Row 1
    'ema': True,  # On Row 1

    'macd': True,  # On Row 3
    'rsi': False,  # On Row 3
    'fisher_rsi': False,  # On Row 3
    'mfi': False,  # On Row 3
    'slow': False,  # On Row 3
    'fast': False  # On Row 3
}


def plot_analyzed_dataframe(args: Namespace) -> None:
    """
    Calls analyze() and plots the returned dataframe
    :return: None
    """

    # Load the configuration
    config = setup_configuration(args)

    # Set the pair to audit
    pair = args.pair

    # Set timerange to use
    timerange = Arguments.parse_timerange(args.timerange)

    # Load the strategy
    try:
        analyze = Analyze(config)
        exchange.init(config)
    except AttributeError:
        logger.critical(
            'Impossible to load the strategy. Please check the file "user_data/strategies/%s.py"',
            args.strategy
        )
        exit()

    # Set the ticker to use
    tick_interval = analyze.get_ticker_interval()

    # Load pair tickers
    tickers = {}
    if args.live:
        logger.info('Downloading pair.')
        tickers[pair] = exchange.get_ticker_history(pair, tick_interval)
    else:
        tickers = optimize.load_data(
            datadir=args.datadir,
            pairs=[pair],
            ticker_interval=tick_interval,
            refresh_pairs=config.get('refresh_pairs', False),
            timerange=timerange
        )

        # No ticker found, or impossible to download
        if tickers == {}:
            exit()

    # Get trades already made from the DB
    trades = []
    if args.db_url:
        engine = create_engine('sqlite:///' + args.db_url)
        persistence.init(_CONF, engine)
        trades = Trade.query.filter(Trade.pair.is_(pair)).all()

    dataframes = analyze.tickerdata_to_dataframe(tickers)
    dataframe = dataframes[pair]
    dataframe = analyze.populate_buy_trend(dataframe)
    dataframe = analyze.populate_sell_trend(dataframe)

    if len(dataframe.index) > 750:
        logger.warning('Ticker contained more than 750 candles, clipping.')

    fig = generate_graph(
        pair=pair,
        trades=trades,
        data=dataframe.tail(750)
    )

    plot(fig, filename=os.path.join('user_data', 'freqtrade-plot.html'))


def generate_graph(pair, trades, data) -> None:
    """
    Generate the graph from the data generated by Backtesting or from DB
    :param pair: Pair to Display on the graph
    :param trades: All trades created
    :param data: Dataframe
    :return: None
    """

    # Define the graph
    fig = tools.make_subplots(
        rows=3,
        cols=1,
        shared_xaxes=True,
        row_width=[1, 1, 4],
        vertical_spacing=0.0001,
    )
    fig['layout'].update(title=pair)

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

    df_buy = data[data['buy'] == 1]
    buys = go.Scattergl(
        x=df_buy.date,
        y=df_buy.close,
        mode='markers',
        name='buy',
        marker=dict(
            symbol='triangle-up-dot',
            size=9,
            line=dict(width=1),
            color='green',
        )
    )
    df_sell = data[data['sell'] == 1]
    sells = go.Scattergl(
        x=df_sell.date,
        y=df_sell.close,
        mode='markers',
        name='sell',
        marker=dict(
            symbol='triangle-down-dot',
            size=9,
            line=dict(width=1),
            color='red',
        )
    )

    trade_buys = go.Scattergl(
        x=[t.open_date.isoformat() for t in trades],
        y=[t.open_rate for t in trades],
        mode='markers',
        name='trade_buy',
        marker=dict(
            symbol='square-open',
            size=11,
            line=dict(width=2),
            color='green'
        )
    )
    trade_sells = go.Scattergl(
        x=[t.close_date.isoformat() for t in trades],
        y=[t.close_rate for t in trades],
        mode='markers',
        name='trade_sell',
        marker=dict(
            symbol='square-open',
            size=11,
            line=dict(width=2),
            color='red'
        )
    )

    # Row 1
    fig.append_trace(candles, 1, 1)

    if 'bb_lowerband' in data and 'bb_upperband' in data:
        bb_lower = go.Scatter(
            x=data.date,
            y=data.bb_lowerband,
            name='BB lower',
            line={'color': "transparent"},
        )
        bb_upper = go.Scatter(
            x=data.date,
            y=data.bb_upperband,
            name='BB upper',
            fill="tonexty",
            fillcolor="rgba(0,176,246,0.2)",
            line={'color': "transparent"},
        )
        fig.append_trace(bb_lower, 1, 1)
        fig.append_trace(bb_upper, 1, 1)

    if TO_DISPLAY['sma'] and 'sma' in data:
        sma = generate_scattergl(index='sma', name='SMA', data=data)
        fig.append_trace(sma, 1, 1)

    if TO_DISPLAY['ema'] and 'ema10' in data and 'ema50' in data:
        ema10 = generate_scattergl(index='ema10', name='EMA10', data=data)
        ema50 = generate_scattergl(index='ema50', name='EMA50', data=data)
        fig.append_trace(ema10, 1, 1)
        fig.append_trace(ema50, 1, 1)

    fig.append_trace(buys, 1, 1)
    fig.append_trace(sells, 1, 1)
    fig.append_trace(trade_buys, 1, 1)
    fig.append_trace(trade_sells, 1, 1)
    fig['layout']['yaxis1'].update(title='Price')

    # Row 2
    volume = go.Bar(
        x=data['date'],
        y=data['volume'],
        name='Volume'
    )
    fig.append_trace(volume, 2, 1)
    fig['layout']['yaxis2'].update(title='Volume')

    # Row 3 (On Row 3, we can only display one indicator)
    if TO_DISPLAY['macd'] and 'macd' in data and 'macdsignal' in data:
        macd = generate_scattergl(index='macd', name='MACD', data=data)
        macdsignal = generate_scattergl(index='macdsignal', name='MACD Signal', data=data)
        fig.append_trace(macd, 3, 1)
        fig.append_trace(macdsignal, 3, 1)
        fig['layout']['yaxis3'].update(title='MACD')

    elif TO_DISPLAY['fast'] and 'fastd' in data and 'fastk' in data:
        fastd = generate_scattergl(index='fastd', name='fastd', data=data)
        fastk = generate_scattergl(index='fastk', name='fastk', data=data)
        fig.append_trace(fastd, 3, 1)
        fig.append_trace(fastk, 3, 1)
        fig['layout']['yaxis3'].update(title='Stoch Fast')

    elif TO_DISPLAY['slow'] and 'slowd' in data and 'slowk' in data:
        slowd = generate_scattergl(index='slowd', name='slowd', data=data)
        slowk = generate_scattergl(index='slowk', name='slowk', data=data)
        fig.append_trace(slowd, 3, 1)
        fig.append_trace(slowk, 3, 1)
        fig['layout']['yaxis3'].update(title='Stoch Slow')

    elif TO_DISPLAY['rsi'] and 'rsi' in data:
        rsi = generate_scattergl(index='rsi', name='RSI', data=data)
        fig.append_trace(rsi, 3, 1)
        fig['layout']['yaxis3'].update(title='RSI')

    elif TO_DISPLAY['mfi'] and 'mfi' in data:
        mfi = generate_scattergl(index='mfi', name='MFI', data=data)
        fig.append_trace(mfi, 3, 1)
        fig['layout']['yaxis3'].update(title='MFI')

    elif TO_DISPLAY['fisher_rsi'] and 'fisher_rsi' in data:
        fisher_rsi = generate_scattergl(index='fisher_rsi', name='Fisher RSI', data=data)
        fig.append_trace(fisher_rsi, 3, 1)
        fig['layout']['yaxis3'].update(title='Fisher RSI')

    return fig


def generate_scattergl(index, name, data) -> go.Scattergl:
    """
    Generate a Scattergl element
    :param index: code of the Indicator to generate
    :param name: Name that will be display in the graph legend
    :param data: Dataframe
    :return: Scattergl
    """
    return go.Scattergl(
        x=data['date'],
        y=data[index],
        name=name
    )


def plot_parse_args(args: List[str]) -> Namespace:
    """
    Parse args passed to the script
    :param args: Cli arguments
    :return: args: Array with all arguments
    """
    arguments = Arguments(args, 'Graph dataframe')
    arguments.scripts_options()
    arguments.common_args_parser()
    arguments.optimizer_shared_options(arguments.parser)
    arguments.backtesting_options(arguments.parser)

    return arguments.parse_args()


def main(sysargv: List[str]) -> None:
    """
    This function will initiate the bot and start the trading loop.
    :return: None
    """
    logger.info('Starting Plot Dataframe')
    plot_analyzed_dataframe(
        plot_parse_args(sysargv)
    )


if __name__ == '__main__':
    main(sys.argv[1:])