DeFi Patterns

This guide covers decentralized finance primitives on Panda: escrow, automated market makers, NFT marketplaces, prediction markets, and cross-contract composition.

Escrow

The Escrow contract holds PRC-20 tokens until both parties agree on fulfillment. It uses cross-contract PRC20 handles to transfer tokens between buyer, seller, and the escrow contract itself.

# Source: contracts/patterns/escrow.py
"""
Escrow — Cross-contract escrow service using PRC-20 tokens.

Demonstrates complex cross-contract call patterns:
- Buyer deposits payment tokens into escrow
- Seller delivers goods/services
- Buyer confirms delivery, releasing funds to seller
- Or buyer requests refund within the grace period
"""

from panda import contract, constructor, call, query, event, PRC20


@contract
class Escrow:
    """Escrow service using PRC-20 tokens via cross-contract calls."""

    class State:
        payment_token: str = ""
        escrows: dict = {}
        next_escrow_id: int = 1
        owner: str = ""
        fee_bps: int = 100  # 1% fee
        total_volume: int = 0

    @constructor
    def deploy(self, ctx, payment_token: str, fee_bps: int = 100):
        self.state.payment_token = payment_token
        self.state.owner = ctx.sender
        self.state.fee_bps = fee_bps
        print(f"Escrow deployed by {ctx.sender}, token={payment_token}, fee={fee_bps}bps")

    @call
    def create_escrow(self, ctx, seller: str, amount: int, grace_period: int):
        """Create an escrow. Buyer deposits tokens via cross-contract call.

        grace_period: seconds after creation during which buyer can refund.
        """
        if amount <= 0:
            raise ValueError("amount must be positive")
        if seller == ctx.sender:
            raise ValueError("buyer and seller must be different")

        # Transfer tokens from buyer to escrow contract
        PRC20(self.state.payment_token).transfer_from(
            owner=ctx.sender, to=ctx.contract_address, value=amount
        )

        eid = self.state.next_escrow_id
        escrows = dict(self.state.escrows)
        escrows[str(eid)] = {
            "buyer": ctx.sender,
            "seller": seller,
            "amount": amount,
            "status": "active",
            "created_at": ctx.block_time,
            "grace_deadline": ctx.block_time + grace_period,
        }
        self.state.escrows = escrows
        self.state.next_escrow_id = eid + 1
        self.state.total_volume = self.state.total_volume + amount

        self.emit(
            event.EscrowCreated(
                escrow_id=eid,
                buyer=ctx.sender,
                seller=seller,
                amount=amount,
            )
        )
        print(f"Escrow #{eid} created: buyer={ctx.sender}, seller={seller}, amount={amount}")

    @call
    def release(self, ctx, escrow_id: int):
        """Release funds to seller. Only buyer can release."""
        key = str(escrow_id)
        escrows = dict(self.state.escrows)
        escrow = escrows.get(key)
        if not escrow:
            raise ValueError("escrow not found")
        if escrow["status"] != "active":
            raise ValueError(f"escrow is {escrow['status']}, not active")
        if ctx.sender != escrow["buyer"]:
            raise ValueError("only buyer can release")

        # Calculate fee
        fee = (escrow["amount"] * self.state.fee_bps) // 10000
        seller_amount = escrow["amount"] - fee

        # Transfer to seller
        token = PRC20(self.state.payment_token)
        token.transfer(to=escrow["seller"], value=seller_amount)
        # Transfer fee to owner
        if fee > 0:
            token.transfer(to=self.state.owner, value=fee)

        escrow["status"] = "released"
        escrows[key] = escrow
        self.state.escrows = escrows

        self.emit(
            event.EscrowReleased(
                escrow_id=escrow_id,
                seller=escrow["seller"],
                amount=seller_amount,
                fee=fee,
            )
        )
        print(f"Escrow #{escrow_id} released: {seller_amount} to {escrow['seller']}, fee={fee}")

    @call
    def refund(self, ctx, escrow_id: int):
        """Refund buyer. Only buyer can refund, and only within grace period."""
        key = str(escrow_id)
        escrows = dict(self.state.escrows)
        escrow = escrows.get(key)
        if not escrow:
            raise ValueError("escrow not found")
        if escrow["status"] != "active":
            raise ValueError(f"escrow is {escrow['status']}, not active")
        if ctx.sender != escrow["buyer"]:
            raise ValueError("only buyer can refund")
        if ctx.block_time > escrow["grace_deadline"]:
            raise ValueError("grace period has expired")

        # Refund full amount to buyer
        PRC20(self.state.payment_token).transfer(to=escrow["buyer"], value=escrow["amount"])

        escrow["status"] = "refunded"
        escrows[key] = escrow
        self.state.escrows = escrows

        self.emit(
            event.EscrowRefunded(
                escrow_id=escrow_id,
                buyer=escrow["buyer"],
                amount=escrow["amount"],
            )
        )
        print(f"Escrow #{escrow_id} refunded: {escrow['amount']} to {escrow['buyer']}")

    @call
    def dispute(self, ctx, escrow_id: int, reason: str):
        """Mark an escrow as disputed. Either buyer or seller can dispute."""
        key = str(escrow_id)
        escrows = dict(self.state.escrows)
        escrow = escrows.get(key)
        if not escrow:
            raise ValueError("escrow not found")
        if escrow["status"] != "active":
            raise ValueError("can only dispute active escrows")
        if ctx.sender != escrow["buyer"] and ctx.sender != escrow["seller"]:
            raise ValueError("only buyer or seller can dispute")

        escrow["status"] = "disputed"
        escrow["dispute_reason"] = reason
        escrow["disputed_by"] = ctx.sender
        escrows[key] = escrow
        self.state.escrows = escrows

        self.emit(
            event.EscrowDisputed(
                escrow_id=escrow_id,
                disputed_by=ctx.sender,
                reason=reason,
            )
        )
        print(f"Escrow #{escrow_id} disputed by {ctx.sender}: {reason}")

    @query
    def get_escrow(self, escrow_id: int) -> dict:
        return self.state.escrows.get(str(escrow_id), {})

    @query
    def get_stats(self) -> dict:
        active = sum(1 for e in self.state.escrows.values() if e["status"] == "active")
        return {
            "total_volume": self.state.total_volume,
            "total_escrows": len(self.state.escrows),
            "active_escrows": active,
        }

