Skip to content

wisdom_capital

Wisdom Capital instrument master ingestion.

Wisdom Capital serves its master through a public POST endpoint, one call per exchange segment. The response is JSON whose 'result' field holds pipe-delimited text with no header line, and the number of fields depends on whether the row is an equity, an option or a future, so the three shapes carry their own column names.

The host serves a certificate issued to a different domain, so the session mounts a relaxed hostname adapter that still verifies the certificate's chain and dates.

Classes:

Name Description
HostnameCheckRelaxedAdapter

A requests transport adapter that verifies the server certificate but does not require its name to match the host being called.
WisdomCapitalInstruments

Downloads Wisdom Capital's daily instrument master.

Functions:

Name Description
build_session

Open a requests session that can reach Wisdom Capital's market data host.

HostnameCheckRelaxedAdapter

Bases: HTTPAdapter

A requests transport adapter that verifies the server certificate but does not require its name to match the host being called.

The certificate must still chain to a trusted authority and must still be in date. Only the name check is dropped, so a self signed or untrusted certificate is still refused.

Example programs

examples/stock_brokers/instruments/wisdom_capital/HostnameCheckRelaxedAdapter/example_1_a_pool_that_skips_the_name_check.py
"""Builds a `HostnameCheckRelaxedAdapter` and reads the TLS settings of the connection pool it makes.

Wisdom Capital's market data host serves a certificate issued to a different domain, so an ordinary `requests` session refuses it on the name check alone. `HostnameCheckRelaxedAdapter` is a `requests` transport adapter whose `init_poolmanager` builds the connection pool with a TLS context that still requires a certificate chaining to a trusted authority and still in date, but skips the check that the certificate's name matches the host.

`requests` calls `init_poolmanager` from the adapter's constructor. The program also calls it again with its own pool sizes, as `requests` does when an adapter is rebuilt, and prints the settings the pool manager was given each time. Nothing connects to anything.

Notice that certificate verification stays `CERT_REQUIRED` while `check_hostname` and `assert_hostname` are both off, and that the second call keeps those settings while changing the pool sizes.

Run it from the project root:

    python examples/stock_brokers/instruments/wisdom_capital/HostnameCheckRelaxedAdapter/example_1_a_pool_that_skips_the_name_check.py
"""

from stock_brokers.instruments.wisdom_capital import (
    HostnameCheckRelaxedAdapter,
)


class APoolThatSkipsTheNameCheckExample:
    """Builds the adapter, rebuilds its pool manager and prints the pool settings each time.

    Attributes:
        adapter (HostnameCheckRelaxedAdapter): The adapter being shown.
    """

    def __init__(self):
        """Builds the adapter, which builds its first pool manager.

        Returns:
            None: This method returns nothing.
        """
        self.adapter = HostnameCheckRelaxedAdapter()

    def show(self, heading):
        """Prints the settings the adapter's pool manager hands to every pool it makes.

        Args:
            heading (str): What the settings follow from.

        Returns:
            None: This method returns nothing.
        """
        pool_settings = self.adapter.poolmanager.connection_pool_kw
        context = pool_settings['ssl_context']
        print(heading)
        print(f'    certificate verification: {context.verify_mode.name}')
        print(f'    check_hostname: {context.check_hostname}')
        print(f'    assert_hostname: {pool_settings["assert_hostname"]}')
        print(f'    maxsize: {pool_settings["maxsize"]}, block: {pool_settings["block"]}')

    def run(self):
        """Prints the pool settings, rebuilds the pool manager with other sizes and prints them again.

        Returns:
            None: This method returns nothing.
        """
        self.show('After construction:')
        self.adapter.init_poolmanager(4, 8, block=True)
        self.show('After init_poolmanager(4, 8, block=True):')


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

It prints:

After construction:
    certificate verification: CERT_REQUIRED
    check_hostname: False
    assert_hostname: False
    maxsize: 10, block: False
After init_poolmanager(4, 8, block=True):
    certificate verification: CERT_REQUIRED
    check_hostname: False
    assert_hostname: False
    maxsize: 8, block: True
examples/stock_brokers/instruments/wisdom_capital/HostnameCheckRelaxedAdapter/example_2_mounted_on_a_session.py
"""Mounts a `HostnameCheckRelaxedAdapter` on a `requests` session and shows which requests it serves.

A `requests` session picks the transport adapter for each request by the longest URL prefix it has mounted. `build_session` in the Wisdom Capital instruments module mounts a `HostnameCheckRelaxedAdapter` for `https://`, so every HTTPS request the session sends goes through a pool whose TLS context skips the certificate's name check, while plain HTTP keeps the default adapter.

This program mounts the adapter the same way, after calling `init_poolmanager` to give it a pool of two connections, asks the session which adapter serves each of two URLs, and asks the adapter for the connection pool it would use for Wisdom Capital's instrument master. Making a pool opens no connection, so nothing reaches the network.

Notice that only the HTTPS URL gets the relaxed adapter, and that the pool for the market data host carries the relaxed settings.

Run it from the project root:

    python examples/stock_brokers/instruments/wisdom_capital/HostnameCheckRelaxedAdapter/example_2_mounted_on_a_session.py
"""

import requests

from stock_brokers.instruments.wisdom_capital import (
    INSTRUMENTS_URL,
    HostnameCheckRelaxedAdapter,
)


class MountedOnASessionExample:
    """Mounts the adapter on a session and prints which adapter and pool serve each URL.

    Attributes:
        adapter (HostnameCheckRelaxedAdapter): The adapter being shown.
        session (requests.Session): The session it is mounted on.
    """

    def __init__(self):
        """Builds the adapter with a pool of two connections and mounts it for HTTPS.

        Returns:
            None: This method returns nothing.
        """
        self.adapter = HostnameCheckRelaxedAdapter()
        self.adapter.init_poolmanager(1, 2)
        self.session = requests.Session()
        self.session.mount('https://', self.adapter)

    def run(self):
        """Prints the adapter serving each URL and the settings of the pool for the instrument master.

        Returns:
            None: This method returns nothing.
        """
        for url in [
            INSTRUMENTS_URL,
            'http://example.com/',
        ]:
            print(f'{url} -> {type(self.session.get_adapter(url)).__name__}')
        pool = self.adapter.poolmanager.connection_from_url(INSTRUMENTS_URL)
        print(f'Pool for {pool.host}:{pool.port}')
        print(f'    assert_hostname: {pool.assert_hostname}')
        print(f'    check_hostname: {pool.conn_kw["ssl_context"].check_hostname}')
        print(f'    certificate verification: {pool.conn_kw["ssl_context"].verify_mode.name}')
        print(f'    connections kept: {pool.pool.maxsize}')


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

It prints:

https://trade.wisdomcapital.in/apimarketdata/instruments/master -> HostnameCheckRelaxedAdapter
http://example.com/ -> HTTPAdapter
Pool for trade.wisdomcapital.in:443
    assert_hostname: False
    check_hostname: False
    certificate verification: CERT_REQUIRED
    connections kept: 2

Methods:

Name Description
init_poolmanager

Build the connection pool with a TLS context whose hostname check is switched off.

init_poolmanager 

init_poolmanager(connections, maxsize, block=False, **pool_arguments)

