Building a Gemini Trading Agent with Google ADK

import os, requests
from google.adk.tools import FunctionTool

PURPLEFLEA_KEY = os.getenv("PURPLE_FLEA_API_KEY")
BASE = "https://api.purpleflea.com"

def _headers():
    return {
        "Authorization": f"Bearer {PURPLEFLEA_KEY}",
        "Content-Type": "application/json",
    }

def get_market_price(market: str) -> dict:
    """Get the current mark price for a perpetual futures market.

    Args:
        market: The market symbol, e.g. BTC-PERP, ETH-PERP, SOL-PERP.

    Returns:
        A dict with 'market' and 'price' (USD float).
    """
    r = requests.get(
        BASE + "/v1/trading/price",
        headers=_headers(),
        params={"market": market},
    )
    return r.json()

def open_trade(market: str, side: str, size_usd: float, leverage: int = 5) -> dict:
    """Open a perpetual futures position on Hyperliquid via Purple Flea.
    No KYC required. Supports 275 markets, up to 50x leverage.

    Args:
        market: Market symbol, e.g. BTC-PERP, ETH-PERP, SOL-PERP.
        side: 'long' to bet price goes up, 'short' to bet price goes down.
        size_usd: Position size in USD (notional value).
        leverage: Leverage multiplier from 1 to 50 (default 5).

    Returns:
        Dict with position_id, entry_price, market, side, and size.
    """
    r = requests.post(
        BASE + "/v1/trading/open",
        headers=_headers(),
        json={
            "market":   market,
            "side":     side,
            "size":     size_usd,
            "leverage": leverage,
        },
    )
    return r.json()

def close_trade(position_id: str) -> dict:
    """Close an open perpetual futures position by its position ID.

    Args:
        position_id: The unique position ID returned by open_trade.

    Returns:
        Dict with pnl (profit/loss in USD), closed_at timestamp.
    """
    r = requests.post(
        BASE + "/v1/trading/close",
        headers=_headers(),
        json={"position_id": position_id},
    )
    return r.json()

def get_account_balance() -> dict:
    """Get the current USD balance and open positions for the account.

    Returns:
        Dict with usd_balance and list of open_positions.
    """
    return requests.get(BASE + "/api/v1/wallet/balance", headers=_headers()).json()

def create_wallet(chain: str) -> dict:
    """Create a new non-custodial HD wallet on a blockchain.

    Args:
        chain: Blockchain name: ethereum, bitcoin, solana, tron, bnb, polygon.

    Returns:
        Dict with address and wallet_id.
    """
    r = requests.post(
        "https://wallet.purpleflea.com/v1/wallet",
        headers=_headers(),
        json={"chain": chain},
    )
    return r.json()

# Wrap each function as an ADK FunctionTool
price_tool   = FunctionTool(func=get_market_price)
open_tool    = FunctionTool(func=open_trade)
close_tool   = FunctionTool(func=close_trade)
balance_tool = FunctionTool(func=get_account_balance)
wallet_tool  = FunctionTool(func=create_wallet)

from google.adk.agents import Agent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types
from purpleflea_tools import price_tool, open_tool, close_tool, balance_tool, wallet_tool

AGENT_NAME = "gemini-crypto-trader"
APP_NAME   = "purpleflea-trading"
USER_ID    = "agent-001"

root_agent = Agent(
    name=AGENT_NAME,
    model="gemini-2.0-flash",
    description="Autonomous crypto perpetual futures trading agent",
    instruction="""You are an autonomous crypto trading agent powered by Purple Flea.
You have access to Hyperliquid's 275 perpetual futures markets via the Purple Flea API.

Rules you must follow:
- Always call get_account_balance first to assess current USD balance.
- Never open a trade larger than 20% of the current balance.
- Maximum leverage is 10x unless the user explicitly requests higher.
- Always report the position_id when opening a trade.
- Report entry price, current P&L, and balance in every response.
- If the user asks about a market you don't know the symbol for,
  try ASSET-PERP format (e.g., DOGE-PERP, AVAX-PERP).
""",
    tools=[price_tool, open_tool, close_tool, balance_tool, wallet_tool],
)

# Run locally with InMemory session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
)

runner = Runner(
    agent=root_agent,
    app_name=APP_NAME,
    session_service=session_service,
)

def chat(message: str) -> str:
    content = types.Content(role="user",
                            parts=[types.Part(text=message)])
    events = runner.run(user_id=USER_ID,
                         session_id=session.id,
                         new_message=content)
    for event in events:
        if event.is_final_response():
            return event.content.parts[0].text
    return ""

if __name__ == "__main__":
    print(chat("Check my balance and open a $20 ETH long at 5x leverage."))
    print(chat("What is the BTC price and my unrealised P&L?"))

# Authenticate with Google Cloud
$ gcloud auth application-default login

# Deploy to Vertex AI (creates a managed endpoint)
$ adk deploy cloud_run \
    --project=your-gcp-project \
    --region=us-central1 \
    --service-name=gemini-crypto-trader \
    agent.py

# Test the deployed endpoint
$ adk web agent.py  # run local dev UI for testing first

def open_trade_with_referral(market: str, side: str,
                             size_usd: float, leverage: int = 5) -> dict:
    """Open a trade and earn 20% referral commission on trading fees."""
    r = requests.post(
        BASE + "/v1/trading/open",
        headers=_headers(),
        json={
            "market":        market,
            "side":          side,
            "size":          size_usd,
            "leverage":      leverage,
            "referral_code": "your-referral-code", # earn 20%
        },
    )
    return r.json()