Escrow lifecycle:

• active -- funds deposited, waiting for fulfillment
• released -- buyer confirmed delivery, funds sent to seller (minus fee)
• refunded -- buyer cancelled within grace period, funds returned
• disputed -- either party raised a dispute (resolution handled off-chain or by a governance contract)

Automated Market Maker (AMM)

The TokenSwap contract implements a constant-product AMM using the formula x * y = k to price trades between two PRC-20 tokens.

# Source: contracts/tokens/token_swap.py
from panda import contract, constructor, call, query, event, PRC20


@contract
class TokenSwap:
    """Simple AMM swap pool for two PRC-20 tokens via cross-contract calls."""

    class State:
        token_a: str = ""
        token_b: str = ""
        reserve_a: int = 0
        reserve_b: int = 0
        total_lp: int = 0
        lp_balances: dict = {}
        fee_bps: int = 30  # 0.3% fee in basis points
        owner: str = ""

    @constructor
    def deploy(self, ctx, token_a: str, token_b: str, fee_bps: int = 30):
        if token_a == token_b:
            raise ValueError("tokens must be different")
        self.state.token_a = token_a
        self.state.token_b = token_b
        self.state.fee_bps = fee_bps
        self.state.owner = ctx.sender

    @call
    def add_liquidity(self, ctx, amount_a: int, amount_b: int):
        """Add liquidity to the pool. Transfers tokens from sender via cross-call."""
        if amount_a <= 0 or amount_b <= 0:
            raise ValueError("amounts must be positive")

        # Transfer tokens into the pool
        PRC20(self.state.token_a).transfer_from(
            owner=ctx.sender, to=ctx.contract_address, value=amount_a
        )
        PRC20(self.state.token_b).transfer_from(
            owner=ctx.sender, to=ctx.contract_address, value=amount_b
        )

        # Calculate LP tokens to mint
        if self.state.total_lp == 0:
            # First deposit: LP = sqrt(a * b) approximated as (a + b) / 2
            lp_amount = (amount_a + amount_b) // 2
        else:
            lp_a = amount_a * self.state.total_lp // self.state.reserve_a
            lp_b = amount_b * self.state.total_lp // self.state.reserve_b
            lp_amount = min(lp_a, lp_b)

        if lp_amount <= 0:
            raise ValueError("insufficient liquidity minted")

        self.state.reserve_a = self.state.reserve_a + amount_a
        self.state.reserve_b = self.state.reserve_b + amount_b
        self.state.total_lp = self.state.total_lp + lp_amount
        lp = dict(self.state.lp_balances)
        lp[ctx.sender] = lp.get(ctx.sender, 0) + lp_amount
        self.state.lp_balances = lp

        self.emit(
            event.LiquidityAdded(
                provider=ctx.sender,
                amount_a=amount_a,
                amount_b=amount_b,
                lp_minted=lp_amount,
            )
        )

    @call
    def swap_a_for_b(self, ctx, amount_in: int):
        """Swap token_a for token_b using constant-product formula."""
        if amount_in <= 0:
            raise ValueError("amount must be positive")
        if self.state.reserve_a == 0 or self.state.reserve_b == 0:
            raise ValueError("no liquidity")

        # Apply fee
        amount_in_with_fee = amount_in * (10000 - self.state.fee_bps)
        amount_out = (amount_in_with_fee * self.state.reserve_b) // (
            self.state.reserve_a * 10000 + amount_in_with_fee
        )

        if amount_out <= 0:
            raise ValueError("insufficient output amount")

        # Transfer token_a from sender into pool
        PRC20(self.state.token_a).transfer_from(
            owner=ctx.sender, to=ctx.contract_address, value=amount_in
        )
        # Transfer token_b from pool to sender
        PRC20(self.state.token_b).transfer(to=ctx.sender, value=amount_out)

        self.state.reserve_a = self.state.reserve_a + amount_in
        self.state.reserve_b = self.state.reserve_b - amount_out

        self.emit(
            event.Swap(
                sender=ctx.sender,
                token_in=self.state.token_a,
                amount_in=amount_in,
                amount_out=amount_out,
            )
        )

    @call
    def swap_b_for_a(self, ctx, amount_in: int):
        """Swap token_b for token_a using constant-product formula."""
        if amount_in <= 0:
            raise ValueError("amount must be positive")
        if self.state.reserve_a == 0 or self.state.reserve_b == 0:
            raise ValueError("no liquidity")

        amount_in_with_fee = amount_in * (10000 - self.state.fee_bps)
        amount_out = (amount_in_with_fee * self.state.reserve_a) // (
            self.state.reserve_b * 10000 + amount_in_with_fee
        )

        if amount_out <= 0:
            raise ValueError("insufficient output amount")

        PRC20(self.state.token_b).transfer_from(
            owner=ctx.sender, to=ctx.contract_address, value=amount_in
        )
        PRC20(self.state.token_a).transfer(to=ctx.sender, value=amount_out)

        self.state.reserve_b = self.state.reserve_b + amount_in
        self.state.reserve_a = self.state.reserve_a - amount_out

        self.emit(
            event.Swap(
                sender=ctx.sender,
                token_in=self.state.token_b,
                amount_in=amount_in,
                amount_out=amount_out,
            )
        )

    @query
    def get_reserves(self) -> dict:
        return {
            "reserve_a": self.state.reserve_a,
            "reserve_b": self.state.reserve_b,
            "total_lp": self.state.total_lp,
        }

    @query
    def quote_swap_a_for_b(self, amount_in: int) -> int:
        """Preview how much token_b you'd get for amount_in of token_a."""
        if self.state.reserve_a == 0 or self.state.reserve_b == 0:
            return 0
        amount_in_with_fee = amount_in * (10000 - self.state.fee_bps)
        return (amount_in_with_fee * self.state.reserve_b) // (
            self.state.reserve_a * 10000 + amount_in_with_fee
        )

