Merge pull request #134 from flashbots/prices

Add support for fetching prices from coinbase and storing

cli.py

@@ -5,12 +5,15 @@ import sys
 import click
 
 from mev_inspect.concurrency import coro
+from mev_inspect.crud.prices import write_prices
 from mev_inspect.db import get_inspect_session, get_trace_session
 from mev_inspect.inspector import MEVInspector
+from mev_inspect.prices import fetch_all_supported_prices
 
 RPC_URL_ENV = "RPC_URL"
 
 logging.basicConfig(stream=sys.stdout, level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 
 @click.group()
@@ -79,6 +82,18 @@ async def inspect_many_blocks_command(
     )
 
 
+@cli.command()
+@coro
+async def fetch_all_prices():
+    inspect_db_session = get_inspect_session()
+
+    logger.info("Fetching prices")
+    prices = await fetch_all_supported_prices()
+
+    logger.info("Writing prices")
+    write_prices(inspect_db_session, prices)
+
+
 def get_rpc_url() -> str:
     return os.environ["RPC_URL"]
 

mev

@@ -56,6 +56,19 @@ case "$1" in
       echo "Fetching block $block_number"
       kubectl exec -ti deploy/mev-inspect -- poetry run fetch-block $block_number
   ;;
+  prices)
+        shift
+        case "$1" in
+          fetch-all)
+                echo "Running price fetch-all"
+                kubectl exec -ti deploy/mev-inspect -- \
+                    poetry run fetch-all-prices
+            ;;
+          *)
+            echo "prices usage: "$1" {fetch-all}"
+            exit 1
+        esac
+  ;;
   exec)
         shift
         kubectl exec -ti deploy/mev-inspect -- $@

mev_inspect/coinbase.py

@@ -0,0 +1,25 @@
+import aiohttp
+
+from mev_inspect.classifiers.specs.weth import WETH_ADDRESS
+from mev_inspect.schemas.transfers import ETH_TOKEN_ADDRESS
+from mev_inspect.schemas.coinbase import CoinbasePrices, CoinbasePricesResponse
+
+
+COINBASE_API_BASE = "https://www.coinbase.com/api/v2"
+COINBASE_TOKEN_NAME_BY_ADDRESS = {
+    WETH_ADDRESS: "weth",
+    ETH_TOKEN_ADDRESS: "ethereum",
+}
+
+
+async def fetch_coinbase_prices(token_address: str) -> CoinbasePrices:
+    if token_address not in COINBASE_TOKEN_NAME_BY_ADDRESS:
+        raise ValueError(f"Unsupported token_address {token_address}")
+
+    coinbase_token_name = COINBASE_TOKEN_NAME_BY_ADDRESS[token_address]
+    url = f"{COINBASE_API_BASE}/assets/prices/{coinbase_token_name}"
+
+    async with aiohttp.ClientSession() as session:
+        async with session.get(url, params={"base": "USD"}) as response:
+            json_data = await response.json()
+            return CoinbasePricesResponse(**json_data).data.prices

mev_inspect/crud/prices.py

@@ -0,0 +1,17 @@
+from typing import List
+
+from sqlalchemy.dialects.postgresql import insert
+
+from mev_inspect.models.prices import PriceModel
+from mev_inspect.schemas.prices import Price
+
+
+def write_prices(db_session, prices: List[Price]) -> None:
+    insert_statement = (
+        insert(PriceModel.__table__)
+        .values([price.dict() for price in prices])
+        .on_conflict_do_nothing()
+    )
+
+    db_session.execute(insert_statement)
+    db_session.commit()

mev_inspect/models/prices.py

@@ -0,0 +1,11 @@
+from sqlalchemy import Column, Numeric, String, TIMESTAMP
+
+from .base import Base
+
+
+class PriceModel(Base):
+    __tablename__ = "prices"
+
+    timestamp = Column(TIMESTAMP, nullable=False, primary_key=True)
+    usd_price = Column(Numeric, nullable=False)
+    token_address = Column(String, nullable=False, primary_key=True)

mev_inspect/prices.py

@@ -0,0 +1,29 @@
+from typing import List
+
+from mev_inspect.classifiers.specs.weth import WETH_ADDRESS
+from mev_inspect.coinbase import fetch_coinbase_prices
+from mev_inspect.schemas.prices import Price
+from mev_inspect.schemas.transfers import ETH_TOKEN_ADDRESS
+
+
+SUPPORTED_TOKENS = [
+    WETH_ADDRESS,
+    ETH_TOKEN_ADDRESS,
+]
+
+
+async def fetch_all_supported_prices() -> List[Price]:
+    prices = []
+
+    for token_address in SUPPORTED_TOKENS:
+        coinbase_prices = await fetch_coinbase_prices(token_address)
+        for usd_price, timestamp_seconds in coinbase_prices.all.prices:
+            price = Price(
+                token_address=token_address,
+                usd_price=usd_price,
+                timestamp=timestamp_seconds,
+            )
+
+            prices.append(price)
+
+    return prices

mev_inspect/schemas/coinbase.py

@@ -0,0 +1,20 @@
+from typing import List, Tuple
+
+from pydantic import BaseModel
+
+
+class CoinbasePricesEntry(BaseModel):
+    # tuple of price and timestamp
+    prices: List[Tuple[float, int]]
+
+
+class CoinbasePrices(BaseModel):
+    all: CoinbasePricesEntry
+
+
+class CoinbasePricesDataResponse(BaseModel):
+    prices: CoinbasePrices
+
+
+class CoinbasePricesResponse(BaseModel):
+    data: CoinbasePricesDataResponse

mev_inspect/schemas/prices.py

@@ -0,0 +1,9 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class Price(BaseModel):
+    token_address: str
+    timestamp: datetime
+    usd_price: float

mev_inspect/schemas/utils.py

@@ -1,8 +1,8 @@
 import json
 
 from hexbytes import HexBytes
-from web3.datastructures import AttributeDict
 from pydantic import BaseModel
+from web3.datastructures import AttributeDict
 
 
 def to_camel(string: str) -> str:

pyproject.toml

@@ -34,6 +34,7 @@ build-backend = "poetry.core.masonry.api"
 inspect-block = 'cli:inspect_block_command'
 inspect-many-blocks = 'cli:inspect_many_blocks_command'
 fetch-block = 'cli:fetch_block_command'
+fetch-all-prices = 'cli:fetch_all_prices'
 
 [tool.black]
 exclude = '''