scripts/metrics.py

import pandas as pd
from tqdm import tqdm

DEFAULT_MECH_FEE = 0.01  # xDAI


def compute_metrics(
    trader_address: str, trader_data: pd.DataFrame, live_metrics: bool = False
) -> dict:

    if len(trader_data) == 0:
        # print("No data to compute metrics")
        return {}

    agg_metrics = {}
    agg_metrics["trader_address"] = trader_address
    total_bet_amounts = trader_data.collateral_amount.sum()
    total_num_mech_calls = trader_data.num_mech_calls.sum()
    agg_metrics["bet_amount"] = total_bet_amounts
    agg_metrics["nr_mech_calls"] = total_num_mech_calls
    agg_metrics["staking"] = trader_data.iloc[0].staking
    if live_metrics:
        return agg_metrics
    total_net_earnings = trader_data.net_earnings.sum()
    agg_metrics["net_earnings"] = total_net_earnings
    agg_metrics["earnings"] = trader_data.earnings.sum()
    agg_metrics["nr_trades"] = len(trader_data)
    total_fee_amounts = trader_data.mech_fee_amount.sum()
    total_costs = (
        total_bet_amounts
        + total_fee_amounts
        + (total_num_mech_calls * DEFAULT_MECH_FEE)
    )
    agg_metrics["roi"] = total_net_earnings / total_costs
    return agg_metrics


def compute_trader_metrics_by_market_creator(
    trader_address: str,
    traders_data: pd.DataFrame,
    market_creator: str = "all",
    live_metrics: bool = False,
) -> dict:
    """This function computes for a specific time window (week or day) the different metrics:
    roi, net_earnings, earnings, bet_amount, nr_mech_calls and nr_trades.
    The global roi of the trader agent by computing the individual net profit and the individual costs values
    achieved per market and dividing both.
    It is possible to filter by market creator: quickstart, pearl, all"""
    assert "market_creator" in traders_data.columns
    filtered_traders_data = traders_data.loc[
        traders_data["trader_address"] == trader_address
    ]
    if market_creator != "all":  # compute only for the specific market creator
        filtered_traders_data = filtered_traders_data.loc[
            filtered_traders_data["market_creator"] == market_creator
        ]
        if len(filtered_traders_data) == 0:
            # tqdm.write(f"No data. Skipping market creator {market_creator}")
            return {}  # No Data

    metrics = compute_metrics(trader_address, filtered_traders_data, live_metrics)
    return metrics


def compute_trader_metrics_by_trader_family(
    trader_address: str, traders_data: pd.DataFrame, trader_family: str = "all"
) -> dict:
    """This function computes for a specific time window (week or day) the different metrics:
    roi, net_earnings, earnings, bet_amount, nr_mech_calls and nr_trades.
    The global roi of the trader agent by computing the individual net profit and the individual costs values
    achieved per market and dividing both.
    It is possible to filter by trader family: quickstart_agent, pearl_agent, non_agent, all
    """
    assert "trader_family" in traders_data.columns
    filtered_traders_data = traders_data.loc[
        traders_data["trader_address"] == trader_address
    ]
    if trader_family != "all":  # compute only for the specific trader family
        filtered_traders_data = filtered_traders_data.loc[
            filtered_traders_data["trader_family"] == trader_family
        ]
        if len(filtered_traders_data) == 0:
            # tqdm.write(f"No data. Skipping trader family {trader_family}")
            return {}  # No Data

    metrics = compute_metrics(trader_address, filtered_traders_data)
    return metrics


def merge_trader_weekly_metrics(
    trader: str, weekly_data: pd.DataFrame, week: str
) -> pd.DataFrame:
    trader_metrics = []
    # computation as specification 1 for all types of markets
    weekly_metrics_all = compute_trader_metrics_by_market_creator(
        trader, weekly_data, market_creator="all"
    )
    weekly_metrics_all["month_year_week"] = week
    weekly_metrics_all["market_creator"] = "all"
    trader_metrics.append(weekly_metrics_all)

    # computation as specification 1 for quickstart markets
    weekly_metrics_qs = compute_trader_metrics_by_market_creator(
        trader, weekly_data, market_creator="quickstart"
    )
    if len(weekly_metrics_qs) > 0:
        weekly_metrics_qs["month_year_week"] = week
        weekly_metrics_qs["market_creator"] = "quickstart"
        trader_metrics.append(weekly_metrics_qs)
    # computation as specification 1 for pearl markets
    weekly_metrics_pearl = compute_trader_metrics_by_market_creator(
        trader, weekly_data, market_creator="pearl"
    )
    if len(weekly_metrics_pearl) > 0:
        weekly_metrics_pearl["month_year_week"] = week
        weekly_metrics_pearl["market_creator"] = "pearl"
        trader_metrics.append(weekly_metrics_pearl)
    result = pd.DataFrame.from_dict(trader_metrics, orient="columns")
    return result


