​
Introduction

In this tutorial, we will build a cryptocurrency address portfolio tracker, as a Shinkai Tool. This tool can be a core component of a crypto AI Agent able to talk about up-to-date blockchain data. The tool presented here interacts with the TapTools API to analyze and visualize portfolio data for a specified Cardano wallet address. It retrieves positions for fungible tokens, NFTs, and liquidity provider positions, calculates their values in Ada and in a specified quote currency, and can generate visualizations based on user-defined parameters.

You can find this tool in the Shinkai AI Store. Its name is : Taptools Cardano Portfolio Tracker.

While we use a Cardano example, the same tool structure can be applied to any blockchain for which an API provides such data.

​
Prerequisites

Before you begin, ensure you have :

  • the lastest version of the Shinkai Desktop App installed
  • the App opened at the tool creation UI
  • a Taptools API key

​
Step 1 : Import components

We import ‘get’ to make API calls. We import ‘matplotlib’ to create graphs. We import ‘get_home_path’ to save some files.

from typing import Any, Optional, Dict, List
from requests import get
import matplotlib.pyplot as plt
from shinkai_local_support import get_home_path

​
Step 2: Define Configuration Class

We start by defining a configuration class to hold our settings :

  • the TapTools API key
  • some minimum positions value to filter out small ones from graphs
class CONFIG:
    taptools_api_key: str
    min_quote_value_ft: str = "1"  # Minimum quote currency value for fungible tokens
    min_quote_value_nft: str = "1"  # Minimum quote currency value for NFTs
    min_quote_value_lp: str = "1"  # Minimum quote currency value for liquidity positions

​
Step 3: Define Input Class

Next, we define an input class to capture user inputs :

  • the address to look at
  • the quote currency to use
  • some flags to generate or not some plots
  • the timeframe to use for the chart of the porfolio value over time
class INPUTS:
    address: str
    quote_currency: str = "USD"
    generate_positions_plots: str = "yes"  # "yes" or "no"
    generate_portfolio_trended_value_graphs: str = "yes"  # "yes" or "no"
    portfolio_trended_value_timeframe: str = "30d"  # Options are 24h, 7d, 30d, 90d, 180d, 1y, all

These flags will be used for optional outputs: the user can choose whether to generate plots and graphs by setting the generate_positions_plots and generate_portfolio_trended_value_graphs attributes.

​
Step 4: Define Output Class

We create an output class to structure the data returned by our tool.

class OUTPUT:
    portfolio_positions: Dict[str, Any]
    ada_price_in_quote: float
    error: Optional[str] = None

The error field is included only if an error occurs during API calls.

​
Step 5: Start of the Main Function to Run the Tool

Now we implement the run function, which encapsulates the main logic of our tool.

async def run(config: CONFIG, inputs: INPUTS) -> OUTPUT:
    api_base_url = "https://openapi.taptools.io/api/v1"
    portfolio_positions_url = f"{api_base_url}/wallet/portfolio/positions?address={inputs.address}"
    token_quote_url = f"{api_base_url}/token/quote?quote={inputs.quote_currency}"
    
    headers = {
        "x-api-key": config.taptools_api_key
    }

Multiple API Calls: this function makes 2 API calls to fetch data, one for portfolio positions and one to get the quote price of Ada, so that we can convert all positions value to a quote currency.

​
Step 6: Fetching ADA Price

The first API call retrieves the price of ADA in the specified quote currency.

    quote_response = get(token_quote_url, headers=headers)
    if quote_response.status_code != 200:
        output = OUTPUT()
        try:
            error_details = quote_response.json()
            error_message = error_details.get("message", "An error occurred")
        except json.JSONDecodeError:
            error_message = quote_response.text
        output.error = f"Error fetching ADA price: {error_message} (Status Code: {quote_response.status_code})"
        return output

Error Handling: If the API call fails, the tool captures the error message and returns it in the output.

And we store the Ada price.

    ada_price_in_quote = quote_response.json().get("price", 0)
    ada_price_attr_name = f"ada_price_in_{inputs.quote_currency}"

​
Step 7: Fetching Portfolio Positions

Next, we fetch the portfolio positions using another API call, and store it in ‘data’.

    response = get(portfolio_positions_url, headers=headers)
    
    output = OUTPUT()
    
    if response.status_code == 200:
        data = response.json()

​
Step 8: Improving the output

​
8.1: Renaming Keys

Once we have the data, an important step is to give explicit names to keys for clarity. Well formatted outputs will help the LLM interpreting the data to better answer the users’ prompts.

        data["Number_of_Fungible_Tokens"] = data.pop("numFTs", 0)
        data["Number_of_NFT_collections"] = data.pop("numNFTs", 0)
        data["Fungible_Tokens_positions"] = data.pop("positionsFt", [])
        data["NFTs_positions"] = data.pop("positionsNft", [])
        data["Liquidity_Provider_Positions"] = data.pop("positionsLp", [])

