Use new database layout

.github/workflows/pull_request.yaml

@@ -6,6 +6,23 @@ on:
 jobs:
   python:
     runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: mainnet
+        # Set health checks to wait until postgres has started
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+        # Copy repo contents into container (needed to populate DB)
+        volumes:
+          - ${{ github.workspace }}/database:/docker-entrypoint-initdb.d
     steps:
       - uses: actions/checkout@v3
       - name: Setup Python 3.12
@@ -29,3 +46,5 @@ jobs:
         run: pytest tests/e2e/test_blockchain_data.py tests/e2e/test_imbalances_script.py
         env:
           NODE_URL: ${{ secrets.NODE_URL }}
+          SOLVER_SLIPPAGE_DB_URL: postgres:postgres@localhost:5432/mainnet
+          CHAIN_SLEEP_TIME: 1

Dockerfile.test_db

@@ -1,4 +1,4 @@
 FROM postgres
 ENV POSTGRES_PASSWORD=postgres
 ENV POSTGRES_DB=mainnet
-COPY ./database/01_table_creation.sql /docker-entrypoint-initdb.d/
+COPY ./database/* /docker-entrypoint-initdb.d/

database/00_legacy_tables.sql

@@ -0,0 +1,36 @@
+-- Database Schema for Token Imbalances and Slippage
+
+-- Table: raw_token_imbalances (for storing raw token imbalances)
+CREATE TABLE raw_token_imbalances (
+    auction_id BIGINT NOT NULL,
+    chain_name VARCHAR(50) NOT NULL,
+    block_number BIGINT NOT NULL,
+    tx_hash BYTEA NOT NULL,
+    token_address BYTEA NOT NULL,
+    imbalance NUMERIC(78,0),
+    PRIMARY KEY (tx_hash, token_address)
+);
+
+-- Table: slippage_prices (for storing per unit token prices in ETH)
+CREATE TABLE slippage_prices (
+    chain_name VARCHAR(50) NOT NULL,
+    source VARCHAR(50) NOT NULL,
+    block_number BIGINT NOT NULL,
+    tx_hash BYTEA NOT NULL,
+    token_address BYTEA NOT NULL,
+    price NUMERIC(42,18),
+    PRIMARY KEY (tx_hash, token_address)
+);
+
+-- Table: Stores fees (i.e. protocol fee, network fee on per token basis)
+CREATE TABLE fees_new (
+    chain_name VARCHAR(50) NOT NULL,
+    auction_id BIGINT NOT NULL,
+    block_number BIGINT NOT NULL,
+    tx_hash BYTEA NOT NULL,
+    order_uid BYTEA NOT NULL,
+    token_address BYTEA NOT NULL,
+    fee_amount NUMERIC(78,0) NOT NULL,
+    fee_type VARCHAR(50) NOT NULL, -- e.g. "protocol" or "network"
+    PRIMARY KEY (tx_hash, order_uid, token_address, fee_type)
+);

requirements.in

@@ -2,10 +2,9 @@ dune-client
 moralis
 pandas
 pandas-stubs
-psycopg2
+psycopg
 python-dotenv
 requests
-types-psycopg2
 types-requests
 SQLAlchemy
 web3

requirements.txt

@@ -2,7 +2,7 @@
 # This file is autogenerated by pip-compile with Python 3.12
 # by the following command:
 #
-#    pip-compile requirements.in
+#    pip-compile
 #
 aiohappyeyeballs==2.4.0
     # via aiohttp
@@ -139,7 +139,7 @@ platformdirs==4.3.6
     #   pylint
 pluggy==1.5.0
     # via pytest
-psycopg2==2.9.9
+psycopg==3.2.3
     # via -r requirements.in
 pycryptodome==3.20.0
     # via
@@ -187,8 +187,6 @@ toolz==0.12.1
     # via cytoolz
 types-deprecated==1.2.9.20240311
     # via dune-client
-types-psycopg2==2.9.21.20240819
-    # via -r requirements.in
 types-python-dateutil==2.9.0.20240906
     # via dune-client
 types-pytz==2024.2.0.20240913
@@ -207,6 +205,7 @@ typing-extensions==4.12.2
     #   eth-typing
     #   moralis
     #   mypy
+    #   psycopg
     #   pydantic
     #   pydantic-core
     #   sqlalchemy

src/helpers/blockchain_data.py

@@ -1,5 +1,8 @@
 from hexbytes import HexBytes
 from web3 import Web3
+from web3.types import HexStr
+
+from contracts.erc20_abi import erc20_abi
 from src.helpers.config import logger
 from src.constants import SETTLEMENT_CONTRACT_ADDRESS, INVALIDATED_ORDER_TOPIC
 
@@ -71,3 +74,34 @@ def get_auction_id(self, tx_hash: str) -> int:
         # convert bytes to int
         auction_id = int.from_bytes(call_data_bytes[-8:], byteorder="big")
         return auction_id
+
+
+def get_transaction_timestamp(tx_hash: str, web3: Web3) -> tuple[str, int]:
+    receipt = web3.eth.get_transaction_receipt(HexStr(tx_hash))
+    block_number = receipt["blockNumber"]
+    block = web3.eth.get_block(block_number)
+    timestamp = block["timestamp"]
+
+    return tx_hash, timestamp
+
+
+def get_transaction_tokens(tx_hash: str, web3: Web3) -> list[tuple[str, str]]:
+    receipt = web3.eth.get_transaction_receipt(HexStr(tx_hash))
+
+    transfer_topic = web3.keccak(text="Transfer(address,address,uint256)")
+
+    token_addresses: set[str] = set()
+    for log in receipt["logs"]:
+        if log["topics"] and log["topics"][0] == transfer_topic:
+            token_address = log["address"]
+            token_addresses.add(token_address)
+
+    return [(tx_hash, token_address) for token_address in token_addresses]
+
+
+def get_token_decimals(token_address: str, web3: Web3) -> int:
+    """Get number of decimals for a token."""
+    contract = web3.eth.contract(
+        address=Web3.to_checksum_address(token_address), abi=erc20_abi
+    )
+    return contract.functions.decimals().call()

src/helpers/config.py

@@ -51,7 +51,7 @@ def create_db_connection(db_type: str) -> Engine:
     if not db_url:
         raise ValueError(f"{db_type} database URL not found in environment variables.")
 
-    return create_engine(f"postgresql+psycopg2://{db_url}")
+    return create_engine(f"postgresql+psycopg://{db_url}")
 
 
 def check_db_connection(engine: Engine, db_type: str) -> Engine:

src/helpers/database.py

@@ -1,5 +1,9 @@
-from sqlalchemy import text
+from datetime import datetime
+
+from hexbytes import HexBytes
+from sqlalchemy import text, insert, Table, Column, Integer, LargeBinary, MetaData
 from sqlalchemy.engine import Engine
+
 from src.helpers.config import check_db_connection, logger
 from src.helpers.helper_functions import read_sql_file
 from src.constants import NULL_ADDRESS_STRING
@@ -124,3 +128,99 @@ def write_fees(
                 "fee_recipient": final_recipient,
             },
         )
+
+    def write_transaction_timestamp(
+        self, transaction_timestamp: tuple[str, int]
+    ) -> None:
+        """Function writes the transaction timestamp to the table."""
+        query = (
+            "INSERT INTO transaction_timestamps (tx_hash, time) "
+            "VALUES (:tx_hash, :time);"
+        )
+        self.execute_and_commit(
+            query,
+            {
+                "tx_hash": bytes.fromhex(transaction_timestamp[0][2:]),
+                "time": datetime.fromtimestamp(transaction_timestamp[1]),
+            },
+        )
+
+    def write_transaction_tokens(
+        self, transaction_tokens: list[tuple[str, str]]
+    ) -> None:
+        """Function writes the transaction timestamp to the table."""
+        query = (
+            "INSERT INTO transaction_tokens (tx_hash, token_address) "
+            "VALUES (:tx_hash, :token_address);"
+        )
+        for tx_hash, token_address in transaction_tokens:
+            self.execute_and_commit(
+                query,
+                {
+                    "tx_hash": bytes.fromhex(tx_hash[2:]),
+                    "token_address": bytes.fromhex(token_address[2:]),
+                },
+            )
+
+    def write_prices_new(self, prices: list[tuple[str, int, float, str]]) -> None:
+        """Write prices to database."""
+        query = (
+            "INSERT INTO prices (token_address, time, price, source) "
+            "VALUES (:token_address, :time, :price, :source);"
+        )
+        for token_address, time, price, source in prices:
+            self.execute_and_commit(
+                query,
+                {
+                    "token_address": bytes.fromhex(token_address[2:]),
+                    "time": datetime.fromtimestamp(time),
+                    "price": price,
+                    "source": source,
+                },
+            )
+
+    def get_latest_transaction(self) -> str | None:
+        """Get latest transaction hash.
+        If no transaction is found, return None."""
+        query = "SELECT tx_hash FROM transaction_timestamps ORDER BY time DESC LIMIT 1;"
+        result = self.execute_query(query, {}).fetchone()
+
+        if result is None:
+            return None
+
+        latest_tx_hash = HexBytes(result[0]).to_0x_hex()
+        return latest_tx_hash
+
+    def get_tokens_without_decimals(self) -> list[str]:
+        """Get tokens without decimals."""
+        query = (
+            "SELECT token_address FROM transaction_tokens "
+            "WHERE token_address not in (SELECT token_address FROM token_decimals);"
+        )
+        result = self.execute_query(query, {}).fetchall()
+        return list({HexBytes(row[0]).to_0x_hex() for row in result})
+
+    def write_token_decimals(self, token_decimals: list[tuple[str, int]]) -> None:
+        self.engine = check_db_connection(self.engine, "solver_slippage")
+
+        # Define the table without creating a model class
+        token_decimals_table = Table(
+            "token_decimals",
+            MetaData(),
+            Column("token_address", LargeBinary, primary_key=True),
+            Column("decimals", Integer, nullable=False),
+        )
+
+        # Prepare the data
+        values = [
+            {"token_address": bytes.fromhex(token_address[2:]), "decimals": decimals}
+            for token_address, decimals in token_decimals
+        ]
+
+        # Create the insert statement
+        stmt = insert(token_decimals_table).values(values)
+
+        # Execute the bulk insert
+        with self.engine.connect() as conn:
+            conn.execute(stmt)
+            conn.commit()

src/imbalances_script.py

@@ -24,7 +24,7 @@
 
 from web3 import Web3
 from web3.datastructures import AttributeDict
-from web3.types import TxReceipt
+from web3.types import HexStr, TxReceipt
 from src.helpers.config import CHAIN_RPC_ENDPOINTS, logger
 from src.constants import (
     SETTLEMENT_CONTRACT_ADDRESS,

src/price_providers/coingecko_pricing.py

@@ -29,7 +29,7 @@ def __init__(self) -> None:
 
     @property
     def name(self) -> str:
-        return "Coingecko"
+        return "coingecko"
 
     def fetch_coingecko_list(self) -> list[dict]:
         """

src/price_providers/dune_pricing.py

@@ -23,7 +23,7 @@ def __init__(self) -> None:
 
     @property
     def name(self) -> str:
-        return "Dune"
+        return "dune"
 
     def initialize_dune_client(self) -> DuneClient | None:
         """

src/price_providers/endpoint_auction_pricing.py

@@ -1,21 +1,24 @@
 import requests
+
 from src.price_providers.pricing_model import AbstractPriceProvider
-from src.helpers.config import logger
-from src.helpers.helper_functions import extract_params, get_token_decimals
+from src.helpers.blockchain_data import get_token_decimals
+from src.helpers.config import get_web3_instance, logger
+from src.helpers.helper_functions import extract_params
 
 
 class AuctionPriceProvider(AbstractPriceProvider):
     """Fetch auction prices."""
 
     def __init__(self) -> None:
+        self.web3 = get_web3_instance()
         self.endpoint_urls = {
             "prod": f"https://api.cow.fi/mainnet/api/v1/solver_competition/by_tx_hash/",
             "barn": f"https://barn.api.cow.fi/mainnet/api/v1/solver_competition/by_tx_hash/",
         }
 
     @property
     def name(self) -> str:
-        return "AuctionPrices"
+        return "native"
 
     def get_price(self, price_params: dict) -> float | None:
         """Function returns Auction price from endpoint for a token address."""
@@ -39,7 +42,7 @@ def get_price(self, price_params: dict) -> float | None:
                     return None
                 # calculation for converting auction price from endpoint to ETH equivalent per token unit
                 price_in_eth = (float(price) / 10**18) * (
-                    10 ** get_token_decimals(token_address) / 10**18
+                    10 ** get_token_decimals(token_address, self.web3) / 10**18
                 )
                 return price_in_eth
 

src/price_providers/moralis_pricing.py

@@ -1,11 +1,14 @@
+import os
+
+from dotenv import load_dotenv
 from moralis import evm_api
+
 from src.helpers.config import get_logger
-import os, dotenv
 from src.price_providers.pricing_model import AbstractPriceProvider
 from src.helpers.helper_functions import extract_params
 
 
-dotenv.load_dotenv()
+load_dotenv()
 
 
 class MoralisPriceProvider(AbstractPriceProvider):
@@ -18,7 +21,7 @@ def __init__(self) -> None:
 
     @property
     def name(self) -> str:
-        return "Moralis"
+        return "moralis"
 
     @staticmethod
     def wei_to_eth(price: str) -> float | None: