Fix USDC const

Inspect many writing 10 blocks at a time - 40s => 30s locally

alembic/versions/99d376cb93cc_error.py

@@ -0,0 +1,23 @@
+"""error column
+
+Revision ID: 99d376cb93cc
+Revises: c4a7620a2d33
+Create Date: 2021-12-21 21:26:12.142484
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "99d376cb93cc"
+down_revision = "c4a7620a2d33"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("arbitrages", sa.Column("error", sa.String(256), nullable=True))
+
+
+def downgrade():
+    op.drop_column("arbitrages", "error")

alembic/versions/c4a7620a2d33_create_tokens_table.py

@@ -0,0 +1,28 @@
+"""Create tokens table
+
+Revision ID: c4a7620a2d33
+Revises: 15ba9c27ee8a
+Create Date: 2021-12-21 19:12:33.940117
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c4a7620a2d33"
+down_revision = "15ba9c27ee8a"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "tokens",
+        sa.Column("token_address", sa.String(256), nullable=False),
+        sa.Column("decimals", sa.Numeric, nullable=False),
+        sa.PrimaryKeyConstraint("token_address"),
+    )
+
+
+def downgrade():
+    op.drop_table("tokens")

listener

@@ -23,6 +23,7 @@ case "$1" in
   stop)
         echo -n "Stopping daemon: "$NAME
 	start-stop-daemon --stop --quiet --oknodo --pidfile $PIDFILE
+    rm $PIDFILE
         echo "."
 	;;
   tail)
@@ -31,6 +32,7 @@ case "$1" in
   restart)
         echo -n "Restarting daemon: "$NAME
 	start-stop-daemon --stop --quiet --oknodo --retry 30 --pidfile $PIDFILE
+    rm $PIDFILE
 	start-stop-daemon \
         --background \
         --chdir /app \

listener.py

