The error you're encountering is due to a httpx.ReadTimeout exception, which is raised when the HTTP request to the Solana RPC endpoint takes longer than the allowed timeout period. This exception is then re-raised as a solana.exceptions.SolanaRpcException by the Solana client library.

Why This Happens:

• Network Latency: The Solana RPC endpoint (https://api.mainnet-beta.solana.com) might be experiencing high load or network latency, causing requests to take longer than expected.
• Default Timeout: The default timeout for the HTTP requests might be too short for your use case.
• Transient Issues: Temporary network issues can also cause timeouts.

How to Fix It:

1. Increase the Timeout: You can specify a longer timeout when initializing the AsyncClient from the solana-py library. This will give the server more time to respond before a timeout exception is raised.

2. Implement Retry Logic: Wrap your request in a try-except block and implement retries with exponential backoff in case of transient failures.

Here's how you can modify your code:

python
import asyncio
import logging
import os
from solana.rpc.async_api import AsyncClient
from solana.publickey import PublicKey

async def get_latest_sol_transactions(address: str) -> list[str]:
    """
    Получить последние n транзакций для указанного адреса в сети Solana.

    :param address: Строка с адресом кошелька Solana.
    :return: Список подписей последних транзакций.
    """
    count = int(os.getenv('SOLANA_WALLETS_NUM', '10'))  # Default to 10 if not set
    timeout = 30  # Set the timeout to 30 seconds
    max_retries = 3  # Maximum number of retries
    retry_delay = 5  # Initial delay between retries

    for attempt in range(max_retries):
        try:
            async with AsyncClient('https://api.mainnet-beta.solana.com', timeout=timeout) as client:
                # Запрос подписей транзакций для указанного адреса
                res = await client.get_signatures_for_address(PublicKey(address))
                if res.value is not None:
                    # Возвращаем только первые `count` транзакций
                    return [str(tx.signature) for tx in res.value[:count]]
                else:
                    logging.info("No transactions found for the given address.")
                    return []
        except Exception as e:
            logging.error(f"Attempt {attempt + 1}: Failed to get transactions due to {e}")
            if attempt < max_retries - 1:
                await asyncio.sleep(retry_delay)
                retry_delay *= 2  # Exponential backoff
            else:
                raise e  # Re-raise the exception if max retries exceeded

Explanation:

• Timeout Parameter: We set the timeout parameter when initializing AsyncClient. This adjusts the maximum time to wait for a response.

  python
  async with AsyncClient('https://api.mainnet-beta.solana.com', timeout=timeout) as client:

• Retry Logic: We implement a retry mechanism with exponential backoff. If a transient error occurs, the code waits and retries the request.

• Logging: We added logging to keep track of failed attempts, which can help with debugging.

Additional Recommendations:

• Environment Variables: Ensure that the SOLANA_WALLETS_NUM environment variable is set appropriately or provide a default value.

• Error Handling: Be cautious with broad exception handling (except Exception as e). If you want to handle specific exceptions (like timeouts), you can catch httpx.ReadTimeout and solana.exceptions.SolanaRpcException separately.

• Thread Safety: If this code runs in a multi-threaded environment, ensure that the AsyncClient is used correctly within each coroutine.

• Update Dependencies: Make sure all your libraries (solana-py, httpx, etc.) are up to date, as newer versions might have fixes for related issues.

Example with Specific Exception Handling:

python
from httpx import ReadTimeout
from solana.exceptions import SolanaRpcException

...

        except (ReadTimeout, SolanaRpcException) as e:
            logging.error(f"Attempt {attempt + 1}: Timeout occurred: {e}")
            ...

Implementing in Your send_notifications Function:

Ensure that any exceptions not handled within get_latest_sol_transactions are appropriately managed in your send_notifications function to prevent the entire notification loop from crashing due to an unhandled exception.

Final Notes:

• By increasing the timeout and adding retries, your application becomes more robust against transient network issues and server load spikes.
• Always test these changes in a controlled environment before deploying them to production.

The new error you're encountering indicates that your get_latest_sol_transactions function is returning None, and when you attempt to create a set from it in send_notifications, Python raises a TypeError because None is not iterable.

Understanding the Error:

• HTTP 429 Too Many Requests: The log shows that you're receiving a 429 Too Many Requests response from the Solana RPC endpoint (https://api.mainnet-beta.solana.com). This means you've exceeded the rate limit for requests to this endpoint.

• Function Returns None: Due to the exception, your function get_latest_sol_transactions does not return any value, which defaults to None in Python.

• TypeError in send_notifications: When send_notifications tries to execute set((await get_latest_sol_transactions(wallet.wallet_hash))), it fails because you cannot create a set from None.

How to Fix It:

1. Ensure the Function Always Returns a List or Raises an Exception:

   Modify your get_latest_sol_transactions function to ensure that it either returns a list or raises an exception in all code paths. This prevents it from returning None.

2. Handle Exceptions in send_notifications:

   In your send_notifications function, wrap the call to get_latest_sol_transactions in a try-except block to handle any exceptions that might occur.

3. Implement Rate Limiting or Use an Alternative RPC Endpoint:

   Since you're hitting rate limits, consider using a different Solana RPC endpoint or implement rate limiting in your code to avoid sending too many requests in a short period.

Modified Code Examples:

1. Modify get_latest_sol_transactions Function:

python
import asyncio
import logging
import os
from solana.rpc.async_api import AsyncClient
from solana.publickey import PublicKey
from solana.exceptions import SolanaRpcException

async def get_latest_sol_transactions(address: str) -> list[str]:
    """
    Get the latest n transactions for the specified Solana address.

    :param address: Solana wallet address as a string.
    :return: List of signatures of the latest transactions.
    """
    count = int(os.getenv('SOLANA_WALLETS_NUM', '10'))  # Default to 10 if not set
    timeout = 30  # Timeout in seconds
    max_retries = 3  # Maximum number of retries
    retry_delay = 5  # Initial delay between retries

    for attempt in range(1, max_retries + 1):
        try:
            async with AsyncClient('https://api.mainnet-beta.solana.com', timeout=timeout) as client:
                res = await client.get_signatures_for_address(PublicKey(address))
                if res.value is not None:
                    return [str(tx.signature) for tx in res.value[:count]]
                else:
                    logging.info("No transactions found for the given address.")
                    return []
        except Exception as e:
            logging.error(f"Attempt {attempt}: Failed to get transactions due to: {repr(e)}")
            if attempt < max_retries:
                await asyncio.sleep(retry_delay)
                retry_delay *= 2  # Exponential backoff
            else:
                logging.error(f"Max retries exceeded for address {address}. Raising exception.")
                raise  # Re-raise the last exception
    # In case all retries are exhausted without raising an exception
    return []

2. Modify send_notifications Function:

python
async def send_notifications(bot: Bot, timeout: int):
    """
    Function to send notifications.

    :param bot: Bot instance that will send messages.
    :param timeout: Time between checks for new transactions (in seconds), configured in .env.
    """
    while True:
        wallets = await PostgresRequests().get_all_wallets()
        for wallet in wallets:
            old_links = await get_links(wallet_hash=wallet.wallet_hash)
            try:
                new_links = set(await get_latest_sol_transactions(wallet.wallet_hash))
            except Exception as e:
                logging.error(f"Failed to get transactions for wallet {wallet.wallet_hash} due to: {repr(e)}")
                continue  # Skip to the next wallet
            diff = new_links - old_links
            for link in diff:
                transaction = await get_sol_transaction(link)
                text = f"Транзакция: {transaction['transaction']} \n"
                text += f"Отправитель: {transaction['sender']} \n"
                text += f"Получатель: {transaction['receiver']} \n"
                text += f"Сумма: {transaction['amount']} SOL \n"
                await bot.send_message(chat_id=wallet.tg_id, text=text)
            await store_links(wallet_hash=wallet.wallet_hash, links=new_links)
        await asyncio.sleep(timeout)

3. Handle Rate Limiting:

Since you're receiving 429 Too Many Requests, you should handle this specifically:

python
from httpx import HTTPStatusError

async def get_latest_sol_transactions(address: str) -> list[str]:
    # ... (previous code) ...

        except SolanaRpcException as e:
            if '429' in str(e):
                logging.error(f"Received 429 Too Many Requests for address {address}. Retrying after delay.")
                if attempt < max_retries:
                    await asyncio.sleep(retry_delay)
                    retry_delay *= 2
                else:
                    logging.error(f"Max retries exceeded due to rate limiting. Raising exception.")
                    raise
            else:
                logging.error(f"Solana RPC Exception occurred: {repr(e)}")
                raise
        except Exception as e:
            logging.error(f"Attempt {attempt}: Unexpected error: {repr(e)}")
            raise

4. Consider Using an Alternative Solana RPC Endpoint:

The public Solana RPC endpoint (https://api.mainnet-beta.solana.com) has strict rate limits. You can use alternative endpoints or services like:

• QuickNode: https://www.quicknode.com/
• Alchemy: https://www.alchemy.com/solana
• Figment DataHub: https://datahub.figment.io/

These services often provide higher rate limits or even unlimited requests for free tiers.

5. Implement Client-Side Rate Limiting:

If changing the endpoint is not an option, implement rate limiting in your code to prevent hitting the server's limits:

python
import asyncio
from asyncio_throttle import Throttler

throttler = Throttler(rate_limit=5, period=1)  # Allow 5 requests per second

async def get_latest_sol_transactions(address: str) -> list[str]:
    # ... (previous code) ...

            async with throttler:
                async with AsyncClient('https://api.mainnet-beta.solana.com', timeout=timeout) as client:
                    # ... (rest of the code) ...

Note: You'll need to install asyncio-throttle with pip install asyncio-throttle.

6. Adjust Logging for Clarity:

Make sure your logging statements include repr(e) to capture the full exception information, especially since some exceptions may not include a message.

Example:

python
logging.error(f"Attempt {attempt}: Failed to get transactions due to: {repr(e)}")

Summary:

• Ensure that get_latest_sol_transactions always returns a list or raises an exception.
• Handle exceptions properly in send_notifications to prevent the entire loop from crashing.
• Address the rate limiting issue by using an alternative RPC endpoint or implementing client-side rate limiting.
• Adjust your logging to capture full exception details for better debugging.

Final Thoughts:

By making these adjustments, your application should become more robust, handle exceptions gracefully, and comply with the rate limits imposed by the Solana RPC endpoint.

The error you're encountering is:

ValueError: expected a sequence of length 32 (got 44)

This error occurs when constructing a PublicKey object with an input that doesn't meet the expected format or length. Specifically, the PublicKey constructor expects a sequence of bytes of length 32 (the length of a Solana public key in bytes), but it's receiving an input of length 44.

Understanding the Error:

• Incorrect Class Used: It appears that you're using Pubkey instead of PublicKey.
• Address Format Mismatch: The address you're providing is a base58-encoded string, which is 44 characters long. When decoded, it should result in 32 bytes.
• Class Expectations: The Pubkey class may be expecting a raw byte sequence, not a base58-encoded string.

Solution:

1. Use the Correct Class:

   Ensure you're importing and using the PublicKey class from the solana.publickey module, not Pubkey.

   python
   from solana.publickey import PublicKey

2. Modify Your Function:

   Update your get_latest_sol_transactions function to use PublicKey:

   python
   async def get_latest_sol_transactions(address: str) -> list[str]:
       """
       Получить последние n транзакций для указанного адреса в сети Solana.
   
       :param address: Строка с адресом кошелька Solana.
       :return: Список подписей последних транзакций.
       """
       count = int(os.getenv('SOLANA_WALLETS_NUM', '10'))  # Default to 10 if not set
       timeout = 30  # Timeout in seconds
       max_retries = 3  # Maximum number of retries
       retry_delay = 5  # Initial delay between retries
   
       for attempt in range(1, max_retries + 1):
           try:
               async with AsyncClient('https://api.mainnet-beta.solana.com', timeout=timeout) as client:
                   res = await client.get_signatures_for_address(PublicKey(address))
                   if res.value is not None:
                       return [str(tx.signature) for tx in res.value[:count]]
                   else:
                       logging.info("No transactions found for the given address.")
                       return []
           except Exception as e:
               logging.error(f"Attempt {attempt}: Failed to get transactions due to: {repr(e)}")
               if attempt < max_retries:
                   await asyncio.sleep(retry_delay)
                   retry_delay *= 2  # Exponential backoff
               else:
                   logging.error(f"Max retries exceeded for address {address}. Raising exception.")
                   raise  # Re-raise the last exception
       # In case all retries are exhausted without raising an exception
       return []

3. Explanation:

   • PublicKey vs. Pubkey: The PublicKey class is designed to handle base58-encoded strings directly, whereas Pubkey might not.
   • Base58 Decoding: The PublicKey constructor will internally decode the base58-encoded address to the required 32-byte sequence.
   • Correct Import: Ensure you're importing PublicKey from the correct module.

4. Check Your Imports:

   Make sure you have the correct import statements:

   python
   from solana.publickey import PublicKey

5. Verify the Address:

   Ensure that the address you're using is a valid Solana public key in base58 format.

Additional Notes:

• Library Version: Ensure you're using a compatible version of the solana-py library. Updating to the latest version can resolve unexpected issues.

  bash
  pip install --upgrade solana

• Error Handling: The exception handling in your function will now catch and log the error, then retry as per your logic.

• Consistent Class Usage: Using PublicKey consistently throughout your code will prevent similar errors.

Alternative Approach (If Needed):

If you still encounter issues, you can manually decode the base58-encoded address:

python
from solana.publickey import PublicKey
from base58 import b58decode

async def get_latest_sol_transactions(address: str) -> list[str]:
    # ... previous code ...
    try:
        pubkey_bytes = b58decode(address)
        res = await client.get_signatures_for_address(PublicKey(pubkey_bytes))
        # ... rest of your code ...

However, this shouldn't be necessary if PublicKey(address) works correctly.

Implementing in Your enable_tracking Function:

Ensure that you update the enable_tracking function accordingly:

python
async def enable_tracking():
    # ... previous code ...
    links = await get_latest_sol_transactions(address)
    # ... rest of your code ...

Summary:

• Replace Pubkey with PublicKey.
• Ensure correct import from solana.publickey.
• Verify that the address is a valid base58-encoded Solana public key.
• Update your code to handle exceptions and retries appropriately.

By making these changes, the ValueError should be resolved, and your application should function as expected.

Final Thoughts:

• Testing: After making these changes, test your application with a known valid address to ensure everything works.
• Logging: Continue to monitor logs for any other exceptions and handle them as needed.
• Documentation: Refer to the solana-py documentation for any additional guidance.

If you have any more questions or encounter further issues, feel free to ask!

The KeyError: 'program' you're encountering indicates that in some of the instructions within the instructions list, the 'program' key does not exist. This is causing your code to fail when it tries to access instruction['program'].

Understanding the Error:

• Instruction Variations: In the Solana transaction data, instructions can vary in structure depending on the type of transaction and how the instruction is encoded.
• Missing Keys: Some instructions might not have the 'program' key but instead have a 'programId' key, especially if they are not parsed or are in a different format.

Solution:

You need to adjust your code to handle cases where the 'program' key might not be present in an instruction. Here's how you can modify your code:

1. Use .get() Method:

   Replace instruction['program'] with instruction.get('program'). This will return None if the key doesn't exist, preventing a KeyError.

2. Add a Check for the Key:

   Before accessing instruction['program'], check if the key exists in the instruction.

3. Handle Instructions with 'programId':

   If the 'program' key is not present, check for the 'programId' key, which might contain the program ID in base58 format.

4. Log the Instruction for Debugging:

   Logging the instruction when the 'program' key is missing can help you understand the structure of such instructions.

Modified Code:

Here's the updated get_sol_transaction function with the necessary adjustments:

python
import aiohttp
import logging

async def get_sol_transaction(tx_signature: str) -> dict | None:
    """
    Получение информации о транзакции Solana по ее подписи.

    :param tx_signature: Подпись транзакции Solana в формате строки
    :return: Словарь с информацией о транзакции (подпись, отправитель, получатель, сумма) / None в случае ошибок
    """
    url = 'https://api.mainnet-beta.solana.com'

    headers = {'Content-Type': 'application/json'}
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getTransaction",
        "params": [
            tx_signature,
            {
                "encoding": "jsonParsed",
                "maxSupportedTransactionVersion": 0
            }
        ]
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=payload, headers=headers) as resp:
            response = await resp.json()
            if 'result' not in response or response['result'] is None:
                logging.error("Транзакция не найдена или произошла ошибка.")
                return

            result = response['result']

            # Информация о транзакции
            transaction = result['transaction']
            message = transaction['message']
            account_keys = message['accountKeys']
            instructions = message['instructions']

            # Инициализация переменных получателя и суммы
            receiver = None
            amount_lamports = None

            # Отправитель — первый в списке accountKeys (fee payer)
            sender = account_keys[0]['pubkey']

            # Поиск инструкции перевода в списке инструкций
            for instruction in instructions:
                program = instruction.get('program')
                if program == 'system':
                    parsed = instruction.get('parsed')
                    if parsed and parsed.get('type') == 'transfer':
                        info = parsed.get('info')
                        if info:
                            receiver = info.get('destination')
                            amount_lamports = int(info.get('lamports', 0))
                            break  # Предполагаем, что нас интересует первая инструкция перевода
                else:
                    # Optionally, handle other types of instructions or log them
                    logging.debug(f"Skipping instruction without 'program' key or non-system program: {instruction}")

            if amount_lamports is not None:
                amount_sol = amount_lamports / 1e9
            else:
                amount_sol = None

            return {
                "transaction": tx_signature,
                "sender": sender,
                "receiver": receiver,
                "amount": amount_sol,
            }

Explanation:

• Using .get() Method:

  • instruction.get('program') safely retrieves the value associated with 'program' or returns None if the key doesn't exist.
  • This prevents the KeyError exception when the key is missing.

• Additional Checks:

  • Before accessing nested keys, use .get() to ensure they exist.
  • For example, parsed.get('type') and info.get('destination').

• Logging:

  • Added a logging.debug statement to log instructions that are skipped. This can help you understand what other types of instructions are present in the transaction.

• Handling Amount:

  • Used info.get('lamports', 0) to provide a default value in case 'lamports' is missing.

Additional Recommendations:

1. Check for 'programId' Key:

   Sometimes, the instruction might have a 'programId' key instead of 'program'. You can adjust your code to handle this:

   python
   program = instruction.get('program') or instruction.get('programId')
   if program == 'system' or program == '11111111111111111111111111111111':
       # Proceed with parsing

   Note: '11111111111111111111111111111111' is the program ID for the system program in Solana.

2. Handle Inner Instructions:

   Some transactions include inner instructions that might contain the transfer details. You might need to check meta.innerInstructions.

   python
   meta = result.get('meta', {})
   inner_instructions = meta.get('innerInstructions', [])
   # Process inner instructions similarly

3. Robust Error Handling:

   Add exception handling around sections that might fail to prevent the function from crashing.

   python
   try:
       # Parsing logic
   except Exception as e:
       logging.error(f"Error parsing transaction {tx_signature}: {e}")
       return

Testing:

• Test with Different Transactions:

  • Test your function with various transaction signatures to ensure it works across different types of transactions.
  • Use both simple transfers and more complex transactions.

• Log Unexpected Instructions:

  • Modify the logging level to capture unexpected instruction formats.
  • This will help you adjust your parsing logic to handle more cases if needed.

Example of Enhanced Logging:

python
for instruction in instructions:
    program = instruction.get('program') or instruction.get('programId')
    if program == 'system' or program == '11111111111111111111111111111111':
        # Proceed with parsing
    else:
        logging.debug(f"Instruction with unhandled program: {program}, instruction: {instruction}")

Handling 'programId' as Base58 String:

• If program is not equal to 'system', it might be the program ID in base58 format.

• You can compare program to the system program ID:

  python
  SYSTEM_PROGRAM_ID = '11111111111111111111111111111111'
  program_id = instruction.get('programId')
  if program_id == SYSTEM_PROGRAM_ID:
      # Proceed with parsing

Final Code Example:

Combining the suggestions, here's a more robust version:

python
import aiohttp
import logging

SYSTEM_PROGRAM_ID = '11111111111111111111111111111111'

async def get_sol_transaction(tx_signature: str) -> dict | None:
    # ... (previous code remains the same) ...

            # Инициализация переменных получателя и суммы
            receiver = None
            amount_lamports = None

            # Отправитель — первый в списке accountKeys (fee payer)
            sender = account_keys[0]['pubkey']

            # Поиск инструкции перевода в списке инструкций
            for instruction in instructions:
                program = instruction.get('program')
                program_id = instruction.get('programId')

                if program == 'system' or program_id == SYSTEM_PROGRAM_ID:
                    parsed = instruction.get('parsed')
                    if parsed and parsed.get('type') == 'transfer':
                        info = parsed.get('info', {})
                        receiver = info.get('destination')
                        amount_lamports = int(info.get('lamports', 0))
                        break  # Предполагаем, что нас интересует первая инструкция перевода
                else:
                    logging.debug(f"Skipping instruction: {instruction}")

            if amount_lamports:
                amount_sol = amount_lamports / 1e9
            else:
                amount_sol = None

            return {
                "transaction": tx_signature,
                "sender": sender,
                "receiver": receiver,
                "amount": amount_sol,
            }

Summary:

• Avoid KeyError by using .get() when accessing dictionary keys.
• Check both 'program' and 'programId' to identify system program instructions.
• Use logging to debug and understand the structure of different instructions.
• Test your function thoroughly with different transaction types.

Additional Considerations:

• Handle Transactions Without Transfers:

  • Some transactions might not involve SOL transfers but might interact with tokens or other programs.
  • Decide how you want to handle such cases—either skip them or extract relevant information.

• Monitor API Limits:

  • Be aware of rate limits on the Solana RPC endpoint.
  • Implement retries with exponential backoff and handle HTTP 429 responses gracefully.

Final Thoughts:

By making these adjustments, your function should handle transactions more robustly and avoid crashing due to unexpected instruction formats. Always ensure that your code can handle various edge cases, especially when dealing with external data sources like blockchain APIs.

If you have any more questions or need further assistance, feel free to ask!

The error you're encountering is:

TypeError: 'NoneType' object is not subscriptable

This error occurs because your get_sol_transaction function is returning None, and your send_notifications function is trying to access it as if it were a dictionary. Specifically, the line:

python
text = f"Транзакция: {transaction['transaction']} \n"

fails because transaction is None, and you cannot subscript ([]) a NoneType object.

Understanding the Error:

• get_sol_transaction Returns None: Your function returns None when it cannot find the transaction or when an error occurs.
• No Check for None: In your send_notifications function, you proceed to use transaction without checking if it's None.
• TypeError Occurs: Accessing transaction['transaction'] when transaction is None raises a TypeError.

Why Did This Happen?

When you swapped Solana for WEN, the transaction might not have the expected structure that your get_sol_transaction function can parse. It could be a more complex transaction involving token swaps, which may not include the standard 'transfer' instruction of the 'system' program.

Solution:

1. Check for None in send_notifications:

   Modify your send_notifications function to handle cases where get_sol_transaction returns None.

   python
   async def send_notifications(bot: Bot, timeout: int):
       """
       Function to send notifications.
       """
       while True:
           wallets = await PostgresRequests().get_all_wallets()
           for wallet in wallets:
               old_links = await get_links(wallet_hash=wallet.wallet_hash)
               try:
                   new_links = set(await get_latest_sol_transactions(wallet.wallet_hash))
               except Exception as e:
                   logging.error(f"Failed to get transactions for wallet {wallet.wallet_hash} due to: {repr(e)}")
                   continue  # Skip to the next wallet
               diff = new_links - old_links
               for link in diff:
                   transaction = await get_sol_transaction(link)
                   if transaction is None:
                       logging.error(f"Failed to retrieve transaction details for {link}. Skipping.")
                       continue  # Skip this transaction
                   text = f"Транзакция: {transaction['transaction']} \n"
                   text += f"Отправитель: {transaction['sender']} \n"
                   text += f"Получатель: {transaction['receiver']} \n"
                   text += f"Сумма: {transaction['amount']} SOL \n"
                   await bot.send_message(chat_id=wallet.tg_id, text=text)
               await store_links(wallet_hash=wallet.wallet_hash, links=new_links)
           await asyncio.sleep(timeout)

   Explanation:

   • Check if transaction is None: Before trying to access its keys, verify that transaction is not None.
   • Logging: If transaction is None, log an error message and continue to the next iteration.

2. Enhance Logging in get_sol_transaction:

   Modify get_sol_transaction to include the transaction signature in the error message. This will help you identify which transactions are causing issues.

   python
   async def get_sol_transaction(tx_signature: str) -> dict | None:
       # ... previous code ...
       async with aiohttp.ClientSession() as session:
           async with session.post(url, json=payload, headers=headers) as resp:
               response = await resp.json()
               if 'result' not in response or response['result'] is None:
                   logging.error(f"Transaction {tx_signature} not found or an error occurred.")
                   return None
       # ... rest of your code ...

3. Handle Complex Transactions:

   Your get_sol_transaction function might not be equipped to handle complex transactions like token swaps. These transactions may not include the standard 'transfer' instruction you're looking for. To handle this:

   • Check for Inner Instructions:

     Some transactions include inner instructions that contain the actual transfer details.

     python
     meta = result.get('meta', {})
     inner_instructions = meta.get('innerInstructions', [])
     # Process inner instructions similar to the main instructions

   • Expand Instruction Parsing:

     Modify your function to parse different types of instructions and handle tokens other than SOL.

     python
     for instruction in instructions:
         program = instruction.get('program') or instruction.get('programId')
         if program in ['system', 'spl-token']:
             # Handle system transfers and token transfers

4. Update get_sol_transaction to Be More Robust:

   Here's an updated version of get_sol_transaction that attempts to handle token transfers and includes additional error handling:

   python
   async def get_sol_transaction(tx_signature: str) -> dict | None:
       """
       Получение информации о транзакции Solana по ее подписи.
       """
       url = 'https://api.mainnet-beta.solana.com'
       headers = {'Content-Type': 'application/json'}
       payload = {
           "jsonrpc": "2.0",
           "id": 1,
           "method": "getTransaction",
           "params": [
               tx_signature,
               {
                   "encoding": "jsonParsed",
                   "maxSupportedTransactionVersion": 0
               }
           ]
       }
   
       async with aiohttp.ClientSession() as session:
           async with session.post(url, json=payload, headers=headers) as resp:
               response = await resp.json()
               if 'result' not in response or response['result'] is None:
                   logging.error(f"Transaction {tx_signature} not found or an error occurred.")
                   return None
   
               result = response['result']
               transaction = result['transaction']
               message = transaction['message']
               account_keys = message['accountKeys']
               instructions = message['instructions']
   
               # Sender (fee payer) is the first in accountKeys
               sender = account_keys[0]['pubkey']
   
               # Initialize variables
               receiver = None
               amount_lamports = None
               token = 'SOL'
   
               # Check both main and inner instructions
               all_instructions = instructions
               meta = result.get('meta', {})
               inner_instructions = meta.get('innerInstructions', [])
               for inner in inner_instructions:
                   all_instructions.extend(inner.get('instructions', []))
   
               for instruction in all_instructions:
                   parsed = instruction.get('parsed')
                   if not parsed:
                       continue
                   info = parsed.get('info', {})
                   instruction_type = parsed.get('type')
   
                   # Handle system transfer
                   if instruction_type == 'transfer':
                       receiver = info.get('destination')
                       amount_lamports = int(info.get('lamports', 0))
                       token = 'SOL'
                       break
                   # Handle token transfer
                   elif instruction_type == 'transferChecked':
                       receiver = info.get('destination')
                       amount_lamports = int(info.get('tokenAmount', {}).get('amount', 0))
                       decimals = int(info.get('tokenAmount', {}).get('decimals', 0))
                       token = info.get('tokenAmount', {}).get('uiAmountString', 'Unknown Token')
                       amount_sol = amount_lamports / (10 ** decimals)
                       break
   
               if amount_lamports is not None:
                   if token == 'SOL':
                       amount = amount_lamports / 1e9
                   else:
                       amount = amount_sol
               else:
                   amount = None
   
               return {
                   "transaction": tx_signature,
                   "sender": sender,
                   "receiver": receiver,
                   "amount": amount,
                   "token": token,
               }

   Explanation:

   • Inner Instructions: We now include inner instructions, which may contain the actual transfer details for token swaps.
   • Parsing Token Transfers: We handle transferChecked instructions, which are commonly used for token transfers.
   • Token Amounts: We calculate the amount based on the token's decimals.
   • Include Token Name: We retrieve the token's symbol or name to include in the message.

5. Update Message Formatting:

   Adjust your message in send_notifications to include the token name:

   python
   text = f"Транзакция: {transaction['transaction']} \n"
   text += f"Отправитель: {transaction['sender']} \n"
   text += f"Получатель: {transaction['receiver']} \n"
   text += f"Сумма: {transaction['amount']} {transaction['token']} \n"

6. Handle Transactions Without Amount:

   If amount is None, decide how you want to handle such transactions. You can choose to:

   • Skip the Transaction:

     python
     if transaction['amount'] is None:
         logging.info(f"No amount found for transaction {transaction['transaction']}. Skipping.")
         continue

   • Send a Different Message:

     python
     text = f"Транзакция: {transaction['transaction']} \n"
     text += f"Отправитель: {transaction['sender']} \n"
     text += f"Получатель: {transaction['receiver']} \n"
     text += "Сумма: Не удалось определить сумму или тип транзакции.\n"

Final Thoughts:

• Error Handling: Always check for None or missing data before accessing dictionary keys.
• Logging: Provide detailed logging to help diagnose issues when they occur.
• Transaction Variability: Be aware that transactions can vary widely in structure, especially with token swaps and more complex operations.
• Testing: Test your code with different types of transactions to ensure it handles various cases.

Example of the Updated send_notifications Function:

python
async def send_notifications(bot: Bot, timeout: int):
    while True:
        wallets = await PostgresRequests().get_all_wallets()
        for wallet in wallets:
            old_links = await get_links(wallet_hash=wallet.wallet_hash)
            try:
                new_links = set(await get_latest_sol_transactions(wallet.wallet_hash))
            except Exception as e:
                logging.error(f"Failed to get transactions for wallet {wallet.wallet_hash} due to: {repr(e)}")
                continue
            diff = new_links - old_links
            for link in diff:
                transaction = await get_sol_transaction(link)
                if transaction is None:
                    logging.error(f"Failed to retrieve transaction details for {link}. Skipping.")
                    continue
                text = f"Транзакция: {transaction['transaction']} \n"
                text += f"Отправитель: {transaction['sender']} \n"
                text += f"Получатель: {transaction['receiver']} \n"
                text += f"Сумма: {transaction['amount']} {transaction['token']} \n"
                await bot.send_message(chat_id=wallet.tg_id, text=text)
            await store_links(wallet_hash=wallet.wallet_hash, links=new_links)
        await asyncio.sleep(timeout)

Summary:

• Check for None Values: Always ensure you're not accessing attributes of None objects.
• Handle Different Transaction Types: Adjust your parsing logic to handle various transaction types, including token transfers and swaps.
• Improve Error Messages: Include more context in your error messages to aid in debugging.
• Robust Code: Write your code to be robust against unexpected data structures.

By implementing these changes, your application should handle transactions more gracefully, and you should no longer encounter the TypeError due to NoneType objects.

If you need further assistance or have additional questions, feel free to ask!

The issue you're encountering stems from the way small floating-point numbers are displayed in Python. When dealing with very small amounts, Python may represent them in scientific notation (e.g., 1e-07), which can be confusing or misleading in user-facing messages.

Additionally, there may be inaccuracies in how the amount is calculated and displayed for token transfers, especially if the transaction involves tokens other than SOL.

Here's how to fix it:

1. Format the Amount Appropriately:

   To avoid displaying amounts in scientific notation, you can format the amount using string formatting to a fixed number of decimal places.

2. Ensure Correct Amount Calculation for Token Transfers:

   For token transfers, you need to extract the amount and token information correctly. Token transactions often involve different decimals and require proper parsing to display accurate information.

3. Map Token Mint Addresses to Token Symbols:

   To display the correct token symbol (e.g., "WEN" instead of the token mint address), you can use a token list to map mint addresses to their corresponding symbols.

Step-by-Step Solution:

1. Update the get_sol_transaction Function

Modify your get_sol_transaction function to handle token transfers correctly and format the amount properly.

python
import aiohttp
import logging

# Load the token list once at startup
token_map = {}

def load_token_list():
    global token_map
    import requests
    token_list_url = "https://raw.githubusercontent.com/solana-labs/token-list/main/src/tokens/solana.tokenlist.json"
    response = requests.get(token_list_url)
    token_list = response.json()
    tokens = token_list.get('tokens', [])
    token_map = {token['address']: token['symbol'] for token in tokens}

def get_token_symbol(mint_address):
    return token_map.get(mint_address, mint_address)  # Return the symbol or the mint address if not found

# Call load_token_list() at startup
load_token_list()

async def get_sol_transaction(tx_signature: str) -> dict | None:
    """
    Retrieve Solana transaction information by its signature.
    """
    url = 'https://api.mainnet-beta.solana.com'
    headers = {'Content-Type': 'application/json'}
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getTransaction",
        "params": [
            tx_signature,
            {
                "encoding": "jsonParsed",
                "maxSupportedTransactionVersion": 0
            }
        ]
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=payload, headers=headers) as resp:
            response = await resp.json()
            if 'result' not in response or response['result'] is None:
                logging.error(f"Transaction {tx_signature} not found or an error occurred.")
                return None

            result = response['result']
            transaction = result['transaction']
            message = transaction['message']
            account_keys = message['accountKeys']
            instructions = message['instructions']

            # Sender (fee payer) is the first in accountKeys
            sender = account_keys[0]['pubkey']

            # Initialize variables
            receiver = None
            amount = None
            token = 'SOL'
            token_mint = None

            # Combine instructions and inner instructions
            all_instructions = instructions.copy()
            meta = result.get('meta', {})
            inner_instructions = meta.get('innerInstructions', [])
            for inner in inner_instructions:
                all_instructions.extend(inner.get('instructions', []))

            for instruction in all_instructions:
                parsed = instruction.get('parsed')
                if not parsed:
                    continue
                info = parsed.get('info', {})
                instruction_type = parsed.get('type')
                program = instruction.get('program') or instruction.get('programId')

                if (program == 'system' or program == '11111111111111111111111111111111') and instruction_type == 'transfer':
                    # Handle SOL transfer
                    receiver = info.get('destination')
                    amount_lamports = int(info.get('lamports', 0))
                    amount = amount_lamports / 1e9
                    token = 'SOL'
                    break
                elif (program == 'spl-token' or program == 'TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA') and instruction_type in ['transfer', 'transferChecked']:
                    # Handle SPL token transfer
                    receiver = info.get('destination')
                    amount_raw = int(info.get('tokenAmount', {}).get('amount', 0))
                    decimals = int(info.get('tokenAmount', {}).get('decimals', 0))
                    amount = amount_raw / (10 ** decimals)
                    token_mint = info.get('mint')
                    token = get_token_symbol(token_mint)
                    break
                else:
                    continue

            if amount is not None:
                return {
                    "transaction": tx_signature,
                    "sender": sender,
                    "receiver": receiver,
                    "amount": amount,
                    "token": token,
                }
            else:
                logging.info(f"No transfer instruction found in transaction {tx_signature}")
                return None

