app.py

from datetime import datetime, timedelta
import gradio as gr
import pandas as pd
import duckdb
import logging
from tabs.trades import (
    prepare_trades,
    get_overall_trades,
    get_overall_winning_trades,
    plot_trades_by_week,
    plot_winning_trades_by_week,
    plot_trade_details,
)
from tabs.tool_win import (
    get_tool_winning_rate,
    get_overall_winning_rate,
    plot_tool_winnings_overall,
    plot_tool_winnings_by_tool,
)
from tabs.error import (
    get_error_data,
    get_error_data_overall,
    plot_error_data,
    plot_tool_error_data,
    plot_week_error_data,
)
from tabs.about import about_olas_predict, about_this_dashboard

from scripts.utils import INC_TOOLS


def get_logger():
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)
    # stream handler and formatter
    stream_handler = logging.StreamHandler()
    stream_handler.setLevel(logging.DEBUG)
    formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    stream_handler.setFormatter(formatter)
    logger.addHandler(stream_handler)
    return logger


logger = get_logger()


def get_last_one_month_data():
    """
    Get the last one month data from the tools.parquet file
    """
    logger.info("Getting last one month data")
    con = duckdb.connect(":memory:")
    one_months_ago = (datetime.now() - timedelta(days=60)).strftime("%Y-%m-%d")

    # Query to fetch data from all_trades_profitability.parquet
    query2 = f"""
    SELECT *
    FROM read_parquet('./data/all_trades_profitability.parquet')
    WHERE creation_timestamp >= '{one_months_ago}'
    """
    df2 = con.execute(query2).fetchdf()
    logger.info("Got last one month data from all_trades_profitability.parquet")

    query1 = f"""
    SELECT *
    FROM read_parquet('./data/tools.parquet')
    WHERE request_time >= '{one_months_ago}'
    """
    df1 = con.execute(query1).fetchdf()
    logger.info("Got last one month data from tools.parquet")

    con.close()

    return df1, df2


def get_all_data():
    """
    Get all data from the tools.parquet and all_trades_profitability.parquet files
    """
    logger.info("Getting all data")
    con = duckdb.connect(":memory:")

    # Query to fetch data from all_trades_profitability.parquet
    query2 = f"""
    SELECT *
    FROM read_parquet('./data/all_trades_profitability.parquet')
    """
    df2 = con.execute(query2).fetchdf()
    logger.info("Got all data from all_trades_profitability.parquet")

    query1 = f"""
    SELECT *
    FROM read_parquet('./data/tools.parquet')
    """
    df1 = con.execute(query1).fetchdf()
    logger.info("Got all data from tools.parquet")

    con.close()

    return df1, df2


def prepare_data():
    """
    Prepare the data for the dashboard
    """
    tools_df, trades_df = get_all_data()

    tools_df["request_time"] = pd.to_datetime(tools_df["request_time"])
    trades_df["creation_timestamp"] = pd.to_datetime(trades_df["creation_timestamp"])

    trades_df = prepare_trades(trades_df)
    return tools_df, trades_df


tools_df, trades_df = prepare_data()


demo = gr.Blocks()

error_df = get_error_data(tools_df=tools_df, inc_tools=INC_TOOLS)
error_overall_df = get_error_data_overall(error_df=error_df)
winning_rate_df = get_tool_winning_rate(tools_df=tools_df, inc_tools=INC_TOOLS)
winning_rate_overall_df = get_overall_winning_rate(wins_df=winning_rate_df)
trades_count_df = get_overall_trades(trades_df=trades_df)
trades_winning_rate_df = get_overall_winning_trades(trades_df=trades_df)

