use parquet instead of csv

app.py

@@ -106,8 +106,8 @@ def refresh_data():
         
         logging.info("Refreshing data...")
         
-        tools_df = pd.read_csv("./data/tools.csv", low_memory=False)
-        trades_df = pd.read_csv("./data/all_trades_profitability.csv")
+        tools_df = pd.read_parquet("./data/tools.parquet")
+        trades_df = pd.read_parquet("./data/all_trades_profitability.parquet")
         trades_df = prepare_trades(trades_df)
         error_df = get_error_data(tools_df=tools_df, inc_tools=INC_TOOLS)
         error_overall_df = get_error_data_overall(error_df=error_df)
@@ -134,8 +134,8 @@ def pull_refresh_data():
     refresh_data()
 
         
-tools_df = pd.read_csv("./data/tools.csv", low_memory=False)
-trades_df = pd.read_csv("./data/all_trades_profitability.csv")
+tools_df = pd.read_parquet("./data/tools.parquet")
+trades_df = pd.read_parquet("./data/all_trades_profitability.parquet")
 trades_df = prepare_trades(trades_df)
 
 

data/{all_trades_profitability.csv → all_trades_profitability.parquet}

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:ea9047bacb53f4f2d396242ae39939517fde4b4061f425c992e981cc92c5b452
-size 34257800
+oid sha256:ae0de6d7e607b8ac33140081ab5415b9c16e7359d23b196e555535af0d78965c
+size 8251611

data/{fpmmTrades.csv → fpmmTrades.parquet}

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:6a297b050b6c7c88c8fe7d0c597f362b95c0af735f65ff3218e9c748bdcbb820
-size 76092671
+oid sha256:bb0cd005a2bb7b37b04e0388538249ab6434c9de532b337fcee775ab9205064c
+size 20528876

data/{fpmms.csv → fpmms.parquet}

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:c313ba383cbff6ed1bdacadade252d9e5ae8b66359336c99b3ad7845842e533d
-size 448312
+oid sha256:5b0b82cf173571152d11bbcabd94f675e8d84c148925f47a96c5192d9b9e2f67
+size 319767

data/requests.csv

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

data/requests.parquet

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

data/{summary_profitability.csv → summary_profitability.parquet}

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:60f49e3d95e3abc0d93b7518e3287f9fbbe65211b3447ed21030ab4b5415c7a9
-size 65116
+oid sha256:0ef6d6a03b5f872d0228881b74e3a2427c4e8a5f7fd02776eb70683605ccbb4b
+size 52394

data/t_map.pkl

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:c1a007c0c28c3ab7cd85597a9cedde455c346b481adec25c6ab0223f50808c9c
-size 7422234
+oid sha256:2738a5a8e98ca83c409251237cc338ed540c0ea58779bf23ea59255fa88b42d5
+size 7749840

scripts/markets.py

@@ -46,7 +46,7 @@ QUESTION_FIELD = "question"
 OUTCOMES_FIELD = "outcomes"
 TITLE_FIELD = "title"
 MAX_UINT_HEX = "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
-DEFAULT_FILENAME = "fpmms.csv"
+DEFAULT_FILENAME = "fpmms.parquet"
 SCRIPTS_DIR = Path(__file__).parent
 ROOT_DIR = SCRIPTS_DIR.parent
 DATA_DIR = ROOT_DIR / "data"
@@ -218,10 +218,11 @@ def etl(filename: Optional[str] = None) -> pd.DataFrame:
     fpmms = transform_fpmms(fpmms)
 
     if filename:
-        fpmms.to_csv(DATA_DIR / filename, index=False)
+        fpmms.to_parquet(DATA_DIR / filename, index=False)
 
     return fpmms
 
 
 if __name__ == "__main__":
     etl(DEFAULT_FILENAME)
+

scripts/profitability.py

@@ -385,7 +385,7 @@ def create_fpmmTrades(rpc: str):
     df.rename(columns={"creator": "trader_address"}, inplace=True)
 
     # save to csv
-    df.to_csv(DATA_DIR / "fpmmTrades.csv", index=False)
+    df.to_parquet(DATA_DIR / "fpmmTrades.parquet", index=False)
 
     return df
 
@@ -396,7 +396,7 @@ def prepare_profitalibity_data(rpc: str):
     # Check if tools.py is in the same directory
     try:
         # load tools.csv
-        tools = pd.read_csv(DATA_DIR / "tools.csv")
+        tools = pd.read_parquet(DATA_DIR / "tools.parquet")
 
         # make sure creator_address is in the columns
         assert "trader_address" in tools.columns, "trader_address column not found"
@@ -407,21 +407,21 @@ def prepare_profitalibity_data(rpc: str):
         # drop duplicates
         tools.drop_duplicates(inplace=True)
 
-        print("tools.csv loaded")
+        print("tools.parquet loaded")
     except FileNotFoundError:
-        print("tools.csv not found. Please run tools.py first.")
+        print("tools.parquet not found. Please run tools.py first.")
         return
 
     # Check if fpmmTrades.csv is in the same directory
     try:
         # load fpmmTrades.csv
-        fpmmTrades = pd.read_csv(DATA_DIR / "fpmmTrades.csv")
-        print("fpmmTrades.csv loaded")
+        fpmmTrades = pd.read_parquet(DATA_DIR / "fpmmTrades.parquet")
+        print("fpmmTrades.parquet loaded")
     except FileNotFoundError:
-        print("fpmmTrades.csv not found. Creating fpmmTrades.csv...")
+        print("fpmmTrades.parquet not found. Creating fpmmTrades.parquet...")
         fpmmTrades = create_fpmmTrades(rpc)
-        fpmmTrades.to_csv(DATA_DIR / "fpmmTrades.csv", index=False)
-        fpmmTrades = pd.read_csv(DATA_DIR / "fpmmTrades.csv")
+        fpmmTrades.to_parquet(DATA_DIR / "fpmmTrades.parquet", index=False)
+        fpmmTrades = pd.read_parquet(DATA_DIR / "fpmmTrades.parquet")
 
     # make sure trader_address is in the columns
     assert "trader_address" in fpmmTrades.columns, "trader_address column not found"
@@ -434,13 +434,13 @@ def prepare_profitalibity_data(rpc: str):
 
 def determine_market_status(trade, current_answer):
     """Determine the market status of a trade."""
-    if current_answer is np.nan and time.time() >= trade["fpmm.openingTimestamp"]:
+    if current_answer is np.nan and time.time() >= int(trade["fpmm.openingTimestamp"]):
         return MarketState.PENDING
     elif current_answer == np.nan:
         return MarketState.OPEN
     elif trade["fpmm.isPendingArbitration"]:
         return MarketState.ARBITRATING
-    elif time.time() < trade["fpmm.answerFinalizedTimestamp"]:
+    elif time.time() < int(trade["fpmm.answerFinalizedTimestamp"]):
         return MarketState.FINALIZING
     return MarketState.CLOSED
 