Explanation:

• Token Mapping:

  • load_token_list(): Loads the Solana token list once at startup and populates the token_map dictionary.
  • get_token_symbol(mint_address): Retrieves the token symbol from token_map based on the mint address.

• Amount Calculation:

  • For SOL Transfers:
    • Amount is calculated by converting lamports to SOL (amount_lamports / 1e9).
  • For Token Transfers:
    • Amount is calculated using the token's decimals (amount_raw / (10 ** decimals)).

• Instruction Parsing:

  • Checks both main and inner instructions to find the transfer instruction.
  • Handles both 'system' program (for SOL transfers) and 'spl-token' program (for token transfers).

2. Update Message Formatting in send_notifications

Modify the message formatting to display the amount with a fixed number of decimal places to prevent scientific notation.

python
async def send_notifications(bot: Bot, timeout: int):
    while True:
        wallets = await PostgresRequests().get_all_wallets()
        for wallet in wallets:
            old_links = await get_links(wallet_hash=wallet.wallet_hash)
            try:
                new_links = set(await get_latest_sol_transactions(wallet.wallet_hash))
            except Exception as e:
                logging.error(f"Failed to get transactions for wallet {wallet.wallet_hash} due to: {repr(e)}")
                continue
            diff = new_links - old_links
            for link in diff:
                transaction = await get_sol_transaction(link)
                if transaction is None:
                    logging.error(f"Failed to retrieve transaction details for {link}. Skipping.")
                    continue
                # Format the amount to avoid scientific notation
                amount_formatted = "{:.9f}".format(transaction['amount']).rstrip('0').rstrip('.')
                text = f"Транзакция: {transaction['transaction']} \n"
                text += f"Отправитель: {transaction['sender']} \n"
                text += f"Получатель: {transaction['receiver']} \n"
                text += f"Сумма: {amount_formatted} {transaction['token']} \n"
                await bot.send_message(chat_id=wallet.tg_id, text=text)
            await store_links(wallet_hash=wallet.wallet_hash, links=new_links)
        await asyncio.sleep(timeout)