Build the connection pool with a TLS context whose hostname check is switched off.

Parameters:

Name Type Description Default
connections int

Number of connection pools to cache.

 required
maxsize int

Number of connections to keep in each pool.

 required
block bool

Whether to block when the pool is full.

 False
**pool_arguments Any

Further arguments passed on to the pool manager.

 {}

Returns:

Type Description
None

The pool manager is stored on the adapter rather than returned.

Example usage

From example_1_a_pool_that_skips_the_name_check.py
self.adapter.init_poolmanager(4, 8, block=True)
From example_2_mounted_on_a_session.py
self.adapter.init_poolmanager(1, 2)
Source code in stock_brokers/instruments/wisdom_capital.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def init_poolmanager(self, connections, maxsize, block=False, **pool_arguments):
    """
    Build the connection pool with a TLS context whose hostname check is switched off.

    Args:
        connections (int): Number of connection pools to cache.
        maxsize (int): Number of connections to keep in each pool.
        block (bool): Whether to block when the pool is full.
        **pool_arguments (Any): Further arguments passed on to the pool manager.

    Returns:
        None: The pool manager is stored on the adapter rather than returned.
    """
    context = ssl.create_default_context()
    context.check_hostname = False
    pool_arguments["ssl_context"] = context
    pool_arguments["assert_hostname"] = False
    return super().init_poolmanager(connections, maxsize, block, **pool_arguments)

WisdomCapitalInstruments 

WisdomCapitalInstruments()

Bases: BrokerInstruments

Downloads Wisdom Capital's daily instrument master.

Attributes:

Name Type Description
BROKER_NAME str

Always "wisdom_capital".
DEDUPE_KEY_COLUMNS list[str]

Exchange segment and exchange instrument identifier together.
DEDUPE_SORT_COLUMN str | None

Series, so the kept row is predictable.

Example programs

examples/stock_brokers/instruments/wisdom_capital/WisdomCapitalInstruments/example_1_downloading_every_segment.py
"""Downloads Wisdom Capital's instrument master for every exchange segment, from canned answers instead of the network.

Wisdom Capital serves its master through a POST endpoint, one call per exchange segment. The answer is JSON whose `result` field holds pipe-delimited lines with no header, and a line has 22, 23 or 21 fields depending on whether it is an equity, an option or a future. `WisdomCapitalInstruments.download` sorts the lines into those three shapes by the instrument type in the third field, names each shape's columns, and stacks them with a `data_category` column saying which shape each row was read as. A segment answered with an empty `result` is reported and skipped, because the broker does that for a segment it is not publishing that day.

The program must not download anything, so the module's `build_session` is replaced, only while `download` runs, by one that returns `RecordedMasterSession`, a stand-in whose `post` answers each segment with a few recorded-looking lines and remembers which segments were asked for. Building the ingester creates a database engine, but the engine opens no connection until a query runs, and this program runs none.

Notice the segments reported as empty, and that the future, the option and the equity each come back with their own shape's columns filled in.

Run it from the project root:

    python examples/stock_brokers/instruments/wisdom_capital/WisdomCapitalInstruments/example_1_downloading_every_segment.py
"""

import json
import unittest.mock

import requests

from stock_brokers.instruments import wisdom_capital as wisdom_capital_instruments
from stock_brokers.instruments.wisdom_capital import (
    WisdomCapitalInstruments,
)

SEGMENT_ANSWERS = {
    'NSECM': (
        'NSECM|2885|8|RELIANCE|RELIANCE-EQ|EQ|RELIANCE-EQ|1100100002885|1545.3|1264.3|50001|0.1|1|1|RELIANCE|INE002A01018|1|1|RELIANCE INDUSTRIES LTD|0|-1|-1\n'
        'NSECM|1594|8|INFY|INFY-EQ|EQ|INFY-EQ|1100100001594|1650.2|1350.2|50001|0.1|1|1|INFY|INE009A01021|1|1|INFOSYS LIMITED|0|-1|-1\n'
    ),
    'NSEFO': (
        'NSEFO|35001|1|NIFTY|NIFTY26OCTFUT|FUTIDX|FUTIDX-NIFTY|2600000035001|27500|22500|1801|0.1|75|1|-1|NIFTY|2026-10-27T14:30:00|NIFTY 27OCT FUT|1|1|NIFTY 26 OCT FUT\n'
        'NSEFO|40001|2|NIFTY|NIFTY26OCT25000CE|OPTIDX|OPTIDX-NIFTY|2600000040001|1200|0.05|1801|0.05|75|1|-1|NIFTY|2026-10-27T14:30:00|25000|3|NIFTY 27OCT 25000 CE|1|1|NIFTY 26 OCT 25000 CE\n'
    ),
    'MCXFO': (
        'MCXFO|451669|1|CRUDEOIL|CRUDEOIL26OCTFUT|FUTCOM|FUTCOM-CRUDEOIL|5100000451669|7200|5400|10000|1|100|1|-1|CRUDEOIL|2026-10-19T23:30:00|CRUDEOIL 19OCT FUT|1|1|CRUDEOIL 26 OCT FUT\n'
    ),
}


class RecordedMasterSession:
    """A stand-in for the `requests` session `download` posts through, answering each segment with canned lines.

    Attributes:
        answers (dict): Each exchange segment mapped to the pipe-delimited text of its master.
        segments_asked (list): Every segment requested, in order.
    """

    def __init__(self, answers):
        """Holds the canned answers.

        Args:
            answers (dict): Each exchange segment mapped to the pipe-delimited text of its master.

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

    def post(self, url, headers=None, json=None, timeout=None):
        """Answers one segment's request as Wisdom Capital does, with the lines in the `result` field.

        Args:
            url (str): The master endpoint.
            headers (dict | None): The request headers.
            json (dict | None): The request body, naming the segment in `exchangeSegmentList`.
            timeout (float | None): The timeout, which the stand-in ignores.

        Returns:
            requests.Response: A successful answer, with an empty `result` for a segment that has no canned lines.
        """
        segment = json['exchangeSegmentList'][0]
        self.segments_asked.append(segment)
        body = {
            'type': 'success',
            'code': 's-instrument-0002',
            'description': 'Instrument master downloaded',
            'result': self.answers.get(segment, ''),
        }
        response = requests.Response()
        response.status_code = 200
        response.headers['Content-Type'] = 'application/json'
        response._content = self.encode(body)
        response.url = url
        return response

    def encode(self, body):
        """Encodes an answer body as JSON bytes.

        Args:
            body (dict): The answer body.

        Returns:
            bytes: The body as UTF-8 JSON.
        """
        return json.dumps(body).encode('utf-8')


class DownloadingEverySegmentExample:
    """Downloads the master from the canned answers and prints what came back.

    Attributes:
        session (RecordedMasterSession): The stand-in for Wisdom Capital's endpoint.
        instruments (WisdomCapitalInstruments): The ingester being shown.
    """

    def __init__(self):
        """Builds the stand-in session and the ingester.

        Returns:
            None: This method returns nothing.
        """
        self.session = RecordedMasterSession(SEGMENT_ANSWERS)
        self.instruments = WisdomCapitalInstruments()

    def run(self):
        """Downloads the master and prints the segments asked for and the rows returned.

        Returns:
            None: This method returns nothing.
        """
        with unittest.mock.patch.object(wisdom_capital_instruments, 'build_session', return_value=self.session):
            frame = self.instruments.download()
        print(f'Segments asked for: {self.session.segments_asked}')
        print(f'Rows: {len(frame)}')
        shown_columns = [
            'ExchangeSegment',
            'ExchangeInstrumentID',
            'InstrumentType',
            'Series',
            'LotSize',
            'StrikePrice',
            'ISIN',
            'data_category',
        ]
        print(frame[shown_columns].to_string())


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

It prints:

wisdom_capital: segment NSECD returned no instruments.
wisdom_capital: segment NSECO returned no instruments.
wisdom_capital: segment BSECM returned no instruments.
wisdom_capital: segment BSEFO returned no instruments.
wisdom_capital: segment BSECD returned no instruments.
wisdom_capital: segment NCDEX returned no instruments.
Segments asked for: ['NSECM', 'NSEFO', 'NSECD', 'NSECO', 'BSECM', 'BSEFO', 'BSECD', 'MCXFO', 'NCDEX']
Rows: 5
  ExchangeSegment ExchangeInstrumentID InstrumentType  Series LotSize StrikePrice          ISIN data_category
0           NSECM                 2885              8      EQ       1         NaN  INE002A01018      equities
1           NSECM                 1594              8      EQ       1         NaN  INE009A01021      equities
2           NSEFO                40001              2  OPTIDX      75       25000           NaN       options
3           NSEFO                35001              1  FUTIDX      75         NaN           NaN       futures
4           MCXFO               451669              1  FUTCOM     100         NaN           NaN       futures
examples/stock_brokers/instruments/wisdom_capital/WisdomCapitalInstruments/example_2_fitting_lines_to_their_shape.py
"""Fits split master lines to the number of fields their shape declares with `pad_or_join_fields`.

