Python + Deriv API: How to Build Your First Automated Trading Bot

pip install websockets asyncio

"""
Trading Bot — Deriv API + Python
IA Trader Pro — iatraderpro.com/
Author: Dan Machado

IMPORTANT: Use only on a DEMO account!
"""

import asyncio
import json
import websockets

# ============================================
# CONFIG — Edit here
# ============================================
API_TOKEN = "YOUR_TOKEN_HERE"  # Paste your Deriv API token
SYMBOL = "R_75"                # Volatility 75 Index
CONTRACT_TYPE = "CALL"         # CALL = Rise, PUT = Fall
DURATION = 5                   # Duration in ticks
DURATION_UNIT = "t"            # t = ticks, s = seconds, m = minutes
STAKE = 1.0                    # Stake amount in USD
CURRENCY = "USD"               # Account currency
NUM_TRADES = 5                 # Number of trades to execute

API_URL = "wss://ws.derivws.com/websockets/v3?app_id=1089"

# ============================================
# BOT FUNCTIONS
# ============================================

async def connect_and_trade():
    """Connects to Deriv API and executes trades."""

    async with websockets.connect(API_URL) as ws:

        # 1. Authenticate with token
        print("🔐 Authenticating...")
        await ws.send(json.dumps({
            "authorize": API_TOKEN
        }))
        auth_response = json.loads(await ws.recv())

        if "error" in auth_response:
            print(f"❌ Authentication error: {auth_response['error']['message']}")
            return

        balance = auth_response["authorize"]["balance"]
        account = auth_response["authorize"]["loginid"]
        print(f"✅ Connected! Account: {account} | Balance: ${balance}")

        # 2. Check if asset is available
        print(f"\n📊 Checking asset {SYMBOL}...")
        await ws.send(json.dumps({
            "active_symbols": "brief",
            "product_type": "basic"
        }))
        symbols_response = json.loads(await ws.recv())

        symbol_found = False
        for sym in symbols_response.get("active_symbols", []):
            if sym["symbol"] == SYMBOL:
                symbol_found = True
                print(f"✅ Asset found: {sym['display_name']}")
                break

        if not symbol_found:
            print(f"❌ Asset {SYMBOL} not found or unavailable")
            return

        # 3. Execute trades
        wins = 0
        losses = 0
        total_profit = 0

        for i in range(NUM_TRADES):
            print(f"\n{'='*40}")
            print(f"🤖 Trade {i+1}/{NUM_TRADES}")
            print(f"{'='*40}")

            # Buy contract
            print(f"📈 Buying {CONTRACT_TYPE} | Stake: ${STAKE} | Duration: {DURATION}{DURATION_UNIT}")

            await ws.send(json.dumps({
                "buy": 1,
                "price": STAKE,
                "parameters": {
                    "contract_type": CONTRACT_TYPE,
                    "symbol": SYMBOL,
                    "duration": DURATION,
                    "duration_unit": DURATION_UNIT,
                    "currency": CURRENCY,
                    "basis": "stake",
                    "amount": STAKE
                }
            }))

            buy_response = json.loads(await ws.recv())

            if "error" in buy_response:
                print(f"❌ Buy error: {buy_response['error']['message']}")
                continue

            contract_id = buy_response["buy"]["contract_id"]
            buy_price = buy_response["buy"]["buy_price"]
            print(f"✅ Contract opened! ID: {contract_id} | Price: ${buy_price}")

            # Wait for result
            print("⏳ Waiting for result...")

            # Subscribe to contract updates
            await ws.send(json.dumps({
                "proposal_open_contract": 1,
                "contract_id": contract_id,
                "subscribe": 1
            }))

            # Wait until contract closes
            while True:
                response = json.loads(await ws.recv())

                if "proposal_open_contract" in response:
                    contract = response["proposal_open_contract"]

                    if contract.get("is_sold"):
                        profit = contract["profit"]
                        total_profit += profit

                        if profit > 0:
                            wins += 1
                            print(f"🟢 PROFIT: +${profit:.2f}")
                        else:
                            losses += 1
                            print(f"🔴 LOSS: ${profit:.2f}")

                        # Cancel subscription
                        if "id" in response.get("subscription", {}):
                            await ws.send(json.dumps({
                                "forget": response["subscription"]["id"]
                            }))
                            await ws.recv()
                        break

            # Pause between trades
            await asyncio.sleep(2)

        # 4. Final summary
        print(f"\n{'='*40}")
        print(f"📊 FINAL SUMMARY")
        print(f"{'='*40}")
        print(f"Total trades: {NUM_TRADES}")
        print(f"Wins: {wins} ✅")
        print(f"Losses: {losses} ❌")
        print(f"Win rate: {(wins/NUM_TRADES)*100:.1f}%")
        print(f"Total profit/loss: ${total_profit:.2f}")
        print(f"{'='*40}")

# ============================================
# RUN
# ============================================
if __name__ == "__main__":
    print("🤖 IA Trader Pro — Deriv API Bot")
    print("📌 MODE: Demo Account")
    print("⚠️  For educational purposes only\n")
    asyncio.run(connect_and_trade())