Purple Flea for Prefect

Why Prefect for Trading Agents?

Trading agents are not one-shot scripts — they run on schedules, they fail, they retry, they branch on market conditions. Prefect is purpose-built for exactly this: Python functions annotated with @flow and @task decorators become observable, retriable, schedulable units of work.

Purple Flea exposes financial primitives — casino bets, perpetual futures, wallets, escrow, domain registration — via clean REST endpoints. Wrapping those calls in Prefect tasks means every API call gets automatic retry logic, every run produces a state record, and every failure generates an alert in Prefect Cloud.

Prefect Flow with Purple Flea APIs

The example below shows a complete BTC momentum trading flow. The get_market_price and execute_trade tasks each have retry policies. The flow is deployed to Prefect Cloud with an hourly interval schedule so it runs automatically without a cron job or external scheduler.

from prefect import flow, task
from prefect.deployments import Deployment
import requests

PURPLE_FLEA_KEY = "your-api-key"
HEADERS = {"Authorization": f"Bearer {PURPLE_FLEA_KEY}", "Content-Type": "application/json"}

@task(retries=3, retry_delay_seconds=30)
def get_market_price(symbol: str) -> dict:
    """Fetch price with automatic retry on failure"""
    r = requests.get(
        f"https://purpleflea.com/api/v1/markets/{symbol}/price",
        headers=HEADERS
    )
    r.raise_for_status()
    return r.json()

@task(retries=2, retry_delay_seconds=60)
def execute_trade(symbol: str, side: str, size_usd: float) -> dict:
    """Execute trade — auto-retried if API fails"""
    r = requests.post(
        "https://purpleflea.com/api/v1/trade",
        json={
            "symbol": symbol,
            "side": side,
            "size": size_usd,
            "leverage": 2
        },
        headers=HEADERS
    )
    r.raise_for_status()
    return r.json()

@task(retries=2)
def get_portfolio_snapshot() -> dict:
    """Fetch current portfolio for risk check"""
    r = requests.get("https://purpleflea.com/api/v1/portfolio", headers=HEADERS)
    r.raise_for_status()
    return r.json()

@flow(name="btc-momentum-trader", log_prints=True)
def momentum_flow():
    # Risk gate: check total exposure before trading
    portfolio = get_portfolio_snapshot()
    if portfolio["total_exposure_usd"] > 500:
        print("Exposure limit reached — skipping this run")
        return

    # Fetch current price with 1h momentum signal
    price = get_market_price("BTC-PERP")
    print(f"BTC: ${price['price']:,.0f}, 1h change: {price['change_1h_pct']}%")

    # Long on positive momentum, short on negative
    if price["change_1h_pct"] > 2.0:
        trade = execute_trade("BTC-PERP", "long", 50)
        print(f"Opened LONG: {trade['trade_id']}, entry: ${trade['entry_price']:,.0f}")
    elif price["change_1h_pct"] < -2.0:
        trade = execute_trade("BTC-PERP", "short", 50)
        print(f"Opened SHORT: {trade['trade_id']}, entry: ${trade['entry_price']:,.0f}")
    else:
        print("No signal — holding flat")

# Deploy to Prefect Cloud with hourly schedule
if __name__ == "__main__":
    deployment = Deployment.build_from_flow(
        flow=momentum_flow,
        name="btc-momentum-hourly",
        schedule={"interval": 3600}
    )
    deployment.apply()
    print("Deployment created — flow will run every hour in Prefect Cloud")

Available Financial APIs

Every Purple Flea endpoint can be wrapped in a Prefect task with custom retry policies. Mix and match across services to build complex, multi-step trading workflows.

Common Trading Workflow Patterns

These patterns are proven Prefect workflow architectures adapted for financial agents on Purple Flea. Each can be deployed to Prefect Cloud with a schedule and monitored via the Prefect UI.

Full Observability from Prefect Cloud

Prefect Cloud is the managed control plane for your flows. Every run of your trading agent is recorded, logged, and visible in the Prefect UI at app.prefect.cloud. The free tier supports unlimited flow runs.

Continue Building