​
8.2: Formatting % changes for fungible tokens positions

We improve the formatting of percentage changes by multiplying by 100, rounding, and adding + sign if positive.

        # Modify percentages
        for position in data["Fungible_Tokens_positions"]:
            for timeframe in ["24h", "7d", "30d"]:
                if timeframe in position:
                    change_value = position[timeframe]
                    sign = "+" if change_value >= 0 else ""
                    position[timeframe] = f"{sign}{round(change_value * 100, 2)} %"

​
8.3: Rouding values and calculations in quote currency

For better formatting, we round some values. We also calculate some values in quote currency. Note the dynamic naming of the output key : using inputs.quote_currency allows to set the name according to the selected quote currency in inputs.

            # Round and rename values
            position["Value_in_Ada"] = round(position.pop("adaValue", 0), 2)
            position["Liquid_Value_in_Ada"] = round(position.pop("liquidValue", 0), 2)
            position["Value_in_" + inputs.quote_currency] = round(position["Value_in_Ada"] * ada_price_in_quote, 2)
            position["Liquid_Value_in_" + inputs.quote_currency] = round(position["Liquid_Value_in_Ada"] * ada_price_in_quote, 2)
            position["Liquid_Balance_in_Ada"] = position.pop("liquidBalance", 0)

​
8.4: Clearing the output of unneccessary elements

We remove from the output some data coming from the API call but that we’re not interested in (unit and fingerprint of the token).

            position.pop("unit", None)
            position.pop("fingerprint", None)

We repeat the above for NFTs positions :

        for position in data["NFTs_positions"]:
            for timeframe in ["24h", "7d", "30d"]:
                if timeframe in position:
                    change_value = position[timeframe]
                    sign = "+" if change_value >= 0 else ""
                    position[timeframe] = f"{sign}{round(change_value * 100, 2)} %"
            
            # Round and rename NFT-specific values
            position["Value_in_Ada"] = round(position.pop("adaValue", 0), 2)
            position["Value_in_" + inputs.quote_currency] = round(position["Value_in_Ada"] * ada_price_in_quote, 2)
            position["Floor_Price_in_Ada"] = position.pop("floorPrice", 0)
            position["Floor_Price_in_" + inputs.quote_currency] = round(position["Floor_Price_in_Ada"] * ada_price_in_quote, 2)

And for liquidity provider positions, we also round and calculate values in quote currency.

        for position in data["Liquidity_Provider_Positions"]:
            position["Value_in_Ada"] = round(position.pop("adaValue", 0), 2)
            position["Value_in_" + inputs.quote_currency] = round(position["Value_in_Ada"] * ada_price_in_quote, 2)

We also improve the output part about the total value. We calculate it in quote currency, and improve the naming plus round.

        # Correct the key used to calculate total portfolio value
        total_value_in_ada = data.pop("adaValue", 0)
        data["Total_Portfolio_Value_in_Ada"] = round(total_value_in_ada, 2)
        data["Total_Portfolio_Value_in_" + inputs.quote_currency] = round(total_value_in_ada * ada_price_in_quote, 2)
        
        setattr(output, ada_price_attr_name, ada_price_in_quote)
        output.portfolio_positions = data

​
Step 9: Generating Plots for the different positions

If requested, the tool generates plots for fungible tokens, NFTs, and liquidity provider positions.

We make the plot generation conditional.

        if inputs.generate_positions_plots.lower() == "yes":
            home_path = await get_home_path()

We set a dark background for all plots.

            plt.style.use('dark_background')

We convert our config parameters to float

# Convert min values to float
            min_value_ft = float(config.min_quote_value_ft)
            min_value_nft = float(config.min_quote_value_nft)
            min_value_lp = float(config.min_quote_value_lp)

We build 3 barplot graphs (NFTs, tokens, liquidity provider positions).

