Building a blockchain data-aware AI Agent

Plot ticker IAG, 1d candles, 365 intervals

How many BTC are bridged to Cardano ?

Show me a table of USD stablecoins in Cardano.

Show me the peg status of all BTC related coins on Cardano.

Show me the top 20 NFTs by volume over the last 30d.
And rank today top NFTs by gainers, show top 15.

Give me the links to check the all time top NFT collections by volume, show me their names and some quick stats too.

quote_currency : EUR
address : addrxxxxxxxxxxxxx
Analyse exhaustively the portfolio in great details, and then generate the text for a long and funny description of that wallet, making fun of the holder. For example, look into NFTs with ridiculously low value, into dust tokens, into poorly performing positions. Make sure to name the positions (name of the LP, of the ticker, of the NFT collection).

analyze ticker FLDT, for trading chart use 1d intervals, and 365 intervals,
focus your analysis on the holdings distribution,
and show me the links where I can know more about the project

write a full report on the current Cardano Native Tokens market

class INPUTS:
    timframe_for_NFT_market_stats: Optional[str] = "30d"
    timeframe_for_top_NFTs_by_volume: Optional[str] = "24h"
    timeframe_for_NFT_volume_trended: Optional[str] = "30d"

class INPUTS:
    Ranking_method_for_today_top_NFTs: Optional[str] = "volume"

async def run(config: CONFIG, inputs: INPUTS) -> OUTPUT:
    api_base_url = "https://openapi.taptools.io/api/v1"nft_market_stats_url = f"{api_base_url}/nft/market/stats/extended?timeframe={inputs.timframe_for_NFT_market_stats}"
    top_nfts_url = f"{api_base_url}/nft/top/timeframe?ranking={inputs.Ranking_method_for_today_top_NFTs}&items={inputs.Number_of_today_top_NFTs_to_retrieve}"

class OUTPUT:
    portfolio_positions: Dict[str, Any]

    # Fetch portfolio positions
    response = get(portfolio_positions_url, headers=headers)
    
    output = OUTPUT()
    
    if response.status_code == 200:
        data = response.json()
        
        # Rename keys
        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", [])

        # Modify percentages and calculate quote currency values
        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)} %"

