Python में Deriv API: शुरुआत से WebSocket कनेक्शन.

01Deriv API को पढ़ना क्यों उपयोगी है

तीन वजहें इसे अच्छा टेक्निकल केस बनाती हैं: साफ़ और पब्लिक डॉक्युमेंटेशन, टेस्ट के लिए फ्री sandbox/डेमो (कुछ भी जमा किए बिना), और WebSocket प्रोटोकॉल — यही समझ आपको किसी भी मॉडर्न क्रिप्टो exchange, bridge के ज़रिए MT5, या रियल-टाइम डेटा स्ट्रीमिंग वाले किसी भी प्लेटफ़ॉर्म से जुड़ने के लिए चाहिए। यहाँ जो सीखते हैं वह सीधे Binance, Bybit, Deribit पर लागू होता है।

02एनवायरनमेंट सेटअप

आपको Python 3.11+, एक virtualenv और दो लाइब्रेरी चाहिए। टर्मिनल में:

# प्रोजेक्ट और isolated environment बनाएं
mkdir bot-deriv && cd bot-deriv
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate     # Windows

# dependencies इंस्टॉल करें
pip install websockets asyncio python-dotenv

इसके बाद Deriv पर एक फ्री डेमो अकाउंट बनाइए (बिना डिपॉजिट), सिर्फ़ एक रीड-ओनली API टोकन पाने के लिए। आपको न डिपॉजिट करना है, न ट्रेड करना है।

1. इस पर जाएँ: app.deriv.com — और डेमो अकाउंट बनाएँ
2. यहाँ जाएँ: सेटिंग्स → API टोकन
3. इस ट्यूटोरियल के लिए ऐसा टोकन बनाएँ जिसका स्कोप हो Read (सिर्फ़ रीड-ओनली)
4. टोकन कॉपी करें और उसे प्रोजेक्ट की फ़ाइल .env में रखें:

# .env
DERIV_TOKEN=seu_token_aqui
DERIV_APP_ID=1089  # पब्लिक टेस्ट app_id

03Hello World: WebSocket से कनेक्ट करना

Deriv का सर्वर WebSocket इस पते पर चलाता है: wss://ws.derivws.com/websockets/v3। सारा कम्युनिकेशन असिंक्रोनस है — आप एक JSON मैसेज भेजते हैं, सर्वर भी JSON में जवाब देता है, अक्सर स्ट्रीमिंग में। पहला संपर्क:

# conectar.py
import asyncio, json, os
import websockets
from dotenv import load_dotenv

load_dotenv()
APP_ID = os.getenv("DERIV_APP_ID")
URL    = f"wss://ws.derivws.com/websockets/v3?app_id={APP_ID}"

async def ping():
    async with websockets.connect(URL) as ws:
        await ws.send(json.dumps({"ping": 1}))
        resposta = await ws.recv()
        print("Servidor respondeu:", resposta)

asyncio.run(ping())

इससे चलाएँ: python conectar.py। आपको दिखना चाहिए: {"echo_req": {"ping": 1}, "msg_type": "pong", "ping": "pong"}। दिख गया तो बधाई — आप कनेक्ट हैं। नहीं दिखा तो फ़ायरवॉल या प्रॉक्सी जाँचिए।

04ऑथेंटिकेशन

handshake के बाद टोकन से authorize किया जाता है:

# autenticar.py
async def autenticar():
    token = os.getenv("DERIV_TOKEN")
    async with websockets.connect(URL) as ws:
        await ws.send(json.dumps({"authorize": token}))
        resp = json.loads(await ws.recv())
        if "error" in resp:
            print("Falha:", resp["error"]["message"])
        else:
            print("Conta:", resp["authorize"]["loginid"])
            print("Saldo:", resp["authorize"]["balance"])

05रियल-टाइम कैंडल स्ट्रीमिंग

यहीं WebSocket चमकता है। हर सेकंड REST पोलिंग करते रहने के बजाय आप एक स्ट्रीम सब्सक्राइब करते हैं और सर्वर खुद अपडेट पुश करता है:

# candles.py
async def stream_candles(symbol="frxEURUSD", granularity=60):
    async with websockets.connect(URL) as ws:
        await ws.send(json.dumps({
            "ticks_history": symbol,
            "adjust_start_time": 1,
            "count": 100,
            "end": "latest",
            "style": "candles",
            "granularity": granularity,
            "subscribe": 1
        }))

        while True:
            msg = json.loads(await ws.recv())
            if msg.get("msg_type") == "ohlc":
                c = msg["ohlc"]
                print(f"{c['epoch']} O:{c['open']} H:{c['high']} L:{c['low']} C:{c['close']}")

asyncio.run(stream_candles())

पैरामीटर subscribe: 1 ही असली जादू है — यह कनेक्शन खुला रखता है और हर नया कैंडल खुद पुश करता है। आपको दोबारा माँगना नहीं पड़ता।

06असली बॉट जैसा ढांचा बनाना

ऊपर का उदाहरण सिखाने के लिए है। असली बॉट में तीन अलग लेयर होती हैं: कनेक्शन (ऑटोमैटिक रीकनेक्ट, heartbeat), स्ट्रैटेजी (प्योर, टेस्ट करने लायक नियम), और एक्ज़ीक्यूशन (ऑर्डर भेजना, स्टेट मैनेज करना)। सुझाया गया ढांचा:

# structure/
# ├── client.py       # WebSocket + reconnect + heartbeat
# ├── strategy.py     # Pure functions: DataFrame दें, सिग्नल मिलेगा
# ├── executor.py     # ऑर्डर भेजता है, खुली position संभालता है
# ├── risk.py         # स्टॉप लॉस, टेक प्रॉफिट, position sizing
# └── main.py         # सब जोड़ता है

# strategy.py - exemplo: cruzamento de médias
import pandas as pd

def calcular_sinal(df: pd.DataFrame) -> str:
    df["ma_curta"] = df["close"].rolling(9).mean()
    df["ma_longa"] = df["close"].rolling(21).mean()

    ultima = df.iloc[-1]
    anterior = df.iloc[-2]

    # ऊपर की ओर crossover
    if anterior["ma_curta"] < anterior["ma_longa"] and ultima["ma_curta"] > ultima["ma_longa"]:
        return "compra"
    # नीचे की ओर crossover
    if anterior["ma_curta"] > anterior["ma_longa"] and ultima["ma_curta"] < ultima["ma_longa"]:
        return "venda"
    return "hold"

यह अलगाव मायने रखता है: प्योर फंक्शन वाली स्ट्रैटेजी को आप pytest से कुछ ही सेकंड में टेस्ट कर सकते हैं, बिना कहीं कनेक्ट किए। आप 5 साल के डेटा पर बैकटेस्ट चलाते हैं, वैलिडेट करते हैं, और उसके बाद ही असली एक्ज़ीक्यूटर से जोड़ते हैं।

07एरर हैंडलिंग और रीकनेक्शन

प्रोडक्शन में WebSocket ज़रूर गिरेगा। सर्वर रीस्टार्ट होता है, नेटवर्क डगमगाता है, timeout हो जाता है। जो बॉट दोबारा कनेक्ट नहीं करता, वह बेकार बॉट है। सुझाया गया पैटर्न:

async def conectar_com_retry():
    backoff = 1
    while True:
        try:
            async with websockets.connect(URL, ping_interval=20) as ws:
                print("Conectado")
                backoff = 1  # reset
                await rotina_principal(ws)
        except (websockets.ConnectionClosed, OSError) as e:
            print(f"Conexão caiu: {e}. Reconectando em {backoff}s")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60)  # exponential, 60s पर cap

08इस जानकारी का इस्तेमाल कहाँ करें

WebSocket + asyncio का जो पैटर्न आपने सीखा, वह स्ट्रीमिंग वाले लगभग हर मॉडर्न प्लेटफ़ॉर्म पर काम करता है। यह आपको कहाँ ले जाता है:

• Binance / Bybit — मिलती-जुलती APIs, वही फ़िलॉसफ़ी। websockets + ऑथेंटिकेशन + कैंडल स्ट्रीम। कोड लगभग वही रहता है।
• Python लाइब्रेरी से MT5 — यह प्योर WebSocket नहीं है, लेकिन "प्योर फंक्शन में स्ट्रैटेजी + एक्ज़ीक्यूटर" वाला ढांचा सीधे लागू होता है।
• TradingView Webhooks — यहाँ आप दूसरी तरफ़ होते हैं: HTTP से अलर्ट पाते हैं और उन्हें प्रोसेस करने के लिए यही पैटर्न इस्तेमाल करते हैं।
• रेगुलेटेड मार्केट (ब्राज़ील की B3) — वहाँ की घरेलू ब्रोकर अपनी प्रोप्राइटरी APIs देती हैं; स्ट्रीम की लॉजिक और लेयर अलग रखने का सिद्धांत वहाँ भी वैसा ही चलता है।