How the constant-product formula works:

The formula ensures reserve_a * reserve_b = k (constant) after every swap. Larger trades get progressively worse prices (slippage), which protects liquidity providers. Fees (default 0.3%) are deducted before the swap calculation, making k grow over time and rewarding LPs.

Liquidity provision uses (a + b) / 2 for the first deposit and proportional minting via min(lp_a, lp_b) for subsequent deposits, ensuring fair LP token distribution.

NFT Marketplace

The NFT Marketplace combines PRC-721 and PRC-20 cross-contract calls to enable atomic buy/sell of NFTs with token payments.

# Source: contracts/tokens/nft_marketplace.py (excerpt: buy_nft method)
from panda import contract, constructor, call, query, event, Contract, PRC20

    @call
    def buy_nft(self, ctx, nft_contract: str, token_id: int):
        """Buy a listed NFT. Buyer must have approved payment token spending."""
        listing_id = f"{nft_contract}:{token_id}"
        listings = dict(self.state.listings)
        listing = listings.get(listing_id)
        if not listing or not listing["active"]:
            raise ValueError("listing not found or inactive")

        price = listing["price"]
        seller = listing["seller"]

        # Calculate fees
        fee = (price * self.state.fee_bps) // 10000
        seller_proceeds = price - fee

        # Transfer payment from buyer to seller
        payment = PRC20(self.state.payment_token)
        payment.transfer_from(owner=ctx.sender, to=seller, value=seller_proceeds)

        # Transfer fee to fee recipient
        if fee > 0:
            payment.transfer_from(owner=ctx.sender, to=self.state.fee_recipient, value=fee)

        # Transfer NFT from seller to buyer
        Contract(nft_contract).transfer_from(
            from_addr=seller, to_addr=ctx.sender, token_id=token_id
        )

        # Mark listing as inactive
        listing["active"] = False
        listings[listing_id] = listing
        self.state.listings = listings
        self.state.total_volume = self.state.total_volume + price
        self.state.total_sales = self.state.total_sales + 1

        self.emit(
            event.NFTSold(
                buyer=ctx.sender,
                seller=seller,
                nft_contract=nft_contract,
                token_id=token_id,
                price=price,
                fee=fee,
            )
        )