def format_price(price: float) -> str:
    """Format price based on its value."""
    if price >= 1:
        return f"{price:.2f}"
    else:
        price_str = f"{price:.10f}".rstrip('0')
        decimal_idx = price_str.index('.')
        for i in range(decimal_idx + 1, len(price_str)):
            if price_str[i] != '0':
                end_idx = min(i + 3, len(price_str))
                return price_str[:end_idx]
        return price_str

    token_asset_data.update({
        f"{inputs.ticker}price_in_{config.quote_currency}": f"{format_price(ticker_price_in_quote)} {config.quote_currency}" if ticker_price_in_quote else None,
        f"{inputs.ticker}fully_dilluted_valuation_in_{config.quote_currency}": f"{round(ticker_fdv_in_quote)} {config.quote_currency}" if ticker_fdv_in_quote else None,
        f"{inputs.ticker}market_cap_in_{config.quote_currency}": f"{round(ticker_mcap_in_quote)} {config.quote_currency}
        ...

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

    # Podcast Script Generation
    if config.generate_podcast_script.lower() == "yes":
        try:
            podcast_input = {
                "format": "text",
                "prompt": (
                    "You are the writer of the podcast Tappy Chats, which is about detailed and insightful Cardano blockchain data."
                    "Using all of the information from the output token_data below, generate a podcast script according to the instructions below."
                    "Make sure to talk about every section of token_data."
                    "When analyzing value, only use Ada or USD or Euro value, not the value in the amount of the token."
                    "<information from output token_data>"
                    f"{token_data}"
                    "</information from output token_data>"
                    "<instructions>"
                    f"{config.podcast_script_instructions}"
                    "</instructions>"
                    "<format instructions>"
                    "Handle numbers gracefully. If you throw too many numbers at the listeners you can lose them. So pick relevant numbers."
                    "Make sure to round numbers to make them clearer to hear and understand. For tiny numbers with many zeros below zero, you can use 'less than a cent', or 'much less than a cent'."
                    "The text must flow. Do not use any formatting, but just plain text. The text should just be one flowing paragraph."
                    "</format instructions>"
                )
            }
            podcast_script = (await shinkai_llm_prompt_processor(podcast_input)).get("message")
        except Exception as err:
            error_messages.append(f"Podcast script generation failed: {err}")
    else:
        status["podcast_script"] = "Skipped (set to 'no')"

    # Audio Generation
    if config.generate_audio.lower() == "yes":
        if podcast_script:
            try:
                home_path = await get_home_path()
                audio_file_name = f"tappy_chats_{inputs.ticker}.mp3"
                # Call the integrated eleven_labs_text_to_speech function
                audio_generation_result = await eleven_labs_text_to_speech(
                    {"text": podcast_script, "fileName": audio_file_name},
                    {"ELEVENLABS_API_KEY": config.ELEVENLABS_API_KEY, "voice": config.voice}  # Pass config
                )
                audio_file_path = audio_generation_result['audio_file'] # Use the returned path
            except Exception as err:
                error_messages.append(f"Audio generation failed: {err}")
        else:
            error_messages.append("Cannot generate audio: podcast script missing.")
    else:
        status["audio_generation"] = "Skipped (set to 'no')"

class CONFIG:
    show_trading_chart: str = "no"  # "yes" or "no"
    create_pdf_report: str = "no"   # "yes" or "no"
    generate_llm_report: str = "no" # "yes" or "no"
    generate_podcast_script: str = "no" # "yes" or "no"
    analyze_holder_distribution: str = "no" # "yes" or "no"
    generate_audio: str = "no" # "yes" or "no"
    show_llm_report_content: str = "no" # "yes" or "no"

    # Optional Features
    # Holder Distribution Analysis
    if config.analyze_holder_distribution.lower() == "yes":
        try:
            holder_analysis_data, holder_graph_path = await analyze_holders(unit, inputs.ticker, taptools_headers, endpoints, ticker_price_in_ada, ticker_price_in_quote, config.quote_currency)
            token_data[f"{inputs.ticker}_holder_distribution"] = holder_analysis_data
        except Exception as e:
            error_messages.append(f"Holder distribution analysis failed: {e}")
    else:
        status["holder_distribution_analysis"] = "Skipped (set to 'no')"

  "audio_file_path": null,
  "chart_file_path": "C:\\Users\\gille\\AppData\\Roaming\\com.shinkai.desktop\\node_storage\\tools_storage\\app-id-1741695406055\\home\\HOSKY_chart.png",
  "error_message": null,
  "holder_graph_path": null,
  "llm_report": null,
  "pdf_file_path": null,
  "podcast_script": null,
  "status": {
    "audio_generation": "Skipped (set to 'no')",
    "holder_distribution_analysis": "Skipped (set to 'no')",
    "llm_report": "Skipped (set to 'no')",
    "pdf_report": "Skipped (set to 'no')",
    "podcast_script": "Skipped (set to 'no')"
  },

def validate_config(config: CONFIG) -> List[str]:
    """
    Validate configuration flags and dependencies.
    Returns a list of error messages if any issues are found.
    """
    errors = []
    flags = [
        "show_trading_chart", "create_pdf_report", "generate_llm_report",
        "generate_podcast_script", "generate_audio", "analyze_holder_distribution"
    ]
    for flag in flags:
        value = getattr(config, flag)
        if value.lower() not in ["yes", "no"]:
            errors.append(f"Invalid config value for {flag}: {value}. Must be 'yes' or 'no'.")

    if config.generate_audio.lower() == "yes" and config.generate_podcast_script.lower() != "yes":
        errors.append("Cannot generate audio without generating podcast script. Set generate_podcast_script to 'yes'.")
    return errors

    # Validate configuration and fail fast if there are issues
    config_errors = validate_config(config)
    if config_errors:
        return OUTPUT(error_message="; ".join(config_errors))

    return {
        f"{ticker}_onchain_address_concentration_top_10": f"{concentration_top_10:.2f} % of the {ticker} token total supply is held by the top 10 on-chain addresses (WARNING: this might include special wallets like project wallets of non-circulating supply, liquidity pools on DEXs, etc.).",
        f"{ticker}_small_onchain_address_ratio": f"{small_holder_ratio:.2f} % of the on-chain addresses holding {ticker} hold less than {CONFIG.small_holder_ratio_threshold}% of the total supply. This metric shows how much of the community consists of small retail investors versus big players, reflecting grassroots support. (WARNING: this metric could be affected by addresses with residual amounts.)",
        ...

    # Fetch 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

class OUTPUT:
    error_message: Optional[str]
    status: Optional[Dict[str, str]] 

    # Initialize error messages and status tracking
    error_messages = []
    status = {}

        except Exception as e:
            error_messages.append(f"Holder distribution analysis failed: {e}")
    else:
        status["holder_distribution_analysis"] = "Skipped (set to 'no')"

    return OUTPUT(
        ...
        error_message="; ".join(error_messages) if error_messages else None,
        status=status if status else None
    )