DEV Community

Alex
Alex

Posted on

How to Build an Autonomous Trading Agent with Python in 2026

How to Build an Autonomous Trading Agent with Python in 2026

AI agents aren't just chatbots anymore. They trade markets, manage portfolios, and make decisions while you sleep. In this guide, I'll show you how to build one from scratch using Python — no ML degree required.


What You'll Build

A fully autonomous trading agent that:

  • Scans prediction markets for mispriced opportunities
  • Researches events using web search
  • Calculates fair probabilities and identifies edges
  • Executes trades with built-in risk management
  • Logs performance and learns from mistakes

This isn't theoretical. This architecture has been running in production on Polymarket, a blockchain-based prediction market.


The Architecture: Let AI Be the Brain

Most "AI trading" tutorials make the same mistake: they use AI as a fancy API wrapper. Instead, we'll use the AI model itself as the decision-making engine.

┌─────────────────────────────────────────┐
│              AI AGENT (Brain)           │
│  - Analyzes market data                 │
│  - Estimates fair probabilities         │
│  - Makes trading decisions              │
│  - Learns from outcomes                 │
└──────────┬──────────────┬───────────────┘
           │              │
    ┌──────▼──────┐  ┌────▼────────┐
    │  Data Layer │  │ Exec Layer  │
    │  - API calls│  │ - Place bets│
    │  - Web scrape│ │ - Risk mgmt │
    │  - Caching  │  │ - Logging   │
    └─────────────┘  └─────────────┘
Enter fullscreen mode Exit fullscreen mode

The Python code handles the plumbing. The AI handles the thinking.


Step 1: Project Structure

mkdir trading-agent && cd trading-agent
python3 -m venv .venv && source .venv/bin/activate
pip install requests python-dotenv web3
Enter fullscreen mode Exit fullscreen mode
trading-agent/
├── core/
│   ├── client.py      # API wrapper
│   ├── cache.py       # Avoid redundant calls
│   └── config.py      # Environment setup
├── agents/
│   ├── researcher.py  # Fetch & format market data
│   ├── trader.py      # Execute trades
│   └── risk.py        # Risk management gates
├── memory/
│   ├── performance.md # Trade log
│   └── strategies.md  # What works / doesn't
└── .env               # API keys (never commit this)
Enter fullscreen mode Exit fullscreen mode

Step 2: The Data Layer

First, build a lightweight API client. Here's the pattern:

# core/client.py
import requests
from core.config import API_BASE_URL

class MarketClient:
    def __init__(self, api_key: str):
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {api_key}"})

    def get_markets(self, limit=20, active=True):
        """Fetch active markets with current prices."""
        params = {"limit": limit, "active": active}
        resp = self.session.get(f"{API_BASE_URL}/markets", params=params)
        resp.raise_for_status()
        return resp.json()

    def get_market_detail(self, market_id: str):
        """Get full details for a specific market."""
        resp = self.session.get(f"{API_BASE_URL}/markets/{market_id}")
        resp.raise_for_status()
        return resp.json()

    def place_order(self, market_id: str, side: str, amount: float):
        """Place a market order."""
        payload = {
            "market_id": market_id,
            "side": side,  # "buy" or "sell"
            "amount": amount,
            "type": "market"
        }
        resp = self.session.post(f"{API_BASE_URL}/orders", json=payload)
        resp.raise_for_status()
        return resp.json()
Enter fullscreen mode Exit fullscreen mode

Add Caching to Avoid Rate Limits

# core/cache.py
import json
import time
from pathlib import Path

CACHE_DIR = Path("cache")
CACHE_TTL = 300  # 5 minutes

def cached(key: str):
    """Simple file-based cache decorator."""
    def decorator(func):
        def wrapper(*args, **kwargs):
            cache_file = CACHE_DIR / f"{key}.json"
            if cache_file.exists():
                data = json.loads(cache_file.read_text())
                if time.time() - data["timestamp"] < CACHE_TTL:
                    return data["result"]

            result = func(*args, **kwargs)
            CACHE_DIR.mkdir(exist_ok=True)
            cache_file.write_text(json.dumps({
                "timestamp": time.time(),
                "result": result
            }))
            return result
        return wrapper
    return decorator
Enter fullscreen mode Exit fullscreen mode

Step 3: Risk Management (The Most Important Part)

This is what separates a trading agent from a gambling bot. Every trade must pass through risk gates.

# agents/risk.py
from datetime import datetime
from pathlib import Path

