Case Study 1: Simulating the Conditional Token Framework Lifecycle

Overview

This case study walks through the complete lifecycle of a prediction market built on the Conditional Token Framework (CTF): from condition preparation and collateral splitting, through trading, to oracle resolution and redemption. We implement the full CTF logic in Python, simulating what happens on-chain at each step, and verify the collateral invariant throughout.

By the end, you will understand exactly how every token balance changes at each stage and why the system is always fully collateralized.

Background

The Conditional Token Framework, developed by Gnosis, is the token standard underlying Polymarket, Omen, and other major decentralized prediction markets. It uses a single ERC-1155 contract to manage outcome tokens for all markets. The key operations are:

  1. prepareCondition: Registers a new market condition (question + oracle + outcome count)
  2. splitPosition: Converts collateral into outcome tokens
  3. mergePositions: Converts a complete set of outcome tokens back into collateral
  4. reportPayouts: Oracle reports the outcome
  5. redeemPositions: Winning token holders claim their collateral

Step 1: Condition Preparation

A market creator wants to create a market on: "Will Protocol X launch mainnet by September 30, 2026?"

"""
Step 1: Prepare the condition.
The conditionId is deterministically derived from the oracle, questionId,
and outcome count.
"""
import hashlib
from dataclasses import dataclass, field
from typing import Optional


def keccak256(data: bytes) -> bytes:
    """Keccak-256 hash (SHA-3 approximation for education)."""
    return hashlib.sha3_256(data).digest()


def compute_condition_id(
    oracle: str, question_id: bytes, outcome_count: int
) -> bytes:
    """Compute the CTF conditionId.

    conditionId = keccak256(abi.encodePacked(oracle, questionId, outcomeCount))
    """
    oracle_bytes = bytes.fromhex(oracle.replace("0x", "").zfill(40))
    count_bytes = outcome_count.to_bytes(32, "big")
    packed = oracle_bytes + question_id + count_bytes
    return keccak256(packed)


# Market parameters
oracle_address = "0x1234567890abcdef1234567890abcdef12345678"
question_id = keccak256(b"Will Protocol X launch mainnet by Sep 30, 2026?")
outcome_count = 2  # Binary: YES (index 0) and NO (index 1)

condition_id = compute_condition_id(oracle_address, question_id, outcome_count)
print(f"Condition ID: 0x{condition_id.hex()}")
print(f"Oracle: {oracle_address}")
print(f"Outcomes: {outcome_count} (YES=0, NO=1)")

At this point, the CTF contract stores that this condition exists with 2 outcome slots. No tokens have been minted yet.

Step 2: Computing Position IDs

Each position in the CTF is identified by a unique token ID derived from the collateral token, the condition, and the index set.

def compute_collection_id(
    condition_id: bytes, index_set: int, parent_collection_id: bytes = b'\x00' * 32
) -> bytes:
    """Compute a collection ID for a given condition and index set."""
    index_bytes = index_set.to_bytes(32, "big")
    raw = keccak256(condition_id + index_bytes)
    # XOR with parent collection ID for deep positions
    result = bytes(a ^ b for a, b in zip(raw, parent_collection_id))
    return result


def compute_position_id(collateral_token: str, collection_id: bytes) -> int:
    """Compute the ERC-1155 token ID for a position."""
    collateral_bytes = bytes.fromhex(collateral_token.replace("0x", "").zfill(40))
    raw = keccak256(collateral_bytes + collection_id)
    return int.from_bytes(raw, "big")


# Collateral: USDC on Polygon
usdc_address = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174"

# YES position: index set = 1 (binary 01, selecting outcome 0)
yes_collection = compute_collection_id(condition_id, 1)
yes_token_id = compute_position_id(usdc_address, yes_collection)

# NO position: index set = 2 (binary 10, selecting outcome 1)
no_collection = compute_collection_id(condition_id, 2)
no_token_id = compute_position_id(usdc_address, no_collection)

print(f"\nYES collection: 0x{yes_collection.hex()[:16]}...")
print(f"YES token ID:   {yes_token_id}")
print(f"NO collection:  0x{no_collection.hex()[:16]}...")
print(f"NO token ID:    {no_token_id}")