Wisdom Capital's master lines are split on the pipe character, and each of the three row shapes declares its own number of columns: 22 for an equity, 23 for an option and 21 for a future. `WisdomCapitalInstruments.pad_or_join_fields` forces one split line to that number. A short line is padded with empty strings, and a line with too many fields, which happens when a text field itself contains a pipe, has its surplus joined back into the last field so that the frame can still be built.

The program fits three equity lines: one of the right length, one missing its last two indicators, and one whose description contains a pipe. It pairs each result with the equity column names from the module and prints the columns that change. It needs no network and no database.

Notice the third line. The pipe was in `Description`, the fifth field, but the surplus is joined into the last field, `GSMIndicator`, so every value from the description onwards lands one column to the left: `Description` holds only the first half of the text and `Series` holds the second half. This looks like a bug and is reported as one; the output records what the code does today.

Run it from the project root:

    python examples/stock_brokers/instruments/wisdom_capital/WisdomCapitalInstruments/example_2_fitting_lines_to_their_shape.py
"""

from stock_brokers.instruments.wisdom_capital import (
    EQUITY_COLUMN_NAMES,
    WisdomCapitalInstruments,
)


class FittingLinesToTheirShapeExample:
    """Fits three equity lines to the equity shape and prints the columns of interest.

    Attributes:
        instruments (WisdomCapitalInstruments): The ingester being shown.
        lines (list): Pairs of (description, pipe-delimited equity line).
        shown_columns (list): The equity columns to print for each line.
    """

    def __init__(self):
        """Builds the ingester and the lines to fit.

        Returns:
            None: This method returns nothing.
        """
        self.instruments = WisdomCapitalInstruments()
        self.lines = [
            (
                'a line of the right length',
                'NSECM|2031|8|M&M|M&M-EQ|EQ|M&M-EQ|1100100002031|3500.5|2800.4|50001|0.1|1|1|M&M|INE101A01026|1|1|MAHINDRA & MAHINDRA LTD|0|-1|-1',
            ),
            (
                'a line missing its last two indicators',
                'NSECM|2031|8|M&M|M&M-EQ|EQ|M&M-EQ|1100100002031|3500.5|2800.4|50001|0.1|1|1|M&M|INE101A01026|1|1|MAHINDRA & MAHINDRA LTD|0',
            ),
            (
                'a line whose description contains a pipe',
                'NSECM|2031|8|M&M|MAHINDRA|MAHINDRA-EQ|EQ|M&M-EQ|1100100002031|3500.5|2800.4|50001|0.1|1|1|M&M|INE101A01026|1|1|MAHINDRA & MAHINDRA LTD|0|-1|-1',
            ),
        ]
        self.shown_columns = [
            'Description',
            'Series',
            'ISIN',
            'CautionIndicator',
            'GSMIndicator',
        ]

    def run(self):
        """Fits each line and prints its field counts and chosen columns.

        Returns:
            None: This method returns nothing.
        """
        expected_count = len(EQUITY_COLUMN_NAMES)
        print(f'Equity shape: {expected_count} fields')
        for description, line in self.lines:
            fields = line.split('|')
            fitted = self.instruments.pad_or_join_fields(fields, expected_count)
            row = dict(zip(EQUITY_COLUMN_NAMES, fitted))
            print(f'{description}: {len(fields)} fields in, {len(fitted)} out')
            for column in self.shown_columns:
                print(f'    {column}: {row[column]!r}')


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

It prints:

Equity shape: 22 fields
a line of the right length: 22 fields in, 22 out
    Description: 'M&M-EQ'
    Series: 'EQ'
    ISIN: 'INE101A01026'
    CautionIndicator: '-1'
    GSMIndicator: '-1'
a line missing its last two indicators: 20 fields in, 22 out
    Description: 'M&M-EQ'
    Series: 'EQ'
    ISIN: 'INE101A01026'
    CautionIndicator: ''
    GSMIndicator: ''
a line whose description contains a pipe: 23 fields in, 22 out
    Description: 'MAHINDRA'
    Series: 'MAHINDRA-EQ'
    ISIN: 'M&M'
    CautionIndicator: '0'
    GSMIndicator: '-1|-1'

Build the broker ingester and its database engine.

The engine is lazy and opens no connection until a query runs, so download() and the cleaning steps can be exercised without a database.

Example programs

examples/stock_brokers/instruments/base/BrokerInstruments/example_1_cleaning_a_master_file.py
"""Writes a small broker ingester on `BrokerInstruments` and runs its cleaning steps one at a time.

A broker's instrument ingester subclasses `BrokerInstruments`, names the broker in `BROKER_NAME`, says which columns identify one instrument in `DEDUPE_KEY_COLUMNS`, and implements `download`. The base class supplies the cleaning steps that `ingest` runs on every download, in this order: `normalize_columns` turns header names into lowercase SQL identifiers, `strip_whitespace` trims padded text, `drop_unnamed_columns` drops the empty column that a trailing comma on every line creates, `drop_garbage_rows` drops exchange test instruments such as `NSETEST`, and `dedupe` keeps one row per instrument.

`DemoInstruments` is a made-up broker whose `download` reads a five-line master file from memory instead of the network. The file carries every blemish the cleaning steps exist for. Building the ingester creates a database engine, but the engine opens no connection until a query runs, and this program runs none.

