# /// script
# requires-python = ">=3.10,<3.12"
# dependencies = [
# "requests",
# "matplotlib",
# ]
# ///
# Import necessary modules and types
from typing import Any, Optional, Dict, List # Type hints for better code clarity
from requests import get # HTTP GET requests
import matplotlib.pyplot as plt # Plotting library
from shinkai_local_support import get_home_path # Utility to get user's home directory
# Configuration class for API and value thresholds
class CONFIG:
taptools_api_key: str # API key for TapTools
min_quote_value_ft: str = "1" # Min value for fungible tokens in quote currency
min_quote_value_nft: str = "1" # Min value for NFTs in quote currency
min_quote_value_lp: str = "1" # Min value for liquidity positions in quote currency
# Input class for user-provided parameters
class INPUTS:
address: str # Wallet address to query
quote_currency: str = "USD" # Currency for value quotes (default: USD)
generate_positions_plots: str = "yes" # Option to generate position plots ("yes"/"no")
generate_portfolio_trended_value_graphs: str = "yes" # Option to generate trend graphs ("yes"/"no")
portfolio_trended_value_timeframe: str = "30d" # Timeframe for trend graphs (24h, 7d, 30d, 90d, 180d, 1y, all)
# Output class for storing results
class OUTPUT:
portfolio_positions: Dict[str, Any] # Portfolio data (positions, values)
ada_price_in_quote: float # ADA price in quote currency
error: Optional[str] = None # Error message if request fails
# Main async function to fetch and process portfolio data
async def run(config: CONFIG, inputs: INPUTS) -> OUTPUT:
api_base_url = "https://openapi.taptools.io/api/v1" # Base URL for TapTools API
portfolio_positions_url = f"{api_base_url}/wallet/portfolio/positions?address={inputs.address}" # URL for portfolio positions
token_quote_url = f"{api_base_url}/token/quote?quote={inputs.quote_currency}" # URL for ADA price in quote currency
headers = {
"x-api-key": config.taptools_api_key # API key header for authentication
}
# Fetch ADA price in quote currency
quote_response = get(token_quote_url, headers=headers)
if quote_response.status_code != 200: # Check if request failed
output = OUTPUT()
try:
error_details = quote_response.json() # Parse error details
error_message = error_details.get("message", "An error occurred") # Extract error message
except json.JSONDecodeError: # Handle non-JSON response
error_message = quote_response.text
output.error = f"Error fetching ADA price: {error_message} (Status Code: {quote_response.status_code})" # Set error
return output
ada_price_in_quote = quote_response.json().get("price", 0) # Extract ADA price
ada_price_attr_name = f"ada_price_in_{inputs.quote_currency}" # Dynamic attribute name for ADA price
# Fetch portfolio positions
response = get(portfolio_positions_url, headers=headers)
output = OUTPUT() # Initialize output object
if response.status_code == 200: # Check if request succeeded
data = response.json() # Parse response JSON
# Rename keys for clarity
data["Number_of_Fungible_Tokens"] = data.pop("numFTs", 0) # Number of fungible tokens
data["Number_of_NFT_collections"] = data.pop("numNFTs", 0) # Number of NFT collections
data["Fungible_Tokens_positions"] = data.pop("positionsFt", []) # Fungible token positions
data["NFTs_positions"] = data.pop("positionsNft", []) # NFT positions
data["Liquidity_Provider_Positions"] = data.pop("positionsLp", []) # Liquidity provider positions
# Process fungible token positions
for position in data["Fungible_Tokens_positions"]:
for timeframe in ["24h", "7d", "30d"]: # Process price changes for timeframes
if timeframe in position:
change_value = position[timeframe] # Get change value
sign = "+" if change_value >= 0 else "" # Add sign for positive/negative
position[timeframe] = f"{sign}{round(change_value * 100, 2)} %" # Format as percentage
# Round and rename values
position["Value_in_Ada"] = round(position.pop("adaValue", 0), 2) # Value in ADA
position["Liquid_Value_in_Ada"] = round(position.pop("liquidValue", 0), 2) # Liquid value in ADA
position["Value_in_" + inputs.quote_currency] = round(position["Value_in_Ada"] * ada_price_in_quote, 2) # Value in quote currency
position["Liquid_Value_in_" + inputs.quote_currency] = round(position["Liquid_Value_in_Ada"] * ada_price_in_quote, 2) # Liquid value in quote currency
position["Liquid_Balance_in_Ada"] = position.pop("liquidBalance", 0) # Liquid balance in ADA
position.pop("unit", None) # Remove unused unit field
position.pop("fingerprint", None) # Remove unused fingerprint field
# Process NFT positions
for position in data["NFTs_positions"]:
for timeframe in ["24h", "7d", "30d"]: # Process price changes for timeframes
if timeframe in position:
change_value = position[timeframe] # Get change value
sign = "+" if change_value >= 0 else "" # Add sign for positive/negative
position[timeframe] = f"{sign}{round(change_value * 100, 2)} %" # Format as percentage
# Round and rename NFT values
position["Value_in_Ada"] = round(position.pop("adaValue", 0), 2) # Value in ADA
position["Value_in_" + inputs.quote_currency] = round(position["Value_in_Ada"] * ada_price_in_quote, 2) # Value in quote currency
position["Floor_Price_in_Ada"] = position.pop("floorPrice", 0) # Floor price in ADA
position["Floor_Price_in_" + inputs.quote_currency] = round(position["Floor_Price_in_Ada"] * ada_price_in_quote, 2) # Floor price in quote currency
# Process liquidity provider positions
for position in data["Liquidity_Provider_Positions"]:
position["Value_in_Ada"] = round(position.pop("adaValue", 0), 2) # Value in ADA
position["Value_in_" + inputs.quote_currency] = round(position["Value_in_Ada"] * ada_price_in_quote, 2) # Value in quote currency
# Calculate total portfolio value
total_value_in_ada = data.pop("adaValue", 0) # Total portfolio value in ADA
data["Total_Portfolio_Value_in_Ada"] = round(total_value_in_ada, 2) # Rounded total in ADA
data["Total_Portfolio_Value_in_" + inputs.quote_currency] = round(total_value_in_ada * ada_price_in_quote, 2) # Total in quote currency
setattr(output, ada_price_attr_name, ada_price_in_quote) # Set ADA price in output
output.portfolio_positions = data # Store portfolio data in output
# Generate position plots if requested
if inputs.generate_positions_plots.lower() == "yes":
home_path = await get_home_path() # Get home directory for saving plots
plt.style.use('dark_background') # Set dark theme for plots
# Convert min values to float for comparison
min_value_ft = float(config.min_quote_value_ft) # Min value for fungible tokens
min_value_nft = float(config.min_quote_value_nft) # Min value for NFTs
min_value_lp = float(config.min_quote_value_lp) # Min value for liquidity positions
# Plot 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] # Token names
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] # Token values
sorted_ft = sorted(zip(ft_values, ft_names), reverse=True) # Sort by value descending
plt.figure(figsize=(10, 6)) # Set plot size
plt.bar([name for _, name in sorted_ft], [value for value, _ in sorted_ft]) # Create bar plot
plt.title("Fungible Tokens Value") # Set title
plt.xlabel("Token") # Set x-axis label
plt.ylabel(f"Value in {inputs.quote_currency}") # Set y-axis label
plt.xticks(rotation=45, ha="right") # Rotate x-axis labels
plt.tight_layout() # Adjust layout
plt.savefig(f"{home_path}/fungible_tokens_value.png") # Save plot
plt.close() # Close plot
# Plot NFTs
nft_names = [p.get("name", "") for p in data["NFTs_positions"] if p.get("Value_in_" + inputs.quote_currency, 0) >= min_value_nft] # NFT names
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] # NFT values
sorted_nft = sorted(zip(nft_values, nft_names), reverse=True) # Sort by value descending
plt.figure(figsize=(10, 6)) # Set plot size
plt.bar([name for _, name in sorted_nft], [value for value, _ in sorted_nft]) # Create bar plot
plt.title("NFTs Value") # Set title
plt.xlabel("NFT") # Set x-axis label
plt.ylabel(f"Value in {inputs.quote_currency}") # Set y-axis label
plt.xticks(rotation=45, ha="right") # Rotate x-axis labels
plt.tight_layout() # Adjust layout
plt.savefig(f"{home_path}/nfts_value.png") # Save plot
plt.close() # Close plot
# Plot 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] # Position names
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] # Position values
sorted_lp = sorted(zip(lp_values, lp_names), reverse=True) # Sort by value descending
plt.figure(figsize=(10, 6)) # Set plot size
plt.bar([name for _, name in sorted_lp], [value for value, _ in sorted_lp]) # Create bar plot
plt.title("Liquidity Provider Positions Value") # Set title
plt.xlabel("Position") # Set x-axis label
plt.ylabel(f"Value in {inputs.quote_currency}") # Set y-axis label
plt.xticks(rotation=45, ha="right") # Rotate x-axis labels
plt.tight_layout() # Adjust layout
plt.savefig(f"{home_path}/liquidity_provider_positions_value.png") # Save plot
plt.close() # Close plot
# Generate trend graphs if requested
if inputs.generate_portfolio_trended_value_graphs.lower() == "yes":
trend_base_url = f"{api_base_url}/wallet/value/trended" # URL for portfolio trend data
# Fetch trend data in ADA
trend_response_ada = get(f"{trend_base_url}?address={inputs.address}&timeframe={inputs.portfolio_trended_value_timeframe}"e=ADA", headers=headers)
if trend_response_ada.status_code == 200: # Check if request succeeded
trend_data_ada = trend_response_ada.json() # Parse response JSON
times_ada = [entry["time"] for entry in trend_data_ada] # Extract time points
values_ada = [entry["value"] for entry in trend_data_ada] # Extract values
plt.figure(figsize=(10, 6)) # Set plot size
plt.plot(times_ada, values_ada, marker='o') # Create line plot
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}" # Set dynamic title
plt.title(title_ada) # Set title
plt.ylabel("Value in ADA") # Set y-axis label
plt.xticks([]) # Remove x-axis labels
plt.tight_layout() # Adjust layout
plt.savefig(f"{home_path}/portfolio_trend_ada.png") # Save plot
plt.close() # Close plot
# Fetch trend data in quote currency
trend_response_quote = get(f"{trend_base_url}?address={inputs.address}&timeframe={inputs.portfolio_trended_value_timeframe}"e={inputs.quote_currency}", headers=headers)
if trend_response_quote.status_code == 200: # Check if request succeeded
trend_data_quote = trend_response_quote.json() # Parse response JSON
times_quote = [entry["time"] for entry in trend_data_quote] # Extract time points
values_quote = [entry["value"] for entry in trend_data_quote] # Extract values
plt.figure(figsize=(10, 6)) # Set plot size
plt.plot(times_quote, values_quote, marker='o') # Create line plot
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}" # Set dynamic title
plt.title(title_quote) # Set title
plt.ylabel(f"Value in {inputs.quote_currency}") # Set y-axis label
plt.xticks([]) # Remove x-axis labels
plt.tight_layout() # Adjust layout
plt.savefig(f"{home_path}/portfolio_trend_{inputs.quote_currency.lower()}.png") # Save plot
plt.close() # Close plot
else: # Handle failed portfolio request
try:
error_details = response.json() # Parse error details
error_message = error_details.get("message", "An error occurred") # Extract error message
except json.JSONDecodeError: # Handle non-JSON response
error_message = response.text
output.error = f"Error: {error_message} (Status Code: {response.status_code})" # Set error
return output # Return output object
Below is an example of thinking process you might get from Shinkai code generator AI. Review it and follow along.
Also, in this video a working prototype of the Cardano portfolio tracker’s data retrieval feature is built in just a few minutes by simply providing instructions and API documentation snippets.
For reference, here is an example of fully annotated code for a Cardano wallet portfolio tracker tool, obtained after prompting a few times for such improvements.
