Skip to content

wisdom_capital

Wisdom Capital historical candle download.

Wisdom Capital runs Symphony's XTS platform, and four things about it are unlike every other broker here. All four were measured on 2026-09-12 rather than taken from documentation.

It needs a different session. XTS splits a broker into two applications with separate credentials, and the interactive token is refused by the chart endpoint, which lives under /apimarketdata. WisdomCapitalAPI establishes both sessions when it is constructed and keeps the market data one in last_login as market_data_access_token, so this module asks that class for the token - the same one bin/wisdom_capital/instruments/websocket_quotes carries - and sends it on every request.

The bars are not JSON. They arrive inside one string under a key the platform spells dataReponse, comma separated between bars and pipe separated within one.

Its timestamps need two corrections. The number is seconds since the epoch computed as though India time were UTC, so decoding it as UTC and relabelling the result as India time gives the India wall clock reading directly. And it names the last second of the bar rather than the first, so the bar's own length less one second is subtracted to reach its start. Both were checked against Flattrade's one minute bars for the same session: the starts line up exactly.

There is no daily interval. A compression of 86,400 seconds returns a rolling twenty four hour bucket running from 09:15 one day to 09:14:59 the next, not the exchange's session, which was confirmed by asking for it. A bar like that would sit in price_history looking like a daily bar and comparing against real ones, so it is not offered at all.

The fifth difference is the dangerous one. An empty response says nothing. An instrument the platform does not recognise, a window before its history begins, and an expired market data token all answer with the same envelope: a type of success and an empty dataReponse. So a run of empty windows cannot be trusted here the way it can elsewhere - a token that expired mid-backfill would look exactly like tens of thousands of dead instruments, and the base class would retire every series it touched. That is what _check_still_alive is for: after a run of empty windows it asks about an instrument known to have bars, and if that comes back empty too it logs in again and asks once more before concluding the session is the problem and stopping the broker.

Classes:

Name Description
WisdomCapitalCandles

Downloads Wisdom Capital's historical candles.

WisdomCapitalCandles 

WisdomCapitalCandles(api=None)

Bases: BrokerCandles

Downloads Wisdom Capital's historical candles.

Wisdom Capital candle downloader.

  • api is an authenticated WisdomCapitalAPI. One is constructed when not supplied. Either way a market data session is established on top of it, because the chart endpoint does not accept the interactive token.

Example programs

examples/stock_brokers/instruments/historical/wisdom_capital/WisdomCapitalCandles/example_1_bar_string_and_end_stamps.py
"""Takes a market data token, fetches one window of RELIANCE one minute bars from Wisdom Capital, and turns the platform's bar string into bars.

Wisdom Capital runs Symphony's XTS platform, whose chart endpoint refuses the interactive trading token and wants a separate market data token. `after_relogin` takes that token from the API class, which is also what the constructor does, so the program calls it once to arm the downloader. The request then carries the token in its `authorization` header, the segment number and instrument id from the identifier `1|2885`, the window in the platform's own `Sep 11 2026 000000` format, and the bar length in seconds.

The answer is one string, bars separated by commas and fields by pipes. Each timestamp needs two corrections: it counts seconds as though India time were UTC, and it names the bar's last second rather than its first. So `1789031759`, which reads as 09:15:59 on 2026-09-10, is the one minute bar that opened at 09:15 India time. Read as a five minute bar, the same stamp would open four minutes earlier, at 09:11.

A `WisdomCapitalCandles` constructor opens a PostgreSQL connection and logs in, so the program builds the object without running its constructor, as `test_runs/candle_parse.py` does, and gives it a stand-in for `WisdomCapitalAPI`. The stand-in hands out a fixed market data token, records the request, and answers with a bar string shaped like the one the live one minute feed returned, wrapped in the `status_code` and `data` envelope the real API class returns, including the platform's own misspelled key `dataReponse`.

Run it from the project root:

    python examples/stock_brokers/instruments/historical/wisdom_capital/WisdomCapitalCandles/example_1_bar_string_and_end_stamps.py
"""

import datetime

from stock_brokers.instruments.historical.wisdom_capital import (
    WisdomCapitalCandles,
)


class RecordedWisdomCapitalAPI:
    """A stand-in for `WisdomCapitalAPI` with a fixed market data token and one recorded bar string.

    Attributes:
        requests (list): The (url, headers, parameters) triples it was asked for.
    """

    def __init__(self):
        """Builds the stand-in with no requests recorded.

        Returns:
            None: This method returns nothing.
        """
        self.requests = []

    def replace_market_data_session(self, stale_access_token=None):
        """Hands out the market data session in force.

        Args:
            stale_access_token (str | None): A token that was refused, or None when none was.

        Returns:
            dict: The session's `access_token` and `user_id`.
        """
        return {
            'access_token': 'market-data-token-1',
            'user_id': 'WC1001',
        }

    def get(self, url, headers=None, params=None):
        """Records a GET request and answers it with two bars.

        Args:
            url (str): The URL the downloader asked for.
            headers (dict): The headers it sent.
            params (dict): The query parameters it sent.

        Returns:
            dict: The response envelope, with the platform's result under `data`.
        """
        self.requests.append((url, headers, params))
        return {
            'status_code': 200,
            'data': {
                'dataReponse': '1789031759|1279.5|1285.3|1279.5|1283.2|149743|0|,1789031819|1283.2|1284.0|1282.0|1283.0|51743|0|',
            },
        }


class BarStringAndEndStampsExample:
    """Arms the downloader with a market data token, fetches one window and parses it.

    Attributes:
        api (RecordedWisdomCapitalAPI): The stand-in Wisdom Capital API.
        candles (WisdomCapitalCandles): The downloader being shown.
    """

    def __init__(self):
        """Builds the downloader around the stand-in API without opening a database connection.

        Returns:
            None: This method returns nothing.
        """
        self.api = RecordedWisdomCapitalAPI()
        self.candles = object.__new__(WisdomCapitalCandles)
        self.candles._api = self.api
        self.candles._empty_streak = 0

    def run(self):
        """Takes the token, fetches 2026-09-10 and prints the request and the bars.

        Returns:
            None: This method returns nothing.
        """
        self.candles.after_relogin()
        day = datetime.date(2026, 9, 10)
        payload = self.candles.fetch_candles('1|2885', '1minute', day, day)
        url, headers, parameters = self.api.requests[0]
        print(f'Requested {url}')
        print(f'  headers: {headers}')
        print(f'  parameters: {parameters}')
        print(f'  answer: {payload}')
        for bar in self.candles.parse_response(payload, '1minute'):
            bar_time, open_price, high, low, close, volume, open_interest = bar
            print(f'  {bar_time.isoformat()} open {open_price} high {high} low {low} close {close} volume {volume} oi {open_interest}')
        first_as_five_minutes = self.candles.parse_response(payload, '5minute')[0][0]
        print(f'The first stamp read as a five minute bar opens at {first_as_five_minutes.isoformat()}')


if __name__ == '__main__':
    BarStringAndEndStampsExample().run()

It prints:

Requested https://trade.wisdomcapital.in/apimarketdata/instruments/ohlc
  headers: {'authorization': 'market-data-token-1'}
  parameters: {'exchangeSegment': '1', 'exchangeInstrumentID': '2885', 'startTime': 'Sep 10 2026 000000', 'endTime': 'Sep 11 2026 000000', 'compressionValue': '60'}
  answer: 1789031759|1279.5|1285.3|1279.5|1283.2|149743|0|,1789031819|1283.2|1284.0|1282.0|1283.0|51743|0|
  2026-09-10T09:15:00+05:30 open 1279.5 high 1285.3 low 1279.5 close 1283.2 volume 149743 oi 0
  2026-09-10T09:16:00+05:30 open 1283.2 high 1284.0 low 1282.0 close 1283.0 volume 51743 oi 0
The first stamp read as a five minute bar opens at 2026-09-10T09:11:00+05:30
examples/stock_brokers/instruments/historical/wisdom_capital/WisdomCapitalCandles/example_2_identifiers_and_refusals.py
"""Builds Wisdom Capital identifiers from its instrument master, reads what each says about its instrument, and classifies the platform's refusals.

One XTS instrument id appears under several segments, so `instruments` joins the segment's number to the instrument id, as in `1|2885`, and turns XTS's numeric instrument type into a word the queue's priority understands (`8` is cash, `1` a future). Rows in a segment the platform has no number for are left out. The program gives the downloader a stand-in database connection whose cursor answers with four rows, one of them in a segment the table does not know. `identifier_parts` splits an identifier again, and `series_context`, a classmethod that needs no downloader, reads the exchange and segments its number names.

The platform raises its conditions with codes of its own. The program fetches through a stand-in for `WisdomCapitalAPI` that raises the `WisdomCapitalAPIException` for each of three: `e-apirl` is a rate limit, `e-session` a dead session and `e-instrument` an instrument the platform will not serve, which retires the series. The stand-in also hands out the market data token that `after_relogin` takes before any request is made.

A `WisdomCapitalCandles` constructor opens a PostgreSQL connection and logs in, so the program builds the object without running its constructor, as `test_runs/candle_parse.py` does.

Run it from the project root:

    python examples/stock_brokers/instruments/historical/wisdom_capital/WisdomCapitalCandles/example_2_identifiers_and_refusals.py
"""