Notice what the output shows about `strip_whitespace` and `drop_garbage_rows`. Both only touch columns whose dtype is `object`, but under pandas 3, which this project pins, `read_csv(dtype=str)` produces string columns whose dtype is not `object`. So on the frame as downloaded both steps change nothing: the padded symbols stay padded, the `NSETEST` row stays, and `dedupe` then treats ` INFY ` and `INFY` as different instruments. This is reported as a bug and the output records what the code does today. The program then runs the same steps again on a copy converted to `object` columns, which shows what they are meant to do: every value trimmed, the test instrument dropped, and one NSE INFY row kept. There `dedupe` sorts on `series` before keeping the first row of each instrument, so the `BE` row survives whatever order the broker listed the rows in.

Run it from the project root:

    python examples/stock_brokers/instruments/base/BrokerInstruments/example_1_cleaning_a_master_file.py
"""

import io

import pandas

from stock_brokers.instruments.base import (
    BrokerInstruments,
)

MASTER_FILE = (
    ' Exchange ,Trading Symbol,Series,Lot Size,\n'
    'NSE, INFY ,EQ,1,\n'
    'NSE,INFY,BE,1,\n'
    'NSE,NSETEST,EQ,1,\n'
    'BSE,INFY,A,1,\n'
    'NSE,TCS ,EQ,1,\n'
)


class DemoInstruments(BrokerInstruments):
    """A made-up broker's instrument ingester whose master file is held in memory."""

    BROKER_NAME = 'demo'
    DEDUPE_KEY_COLUMNS = [
        'exchange',
        'trading_symbol',
    ]
    DEDUPE_SORT_COLUMN = 'series'

    def download(self):
        """Reads the master file, every value as text, as a real ingester reads the broker's file.

        Returns:
            pandas.DataFrame: Every row of the file.
        """
        return pandas.read_csv(io.StringIO(MASTER_FILE), dtype=str)


class CleaningAMasterFileExample:
    """Runs each cleaning step on the downloaded file and prints the result after each.

    Attributes:
        instruments (DemoInstruments): The ingester being shown.
    """

    def __init__(self):
        """Builds the ingester.

        Returns:
            None: This method returns nothing.
        """
        self.instruments = DemoInstruments()

    def show(self, step, frame):
        """Prints a frame under the name of the step that produced it.

        Args:
            step (str): The step's name.
            frame (pandas.DataFrame): The frame after that step.

        Returns:
            None: This method returns nothing.
        """
        print(f'After {step}:')
        print(frame.to_string())
        print()

    def run(self):
        """Downloads the file and cleans it step by step.

        Returns:
            None: This method returns nothing.
        """
        print(f'Table: {self.instruments.table}')
        frame = self.instruments.download()
        print(f'Downloaded columns: {list(frame.columns)}')
        frame = self.instruments.normalize_columns(frame)
        self.show('normalize_columns', frame)
        frame = self.instruments.strip_whitespace(frame)
        print(f'Trading symbols after strip_whitespace: {list(frame["trading_symbol"])}')
        print()
        frame = self.instruments.drop_unnamed_columns(frame)
        self.show('drop_unnamed_columns', frame)
        frame = self.instruments.drop_garbage_rows(frame)
        self.show('drop_garbage_rows', frame)
        frame = self.instruments.dedupe(frame)
        self.show('dedupe', frame)
        print('The same steps on a copy with object columns:')
        frame = self.instruments.normalize_columns(self.instruments.download()).astype(object)
        frame = self.instruments.strip_whitespace(frame)
        frame = self.instruments.drop_unnamed_columns(frame)
        frame = self.instruments.drop_garbage_rows(frame)
        frame = self.instruments.dedupe(frame)
        self.show('all five steps', frame)


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

It prints:

Table: demo.instruments
Downloaded columns: [' Exchange ', 'Trading Symbol', 'Series', 'Lot Size', 'Unnamed: 4']
After normalize_columns:
  exchange trading_symbol series lot_size unnamed_4
0      NSE          INFY      EQ        1       NaN
1      NSE           INFY     BE        1       NaN
2      NSE        NSETEST     EQ        1       NaN
3      BSE           INFY      A        1       NaN
4      NSE           TCS      EQ        1       NaN

Trading symbols after strip_whitespace: [' INFY ', 'INFY', 'NSETEST', 'INFY', 'TCS ']

demo: dropped empty artifact column 'unnamed_4'.
After drop_unnamed_columns:
  exchange trading_symbol series lot_size
0      NSE          INFY      EQ        1
1      NSE           INFY     BE        1
2      NSE        NSETEST     EQ        1
3      BSE           INFY      A        1
4      NSE           TCS      EQ        1

After drop_garbage_rows:
  exchange trading_symbol series lot_size
0      NSE          INFY      EQ        1
1      NSE           INFY     BE        1
2      NSE        NSETEST     EQ        1
3      BSE           INFY      A        1
4      NSE           TCS      EQ        1

After dedupe:
  exchange trading_symbol series lot_size
0      BSE           INFY      A        1
1      NSE           INFY     BE        1
2      NSE          INFY      EQ        1
3      NSE        NSETEST     EQ        1
4      NSE           TCS      EQ        1

The same steps on a copy with object columns:
demo: dropped empty artifact column 'unnamed_4'.
demo: dropped 1 placeholder row(s) matching ['NSETEST'].
demo: dropped 1 duplicate row(s) on ['exchange', 'trading_symbol'].
After all five steps:
  exchange trading_symbol series lot_size
0      BSE           INFY      A        1
1      NSE           INFY     BE        1
2      NSE            TCS     EQ        1
examples/stock_brokers/instruments/base/BrokerInstruments/example_2_storing_daily_snapshots.py
"""Stores one day's instrument master with `ingest` and checks its row count against earlier days.

`ingest` is the whole daily job for one broker: it skips a date already stored, downloads the file, runs the cleaning steps, refuses a column the table does not have, and appends the rows to `<broker>.instruments` with the snapshot's `download_date`. `has_data_for` says whether a date is already stored, `table_columns` reads the table's columns from the database itself, and `check_row_count_deviation` compares a day's row count with the average of every earlier day, reporting but never raising, because a file that suddenly shrinks or grows is worth a look rather than a failed job.

The real engine points at TimescaleDB. This program replaces it with an in-memory SQLite database, `InMemoryInstrumentDatabase`, arranged to answer the same SQL: the broker's schema is an attached database named `demo`, a second attached database named `information_schema` holds a `columns` table describing `demo.instruments`, and a small SQL function stands in for PostgreSQL's `to_regclass`. The table already holds two earlier snapshots, of four and five rows. `DemoInstruments` is a made-up broker whose master file is held in memory, so nothing is downloaded.

Notice that the second `ingest` of the same date is skipped, that `bootstrap=True` replaces the stored rows instead, and that three rows against an average of four and a half is an alarm at the default ten percent but not at fifty.

Run it from the project root:

    python examples/stock_brokers/instruments/base/BrokerInstruments/example_2_storing_daily_snapshots.py
"""

import datetime
import io
import sqlite3

import pandas
import sqlalchemy
import sqlalchemy.pool

from stock_brokers.instruments.base import (
    BrokerInstruments,
)

