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

SWAP details for native input

Browse Source
This commit is contained in:
Dobromir Popov 2024-11-05 20:08:03 +02:00
parent 0780878c50
commit 4773e74caf
1 changed files with 42 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
crypto/sol/modules/webui.py
View File

@ -136,42 +136,48 @@ def init_app(tr_handler=None):
usdcMint = LIQUIDITY_TOKENS[0]
tr = {}
if not token_inputs or len(token_inputs) == 0:
logging.info("Assumed USDC as first token. BUY transaction detected")
tr["token_in"] = usdcMint
tr["type"] = "BUY"
tr["amount_in"] = await SolanaAPI.DEX.get_token_prices(
token_outputs[0]["mint"],
int(token_outputs[0]["rawTokenAmount"]["tokenAmount"])
)
else:
token_in = token_inputs[0]
tr["token_in"] = token_in["mint"]
tr["token_in_decimals"] = token_in["rawTokenAmount"]["decimals"]
tr["amount_in"] = float(int(token_in["rawTokenAmount"]["tokenAmount"]) / 10**token_in["rawTokenAmount"]["decimals"])
# 'amount_in': float(token_inputs[0]['rawTokenAmount']['tokenAmount']) / 10**token_inputs[0]['rawTokenAmount']['decimals'],
try:
if not token_inputs or len(token_inputs) == 0:
logging.info("Assumed USDC as first token. BUY transaction detected")
tr["token_in"] = usdcMint
tr["type"] = "BUY"
tr["amount_in"] = await SolanaAPI.DEX.get_token_prices(
token_outputs[0]["mint"],
int(token_outputs[0]["rawTokenAmount"]["tokenAmount"])
)
else:
token_in = token_inputs[0]
tr["token_in"] = token_in["mint"]
tr["token_in_decimals"] = token_in["rawTokenAmount"]["decimals"]
tr["amount_in"] = float(int(token_in["rawTokenAmount"]["tokenAmount"]) / 10**token_in["rawTokenAmount"]["decimals"])
# 'amount_in': float(token_inputs[0]['rawTokenAmount']['tokenAmount']) / 10**token_inputs[0]['rawTokenAmount']['decimals'],
if not token_outputs or len(token_outputs) == 0:
logging.info("Assumed USDC as second token. SELL transaction detected")
tr["token_out"] = usdcMint
tr["type"] = "SELL"
tr["amount_out"] = await SolanaAPI.DEX.get_token_prices(
token_inputs[0]["mint"],
int(token_inputs[0]["rawTokenAmount"]["tokenAmount"])
)
else:
token_out = token_outputs[0]
tr["token_out"] = token_out["mint"]
tr["token_out_decimals"] = token_out["rawTokenAmount"]["decimals"]
tr["amount_out"] = float(int(token_out["rawTokenAmount"]["tokenAmount"]) / 10**token_out["rawTokenAmount"]["decimals"])
#'amount_out': float(token_outputs[0]['rawTokenAmount']['tokenAmount']) / 10**token_outputs[0]['rawTokenAmount']['decimals'],
if not token_outputs or len(token_outputs) == 0:
logging.info("Assumed USDC as second token. SELL transaction detected")
tr["token_out"] = usdcMint
tr["type"] = "SELL"
tr["amount_out"] = await SolanaAPI.DEX.get_token_prices(
token_inputs[0]["mint"],
int(token_inputs[0]["rawTokenAmount"]["tokenAmount"])
)
else:
token_out = token_outputs[0]
tr["token_out"] = token_out["mint"]
tr["token_out_decimals"] = token_out["rawTokenAmount"]["decimals"]
tr["amount_out"] = float(int(token_out["rawTokenAmount"]["tokenAmount"]) / 10**token_out["rawTokenAmount"]["decimals"])
#'amount_out': float(token_outputs[0]['rawTokenAmount']['tokenAmount']) / 10**token_outputs[0]['rawTokenAmount']['decimals'],
if swap_event.get('nativeInput'): # SOL
token_in = swap_event.get('nativeInput', [])
logger.info(f"Native input (SOL) detected ({token_in["amount"]})")
if not tr["token_in"] or not tr["token_out"] or tr["amount_in"] == 0 or tr["amount_out"] == 0:
logging.warning("Incomplete swap details found in logs. Getting details from transaction")
tx_signature = data[0].get('signature')
logs = data[0].get('logs', [])
tr = await SolanaAPI.SAPI.get_transaction_details_info(tx_signature, logs)
if not tr["token_in"] or not tr["token_out"] or tr["amount_in"] == 0 or tr["amount_out"] == 0:
logging.warning("Incomplete swap details found in logs. Getting details from transaction")
tx_signature = data[0].get('signature')
logs = data[0].get('logs', [])
tr = await SolanaAPI.SAPI.get_transaction_details_info(tx_signature, logs)
except Exception as e:
logging.error(f"Error loading transaction token data: {str(e)}")
wallet = data[0]['feePayer'] # Using feePayer as the wallet address
tx_signature = data[0]['signature']