import datetime

from stock_brokers.api.wisdom_capital import (
    WisdomCapitalAPIException,
)
from stock_brokers.instruments.historical.base import (
    CandleError,
)
from stock_brokers.instruments.historical.wisdom_capital import (
    WisdomCapitalCandles,
)


class InstrumentMasterCursor:
    """A stand-in database cursor that answers the instrument master query with fixed rows.

    Attributes:
        rows (list): The rows `fetchall` returns.
    """

    def __init__(self, rows):
        """Holds the rows to answer with.

        Args:
            rows (list): Tuples of (segment name, instrument id, instrument type, expiry).

        Returns:
            None: This method returns nothing.
        """
        self.rows = rows

    def __enter__(self):
        """Enters the `with` block.

        Returns:
            InstrumentMasterCursor: This cursor.
        """
        return self

    def __exit__(self, exception_type, exception, traceback):
        """Leaves the `with` block without suppressing an exception.

        Args:
            exception_type (type | None): The exception's class, if one was raised.
            exception (BaseException | None): The exception, if one was raised.
            traceback (types.TracebackType | None): Its traceback, if one was raised.

        Returns:
            bool: Always False.
        """
        return False

    def execute(self, query):
        """Accepts the query without running it.

        Args:
            query (str): The SQL the downloader sent.

        Returns:
            None: This method returns nothing.
        """

    def fetchall(self):
        """Returns the fixed rows.

        Returns:
            list: The rows.
        """
        return self.rows


class InstrumentMasterConnection:
    """A stand-in PostgreSQL connection over four rows of the XTS instrument master.

    Attributes:
        rows (list): The rows every cursor answers with.
    """

    def __init__(self):
        """Builds the connection with four instrument master rows.

        Returns:
            None: This method returns nothing.
        """
        self.rows = [
            (
                'NSECM',
                '2885',
                8,
                None,
            ),
            (
                'BSECM',
                '500325',
                8,
                None,
            ),
            (
                'MCXFO',
                '565899',
                1,
                '2026-10-19T23:59:59',
            ),
            (
                'NSEXX',
                '777',
                8,
                None,
            ),
        ]

    def cursor(self):
        """Opens a cursor.

        Returns:
            InstrumentMasterCursor: A cursor over the rows.
        """
        return InstrumentMasterCursor(self.rows)

    def commit(self):
        """Accepts a commit.

        Returns:
            None: This method returns nothing.
        """


class RefusingWisdomCapitalAPI:
    """A stand-in for `WisdomCapitalAPI` that hands out a market data token and refuses every chart request.

    Attributes:
        code (str | None): The platform code the next refusal carries.
        message (str | None): The message the next refusal carries.
    """

    def __init__(self):
        """Builds the stand-in with no refusal chosen yet.

        Returns:
            None: This method returns nothing.
        """
        self.code = None
        self.message = None

    def replace_market_data_session(self, stale_access_token=None):
        """Hands out the market data session in force.

        Args:
            stale_access_token (str | None): A token that was refused, or None when none was.

        Returns:
            dict: The session's `access_token` and `user_id`.
        """
        return {
            'access_token': 'market-data-token-1',
            'user_id': 'WC1001',
        }

    def get(self, url, headers=None, params=None):
        """Refuses a GET request with the chosen code.

        Args:
            url (str): The URL the downloader asked for.
            headers (dict): The headers it sent.
            params (dict): The query parameters it sent.

        Returns:
            dict: Never returns.

        Raises:
            WisdomCapitalAPIException: Always, carrying the chosen code and message.
        """
        raise WisdomCapitalAPIException(code=self.code, message=self.message)


class IdentifiersAndRefusalsExample:
    """Lists identifiers, reads them, and classifies three refusals.

    Attributes:
        api (RefusingWisdomCapitalAPI): The stand-in Wisdom Capital API.
        candles (WisdomCapitalCandles): The downloader being shown.
    """

    def __init__(self):
        """Builds the downloader around the stand-ins without running its constructor.

        Returns:
            None: This method returns nothing.
        """
        self.api = RefusingWisdomCapitalAPI()
        self.candles = object.__new__(WisdomCapitalCandles)
        self.candles._connection = InstrumentMasterConnection()
        self.candles._api = self.api
        self.candles._empty_streak = 0

    def show_identifiers(self):
        """Lists the instruments and reads each identifier.

        Returns:
            None: This method returns nothing.
        """
        instruments = self.candles.instruments()
        print(f'Instruments to download: {len(instruments)} of 4 rows')
        for identifier, instrument_type, expiry in instruments:
            segment_number, instrument_identifier = self.candles.identifier_parts(identifier)
            context = WisdomCapitalCandles.series_context(identifier)
            print(f'{identifier}: segment {segment_number}, instrument {instrument_identifier}, {instrument_type}, expiry {expiry}, exchange {context.exchange}, {len(context.segments)} segments to search')

    def show_refusals(self):
        """Fetches through three refusals and prints what each became.

        Returns:
            None: This method returns nothing.
        """
        self.candles.after_relogin()
        refusals = [
            (
                'e-apirl-0004',
                'Rate limit exceeded',
            ),
            (
                'e-session-0002',
                'Invalid Token',
            ),
            (
                'e-instrument-0001',
                'Instrument not found',
            ),
        ]
        for code, message in refusals:
            self.api.code = code
            self.api.message = message
            try:
                self.candles.fetch_candles(
                    '2|35001',
                    '5minute',
                    datetime.date(2025, 1, 1),
                    datetime.date(2025, 3, 31),
                )
                outcome = 'answered'
            except CandleError as error:
                outcome = type(error).__name__
            print(f'{code} "{message}": {outcome}')

    def run(self):
        """Runs both parts in turn.

        Returns:
            None: This method returns nothing.
        """
        self.show_identifiers()
        self.show_refusals()


if __name__ == '__main__':
    IdentifiersAndRefusalsExample().run()

It prints:

Instruments to download: 3 of 4 rows
1|2885: segment 1, instrument 2885, EQ, expiry None, exchange nse, 6 segments to search
11|500325: segment 11, instrument 500325, EQ, expiry None, exchange bse, 6 segments to search
51|565899: segment 51, instrument 565899, FUT, expiry 2026-10-19T23:59:59, exchange mcx, 16 segments to search
e-apirl-0004 "Rate limit exceeded": CandleThrottled
e-session-0002 "Invalid Token": CandleAuthenticationError
e-instrument-0001 "Instrument not found": CandleInstrumentUnknown
examples/stock_brokers/instruments/historical/wisdom_capital/WisdomCapitalCandles/example_3_fifty_empty_windows_check_the_session.py
"""Shows what Wisdom Capital's downloader does after fifty empty windows in a row: it asks about an instrument that always has bars before deciding the session is dead.

The XTS platform answers an unknown instrument, a window before the history begins and an expired market data token in exactly the same way, with an empty bar string. So after fifty empty windows `fetch_candles` asks about NSE cash RELIANCE for the past week. If that is empty too it takes a fresh market data token, handing the refused one back so a token another process already replaced is recognised, and asks once more. Only if the second answer is empty as well does it raise `CandleAuthenticationError` and stop the broker.

The program runs two stand-ins for `WisdomCapitalAPI` through this. In the first, the market data token has expired: every answer is empty until a new token is taken, after which RELIANCE has bars, so the fiftieth window returns normally with the streak reset. In the second, the platform answers everything empty whatever the token, so the fiftieth window raises. The stand-ins count requests and record which stale token each replacement was given; the canary's window depends on today's date, so it is not printed.

A `WisdomCapitalCandles` constructor opens a PostgreSQL connection and logs in, so the program builds the object without running its constructor, as `test_runs/candle_parse.py` does, and supplies the rate limiter and logger the constructor would have made. The limiter is set to a thousand requests a second so the program does not wait, and the logger is silent so the warning the check logs does not clutter the output.

Run it from the project root:

    python examples/stock_brokers/instruments/historical/wisdom_capital/WisdomCapitalCandles/example_3_fifty_empty_windows_check_the_session.py
"""

import datetime
import logging

from stock_brokers.instruments.historical.base import (
    CandleAuthenticationError,
    RateLimiter,
)
from stock_brokers.instruments.historical.wisdom_capital import (
    WisdomCapitalCandles,
)