MASTER_FILE = (
    'Exchange,Trading Symbol,Series,Lot Size\n'
    'NSE,INFY,EQ,1\n'
    'BSE,INFY,A,1\n'
    'NSE,TCS,EQ,1\n'
)

TABLE_COLUMNS = [
    'exchange',
    'trading_symbol',
    'series',
    'lot_size',
    'download_date',
]


class DemoInstruments(BrokerInstruments):
    """A made-up broker's instrument ingester whose master file is held in memory."""

    BROKER_NAME = 'demo'
    DEDUPE_KEY_COLUMNS = [
        'exchange',
        'trading_symbol',
    ]
    DEDUPE_SORT_COLUMN = 'series'

    def download(self):
        """Reads the master file, every value as text, as a real ingester reads the broker's file.

        Returns:
            pandas.DataFrame: Every row of the file.
        """
        return pandas.read_csv(io.StringIO(MASTER_FILE), dtype=str)


class InMemoryInstrumentDatabase:
    """An in-memory SQLite database arranged to answer the PostgreSQL queries `BrokerInstruments` sends.

    Attributes:
        engine (sqlalchemy.engine.Engine): The engine to hand to the ingester.
        connection (sqlite3.Connection | None): The one SQLite connection behind the engine, once it is open.
    """

    def __init__(self):
        """Builds the engine, the attached schemas, the instrument table and two earlier snapshots.

        Returns:
            None: This method returns nothing.
        """
        sqlite3.register_adapter(datetime.date, self.adapt_date)
        sqlite3.register_converter('DATE', self.convert_date)
        self.connection = None
        self.engine = sqlalchemy.create_engine(
            'sqlite://',
            poolclass=sqlalchemy.pool.StaticPool,
            connect_args={
                'detect_types': sqlite3.PARSE_DECLTYPES,
            },
        )
        sqlalchemy.event.listen(self.engine, 'connect', self.prepare_connection)
        with self.engine.begin() as connection:
            connection.execute(sqlalchemy.text('create table information_schema.columns (table_schema text, table_name text, column_name text, ordinal_position integer)'))
            position = 1
            for column_name in TABLE_COLUMNS:
                connection.execute(
                    sqlalchemy.text('insert into information_schema.columns values (:schema, :table, :column, :position)'),
                    {
                        'schema': 'demo',
                        'table': 'instruments',
                        'column': column_name,
                        'position': position,
                    },
                )
                position = position + 1
            connection.execute(sqlalchemy.text('create table demo.instruments (exchange text, trading_symbol text, series text, lot_size text, download_date DATE)'))
            self.insert_snapshot(connection, datetime.date(2026, 9, 28), 4)
            self.insert_snapshot(connection, datetime.date(2026, 9, 29), 5)

    def adapt_date(self, value):
        """Writes a date as ISO text, the way SQLite stores it.

        Args:
            value (datetime.date): The date to store.

        Returns:
            str: The date as `YYYY-MM-DD`.
        """
        return value.isoformat()

    def convert_date(self, value):
        """Reads a stored date back as a `datetime.date`, as PostgreSQL returns it.

        Args:
            value (bytes): The stored text.

        Returns:
            datetime.date: The date.
        """
        return datetime.date.fromisoformat(value.decode('ascii'))

    def prepare_connection(self, connection, connection_record):
        """Attaches the two schemas and adds the `to_regclass` function when SQLite opens its connection.

        Args:
            connection (sqlite3.Connection): The new SQLite connection.
            connection_record (Any): SQLAlchemy's record of the connection, unused here.

        Returns:
            None: This method returns nothing.
        """
        self.connection = connection
        connection.execute("attach database ':memory:' as demo")
        connection.execute("attach database ':memory:' as information_schema")
        connection.create_function('to_regclass', 1, self.to_regclass)

    def to_regclass(self, qualified_name):
        """Stands in for PostgreSQL's `to_regclass`, naming the table when it exists.

        Args:
            qualified_name (str): The table name with its schema, such as `demo.instruments`.

        Returns:
            str | None: The name when the table exists, otherwise None.
        """
        schema, table = qualified_name.split('.')
        cursor = self.connection.execute(f'select count(*) from {schema}.sqlite_master where type = ? and name = ?', ('table', table))
        if cursor.fetchone()[0] == 0:
            return None
        return qualified_name

    def insert_snapshot(self, connection, download_date, row_count):
        """Stores an earlier day's snapshot of made-up rows.

        Args:
            connection (sqlalchemy.engine.Connection): An open connection.
            download_date (datetime.date): The snapshot's date.
            row_count (int): How many rows to store.

        Returns:
            None: This method returns nothing.
        """
        for index in range(row_count):
            connection.execute(
                sqlalchemy.text('insert into demo.instruments values (:exchange, :symbol, :series, :lot_size, :download_date)'),
                {
                    'exchange': 'NSE',
                    'symbol': f'SYMBOL{index}',
                    'series': 'EQ',
                    'lot_size': '1',
                    'download_date': download_date,
                },
            )


class StoringDailySnapshotsExample:
    """Ingests one date three ways and checks its row count.

    Attributes:
        instruments (DemoInstruments): The ingester being shown, with its engine replaced.
        download_date (datetime.date): The date being ingested.
    """

    def __init__(self):
        """Builds the ingester and points it at the in-memory database.

        Returns:
            None: This method returns nothing.
        """
        self.instruments = DemoInstruments()
        self.instruments.engine = InMemoryInstrumentDatabase().engine
        self.download_date = datetime.date(2026, 9, 30)

    def run(self):
        """Ingests the date, ingests it again, replaces it, and checks the row counts.

        Returns:
            None: This method returns nothing.
        """
        print(f'Columns of {self.instruments.table}: {self.instruments.table_columns()}')
        print(f'Stored for {self.download_date} before: {self.instruments.has_data_for(self.download_date)}')
        stored = self.instruments.ingest(download_date=self.download_date)
        print(f'First ingest returned {stored}')
        print(f'Stored for {self.download_date} after: {self.instruments.has_data_for(self.download_date)}')
        stored = self.instruments.ingest(download_date=self.download_date)
        print(f'Second ingest returned {stored}')
        stored = self.instruments.ingest(download_date=self.download_date, bootstrap=True)
        print(f'Bootstrap ingest returned {stored}')
        print()
        report = self.instruments.check_row_count_deviation(self.download_date)
        print(f'Alarm at 10%: {report["alarm"]}')
        report = self.instruments.check_row_count_deviation(self.download_date, threshold=0.5)
        print(f'Alarm at 50%: {report["alarm"]}')
        report = self.instruments.check_row_count_deviation(datetime.date(2026, 10, 1))
        print(f'Alarm for a date with no rows: {report["alarm"]}')


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

It prints:

Columns of demo.instruments: ['exchange', 'trading_symbol', 'series', 'lot_size', 'download_date']
Stored for 2026-09-30 before: False
demo: downloading instruments for 2026-09-30 ...
demo: ingested 3 row(s) for 2026-09-30.
First ingest returned 3
Stored for 2026-09-30 after: True
demo: already ingested for 2026-09-30, skipping.
Second ingest returned 0
demo: downloading instruments for 2026-09-30 ...
demo: removed 3 existing row(s) for 2026-09-30 before re-ingesting.
demo: ingested 3 row(s) for 2026-09-30.
Bootstrap ingest returned 3

