Python + API Deriv — Construire Votre Premier Bot Trading

"""
Deriv Trading Bot - Python WebSocket
Author: Dan Machado / IA Trader Pro
License: MIT (use at own risk)
"""
import asyncio
import json
import websockets
import logging
import csv
from datetime import datetime
from strategy import RSIStrategy
from risk_manager import RiskManager

# === CONFIGURATION ===
API_TOKEN = "VOTRE_TOKEN_DERIV_ICI"  # Remplacer
APP_ID = "1089"  # Deriv public app_id (OK pour debut)
SYMBOL = "R_75"  # Volatility 75 Index
WS_URL = f"wss://ws.derivws.com/websockets/v3?app_id={APP_ID}"

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[logging.FileHandler('bot.log'), logging.StreamHandler()]
)
log = logging.getLogger(__name__)


class DerivBot:
    def __init__(self):
        self.ws = None
        self.balance = 0
        self.starting_balance = 0
        self.strategy = RSIStrategy(period=14, oversold=30, overbought=70)
        self.risk = RiskManager(max_daily_loss_pct=5.0, position_size_pct=2.0)
        self.csv_init()
    
    def csv_init(self):
        """Init CSV log file"""
        try:
            with open('trades.csv', 'x', newline='') as f:
                w = csv.writer(f)
                w.writerow(['timestamp', 'signal', 'price', 'stake',
                           'balance_before', 'result', 'balance_after'])
        except FileExistsError:
            pass  # File exists, append mode
    
    def log_trade(self, signal, price, stake, balance_before, result, balance_after):
        with open('trades.csv', 'a', newline='') as f:
            w = csv.writer(f)
            w.writerow([datetime.now().isoformat(), signal, price, stake,
                       balance_before, result, balance_after])
    
    async def authenticate(self):
        await self.ws.send(json.dumps({"authorize": API_TOKEN}))
        msg = json.loads(await self.ws.recv())
        if "error" in msg:
            log.error(f"Auth failed: {msg['error']['message']}")
            raise Exception("Authentication failed")
        self.balance = msg["authorize"]["balance"]
        self.starting_balance = self.balance
        log.info(f"Authenticated. Balance: ${self.balance:.2f}")
    
    async def subscribe_ticks(self):
        await self.ws.send(json.dumps({
            "ticks": SYMBOL,
            "subscribe": 1
        }))
        log.info(f"Subscribed to {SYMBOL} ticks")
    
    async def place_trade(self, contract_type, stake):
        """contract_type: 'CALL' (up) or 'PUT' (down)"""
        balance_before = self.balance
        proposal_req = {
            "buy": 1,
            "price": stake,
            "parameters": {
                "amount": stake,
                "basis": "stake",
                "contract_type": contract_type,
                "currency": "USD",
                "duration": 5,
                "duration_unit": "t",  # ticks
                "symbol": SYMBOL
            }
        }
        await self.ws.send(json.dumps(proposal_req))
        msg = json.loads(await self.ws.recv())
        
        if "error" in msg:
            log.error(f"Trade error: {msg['error']['message']}")
            return None
        
        contract_id = msg["buy"]["contract_id"]
        log.info(f"{contract_type} placed: ${stake} @ {SYMBOL}, contract {contract_id}")
        
        # Wait for result (poll contract status)
        await asyncio.sleep(6)  # 5 tick duration + buffer
        await self.ws.send(json.dumps({
            "proposal_open_contract": 1,
            "contract_id": contract_id
        }))
        result_msg = json.loads(await self.ws.recv())
        
        if "proposal_open_contract" in result_msg:
            profit = result_msg["proposal_open_contract"].get("profit", 0)
            self.balance += profit
            self.log_trade(contract_type, msg["buy"].get("buy_price"),
                          stake, balance_before, profit, self.balance)
            log.info(f"Trade result: P/L=${profit:.2f}, Balance=${self.balance:.2f}")
            return profit
        return None
    
    async def process_tick(self, msg):
        if "tick" not in msg:
            return
        
        tick = msg["tick"]
        price = float(tick["quote"])
        
        # Update strategy
        signal = self.strategy.update(price)
        
        if signal is None:
            return  # Not enough data yet, or no signal
        
        # Check risk manager
        if not self.risk.can_trade(self.balance, self.starting_balance):
            log.warning("Risk limit hit. Skipping trade.")
            return
        
        # Calculate stake
        stake = self.risk.calculate_stake(self.balance)
        
        # Place trade
        contract_type = "CALL" if signal == "BUY" else "PUT"
        await self.place_trade(contract_type, stake)
    
    async def run(self):
        log.info("Starting Deriv Bot")
        async with websockets.connect(WS_URL) as ws:
            self.ws = ws
            await self.authenticate()
            await self.subscribe_ticks()
            
            try:
                async for message in ws:
                    msg = json.loads(message)
                    await self.process_tick(msg)
            except websockets.exceptions.ConnectionClosed:
                log.warning("Connection lost. Reconnecting in 5s...")
                await asyncio.sleep(5)
                await self.run()  # Restart