Explanation:

• Formatting the Amount:
  • "{:.9f}".format(transaction['amount']): Formats the amount to 9 decimal places.
  • rstrip('0').rstrip('.'): Removes trailing zeros and any trailing decimal point if the number is an integer.
• Example:
  • If the amount is 0.0000001, it will display as 0.0000001 instead of 1e-07.
  • If the amount is 1.0000000, it will display as 1.

3. Verify the Results

After making these changes, your bot messages should display the amount accurately and avoid scientific notation.

Example Messages:

1. SOL Transfer:

   text
   Транзакция: 3ua66VcYeHMoYyMvP4pRzcw8NXz2irAgSEX6YsjaEQLE7MMLKa4d2jQ5hAAvMHxwe287uS2Fkx6PrC4WztwBiWgG 
   Отправитель: 76a3CxHZYeGoTJLXopkKqE53zMBwgNriDyDgPZ456gmq 
   Получатель: AFFBbhjp6RQtaobLizaRh4Su39GN5uCCRQY7v5LRbYGe 
   Сумма: 0.00015 SOL

2. Token Transfer:

   text
   Транзакция: 3DtMVrBVetkyq4JxNG7yUoGAbuiVSa4jUXPPbLxNGEPBZba644nUcsG8WvpDDpYRx2sMNoST52cjhk5cCDoS2gRY 
   Отправитель: fLiPgg2yTvmgfhiPkKriAHkDmmXGP6CdeFX9UF5o7Zc 
   Получатель: GJ2RJuf8BFZZHU4EjoVT9Br4gMsmgMQaVLENimxjEYw 
   Сумма: 0.0000001 WEN

