import asyncio import uvicorn from asgiref.wsgi import WsgiToAsgi import websockets import json import datetime import os from dotenv import load_dotenv from threading import Thread from solana.rpc.async_api import AsyncClient from solders.transaction import VersionedTransaction from solana.rpc.types import TxOpts from solders.keypair import Keypair from jupiter_python_sdk.jupiter import Jupiter from solana.rpc.commitment import Processed from modules.webui import init_app from modules.storage import init_db, store_transaction from modules.utils import telegram_utils, logging, get_pk from modules.log_processor import watch_for_new_logs from modules.SolanaAPI import SAPI # config = load_config() load_dotenv() load_dotenv('.env.secret') # Configuration from config import (DO_WATCH_WALLET, logging, logger) # # # # # # # # # # TELEGRAM # # # # # # # # # # # if not telegram_utils.bot: # try: # asyncio.run(telegram_utils.initialize()) # except Exception as e: # logging.error(f"Error initializing Telegram bot: {str(e)}") # async def telegram_utils.send_telegram_message(message): # try: # await bot.send_message(chat_id=DEVELOPER_CHAT_ID, text=f"[{BOT_NAME}] {message}", parse_mode=ParseMode.HTML) # logging.info(f"Telegram message sent: {message}") # # logging.info(f"Telegram message dummy sent: {message}") # except Exception as e: # logging.error(f"Error sending Telegram message: {str(e)}") # # # # # # # # # # DATABASE # # # # # # # # # # # # # # # # # # # # CRYPTO PUBLIC # # # # # # # # # # # # # # # # # # # # SOLANA BLOCKCHAIN # # # # # # # # # # # # # # # # # # # # Functionality # # # # # # # # # # async def save_log(log): try: os.makedirs('./logs', exist_ok=True) timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f") filename = f"./logs/log_{timestamp}.json" with open(filename, 'w') as f: json.dump(log, f, indent=2) except Exception as e: logging.error(f"Error saving RPC log: {e}") PROCESSING_LOG = False async def process_log(log_result): global PROCESSING_LOG tr_details = { "order_id": None, "token_in": None, "token_out": None, "amount_in": 0, "amount_out": 0, "amount_in_USD": 0, "amount_out_USD": 0, "percentage_swapped": 0 } if log_result['value']['err']: return logs = log_result['value']['logs'] try: # Detect swap operations in logs PROCESSING_LOG = True swap_operations = ['Program log: Instruction: Swap', 'Program log: Instruction: Swap2', 'Program log: Instruction: SwapExactAmountIn', 'Program log: Instruction: SwapV2'] if any(op in logs for op in swap_operations): # Save the log to a file await save_log(log_result) tx_signature_str = log_result['value']['signature'] before_source_balance = 0 source_token_change = 0 i = 0 while i < len(logs): log_entry = logs[i] # Check if we found the 'order_id' if tr_details["order_id"] is None and "order_id" in log_entry: # Extract the order_id tr_details["order_id"] = log_entry.split(":")[-1].strip() tr_details["token_in"] = logs[i + 1].split(":")[-1].strip() tr_details["token_out"] = logs[i + 2].split(":")[-1].strip() # Look for the token change amounts after tokens have been found if "source_token_change" in log_entry: parts = log_entry.split(", ") for part in parts: if "source_token_change" in part: tr_details["amount_in"] = float(part.split(":")[-1].strip()) / 10 ** 6 # Assuming 6 decimals elif "destination_token_change" in part: tr_details["amount_out"] = float(part.split(":")[-1].strip()) / 10 ** 6 # Assuming 6 decimals i += 1 # calculate percentage swapped by digging before_source_balance, source_token_change and after_source_balance # "Program log: before_source_balance: 19471871, before_destination_balance: 0, amount_in: 19471871, expect_amount_out: 770877527, min_return: 763168752", # "Program log: after_source_balance: 0, after_destination_balance: 770570049", # "Program log: source_token_change: 19471871, destination_token_change: 770570049", if "before_source_balance" in log_entry: parts = log_entry.split(", ") for part in parts: if "before_source_balance" in part: before_source_balance = float(part.split(":")[-1].strip()) / 10 ** 6 if "source_token_change" in log_entry: parts = log_entry.split(", ") for part in parts: if "source_token_change" in part: source_token_change = float(part.split(":")[-1].strip()) / 10 ** 6 # GET DETAILS FROM TRANSACTION IF NOT FOUND IN LOGS try: if tr_details["token_in"] is None or tr_details["token_out"] is None or tr_details["amount_in"] == 0 or tr_details["amount_out"] == 0: logging.warning("Incomplete swap details found in logs. Getting details from transaction") tr_details = await SAPI.get_transaction_details_info(tx_signature_str, logs) # onlt needed if no details got if before_source_balance > 0 and source_token_change > 0: tr_details["percentage_swapped"] = (source_token_change / before_source_balance) * 100 #dirty fix for percentage > 100 (decimals 9 but expecting 6) if tr_details["percentage_swapped"] > 100: tr_details["percentage_swapped"] = tr_details["percentage_swapped"] / 1000 # update token info: ToDo: check, but already did # all_token_addresses = list(set([tr_details["token_in"], tr_details["token_out"]])) # await get_token_prices(all_token_addresses) try: token_in = SAPI.dex.TOKENS_INFO[tr_details["token_in"]] token_out = SAPI.dex.TOKENS_INFO[tr_details["token_out"]] tr_details["symbol_in"] = token_in.get('symbol') tr_details["symbol_out"] = token_out.get('symbol') tr_details['amount_in_USD'] = tr_details['amount_in'] * token_in.get('price', 0) tr_details['amount_out_USD'] = tr_details['amount_out'] * token_out.get('price', 0) except Exception as e: logging.error(f"Error fetching token prices: {e}") message_text = ( f"Swap detected: \n" f"{tr_details['amount_in_USD']:.2f} worth of {tr_details['symbol_in']} ({tr_details['percentage_swapped']:.2f}% ) swapped for " # ({tr_details['token_in']}) ({tr_details['token_out']}) f"{tr_details['symbol_out']} \n" ) await telegram_utils.send_telegram_message(message_text) await SAPI.follow_move(tr_details) await SAPI.save_token_info() except Exception as e: logging.error(f"Error aquiring log details and following: {e}") await telegram_utils.send_telegram_message(f"Not followed! Error following move.") except Exception as e: logging.error(f"Error processing log: {e}") PROCESSING_LOG = False return tr_details async def process_messages(websocket): try: while True: response = await websocket.recv() response_data = json.loads(response) logger.debug(f"Received response: {response_data}") if 'result' in response_data: new_sub_id = response_data['result'] if int(new_sub_id) > 1: subscription_id = new_sub_id logger.info(f"Subscription successful. New id: {subscription_id}") elif new_sub_id: logger.info(f"Existing subscription confirmed: {subscription_id}") else: return None return subscription_id elif 'params' in response_data: log = response_data['params']['result'] logger.debug(f"Received transaction log: {log}") asyncio.create_task(process_log(log)) else: logger.warning(f"Unexpected response: {response_data}") except websockets.exceptions.ConnectionClosedError as e: logger.error(f"Connection closed unexpectedly: {e}") # await send_telegram_message("Connection to Solana network was closed. Not listening for transactions right now. Attempting to reconnect...") pass except json.JSONDecodeError as e: logger.error(f"Failed to decode JSON: {e}") except Exception as e: logger.error(f"An unexpected error occurred: {e}") pk = None app = init_app() # app = init_app(follow_move_legacy) # Convert Flask app to ASGI asgi_app = WsgiToAsgi(app) async def restartable_task(task_func, *args, **kwargs): while True: try: await task_func(*args, **kwargs) except Exception as e: logging.error(f"Error in task {task_func.__name__}: {e}") await telegram_utils.send_telegram_message(f"Error in task {task_func.__name__}: {e}") await asyncio.sleep(5) # Wait before retrying async def main(): global bot, PROCESSING_LOG, pk pk = await get_pk() await telegram_utils.initialize() await telegram_utils.send_telegram_message("Solana Agent Started. Connecting to mainnet...") # Start the log processor asyncio.create_task(restartable_task(watch_for_new_logs)) # process_transaction if DO_WATCH_WALLET: asyncio.create_task(restartable_task(SAPI.wallet_watch_loop)) await SAPI.wallet_watch_loop() def run_asyncio_tasks(): asyncio.run(main()) if __name__ == '__main__': import multiprocessing # Start the asyncio tasks in a separate process process = multiprocessing.Process(target=run_asyncio_tasks) process.start() # Run the ASGI server uvicorn.run( "app:asgi_app", host="0.0.0.0", port=3001, log_level="info", # "debug" reload=True ) # Wait for the asyncio tasks to complete process.join()