class EmptyAnsweringAPI:
    """A stand-in for `WisdomCapitalAPI` that answers empty until its token is replaced, or for ever.

    Attributes:
        recovers_after_new_token (bool): True when a fresh token makes the canary answer with bars.
        tokens_issued (int): How many market data tokens it has handed out.
        stale_tokens (list): The stale token each replacement was given.
        requests (int): How many chart requests it answered.
    """

    def __init__(self, recovers_after_new_token):
        """Builds the stand-in.

        Args:
            recovers_after_new_token (bool): True when a fresh token makes the canary answer with bars.

        Returns:
            None: This method returns nothing.
        """
        self.recovers_after_new_token = recovers_after_new_token
        self.tokens_issued = 0
        self.stale_tokens = []
        self.requests = 0

    def replace_market_data_session(self, stale_access_token=None):
        """Hands out a new market data token.

        Args:
            stale_access_token (str | None): The token that was refused, or None when none was.

        Returns:
            dict: The session's `access_token` and `user_id`.
        """
        self.tokens_issued += 1
        self.stale_tokens.append(stale_access_token)
        return {
            'access_token': f'market-data-token-{self.tokens_issued}',
            'user_id': 'WC1001',
        }

    def get(self, url, headers=None, params=None):
        """Answers a chart request, with bars only for the canary on a fresh token when it recovers.

        Args:
            url (str): The URL the downloader asked for.
            headers (dict): The headers it sent.
            params (dict): The query parameters it sent.

        Returns:
            dict: The response envelope, with the platform's result under `data`.
        """
        self.requests += 1
        bars = ''
        is_canary = params['exchangeInstrumentID'] == '2885'
        fresh_token = headers['authorization'] != 'market-data-token-1'
        if self.recovers_after_new_token and is_canary and fresh_token:
            bars = '1789031759|1279.5|1285.3|1279.5|1283.2|149743|0|'
        return {
            'status_code': 200,
            'data': {
                'dataReponse': bars,
            },
        }


class FiftyEmptyWindowsCheckTheSessionExample:
    """Runs fifty empty windows against a recovering platform and against a dead one."""

    def build_downloader(self, api):
        """Builds a downloader around a stand-in API without opening a database connection.

        Args:
            api (EmptyAnsweringAPI): The stand-in API.

        Returns:
            WisdomCapitalCandles: The downloader, holding its first market data token.
        """
        candles = object.__new__(WisdomCapitalCandles)
        candles._api = api
        candles._empty_streak = 0
        candles._relogin_used = False
        candles._limiter = RateLimiter(1000.0)
        candles._logger = logging.getLogger('example.wisdom_capital')
        candles._logger.disabled = True
        candles.after_relogin()
        return candles

    def walk_fifty_empty_windows(self, description, api):
        """Asks for fifty expired contracts' windows and prints how the fiftieth ends.

        Args:
            description (str): What the stand-in pretends.
            api (EmptyAnsweringAPI): The stand-in API.

        Returns:
            None: This method returns nothing.
        """
        candles = self.build_downloader(api)
        day = datetime.date(2025, 6, 2)
        print(description)
        try:
            for instrument_number in range(50):
                candles.fetch_candles(f'2|{40000 + instrument_number}', '5minute', day, day)
            print('  the fiftieth window returned normally')
        except CandleAuthenticationError as error:
            print(f'  {type(error).__name__}: {error}')
        print(f'  chart requests: {api.requests}')
        print(f'  market data tokens issued: {api.tokens_issued}, stale tokens handed back: {api.stale_tokens}')

    def run(self):
        """Runs the recovering platform, then the dead one.

        Returns:
            None: This method returns nothing.
        """
        self.walk_fifty_empty_windows('Expired market data token:', EmptyAnsweringAPI(True))
        self.walk_fifty_empty_windows('Platform answering nothing:', EmptyAnsweringAPI(False))


if __name__ == '__main__':
    FiftyEmptyWindowsCheckTheSessionExample().run()

It prints:

Expired market data token:
  the fiftieth window returned normally
  chart requests: 52
  market data tokens issued: 2, stale tokens handed back: [None, 'market-data-token-1']
Platform answering nothing:
  CandleAuthenticationError: wisdom_capital answers every window empty, including one for an instrument that always has bars, and a fresh market data login did not change it. Stopping rather than retiring series that may be perfectly good.
  chart requests: 52
  market data tokens issued: 2, stale tokens handed back: [None, 'market-data-token-1']

Methods:

Name Description
after_relogin

Take the market data token again after the broker session has been logged in again.
instruments

Every instrument Wisdom Capital has published, identified the way the wire needs.
identifier_parts

Split a stored Wisdom Capital identifier into its segment number and instrument id.
series_context

Read the exchange and segment out of a Wisdom Capital identifier.
fetch_candles

Fetch one window of Wisdom Capital candles.
parse_response

Turn Wisdom Capital's bar string into bars.
priority_for

Where a series sits in the queue. Lower is worked first.
seed

Make sure every instrument and interval this broker serves has a progress row.
claim

The next series due for a window, or None when there is nothing to do.
next_window

The date range to ask for next, as a (start, end, direction) triple.
relogin

Establish a fresh session and rebuild everything built on the old one.
visit

Fetch one window for one series and store it, advancing the watermark.
run

Work the queue until stopped, out of work, or out of time.
summary

How far the download has got, for a status display.
close

Close the database connection.
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def __init__(self, api=None):
    """
    Wisdom Capital candle downloader.

    - `api` is an authenticated `WisdomCapitalAPI`. One is constructed when not supplied.
      Either way a market data session is established on top of it, because the chart
      endpoint does not accept the interactive token.
    """
    super().__init__()
    if api is None:
        from stock_brokers.api.utilities.session import ensure_session
        from stock_brokers.api.wisdom_capital import WisdomCapitalAPI

        ensure_session(self.BROKER_NAME, logger=self._logger)
        api = WisdomCapitalAPI()
    self._api = api
    self._market_data_token = None
    self._empty_streak = 0
    self._log_in_to_market_data()

_log_in_to_market_data 

_log_in_to_market_data(refresh=False)

Take the market data token the broker session holds, replacing it when it has been refused.

Separate from the broker's ordinary session: XTS issues one token for trading and another for market data, and the chart endpoint refuses the first. WisdomCapitalAPI establishes both and publishes them in the shared last_login, so this asks that class for the token rather than minting one - a second market data login invalidates the first, which would take the live quotes feed's token with it.

  • refresh replaces the token in hand, for one that has stopped working. Rotating it is not destructive: whoever reads it next gets the new one.
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def _log_in_to_market_data(self, refresh=False):
    """
    Take the market data token the broker session holds, replacing it when it has been refused.

    Separate from the broker's ordinary session: XTS issues one token for trading and another
    for market data, and the chart endpoint refuses the first. `WisdomCapitalAPI` establishes
    both and publishes them in the shared `last_login`, so this asks that class for the token
    rather than minting one - a second market data login invalidates the first, which would
    take the live quotes feed's token with it.

    - `refresh` replaces the token in hand, for one that has stopped working. Rotating it is
      not destructive: whoever reads it next gets the new one.
    """
    try:
        session = self._api.replace_market_data_session(
            stale_access_token=self._market_data_token if refresh else None)
    except Exception as exception:
        raise CandleAuthenticationError(
            f"wisdom_capital market data login failed: "
            f"{type(exception).__name__}: {str(exception)[:200]}") from exception
    self._market_data_token = session["access_token"]

after_relogin 

after_relogin()

Take the market data token again after the broker session has been logged in again.

The chart endpoint takes the market data token, not the interactive one, so a fresh interactive session alone would leave a refused credential in place. Building the API class checks the market data token as well and replaces it only when it has stopped working, so this takes what that check settled on rather than forcing a new login, which would invalidate the token the live quotes feed is using. The empty-window check does force one, because there the market data token is what is suspect.

Example usage

From example_1_bar_string_and_end_stamps.py
self.candles.after_relogin()
From example_2_identifiers_and_refusals.py
self.candles.after_relogin()
From example_3_fifty_empty_windows_check_the_session.py
candles.after_relogin()
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
186
187
188
189
190
191
192
193
194
195
196
197
def after_relogin(self):
    """
    Take the market data token again after the broker session has been logged in again.

    The chart endpoint takes the market data token, not the interactive one, so a fresh
    interactive session alone would leave a refused credential in place. Building the API
    class checks the market data token as well and replaces it only when it has stopped
    working, so this takes what that check settled on rather than forcing a new login, which
    would invalidate the token the live quotes feed is using. The empty-window check does
    force one, because there the market data token is what is suspect.
    """
    self._log_in_to_market_data()

instruments 

instruments()

Every instrument Wisdom Capital has published, identified the way the wire needs.

The identifier is the segment's number joined to the instrument id, because one instrument id appears under several segments and the request carries both. The segment number comes from the platform's own segment name rather than from anything translated, since the names map one to one onto the numbers.

