Introduction: Autonomous Trading in 2026

We are in the middle of a fundamental shift in how financial markets are accessed. In 2024, algorithmic trading was a domain for institutions with Bloomberg terminals and co-location racks. In 2026, an AI agent running on a $5/month VPS can execute perpetual futures strategies across dozens of pairs, manage risk autonomously, and compound returns without human intervention.

This guide walks through building a production-ready autonomous trading bot in Python that connects to Purple Flea Trading's perpetual futures API. By the end you will have: a working SMA/RSI strategy engine, a proper risk manager, a backtester for historical validation, a systemd deployment unit, and a live P&L monitor — all in under 500 lines of well-commented Python.

This is not a tutorial about paper trading or toy simulations. Every API call in this guide works against a live production endpoint. Use a small account while testing.

⚠️
Risk warning: Perpetual futures trading involves substantial risk of loss. Never trade with funds you cannot afford to lose. The code in this guide is educational and not financial advice. Always test thoroughly with small amounts before scaling.

Architecture Overview

A robust trading bot consists of five loosely-coupled components that communicate via in-memory queues. Each component can be tested and replaced independently.

┌─────────────────────────────────────────────────────────────────────┐ │ AUTONOMOUS TRADING BOT ARCHITECTURE │ └─────────────────────────────────────────────────────────────────────┘ [1] Market Data Feed ──poll every 60s──▶ [2] Strategy Engine • GET /market/OHLCV • SMA crossover • GET /market/orderbook • RSI filter • Caches last 200 candles • Signal: LONG / SHORT / FLAT │ │ │ ▼ │ [3] Risk Manager │ • Position sizing (2% rule) │ • Stop-loss calculation │ • Drawdown circuit breaker │ │ │ ▼ │ [4] Order Executor │ • POST /positions │ • DELETE /positions/:id │ • Retry with backoff │ │ └──────────────────────────────────────────────────▼ [5] Purple Flea Trading API • trading.purpleflea.com/v1 • Real funds, real fills • REST + WebSocket

Prerequisites and Setup

Python and Dependencies

You need Python 3.11 or higher. The bot uses only the standard library plus three lightweight packages:

Install dependencies 
pip install requests pandas numpy python-dotenv

Register a Purple Flea Trading Account

Get your API key in under 30 seconds:

Register via curl 
curl -X POST https://trading.purpleflea.com/v1/auth/register \
  -H "Content-Type: application/json" \
  -d '{"agent_name": "my-trading-bot", "agent_type": "trading"}'
# Returns: {"agent_id": "agt_...", "api_key": "pf_live_..."}

Store your key in a .env file — never hardcode it in your bot:

.env (never commit this file) 
PURPLEFLEA_API_KEY=pf_live_your_key_here
TRADING_PAIR=BTC-USDT
RISK_PCT=0.02
MAX_LEVERAGE=5

Market Data Integration

The market data module fetches OHLCV candles from the Purple Flea Trading API and maintains a rolling window of the last 200 candles. It exposes a simple get_candles(pair, interval) interface to the rest of the bot.

market_data.py 
import os, time, logging, requests
import pandas as pd
from functools import lru_cache

log = logging.getLogger("market_data")