The buy_nft method performs three cross-contract calls atomically:

1. PRC20.transfer_from -- move payment tokens from buyer to seller
2. PRC20.transfer_from -- move marketplace fee to fee recipient
3. Contract.transfer_from -- transfer the NFT from seller to buyer

If any call fails, the entire transaction reverts -- no partial state changes.

Prediction Market

A binary prediction market where users bet on YES or NO outcomes. Uses time-based logic for betting deadlines and proportional payout calculations.

# Source: contracts/examples/prediction_market.py
from panda import contract, constructor, call, query, event


@contract
class PredictionMarket:
    """Binary prediction market where users bet on YES or NO outcomes."""

    class State:
        question: str = ""
        owner: str = ""
        deadline: int = 0
        resolved: bool = False
        outcome: str = ""  # "yes" or "no" after resolution
        yes_pool: int = 0
        no_pool: int = 0
        bets: dict = {}  # {addr: {"side": "yes"|"no", "amount": int}}
        claimed: dict = {}  # {addr: True} after payout claimed

    @constructor
    def deploy(self, ctx, question: str, deadline: int):
        """Create a new prediction market.

        Args:
            question: The yes/no question being predicted.
            deadline: Block timestamp after which no new bets are accepted.
        """
        if not question:
            raise ValueError("Question cannot be empty")
        if deadline <= ctx.block_time:
            raise ValueError("Deadline must be in the future")
        self.state.question = question
        self.state.owner = ctx.sender
        self.state.deadline = deadline
        print(f"PredictionMarket deployed: '{question}' by {ctx.sender}")

    @call
    def place_bet(self, ctx, side: str, amount: int):
        """Place a bet on YES or NO.

        Args:
            side: "yes" or "no"
            amount: Amount to wager (must be positive)
        """
        if self.state.resolved:
            raise ValueError("Market already resolved")
        if ctx.block_time >= self.state.deadline:
            raise ValueError("Betting period has ended")
        if side not in ("yes", "no"):
            raise ValueError("Side must be 'yes' or 'no'")
        if amount <= 0:
            raise ValueError("Amount must be positive")

        addr = ctx.sender
        bets = dict(self.state.bets)
        if addr in bets:
            raise ValueError("Already placed a bet")

        bets[addr] = {"side": side, "amount": amount}
        self.state.bets = bets

        if side == "yes":
            self.state.yes_pool = self.state.yes_pool + amount
        else:
            self.state.no_pool = self.state.no_pool + amount

        self.emit(event.BetPlaced(
            bettor=addr,
            side=side,
            amount=amount,
            yes_pool=self.state.yes_pool,
            no_pool=self.state.no_pool,
        ))

    @call
    def resolve(self, ctx, outcome: str):
        """Resolve the market. Only the owner can resolve, and only after deadline.

        Args:
            outcome: "yes" or "no"
        """
        if ctx.sender != self.state.owner:
            raise ValueError("Only the owner can resolve")
        if self.state.resolved:
            raise ValueError("Already resolved")
        if ctx.block_time < self.state.deadline:
            raise ValueError("Cannot resolve before deadline")
        if outcome not in ("yes", "no"):
            raise ValueError("Outcome must be 'yes' or 'no'")

        self.state.resolved = True
        self.state.outcome = outcome
        self.emit(event.MarketResolved(
            outcome=outcome,
            yes_pool=self.state.yes_pool,
            no_pool=self.state.no_pool,
        ))

    @call
    def claim_payout(self, ctx):
        """Claim winnings after market resolution."""
        if not self.state.resolved:
            raise ValueError("Market not resolved yet")

        addr = ctx.sender
        bets = dict(self.state.bets)
        claimed = dict(self.state.claimed)

        if addr not in bets:
            raise ValueError("No bet placed")
        if addr in claimed:
            raise ValueError("Already claimed")

        bet = bets[addr]
        if bet["side"] != self.state.outcome:
            raise ValueError("You lost this bet")

        # Calculate proportional payout from total pool
        total_pool = self.state.yes_pool + self.state.no_pool
        winning_pool = self.state.yes_pool if self.state.outcome == "yes" else self.state.no_pool

        if winning_pool == 0:
            raise ValueError("No winners")

        payout = (bet["amount"] * total_pool) // winning_pool

        claimed[addr] = True
        self.state.claimed = claimed

        self.emit(event.PayoutClaimed(
            bettor=addr,
            payout=payout,
            bet_amount=bet["amount"],
        ))
        return payout

    @query
    def get_market(self) -> dict:
        """Return market status."""
        return {
            "question": self.state.question,
            "deadline": self.state.deadline,
            "resolved": self.state.resolved,
            "outcome": self.state.outcome,
            "yes_pool": self.state.yes_pool,
            "no_pool": self.state.no_pool,
            "total_bets": len(self.state.bets),
        }

    @query
    def get_odds(self) -> dict:
        """Return current implied odds (percentage of pool)."""
        total = self.state.yes_pool + self.state.no_pool
        if total == 0:
            return {"yes_pct": 50, "no_pct": 50}
        return {
            "yes_pct": (self.state.yes_pool * 100) // total,
            "no_pct": (self.state.no_pool * 100) // total,
        }

