Skip to content

NFT Portfolio Tracker Bot on Discord with Napkin

Build a bot to track the value of your NFT portfolio

Post written by Nick Sypteras

Nick Sypteras

@NickSypteras
NFT Portfolio Tracker Bot on Discord with Napkin

I was browsing Web3 Twitter last week and came across a cool project just launched by @Traf: value.app. It's a site that estimates the current total value of all NFTs belonging to any Ethereum wallet.

I loved the idea so much that I decided to build my own “bot version” on Napkin. This "NFT Portfolio Tracker Bot" sends me updates on my/others’ NFT portfolio performance each day on Discord. Here's how it looks.

nft tracker bot message

And no, that’s not the value of my NFT collection, sadly. It’s Serena Williams’, who I recently found out is a fellow JPEG collector.

degen serena

But back to the tracker bot. In this tutorial, we’ll go through how to make one yourself for free using Napkin! You can follow the tutorial below or simply fork the function here.

Set Up The Discord Bot

First, lets set up a bot on Discord to send us messages from our Napkin function. Go to your Discord server settings, then

Integrations -> Webhooks
, then click
New Webhook
.

new webhook button

Give your Webhook Integration a name and select what channel it will post in. Don’t forget to give your bot a beautiful face!

new webhook button

Save your changes, copy the Webhook URL and then head over to napkin.io/dashboard/new to create a new Python function.

new function

Setting Up Your Napkin Environment

First off, let’s add our webhook URL as an environment variable for our function. Go to the Other tab and add a new Environment Variable called

WEBHOOK_URL
. Let’s also add an environment variable for the wallet address we want to track (Serena Williams is 0x0864224f3cc570ab909ebf619f7583ef4a50b826 🎾).

environment variables

As for modules, we’ll only need one: requests, to make HTTP requests to the OpenSea API*. You can easily add it (and any other modules you want) in the Modules tab of the Napkin editor.

modules

*Quick note on the OpenSea API: You can use it without an API key, but for production use cases you should request a free one.

What’s A NFT Actually Worth?

Assessing the value of a NFT is a highly... ahem... debated topic. There’s no purely objective way to do it, so feel free to use your own logic. The simple logic my function uses is to assign each NFT the value of either the NFT collection’s Floor Price or the last amount paid for the NFT if the collection floor isn’t known. In pseudo code, we can define this as:

nft_value = collection_floor_price(nft)


if nft_value == null:
    nft_value = last_purchase_price(nft)

The OpenSea API

Given our value definition above, we’ll need 2 pieces of data from the OpenSea API: the floor of a given NFT collection, and the price that the owner paid for a NFT. We can get these using the following endpoints:

  • Floor (
    https://api.opensea.io/api/v1/collection/{collection_slug}/stats
    )
  • Last Price Paid (
    https://api.opensea.io/api/v1/assets
    )

Our API requests will then look like this

def opensea_api(path, **kwargs) -> dict:
   res = requests.get(f"https://api.opensea.io/api/v1/{path}", **kwargs)


   return res.json()


def get_nfts(address) -> dict:
        # if you're address has > 50 NFTs, you'll also need pagination logic here
   data = opensea_api(f"assets?order_direction=desc&limit=50&owner={address}")


   nft_holdings = {}


   for asset in data['assets']:
       slug = asset['collection']['slug']
       last_sale = asset.get('last_sale') or {}
       str_price = last_sale.get('total_price', '0')
       int_price = int(str_price) / 1000000000000000000


       if slug in nft_holdings:
           nft_holdings[slug].append(int_price)
       else:
           nft_holdings[slug] = [int_price]


                return nft_holdings


def update_with_floor_prices(nft_holdings) -> None:


   slugs = [s for s in nft_holdings.keys()]
   for slug in slugs:
       floor_price = get_floor(slug)


       if floor_price > 0.0:
           nft_holdings[slug] = [floor_price for h in nft_holdings[slug]]


def get_floor(slug) -> float:
   data = opensea_api(f"collection/{slug}/stats")


   # if opensea can't find it, then we can't guess the value
   if not data.get('success', True):
       return 0


   floor_price = data['stats']['floor_price'] or 0


   return floor_price

Tracking Value Over Time

We’ll be wanting to compare the portfolio’s current value to it’s historical value. We can cache and retrieve the historical value by saving data to the Napkin key-value store.