class RiskManager:
    def __init__(self, balance: float):
        self.balance = balance
        self.daily_pnl = 0.0
        self.positions = {}

    # All limits are percentage-based — scales with any balance
    MAX_BET_PCT = 0.20       # 20% of balance per trade
    DAILY_LOSS_PCT = 0.30    # Stop trading at 30% daily loss
    MAX_POSITION_PCT = 0.50  # 50% max in one market

    def check_trade(self, market_id: str, amount: float) -> dict:
        """Returns {allowed: bool, reason: str}"""

        # Gate 1: Daily loss limit
        if self.daily_pnl <= -(self.balance * self.DAILY_LOSS_PCT):
            return {"allowed": False, "reason": "Daily loss limit hit"}

        # Gate 2: Single bet size
        max_bet = self.balance * self.MAX_BET_PCT
        if amount > max_bet:
            return {"allowed": False, "reason": f"Bet ${amount} exceeds max ${max_bet:.2f}"}

        # Gate 3: Position concentration
        current_position = self.positions.get(market_id, 0)
        max_position = self.balance * self.MAX_POSITION_PCT
        if current_position + amount > max_position:
            return {"allowed": False,
                    "reason": f"Would exceed max position in this market"}

        return {"allowed": True, "reason": "All checks passed"}

    def log_trade(self, market: str, side: str, amount: float,
                  price: float, reasoning: str):
        """Append trade to performance log."""
        log_line = (f"| {datetime.now().strftime('%Y-%m-%d %H:%M')} "
                   f"| {market} | {side} | ${amount:.2f} "
                   f"| {price:.2f} | LIVE | {reasoning} |\n")

        log_file = Path("memory/performance.md")
        with open(log_file, "a") as f:
            f.write(log_line)
Enter fullscreen mode Exit fullscreen mode

Key design decisions:

  • All limits are percentage-based — works whether you have $20 or $20,000
  • Risk checks happen before execution, not after
  • Every trade is logged automatically — you can't skip it
  • The AI doesn't get to override risk limits

Step 4: The Researcher Agent

This is where we prepare data for the AI brain:

# agents/researcher.py
from datetime import datetime, timedelta
from core.client import MarketClient
from core.cache import cached

client = MarketClient(api_key=os.getenv("API_KEY"))

@cached("markets_scan")
def scan_markets(hours: int = 48):
    """Fetch markets resolving within the given timeframe."""
    markets = client.get_markets(limit=100)
    cutoff = datetime.now() + timedelta(hours=hours)

    short_term = []
    for m in markets:
        end_date = datetime.fromisoformat(m["end_date"])
        if end_date <= cutoff:
            short_term.append({
                "id": m["id"],
                "question": m["question"],
                "end_date": m["end_date"],
                "current_price": m["price"],
                "volume": m["volume"],
                "tokens": m["tokens"]  # YES/NO token IDs
            })

    # Sort by resolution time (nearest first)
    short_term.sort(key=lambda x: x["end_date"])

    return {
        "count": len(short_term),
        "markets": short_term,
        "scanned_at": datetime.now().isoformat()
    }
Enter fullscreen mode Exit fullscreen mode

Step 5: The Trading Agent

# agents/trader.py
from agents.risk import RiskManager
from core.client import MarketClient

def execute_trade(token_id: str, amount: float, side: str,
                  market_name: str, reasoning: str):
    """Execute a trade after passing risk checks."""

    # Get current balance
    balance = client.get_balance()
    risk = RiskManager(balance)

    # Risk check
    check = risk.check_trade(token_id, amount)
    if not check["allowed"]:
        return {"success": False, "error": check["reason"]}

    # Execute
    try:
        result = client.place_order(
            market_id=token_id,
            side=side,
            amount=amount
        )

        # Log the trade
        risk.log_trade(market_name, side, amount,
                      result.get("price", 0), reasoning)

        return {"success": True, "order": result}

    except Exception as e:
        return {"success": False, "error": str(e)}
Enter fullscreen mode Exit fullscreen mode

Step 6: The AI Decision Loop

Here's the core insight: the AI model IS the trading strategy. Instead of coding rules, you give it data and let it reason:

