weekly data and some fixes

data/all_trades_profitability.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:08d3f27f13f20022a8aa176e3f84559a38c001d03753fd13101eae15ab4a1123
-size 6500230
+oid sha256:04134fbaceed13c1c02eeb952134df58ec787afddcba0e09b69b7ea5ae8693cd
+size 6590780

data/daily_info.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:689d0f6abf1886fb0b089db0294e07c551485005fabbaf73ea36083d6388ce5f
-size 78026
+oid sha256:880e896031e8e87ddc4e9fc952941d7bbe5196b86e56ebdf003e572182db6b76
+size 1050038

data/error_by_markets.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:7832acf3c67797c32225d4c253d651ffcbd4aba4bb19354ea00ad0c1d7200cc1
-size 13489
+oid sha256:5b0f69c6df66a53208bd47167d39db8ef7ec965a47a1a90fe429815cc44cbad9
+size 14168

data/invalid_trades.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:6156d839803a0c09606450ead26fb3b31ef6ef30bf3507ccbd0dc0399ad366f7
-size 152273
+oid sha256:f29e5a59d7a2c60ef8408cddf58785f601a472105a99bd7339a45de8389fa36f
+size 159829

data/tools_accuracy.csv

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

data/unknown_traders.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:3620eaba76778060f41059fb2b6ff6e92a6000eedfd9a9119b703f84cdda11ff
-size 194084
+oid sha256:5c237838416f8339b145719e5b370df1d9763099f9e5fa3e75d4d69053e5d311
+size 200722

data/winning_df.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:1ad71600af38478d3dc275bf7ad0f3cd3cd46d44d1bf54f06bb6f9d9540cc041
-size 13714
+oid sha256:6c31991027458844d3c8e72da29bb9acd7bcef8ad31ae7c95a59ee168a34ba58
+size 14407

notebooks/mech_calls_analysis.ipynb

@@ -2,26 +2,132 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 1,
+   "execution_count": 7,
    "metadata": {},
    "outputs": [],
    "source": [
     "import pandas as pd\n",
     "import matplotlib.pyplot as plt\n",
     "import seaborn as sns\n",
-    "import gc\n",
+    "from datetime import datetime, timezone\n",
     "sns.set_style(\"darkgrid\")"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": 8,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "new_trades =  pd.read_parquet('../tmp/new_fpmmTrades.parquet')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 9,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def transform_to_datetime(x):\n",
+    "    return datetime.fromtimestamp(int(x), tz=timezone.utc)\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 10,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "new_trades[\"creationTimestamp\"] = new_trades[\"creationTimestamp\"].apply(\n",
+    "    lambda x: transform_to_datetime(x)\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "Index(['collateralAmount', 'collateralAmountUSD', 'collateralToken',\n",
+       "       'creationTimestamp', 'trader_address', 'feeAmount', 'id',\n",
+       "       'oldOutcomeTokenMarginalPrice', 'outcomeIndex',\n",
+       "       'outcomeTokenMarginalPrice', 'outcomeTokensTraded', 'title',\n",
+       "       'transactionHash', 'type', 'market_creator',\n",
+       "       'fpmm.answerFinalizedTimestamp', 'fpmm.arbitrationOccurred',\n",
+       "       'fpmm.currentAnswer', 'fpmm.id', 'fpmm.isPendingArbitration',\n",
+       "       'fpmm.openingTimestamp', 'fpmm.outcomes', 'fpmm.title',\n",
+       "       'fpmm.condition.id'],\n",
+       "      dtype='object')"
+      ]
+     },
+     "execution_count": 5,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "new_trades.columns"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 11,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "Timestamp('2025-01-07 09:01:05+0000', tz='UTC')"
+      ]
+     },
+     "execution_count": 11,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "min(new_trades.creationTimestamp)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 12,
    "metadata": {},
    "outputs": [],
    "source": [
     "all_trades = pd.read_parquet('../data/all_trades_profitability.parquet')"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": 13,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "Timestamp('2025-01-07 04:43:00+0000', tz='UTC')"
+      ]
+     },
+     "execution_count": 13,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "max(all_trades.creation_timestamp)"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": 3,

scripts/daily_data.py

@@ -10,6 +10,7 @@ from nr_mech_calls import (
     compute_daily_mech_calls,
     transform_to_datetime,
 )
