Python + Deriv API — Build Your First Trading Bot

pip install websockets python-dotenv pandas numpy

deriv-bot/
├── .env                    # API token (gitignored)
├── .gitignore
├── bot.py                  # Main bot logic
├── indicators.py           # RSI and other indicators
├── risk_manager.py         # Position sizing, daily loss limit
├── logger.py               # Trade logging to CSV
└── trades.csv              # Trade history output

"""RSI indicator using numpy — fast, no external libs."""
import numpy as np
from collections import deque


class RSI:
    def __init__(self, period=14):
        self.period = period
        self.prices = deque(maxlen=period + 1)
        self.last_rsi = None

    def update(self, price: float) -> float | None:
        """Add new price, return current RSI (or None if not enough data)."""
        self.prices.append(price)
        if len(self.prices) < self.period + 1:
            return None

        deltas = np.diff(np.array(self.prices))
        gains = np.where(deltas > 0, deltas, 0)
        losses = np.where(deltas < 0, -deltas, 0)

        avg_gain = gains.mean()
        avg_loss = losses.mean()

        if avg_loss == 0:
            self.last_rsi = 100.0
        else:
            rs = avg_gain / avg_loss
            self.last_rsi = 100 - (100 / (1 + rs))

        return self.last_rsi

"""Position sizing + daily loss limits. THE most important file."""
from datetime import date


class RiskManager:
    MAX_RISK_PER_TRADE = 0.02      # 2% per trade
    MAX_DAILY_LOSS_PCT = 0.05      # 5% daily stop
    MAX_DAILY_TRADES = 8           # Quality over quantity
    MAX_CONSECUTIVE_LOSSES = 3     # Cooldown trigger

    def __init__(self, starting_balance: float):
        self.balance = starting_balance
        self.day_start_balance = starting_balance
        self.current_date = date.today()
        self.trades_today = 0
        self.consecutive_losses = 0
        self.daily_pnl = 0.0

    def _reset_if_new_day(self):
        if date.today() != self.current_date:
            self.current_date = date.today()
            self.day_start_balance = self.balance
            self.trades_today = 0
            self.consecutive_losses = 0
            self.daily_pnl = 0.0

    def calculate_stake(self) -> float:
        """Returns USD amount to stake (2% of current balance)."""
        return round(self.balance * self.MAX_RISK_PER_TRADE, 2)

    def can_trade(self) -> tuple[bool, str]:
        """Check if conditions allow new trade. Returns (allowed, reason)."""
        self._reset_if_new_day()

        if self.trades_today >= self.MAX_DAILY_TRADES:
            return False, f"Max trades reached ({self.MAX_DAILY_TRADES}/day)"

        if self.consecutive_losses >= self.MAX_CONSECUTIVE_LOSSES:
            return False, f"Cooldown: {self.consecutive_losses} losses in a row"

        loss_threshold = self.day_start_balance * self.MAX_DAILY_LOSS_PCT
        if self.daily_pnl < -loss_threshold:
            return False, f"Daily loss limit hit: ${self.daily_pnl:.2f}"

        return True, "OK"

    def record_trade(self, pnl: float):
        """Update balance and counters after trade closes."""
        self.balance += pnl
        self.daily_pnl += pnl
        self.trades_today += 1
        if pnl > 0:
            self.consecutive_losses = 0
        else:
            self.consecutive_losses += 1

"""Deriv RSI bot — Python + WebSocket API."""
import asyncio
import json
import os
from datetime import datetime
import websockets
from dotenv import load_dotenv

from indicators import RSI
from risk_manager import RiskManager

load_dotenv()
TOKEN = os.getenv("DERIV_TOKEN")
APP_ID = 1089  # Public Deriv app ID for testing

# Settings
ASSET = "R_75"          # Volatility 75 Index
DURATION = 5            # 5 ticks per trade
DURATION_UNIT = "t"     # "t" = ticks
RSI_PERIOD = 14
OS_LEVEL = 30           # Oversold → BUY CALL
OB_LEVEL = 70           # Overbought → BUY PUT


