tabs/trades.py

import gradio as gr
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots


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_date"] = trades_df["creation_timestamp"].dt.date
    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 = 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 get_overall_winning_by_market_and_trader_type(
    trades_df: pd.DataFrame,
) -> pd.DataFrame:
    """Gets the overall winning trades data for the given tools and calculates the winning percentage."""
    # Group by week, market_creator and staking_type
    winning_trades = (
        trades_df.groupby(
            ["month_year_week", "market_creator", "staking_type"], sort=False
        )["winning_trade"].sum()
        / trades_df.groupby(
            ["month_year_week", "market_creator", "staking_type"], sort=False
        )["winning_trade"].count()
        * 100
    )

    winning_trades = winning_trades.reset_index()
    winning_trades.columns = winning_trades.columns.astype(str)
    winning_trades.columns = [
        "month_year_week",
        "market_creator",
        "staking_type",
        "winning_trade",
    ]
    return winning_trades


def plot_trades_by_week(trades_df: pd.DataFrame) -> gr.BarPlot:
    """Plots the weekly trades data ."""
    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 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=["purple", "goldenrod", "darkgreen"],
        category_orders={"market_creator": ["pearl", "quickstart", "all"]},
    )

    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_trades_per_market_by_week_v2(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
    )

    # Create binary staking category
    all_filtered_trades["staking_type"] = all_filtered_trades["staking"].apply(
        lambda x: "non_agent" if x == "non_agent" else "agent"
    )

    # Group by week, market_creator and staking_type
    trades = (
        all_filtered_trades.groupby(
            ["month_year_week", "market_creator", "staking_type"], sort=False
        )
        .size()
        .reset_index(name="trades")
    )

    # Combine the traces
    final_traces = []
    market_colors = {"pearl": "darkviolet", "quickstart": "goldenrod", "all": "green"}
    # First add 'agent' traces
    for market in ["pearl", "quickstart", "all"]:
        market_data = trades[trades["market_creator"] == market]
        agent_data = market_data[market_data["staking_type"] == "agent"]

        trace = go.Bar(
            x=agent_data["month_year_week"],
            y=agent_data["trades"],
            name=f"{market}-agent",
            marker_color=market_colors[market],
            offsetgroup=market,
            showlegend=True,
        )
        final_traces.append(trace)

    # Then add 'non_agent' traces
    market_darker_colors = {
        "pearl": "purple",
        "quickstart": "darkgoldenrod",
        "all": "darkgreen",
    }
    for market in ["pearl", "quickstart", "all"]:
        market_data = trades[trades["market_creator"] == market]
        non_agent_data = market_data[market_data["staking_type"] == "non_agent"]

        trace = go.Bar(
            x=non_agent_data["month_year_week"],
            y=non_agent_data["trades"],
            name=f"{market}-non_agent",
            marker_color=market_darker_colors[market],
            offsetgroup=market,
            showlegend=True,
        )
        final_traces.append(trace)

    # Create new figure with the combined traces
    fig = go.Figure(data=final_traces)

    # Update layout
    fig.update_layout(
        xaxis_title="Week",
        yaxis_title="Weekly nr of trades",
        legend=dict(yanchor="top", y=0.5),
        width=WIDTH,
        height=HEIGHT,
        barmode="group",
    )

    # Update x-axis format
    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=["purple", "goldenrod", "darkgreen"],
        category_orders={"market_creator": ["pearl", "quickstart", "all"]},
    )
    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 integrated_plot_winning_trades_per_market_by_week_v2(
    trades_df: pd.DataFrame, trader_filter: str = None
) -> 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
    )
    # Create binary staking category
    all_filtered_trades["staking_type"] = all_filtered_trades["staking"].apply(
        lambda x: "non_agent" if x == "non_agent" else "agent"
    )
    if trader_filter is None:
        final_df = get_overall_winning_by_market_trades(all_filtered_trades)
    else:
        final_df = get_overall_winning_by_market_and_trader_type(all_filtered_trades)

    color_discrete_sequence = ["darkviolet", "goldenrod", "green"]
    if trader_filter == "agent":
        final_df = final_df[final_df["staking_type"] == "agent"]
    elif trader_filter == "non_agent":
        final_df = final_df[final_df["staking_type"] == "non_agent"]
        color_discrete_sequence = ["purple", "darkgoldenrod", "darkgreen"]

    fig = px.bar(
        final_df,
        x="month_year_week",
        y="winning_trade",
        color="market_creator",
        barmode="group",
        color_discrete_sequence=color_discrete_sequence,
        category_orders={"market_creator": ["pearl", "quickstart", "all"]},
    )
    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,
    )