from datetime import datetime
from napkin import store


TIMESTAMP_FMT = "%Y-%m-%d-%H:%M"
store_key = f"{ADDRESS}_value"


def save(current, **kwargs):
    time_now = datetime.now().strftime(TIMESTAMP_FMT)
    store.put(store_key, {
        'previous': current,
        'last_checked': time_now,
        'initial_checked': kwargs.get('initial_checked', time_now),
        'initial': kwargs.get('initial', current)
    })


def previous_value_eth(address):
    data = store.get(store_key)['data'] or {}


    return {
        'last_checked': data['last_checked'],
        'initial_checked': data['initial_checked'],
        'previous': data['previous'],
        'initial':data['initial']
    }

Every time our function runs, we’ll see what the previous value was, compare it to the current value, and then save the current value in the database to be compared with next time.

Sending a Message on Discord

Here, we compare the previous value to the current value, and then POST the results to our Discord webhook URL. To make things look nice, I also included some extra formatting in the request payload. Here’s a nice cheatsheet I found if you want to change how the message looks.

import os
webhook_url = os.getenv('WEBHOOK_URL')


def send_message(current_value, previous=None, initial=None, initial_checked=None, last_checked=None):


   gain_from_prev = current_value - previous
   gain_from_init = current_value - initial


   trend_from_prev = 'upwards' if gain_from_prev >= 0 else 'downwards'
   trend_from_init = 'upwards' if gain_from_init >= 0 else 'downwards'


   pct_from_prev = round(gain_from_prev / previous, 2)
   pct_from_init = round(gain_from_init / initial, 2)


   if trend_from_init == trend_from_prev == 'upwards':
       color = 2206495
   elif trend_from_init == trend_from_prev == 'downwards':
       color = 15547966
   else:
       color = 7237230


   def fmt(timestamp):
       now = datetime.now()
       dt = datetime.strptime(timestamp, TIMESTAMP_FMT)


       delta = now - dt


       if delta.days > 0:
           return dt.strftime("%b %d, %Y")


       if delta.seconds > 3600:
           return f"{delta.seconds // 3600} hours ago"


       if delta.seconds > 60:
           return f"{delta.seconds // 60} minutes ago"


       return "Just now"




   requests.post(webhook_url, json={
       'content': "gm",
       'embeds': [
           {
               'title': f"Total Value: {round(current_value, 2)} Ξ :sparkles:",
               'description': f"Last checked {fmt(last_checked)}. Started tracking {fmt(initial_checked)}.",
               'color': color,
               'fields': [
                   {
                       'name': f'Since Last Checked',
                       'value': f"{pct_from_prev*100}% :chart_with_{trend_from_prev}_trend:",
                       'inline': False
                   },
                   {
                       'name': f'All Time',
                       'value': f"{pct_from_init*100}% :chart_with_{trend_from_init}_trend:",
                       'inline': False
                   }
               ],
               'footer': {
                   'text': f"https://opensea.io/assets/{ADDRESS}"
               }
           },
       ],


   })

And that’s the gist of it! Here’s what all the code looks like together (you can also fork it here).

import os
import requests
from napkin import store, response
from datetime import datetime


# serena williams wallet
# 0x0864224f3cc570ab909ebf619f7583ef4a50b826




ADDRESS = os.getenv('WALLET_ADDRESS')
WEBHOOK_URL = os.getenv('WEBHOOK_URL')
TIMESTAMP_FMT = "%Y-%m-%d-%H:%M"


store_key = f"{ADDRESS}_value"


def opensea_api(path, **kwargs) -> dict:
   res = requests.get(f"https://api.opensea.io/api/v1/{path}", **kwargs)


   return res.json()


def get_nfts(address):
   data = opensea_api(f"assets?order_direction=desc&limit=50&owner={address}")


   nft_holdings = {}


   for asset in data['assets']:
       slug = asset['collection']['slug']
       last_sale = asset.get('last_sale') or {}
       str_price = last_sale.get('total_price', '0')
       int_price = int(str_price) / 1000000000000000000


       if slug in nft_holdings:
           nft_holdings[slug].append(int_price)
       else:
           nft_holdings[slug] = [int_price]


   return nft_holdings