Note:

• The token symbol (e.g., "WEN") should now be displayed correctly if the token mint address is present in the token list.
• The amount should be displayed without scientific notation and with appropriate decimal places.

4. Additional Considerations

Handling Missing Tokens:

• If a token is not found in the token list, get_token_symbol will return the mint address.
• You may want to handle this by indicating that the token is unknown or by adding it to your token list.

Caching the Token List:

• Since the token list is loaded at startup, make sure your application reloads the token list if it runs for extended periods, or consider updating it periodically.

Error Handling:

• Ensure that all potential exceptions are caught and logged appropriately to prevent the application from crashing.

Summary

• Formatting Amounts: By formatting the amount with fixed decimal places, you avoid displaying confusing scientific notation for small amounts.
• Correct Amount Calculation: Ensuring that amounts are calculated correctly for both SOL and token transfers prevents misinformation.
• Token Symbols: Mapping token mint addresses to symbols allows users to understand which token is involved in the transaction.
• Robust Parsing: Handling both main and inner instructions ensures that all relevant transfer information is captured.

Final Thoughts:

By implementing these changes, your bot should now accurately display transaction amounts and tokens, providing clear and reliable information to users. Always test your application with various transaction types to ensure it handles different scenarios correctly.

If you need further assistance or have additional questions, feel free to ask!