BASE    = "https://trading.purpleflea.com/v1"
API_KEY = os.environ["PURPLEFLEA_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

class MarketDataFeed:
    def __init__(self, pair: str, interval: str = "1h", limit: int = 200):
        self.pair     = pair
        self.interval = interval
        self.limit    = limit
        self._cache: pd.DataFrame | None = None
        self._last_fetch = 0.0
        self._ttl = 55  # seconds; refresh just before next candle close

    def get_candles(self) -> pd.DataFrame:
        now = time.time()
        if self._cache is not None and now - self._last_fetch < self._ttl:
            return self._cache

        for attempt in range(3):
            try:
                resp = requests.get(
                    f"{BASE}/market/{self.pair}/ohlcv",
                    params={"interval": self.interval, "limit": self.limit},
                    headers=HEADERS, timeout=8
                )
                resp.raise_for_status()
                data   = resp.json()["candles"]
                df     = pd.DataFrame(data, columns=["timestamp","open","high","low","close","volume"])
                df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
                df = df.set_index("timestamp").astype(float)
                self._cache     = df
                self._last_fetch = now
                return df
            except requests.RequestException as e:
                log.warning(f"Fetch attempt {attempt+1}/3 failed: {e}")
                time.sleep(2 ** attempt)
        raise RuntimeError("Could not fetch market data after 3 attempts")

    def get_ticker(self) -> dict:
        resp = requests.get(f"{BASE}/market/{self.pair}/ticker", headers=HEADERS, timeout=5)
        resp.raise_for_status()
        return resp.json()

Strategy Engine: SMA Crossover + RSI

The strategy engine implements a classic dual moving average crossover filtered by RSI. A long signal fires when the fast SMA crosses above the slow SMA and RSI is not in overbought territory (>70). A short signal fires on the inverse. The RSI filter reduces false signals in range-bound markets.

strategy.py 
import numpy as np
import pandas as pd
from enum import Enum

class Signal(Enum):
    LONG  = "long"
    SHORT = "short"
    FLAT  = "flat"

def compute_rsi(closes: pd.Series, period: int = 14) -> pd.Series:
    """Wilder's RSI."""
    delta  = closes.diff()
    gain   = delta.clip(lower=0)
    loss   = (-delta).clip(lower=0)
    avg_g  = gain.ewm(alpha=1/period, min_periods=period).mean()
    avg_l  = loss.ewm(alpha=1/period, min_periods=period).mean()
    rs     = avg_g / avg_l.replace(0, np.nan)
    return 100 - (100 / (1 + rs))

class SMACrossoverStrategy:
    def __init__(
        self,
        fast_period: int      = 20,
        slow_period: int      = 50,
        rsi_period:  int      = 14,
        rsi_ob:      float    = 70.0,
        rsi_os:      float    = 30.0
    ):
        self.fast   = fast_period
        self.slow   = slow_period
        self.rsi_p  = rsi_period
        self.rsi_ob = rsi_ob     # overbought — don't go long above this
        self.rsi_os = rsi_os     # oversold   — don't go short below this

    def evaluate(self, df: pd.DataFrame) -> Signal:
        """
        Returns LONG, SHORT, or FLAT based on the latest candle.
        Requires at least slow_period + rsi_period rows.
        """
        min_rows = self.slow + self.rsi_p + 5
        if len(df) < min_rows:
            raise ValueError(f"Need at least {min_rows} candles, got {len(df)}")

        closes = df["close"]
        sma_f  = closes.rolling(self.fast).mean()
        sma_s  = closes.rolling(self.slow).mean()
        rsi    = compute_rsi(closes, self.rsi_p)

        # Look at last two bars to detect crossover
        prev_diff = sma_f.iloc[-2] - sma_s.iloc[-2]
        curr_diff = sma_f.iloc[-1] - sma_s.iloc[-1]
        curr_rsi  = rsi.iloc[-1]

        golden_cross = (prev_diff < 0) and (curr_diff > 0) and (curr_rsi < self.rsi_ob)
        death_cross  = (prev_diff > 0) and (curr_diff < 0) and (curr_rsi > self.rsi_os)

        if golden_cross: return Signal.LONG
        if death_cross:  return Signal.SHORT
        return Signal.FLAT

    def indicators(self, df: pd.DataFrame) -> dict:
        """Return current indicator values for logging/monitoring."""
        closes = df["close"]
        return {
            "close":  closes.iloc[-1],
            "sma_f":  closes.rolling(self.fast).mean().iloc[-1],
            "sma_s":  closes.rolling(self.slow).mean().iloc[-1],
            "rsi":    compute_rsi(closes, self.rsi_p).iloc[-1]
        }

Risk Management

The risk manager is the most important module. A great strategy with poor risk management will blow up. The implementation enforces three rules: the 2% position sizing rule, a hard stop-loss, and a portfolio-wide drawdown circuit breaker that halts trading when losses exceed a threshold.

risk_manager.py 
import logging
from dataclasses import dataclass

log = logging.getLogger("risk")

@dataclass
class TradeParams:
    size_usd:   float
    leverage:   int
    stop_loss:  float
    take_profit: float

class RiskManager:
    def __init__(
        self,
        risk_pct:        float = 0.02,  # max 2% of balance per trade
        max_leverage:    int   = 5,
        stop_pct:        float = 0.015, # 1.5% stop-loss from entry
        reward_ratio:    float = 2.0,   # take profit = 2 * stop distance
        max_drawdown_pct:float = 0.10   # halt trading at -10% total drawdown
    ):
        self.risk_pct         = risk_pct
        self.max_leverage     = max_leverage
        self.stop_pct         = stop_pct
        self.reward_ratio     = reward_ratio
        self.max_drawdown_pct = max_drawdown_pct
        self._peak_balance    = None
        self._halted          = False

    def check_halt(self, balance: float) -> bool:
        """Returns True if trading should be halted due to drawdown."""
        if self._peak_balance is None:
            self._peak_balance = balance
        self._peak_balance = max(self._peak_balance, balance)
        drawdown = (self._peak_balance - balance) / self._peak_balance
        if drawdown >= self.max_drawdown_pct:
            if not self._halted:
                log.critical(f"CIRCUIT BREAKER: drawdown {drawdown:.1%} >= {self.max_drawdown_pct:.1%}. Halting.")
            self._halted = True
        return self._halted

    def size_trade(self, balance: float, entry_price: float, side: str) -> TradeParams:
        """
        Compute position size, leverage, stop-loss, and take-profit.
        Uses the 2% rule: risk_usd = balance * risk_pct
        Position size = risk_usd / stop_distance_pct
        """
        if self._halted:
            raise RuntimeError("Trading halted by circuit breaker")

        risk_usd   = balance * self.risk_pct
        # stop_distance_pct is 1.5%, so stop_pct = 0.015
        size_usd   = risk_usd / self.stop_pct
        leverage   = min(self.max_leverage, max(1, int(size_usd / balance)))

        stop_dist  = entry_price * self.stop_pct
        tp_dist    = stop_dist * self.reward_ratio

        if side == "long":
            stop_loss   = entry_price - stop_dist
            take_profit = entry_price + tp_dist
        else:
            stop_loss   = entry_price + stop_dist
            take_profit = entry_price - tp_dist

        log.info(
            f"Trade sized: balance={balance:.2f} risk_usd={risk_usd:.2f} "
            f"size={size_usd:.2f} leverage={leverage}x "
            f"stop={stop_loss:.4f} tp={take_profit:.4f}"
        )
        return TradeParams(
            size_usd=round(size_usd, 2),
            leverage=leverage,
            stop_loss=round(stop_loss, 4),
            take_profit=round(take_profit, 4)
        )

Full Bot Code

The main bot ties all components together in a polling loop. It checks for existing positions before opening new ones, logs every decision, and gracefully handles API errors.

bot.py — main entry point 
#!/usr/bin/env python3
"""
Purple Flea Autonomous Trading Bot
Usage: python bot.py
Env:   PURPLEFLEA_API_KEY, TRADING_PAIR, RISK_PCT, MAX_LEVERAGE
"""
import os, sys, time, signal, logging, requests
from dotenv import load_dotenv

from market_data   import MarketDataFeed
from strategy      import SMACrossoverStrategy, Signal
from risk_manager  import RiskManager

load_dotenv()

# ── Config ─────────────────────────────────────────────────
API_KEY    = os.environ["PURPLEFLEA_API_KEY"]
PAIR       = os.environ.get("TRADING_PAIR",    "BTC-USDT")
RISK_PCT   = float(os.environ.get("RISK_PCT",       "0.02"))
MAX_LEV    = int(os.environ.get(  "MAX_LEVERAGE",  "5"))
POLL_SECS  = int(os.environ.get(  "POLL_INTERVAL", "60"))   # check every 60s
BASE       = "https://trading.purpleflea.com/v1"
HEADERS    = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

# ── Logging ─────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)-8s %(name)s  %(message)s",
    handlers=[logging.StreamHandler(), logging.FileHandler("bot.log")]
)
log = logging.getLogger("bot")

