new scripts and new live data

live_data/markets_live_data.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:234dbf259fdc32ce709e52ccf7a571af34045398c6853a579251bbe3ee682ac7
-size 52537
+oid sha256:6284205c09e9328083c363b6e7217bff4d318f9d1e366f191693e200f8b5810a
+size 55976

scripts/closed_markets_data.py

@@ -0,0 +1,678 @@
+#   -*- coding: utf-8 -*-
+#   ------------------------------------------------------------------------------
+#
+#     Copyright 2024 Valory AG
+#
+#     Licensed under the Apache License, Version 2.0 (the "License");
+#     you may not use this file except in compliance with the License.
+#     You may obtain a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#     Unless required by applicable law or agreed to in writing, software
+#     distributed under the License is distributed on an "AS IS" BASIS,
+#     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#     See the License for the specific language governing permissions and
+#     limitations under the License.
+#
+#   ------------------------------------------------------------------------------
+
+import functools
+import warnings
+from typing import Optional, Generator, Callable
+import os
+import logging
+import pandas as pd
+from collections import defaultdict
+from typing import Any, Optional, Tuple, List, Dict
+from tqdm import tqdm
+from datetime import datetime, timedelta, UTC
+import requests
+from math import prod
+
+from utils import (
+    SUBGRAPH_API_KEY,
+    measure_execution_time,
+    OMEN_SUBGRAPH_URL,
+    CREATOR,
+    BATCH_SIZE,
+    DATA_DIR,
+    _to_content,
+)
+from queries import (
+    FPMMS_CLOSED_MARKETS_QUERY,
+    ID_FIELD,
+    DATA_FIELD,
+    QUERY_FIELD,
+    TITLE_FIELD,
+    OPENING_TIMESTAMP_FIELD,
+    CREATION_TIMESTAMP_FIELD,
+    LIQUIDITY_FIELD,
+    LIQUIDIY_MEASURE_FIELD,
+    TOKEN_AMOUNTS_FIELD,
+    ERROR_FIELD,
+    FPMMS_FIELD,
+    omen_market_sorted_trades_query,
+    get_initial_amount_tokens_query,
+)
+
+headers = {
+    "Accept": "application/json, multipart/mixed",
+    "Content-Type": "application/json",
+}
+logger = logging.getLogger(__name__)
+
+ResponseItemType = List[Dict[str, str]]
+SubgraphResponseType = Dict[str, ResponseItemType]
+
+
+class RetriesExceeded(Exception):
+    """Exception to raise when retries are exceeded during data-fetching."""
+
+    def __init__(
+        self, msg="Maximum retries were exceeded while trying to fetch the data!"
+    ):
+        super().__init__(msg)
+
+
+def hacky_retry(func: Callable, n_retries: int = 3) -> Callable:
+    """Create a hacky retry strategy.
+        Unfortunately, we cannot use `requests.packages.urllib3.util.retry.Retry`,
+        because the subgraph does not return the appropriate status codes in case of failure.
+        Instead, it always returns code 200. Thus, we raise exceptions manually inside `make_request`,
+        catch those exceptions in the hacky retry decorator and try again.
+        Finally, if the allowed number of retries is exceeded, we raise a custom `RetriesExceeded` exception.
+
+    :param func: the input request function.
+    :param n_retries: the maximum allowed number of retries.
+    :return: The request method with the hacky retry strategy applied.
+    """
+
+    @functools.wraps(func)
+    def wrapper_hacky_retry(*args, **kwargs) -> SubgraphResponseType:
+        """The wrapper for the hacky retry.
+
+        :return: a response dictionary.
+        """
+        retried = 0
+
+        while retried <= n_retries:
+            try:
+                if retried > 0:
+                    warnings.warn(f"Retrying {retried}/{n_retries}...")
+
+                return func(*args, **kwargs)
+            except (ValueError, ConnectionError) as e:
+                warnings.warn(e.args[0])
+            finally:
+                retried += 1
+
+        raise RetriesExceeded()
+
+    return wrapper_hacky_retry
+
+
+@hacky_retry
+def query_subgraph(url: str, query: str, key: str) -> SubgraphResponseType:
+    """Query a subgraph.
+
+    Args:
+        url: the subgraph's URL.
+        query: the query to be used.
+        key: the key to use in order to access the required data.
+
+    Returns:
+        a response dictionary.
+    """
+    content = {QUERY_FIELD: query}
+    headers = {
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+    }
+    res = requests.post(url, json=content, headers=headers)
+
+    if res.status_code != 200:
+        raise ConnectionError(
+            "Something went wrong while trying to communicate with the subgraph "
+            f"(Error: {res.status_code})!\n{res.text}"
+        )
+
+    body = res.json()
+    if ERROR_FIELD in body.keys():
+        raise ValueError(f"The given query is not correct: {body[ERROR_FIELD]}")
+
+    data = body.get(DATA_FIELD, {}).get(key, None)
+    if data is None:
+        raise ValueError(f"Unknown error encountered!\nRaw response: \n{body}")
+
+    return data
+
+
+def fpmms_fetcher(
+    start_timestamp: int, end_timestamp: int
+) -> Generator[ResponseItemType, int, None]:
+    """An indefinite fetcher for the FPMMs."""
+
+    omen_subgraph = OMEN_SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
+    logger.debug(f"omen_subgraph = {omen_subgraph}")
+    while True:
+        fpmm_id = yield
+        fpmms_query = FPMMS_CLOSED_MARKETS_QUERY.substitute(
+            creator=CREATOR,
+            fpmm_id=fpmm_id,
+            start_timestamp=start_timestamp,
+            end_timestamp=end_timestamp,
+            fpmms_field=FPMMS_FIELD,
+            id_field=ID_FIELD,
+            title_field=TITLE_FIELD,
+            opening_timestamp_field=OPENING_TIMESTAMP_FIELD,
+            creation_timestamp_field=CREATION_TIMESTAMP_FIELD,
+            liquidity_field=LIQUIDITY_FIELD,
+            liquidity_measure_field=LIQUIDIY_MEASURE_FIELD,
+            token_amounts_field=TOKEN_AMOUNTS_FIELD,
+        )
+        logger.debug(f"Executing query {fpmms_query}")
+        yield query_subgraph(omen_subgraph, fpmms_query, FPMMS_FIELD)
+
+
+def fetch_fpmms(start_timestamp: int, end_timestamp: int) -> pd.DataFrame:
+    """Fetch all the fpmms of the creator."""
+    logger.info("Fetching all markets")
+    latest_id = ""
+    fpmms = []
+    fetcher = fpmms_fetcher(start_timestamp, end_timestamp)
+    for _ in tqdm(fetcher, unit="fpmms", unit_scale=BATCH_SIZE):
+        batch = fetcher.send(latest_id)
+        if len(batch) == 0:
+            logger.debug("no data")
+            break
+
+        logger.debug(f"length of the data received = {len(batch)}")
+        latest_id = batch[-1].get(ID_FIELD, "")
+        if latest_id == "":
+            raise ValueError(f"Unexpected data format retrieved: {batch}")
+
+        fpmms.extend(batch)
+
+    logger.info("Finished collecting data")
+    return pd.DataFrame(fpmms)
+
+
+def get_first_token_perc(row):
+    """To compute from the total amount of tokens bought how many are for first outcome"""
+    if row["total_tokens"] == 0.0:
+        return 0
+
+    return 100.0 - round((row["token_first_amount"] / row["total_tokens"]) * 100, 2)
+
+
+def get_second_token_perc(row):
+    """To compute from the total amount of tokens bought how many are for second outcome"""
+    if row["total_tokens"] == 0.0:
+        return 0
+    return 100.0 - round((row["token_second_amount"] / row["total_tokens"]) * 100, 2)
+
+
+def remove_testing_market(markets_data: pd.DataFrame) -> pd.DataFrame:
+    return markets_data.loc[
+        markets_data["id"] != "0xbfa584b29891941c8950ce975c1f7fa595ce1b99"
+    ]
+
+
+def transform_fpmms(fpmms: pd.DataFrame) -> pd.DataFrame:
+    """Transform an FPMMS dataframe."""
+
+    fpmms["liquidityMeasure"] = fpmms["liquidityMeasure"].apply(lambda x: int(x))
+    fpmms["creationTimestamp"] = fpmms["creationTimestamp"].apply(
+        lambda x: datetime.fromtimestamp(int(x))
+    )
+    fpmms["openingTimestamp"] = fpmms["openingTimestamp"].apply(
+        lambda x: datetime.fromtimestamp(int(x))
+    )
+    fpmms["token_first_amount"] = fpmms.outcomeTokenAmounts.apply(lambda x: int(x[0]))
+    fpmms["token_second_amount"] = fpmms.outcomeTokenAmounts.apply(lambda x: int(x[1]))
+    fpmms["total_tokens"] = fpmms.apply(
+        lambda x: x.token_first_amount + x.token_second_amount, axis=1
+    )
+    fpmms["first_token_perc"] = fpmms.apply(lambda x: get_first_token_perc(x), axis=1)
+    fpmms["second_token_perc"] = fpmms.apply(lambda x: get_second_token_perc(x), axis=1)
+    fpmms.drop(
+        columns=[
+            "token_first_amount",
+            "token_second_amount",
+            "total_tokens",
+        ],
+        inplace=True,
+    )
+    return fpmms
+
+
+@measure_execution_time
+def collect_closed_markets(filename: Optional[str]) -> pd.DataFrame:
+    """Fetch, process, store and return the markets as a Dataframe."""
+
+    logger.info("fetching new markets information")
+    today = datetime.now(UTC)
+    delta = timedelta(days=150)  # 4 months approx
+    ref_day = today - delta
+
+    start_timestamp = int(ref_day.timestamp())  # seconds
+    end_timestamp = int((today - timedelta(days=20)).timestamp())
+    fpmms = fetch_fpmms(start_timestamp, end_timestamp)
+    logger.debug("Finished markets data collection")
+    logger.debug(fpmms.info())
+
+    logger.info("transforming data")
+    fpmms = transform_fpmms(fpmms)
+
+    logger.info("saving the data")
+    # logger.debug(fpmms.info())
+    if filename:
+        fpmms.to_parquet(DATA_DIR / filename, index=False)
+
+    return fpmms
+
+
+def collect_liquidity_info(fpmm_id: str, logger) -> dict[str, Any]:
+    omen_subgraph = OMEN_SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
+    query = get_initial_amount_tokens_query.substitute(
+        fpmm_creator=CREATOR.lower(),
+        fpmm_id=fpmm_id,
+    )
+    content_json = _to_content(query)
+    logger.info(f"Executing liquidity query {query}")
+    res = requests.post(omen_subgraph, headers=headers, json=content_json)
+    result_json = res.json()
+    tokens_info = result_json.get("data", {}).get("fpmmLiquidities", [])
+    if not tokens_info:
+        return None
+
+    # the first item is the initial information of the market
+    first_info = tokens_info[0]
+    token_amounts = [int(x) for x in first_info["outcomeTokenAmounts"]]
+    return token_amounts
+
+
+@measure_execution_time
+def collect_trades_subgraph(fpmm_id: str, logger) -> dict[str, Any]:
+    """Query the subgraph."""
+    omen_subgraph = OMEN_SUBGRAPH_URL.substitute(subgraph_api_key=SUBGRAPH_API_KEY)
+    logger.info(f"omen_subgraph = {omen_subgraph}")
+    grouped_results = defaultdict(list)
+    id_gt = ""
+
+    while True:
+        query = omen_market_sorted_trades_query.substitute(
+            fpmm_creator=CREATOR.lower(),
+            first=BATCH_SIZE,
+            id_gt=id_gt,
+            fpmm_id=fpmm_id,
+        )
+        # logger.debug(f"query for the omen to collect trades {query}")
+        content_json = _to_content(query)
+
+        res = requests.post(omen_subgraph, headers=headers, json=content_json)
+        result_json = res.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 compute_price_amounts(
+    trades_outcome_0: pd.DataFrame,
+    trades_outcome_1: pd.DataFrame,
+    prev_usd_outcome_0: float = None,
+    prev_usd_outcome_1: float = None,
+) -> Tuple:
+    total_usd_outcome_0 = trades_outcome_0.collateralAmountUSD.sum()
+    total_usd_outcome_1 = trades_outcome_1.collateralAmountUSD.sum()
+    if prev_usd_outcome_0 is not None:
+        total_usd_outcome_0 += prev_usd_outcome_0
+        total_usd_outcome_1 += prev_usd_outcome_1
+    return total_usd_outcome_0, total_usd_outcome_1
+
+
+def compute_price_weighted_distributions(
+    total_usd_outcome_0, total_usd_outcome_1, logger
+) -> Tuple:
+    total_usd = total_usd_outcome_0 + total_usd_outcome_1
+    percentage_pwc_outcome_0 = round((total_usd_outcome_0 / total_usd) * 100, 2)
+    logger.debug(f"total amount for outcome 0 = {total_usd_outcome_0}")
+    logger.debug(f"total amount for outcome 1 = {total_usd_outcome_1}")
+    return percentage_pwc_outcome_0, 100 - percentage_pwc_outcome_0
+
+
+def get_updated_amounts(
+    k: int,
+    price_amount_constant: float,
+    tokens_outcome_0: int,
+    tokens_outcome_1: int,
+    bet_per_token: int,
+    outcome_index: int,
+    old_price_outcome: float,
+    new_price_outcome: float,
+    first_trade: bool,
+):
+    if first_trade:
+        # first trade ever, same price both tokens
+        other_shares = bet_per_token / old_price_outcome
+        if outcome_index == 0:
+            # other is outcome 1
+            new_other_tokens_amount = tokens_outcome_1 + other_shares
+            return int(k / new_other_tokens_amount), new_other_tokens_amount
+        else:
+            # other is outcome 0
+            new_other_tokens_amount = tokens_outcome_0 + other_shares
+            return new_other_tokens_amount, int(k / new_other_tokens_amount)
+
+    # we can use the price_amount constant to get the new final amount of tokens in the market
+    # for the selected outcome
+    new_voted_tokens_amount = price_amount_constant / new_price_outcome
+    new_other_tokens_amount = k / new_voted_tokens_amount
+    if outcome_index == 0:
+        return new_voted_tokens_amount, new_other_tokens_amount
+    # outcome_index 1
+    return new_other_tokens_amount, new_voted_tokens_amount
+
+
+def compute_token_amounts(
+    logger,
+    initial_amount_tokens: list,
+    initial_token_prices: list,
+    selected_trades: pd.DataFrame,
+    prev_tokens_first_outcome: int,
+    prev_tokens_second_outcome: int,
+) -> Tuple:
+    """Function to compute the final distribution of tokens depending on the trades done that day"""
+    # traverse the trades and adjust the initial amount of tokens
+    tokens_outcome_0, tokens_outcome_1 = initial_amount_tokens
+    # market constant
+    k = prod(initial_amount_tokens)
+    # price * token_amount constant, used to adjust the price of tokens
+    price_amount_constant = initial_amount_tokens[0] * initial_token_prices[0]
+    if prev_tokens_first_outcome != 0:
+        # reset the variables
+        tokens_outcome_0 = prev_tokens_first_outcome
+        tokens_outcome_1 = prev_tokens_second_outcome
+
+    for i in range(len(selected_trades)):
+        trade = selected_trades.iloc[i]
+        outcome_index = int(trade.outcomeIndex)
+        collateral_amount = int(trade.collateralAmount)
+        fee_amount = int(trade.feeAmount)
+        net_amount = collateral_amount - fee_amount
+        bet_per_token = int(net_amount / 2)
+        old_price_outcome = float(trade.oldOutcomeTokenMarginalPrice)
+        new_price_outcome = float(trade.outcomeTokenMarginalPrice)
+        tokens_outcome_0, tokens_outcome_1 = get_updated_amounts(
+            k,
+            price_amount_constant,
+            tokens_outcome_0,
+            tokens_outcome_1,
+            bet_per_token,
+            outcome_index,
+            old_price_outcome,
+            new_price_outcome,
+            prev_tokens_first_outcome == 0,
+        )
+
+    logger.info(f"New token amounts [{tokens_outcome_0}, {tokens_outcome_1}]")
+    return tokens_outcome_0, tokens_outcome_1
+
+
+def compute_tokens_distribution(tokens_outcome_0, tokens_outcome_1) -> Tuple:
+    total_amount_tokens = tokens_outcome_0 + tokens_outcome_1
+    # the inverse of the final tokens distribution is the emergent probability
+    tokens_perc_first_outcome = 100.0 - round(
+        (tokens_outcome_0 / total_amount_tokens) * 100, 2
+    )
+    return tokens_perc_first_outcome, 100 - tokens_perc_first_outcome
+
+
+def compute_daily_gap_metric(
+    market_trades: pd.DataFrame, market_id: str, logger
+) -> pd.DataFrame:
+
+    new_market_daily_metrics = pd.DataFrame()
+    # group by days, they are already sorted by creationtimestamp
+    market_trades["creation_date"] = market_trades["creationTimestamp"].apply(
+        lambda x: x.date()
+    )
+
+    # list of days to traverse
+    market_days = list(market_trades.creation_date.unique())
+    new_market_daily_metrics["market_day"] = market_days
+    new_market_daily_metrics["market_id"] = market_id
+    new_market_daily_metrics["daily_trades"] = 0
+    new_market_daily_metrics["mean_trade_size"] = 0
+    new_market_daily_metrics["total_bet_amount"] = 0.0
+    new_market_daily_metrics["dist_gap_perc"] = 0.0
+    new_market_daily_metrics["perc_total_trades"] = 0.0
+    new_market_daily_metrics["creator_id"] = (
+        "0x89c5cc945dd550BcFfb72Fe42BfF002429F46Fec"
+    )
+
+    logger.info(f"Total days to compute the metrics = {len(market_days)}")
+    total_nr_trades = len(market_trades)
+
+    # compute the initial amount of tokens for that market
+    initial_amount_tokens = collect_liquidity_info(market_id, logger)
+    if initial_amount_tokens is None:
+        logger.info("No initial tokens information for this market. Skipping it")
+        return None
+
+    logger.info(f"Initial amount of tokens {initial_amount_tokens}")
+    total_cum_trades = 0
+    # traverse the market days
+    for i, row in tqdm(
+        new_market_daily_metrics.iterrows(),
+        total=len(new_market_daily_metrics),
+        desc="Computing daily distributions",
+    ):
+        day = row["market_day"]
+        selected_trades = market_trades.loc[market_trades["creation_date"] == day]
+        logger.info(
+            f"Adding aggregated metrics for day={day}. Number of trades = {len(selected_trades)}"
+        )
+        daily_trades = len(selected_trades)
+        total_cum_trades += daily_trades
+        new_market_daily_metrics.at[i, "daily_trades"] = daily_trades
+        mean_trade_size = selected_trades.collateralAmountUSD.mean()
+        new_market_daily_metrics.at[i, "mean_trade_size"] = round(mean_trade_size, 3)
+        total_bet_amount = sum(selected_trades.collateralAmountUSD)
+        new_market_daily_metrics.at[i, "total_bet_amount"] = total_bet_amount
+        new_market_daily_metrics.at[i, "perc_total_trades"] = round(
+            (total_cum_trades / total_nr_trades) * 100, 2
+        )
+        # trades for outcome 0
+        trades_outcome_0 = selected_trades.loc[selected_trades["outcomeIndex"] == 0]
+        logger.debug(f"Total trades for outcome 0 = {len(trades_outcome_0)}")
+        # trades for outcome 1
+        trades_outcome_1 = selected_trades.loc[selected_trades["outcomeIndex"] == 1]
+        logger.debug(f"Total trades for outcome 1 = {len(trades_outcome_1)}")
+        # initial prices for tokens in the market
+        first_trade = selected_trades.iloc[0]
+        price_token_0 = float(
+            first_trade.oldOutcomeTokenMarginalPrice
+        )  # same price both
+        initial_token_prices = [price_token_0, price_token_0]
+        try:
+
+            # token amounts and price amounts
+            if i == 0:
+                # first day no previous information
+                tokens_outcome_0, tokens_outcome_1 = compute_token_amounts(
+                    logger,
+                    initial_amount_tokens,
+                    initial_token_prices,
+                    selected_trades,
+                    0,
+                    0,
+                )
+                total_usd_outcome_0, total_usd_outcome_1 = compute_price_amounts(
+                    trades_outcome_0, trades_outcome_1
+                )
+            else:
+                tokens_outcome_0, tokens_outcome_1 = compute_token_amounts(
+                    logger,
+                    initial_amount_tokens,
+                    initial_token_prices,
+                    selected_trades,
+                    prev_tokens_outcome_0,
+                    prev_tokens_outcome_1,
+                )
+                total_usd_outcome_0, total_usd_outcome_1 = compute_price_amounts(
+                    trades_outcome_0,
+                    trades_outcome_1,
+                    prev_usd_outcome_0,
+                    prev_usd_outcome_1,
+                )
+
+            prev_tokens_outcome_0 = tokens_outcome_0
+            prev_tokens_outcome_1 = tokens_outcome_1
+            prev_usd_outcome_0 = total_usd_outcome_0
+            prev_usd_outcome_1 = total_usd_outcome_1
+            # price weighted distribution at the end of the day
+            pwc_first_outcome, pwc_second_outcome = (
+                compute_price_weighted_distributions(
+                    total_usd_outcome_0, total_usd_outcome_1, logger
+                )
+            )
+            logger.info(
+                f"pwc distributions eod = {pwc_first_outcome, pwc_second_outcome}"
+            )
+            #  distribution based on initial token amounts
+            tokens_perc_first_outcome, tokens_perc_second_outcome = (
+                compute_tokens_distribution(
+                    tokens_outcome_0,
+                    tokens_outcome_1,
+                )
+            )
+            logger.info(
+                f"token distributions eod = {tokens_perc_first_outcome, tokens_perc_second_outcome}"
+            )
+            # bought tokens distribution at the end of the day
+            # bought_tokens_first, bought_tokens_second = compute_bought_tokens_distribution(
+            #     trades_outcome_0, trades_outcome_1
+            # )
+            metric = abs(tokens_perc_first_outcome - pwc_first_outcome)
+            logger.info(f"eod {day}: gap metric {metric}")
+            new_market_daily_metrics.at[i, "dist_gap_perc"] = metric
+        except Exception as e:
+            logger.error(e)
+            logger.warning(
+                f"It was not possible to compute the distributions for the day {day} and market id {market_id}"
+            )
+            continue
+    return new_market_daily_metrics
+
+
+def collect_trades_info(filename, logger):
+    if os.path.exists(DATA_DIR / filename):
+        fpmms = pd.read_parquet(DATA_DIR / filename)
+    else:
+        raise FileNotFoundError("Parquet file with trades not found")
+
+    fpmms = remove_testing_market(fpmms)
+    # Iterate over the markets
+    markets_with_daily_metrics = None
+
+    for i, fpmm in tqdm(fpmms.iterrows(), total=len(fpmms), desc="Analysing trades"):
+        # get the market id
+        market_id = fpmm["id"]
+        logger.info(f"Adding trades information for the market {market_id}")
+        try:
+            market_trades_json = collect_trades_subgraph(
+                fpmm_id=market_id, logger=logger
+            )
+            market_trades = pd.DataFrame(market_trades_json["data"]["fpmmTrades"])
+            if len(market_trades) == 0:
+                logger.warning("No trades for this market")
+                continue
+
+            market_trades.outcomeIndex = pd.to_numeric(
+                market_trades.outcomeIndex, errors="coerce"
+            )
+            fpmms.at[i, "total_trades"] = len(market_trades)
+        except Exception:
+            fpmms.at[i, "total_trades"] = 0
+            continue
+
+        if len(market_trades) > 0:
+            # formatting types
+            market_trades["collateralAmountUSD"] = (
+                market_trades.collateralAmountUSD.apply(lambda x: round(float(x), 3))
+            )
+            market_trades["outcomeTokensTraded"] = (
+                market_trades.outcomeTokensTraded.apply(lambda x: int(x))
+            )
+            market_trades["oldOutcomeTokenMarginalPrice"] = (
+                market_trades.oldOutcomeTokenMarginalPrice.apply(
+                    lambda x: round(float(x), 9)
+                )
+            )
+            market_trades["outcomeTokenMarginalPrice"] = (
+                market_trades.outcomeTokenMarginalPrice.apply(
+                    lambda x: round(float(x), 9)
+                )
+            )
+            # datetimes
+            market_trades["creationTimestamp"] = market_trades[
+                "creationTimestamp"
+            ].apply(lambda x: datetime.fromtimestamp(int(x)))
+
+            new_data = compute_daily_gap_metric(market_trades, market_id, logger)
+            if new_data is None:
+                continue
+
+            if markets_with_daily_metrics is None:
+                markets_with_daily_metrics = new_data
+            else:
+                markets_with_daily_metrics = pd.concat(
+                    [markets_with_daily_metrics, new_data], axis=0, ignore_index=True
+                )
+            # saving the trades information into the file
+            logger.info(
+                f"Saving trades info into the parquet file. Length of file = {len(markets_with_daily_metrics)}"
+            )
+            markets_with_daily_metrics.to_parquet(
+                DATA_DIR / "markets_with_daily_metrics.parquet", index=False
+            )
+
+            fpmms.to_parquet(DATA_DIR / filename, index=False)
+
+
+if __name__ == "__main__":
+
+    logger.setLevel(logging.DEBUG)
+    # stream handler and formatter
+    stream_handler = logging.StreamHandler()
+    stream_handler.setLevel(logging.DEBUG)
+    formatter = logging.Formatter(
+        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    )
+    stream_handler.setFormatter(formatter)
+    logging.basicConfig(
+        handlers=[logging.FileHandler("live_markets.log"), stream_handler]
+    )
+    # collect_closed_markets("closed_markets_data.parquet")
+    collect_trades_info("closed_markets_data.parquet", logger)