The instrument type is turned from XTS's number into a word so that priority_for can tell a future from an option, and the expiry is already ISO-8601 and needs nothing.

Example usage

From example_2_identifiers_and_refusals.py
instruments = self.candles.instruments()
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def instruments(self):
    """
    Every instrument Wisdom Capital has published, identified the way the wire needs.

    The identifier is the segment's number joined to the instrument id, because one
    instrument id appears under several segments and the request carries both. The segment
    number comes from the platform's own segment name rather than from anything translated,
    since the names map one to one onto the numbers.

    The instrument type is turned from XTS's number into a word so that `priority_for` can
    tell a future from an option, and the expiry is already ISO-8601 and needs nothing.
    """
    query = f"""
        select distinct on (i.exchangesegment, i."{self.TOKEN_COLUMN}")
               i.exchangesegment, i."{self.TOKEN_COLUMN}",
               i."{self.TYPE_COLUMN}", i."{self.EXPIRY_COLUMN}"
        from {self.BROKER_NAME}.instruments i
        where i."{self.TOKEN_COLUMN}" is not null and i."{self.TOKEN_COLUMN}" <> ''
        order by i.exchangesegment, i."{self.TOKEN_COLUMN}", i.download_date desc
    """
    with self._connection.cursor() as cursor:
        cursor.execute(query)
        rows = cursor.fetchall()
    self._connection.commit()

    instruments = []
    for segment, token, instrument_type, expiry in rows:
        number = EXCHANGE_SEGMENT_NUMBERS.get(segment)
        if number is None:
            continue
        instruments.append((f"{number}|{token}",
                            INSTRUMENT_TYPES.get(str(instrument_type), instrument_type),
                            expiry))
    return instruments

identifier_parts staticmethod 

identifier_parts(identifier)

Split a stored Wisdom Capital identifier into its segment number and instrument id.

The identifier is built in instruments and taken apart here and nowhere else, so the two cannot drift apart.

Parameters:

Name Type Description Default
identifier str

For example "11|500325".

 required

Returns:

Type Description
tuple

The segment number and the exchange instrument id.

Example usage

From example_2_identifiers_and_refusals.py
segment_number, instrument_identifier = self.candles.identifier_parts(identifier)
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
@staticmethod
def identifier_parts(identifier):
    """
    Split a stored Wisdom Capital identifier into its segment number and instrument id.

    The identifier is built in `instruments` and taken apart here and nowhere else, so the two
    cannot drift apart.

    Args:
        identifier (str): For example "11|500325".

    Returns:
        tuple: The segment number and the exchange instrument id.
    """
    segment_number, instrument_identifier = str(identifier).split("|", 1)
    return segment_number, instrument_identifier

series_context classmethod 

series_context(identifier)

Read the exchange and segment out of a Wisdom Capital identifier.

Parameters:

Name Type Description Default
identifier str

For example "11|500325".

 required

Returns:

Type Description
SeriesContext

The instrument id, with the exchange and segments its number names.

Example usage

From example_2_identifiers_and_refusals.py
context = WisdomCapitalCandles.series_context(identifier)
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
@classmethod
def series_context(cls, identifier):
    """
    Read the exchange and segment out of a Wisdom Capital identifier.

    Args:
        identifier (str): For example "11|500325".

    Returns:
        SeriesContext: The instrument id, with the exchange and segments its number names.
    """
    segment_number, instrument_identifier = cls.identifier_parts(identifier)
    exchange, family = cls.SEGMENT_NUMBERS.get(segment_number, (None, None))
    if family is None:
        return SeriesContext(instrument_identifier)
    return SeriesContext(instrument_identifier, exchange, segments_of(exchange, family),
                         exchange_token=instrument_identifier)

fetch_candles 

fetch_candles(token, interval, start_date, end_date)

Fetch one window of Wisdom Capital candles.

  • token is the segmentNumber|instrumentId identifier from instruments.
  • interval is the stored interval name.
  • start_date and end_date bound the window, inclusive.

Example usage

From example_1_bar_string_and_end_stamps.py
payload = self.candles.fetch_candles('1|2885', '1minute', day, day)
From example_2_identifiers_and_refusals.py
self.candles.fetch_candles(
    '2|35001',
    '5minute',
    datetime.date(2025, 1, 1),
    datetime.date(2025, 3, 31),
)
From example_3_fifty_empty_windows_check_the_session.py
candles.fetch_candles(f'2|{40000 + instrument_number}', '5minute', day, day)
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def fetch_candles(self, token, interval, start_date, end_date):
    """
    Fetch one window of Wisdom Capital candles.

    - `token` is the `segmentNumber|instrumentId` identifier from `instruments`.
    - `interval` is the stored interval name.
    - `start_date` and `end_date` bound the window, inclusive.
    """
    segment_number, instrument_identifier = self.identifier_parts(token)
    window_start = datetime.datetime(start_date.year, start_date.month, start_date.day)
    window_end = datetime.datetime.combine(end_date + datetime.timedelta(days=1),
                                           datetime.time(0, 0))

    payload = self._ask(segment_number, instrument_identifier, self.INTERVALS[interval],
                        window_start, window_end)

    if payload:
        self._empty_streak = 0
    else:
        self._empty_streak += 1
        if self._empty_streak >= EMPTY_WINDOWS_BEFORE_CHECKING:
            self._check_still_alive()

    return payload

_ask 

_ask(segment_number, instrument_identifier, compression, window_start, window_end)

Make one request and return the bar string it carried, which may be empty.

  • segment_number is the XTS exchange segment number.
  • instrument_identifier is the instrument id within that segment.
  • compression is the bar length in seconds.
  • window_start and window_end are India wall clock times bounding the window.
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def _ask(self, segment_number, instrument_identifier, compression, window_start, window_end):
    """
    Make one request and return the bar string it carried, which may be empty.

    - `segment_number` is the XTS exchange segment number.
    - `instrument_identifier` is the instrument id within that segment.
    - `compression` is the bar length in seconds.
    - `window_start` and `window_end` are India wall clock times bounding the window.
    """
    parameters = {"exchangeSegment": segment_number,
                  "exchangeInstrumentID": instrument_identifier,
                  "startTime": window_start.strftime(self.TIME_FORMAT),
                  "endTime": window_end.strftime(self.TIME_FORMAT),
                  "compressionValue": compression}

    try:
        response = self._api.get(url=HISTORICAL_URL,
                                 headers={"authorization": self._market_data_token},
                                 params=parameters)
    except Exception as exception:
        raise self._classify(exception) from exception

    # The API class unwraps the envelope's `result`, which is where the bars live. The
    # platform's spelling of the key is kept, and the correct spelling accepted too in case
    # that is ever fixed.
    result = (response or {}).get("data") or {}
    return result.get("dataReponse") or result.get("dataResponse") or ""

_check_still_alive 

_check_still_alive()

Decide whether a run of empty windows means dead instruments or a dead session.

This platform answers an unknown instrument, a window before the history begins and an expired market data token identically, so the only way to tell them apart is to ask about an instrument that is known to have bars. If that is empty too the token is logged in again and asked once more, because expiry is the likeliest explanation and it is cheap to fix. Only if the second answer is empty as well is the broker stopped.

Source code in stock_brokers/instruments/historical/wisdom_capital.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
def _check_still_alive(self):
    """
    Decide whether a run of empty windows means dead instruments or a dead session.

    This platform answers an unknown instrument, a window before the history begins and an
    expired market data token identically, so the only way to tell them apart is to ask about
    an instrument that is known to have bars. If that is empty too the token is logged in
    again and asked once more, because expiry is the likeliest explanation and it is cheap to
    fix. Only if the second answer is empty as well is the broker stopped.
    """
    self._empty_streak = 0
    yesterday = datetime.datetime.now(INDIA_TIMEZONE).replace(tzinfo=None)
    window_start = yesterday - datetime.timedelta(days=7)

    self._limiter.take()
    if self._ask(str(CANARY_SEGMENT), str(CANARY_INSTRUMENT), "3600",
                 window_start, yesterday):
        return

    self._logger.warning(f"{EMPTY_WINDOWS_BEFORE_CHECKING} empty windows in a row and the "
                         f"session check came back empty too. Logging in again.")
    self._log_in_to_market_data(refresh=True)

    self._limiter.take()
    if self._ask(str(CANARY_SEGMENT), str(CANARY_INSTRUMENT), "3600",
                 window_start, yesterday):
        return

    # This check has already logged the market data session in again, which is the one login
    # the rule allows. Marking it spent stops the base class logging in a second time.
    self._relogin_used = True
    raise CandleAuthenticationError(
        "wisdom_capital answers every window empty, including one for an instrument that "
        "always has bars, and a fresh market data login did not change it. Stopping rather "
        "than retiring series that may be perfectly good.")

