Build a Trading Bot with Python

#!/usr/bin/env python3
"""
Prediction market trading bot using the Rekko AI API.

Screens markets for high-edge opportunities, runs AI analysis,
generates Kelly-sized trading signals, and paper trades via shadow mode.

Usage:
    pip install httpx
    export REKKO_API_KEY="your_key_here"
    python trading_bot.py
"""

import os
import sys
import time

import httpx

API_KEY = os.environ.get("REKKO_API_KEY")
if not API_KEY:
    print("Set REKKO_API_KEY environment variable first.")
    print("Sign up free at https://rekko.ai/dashboard")
    sys.exit(1)

BANKROLL = 10_000  # Starting bankroll in dollars
MIN_SCORE = 50     # Minimum screening score to analyze
MIN_EDGE = 0.05    # Minimum edge (5%) to trade
PLATFORM = "kalshi" # "kalshi" or "polymarket"

client = httpx.Client(
    base_url="https://api.rekko.ai/v1",
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=120.0,
)


def screen_markets() -> list[dict]:
    """Screen markets and return candidates sorted by score."""
    results = client.post("/screen", json={
        "platform": PLATFORM,
        "min_volume_24h": 10000,
        "min_score": MIN_SCORE,
        "limit": 10,
    }).json()
    print(f"Screened {len(results)} markets above score {MIN_SCORE}")
    return results


def analyze_market(platform: str, market_id: str) -> dict:
    """Trigger analysis, poll until complete, and return the result."""
    resp = client.post(f"/markets/{platform}/{market_id}/analyze")
    analysis_id = resp.json()["analysis_id"]
    print(f"  Analysis started: {analysis_id}")

    while True:
        status = client.get(
            f"/markets/{platform}/{market_id}/analyze/{analysis_id}/status"
        ).json()
        if status["status"] == "complete":
            break
        if status["status"] == "error":
            raise Exception(status.get("error_message", "Analysis failed"))
        time.sleep(5)

    return client.get(f"/markets/{platform}/{market_id}/analysis").json()


def get_signal(platform: str, market_id: str) -> dict:
    """Generate a trading signal with Kelly position sizing."""
    return client.post("/signals", params={"wait": "true"}, json={
        "market": f"{platform}/{market_id}",
        "risk_limit": "medium",
    }).json()


def place_shadow_trade(platform: str, market_id: str, side: str, size_usd: float) -> dict:
    """Place a paper trade via shadow trading."""
    return client.post("/trades/shadow", json={
        "ticker": market_id,
        "platform": platform,
        "side": side,
        "size_usd": round(size_usd, 2),
    }).json()


def show_portfolio():
    """Print current shadow portfolio summary."""
    portfolio = client.get("/portfolio", params={"mode": "shadow"}).json()
    print(f"\n--- Portfolio ---")
    print(f"Open: {portfolio['open_count']} | Closed: {portfolio['closed_count']}")
    print(f"Win rate: {portfolio['win_rate']:.0%} | P&L: ${portfolio['total_pnl']:+.2f}")


def main():
    print(f"Screening {PLATFORM} markets...\n")
    candidates = screen_markets()

    if not candidates:
        print("No markets above threshold. Try lowering MIN_SCORE.")
        return

    for candidate in candidates[:3]:  # Analyze top 3
        market_id = candidate["market_id"]
        platform = candidate["platform"]
        print(f"\n{'='*60}")
        print(f"Analyzing: {candidate['title'][:55]}")
        print(f"Ticker: {market_id} | Score: {candidate['score']}")
        print(f"{'='*60}")

        analysis = analyze_market(platform, market_id)
        print(f"  Probability: {analysis['probability']:.0%}")
        print(f"  Edge: {analysis['edge']:+.2f}")
        print(f"  Risk: {analysis['risk_rating']}")

        if abs(analysis["edge"]) < MIN_EDGE:
            print(f"  Edge below {MIN_EDGE:.0%} threshold — skipping.")
            continue

        signal = get_signal(platform, market_id)
        print(f"  Signal: {signal['recommendation']}")
        print(f"  Kelly size: {signal['size_pct']:.1%} (${signal['size_pct'] * BANKROLL:.0f})")

        if signal["recommendation"] == "NO_TRADE":
            print("  No trade recommended.")
            continue

        side = "yes" if "YES" in signal["recommendation"] else "no"
        size_usd = signal["size_pct"] * BANKROLL
        trade = place_shadow_trade(platform, market_id, side, size_usd)
        print(f"  Shadow trade placed: {trade['trade_id']}")
        print(f"  {side.upper()} @ ${size_usd:.2f}")

    show_portfolio()


if __name__ == "__main__":
    main()

signal = client.post("/signals/portfolio", params={"wait": "true"}, json={
    "market": f"{platform}/{market_id}",
    "risk_limit": "medium",
    "positions": portfolio["positions"],  # Pass your current holdings
    "bankroll": BANKROLL,
}).json()

print(f"Adjusted size: {signal['size_pct']:.1%}")  # May be smaller to reduce correlation
if signal.get("concentration_warning"):
    print(f"Warning: {signal['concentration_warning']}")

arb = client.get("/arbitrage").json()

for opp in arb:
    print(f"{opp['title']}")
    print(f"  Kalshi YES: {opp['kalshi_yes_price']:.2f}")
    print(f"  Polymarket YES: {opp['polymarket_yes_price']:.2f}")
    print(f"  Spread: {opp['spread']:+.2f}")