Solana Trading for AI Agents: High-Speed Execution on the Fastest Chain

"""
Fetch Jupiter v6 quote for optimal swap routing.
Uses Purple Flea Wallet API to execute the swap on Solana.
"""
import aiohttp
import asyncio
from dataclasses import dataclass

# Token mints (Solana mainnet)
SOL_MINT  = "So11111111111111111111111111111111111111112"
USDC_MINT = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"
BONK_MINT = "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263"

JUPITER_API       = "https://quote-api.jup.ag/v6"
PURPLE_FLEA_API   = "https://purpleflea.com/wallet-api"
PURPLE_FLEA_KEY   = "YOUR_API_KEY"
AGENT_WALLET      = "YourSolanaWalletAddressHere"


@dataclass
class SwapQuote:
    input_mint: str
    output_mint: str
    in_amount: int           # lamports / token atoms
    out_amount: int          # minimum out
    price_impact_pct: float
    route_plan: list
    slippage_bps: int


async def get_jupiter_quote(
    input_mint: str,
    output_mint: str,
    amount_atoms: int,
    slippage_bps: int = 50,   # 0.5% slippage tolerance
    session: aiohttp.ClientSession = None,
) -> SwapQuote:
    """
    Query Jupiter v6 quote API.
    amount_atoms: amount in the token's base unit (lamports for SOL, 1e6 for USDC).
    """
    params = {
        "inputMint":       input_mint,
        "outputMint":      output_mint,
        "amount":          amount_atoms,
        "slippageBps":     slippage_bps,
        "onlyDirectRoutes": "false",
        "asLegacyTransaction": "false",
    }
    async with session.get(f"{JUPITER_API}/quote", params=params) as resp:
        data = await resp.json()

    return SwapQuote(
        input_mint=input_mint,
        output_mint=output_mint,
        in_amount=int(data["inAmount"]),
        out_amount=int(data["outAmount"]),
        price_impact_pct=float(data["priceImpactPct"]),
        route_plan=data["routePlan"],
        slippage_bps=slippage_bps,
    )


async def execute_swap_via_purple_flea(
    quote: SwapQuote,
    session: aiohttp.ClientSession,
) -> dict:
    """
    Send the swap to Purple Flea Wallet API for signing and submission.
    Purple Flea handles key management, priority fees, and retry logic.
    """
    payload = {
        "wallet":       AGENT_WALLET,
        "chain":        "solana",
        "action":       "swap",
        "input_mint":   quote.input_mint,
        "output_mint":  quote.output_mint,
        "amount":       quote.in_amount,
        "slippage_bps": quote.slippage_bps,
        "priority":     "high",  # auto-sets priority fee
    }
    headers = {"X-API-Key": PURPLE_FLEA_KEY, "Content-Type": "application/json"}

    async with session.post(
        f"{PURPLE_FLEA_API}/solana/swap",
        json=payload, headers=headers
    ) as resp:
        result = await resp.json()

    return result  # { "signature": "...", "confirmed": true, "slot": 12345 }


async def main():
    async with aiohttp.ClientSession() as session:
        # Swap 1 SOL → USDC
        sol_amount_lamports = 1_000_000_000  # 1 SOL = 1e9 lamports

        quote = await get_jupiter_quote(
            input_mint=SOL_MINT,
            output_mint=USDC_MINT,
            amount_atoms=sol_amount_lamports,
            slippage_bps=30,  # 0.3% slippage tolerance
            session=session,
        )

        usdc_received = quote.out_amount / 1_000_000  # USDC has 6 decimals
        print(f"Quote: 1 SOL → {usdc_received:.2f} USDC")
        print(f"Price impact: {quote.price_impact_pct:.4f}%")
        print(f"Route hops: {len(quote.route_plan)}")

        # Only execute if price impact is acceptable
        if quote.price_impact_pct < 0.5:
            result = await execute_swap_via_purple_flea(quote, session)
            print(f"Swap confirmed! Signature: {result['signature']}")
            print(f"Slot: {result['slot']}")
        else:
            print(f"Price impact too high ({quote.price_impact_pct:.2f}%), skipping.")