_classify staticmethod 

_classify(exception)

Decide what a Wisdom Capital refusal means for the series being downloaded.

The platform names its conditions in the body with codes of its own, which the API class raises as the exception's code.

  • exception is what the API call raised.
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
@staticmethod
def _classify(exception):
    """
    Decide what a Wisdom Capital refusal means for the series being downloaded.

    The platform names its conditions in the body with codes of its own, which the API class
    raises as the exception's code.

    - `exception` is what the API call raised.
    """
    text = str(exception)
    lowered = text.lower()

    if "e-apirl" in lowered or "429" in text or ("rate" in lowered and "limit" in lowered):
        return CandleThrottled(text[:200])
    if "e-session" in lowered or "e-token" in lowered:
        return CandleAuthenticationError(text[:200])
    if "e-instrument" in lowered:
        return CandleInstrumentUnknown(text[:200])
    return exception

parse_response 

parse_response(payload, interval)

Turn Wisdom Capital's bar string into bars.

Each bar is timestamp|open|high|low|close|volume|openInterest, bars separated by commas, with a trailing pipe leaving an empty last field.

  • payload is the bar string fetch_candles returned.
  • interval is the stored interval name. It is needed rather than ignored, because the timestamp names the end of the bar and the bar's own length is what converts that into its start.

Example usage

From example_1_bar_string_and_end_stamps.py
self.candles.parse_response(payload, '1minute')
From example_1_bar_string_and_end_stamps.py
first_as_five_minutes = self.candles.parse_response(payload, '5minute')[0][0]
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def parse_response(self, payload, interval):
    """
    Turn Wisdom Capital's bar string into bars.

    Each bar is `timestamp|open|high|low|close|volume|openInterest`, bars separated by commas,
    with a trailing pipe leaving an empty last field.

    - `payload` is the bar string `fetch_candles` returned.
    - `interval` is the stored interval name. It is needed rather than ignored, because the
      timestamp names the end of the bar and the bar's own length is what converts that into
      its start.
    """
    bar_seconds = int(self.INTERVALS[interval])

    bars = []
    for row in (payload or "").split(","):
        if not row:
            continue
        fields = row.split("|")
        if len(fields) < 7:
            continue
        bars.append((self._bar_start_time(int(fields[0]), bar_seconds),
                     float(fields[1]), float(fields[2]), float(fields[3]), float(fields[4]),
                     int(float(fields[5])), int(float(fields[6]))))
    bars.sort(key=lambda bar: bar[0])
    return bars

_bar_start_time staticmethod 

_bar_start_time(stamp, bar_seconds)

Turn one of the platform's timestamps into the time the bar opened.

Two corrections, both measured. The number is seconds since the epoch computed as though India time were UTC, so it is decoded as UTC and the result relabelled as India time, which yields the India wall clock reading directly. And it names the bar's last second, so the bar's length less one second is subtracted to reach its start.

  • stamp is the epoch value as the platform reported it.
  • bar_seconds is the length of the bar in seconds.
Source code in stock_brokers/instruments/historical/wisdom_capital.py 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
@staticmethod
def _bar_start_time(stamp, bar_seconds):
    """
    Turn one of the platform's timestamps into the time the bar opened.

    Two corrections, both measured. The number is seconds since the epoch computed as though
    India time were UTC, so it is decoded as UTC and the result relabelled as India time,
    which yields the India wall clock reading directly. And it names the bar's last second,
    so the bar's length less one second is subtracted to reach its start.

    - `stamp` is the epoch value as the platform reported it.
    - `bar_seconds` is the length of the bar in seconds.
    """
    india_reading = datetime.datetime.fromtimestamp(
        stamp, datetime.timezone.utc).replace(tzinfo=INDIA_TIMEZONE)
    return india_reading - datetime.timedelta(seconds=bar_seconds - 1)

priority_for 

priority_for(instrument_type, expiry, interval)

Where a series sits in the queue. Lower is worked first.

The ordering is the difference between useful data in days and useful data in months. Cash instruments come first because they have the deepest history and are wanted most; expired options come last because there are tens of thousands of them, each holding a few weeks of data, and most will be refused outright. Coarse intervals beat fine ones within each tier, since a day bar covers a year in one request and a minute bar covers two months.

  • instrument_type is the broker's own type string.
  • expiry is the contract expiry, or None for cash.
  • interval is the stored interval name.

Example usage

From example_1_one_window_of_daily_bars.py
priority = self.downloader.priority_for(instrument_type, expiry, interval)
Source code in stock_brokers/instruments/historical/base.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
def priority_for(self, instrument_type, expiry, interval):
    """
    Where a series sits in the queue. Lower is worked first.

    The ordering is the difference between useful data in days and useful data in months.
    Cash instruments come first because they have the deepest history and are wanted most;
    expired options come last because there are tens of thousands of them, each holding a few
    weeks of data, and most will be refused outright. Coarse intervals beat fine ones within
    each tier, since a day bar covers a year in one request and a minute bar covers two
    months.

    - `instrument_type` is the broker's own type string.
    - `expiry` is the contract expiry, or None for cash.
    - `interval` is the stored interval name.
    """
    kind = (instrument_type or "").upper()
    expired = False
    if expiry:
        try:
            expired = datetime.date.fromisoformat(str(expiry)[:10]) < datetime.date.today()
        except ValueError:
            expired = False

    if not expiry:
        base = 10                                   # cash: equities, indices
    elif kind in ("FUT", "FUTSTK", "FUTIDX", "FUTCOM", "FUTCUR"):
        base = 80 if expired else 30
    else:
        base = 90 if expired else 60                # options, live or dead

    if interval in ("day", "week", "month"):
        return base
    minutes = self._interval_minutes(interval)
    return base + (5 if minutes >= 15 else 15)

_interval_minutes staticmethod 

_interval_minutes(interval)

Minutes in an interval name like 15minute, or 0 when it is not a minute interval.

  • interval is the stored interval name.
Source code in stock_brokers/instruments/historical/base.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
@staticmethod
def _interval_minutes(interval):
    """
    Minutes in an interval name like `15minute`, or 0 when it is not a minute interval.

    - `interval` is the stored interval name.
    """
    if not interval.endswith("minute"):
        return 0
    try:
        return int(interval[:-len("minute")])
    except ValueError:
        return 0

seed 

seed()

Make sure every instrument and interval this broker serves has a progress row.

Safe to run again: existing rows are left exactly as they are, so re-seeding after a new instrument master picks up what is new without disturbing what is already downloaded. Returns how many rows were added.

Example usage

From example_1_one_window_of_daily_bars.py
print(f'Series added by the first seed: {self.downloader.seed()}')
From example_1_one_window_of_daily_bars.py
print(f'Series added by a second seed: {self.downloader.seed()}')
Source code in stock_brokers/instruments/historical/base.py 
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
def seed(self):
    """
    Make sure every instrument and interval this broker serves has a progress row.

    Safe to run again: existing rows are left exactly as they are, so re-seeding after a new
    instrument master picks up what is new without disturbing what is already downloaded.
    Returns how many rows were added.
    """
    instruments = self.instruments()
    if not instruments:
        self._logger.warning(f"No instruments for {self.BROKER_NAME}; nothing to seed. "
                             f"Has the instrument master been downloaded?")
        return 0

    today = datetime.date.today()
    considered = 0
    batch = []

    with self._connection.cursor() as cursor:
        # Counted by measuring the table, not by summing cursor.rowcount: execute_values
        # splits a batch into pages of its own, and rowcount then reports only the last page,
        # which made a complete seed of 1,149,064 rows announce itself as 22,964.
        cursor.execute(f"select count(*) from {self._progress_table}")
        before = cursor.fetchone()[0]

        # Written a batch at a time as the instruments are walked rather than assembled in
        # full first. Wisdom Capital alone is 372,545 instruments across thirteen intervals,
        # and holding all 4.8 million rows in memory to write them costs a gigabyte for
        # nothing.
        for token, instrument_type, expiry in instruments:
            for interval in self.INTERVALS:
                batch.append((str(token), interval,
                              self.priority_for(instrument_type, expiry, interval), today))
            considered += len(self.INTERVALS)
            if len(batch) >= WRITE_BATCH_ROWS:
                self._insert_series(cursor, batch)
                batch = []
        if batch:
            self._insert_series(cursor, batch)

        cursor.execute(f"select count(*) from {self._progress_table}")
        added = cursor.fetchone()[0] - before
    self._connection.commit()
    self._logger.info(f"Seeded {added} new series ({considered} considered).")
    return added

_insert_series 

_insert_series(cursor, batch)

Register one batch of series, leaving any that already exist exactly as they are.

  • cursor is an open cursor on this broker's connection.
  • batch is a list of (token, interval, priority, seeded_date) tuples.
