428-backtester/scripts/plot_trades_chart.py

#!/usr/bin/env python3
"""
Clean trade chart: price, signals, SL/TP levels, filled trades, volume.
Price / signals / SL·TP / fills — no indicator clutter.
"""
from __future__ import annotations

import argparse
import logging
import sys
from datetime import UTC, datetime
from pathlib import Path

import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots

from freqtrade.configuration import Configuration
from freqtrade.data.btanalysis import extract_trades_of_period
from freqtrade.data.converter import trim_dataframe
from freqtrade.data.dataprovider import DataProvider
from freqtrade.misc import pair_to_filename
from freqtrade.plot.plotting import init_plotscript, store_plot_file
from freqtrade.resolvers import ExchangeResolver, StrategyResolver
from freqtrade.strategy import IStrategy
from freqtrade.strategy.strategy_wrapper import strategy_safe_wrapper

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("plot_trades_chart")

SIGNAL_SIZE = 16
TRADE_SIZE = 14

LEVEL_STYLES = {
    "sl": ("SL", "#e74c3c", "solid", 2),
    "tp1": ("TP1", "#27ae60", "solid", 1.5),
    "tp2": ("TP2", "#2ecc71", "dash", 1.2),
    "tp3": ("TP3", "#1abc9c", "dot", 1.2),
}


def _signal_scatter(
    data: pd.DataFrame, column: str, color: str, direction: str, size: int
) -> go.Scatter | None:
    if column not in data.columns:
        return None
    df = data[data[column] == 1]
    if df.empty:
        return None
    return go.Scatter(
        x=df["date"],
        y=df["close"],
        mode="markers",
        name=column,
        marker=dict(
            symbol=f"triangle-{direction}-dot",
            size=size,
            line=dict(width=1.5, color=color),
            color=color,
        ),
    )


def _levels_for_trade(row: pd.Series, candles: pd.DataFrame) -> dict[str, float] | None:
    """Read frozen SL/TP from the entry candle (strategy columns)."""
    open_ts = pd.Timestamp(row["open_date"])
    if open_ts.tzinfo is None:
        open_ts = open_ts.tz_localize("UTC")
    dates = pd.to_datetime(candles["date"], utc=True)
    matched = candles.loc[dates <= open_ts]
    if matched.empty:
        return None
    candle = matched.iloc[-1]
    prefix = "short" if bool(row.get("is_short", False)) else "long"
    out: dict[str, float] = {}
    for key in ("sl", "tp1", "tp2", "tp3"):
        col = f"{prefix}_{key}"
        if col not in candle.index or pd.isna(candle[col]):
            return None
        out[key] = float(candle[col])
    return out


def _tp1_fill_time(row: pd.Series) -> pd.Timestamp | None:
    orders = row.get("orders")
    if not isinstance(orders, list):
        return None
    for order in orders:
        tag = str(order.get("ft_order_tag") or "")
        if tag == "tp1" or tag.startswith("tp1"):
            ts = order.get("order_filled_timestamp") or order.get("order_filled_date")
            if ts is None:
                return None
            if isinstance(ts, (int, float)):
                return pd.to_datetime(ts, unit="ms", utc=True)
            return pd.to_datetime(ts, utc=True)
    return None


def _add_trade_levels(fig: go.Figure, trades: pd.DataFrame, candles: pd.DataFrame) -> None:
    seen: set[str] = set()
    for _, row in trades.iterrows():
        levels = _levels_for_trade(row, candles)
        if not levels:
            continue
        x0, x1 = row["open_date"], row["close_date"]
        if pd.isna(x1):
            continue
        be_from = _tp1_fill_time(row)

        for key, price in levels.items():
            name, color, dash, width = LEVEL_STYLES[key]
            show = name not in seen
            if show:
                seen.add(name)

            if key == "sl" and be_from is not None and be_from > x0:
                fig.add_trace(
                    go.Scatter(
                        x=[x0, be_from],
                        y=[price, price],
                        mode="lines",
                        name=name,
                        showlegend=show,
                        line=dict(color=color, width=width, dash=dash),
                        hovertemplate=f"{name}: %{{y:.1f}}<extra></extra>",
                    ),
                    row=1,
                    col=1,
                )
                be_name = "SL (BE)"
                be_show = be_name not in seen
                if be_show:
                    seen.add(be_name)
                entry = float(row["open_rate"])
                fig.add_trace(
                    go.Scatter(
                        x=[be_from, x1],
                        y=[entry, entry],
                        mode="lines",
                        name=be_name,
                        showlegend=be_show,
                        line=dict(color="#f39c12", width=2, dash="dash"),
                        hovertemplate=f"{be_name}: %{{y:.1f}}<extra></extra>",
                    ),
                    row=1,
                    col=1,
                )
            else:
                fig.add_trace(
                    go.Scatter(
                        x=[x0, x1],
                        y=[price, price],
                        mode="lines",
                        name=name,
                        showlegend=show,
                        line=dict(color=color, width=width, dash=dash),
                        hovertemplate=f"{name}: %{{y:.1f}}<extra></extra>",
                    ),
                    row=1,
                    col=1,
                )