@@ -468,9 +468,12 @@ def analyse_trader(
     # Iterate over the trades
     for i, trade in tqdm(trades.iterrows(), total=len(trades), desc="Analysing trades"):
         try:
+            if not trade['fpmm.currentAnswer']:
+                print(f"Skipping trade {i} because currentAnswer is NaN")
+                continue
             # Parsing and computing shared values
             creation_timestamp_utc = datetime.datetime.fromtimestamp(
-                trade["creationTimestamp"], tz=datetime.timezone.utc
+                int(trade["creationTimestamp"]), tz=datetime.timezone.utc
             )
             collateral_amount = wei_to_unit(float(trade["collateralAmount"]))
             fee_amount = wei_to_unit(float(trade["feeAmount"]))
@@ -497,7 +500,7 @@ def analyse_trader(
             if is_invalid:
                 earnings = collateral_amount
                 winner_trade = False
-            elif trade["outcomeIndex"] == current_answer:
+            elif int(trade["outcomeIndex"]) == current_answer:
                 earnings = outcome_tokens_traded
                 winner_trade = True
 
@@ -610,6 +613,7 @@ def run_profitability_analysis(rpc):
     # load dfs from csv for analysis
     print("Preparing data...")
     fpmmTrades, tools = prepare_profitalibity_data(rpc)
+    tools['trader_address'] = tools['trader_address'].str.lower()
 
     # all trades profitability df
     print("Analysing trades...")
@@ -620,8 +624,8 @@ def run_profitability_analysis(rpc):
     summary_df = summary_analyse(all_trades_df)
 
     # save to csv
-    all_trades_df.to_csv(DATA_DIR / "all_trades_profitability.csv", index=False)
-    summary_df.to_csv(DATA_DIR / "summary_profitability.csv", index=False)
+    all_trades_df.to_parquet(DATA_DIR / "all_trades_profitability.parquet", index=False)
+    summary_df.to_parquet(DATA_DIR / "summary_profitability.parquet", index=False)
 
     print("Done!")
 

scripts/pull_data.py

@@ -85,16 +85,16 @@ def weekly_analysis():
 
     # Run profitability analysis
     logging.info("Running profitability analysis")
-    if os.path.exists(DATA_DIR / "fpmmTrades.csv"):
-        os.remove(DATA_DIR / "fpmmTrades.csv")
+    if os.path.exists(DATA_DIR / "fpmmTrades.parquet"):
+        os.remove(DATA_DIR / "fpmmTrades.parquet")
     run_profitability_analysis(
         rpc=rpc,
     )
     logging.info("Profitability analysis completed")
 
     # Get currentAnswer from FPMMS
-    fpmms = pd.read_csv(DATA_DIR / MARKETS_FILENAME)
-    tools = pd.read_csv(DATA_DIR / TOOLS_FILENAME)
+    fpmms = pd.read_parquet(DATA_DIR / MARKETS_FILENAME)
+    tools = pd.read_parquet(DATA_DIR / TOOLS_FILENAME)
 
     # Get the question from the tools
     logging.info("Getting the question and current answer for the tools")
@@ -123,7 +123,7 @@ def weekly_analysis():
     tools['request_month_year_week'] = pd.to_datetime(tools['request_time']).dt.to_period('W').astype(str)
 
     # Save the tools
-    tools.to_csv(DATA_DIR / TOOLS_FILENAME, index=False)
+    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']

scripts/tools.py

@@ -86,7 +86,7 @@ IPFS_ADDRESS = f"{HTTPS}gateway.autonolas.tech/ipfs/"
 IPFS_LINKS_SERIES_NAME = "ipfs_links"
 BACKOFF_FACTOR = 1
 STATUS_FORCELIST = [404, 500, 502, 503, 504]
-DEFAULT_FILENAME = "tools.csv"
+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"
 SLEEP = 0.5
@@ -580,7 +580,7 @@ def transform_deliver(contents: pd.DataFrame, full_contents=False) -> pd.DataFra
 
 def gen_event_filename(event_name: MechEventName) -> str:
     """Generate the filename of an event."""
-    return f"{event_name.value.lower()}s.csv"
+    return f"{event_name.value.lower()}s.parquet"
 
 
 def read_n_last_lines(filename: str, n: int = 1) -> str:
@@ -605,33 +605,38 @@ def get_earliest_block(event_name: MechEventName) -> int:
     if not os.path.exists(DATA_DIR / filename):
         return 0
 
-    cols = pd.read_csv(DATA_DIR / filename, index_col=0, nrows=0).columns.tolist()
-    last_line_buff = StringIO(read_n_last_lines(DATA_DIR/filename))
-    last_line_series = pd.read_csv(last_line_buff, names=cols)
+    df = pd.read_parquet(DATA_DIR / filename)
     block_field = f"{event_name.value.lower()}_{BLOCK_FIELD}"
-    return int(last_line_series[block_field].values[0])
+    return int(df[block_field].max())
 
 
 def store_progress(
     filename: str,
-    event_to_contents: Dict[MechEventName, pd.DataFrame],
+    event_to_contents: Dict[str, pd.DataFrame],
     tools: pd.DataFrame,
 ) -> None:
     """Store the given progress."""
     if filename:
+        DATA_DIR.mkdir(parents=True, exist_ok=True)  # Ensure the directory exists
         for event_name, content in event_to_contents.items():
-            event_filename = gen_event_filename(event_name)
-
-            if "result" in content.columns:
-                content.drop(columns=["result"], inplace=True)
-
-            content.to_csv(DATA_DIR / event_filename, index=False, escapechar="\\")
-
-        # drop result and error columns
-        if "result" in tools.columns:
-            tools.drop(columns=["result"], inplace=True)
-
-        tools.to_csv(DATA_DIR / filename, index=False, escapechar="\\")
+            event_filename = gen_event_filename(event_name)  # Ensure this function returns a valid filename string
+            try:
+                if "result" in content.columns:
+                    content = content.drop(columns=["result"])  # Avoid in-place modification
+                if 'error' in content.columns:
+                    content['error'] = content['error'].astype(bool)
+                content.to_parquet(DATA_DIR / event_filename, index=False)
+            except Exception as e:
+                print(f"Failed to write {event_name}: {e}")
+        # Drop result and error columns for tools DataFrame
+        try:
+            if "result" in tools.columns:
+                tools = tools.drop(columns=["result"])
+            if 'error' in tools.columns:
+                tools['error'] = tools['error'].astype(bool)
+            tools.to_parquet(DATA_DIR / filename, index=False)
+        except Exception as e:
+            print(f"Failed to write tools data: {e}")
 
 
 def etl(
@@ -736,7 +741,7 @@ def etl(
         events_filename = gen_event_filename(event_name)
 
         if os.path.exists(DATA_DIR / events_filename):
-            old = pd.read_csv(DATA_DIR / events_filename)
+            old = pd.read_parquet(DATA_DIR / events_filename)
 
             # Reset index to avoid index conflicts
             old.reset_index(drop=True, inplace=True)