Source code in stock_brokers/instruments/historical/base.py 
449
450
451
452
453
454
455
456
457
458
459
460
461
def _insert_series(self, cursor, batch):
    """
    Register one batch of series, leaving any that already exist exactly as they are.

    - `cursor` is an open cursor on this broker's connection.
    - `batch` is a list of `(token, interval, priority, seeded_date)` tuples.
    """
    execute_values(cursor, f"""
        insert into {self._progress_table}
            (instrument_token, "interval", priority, seeded_date)
        values %s
        on conflict (instrument_token, "interval") do nothing
    """, batch, page_size=WRITE_BATCH_ROWS)

claim 

claim()

The next series due for a window, or None when there is nothing to do.

Ordered by priority and then by how long a series has been waiting. Series that have reached the broker's limit, or that are backing off after failures, are skipped.

Example usage

From example_1_one_window_of_daily_bars.py
series = self.downloader.claim()
Source code in stock_brokers/instruments/historical/base.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def claim(self):
    """
    The next series due for a window, or None when there is nothing to do.

    Ordered by priority and then by how long a series has been waiting. Series that have
    reached the broker's limit, or that are backing off after failures, are skipped.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            select instrument_token, "interval", earliest_bar_time, latest_bar_time,
                   oldest_requested_time, empty_window_streak, consecutive_failures
            from {self._progress_table}
            where reached_broker_limit = false
              and consecutive_failures < %s
              and (next_attempt_after is null or next_attempt_after <= now())
            order by priority, last_attempt_at nulls first
            limit 1
        """, (MAXIMUM_CONSECUTIVE_FAILURES,))
        row = cursor.fetchone()
    self._connection.commit()
    return row

next_window 

next_window(series)

The date range to ask for next, as a (start, end, direction) triple.

The backward walk comes first and runs to completion. A series starts at today and steps back a window at a time until it reaches the broker's earliest date or the broker stops returning anything, and only then does it start filling forward with whatever has happened since.

Doing it the other way round - preferring forward whenever the last stored bar is older than today - looks reasonable and does not work: outside market hours the last bar is always older than today, so the series asks for the same empty recent window forever and never walks back at all.

  • series is a row from claim.

Example usage

From example_1_one_window_of_daily_bars.py
start_date, end_date, direction = self.downloader.next_window(series)
Source code in stock_brokers/instruments/historical/base.py 
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
515
516
517
518
519
520
521
522
523
def next_window(self, series):
    """
    The date range to ask for next, as a `(start, end, direction)` triple.

    The backward walk comes first and runs to completion. A series starts at today and steps
    back a window at a time until it reaches the broker's earliest date or the broker stops
    returning anything, and only then does it start filling forward with whatever has
    happened since.

    Doing it the other way round - preferring forward whenever the last stored bar is older
    than today - looks reasonable and does not work: outside market hours the last bar is
    always older than today, so the series asks for the same empty recent window forever and
    never walks back at all.

    - `series` is a row from `claim`.
    """
    _, interval, earliest, latest, oldest_requested, _, _ = series
    span = datetime.timedelta(days=self.MAXIMUM_WINDOW_DAYS.get(interval, 60) - 1)
    today = datetime.date.today()

    backward_done = (oldest_requested is not None
                     and oldest_requested.date() <= self.EARLIEST_AVAILABLE_DATE)

    if not backward_done:
        end = today if oldest_requested is None else (oldest_requested.date()
                                                      - datetime.timedelta(days=1))
        start = max(end - span, self.EARLIEST_AVAILABLE_DATE)
        if start > end:
            return None, None, "exhausted"
        return start, end, "backward"

    if latest is None:
        return None, None, "exhausted"
    start = latest.date()
    if start >= today:
        return None, None, "current"
    return start, min(start + span, today), "forward"

_fetch_bars 

_fetch_bars(token, interval, start_date, end_date)

Fetch and parse one window, logging in once first if the broker refuses the session.

A token expiring does not mean the download is over, and it does not mean this series is at fault. The rule is to log in first and then carry on: one login, then the same window again. Only if that login fails, or the fresh session is refused as well, does the error go back up - where visit and run stop the broker rather than carry on with a dead token.

Both alternatives are worse. Stopping outright leaves the download waiting for systemd to restart it into a login, and recording the refusal against whichever series it was on and moving to the next sends thousands of refused requests an hour - enough for Cloudflare to ban the machine from Fyers.

The login is re-armed by any window that comes back without a session error, so a backfill that runs for weeks gets its one login at each daily expiry.

  • token, interval, start_date and end_date are as for fetch_candles.
Source code in stock_brokers/instruments/historical/base.py 
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
def _fetch_bars(self, token, interval, start_date, end_date):
    """
    Fetch and parse one window, logging in once first if the broker refuses the session.

    A token expiring does not mean the download is over, and it does not mean this series is at
    fault. The rule is to log in first and then carry on: one login, then the same window
    again. Only if that login fails, or the fresh session is refused as well, does the error go
    back up - where `visit` and `run` stop the broker rather than carry on with a dead token.

    Both alternatives are worse. Stopping outright leaves the download waiting for systemd to
    restart it into a login, and recording the refusal against whichever series it was on and
    moving to the next sends thousands of refused requests an hour - enough for Cloudflare to
    ban the machine from Fyers.

    The login is re-armed by any window that comes back without a session error, so a
    backfill that runs for weeks gets its one login at each daily expiry.

    - `token`, `interval`, `start_date` and `end_date` are as for `fetch_candles`.
    """
    try:
        bars = self.parse_response(self.fetch_candles(token, interval, start_date, end_date),
                                   interval)
    except CandleAuthenticationError as failure:
        if self._relogin_used:
            raise
        self._relogin_used = True
        self._logger.warning(f"{self.BROKER_NAME} refused the session ({failure}). Logging in "
                             f"once before continuing.")
        self.relogin()
        self._limiter.take()
        bars = self.parse_response(self.fetch_candles(token, interval, start_date, end_date),
                                   interval)
    self._relogin_used = False
    return bars

relogin 

relogin()

Establish a fresh session and rebuild everything built on the old one.

Through ensure_session, so it takes the same Redis lock and obeys the same floor on login attempts as any other process using it - which means it may wait, and may find another process has already logged in, in which case that session is simply used.

Raises:

Type Description
CandleAuthenticationError

If no working session could be established.

Example usage

From example_3_logging_in_again_after_a_refused_session.py
downloader.relogin()
Source code in stock_brokers/instruments/historical/base.py 
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
def relogin(self):
    """
    Establish a fresh session and rebuild everything built on the old one.

    Through `ensure_session`, so it takes the same Redis lock and obeys the same floor on login
    attempts as any other process using it - which means it may wait, and may find another
    process has already logged in, in which case that session is simply used.

    Raises:
        CandleAuthenticationError: If no working session could be established.
    """
    from stock_brokers.api.utilities.session import ensure_session

    try:
        ensure_session(self.BROKER_NAME, logger=self._logger)
        self._api = self._build_api()
        self.after_relogin()
    except CandleAuthenticationError:
        raise
    except Exception as exception:
        raise CandleAuthenticationError(
            f"logging in again did not produce a working session: "
            f"{type(exception).__name__}: {str(exception)[:200]}") from exception
    self._logger.info(f"{self.BROKER_NAME} logged in again. Continuing.")

_build_api 

_build_api()

Construct this broker's authenticated API class, around whatever session is now stored.

The broker's own class from api_class_for, which is what every downloader builds. A downloader that builds its API differently overrides this.

Source code in stock_brokers/instruments/historical/base.py 
589
590
591
592
593
594
595
596
597
598
def _build_api(self):
    """
    Construct this broker's authenticated API class, around whatever session is now stored.

    The broker's own class from `api_class_for`, which is what every downloader builds. A
    downloader that builds its API differently overrides this.
    """
    from stock_brokers.api.utilities.session import api_class_for

    return api_class_for(self.BROKER_NAME)()

visit 

visit(series)

Fetch one window for one series and store it, advancing the watermark.

Returns the number of bars stored. The bars and the watermark are written in the same transaction, so an interruption costs at most the window in flight - and repeating that window is harmless, because bars upsert on their key.

  • series is a row from claim.

Example usage