The error you're encountering is due to a mismatch between the expected Content-Type header and the actual Content-Type of the response when using aiohttp to fetch the JSON file from the URL. Here's the specific error message:

aiohttp.client_exceptions.ContentTypeError: 200, message='Attempt to decode JSON with unexpected mimetype: text/plain; charset=utf-8', url='https://raw.githubusercontent.com/solana-labs/token-list/main/src/tokens/solana.tokenlist.json'

Understanding the Error:

• aiohttp Behavior: When you use await response.json(), aiohttp expects the Content-Type header of the response to be 'application/json' by default. If the response has a different Content-Type, it raises a ContentTypeError.

• Actual Content-Type: The URL you're accessing returns a Content-Type of 'text/plain; charset=utf-8', not 'application/json'. This is common for raw files served from GitHub's raw content URLs.

Solution:

You can instruct aiohttp to ignore the Content-Type header by passing content_type=None to response.json(). This tells aiohttp to attempt to parse the response body as JSON regardless of the Content-Type header.

Modified load_token_list Function:

python
import aiohttp
import logging

token_map = {}

async def load_token_list():
    global token_map
    token_list_url = "https://raw.githubusercontent.com/solana-labs/token-list/main/src/tokens/solana.tokenlist.json"
    async with aiohttp.ClientSession() as session:
        async with session.get(token_list_url) as response:
            # Pass content_type=None to ignore the Content-Type header
            token_list = await response.json(content_type=None)
            tokens = token_list.get('tokens', [])
            token_map = {token['address']: token['symbol'] for token in tokens}