if __name__ == "__main__":
    bot = DerivBot()
    asyncio.run(bot.run())

"""RSI Strategy Module"""
from collections import deque

class RSIStrategy:
    def __init__(self, period=14, oversold=30, overbought=70):
        self.period = period
        self.oversold = oversold
        self.overbought = overbought
        self.prices = deque(maxlen=period + 1)
        self.gains = deque(maxlen=period)
        self.losses = deque(maxlen=period)
        self.last_rsi = None
        self.prev_rsi = None
    
    def calculate_rsi(self):
        if len(self.prices) < self.period + 1:
            return None
        
        # Calculate price changes
        changes = []
        for i in range(1, len(self.prices)):
            changes.append(self.prices[i] - self.prices[i-1])
        
        # Separate gains and losses
        gains = [c if c > 0 else 0 for c in changes]
        losses = [-c if c < 0 else 0 for c in changes]
        
        # Average
        avg_gain = sum(gains) / self.period
        avg_loss = sum(losses) / self.period
        
        if avg_loss == 0:
            return 100
        
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        return rsi
    
    def update(self, price):
        self.prices.append(price)
        rsi = self.calculate_rsi()
        
        if rsi is None:
            return None
        
        self.prev_rsi = self.last_rsi
        self.last_rsi = rsi
        
        # Signal logic: CROSS oversold/overbought
        if self.prev_rsi is None:
            return None
        
        # BUY signal: RSI crosses ABOVE oversold (was below, now above)
        if self.prev_rsi < self.oversold and rsi >= self.oversold:
            return "BUY"
        
        # SELL signal: RSI crosses BELOW overbought
        if self.prev_rsi > self.overbought and rsi <= self.overbought:
            return "SELL"
        
        return None

"""Risk Management Module"""
from datetime import datetime, date

class RiskManager:
    def __init__(self, max_daily_loss_pct=5.0, position_size_pct=2.0,
                 max_trades_per_day=8):
        self.max_daily_loss_pct = max_daily_loss_pct
        self.position_size_pct = position_size_pct
        self.max_trades_per_day = max_trades_per_day
        self.trades_today = 0
        self.last_trade_date = date.today()
        self.consecutive_losses = 0
        self.cooldown_until = None
    
    def reset_daily(self):
        """Reset counters at midnight"""
        today = date.today()
        if today != self.last_trade_date:
            self.trades_today = 0
            self.last_trade_date = today
    
    def can_trade(self, current_balance, starting_balance):
        self.reset_daily()
        
        # Check daily loss limit
        loss_pct = ((starting_balance - current_balance) / starting_balance) * 100
        if loss_pct >= self.max_daily_loss_pct:
            return False
        
        # Check trades per day
        if self.trades_today >= self.max_trades_per_day:
            return False
        
        # Check cooldown after 3 consecutive losses
        if self.consecutive_losses >= 3:
            if self.cooldown_until and datetime.now() < self.cooldown_until:
                return False
            else:
                self.consecutive_losses = 0
                self.cooldown_until = None
        
        return True
    
    def calculate_stake(self, balance):
        """Position size = balance * position_size_pct / 100"""
        stake = balance * (self.position_size_pct / 100)
        # Round to 2 decimals, minimum $1
        stake = max(round(stake, 2), 1.0)
        # Cap at $100 (safety)
        stake = min(stake, 100.0)
        self.trades_today += 1
        return stake
    
    def record_result(self, profit):
        if profit < 0:
            self.consecutive_losses += 1
            if self.consecutive_losses >= 3:
                # 60min cooldown
                from datetime import timedelta
                self.cooldown_until = datetime.now() + timedelta(minutes=60)
        else:
            self.consecutive_losses = 0

# 1. SSH dans VPS
ssh root@VOTRE_IP_VPS

# 2. Installer Python + git
apt update && apt install -y python3 python3-pip git screen

# 3. Cloner ou copier votre code
scp -r ./deriv-bot/ root@VOTRE_IP:/root/

# 4. Installer dependencies
cd /root/deriv-bot
pip3 install -r requirements.txt

# 5. Lancer en background avec screen
screen -S deriv-bot
python3 bot_main.py
# Ctrl+A puis D pour détacher (bot continue tourner)

# 6. Reconnecter session:
screen -r deriv-bot

# 7. Monitor logs:
tail -f bot.log