无法使用货币的 api 获取市场数据

如何解决无法使用货币的 api 获取市场数据

我的 Algo 交易系统使用 python 3.9，我有 API，但我无法获取货币市场的市场数据。交换是 IIFL，我使用的是 XTS API。下面是 conenct.py 文件

from six.moves.urllib.parse import urljoin
import json
import logging
import requests
import Exception as ex
from requests.adapters import HTTPAdapter
import configparser

log = logging.getLogger(__name__)


class XTSCommon:
    """
    Base variables class
    """

    def __init__(self,token=None,userID=None,isInvestorClient=None):
        """Initialize the common variables."""
        self.token = token
        self.userID = userID
        self.isInvestorClient = isInvestorClient


class XTSConnect(XTSCommon):
    """
    The XTS Connect API wrapper class.
    In production,you may initialise a single instance of this class per `api_key`.
    """
    """Get the configurations from config.ini"""
    cfg = configparser.ConfigParser()
    cfg.read('config.ini')

    # Default root API endpoint. It's possible to
    # override this by passing the `root` parameter during initialisation.
    _default_root_uri = cfg.get('root_url','root')
    _default_login_uri = _default_root_uri + "/user/session"
    _default_timeout = 7  # In seconds

    # SSL Flag
    _ssl_flag = cfg.get('SSL','disable_ssl')

    # Constants
    # Products
    PRODUCT_MIS = "MIS"
    PRODUCT_NRML = "NRML"

    # Order types
    ORDER_TYPE_MARKET = "MARKET"
    ORDER_TYPE_LIMIT = "LIMIT"

    # Transaction type
    TRANSACTION_TYPE_BUY = "BUY"
    TRANSACTION_TYPE_SELL = "SELL"

    # Squareoff mode
    SQUAREOFF_DAYWISE = "DayWise"
    SQUAREOFF_NETWISE = "Netwise"

    # Squareoff position quantity types
    SQUAREOFFQUANTITY_EXACTQUANTITY = "ExactQty"
    SQUAREOFFQUANTITY_PERCENTAGE = "Percentage"

    # Validity
    VALIDITY_DAY = "DAY"

    # Exchange Segments

    EXCHANGE_NSECD = "NSECD"

    # URIs to various calls
    _routes = {
        # Market API endpoints
        "marketdata.prefix": "marketdata","market.login": "/marketdata/auth/login","market.logout": "/marketdata/auth/logout","market.config": "/marketdata/config/clientConfig","market.instruments.master": "/marketdata/instruments/master","market.instruments.subscription": "/marketdata/instruments/subscription","market.instruments.unsubscription": "/marketdata/instruments/subscription","market.instruments.ohlc": "/marketdata/instruments/ohlc","market.instruments.indexlist": "/marketdata/instruments/indexlist","market.instruments.quotes": "/marketdata/instruments/quotes","market.search.instrumentsbyid": '/marketdata/search/instrumentsbyid',"market.search.instrumentsbystring": '/marketdata/search/instruments',"market.instruments.instrument.series": "/marketdata/instruments/instrument/series","market.instruments.instrument.equitysymbol": "/marketdata/instruments/instrument/symbol","market.instruments.instrument.futuresymbol": "/marketdata/instruments/instrument/futureSymbol","market.instruments.instrument.optionsymbol": "/marketdata/instruments/instrument/optionsymbol","market.instruments.instrument.optiontype": "/marketdata/instruments/instrument/optionType","market.instruments.instrument.expirydate": "/marketdata/instruments/instrument/expiryDate"
    }

    def __init__(self,apiKey,secretKey,source,root=None,debug=False,timeout=None,pool=None,disable_ssl=_ssl_flag):
        """
        Initialise a new XTS Connect client instance.

        - `api_key` is the key issued to you
        - `token` is the token obtained after the login flow. Pre-login,this will default to None,but once you have obtained it,you should persist it in a database or session to pass
        to the XTS Connect class initialisation for subsequent requests.
        - `root` is the API end point root. Unless you explicitly
        want to send API requests to a non-default endpoint,this
        can be ignored.
        - `debug`,if set to True,will serialise and print requests
        and responses to stdout.
        - `timeout` is the time (seconds) for which the API client will wait for
        a request to complete before it fails. Defaults to 7 seconds
        - `pool` is manages request pools. It takes a dict of params accepted by HTTPAdapter
        - `disable_ssl` disables the SSL verification while making a request.
        If set requests won't throw SSLError if its set to custom `root` url without SSL.
        """
        self.debug = debug
        self.apiKey = apiKey
        self.secretKey = secretKey
        self.source = source
        self.disable_ssl = disable_ssl
        self.root = root or self._default_root_uri
        self.timeout = timeout or self._default_timeout

        super().__init__()

        # Create requests session only if pool exists. Reuse session
        # for every request. Otherwise create session for each request
        if pool:
            self.reqsession = requests.Session()
            reqadapter = requests.adapters.HTTPAdapter(**pool)
            self.reqsession.mount("https://",reqadapter)
        else:
            self.reqsession = requests

        # disable requests SSL warning
        requests.packages.urllib3.disable_warnings()

    def _set_common_variables(self,access_token,userID,isInvestorClient=None):
        """Set the `access_token` received after a successful authentication."""
        super().__init__(access_token,isInvestorClient)

    def _login_url(self):
        """Get the remote login url to which a user should be redirected to initiate the login flow."""
        return self._default_login_uri

    def marketdata_login(self):
        try:
            params = {
                "appKey": self.apiKey,"secretKey": self.secretKey,"source": self.source
            }
            response = self._post("market.login",params)
            #print(response)
            if "token" in response['result']:
                self._set_common_variables(response['result']['token'],response['result']['userID'])
            return response
        except Exception as e:
            return response['description']
    def get_config(self):
        try:
            params = {}
            response = self._get('market.config',params)
            return response
        except Exception as e:
            return response['description']

    def get_future_symbol(self,exchangeSegment,series,symbol,expiryDate):
        try:
            params = {'exchangeSegment': exchangeSegment,'series': series,'symbol': symbol,'expiryDate': expiryDate}
            response = self._get(
                'market.instruments.instrument.futuresymbol',params)
            return response
        except Exception as e:
            return response['description']

    def marketdata_logout(self):
        try:
            params = {}
            response = self._delete('market.logout',params)
            return response
        except Exception as e:
            return response['description']

    ########################################################################################################
    # Common Methods
    ########################################################################################################

    def _get(self,route,params=None):
        """Alias for sending a GET request."""
        return self._request(route,"GET",params)

    def _post(self,params=None):
        """Alias for sending a POST request."""
        return self._request(route,"POST",params)

    def _put(self,params=None):
        """Alias for sending a PUT request."""
        return self._request(route,"PUT",params)

    def _delete(self,params=None):
        """Alias for sending a DELETE request."""
        return self._request(route,"DELETE",params)

    def _request(self,method,parameters=None):
        """Make an HTTP request."""
        params = parameters if parameters else {}

        # Form a restful URL
        uri = self._routes[route].format(params)
        url = urljoin(self.root,uri)
        headers = {}

        if self.token:
            # set authorization header
            headers.update({'Content-Type': 'application/json','Authorization': self.token})

        try:
            r = self.reqsession.request(method,url,data=params if method in [
                                            "POST","PUT"] else None,params=params if method in [
                                            "GET","DELETE"] else None,headers=headers,verify=not self.disable_ssl)

        except Exception as e:
            raise e

        if self.debug:
            log.debug("Response: {code} {content}".format(
                code=r.status_code,content=r.content))

        # Validate the content type.
        if "json" in r.headers["content-type"]:
            try:
                data = json.loads(r.content.decode("utf8"))
            except ValueError:
                raise ex.XTSDataException("Couldn't parse the JSON response received from the server: {content}".format(
                    content=r.content))

            # api error
            if data.get("type"):

                if r.status_code == 400 and data["type"] == "error" and data["description"] == "Invalid Token":
                    raise ex.XTSTokenException(data["description"])

                if r.status_code == 400 and data["type"] == "error" and data["description"] == "Bad Request":
                    message = "Description: " + \
                        data["description"] + " errors: " + \
                        data['result']["errors"]
                    raise ex.XTSInputException(str(message))

            return data
        else:
            raise ex.XTSDataException("Unknown Content-Type ({content_type}) with response: ({content})".format(
                content_type=r.headers["content-type"],content=r.content))

