popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
Dobromir Popov 2024-11-06 13:27:19 +02:00
parent efbe12f173
commit efd7d8df5a
2 changed files with 3 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
crypto/sol/modules/SolanaAPI.py
View File

@ -76,10 +76,6 @@ solana_client = AsyncClient(SOLANA_HTTP_URL)
dexscreener_client = DexscreenerClient() dexscreener_client = DexscreenerClient()
class SolanaWS: class SolanaWS:
def __init__(self, on_message: Optional[callable] = None): def __init__(self, on_message: Optional[callable] = None):
self.websocket = None self.websocket = None
@ -411,9 +407,6 @@ class SolanaAPI:
logging.error(f"Error fetching token metadata for {mint_address}: {str(e)}") logging.error(f"Error fetching token metadata for {mint_address}: {str(e)}")
return None return None
async def get_transaction_details_rpc(self, tx_signature, readfromDump=False): async def get_transaction_details_rpc(self, tx_signature, readfromDump=False):
try: try:
@ -564,7 +557,6 @@ class SolanaAPI:
logging.error(f"Error calculating percentage swapped: {e}") logging.error(f"Error calculating percentage swapped: {e}")
return tr_info return tr_info
# def _get_pre_balance(transaction_details: Dict[str, Any], token: str) -> float: # def _get_pre_balance(transaction_details: Dict[str, Any], token: str) -> float:
# pre_balances = transaction_details.get('meta', {}).get('preTokenBalances', []) # pre_balances = transaction_details.get('meta', {}).get('preTokenBalances', [])
# for balance in pre_balances: # for balance in pre_balances:
@ -589,8 +581,7 @@ class SolanaAPI:
if backoff: if backoff:
retry_delay = retry_delay * 1.2 retry_delay = retry_delay * 1.2
return tx_details return tx_details
async def get_swap_transaction_details(self, tx_signature_str): async def get_swap_transaction_details(self, tx_signature_str):
t = await self.get_transaction(Signature.from_string(tx_signature_str), max_supported_transaction_version=0) t = await self.get_transaction(Signature.from_string(tx_signature_str), max_supported_transaction_version=0)
try: try:
@ -632,7 +623,6 @@ class SolanaAPI:
logging.error(f"Error fetching transaction details: {e}") logging.error(f"Error fetching transaction details: {e}")
return None return None
async def get_token_balance_rpc(self, wallet_address, token_address): async def get_token_balance_rpc(self, wallet_address, token_address):
try: try:
@ -855,7 +845,6 @@ class SolanaAPI:
except Exception as e: except Exception as e:
logging.error(f"Error following move: {e}") logging.error(f"Error following move: {e}")
async def calculate_priority_fee(async_client, priority_level=5): async def calculate_priority_fee(async_client, priority_level=5):
recent_fees = await async_client.get_recent_prioritization_fees() recent_fees = await async_client.get_recent_prioritization_fees()
@ -881,6 +870,7 @@ class SolanaAPI:
# Set minimum fee to avoid too low values # Set minimum fee to avoid too low values
return max(final_fee, 100) # minimum 100 microlamports return max(final_fee, 100) # minimum 100 microlamports
class SolanaDEX: class SolanaDEX:
def __init__(self, DISPLAY_CURRENCY: str): def __init__(self, DISPLAY_CURRENCY: str):
self.DISPLAY_CURRENCY = DISPLAY_CURRENCY self.DISPLAY_CURRENCY = DISPLAY_CURRENCY
@ -1201,7 +1191,5 @@ class SolanaDEX:
return obj.get(prop, 'N/A') return obj.get(prop, 'N/A')
DEX = SolanaDEX(DISPLAY_CURRENCY) DEX = SolanaDEX(DISPLAY_CURRENCY)
SAPI = SolanaAPI( on_initial_subscription_callback=DEX.list_initial_wallet_states(FOLLOWED_WALLET,YOUR_WALLET)) SAPI = SolanaAPI( on_initial_subscription_callback=DEX.list_initial_wallet_states(FOLLOWED_WALLET,YOUR_WALLET))

2
crypto/sol/modules/log_processor.py
View File

@ -4,7 +4,7 @@ from pathlib import Path
from .storage import store_transaction, prisma_client from .storage import store_transaction, prisma_client
from .SolanaAPI import SolanaAPI from .SolanaAPI import SolanaAPI
LOG_DIRECTORY = "logs" LOG_DIRECTORY = "./logs"
FILE_MASK = "wh_*.json" FILE_MASK = "wh_*.json"
async def process_log_file(file_path): async def process_log_file(file_path):