base_client.py

from base64 import b64encode
from pathlib import Path
import random
from typing import Dict, Optional, List, Tuple, Union, Any

import asyncio
import hashlib
import hmac
import time
from Crypto.PublicKey import RSA, ECC
from Crypto.Hash import SHA256
from Crypto.Signature import pkcs1_15, eddsa
import urllib.parse as _urlencode
from operator import itemgetter
from urllib.parse import urlencode

from binance.ws.websocket_api import WebsocketAPI

from .helpers import get_loop


class BaseClient:
    API_URL = "https://api{}.binance.{}/api"
    API_TESTNET_URL = "https://testnet.binance.vision/api"
    API_DEMO_URL = "https://demo-api.binance.com/api"
    MARGIN_API_URL = "https://api{}.binance.{}/sapi"
    WEBSITE_URL = "https://www.binance.{}"
    FUTURES_URL = "https://fapi.binance.{}/fapi"
    FUTURES_TESTNET_URL = "https://testnet.binancefuture.com/fapi"
    FUTURES_DEMO_URL = "https://demo-fapi.binance.com/fapi"
    FUTURES_DATA_URL = "https://fapi.binance.{}/futures/data"
    FUTURES_DATA_TESTNET_URL = "https://testnet.binancefuture.com/futures/data"
    FUTURES_COIN_URL = "https://dapi.binance.{}/dapi"
    FUTURES_COIN_TESTNET_URL = "https://testnet.binancefuture.com/dapi"
    FUTURES_COIN_DEMO_URL = "https://demo-dapi.binance.com/dapi"
    FUTURES_COIN_DATA_URL = "https://dapi.binance.{}/futures/data"
    FUTURES_COIN_DATA_TESTNET_URL = "https://testnet.binancefuture.com/futures/data"
    OPTIONS_URL = "https://eapi.binance.{}/eapi"
    OPTIONS_TESTNET_URL = "https://testnet.binanceops.{}/eapi"
    PAPI_URL = "https://papi.binance.{}/papi"
    WS_API_URL = "wss://ws-api.binance.{}/ws-api/v3"
    WS_API_TESTNET_URL = "wss://ws-api.testnet.binance.vision/ws-api/v3"
    WS_API_DEMO_URL = "wss://demo-ws-api.binance.com/ws-api/v3"
    WS_FUTURES_URL = "wss://ws-fapi.binance.{}/ws-fapi/v1"
    WS_FUTURES_TESTNET_URL = "wss://testnet.binancefuture.com/ws-fapi/v1"
    WS_FUTURES_DEMO_URL = "wss://testnet.binancefuture.com/ws-fapi/v1"
    PUBLIC_API_VERSION = "v3"
    PRIVATE_API_VERSION = "v3"
    MARGIN_API_VERSION = "v1"
    MARGIN_API_VERSION2 = "v2"
    MARGIN_API_VERSION3 = "v3"
    MARGIN_API_VERSION4 = "v4"
    FUTURES_API_VERSION = "v1"
    FUTURES_API_VERSION2 = "v2"
    FUTURES_API_VERSION3 = "v3"
    OPTIONS_API_VERSION = "v1"
    PORTFOLIO_API_VERSION = "v1"
    PORTFOLIO_API_VERSION2 = "v2"

    BASE_ENDPOINT_DEFAULT = ""
    BASE_ENDPOINT_1 = "1"
    BASE_ENDPOINT_2 = "2"
    BASE_ENDPOINT_3 = "3"
    BASE_ENDPOINT_4 = "4"

    REQUEST_TIMEOUT: float = 10

    REQUEST_RECVWINDOW: int = 10000  # 10 seconds

    SYMBOL_TYPE_SPOT = "SPOT"

    ORDER_STATUS_NEW = "NEW"
    ORDER_STATUS_PARTIALLY_FILLED = "PARTIALLY_FILLED"
    ORDER_STATUS_FILLED = "FILLED"
    ORDER_STATUS_CANCELED = "CANCELED"
    ORDER_STATUS_PENDING_CANCEL = "PENDING_CANCEL"
    ORDER_STATUS_REJECTED = "REJECTED"
    ORDER_STATUS_EXPIRED = "EXPIRED"

    KLINE_INTERVAL_1SECOND = "1s"
    KLINE_INTERVAL_1MINUTE = "1m"
    KLINE_INTERVAL_3MINUTE = "3m"
    KLINE_INTERVAL_5MINUTE = "5m"
    KLINE_INTERVAL_15MINUTE = "15m"
    KLINE_INTERVAL_30MINUTE = "30m"
    KLINE_INTERVAL_1HOUR = "1h"
    KLINE_INTERVAL_2HOUR = "2h"
    KLINE_INTERVAL_4HOUR = "4h"
    KLINE_INTERVAL_6HOUR = "6h"
    KLINE_INTERVAL_8HOUR = "8h"
    KLINE_INTERVAL_12HOUR = "12h"
    KLINE_INTERVAL_1DAY = "1d"
    KLINE_INTERVAL_3DAY = "3d"
    KLINE_INTERVAL_1WEEK = "1w"
    KLINE_INTERVAL_1MONTH = "1M"

    SIDE_BUY = "BUY"
    SIDE_SELL = "SELL"

    ORDER_TYPE_LIMIT = "LIMIT"
    ORDER_TYPE_MARKET = "MARKET"
    ORDER_TYPE_STOP_LOSS = "STOP_LOSS"
    ORDER_TYPE_STOP_LOSS_LIMIT = "STOP_LOSS_LIMIT"
    ORDER_TYPE_TAKE_PROFIT = "TAKE_PROFIT"
    ORDER_TYPE_TAKE_PROFIT_LIMIT = "TAKE_PROFIT_LIMIT"
    ORDER_TYPE_LIMIT_MAKER = "LIMIT_MAKER"

    FUTURE_ORDER_TYPE_LIMIT = "LIMIT"
    FUTURE_ORDER_TYPE_MARKET = "MARKET"
    FUTURE_ORDER_TYPE_STOP = "STOP"
    FUTURE_ORDER_TYPE_STOP_MARKET = "STOP_MARKET"
    FUTURE_ORDER_TYPE_TAKE_PROFIT = "TAKE_PROFIT"
    FUTURE_ORDER_TYPE_TAKE_PROFIT_MARKET = "TAKE_PROFIT_MARKET"
    FUTURE_ORDER_TYPE_LIMIT_MAKER = "LIMIT_MAKER"

    TIME_IN_FORCE_GTC = "GTC"  # Good till cancelled
    TIME_IN_FORCE_IOC = "IOC"  # Immediate or cancel
    TIME_IN_FORCE_FOK = "FOK"  # Fill or kill

    ORDER_RESP_TYPE_ACK = "ACK"
    ORDER_RESP_TYPE_RESULT = "RESULT"
    ORDER_RESP_TYPE_FULL = "FULL"

    # For accessing the data returned by Client.aggregate_trades().
    AGG_ID = "a"
    AGG_PRICE = "p"
    AGG_QUANTITY = "q"
    AGG_FIRST_TRADE_ID = "f"
    AGG_LAST_TRADE_ID = "l"
    AGG_TIME = "T"
    AGG_BUYER_MAKES = "m"
    AGG_BEST_MATCH = "M"

    # new asset transfer api enum
    SPOT_TO_FIAT = "MAIN_C2C"
    SPOT_TO_USDT_FUTURE = "MAIN_UMFUTURE"
    SPOT_TO_COIN_FUTURE = "MAIN_CMFUTURE"
    SPOT_TO_MARGIN_CROSS = "MAIN_MARGIN"
    SPOT_TO_MINING = "MAIN_MINING"
    FIAT_TO_SPOT = "C2C_MAIN"
    FIAT_TO_USDT_FUTURE = "C2C_UMFUTURE"
    FIAT_TO_MINING = "C2C_MINING"
    USDT_FUTURE_TO_SPOT = "UMFUTURE_MAIN"
    USDT_FUTURE_TO_FIAT = "UMFUTURE_C2C"
    USDT_FUTURE_TO_MARGIN_CROSS = "UMFUTURE_MARGIN"
    COIN_FUTURE_TO_SPOT = "CMFUTURE_MAIN"
    MARGIN_CROSS_TO_SPOT = "MARGIN_MAIN"
    MARGIN_CROSS_TO_USDT_FUTURE = "MARGIN_UMFUTURE"
    MINING_TO_SPOT = "MINING_MAIN"
    MINING_TO_USDT_FUTURE = "MINING_UMFUTURE"
    MINING_TO_FIAT = "MINING_C2C"

    ## order ids
    SPOT_ORDER_PREFIX = "x-HNA2TXFJ"
    CONTRACT_ORDER_PREFIX = "x-Cb7ytekJ"

    def __init__(
        self,
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
        requests_params: Optional[Dict[str, Any]] = None,
        tld: str = "com",
        base_endpoint: str = BASE_ENDPOINT_DEFAULT,
        testnet: bool = False,
        demo: bool = False,
        private_key: Optional[Union[str, Path]] = None,
        private_key_pass: Optional[str] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
        time_unit: Optional[str] = None,
    ):
        """Binance API Client constructor

        :param api_key: Api Key
        :type api_key: str.
        :param api_secret: Api Secret
        :type api_secret: str.
        :param requests_params: optional - Dictionary of requests params to use for all calls
        :type requests_params: dict.
        :param testnet: Use testnet environment - only available for vanilla options at the moment
        :type testnet: bool
        :param private_key: Path to private key, or string of file contents
        :type private_key: optional - str or Path
        :param private_key_pass: Password of private key
        :type private_key_pass: optional - str
        :param time_unit: Time unit to use for requests. Supported values: "MILLISECOND", "MICROSECOND"
        :type time_unit: optional - str

        """

        self.tld = tld
        self.API_URL = self.API_URL.format(base_endpoint, tld)
        self.MARGIN_API_URL = self.MARGIN_API_URL.format(base_endpoint, tld)
        self.WEBSITE_URL = self.WEBSITE_URL.format(tld)
        self.FUTURES_URL = self.FUTURES_URL.format(tld)
        self.FUTURES_DATA_URL = self.FUTURES_DATA_URL.format(tld)
        self.FUTURES_COIN_URL = self.FUTURES_COIN_URL.format(tld)
        self.FUTURES_COIN_DATA_URL = self.FUTURES_COIN_DATA_URL.format(tld)
        self.OPTIONS_URL = self.OPTIONS_URL.format(tld)
        self.OPTIONS_TESTNET_URL = self.OPTIONS_TESTNET_URL.format(tld)

        self.API_KEY = api_key
        self.API_SECRET = api_secret
        self.TIME_UNIT = time_unit
        self._is_rsa = False
        self.PRIVATE_KEY: Any = self._init_private_key(private_key, private_key_pass)
        self.session = self._init_session()
        self._requests_params = requests_params
        self.response = None
        self.testnet = testnet
        self.demo = demo
        self.timestamp_offset = 0
        ws_api_url = self.WS_API_URL.format(tld)
        if testnet:
            ws_api_url = self.WS_API_TESTNET_URL
        elif demo:
            ws_api_url = self.WS_API_DEMO_URL
        if self.TIME_UNIT:
            ws_api_url += f"?timeUnit={self.TIME_UNIT}"
        # Extract proxy from requests_params for WebSocket connections
        https_proxy = None
        if requests_params and 'proxies' in requests_params:
            https_proxy = requests_params['proxies'].get('https') or requests_params['proxies'].get('http')
        
        self.ws_api = WebsocketAPI(url=ws_api_url, tld=tld, https_proxy=https_proxy)
        ws_future_url = self.WS_FUTURES_URL.format(tld)
        if testnet:
            ws_future_url = self.WS_FUTURES_TESTNET_URL
        elif demo:
            ws_future_url = self.WS_FUTURES_DEMO_URL
        self.ws_future = WebsocketAPI(url=ws_future_url, tld=tld, https_proxy=https_proxy)
        self.loop = loop or get_loop()

    def _get_headers(self) -> Dict:
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36",  # noqa
        }
        if self.API_KEY:
            assert self.API_KEY
            headers["X-MBX-APIKEY"] = self.API_KEY
        if self.TIME_UNIT:
            assert self.TIME_UNIT
            headers["X-MBX-TIME-UNIT"] = self.TIME_UNIT
        return headers

    def _init_session(self):
        raise NotImplementedError

    def _init_private_key(
        self,
        private_key: Optional[Union[str, Path]],
        private_key_pass: Optional[str] = None,
    ):
        if not private_key:
            return
        if isinstance(private_key, Path):
            with open(private_key, "r") as f:
                private_key = f.read()
        if len(private_key) > 120:
            self._is_rsa = True
            return RSA.import_key(private_key, passphrase=private_key_pass)
        return ECC.import_key(private_key)

    def _create_api_uri(
        self, path: str, signed: bool = True, version: str = PUBLIC_API_VERSION
    ) -> str:
        url = self.API_URL
        if self.testnet:
            url = self.API_TESTNET_URL
        elif self.demo:
            url = self.API_DEMO_URL
        v = self.PRIVATE_API_VERSION if signed else version
        return url + "/" + v + "/" + path

    def _create_margin_api_uri(self, path: str, version: int = 1) -> str:
        options = {
            1: self.MARGIN_API_VERSION,
            2: self.MARGIN_API_VERSION2,
            3: self.MARGIN_API_VERSION3,
            4: self.MARGIN_API_VERSION4,
        }
        return self.MARGIN_API_URL + "/" + options[version] + "/" + path

    def _create_papi_api_uri(self, path: str, version: int = 1) -> str:
        options = {1: self.PORTFOLIO_API_VERSION, 2: self.PORTFOLIO_API_VERSION2}
        return self.PAPI_URL.format(self.tld) + "/" + options[version] + "/" + path

    def _create_website_uri(self, path: str) -> str:
        return self.WEBSITE_URL + "/" + path

    def _create_futures_api_uri(self, path: str, version: int = 1) -> str:
        url = self.FUTURES_URL
        if self.testnet:
            url = self.FUTURES_TESTNET_URL
        elif self.demo:
            url = self.FUTURES_DEMO_URL
        options = {
            1: self.FUTURES_API_VERSION,
            2: self.FUTURES_API_VERSION2,
            3: self.FUTURES_API_VERSION3,
        }
        return url + "/" + options[version] + "/" + path

    def _create_futures_data_api_uri(self, path: str) -> str:
        url = self.FUTURES_DATA_URL
        if self.testnet:
            url = self.FUTURES_DATA_TESTNET_URL
        return url + "/" + path

    def _create_futures_coin_api_url(self, path: str, version: int = 1) -> str:
        url = self.FUTURES_COIN_URL
        if self.testnet:
            url = self.FUTURES_COIN_TESTNET_URL
        elif self.demo:
            url = self.FUTURES_COIN_DEMO_URL
        options = {
            1: self.FUTURES_API_VERSION,
            2: self.FUTURES_API_VERSION2,
            3: self.FUTURES_API_VERSION3,
        }
        return url + "/" + options[version] + "/" + path

    def _create_futures_coin_data_api_url(self, path: str, version: int = 1) -> str:
        url = self.FUTURES_COIN_DATA_URL
        if self.testnet:
            url = self.FUTURES_COIN_DATA_TESTNET_URL
        return url + "/" + path

    def _create_options_api_uri(self, path: str) -> str:
        url = self.OPTIONS_URL
        if self.testnet:
            url = self.OPTIONS_TESTNET_URL
        return url + "/" + self.OPTIONS_API_VERSION + "/" + path

    def _rsa_signature(self, query_string: str):
        assert self.PRIVATE_KEY
        h = SHA256.new(query_string.encode("utf-8"))
        signature = pkcs1_15.new(self.PRIVATE_KEY).sign(h)  # type: ignore
        res = b64encode(signature).decode()
        # return self.encode_uri_component(res)
        return res

    @staticmethod
    def encode_uri_component(uri, safe="~()*!.'"):
        return _urlencode.quote(uri, safe=safe)

    @staticmethod
    def convert_to_dict(list_tuples):
        dictionary = dict((key, value) for key, value in list_tuples)
        return dictionary

    def _require_tld(self, required_tld: str, endpoint_name: str = "endpoint") -> None:
        """Validate client is configured for required TLD.

        :param required_tld: The required TLD (e.g., "us")
        :param endpoint_name: Description of the endpoint for error messages
        :raises BinanceRegionException: If the client TLD doesn't match
        """
        if self.tld != required_tld:
            from binance.exceptions import BinanceRegionException

            raise BinanceRegionException(required_tld, self.tld, endpoint_name)

    def _ed25519_signature(self, query_string: str):
        assert self.PRIVATE_KEY
        res = b64encode(
            eddsa.new(self.PRIVATE_KEY, "rfc8032").sign(query_string.encode())
        ).decode()  # type: ignore
        # return self.encode_uri_component(res)
        return res

    def _hmac_signature(self, query_string: str) -> str:
        assert self.API_SECRET, "API Secret required for private endpoints"
        m = hmac.new(
            self.API_SECRET.encode("utf-8"),
            query_string.encode("utf-8"),
            hashlib.sha256,
        )
        return m.hexdigest()

    def _generate_signature(self, data: Dict, uri_encode=True) -> str:
        sig_func = self._hmac_signature
        if self.PRIVATE_KEY:
            if self._is_rsa:
                sig_func = self._rsa_signature
            else:
                sig_func = self._ed25519_signature
        query_string = "&".join([f"{d[0]}={_urlencode.quote(d[1]) if d[0] == 'symbol' else d[1]}" for d in self._order_params(data)])
        res = sig_func(query_string)
        return self.encode_uri_component(res) if uri_encode else res

    def _sign_ws_params(self, params, signature_func):
        if "signature" in params:
            return params
        params.setdefault("apiKey", self.API_KEY)
        params.setdefault("timestamp", int(time.time() * 1000 + self.timestamp_offset))
        params = dict(sorted(params.items()))
        return {**params, "signature": signature_func(params)}

    def _generate_ws_api_signature(self, data: Dict) -> str:
        sig_func = self._hmac_signature
        if self.PRIVATE_KEY:
            if self._is_rsa:
                sig_func = self._rsa_signature
            else:
                sig_func = self._ed25519_signature
        query_string = urlencode(data)
        return sig_func(query_string)

    async def _ws_futures_api_request(self, method: str, signed: bool, params: dict):
        """Send request and wait for response"""
        id = params.pop("id", self.uuid22())
        payload = {
            "id": id,
            "method": method,
            "params": params,
        }
        if signed:
            payload["params"] = self._sign_ws_params(params, self._generate_signature)
        return await self.ws_future.request(id, payload)

    def _ws_futures_api_request_sync(self, method: str, signed: bool, params: dict):
        self.loop = get_loop()
        return self.loop.run_until_complete(
            self._ws_futures_api_request(method, signed, params)
        )

    async def _make_sync(self, method):
        return asyncio.run(method)

    async def _ws_api_request(self, method: str, signed: bool, params: dict):
        """Send request and wait for response"""
        id = params.pop("id", self.uuid22())
        payload = {
            "id": id,
            "method": method,
            "params": params,
        }
        if signed:
            payload["params"] = self._sign_ws_params(
                params, self._generate_ws_api_signature
            )
        return await self.ws_api.request(id, payload)

    def _ws_api_request_sync(self, method: str, signed: bool, params: dict):
        """Send request to WS API and wait for response"""
        self.loop = get_loop()
        return self.loop.run_until_complete(
            self._ws_api_request(method, signed, params)
        )

    @staticmethod
    def _get_version(version: int, **kwargs) -> int:
        if "data" in kwargs and "version" in kwargs["data"]:
            version_override = kwargs["data"].get("version")
            del kwargs["data"]["version"]
            return version_override
        return version

    @staticmethod
    def uuid22(length=22):
        return format(random.getrandbits(length * 4), "x")

    @staticmethod
    def _order_params(data: Dict) -> List[Tuple[str, str]]:
        """Convert params to list with signature as last element

        :param data:
        :return:

        """
        data = dict(filter(lambda el: el[1] is not None, data.items()))
        has_signature = False
        params = []
        for key, value in data.items():
            if key == "signature":
                has_signature = True
            else:
                params.append((key, str(value)))
        # sort parameters by key
        params.sort(key=itemgetter(0))
        if has_signature:
            params.append(("signature", data["signature"]))
        return params

    def _get_request_kwargs(
        self, method, signed: bool, force_params: bool = False, **kwargs
    ) -> Dict:
        # set default requests timeout
        kwargs["timeout"] = self.REQUEST_TIMEOUT

        # add our global requests params
        if self._requests_params:
            kwargs.update(self._requests_params)

        data = kwargs.get("data", None)
        if data and isinstance(data, dict):
            kwargs["data"] = data
            # find any requests params passed and apply them
            if "requests_params" in kwargs["data"]:
                # merge requests params into kwargs
                kwargs.update(kwargs["data"]["requests_params"])
                del kwargs["data"]["requests_params"]

        if signed:
            # generate signature
            kwargs["data"]["timestamp"] = int(
                time.time() * 1000 + self.timestamp_offset
            )
            if self.REQUEST_RECVWINDOW:
                kwargs["data"]["recvWindow"] = self.REQUEST_RECVWINDOW
            kwargs["data"]["signature"] = self._generate_signature(kwargs["data"])

        # sort get and post params to match signature order
        if data:
            # sort post params and remove any arguments with values of None
            kwargs["data"] = self._order_params(kwargs["data"])
            # Remove any arguments with values of None.
            null_args = [
                i for i, (key, value) in enumerate(kwargs["data"]) if value is None
            ]
            for i in reversed(null_args):
                del kwargs["data"][i]

        # if get request assign data array to params value for requests lib
        if data and (method == "get" or force_params):
            kwargs["params"] = "&".join(
                "%s=%s" % (data[0], _urlencode.quote(data[1]) if data[0] == 'symbol' else data[1]) for data in kwargs["data"]
            )
            del kwargs["data"]

        # Temporary fix for Signature issue while using batchOrders in AsyncClient
        if "params" in kwargs.keys():
            if (
                "batchOrders" in kwargs["params"]
                or "orderidlist" in kwargs["params"]
                or "origclientorderidlist" in kwargs["params"]
            ):
                kwargs["data"] = kwargs["params"]
                del kwargs["params"]

        return kwargs