Python + Deriv API — Cara Bikin Trading Bot Pertama Kamu

# Install library Deriv API official
pip install python_deriv_api

# Install library tambahan untuk strategi
pip install pandas numpy

# Untuk technical indicators (RSI, EMA, dll)
pip install ta

import deriv_api
import pandas as pd
import ta

print("✓ deriv_api installed")
print("✓ pandas installed")
print("✓ ta installed")
print("\nAll dependencies ready!")

"""
First Connection — Deriv API Tutorial
IA Trader Pro
"""

import asyncio
import os
from deriv_api import DerivAPI

# === CONFIG ===
APP_ID = 1089  # Demo app_id (or get your own at developers.deriv.com)
API_TOKEN = "YOUR_TOKEN_HERE"  # Replace with your real token


async def main():
    """Main entry point"""
    api = DerivAPI(app_id=APP_ID)
    
    # Authorize with token
    auth = await api.authorize(API_TOKEN)
    print(f"✓ Connected as: {auth['authorize']['loginid']}")
    print(f"  Currency: {auth['authorize']['currency']}")
    print(f"  Balance: {auth['authorize']['balance']}")
    
    # Get account info
    balance = await api.balance()
    print(f"\n💰 Current balance: {balance['balance']['balance']} {balance['balance']['currency']}")
    
    # Clean disconnect
    await api.disconnect()
    print("\n✓ Disconnected")


if __name__ == "__main__":
    asyncio.run(main())

python deriv_bot.py

✓ Connected as: VRTC1234567
  Currency: USD
  Balance: 10000.0

💰 Current balance: 10000.0 USD

✓ Disconnected

"""
Subscribe to real-time ticks — Volatility 75 Index
"""

import asyncio
from deriv_api import DerivAPI

APP_ID = 1089
API_TOKEN = "YOUR_TOKEN_HERE"
SYMBOL = "R_75"  # Volatility 75 Index


async def main():
    api = DerivAPI(app_id=APP_ID)
    await api.authorize(API_TOKEN)
    
    print(f"📡 Subscribing to {SYMBOL} ticks...\n")
    
    # Subscribe to ticks
    source = await api.subscribe({"ticks": SYMBOL})
    
    tick_count = 0
    async for tick_data in source:
        tick = tick_data.get("tick", {})
        if tick:
            tick_count += 1
            print(f"Tick #{tick_count}: {SYMBOL} = {tick['quote']} @ {tick['epoch']}")
            
            # Stop after 10 ticks for this demo
            if tick_count >= 10:
                break
    
    await api.disconnect()


if __name__ == "__main__":
    asyncio.run(main())

📡 Subscribing to R_75 ticks...

Tick #1: R_75 = 1234.56 @ 1715433601
Tick #2: R_75 = 1234.78 @ 1715433603
Tick #3: R_75 = 1234.65 @ 1715433605
...

"""
First Trading Bot — RSI Strategy on Volatility 75
IA Trader Pro - Educational example only

Strategy:
- BUY (CALL) when RSI < 30 (oversold)
- SELL (PUT) when RSI > 70 (overbought)
- Duration: 5 ticks
- Stake: $1
- Max 10 trades per session
"""

import asyncio
import pandas as pd
import ta
from deriv_api import DerivAPI

# === CONFIG ===
APP_ID = 1089
API_TOKEN = "YOUR_TOKEN_HERE"
SYMBOL = "R_75"
STAKE = 1.0
DURATION = 5  # ticks
DURATION_UNIT = "t"  # ticks
RSI_PERIOD = 14
RSI_OVERSOLD = 30
RSI_OVERBOUGHT = 70
MAX_TRADES = 10


class TradingBot:
    def __init__(self, api):
        self.api = api
        self.prices = []
        self.trade_count = 0
        self.wins = 0
        self.losses = 0
    
    def calculate_rsi(self):
        """Calculate current RSI from price history"""
        if len(self.prices) < RSI_PERIOD + 1:
            return None
        
        df = pd.DataFrame({"close": self.prices})
        rsi_indicator = ta.momentum.RSIIndicator(df["close"], window=RSI_PERIOD)
        return rsi_indicator.rsi().iloc[-1]
    
    async def place_trade(self, contract_type):
        """Place a Rise/Fall trade"""
        try:
            proposal = await self.api.proposal({
                "proposal": 1,
                "amount": STAKE,
                "basis": "stake",
                "contract_type": contract_type,
                "currency": "USD",
                "duration": DURATION,
                "duration_unit": DURATION_UNIT,
                "symbol": SYMBOL,
            })
            
            proposal_id = proposal["proposal"]["id"]
            
            buy = await self.api.buy({
                "buy": proposal_id,
                "price": STAKE,
            })
            
            contract_id = buy["buy"]["contract_id"]
            print(f"  → Trade #{self.trade_count + 1}: {contract_type} placed (ID: {contract_id})")
            
            # Wait for contract to finish
            await asyncio.sleep(DURATION * 2.5)  # ticks are ~2 seconds on V75
            
            # Get result
            result = await self.api.proposal_open_contract({
                "proposal_open_contract": 1,
                "contract_id": contract_id,
            })
            
            profit = result["proposal_open_contract"]["profit"]
            
            if profit > 0:
                self.wins += 1
                print(f"  ✓ WIN: +${profit:.2f}")
            else:
                self.losses += 1
                print(f"  ✗ LOSS: ${profit:.2f}")
            
            self.trade_count += 1
            
        except Exception as e:
            print(f"  ✗ Trade error: {e}")
    
    async def run(self):
        """Main bot loop"""
        print(f"🤖 RSI Bot starting on {SYMBOL}")
        print(f"   Stake: ${STAKE} · Duration: {DURATION} ticks")
        print(f"   RSI: oversold={RSI_OVERSOLD}, overbought={RSI_OVERBOUGHT}\n")
        
        # Subscribe to ticks
        source = await self.api.subscribe({"ticks": SYMBOL})
        
        async for tick_data in source:
            tick = tick_data.get("tick", {})
            if not tick:
                continue
            
            price = tick["quote"]
            self.prices.append(price)
            
            # Keep only last 50 prices (rolling window)
            if len(self.prices) > 50:
                self.prices = self.prices[-50:]
            
            rsi = self.calculate_rsi()
            
            if rsi is None:
                # Still gathering data
                if len(self.prices) % 5 == 0:
                    print(f"  📊 Gathering data... ({len(self.prices)}/{RSI_PERIOD + 1})")
                continue
            
            print(f"Price: {price:.2f} · RSI: {rsi:.2f}")
            
            # Check signals
            if rsi < RSI_OVERSOLD:
                print(f"  🟢 OVERSOLD signal — placing CALL...")
                await self.place_trade("CALL")
                self.prices = self.prices[-RSI_PERIOD:]  # reset
            elif rsi > RSI_OVERBOUGHT:
                print(f"  🔴 OVERBOUGHT signal — placing PUT...")
                await self.place_trade("PUT")
                self.prices = self.prices[-RSI_PERIOD:]  # reset
            
            # Stop after max trades
            if self.trade_count >= MAX_TRADES:
                print(f"\n🏁 Max trades reached ({MAX_TRADES})")
                break
        
        # Final stats
        print(f"\n=== SESSION SUMMARY ===")
        print(f"Total trades: {self.trade_count}")
        print(f"Wins: {self.wins}")
        print(f"Losses: {self.losses}")
        if self.trade_count > 0:
            win_rate = (self.wins / self.trade_count) * 100
            print(f"Win rate: {win_rate:.1f}%")