class DerivBot:
    def __init__(self):
        self.ws = None
        self.rsi = RSI(RSI_PERIOD)
        self.risk = RiskManager(starting_balance=1000.0)
        self.req_id = 0
        self.last_signal = None  # avoid duplicate signals
        self.active_contract_id = None

    async def send(self, payload: dict) -> dict:
        """Send request and wait for response."""
        self.req_id += 1
        payload["req_id"] = self.req_id
        await self.ws.send(json.dumps(payload))
        async for msg in self.ws:
            data = json.loads(msg)
            if data.get("req_id") == self.req_id:
                return data

    async def authorize(self):
        result = await self.send({"authorize": TOKEN})
        if "error" in result:
            raise Exception(f"Auth failed: {result['error']['message']}")
        balance = result["authorize"]["balance"]
        self.risk.balance = balance
        print(f"[AUTH] Connected. Balance: ${balance:.2f}")

    async def subscribe_ticks(self):
        await self.send({"ticks": ASSET, "subscribe": 1})
        print(f"[SUB] Subscribed to {ASSET} ticks")

    async def buy_contract(self, contract_type: str):
        """Place CALL or PUT contract."""
        stake = self.risk.calculate_stake()
        payload = {
            "buy": 1,
            "subscribe": 1,
            "price": stake,
            "parameters": {
                "amount": stake,
                "basis": "stake",
                "contract_type": contract_type,
                "currency": "USD",
                "duration": DURATION,
                "duration_unit": DURATION_UNIT,
                "symbol": ASSET,
            }
        }
        result = await self.send(payload)
        if "error" in result:
            print(f"[ERR] Buy failed: {result['error']['message']}")
            return None
        contract_id = result["buy"]["contract_id"]
        print(f"[TRADE] {contract_type} stake=${stake} id={contract_id}")
        return contract_id

    async def handle_tick(self, tick: dict):
        """Process incoming price tick."""
        price = tick.get("quote")
        if price is None:
            return

        rsi = self.rsi.update(price)
        if rsi is None:
            return  # Need more data

        # Check risk first
        allowed, reason = self.risk.can_trade()
        if not allowed:
            return

        # Determine signal
        signal = None
        if rsi < OS_LEVEL and self.last_signal != "CALL":
            signal = "CALL"
        elif rsi > OB_LEVEL and self.last_signal != "PUT":
            signal = "PUT"

        if signal and self.active_contract_id is None:
            contract_id = await self.buy_contract(signal)
            if contract_id:
                self.last_signal = signal
                self.active_contract_id = contract_id

    async def run(self):
        url = f"wss://ws.derivws.com/websockets/v3?app_id={APP_ID}"
        async with websockets.connect(url) as ws:
            self.ws = ws
            await self.authorize()
            await self.subscribe_ticks()

            async for msg in ws:
                data = json.loads(msg)
                msg_type = data.get("msg_type")

                if msg_type == "tick":
                    await self.handle_tick(data["tick"])
                elif msg_type == "proposal_open_contract":
                    contract = data["proposal_open_contract"]
                    if contract.get("is_sold"):
                        pnl = float(contract["profit"])
                        self.risk.record_trade(pnl)
                        self.active_contract_id = None
                        emoji = "✓" if pnl > 0 else "✗"
                        print(f"[CLOSED] {emoji} P/L=${pnl:.2f} "
                              f"Balance=${self.risk.balance:.2f}")


if __name__ == "__main__":
    bot = DerivBot()
    try:
        asyncio.run(bot.run())
    except KeyboardInterrupt:
        print("\n[STOP] Bot stopped by user")

python bot.py

[AUTH] Connected. Balance: $10000.00
[SUB] Subscribed to R_75 ticks
[TRADE] CALL stake=$200.00 id=234567891
[CLOSED] ✓ P/L=$181.00 Balance=$10181.00
[TRADE] PUT stake=$203.62 id=234567892
[CLOSED] ✗ P/L=-$203.62 Balance=$9977.38
[TRADE] CALL stake=$199.55 id=234567893
[CLOSED] ✓ P/L=$180.59 Balance=$10157.97

import csv

def log_trade(self, signal, stake, pnl, balance):
    file_exists = os.path.exists("trades.csv")
    with open("trades.csv", "a", newline="") as f:
        writer = csv.writer(f)
        if not file_exists:
            writer.writerow(["timestamp", "signal", "stake",
                             "pnl", "balance"])
        writer.writerow([
            datetime.now().isoformat(),
            signal, stake, pnl, balance
        ])