# Plot for Fungible Tokens
            ft_names = [p.get("ticker", "") for p in data["Fungible_Tokens_positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_ft]
            ft_values = [p.get("Value_in_" + inputs.quote_currency, 0) for p in data["Fungible_Tokens_positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_ft]
            sorted_ft = sorted(zip(ft_values, ft_names), reverse=True)
            plt.figure(figsize=(10, 6))
            plt.bar([name for _, name in sorted_ft], [value for value, _ in sorted_ft])
            plt.title("Fungible Tokens Value")
            plt.xlabel("Token")
            plt.ylabel(f"Value in {inputs.quote_currency}")
            plt.xticks(rotation=45, ha="right")
            plt.tight_layout()
            plt.savefig(f"{home_path}/fungible_tokens_value.png")
            plt.close()
            
            # Plot for NFTs using the "name" field
            nft_names = [p.get("name", "") for p in data["NFTs_positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_nft]
            nft_values = [p.get("Value_in_" + inputs.quote_currency, 0) for p in data["NFTs_positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_nft]
            sorted_nft = sorted(zip(nft_values, nft_names), reverse=True)
            plt.figure(figsize=(10, 6))
            plt.bar([name for _, name in sorted_nft], [value for value, _ in sorted_nft])
            plt.title("NFTs Value")
            plt.xlabel("NFT")
            plt.ylabel(f"Value in {inputs.quote_currency}")
            plt.xticks(rotation=45, ha="right")
            plt.tight_layout()
            plt.savefig(f"{home_path}/nfts_value.png")
            plt.close()
            
            # Plot for Liquidity Provider Positions
            lp_names = [p.get("ticker", "") for p in data["Liquidity_Provider_Positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_lp]
            lp_values = [p.get("Value_in_" + inputs.quote_currency, 0) for p in data["Liquidity_Provider_Positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_lp]
            sorted_lp = sorted(zip(lp_values, lp_names), reverse=True)
            plt.figure(figsize=(10, 6))
            plt.bar([name for _, name in sorted_lp], [value for value, _ in sorted_lp])
            plt.title("Liquidity Provider Positions Value")
            plt.xlabel("Position")
            plt.ylabel(f"Value in {inputs.quote_currency}")
            plt.xticks(rotation=45, ha="right")
            plt.tight_layout()
            plt.savefig(f"{home_path}/liquidity_provider_positions_value.png")
            plt.close()

Note these 2 lines at the beginning of each graph, which allows to filter out some of the positions if they are below the minimum thresholds set in configurations.

            ft_names = [p.get("ticker", "") for p in data["Fungible_Tokens_positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_ft]
            ft_values = [p.get("Value_in_" + inputs.quote_currency, 0) for p in data["Fungible_Tokens_positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_ft]

​
Step 10: Generating Trend Graphs

Finally, if requested, we can generate trend graphs based on the portfolio value over time. We call the corresponding endpoint twice, once for the value in Ada and once for the value in quote currency. We generate a graph for each.

        # Generate trend graphs if requested
        if inputs.generate_portfolio_trended_value_graphs.lower() == "yes":
            trend_base_url = f"{api_base_url}/wallet/value/trended"
            
            # Fetch data in ADA
            trend_response_ada = get(f"{trend_base_url}?address={inputs.address}&timeframe={inputs.portfolio_trended_value_timeframe}&quote=ADA", headers=headers)
            if trend_response_ada.status_code == 200:
                trend_data_ada = trend_response_ada.json()
                times_ada = [entry["time"] for entry in trend_data_ada]
                values_ada = [entry["value"] for entry in trend_data_ada]
                
                plt.figure(figsize=(10, 6))
                plt.plot(times_ada, values_ada, marker='o')
                title_ada = "Portfolio Value Trend in ADA (all time)" if inputs.portfolio_trended_value_timeframe == "all" else f"Portfolio Value Trend in ADA over {inputs.portfolio_trended_value_timeframe}"
                plt.title(title_ada)
                plt.ylabel("Value in ADA")
                plt.xticks([])  # Remove x-axis labels
                plt.tight_layout()
                plt.savefig(f"{home_path}/portfolio_trend_ada.png")
                plt.close()
            
            # Fetch data in quote currency
            trend_response_quote = get(f"{trend_base_url}?address={inputs.address}&timeframe={inputs.portfolio_trended_value_timeframe}&quote={inputs.quote_currency}", headers=headers)
            if trend_response_quote.status_code == 200:
                trend_data_quote = trend_response_quote.json()
                times_quote = [entry["time"] for entry in trend_data_quote]
                values_quote = [entry["value"] for entry in trend_data_quote]
                
                plt.figure(figsize=(10, 6))
                plt.plot(times_quote, values_quote, marker='o')
                title_quote = f"Portfolio Value Trend in {inputs.quote_currency} (all time)" if inputs.portfolio_trended_value_timeframe == "all" else f"Portfolio Value Trend in {inputs.quote_currency} over {inputs.portfolio_trended_value_timeframe}"
                plt.title(title_quote)
                plt.ylabel(f"Value in {inputs.quote_currency}")
                plt.xticks([])  # Remove x-axis labels
                plt.tight_layout()
                plt.savefig(f"{home_path}/portfolio_trend_{inputs.quote_currency.lower()}.png")
                plt.close()

The rest of the code logs errors and returns the output.

    else:
        try:
            error_details = response.json()
            error_message = error_details.get("message", "An error occurred")
        except json.JSONDecodeError:
            error_message = response.text
        
        output.error = f"Error: {error_message} (Status Code: {response.status_code})"
    
    return output

Feel free to extend this tool further by adding additional features or integrating with other APIs!

​
Screenshots

Below are 4 screenshots of our portofiolio tracker tool being used in Shinkai.

Was this page helpful?

Building an AI Email AssistantOverview