Purple Flea for Anyscale

Why Ray + Purple Flea

Sequential trading agents are bottlenecked by network I/O and API rate limits per symbol. With Ray actors, you break out of that constraint — each actor manages a different market or strategy, all running simultaneously across a cluster. Purple Flea's per-agent API key model maps cleanly to this: each Ray actor is an independent agent with its own registered identity and wallet.

Ray Actor Pattern for Purple Flea

The Ray actor model maps cleanly to autonomous trading agents. Each actor maintains its own state (symbol, strategy parameters, position tracking), makes independent API calls to Purple Flea, and runs truly concurrently — not just asynchronously — across CPU cores or cluster nodes.

import ray
import requests
import os
import time
from typing import Optional

API_KEY = os.environ.get("PURPLE_FLEA_API_KEY", "")
PF_BASE = "https://purpleflea.com/api/v1"

@ray.remote
class TradingAgent:
    def __init__(self, agent_id: str, api_key: str, symbol: str, strategy: str = "momentum"):
        self.agent_id = agent_id
        self.api_key  = api_key
        self.symbol   = symbol
        self.strategy = strategy
        self.headers  = {"Authorization": f"Bearer {api_key}"}
        self.position = None
        self.pnl      = 0.0

    def get_price(self) -> Optional[dict]:
        try:
            resp = requests.get(
                f"{PF_BASE}/markets/{self.symbol}/price",
                headers=self.headers,
                timeout=5
            )
            return resp.json()
        except Exception:
            return None

    def check_and_trade(self) -> dict:
        # Get current market price and momentum
        price_data = self.get_price()
        if not price_data:
            return {"agent": self.agent_id, "error": "price fetch failed"}

        change_1h  = price_data.get("change_1h_pct", 0)
        change_24h = price_data.get("change_24h_pct", 0)
        price      = price_data.get("price", 0)

        action = "hold"
        trade  = None

        if self.strategy == "momentum":
            # Simple momentum: buy if price up 2%+ in last hour
            if change_1h > 2.0 and not self.position:
                trade = requests.post(
                    f"{PF_BASE}/trade",
                    json={"symbol": self.symbol, "side": "long", "size_usd": 50},
                    headers=self.headers, timeout=8
                ).json()
                self.position = trade.get("position_id")
                action = "buy"

            # Exit if reversal signal
            elif change_1h < -1.5 and self.position:
                trade = requests.delete(
                    f"{PF_BASE}/trade/{self.position}",
                    headers=self.headers, timeout=8
                ).json()
                self.pnl += trade.get("realized_pnl", 0)
                self.position = None
                action = "close"

        return {
            "agent":    self.agent_id,
            "symbol":   self.symbol,
            "price":    price,
            "change_1h": change_1h,
            "action":   action,
            "trade":    trade,
            "total_pnl": self.pnl,
        }

    def get_state(self) -> dict:
        return {
            "agent_id": self.agent_id,
            "symbol":   self.symbol,
            "position": self.position,
            "pnl":      self.pnl,
        }


# --- Launch 10 agents in parallel across different symbols ---

ray.init()  # or ray.init(address="auto") for Anyscale cluster

symbols = [
    "BTC-PERP", "ETH-PERP", "SOL-PERP", "AVAX-PERP", "ARB-PERP",
    "OP-PERP",  "MATIC-PERP", "DOGE-PERP", "APT-PERP", "SUI-PERP"
]

# Create agents — each is a persistent Ray actor with its own state
agents = [
    TradingAgent.remote(
        agent_id=f"agent-{i:03d}",
        api_key=API_KEY,
        symbol=sym,
        strategy="momentum"
    )
    for i, sym in enumerate(symbols)
]

# Run all agents simultaneously — true parallel execution
while True:
    # Fire all 10 agents at once, collect results concurrently
    futures = [agent.check_and_trade.remote() for agent in agents]
    results = ray.get(futures)

    # Aggregate results
    total_pnl = sum(r.get("total_pnl", 0) for r in results)
    active    = [r for r in results if r.get("action") != "hold"]

    print(f"Cycle complete | Active signals: {len(active)} | Total P&L: ${total_pnl:.2f}")
    for r in active:
        print(f"  {r['symbol']}: {r['action']} @ ${r['price']:,.2f} (1h: {r['change_1h']:+.1f}%)")

    time.sleep(60)  # poll every minute

Available Financial APIs

Every Purple Flea service is a REST endpoint callable from any Ray actor. Scale your agent fleet across all six services simultaneously.

Scaling Patterns

Ray's actor model supports several architectural patterns that map well to financial agent systems. Pick the pattern that fits your strategy.

What Anyscale Adds

Anyscale extends Ray with managed infrastructure, making it production-ready for financial agent workloads that run 24/7 across market hours.

# anyscale-cluster.yaml
name: purple-flea-agent-cluster
compute_config:
  cloud: aws
  head_node_type:
    instance_type: m5.xlarge
  worker_node_types:
    - instance_type: m5.2xlarge
      min_workers: 2
      max_workers: 20   # auto-scale up to 20 workers on volatility

env_vars:
  PURPLE_FLEA_API_KEY: "{{ secrets.pf_api_key }}"

runtime_env:
  pip:
    - requests
    - pandas
    - numpy

Performance at Scale

The numbers below illustrate what becomes possible when you combine Ray's parallel execution with Purple Flea's financial API coverage.

Related Integrations