async def main():
    api = DerivAPI(app_id=APP_ID)
    await api.authorize(API_TOKEN)
    
    bot = TradingBot(api)
    await bot.run()
    
    await api.disconnect()


if __name__ == "__main__":
    asyncio.run(main())

python rsi_bot.py

"""
Risk Management Utilities
"""

class RiskManager:
    def __init__(self, initial_balance, max_daily_loss_percent=5.0,
                 max_drawdown_percent=15.0, max_trades_per_day=20):
        self.initial_balance = initial_balance
        self.current_balance = initial_balance
        self.peak_balance = initial_balance
        self.daily_loss = 0.0
        self.trade_count_today = 0
        
        self.max_daily_loss = initial_balance * (max_daily_loss_percent / 100)
        self.max_drawdown = max_drawdown_percent / 100
        self.max_trades_per_day = max_trades_per_day
    
    def can_trade(self):
        """Check if bot is allowed to trade"""
        # Check daily loss limit
        if self.daily_loss >= self.max_daily_loss:
            print(f"  🛑 STOP: Daily loss limit reached (${self.daily_loss:.2f})")
            return False
        
        # Check max drawdown
        drawdown = (self.peak_balance - self.current_balance) / self.peak_balance
        if drawdown >= self.max_drawdown:
            print(f"  🛑 STOP: Max drawdown reached ({drawdown*100:.1f}%)")
            return False
        
        # Check max trades per day
        if self.trade_count_today >= self.max_trades_per_day:
            print(f"  🛑 STOP: Max trades per day reached ({self.trade_count_today})")
            return False
        
        return True
    
    def update_balance(self, profit):
        """Update balance after each trade"""
        self.current_balance += profit
        self.trade_count_today += 1
        
        if profit < 0:
            self.daily_loss += abs(profit)
        
        if self.current_balance > self.peak_balance:
            self.peak_balance = self.current_balance
    
    def get_stats(self):
        """Return current risk stats"""
        return {
            "balance": self.current_balance,
            "peak": self.peak_balance,
            "daily_loss": self.daily_loss,
            "trades_today": self.trade_count_today,
            "drawdown_pct": ((self.peak_balance - self.current_balance) / self.peak_balance) * 100,
        }


# Example usage
if __name__ == "__main__":
    rm = RiskManager(initial_balance=10000, max_daily_loss_percent=3.0)
    
    # Simulate trades
    rm.update_balance(-50)  # loss
    rm.update_balance(+30)  # win
    rm.update_balance(-200) # big loss
    
    if rm.can_trade():
        print("✓ Safe to trade")
    else:
        print("⛔ Should stop")
    
    print(rm.get_stats())

# Run 3 strategies simultaneously
async def main():
    api = DerivAPI(app_id=APP_ID)
    await api.authorize(API_TOKEN)
    
    bot1 = TradingBot(api, symbol="R_75", strategy="rsi")
    bot2 = TradingBot(api, symbol="R_100", strategy="ema_cross")
    bot3 = TradingBot(api, symbol="frxEURUSD", strategy="bollinger")
    
    # Run all bots concurrently
    await asyncio.gather(bot1.run(), bot2.run(), bot3.run())

# pip install python-telegram-bot

import telegram

async def send_notification(message):
    bot = telegram.Bot(token="YOUR_TELEGRAM_BOT_TOKEN")
    await bot.send_message(chat_id="YOUR_CHAT_ID", text=message)

# Usage
await send_notification(f"✓ WIN: +${profit:.2f}")

import csv
from datetime import datetime

def log_trade(trade_data):
    filename = "trades.csv"
    fieldnames = ["timestamp", "symbol", "type", "stake", "profit", "rsi"]
    
    with open(filename, "a", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        if f.tell() == 0:
            writer.writeheader()
        writer.writerow({
            "timestamp": datetime.now().isoformat(),
            **trade_data
        })