From example_1_one_window_of_daily_bars.py
stored = self.downloader.visit(series)
From example_3_logging_in_again_after_a_refused_session.py
stored = downloader.visit(self.series('738561'))
Source code in stock_brokers/instruments/historical/base.py 
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
def visit(self, series):
    """
    Fetch one window for one series and store it, advancing the watermark.

    Returns the number of bars stored. The bars and the watermark are written in the same
    transaction, so an interruption costs at most the window in flight - and repeating that
    window is harmless, because bars upsert on their key.

    - `series` is a row from `claim`.
    """
    token, interval, _, previous_latest, _, empty_streak, _ = series
    start_date, end_date, direction = self.next_window(series)

    if direction == "exhausted":
        if previous_latest is None:
            # Walked the whole range the broker serves and never saw a bar. A contract that
            # listed and never traded is the usual case, and there are tens of thousands of
            # them. Without this they would be claimed again the moment they were put down -
            # `next_window` answers "exhausted" every time, so nothing would move - and the
            # queue would spin through them at database speed making no requests at all.
            self._retire(token, interval,
                         "the broker served no bars anywhere in the range it offers")
            return 0
        # The backward walk is finished. Not retired: the series still gets new bars each
        # day, it simply has no more past to collect.
        self._finish_backfill(token, interval, "reached the broker's earliest date")
        return 0
    if direction == "current":
        self._defer_until_tomorrow(token, interval, "up to date")
        return 0

    self._limiter.take()
    try:
        bars = self._fetch_bars(token, interval, start_date, end_date)
    except CandleInstrumentUnknown as exception:
        self._retire(token, interval, str(exception))
        return 0
    except CandleThrottled as exception:
        # A throttle says something about the pace, not about this series, so it costs the
        # limiter five seconds and leaves the series claimable. Counting it as a failure
        # instead would push a perfectly good series an hour into the future - and a broker
        # that throttles does it to a run of series, not to one.
        self._limiter.back_off(5.0)
        self._record_throttle(token, interval, str(exception))
        return 0
    except (CandleAuthenticationError, CandleBlocked):
        # Neither says anything about this series, so neither is recorded against it. Before a
        # dead token was re-raised here, every refused request landed on whichever series was
        # claimed, and after five of those a good series stopped being claimed at all.
        raise
    except Exception as exception:
        self._record_failure(token, interval,
                             f"{type(exception).__name__}: {str(exception)[:200]}")
        return 0

    if not bars:
        streak = (empty_streak or 0) + 1
        if direction == "backward" and streak >= MAXIMUM_EMPTY_WINDOWS:
            # The broker has nothing older. Stop walking back, keep collecting forward.
            self._finish_backfill(token, interval,
                                  f"{streak} empty windows walking back from {end_date}")
        elif direction == "forward":
            self._defer_until_tomorrow(token, interval, "no new bars yet")
        else:
            self._record_empty(token, interval, streak, start_date, direction)
        return 0

    self._store(token, interval, bars, start_date, direction, previous_latest)
    return len(bars)

_store 

_store(token, interval, bars, requested_from, direction, previous_latest=None)

Write a window's bars and move the watermark, in one transaction.

  • token is the broker's instrument identifier.
  • interval is the stored interval name.
  • bars are the parsed (time, open, high, low, close, volume, oi) tuples.
  • requested_from is the start of the window that was asked for.
  • direction is whether the walk was forward or backward.
  • previous_latest is the newest bar already stored, used to tell a forward window that gained ground from one that returned only what was already known.
Source code in stock_brokers/instruments/historical/base.py 
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
def _store(self, token, interval, bars, requested_from, direction, previous_latest=None):
    """
    Write a window's bars and move the watermark, in one transaction.

    - `token` is the broker's instrument identifier.
    - `interval` is the stored interval name.
    - `bars` are the parsed `(time, open, high, low, close, volume, oi)` tuples.
    - `requested_from` is the start of the window that was asked for.
    - `direction` is whether the walk was forward or backward.
    - `previous_latest` is the newest bar already stored, used to tell a forward window that
      gained ground from one that returned only what was already known.
    """
    # Collapsed on bar time before the write, last one winning. A broker can answer with the
    # same timestamp twice in one window - Dhan does - and Postgres refuses a batch that
    # touches the same row twice with "ON CONFLICT DO UPDATE command cannot affect row a
    # second time", which killed the whole process rather than the window.
    by_time = {}
    for bar in bars:
        volume = self._value_within_bigint(token, interval, bar[0], "volume", bar[5])
        open_interest = self._value_within_bigint(token, interval, bar[0], "oi", bar[6])
        by_time[bar[0]] = (bar[0], str(token), interval,
                           bar[1], bar[2], bar[3], bar[4], volume, open_interest)
    rows = list(by_time.values())
    times = list(by_time)

    try:
        with self._connection.cursor() as cursor:
            for start in range(0, len(rows), WRITE_BATCH_ROWS):
                execute_values(cursor, f"""
                    insert into {self._table}
                        ("time", instrument_token, "interval", open, high, low, close,
                         volume, oi)
                    values %s
                    on conflict (instrument_token, "interval", "time") do update set
                        open = excluded.open, high = excluded.high, low = excluded.low,
                        close = excluded.close, volume = excluded.volume, oi = excluded.oi,
                        downloaded_at = now()
                """, rows[start:start + WRITE_BATCH_ROWS])

            cursor.execute(f"""
                update {self._progress_table} set
                    earliest_bar_time = least(coalesce(earliest_bar_time, %s), %s),
                    latest_bar_time = greatest(coalesce(latest_bar_time, %s), %s),
                    oldest_requested_time = case when %s = 'backward'
                        then least(coalesce(oldest_requested_time, %s), %s)
                        else oldest_requested_time end,
                    bar_count = bar_count + %s,
                    empty_window_streak = 0,
                    consecutive_failures = 0,
                    next_attempt_after = null,
                    last_attempt_at = now(),
                    last_outcome = 'stored',
                    last_failure_reason = null
                where instrument_token = %s and "interval" = %s
            """, (min(times), min(times), max(times), max(times),
                  direction, requested_from, requested_from,
                  len(rows), str(token), interval))
        self._connection.commit()
    except Exception:
        self._connection.rollback()
        raise

    # A forward window that returned nothing newer than what was already stored means the
    # market has not produced more yet. Without this the series would be claimed again
    # immediately and spend the rate limit re-fetching the same bars all weekend.
    if direction == "forward" and previous_latest is not None and max(times) <= previous_latest:
        self._defer_until_tomorrow(token, interval, "forward window gained no new bars")

_value_within_bigint 

_value_within_bigint(token, interval, bar_time, column, value)

Returns a bar's count, or None when the count cannot fit in a BIGINT column.

Parameters:

Name Type Description Default
token str

The broker's instrument identifier.

 required
interval str

The stored interval name.

 required
bar_time datetime

The time of the bar the value belongs to.

 required
column str

The name of the column the value is written to, either "volume" or "oi".

 required
value int | float | None

The count exactly as the broker sent it.

 required

Returns:

Type Description
int | float | None

The value unchanged when it fits in a BIGINT or is not a number, and None when it does not fit.
Source code in stock_brokers/instruments/historical/base.py 
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
def _value_within_bigint(self, token, interval, bar_time, column, value):
    """Returns a bar's count, or None when the count cannot fit in a BIGINT column.

    Args:
        token (str): The broker's instrument identifier.
        interval (str): The stored interval name.
        bar_time (datetime.datetime): The time of the bar the value belongs to.
        column (str): The name of the column the value is written to, either "volume" or "oi".
        value (int | float | None): The count exactly as the broker sent it.

    Returns:
        int | float | None: The value unchanged when it fits in a BIGINT or is not a number, and None when it does not fit.
    """
    if not isinstance(value, (int, float)):
        return value
    if BIGINT_MINIMUM <= value <= BIGINT_MAXIMUM:
        return value
    self._logger.warning(
        "%s %s %s: %s %s does not fit in BIGINT, storing NULL.",
        token,
        interval,
        bar_time.isoformat(),
        column,
        value,
    )
    return None

_record_empty 

_record_empty(token, interval, streak, requested_from, direction)

Note that a window came back with nothing, and step past it.

  • token, interval identify the series.
  • streak is how many empty windows in a row this makes.
  • requested_from is the start of the window that was asked for.
  • direction is whether the walk was forward or backward.
Source code in stock_brokers/instruments/historical/base.py 
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
def _record_empty(self, token, interval, streak, requested_from, direction):
    """
    Note that a window came back with nothing, and step past it.

    - `token`, `interval` identify the series.
    - `streak` is how many empty windows in a row this makes.
    - `requested_from` is the start of the window that was asked for.
    - `direction` is whether the walk was forward or backward.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            update {self._progress_table} set
                empty_window_streak = %s,
                oldest_requested_time = case when %s = 'backward'
                    then least(coalesce(oldest_requested_time, %s), %s)
                    else oldest_requested_time end,
                last_attempt_at = now(),
                last_outcome = 'empty',
                consecutive_failures = 0,
                next_attempt_after = null
            where instrument_token = %s and "interval" = %s
        """, (streak, direction, requested_from, requested_from, str(token), interval))
    self._connection.commit()

_finish_backfill 

_finish_backfill(token, interval, reason)

Mark the backward walk complete, leaving the series collecting forward.

