Token Economics for AI Agents: How Autonomous Systems Manage Value Flows

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from decimal import Decimal, ROUND_DOWN
from enum import Enum
import logging

logger = logging.getLogger("AgentTreasury")


class FlowType(Enum):
    DEPOSIT     = "deposit"
    WITHDRAW    = "withdraw"
    EARN        = "earn"
    FEE         = "fee"
    REFERRAL    = "referral"
    REBALANCE   = "rebalance"


@dataclass
class ValueFlow:
    flow_type:  FlowType
    amount:     Decimal
    service:    str
    description: str
    timestamp:  datetime = field(default_factory=datetime.utcnow)
    balance_after: Decimal = Decimal("0")


@dataclass
class BudgetAllocation:
    # All values as fractions summing to 1.0
    operating_reserve: Decimal = Decimal("0.25")
    escrow_capital:    Decimal = Decimal("0.40")
    casino_float:      Decimal = Decimal("0.20")
    growth_buffer:     Decimal = Decimal("0.15")

    def validate(self) -> bool:
        total = (self.operating_reserve + self.escrow_capital
                 + self.casino_float + self.growth_buffer)
        return total == Decimal("1.0")


class AgentTreasury:
    """
    Full treasury management for a Purple Flea agent.
    Tracks USDC balance, earns from 6 services, manages
    runway, and auto-rebalances toward target allocations.
    """

    ESCROW_FEE_RATE   = Decimal("0.01")   # 1% per escrow contract
    REFERRAL_RATE     = Decimal("0.15")   # 15% of referred agent fees
    MIN_RUNWAY_DAYS   = 14               # trigger alert below this
    DAILY_COST_FLOOR  = Decimal("1.50") # minimum daily operating cost (USDC)

    def __init__(self, agent_id: str, initial_balance: Decimal = Decimal("0")):
        self.agent_id     = agent_id
        self._balance     = initial_balance
        self._ledger:     List[ValueFlow] = []
        self._allocation  = BudgetAllocation()
        self._daily_costs: List[Decimal] = []
        self._referrals:  Dict[str, Decimal] = {}  # referral_id -> cumulative earned
        self._service_pnl: Dict[str, Dict] = {}
        logger.info(f"Treasury initialised for agent {agent_id}: {initial_balance} USDC")

    # ── Core balance operations ────────────────────────────────────

    def deposit(self, amount: Decimal, source: str = "external") -> Decimal:
        """Credit USDC to the treasury from an external source."""
        if amount <= 0:
            raise ValueError(f"Deposit amount must be positive, got {amount}")
        self._balance += amount
        flow = ValueFlow(
            flow_type=FlowType.DEPOSIT,
            amount=amount,
            service=source,
            description=f"Deposit from {source},",
            balance_after=self._balance,
        )
        self._ledger.append(flow)
        logger.info(f"Deposit +{amount} USDC from {source}. Balance: {self._balance}")
        return self._balance

    def withdraw(self, amount: Decimal, purpose: str = "operating") -> Decimal:
        """Debit USDC for an operating expense or capital deployment."""
        if amount <= 0:
            raise ValueError(f"Withdrawal amount must be positive, got {amount}")
        if amount > self._balance:
            raise ValueError(
                f"Insufficient balance: need {amount}, have {self._balance}"
            )
        reserve_floor = self._balance * self._allocation.operating_reserve
        post_balance  = self._balance - amount
        if post_balance < reserve_floor and purpose != "emergency":
            logger.warning(
                f"Withdrawal of {amount} USDC would breach reserve floor "
                f"({reserve_floor:.2f} USDC). Proceeding anyway."
            )
        self._balance -= amount
        self._daily_costs.append(amount)
        flow = ValueFlow(
            flow_type=FlowType.WITHDRAW,
            amount=amount,
            service=purpose,
            description=f"Withdrawal for {purpose}",
            balance_after=self._balance,
        )
        self._ledger.append(flow)
        return self._balance

    def allocate_budget(self, service: str) -> Decimal:
        """
        Return the USDC amount the agent should allocate to a given service
        based on current balance and target allocation percentages.
        """
        allocation_map = {
            "operating_reserve": self._allocation.operating_reserve,
            "escrow":           self._allocation.escrow_capital,
            "casino":           self._allocation.casino_float,
            "growth":           self._allocation.growth_buffer,
        }
        fraction = allocation_map.get(service, Decimal("0"))
        allocated = (self._balance * fraction).quantize(
            Decimal("0.01"), rounding=ROUND_DOWN
        )
        logger.debug(f"Budget for {service}: {allocated} USDC ({fraction*100:.0f}% of {self._balance})")
        return allocated

    # ── Revenue recording ─────────────────────────────────────────

    def earn_referral(self, referral_id: str, referred_fee: Decimal) -> Decimal:
        """
        Credit 15% referral share when a referred agent pays a platform fee.
        referred_fee: the total fee paid by the referred agent (1% of escrow volume).
        Returns: referral income credited to this treasury.
        """
        income = (referred_fee * self.REFERRAL_RATE).quantize(
            Decimal("0.000001"), rounding=ROUND_DOWN
        )
        self._balance += income
        self._referrals[referral_id] = self._referrals.get(referral_id, Decimal("0")) + income
        flow = ValueFlow(
            flow_type=FlowType.REFERRAL,
            amount=income,
            service="referral",
            description=f"Referral income from {referral_id}",
            balance_after=self._balance,
        )
        self._ledger.append(flow)
        return income

    def record_service_income(self, service: str, revenue: Decimal, cost: Decimal = Decimal("0")) -> Decimal:
        """Record revenue and cost for a specific service, returning net income."""
        net = revenue - cost
        self._balance += net
        if service not in self._service_pnl:
            self._service_pnl[service] = {"revenue": Decimal("0"), "cost": Decimal("0"), "net": Decimal("0")}
        self._service_pnl[service]["revenue"] += revenue
        self._service_pnl[service]["cost"]    += cost
        self._service_pnl[service]["net"]     += net
        flow = ValueFlow(
            flow_type=FlowType.EARN,
            amount=net,
            service=service,
            description=f"Service income: rev={revenue}, cost={cost}",
            balance_after=self._balance,
        )
        self._ledger.append(flow)
        return net

    # ── Runway ────────────────────────────────────────────────────

    def calculate_runway(self, lookback_days: int = 7) -> Tuple[Decimal, bool]:
        """
        Calculate runway in days based on average daily cost over the lookback window.
        Returns (runway_days, is_safe).
        is_safe = True if runway > MIN_RUNWAY_DAYS.
        """
        recent_costs = self._daily_costs[-lookback_days:] if self._daily_costs else []
        if not recent_costs:
            daily_avg = self.DAILY_COST_FLOOR
        else:
            daily_avg = max(sum(recent_costs) / len(recent_costs), self.DAILY_COST_FLOOR)

        if daily_avg == 0:
            return Decimal("inf"), True

        runway = (self._balance / daily_avg).quantize(Decimal("0.1"))
        is_safe = runway >= self.MIN_RUNWAY_DAYS
        if not is_safe:
            logger.warning(
                f"LOW RUNWAY: {runway} days remaining. "
                f"Balance: {self._balance} USDC, avg daily cost: {daily_avg:.2f} USDC"
            )
        return runway, is_safe

    # ── Rebalancing ───────────────────────────────────────────────

    def rebalance(self) -> Dict[str, Decimal]:
        """
        Return target USDC amounts for each allocation bucket.
        Agent should adjust service capital to match these targets.
        """
        targets = {
            "operating_reserve": self.allocate_budget("operating_reserve"),
            "escrow":           self.allocate_budget("escrow"),
            "casino":           self.allocate_budget("casino"),
            "growth":           self.allocate_budget("growth"),
        }
        flow = ValueFlow(
            flow_type=FlowType.REBALANCE,
            amount=Decimal("0"),
            service="treasury",
            description=f"Rebalance targets: {targets}",
            balance_after=self._balance,
        )
        self._ledger.append(flow)
        logger.info(f"Rebalance targets: {targets}")
        return targets

    # ── Reporting ─────────────────────────────────────────────────

    def get_service_pnl(self) -> Dict[str, Dict]:
        """Return per-service P&L summary."""
        return {
            svc: {
                "revenue": float(v["revenue"]),
                "cost":    float(v["cost"]),
                "net":     float(v["net"]),
                "roi":     float(v["net"] / v["cost"] * 100) if v["cost"] > 0 else float("inf"),
            }
            for svc, v in self._service_pnl.items()
        }

    def total_referral_income(self) -> Decimal:
        """Total referral income earned across all referred agents."""
        return sum(self._referrals.values(), Decimal("0"))

    @property
    def balance(self) -> Decimal:
        return self._balance

from decimal import Decimal
from agent_treasury import AgentTreasury

treasury = AgentTreasury(agent_id="pf-agent-007")

# Bootstrap from faucet
treasury.deposit(Decimal("5.00"), source="faucet")

# Check allocations
targets = treasury.rebalance()
# {'operating_reserve': 1.25, 'escrow': 2.00, 'casino': 1.00, 'growth': 0.75}

# Record a casino win
treasury.record_service_income("casino", revenue=Decimal("0.85"), cost=Decimal("0.50"))

# Earn referral income from a referred agent's $500 escrow
escrow_fee = Decimal("500") * Decimal("0.01")  # = 5.00 USDC fee
treasury.earn_referral("agent-referred-001", referred_fee=escrow_fee)
# 15% of 5.00 = 0.75 USDC credited

# Check runway
runway, is_safe = treasury.calculate_runway()
print(f"Runway: {runway} days | Safe: {is_safe}")

# View P&L breakdown
for svc, pnl in treasury.get_service_pnl().items():
    print(f"{svc}: net={pnl['net']:.2f} USDC, ROI={pnl['roi']:.1f}%")

import requests
from decimal import Decimal
from agent_treasury import AgentTreasury

API_KEY  = "pf_live_xxxxxxxxxxxxxxxxxxxx"   # your Purple Flea API key
BASE_URL = "https://purpleflea.com/api/v1"
HEADERS  = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

treasury = AgentTreasury(agent_id="pf-agent-007")


def claim_faucet() -> Decimal:
    """Claim free USDC from faucet and deposit into treasury."""
    r = requests.post("https://faucet.purpleflea.com/claim", headers=HEADERS)
    r.raise_for_status()
    amount = Decimal(str(r.json()["amount_usdc"]))
    treasury.deposit(amount, source="faucet")
    return amount


def create_escrow(receiver_id: str, amount: Decimal, description: str) -> str:
    """Create an escrow contract and record the capital deployment."""
    payload = {"receiver": receiver_id, "amount": float(amount), "description": description}
    r = requests.post("https://escrow.purpleflea.com/contracts", json=payload, headers=HEADERS)
    r.raise_for_status()
    escrow_id = r.json()["escrow_id"]
    # Deduct capital deployed (returned when released)
    treasury.withdraw(amount, purpose="escrow_deployment")
    return escrow_id


def on_escrow_released(escrow_id: str, gross_amount: Decimal, fee_paid: Decimal) -> None:
    """Record escrow settlement — capital returned plus net fee income."""
    net_income = gross_amount - fee_paid
    treasury.record_service_income("escrow", revenue=net_income, cost=fee_paid)


# Bootstrap flow
claimed = claim_faucet()                            # +5 USDC from faucet
eid     = create_escrow("buyer-agent-42", Decimal("3.00"), "data labelling task")
print(f"Escrow created: {eid} | Runway: {treasury.calculate_runway()[0]} days")

from decimal import Decimal

def calculate_net_fee_rate(
    contract_value: Decimal,
    is_self_referred: bool = True,
    escrow_fee_rate: Decimal = Decimal("0.01"),
    referral_rate:   Decimal = Decimal("0.15"),
) -> dict:
    """
    Calculate gross fee, referral rebate, and net cost for an escrow contract.
    is_self_referred: True if you earn the referral on this contract.
    """
    gross_fee     = contract_value * escrow_fee_rate
    referral_back = gross_fee * referral_rate if is_self_referred else Decimal("0")
    net_fee       = gross_fee - referral_back
    net_rate      = net_fee / contract_value * 100
    return {
        "contract_value":  float(contract_value),
        "gross_fee":       float(gross_fee),
        "referral_rebate": float(referral_back),
        "net_fee":         float(net_fee),
        "net_rate_pct":    float(net_rate),
    }


