new weekly data using new pipeline

data/all_trades_profitability.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:54d1564ac5393d51c6a7218a9d6afabb78c6166f2661afa5c68fe8ec166ba213
-size 2885806
+oid sha256:1df952a693ba00cc0b11bca8ff4c6805415f2d006b3cd47242e43e7cdc7d5fe1
+size 3266876

data/fpmmTrades.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:6c95bdbd08b0fa875044c53bfc983c0e4a76d861c703da2f4a04cdec20a9f13c
-size 8600942
+oid sha256:390f96495582e69ae82225a61e6473c1fe6536081b326a6bd11617be45ce672a
+size 10816943

data/fpmms.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:2b4ad4e1780f6dcb812787d397a738141e53db0fc6b588f386586dddabca034e
-size 494787
+oid sha256:3b5dcd19c7922e3f7168a139b0d63c335c921343faa15852b6ae04888f7e006a
+size 504817

data/invalid_trades.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:8199109e6224b609408037098e23f11f6e38a2f2526f9376fca895ee2728edd9
-size 77169
+oid sha256:30e0fa52d0c605961b5a12bec47bc3b0288b02b814c61cc7f8a33ad793f8bd30
+size 84013

data/summary_profitability.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:2c7253ea40d66ba0f06955bb2b96a589fb20e154eb2cfd95db85872d80d80a4b
-size 71470
+oid sha256:a55a6c4c7ef5da8db27b61c268eccbd1d426c456a0d17efa4b22b7c69ed1454d
+size 78788

data/t_map.pkl

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:b0705cfa2166283351129e1dac63aab41a4231e60f6873d3026eb23da5cdbf27
-size 21870626
+oid sha256:7b67cf178943b82b5286b7c2adb6329e1e23fffce807ebf299684746813f55de
+size 22992649

data/tools.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:c5594bd432a7db7fc7bf8ccbf5c40a10bd452643e2a884b5ae221a9bfe21271e
-size 342399362
+oid sha256:7e4ace8d172836c379ee23bde678f19d9eeec28e7bd31bf9e95dc914ac5c9bc5
+size 407088092

data/tools_accuracy.csv

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:9ca0e77ced7ed3d627b0cbc2028af4116629135ed781b4e12c3c410316208a72
-size 1121
+oid sha256:f7a3622338d1eb2f23824031733ecdd77ae77eff7cb2b1c879aba05b0966d2cc
+size 1133

scripts/get_mech_info.py

@@ -120,7 +120,7 @@ def update_json_files():
     merge_json_files("tools_info.json", "new_tools_info.json")
 
 
-def update_trades_parquet(new_trades_df: pd.DataFrame) -> pd.DataFrame:
+def update_fpmmTrades_parquet(trades_filename: str) -> pd.DataFrame:
     # Read old trades parquet file
     try:
         old_trades_df = pd.read_parquet(DATA_DIR / "fpmmTrades.parquet")
@@ -128,58 +128,78 @@ def update_trades_parquet(new_trades_df: pd.DataFrame) -> pd.DataFrame:
         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
+
     # 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
-    duplicates = merge_df.duplicated()
-
-    # Print the duplicates
-    print(duplicates)
-
-    # Get the number of duplicates
-    num_duplicates = duplicates.sum()
-    print("Number of duplicates:", num_duplicates)
-
-    # Get the rows with duplicates
-    duplicate_rows = merge_df[duplicates]
-    print("Duplicate rows:\n", duplicate_rows)
+    print(f"Initial length before removing duplicates= {len(merge_df)}")
 
     # Remove duplicates
-    merge_df.drop_duplicates(inplace=True)
+    # fpmm.outcomes is a numpy array
+    merge_df = merge_df.drop_duplicates(
+        subset=[col for col in merge_df.columns if col != "fpmm.outcomes"]
+    )
+    print(f"Final length before removing duplicates= {len(merge_df)}")
 
     # save the parquet file
     merge_df.to_parquet(DATA_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:
+        old_trades_df = pd.read_parquet(DATA_DIR / "all_trades_profitability.parquet")
+    except Exception as e:
+        print(f"Error reading old trades parquet file {e}")
+        return None
+    # merge two dataframes
+    merge_df = pd.concat([old_trades_df, new_trades_df], ignore_index=True)
+
+    # Check for duplicates
+    print(f"Initial length before removing duplicates= {len(merge_df)}")
+
+    # Remove duplicates
+    merge_df.drop_duplicates(inplace=True)
+    print(f"Final length before removing duplicates= {len(merge_df)}")
     return merge_df
 
 
-def update_tools_parquet(new_tools_df: pd.DataFrame):
+def update_tools_parquet(new_tools_filename: pd.DataFrame):
     try:
         old_tools_df = pd.read_parquet(DATA_DIR / "tools.parquet")
     except Exception as e:
         print(f"Error reading old tools parquet file {e}")
         return None
+    try:
+        new_tools_df = pd.read_parquet(DATA_DIR / new_tools_filename)
+    except Exception as e:
+        print(f"Error reading new trades parquet file {e}")
+        return None
 
     # merge two dataframes
     merge_df = pd.concat([old_tools_df, new_tools_df], ignore_index=True)
 
     # Check for duplicates
-    duplicates = merge_df.duplicated()
-
-    # Print the duplicates
-    print(duplicates)
-
-    # Get the number of duplicates
-    num_duplicates = duplicates.sum()
-    print("Number of duplicates:", num_duplicates)
-
-    # Get the rows with duplicates
-    duplicate_rows = merge_df[duplicates]
-    print("Duplicate rows:\n", duplicate_rows)
+    print(f"Initial length before removing duplicates= {len(merge_df)}")
 
     # Remove duplicates
     merge_df.drop_duplicates(inplace=True)
+    print(f"Final length before removing duplicates= {len(merge_df)}")
 
     # save the parquet file
     merge_df.to_parquet(DATA_DIR / "tools.parquet", index=False)
@@ -290,6 +310,13 @@ def get_mech_events_since_last_run():
         int(latest_timestamp.timestamp()),
         int((latest_timestamp + five_seconds).timestamp()),
     )