ALARM: demo: 3 row(s) for 2026-09-30, average 4 over 2 earlier day(s), deviation -33.33%, allowed range [4, 5].
Alarm at 10%: True
OK: demo: 3 row(s) for 2026-09-30, average 4 over 2 earlier day(s), deviation -33.33%, allowed range [2, 7].
Alarm at 50%: False
INFO: demo: no rows stored for 2026-10-01, nothing to check.
Alarm for a date with no rows: False
examples/stock_brokers/instruments/base/BrokerInstruments/example_3_what_a_subclass_must_supply.py
"""Shows the two things `BrokerInstruments` refuses to guess: the broker's name and how to download its file.

`BrokerInstruments` is a base class. Its constructor raises `NotImplementedError` when a subclass has not set `BROKER_NAME`, because the name is also the PostgreSQL schema the rows are written to, and its `download` raises `NotImplementedError` until a subclass overrides it. Both fail at once, before any file is fetched or any row is written.

The program defines one subclass that forgets the name and one that sets it but forgets `download`, and prints what each refusal says. Building the second ingester creates a database engine, but the engine opens no connection until a query runs, and this program runs none.

Notice that the second ingester is built without complaint and fails only when `download` is called, and that both messages name the subclass at fault.

Run it from the project root:

    python examples/stock_brokers/instruments/base/BrokerInstruments/example_3_what_a_subclass_must_supply.py
"""

from stock_brokers.instruments.base import (
    BrokerInstruments,
)


class NamelessInstruments(BrokerInstruments):
    """An ingester that forgot to set `BROKER_NAME`."""


class UndownloadableInstruments(BrokerInstruments):
    """An ingester that names its broker but forgot to implement `download`."""

    BROKER_NAME = 'demo'


class WhatASubclassMustSupplyExample:
    """Builds the two incomplete ingesters and prints each refusal."""

    def run(self):
        """Tries each incomplete ingester and prints what it raises.

        Returns:
            None: This method returns nothing.
        """
        try:
            NamelessInstruments()
        except NotImplementedError as error:
            print(f'Building NamelessInstruments: {error}')
        instruments = UndownloadableInstruments()
        print(f'Built UndownloadableInstruments for table {instruments.table}')
        try:
            instruments.download()
        except NotImplementedError as error:
            print(f'Calling download: {error}')


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

It prints:

Building NamelessInstruments: NamelessInstruments must set BROKER_NAME
Built UndownloadableInstruments for table demo.instruments
Calling download: UndownloadableInstruments must implement download()

Methods:

Name Description
pad_or_join_fields

Force one split line to the number of fields its shape expects.
download

Fetch Wisdom Capital's master for every exchange segment and stack the three row shapes.
normalize_columns

Rewrite column names into a stable lowercase form usable as a SQL identifier.
strip_whitespace

Remove leading and trailing whitespace from every text column.
drop_unnamed_columns

Drop the empty placeholder columns a trailing delimiter on every line creates.
drop_garbage_rows

Drop the placeholder instruments some brokers ship in their master files.
dedupe

Drop rows repeating this broker's natural key.
table_columns

The column names of this broker's table as it currently exists in the database.
has_data_for

Whether this broker's table already holds a snapshot for a given date.
ingest

Download, clean and store one day's instrument master for this broker.
check_row_count_deviation

Compare a day's row count against the average of every earlier day.
Source code in stock_brokers/instruments/base.py 
55
56
57
58
59
60
61
62
63
64
65
def __init__(self):
    """
    Build the broker ingester and its database engine.

    The engine is lazy and opens no connection until a query runs, so `download()` and the
    cleaning steps can be exercised without a database.
    """
    if not self.BROKER_NAME:
        raise NotImplementedError(f"{type(self).__name__} must set BROKER_NAME")
    self.engine = get_postgres_engine()
    self.table = instrument_table_name(self.BROKER_NAME)

pad_or_join_fields 

pad_or_join_fields(fields, expected_count)

Force one split line to the number of fields its shape expects.

A description containing a pipe character splits into too many fields, so the surplus is joined back into the last one. A short line is padded with empty strings.

Parameters:

Name Type Description Default
fields list[str]

The pipe-split values of one line.

 required
expected_count int

How many values that row shape declares.

 required

Returns:

Type Description
list[str]

Exactly expected_count values.

Example usage

From example_2_fitting_lines_to_their_shape.py
fitted = self.instruments.pad_or_join_fields(fields, expected_count)
Source code in stock_brokers/instruments/wisdom_capital.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def pad_or_join_fields(self, fields, expected_count):
    """
    Force one split line to the number of fields its shape expects.

    A description containing a pipe character splits into too many fields, so the surplus is joined back into the last one. A short line is padded with empty strings.

    Args:
        fields (list[str]): The pipe-split values of one line.
        expected_count (int): How many values that row shape declares.

    Returns:
        list[str]: Exactly expected_count values.
    """
    if len(fields) > expected_count:
        return fields[: expected_count - 1] + ["|".join(fields[expected_count - 1:])]
    return fields + [""] * (expected_count - len(fields))

download 

download()

Fetch Wisdom Capital's master for every exchange segment and stack the three row shapes.

Returns:

Type Description
DataFrame

Every row Wisdom Capital published, read as text, with a data_category column recording which shape each row was parsed as.

Raises:

Type Description
HTTPError

If any segment request fails.
ValueError

If no segment returned any instrument data.

Example usage

From example_1_downloading_every_segment.py
frame = self.instruments.download()
Source code in stock_brokers/instruments/wisdom_capital.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
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
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def download(self):
    """
    Fetch Wisdom Capital's master for every exchange segment and stack the three row shapes.

    Returns:
        pandas.DataFrame: Every row Wisdom Capital published, read as text, with a data_category column recording which shape each row was parsed as.

    Raises:
        requests.HTTPError: If any segment request fails.
        ValueError: If no segment returned any instrument data.
    """
    session = build_session()
    frames = []
    for exchange_segment in EXCHANGE_SEGMENTS:
        response = session.post(
            INSTRUMENTS_URL,
            headers={"Content-Type": "application/json"},
            json={"exchangeSegmentList": [exchange_segment]},
            timeout=DOWNLOAD_TIMEOUT_SECONDS,
        )
        response.raise_for_status()
        result_text = response.json().get("result", "")
        if not result_text:
            # The broker answers 'success' with an empty body for a segment it is not
            # publishing that day, so this is not an error - but a segment quietly going
            # empty is exactly what a shrinking row count later turns out to have been.
            print(f"{self.BROKER_NAME}: segment {exchange_segment} returned no instruments.")
            continue

        rows_by_category = {
            "equities": [],
            "options": [],
            "futures": [],
        }
        for line in result_text.splitlines():
            line = line.strip()
            if not line or line.lower().startswith("exchangesegment|"):
                continue
            fields = [field.strip() for field in line.split("|")]
            if len(fields) < 3:
                continue
            instrument_type = fields[2]
            series = fields[5] if len(fields) > 5 else ""
            if instrument_type == "8":
                rows_by_category["equities"].append(self.pad_or_join_fields(fields, len(EQUITY_COLUMN_NAMES)))
            elif instrument_type == "2" or str(series).upper().startswith("OPT"):
                rows_by_category["options"].append(self.pad_or_join_fields(fields, len(OPTION_COLUMN_NAMES)))
            else:
                rows_by_category["futures"].append(self.pad_or_join_fields(fields, len(FUTURE_COLUMN_NAMES)))

        for category, column_names in [
            ("equities", EQUITY_COLUMN_NAMES),
            ("options", OPTION_COLUMN_NAMES),
            ("futures", FUTURE_COLUMN_NAMES),
        ]:
            if not rows_by_category[category]:
                continue
            frame = pandas.DataFrame(rows_by_category[category], columns=column_names, dtype=str)
            frame["data_category"] = category
            frames.append(frame)

    if not frames:
        raise ValueError("Wisdom Capital returned no instrument data for any exchange segment.")
    return pandas.concat(frames, ignore_index=True)