+from markets import check_current_week_data
 
 logging.basicConfig(level=logging.INFO)
 
@@ -21,12 +22,15 @@ def prepare_live_metrics(
     fpmmTrades = pd.read_parquet(TMP_DIR / trades_filename)
     tools = pd.read_parquet(TMP_DIR / tools_filename)
 
+    # TODO if monday data of the week is missing in new_fpmmTrades then take it from the general file
     try:
         fpmmTrades["creationTimestamp"] = fpmmTrades["creationTimestamp"].apply(
             lambda x: transform_to_datetime(x)
         )
     except Exception as e:
         print(f"Transformation not needed")
+    # check missing data from Monday
+    fpmmTrades = check_current_week_data(fpmmTrades)
 
     print("Computing the estimated mech calls dataset")
     trader_mech_calls = compute_daily_mech_calls(fpmmTrades=fpmmTrades, tools=tools)

scripts/get_mech_info.py

@@ -7,11 +7,11 @@ from utils import (
     DATA_DIR,
     TMP_DIR,
     NETWORK_SUBGRAPH_URL,
+    transform_to_datetime,
 )
 import requests
 import pandas as pd
 import numpy as np
-from gnosis_timestamps import compute_request_time
 from mech_request_utils import (
     collect_all_mech_delivers,
     collect_all_mech_requests,
@@ -21,8 +21,6 @@ from mech_request_utils import (
     get_ipfs_data,
     merge_json_files,
 )
-from web3_utils import updating_timestamps
-from nr_mech_calls import transform_to_datetime
 
 SUBGRAPH_HEADERS = {
     "Accept": "application/json, multipart/mixed",
@@ -311,7 +309,7 @@ def get_mech_info_last_60_days() -> dict[str, Any]:
 
 
 @measure_execution_time
-def get_mech_events_since_last_run():
+def get_mech_events_since_last_run(logger):
     """Function to download only the new events since the last execution."""
 
     # Read the latest date from stored data
@@ -371,7 +369,7 @@ def get_mech_events_since_last_run():
     )
 
     # Add ipfs contents
-    get_ipfs_data("new_merged_requests.json", "new_tools_info.json")
+    get_ipfs_data("new_merged_requests.json", "new_tools_info.json", logger)
     return latest_timestamp
 
 

scripts/markets.py

@@ -19,13 +19,13 @@
 
 import functools
 import warnings
-from string import Template
+from datetime import datetime, timedelta
 from typing import Optional, Generator, Callable
 import pandas as pd
 import requests
 from tqdm import tqdm
 from typing import List, Dict
-from utils import SUBGRAPH_API_KEY, DATA_DIR, TMP_DIR
+from utils import SUBGRAPH_API_KEY, DATA_DIR, TMP_DIR, transform_to_datetime
 from web3_utils import (
     FPMM_QS_CREATOR,
     FPMM_PEARL_CREATOR,
@@ -299,14 +299,19 @@ def etl(filename: Optional[str] = None) -> pd.DataFrame:
     return fpmms
 
 
-def add_market_creator(tools: pd.DataFrame) -> None:
-    # Check if fpmmTrades.parquet is in the same directory
+def read_global_trades_file() -> pd.DataFrame:
     try:
         trades_filename = "fpmmTrades.parquet"
         fpmms_trades = pd.read_parquet(TMP_DIR / trades_filename)
     except FileNotFoundError:
         print("Error: fpmmTrades.parquet not found. No market creator added")
         return
+    return fpmms_trades
+
+
+def add_market_creator(tools: pd.DataFrame) -> None:
+    # Check if fpmmTrades.parquet is in the same directory
+    fpmms_trades = read_global_trades_file()
     tools["market_creator"] = ""
     # traverse the list of traders
     tools_no_market_creator = 0
@@ -346,5 +351,35 @@ def fpmmTrades_etl(rpc: str, trades_filename: str, from_timestamp: str) -> None:
     return
 
 
+def check_current_week_data(trades_df: pd.DataFrame) -> pd.DataFrame:
+    """Function to check if all current weeks data is present, if not, then add the missing data from previous file"""
+    # Get current date
+    now = datetime.now()
+
+    # Get start of the current week (Monday)
+    start_of_week = now - timedelta(days=now.weekday())
+    start_of_week = start_of_week.replace(hour=0, minute=0, second=0, microsecond=0)
+    print(f"start of the week = {start_of_week}")
+
+    trades_df["creation_timestamp"] = pd.to_datetime(trades_df["creationTimestamp"])
+    trades_df["creation_date"] = trades_df["creation_timestamp"].dt.date
+    trades_df["creation_date"] = pd.to_datetime(trades_df["creation_date"])
+    # Check dataframe
+    min_date = min(trades_df.creation_date)
+    if min_date > start_of_week:
+        # missing data of current week in the trades file
+        fpmms_trades = read_global_trades_file()
+        # get missing data
+        missing_data = fpmms_trades[
+            (fpmms_trades["creation_date"] >= start_of_week)
+            & (fpmms_trades["creation_date"] < min_date)
+        ]
+        merge_df = pd.concat([trades_df, missing_data], ignore_index=True)
+        merge_df.drop_duplicates("id", keep="last", inplace=True)
+        return merge_df
+    # no update needed
+    return trades_df
+
+
 if __name__ == "__main__":
     etl("all_fpmms.parquet")

scripts/mech_request_utils.py

@@ -40,14 +40,20 @@ from web3_utils import (
     SUBGRAPH_POLL_INTERVAL,
 )
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from utils import DATA_DIR, JSON_DATA_DIR, MECH_SUBGRAPH_URL, SUBGRAPH_API_KEY
+from utils import (
+    DATA_DIR,
+    JSON_DATA_DIR,
+    MECH_SUBGRAPH_URL,
+    SUBGRAPH_API_KEY,
+    IPFS_ADDRESS,
+)
 
 NUM_WORKERS = 10
 BLOCKS_CHUNK_SIZE = 10000
 TEXT_ALIGNMENT = 30
 MINIMUM_WRITE_FILE_DELAY_SECONDS = 20
 MECH_FROM_BLOCK_RANGE = 50000
-IPFS_ADDRESS = "https://gateway.autonolas.tech/ipfs/"
+
 last_write_time = 0.0
 
 REQUESTS_QUERY_FILTER = """
@@ -501,14 +507,14 @@ def merge_requests_delivers(
     return
 
 
-def get_ipfs_data(input_filename: str, output_filename: str):
+def get_ipfs_data(input_filename: str, output_filename: str, logger):
     with open(JSON_DATA_DIR / input_filename, "r") as file:
         mech_requests = json.load(file)
 
     total_keys_to_traverse = list(mech_requests.keys())
     updated_mech_requests = dict()
     session = create_session()
-    print("UPDATING IPFS CONTENTS OF REQUESTS")
+    logger.info("UPDATING IPFS CONTENTS OF REQUESTS")
     # requests
     nr_errors = 0
     with ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
@@ -533,11 +539,11 @@ def get_ipfs_data(input_filename: str, output_filename: str):
             updated_mech_requests.update(partial_dict)
 
     save_json_file(updated_mech_requests, output_filename)
-    print(f"NUMBER OF MECH REQUEST IPFS ERRORS={nr_errors}")
+    logger.info(f"NUMBER OF MECH REQUEST IPFS ERRORS={nr_errors}")
 
     # delivers
     nr_deliver_errors = 0
-    print("UPDATING IPFS CONTENTS OF DELIVERS")
+    logger.info("UPDATING IPFS CONTENTS OF DELIVERS")
     total_keys_to_traverse = list(updated_mech_requests.keys())
     final_tools_content = {}
     with ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
@@ -562,7 +568,7 @@ def get_ipfs_data(input_filename: str, output_filename: str):
             final_tools_content.update(partial_dict)
 
     save_json_file(final_tools_content, output_filename)
-    print(f"NUMBER OF MECH DELIVERS IPFS ERRORS={nr_deliver_errors}")
+    logger.info(f"NUMBER OF MECH DELIVERS IPFS ERRORS={nr_deliver_errors}")
 
 
 def only_delivers_loop():

scripts/nr_mech_calls.py

@@ -1,17 +1,13 @@
 import pandas as pd
-from utils import DATA_DIR, DEFAULT_MECH_FEE, TMP_DIR
+from utils import DATA_DIR, DEFAULT_MECH_FEE, TMP_DIR, transform_to_datetime
 from tqdm import tqdm
-from datetime import datetime, timezone
+
 from typing import Dict, Any
 from collections import defaultdict
 from tools import IRRELEVANT_TOOLS
 import re
 
 
-def transform_to_datetime(x):
-    return datetime.fromtimestamp(int(x), tz=timezone.utc)
-
-
 def update_roi(row: pd.DataFrame) -> float:
     new_value = row.net_earnings / (
         row.collateral_amount

scripts/pull_data.py

@@ -26,7 +26,12 @@ from cloud_storage import load_historical_file
 from tools_metrics import compute_tools_based_datasets
 
 
-logging.basicConfig(level=logging.INFO)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger(__name__)
 
 
 def add_current_answer(tools_filename: str):
@@ -88,7 +93,7 @@ def only_new_weekly_analysis():
     # Mech events ETL
     logging.info("Generating the mech json files")
     # get only new data
-    latest_timestamp = get_mech_events_since_last_run()
+    latest_timestamp = get_mech_events_since_last_run(logger)
     if latest_timestamp == None:
         print("Error while getting the mech events")
         return
@@ -127,7 +132,7 @@ def only_new_weekly_analysis():
 
     save_historical_data()
     try:
-        clean_old_data_from_parquet_files("2024-11-06")
+        clean_old_data_from_parquet_files("2024-11-12")
     except Exception as e:
         print("Error cleaning the oldest information from parquet files")
         print(f"reason = {e}")

scripts/tools.py

@@ -27,6 +27,7 @@ from typing import (
 )
 import pandas as pd
 import requests
+from datetime import datetime
 from gnosis_timestamps import transform_timestamp_to_datetime
 from requests.adapters import HTTPAdapter
 from tqdm import tqdm
@@ -96,11 +97,17 @@ NUM_WORKERS = 10
 GET_CONTENTS_BATCH_SIZE = 1000
 
 
+class TimestampedRetry(Retry):
+    def increment(self, *args, **kwargs):
+        print(f"Retry attempt at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        return super().increment(*args, **kwargs)
+
+
 def create_session() -> requests.Session:
     """Create a session with a retry strategy."""
     session = requests.Session()
-    retry_strategy = Retry(
-        total=N_IPFS_RETRIES + 1,
+    retry_strategy = TimestampedRetry(
+        total=N_IPFS_RETRIES,
         backoff_factor=BACKOFF_FACTOR,
         status_forcelist=STATUS_FORCELIST,
     )

scripts/utils.py

@@ -7,6 +7,7 @@ import pandas as pd
 import gc
 import re
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from pathlib import Path
 from enum import Enum
 from string import Template
@@ -29,11 +30,12 @@ REQUEST_SENDER = "sender"
 PROMPT_FIELD = "prompt"
 HTTP = "http://"
 HTTPS = HTTP[:4] + "s" + HTTP[4:]
-IPFS_ADDRESS = f"{HTTPS}gateway.autonolas.tech/ipfs/"
 FORMAT_UPDATE_BLOCK_NUMBER = 30411638
 INVALID_ANSWER_HEX = (
     "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
 )
+OLD_IPFS_ADDRESS = "https://gateway.autonolas.tech/ipfs/"
+IPFS_ADDRESS = "https://gateway.gcp.autonolas.tech/ipfs/"
 
 INC_TOOLS = [
     "prediction-online",
@@ -241,6 +243,10 @@ EVENT_TO_MECH_STRUCT = {
 }
 
 
+def transform_to_datetime(x):
+    return datetime.fromtimestamp(int(x), tz=timezone.utc)
+
+
 def measure_execution_time(func):
     def wrapper(*args, **kwargs):
         start_time = time.time()

scripts/web3_utils.py

@@ -33,7 +33,7 @@ LATEST_BLOCK: Optional[int] = None
 LATEST_BLOCK_NAME: BlockParams = "latest"
 BLOCK_DATA_NUMBER = "number"
 BLOCKS_CHUNK_SIZE = 10_000
-N_IPFS_RETRIES = 2
+N_IPFS_RETRIES = 4
 N_RPC_RETRIES = 100
 RPC_POLL_INTERVAL = 0.05
 SUBGRAPH_POLL_INTERVAL = 0.05