Purple Flea for Temporal

from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker
from datetime import timedelta
import requests
import os

PURPLE_FLEA_KEY = os.environ["PURPLE_FLEA_API_KEY"]
BASE_URL = "https://purpleflea.com/api/v1"

# ─── ACTIVITIES ───────────────────────────────────────────────────────────────

@activity.defn
async def execute_trade(symbol: str, side: str, size_usd: float) -> dict:
    """Purple Flea trade wrapped as Temporal activity — auto-retried on failure"""
    response = requests.post(
        f"{BASE_URL}/trade",
        json={"symbol": symbol, "side": side, "size": size_usd, "leverage": 2},
        headers={"Authorization": f"Bearer {PURPLE_FLEA_KEY}"}
    )
    response.raise_for_status()
    return response.json()


@activity.defn
async def get_portfolio() -> dict:
    """Fetch all open positions and unrealized PnL"""
    response = requests.get(
        f"{BASE_URL}/portfolio",
        headers={"Authorization": f"Bearer {PURPLE_FLEA_KEY}"}
    )
    response.raise_for_status()
    return response.json()


@activity.defn
async def close_all_positions() -> dict:
    """Emergency close — exits all positions at market"""
    response = requests.post(
        f"{BASE_URL}/trade/close-all",
        headers={"Authorization": f"Bearer {PURPLE_FLEA_KEY}"}
    )
    response.raise_for_status()
    return response.json()


@activity.defn
async def create_escrow(amount_usd: float, counterparty_id: str, condition: str) -> dict:
    """Create trustless escrow with Purple Flea (1% fee, 15% referral)"""
    response = requests.post(
        f"{BASE_URL}/escrow/create",
        json={"amount_usd": amount_usd, "counterparty": counterparty_id, "condition": condition},
        headers={"Authorization": f"Bearer {PURPLE_FLEA_KEY}"}
    )
    response.raise_for_status()
    return response.json()


# ─── WORKFLOWS ────────────────────────────────────────────────────────────────

@workflow.defn
class DCAWorkflow:
    """Dollar-cost average into BTC every week for 52 weeks"""

    @workflow.run
    async def run(self, weekly_amount: float):
        for week in range(52):
            result = await workflow.execute_activity(
                execute_trade,
                args=["BTC-PERP", "long", weekly_amount],
                schedule_to_close_timeout=timedelta(minutes=5),
                retry_policy=RetryPolicy(
                    backoff_coefficient=2.0,
                    maximum_attempts=10,
                    maximum_interval=timedelta(minutes=2)
                )
            )
            # Persist weekly result in workflow history
            workflow.logger.info(f"Week {week+1}: bought BTC at {result['entry_price']}")
            # Sleep for exactly 7 days — Temporal timer, not OS sleep
            await workflow.sleep(timedelta(days=7))


@workflow.defn
class RiskMonitorWorkflow:
    """Monitor portfolio and close all if drawdown exceeds threshold"""

    @workflow.run
    async def run(self, max_drawdown_pct: float = 0.15):
        while True:
            portfolio = await workflow.execute_activity(
                get_portfolio,
                schedule_to_close_timeout=timedelta(seconds=30)
            )
            drawdown = portfolio.get("drawdown_pct", 0)
            if drawdown >= max_drawdown_pct:
                await workflow.execute_activity(
                    close_all_positions,
                    schedule_to_close_timeout=timedelta(minutes=2)
                )
                workflow.logger.warning(f"Drawdown {drawdown:.1%} — closed all positions")
                return
            # Check every 5 minutes
            await workflow.sleep(timedelta(minutes=5))