normalize_columns 

normalize_columns(frame)

Rewrite column names into a stable lowercase form usable as a SQL identifier.

Some brokers bake stray whitespace and punctuation into the header itself rather than it being a parsing artifact, so the raw name is not safe to key on.

  • frame is the frame whose column names are to be normalized.

Example usage

From example_1_cleaning_a_master_file.py
frame = self.instruments.normalize_columns(frame)
From example_1_cleaning_a_master_file.py
frame = self.instruments.normalize_columns(self.instruments.download()).astype(object)
Source code in stock_brokers/instruments/base.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def normalize_columns(self, frame):
    """
    Rewrite column names into a stable lowercase form usable as a SQL identifier.

    Some brokers bake stray whitespace and punctuation into the header itself rather than it
    being a parsing artifact, so the raw name is not safe to key on.

    - `frame` is the frame whose column names are to be normalized.
    """
    def clean(name):
        name = re.sub(r"[^0-9a-zA-Z]+", "_", str(name).strip())
        return name.strip("_").lower()

    frame = frame.copy()
    frame.columns = [clean(column) for column in frame.columns]
    return frame

strip_whitespace 

strip_whitespace(frame)

Remove leading and trailing whitespace from every text column.

Brokers intermittently pad text fields, so the same value can arrive padded one day and bare the next.

  • frame is the frame to strip.

Example usage

From example_1_cleaning_a_master_file.py
frame = self.instruments.strip_whitespace(frame)
Source code in stock_brokers/instruments/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def strip_whitespace(self, frame):
    """
    Remove leading and trailing whitespace from every text column.

    Brokers intermittently pad text fields, so the same value can arrive padded one day and
    bare the next.

    - `frame` is the frame to strip.
    """
    frame = frame.copy()
    for column in frame.columns:
        if frame[column].dtype == object:
            frame[column] = frame[column].str.strip()
    return frame

drop_unnamed_columns 

drop_unnamed_columns(frame)

Drop the empty placeholder columns a trailing delimiter on every line creates.

A column is only dropped when it is empty on every row. One carrying real values is kept and reported instead, so nothing is discarded silently.

  • frame is the frame to examine.

Example usage

From example_1_cleaning_a_master_file.py
frame = self.instruments.drop_unnamed_columns(frame)
Source code in stock_brokers/instruments/base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def drop_unnamed_columns(self, frame):
    """
    Drop the empty placeholder columns a trailing delimiter on every line creates.

    A column is only dropped when it is empty on every row. One carrying real values is kept
    and reported instead, so nothing is discarded silently.

    - `frame` is the frame to examine.
    """
    frame = frame.copy()
    for column in [c for c in frame.columns if re.match(r"^unnamed(_\d+)?$", c)]:
        populated = frame[column].notna() & (frame[column].astype(str).str.strip() != "")
        if populated.any():
            print(f"{self.BROKER_NAME}: column '{column}' looks like a parsing artifact but holds "
                  f"{int(populated.sum())} value(s), keeping it.")
            continue
        frame = frame.drop(columns=[column])
        print(f"{self.BROKER_NAME}: dropped empty artifact column '{column}'.")
    return frame

drop_garbage_rows 

drop_garbage_rows(frame)

Drop the placeholder instruments some brokers ship in their master files.

  • frame is the frame to filter.

Example usage

From example_1_cleaning_a_master_file.py
frame = self.instruments.drop_garbage_rows(frame)
Source code in stock_brokers/instruments/base.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def drop_garbage_rows(self, frame):
    """
    Drop the placeholder instruments some brokers ship in their master files.

    - `frame` is the frame to filter.
    """
    if frame.empty:
        return frame
    matches = pandas.Series(False, index=frame.index)
    for column in frame.columns:
        if frame[column].dtype == object:
            for pattern in GARBAGE_SYMBOL_PATTERNS:
                matches = matches | frame[column].astype(str).str.contains(pattern, case=False, na=False)
    dropped = int(matches.sum())
    if dropped:
        print(f"{self.BROKER_NAME}: dropped {dropped} placeholder row(s) matching {GARBAGE_SYMBOL_PATTERNS}.")
    return frame[~matches].reset_index(drop=True)

dedupe 

dedupe(frame)

Drop rows repeating this broker's natural key.

  • frame is the frame to de-duplicate.

Example usage

From example_1_cleaning_a_master_file.py
frame = self.instruments.dedupe(frame)
Source code in stock_brokers/instruments/base.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def dedupe(self, frame):
    """
    Drop rows repeating this broker's natural key.

    - `frame` is the frame to de-duplicate.
    """
    if not self.DEDUPE_KEY_COLUMNS:
        return frame
    if self.DEDUPE_SORT_COLUMN and self.DEDUPE_SORT_COLUMN in frame.columns:
        frame = frame.sort_values(by=self.DEDUPE_SORT_COLUMN, kind="stable",
                                  key=lambda column: column.astype(str))
    before = len(frame)
    frame = frame.drop_duplicates(subset=self.DEDUPE_KEY_COLUMNS, keep="first").reset_index(drop=True)
    dropped = before - len(frame)
    if dropped:
        print(f"{self.BROKER_NAME}: dropped {dropped} duplicate row(s) on {self.DEDUPE_KEY_COLUMNS}.")
    return frame

table_columns 

table_columns()

The column names of this broker's table as it currently exists in the database.

The live table is the single source of truth for its own shape: the DDL lives in stock_brokers/instruments/sql/ddl and nothing in Python restates it.

Example usage

From example_2_storing_daily_snapshots.py
print(f'Columns of {self.instruments.table}: {self.instruments.table_columns()}')
Source code in stock_brokers/instruments/base.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def table_columns(self):
    """
    The column names of this broker's table as it currently exists in the database.

    The live table is the single source of truth for its own shape: the DDL lives in
    `stock_brokers/instruments/sql/ddl` and nothing in Python restates it.
    """
    with self.engine.connect() as connection:
        rows = connection.execute(
            text("select column_name from information_schema.columns "
                 "where table_schema = :schema and table_name = 'instruments' "
                 "order by ordinal_position"),
            {"schema": self.BROKER_NAME},
        ).all()
    if not rows:
        raise ValueError(f"Table {self.table} does not exist. Run 'python -m stock_brokers.instruments.sql.apply_ddl' first.")
    return [row[0] for row in rows]

has_data_for 

has_data_for(download_date)

