AI Agent Prediction Markets API

"""
Purple Flea Prediction Markets Agent
Uses Kelly criterion to size bets optimally.
Runs as a daemon, polling every 60 seconds.
"""

import requests
import time
import json
from typing import Optional
from dataclasses import dataclass

API_KEY = "pf_live_your_key_here"
BASE_URL = "https://purpleflea.com/v1/prediction-markets"
BANKROLL_USDC = 1000.0   # Total bankroll in USDC
MAX_KELLY_FRACTION = 0.25   # Cap at 25% of Kelly to be conservative

@dataclass
class Market:
    id: str
    question: str
    protocol: str
    yes_price: float    # current probability [0,1]
    liquidity: float
    volume_24h: float
    resolution_date: str

def get_headers() -> dict:
    return {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }

def fetch_markets(min_liquidity: float = 10000) -> list[Market]:
    """Fetch all active markets above a liquidity threshold."""
    resp = requests.get(
        f"{BASE_URL}/markets",
        params={"min_liquidity": min_liquidity, "status": "active"},
        headers=get_headers(),
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()
    return [Market(**m) for m in data["markets"]]

def kelly_fraction(p_edge: float, current_price: float) -> float:
    """
    Calculate Kelly fraction given agent's estimated probability (p_edge)
    and current market price.

    f* = (p*b - q) / b  where b = (1/price - 1)
    """
    if current_price <= 0 or current_price >= 1:
        return 0.0

    b = (1.0 / current_price) - 1.0   # Net odds
    q = 1.0 - p_edge               # Probability of losing
    f_star = (p_edge * b - q) / b
    return max(0.0, min(f_star, MAX_KELLY_FRACTION))

def estimate_edge(market: Market) -> Optional[float]:
    """
    Agent's edge estimation model.
    In production: use ML model, news sentiment, or fundamental analysis.
    Returns agent's own probability estimate, or None if no edge found.
    """
    # Example: simple mean-reversion from consensus
    # Real agents would use much more sophisticated models here
    consensus = market.yes_price
    agent_estimate = consensus  # placeholder — replace with your model

    edge = abs(agent_estimate - consensus)
    if edge < 0.03:   # Less than 3% edge, skip
        return None
    return agent_estimate

def open_position(market_id: str, direction: str, usdc_amount: float) -> dict:
    """Open a YES or NO position on a market."""
    payload = {
        "market_id": market_id,
        "direction": direction,       # "YES" or "NO"
        "amount_usdc": round(usdc_amount, 2),
        "slippage_tolerance": 0.02,   # 2% max slippage
    }
    resp = requests.post(
        f"{BASE_URL}/positions/open",
        json=payload,
        headers=get_headers(),
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()

def run_agent(poll_interval: int = 60):
    """Main agent loop."""
    print("[PurpleFlea] Prediction Markets Agent started.")
    print(f"[PurpleFlea] Bankroll: ${BANKROLL_USDC} USDC | Poll: {poll_interval}s")

    while True:
        try:
            markets = fetch_markets(min_liquidity=50000)
            print(f"[PurpleFlea] Scanning {len(markets)} markets...")

            for market in markets:
                p_agent = estimate_edge(market)
                if p_agent is None:
                    continue

                direction = "YES" if p_agent > market.yes_price else "NO"
                f = kelly_fraction(p_agent, market.yes_price)
                bet_size = BANKROLL_USDC * f

                if bet_size < 5.0:   # Minimum $5 position
                    continue

                print(f"  >> Edge found: {market.question[:50]}...")
                print(f"     Market: {market.yes_price:.1%} | Agent: {p_agent:.1%} | Kelly: {f:.1%}")
                print(f"     Placing {direction} ${bet_size:.2f} on {market.protocol}")

                result = open_position(market.id, direction, bet_size)
                print(f"     Position opened: {result['position_id']}")

        except Exception as e:
            print(f"[PurpleFlea] Error: {e}")

        time.sleep(poll_interval)

if __name__ == "__main__":
    run_agent()