asyncio.run(main())

"""
Solana Momentum Trading Agent — Purple Flea
Monitors SOL price, takes positions via Jupiter through Purple Flea Wallet API.
Manages stop-loss, take-profit, and position sizing automatically.
"""
import asyncio
import aiohttp
import logging
from datetime import datetime, timezone
from enum import Enum
from dataclasses import dataclass, field
from collections import deque

log = logging.getLogger("sol_agent")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

PURPLE_FLEA_API = "https://purpleflea.com/wallet-api"
PURPLE_FLEA_KEY = "YOUR_API_KEY"
AGENT_WALLET    = "YourSolanaPublicKey"

SOL_MINT  = "So11111111111111111111111111111111111111112"
USDC_MINT = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"


class Signal(Enum):
    BUY  = "buy"
    SELL = "sell"
    HOLD = "hold"


@dataclass
class Position:
    entry_price: float
    sol_amount: float
    usdc_spent: float
    entry_time: datetime
    stop_loss: float    # e.g. entry * 0.97
    take_profit: float  # e.g. entry * 1.04
    tx_signature: str = ""


@dataclass
class AgentState:
    usdc_balance: float = 1000.0
    sol_balance: float = 0.0
    position: Position | None = None
    price_history: deque = field(default_factory=lambda: deque(maxlen=50))
    trades: list = field(default_factory=list)
    total_pnl: float = 0.0