Explanation:

• Passing content_type=None: This parameter overrides the default behavior of aiohttp and tells it to parse the response as JSON regardless of the Content-Type header.

  python
  token_list = await response.json(content_type=None)

• Why This Works: Since you know the content at the URL is valid JSON, it's safe to bypass the Content-Type check.

Alternative Solution:

If you prefer not to bypass the Content-Type check, you can read the response as text and then parse it using the json module:

python
import aiohttp
import json

token_map = {}

async def load_token_list():
    global token_map
    token_list_url = "https://raw.githubusercontent.com/solana-labs/token-list/main/src/tokens/solana.tokenlist.json"
    async with aiohttp.ClientSession() as session:
        async with session.get(token_list_url) as response:
            text = await response.text()
            token_list = json.loads(text)
            tokens = token_list.get('tokens', [])
            token_map = {token['address']: token['symbol'] for token in tokens}

Explanation:

• Reading as Text: await response.text() reads the response body as a text string.

• Parsing JSON Manually: json.loads(text) parses the text string into a Python dictionary.

Additional Recommendations:

1. Error Handling:

   Add exception handling to catch any potential errors during the network request or JSON parsing.

   python
   import aiohttp
   import json
   import logging
   
   token_map = {}
   
   async def load_token_list():
       global token_map
       token_list_url = "https://raw.githubusercontent.com/solana-labs/token-list/main/src/tokens/solana.tokenlist.json"
       try:
           async with aiohttp.ClientSession() as session:
               async with session.get(token_list_url) as response:
                   if response.status != 200:
                       logging.error(f"Failed to fetch token list. HTTP status code: {response.status}")
                       return
                   text = await response.text()
                   token_list = json.loads(text)
                   tokens = token_list.get('tokens', [])
                   token_map = {token['address']: token['symbol'] for token in tokens}
       except Exception as e:
           logging.error(f"An error occurred while loading the token list: {e}")

2. Asynchronous Execution:

   Ensure that load_token_list() is called in an asynchronous context since it's an async function.

   python
   import asyncio
   
   # Call this in your main coroutine or setup function
   asyncio.create_task(load_token_list())

   Or, if you're in an asynchronous context:

   python
   await load_token_list()

3. Update Token List Periodically:

   If your application runs continuously, you might want to update the token list periodically to account for new tokens.

   python
   async def update_token_list_periodically(interval_hours=24):
       while True:
           await load_token_list()
           await asyncio.sleep(interval_hours * 3600)  # Sleep for specified hours

   Start this coroutine in your main event loop:

   python
   asyncio.create_task(update_token_list_periodically())

Summary:

• Problem: aiohttp raises ContentTypeError because the response Content-Type is 'text/plain', not 'application/json'.

• Solution: Override the default behavior by passing content_type=None to response.json(), or read the response as text and parse it manually using the json module.

• Implementation: Modify your load_token_list function accordingly.

Final Thoughts:

By making these changes, your application should successfully fetch and parse the token list from the provided URL without encountering the ContentTypeError. Always ensure to handle exceptions gracefully to make your application robust.

If you have any further questions or need additional assistance, feel free to ask!

Здравствуйте,

В предыдущих ответах я предоставил изменения в коде, предназначенные для обработки как SOL, так и токеновых переводов. Цель состояла в том, чтобы вы получали уведомления о транзакциях, связанных с SOL и другими токенами, с указанием токена в сообщениях.

Из вашего описания следует, что вы получаете только уведомления о транзакциях SOL, но не получаете уведомлений о транзакциях с другими токенами. Давайте вместе выясним, почему уведомления о токеновых переводах не приходят.

Проверка и отладка кода

1. Проверка функции get_sol_transaction

В функции get_sol_transaction мы пытаемся обработать как SOL, так и токеновые переводы. Возможно, токеновые переводы не обнаруживаются из-за ошибок в логике разбора инструкций.

Добавим дополнительные логи для отладки:

python
import logging

async def get_sol_transaction(tx_signature: str) -> dict | None:
    # ... существующий код ...

    # Инициализация переменных
    receiver = None
    amount = None
    token = 'SOL'
    token_mint = None

    # Объединение инструкций и внутренних инструкций
    all_instructions = instructions.copy()
    meta = result.get('meta', {})
    inner_instructions = meta.get('innerInstructions', [])
    for inner in inner_instructions:
        all_instructions.extend(inner.get('instructions', []))

    for instruction in all_instructions:
        parsed = instruction.get('parsed')
        if not parsed:
            logging.debug(f"Инструкция без поля 'parsed': {instruction}")
            continue
        info = parsed.get('info', {})
        instruction_type = parsed.get('type')
        program = instruction.get('program') or instruction.get('programId')

        if (program == 'system' or program == '11111111111111111111111111111111') and instruction_type == 'transfer':
            # Обработка перевода SOL
            logging.debug(f"SOL перевод обнаружен в транзакции {tx_signature}")
            receiver = info.get('destination')
            amount_lamports = int(info.get('lamports', 0))
            amount = amount_lamports / 1e9
            token = 'SOL'
            break
        elif (program == 'spl-token' or program == 'TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA') and instruction_type in ['transfer', 'transferChecked']:
            # Обработка токенового перевода
            logging.debug(f"Токеновый перевод обнаружен в транзакции {tx_signature}")
            receiver = info.get('destination')
            sender = info.get('source')  # Для токеновых переводов отправитель — 'source'
            amount_raw = int(info.get('tokenAmount', {}).get('amount', 0))
            decimals = int(info.get('tokenAmount', {}).get('decimals', 0))
            amount = amount_raw / (10 ** decimals)
            token_mint = info.get('mint')
            token = get_token_symbol(token_mint)
            break
        else:
            logging.debug(f"Необработанная инструкция в транзакции {tx_signature}: {instruction}")
            continue

    if amount is not None:
        logging.debug(f"Транзакция {tx_signature} успешно обработана: {amount} {token}")
        return {
            "transaction": tx_signature,
            "sender": sender,
            "receiver": receiver,
            "amount": amount,
            "token": token,
        }
    else:
        logging.info(f"В транзакции {tx_signature} не найдено инструкции перевода")
        return None

2. Проверка функции send_notifications

Убедитесь, что функция send_notifications не пропускает токеновые транзакции:

python
if transaction is None:
    logging.error(f"Не удалось получить детали транзакции для {link}. Пропуск.")
    continue  # Переходим к следующей транзакции

3. Тестирование с известными токеновыми транзакциями

Попробуйте вручную вызвать get_sol_transaction с подписью транзакции, которая содержит токеновый перевод, и проверьте вывод:

python
test_signature = '...подпись транзакции с токеновым переводом...'
transaction = await get_sol_transaction(test_signature)
print(transaction)

4. Проверка отображения токенов

Убедитесь, что адрес токена (mint address) присутствует в token_map. Если токен отсутствует в списке, функция get_token_symbol вернет адрес mint, и это может вызвать недопонимание.

Возможные причины и решения

1. Инструкции не обнаруживаются

• Проблема: Токеновые переводы могут находиться во внутренних инструкциях (innerInstructions), и их нужно правильно объединить с основными инструкциями.

• Решение: Убедитесь, что вы правильно объединяете все инструкции:

  python
  all_instructions = instructions.copy()
  meta = result.get('meta', {})
  inner_instructions = meta.get('innerInstructions', [])
  for inner in inner_instructions:
      all_instructions.extend(inner.get('instructions', []))

2. Неправильный парсинг инструкций

• Проблема: Возможно, инструкции не парсятся должным образом из-за различий в структуре токеновых переводов.

• Решение: Убедитесь, что вы правильно извлекаете sender, receiver, amount и token из инструкции.

3. Отсутствие токена в token_map

• Проблема: Если токен отсутствует в загруженном списке токенов, символ токена не будет найден.

• Решение: Добавьте обработку случая, когда символ токена не найден:

  python
  def get_token_symbol(mint_address):
      return token_map.get(mint_address, 'Unknown Token')