@@ -65,14 +65,10 @@ async def inspect_next_block(
 
     if last_written_block is None:
         # maintain lag if no blocks written yet
-        last_written_block = latest_block_number - 1
+        last_written_block = latest_block_number - BLOCK_NUMBER_LAG - 1
 
     if last_written_block < (latest_block_number - BLOCK_NUMBER_LAG):
-        block_number = (
-            latest_block_number
-            if last_written_block is None
-            else last_written_block + 1
-        )
+        block_number = last_written_block + 1
 
         logger.info(f"Writing block: {block_number}")
 

mev

@@ -13,7 +13,7 @@ function db(){
     username=$(get_kube_db_secret "username")
     password=$(get_kube_db_secret "password")
 
-    kubectl run -i --rm --tty postgres-client \
+    kubectl run -i --rm --tty postgres-client-$RANDOM \
         --env="PGPASSWORD=$password"  \
         --image=jbergknoff/postgresql-client \
         -- $DB_NAME --host=$host --user=$username

mev_inspect/arbitrages.py

@@ -56,6 +56,10 @@ def _get_arbitrages_from_swaps(swaps: List[Swap]) -> List[Arbitrage]:
             start_amount = route[0].token_in_amount
             end_amount = route[-1].token_out_amount
             profit_amount = end_amount - start_amount
+            error = None
+            for swap in route:
+                if swap.error is not None:
+                    error = swap.error
 
             arb = Arbitrage(
                 swaps=route,
@@ -66,6 +70,7 @@ def _get_arbitrages_from_swaps(swaps: List[Swap]) -> List[Arbitrage]:
                 start_amount=start_amount,
                 end_amount=end_amount,
                 profit_amount=profit_amount,
+                error=error,
             )
             all_arbitrages.append(arb)
     if len(all_arbitrages) == 1:

mev_inspect/classifiers/specs/zero_ex.py

@@ -221,10 +221,16 @@ ZEROX_CLASSIFIER_SPECS = ZEROX_CONTRACT_SPECS + ZEROX_GENERIC_SPECS
 
 
 def _get_taker_token_in_amount(
-    taker_address: str, token_in_address: str, child_transfers: List[Transfer]
+    trace: DecodedCallTrace,
+    taker_address: str,
+    token_in_address: str,
+    child_transfers: List[Transfer],
 ) -> int:
 
-    if len(child_transfers) != 2:
+    if trace.error is not None:
+        return 0
+
+    if len(child_transfers) < 2:
         raise ValueError(
             f"A settled order should consist of 2 child transfers, not {len(child_transfers)}."
         )
@@ -237,7 +243,8 @@ def _get_taker_token_in_amount(
         for transfer in child_transfers:
             if transfer.to_address == taker_address:
                 return transfer.amount
-    return 0
+
+    raise RuntimeError("Unable to find transfers matching 0x order.")
 
 
 def _get_0x_token_in_data(
@@ -259,7 +266,7 @@ def _get_0x_token_in_data(
         )
 
     token_in_amount = _get_taker_token_in_amount(
-        taker_address, token_in_address, child_transfers
+        trace, taker_address, token_in_address, child_transfers
     )
 
     return token_in_address, token_in_amount

mev_inspect/coinbase.py

@@ -4,10 +4,13 @@ from mev_inspect.classifiers.specs.weth import WETH_ADDRESS
 from mev_inspect.schemas.coinbase import CoinbasePrices, CoinbasePricesResponse
 from mev_inspect.schemas.prices import (
     AAVE_TOKEN_ADDRESS,
+    CDAI_TOKEN_ADDRESS,
+    CUSDC_TOKEN_ADDRESS,
+    DAI_TOKEN_ADDRESS,
     LINK_TOKEN_ADDRESS,
     REN_TOKEN_ADDRESS,
     UNI_TOKEN_ADDRESS,
-    USDC_TOKEN_ADDRESS_ADDRESS,
+    USDC_TOKEN_ADDRESS,
     WBTC_TOKEN_ADDRESS,
     YEARN_TOKEN_ADDRESS,
 )
@@ -22,8 +25,11 @@ COINBASE_TOKEN_NAME_BY_ADDRESS = {
     YEARN_TOKEN_ADDRESS: "yearn-finance",
     AAVE_TOKEN_ADDRESS: "aave",
     UNI_TOKEN_ADDRESS: "uniswap",
-    USDC_TOKEN_ADDRESS_ADDRESS: "usdc",
+    USDC_TOKEN_ADDRESS: "usdc",
+    DAI_TOKEN_ADDRESS: "dai",
     REN_TOKEN_ADDRESS: "ren",
+    CUSDC_TOKEN_ADDRESS: "compound-usd-coin",
+    CDAI_TOKEN_ADDRESS: "compound-dai",
 }
 
 

mev_inspect/compound_liquidations.py

@@ -44,6 +44,7 @@ def get_compound_liquidations(
                             debt_purchase_amount=trace.value,
                             protocol=trace.protocol,
                             received_amount=seize_trace.inputs["seizeTokens"],
+                            received_token_address=trace.to_address,
                             transaction_hash=trace.transaction_hash,
                             trace_address=trace.trace_address,
                             block_number=trace.block_number,
@@ -60,6 +61,7 @@ def get_compound_liquidations(
                             debt_purchase_amount=trace.inputs["repayAmount"],
                             protocol=trace.protocol,
                             received_amount=seize_trace.inputs["seizeTokens"],
+                            received_token_address=trace.to_address,
                             transaction_hash=trace.transaction_hash,
                             trace_address=trace.trace_address,
                             block_number=trace.block_number,

mev_inspect/crud/arbitrages.py

@@ -4,17 +4,20 @@ from uuid import uuid4
 from mev_inspect.models.arbitrages import ArbitrageModel
 from mev_inspect.schemas.arbitrages import Arbitrage
 
+from .shared import delete_by_block_range
 
-def delete_arbitrages_for_block(
+
+def delete_arbitrages_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(ArbitrageModel)
-        .filter(ArbitrageModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        ArbitrageModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 
@@ -37,6 +40,7 @@ def write_arbitrages(
                 start_amount=arbitrage.start_amount,
                 end_amount=arbitrage.end_amount,
                 profit_amount=arbitrage.profit_amount,
+                error=arbitrage.error,
             )
         )
 

mev_inspect/crud/blocks.py

@@ -3,13 +3,22 @@ from datetime import datetime
 from mev_inspect.schemas.blocks import Block
 
 
-def delete_block(
+def delete_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
     db_session.execute(
-        "DELETE FROM blocks WHERE block_number = :block_number",
-        params={"block_number": block_number},
+        """
+        DELETE FROM blocks
+        WHERE
+            block_number >= :after_block_number AND
+            block_number < :before_block_number
+        """,
+        params={
+            "after_block_number": after_block_number,
+            "before_block_number": before_block_number,
+        },
     )
     db_session.commit()
 

mev_inspect/crud/liquidations.py

@@ -4,17 +4,20 @@ from typing import List
 from mev_inspect.models.liquidations import LiquidationModel
 from mev_inspect.schemas.liquidations import Liquidation
 
+from .shared import delete_by_block_range
 
-def delete_liquidations_for_block(
+
+def delete_liquidations_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(LiquidationModel)
-        .filter(LiquidationModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        LiquidationModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 

mev_inspect/crud/miner_payments.py

@@ -4,17 +4,20 @@ from typing import List
 from mev_inspect.models.miner_payments import MinerPaymentModel
 from mev_inspect.schemas.miner_payments import MinerPayment
 
+from .shared import delete_by_block_range
 
-def delete_miner_payments_for_block(
+
+def delete_miner_payments_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(MinerPaymentModel)
-        .filter(MinerPaymentModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        MinerPaymentModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 

mev_inspect/crud/punks.py

@@ -10,17 +10,20 @@ from mev_inspect.schemas.punk_accept_bid import PunkBidAcceptance
 from mev_inspect.schemas.punk_bid import PunkBid
 from mev_inspect.schemas.punk_snipe import PunkSnipe
 
+from .shared import delete_by_block_range
 
-def delete_punk_bid_acceptances_for_block(
+
+def delete_punk_bid_acceptances_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(PunkBidAcceptanceModel)
-        .filter(PunkBidAcceptanceModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        PunkBidAcceptanceModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 
@@ -37,16 +40,17 @@ def write_punk_bid_acceptances(
     db_session.commit()
 
 
-def delete_punk_bids_for_block(
+def delete_punk_bids_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(PunkBidModel)
-        .filter(PunkBidModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        PunkBidModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 
@@ -60,16 +64,17 @@ def write_punk_bids(
     db_session.commit()
 
 
-def delete_punk_snipes_for_block(
+def delete_punk_snipes_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(PunkSnipeModel)
-        .filter(PunkSnipeModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        PunkSnipeModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 

mev_inspect/crud/sandwiches.py

@@ -4,17 +4,20 @@ from uuid import uuid4
 from mev_inspect.models.sandwiches import SandwichModel
 from mev_inspect.schemas.sandwiches import Sandwich
 
+from .shared import delete_by_block_range
 
-def delete_sandwiches_for_block(
+
+def delete_sandwiches_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(SandwichModel)
-        .filter(SandwichModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        SandwichModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 

mev_inspect/crud/shared.py

@@ -0,0 +1,20 @@
+from typing import Type
+
+from mev_inspect.models.base import Base
+
+
+def delete_by_block_range(
+    db_session,
+    model_class: Type[Base],
+    after_block_number,
+    before_block_number,
+) -> None:
+
+    (
+        db_session.query(model_class)
+        .filter(model_class.block_number >= after_block_number)
+        .filter(model_class.block_number < before_block_number)
+        .delete()
+    )
+
+    db_session.commit()

mev_inspect/crud/swaps.py

@@ -4,17 +4,20 @@ from typing import List
 from mev_inspect.models.swaps import SwapModel
 from mev_inspect.schemas.swaps import Swap
 
+from .shared import delete_by_block_range
 
-def delete_swaps_for_block(
+
+def delete_swaps_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(SwapModel)
-        .filter(SwapModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        SwapModel,
+        after_block_number,
+        before_block_number,
     )
-
     db_session.commit()
 
 

mev_inspect/crud/traces.py

@@ -4,15 +4,19 @@ from typing import List
 from mev_inspect.models.traces import ClassifiedTraceModel
 from mev_inspect.schemas.traces import ClassifiedTrace
 
+from .shared import delete_by_block_range
 
-def delete_classified_traces_for_block(
+
+def delete_classified_traces_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(ClassifiedTraceModel)
-        .filter(ClassifiedTraceModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        ClassifiedTraceModel,
+        after_block_number,
+        before_block_number,
     )
 
     db_session.commit()

mev_inspect/crud/transfers.py

@@ -4,15 +4,19 @@ from typing import List
 from mev_inspect.models.transfers import TransferModel
 from mev_inspect.schemas.transfers import Transfer
 
+from .shared import delete_by_block_range
 
-def delete_transfers_for_block(
+
+def delete_transfers_for_blocks(
     db_session,
-    block_number: int,
+    after_block_number: int,
+    before_block_number: int,
 ) -> None:
-    (
-        db_session.query(TransferModel)
-        .filter(TransferModel.block_number == block_number)
-        .delete()
+    delete_by_block_range(
+        db_session,
+        TransferModel,
+        after_block_number,
+        before_block_number,
     )
 
     db_session.commit()

mev_inspect/db.py

@@ -12,7 +12,7 @@ def get_trace_database_uri() -> Optional[str]:
     db_name = "trace_db"
 
     if all(field is not None for field in [username, password, host]):
-        return f"postgresql://{username}:{password}@{host}/{db_name}"
+        return f"postgresql+psycopg2://{username}:{password}@{host}/{db_name}"
 
     return None
 
@@ -22,11 +22,16 @@ def get_inspect_database_uri():
     password = os.getenv("POSTGRES_PASSWORD")
     host = os.getenv("POSTGRES_HOST")
     db_name = "mev_inspect"
-    return f"postgresql://{username}:{password}@{host}/{db_name}"
+    return f"postgresql+psycopg2://{username}:{password}@{host}/{db_name}"
 
 
 def _get_engine(uri: str):
-    return create_engine(uri)
+    return create_engine(
+        uri,
+        executemany_mode="values",
+        executemany_values_page_size=10000,
+        executemany_batch_page_size=500,
+    )
 
 
 def _get_session(uri: str):

mev_inspect/inspect_block.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Optional
+from typing import List, Optional
 
 from sqlalchemy import orm
 from web3 import Web3
@@ -7,35 +7,46 @@ from web3 import Web3
 from mev_inspect.arbitrages import get_arbitrages
 from mev_inspect.block import create_from_block_number
 from mev_inspect.classifiers.trace import TraceClassifier
-from mev_inspect.crud.arbitrages import delete_arbitrages_for_block, write_arbitrages
-from mev_inspect.crud.blocks import delete_block, write_block
+from mev_inspect.crud.arbitrages import delete_arbitrages_for_blocks, write_arbitrages
+from mev_inspect.crud.blocks import delete_blocks, write_block
 from mev_inspect.crud.liquidations import (
-    delete_liquidations_for_block,
+    delete_liquidations_for_blocks,
     write_liquidations,
 )
 from mev_inspect.crud.miner_payments import (
-    delete_miner_payments_for_block,
+    delete_miner_payments_for_blocks,
     write_miner_payments,
 )
 from mev_inspect.crud.punks import (
-    delete_punk_bid_acceptances_for_block,
-    delete_punk_bids_for_block,
-    delete_punk_snipes_for_block,
+    delete_punk_bid_acceptances_for_blocks,
+    delete_punk_bids_for_blocks,
+    delete_punk_snipes_for_blocks,
     write_punk_bid_acceptances,
     write_punk_bids,
     write_punk_snipes,
 )
-from mev_inspect.crud.sandwiches import delete_sandwiches_for_block, write_sandwiches
-from mev_inspect.crud.swaps import delete_swaps_for_block, write_swaps
+from mev_inspect.crud.sandwiches import delete_sandwiches_for_blocks, write_sandwiches
+from mev_inspect.crud.swaps import delete_swaps_for_blocks, write_swaps
 from mev_inspect.crud.traces import (
-    delete_classified_traces_for_block,
+    delete_classified_traces_for_blocks,
     write_classified_traces,
 )
-from mev_inspect.crud.transfers import delete_transfers_for_block, write_transfers
+from mev_inspect.crud.transfers import delete_transfers_for_blocks, write_transfers
 from mev_inspect.liquidations import get_liquidations
 from mev_inspect.miner_payments import get_miner_payments
 from mev_inspect.punks import get_punk_bid_acceptances, get_punk_bids, get_punk_snipes
 from mev_inspect.sandwiches import get_sandwiches
+from mev_inspect.schemas.arbitrages import Arbitrage
+from mev_inspect.schemas.blocks import Block
+from mev_inspect.schemas.liquidations import Liquidation
+from mev_inspect.schemas.miner_payments import MinerPayment
+from mev_inspect.schemas.punk_accept_bid import PunkBidAcceptance
+from mev_inspect.schemas.punk_bid import PunkBid
+from mev_inspect.schemas.punk_snipe import PunkSnipe
+from mev_inspect.schemas.sandwiches import Sandwich
+from mev_inspect.schemas.swaps import Swap
+from mev_inspect.schemas.traces import ClassifiedTrace
+from mev_inspect.schemas.transfers import Transfer
 from mev_inspect.swaps import get_swaps
 from mev_inspect.transfers import get_transfers
 
@@ -51,79 +62,154 @@ async def inspect_block(
     trace_db_session: Optional[orm.Session],
     should_write_classified_traces: bool = True,
 ):
-    block = await create_from_block_number(
+    await inspect_many_blocks(
+        inspect_db_session,
         base_provider,
         w3,
+        trace_classifier,
         block_number,
+        block_number + 1,
         trace_db_session,
+        should_write_classified_traces,
     )
 
-    logger.info(f"Block: {block_number} -- Total traces: {len(block.traces)}")
 
-    delete_block(inspect_db_session, block_number)
+async def inspect_many_blocks(
+    inspect_db_session: orm.Session,
+    base_provider,
+    w3: Web3,
+    trace_classifier: TraceClassifier,
+    after_block_number: int,
+    before_block_number: int,
+    trace_db_session: Optional[orm.Session],
+    should_write_classified_traces: bool = True,
+):
+    all_blocks: List[Block] = []
+    all_classified_traces: List[ClassifiedTrace] = []
+    all_transfers: List[Transfer] = []
+    all_swaps: List[Swap] = []
+    all_arbitrages: List[Arbitrage] = []
+    all_liqudations: List[Liquidation] = []
+    all_sandwiches: List[Sandwich] = []
+
+    all_punk_bids: List[PunkBid] = []
+    all_punk_bid_acceptances: List[PunkBidAcceptance] = []
+    all_punk_snipes: List[PunkSnipe] = []
+
+    all_miner_payments: List[MinerPayment] = []
+
+    for block_number in range(after_block_number, before_block_number):
+        block = await create_from_block_number(
+            base_provider,
+            w3,
+            block_number,
+            trace_db_session,
+        )
+
+        logger.info(f"Block: {block_number} -- Total traces: {len(block.traces)}")
+
+        total_transactions = len(
+            set(
+                t.transaction_hash
+                for t in block.traces
+                if t.transaction_hash is not None
+            )
+        )
+        logger.info(
+            f"Block: {block_number} -- Total transactions: {total_transactions}"
+        )
+
+        classified_traces = trace_classifier.classify(block.traces)
+        logger.info(
+            f"Block: {block_number} -- Returned {len(classified_traces)} classified traces"
+        )
+
+        transfers = get_transfers(classified_traces)
+        logger.info(f"Block: {block_number} -- Found {len(transfers)} transfers")
+
+        swaps = get_swaps(classified_traces)
+        logger.info(f"Block: {block_number} -- Found {len(swaps)} swaps")
+
+        arbitrages = get_arbitrages(swaps)
+        logger.info(f"Block: {block_number} -- Found {len(arbitrages)} arbitrages")
+
+        liquidations = get_liquidations(classified_traces)
+        logger.info(f"Block: {block_number} -- Found {len(liquidations)} liquidations")
+
+        sandwiches = get_sandwiches(swaps)
+        logger.info(f"Block: {block_number} -- Found {len(sandwiches)} sandwiches")
+
+        punk_bids = get_punk_bids(classified_traces)
+        punk_bid_acceptances = get_punk_bid_acceptances(classified_traces)
+        punk_snipes = get_punk_snipes(punk_bids, punk_bid_acceptances)
+        logger.info(f"Block: {block_number} -- Found {len(punk_snipes)} punk snipes")
+
+        miner_payments = get_miner_payments(
+            block.miner, block.base_fee_per_gas, classified_traces, block.receipts
+        )
+
+        all_blocks.append(block)
+        all_classified_traces.extend(classified_traces)
+        all_transfers.extend(transfers)
+        all_swaps.extend(swaps)
+        all_arbitrages.extend(arbitrages)
+        all_liqudations.extend(liquidations)
+        all_sandwiches.extend(sandwiches)
+
+        all_punk_bids.extend(punk_bids)
+        all_punk_bid_acceptances.extend(punk_bid_acceptances)
+        all_punk_snipes.extend(punk_snipes)
+
+        all_miner_payments.extend(miner_payments)
+
+    delete_blocks(inspect_db_session, after_block_number, before_block_number)
     write_block(inspect_db_session, block)
 
-    total_transactions = len(
-        set(t.transaction_hash for t in block.traces if t.transaction_hash is not None)
-    )
-    logger.info(f"Block: {block_number} -- Total transactions: {total_transactions}")
-
-    classified_traces = trace_classifier.classify(block.traces)
-    logger.info(
-        f"Block: {block_number} -- Returned {len(classified_traces)} classified traces"
-    )
-
     if should_write_classified_traces:
-        delete_classified_traces_for_block(inspect_db_session, block_number)
+        delete_classified_traces_for_blocks(
+            inspect_db_session, after_block_number, before_block_number
+        )
         write_classified_traces(inspect_db_session, classified_traces)
 
-    transfers = get_transfers(classified_traces)
-    logger.info(f"Block: {block_number} -- Found {len(transfers)} transfers")
-
-    delete_transfers_for_block(inspect_db_session, block_number)
+    delete_transfers_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
+    )
     write_transfers(inspect_db_session, transfers)
 
-    swaps = get_swaps(classified_traces)
-    logger.info(f"Block: {block_number} -- Found {len(swaps)} swaps")
-
-    delete_swaps_for_block(inspect_db_session, block_number)
+    delete_swaps_for_blocks(inspect_db_session, after_block_number, before_block_number)
     write_swaps(inspect_db_session, swaps)
 
-    arbitrages = get_arbitrages(swaps)
-    logger.info(f"Block: {block_number} -- Found {len(arbitrages)} arbitrages")
-
-    delete_arbitrages_for_block(inspect_db_session, block_number)
+    delete_arbitrages_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
+    )
     write_arbitrages(inspect_db_session, arbitrages)
 
-    liquidations = get_liquidations(classified_traces)
-    logger.info(f"Block: {block_number} -- Found {len(liquidations)} liquidations")
-
-    delete_liquidations_for_block(inspect_db_session, block_number)
+    delete_liquidations_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
+    )
     write_liquidations(inspect_db_session, liquidations)
 
-    sandwiches = get_sandwiches(swaps)
-    logger.info(f"Block: {block_number} -- Found {len(sandwiches)} sandwiches")
-
-    delete_sandwiches_for_block(inspect_db_session, block_number)
+    delete_sandwiches_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
+    )
     write_sandwiches(inspect_db_session, sandwiches)
 
-    punk_bids = get_punk_bids(classified_traces)
-    delete_punk_bids_for_block(inspect_db_session, block_number)
+    delete_punk_bids_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
+    )
     write_punk_bids(inspect_db_session, punk_bids)
 
-    punk_bid_acceptances = get_punk_bid_acceptances(classified_traces)
-    delete_punk_bid_acceptances_for_block(inspect_db_session, block_number)
+    delete_punk_bid_acceptances_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
+    )
     write_punk_bid_acceptances(inspect_db_session, punk_bid_acceptances)
 
-    punk_snipes = get_punk_snipes(punk_bids, punk_bid_acceptances)
-    logger.info(f"Block: {block_number} -- Found {len(punk_snipes)} punk snipes")
-
-    delete_punk_snipes_for_block(inspect_db_session, block_number)
+    delete_punk_snipes_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
+    )
     write_punk_snipes(inspect_db_session, punk_snipes)
 
-    miner_payments = get_miner_payments(
-        block.miner, block.base_fee_per_gas, classified_traces, block.receipts
+    delete_miner_payments_for_blocks(
+        inspect_db_session, after_block_number, before_block_number
     )
-
-    delete_miner_payments_for_block(inspect_db_session, block_number)
     write_miner_payments(inspect_db_session, miner_payments)

mev_inspect/inspector.py

@@ -10,7 +10,7 @@ from web3.eth import AsyncEth
 
 from mev_inspect.block import create_from_block_number
 from mev_inspect.classifiers.trace import TraceClassifier
-from mev_inspect.inspect_block import inspect_block
+from mev_inspect.inspect_block import inspect_block, inspect_many_blocks
 from mev_inspect.provider import get_base_provider
 
 logger = logging.getLogger(__name__)
@@ -50,12 +50,23 @@ class MEVInspector:
             trace_db_session=self.trace_db_session,
         )
 
-    async def inspect_many_blocks(self, after_block: int, before_block: int):
+    async def inspect_many_blocks(
+        self,
+        after_block: int,
+        before_block: int,
+        block_batch_size: int = 10,
+    ):
         tasks = []
-        for block_number in range(after_block, before_block):
+        for block_number in range(after_block, before_block, block_batch_size):
+            batch_after_block = block_number
+            batch_before_block = min(block_number + block_batch_size, before_block)
+
             tasks.append(
                 asyncio.ensure_future(
-                    self.safe_inspect_block(block_number=block_number)
+                    self.safe_inspect_many_blocks(
+                        after_block_number=batch_after_block,
+                        before_block_number=batch_before_block,
+                    )
                 )
             )
         logger.info(f"Gathered {len(tasks)} blocks to inspect")
@@ -67,13 +78,18 @@ class MEVInspector:
             logger.error(f"Existed due to {type(e)}")
             traceback.print_exc()
 
-    async def safe_inspect_block(self, block_number: int):
+    async def safe_inspect_many_blocks(
+        self,
+        after_block_number: int,
+        before_block_number: int,
+    ):
         async with self.max_concurrency:
-            return await inspect_block(
+            return await inspect_many_blocks(
                 self.inspect_db_session,
                 self.base_provider,
                 self.w3,
                 self.trace_classifier,
-                block_number,
+                after_block_number,
+                before_block_number,
                 trace_db_session=self.trace_db_session,
             )

mev_inspect/models/arbitrages.py

@@ -14,3 +14,4 @@ class ArbitrageModel(Base):
     start_amount = Column(Numeric, nullable=False)
     end_amount = Column(Numeric, nullable=False)
     profit_amount = Column(Numeric, nullable=False)
+    error = Column(String, nullable=True)

mev_inspect/prices.py

@@ -7,7 +7,7 @@ from mev_inspect.schemas.prices import (
     LINK_TOKEN_ADDRESS,
     REN_TOKEN_ADDRESS,
     UNI_TOKEN_ADDRESS,
-    USDC_TOKEN_ADDRESS_ADDRESS,
+    USDC_TOKEN_ADDRESS,
     WBTC_TOKEN_ADDRESS,
     YEARN_TOKEN_ADDRESS,
     Price,
@@ -19,7 +19,7 @@ SUPPORTED_TOKENS = [
     ETH_TOKEN_ADDRESS,
     LINK_TOKEN_ADDRESS,
     AAVE_TOKEN_ADDRESS,
-    USDC_TOKEN_ADDRESS_ADDRESS,
+    USDC_TOKEN_ADDRESS,
     REN_TOKEN_ADDRESS,
     WBTC_TOKEN_ADDRESS,
     YEARN_TOKEN_ADDRESS,

mev_inspect/sandwiches.py

@@ -48,12 +48,13 @@ def _get_sandwich_starting_with_swap(
                 and other_swap.token_in_address == front_swap.token_out_address
                 and other_swap.from_address == sandwicher_address
             ):
-                return Sandwich(
-                    block_number=front_swap.block_number,
-                    sandwicher_address=sandwicher_address,
-                    frontrun_swap=front_swap,
-                    backrun_swap=other_swap,
-                    sandwiched_swaps=sandwiched_swaps,
-                )
+                if len(sandwiched_swaps) > 0:
+                    return Sandwich(
+                        block_number=front_swap.block_number,
+                        sandwicher_address=sandwicher_address,
+                        frontrun_swap=front_swap,
+                        backrun_swap=other_swap,
+                        sandwiched_swaps=sandwiched_swaps,
+                    )
 
     return None

mev_inspect/schemas/arbitrages.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import List, Optional
 
 from pydantic import BaseModel
 
@@ -14,3 +14,4 @@ class Arbitrage(BaseModel):
     start_amount: int
     end_amount: int
     profit_amount: int
+    error: Optional[str]

mev_inspect/schemas/prices.py

@@ -1,17 +1,24 @@
 from datetime import datetime
 
-from pydantic import BaseModel
+from pydantic import BaseModel, validator
 
 WBTC_TOKEN_ADDRESS = "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599"
 LINK_TOKEN_ADDRESS = "0x514910771af9ca656af840dff83e8264ecf986ca"
 YEARN_TOKEN_ADDRESS = "0x0bc529c00c6401aef6d220be8c6ea1667f6ad93e"
 AAVE_TOKEN_ADDRESS = "0x7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9"
 UNI_TOKEN_ADDRESS = "0x1f9840a85d5af5bf1d1762f925bdaddc4201f984"
-USDC_TOKEN_ADDRESS_ADDRESS = "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"
+USDC_TOKEN_ADDRESS = "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"
+DAI_TOKEN_ADDRESS = "0x6b175474e89094c44da98b954eedeac495271d0f"
 REN_TOKEN_ADDRESS = "0x408e41876cccdc0f92210600ef50372656052a38"
+CUSDC_TOKEN_ADDRESS = "0x39aa39c021dfbae8fac545936693ac917d5e7563"
+CDAI_TOKEN_ADDRESS = "0x5d3a536e4d6dbd6114cc1ead35777bab948e3643"
 
 
 class Price(BaseModel):
     token_address: str
     timestamp: datetime
     usd_price: float
+
+    @validator("token_address")
+    def lower_token_address(cls, v: str) -> str:
+        return v.lower()

tests/test_arbitrage_integration.py

@@ -57,3 +57,23 @@ def test_arbitrage_real_block(trace_classifier: TraceClassifier):
         == "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"
     )
     assert arbitrage_2.profit_amount == 53560707941943273628
+
+
+def test_reverting_arbitrage(trace_classifier: TraceClassifier):
+    block = load_test_block(11473321)
+    classified_traces = trace_classifier.classify(block.traces)
+
+    swaps = get_swaps(classified_traces)
+    assert len(swaps) == 38
+
+    arbitrages = get_arbitrages(list(swaps))
+    assert len(arbitrages) == 21
+
+    arbitrage_1 = [
+        arb
+        for arb in arbitrages
+        if arb.transaction_hash
+        == "0x565146ec57af69208b4a37e3a138ab85c6a6ff358fffb0077824a7378a67c4d6"
+    ][0]
+
+    assert arbitrage_1.error == "Reverted"

tests/test_compound.py

@@ -21,6 +21,7 @@ def test_c_ether_liquidations(trace_classifier: TraceClassifier):
             debt_token_address="0x39aa39c021dfbae8fac545936693ac917d5e7563",
             debt_purchase_amount=268066492249420078,
             received_amount=4747650169097,
+            received_token_address="0x4ddc2d193948926d02f9b1fe9e1daa0718270ed5",
             protocol=Protocol.compound_v2,
             transaction_hash=transaction_hash,
             trace_address=[1],
@@ -44,6 +45,7 @@ def test_c_ether_liquidations(trace_classifier: TraceClassifier):
             debt_token_address="0x35a18000230da775cac24873d00ff85bccded550",
             debt_purchase_amount=414547860568297082,
             received_amount=321973320649,
+            received_token_address="0x4ddc2d193948926d02f9b1fe9e1daa0718270ed5",
             protocol=Protocol.compound_v2,
             transaction_hash=transaction_hash,
             trace_address=[1],
@@ -68,6 +70,7 @@ def test_c_ether_liquidations(trace_classifier: TraceClassifier):
             debt_token_address="0x35a18000230da775cac24873d00ff85bccded550",
             debt_purchase_amount=1106497772527562662,
             received_amount=910895850496,
+            received_token_address="0x4ddc2d193948926d02f9b1fe9e1daa0718270ed5",
             protocol=Protocol.compound_v2,
             transaction_hash=transaction_hash,
             trace_address=[1],
@@ -93,6 +96,7 @@ def test_c_token_liquidation(trace_classifier: TraceClassifier):
             debt_token_address="0x70e36f6bf80a52b3b46b3af8e106cc0ed743e8e4",
             debt_purchase_amount=1207055531,
             received_amount=21459623305,
+            received_token_address="0x39aa39c021dfbae8fac545936693ac917d5e7563",
             protocol=Protocol.compound_v2,
             transaction_hash=transaction_hash,
             trace_address=[1],
@@ -118,6 +122,7 @@ def test_cream_token_liquidation(trace_classifier: TraceClassifier):
             debt_token_address="0x44fbebd2f576670a6c33f6fc0b00aa8c5753b322",
             debt_purchase_amount=14857434973806369550,
             received_amount=1547215810826,
+            received_token_address="0x697256caa3ccafd62bb6d3aa1c7c5671786a5fd9",
             protocol=Protocol.cream,
             transaction_hash=transaction_hash,
             trace_address=[],