tabs/trades.py

import gradio as gr
import pandas as pd
import plotly.express as px


HEIGHT = 400
WIDTH = 1100


def prepare_trades(trades_df: pd.DataFrame) -> pd.DataFrame:
    """Prepares the trades data for analysis."""
    trades_df["creation_timestamp"] = pd.to_datetime(trades_df["creation_timestamp"])
    trades_df["creation_timestamp"] = trades_df["creation_timestamp"].dt.tz_convert(
        "UTC"
    )
    trades_df = trades_df.sort_values(by="creation_timestamp", ascending=True)
    trades_df["month_year"] = (
        trades_df["creation_timestamp"].dt.to_period("M").astype(str)
    )
    trades_df["month_year_week"] = (
        trades_df["creation_timestamp"].dt.to_period("W").dt.strftime("%b-%d")
    )
    trades_df["winning_trade"] = trades_df["winning_trade"].astype(int)
    return trades_df


def get_overall_trades(trades_df: pd.DataFrame) -> pd.DataFrame:
    """Gets the overall trades data"""
    trades_count = trades_df.groupby("month_year_week").size().reset_index()
    trades_count.columns = trades_count.columns.astype(str)
    trades_count.rename(columns={"0": "trades"}, inplace=True)
    return trades_count


def get_overall_by_market_trades(trades_df: pd.DataFrame) -> pd.DataFrame:
    """Gets the overall trades data"""
    trades_count = (
        trades_df.groupby(["month_year_week", "market_creator"], sort=False)
        .size()
        .reset_index()
    )
    trades_count.columns = trades_count.columns.astype(str)
    trades_count.rename(columns={"0": "trades"}, inplace=True)
    return trades_count


def get_overall_winning_trades(trades_df: pd.DataFrame) -> pd.DataFrame:
    """Gets the overall winning trades data for the given tools and calculates the winning percentage."""
    winning_trades = (
        trades_df.groupby(["month_year_week"])["winning_trade"].sum()
        / trades_df.groupby(["month_year_week"])["winning_trade"].count()
        * 100
    )
    # winning_trades is a series, give it a dataframe
    winning_trades = winning_trades.reset_index()
    winning_trades.columns = winning_trades.columns.astype(str)
    winning_trades.columns = ["month_year_week", "winning_trade"]
    return winning_trades


def get_overall_winning_by_market_trades(trades_df: pd.DataFrame) -> pd.DataFrame:
    """Gets the overall winning trades data for the given tools and calculates the winning percentage."""
    winning_trades = (
        trades_df.groupby(["month_year_week", "market_creator"], sort=False)[
            "winning_trade"
        ].sum()
        / trades_df.groupby(["month_year_week", "market_creator"], sort=False)[
            "winning_trade"
        ].count()
        * 100
    )
    # winning_trades is a series, give it a dataframe
    winning_trades = winning_trades.reset_index()
    winning_trades.columns = winning_trades.columns.astype(str)
    winning_trades.columns = ["month_year_week", "market_creator", "winning_trade"]
    return winning_trades


def plot_trades_by_week(trades_df: pd.DataFrame) -> gr.BarPlot:
    """Plots the trades data for the given tools and calculates the winning percentage."""
    return gr.BarPlot(
        value=trades_df,
        x="month_year_week",
        y="trades",
        show_label=True,
        interactive=True,
        show_actions_button=True,
        tooltip=["month_year_week", "trades"],
        height=HEIGHT,
        width=WIDTH,
    )


def plot_trades_per_market_by_week(
    trades_df: pd.DataFrame, market_type: str
) -> gr.Plot:
    """Plots the trades data for the given tools and calculates the winning percentage."""
    assert "market_creator" in trades_df.columns
    # if market_type is "all then no filter is applied"
    if market_type == "quickstart":
        trades = trades_df.loc[trades_df["market_creator"] == "quickstart"]
        color_sequence = ["goldenrod"]

    elif market_type == "pearl":
        trades = trades_df.loc[trades_df["market_creator"] == "pearl"]
        color_sequence = ["purple"]
    else:
        trades = trades_df
        color_sequence = ["darkgreen"]

    fig = px.bar(
        trades,
        x="month_year_week",
        y="trades",
        color_discrete_sequence=color_sequence,
        title=market_type + " trades",
    )
    fig.update_layout(
        xaxis_title="Week",
        yaxis_title="Weekly nr of trades",
    )
    fig.update_xaxes(tickformat="%b %d\n%Y")
    return gr.Plot(
        value=fig,
    )


