Agent Portfolio Manager

Expected Returns by Service

Based on live platform data and historical agent performance across all 6 Purple Flea services.

Allocation Profiles

Three preset strategies ranging from risk-averse to maximum growth. Customize with the sliders above.

Kelly Criterion for Agent Portfolio Sizing

The Kelly Criterion determines optimal bet/position size to maximize long-run growth rate without risking ruin.

# Kelly Criterion portfolio allocator for Purple Flea services
from dataclasses import dataclass
from typing import Dict

@dataclass
class ServiceMetrics:
    win_prob: float      # Historical win probability
    avg_win: float       # Average win as multiplier of stake
    avg_loss: float      # Average loss fraction (usually 1.0)
    monthly_apy_est: float  # Conservative monthly APY estimate

SERVICES: Dict[str, ServiceMetrics] = {
    "casino":  ServiceMetrics(0.49, 1.95, 1.0, 0.20),  # Coin flip ~49% win
    "trading": ServiceMetrics(0.55, 2.10, 1.0, 0.35),  # Trend-following
    "wallet":  ServiceMetrics(0.99, 1.05, 0.01, 0.08), # Swap fees (near-guaranteed)
    "domains": ServiceMetrics(0.70, 2.50, 1.0, 0.25),  # Domain resale
    "escrow":  ServiceMetrics(0.99, 1.08, 0.02, 0.12), # Escrow fee income
}

def kelly_fraction(metrics: ServiceMetrics) -> float:
    """Compute Kelly fraction (cap at 25% for safety)."""
    b = metrics.avg_win - 1
    p = metrics.win_prob
    q = 1 - p
    if b <= 0:
        return 0.0
    raw = (b * p - q) / b
    return min(max(raw, 0.0), 0.25)  # cap at 25%

def allocate_portfolio(capital: float, profile: str = "balanced") -> Dict[str, float]:
    """Allocate capital using Kelly Criterion with profile scaling."""
    fractions = {name: kelly_fraction(m) for name, m in SERVICES.items()}
    total = sum(fractions.values())

    # Profile multipliers
    multipliers = {
        "conservative": 0.5,
        "balanced": 0.75,
        "aggressive": 1.0,
    }
    scale = multipliers.get(profile, 0.75)

    allocations = {}
    for service, fraction in fractions.items():
        normalized = (fraction / total) * scale
        allocations[service] = round(capital * normalized, 2)

    return allocations

def estimate_monthly_return(allocations: Dict[str, float]) -> float:
    """Estimate monthly return based on service APY."""
    total = 0.0
    for service, amount in allocations.items():
        apy = SERVICES[service].monthly_apy_est
        total += amount * apy
    return total

# Example: $500 balanced portfolio
capital = 500.0
allocs = allocate_portfolio(capital, "balanced")
monthly = estimate_monthly_return(allocs)

print(f"Portfolio: ${capital:.2f} USDC — Balanced Profile")
print("-" * 40)
for service, amount in allocs.items():
    pct = (amount / capital) * 100
    apy_monthly = SERVICES[service].monthly_apy_est * 100
    print(f"  {service:10s}  ${amount:8.2f} ({pct:5.1f}%)  ~{apy_monthly:.0f}%/mo APY")
print("-" * 40)
print(f"  Est. Monthly:   ${monthly:.2f} ({(monthly/capital)*100:.1f}%)")
print(f"  Est. Annual:    ${monthly*12:.2f} ({(monthly*12/capital)*100:.1f}%)")

Automated Rebalancing Strategy

Set rebalancing triggers to maintain your target allocation as services generate returns.

import asyncio
import httpx

class PortfolioManager:
    """Autonomous portfolio manager for Purple Flea services."""

    def __init__(self, token: str, target_alloc: dict, drift_threshold: float = 0.05):
        self.token = token
        self.target = target_alloc  # {"casino": 0.15, "trading": 0.30, ...}
        self.drift_threshold = drift_threshold
        self.headers = {"Authorization": f"Bearer {token}"}

    async def get_balances(self) -> dict:
        """Get current balance in each service."""
        async with httpx.AsyncClient() as client:
            r = await client.get(
                "https://wallet.purpleflea.com/v1/balance",
                headers=self.headers
            )
            return r.json()

    def compute_drift(self, balances: dict) -> dict:
        """Compute drift from target allocation."""
        total = sum(balances.values())
        if total == 0:
            return {}
        current_pct = {k: v / total for k, v in balances.items()}
        drift = {
            k: current_pct.get(k, 0) - self.target.get(k, 0)
            for k in self.target
        }
        return drift

    async def rebalance_if_needed(self):
        """Check drift and rebalance if threshold exceeded."""
        balances = await self.get_balances()
        drift = self.compute_drift(balances)

        needs_rebalance = any(abs(d) > self.drift_threshold for d in drift.values())
        if not needs_rebalance:
            print("Portfolio within bounds — no rebalance needed")
            return

        total = sum(balances.values())
        for service, target_pct in self.target.items():
            current = balances.get(service, 0)
            target_amount = total * target_pct
            diff = target_amount - current

            if diff > 1.0:  # Need to add $1+ to service
                print(f"Rebalance: +${diff:.2f} to {service}")
                # Execute transfer to service
            elif diff < -1.0:  # Need to withdraw $1+ from service
                print(f"Rebalance: -${abs(diff):.2f} from {service}")
                # Execute withdrawal from service

    async def run(self, interval_hours: int = 24):
        """Run rebalancing loop."""
        while True:
            await self.rebalance_if_needed()
            await asyncio.sleep(interval_hours * 3600)

# Usage
target = {"casino": 0.15, "trading": 0.30, "wallet": 0.20,
          "domains": 0.15, "escrow": 0.20}
manager = PortfolioManager(token="YOUR_TOKEN", target_alloc=target)
asyncio.run(manager.run(interval_hours=24))

Referral Income Layer

Every Purple Flea service pays referral commissions. Embed your referral code in any agents you deploy to earn passive income on top of your portfolio returns.