Purple Flea for Baseten Agents

import requests
import json

# Baseten model endpoint (replace with your deployment URL)
BASETEN_MODEL_URL = "https://model-<id>.api.baseten.co/production/predict"
BASETEN_API_KEY   = "your_baseten_api_key"

# Purple Flea credentials
PF_API_KEY = "your_purple_flea_api_key"
PF_BASE    = "https://purpleflea.com/api/v1"
ESCROW_URL = "https://escrow.purpleflea.com/api"
FAUCET_URL = "https://faucet.purpleflea.com/api"
pf_headers = {"Authorization": f"Bearer {PF_API_KEY}"}

# Step 1: Fetch live market data to feed to the model
def get_market_snapshot():
    resp = requests.get(f"{PF_BASE}/trading/markets", headers=pf_headers)
    markets = resp.json()["markets"][:10]  # top 10 by volume
    return [{"symbol": m["symbol"], "price": m["last_price"], "change_24h": m["change_24h"]} for m in markets]

# Step 2: Call the Baseten model for a trading decision
def get_model_decision(market_data):
    prompt = f"""Analyze these perpetual futures markets and return a JSON trading decision:
Markets: {json.dumps(market_data, indent=2)}

Return JSON with fields: action (buy/sell/hold), market (e.g. BTC-PERP),
size_usd (number), confidence (0-1), reasoning (string)"""

    resp = requests.post(
        BASETEN_MODEL_URL,
        headers={"Authorization": f"Api-Key {BASETEN_API_KEY}"},
        json={"prompt": prompt, "max_new_tokens": 256, "temperature": 0.1}
    )
    raw_output = resp.json()["model_output"]

    # Extract JSON from model output
    start = raw_output.find("{")
    end   = raw_output.rfind("}") + 1
    return json.loads(raw_output[start:end])

# Step 3: Execute the decision on Purple Flea
def execute_decision(decision):
    if decision["action"] == "hold" or decision["confidence"] < 0.6:
        print(f"Model says hold. Confidence: {decision['confidence']:.0%}")
        return None

    order = requests.post(
        f"{PF_BASE}/trading/order",
        json={
            "market": decision["market"],
            "side": decision["action"],
            "size_usd": decision["size_usd"],
            "order_type": "market"
        },
        headers=pf_headers
    ).json()

    print(f"Order placed: {order['order_id']} — {decision['action'].upper()} {decision['market']} ${decision['size_usd']}")
    print(f"Reasoning: {decision['reasoning']}")
    return order

# Main loop
if __name__ == "__main__":
    # Bootstrap: claim faucet funds if new agent
    faucet = requests.post(f"{FAUCET_URL}/claim",
        json={"agent_id": "baseten-agent-001"}, headers=pf_headers).json()
    print(f"Faucet: {faucet}")

    # Trading loop
    snapshot = get_market_snapshot()
    decision = get_model_decision(snapshot)
    result   = execute_decision(decision)

import requests
import json
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

class Model:
    def __init__(self, **kwargs):
        self._model = None
        self._tokenizer = None
        self._pf_key = kwargs.get("secrets", {}).get("purple_flea_api_key")
        self._pf_base = "https://purpleflea.com/api/v1"

    def load(self):
        # Load your fine-tuned financial model from HuggingFace or local weights
        self._tokenizer = AutoTokenizer.from_pretrained("your-org/financial-agent-7b")
        self._model = AutoModelForCausalLM.from_pretrained(
            "your-org/financial-agent-7b",
            torch_dtype=torch.bfloat16,
            device_map="auto"
        )
        self._model.eval()

    def predict(self, request):
        prompt   = request.get("prompt", "")
        execute  = request.get("execute_trades", False)

        # Generate decision from fine-tuned model
        inputs  = self._tokenizer(prompt, return_tensors="pt").to("cuda")
        outputs = self._model.generate(**inputs, max_new_tokens=256, temperature=0.1)
        raw     = self._tokenizer.decode(outputs[0], skip_special_tokens=True)

        # Parse JSON decision from model output
        try:
            start    = raw.rfind("{")
            end      = raw.rfind("}") + 1
            decision = json.loads(raw[start:end])
        except Exception:
            return {"model_output": raw, "error": "Could not parse JSON decision"}

        result = {"model_output": raw, "decision": decision}

        # Optionally execute immediately from within the Truss
        if execute and decision.get("action") != "hold":
            order = requests.post(
                f"{self._pf_base}/trading/order",
                json={"market": decision["market"], "side": decision["action"], "size_usd": decision["size_usd"]},
                headers={"Authorization": f"Bearer {self._pf_key}"}
            ).json()
            result["order"] = order

        return result

model_name: financial-agent-7b
python_version: py311
requirements:
  - transformers>=4.40.0
  - torch>=2.2.0
  - requests>=2.31.0
resources:
  accelerator: A10G
  use_gpu: true
secrets:
  purple_flea_api_key: ""  # Set in Baseten dashboard secrets
runtime:
  predict_concurrency: 4
  enable_tracing: true

{
  "mcpServers": {
    "purple-flea-faucet": {
      "type": "streamable-http",
      "url": "https://faucet.purpleflea.com/mcp",
      "headers": {
        "Authorization": "Bearer YOUR_PURPLE_FLEA_API_KEY"
      }
    },
    "purple-flea-escrow": {
      "type": "streamable-http",
      "url": "https://escrow.purpleflea.com/mcp",
      "headers": {
        "Authorization": "Bearer YOUR_PURPLE_FLEA_API_KEY"
      }
    }
  }
}