updating dashboards with new data

data/all_trades_profitability.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:6256840b7a7704aa5618fd5a4fed41b9444bbf80ea1dcaae068715026c8d52b0
-size 8218375
+oid sha256:6a9fa6fcb351f21334b3a79194c82d9b344be2549d7cd9398c3dea84646291dc
+size 11576111

data/all_trades_profitability_bak.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:ebdebf4884de2bb27a71c4d5144a6dad8f5f5c6f2675ac3b12993c4240b0e2de
+size 7910144

data/daily_info.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:f3d8ec77951dad3d522c90ea0009c15e5ab717c3f34624b4f0d205ad58cfa16e
-size 1054780
+oid sha256:dafbb1cf7614f3040c27aeb6f130e19068e9ca56159ed21a584abd4a925db977
+size 566350

data/error_by_markets.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:9dff09a27b7b5ac4a527d679c446627c6ca4fb2653c6bc50e818d79e29e3c1be
-size 12928
+oid sha256:026835121a261b46a391e397160f878413bd1f337903324bb1cd6ef88bc2d90c
+size 12990

data/invalid_trades.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:099e999dc46d4a2d7086838f3645475aecf27fa88331a8b2d5fd4c9937f1ad81
-size 782151
+oid sha256:8a47b04bfae501edb6051f0089117b51fc96cdb4eeb4ad5ef3ebfbcd7ee19590
+size 755966

data/service_map.pkl

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:32d288a076f719a659159ffdb2bca3f132c3efe3f62ee0412c11e8094c36ffc8
-size 164076
+oid sha256:da35d7c845c32bc90c5f298210458dfa01b8745bf95bd23b62a49c044ca06ac7
+size 167913

data/tools_accuracy.csv

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:bb5a70b32e6a7dbd75c7924a2fa887612bf7523a62f6710f2e2397cdc3664fa2
-size 1100
+oid sha256:3a26951af78022999c8651278aff7a96ed271a6d325907096982b5333a7af7b7
+size 1099

data/unknown_traders.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:1633afc5d408263251ae5290e1f45972abaf0d3f0358ab880604de8a0baae559
-size 283140
+oid sha256:cd2aa16659f93661ac3e9744f55ba17d921bf5f65c843a60f01232dd587254bf
+size 365566

data/winning_df.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:f394838074669231dc3f8dc46167bb05019ae12eb798933e99b2c2de9b9a2c1f
-size 12636
+oid sha256:10f727eac679040aa5f1dc1856580010c315b3f72810cdbcaa6175f5a8343c0e
+size 12528

scripts/active_traders.py

@@ -0,0 +1,91 @@
+import pandas as pd
+import pickle
+from web3_utils import DATA_DIR, TMP_DIR
+from staking import check_list_addresses
+
+
+def get_trader_type(address: str, service_map: dict) -> str:
+    # check if it is part of any service id on the map
+    keys = service_map.keys()
+    last_key = max(keys)
+
+    for key, value in service_map.items():
+        if value["safe_address"].lower() == address.lower():
+            # found a service
+            return "Olas"
+
+    return "non_Olas"
+
+
+def compute_active_traders_dataset():
+    """Function to prepare the active traders dataset"""
+    with open(DATA_DIR / "service_map.pkl", "rb") as f:
+        service_map = pickle.load(f)
+    # read tools info
+    tools_df = pd.read_parquet(TMP_DIR / "tools.parquet")
+    # rename the request_month_year_week
+    tools_df.rename(
+        columns={"request_month_year_week": "month_year_week"}, inplace=True
+    )
+    tool_traders = tools_df.trader_address.unique()
+    mapping = check_list_addresses(tool_traders)
+    # add trader type to tools_df
+    tools_df["trader_type"] = tools_df.trader_address.apply(lambda x: mapping[x])
+    tools_df = tools_df[
+        ["month_year_week", "market_creator", "trader_type", "trader_address"]
+    ]
+    tools_df.drop_duplicates(inplace=True)
+    # read trades info
+    all_trades = pd.read_parquet(DATA_DIR / "all_trades_profitability.parquet")
+
+    # read unknown info
+    unknown_traders = pd.read_parquet(DATA_DIR / "unknown_traders.parquet")
+    unknown_traders["creation_timestamp"] = pd.to_datetime(
+        unknown_traders["creation_timestamp"]
+    )
+    unknown_traders["creation_timestamp"] = unknown_traders[
+        "creation_timestamp"
+    ].dt.tz_convert("UTC")
+    unknown_traders = unknown_traders.sort_values(
+        by="creation_timestamp", ascending=True
+    )
+    unknown_traders["month_year_week"] = (
+        unknown_traders["creation_timestamp"]
+        .dt.to_period("W")
+        .dt.start_time.dt.strftime("%b-%d-%Y")
+    )
+    unknown_traders["trader_type"] = "unknown"
+    unknown_traders = unknown_traders[
+        ["month_year_week", "trader_type", "market_creator", "trader_address"]
+    ]
+    unknown_traders.drop_duplicates(inplace=True)
+
+    all_trades["creation_timestamp"] = pd.to_datetime(all_trades["creation_timestamp"])
+    all_trades["creation_timestamp"] = all_trades["creation_timestamp"].dt.tz_convert(
+        "UTC"
+    )
+    all_trades = all_trades.sort_values(by="creation_timestamp", ascending=True)
+    all_trades["month_year_week"] = (
+        all_trades["creation_timestamp"]
+        .dt.to_period("W")
+        .dt.start_time.dt.strftime("%b-%d-%Y")
+    )
+    all_trades["trader_type"] = all_trades["staking"].apply(
+        lambda x: "non_Olas" if x == "non_Olas" else "Olas"
+    )
+    all_trades = all_trades[
+        ["month_year_week", "market_creator", "trader_type", "trader_address"]
+    ]
+    all_trades.drop_duplicates(inplace=True)
+    filtered_traders_data = pd.concat([all_trades, tools_df], axis=0)
+    filtered_traders_data.drop_duplicates(inplace=True)
+    if len(unknown_traders) > 0:
+        # merge
+        filtered_traders_data = pd.concat(
+            [filtered_traders_data, unknown_traders], axis=0
+        )
+    filtered_traders_data.to_parquet(TMP_DIR / "active_traders.parquet")
+
+
+if __name__ == "__main__":
+    compute_active_traders_dataset()