def _add_filled_trades(fig: go.Figure, trades: pd.DataFrame) -> None:
    if trades is None or trades.empty:
        return

    desc = trades.apply(
        lambda r: (
            f"{r['profit_ratio']:.2%}, "
            + (f"{r['enter_tag']}, " if pd.notna(r.get('enter_tag')) else "")
            + f"{r['exit_reason']}, "
            + f"{r['trade_duration']} min"
        ),
        axis=1,
    )

    fig.add_trace(
        go.Scatter(
            x=trades["open_date"],
            y=trades["open_rate"],
            mode="markers",
            name="Trade entry",
            text=desc,
            marker=dict(symbol="circle-open", size=TRADE_SIZE, line=dict(width=2.5), color="cyan"),
        ),
        row=1,
        col=1,
    )

    wins = trades["profit_ratio"] > 0
    losses = ~wins
    if wins.any():
        fig.add_trace(
            go.Scatter(
                x=trades.loc[wins, "close_date"],
                y=trades.loc[wins, "close_rate"],
                mode="markers",
                name="Exit - Profit",
                text=desc[wins],
                marker=dict(
                    symbol="square-open", size=TRADE_SIZE, line=dict(width=2.5), color="green"
                ),
            ),
            row=1,
            col=1,
        )
    if losses.any():
        fig.add_trace(
            go.Scatter(
                x=trades.loc[losses, "close_date"],
                y=trades.loc[losses, "close_rate"],
                mode="markers",
                name="Exit - Loss",
                text=desc[losses],
                marker=dict(
                    symbol="square-open", size=TRADE_SIZE, line=dict(width=2.5), color="red"
                ),
            ),
            row=1,
            col=1,
        )


def build_figure(pair: str, data: pd.DataFrame, trades: pd.DataFrame) -> go.Figure:
    fig = make_subplots(
        rows=2,
        cols=1,
        shared_xaxes=True,
        row_width=[1, 4],
        vertical_spacing=0.02,
    )
    fig.update_layout(
        title=f"{pair} — trades",
        xaxis_rangeslider_visible=False,
        legend=dict(orientation="h", yanchor="bottom", y=1.02, x=0),
        margin=dict(t=80, b=40),
        modebar_add=["v1hovermode", "toggleSpikeLines"],
    )
    fig.update_yaxes(title_text="Price", row=1, col=1)
    fig.update_yaxes(title_text="Volume", row=2, col=1)

    fig.add_trace(
        go.Candlestick(
            x=data["date"],
            open=data["open"],
            high=data["high"],
            low=data["low"],
            close=data["close"],
            name="Price",
            increasing_line_color="#26a69a",
            decreasing_line_color="#ef5350",
        ),
        row=1,
        col=1,
    )

    for scatter in (
        _signal_scatter(data, "enter_long", "#2ecc71", "up", SIGNAL_SIZE),
        _signal_scatter(data, "exit_long", "#e74c3c", "down", SIGNAL_SIZE),
        _signal_scatter(data, "enter_short", "#3498db", "down", SIGNAL_SIZE),
        _signal_scatter(data, "exit_short", "#9b59b6", "up", SIGNAL_SIZE),
    ):
        if scatter is not None:
            fig.add_trace(scatter, row=1, col=1)

    if trades is not None and not trades.empty:
        _add_trade_levels(fig, trades, data)
        _add_filled_trades(fig, trades)

    fig.add_trace(
        go.Bar(
            x=data["date"],
            y=data["volume"],
            name="Volume",
            marker_color="DarkSlateGrey",
            marker_line_color="DarkSlateGrey",
        ),
        row=2,
        col=1,
    )
    return fig


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", default="user_data/config.json")
    parser.add_argument("--strategy", default="SampleStrategy")
    parser.add_argument("--timerange", default=None)
    parser.add_argument("--pair", default="BTC/USDT:USDT")
    parser.add_argument("--timeframe", default="15m")
    parser.add_argument("--outfile", default=None)
    args = parser.parse_args()

    cfg = Configuration.from_files([args.config])
    cfg["strategy"] = args.strategy
    cfg["timeframe"] = args.timeframe
    cfg["pairs"] = [args.pair]
    if args.timerange:
        cfg["timerange"] = args.timerange
    cfg.setdefault("trade_source", "file")

    strategy = StrategyResolver.load_strategy(cfg)
    exchange = ExchangeResolver.load_exchange(cfg)
    IStrategy.dp = DataProvider(cfg, exchange)
    strategy.ft_bot_start()
    strategy_safe_wrapper(strategy.bot_loop_start)(current_time=datetime.now(UTC))

    plot_elements = init_plotscript(cfg, list(exchange.markets), strategy.startup_candle_count)
    timerange = plot_elements["timerange"]
    trades = plot_elements["trades"]

    pair = args.pair
    if pair not in plot_elements["ohlcv"]:
        raise SystemExit(f"No OHLCV for {pair}")

    data = strategy.analyze_ticker(plot_elements["ohlcv"][pair], {"pair": pair})
    data = trim_dataframe(data, timerange)

    if not trades.empty:
        trades_pair = trades.loc[trades["pair"] == pair]
        trades_pair = extract_trades_of_period(data, trades_pair)
    else:
        trades_pair = trades

    fig = build_figure(pair, data, trades_pair)
    out_name = args.outfile or f"freqtrade-plot-{pair_to_filename(pair)}-{args.timeframe}.html"
    store_plot_file(fig, filename=out_name, directory=Path(cfg["user_data_dir"]) / "plot")
    logger.info("Open: user_data/plot/%s", out_name)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())