AWS Lambda function for periodically retrieving the best offers from opensea, looksrare, x2y2, sudoswap, nftx

lambda_function.py

import threading
from datetime import datetime
from decimal import Decimal

import boto3
import cloudscraper
import pymysql
import requests
from bs4 import BeautifulSoup
from pytz import timezone

class UpdateDB:
    def __init__(self, collections, test=False):
        self.opensea_data = {}
        self.lookrare_data = {}
        self.x2y2_data = {}
        self.sudoswap_data = {}
        self.nftx_data = {}
        format = "%Y-%m-%d %H:%M:%S"
        timestamp = datetime.now(timezone("US/Pacific")).strftime(format)
        self.connection = pymysql.connect(
            host="collection-db.cspjhlwr0o1p.us-west-1.rds.amazonaws.com",
            user="admin",
            passwd="YeZ0Nw3V8TDnp923CSDY",
            db="collection",
        )
        for collection in collections:
            opensea_thread = threading.Thread(
                target=self.get_opensea, args=(collection,)
            )
            looksrare_thread = threading.Thread(
                target=self.get_looksrare, args=(collection,)
            )
            x2y2_thread = threading.Thread(target=self.get_x2y2, args=(collection,))
            sudoswap_thread = threading.Thread(
                target=self.get_sudoswap, args=(collection,)
            )
            nftx_thread = threading.Thread(target=self.get_nftx, args=(collection,))
            opensea_thread.start()
            looksrare_thread.start()
            x2y2_thread.start()
            sudoswap_thread.start()
            nftx_thread.start()
            opensea_thread.join()
            x2y2_thread.join()
            looksrare_thread.join()
            sudoswap_thread.join()
            nftx_thread.join()
            self.data = {
                "timestamp": timestamp,
                "opensea": self.opensea_data,
                "looksrare": self.lookrare_data,
                "x2y2": self.x2y2_data,
                "sudoswap": self.sudoswap_data,
                "nftx": self.nftx_data,
            }
            if not test:
                self.update_db(collection["name"])
            else:
                print(self.data)

    def update_db(self, table):
        client_dynamo = boto3.resource("dynamodb")
        table_dynamo = client_dynamo.Table(table)
        try:
            table_dynamo.put_item(Item=self.data)
            for attr, value in self.data.items():
                if attr == "timestamp":
                    continue
                if value["best_offer"]:
                    self.data[attr]["best_offer"] = f"'{value['best_offer']}'"
                else:
                    self.data[attr]["best_offer"] = "null"
            cursor = self.connection.cursor()
            cursor.execute(
                f"INSERT INTO `{table}` (`time_stamp`, `sudoswap`, `opensea`, `looksrare`, `x2y2`, `nftx`) VALUES ('{self.data['timestamp']}', {self.data['sudoswap']['best_offer']}, {self.data['opensea']['best_offer']}, {self.data['looksrare']['best_offer']}, {self.data['x2y2']['best_offer']}, {self.data['nftx']['best_offer']})",
            )
            self.connection.commit()
        except Exception:
            raise

    def get_opensea(self, collection):
        try:
            scraper = cloudscraper.create_scraper()
            html_res = scraper.get(
                f'https://opensea.io/collection/{collection["name"]}'
            ).text
            soup = BeautifulSoup(html_res, "html.parser")
            element = soup.find_all(
                "span",
                class_="sc-1xf18x6-0 sc-1aqfqq9-0 haVRLx cUPmoY styledPhoenixText",
            )[4]
            best_offer = Decimal(element.text)
        except Exception:
            best_offer = None
        url = f'https://api.opensea.io/api/v1/collection/{collection["name"]}/stats'
        headers = {
            "Accept": "application/json",
            "X-API-KEY": "",
        }
        try:
            response = requests.get(url, headers=headers, timeout=5)
            self.opensea_data = {
                "best_offer": best_offer,
                "floor_price": Decimal(
                    "%.3f" % response.json()["stats"]["floor_price"]
                ),
                "total_volume": Decimal(
                    "%.3f" % response.json()["stats"]["total_volume"]
                ),
            }
        except Exception:
            self.opensea_data = {
                "best_offer": best_offer,
                "floor_price": None,
                "total_volume": None,
            }

    def get_looksrare(self, collection):
        url = f'https://api.looksrare.org/api/v1/orders?isOrderAsk=true&collection={collection["collectionAddress"]}&status%5B%5D=VALID&sort=PRICE_ASC'
        headers = {
            "Accept": "application/json",
            "X-API-KEY": "",
        }
        try:
            response = requests.get(url, headers=headers, timeout=5)
            best_offer = Decimal(
                "%.3f" % (float(response.json()["data"][0]["price"]) / 1e18)
            )
        except Exception:
            best_offer = None
        url = f'https://api.looksrare.org/api/v1/collections/stats?address={collection["collectionAddress"]}'
        try:
            response = requests.get(url, headers=headers, timeout=5)
            self.lookrare_data = {
                "best_offer": best_offer,
                "floor_price": Decimal(
                    "%.3f" % (float(response.json()["data"]["floorPrice"]) / 1e18)
                ),
                "total_volume": Decimal(
                    "%.3f" % (float(response.json()["data"]["volumeAll"]) / 1e18)
                ),
            }
        except Exception:
            self.lookrare_data = {
                "best_offer": best_offer,
                "floor_price": None,
                "total_volume": None,
            }

    def get_x2y2(self, collection):
        try:
            scraper = cloudscraper.create_scraper()
            html_res = scraper.get(
                f'https://x2y2.io/collection/{collection["name"]}/items'
            ).text
            soup = BeautifulSoup(html_res, "html.parser")
            element = soup.find_all("div", class_="css-1asyqbr")[2]
            if element.text.endswith("K"):
                total_volume = Decimal(float(element.text.replace("K", "")) * 1000)
            else:
                total_volume = Decimal(element.text)
        except Exception:
            total_volume = None
        url = (
            f'https://api.x2y2.org/v1/contracts/{collection["collectionAddress"]}/stats'
        )
        headers = {
            "Accept": "application/json",
            "X-API-KEY": "",
        }
        try:
            response = requests.get(url, headers=headers, timeout=5)
            params = {
                "contract": collection["collectionAddress"],
                "sort": "price",
            }
            best_offer = Decimal(
                "%.3f"
                % (
                    float(
                        requests.get(
                            "https://api.x2y2.org/v1/offers",
                            headers=headers,
                            params=params,
                            timeout=5,
                        ).json()["data"][-1]["price"]
                    )
                    / 1e18
                )
            )
            self.x2y2_data = {
                "best_offer": best_offer,
                "floor_price": Decimal(
                    "%.3f" % (float(response.json()["data"]["floor_price"]) / 1e18)
                ),
                "total_volume": total_volume,
            }
        except Exception:
            self.x2y2_data = {
                "best_offer": None,
                "floor_price": None,
                "total_volume": None,
            }

    def get_sudoswap(self, collection):
        url = f'https://sudoapi.xyz/v1/analytics/collection/{collection["collectionAddress"]}'
        try:
            total_volume = requests.get(url, timeout=5).json()["volume_all_time"]
            floor_price = Decimal(0)
            best_offer = Decimal(0)
            response = requests.get(
                "https://sudoapi.xyz/v1/collections?sort=offer_tvl&desc=true", timeout=5
            ).json()["collections"]
            for col in response:
                if col["address"] == collection["collectionAddress"]:
                    floor_price = Decimal("%.3f" % (float(col["buy_quote"]) / 1e18))
                    best_offer = Decimal("%.3f" % (float(col["sell_quote"]) / 1e18))
            self.sudoswap_data = {
                "best_offer": best_offer,
                "floor_price": floor_price,
                "total_volume": Decimal("%.3f" % float(total_volume)),
            }
        except Exception:
            self.sudoswap_data = {
                "best_offer": None,
                "floor_price": None,
                "total_volume": None,
            }

    def get_nftx(self, collection):
        url = f'http://api.spicyest.com/nftx_sell_price?address={collection["collectionAddress"]}'
        headers = {"Accept": "*/*", "X-API-Key": ""}
        try:
            response = requests.get(url, headers=headers, timeout=5)
            self.nftx_data = {
                "best_offer": Decimal("%.3f" % (float(response.json()["price"]))),
                "floor_price": None,
                "total_volume": None,
            }
        except Exception:
            self.nftx_data = {
                "best_offer": None,
                "floor_price": None,
                "total_volume": None,
            }


