Classes:
| Name | Description |
|---|---|
WisdomCapitalAPIException |
Raised for Wisdom Capital broker API errors. |
WisdomCapitalAPI |
Wisdom Capital API class |
Bases: BrokerAPIException
Raised for Wisdom Capital broker API errors.
Example programs
examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPIException/example_1_catching_a_refused_token.py
"""Catches the `WisdomCapitalAPIException` that `WisdomCapitalAPI` raises when Wisdom Capital refuses the access token.
When Wisdom Capital answers with an HTTP error, `WisdomCapitalAPI`'s `_request` raises `WisdomCapitalAPIException` instead of returning an answer. Symphony XTS names the error in `code` and explains it in `description`, and `_request` copies both into the exception.
The real constructor reads the broker's settings from MongoDB, probes `GET /interactive/user/balance` and the market data application's `clientConfig`, and logs in to Wisdom Capital with a session request for the interactive application and a second login for the market data application when the stored token no longer works. This program must never log in, so a small subclass, `WisdomCapitalAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Wisdom Capital response, so nothing reaches the network. Here the stand-in server answers with the HTTP 400 Wisdom Capital sends for a token it no longer accepts.
Notice that `code` holds Wisdom Capital's own error code rather than the HTTP status, so a caller can tell an expired token from other refusals.
Run it from the project root:
python examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPIException/example_1_catching_a_refused_token.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.wisdom_capital import (
WisdomCapitalAPI,
WisdomCapitalAPIException,
)
STORED_LOGIN = {
'broker_name': 'wisdom_capital',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
REFUSAL = {
'type': 'error',
'code': 'e-token-0002',
'description': 'Please Provide token to Authenticate',
}
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Wisdom Capital response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class WisdomCapitalAPIWithoutLogin(WisdomCapitalAPI):
"""A `WisdomCapitalAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `wisdom_capital` settings document from MongoDB and then logs in to Wisdom Capital if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `WisdomCapitalAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'wisdom_capital'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('wisdom_capital')
self._settings = settings
self._last_login = stored_login
self.market_data_session_error = None
class CatchingARefusedTokenExample:
"""Sends a request that Wisdom Capital refuses and prints the exception it raises.
Attributes:
server (RecordedBrokerServer): The stand-in for Wisdom Capital's API, answering with a refusal.
api (WisdomCapitalAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores an expired login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
cache = SharedLoginCache()
cache.hset('last_login', 'wisdom_capital', json.dumps(STORED_LOGIN))
settings = {
'ucc_code': 'WC0001',
}
self.server = RecordedBrokerServer(400, 'application/json', json.dumps(REFUSAL))
self.api = WisdomCapitalAPIWithoutLogin(cache, settings, STORED_LOGIN)
def run(self):
"""Sends the request, catches the exception and prints its fields.
Returns:
None: This method returns nothing.
"""
print(f'Wisdom Capital answers HTTP {self.server.status_code} with: {self.server.body}')
with unittest.mock.patch('requests.request', self.server.request):
try:
self.api.get(url='https://trade.wisdomcapital.in/interactive/user/balance?clientID=WC0001')
except WisdomCapitalAPIException as error:
print(f'Caught: {type(error).__name__}')
print(f'code: {error.code}')
print(f'message: {error.message}')
if __name__ == '__main__':
CatchingARefusedTokenExample().run()
It prints:
examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPIException/example_2_catching_a_gateway_error.py
"""Catches a gateway error from Wisdom Capital through the base class that every broker's exception shares.
A load balancer in front of Wisdom Capital's API sometimes answers with an HTML page instead of JSON, for example a 502 Bad Gateway while the service behind it restarts. `WisdomCapitalAPI`'s `_request` cannot read an error code from such a page, so it raises `WisdomCapitalAPIException` with the HTTP status as `code` and the page's text as `message`.
Every broker's exception class derives from `BrokerAPIException`, so code that talks to several brokers can catch that one class and still read `code` and `message`. This program catches it that way.
The real constructor reads the broker's settings from MongoDB, probes `GET /interactive/user/balance` and the market data application's `clientConfig`, and logs in to Wisdom Capital with a session request for the interactive application and a second login for the market data application when the stored token no longer works. This program must never log in, so a small subclass, `WisdomCapitalAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Wisdom Capital response, so nothing reaches the network.
Notice that the exception caught as `BrokerAPIException` is still a `WisdomCapitalAPIException`, and that its `code` is the integer 502 rather than a code of Wisdom Capital's own.
Run it from the project root:
python examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPIException/example_2_catching_a_gateway_error.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.base import (
BrokerAPIException,
)
from stock_brokers.api.wisdom_capital import (
WisdomCapitalAPI,
WisdomCapitalAPIException,
)
STORED_LOGIN = {
'broker_name': 'wisdom_capital',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
GATEWAY_PAGE = '<html><head><title>502 Bad Gateway</title></head><body><center><h1>502 Bad Gateway</h1></center></body></html>'
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Wisdom Capital response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class WisdomCapitalAPIWithoutLogin(WisdomCapitalAPI):
"""A `WisdomCapitalAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `wisdom_capital` settings document from MongoDB and then logs in to Wisdom Capital if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `WisdomCapitalAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'wisdom_capital'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('wisdom_capital')
self._settings = settings
self._last_login = stored_login
self.market_data_session_error = None
class CatchingAGatewayErrorExample:
"""Sends a request that a gateway answers with an HTML error page and catches the result as a `BrokerAPIException`.
Attributes:
server (RecordedBrokerServer): The stand-in for Wisdom Capital's API, answering with a gateway error.
api (WisdomCapitalAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores a login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
cache = SharedLoginCache()
cache.hset('last_login', 'wisdom_capital', json.dumps(STORED_LOGIN))
settings = {
'ucc_code': 'WC0001',
}
self.server = RecordedBrokerServer(502, 'text/html; charset=utf-8', GATEWAY_PAGE)
self.api = WisdomCapitalAPIWithoutLogin(cache, settings, STORED_LOGIN)
def run(self):
"""Sends the request, catches the exception through the base class and prints what it holds.
Returns:
None: This method returns nothing.
"""
with unittest.mock.patch('requests.request', self.server.request):
try:
self.api.get(url='https://trade.wisdomcapital.in/interactive/user/balance?clientID=WC0001')
except BrokerAPIException as error:
print(f'Caught as BrokerAPIException: {type(error).__name__}')
print(f'Is an instance of WisdomCapitalAPIException: {isinstance(error, WisdomCapitalAPIException)}')
print(f'code: {error.code!r}')
print(f'message: {error.message}')
if __name__ == '__main__':
CatchingAGatewayErrorExample().run()
It prints:
BrokerAPIException represents an exception condition related to Broker API calls
-code: (string) error code
-message: (string) error message
Example programs
examples/stock_brokers/api/base/BrokerAPIException/example_1_raising_and_reading_an_error.py
"""Raises a `BrokerAPIException` and reads the code and message it carries.
`BrokerAPIException` is what every broker class raises, through its own subclass, when a broker refuses a request. It holds the broker's error `code` and a `message`, both optional, and passes both to `Exception` as well, so they also appear in `args` and in the exception's text.
This program raises one with a Kite-style code and message, as `ZerodhaAPI` does for an expired token, and one with neither, and prints what a handler can read from each. It needs no data store, no network and no broker.
Notice that the code and message are plain attributes, and that an exception raised without them still has both, set to None.
Run it from the project root:
python examples/stock_brokers/api/base/BrokerAPIException/example_1_raising_and_reading_an_error.py
"""
from stock_brokers.api.base import (
BrokerAPIException,
)
class RaisingAndReadingAnErrorExample:
"""Raises two exceptions and prints what each holds.
Attributes:
errors (list): The exceptions to raise, one with a code and message and one without.
"""
def __init__(self):
"""Builds the exceptions to raise.
Returns:
None: This method returns nothing.
"""
self.errors = [
BrokerAPIException(code='TokenException', message='Incorrect `api_key` or `access_token`.'),
BrokerAPIException(),
]
def run(self):
"""Raises each exception, catches it and prints its fields.
Returns:
None: This method returns nothing.
"""
for error_to_raise in self.errors:
try:
raise error_to_raise
except BrokerAPIException as error:
print(f'code: {error.code!r}')
print(f'message: {error.message!r}')
print(f'args: {error.args!r}')
print(f'text: {error}')
print()
if __name__ == '__main__':
RaisingAndReadingAnErrorExample().run()
It prints:
examples/stock_brokers/api/base/BrokerAPIException/example_2_one_handler_for_every_broker.py
"""Handles the errors of several brokers with one `except BrokerAPIException` clause.
Each broker module defines its own exception class, such as `DhanAPIException` or `ZerodhaAPIException`, and every one of them derives from `BrokerAPIException`. Code that works across brokers, such as a script that polls every broker's funds, can therefore catch the base class once and still read the broker's `code` and `message`, and use the exception's class to say which broker refused.
This program raises three brokers' exceptions with the codes and messages those brokers send for an expired token, catches them all with one handler, and prints what the handler sees. It needs no data store, no network and no broker.
Notice that every exception is caught by the same clause, and that each keeps its broker's own code: Dhan's `errorType`, Kite's `error_type` and Noren's `stat`.
Run it from the project root:
python examples/stock_brokers/api/base/BrokerAPIException/example_2_one_handler_for_every_broker.py
"""
from stock_brokers.api.base import (
BrokerAPIException,
)
from stock_brokers.api.dhan import (
DhanAPIException,
)
from stock_brokers.api.flattrade import (
FlattradeAPIException,
)
from stock_brokers.api.zerodha import (
ZerodhaAPIException,
)
class RefusingBroker:
"""A stand-in for one broker's API object whose every request is refused with that broker's exception.
Attributes:
broker_name (str): The broker's name.
refusal (BrokerAPIException): The exception every request raises.
"""
def __init__(self, broker_name, refusal):
"""Holds the broker's name and the refusal.
Args:
broker_name (str): The broker's name.
refusal (BrokerAPIException): The exception every request raises.
Returns:
None: This method returns nothing.
"""
self.broker_name = broker_name
self.refusal = refusal
def get(self, url):
"""Refuses the request.
Args:
url (str): The URL requested.
Returns:
dict: Never returns.
Raises:
BrokerAPIException: Always, the broker's own subclass of it.
"""
raise self.refusal
class OneHandlerExample:
"""Asks every broker for its funds and handles every refusal in one place.
Attributes:
brokers (list): The stand-in brokers to ask.
"""
def __init__(self):
"""Builds the refusing brokers.
Returns:
None: This method returns nothing.
"""
self.brokers = [
RefusingBroker('dhan', DhanAPIException(code='Invalid_Authentication', message='Client ID or user generated access token is invalid or expired.')),
RefusingBroker('zerodha', ZerodhaAPIException(code='TokenException', message='Incorrect `api_key` or `access_token`.')),
RefusingBroker('flattrade', FlattradeAPIException(code='Not_Ok', message='Session Expired : Invalid Session Key')),
]
def run(self):
"""Sends one request to each broker and prints each refusal.
Returns:
None: This method returns nothing.
"""
for broker in self.brokers:
try:
broker.get(url='/funds')
except BrokerAPIException as error:
print(f'{broker.broker_name}: {type(error).__name__} code={error.code} message={error.message}')
if __name__ == '__main__':
OneHandlerExample().run()
It prints:
dhan: DhanAPIException code=Invalid_Authentication message=Client ID or user generated access token is invalid or expired.
zerodha: ZerodhaAPIException code=TokenException message=Incorrect `api_key` or `access_token`.
flattrade: FlattradeAPIException code=Not_Ok message=Session Expired : Invalid Session Key
Source code in stock_brokers/api/base.py
Bases: BrokerAPI
Wisdom Capital API class
Establishes both of the sessions Wisdom Capital's platform needs, logging in only where the stored token has stopped working.
Symphony XTS splits a broker into two applications with separate credentials. The interactive application places orders and reports the account, and the market data application serves quotes and charts; each issues its own token and refuses the other's. Constructing this class establishes both, and leaves them in the last_login document as access_token and market_data_access_token.
Neither token is minted when the stored one still works. The interactive token is checked with GET /user/balance, the funds endpoint. It is not /user/profile, because Wisdom Capital allows only about one profile call a day, which is a budget an object that is constructed several times a day cannot live inside. The funds endpoint authenticates identically, so it proves the same thing about the token without spending that allowance. The market data token is checked with GET /apimarketdata/config/clientConfig in the same spirit.
Wisdom Capital rate limits on a rolling window regardless of endpoint, and a rate-limit rejection (HTTP 429, or an error code containing e-apirl) is not a reason to log in again. The rate limiter runs only after the request has authenticated, and its error names the authenticated user, so a throttled check still proves the token is good. An expired or missing token instead comes back as e-token-0002 ("Please Provide token to Authenticate"), which does mean a fresh login is needed. A previously failed login stored the string "None" rather than a token, which also means a fresh login.
A market data login that fails is recorded in market_data_session_error rather than raised, because the scripts that poll orders, positions and funds need only the interactive token and must not be stopped by a market data credential they never use. bin/wisdom_capital/session/connect reports that error and exits non-zero, so the morning login is still where a broken market data credential is caught.
Raises:
| Type | Description |
|---|---|
WisdomCapitalAPIException
|
The legacy login found no access token on the login page, the host lookup returned no unique key, or the interactive session request returned no token. |
Example programs
examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_1_reading_the_balance.py
"""Reads the account balance from Wisdom Capital's REST API with a stored access token.
`WisdomCapitalAPI` is the object every Wisdom Capital script talks to the broker through. Its `get`, `post`, `put`, `patch` and `delete` methods all go through the broker's own `_request`, which adds the credentials of the login in force, sends the request, and wraps a successful answer in a dictionary with `status`, `code`, `data` and a `timestamp`.
The real constructor reads the broker's settings from MongoDB, probes `GET /interactive/user/balance` and the market data application's `clientConfig`, and logs in to Wisdom Capital with a session request for the interactive application and a second login for the market data application when the stored token no longer works. This program must never log in, so a small subclass, `WisdomCapitalAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Wisdom Capital response, so nothing reaches the network.
Notice the headers the request carried, which were built from the stored login and, where the broker needs them, from the settings, and that `data` holds only the `result` object, because `_request` unwraps the envelope Symphony XTS puts around every answer, and that the request went out with certificate checking switched off (`verify=False`), because Wisdom Capital's host serves a certificate issued to another domain. The `timestamp` is left out of the output because it is the moment the program ran.
Run it from the project root:
python examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_1_reading_the_balance.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.wisdom_capital import (
WisdomCapitalAPI,
)
STORED_LOGIN = {
'broker_name': 'wisdom_capital',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
RECORDED_ANSWER = {
'type': 'success',
'code': 's-user-0001',
'description': 'Get Balance Success',
'result': {
'BalanceList': [
{
'limitHeader': 'ALL|ALL|ALL',
'limitObject': {
'RMSSubLimits': {
'cashAvailable': '250000',
'netMarginAvailable': '248500.5',
'marginUtilized': '1499.5',
},
},
},
],
},
}
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Wisdom Capital response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class WisdomCapitalAPIWithoutLogin(WisdomCapitalAPI):
"""A `WisdomCapitalAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `wisdom_capital` settings document from MongoDB and then logs in to Wisdom Capital if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `WisdomCapitalAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'wisdom_capital'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('wisdom_capital')
self._settings = settings
self._last_login = stored_login
self.market_data_session_error = None
class ReadingTheBalanceExample:
"""Sends one authenticated request through `WisdomCapitalAPI` and prints what went out and what came back.
Attributes:
cache (SharedLoginCache): The stand-in for Redis, holding the stored login.
server (RecordedBrokerServer): The stand-in for Wisdom Capital's API.
api (WisdomCapitalAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores the morning login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
self.cache = SharedLoginCache()
self.cache.hset('last_login', 'wisdom_capital', json.dumps(STORED_LOGIN))
settings = {
'ucc_code': 'WC0001',
}
self.server = RecordedBrokerServer(200, 'application/json', json.dumps(RECORDED_ANSWER))
self.api = WisdomCapitalAPIWithoutLogin(self.cache, settings, STORED_LOGIN)
def run(self):
"""Sends the request and prints the request and the answer.
Returns:
None: This method returns nothing.
"""
with unittest.mock.patch('requests.request', self.server.request):
answer = self.api.get(url='https://trade.wisdomcapital.in/interactive/user/balance?clientID=WC0001')
call = self.server.calls[0]
print(f'Request: {call["method"]} {call["url"]}')
print('Headers sent:')
for name, value in call['headers'].items():
print(f' {name}: {value}')
print(f'Certificate checked: {call["verify"]}')
print(f'Status: {answer["status"]}')
print(f'HTTP code: {answer["code"]}')
print('Data:')
print(json.dumps(answer['data'], indent=4))
if __name__ == '__main__':
ReadingTheBalanceExample().run()
It prints:
Request: GET https://trade.wisdomcapital.in/interactive/user/balance?clientID=WC0001
Headers sent:
Content-Type: application/json
authorization: morning-token
Certificate checked: False
Status: success
HTTP code: 200
Data:
{
"BalanceList": [
{
"limitHeader": "ALL|ALL|ALL",
"limitObject": {
"RMSSubLimits": {
"cashAvailable": "250000",
"netMarginAvailable": "248500.5",
"marginUtilized": "1499.5"
}
}
}
]
}
examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_2_using_a_newer_login.py
"""Picks up an access token another process obtained, on the very next request.
Every Wisdom Capital script builds its own `WisdomCapitalAPI`, and more than one of them may log in during a day. A login writes the new token to MongoDB and then to the Redis `last_login` hash, and `WisdomCapitalAPI`'s `_request` reads that hash again before every request, keeping whichever login is newer by its `last_login` time. So a token obtained by any process is used by every other one without restarting anything.
The program sends two requests through one API object. Between them it writes a newer login into the stand-in for Redis, as another process's login would.
The real constructor reads the broker's settings from MongoDB, probes `GET /interactive/user/balance` and the market data application's `clientConfig`, and logs in to Wisdom Capital with a session request for the interactive application and a second login for the market data application when the stored token no longer works. This program must never log in, so a small subclass, `WisdomCapitalAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Wisdom Capital response, so nothing reaches the network.
Notice that the second request carries the newer token although the object was never rebuilt.
Run it from the project root:
python examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_2_using_a_newer_login.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.wisdom_capital import (
WisdomCapitalAPI,
)
STORED_LOGIN = {
'broker_name': 'wisdom_capital',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
NEWER_LOGIN = {
'broker_name': 'wisdom_capital',
'access_token': 'noon-token',
'last_login': '2026-09-30 12:30:41',
}
RECORDED_ANSWER = {
'type': 'success',
'code': 's-user-0001',
'description': 'Get Balance Success',
'result': {
'BalanceList': [],
},
}
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Wisdom Capital response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class WisdomCapitalAPIWithoutLogin(WisdomCapitalAPI):
"""A `WisdomCapitalAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `wisdom_capital` settings document from MongoDB and then logs in to Wisdom Capital if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `WisdomCapitalAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'wisdom_capital'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('wisdom_capital')
self._settings = settings
self._last_login = stored_login
self.market_data_session_error = None
class UsingANewerLoginExample:
"""Sends a request, lets another process log in, and sends another through the same object.
Attributes:
cache (SharedLoginCache): The stand-in for Redis, holding the shared login.
server (RecordedBrokerServer): The stand-in for Wisdom Capital's API.
api (WisdomCapitalAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores the morning login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
self.cache = SharedLoginCache()
self.cache.hset('last_login', 'wisdom_capital', json.dumps(STORED_LOGIN))
settings = {
'ucc_code': 'WC0001',
}
self.server = RecordedBrokerServer(200, 'application/json', json.dumps(RECORDED_ANSWER))
self.api = WisdomCapitalAPIWithoutLogin(self.cache, settings, STORED_LOGIN)
def credential(self, call):
"""The credential one recorded request carried.
Args:
call (dict): One request recorded by the stand-in server.
Returns:
str: The `authorization` header.
"""
return call['headers']['authorization']
def run(self):
"""Sends a request, writes a newer login, sends another request and prints what each carried.
Returns:
None: This method returns nothing.
"""
with unittest.mock.patch('requests.request', self.server.request):
self.api.get(url='https://trade.wisdomcapital.in/interactive/user/balance?clientID=WC0001')
print(f'First request carried: {self.credential(self.server.calls[0])}')
self.cache.hset('last_login', 'wisdom_capital', json.dumps(NEWER_LOGIN))
print('Another process logged in again at 12:30:41 and wrote its login to Redis.')
with unittest.mock.patch('requests.request', self.server.request):
answer = self.api.get(url='https://trade.wisdomcapital.in/interactive/user/balance?clientID=WC0001')
print(f'Second request carried: {self.credential(self.server.calls[1])}')
print(f'Request: {self.server.calls[1]["method"]} {self.server.calls[1]["url"]}')
print(f'Status: {answer["status"]}, HTTP code: {answer["code"]}')
print('Data:')
print(json.dumps(answer['data'], indent=4))
if __name__ == '__main__':
UsingANewerLoginExample().run()
It prints:
examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_3_telling_rate_limits_from_refused_sessions.py
"""Sorts Wisdom Capital's refusals into rate limits, refused sessions and everything else.
Wisdom Capital's platform, Symphony XTS, rate limits every request on a rolling window, and it checks the token before it checks the rate. So a request refused for rate still proves the token is good, while one refused for its token means a fresh login is needed. `WisdomCapitalAPI.is_rate_limited` and `WisdomCapitalAPI.is_session_refused` make that call from whatever a request raised, by its code (429 or 401) or by the XTS error codes in its text (`e-apirl` for rate, `e-session` and `e-token` for the session).
The constructor uses them to decide whether to log in again: it keeps the stored token after a rate limit, logs in after a refused session, and keeps the token when the error says nothing either way, such as a gateway error or a dropped connection, because a new market data login would invalidate the token every other process is using.
Both are static methods and need no logged-in object, so this program builds only exceptions, of the kinds the platform and the network produce. It reaches no broker.
Notice that the gateway error and the dropped connection are neither, which is what keeps a passing outage from triggering a login.
Run it from the project root:
python examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_3_telling_rate_limits_from_refused_sessions.py
"""
import requests
from stock_brokers.api.wisdom_capital import (
WisdomCapitalAPI,
WisdomCapitalAPIException,
)
class TellingRefusalsApartExample:
"""Classifies a list of errors and says what the constructor would do after each.
Attributes:
errors (list): Pairs of (description, exception) to classify.
"""
def __init__(self):
"""Builds the errors to classify.
Returns:
None: This method returns nothing.
"""
self.errors = [
(
'HTTP 429 from the gateway',
WisdomCapitalAPIException(code=429, message='Too Many Requests'),
),
(
'XTS rate limit code',
WisdomCapitalAPIException(code='e-apirl-0004', message='Rate limit exceeded for user WC0001'),
),
(
'XTS missing or expired token',
WisdomCapitalAPIException(code='e-token-0002', message='Please Provide token to Authenticate'),
),
(
'HTTP 401',
WisdomCapitalAPIException(code=401, message='Unauthorized'),
),
(
'HTTP 502 gateway page',
WisdomCapitalAPIException(code=502, message='<html><body><h1>502 Bad Gateway</h1></body></html>'),
),
(
'dropped connection',
requests.ConnectionError('Connection aborted.'),
),
]
def run(self):
"""Prints each error's classification and the resulting decision.
Returns:
None: This method returns nothing.
"""
for description, error in self.errors:
rate_limited = WisdomCapitalAPI.is_rate_limited(error)
session_refused = WisdomCapitalAPI.is_session_refused(error)
if rate_limited:
decision = 'keep the token, it authenticated'
elif session_refused:
decision = 'log in again'
else:
decision = 'keep the token, the error says nothing about it'
print(f'{description}:')
print(f' rate limited={rate_limited} session refused={session_refused} -> {decision}')
if __name__ == '__main__':
TellingRefusalsApartExample().run()
It prints:
HTTP 429 from the gateway:
rate limited=True session refused=False -> keep the token, it authenticated
XTS rate limit code:
rate limited=True session refused=False -> keep the token, it authenticated
XTS missing or expired token:
rate limited=False session refused=True -> log in again
HTTP 401:
rate limited=False session refused=True -> log in again
HTTP 502 gateway page:
rate limited=False session refused=False -> keep the token, the error says nothing about it
dropped connection:
rate limited=False session refused=False -> keep the token, the error says nothing about it
examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_4_replacing_a_refused_market_data_session.py
"""Replaces a refused market data session, reusing one another process already obtained where it can.
Wisdom Capital's platform issues exactly one market data session per application key, and a second login invalidates the first token. So when a process finds its market data token refused, it calls `replace_market_data_session` with that stale token. If another process has already published a different token in the shared login, that token is returned and nobody logs in. Otherwise the process takes a Redis lock, checks once more, and logs in itself, releasing the lock afterwards whatever happens. `market_data_session` reads the market data token and user id in force from the shared login.
The program shows both paths. In the first, another process has already replaced the token, so the call returns it without taking the lock. In the second, nothing has replaced it, so the lock is taken and a login is attempted; the settings here deliberately lack the market data key pair, so the login stops with a `WisdomCapitalAPIException` before sending anything, and the lock is released.
The real constructor reads the settings from MongoDB and logs in to both of Wisdom Capital's applications when their stored tokens no longer work. This program must never log in, so a small subclass, `WisdomCapitalAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that holds the shared login and the lock. No request reaches the network.
Notice that the first path takes no lock, and that the second path leaves no lock behind although the login failed.
Run it from the project root:
python examples/stock_brokers/api/wisdom_capital/WisdomCapitalAPI/example_4_replacing_a_refused_market_data_session.py
"""
import json
import logging
from stock_brokers.api.wisdom_capital import (
MARKET_DATA_LOCK_KEY,
WisdomCapitalAPI,
WisdomCapitalAPIException,
)
MORNING_LOGIN = {
'broker_name': 'wisdom_capital',
'access_token': 'interactive-token',
'last_login': '2026-09-30 07:00:05',
'market_data_access_token': 'market-data-token-morning',
'market_data_user_id': 'WC0001',
}
REPLACED_LOGIN = {
'broker_name': 'wisdom_capital',
'access_token': 'interactive-token',
'last_login': '2026-09-30 07:00:05',
'market_data_access_token': 'market-data-token-noon',
'market_data_user_id': 'WC0001',
}
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the shared login and the market data lock.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
keys (dict): Each plain key mapped to its value.
locks_taken (list): Every key set with `nx=True` that was not already present.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
self.keys = {}
self.locks_taken = []
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
def set(self, name, value, nx=False, ex=None):
"""Sets a plain key, only when it is absent if `nx` is given, as the lock is taken.
Args:
name (str): The key.
value (str): The value to store.
nx (bool): True to set the key only when it does not exist.
ex (int | None): The expiry in seconds, which the stand-in ignores.
Returns:
bool: True when the key was set.
"""
if nx and name in self.keys:
return False
self.keys[name] = value
if nx:
self.locks_taken.append(name)
return True
def delete(self, name):
"""Removes a plain key, as the lock is released.
Args:
name (str): The key.
Returns:
int: 1 when the key existed, otherwise 0.
"""
if name in self.keys:
del self.keys[name]
return 1
return 0
class WisdomCapitalAPIWithoutLogin(WisdomCapitalAPI):
"""A `WisdomCapitalAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `wisdom_capital` settings document from MongoDB and then establishes both of the platform's sessions. This subclass sets the same attributes from what it is given and does nothing else.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the shared login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'wisdom_capital'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('wisdom_capital')
self._settings = settings
self._last_login = stored_login
self.market_data_session_error = None
class ReplacingAMarketDataSessionExample:
"""Replaces a refused market data token once when another process already has, and once when nobody has.
Attributes:
settings (dict): The broker's settings document, without the market data key pair.
"""
def __init__(self):
"""Holds the settings both paths use.
Returns:
None: This method returns nothing.
"""
self.settings = {
'ucc_code': 'WC0001',
}
def build_api(self, shared_login):
"""Builds an API object whose shared login is the one given.
Args:
shared_login (dict): The login document another process last wrote to Redis.
Returns:
tuple: (SharedLoginCache, WisdomCapitalAPIWithoutLogin), the stand-in cache and the API object.
"""
cache = SharedLoginCache()
cache.hset('last_login', 'wisdom_capital', json.dumps(shared_login))
api = WisdomCapitalAPIWithoutLogin(cache, self.settings, MORNING_LOGIN)
return cache, api
def run(self):
"""Runs both paths and prints what each returned or raised.
Returns:
None: This method returns nothing.
"""
print('Path 1: another process has already replaced the refused token.')
cache, api = self.build_api(MORNING_LOGIN)
print(f' Session in force: {api.market_data_session()}')
cache.hset('last_login', 'wisdom_capital', json.dumps(REPLACED_LOGIN))
session = api.replace_market_data_session(stale_access_token='market-data-token-morning')
print(f' Replacement returned: {session}')
print(f' Locks taken: {cache.locks_taken}')
print('Path 2: nobody has replaced it, and the settings lack the market data key pair.')
cache, api = self.build_api(MORNING_LOGIN)
try:
api.replace_market_data_session(stale_access_token='market-data-token-morning')
except WisdomCapitalAPIException as error:
print(f' Raised: {type(error).__name__} code={error.code}')
print(f' message: {error.message}')
print(f' Locks taken: {cache.locks_taken}')
print(f' Lock still held: {MARKET_DATA_LOCK_KEY in cache.keys}')
if __name__ == '__main__':
ReplacingAMarketDataSessionExample().run()
It prints:
Path 1: another process has already replaced the refused token.
Session in force: {'access_token': 'market-data-token-morning', 'user_id': 'WC0001'}
Replacement returned: {'access_token': 'market-data-token-noon', 'user_id': 'WC0001'}
Locks taken: []
Path 2: nobody has replaced it, and the settings lack the market data key pair.
Raised: WisdomCapitalAPIException code=500
message: wisdom_capital has no price_api_key or price_api_secret in its settings, so the market data session cannot be established.
Locks taken: ['wisdom_capital:session:marketdata:lock']
Lock still held: False
Methods:
| Name | Description |
|---|---|
is_rate_limited |
Whether Wisdom Capital refused a request for rate rather than for the token it carried. |
is_session_refused |
Whether Wisdom Capital refused a request because the token it carried is expired, unknown or missing. |
market_data_session |
The market data token and user id in force now, read from the shared login rather than from this object. |
replace_market_data_session |
Establishes a market data session, unless another process has already replaced the one that failed. |
get |
HTTP GET API call to Broker's API server. |
post |
HTTP POST API call to Broker's API server. |
put |
HTTP PUT API call to Broker's API server. |
patch |
HTTP PATCH API call to Broker's API server. |
delete |
HTTP DELETE API call to Broker's API server. |
Source code in stock_brokers/api/wisdom_capital.py
staticmethod
¶
Whether Wisdom Capital refused a request for rate rather than for the token it carried.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exception
|
Exception
|
Whatever the request raised. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True when the refusal came from the rate limiter, which runs only after the request has authenticated and so still proves the token is good. |
Example usage
From example_3_telling_rate_limits_from_refused_sessions.py
rate_limited = WisdomCapitalAPI.is_rate_limited(error)
Source code in stock_brokers/api/wisdom_capital.py
staticmethod
¶
Whether Wisdom Capital refused a request because the token it carried is expired, unknown or missing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exception
|
Exception
|
Whatever the request raised. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True when the platform named the session or the token as the problem. |
Example usage
From example_3_telling_rate_limits_from_refused_sessions.py
session_refused = WisdomCapitalAPI.is_session_refused(error)
Source code in stock_brokers/api/wisdom_capital.py
The market data token and user id in force now, read from the shared login rather than from this object.
Returns:
| Type | Description |
|---|---|
dict
|
|
Example usage
From example_4_replacing_a_refused_market_data_session.py
print(f' Session in force: {api.market_data_session()}')
Source code in stock_brokers/api/wisdom_capital.py
Establishes a market data session, unless another process has already replaced the one that failed.
XTS issues exactly one market data session per application key, and a second login invalidates the first token rather than handing out a second. Two processes that each log in therefore take turns invalidating each other's session, so whoever finds the stored token unusable takes a Redis lock, logs in once, and publishes the result for everyone else to read.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stale_access_token
|
str | None
|
The market data token that was just refused, so that a token another process has already published in its place is recognised as a replacement rather than as the same failure. |
None
|
Returns:
| Type | Description |
|---|---|
dict
|
|
Raises:
| Type | Description |
|---|---|
WisdomCapitalAPIException
|
The market data credentials are missing, the login returned no token, or another process held the lock for longer than |
Example usage
From example_4_replacing_a_refused_market_data_session.py
session = api.replace_market_data_session(stale_access_token='market-data-token-morning')
From example_4_replacing_a_refused_market_data_session.py
api.replace_market_data_session(stale_access_token='market-data-token-morning')
Source code in stock_brokers/api/wisdom_capital.py
Makes sure the interactive token in the shared login is one Wisdom Capital still accepts, logging in again when it is not.
Returns:
| Type | Description |
|---|---|
None
|
The token in force is left in |
Raises:
| Type | Description |
|---|---|
WisdomCapitalAPIException
|
The login produced no token. |
Source code in stock_brokers/api/wisdom_capital.py
Asks Wisdom Capital for the account balance to find out whether the stored interactive token is still accepted.
Returns:
| Type | Description |
|---|---|
bool
|
True when the balance came back, or when the request was refused for rate, which happens only after it has authenticated. |
Source code in stock_brokers/api/wisdom_capital.py
Exchanges the interactive credentials for a token and stores it.
Returns:
| Type | Description |
|---|---|
dict
|
The stored login after the token was written into it. |
Raises:
| Type | Description |
|---|---|
WisdomCapitalAPIException
|
The legacy login found no access token on the login page, the host lookup returned no unique key, or the session request returned no token. |
Source code in stock_brokers/api/wisdom_capital.py
Drives Wisdom Capital's web login page with Selenium and returns the extra fields the legacy session request carries.
Returns:
| Type | Description |
|---|---|
dict
|
|
Raises:
| Type | Description |
|---|---|
WisdomCapitalAPIException
|
No access token could be extracted from the login page, or the host lookup returned no unique key. |
Source code in stock_brokers/api/wisdom_capital.py
216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 | |
Makes sure the market data token in the shared login is one Wisdom Capital still accepts, logging in again when it is not.
Returns:
| Type | Description |
|---|---|
None
|
The token in force is left in |
Raises:
| Type | Description |
|---|---|
WisdomCapitalAPIException
|
No market data session could be established. |
Source code in stock_brokers/api/wisdom_capital.py
Asks the market data application for its client configuration to find out whether a token is still accepted.
A failure that does not name the session is deliberately read as "cannot tell" and the token is kept. Minting a new one invalidates whatever the live quotes feed and the candle downloader are holding, which is too destructive to do on the strength of a timeout or a bad gateway.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
access_token
|
str
|
The market data token to check. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True when the token is still accepted, or when the answer said nothing about the session. |
Source code in stock_brokers/api/wisdom_capital.py
The headers a market data request carries, which authorize with the market data token rather than the interactive one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
access_token
|
str
|
The market data token. |
required |
Returns:
| Type | Description |
|---|---|
dict
|
The headers for a market data request. |
Source code in stock_brokers/api/wisdom_capital.py
The market data session another process has published, when it is not the one that has just failed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stale_access_token
|
str | None
|
The market data token that was refused, or None when there was none to refuse. |
required |
Returns:
| Type | Description |
|---|---|
dict
|
|
Source code in stock_brokers/api/wisdom_capital.py
Exchanges the market data credentials for a token and stores it.
Returns:
| Type | Description |
|---|---|
dict
|
|
Raises:
| Type | Description |
|---|---|
WisdomCapitalAPIException
|
The market data credentials are missing from the broker's settings, or the login returned no token. |
Source code in stock_brokers/api/wisdom_capital.py
Merges freshly minted session fields into the shared login, MongoDB first and Redis second.
The two tokens are minted by different paths and at different times, so a login is merged into whatever is stored rather than replacing it. Without that, establishing one session would erase the other's token for every process reading the same document.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fields
|
dict
|
The fields to set on the |
required |
Returns:
| Type | Description |
|---|---|
dict
|
The stored login after the merge, also left in |
Source code in stock_brokers/api/wisdom_capital.py
_request(method, url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
Private generic REST API request to Broker API server. This is specialized into GET, POST, PUT, PATCH and DELETE by public methods.
method: HTTP method to use for the request.- `url
params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Source code in stock_brokers/api/wisdom_capital.py
429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 | |
The login in force right now, read on every request rather than once at construction.
Any process that logs a broker in writes the new login to MongoDB and then to the Redis
last_login hash. Reading that hash here, on every request, means a token another process
has just obtained is used by this object's very next request, rather than a token loaded at
construction being sent long after it has been replaced.
Redis is the one place a token is read from, and a login the only thing that writes it. Constructors do not copy MongoDB into it, because that would race a login: a process that read MongoDB just before another logged in, and wrote Redis just after, would put the old token back for everyone.
The newer of this object's login and Redis's is used, by their last_login time. The
shared login normally wins. This object's own wins only when it is genuinely newer - a
login this object has just made, or one whose write to Redis failed while its write to
MongoDB, which this object was built from, succeeded. When either time cannot be read,
Redis is trusted.
When Redis holds nothing for the broker, the login is read from MongoDB and written into Redis only if that field is still empty, so it can fill the cache but never overwrite a login that lands in between. When neither store can be read, the object keeps what it has.
Returns:
| Type | Description |
|---|---|
dict | None
|
The |
Source code in stock_brokers/api/base.py
staticmethod
¶
When a login document was made, or None when that cannot be read.
Brokers stamp last_login with local time, some with microseconds and some without.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
document
|
dict | None
|
A |
required |
Returns:
| Type | Description |
|---|---|
datetime | None
|
The moment of the login. |
Source code in stock_brokers/api/base.py
get(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP GET API call to Broker's API server. This queries the Broker's API server for some information. Most responses are JSON encoded. Only a few are CSV encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
post(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP POST API call to Broker's API server. This carries out some action using Broker's API server such as placing an order. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
put(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP PUT API call to Broker's API server. This modifies something using Broker's API server such as modifying an order. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
patch(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP PATCH API call to Broker's API server. This updates something using Broker's API server such as modifying an order. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
delete(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP DELETE API call to Broker's API server. This deletes something using Broker's API server such as cancelling an order or invalidating an access token. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage