wip - still not working

crypto/sol/app.py

@@ -432,9 +432,9 @@ async def process_messages(websocket):
 
 
 pk = None
-
+app = init_app()
 # Convert Flask app to ASGI
-asgi_app = WsgiToAsgi(init_app)
+asgi_app = WsgiToAsgi(app)
 
 async def main():
     global solanaAPI, bot, PROCESSING_LOG, pk

crypto/sol/modules/SolanaAPI.py

@@ -53,10 +53,10 @@ PING_INTERVAL = 30
 SUBSCRIBE_INTERVAL = 10*60  # Resubscribe every 1 minute
 
 from config import (
-    FOLLOWED_WALLET, SOLANA_HTTP_URL, DISPLAY_CURRENCY, SOLANA_ENDPOINTS
+    FOLLOWED_WALLET, SOLANA_HTTP_URL, DISPLAY_CURRENCY, SOLANA_ENDPOINTS, YOUR_WALLET
 )
 
-from modules.utils import telegram_utils
+from modules.utils import telegram_utils, async_safe_call
 
 # Use the production Solana RPC endpoint
 solana_client = AsyncClient(SOLANA_HTTP_URL)
@@ -124,7 +124,9 @@ class SolanaWS:
         ]
         # define onmessage as inline callback to get subscription_id which waits for last_msg_responded
         # self.on_message = lambda message: self.subscription_id = message.get('result')
-        result = await self.ws_jsonrpc("logsSubscribe", params)
+        await self.ws_jsonrpc("logsSubscribe", params, False)
+        await self.receive_messages(True) 
+        result = await self.process_messages(True)    
         
         if result is not None and result > 0:
             self.subscription_id = result
@@ -143,11 +145,17 @@ class SolanaWS:
             else:
                 logger.error(f"Failed to unsubscribe from subscription id: {self.subscription_id}")
 
-    async def receive_messages(self):
+    async def receive_messages(self, one = False):
         while True:
             try:
-                message = await self.websocket.recv()
+                response = await self.websocket.recv()
-                await self.message_queue.put(message)
+                response_data = json.loads(response)
+                self.last_msg_responded = True
+                
+                if 'result' in response_data:
+                    await self.message_queue.put(response_data['result'])
+                if one:
+                    break
             except websockets.exceptions.ConnectionClosedError:
                 logger.error("WebSocket connection closed")
                 break
@@ -155,12 +163,15 @@ class SolanaWS:
                 logger.error(f"Error receiving message: {e}")
                 break
 
-    async def process_messages(self):
+    async def process_messages(self, one = False):
         while True:
             message = await self.message_queue.get()
+            message = json.loads(message)
             if self.on_message:
                 await self.on_message(message)
             logger.info(f"Received message: {message}")
+            if one:
+                return message 
 
     async def close(self):
         if self.websocket:
@@ -206,7 +217,7 @@ class SolanaAPI:
         # Use the provided on_bot_message if it's callable, otherwise use the default
         self.on_bot_message = on_bot_message if callable(on_bot_message) else default_on_bot_message
 
-        self.dex = SolanaDEX(DISPLAY_CURRENCY)
+        self.dex = DEX
         self.solana_ws = SolanaWS(on_message=self.process_transaction)
 
     async def process_messages(self, solana_ws):
@@ -226,11 +237,11 @@ class SolanaAPI:
                 await solana_ws.connect()
                 await solana_ws.subscribe()
 
-                if first_subscription and self.on_initial_subscription is not None:
+                if first_subscription:
-                    await self.on_initial_subscription
+                    await async_safe_call( self.on_initial_subscription, solana_ws.subscription_id)
                     first_subscription = False
 
-                self.on_bot_message(f"Solana mainnet connected ({solana_ws.subscription_id})...")
+                await async_safe_call(self.on_bot_message,f"Solana mainnet connected ({solana_ws.subscription_id})...")
 
                 receive_task = asyncio.create_task(solana_ws.receive_messages()) 
                 process_task = asyncio.create_task(solana_ws.process_messages())
@@ -249,8 +260,7 @@ class SolanaAPI:
                 await solana_ws.unsubscribe()
                 if solana_ws.websocket:
                     await solana_ws.close()
-                if self.on_bot_message:
+                await async_safe_call(self.on_bot_message,"Reconnecting...")
-                    await self.on_bot_message("Reconnecting...")
                 await asyncio.sleep(5)
 
     async def get_last_transactions(self, account_address, check_interval=300, limit=1000):
@@ -559,75 +569,69 @@ class SolanaAPI:
 
 
 class SolanaDEX:
-    def __init__(self, DISPLAY_CURRENCY):
+    def __init__(self, DISPLAY_CURRENCY: str):
         self.DISPLAY_CURRENCY = DISPLAY_CURRENCY
-        pass    
+        self.solana_client = AsyncClient("https://api.mainnet-beta.solana.com")
+        self.TOKENS_INFO = {}
+        self.TOKEN_PRICES = {}
+        self.TOKEN_ADDRESSES = {}
+        self.FOLLOWED_WALLET_VALUE = 0
+        self.YOUR_WALLET_VALUE = 0
 
-    async def get_token_prices(token_addresses: List[str]) -> Dict[str, float]:
+    async def get_token_prices(self, token_addresses: List[str]) -> Dict[str, float]:
-        global TOKENS_INFO
-        
-    # Skip for USD
         prices = {addr: 1.0 for addr in token_addresses if addr == "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"}
         remaining_tokens = [addr for addr in token_addresses if addr not in prices]
         
-        # Try CoinGecko
         coingecko_prices = await self.get_prices_from_coingecko(remaining_tokens)
         prices.update(coingecko_prices)
         
-        
-        # For remaining missing tokens, try Jupiter
         missing_tokens = set(remaining_tokens) - set(prices.keys())
         if missing_tokens:
-            jupiter_prices = await get_prices_from_jupiter(list(missing_tokens))
+            jupiter_prices = await self.get_prices_from_jupiter(list(missing_tokens))
             prices.update(jupiter_prices)
             
-            
-        # For tokens not found in CoinGecko, use DexScreener
         missing_tokens = set(remaining_tokens) - set(coingecko_prices.keys())
         if missing_tokens:
-            dexscreener_prices = await get_prices_from_dexscreener(list(missing_tokens))
+            dexscreener_prices = await self.get_prices_from_dexscreener(list(missing_tokens))
             prices.update(dexscreener_prices)
         
-        # For remaining missing tokens, try Raydium
         missing_tokens = set(remaining_tokens) - set(prices.keys())
         if missing_tokens:
-            raydium_prices = await get_prices_from_raydium(list(missing_tokens))
+            raydium_prices = await self.get_prices_from_raydium(list(missing_tokens))
             prices.update(raydium_prices)
         
-        # For remaining missing tokens, try Orca
         missing_tokens = set(remaining_tokens) - set(prices.keys())
         if missing_tokens:
-            orca_prices = await get_prices_from_orca(list(missing_tokens))
+            orca_prices = await self.get_prices_from_orca(list(missing_tokens))
             prices.update(orca_prices)
         
-        # If any tokens are still missing, set their prices to 0
         for token in set(token_addresses) - set(prices.keys()):
             prices[token] = 0.0
             logging.warning(f"Price not found for token {token}. Setting to 0.")
         
         for token, price in prices.items():
-            token_info = TOKENS_INFO.setdefault(token, {})
+            token_info = self.TOKENS_INFO.setdefault(token, {})
             if 'symbol' not in token_info:
-                token_info['symbol'] = await get_token_metadata_symbol(token)
+                token_info['symbol'] = await self.get_token_metadata_symbol(token)
             token_info['price'] = price                       
         
         return prices
 
-    async def get_prices_from_coingecko(token_addresses: List[str]) -> Dict[str, float]:
+    async def get_prices_from_coingecko(self, token_addresses: List[str]) -> Dict[str, float]:
         base_url = "https://api.coingecko.com/api/v3/simple/token_price/solana"
         prices = {}
 
         async def fetch_single_price(session, address):
             params = {
                 "contract_addresses": address,
-                "vs_currencies": DISPLAY_CURRENCY.lower()
+                "vs_currencies": self.DISPLAY_CURRENCY.lower()
             }
             try:
                 async with session.get(base_url, params=params) as response:
                     if response.status == 200:
                         data = await response.json()
-                        if address in data and DISPLAY_CURRENCY.lower() in data[address]:
+                        if address in data and self.DISPLAY_CURRENCY.lower() in data[address]:
-                            return address, data[address][DISPLAY_CURRENCY.lower()]
+                            return address, data[address][self.DISPLAY_CURRENCY.lower()]
                     else:
                         logging.warning(f"Failed to get price for {address} from CoinGecko. Status: {response.status}")
             except Exception as e:
@@ -644,13 +648,13 @@ class SolanaDEX:
 
         return prices
 
-    async def get_prices_from_dexscreener(token_addresses: List[str]) -> Dict[str, float]:
+    async def get_prices_from_dexscreener(self, token_addresses: List[str]) -> Dict[str, float]:
         base_url = "https://api.dexscreener.com/latest/dex/tokens/"
         prices = {}
 
         try:
             async with aiohttp.ClientSession() as session:
