new_features_examples/solana_agent.py

from dataclasses import dataclass
from typing import List, Optional, Dict, Any
from datetime import datetime
import asyncio
from loguru import logger
import json
import base58
from decimal import Decimal

# Swarms imports
from swarms import Agent

# Solana imports
from solders.rpc.responses import GetTransactionResp
from solders.transaction import Transaction
from anchorpy import Provider, Wallet
from solders.keypair import Keypair
import aiohttp

# Specialized Solana Analysis System Prompt
SOLANA_ANALYSIS_PROMPT = """You are a specialized Solana blockchain analyst agent. Your role is to:

1. Analyze real-time Solana transactions for patterns and anomalies
2. Identify potential market-moving transactions and whale movements
3. Detect important DeFi interactions across major protocols
4. Monitor program interactions for suspicious or notable activity
5. Track token movements across significant protocols like:
   - Serum DEX
   - Raydium
   - Orca
   - Marinade
   - Jupiter
   - Other major Solana protocols

When analyzing transactions, consider:
- Transaction size relative to protocol norms
- Historical patterns for involved addresses
- Impact on protocol liquidity
- Relationship to known market events
- Potential wash trading or suspicious patterns
- MEV opportunities and arbitrage patterns
- Program interaction sequences

Provide analysis in the following format:
{
    "analysis_type": "[whale_movement|program_interaction|defi_trade|suspicious_activity]",
    "severity": "[high|medium|low]",
    "details": {
        "transaction_context": "...",
        "market_impact": "...",
        "recommended_actions": "...",
        "related_patterns": "..."
    }
}

Focus on actionable insights that could affect:
1. Market movements
2. Protocol stability
3. Trading opportunities
4. Risk management
"""


@dataclass
class TransactionData:
    """Data structure for parsed Solana transaction information"""

    signature: str
    block_time: datetime
    slot: int
    fee: int
    lamports: int
    from_address: str
    to_address: str
    program_id: str
    instruction_data: Optional[str] = None
    program_logs: List[str] = None

    @property
    def sol_amount(self) -> Decimal:
        """Convert lamports to SOL"""
        return Decimal(self.lamports) / Decimal(1e9)

    def to_dict(self) -> Dict[str, Any]:
        """Convert transaction data to dictionary for agent analysis"""
        return {
            "signature": self.signature,
            "timestamp": self.block_time.isoformat(),
            "slot": self.slot,
            "fee": self.fee,
            "amount_sol": str(self.sol_amount),
            "from_address": self.from_address,
            "to_address": self.to_address,
            "program_id": self.program_id,
            "instruction_data": self.instruction_data,
            "program_logs": self.program_logs,
        }


class SolanaSwarmAgent:
    """Intelligent agent for analyzing Solana transactions using swarms"""

    def __init__(
        self,
        agent_name: str = "Solana-Analysis-Agent",
        model_name: str = "gpt-4",
    ):
        self.agent = Agent(
            agent_name=agent_name,
            system_prompt=SOLANA_ANALYSIS_PROMPT,
            model_name=model_name,
            max_loops=1,
            autosave=True,
            dashboard=False,
            verbose=True,
            dynamic_temperature_enabled=True,
            saved_state_path="solana_agent.json",
            user_name="solana_analyzer",
            retry_attempts=3,
            context_length=4000,
        )

        # Initialize known patterns database
        self.known_patterns = {
            "whale_addresses": set(),
            "program_interactions": {},
            "recent_transactions": [],
        }
        logger.info(
            f"Initialized {agent_name} with specialized Solana analysis capabilities"
        )

    async def analyze_transaction(
        self, tx_data: TransactionData
    ) -> Dict[str, Any]:
        """Analyze a transaction using the specialized agent"""
        try:
            # Update recent transactions for pattern analysis
            self.known_patterns["recent_transactions"].append(
                tx_data.signature
            )
            if len(self.known_patterns["recent_transactions"]) > 1000:
                self.known_patterns["recent_transactions"].pop(0)

            # Prepare context for agent
            context = {
                "transaction": tx_data.to_dict(),
                "known_patterns": {
                    "recent_similar_transactions": [
                        tx
                        for tx in self.known_patterns[
                            "recent_transactions"
                        ][-5:]
                        if abs(
                            TransactionData(tx).sol_amount
                            - tx_data.sol_amount
                        )
                        < 1
                    ],
                    "program_statistics": self.known_patterns[
                        "program_interactions"
                    ].get(tx_data.program_id, {}),
                },
            }

            # Get analysis from agent
            analysis = await self.agent.run_async(
                f"Analyze the following Solana transaction and provide insights: {json.dumps(context, indent=2)}"
            )

            # Update pattern database
            if tx_data.sol_amount > 1000:  # Track whale addresses
                self.known_patterns["whale_addresses"].add(
                    tx_data.from_address
                )

            # Update program interaction statistics
            if (
                tx_data.program_id
                not in self.known_patterns["program_interactions"]
            ):
                self.known_patterns["program_interactions"][
                    tx_data.program_id
                ] = {"total_interactions": 0, "total_volume": 0}
            self.known_patterns["program_interactions"][
                tx_data.program_id
            ]["total_interactions"] += 1
            self.known_patterns["program_interactions"][
                tx_data.program_id
            ]["total_volume"] += float(tx_data.sol_amount)

            return json.loads(analysis)

        except Exception as e:
            logger.error(f"Error in agent analysis: {str(e)}")
            return {
                "analysis_type": "error",
                "severity": "low",
                "details": {
                    "error": str(e),
                    "transaction": tx_data.signature,
                },
            }