def integrated_plot_trades_per_market_by_week(trades_df: pd.DataFrame) -> gr.Plot:

    # adding the total
    trades_all = trades_df.copy(deep=True)
    trades_all["market_creator"] = "all"

    # merging both dataframes
    all_filtered_trades = pd.concat([trades_df, trades_all], ignore_index=True)
    all_filtered_trades = all_filtered_trades.sort_values(
        by="creation_timestamp", ascending=True
    )

    trades = get_overall_by_market_trades(all_filtered_trades)
    fig = px.bar(
        trades,
        x="month_year_week",
        y="trades",
        color="market_creator",
        barmode="group",
        color_discrete_sequence=["goldenrod", "darkgreen", "purple"],
    )
    fig.update_layout(
        xaxis_title="Week",
        yaxis_title="Weekly nr of trades",
        legend=dict(yanchor="top", y=0.5),
    )
    fig.update_layout(width=WIDTH, height=HEIGHT)
    fig.update_xaxes(tickformat="%b %d\n%Y")
    return gr.Plot(value=fig)


def integrated_plot_winning_trades_per_market_by_week(
    trades_df: pd.DataFrame,
) -> gr.Plot:
    # adding the total
    trades_all = trades_df.copy(deep=True)
    trades_all["market_creator"] = "all"

    # merging both dataframes
    all_filtered_trades = pd.concat([trades_df, trades_all], ignore_index=True)
    all_filtered_trades = all_filtered_trades.sort_values(
        by="creation_timestamp", ascending=True
    )
    final_df = get_overall_winning_by_market_trades(all_filtered_trades)
    fig = px.bar(
        final_df,
        x="month_year_week",
        y="winning_trade",
        color="market_creator",
        barmode="group",
        color_discrete_sequence=["goldenrod", "darkgreen", "purple"],
    )
    fig.update_layout(
        xaxis_title="Week",
        yaxis_title="Weekly % of winning trades",
        legend=dict(yanchor="top", y=0.5),
    )
    fig.update_layout(width=WIDTH, height=HEIGHT)
    fig.update_xaxes(tickformat="%b %d\n%Y")
    return gr.Plot(
        value=fig,
    )


def plot_winning_trades_by_week(trades_df: pd.DataFrame) -> gr.BarPlot:
    """Plots the winning trades data for the given tools and calculates the winning percentage."""
    return gr.BarPlot(
        value=trades_df,
        x="month_year_week",
        y="winning_trade",
        show_label=True,
        interactive=True,
        show_actions_button=True,
        tooltip=["month_year_week", "winning_trade"],
        height=HEIGHT,
        width=WIDTH,
    )


def plot_winning_trades_per_market_by_week(
    trades_df: pd.DataFrame, market_type: str
) -> gr.Plot:
    """Plots the winning trades data for the given tools and calculates the winning percentage."""
    # if market_type is "all then no filter is applied"
    if market_type == "quickstart":
        trades = trades_df.loc[trades_df["market_creator"] == "quickstart"]
        color_sequence = ["goldenrod"]

    elif market_type == "pearl":
        trades = trades_df.loc[trades_df["market_creator"] == "pearl"]
        color_sequence = ["purple"]
    else:
        trades = trades_df
        color_sequence = ["darkgreen"]

    fig = px.bar(
        trades,
        x="month_year_week",
        y="winning_trade",
        color_discrete_sequence=color_sequence,
        title=market_type + " winning trades",
    )
    fig.update_layout(
        xaxis_title="Week",
        yaxis_title="Weekly % of winning trades",
    )
    fig.update_xaxes(tickformat="%b %d\n%Y")
    return gr.Plot(
        value=fig,
    )