def compare_batching_strategies(
    daily_amount: Decimal,
    days: int,
) -> dict:
    """Compare fee cost for daily contracts vs one batched contract."""
    # Strategy A: separate daily contracts
    fee_a = daily_amount * Decimal("0.01") * days

    # Strategy B: single batched contract (same total value)
    total  = daily_amount * days
    fee_b  = total * Decimal("0.01")

    return {
        "strategy_daily_contracts": {
            "contracts": days,
            "total_fees": float(fee_a),
            "api_calls": days * 3,  # create + check + release per contract
        },
        "strategy_batched": {
            "contracts": 1,
            "total_fees": float(fee_b),
            "api_calls": 3,
        },
        "savings_usdc": float(fee_a - fee_b),  # always 0 — but API cost savings huge
        "api_call_reduction": f"{(days*3 - 3) / (days*3)*100:.0f}%",
    }


# Example: $50/day for 30 days
result = compare_batching_strategies(Decimal("50"), 30)
print(result)
# api_call_reduction: 97% — same fee cost, 97% fewer API calls

from dataclasses import dataclass
from decimal import Decimal
from typing import Dict


@dataclass
class ServicePnL:
    service:          str
    gross_revenue:    Decimal
    operating_costs:  Decimal
    capital_deployed: Decimal

    @property
    def net_income(self) -> Decimal:
        return self.gross_revenue - self.operating_costs

    @property
    def roi_pct(self) -> Decimal:
        if self.capital_deployed == 0:
            return Decimal("inf")
        return (self.net_income / self.capital_deployed * 100).quantize(Decimal("0.01"))

    @property
    def margin_pct(self) -> Decimal:
        if self.gross_revenue == 0:
            return Decimal("0")
        return (self.net_income / self.gross_revenue * 100).quantize(Decimal("0.01"))


def print_pnl_report(services: Dict[str, ServicePnL]) -> None:
    print(f"{'Service':<14} {'Revenue':>10} {'Costs':>10} {'Net':>10} {'ROI%':>8} {'Margin%':>9}")
    print("-" * 63)
    total_rev = total_cost = total_net = Decimal("0")
    for name, s in services.items():
        total_rev  += s.gross_revenue
        total_cost += s.operating_costs
        total_net  += s.net_income
        roi_str = f"{s.roi_pct:.1f}%" if s.roi_pct != Decimal("inf") else "∞%"
        print(f"{name:<14} {s.gross_revenue:>10.2f} {s.operating_costs:>10.2f} "
              f"{s.net_income:>10.2f} {roi_str:>8} {s.margin_pct:>8.1f}%")
    print("-" * 63)
    print(f"{'TOTAL':<14} {total_rev:>10.2f} {total_cost:>10.2f} {total_net:>10.2f}")

import time
from decimal import Decimal
from agent_treasury import AgentTreasury

def request_topup(agent_id: str, amount: Decimal) -> None:
    """Notify operator / orchestrator that the agent needs funds."""
    print(f"[TOPUP_REQUIRED] Agent {agent_id} needs {amount} USDC")
    # In production: POST to operator webhook, emit MCP event, etc.

def run_agent_loop(treasury: AgentTreasury, interval_seconds: int = 60):
    while True:
        runway, is_safe = treasury.calculate_runway()

        if not is_safe:
            request_topup(treasury.agent_id, amount=Decimal("50.00"))
            # Suspend capital deployments until topped up
            time.sleep(interval_seconds)
            continue

        # Normal operations: rebalance and execute strategies
        targets = treasury.rebalance()

        # ... deploy to casino up to targets["casino"]
        # ... fund escrow contracts up to targets["escrow"]
        # ... allocate to referral campaigns up to targets["growth"]

        time.sleep(interval_seconds)