def update_with_floor_prices(nft_holdings):


   slugs = [s for s in nft_holdings.keys()]
   for slug in slugs:
       floor_price = get_floor(slug)


       if floor_price > 0.0:
           nft_holdings[slug] = [floor_price for h in nft_holdings[slug]]


def get_floor(slug) -> float:
   data = opensea_api(f"collection/{slug}/stats")


   # if opensea can't find it, then we can't guess the value
   if not data.get('success', True):
       return 0


   floor_price = data['stats']['floor_price'] or 0


   return floor_price




def portfolio_value_eth(address):
   nft_holdings = get_nfts(address)
   total_value_eth = 0


   # for each nft, get the floor price if available
   update_with_floor_prices(nft_holdings)


   for slug, holdings in nft_holdings.items():
       total_value_eth += sum(holdings)


   return total_value_eth




def previous_value_eth(address):
   data = store.get(store_key)['data']


   if data is None:
       return {}


   return {
       'last_checked': data['last_checked'],
       'initial_checked': data['initial_checked'],
       'previous': data['previous'],
       'initial':data['initial']
   }


def save(current, **kwargs):
   time_now = datetime.now().strftime(TIMESTAMP_FMT)
   store.put(store_key, {
       'previous': current,
       'last_checked': time_now,
       'initial_checked': kwargs.get('initial_checked', time_now),
       'initial': kwargs.get('initial', current)
   })


def send_message(current_value, previous=None, initial=None, initial_checked=None, last_checked=None):


   gain_from_prev = current_value - previous
   gain_from_init = current_value - initial


   trend_from_prev = 'upwards' if gain_from_prev >= 0 else 'downwards'
   trend_from_init = 'upwards' if gain_from_init >= 0 else 'downwards'


   pct_from_prev = round(gain_from_prev / previous, 2)
   pct_from_init = round(gain_from_init / initial, 2)


   if trend_from_init == trend_from_prev == 'upwards':
       color = 2206495
   elif trend_from_init == trend_from_prev == 'downwards':
       color = 15547966
   else:
       color = 7237230


   def fmt(timestamp):
       now = datetime.now()
       dt = datetime.strptime(timestamp, TIMESTAMP_FMT)


       delta = now - dt


       if delta.days > 0:
           return dt.strftime("%b %d, %Y")


       if delta.seconds > 3600:
           return f"{delta.seconds // 3600} hours ago"


       if delta.seconds > 60:
           return f"{delta.seconds // 60} minutes ago"


       return "Just now"




   payload = {
       'content': "gm",
       'embeds': [
           {
               'title': f"Total Value: {round(current_value, 2)} Ξ :sparkles:",
               'description': f"Last checked {fmt(last_checked)}. Started tracking {fmt(initial_checked)}.",
               'color': color,
               'fields': [
                   {
                       'name': 'Since Last Checked',
                       'value': f"{pct_from_prev*100}% :chart_with_{trend_from_prev}_trend:",
                       'inline': False
                   },
                   {
                       'name': 'All Time',
                       'value': f"{pct_from_init*100}% :chart_with_{trend_from_init}_trend:",
                       'inline': False
                   }
               ],
               'footer': {
                   'text': f"https://opensea.io/assets/{ADDRESS}"
               }
           },
       ],


   }
   print(payload)
   res = requests.post(WEBHOOK_URL, json=payload)


   print(res.content)




# main
current_value = portfolio_value_eth(ADDRESS)
prev_data = previous_value_eth(ADDRESS)
if len(prev_data) > 0:
   send_message(current_value, **prev_data)
save(current_value, **prev_data)

Make sure to run the function once manually to populate the key-value store with initial values - that way when the schedule runs there will be something to compare to.

Running On A Schedule

Finally let’s schedule our function so it automatically gives us updates. Napkin lets you schedule functions to run every hour, every minute - but for our sanity, let’s stick to once a day 🙂

schedule

And that’s it! You now have a personal assistant tracking your NFT portfolio. There’s a lot of directions you could go from here. If you want your bot to respond to slash commands, check out my previous post, Build a Discord Bot to Track the ISS, for a guide on how to set that up.

That’s it! Have fun, be safe. gn 🌜

Join our Serverless Community

Become part of our growing community of serverless developers on Discord. Got questions about serverless, bots or backend development in general? We got you covered!

Write and deploy cloud functions from your browser.

Napkin is the quickest way to build your backend

Create an AccountRead Docs