问题描述
我正在使用 python 和 IIFL 证券账户进行算法交易，我们正在使用 xts-pythonclient-api-sdk 我正在运行 Example.py 文件
#Marketdata API Credentials
API_KEY = "YOUR_API_KEY_HERE"
API_SECRET = "YOUR_API_SECRET_HERE"
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)
下面我附上了connect.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_NSECM = "NSECM"
EXCHANGE_NSEFO = "NSEFO"
EXCHANGE_NSECD = "NSECD"
EXCHANGE_MCXFO = "MCXFO"
EXCHANGE_BSECM = "BSECM"
# 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
########################################################################################################
# Market data API
########################################################################################################
def marketdata_login(self):
try:
params = {
"appKey": self.apiKey,"secretKey": self.secretKey,"source": self.source
}
response = self._post("market.login",params)
if "token" in response['result']:
self._set_common_variables(response['result']['token'],response['result']['userID'])
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))
我得到的错误：
{'type': 'error','code': 'e-response-0005','description': 'Your credentials are invalid'}
MarketData Login: Your credentials are invalid
Config : {'type': 'error','code': 'e-token-0006','description': 'Please Provide token to Authenticate'}
Quote : {'type': 'error','description': 'Please Provide token to Authenticate'}
Subscribe : {'type': 'error','description': 'Please Provide token to Authenticate'}
Unsubscribe : {'type': 'error','description': 'Please Provide token to Authenticate'}
Traceback (most recent call last):
File "Connect.py",line 526,in get_master
response = self._post('market.instruments.master',json.dumps(params))
File "Connect.py",line 636,in _post
return self._request(route,params)
File "Connect.py",line 688,in _request
message = "Description: " + data["description"] + " errors: " + data['result']["errors"]
TypeError: must be str,not list
During handling of the above exception,another exception occurred:
Traceback (most recent call last):
File "Example.py",line 199,in <module>
response = xt.get_master(exchangeSegmentList=exchangesegments)
et_master
return response['description']
UnboundLocalError: local variable 'response' referenced before assignment
如需进一步参考，请参阅下面的 github 代码链接： https://github.com/symphonyfintech/xts-pythonclient-api-sdk
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）