Crypto Finance Plugins
for Microsoft TaskWeaver

# plugin/purple_flea.py — TaskWeaver plugin definition
from taskweaver.plugin import Plugin, register_plugin
import httpx

@register_plugin
class PurpleFleatPlugin(Plugin):
    """Purple Flea financial infrastructure for AI agents."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.api_key = self.config.get_required("api_key")
        self.base_url = "https://purpleflea.com/api"

    def get_wallet_balance(self) -> dict:
        """Get current wallet balance across all chains.

        Returns:
            dict: {
                "balance_usd": float,
                "chains": {"ETH": float, "SOL": float, ...}
            }
        """
        r = httpx.get(
            f"{self.base_url}/wallet/balance",
            headers={"X-API-Key": self.api_key}
        )
        return r.json()

    def get_market_price(self, symbol: str) -> dict:
        """Get current price for a crypto asset.

        Args:
            symbol: Asset symbol (BTC, ETH, SOL, etc.)

        Returns:
            dict: {"symbol": str, "price_usd": float, "24h_change_pct": float}
        """
        r = httpx.get(
            f"{self.base_url}/market/price/{symbol}",
            headers={"X-API-Key": self.api_key}
        )
        return r.json()

    def place_trade(self, symbol: str, side: str, size_usd: float, leverage: float = 1.0) -> dict:
        """Place a perpetual futures trade on Hyperliquid.

        Args:
            symbol: Trading pair (BTC, ETH, SOL, etc.)
            side: "long" or "short"
            size_usd: Position size in USD
            leverage: Leverage multiplier (1-50x)

        Returns:
            dict: {"trade_id": str, "fill_price": float, "fees_usd": float}
        """
        r = httpx.post(
            f"{self.base_url}/trade",
            json={"symbol": symbol, "side": side, "size_usd": size_usd, "leverage": leverage},
            headers={"X-API-Key": self.api_key}
        )
        return r.json()

    def send_payment(self, to_address: str, amount_usdc: float, memo: str = "") -> dict:
        """Send USDC payment to another wallet address.

        Args:
            to_address: Recipient wallet address
            amount_usdc: Amount in USDC
            memo: Optional transaction memo

        Returns:
            dict: {"tx_hash": str, "status": str}
        """
        r = httpx.post(
            f"{self.base_url}/wallet/send",
            json={"to": to_address, "amount_usdc": amount_usdc, "memo": memo},
            headers={"X-API-Key": self.api_key}
        )
        return r.json()

{
  "plugins": {
    "purple_flea": {
      "enabled": true,
      "config": {
        "api_key": "your-purple-flea-api-key"
      }
    }
  }
}

# When you ask TaskWeaver:
# "Buy $200 of ETH with 3x leverage if RSI is below 40"
#
# TaskWeaver generates and executes:

import pandas as pd
import numpy as np

# Get price data
price_data = purple_flea.get_market_price("ETH")
eth_price = price_data["price_usd"]

# Calculate RSI from recent prices (TaskWeaver fetches OHLCV data)
closes = pd.Series(recent_eth_prices["close"])
delta = closes.diff()
gain = delta.clip(lower=0)
loss = -delta.clip(upper=0)
avg_gain = gain.ewm(span=14).mean()
avg_loss = loss.ewm(span=14).mean()
rsi = 100 - (100 / (1 + avg_gain / avg_loss))
current_rsi = rsi.iloc[-1]

# Conditional execution
if current_rsi < 40:
    result = purple_flea.place_trade(
        symbol="ETH",
        side="long",
        size_usd=200,
        leverage=3.0
    )
    print(f"Trade executed: {result}")
else:
    print(f"RSI is {current_rsi:.1f} — condition not met, no trade placed")