updating scripts

scripts/cleaning_old_info.py

@@ -1,5 +1,6 @@
 import pandas as pd
-from profitability import DATA_DIR, summary_analyse
+from profitability import summary_analyse
+from utils import DATA_DIR
 from staking import label_trades_by_staking
 
 
@@ -39,6 +40,9 @@ def clean_old_data_from_parquet_files(cutoff_date: str):
         print(f"length before filtering {len(all_trades)}")
         all_trades = all_trades.loc[all_trades["creation_timestamp"] > min_date_utc]
         print(f"length after filtering {len(all_trades)}")
+        all_trades.to_parquet(
+            DATA_DIR / "all_trades_profitability.parquet", index=False
+        )
 
     except Exception as e:
         print(f"Error cleaning all trades profitability file {e}")

scripts/cloud_storage.py

@@ -0,0 +1,83 @@
+from minio import Minio
+from minio.error import S3Error
+import os
+import argparse
+
+from utils import HIST_DIR
+
+MINIO_ENDPOINT = "minio.autonolas.tech"
+ACCESS_KEY = os.environ.get("CLOUD_ACCESS_KEY", None)
+SECRET_KEY = os.environ.get("CLOUD_SECRET_KEY", None)
+BUCKET_NAME = "weekly-stats"
+FOLDER_NAME = "historical_data"
+
+
+def initialize_client():
+    # Initialize the MinIO client
+    client = Minio(
+        MINIO_ENDPOINT,
+        access_key=ACCESS_KEY,
+        secret_key=SECRET_KEY,
+        secure=True,  # Set to False if not using HTTPS
+    )
+    return client
+
+
+def upload_file(client, filename: str, file_path: str):
+    """Upload a file to the bucket"""
+    try:
+        OBJECT_NAME = FOLDER_NAME + "/" + filename
+        print(
+            f"filename={filename}, object_name={OBJECT_NAME} and file_path={file_path}"
+        )
+        client.fput_object(
+            BUCKET_NAME, OBJECT_NAME, file_path, part_size=10 * 1024 * 1024
+        )  # 10MB parts
+        print(f"File '{file_path}' uploaded as '{OBJECT_NAME}'.")
+    except S3Error as err:
+        print(f"Error uploading file: {err}")
+
+
+def download_file(client, filename: str, file_path: str):
+    """Download the file back"""
+    try:
+        OBJECT_NAME = FOLDER_NAME + "/" + filename
+        client.fget_object(BUCKET_NAME, OBJECT_NAME, "downloaded_" + file_path)
+        print(f"File '{OBJECT_NAME}' downloaded as 'downloaded_{file_path}'.")
+    except S3Error as err:
+        print(f"Error downloading file: {err}")
+
+
+def load_historical_file(client, filename: str):
+    """Function to load one file into the cloud storage"""
+    file_path = filename
+    upload_file(client, filename, file_path)
+
+
+def process_historical_files(client):
+    """Process all parquet files in historical_data folder"""
+
+    # Walk through all files in the folder
+    for filename in os.listdir(HIST_DIR):
+        # Check if file is a parquet file
+        if filename.endswith(".parquet"):
+            try:
+                load_historical_file(client, filename)
+                print(f"Successfully processed {filename}")
+            except Exception as e:
+                print(f"Error processing {filename}: {str(e)}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Load files to the cloud storate for historical data"
+    )
+    parser.add_argument("param_1", type=str, help="Name of the file to upload")
+
+    # Parse the arguments
+    args = parser.parse_args()
+    filename = args.param_1
+
+    client = initialize_client()
+    # load_historical_file(client, filename)
+    process_historical_files(client)

scripts/daily_data.py

@@ -1,7 +1,5 @@
 import logging