Step 3: Splitting Collateral into Outcome Tokens

Five traders enter the market by splitting USDC into YES and NO tokens.

@dataclass
class CTFState:
    """Tracks the state of the CTF simulation."""

    # ERC-1155 balances: {token_id: {address: balance}}
    balances: dict[int, dict[str, int]] = field(default_factory=dict)
    # Collateral balances (simulated USDC)
    collateral: dict[str, int] = field(default_factory=dict)
    # Collateral held by CTF
    ctf_collateral: int = 0
    # Resolution state
    payout_numerators: Optional[list[int]] = None
    payout_denominator: int = 0

    def split(self, user: str, amount: int,
              yes_id: int, no_id: int) -> None:
        """Split collateral into YES + NO tokens."""
        if self.collateral.get(user, 0) < amount:
            raise ValueError(f"{user} has insufficient collateral")

        self.collateral[user] -= amount
        self.ctf_collateral += amount

        for token_id in [yes_id, no_id]:
            if token_id not in self.balances:
                self.balances[token_id] = {}
            self.balances[token_id][user] = (
                self.balances[token_id].get(user, 0) + amount
            )

    def merge(self, user: str, amount: int,
              yes_id: int, no_id: int) -> None:
        """Merge YES + NO tokens back into collateral."""
        for token_id in [yes_id, no_id]:
            bal = self.balances.get(token_id, {}).get(user, 0)
            if bal < amount:
                raise ValueError(f"{user} has insufficient tokens")

        for token_id in [yes_id, no_id]:
            self.balances[token_id][user] -= amount

        self.ctf_collateral -= amount
        self.collateral[user] = self.collateral.get(user, 0) + amount

    def transfer(self, token_id: int, from_addr: str,
                 to_addr: str, amount: int) -> None:
        """Transfer outcome tokens between addresses."""
        bal = self.balances.get(token_id, {}).get(from_addr, 0)
        if bal < amount:
            raise ValueError("Insufficient token balance")

        self.balances[token_id][from_addr] -= amount
        if token_id not in self.balances:
            self.balances[token_id] = {}
        self.balances[token_id][to_addr] = (
            self.balances[token_id].get(to_addr, 0) + amount
        )

    def resolve(self, payouts: list[int]) -> None:
        """Oracle reports the outcome."""
        self.payout_numerators = payouts
        self.payout_denominator = sum(payouts)

    def redeem(self, user: str, yes_id: int, no_id: int) -> int:
        """Redeem winning tokens for collateral."""
        if self.payout_denominator == 0:
            raise ValueError("Market not resolved")

        token_ids = [yes_id, no_id]
        total_payout = 0

        for i, token_id in enumerate(token_ids):
            bal = self.balances.get(token_id, {}).get(user, 0)
            if bal > 0:
                payout = bal * self.payout_numerators[i] // self.payout_denominator
                total_payout += payout
                self.balances[token_id][user] = 0

        self.ctf_collateral -= total_payout
        self.collateral[user] = self.collateral.get(user, 0) + total_payout
        return total_payout

    def verify_invariant(self, yes_id: int, no_id: int) -> bool:
        """Verify the collateral invariant."""
        yes_supply = sum(self.balances.get(yes_id, {}).values())
        no_supply = sum(self.balances.get(no_id, {}).values())
        return (yes_supply == no_supply == self.ctf_collateral)


# Initialize state
state = CTFState()

# Give traders USDC
traders = {
    "Alice": 5_000_000,   # 5,000 USDC (6 decimals)
    "Bob": 3_000_000,
    "Carol": 8_000_000,
    "Dave": 2_000_000,
    "Eve": 4_000_000,
}
for name, balance in traders.items():
    state.collateral[name] = balance

# Each trader splits collateral
splits = [
    ("Alice", 2_000_000),
    ("Bob", 1_500_000),
    ("Carol", 3_000_000),
    ("Dave", 1_000_000),
    ("Eve", 2_500_000),
]

print("\n--- SPLITTING PHASE ---")
for name, amount in splits:
    state.split(name, amount, yes_token_id, no_token_id)
    print(f"{name} splits {amount/1e6:.2f} USDC -> "
          f"{amount/1e6:.2f} YES + {amount/1e6:.2f} NO")

print(f"\nInvariant check: {state.verify_invariant(yes_token_id, no_token_id)}")
print(f"CTF collateral: {state.ctf_collateral/1e6:.2f} USDC")

Step 4: Secondary Market Trading

Traders exchange outcome tokens based on their beliefs about the outcome.

print("\n--- TRADING PHASE ---")

# Alice is bullish: buys Bob's NO tokens (gives USDC via separate trade)
# Simulated: Bob sells his NO tokens to Alice at $0.35 each
trade_amount = 1_500_000  # 1.5M NO tokens
state.transfer(no_token_id, "Bob", "Alice", trade_amount)
print(f"Bob sells {trade_amount/1e6:.2f} NO to Alice")

# Carol is bearish: sells YES to Eve
trade_amount_2 = 1_000_000
state.transfer(yes_token_id, "Carol", "Eve", trade_amount_2)
print(f"Carol sells {trade_amount_2/1e6:.2f} YES to Eve")

# Dave merges his positions (exits market)
state.merge("Dave", 1_000_000, yes_token_id, no_token_id)
print(f"Dave merges 1.00 USDC worth of YES+NO (exits)")

print(f"\nInvariant check: {state.verify_invariant(yes_token_id, no_token_id)}")
print(f"CTF collateral: {state.ctf_collateral/1e6:.2f} USDC")

# Final positions
print("\n--- FINAL POSITIONS ---")
print(f"{'Trader':<8} {'USDC':>10} {'YES':>10} {'NO':>10}")
print("-" * 42)
for name in traders:
    usdc = state.collateral.get(name, 0) / 1e6
    yes_bal = state.balances.get(yes_token_id, {}).get(name, 0) / 1e6
    no_bal = state.balances.get(no_token_id, {}).get(name, 0) / 1e6
    print(f"{name:<8} {usdc:>10.2f} {yes_bal:>10.2f} {no_bal:>10.2f}")

Step 5: Resolution and Redemption

The oracle reports that Protocol X successfully launched. YES wins.

print("\n--- RESOLUTION ---")
state.resolve([1, 0])  # YES wins: payout = [1, 0]
print("Oracle reports: YES wins (Protocol X launched)")
print(f"Payout vector: {state.payout_numerators}")

print("\n--- REDEMPTION ---")
print(f"{'Trader':<8} {'Redeemed':>10} {'Final USDC':>12} {'P&L':>10}")
print("-" * 44)

for name in traders:
    initial = traders[name] / 1e6
    payout = state.redeem(name, yes_token_id, no_token_id)
    final = state.collateral.get(name, 0) / 1e6
    pnl = final - initial
    print(f"{name:<8} {payout/1e6:>10.2f} {final:>12.2f} {pnl:>+10.2f}")

# Final invariant
print(f"\nFinal CTF collateral: {state.ctf_collateral/1e6:.2f} USDC (should be 0)")

Key Takeaways

  1. The collateral invariant holds at every step. After every split, merge, trade, and redemption, the total collateral in the CTF contract equals the total supply of each outcome token. This is enforced by the smart contract and can be verified at any time.

  2. Splits and merges are inverses. Dave demonstrated that merging a complete set always returns exactly the original collateral, regardless of market price movements.

  3. Trading redistributes risk but does not change the total pool. When Alice buys Bob's NO tokens, the CTF collateral does not change --- only the ownership of outcome tokens shifts.

  4. Resolution zeroes out losing tokens. After YES wins, all NO tokens become worthless. The total collateral flows to YES holders in proportion to their holdings.

  5. The CTF is a zero-sum game. Total payouts exactly equal total collateral deposited (minus protocol fees, which we omitted for clarity). Winners' gains come from losers' losses.

Extensions

  • Multi-outcome markets: Extend the simulation to 3+ outcomes and verify that index sets and partitions work correctly.
  • Deep positions: Simulate positions conditional on multiple events (e.g., "YES on Event A AND YES on Event B") using collection ID XOR.
  • Fee modeling: Add a 2% protocol fee on splits and measure how it affects the payout distribution.
  • Gas cost tracking: Estimate the on-chain gas cost at each step and compute the break-even trade size.