Prediction market mechanics:

• Betting: users place bets on "yes" or "no" before the deadline. Each address can only bet once.
• Resolution: only the owner (oracle) can resolve the market after the deadline passes.
• Payout: winners receive a proportional share of the total pool. If you bet 100 on "yes" and the yes pool is 500 out of a total 800, your payout is (100 * 800) / 500 = 160.
• Implied odds: the get_odds query shows the current market consensus as percentages.

Cross-Contract Composition

Panda's cross-contract call system enables composability -- contracts can build on each other:

from panda import Contract, PRC20

# PRC20 handle -- typed interface for token contracts
token = PRC20("0xABC...")
balance = token.balance_of(owner="0x123...")
token.transfer(to="0x456...", value=100)
token.transfer_from(owner="0x123...", to="0x456...", value=100)

# Generic Contract handle -- call any method on any contract
nft = Contract("0xDEF...")
owner = nft.owner_of(token_id=42)
nft.transfer_from(from_addr="0x123...", to_addr="0x456...", token_id=42)

This enables composability patterns seen throughout the DeFi contracts:

• Escrow uses PRC20.transfer_from to deposit and PRC20.transfer to release
• TokenSwap uses PRC20 handles for both token_a and token_b reserves
• NFT Marketplace uses Contract for NFT queries/transfers and PRC20 for payments
• DAO Voting (in the Governance guide) uses PRC20.balance_of and PRC20.total_supply for voting power

Try It

• Open the Playground and select the Token Swap (AMM), Escrow, or Prediction Market template
• Follow the DeFi: Token Swap tutorial for a guided AMM walkthrough
• See the Tokens & DeFi guide for PRC-20 and PRC-721 details