with demo:
    gr.HTML("<h1>Olas Predict Actual Performance</h1>")
    gr.Markdown(
        "This app shows the actual performance of Olas Predict tools on the live market."
    )

    with gr.Tabs():
        with gr.TabItem("🔥Trades Dashboard"):
            with gr.Row():
                gr.Markdown("# Number of trades per week")
            with gr.Row():
                trades_by_week_plot = plot_trades_by_week(trades_df=trades_count_df)
            with gr.Row():
                gr.Markdown("# Percentage of winning trades per week")
            with gr.Row():
                winning_trades_by_week_plot = plot_winning_trades_by_week(
                    trades_df=trades_winning_rate_df
                )
            with gr.Row():
                gr.Markdown("# Trading metrics")
            with gr.Row():
                trade_details_selector = gr.Dropdown(
                    label="Select a trade metric",
                    choices=[
                        "mech calls",
                        "collateral amount",
                        "earnings",
                        "net earnings",
                        "ROI",
                    ],
                    value="mech calls",
                )
            with gr.Row():
                trade_details_plot = plot_trade_details(
                    trade_detail="mech calls", trades_df=trades_df
                )

            def update_trade_details(trade_detail):
                return plot_trade_details(
                    trade_detail=trade_detail, trades_df=trades_df
                )

            trade_details_selector.change(
                update_trade_details,
                inputs=trade_details_selector,
                outputs=trade_details_plot,
            )

            with gr.Row():
                trade_details_selector
            with gr.Row():
                trade_details_plot

        with gr.TabItem("🚀 Tool Winning Dashboard"):
            with gr.Row():
                gr.Markdown("# All tools winning performance")

            with gr.Row():
                winning_selector = gr.Dropdown(
                    label="Select the tool metric",
                    choices=["losses", "wins", "total_request", "win_perc"],
                    value="win_perc",
                )

            with gr.Row():
                winning_plot = plot_tool_winnings_overall(
                    wins_df=winning_rate_overall_df, winning_selector="win_perc"
                )

            def update_tool_winnings_overall_plot(winning_selector):
                return plot_tool_winnings_overall(
                    wins_df=winning_rate_overall_df, winning_selector=winning_selector
                )

            winning_selector.change(
                update_tool_winnings_overall_plot,
                inputs=winning_selector,
                outputs=winning_plot,
            )

            with gr.Row():
                winning_selector
            with gr.Row():
                winning_plot

            with gr.Row():
                gr.Markdown("# Winning performance by each tool")

            with gr.Row():
                sel_tool = gr.Dropdown(
                    label="Select a tool", choices=INC_TOOLS, value=INC_TOOLS[0]
                )

            with gr.Row():
                tool_winnings_by_tool_plot = plot_tool_winnings_by_tool(
                    wins_df=winning_rate_df, tool=INC_TOOLS[0]
                )

            def update_tool_winnings_by_tool_plot(tool):
                return plot_tool_winnings_by_tool(wins_df=winning_rate_df, tool=tool)

            sel_tool.change(
                update_tool_winnings_by_tool_plot,
                inputs=sel_tool,
                outputs=tool_winnings_by_tool_plot,
            )

            with gr.Row():
                sel_tool
            with gr.Row():
                tool_winnings_by_tool_plot

        with gr.TabItem("🏥 Tool Error Dashboard"):
            with gr.Row():
                gr.Markdown("# All tools errors")
            with gr.Row():
                error_overall_plot = plot_error_data(error_all_df=error_overall_df)
            with gr.Row():
                gr.Markdown("# Error percentage per tool")
            with gr.Row():
                sel_tool = gr.Dropdown(
                    label="Select a tool", choices=INC_TOOLS, value=INC_TOOLS[0]
                )

            with gr.Row():
                tool_error_plot = plot_tool_error_data(
                    error_df=error_df, tool=INC_TOOLS[0]
                )

            def update_tool_error_plot(tool):
                return plot_tool_error_data(error_df=error_df, tool=tool)

            sel_tool.change(
                update_tool_error_plot, inputs=sel_tool, outputs=tool_error_plot
            )
            with gr.Row():
                sel_tool
            with gr.Row():
                tool_error_plot

            with gr.Row():
                gr.Markdown("# Tools distribution of errors per week")

            with gr.Row():
                choices = error_overall_df["request_month_year_week"].unique().tolist()
                # sort the choices by the latest week to be on the top
                choices = sorted(choices)
                sel_week = gr.Dropdown(
                    label="Select a week", choices=choices, value=choices[-1]
                )

            with gr.Row():
                week_error_plot = plot_week_error_data(
                    error_df=error_df, week=choices[-1]
                )

            def update_week_error_plot(selected_week):
                return plot_week_error_data(error_df=error_df, week=selected_week)

            sel_tool.change(
                update_tool_error_plot, inputs=sel_tool, outputs=tool_error_plot
            )
            sel_week.change(
                update_week_error_plot, inputs=sel_week, outputs=week_error_plot
            )

            with gr.Row():
                sel_tool
            with gr.Row():
                tool_error_plot
            with gr.Row():
                sel_week
            with gr.Row():
                week_error_plot

        with gr.TabItem("ℹ️ About"):
            with gr.Accordion("About Olas Predict"):
                gr.Markdown(about_olas_predict)

            with gr.Accordion("About this dashboard"):
                gr.Markdown(about_this_dashboard)

demo.queue(default_concurrency_limit=40).launch()