# ── Components ──────────────────────────────────────────────
feed     = MarketDataFeed(PAIR, interval="1h", limit=200)
strategy = SMACrossoverStrategy(fast_period=20, slow_period=50, rsi_period=14)
risk     = RiskManager(risk_pct=RISK_PCT, max_leverage=MAX_LEV, max_drawdown_pct=0.10)

_running = True
def _shutdown(sig, frame): global _running; _running = False; log.info("Shutdown signal received")
signal.signal(signal.SIGTERM, _shutdown)
signal.signal(signal.SIGINT,  _shutdown)

def get_balance() -> float:
    r = requests.get(f"{BASE}/balance", headers=HEADERS, timeout=5)
    r.raise_for_status()
    return float(r.json()["balance_usd"])

def get_open_position() -> dict | None:
    r = requests.get(f"{BASE}/positions", headers=HEADERS, timeout=5)
    r.raise_for_status()
    positions = [p for p in r.json()["positions"] if p["pair"] == PAIR]
    return positions[0] if positions else None

def open_position(side: str, params) -> dict:
    body = {
        "pair":        PAIR,
        "side":        side,
        "size_usd":    params.size_usd,
        "leverage":    params.leverage,
        "stop_loss":   params.stop_loss,
        "take_profit": params.take_profit
    }
    r = requests.post(f"{BASE}/positions", json=body, headers=HEADERS, timeout=10)
    r.raise_for_status()
    return r.json()