-from utils import measure_execution_time
-from pull_data import DATA_DIR
-from tools import DEFAULT_FILENAME as TOOLS_FILENAME, generate_tools_file
+from utils import measure_execution_time, DATA_DIR, TMP_DIR
 from profitability import (
     analyse_all_traders,
     label_trades_by_staking,
@@ -15,8 +13,8 @@ logging.basicConfig(level=logging.INFO)
 def prepare_live_metrics(
     tools_filename="new_tools.parquet", trades_filename="new_fpmmTrades.parquet"
 ):
-    fpmmTrades = pd.read_parquet(DATA_DIR / trades_filename)
-    tools = pd.read_parquet(DATA_DIR / tools_filename)
+    fpmmTrades = pd.read_parquet(TMP_DIR / trades_filename)
+    tools = pd.read_parquet(TMP_DIR / tools_filename)
     print("Analysing trades...")
     all_trades_df = analyse_all_traders(fpmmTrades, tools, daily_info=True)
 

scripts/markets.py

@@ -21,14 +21,12 @@ import functools
 import warnings
 from string import Template
 from typing import Optional, Generator, Callable
-
 import pandas as pd
 import requests
 from tqdm import tqdm
-
 from typing import List, Dict
-from pathlib import Path
-from utils import SUBGRAPH_API_KEY
+from utils import SUBGRAPH_API_KEY, DATA_DIR, TMP_DIR
+from web3_utils import FPMM_QS_CREATOR, FPMM_PEARL_CREATOR, query_omen_xdai_subgraph
 from queries import (
     FPMMS_QUERY,
     ID_FIELD,
@@ -44,22 +42,16 @@ from queries import (
 
 ResponseItemType = List[Dict[str, str]]
 SubgraphResponseType = Dict[str, ResponseItemType]
-
-
-CREATOR = "0x89c5cc945dd550BcFfb72Fe42BfF002429F46Fec"
-PEARL_CREATOR = "0xFfc8029154ECD55ABED15BD428bA596E7D23f557"
 BATCH_SIZE = 1000
-
+DEFAULT_TO_TIMESTAMP = 2147483647  # around year 2038
+DEFAULT_FROM_TIMESTAMP = 0
 OMEN_SUBGRAPH_URL = Template(
     """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/9fUVQpFwzpdWS9bq5WkAnmKbNNcoBwatMR4yZq81pbbz"""
 )
 
 MAX_UINT_HEX = "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
 DEFAULT_FILENAME = "fpmms.parquet"
-SCRIPTS_DIR = Path(__file__).parent
-ROOT_DIR = SCRIPTS_DIR.parent
-DATA_DIR = ROOT_DIR / "data"
-market_creators_map = {"quickstart": CREATOR, "pearl": PEARL_CREATOR}
+market_creators_map = {"quickstart": FPMM_QS_CREATOR, "pearl": FPMM_PEARL_CREATOR}
 
 
 class RetriesExceeded(Exception):
@@ -144,15 +136,72 @@ def query_subgraph(url: str, query: str, key: str) -> SubgraphResponseType:
     return data
 
 
+def transform_fpmmTrades(df: pd.DataFrame) -> pd.DataFrame:
+    print("Transforming trades dataframe")
+    # convert creator to address
+    df["creator"] = df["creator"].apply(lambda x: x["id"])
+
+    # normalize fpmm column
+    fpmm = pd.json_normalize(df["fpmm"])
+    fpmm.columns = [f"fpmm.{col}" for col in fpmm.columns]
+    df = pd.concat([df, fpmm], axis=1)
+
+    # drop fpmm column
+    df.drop(["fpmm"], axis=1, inplace=True)
+
+    # change creator to creator_address
+    df.rename(columns={"creator": "trader_address"}, inplace=True)
+    print(df.head())
+    print(df.info())
+    return df
+
+
+def create_fpmmTrades(rpc: str, from_timestamp: float = DEFAULT_FROM_TIMESTAMP):
+    """Create fpmmTrades for all trades."""
+    # Quickstart trades
+    qs_trades_json = query_omen_xdai_subgraph(
+        trader_category="quickstart",
+        from_timestamp=from_timestamp,
+        to_timestamp=DEFAULT_TO_TIMESTAMP,
+        fpmm_from_timestamp=from_timestamp,
+        fpmm_to_timestamp=DEFAULT_TO_TIMESTAMP,
+    )
+
+    print(f"length of the qs_trades_json dataset {len(qs_trades_json)}")
+
+    # convert to dataframe
+    qs_df = pd.DataFrame(qs_trades_json["data"]["fpmmTrades"])
+    qs_df["market_creator"] = "quickstart"
+    qs_df = transform_fpmmTrades(qs_df)
+
+    # Pearl trades
+    pearl_trades_json = query_omen_xdai_subgraph(
+        trader_category="pearl",
+        from_timestamp=from_timestamp,
+        to_timestamp=DEFAULT_TO_TIMESTAMP,
+        fpmm_from_timestamp=from_timestamp,
+        fpmm_to_timestamp=DEFAULT_TO_TIMESTAMP,
+    )
+
+    print(f"length of the pearl_trades_json dataset {len(pearl_trades_json)}")
+
+    # convert to dataframe
+    pearl_df = pd.DataFrame(pearl_trades_json["data"]["fpmmTrades"])
+    pearl_df["market_creator"] = "pearl"
+    pearl_df = transform_fpmmTrades(pearl_df)
+
+    return pd.concat([qs_df, pearl_df], ignore_index=True)
+
+
 def fpmms_fetcher(trader_category: str) -> Generator[ResponseItemType, int, None]:
     """An indefinite fetcher for the FPMMs."""
     omen_subgraph = OMEN_SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
     print(f"omen_subgraph = {omen_subgraph}")
 
     if trader_category == "pearl":
-        creator_id = PEARL_CREATOR
+        creator_id = FPMM_PEARL_CREATOR
     else:  # quickstart
-        creator_id = CREATOR
+        creator_id = FPMM_QS_CREATOR
     while True:
         fpmm_id = yield
         fpmms_query = FPMMS_QUERY.substitute(
@@ -254,7 +303,7 @@ def add_market_creator(tools: pd.DataFrame) -> None:
     # Check if fpmmTrades.parquet is in the same directory
     try:
         trades_filename = "fpmmTrades.parquet"
-        fpmms_trades = pd.read_parquet(DATA_DIR / trades_filename)
+        fpmms_trades = pd.read_parquet(TMP_DIR / trades_filename)
     except FileNotFoundError:
         print("Error: fpmmTrades.parquet not found. No market creator added")
         return
@@ -278,5 +327,21 @@ def add_market_creator(tools: pd.DataFrame) -> None:
     return tools
 
 
+def fpmmTrades_etl(rpc: str, trades_filename: str, from_timestamp: str) -> None:
+    print("Generating the trades file")
+    try:
+        fpmmTrades = create_fpmmTrades(rpc, from_timestamp=from_timestamp)
+    except FileNotFoundError:
+        print(f"Error creating {trades_filename} file .")
+
+    # make sure trader_address is in the columns
+    assert "trader_address" in fpmmTrades.columns, "trader_address column not found"
+
+    # 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
+
+
 if __name__ == "__main__":
     etl("all_fpmms.parquet")

scripts/mech_request_utils.py

@@ -22,31 +22,27 @@ import json
 import time
 import pickle
 from random import uniform
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, Tuple
 from pathlib import Path
 import requests
 from gql import Client, gql
 from gql.transport.requests import RequestsHTTPTransport
 from tools import (
-    IPFS_POLL_INTERVAL,
     GET_CONTENTS_BATCH_SIZE,
     IRRELEVANT_TOOLS,
     create_session,
     request,
 )
 from tqdm import tqdm
-from markets import PEARL_CREATOR, CREATOR
+from web3_utils import FPMM_QS_CREATOR, FPMM_PEARL_CREATOR, IPFS_POLL_INTERVAL
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from utils import DATA_DIR, JSON_DATA_DIR
 
 NUM_WORKERS = 10
 BLOCKS_CHUNK_SIZE = 10000
 TEXT_ALIGNMENT = 30
 MINIMUM_WRITE_FILE_DELAY_SECONDS = 20
 MECH_FROM_BLOCK_RANGE = 50000
-SCRIPTS_DIR = Path(__file__).parent
-ROOT_DIR = SCRIPTS_DIR.parent
-JSON_DATA_DIR = ROOT_DIR / "json_data"
-DATA_DIR = ROOT_DIR / "data"
 IPFS_ADDRESS = "https://gateway.autonolas.tech/ipfs/"
 THEGRAPH_ENDPOINT = "https://api.studio.thegraph.com/query/57238/mech/0.0.2"
 last_write_time = 0.0
@@ -119,7 +115,7 @@ def collect_all_mech_requests(from_block: int, to_block: int, filename: str) ->
     id_gt = "0x00"
     while True:
         variables = {
-            "sender_not_in": [CREATOR, PEARL_CREATOR],
+            "sender_not_in": [FPMM_QS_CREATOR, FPMM_PEARL_CREATOR],
             "id_gt": id_gt,
             "blockNumber_gte": str(from_block),  # str
             "blockNumber_lte": str(to_block),  # str
@@ -417,6 +413,7 @@ def update_block_request_map(block_request_id_map: dict) -> None:
 
 
 def fix_duplicate_requestIds(requests_filename: str, delivers_filename: str) -> dict:
+    print("Fix duplicated request Ids")
     with open(JSON_DATA_DIR / delivers_filename, "r") as file:
         data_delivers = json.load(file)
 
@@ -455,7 +452,7 @@ def fix_duplicate_requestIds(requests_filename: str, delivers_filename: str) ->
 def merge_requests_delivers(
     requests_filename: str, delivers_filename: str, filename: str
 ) -> None:
-    # TODO
+    print("Merge request delivers")
     """Function to map requests and delivers"""
     with open(JSON_DATA_DIR / delivers_filename, "r") as file:
         mech_delivers = json.load(file)

scripts/profitability.py

@@ -18,17 +18,14 @@
 #   ------------------------------------------------------------------------------
 
 import time
-import requests
 import datetime
 import pandas as pd
-from collections import defaultdict
 from typing import Any
-from string import Template
 from enum import Enum
 from tqdm import tqdm
 import numpy as np
 import os
-
+from web3_utils import query_conditional_tokens_gc_subgraph
 from get_mech_info import (
     DATETIME_60_DAYS_AGO,
     update_fpmmTrades_parquet,
@@ -36,26 +33,20 @@ from get_mech_info import (
     update_all_trades_parquet,
 )
 from utils import (
-    SUBGRAPH_API_KEY,
     wei_to_unit,
     convert_hex_to_int,
-    _to_content,
     JSON_DATA_DIR,
     DATA_DIR,
 )
-from queries import omen_xdai_trades_query, conditional_tokens_gc_user_query
 from staking import label_trades_by_staking
 
-QUERY_BATCH_SIZE = 1000
 DUST_THRESHOLD = 10000000000000
 INVALID_ANSWER = -1
-FPMM_QS_CREATOR = "0x89c5cc945dd550bcffb72fe42bff002429f46fec"
-FPMM_PEARL_CREATOR = "0xFfc8029154ECD55ABED15BD428bA596E7D23f557"
 DEFAULT_FROM_DATE = "1970-01-01T00:00:00"
 DEFAULT_TO_DATE = "2038-01-19T03:14:07"
-DEFAULT_FROM_TIMESTAMP = 0
+
 DEFAULT_60_DAYS_AGO_TIMESTAMP = (DATETIME_60_DAYS_AGO).timestamp()
-DEFAULT_TO_TIMESTAMP = 2147483647  # around year 2038
+
 WXDAI_CONTRACT_ADDRESS = "0xe91D153E0b41518A2Ce8Dd3D7944Fa863463a97d"
 DEFAULT_MECH_FEE = 0.01
 DUST_THRESHOLD = 10000000000000
@@ -148,107 +139,6 @@ SUMMARY_STATS_DF_COLS = [
     "mean_mech_calls_per_trade",
     "mean_mech_fee_amount_per_trade",
 ]
-headers = {
-    "Accept": "application/json, multipart/mixed",
-    "Content-Type": "application/json",
-}
-
-
-def _query_omen_xdai_subgraph(
-    trader_category: str,
-    from_timestamp: float,
-    to_timestamp: float,
-    fpmm_from_timestamp: float,
-    fpmm_to_timestamp: float,
-) -> dict[str, Any]:
-    """Query the subgraph."""
-    OMEN_SUBGRAPH_URL = Template(
-        """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/9fUVQpFwzpdWS9bq5WkAnmKbNNcoBwatMR4yZq81pbbz"""
-    )
-    omen_subgraph = OMEN_SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
-    print(f"omen_subgraph = {omen_subgraph}")
-    grouped_results = defaultdict(list)
-    id_gt = ""
-    if trader_category == "quickstart":
-        creator_id = FPMM_QS_CREATOR.lower()
-    else:  # pearl
-        creator_id = FPMM_PEARL_CREATOR.lower()
-
-    while True:
-        query = omen_xdai_trades_query.substitute(
-            fpmm_creator=creator_id,
-            creationTimestamp_gte=int(from_timestamp),
-            creationTimestamp_lte=int(to_timestamp),
-            fpmm_creationTimestamp_gte=int(fpmm_from_timestamp),
-            fpmm_creationTimestamp_lte=int(fpmm_to_timestamp),
-            first=QUERY_BATCH_SIZE,
-            id_gt=id_gt,
-        )
-        content_json = _to_content(query)
-
-        res = requests.post(omen_subgraph, headers=headers, json=content_json)
-        result_json = res.json()
-        # print(f"result = {result_json}")
-        user_trades = result_json.get("data", {}).get("fpmmTrades", [])
-
-        if not user_trades:
-            break
-
-        for trade in user_trades:
-            fpmm_id = trade.get("fpmm", {}).get("id")
-            grouped_results[fpmm_id].append(trade)
-
-        id_gt = user_trades[len(user_trades) - 1]["id"]
-
-    all_results = {
-        "data": {
-            "fpmmTrades": [
-                trade
-                for trades_list in grouped_results.values()
-                for trade in trades_list
-            ]
-        }
-    }
-
-    return all_results
-
-
-def _query_conditional_tokens_gc_subgraph(creator: str) -> dict[str, Any]:
-    """Query the subgraph."""
-    SUBGRAPH_URL = Template(
-        """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/7s9rGBffUTL8kDZuxvvpuc46v44iuDarbrADBFw5uVp2"""
-    )
-    subgraph = SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
-    all_results: dict[str, Any] = {"data": {"user": {"userPositions": []}}}
-    userPositions_id_gt = ""
-    while True:
-        query = conditional_tokens_gc_user_query.substitute(
-            id=creator.lower(),
-            first=QUERY_BATCH_SIZE,
-            userPositions_id_gt=userPositions_id_gt,
-        )
-        content_json = {"query": query}
-        print("sending query to subgraph")
-        res = requests.post(subgraph, headers=headers, json=content_json)
-        result_json = res.json()
-        # print(f"result = {result_json}")
-        user_data = result_json.get("data", {}).get("user", {})
-
-        if not user_data:
-            break
-
-        user_positions = user_data.get("userPositions", [])
-
-        if user_positions:
-            all_results["data"]["user"]["userPositions"].extend(user_positions)
-            userPositions_id_gt = user_positions[len(user_positions) - 1]["id"]
-        else:
-            break
-
-    if len(all_results["data"]["user"]["userPositions"]) == 0:
-        return {"data": {"user": None}}
-
-    return all_results
 
 
 def _is_redeemed(user_json: dict[str, Any], fpmmTrade: dict[str, Any]) -> bool:
@@ -267,68 +157,10 @@ def _is_redeemed(user_json: dict[str, Any], fpmmTrade: dict[str, Any]) -> bool:
     return False
 
 
-def transform_fpmmTrades(df: pd.DataFrame) -> pd.DataFrame:
-    print("Transforming trades dataframe")
-    # convert creator to address
-    df["creator"] = df["creator"].apply(lambda x: x["id"])
-
-    # normalize fpmm column
-    fpmm = pd.json_normalize(df["fpmm"])
-    fpmm.columns = [f"fpmm.{col}" for col in fpmm.columns]
-    df = pd.concat([df, fpmm], axis=1)
-
-    # drop fpmm column
-    df.drop(["fpmm"], axis=1, inplace=True)
-
-    # change creator to creator_address
-    df.rename(columns={"creator": "trader_address"}, inplace=True)
-    print(df.head())
-    print(df.info())
-    return df
-
-
-def create_fpmmTrades(rpc: str, from_timestamp: float = DEFAULT_FROM_TIMESTAMP):
-    """Create fpmmTrades for all trades."""
-    # Quickstart trades
-    qs_trades_json = _query_omen_xdai_subgraph(
-        trader_category="quickstart",
-        from_timestamp=from_timestamp,
-        to_timestamp=DEFAULT_TO_TIMESTAMP,
-        fpmm_from_timestamp=from_timestamp,
-        fpmm_to_timestamp=DEFAULT_TO_TIMESTAMP,
-    )
-
-    print(f"length of the qs_trades_json dataset {len(qs_trades_json)}")
-
-    # convert to dataframe
-    qs_df = pd.DataFrame(qs_trades_json["data"]["fpmmTrades"])
-    qs_df["market_creator"] = "quickstart"
-    qs_df = transform_fpmmTrades(qs_df)
-
-    # Pearl trades
-    pearl_trades_json = _query_omen_xdai_subgraph(
-        trader_category="pearl",
-        from_timestamp=from_timestamp,
-        to_timestamp=DEFAULT_TO_TIMESTAMP,
-        fpmm_from_timestamp=from_timestamp,
-        fpmm_to_timestamp=DEFAULT_TO_TIMESTAMP,
-    )
-
-    print(f"length of the pearl_trades_json dataset {len(pearl_trades_json)}")
-
-    # convert to dataframe
-    pearl_df = pd.DataFrame(pearl_trades_json["data"]["fpmmTrades"])
-    pearl_df["market_creator"] = "pearl"
-    pearl_df = transform_fpmmTrades(pearl_df)
-
-    return pd.concat([qs_df, pearl_df], ignore_index=True)
-
-
 def prepare_profitalibity_data(
     rpc: str,
     tools_filename: str,
     trades_filename: str,
-    from_timestamp: float,
 ) -> pd.DataFrame:
     """Prepare data for profitalibity analysis."""
 
@@ -354,10 +186,9 @@ def prepare_profitalibity_data(
     # Check if fpmmTrades.parquet is in the same directory
     print("Generating the trades file")
     try:
-        fpmmTrades = create_fpmmTrades(rpc, from_timestamp=from_timestamp)
-        fpmmTrades.to_parquet(DATA_DIR / trades_filename, index=False)
+        fpmmTrades = pd.read_parquet(DATA_DIR / trades_filename)
     except FileNotFoundError:
-        print(f"Error creating {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"
@@ -401,7 +232,7 @@ def analyse_trader(
 
     # Fetch user's conditional tokens gc graph
     try:
-        user_json = _query_conditional_tokens_gc_subgraph(trader_address)
+        user_json = query_conditional_tokens_gc_subgraph(trader_address)
     except Exception as e:
         print(f"Error fetching user data: {e}")
         return trades_df
@@ -575,16 +406,13 @@ def run_profitability_analysis(
     rpc: str,
     tools_filename: str,
     trades_filename: str,
-    from_timestamp: float,
     merge: 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(
-        rpc, tools_filename, trades_filename, from_timestamp
-    )
+    fpmmTrades = prepare_profitalibity_data(rpc, tools_filename, trades_filename)
     if merge:
         update_tools_parquet(rpc, tools_filename)
     tools = pd.read_parquet(DATA_DIR / "tools.parquet")

scripts/pull_data.py

@@ -1,10 +1,7 @@
 import logging
 from datetime import datetime
 import pandas as pd
-from markets import (
-    etl as mkt_etl,
-    DEFAULT_FILENAME as MARKETS_FILENAME,
-)
+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 profitability import run_profitability_analysis, DEFAULT_60_DAYS_AGO_TIMESTAMP
 from utils import (
@@ -23,6 +20,8 @@ from get_mech_info import (
 from update_tools_accuracy import compute_tools_accuracy
 from cleaning_old_info import clean_old_data_from_parquet_files
 from web3_utils import updating_timestamps
+from manage_space_files import move_files
+from cloud_storage import load_historical_file
 
 
 logging.basicConfig(level=logging.INFO)
@@ -57,7 +56,8 @@ def save_historical_data():
         tools = pd.read_parquet(DATA_DIR / "tools.parquet")
         filename = f"tools_{timestamp}.parquet"
         tools.to_parquet(HIST_DIR / filename, index=False)
-
+        # save into cloud storage
+        # load_historical_file(filename)
     except Exception as e:
         print(f"Error saving tools file in the historical folder {e}")
 
@@ -65,6 +65,8 @@ def save_historical_data():
         all_trades = pd.read_parquet(DATA_DIR / "all_trades_profitability.parquet")
         filename = f"all_trades_profitability_{timestamp}.parquet"
         all_trades.to_parquet(HIST_DIR / filename, index=False)
+        # save into cloud storage
+        # load_historical_file(filename)
 
     except Exception as e:
         print(
@@ -81,7 +83,7 @@ def only_new_weekly_analysis():
     mkt_etl(MARKETS_FILENAME)
     logging.info("Markets ETL completed")
 
-    # New tools ETL
+    # Mech events ETL
     logging.info("Generating the mech json files")
     # get only new data
     latest_timestamp = get_mech_events_since_last_run()
@@ -90,6 +92,12 @@ def only_new_weekly_analysis():
         return
     logging.info(f"Finished generating the mech json files from {latest_timestamp}")
 
+    # FpmmTrades ETL
+    fpmmTrades_etl(
+        rpc=rpc,
+        trades_filename="new_fpmmTrades.parquet",
+        from_timestamp=int(latest_timestamp.timestamp()),
+    )
     # Run tools ETL
     logging.info("Generate and parse the tools content")
     # generate only new file
@@ -104,7 +112,6 @@ def only_new_weekly_analysis():
         rpc=rpc,
         tools_filename="new_tools.parquet",
         trades_filename="new_fpmmTrades.parquet",
-        from_timestamp=int(latest_timestamp.timestamp()),
         merge=True,
     )
 
@@ -121,58 +128,14 @@ def only_new_weekly_analysis():
 
     save_historical_data()
 
-    clean_old_data_from_parquet_files("2024-10-06")
-
-    compute_tools_accuracy()
-
-    logging.info("Weekly analysis files generated and saved")
-
-
-def old_weekly_analysis():
-    """Run weekly analysis for the FPMMS project."""
-    rpc = RPC
-    # Run markets ETL
-    logging.info("Running markets ETL")
-    mkt_etl(MARKETS_FILENAME)
-    logging.info("Markets ETL completed")
-
-    # New tools ETL
-    logging.info("Generating the mech json files")
-
-    get_mech_events_last_60_days()
-    logging.info("Finished generating the mech json files")
-
-    # Run tools ETL
-    logging.info("Generate and parse the tools content")
-
-    generate_tools_file("tools_info.json", TOOLS_FILENAME)
-    logging.info("Tools ETL completed")
-    add_current_answer(TOOLS_FILENAME)
-
-    # Run profitability analysis
-    logging.info("Running profitability analysis")
-    run_profitability_analysis(
-        rpc=rpc,
-        tools_filename=TOOLS_FILENAME,
-        trades_filename="fpmmTrades.parquet",
-        from_timestamp=DEFAULT_60_DAYS_AGO_TIMESTAMP,
-    )
-    logging.info("Profitability analysis completed")
-
-    try:
-        updating_timestamps(rpc, TOOLS_FILENAME)
-    except Exception as e:
-        logging.error("Error while updating timestamps of tools")
-        print(e)
+    clean_old_data_from_parquet_files("2024-10-13")
 
     compute_tools_accuracy()
 
+    # move to tmp folder the new generated files
+    move_files()
     logging.info("Weekly analysis files generated and saved")
 
 
 if __name__ == "__main__":
     only_new_weekly_analysis()
-    # weekly_analysis()
-    # rpc = RPC
-    # updating_timestamps(rpc)
-    # compute_tools_accuracy()

scripts/staking.py

@@ -1,6 +1,5 @@
 import json
 import sys
-from pathlib import Path
 from typing import Any, List
 from utils import RPC, DATA_DIR
 import requests

scripts/tools.py

@@ -44,17 +44,25 @@ from urllib3.exceptions import (
     HTTPError as Urllib3HTTPError,
 )
 from web3 import Web3, HTTPProvider
-from web3.exceptions import MismatchedABI
 from markets import add_market_creator
-from web3.types import BlockParams
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from web3_utils import (
+    read_abi,
+    SLEEP,
+    reduce_window,
+    LATEST_BLOCK,
+    LATEST_BLOCK_NAME,
+    BLOCK_DATA_NUMBER,
+    BLOCKS_CHUNK_SIZE,
+    N_RPC_RETRIES,
+    N_IPFS_RETRIES,
+    RPC_POLL_INTERVAL,
+    IPFS_POLL_INTERVAL,
+)
 from utils import (
     clean,
     BLOCK_FIELD,
     gen_event_filename,
-    read_abi,
-    SLEEP,
-    reduce_window,
     limit_text,
     DATA_DIR,
     JSON_DATA_DIR,
@@ -82,10 +90,7 @@ MECH_TO_INFO = {
     "0x77af31de935740567cf4ff1986d04b2c964a786a": ("new_mech_abi.json", 30776879),
 }
 # optionally set the latest block to stop searching for the delivered events
-LATEST_BLOCK: Optional[int] = None
-LATEST_BLOCK_NAME: BlockParams = "latest"
-BLOCK_DATA_NUMBER = "number"
-BLOCKS_CHUNK_SIZE = 10_000
+
 EVENT_ARGUMENTS = "args"
 DATA = "data"
 IPFS_LINKS_SERIES_NAME = "ipfs_links"
@@ -95,11 +100,7 @@ DEFAULT_FILENAME = "tools.parquet"
 RE_RPC_FILTER_ERROR = r"Filter with id: '\d+' does not exist."
 ABI_ERROR = "The event signature did not match the provided ABI"
 HTTP_TIMEOUT = 10
-N_IPFS_RETRIES = 1
-N_RPC_RETRIES = 100
-RPC_POLL_INTERVAL = 0.05
-# IPFS_POLL_INTERVAL = 0.05  # low speed
-IPFS_POLL_INTERVAL = 0.2  # high speed
+
 IRRELEVANT_TOOLS = [
     "openai-text-davinci-002",
     "openai-text-davinci-003",

scripts/utils.py

@@ -1,8 +1,6 @@
-import sys
 import json
 import os
 import time
-from tqdm import tqdm
 from typing import List, Any, Optional, Union, Tuple
 import numpy as np
 import pandas as pd
@@ -20,6 +18,8 @@ SCRIPTS_DIR = Path(__file__).parent
 ROOT_DIR = SCRIPTS_DIR.parent
 DATA_DIR = ROOT_DIR / "data"
 JSON_DATA_DIR = ROOT_DIR / "json_data"
+HIST_DIR = ROOT_DIR / "historical_data"
+TMP_DIR = ROOT_DIR / "tmp"
 BLOCK_FIELD = "block"
 CID_PREFIX = "f01701220"
 REQUEST_ID = "requestId"
@@ -236,31 +236,6 @@ def measure_execution_time(func):
     return wrapper
 
 
-def parse_args() -> str:
-    """Parse the arguments and return the RPC."""
-    if len(sys.argv) != 2:
-        raise ValueError("Expected the RPC as a positional argument.")
-    return sys.argv[1]
-
-
-def read_abi(abi_path: str) -> str:
-    """Read and return the wxDAI contract's ABI."""
-    with open(abi_path) as abi_file:
-        return abi_file.read()
-
-
-def reduce_window(contract_instance, event, from_block, batch_size, latest_block):
-    """Dynamically reduce the batch size window."""
-    keep_fraction = 1 - REDUCE_FACTOR
-    events_filter = contract_instance.events[event].build_filter()
-    events_filter.fromBlock = from_block
-    batch_size = int(batch_size * keep_fraction)
-    events_filter.toBlock = min(from_block + batch_size, latest_block)
-    tqdm.write(f"RPC timed out! Resizing batch size to {batch_size}.")
-    time.sleep(SLEEP)
-    return events_filter, batch_size
-
-
 def limit_text(text: str, limit: int = 200) -> str:
     """Limit the given text"""
     if len(text) > limit:
@@ -314,21 +289,6 @@ def read_n_last_lines(filename: str, n: int = 1) -> str:
     return last_line
 
 
-def get_earliest_block(event_name: MechEventName) -> int:
-    """Get the earliest block number to use when filtering for events."""
-    filename = gen_event_filename(event_name)
-    if not os.path.exists(DATA_DIR / filename):
-        return 0
-
-    df = pd.read_parquet(DATA_DIR / filename)
-    block_field = f"{event_name.value.lower()}_{BLOCK_FIELD}"
-    earliest_block = int(df[block_field].max())
-    # clean and release all memory
-    del df
-    gc.collect()
-    return earliest_block
-
-
 def get_question(text: str) -> str:
     """Get the question from a text."""
     # Regex to find text within double quotes
@@ -366,18 +326,6 @@ def wei_to_unit(wei: int) -> float:
     return wei / 10**18
 
 
-def measure_execution_time(func):
-    def wrapper(*args, **kwargs):
-        start_time = time.time()
-        result = func(*args, **kwargs)
-        end_time = time.time()
-        execution_time = end_time - start_time
-        print(f"Execution time: {execution_time:.6f} seconds")
-        return result
-
-    return wrapper
-
-
 def get_vote(p_yes, p_no) -> Optional[str]:
     """Return the vote."""
     if p_no == p_yes:
@@ -420,7 +368,7 @@ def get_prediction_values(params: dict) -> Tuple:
     return p_yes, p_no, confidence, info_utility
 
 
-def _to_content(q: str) -> dict[str, Any]:
+def to_content(q: str) -> dict[str, Any]:
     """Convert the given query string to payload content, i.e., add it under a `queries` key and convert it to bytes."""
     finalized_query = {
         "query": q,

scripts/web3_utils.py

@@ -0,0 +1,258 @@
+import sys
+import pickle
+import gc
+import time
+import requests
+from functools import partial
+from string import Template
+from datetime import datetime
+from concurrent.futures import ThreadPoolExecutor
+from collections import defaultdict
+from tqdm import tqdm
+from web3 import Web3
+from typing import Any, Optional
+from web3.types import BlockParams
+from utils import JSON_DATA_DIR, DATA_DIR, SUBGRAPH_API_KEY, to_content
+from queries import conditional_tokens_gc_user_query, omen_xdai_trades_query
+import pandas as pd
+
+REDUCE_FACTOR = 0.25
+SLEEP = 0.5
+QUERY_BATCH_SIZE = 1000
+FPMM_QS_CREATOR = "0x89c5cc945dd550bcffb72fe42bff002429f46fec"
+FPMM_PEARL_CREATOR = "0xFfc8029154ECD55ABED15BD428bA596E7D23f557"
+LATEST_BLOCK: Optional[int] = None
+LATEST_BLOCK_NAME: BlockParams = "latest"
+BLOCK_DATA_NUMBER = "number"
+BLOCKS_CHUNK_SIZE = 10_000
+N_IPFS_RETRIES = 1
+N_RPC_RETRIES = 100
+RPC_POLL_INTERVAL = 0.05
+# IPFS_POLL_INTERVAL = 0.05  # low speed
+IPFS_POLL_INTERVAL = 0.2  # high speed
+
+headers = {
+    "Accept": "application/json, multipart/mixed",
+    "Content-Type": "application/json",
+}
+
+
+def parse_args() -> str:
+    """Parse the arguments and return the RPC."""
+    if len(sys.argv) != 2:
+        raise ValueError("Expected the RPC as a positional argument.")
+    return sys.argv[1]
+
+
+def read_abi(abi_path: str) -> str:
+    """Read and return the wxDAI contract's ABI."""
+    with open(abi_path) as abi_file:
+        return abi_file.read()
+
+
+def update_block_request_map(block_request_id_map: dict) -> None:
+    print("Saving block request id map info")
+    with open(JSON_DATA_DIR / "block_request_id_map.pickle", "wb") as handle:
+        pickle.dump(block_request_id_map, handle, protocol=pickle.HIGHEST_PROTOCOL)
+
+
+def reduce_window(contract_instance, event, from_block, batch_size, latest_block):
+    """Dynamically reduce the batch size window."""
+    keep_fraction = 1 - REDUCE_FACTOR
+    events_filter = contract_instance.events[event].build_filter()
+    events_filter.fromBlock = from_block
+    batch_size = int(batch_size * keep_fraction)
+    events_filter.toBlock = min(from_block + batch_size, latest_block)
+    tqdm.write(f"RPC timed out! Resizing batch size to {batch_size}.")
+    time.sleep(SLEEP)
+    return events_filter, batch_size
+
+
+def block_number_to_timestamp(block_number: int, web3: Web3) -> str:
+    """Convert a block number to a timestamp."""
+    block = web3.eth.get_block(block_number)
+    timestamp = datetime.utcfromtimestamp(block["timestamp"])
+    try:
+        timestamp_str = timestamp.strftime("%Y-%m-%d %H:%M:%S")
+        timestamp = datetime.strptime(timestamp_str, "%Y-%m-%dT%H:%M:%S.%f")
+    except Exception as e:
+        timestamp = datetime.utcfromtimestamp(block["timestamp"])
+    return timestamp.strftime("%Y-%m-%d %H:%M:%S")
+
+
+def parallelize_timestamp_conversion(df: pd.DataFrame, function: callable) -> list:
+    """Parallelize the timestamp conversion."""
+    block_numbers = df["request_block"].tolist()
+    with ThreadPoolExecutor(max_workers=10) as executor:
+        results = list(
+            tqdm(executor.map(function, block_numbers), total=len(block_numbers))
+        )
+    return results
+
+
+def updating_timestamps(rpc: str, tools_filename: str):
+    web3 = Web3(Web3.HTTPProvider(rpc))
+
+    tools = pd.read_parquet(DATA_DIR / tools_filename)
+
+    # Convert block number to timestamp
+    print("Converting block number to timestamp")
+    t_map = pickle.load(open(DATA_DIR / "t_map.pkl", "rb"))
+    tools["request_time"] = tools["request_block"].map(t_map)
+
+    no_data = tools["request_time"].isna().sum()
+    print(f"Total rows with no request time info = {no_data}")
+
+    # Identify tools with missing request_time and fill them
+    missing_time_indices = tools[tools["request_time"].isna()].index
+    if not missing_time_indices.empty:
+        partial_block_number_to_timestamp = partial(
+            block_number_to_timestamp, web3=web3
+        )
+        missing_timestamps = parallelize_timestamp_conversion(
+            tools.loc[missing_time_indices], partial_block_number_to_timestamp
+        )
+
+        # Update the original DataFrame with the missing timestamps
+        for i, timestamp in zip(missing_time_indices, missing_timestamps):
+            tools.at[i, "request_time"] = timestamp
+
+    tools["request_month_year"] = pd.to_datetime(tools["request_time"]).dt.strftime(
+        "%Y-%m"
+    )
+    tools["request_month_year_week"] = (
+        pd.to_datetime(tools["request_time"]).dt.to_period("W").astype(str)
+    )
+
+    # Save the tools data after the updates on the content
+    print(f"Updating file {tools_filename} with timestamps")
+    tools.to_parquet(DATA_DIR / tools_filename, index=False)
+
+    # Update t_map with new timestamps
+    new_timestamps = (
+        tools[["request_block", "request_time"]]
+        .dropna()
+        .set_index("request_block")
+        .to_dict()["request_time"]
+    )
+    t_map.update(new_timestamps)
+
+    with open(DATA_DIR / "t_map.pkl", "wb") as f:
+        pickle.dump(t_map, f)
+
+    # clean and release all memory
+    del tools
+    del t_map
+    gc.collect()
+
+
+def query_conditional_tokens_gc_subgraph(creator: str) -> dict[str, Any]:
+    """Query the subgraph."""
+    SUBGRAPH_URL = Template(
+        """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/7s9rGBffUTL8kDZuxvvpuc46v44iuDarbrADBFw5uVp2"""
+    )
+    subgraph = SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
+    all_results: dict[str, Any] = {"data": {"user": {"userPositions": []}}}
+    userPositions_id_gt = ""
+    while True:
+        query = conditional_tokens_gc_user_query.substitute(
+            id=creator.lower(),
+            first=QUERY_BATCH_SIZE,
+            userPositions_id_gt=userPositions_id_gt,
+        )
+        content_json = {"query": query}
+        print("sending query to subgraph")
+        res = requests.post(subgraph, headers=headers, json=content_json)
+        result_json = res.json()
+        # print(f"result = {result_json}")
+        user_data = result_json.get("data", {}).get("user", {})
+
+        if not user_data:
+            break
+
+        user_positions = user_data.get("userPositions", [])
+
+        if user_positions:
+            all_results["data"]["user"]["userPositions"].extend(user_positions)
+            userPositions_id_gt = user_positions[len(user_positions) - 1]["id"]
+        else:
+            break
+
+    if len(all_results["data"]["user"]["userPositions"]) == 0:
+        return {"data": {"user": None}}
+
+    return all_results
+
+
+def query_omen_xdai_subgraph(
+    trader_category: str,
+    from_timestamp: float,
+    to_timestamp: float,
+    fpmm_from_timestamp: float,
+    fpmm_to_timestamp: float,
+) -> dict[str, Any]:
+    """Query the subgraph."""
+    OMEN_SUBGRAPH_URL = Template(
+        """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/9fUVQpFwzpdWS9bq5WkAnmKbNNcoBwatMR4yZq81pbbz"""
+    )
+    omen_subgraph = OMEN_SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
+    print(f"omen_subgraph = {omen_subgraph}")
+    grouped_results = defaultdict(list)
+    id_gt = ""
+    if trader_category == "quickstart":
+        creator_id = FPMM_QS_CREATOR.lower()
+    else:  # pearl
+        creator_id = FPMM_PEARL_CREATOR.lower()
+
+    while True:
+        query = omen_xdai_trades_query.substitute(
+            fpmm_creator=creator_id,
+            creationTimestamp_gte=int(from_timestamp),
+            creationTimestamp_lte=int(to_timestamp),
+            fpmm_creationTimestamp_gte=int(fpmm_from_timestamp),
+            fpmm_creationTimestamp_lte=int(fpmm_to_timestamp),
+            first=QUERY_BATCH_SIZE,
+            id_gt=id_gt,
+        )
+        content_json = to_content(query)
+
+        res = requests.post(omen_subgraph, headers=headers, json=content_json)
+        result_json = res.json()
+        # print(f"result = {result_json}")
+        user_trades = result_json.get("data", {}).get("fpmmTrades", [])
+
+        if not user_trades:
+            break
+
+        for trade in user_trades:
+            fpmm_id = trade.get("fpmm", {}).get("id")
+            grouped_results[fpmm_id].append(trade)
+
+        id_gt = user_trades[len(user_trades) - 1]["id"]
+
+    all_results = {
+        "data": {
+            "fpmmTrades": [
+                trade
+                for trades_list in grouped_results.values()
+                for trade in trades_list
+            ]
+        }
+    }
+
+    return all_results
+
+
+# def get_earliest_block(event_name: MechEventName) -> int:
+#     """Get the earliest block number to use when filtering for events."""
+#     filename = gen_event_filename(event_name)
+#     if not os.path.exists(DATA_DIR / filename):
+#         return 0
+
+#     df = pd.read_parquet(DATA_DIR / filename)
+#     block_field = f"{event_name.value.lower()}_{BLOCK_FIELD}"
+#     earliest_block = int(df[block_field].max())
+#     # clean and release all memory
+#     del df
+#     gc.collect()
+#     return earliest_block