Whether this broker's table already holds a snapshot for a given date.

  • download_date is the snapshot date to look for.

Example usage

From example_2_storing_daily_snapshots.py
print(f'Stored for {self.download_date} before: {self.instruments.has_data_for(self.download_date)}')
From example_2_storing_daily_snapshots.py
print(f'Stored for {self.download_date} after: {self.instruments.has_data_for(self.download_date)}')
Source code in stock_brokers/instruments/base.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def has_data_for(self, download_date):
    """
    Whether this broker's table already holds a snapshot for a given date.

    - `download_date` is the snapshot date to look for.
    """
    with self.engine.connect() as connection:
        exists = connection.execute(
            text("select to_regclass(:qualified_name) is not null"),
            {"qualified_name": self.table},
        ).scalar()
        if not exists:
            return False
        row = connection.execute(
            text(f"select 1 from {self.table} where download_date = :download_date limit 1"),
            {"download_date": download_date},
        ).first()
    return row is not None

ingest 

ingest(download_date=None, bootstrap=False)

Download, clean and store one day's instrument master for this broker.

Re-running for a date already stored is a no-op unless bootstrap is set, so the daily job is safe to run twice.

  • download_date is the snapshot date to record, defaulting to today.
  • bootstrap replaces that date's stored rows rather than skipping.

Example usage

From example_2_storing_daily_snapshots.py
stored = self.instruments.ingest(download_date=self.download_date)
From example_2_storing_daily_snapshots.py
stored = self.instruments.ingest(download_date=self.download_date, bootstrap=True)
Source code in stock_brokers/instruments/base.py 
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
233
234
235
236
237
238
239
240
241
242
243
def ingest(self, download_date=None, bootstrap=False):
    """
    Download, clean and store one day's instrument master for this broker.

    Re-running for a date already stored is a no-op unless `bootstrap` is set, so the daily
    job is safe to run twice.

    - `download_date` is the snapshot date to record, defaulting to today.
    - `bootstrap` replaces that date's stored rows rather than skipping.
    """
    download_date = download_date or pandas.Timestamp.today().date()
    already_stored = self.has_data_for(download_date)
    if already_stored and not bootstrap:
        print(f"{self.BROKER_NAME}: already ingested for {download_date}, skipping.")
        return 0

    print(f"{self.BROKER_NAME}: downloading instruments for {download_date} ...")
    frame = self.download()
    frame = self.normalize_columns(frame)
    frame = self.strip_whitespace(frame)
    frame = self.drop_unnamed_columns(frame)
    frame = self.drop_garbage_rows(frame)
    frame = self.dedupe(frame)
    frame["download_date"] = download_date

    unknown_columns = [column for column in frame.columns if column not in self.table_columns()]
    if unknown_columns:
        raise ValueError(
            f"{self.BROKER_NAME}: the downloaded file carries column(s) {unknown_columns} that "
            f"{self.table} does not have. Add them to the DDL rather than dropping them.")

    if already_stored:
        with self.engine.begin() as connection:
            deleted = connection.execute(
                text(f"delete from {self.table} where download_date = :download_date"),
                {"download_date": download_date},
            ).rowcount
        print(f"{self.BROKER_NAME}: removed {deleted} existing row(s) for {download_date} before re-ingesting.")

    frame.to_sql("instruments", self.engine, schema=self.BROKER_NAME,
                 if_exists="append", index=False, chunksize=10000)
    print(f"{self.BROKER_NAME}: ingested {len(frame)} row(s) for {download_date}.")
    return len(frame)

check_row_count_deviation 

check_row_count_deviation(download_date, threshold=0.1)

Compare a day's row count against the average of every earlier day.

A large swing usually means the broker's file came back truncated or unexpectedly bloated. This reports and never raises: a deviation is a signal to investigate, not a failure.

  • download_date is the snapshot date to check.
  • threshold is the allowed fractional deviation, so 0.10 permits ten percent either way.

Example usage

From example_2_storing_daily_snapshots.py
report = self.instruments.check_row_count_deviation(self.download_date)
From example_2_storing_daily_snapshots.py
report = self.instruments.check_row_count_deviation(self.download_date, threshold=0.5)
From example_2_storing_daily_snapshots.py
report = self.instruments.check_row_count_deviation(datetime.date(2026, 10, 1))
Source code in stock_brokers/instruments/base.py 
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
def check_row_count_deviation(self, download_date, threshold=0.10):
    """
    Compare a day's row count against the average of every earlier day.

    A large swing usually means the broker's file came back truncated or unexpectedly bloated.
    This reports and never raises: a deviation is a signal to investigate, not a failure.

    - `download_date` is the snapshot date to check.
    - `threshold` is the allowed fractional deviation, so 0.10 permits ten percent either way.
    """
    with self.engine.connect() as connection:
        rows = connection.execute(
            text(f"select download_date, count(*) from {self.table} group by download_date")).all()

    counts_by_date = {row[0]: row[1] for row in rows}
    current_rows = counts_by_date.get(download_date)
    if current_rows is None:
        message = f"INFO: {self.BROKER_NAME}: no rows stored for {download_date}, nothing to check."
        print(message)
        return {"broker": self.BROKER_NAME, "alarm": False, "message": message}

    earlier_counts = [count for date, count in counts_by_date.items() if date < download_date]
    if not earlier_counts:
        message = (f"INFO: {self.BROKER_NAME}: {current_rows} row(s) for {download_date}, "
                   f"no earlier day to compare against yet.")
        print(message)
        return {"broker": self.BROKER_NAME, "alarm": False, "message": message,
                "current_rows": current_rows}

    average_rows = sum(earlier_counts) / len(earlier_counts)
    lower_limit = average_rows * (1 - threshold)
    upper_limit = average_rows * (1 + threshold)
    is_alarm = current_rows < lower_limit or current_rows > upper_limit
    deviation_percent = (current_rows - average_rows) / average_rows * 100

    message = (
        f"{'ALARM' if is_alarm else 'OK'}: {self.BROKER_NAME}: {current_rows} row(s) for "
        f"{download_date}, average {average_rows:.0f} over {len(earlier_counts)} earlier day(s), "
        f"deviation {deviation_percent:.2f}%, allowed range [{lower_limit:.0f}, {upper_limit:.0f}].")
    print(message)
    return {"broker": self.BROKER_NAME, "alarm": is_alarm, "message": message,
            "current_rows": current_rows, "average_rows": average_rows,
            "earlier_days": len(earlier_counts), "deviation_percent": deviation_percent}

build_session 

build_session()

Open a requests session that can reach Wisdom Capital's market data host.

The host serves a certificate issued to a different domain, so ordinary verification fails on the name check alone.

Returns:

Type Description
Session

A session with the relaxed hostname adapter mounted.
Source code in stock_brokers/instruments/wisdom_capital.py 
135
136
137
138
139
140
141
142
143
144
145
146
def build_session():
    """
    Open a requests session that can reach Wisdom Capital's market data host.

    The host serves a certificate issued to a different domain, so ordinary verification fails on the name check alone.

    Returns:
        requests.Session: A session with the relaxed hostname adapter mounted.
    """
    session = requests.Session()
    session.mount("https://", HostnameCheckRelaxedAdapter())
    return session