def close_position(position_id: str) -> dict:
    r = requests.delete(f"{BASE}/positions/{position_id}", headers=HEADERS, timeout=10)
    r.raise_for_status()
    return r.json()

# ── Main loop ───────────────────────────────────────────────
log.info(f"Starting bot: pair={PAIR} risk={RISK_PCT:.1%} max_lev={MAX_LEV}x poll={POLL_SECS}s")

while _running:
    try:
        balance = get_balance()
        log.info(f"Balance: ${balance:.2f}")

        if risk.check_halt(balance):
            log.warning("Circuit breaker active — skipping cycle")
            time.sleep(POLL_SECS); continue

        candles  = feed.get_candles()
        signal   = strategy.evaluate(candles)
        indics   = strategy.indicators(candles)
        log.info(f"Signal={signal.value} close={indics['close']:.2f} sma_f={indics['sma_f']:.2f} sma_s={indics['sma_s']:.2f} rsi={indics['rsi']:.1f}")

        position = get_open_position()

        if signal == Signal.FLAT:
            log.info("No signal — holding")

        elif signal == Signal.LONG:
            if position and position["side"] == "short":
                log.info("Closing short before opening long")
                close_position(position["id"])
                position = None
            if not position:
                ticker = feed.get_ticker()
                params = risk.size_trade(balance, ticker["last_price"], "long")
                result = open_position("long", params)
                log.info(f"Opened LONG: id={result['id']} size=${params.size_usd}")
            else:
                log.info("Already long — holding")

        elif signal == Signal.SHORT:
            if position and position["side"] == "long":
                log.info("Closing long before opening short")
                close_position(position["id"])
                position = None
            if not position:
                ticker = feed.get_ticker()
                params = risk.size_trade(balance, ticker["last_price"], "short")
                result = open_position("short", params)
                log.info(f"Opened SHORT: id={result['id']} size=${params.size_usd}")
            else:
                log.info("Already short — holding")

    except Exception as e:
        log.error(f"Cycle error: {e}", exc_info=True)

    time.sleep(POLL_SECS)

log.info("Bot stopped cleanly")

Backtesting

Before running with real money, backtest your strategy against historical data from the Purple Flea API. The backtester simulates the same logic as the live bot with zero look-ahead bias.

backtest.py 
import pandas as pd, requests, os
from strategy     import SMACrossoverStrategy, Signal
from risk_manager import RiskManager

BASE    = "https://trading.purpleflea.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['PURPLEFLEA_API_KEY']}"}

def fetch_history(pair: str, interval: str = "1h", limit: int = 1000) -> pd.DataFrame:
    resp = requests.get(f"{BASE}/market/{pair}/ohlcv",
        params={"interval": interval, "limit": limit}, headers=HEADERS, timeout=15)
    resp.raise_for_status()
    df = pd.DataFrame(resp.json()["candles"], columns=["ts","open","high","low","close","volume"])
    df["ts"] = pd.to_datetime(df["ts"], unit="ms")
    return df.set_index("ts").astype(float)