scripts/daily_data.py

@@ -58,3 +58,4 @@ def prepare_live_metrics(
 
 if __name__ == "__main__":
     prepare_live_metrics()
+    # generate_retention_activity_file()

scripts/get_mech_info.py

@@ -117,63 +117,6 @@ def update_json_files():
     merge_json_files("tools_info.json", "new_tools_info.json")
 
 
-def update_fpmmTrades_parquet(trades_filename: str) -> pd.DataFrame:
-    # Read old trades parquet file
-    try:
-        old_trades_df = pd.read_parquet(TMP_DIR / "fpmmTrades.parquet")
-    except Exception as e:
-        print(f"Error reading old trades parquet file {e}")
-        return None
-
-    try:
-        new_trades_df = pd.read_parquet(DATA_DIR / trades_filename)
-    except Exception as e:
-        print(f"Error reading new trades parquet file {e}")
-        return None
-
-    # lowercase and strip creator_address
-    new_trades_df["trader_address"] = (
-        new_trades_df["trader_address"].str.lower().str.strip()
-    )
-    # ensure creationTimestamp compatibility
-    try:
-        new_trades_df["creationTimestamp"] = new_trades_df["creationTimestamp"].apply(
-            lambda x: transform_to_datetime(x)
-        )
-
-    except Exception as e:
-        print(f"Transformation not needed")
-    try:
-        old_trades_df["creationTimestamp"] = old_trades_df["creationTimestamp"].apply(
-            lambda x: transform_to_datetime(x)
-        )
-    except Exception as e:
-        print(f"Transformation not needed")
-
-    # merge two dataframes
-    merge_df = pd.concat([old_trades_df, new_trades_df], ignore_index=True)
-    # avoid numpy objects
-    merge_df["fpmm.arbitrationOccurred"] = merge_df["fpmm.arbitrationOccurred"].astype(
-        bool
-    )
-    merge_df["fpmm.isPendingArbitration"] = merge_df[
-        "fpmm.isPendingArbitration"
-    ].astype(bool)
-
-    # Check for duplicates
-    print(f"Initial length before removing duplicates in fpmmTrades= {len(merge_df)}")
-
-    # Remove duplicates
-    # fpmm.outcomes is a numpy array
-    merge_df.drop_duplicates("id", keep="last", inplace=True)
-    print(f"Final length after removing duplicates in fpmmTrades= {len(merge_df)}")
-
-    # save the parquet file
-    merge_df.to_parquet(TMP_DIR / "fpmmTrades.parquet", index=False)
-
-    return
-
-
 def update_all_trades_parquet(new_trades_df: pd.DataFrame) -> pd.DataFrame:
     # Read old all_trades parquet file
     try:
@@ -315,11 +258,11 @@ def get_mech_events_since_last_run(logger):
     # Read the latest date from stored data
     try:
         all_trades = pd.read_parquet(DATA_DIR / "all_trades_profitability.parquet")
-        latest_timestamp = max(all_trades.creation_timestamp)
-        # cutoff_date = "2024-12-22"
-        # latest_timestamp = pd.Timestamp(
-        #     datetime.strptime(cutoff_date, "%Y-%m-%d")
-        # ).tz_localize("UTC")
+        # latest_timestamp = max(all_trades.creation_timestamp)
+        cutoff_date = "2025-01-13"
+        latest_timestamp = pd.Timestamp(
+            datetime.strptime(cutoff_date, "%Y-%m-%d")
+        ).tz_localize("UTC")
         print(f"Updating data since {latest_timestamp}")
     except Exception:
         print("Error while reading the profitability parquet file")

scripts/markets.py

@@ -357,7 +357,7 @@ def fpmmTrades_etl(
     # lowercase and strip creator_address
     fpmmTrades["trader_address"] = fpmmTrades["trader_address"].str.lower().str.strip()
     fpmmTrades.to_parquet(DATA_DIR / trades_filename, index=False)
-    return
+    return fpmmTrades
 
 
 def check_current_week_data(trades_df: pd.DataFrame) -> pd.DataFrame:
@@ -390,5 +390,75 @@ def check_current_week_data(trades_df: pd.DataFrame) -> pd.DataFrame:
     return trades_df
 
 
+def update_fpmmTrades_parquet(trades_filename: str) -> pd.DataFrame:
+    # Read old trades parquet file
+    try:
+        old_trades_df = pd.read_parquet(TMP_DIR / "fpmmTrades.parquet")
+    except Exception as e:
+        print(f"Error reading old trades parquet file {e}")
+        return None
+
+    try:
+        new_trades_df = pd.read_parquet(DATA_DIR / trades_filename)
+    except Exception as e:
+        print(f"Error reading new trades parquet file {e}")
+        return None
+
+    # lowercase and strip creator_address
+    new_trades_df["trader_address"] = (
+        new_trades_df["trader_address"].str.lower().str.strip()
+    )
+    # ensure creationTimestamp compatibility
+    try:
+        new_trades_df["creationTimestamp"] = new_trades_df["creationTimestamp"].apply(
+            lambda x: transform_to_datetime(x)
+        )
+
+    except Exception as e:
+        print(f"Transformation not needed")
+    try:
+        old_trades_df["creationTimestamp"] = old_trades_df["creationTimestamp"].apply(
+            lambda x: transform_to_datetime(x)
+        )
+    except Exception as e:
+        print(f"Transformation not needed")
+
+    # merge two dataframes
+    merge_df = pd.concat([old_trades_df, new_trades_df], ignore_index=True)
+    # avoid numpy objects
+    merge_df["fpmm.arbitrationOccurred"] = merge_df["fpmm.arbitrationOccurred"].astype(
+        bool
+    )
+    merge_df["fpmm.isPendingArbitration"] = merge_df[
+        "fpmm.isPendingArbitration"
+    ].astype(bool)
+
+    # Check for duplicates
+    print(f"Initial length before removing duplicates in fpmmTrades= {len(merge_df)}")
+
+    # Remove duplicates
+    # fpmm.outcomes is a numpy array
+    merge_df.drop_duplicates("id", keep="last", inplace=True)
+    print(f"Final length after removing duplicates in fpmmTrades= {len(merge_df)}")
+
+    # save the parquet file
+    merge_df.to_parquet(TMP_DIR / "fpmmTrades.parquet", index=False)
+
+    return
+
+
+def update_fpmmTrades(from_date: str):
+
+    from_timestamp = pd.Timestamp(datetime.strptime(from_date, "%Y-%m-%d")).tz_localize(
+        "UTC"
+    )
+    fpmmTrades_etl(
+        trades_filename="new_fpmmTrades.parquet",
+        from_timestamp=int(from_timestamp.timestamp()),
+    )
+    update_fpmmTrades_parquet("new_fpmmTrades.parquet")
+
+
 if __name__ == "__main__":
-    etl("all_fpmms.parquet")
+    cutoff_date = "2025-01-13"
+    update_fpmmTrades(cutoff_date)

scripts/profitability.py

@@ -37,6 +37,7 @@ from utils import (
     DATA_DIR,
     DEFAULT_MECH_FEE,
     TMP_DIR,
+    measure_execution_time,
 )
 from staking import label_trades_by_staking
 from nr_mech_calls import (
@@ -141,13 +142,16 @@ def _is_redeemed(user_json: dict[str, Any], fpmmTrade: dict[str, Any]) -> bool:
 def prepare_profitalibity_data(
     tools_filename: str,
     trades_filename: str,
+    tmp_dir: bool = False,
 ) -> pd.DataFrame:
     """Prepare data for profitalibity analysis."""
 
     # Check if tools.parquet is in the same directory
     try:
-        # tools parquet file
-        tools = pd.read_parquet(DATA_DIR / tools_filename)
+        if tmp_dir:
+            tools = pd.read_parquet(TMP_DIR / tools_filename)
+        else:
+            tools = pd.read_parquet(DATA_DIR / tools_filename)
 
         # make sure creator_address is in the columns
         assert "trader_address" in tools.columns, "trader_address column not found"
@@ -167,9 +171,12 @@ def prepare_profitalibity_data(
     # Check if fpmmTrades.parquet is in the same directory
     print("Reading the new trades file")
     try:
-        fpmmTrades = pd.read_parquet(DATA_DIR / trades_filename)
+        if tmp_dir:
+            fpmmTrades = pd.read_parquet(TMP_DIR / trades_filename)
+        else:
+            fpmmTrades = pd.read_parquet(DATA_DIR / trades_filename)
     except FileNotFoundError:
-        print(f"Error reading {trades_filename} file .")
+        print(f"Error reading {trades_filename} file.")
 
     # make sure trader_address is in the columns
     assert "trader_address" in fpmmTrades.columns, "trader_address column not found"
@@ -352,16 +359,18 @@ def analyse_all_traders(
     return all_creators_df
 
 
+@measure_execution_time
 def run_profitability_analysis(
     tools_filename: str,
     trades_filename: str,
     merge: bool = False,
+    tmp_dir: bool = False,
 ):
     """Create all trades analysis."""
-
-    # load dfs from data folder for analysis
     print(f"Preparing data with {tools_filename} and {trades_filename}")
-    fpmmTrades = prepare_profitalibity_data(tools_filename, trades_filename)
+    fpmmTrades = prepare_profitalibity_data(
+        tools_filename, trades_filename, tmp_dir=tmp_dir
+    )
 
     if merge:
         update_tools_parquet(tools_filename)
@@ -391,6 +400,7 @@ def run_profitability_analysis(
 
     # debugging purposes
     all_trades_df.to_parquet(JSON_DATA_DIR / "all_trades_df.parquet", index=False)
+    # all_trades_df = pd.read_parquet(JSON_DATA_DIR / "all_trades_df.parquet")
 
     # filter invalid markets. Condition: "is_invalid" is True
     invalid_trades = all_trades_df.loc[all_trades_df["is_invalid"] == True]
@@ -512,12 +522,9 @@ def add_trades_profitability(trades_filename: str):
 
 
 if __name__ == "__main__":
-    # updating the whole fpmmTrades parquet file instead of just the new ones
-    # trade_mech_calls = pd.read_parquet(TMP_DIR / "result_df.parquet")
-    # fpmmTrades = pd.read_parquet(TMP_DIR / "fpmmTrades.parquet")
-    # fpmmTrades["creationTimestamp"] = fpmmTrades["creationTimestamp"].apply(
-    #     lambda x: transform_to_datetime(x)
-    # )
-    # all_trades_df = analyse_all_traders(fpmmTrades, trade_mech_calls)
-    # all_trades_df.to_parquet(TMP_DIR / "all_trades_df.parquet", index=False)
-    run_profitability_analysis("file1", "file2")
+    run_profitability_analysis(
+        tools_filename="tools.parquet",
+        trades_filename="fpmmTrades.parquet",
+        merge=False,
+        tmp_dir=True,
+    )

scripts/pull_data.py

@@ -1,8 +1,13 @@
 import logging
 from datetime import datetime
 import pandas as pd
-from markets import etl as mkt_etl, DEFAULT_FILENAME as MARKETS_FILENAME, fpmmTrades_etl
-from tools import DEFAULT_FILENAME as TOOLS_FILENAME, generate_tools_file
+from markets import (
+    etl as mkt_etl,
+    DEFAULT_FILENAME as MARKETS_FILENAME,
+    fpmmTrades_etl,
+    update_fpmmTrades_parquet,
+)
+from tools import generate_tools_file
 from profitability import run_profitability_analysis, add_trades_profitability
 from utils import (
     get_question,
@@ -15,7 +20,6 @@ from utils import (
 )
 from get_mech_info import (
     get_mech_events_since_last_run,
-    update_fpmmTrades_parquet,
     update_json_files,
 )
 from update_tools_accuracy import compute_tools_accuracy
@@ -131,7 +135,7 @@ def only_new_weekly_analysis():
 
     save_historical_data()
     try:
-        clean_old_data_from_parquet_files("2024-11-19")
+        clean_old_data_from_parquet_files("2024-11-26")
     except Exception as e:
         print("Error cleaning the oldest information from parquet files")
         print(f"reason = {e}")

scripts/staking.py

@@ -1,7 +1,7 @@
 import json
 import sys
 from typing import Any, List
-from utils import RPC, DATA_DIR, TMP_DIR
+from utils import RPC, DATA_DIR, TMP_DIR, JSON_DATA_DIR
 import requests
 from tqdm import tqdm
 from web3 import Web3
@@ -148,6 +148,7 @@ def update_service_map(start: int = 1, end: int = 2000):
             if partial_dict:
                 service_map.update(partial_dict)
 
+    print(f"length of service map {len(service_map)}")
     with open(DATA_DIR / "service_map.pkl", "wb") as f:
         pickle.dump(service_map, f)
 
@@ -223,6 +224,7 @@ def generate_retention_activity_file():
     ]
     print(f"length of reduced tools before labeling = {len(reduced_tools_df)}")
     reduced_tools_df = label_trades_by_staking(trades_df=reduced_tools_df)
+    print(f"labeling of tools activity. {reduced_tools_df.staking.value_counts()}")
     print(f"length of reduced tools after labeling = {len(reduced_tools_df)}")
     reduced_tools_df = reduced_tools_df.sort_values(by="request_time", ascending=True)
     reduced_tools_df["month_year_week"] = (
@@ -246,17 +248,20 @@ def check_list_addresses(address_list: list):
     print(f"last service key = {last_key}")
     update_service_map(start=last_key)
     found_key = -1
+    trader_types = []
     for trader_address in address_list:
         for key, value in service_map.items():
             if value["safe_address"].lower() == trader_address.lower():
                 # found a service
                 found_key = key
                 mapping[trader_address] = "Olas"
+                trader_types.append("Olas")
+                break
 
         if found_key == -1:
             mapping[trader_address] = "non_Olas"
-    print("mapping")
-    print(mapping)
+            trader_types.append("non_Olas")
+    return mapping
 
 
 def check_service_map():
@@ -278,20 +283,20 @@ def check_service_map():
 
 if __name__ == "__main__":
     # create_service_map()
-    # trades_df = pd.read_parquet(TMP_DIR / "all_trades_df.parquet")
-    # trades_df = trades_df.loc[trades_df["is_invalid"] == False]
+    trades_df = pd.read_parquet(JSON_DATA_DIR / "all_trades_df.parquet")
+    trades_df = trades_df.loc[trades_df["is_invalid"] == False]
 
-    # trades_df = label_trades_by_staking(trades_df=trades_df, start=8)
-    # print(trades_df.staking.value_counts())
+    trades_df = label_trades_by_staking(trades_df=trades_df)
+    print(trades_df.staking.value_counts())
     # trades_df.to_parquet(TMP_DIR / "result_staking.parquet", index=False)
     # generate_retention_activity_file()
-    a_list = [
-        "0x027592700fafc4db3221bb662d7bdc7f546a2bb5",
-        "0x0845f4ad01a2f41da618848c7a9e56b64377965e",
-    ]
+    # a_list = [
+    #     "0x027592700fafc4db3221bb662d7bdc7f546a2bb5",
+    #     "0x0845f4ad01a2f41da618848c7a9e56b64377965e",
+    # ]
     # check_list_addresses(address_list=a_list)
     # update_service_map()
     # check_service_map()
-    unknown_traders = pd.read_parquet(DATA_DIR / "unknown_traders.parquet")
-    unknown_traders = label_trades_by_staking(trades_df=unknown_traders)
-    unknown_traders.to_parquet(DATA_DIR / "unknown_traders.parquet", index=False)
+    # unknown_traders = pd.read_parquet(DATA_DIR / "unknown_traders.parquet")
+    # unknown_traders = label_trades_by_staking(trades_df=unknown_traders)
+    # unknown_traders.to_parquet(DATA_DIR / "unknown_traders.parquet", index=False)