def lambda_handler(event, context):
    COLLECTIONS = [
        {
            "name": "clonex",
            "collectionAddress": "0x49cF6f5d44E70224e2E23fDcdd2C053F30aDA28B",
        },
        {
            "name": "cryptodickbutts-s3",
            "collectionAddress": "0x42069ABFE407C60cf4ae4112bEDEaD391dBa1cdB",
        },
        {
            "name": "thepotatoz",
            "collectionAddress": "0x39ee2c7b3cb80254225884ca001F57118C8f21B6",
        },
        {
            "name": "cryptopunks",
            "collectionAddress": "0xb47e3cd837dDF8e4c57F05d70Ab865de6e193BBB",
        },
        {
            "name": "art-blocks",
            "collectionAddress": "0x059EDD72Cd353dF5106D2B9cC5ab83a52287aC3a",
        },
        {
            "name": "doodles",
            "collectionAddress": "0x8a90CAb2b38dba80c64b7734e58Ee1dB38B8992e",
        },
        {
            "name": "boredapeyachtclub",
            "collectionAddress": "0xBC4CA0EdA7647A8aB7C2061c2E118A18a936f13D",
        },
        {
            "name": "mutant-ape-yacht-club",
            "collectionAddress": "0x60E4d786628Fea6478F785A6d7e704777c86a7c6",
        },
        {
            "name": "world-of-women-nft",
            "collectionAddress": "0xe785E82358879F061BC3dcAC6f0444462D4b5330",
        },
    ]
    UpdateDB(COLLECTIONS)
    return "success"