def merge_trader_daily_metrics(
    trader: str, daily_data: pd.DataFrame, day: str, live_metrics: bool = False
) -> pd.DataFrame:
    trader_metrics = []
    # computation as specification 1 for all types of markets
    daily_metrics_all = compute_trader_metrics_by_market_creator(
        trader, daily_data, market_creator="all", live_metrics=live_metrics
    )
    daily_metrics_all["creation_date"] = day
    # staking label is at the trader level
    daily_metrics_all["market_creator"] = "all"
    trader_metrics.append(daily_metrics_all)

    # computation as specification 1 for quickstart markets
    daily_metrics_qs = compute_trader_metrics_by_market_creator(
        trader, daily_data, market_creator="quickstart", live_metrics=live_metrics
    )
    if len(daily_metrics_qs) > 0:
        daily_metrics_qs["creation_date"] = day
        daily_metrics_qs["market_creator"] = "quickstart"
        trader_metrics.append(daily_metrics_qs)
    # computation as specification 1 for pearl markets
    daily_metrics_pearl = compute_trader_metrics_by_market_creator(
        trader, daily_data, market_creator="pearl", live_metrics=live_metrics
    )
    if len(daily_metrics_pearl) > 0:
        daily_metrics_pearl["creation_date"] = day
        daily_metrics_pearl["market_creator"] = "pearl"
        trader_metrics.append(daily_metrics_pearl)
    result = pd.DataFrame.from_dict(trader_metrics, orient="columns")
    return result


def win_metrics_trader_level(weekly_data):
    winning_trades = (
        weekly_data.groupby(
            ["month_year_week", "market_creator", "trader_address"], sort=False
        )["winning_trade"].sum()
        / weekly_data.groupby(
            ["month_year_week", "market_creator", "trader_address"], 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.rename(columns={"winning_trade": "winning_perc"}, inplace=True)
    return winning_trades


def compute_weekly_metrics_by_market_creator(
    trader_agents_data: pd.DataFrame, trader_filter: str = None
) -> pd.DataFrame:
    """Function to compute the metrics at the trader level per week
    and with different categories by market creator"""
    contents = []
    all_weeks = list(trader_agents_data.month_year_week.unique())
    for week in all_weeks:
        weekly_data = trader_agents_data.loc[
            trader_agents_data["month_year_week"] == week
        ]
        print(f"Computing weekly metrics for week ={week} by market creator")
        # traverse each trader agent
        traders = list(weekly_data.trader_address.unique())
        for trader in tqdm(traders, desc=f"Trader' metrics", unit="metrics"):
            if trader_filter is None:
                contents.append(merge_trader_weekly_metrics(trader, weekly_data, week))
            elif trader_filter == "agent":
                filtered_data = weekly_data.loc[weekly_data["staking"] != "non_agent"]
                contents.append(
                    merge_trader_weekly_metrics(trader, filtered_data, week)
                )
            else:  # non_agent traders
                filtered_data = weekly_data.loc[weekly_data["staking"] == "non_agent"]
                contents.append(
                    merge_trader_weekly_metrics(trader, filtered_data, week)
                )

    print("End computing all weekly metrics by market creator")
    return pd.concat(contents, ignore_index=True)


def compute_daily_metrics_by_market_creator(
    trader_agents_data: pd.DataFrame,
    trader_filter: str = None,
    live_metrics: bool = False,
) -> pd.DataFrame:
    """Function to compute the metrics at the trader level per day
    and with different categories by market creator"""
    contents = []

    all_days = list(trader_agents_data.creation_date.unique())
    for day in all_days:
        daily_data = trader_agents_data.loc[trader_agents_data["creation_date"] == day]
        print(f"Computing daily metrics for {day}")
        # traverse each trader agent
        traders = list(daily_data.trader_address.unique())
        for trader in tqdm(traders, desc=f"Trader' daily metrics", unit="metrics"):
            if trader_filter is None:
                contents.append(
                    merge_trader_daily_metrics(trader, daily_data, day, live_metrics)
                )
            elif trader_filter == "agentic":
                filtered_data = daily_data.loc[daily_data["staking"] != "non_agent"]
                contents.append(
                    merge_trader_daily_metrics(trader, filtered_data, day, live_metrics)
                )
            else:  # non_agent traders
                filtered_data = daily_data.loc[daily_data["staking"] == "non_agent"]
                contents.append(
                    merge_trader_daily_metrics(trader, filtered_data, day, live_metrics)
                )
    print("End computing all daily metrics by market creator")
    return pd.concat(contents, ignore_index=True)


def compute_winning_metrics_by_trader(
    trader_agents_data: pd.DataFrame, trader_filter: str = None
) -> pd.DataFrame:
    """Function to compute the winning metrics at the trader level per week and with different market creators"""
    market_all = trader_agents_data.copy(deep=True)
    market_all["market_creator"] = "all"

    # merging both dataframes
    final_traders = pd.concat([market_all, trader_agents_data], ignore_index=True)
    final_traders = final_traders.sort_values(by="creation_timestamp", ascending=True)

    if trader_filter == "agentic":
        final_traders = final_traders.loc[final_traders["staking"] != "non_agent"]
    else:  # non_agent traders
        final_traders = final_traders.loc[final_traders["staking"] == "non_agent"]

    winning_df = win_metrics_trader_level(final_traders)
    winning_df.head()
    return winning_df