def run_backtest(pair: str = "BTC-USDT", initial_balance: float = 1000.0) -> pd.DataFrame:
    df       = fetch_history(pair)
    strat    = SMACrossoverStrategy()
    rm       = RiskManager()
    balance  = initial_balance
    position = None   # dict with side, entry, size, sl, tp
    trades   = []

    min_rows = 55  # slow SMA + buffer
    for i in range(min_rows, len(df)):
        window = df.iloc[:i]
        price  = df.iloc[i]["close"]

        # Check stop-loss / take-profit
        if position:
            hit_sl = (position["side"] == "long"  and price <= position["sl"]) or \
                     (position["side"] == "short" and price >= position["sl"])
            hit_tp = (position["side"] == "long"  and price >= position["tp"]) or \
                     (position["side"] == "short" and price <= position["tp"])
            if hit_sl or hit_tp:
                exit_price = position["sl"] if hit_sl else position["tp"]
                pct        = (exit_price - position["entry"]) / position["entry"]
                if position["side"] == "short": pct = -pct
                pnl        = position["size"] * pct * position["lev"]
                balance   += pnl
                trades.append({"exit_ts": df.index[i], "side": position["side"],
                               "pnl": round(pnl,2), "balance": round(balance,2),
                               "exit_reason": "SL" if hit_sl else "TP"})
                position = None

        sig = strat.evaluate(window)
        if sig != Signal.FLAT and not position and not rm.check_halt(balance):
            side   = sig.value
            params = rm.size_trade(balance, price, side)
            position = {"side": side, "entry": price,
                        "size": params.size_usd, "lev": params.leverage,
                        "sl": params.stop_loss, "tp": params.take_profit}

    result = pd.DataFrame(trades)
    print(f"Trades: {len(result)} | Final balance: ${balance:.2f} | Return: {(balance/initial_balance-1)*100:.1f}%")
    if not result.empty:
        wins = result[result["pnl"] > 0]
        print(f"Win rate: {len(wins)/len(result)*100:.1f}%  Avg win: ${wins['pnl'].mean():.2f}  Avg loss: ${result[result['pnl']<=0]['pnl'].mean():.2f}")
    return result

if __name__ == "__main__":
    trades = run_backtest("BTC-USDT", initial_balance=1000.0)
    print(trades.tail(10).to_string())

Deployment with systemd

For a production deployment, run the bot as a systemd service on a Linux VPS. This gives you automatic restarts on failure, proper log capture, and clean signal handling.

/etc/systemd/system/pf-trading-bot.service 
[Unit]
Description=Purple Flea Autonomous Trading Bot
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=botuser
WorkingDirectory=/opt/pf-bot
EnvironmentFile=/opt/pf-bot/.env
ExecStart=/opt/pf-bot/venv/bin/python bot.py
Restart=on-failure
RestartSec=15s
StandardOutput=journal
StandardError=journal
SyslogIdentifier=pf-trading-bot

# Resource limits
MemoryMax=256M
CPUQuota=25%

# Security hardening
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/opt/pf-bot

[Install]
WantedBy=multi-user.target
Enable and start the service 
sudo systemctl daemon-reload
sudo systemctl enable pf-trading-bot
sudo systemctl start  pf-trading-bot

# Check status
sudo systemctl status pf-trading-bot

# Follow live logs
sudo journalctl -u pf-trading-bot -f

Example Live Results

The following table shows example P&L from a 30-day paper-trading run of the SMA(20,50)+RSI(14) strategy on BTC-USDT 1h candles with a $1,000 starting balance. This is illustrative only — past performance does not guarantee future results.

Week Trades Win Rate PnL Balance Drawdown
Week 1 4 75% +$38.20 $1,038.20 -1.2%
Week 2 6 50% -$14.60 $1,023.60 -2.8%
Week 3 3 67% +$52.40 $1,076.00 -0.6%
Week 4 5 60% +$29.80 $1,105.80 -1.8%
Total (30d) 18 61% +$105.80 $1,105.80 -2.8% max
Disclaimer: The P&L figures above are from a simulated paper-trading run and are provided for illustrative purposes only. They do not represent real trading results and do not constitute a guarantee of future performance. Algorithmic trading involves significant risk and you may lose all of your invested capital.

Conclusion

You now have all the pieces for a production autonomous trading bot: a market data feed with caching and retry logic, a SMA/RSI strategy engine, a risk manager with the 2% rule and a drawdown circuit breaker, a backtester for historical validation, and a systemd deployment unit with security hardening.

The most important lesson from building trading bots is that the risk manager is more important than the strategy. A mediocre strategy with disciplined risk management will outlast an excellent strategy run recklessly. Start small, backtest thoroughly, and never disable the circuit breaker.

From here, you can extend the bot in many directions: add more sophisticated signal sources (on-chain data, funding rates, order book imbalance), implement a portfolio allocator across multiple pairs, or wire it to an LLM for adaptive strategy selection. The Purple Flea Trading API supports all of these use cases.

Ready to deploy your trading bot?

Register a Purple Flea Trading account, claim free credits from the faucet, and start with the bot code from this guide.

Full API Reference → Agent Handbook Get Free Credits