class SolanaAgent:
    def __init__(
        self,
        initial_usdc: float = 1000.0,
        position_size_pct: float = 0.25,  # Use 25% of balance per trade
        stop_loss_pct: float = 0.03,      # 3% stop loss
        take_profit_pct: float = 0.05,    # 5% take profit
        ema_short: int = 9,
        ema_long: int = 21,
        check_interval: int = 60,         # seconds between checks
    ):
        self.state = AgentState(usdc_balance=initial_usdc)
        self.position_size_pct = position_size_pct
        self.stop_loss_pct = stop_loss_pct
        self.take_profit_pct = take_profit_pct
        self.ema_short = ema_short
        self.ema_long = ema_long
        self.check_interval = check_interval
        self.session: aiohttp.ClientSession | None = None

    def _compute_ema(self, prices: list[float], period: int) -> float:
        """Exponential moving average."""
        if len(prices) < period:
            return prices[-1] if prices else 0
        k = 2 / (period + 1)
        ema = prices[0]
        for price in prices[1:]:
            ema = price * k + ema * (1 - k)
        return ema

    def _generate_signal(self) -> Signal:
        """EMA crossover signal: EMA9 crosses above EMA21 → BUY."""
        prices = list(self.state.price_history)
        if len(prices) < self.ema_long:
            return Signal.HOLD

        ema_s = self._compute_ema(prices, self.ema_short)
        ema_l = self._compute_ema(prices, self.ema_long)

        ema_s_prev = self._compute_ema(prices[:-1], self.ema_short)
        ema_l_prev = self._compute_ema(prices[:-1], self.ema_long)

        bullish_cross = ema_s_prev < ema_l_prev and ema_s > ema_l
        bearish_cross = ema_s_prev > ema_l_prev and ema_s < ema_l

        if bullish_cross:
            return Signal.BUY
        elif bearish_cross:
            return Signal.SELL
        return Signal.HOLD

    async def _get_sol_price(self) -> float:
        """Fetch SOL price from Purple Flea Trading API."""
        async with self.session.get(
            f"{PURPLE_FLEA_API}/price",
            params={"chain": "solana", "token": "SOL"},
            headers={"X-API-Key": PURPLE_FLEA_KEY},
        ) as resp:
            data = await resp.json()
        return data["price_usd"]

    async def _execute_buy(self, current_price: float):
        usdc_to_spend = self.state.usdc_balance * self.position_size_pct
        usdc_atoms = int(usdc_to_spend * 1_000_000)  # USDC has 6 decimals

        payload = {
            "wallet": AGENT_WALLET, "chain": "solana",
            "action": "swap", "input_mint": USDC_MINT,
            "output_mint": SOL_MINT, "amount": usdc_atoms,
            "slippage_bps": 50, "priority": "high",
        }
        async with self.session.post(
            f"{PURPLE_FLEA_API}/solana/swap", json=payload,
            headers={"X-API-Key": PURPLE_FLEA_KEY}
        ) as resp:
            result = await resp.json()

        sol_received = usdc_to_spend / current_price
        self.state.position = Position(
            entry_price=current_price,
            sol_amount=sol_received,
            usdc_spent=usdc_to_spend,
            entry_time=datetime.now(timezone.utc),
            stop_loss=current_price * (1 - self.stop_loss_pct),
            take_profit=current_price * (1 + self.take_profit_pct),
            tx_signature=result.get("signature", ""),
        )
        self.state.usdc_balance -= usdc_to_spend
        log.info(f"BUY {sol_received:.4f} SOL @ ${current_price:.2f} | tx: {result.get('signature', '')[:16]}...")

    async def _execute_sell(self, current_price: float, reason: str):
        pos = self.state.position
        sol_atoms = int(pos.sol_amount * 1_000_000_000)  # SOL has 9 decimals

        payload = {
            "wallet": AGENT_WALLET, "chain": "solana",
            "action": "swap", "input_mint": SOL_MINT,
            "output_mint": USDC_MINT, "amount": sol_atoms,
            "slippage_bps": 50, "priority": "high",
        }
        async with self.session.post(
            f"{PURPLE_FLEA_API}/solana/swap", json=payload,
            headers={"X-API-Key": PURPLE_FLEA_KEY}
        ) as resp:
            result = await resp.json()

        usdc_received = pos.sol_amount * current_price
        pnl = usdc_received - pos.usdc_spent
        pnl_pct = pnl / pos.usdc_spent * 100
        self.state.usdc_balance += usdc_received
        self.state.total_pnl += pnl
        self.state.trades.append({
            "reason": reason, "pnl": pnl, "pnl_pct": pnl_pct,
            "time": datetime.now(timezone.utc).isoformat(),
        })
        self.state.position = None
        log.info(f"SELL [{reason}] @ ${current_price:.2f} | PnL: ${pnl:.2f} ({pnl_pct:+.2f}%)")

    async def run(self):
        self.session = aiohttp.ClientSession()
        log.info("Solana agent started. Waiting for signals...")

        try:
            while True:
                price = await self._get_sol_price()
                self.state.price_history.append(price)
                pos = self.state.position

                if pos:
                    # Check stop-loss / take-profit first
                    if price <= pos.stop_loss:
                        await self._execute_sell(price, "stop_loss")
                    elif price >= pos.take_profit:
                        await self._execute_sell(price, "take_profit")
                    else:
                        signal = self._generate_signal()
                        if signal == Signal.SELL:
                            await self._execute_sell(price, "signal")
                else:
                    signal = self._generate_signal()
                    if signal == Signal.BUY and self.state.usdc_balance > 50:
                        await self._execute_buy(price)

                log.debug(f"SOL: ${price:.2f} | USDC: ${self.state.usdc_balance:.2f} | Total PnL: ${self.state.total_pnl:.2f}")
                await asyncio.sleep(self.check_interval)

        finally:
            await self.session.close()


if __name__ == "__main__":
    agent = SolanaAgent(
        initial_usdc=1000.0,
        position_size_pct=0.25,
        stop_loss_pct=0.03,
        take_profit_pct=0.05,
        check_interval=60,
    )
    asyncio.run(agent.run())