-                tasks = [fetch_token_data(session, f"{base_url}{address}") for address in token_addresses]
+                tasks = [self.fetch_token_data(session, f"{base_url}{address}") for address in token_addresses]
                 results = await asyncio.gather(*tasks)
                 
                 for address, result in zip(token_addresses, results):
@@ -664,7 +668,7 @@ class SolanaDEX:
 
         return prices
 
-    async def get_prices_from_jupiter(token_addresses: List[str]) -> Dict[str, float]:
+    async def get_prices_from_jupiter(self, token_addresses: List[str]) -> Dict[str, float]:
         url = "https://price.jup.ag/v4/price"
         params = {
             "ids": ",".join(token_addresses)
@@ -685,8 +689,7 @@ class SolanaDEX:
             logging.error(f"Error fetching token prices from Jupiter: {str(e)}")
         return prices
 
-    # New function for Raydium
+    async def get_prices_from_raydium(self, token_addresses: List[str]) -> Dict[str, float]:
-    async def get_prices_from_raydium(token_addresses: List[str]) -> Dict[str, float]:
         url = "https://api.raydium.io/v2/main/price"
         prices = {}
         
@@ -704,8 +707,7 @@ class SolanaDEX:
             logging.error(f"Error fetching token prices from Raydium: {str(e)}")
         return prices
 
-    # New function for Orca
+    async def get_prices_from_orca(self, token_addresses: List[str]) -> Dict[str, float]:
-    async def get_prices_from_orca(token_addresses: List[str]) -> Dict[str, float]:
         url = "https://api.orca.so/allTokens"
         prices = {}
         
@@ -723,6 +725,7 @@ class SolanaDEX:
             logging.error(f"Error fetching token prices from Orca: {str(e)}")
         return prices
 
+    @staticmethod
     async def fetch_token_data(session, url):
         try:
             async with session.get(url) as response:
@@ -735,17 +738,16 @@ class SolanaDEX:
             logging.error(f"Error fetching data from {url}: {str(e)}")
             return None
 
-    async def get_sol_price() -> float:
+    async def get_sol_price(self) -> float:
         sol_address = "So11111111111111111111111111111111111111112"  # Solana's wrapped SOL address
-        return await get_token_prices([sol_address]).get(sol_address, 0.0)
+        prices = await self.get_token_prices([sol_address])
+        return prices.get(sol_address, 0.0)
 
-
+    async def get_wallet_balances(self, wallet_address, doGetTokenName=True):
-    async def get_wallet_balances(wallet_address, doGetTokenName=True):
         balances = {}
         logging.info(f"Getting balances for wallet: {wallet_address}")
-        global TOKENS_INFO
         try:
-            response = await solana_client.get_token_accounts_by_owner_json_parsed(
+            response = await self.solana_client.get_token_accounts_by_owner_json_parsed(
                 Pubkey.from_string(wallet_address),
                 opts=TokenAccountOpts(
                     program_id=Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
@@ -764,29 +766,27 @@ class SolanaDEX:
                                 decimals = info['tokenAmount']['decimals']
                                 amount =  float(info['tokenAmount']['amount'])/10**decimals
                                 if amount > 0:
-                                    if mint in TOKENS_INFO:
+                                    if mint in self.TOKENS_INFO:
-                                        token_name = TOKENS_INFO[mint].get('symbol')
+                                        token_name = self.TOKENS_INFO[mint].get('symbol')
                                     elif doGetTokenName:
-                                        token_name = await get_token_metadata_symbol(mint) or 'N/A'
+                                        token_name = await self.get_token_metadata_symbol(mint) or 'N/A'
-                                        # sleep for 1 second to avoid rate limiting
                                         await asyncio.sleep(2)
                                         
-                                    TOKENS_INFO[mint]['holdedAmount'] = round(amount,decimals)
+                                    self.TOKENS_INFO[mint]['holdedAmount'] = round(amount,decimals)
-                                    TOKENS_INFO[mint]['decimals'] = decimals
+                                    self.TOKENS_INFO[mint]['decimals'] = decimals
                                     balances[mint] = {
                                         'name': token_name or 'N/A',
                                         'address': mint,
                                         'amount': amount,
                                         'decimals': decimals
                                     }
-                                    # sleep for 1 second to avoid rate limiting
                                     logging.debug(f"Account balance for {token_name} ({mint}): {amount}")
                         else:
                             logging.warning(f"Unexpected data format for account: {account}")
                     except Exception as e:
                         logging.error(f"Error parsing account data: {str(e)}")
 
-            sol_balance = await solana_client.get_balance(Pubkey.from_string(wallet_address))
+            sol_balance = await self.solana_client.get_balance(Pubkey.from_string(wallet_address))
             if sol_balance.value is not None:
                 balances['SOL'] = {
                     'name': 'SOL',
@@ -801,80 +801,87 @@ class SolanaDEX:
         logging.info(f"Found {len(response.value)} ({len(balances)} non zero) token accounts for wallet: {wallet_address}")
         return balances
 
-    async def convert_balances_to_currency(balances , sol_price):
+    async def convert_balances_to_currency(self, balances, sol_price):
         converted_balances = {}
         for address, info in balances.items():
-            converted_balance = info.copy()  # Create a copy of the original info
+            converted_balance = info.copy()
             if info['name'] == 'SOL':
                 converted_balance['value'] = info['amount'] * sol_price
-            elif address in TOKEN_PRICES:
+            elif address in self.TOKEN_PRICES:
-                converted_balance['value'] = info['amount'] * TOKEN_PRICES[address]
+                converted_balance['value'] = info['amount'] * self.TOKEN_PRICES[address]
             else:
-                converted_balance['value'] = None  # Price not available
+                converted_balance['value'] = None
                 logging.warning(f"Price not available for token {info['name']} ({address})")
             converted_balances[address] = converted_balance
         return converted_balances
 
-
+    async def list_initial_wallet_states(self, FOLLOWED_WALLET, YOUR_WALLET):
-    async def list_initial_wallet_states():
+        followed_wallet_balances = await self.get_wallet_balances(FOLLOWED_WALLET)
-        global TOKEN_ADDRESSES, FOLLOWED_WALLET_VALUE, YOUR_WALLET_VALUE, TOKEN_PRICES
+        your_wallet_balances = await self.get_wallet_balances(YOUR_WALLET)
-        global TOKENS_INFO # new 
-        
-        followed_wallet_balances = await get_wallet_balances(FOLLOWED_WALLET)
-        your_wallet_balances = await get_wallet_balances(YOUR_WALLET)
         
         all_token_addresses = list(set(followed_wallet_balances.keys()) | 
                             set(your_wallet_balances.keys()) | 
-                            set(TOKEN_ADDRESSES.values()))
+                            set(self.TOKEN_ADDRESSES.values()))
 
-        TOKEN_PRICES = await get_token_prices(all_token_addresses)
+        self.TOKEN_PRICES = await self.get_token_prices(all_token_addresses)
-        sol_price = await get_sol_price()
+        sol_price = await self.get_sol_price()
         
-        followed_converted_balances = await convert_balances_to_currency(followed_wallet_balances, sol_price)
+        followed_converted_balances = await self.convert_balances_to_currency(followed_wallet_balances, sol_price)
-        your_converted_balances = await convert_balances_to_currency(your_wallet_balances, sol_price)
+        your_converted_balances = await self.convert_balances_to_currency(your_wallet_balances, sol_price)
 
-
+        self.TOKEN_ADDRESSES = {
-        TOKEN_ADDRESSES = {
             address: info for address, 
             info in {**followed_converted_balances, **your_converted_balances}.items() if info['value'] is not None and info['value'] > 0
             }
-        logging.info(f"Monitoring balances for tokens: {[info['name'] for info in TOKEN_ADDRESSES.values()]}")
+        logging.info(f"Monitoring balances for tokens: {[info['name'] for info in self.TOKEN_ADDRESSES.values()]}")
         
         followed_wallet_state = []
-        FOLLOWED_WALLET_VALUE = 0
+        self.FOLLOWED_WALLET_VALUE = 0
         for address, info in followed_converted_balances.items():
             if info['value'] is not None and info['value'] > 0:
-                followed_wallet_state.append(f"{info['name']}: {info['value']:.2f} {DISPLAY_CURRENCY} ({info['address']})")
+                followed_wallet_state.append(f"{info['name']}: {info['value']:.2f} {self.DISPLAY_CURRENCY} ({info['address']})")
-                FOLLOWED_WALLET_VALUE += info['value']
+                self.FOLLOWED_WALLET_VALUE += info['value']
 
         your_wallet_state = []
-        YOUR_WALLET_VALUE = 0
+        self.YOUR_WALLET_VALUE = 0
         for address, info in your_converted_balances.items():
             if info['value'] is not None and info['value'] > 0:
-                your_wallet_state.append(f"{info['name']}: {info['value']:.2f} {DISPLAY_CURRENCY}")
+                your_wallet_state.append(f"{info['name']}: {info['value']:.2f} {self.DISPLAY_CURRENCY}")
-                YOUR_WALLET_VALUE += info['value']
+                self.YOUR_WALLET_VALUE += info['value']
         
         message = (
-            f"<b>Initial Wallet States (All balances in {DISPLAY_CURRENCY}):</b>\n\n"
+            f"<b>Initial Wallet States (All balances in {self.DISPLAY_CURRENCY}):</b>\n\n"
             f"<b>Followed Wallet ({FOLLOWED_WALLET}):</b>\n"
             f"{chr(10).join(followed_wallet_state)}\n"
-            f"<b>Total Value:</b> {FOLLOWED_WALLET_VALUE:.2f} {DISPLAY_CURRENCY}\n\n"
+            f"<b>Total Value:</b> {self.FOLLOWED_WALLET_VALUE:.2f} {self.DISPLAY_CURRENCY}\n\n"
             f"<b>Your Wallet ({YOUR_WALLET}):</b>\n"
             f"{chr(10).join(your_wallet_state)}\n"
-            f"<b>Total Value:</b> {YOUR_WALLET_VALUE:.2f} {DISPLAY_CURRENCY}\n\n"
+            f"<b>Total Value:</b> {self.YOUR_WALLET_VALUE:.2f} {self.DISPLAY_CURRENCY}\n\n"
             f"<b>Monitored Tokens:</b>\n"
-            f"{', '.join([safe_get_property(info, 'name') for info in TOKEN_ADDRESSES.values()])}"
+            f"{', '.join([self.safe_get_property(info, 'name') for info in self.TOKEN_ADDRESSES.values()])}"
         )
         
         logging.info(message)
-        await telegram_utils.send_telegram_message(message)
+        # await telegram_utils.send_telegram_message(message)
 
         # save token info to file
-        await save_token_info()
+        await self.save_token_info()
+
+    @staticmethod
+    def safe_get_property(obj, prop):
+        return obj.get(prop, 'N/A')
+
+    async def get_token_metadata_symbol(self, token_address):
+        # Implement this method to fetch token metadata symbol
+        pass
+
+    async def save_token_info(self):
+        # Implement this method to save token info to a file
+        pass
      
 async def save_token_info():
     with open('./logs/token_info.json', 'w') as f:
         json.dump(TOKENS_INFO, f, indent=2)
 
-
+DEX = SolanaDEX(DISPLAY_CURRENCY)
-SAPI = SolanaAPI( on_initial_subscription_callback=SolanaDEX.list_initial_wallet_states())
+SAPI = SolanaAPI( on_initial_subscription_callback=DEX.list_initial_wallet_states(FOLLOWED_WALLET,YOUR_WALLET))

crypto/sol/modules/utils.py

@@ -10,6 +10,8 @@ from telegram.constants import ParseMode
 from config import TELEGRAM_BOT_TOKEN, DEVELOPER_CHAT_ID, BOT_NAME
 
 
+import asyncio
+from typing import Callable, Any
 import time
 import logging
 from logging.handlers import RotatingFileHandler
@@ -67,7 +69,7 @@ class Log:
     # Set up success logger for accounting CSV
     
     def __init__(self):
-        logger = logging.getLogger(__name__)    
+        self.logger = logging.getLogger(__name__)    
         logging.basicConfig(level=logging.DEBUG)
         #logging.basicConfig(level=logging.INFO)
 
@@ -109,11 +111,34 @@ def safe_get_property(info, property_name, default='Unknown'):
     value = info.get(property_name, default)
     return str(value) if value is not None else str(default)
 
+async def async_safe_call(func: Callable, *args: Any, **kwargs: Any) -> Any:
+    """
+    Safely call a function that might be synchronous, asynchronous, or a coroutine object.
+    
+    :param func: The function to call, or a coroutine object
+    :param args: Positional arguments to pass to the function
+    :param kwargs: Keyword arguments to pass to the function
+    :return: The result of the function call, or None if func is not callable or a coroutine
+    """
+    if func is None:
+        return None
+
+    if callable(func):
+        if asyncio.iscoroutinefunction(func):
+            return await func(*args, **kwargs)
+        else:
+            return func(*args, **kwargs)
+    elif asyncio.iscoroutine(func):
+        # If func is already a coroutine object, just await it
+        return await func
+    else:
+        logging.warning(f"Expected a callable or coroutine, but got {type(func)}: {func}")
+        return None
 
 
 # Create a global instance of TelegramUtils
 telegram_utils = TelegramUtils()
-log = Log()
+log = Log().logger
 # You can add more Telegram-related methods to the TelegramUtils class if needed