Distinct from _retire: the broker has no more past for this series, but it will keep producing new bars. Retiring it here would quietly stop a live instrument updating.

  • token, interval identify the series.
  • reason is why the walk stopped, kept for review.
Source code in stock_brokers/instruments/historical/base.py 
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
def _finish_backfill(self, token, interval, reason):
    """
    Mark the backward walk complete, leaving the series collecting forward.

    Distinct from `_retire`: the broker has no more *past* for this series, but it will keep
    producing new bars. Retiring it here would quietly stop a live instrument updating.

    - `token`, `interval` identify the series.
    - `reason` is why the walk stopped, kept for review.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            update {self._progress_table} set
                oldest_requested_time = %s,
                empty_window_streak = 0,
                consecutive_failures = 0,
                next_attempt_after = null,
                last_attempt_at = now(),
                last_outcome = 'backfilled',
                limit_reason = %s
            where instrument_token = %s and "interval" = %s
        """, (self.EARLIEST_AVAILABLE_DATE, reason[:500], str(token), interval))
    self._connection.commit()
    self._logger.info(f"{token} {interval}: backfill complete ({reason}).")

_defer_until_tomorrow 

_defer_until_tomorrow(token, interval, reason)

Put a series aside until the next day.

Reached when a forward window brings nothing new, which outside market hours is every forward window. Without it the highest priority series would be re-fetched continuously and spend the whole rate limit re-reading bars already stored.

  • token, interval identify the series.
  • reason is recorded as the outcome.
Source code in stock_brokers/instruments/historical/base.py 
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
def _defer_until_tomorrow(self, token, interval, reason):
    """
    Put a series aside until the next day.

    Reached when a forward window brings nothing new, which outside market hours is every
    forward window. Without it the highest priority series would be re-fetched continuously
    and spend the whole rate limit re-reading bars already stored.

    - `token`, `interval` identify the series.
    - `reason` is recorded as the outcome.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            update {self._progress_table} set
                next_attempt_after = date_trunc('day', now()) + interval '1 day',
                last_attempt_at = now(),
                last_outcome = %s,
                consecutive_failures = 0
            where instrument_token = %s and "interval" = %s
        """, (reason[:100], str(token), interval))
    self._connection.commit()

_record_throttle 

_record_throttle(token, interval, reason)

Note that the broker asked us to slow down, without holding it against the series.

The window is left exactly as it was, so the same series is claimed again once the limiter's back off has passed and the work is simply redone.

  • token, interval identify the series.
  • reason is what the broker said.
Source code in stock_brokers/instruments/historical/base.py 
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
def _record_throttle(self, token, interval, reason):
    """
    Note that the broker asked us to slow down, without holding it against the series.

    The window is left exactly as it was, so the same series is claimed again once the
    limiter's back off has passed and the work is simply redone.

    - `token`, `interval` identify the series.
    - `reason` is what the broker said.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            update {self._progress_table} set
                last_attempt_at = now(),
                last_outcome = 'throttled',
                last_failure_reason = %s
            where instrument_token = %s and "interval" = %s
        """, (reason[:500], str(token), interval))
    self._connection.commit()
    self._logger.warning(f"{token} {interval}: throttled: {reason[:150]}")

_record_failure 

_record_failure(token, interval, reason)

Note a failure and push the next attempt out exponentially.

An hour, then two, four, eight and so on, so a series that is permanently broken stops crowding out the ones that are not.

  • token, interval identify the series.
  • reason is what went wrong, stored for a human to read.
Source code in stock_brokers/instruments/historical/base.py 
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
def _record_failure(self, token, interval, reason):
    """
    Note a failure and push the next attempt out exponentially.

    An hour, then two, four, eight and so on, so a series that is permanently broken stops
    crowding out the ones that are not.

    - `token`, `interval` identify the series.
    - `reason` is what went wrong, stored for a human to read.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            update {self._progress_table} set
                consecutive_failures = consecutive_failures + 1,
                next_attempt_after = now() + interval '1 hour'
                                     * power(2, least(consecutive_failures, 6)),
                last_attempt_at = now(),
                last_outcome = 'failed',
                last_failure_reason = %s
            where instrument_token = %s and "interval" = %s
        """, (reason[:500], str(token), interval))
    self._connection.commit()
    self._logger.warning(f"{token} {interval}: {reason}")

_retire 

_retire(token, interval, reason)

Stop asking for a series the broker will not serve.

  • token, interval identify the series.
  • reason is why, stored so the decision can be reviewed.
Source code in stock_brokers/instruments/historical/base.py 
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
def _retire(self, token, interval, reason):
    """
    Stop asking for a series the broker will not serve.

    - `token`, `interval` identify the series.
    - `reason` is why, stored so the decision can be reviewed.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            update {self._progress_table} set
                reached_broker_limit = true,
                limit_reason = %s,
                last_attempt_at = now(),
                last_outcome = 'retired'
            where instrument_token = %s and "interval" = %s
        """, (reason[:500], str(token), interval))
    self._connection.commit()

run 

run(stop_event=None, deadline_seconds=None)

Work the queue until stopped, out of work, or out of time.

Returns a (visits, bars) pair.

  • stop_event is a threading.Event that ends the loop when set.
  • deadline_seconds ends the loop after this long, for a job with a nightly window.

Example usage

From example_2_run_until_the_queue_is_drained.py
visits, bars = self.downloader.run()
From example_2_run_until_the_queue_is_drained.py
print(f'A run asked to stop before it starts: {self.downloader.run(stop_event=stop_event)}')
From example_3_logging_in_again_after_a_refused_session.py
visits, bars = downloader.run()
Source code in stock_brokers/instruments/historical/base.py 
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
def run(self, stop_event=None, deadline_seconds=None):
    """
    Work the queue until stopped, out of work, or out of time.

    Returns a `(visits, bars)` pair.

    - `stop_event` is a `threading.Event` that ends the loop when set.
    - `deadline_seconds` ends the loop after this long, for a job with a nightly window.
    """
    started = time.monotonic()
    visits = 0
    bars = 0

    while True:
        if stop_event is not None and stop_event.is_set():
            self._logger.info("Stopping: asked to.")
            break
        if deadline_seconds is not None and time.monotonic() - started > deadline_seconds:
            self._logger.info(f"Stopping: reached the {deadline_seconds}s deadline.")
            break

        series = self.claim()
        if series is None:
            self._logger.info("Nothing due. The queue is drained or backing off.")
            break

        try:
            stored = self.visit(series)
        except (CandleAuthenticationError, CandleBlocked) as exception:
            self._logger.error(f"Stopping: {type(exception).__name__}: {exception}")
            break
        except CandleThrottled as exception:
            self._logger.warning(f"Stopping: {exception}")
            break
        except Exception as exception:
            # Anything `visit` did not expect - a response shaped in a way no probe produced,
            # a write the database refuses - costs this one series and not the run. Before
            # this, a single window that failed on the way into Postgres took the process
            # down, and since the same window would be claimed again on restart it took it
            # down every ten minutes thereafter.
            self._connection.rollback()
            self._record_failure(series[0], series[1],
                                 f"{type(exception).__name__}: {str(exception)[:200]}")
            self._logger.exception(f"{series[0]} {series[1]}: unexpected failure, "
                                   f"carrying on with the next series.")
            stored = 0

        visits += 1
        bars += stored
        if visits % 200 == 0:
            self._logger.info(f"{visits} windows, {bars} bars so far.")

    self._logger.info(f"Finished: {visits} windows, {bars} bars.")
    return visits, bars

summary 

summary()

How far the download has got, for a status display.

Example usage

From example_1_one_window_of_daily_bars.py
print(f'Summary: {self.downloader.summary()}')
From example_2_run_until_the_queue_is_drained.py
print(f'Summary: {self.downloader.summary()}')
Source code in stock_brokers/instruments/historical/base.py 
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
def summary(self):
    """
    How far the download has got, for a status display.
    """
    with self._connection.cursor() as cursor:
        cursor.execute(f"""
            select count(*),
                   count(*) filter (where reached_broker_limit),
                   count(*) filter (where bar_count > 0),
                   coalesce(sum(bar_count), 0),
                   count(*) filter (where consecutive_failures >= %s)
            from {self._progress_table}
        """, (MAXIMUM_CONSECUTIVE_FAILURES,))
        total, retired, started, stored, stuck = cursor.fetchone()
    self._connection.commit()
    return {"series": total, "retired": retired, "with_bars": started,
            "bars": int(stored), "stuck": stuck}

close 

close()

Close the database connection.

Example usage

From example_1_one_window_of_daily_bars.py
self.downloader.close()
From example_2_run_until_the_queue_is_drained.py
self.downloader.close()
From example_3_logging_in_again_after_a_refused_session.py
downloader.close()
Source code in stock_brokers/instruments/historical/base.py 
982
983
984
985
986
def close(self):
    """
    Close the database connection.
    """
    self._connection.close()