# main.py — The decision loop
def trading_cycle():
    """
    1. Fetch market data
    2. Feed to AI for analysis
    3. AI identifies mispriced markets
    4. Execute trades that pass risk checks
    5. Log and learn
    """

    # Get fresh data
    markets = scan_markets(hours=48)
    portfolio = get_portfolio()
    past_trades = read_performance_log()
    strategies = read_strategies()

    # Package for AI analysis
    context = {
        "markets": markets,
        "balance": portfolio["balance"],
        "positions": portfolio["positions"],
        "past_performance": past_trades,
        "strategies": strategies,
        "current_time": datetime.now().isoformat()
    }

    # The AI analyzes and returns trade decisions
    # This is where YOUR AI model evaluates probabilities
    decisions = ai_analyze(context)

    # Execute each decision through risk gates
    for trade in decisions:
        result = execute_trade(
            token_id=trade["token_id"],
            amount=trade["amount"],
            side=trade["side"],
            market_name=trade["market"],
            reasoning=trade["reasoning"]
        )
        print(f"{'' if result['success'] else ''} {trade['market']}: {result}")
Enter fullscreen mode Exit fullscreen mode

The AI Analysis Prompt

The prompt is where strategy lives. Here's an effective one:

You are a prediction market trader. Analyze these markets and identify
mispriced opportunities.

For each market:
1. Estimate the TRUE probability based on your knowledge
2. Compare to the market price
3. Calculate the edge (your price - market price)
4. Only recommend trades where edge > 5%

Rules:
- Near-certainties (>90%) priced below fair value = free money
- Events that already happened but market hasn't settled = guaranteed
- Check resolution dates — prioritize markets closing tonight
- Never recommend a trade without specific reasoning

Output format: JSON array of {token_id, amount, side, market, reasoning}
Enter fullscreen mode Exit fullscreen mode

Step 7: Memory and Learning

The agent gets smarter over time by maintaining a memory system:

# After each cycle, update strategy notes
def update_strategies(trade_results):
    """Record what worked and what didn't."""
    strategies_file = Path("memory/strategies.md")

    for result in trade_results:
        if result["resolved"]:
            outcome = "WIN" if result["profit"] > 0 else "LOSS"
            lesson = f"- {outcome}: {result['market']}{result['reasoning']}\n"

            with open(strategies_file, "a") as f:
                f.write(lesson)
Enter fullscreen mode Exit fullscreen mode

What to track:

  • Win rate by market type (sports, politics, crypto)
  • Average edge captured vs expected
  • Which research methods led to the best trades
  • Markets where you consistently lose (avoid them)

Step 8: Running It Overnight

Set up a cron job or loop to run the agent on a schedule:

import time

CYCLE_INTERVAL = 600  # 10 minutes

while True:
    try:
        print(f"\n{'='*50}")
        print(f"Trading cycle: {datetime.now()}")
        trading_cycle()
    except Exception as e:
        print(f"Error in cycle: {e}")
        # Log error but don't crash

    time.sleep(CYCLE_INTERVAL)
Enter fullscreen mode Exit fullscreen mode

Or use cron:

# Run every 10 minutes
*/10 * * * * cd /path/to/trading-agent && python main.py >> logs/trading.log 2>&1
Enter fullscreen mode Exit fullscreen mode

Real Performance: What to Expect

After running this architecture in production:

  • Best edge: Events that already happened but markets haven't settled (near 100% win rate)
  • Bread and butter: Near-certainties priced at 85-95¢ — small but consistent returns
  • Biggest risk: Markets where you think you have an edge but don't
  • Key lesson: Short-dated markets (resolving in hours) are easier to price than long-dated ones

Common Pitfalls

  1. Over-betting: The risk manager exists for a reason. Don't override it.
  2. Ignoring liquidity: A 20% edge means nothing if you can't get filled.
  3. Confirmation bias: Research both sides before betting.
  4. Chasing losses: If you hit the daily loss limit, stop. Period.
  5. Not logging: Every trade should be logged. Your future self will thank you.

What's Next

  • Add sentiment analysis from social media
  • Implement portfolio optimization (Kelly criterion for bet sizing)
  • Build a dashboard to visualize performance
  • Add multi-market arbitrage detection
  • Integrate news feeds for real-time event detection

Conclusion

Building an autonomous trading agent isn't about writing a perfect algorithm. It's about building the right architecture: clean data pipelines, strict risk management, and letting a powerful AI model do the reasoning.

The code handles the plumbing. The AI handles the thinking. The risk manager keeps you alive.

Start small, log everything, and iterate.


This article was written for educational purposes. Trading involves risk. Never trade with money you can't afford to lose. Past performance does not guarantee future results.

Top comments (0)