Spaces:
No application file
No application file
File size: 4,595 Bytes
13fb76c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
Minimal async Deriv WebSocket client.
Supports the read-only operations needed by the strategy engine plus
contract buy for live mode. Token is only attached when the operation
requires it (balance, buy, portfolio).
Docs: https://api.deriv.com/api-explorer/
"""
from __future__ import annotations
import asyncio
import json
import uuid
from contextlib import asynccontextmanager
from typing import Any, Optional
import websockets
from .config import get_settings
from .schemas import Candle
class DerivError(RuntimeError):
pass
class DerivClient:
def __init__(self, app_id: Optional[str] = None,
token: Optional[str] = None):
s = get_settings()
self.app_id = app_id or s.DERIV_APP_ID
self.token = token # may be None for public ops
self.url = f"{s.DERIV_WS_URL}?app_id={self.app_id}"
@asynccontextmanager
async def _connect(self):
async with websockets.connect(self.url, max_size=2**22) as ws:
if self.token:
await self._call(ws, {"authorize": self.token})
yield ws
async def _call(self, ws, payload: dict[str, Any]) -> dict[str, Any]:
req_id = payload.get("req_id") or str(uuid.uuid4())
payload = {**payload, "req_id": req_id}
await ws.send(json.dumps(payload))
while True:
msg = json.loads(await ws.recv())
if msg.get("req_id") == req_id or msg.get("echo_req", {}).get("req_id") == req_id:
if "error" in msg:
raise DerivError(msg["error"].get("message", "Deriv error"))
return msg
# ---------- public ----------
async def candles(self, symbol: str, granularity: int = 60,
count: int = 200) -> list[Candle]:
async with self._connect() as ws:
resp = await self._call(ws, {
"ticks_history": symbol,
"adjust_start_time": 1,
"count": count,
"end": "latest",
"granularity": granularity,
"style": "candles",
})
rows = resp.get("candles", [])
return [Candle(epoch=r["epoch"], open=float(r["open"]),
high=float(r["high"]), low=float(r["low"]),
close=float(r["close"]), volume=float(r.get("volume", 0)))
for r in rows]
async def tick(self, symbol: str) -> dict[str, Any]:
async with self._connect() as ws:
resp = await self._call(ws, {"ticks": symbol})
return resp.get("tick", {})
async def active_symbols(self) -> list[dict[str, Any]]:
async with self._connect() as ws:
resp = await self._call(ws, {
"active_symbols": "brief", "product_type": "basic"})
return resp.get("active_symbols", [])
# ---------- authenticated ----------
async def balance(self) -> dict[str, Any]:
if not self.token:
raise DerivError("Token required for balance()")
async with self._connect() as ws:
resp = await self._call(ws, {"balance": 1})
return resp.get("balance", {})
async def buy_contract(self, *, symbol: str, contract_type: str,
amount: float, duration: int = 5,
duration_unit: str = "m",
currency: str = "USD") -> dict[str, Any]:
"""
contract_type: 'CALL' (BUY) or 'PUT' (SELL) for rise/fall contracts.
"""
if not self.token:
raise DerivError("Token required for buy_contract()")
proposal = {
"proposal": 1, "amount": amount, "basis": "stake",
"contract_type": contract_type, "currency": currency,
"duration": duration, "duration_unit": duration_unit,
"symbol": symbol,
}
async with self._connect() as ws:
p = await self._call(ws, proposal)
prop = p.get("proposal", {})
if "id" not in prop:
raise DerivError("No proposal id returned")
b = await self._call(ws, {"buy": prop["id"], "price": amount})
return b.get("buy", {})
async def portfolio(self) -> dict[str, Any]:
if not self.token:
raise DerivError("Token required for portfolio()")
async with self._connect() as ws:
resp = await self._call(ws, {"portfolio": 1})
return resp.get("portfolio", {})
|