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

parsing logs only - no details

Browse Source
This commit is contained in:
Dobromir Popov 2024-10-05 21:49:32 +03:00
parent 019e9f66d4
commit 52c4877ed4
1 changed files with 96 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
crypto/sol/app.py
View File

@ -644,7 +644,6 @@ async def process_log(log_result):
tx_signature_str = log_result['value']['signature'] tx_signature_str = log_result['value']['signature']
logs = log_result['value']['logs'] logs = log_result['value']['logs']
try: try:
# Detect swap operations in logs # Detect swap operations in logs
swap_operations = ['Program log: Instruction: Swap', 'Program log: Instruction: Swap2'] swap_operations = ['Program log: Instruction: Swap', 'Program log: Instruction: Swap2']
@ -652,49 +651,64 @@ async def process_log(log_result):
for log_entry in logs: for log_entry in logs:
if any(op in log_entry for op in swap_operations): if any(op in log_entry for op in swap_operations):
try: try:
watched_tokens = await get_non_zero_token_balances(FOLLOWED_WALLET) watched_tokens = await get_non_zero_token_balances(FOLLOWED_WALLET)
details = parse_swap_logs(logs, watched_tokens) details = parse_swap_logs(logs)
transaction = await get_transaction_details_rpc(tx_signature_str, True) # transaction = await get_transaction_details_rpc(tx_signature_str, True)
tokens = [] # tokens = []
# Check inner instructions for transfers and mints # source_token = None
for instruction_set in transaction.get('meta', {}).get('innerInstructions', []): # target_token = None
for instruction in instruction_set.get('instructions', []):
if 'parsed' in instruction and 'info' in instruction['parsed']:
info = instruction['parsed']['info']
amount = None
mint = 'Unknown'
# Check for amount in transfer and transferChecked instructions # # Check inner instructions for transfers and mints
if 'amount' in info: # for instruction_set in transaction.get('meta', {}).get('innerInstructions', []):
amount = info['amount'] # for instruction in instruction_set.get('instructions', []):
elif 'tokenAmount' in info and 'amount' in info['tokenAmount']: # if 'parsed' in instruction and 'info' in instruction['parsed']:
amount = info['tokenAmount']['amount'] # info = instruction['parsed']['info']
# amount = None
# mint = 'Unknown'
# Get mint if available # # Check for amount in transfer and transferChecked instructions
if 'mint' in info: # if 'amount' in info:
mint = info['mint'] # amount = info['amount']
# elif 'tokenAmount' in info and 'amount' in info['tokenAmount']:
# amount = info['tokenAmount']['amount']
if amount is not None: # # Get mint if available
tokens.append({'amount': amount, 'mint': mint}) # if 'mint' in info:
# mint = info['mint']
# Check post token balances for final token states # if amount is not None:
for balance in transaction.get('postTokenBalances', []): # tokens.append({'amount': amount, 'mint': mint})
amount = balance['uiTokenAmount']['amount']
mint = balance['mint'] # # Identify source and target tokens
tokens.append({'amount': amount, 'mint': mint}) # if 'source' in info:
# source_token = info['source']
# if 'destination' in info:
# target_token = info['destination']
# # Check post token balances for final token states
# for balance in transaction.get('postTokenBalances', []):
# amount = balance['uiTokenAmount']['amount']
# mint = balance['mint']
# tokens.append({'amount': amount, 'mint': mint})
# Get amount_in, amount_out, tokens, and USD value # Get amount_in, amount_out, tokens, and USD value
swap_details = { swap_details = {
'amount_in': details['total_amount_in'], 'amount_in': details['total_amount_in'],
'amount_out': details['total_amount_out'], 'amount_out': details['total_amount_out'],
'tokens': tokens 'tokens': tokens,
'source_token': source_token,
'target_token': target_token
} }
message_text = ( message_text = (
f"Swap detected:\n" f"Swap detected:\n"
f"Amount In: {swap_details['amount_in']}\n" f"Amount In: {swap_details['amount_in']}\n"
f"Amount Out: {swap_details['amount_out']}\n" f"Amount Out: {swap_details['amount_out']}\n"
f"Source Token: {swap_details['source_token']}\n"
f"Target Token: {swap_details['target_token']}\n"
f"Tokens: {tokens}" f"Tokens: {tokens}"
) )
@ -708,39 +722,59 @@ async def process_log(log_result):
except Exception as e: except Exception as e:
logging.error(f"Error processing log: {e}") logging.error(f"Error processing log: {e}")
def parse_swap_logs(logs, watched_tokens):
total_amount_in = 0
total_amount_out = 0
token_addresses = []
# "Program log: Instruction: Swap2",
# "Program log: order_id: 13985890735038016",
# "Program log: AbrMJWfDVRZ2EWCQ1xSCpoVeVgZNpq1U2AoYG98oRXfn", source
# "Program log: EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", target
def parse_swap_logs(logs):
# Initialize variables to store the details
token_in = None
token_out = None
amount_in = 0
amount_out_expected = 0
amount_out_actual = 0
# Parse through each log entry
for log in logs: for log in logs:
if "SwapEvent" in log: # Check for source and target tokens
parts = log.split("{ ")[1].strip(" }").split(", ") if "Program log:" in log:
event_details = {} if "Swap2" in log:
for part in parts: # This log indicates the start of a swap, resetting details
key, value = part.split(": ") token_in = None
event_details[key.strip()] = value.strip() token_out = None
elif "order_id" in log:
order_id = log.split("order_id: ")[-1]
else:
# Check for source and target tokens
if not token_in:
token_in = log.split("Program log: ")[-1].strip()
elif not token_out:
token_out = log.split("Program log: ")[-1].strip()
# Aggregate amounts # Example assuming token changes can be parsed as "source_token_change:" and "destination_token_change:"
total_amount_in += int(event_details.get("amount_in", 0)) elif "source_token_change:" in log:
total_amount_out += int(event_details.get("amount_out", 0))
if "source_token_change:" in log:
# Extract final source and destination token changes
changes = log.split(", ") changes = log.split(", ")
for change in changes: for change in changes:
key_value = change.split(": ", 1) if "source_token_change" in change:
if len(key_value) != 2: amount_in = int(change.split(": ")[-1])
continue elif "destination_token_change" in change:
key, value = key_value amount_out_expected = int(change.split(": ")[-1])
if key == "source_token_change":
total_amount_in = int(value)
elif key == "destination_token_change":
total_amount_out = int(value)
# Assuming amount_out_actual is derived in a similar way to amount_out_expected
amount_out_actual = amount_out_expected # Modify if there is a separate way to determine actual amount
# Return parsed details as a dictionary
return { return {
"total_amount_in": total_amount_in, "token_in": token_in,
"total_amount_out": total_amount_out, "token_out": token_out,
"amount_in": amount_in,
"amount_out_expected": amount_out_expected,
"amount_out_actual": amount_out_actual,
"amount_in_USD": amount_out_actual, # Assuming conversion logic elsewhere
"amount_out_USD": amount_out_actual, # Assuming conversion logic elsewhere
} }
async def on_logs(log): async def on_logs(log):