solanaarbitragedetector-agu7bj/app/parsers/raydium.py

"""
Parser for Raydium AMM instructions
"""
import base64
from typing import Dict, List, Optional, Tuple, Any

from app.parsers.base import SwapParser, LiquidityParser


# Raydium Program IDs
RAYDIUM_LIQUIDITY_PROGRAM_ID = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"
RAYDIUM_SWAP_PROGRAM_ID = "9qvG1zUp8xF1Bi4m6UdRNby1BAAuaDrUxSpv4CmRRMjL"


class RaydiumSwapParser(SwapParser):
    """
    Parser for Raydium swap instructions
    """
    
    @property
    def program_id(self) -> str:
        return RAYDIUM_SWAP_PROGRAM_ID
    
    def parse_instruction(self, instruction: Dict, accounts: List[str], instruction_data: bytes) -> Dict:
        """
        Parse a Raydium swap instruction
        """
        if not instruction_data:
            return {"type": "unknown", "error": "No instruction data"}
            
        # Get accounts referenced by the instruction
        instruction_accounts = instruction.get("accounts", [])
        referenced_accounts = [accounts[idx] for idx in instruction_accounts if idx < len(accounts)]
        
        # For Raydium swap, the account list typically follows this pattern:
        # 0: Token program
        # 1: AMM ID
        # 2: AMM authority
        # 3: AMM open orders
        # 4: AMM target orders
        # 5: Pool coin token account
        # 6: Pool pc token account
        # 7: Serum program
        # 8: Serum market
        # 9: Serum bids
        # 10: Serum asks
        # 11: Serum event queue
        # 12: Serum coin vault
        # 13: Serum pc vault
        # 14: Serum vault signer
        # 15: User source token account
        # 16: User destination token account
        # 17: User owner
        
        # Basic metadata about the instruction
        result = {
            "type": "raydium_swap",
            "program": "raydium_swap",
            "program_id": RAYDIUM_SWAP_PROGRAM_ID,
            "accounts": referenced_accounts,
            "data": base64.b64encode(instruction_data).decode("utf-8"),
        }
        
        # Add specific account references if available
        if len(referenced_accounts) >= 18:
            result.update({
                "amm_id": referenced_accounts[1],
                "pool_coin_account": referenced_accounts[5],
                "pool_pc_account": referenced_accounts[6],
                "serum_market": referenced_accounts[8],
                "user_source_account": referenced_accounts[15],
                "user_destination_account": referenced_accounts[16],
                "user_owner": referenced_accounts[17],
            })
            
        # Parse instruction data to get amount_in
        # Note: This is a simplified version. Actual layout may vary.
        try:
            if len(instruction_data) >= 9:
                # First byte is instruction type, next 8 bytes are the amount
                amount_in = int.from_bytes(instruction_data[1:9], byteorder="little")
                result["amount_in"] = amount_in
                
                # For demonstration - in a real implementation, we would parse
                # minimum amount out as well
        except Exception as e:
            result["parse_error"] = str(e)
            
        return result
    
    def extract_swap_info(self, parsed_instruction: Dict) -> Tuple[Optional[str], Optional[str], Optional[float], Optional[float]]:
        """
        Extract swap information from a parsed instruction
        
        Returns:
            Tuple of (token_in_address, token_out_address, amount_in, amount_out)
        """
        if parsed_instruction.get("type") != "raydium_swap":
            return None, None, None, None
            
        # For Raydium, we don't directly get token mint addresses from the instruction
        # We would need to look up the token accounts' mint addresses
        source_account = parsed_instruction.get("user_source_account")
        destination_account = parsed_instruction.get("user_destination_account")
        amount_in = parsed_instruction.get("amount_in")
        
        # We don't know the amount_out from just the instruction
        return None, source_account, destination_account, amount_in


class RaydiumLiquidityParser(LiquidityParser):
    """
    Parser for Raydium liquidity instructions
    """
    
    @property
    def program_id(self) -> str:
        return RAYDIUM_LIQUIDITY_PROGRAM_ID
    
    def parse_instruction(self, instruction: Dict, accounts: List[str], instruction_data: bytes) -> Dict:
        """
        Parse a Raydium liquidity instruction
        """
        if not instruction_data:
            return {"type": "unknown", "error": "No instruction data"}
            
        # First byte is the instruction type
        instruction_type = instruction_data[0]
        
        # Get accounts referenced by the instruction
        instruction_accounts = instruction.get("accounts", [])
        referenced_accounts = [accounts[idx] for idx in instruction_accounts if idx < len(accounts)]
        
        # Basic metadata about the instruction
        result = {
            "program": "raydium_liquidity",
            "program_id": RAYDIUM_LIQUIDITY_PROGRAM_ID,
            "accounts": referenced_accounts,
            "data": base64.b64encode(instruction_data).decode("utf-8"),
        }
        
        # Parse specific instruction types
        if instruction_type == 1:  # Deposit liquidity
            result["type"] = "raydium_deposit_liquidity"
        elif instruction_type == 2:  # Withdraw liquidity
            result["type"] = "raydium_withdraw_liquidity"
        else:
            result["type"] = f"raydium_instruction_{instruction_type}"
            
        return result
    
    def extract_liquidity_info(self, parsed_instruction: Dict) -> Dict[str, Any]:
        """
        Extract liquidity information from a parsed instruction
        """
        result = {
            "action": None,
            "pool_address": None,
            "tokens": [],
            "amounts": [],
            "user": None,
        }
        
        if parsed_instruction.get("type") == "raydium_deposit_liquidity":
            result["action"] = "add"
        elif parsed_instruction.get("type") == "raydium_withdraw_liquidity":
            result["action"] = "remove"
        else:
            return result
            
        # In a full implementation, we would extract pool address, tokens, amounts
        # and user from the instruction accounts and data
        
        return result