下面是Example.py文件

from Connect import XTSConnect
API_KEY = ""
API_SECRET = ""
XTS_API_BASE_URL = "https://developers.symphonyfintech.in"
source = "WEBAPI"

"""Make the XTSConnect Object with Marketdata API appKey,secretKey and source"""
xt = XTSConnect(API_KEY,API_SECRET,source)

"""Using the object we call the login function Request"""
response = xt.marketdata_login()
print("MarketData Login: ",response)

"""Get Config Request"""
response = xt.get_config()
print('Config :',response)

"""Get Future Symbol Request"""
response = xt.get_future_symbol(
    exchangeSegment=3,series='FUTCUR',symbol='USDINR',expiryDate='28MAY')
print('Future Symbol:',str(response))

"""Marketdata Logout Request"""
response = xt.marketdata_logout()
print('Marketdata Logout :',str(response))

下面是config.ini文件

[user]
source = WEBAPI

[SSL]
disable_ssl=True

[root_url]
root = https://developers.symphonyfintech.in
;root=http://103.69.170.14:10332
broadcastMode=Full

我还收到了这两个错误，即“无法从源代码解析导入“six.moves.urllib.parse”，“响应”未定义

如果有人对如何获取 USDINR 的市场数据作为未来交易品种有任何其他想法，请告诉我。

提前致谢

无法使用货币的 api 获取市场数据

如何解决无法使用货币的 api 获取市场数据

相关推荐