+    # expecting only one block
+    last_run_block_number = last_run_block_number.get("number", "")
+    if last_run_block_number.isdigit():
+        last_run_block_number = int(last_run_block_number)
+
+    if last_run_block_number == "":
+        raise ValueError("Could not find a valid block number for last collected data")
     last_block_number = get_last_block_number()
 
     # mech requests

scripts/profitability.py

@@ -31,10 +31,18 @@ import os
 from pathlib import Path
 from get_mech_info import (
     DATETIME_60_DAYS_AGO,
-    update_trades_parquet,
+    update_fpmmTrades_parquet,
     update_tools_parquet,
+    update_all_trades_parquet,
+)
+from utils import (
+    SUBGRAPH_API_KEY,
+    wei_to_unit,
+    convert_hex_to_int,
+    _to_content,
+    read_parquet_files,
+    JSON_DATA_DIR,
 )
-from utils import SUBGRAPH_API_KEY, wei_to_unit, convert_hex_to_int, _to_content
 from queries import omen_xdai_trades_query, conditional_tokens_gc_user_query
 from staking import label_trades_by_staking
 
@@ -567,13 +575,19 @@ def run_profitability_analysis(
     print("Analysing trades...")
     all_trades_df = analyse_all_traders(fpmmTrades, tools)
 
-    # TODO merge previous files if requested
+    # merge previous files if requested
     if merge:
-        all_trades_df = update_trades_parquet(all_trades_df)
+        update_fpmmTrades_parquet(trades_filename)
         update_tools_parquet(tools_filename)
+        all_trades_df = update_all_trades_parquet(all_trades_df)
 
     # filter invalid markets. Condition: "is_invalid" is True
     invalid_trades = all_trades_df.loc[all_trades_df["is_invalid"] == True]
+    if merge:
+        try:
+            old_invalid_trades = pd.read_parquet(DATA_DIR / "invalid_trades.parquet")
+            merge_df = pd.concat([old_invalid_trades, invalid_trades], ignore_index=True)
+            invalid_trades = merge_df.drop_duplicates()
     invalid_trades.to_parquet(DATA_DIR / "invalid_trades.parquet", index=False)
 
     all_trades_df = all_trades_df.loc[all_trades_df["is_invalid"] == False]

scripts/pull_data.py

@@ -141,9 +141,8 @@ def only_new_weekly_analysis():
 
     add_current_answer("new_tools.parquet")
 
-    # Run profitability analysis
+    # # Run profitability analysis
     logging.info("Running profitability analysis")
-
     run_profitability_analysis(
         rpc=rpc,
         tools_filename="new_tools.parquet",
@@ -213,7 +212,8 @@ def weekly_analysis():
 
 
 if __name__ == "__main__":
-    weekly_analysis()
+    only_new_weekly_analysis()
+    # weekly_analysis()
     # rpc = RPC
     # updating_timestamps(rpc)
     # compute_tools_accuracy()

scripts/utils.py

@@ -428,3 +428,33 @@ def _to_content(q: str) -> dict[str, Any]:
         "extensions": {"headers": None},
     }
     return finalized_query
+
+
+def read_parquet_files(tools_filename: str, trades_filename: str):
+    # Check if tools.parquet is in the same directory
+    try:
+        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"
+
+        # lowercase and strip creator_address
+        tools["trader_address"] = tools["trader_address"].str.lower().str.strip()
+
+        # drop duplicates
+        tools.drop_duplicates(inplace=True)
+
+        print(f"{tools_filename} loaded")
+    except FileNotFoundError:
+        print("tools.parquet not found. Please run tools.py first.")
+        return
+    try:
+        fpmmTrades = pd.read_parquet(DATA_DIR / trades_filename)
+        fpmmTrades["trader_address"] = (
+            fpmmTrades["trader_address"].str.lower().str.strip()
+        )
+    except FileNotFoundError:
+        print("fpmmsTrades.parquet not found.")
+        return
+
+    return tools, fpmmTrades