""" Parser for the Serum DEX v3 instructions """ import base64 from typing import Dict, List, Optional, Tuple from app.parsers.base import SwapParser # Serum DEX v3 Program ID SERUM_DEX_V3_PROGRAM_ID = "9xQeWvG816bUx9EPjHmaT23yvVM2ZWbrrpZb9PusVFin" # Serum DEX Instructions SERUM_DEX_MATCH_ORDERS = 2 SERUM_DEX_NEW_ORDER = 0 SERUM_DEX_CANCEL_ORDER = 1 class SerumDexParser(SwapParser): """ Parser for the Serum DEX v3 instructions """ @property def program_id(self) -> str: return SERUM_DEX_V3_PROGRAM_ID def parse_instruction(self, instruction: Dict, accounts: List[str], instruction_data: bytes) -> Dict: """ Parse a Serum DEX 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)] result = { "program": "serum_dex", "program_id": SERUM_DEX_V3_PROGRAM_ID, "accounts": referenced_accounts, "data": base64.b64encode(instruction_data).decode("utf-8"), } # Parse specific instruction types if instruction_type == SERUM_DEX_NEW_ORDER: # New order instruction result.update(self._parse_new_order(instruction_data, referenced_accounts)) elif instruction_type == SERUM_DEX_MATCH_ORDERS: # Match orders instruction result.update(self._parse_match_orders(instruction_data, referenced_accounts)) elif instruction_type == SERUM_DEX_CANCEL_ORDER: # Cancel order instruction result.update(self._parse_cancel_order(instruction_data, referenced_accounts)) else: # Unknown instruction type result.update({ "type": f"serum_instruction_{instruction_type}", "error": "Unknown instruction type", }) return result def _parse_new_order(self, instruction_data: bytes, accounts: List[str]) -> Dict: """ Parse a Serum DEX new order instruction """ # New order layout is complex, simplified for this example try: side = "buy" if instruction_data[1] == 0 else "sell" # Note: In a real implementation, we would parse: # - limit price # - max base qty # - max quote qty # - self trade behavior # - order type # - client order ID # For demo purposes we'll create a simplified result return { "type": "serum_new_order", "side": side, "market": accounts[0] if len(accounts) > 0 else None, "open_orders": accounts[1] if len(accounts) > 1 else None, "request_queue": accounts[2] if len(accounts) > 2 else None, "payer": accounts[3] if len(accounts) > 3 else None, "owner": accounts[4] if len(accounts) > 4 else None, "base_vault": accounts[5] if len(accounts) > 5 else None, "quote_vault": accounts[6] if len(accounts) > 6 else None, # Additional fields would be parsed from instruction_data } except Exception as e: return { "type": "serum_new_order", "error": f"Failed to parse new order: {str(e)}", } def _parse_match_orders(self, instruction_data: bytes, accounts: List[str]) -> Dict: """ Parse a Serum DEX match orders instruction """ # Simplified implementation for demo purposes try: # Real implementation would extract more fields return { "type": "serum_match_orders", "market": accounts[0] if len(accounts) > 0 else None, "request_queue": accounts[1] if len(accounts) > 1 else None, "event_queue": accounts[2] if len(accounts) > 2 else None, "bids": accounts[3] if len(accounts) > 3 else None, "asks": accounts[4] if len(accounts) > 4 else None, # Additional accounts would be included } except Exception as e: return { "type": "serum_match_orders", "error": f"Failed to parse match orders: {str(e)}", } def _parse_cancel_order(self, instruction_data: bytes, accounts: List[str]) -> Dict: """ Parse a Serum DEX cancel order instruction """ # Simplified implementation for demo purposes try: side = "buy" if instruction_data[1] == 0 else "sell" return { "type": "serum_cancel_order", "side": side, "market": accounts[0] if len(accounts) > 0 else None, "open_orders": accounts[1] if len(accounts) > 1 else None, "owner": accounts[2] if len(accounts) > 2 else None, # Additional fields would be parsed from instruction_data } except Exception as e: return { "type": "serum_cancel_order", "error": f"Failed to parse cancel order: {str(e)}", } 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) """ # For Serum, we need to look at both the new order and match orders instructions # to get a complete swap picture, which is more complex than can be done # with a single instruction. # This is a simplified implementation that would need to be enhanced if parsed_instruction.get("type") == "serum_match_orders": # We would need more context to determine the tokens and amounts # For now, return None values return None, None, None, None return None, None, None, None