class SolanaTransactionMonitor:
    """Main class for monitoring and analyzing Solana transactions"""

    def __init__(
        self,
        rpc_url: str,
        swarm_agent: SolanaSwarmAgent,
        min_sol_threshold: Decimal = Decimal("100"),
    ):
        self.rpc_url = rpc_url
        self.swarm_agent = swarm_agent
        self.min_sol_threshold = min_sol_threshold
        self.wallet = Wallet(Keypair())
        self.provider = Provider(rpc_url, self.wallet)
        logger.info("Initialized Solana transaction monitor")

    async def parse_transaction(
        self, tx_resp: GetTransactionResp
    ) -> Optional[TransactionData]:
        """Parse transaction response into TransactionData object"""
        try:
            if not tx_resp.value:
                return None

            tx_value = tx_resp.value
            meta = tx_value.transaction.meta
            if not meta:
                return None

            tx: Transaction = tx_value.transaction.transaction

            # Extract transaction details
            from_pubkey = str(tx.message.account_keys[0])
            to_pubkey = str(tx.message.account_keys[1])
            program_id = str(tx.message.account_keys[-1])

            # Calculate amount from balance changes
            amount = abs(meta.post_balances[0] - meta.pre_balances[0])

            return TransactionData(
                signature=str(tx_value.transaction.signatures[0]),
                block_time=datetime.fromtimestamp(
                    tx_value.block_time or 0
                ),
                slot=tx_value.slot,
                fee=meta.fee,
                lamports=amount,
                from_address=from_pubkey,
                to_address=to_pubkey,
                program_id=program_id,
                program_logs=(
                    meta.log_messages if meta.log_messages else []
                ),
            )
        except Exception as e:
            logger.error(f"Failed to parse transaction: {str(e)}")
            return None

    async def start_monitoring(self):
        """Start monitoring for new transactions"""
        logger.info(
            "Starting transaction monitoring with swarm agent analysis"
        )

        async with aiohttp.ClientSession() as session:
            async with session.ws_connect(self.rpc_url) as ws:
                await ws.send_json(
                    {
                        "jsonrpc": "2.0",
                        "id": 1,
                        "method": "transactionSubscribe",
                        "params": [
                            {"commitment": "finalized"},
                            {
                                "encoding": "jsonParsed",
                                "commitment": "finalized",
                            },
                        ],
                    }
                )

                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        try:
                            data = json.loads(msg.data)
                            if "params" in data:
                                signature = data["params"]["result"][
                                    "value"
                                ]["signature"]

                                # Fetch full transaction data
                                tx_response = await self.provider.connection.get_transaction(
                                    base58.b58decode(signature)
                                )

                                if tx_response:
                                    tx_data = (
                                        await self.parse_transaction(
                                            tx_response
                                        )
                                    )
                                    if (
                                        tx_data
                                        and tx_data.sol_amount
                                        >= self.min_sol_threshold
                                    ):
                                        # Get agent analysis
                                        analysis = await self.swarm_agent.analyze_transaction(
                                            tx_data
                                        )

                                        logger.info(
                                            f"Transaction Analysis:\n"
                                            f"Signature: {tx_data.signature}\n"
                                            f"Amount: {tx_data.sol_amount} SOL\n"
                                            f"Analysis: {json.dumps(analysis, indent=2)}"
                                        )

                        except Exception as e:
                            logger.error(
                                f"Error processing message: {str(e)}"
                            )
                            continue


async def main():
    """Example usage"""

    # Start monitoring
    try:
        # Initialize swarm agent
        swarm_agent = SolanaSwarmAgent(
            agent_name="Solana-Whale-Detector", model_name="gpt-4"
        )

        # Initialize monitor
        monitor = SolanaTransactionMonitor(
            rpc_url="wss://api.mainnet-beta.solana.com",
            swarm_agent=swarm_agent,
            min_sol_threshold=Decimal("100"),
        )

        await monitor.start_monitoring()
    except KeyboardInterrupt:
        logger.info("Shutting down gracefully...")


if __name__ == "__main__":
    asyncio.run(main())