scripts/queries.py

@@ -63,6 +63,7 @@ FPMMS_WITH_TOKENS_QUERY = Template(
         ${title_field}
         ${opening_timestamp_field}
         ${creation_timestamp_field}
+        resolutionTimestamp
         ${liquidity_field}
         ${liquidity_measure_field}
         ${token_amounts_field}
@@ -71,6 +72,33 @@ FPMMS_WITH_TOKENS_QUERY = Template(
     """
 )
 
+FPMMS_CLOSED_MARKETS_QUERY = Template(
+    """
+    {
+      ${fpmms_field}(
+        where: {
+          creator: "${creator}",
+          id_gt: "${fpmm_id}",
+          creationTimestamp_gt: ${start_timestamp}
+          creationTimestamp_lt: ${end_timestamp}
+        },
+        orderBy: ${id_field}
+        orderDirection: asc
+        first: 1000
+      ){
+        ${id_field}
+        isPendingArbitration
+
+        ${title_field}
+        ${opening_timestamp_field}
+        ${creation_timestamp_field}
+        ${liquidity_field}
+        ${liquidity_measure_field}
+        ${token_amounts_field}
+      }
+    }
+    """
+)
 
 omen_market_trades_query = Template(
     """
@@ -116,3 +144,61 @@ omen_market_trades_query = Template(
     }
     """
 )
+
+omen_market_sorted_trades_query = Template(
+    """
+    {
+        fpmmTrades(
+            where: {
+                type: Buy,
+                fpmm_: {
+                    creator: "${fpmm_creator}",
+                    id: "${fpmm_id}",
+                },
+                id_gt: "${id_gt}"
+            }
+            first: 1000
+            orderBy: creationTimestamp
+            orderDirection: asc
+        ) {
+            id
+            title
+            collateralToken
+            outcomeTokenMarginalPrice
+            oldOutcomeTokenMarginalPrice
+            creationTimestamp
+            collateralAmount
+            collateralAmountUSD
+            feeAmount
+            outcomeIndex
+            outcomeTokensTraded
+            transactionHash
+        }
+    }
+    """
+)
+
+get_initial_amount_tokens_query = Template(
+    """
+    {
+  
+      fpmmLiquidities(
+        where: {
+                fpmm_: {
+                        creator: "${fpmm_creator}",
+                        id: "${fpmm_id}",
+                    },
+                id_gt: ""
+                }
+                orderBy: creationTimestamp
+                orderDirection: asc
+        )
+      {
+        id
+        outcomeTokenAmounts
+        creationTimestamp
+        additionalLiquidityParameter
+      }
+    }
+"""
+)

scripts/utils.py

@@ -17,7 +17,6 @@ from json.decoder import JSONDecodeError
 
 CREATOR = "0x89c5cc945dd550BcFfb72Fe42BfF002429F46Fec"
 BATCH_SIZE = 1000
-# OMEN_SUBGRAPH = "https://api.thegraph.com/subgraphs/name/protofire/omen-xdai"
 OMEN_SUBGRAPH_URL = Template(
     """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/9fUVQpFwzpdWS9bq5WkAnmKbNNcoBwatMR4yZq81pbbz"""
 )