indmoney

IND Money instrument master ingestion.

IND Money is the only broker here whose master is not public. It serves three CSV files behind an Authorization header carrying an access token, which lasts twenty-four hours.

The token is read from the MongoDB last_login collection, where the broker login records it, so this module keeps no second copy of the token in the environment. A token issued on an earlier day is rejected rather than sent, because IND Money would only answer it with an authentication error that says nothing about the cause. When no usable token is stored, the ingester logs in itself through ensure_session rather than failing: the daily instrument download runs before the morning session refresh, so depending on that ordering would mean failing every weekday.

Classes:

Name	Description
IndMoneyInstruments	Downloads IND Money's daily instrument master.

Functions:

Name	Description
stored_access_token	Read today's IND Money access token from the MongoDB last_login collection.

IndMoneyInstruments

IndMoneyInstruments(access_token=None)

Bases: BrokerInstruments

Downloads IND Money's daily instrument master.

Attributes:

Name	Type	Description
BROKER_NAME	str	Always "indmoney".
DEDUPE_KEY_COLUMNS	list[str]	Exchange, segment and security identifier together.
DEDUPE_SORT_COLUMN	str | None	Series, so the kept row is predictable.

Build the IND Money ingester.

Parameters:

Name	Type	Description	Default
access_token	str | None	Token to authenticate with. When omitted, today's token is read from the MongoDB last_login collection.	None

Example programs

Downloading with a stored tokenLogging in when no token is storedWhen no token can be had

examples/stock_brokers/instruments/indmoney/IndMoneyInstruments/example_1_downloading_with_a_stored_token.py

"""Downloads IND Money's three instrument master files with an access token, served from memory instead of the network.

IND Money is the one broker here whose master is not public. `IndMoneyInstruments.download` asks the same endpoint three times, once for each `source` (`equity`, `fno` and `index`), with the access token in the `Authorization` header, reads each CSV answer as text and stacks them. The token is normally read from MongoDB when the ingester is built; passing `access_token` to the constructor, as this program does, uses that token instead and reads nothing.

The program must not download anything, so `requests.get` is replaced, only while `download` runs, by `InstrumentFileServer`, a stand-in that answers each source with a few recorded-looking lines and remembers what each request carried. Building the ingester creates a database engine, but the engine opens no connection until a query runs, and this program runs none.

Notice that all three requests carry the token given to the constructor, and that each source's rows follow the previous one's in the stacked frame.

Run it from the project root:

    python examples/stock_brokers/instruments/indmoney/IndMoneyInstruments/example_1_downloading_with_a_stored_token.py
"""

import unittest.mock

import requests

from stock_brokers.instruments.indmoney import (
    IndMoneyInstruments,
)

SOURCE_FILES = {
    'equity': (
        'exch,segment,security_id,instrument_name,trading_symbol,lot_units,tick_size,series,symbol_name,isin\n'
        'NSE,E,1594,EQUITY,INFY,1,0.1,EQ,INFOSYS LIMITED,INE009A01021\n'
        'BSE,E,500209,EQUITY,INFY,1,0.05,A,INFOSYS LTD,INE009A01021\n'
    ),
    'fno': (
        'exch,segment,security_id,instrument_name,trading_symbol,lot_units,tick_size,series,symbol_name,isin\n'
        'NSE,D,52175,FUTIDX,NIFTY-Oct2026-FUT,75,0.1,,NIFTY,\n'
    ),
    'index': (
        'exch,segment,security_id,instrument_name,trading_symbol,lot_units,tick_size,series,symbol_name,isin\n'
        'NSE,I,13,INDEX,NIFTY,1,0.05,,NIFTY 50,\n'
    ),
}


class InstrumentFileServer:
    """A stand-in for `requests.get` that answers each IND Money source with a canned CSV file.

    Attributes:
        files (dict): Each source mapped to the text of its CSV file.
        requests_made (list): The `source` parameter and `Authorization` header of every request, in order.
    """

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

        Args:
            files (dict): Each source mapped to the text of its CSV file.

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

    def get(self, url, params=None, headers=None, timeout=None):
        """Answers one source's request, in place of `requests.get`.

        Args:
            url (str): The instruments endpoint.
            params (dict | None): The query parameters, naming the source.
            headers (dict | None): The request headers, carrying the token.
            timeout (float | None): The timeout, which the stand-in ignores.

        Returns:
            requests.Response: A successful answer carrying the source's CSV file.
        """
        source = params['source']
        self.requests_made.append((source, headers['Authorization']))
        response = requests.Response()
        response.status_code = 200
        response.headers['Content-Type'] = 'text/csv; charset=utf-8'
        response.encoding = 'utf-8'
        response._content = self.files[source].encode('utf-8')
        response.url = url
        return response


class DownloadingWithAStoredTokenExample:
    """Downloads the master with a given token and prints the requests and the rows.

    Attributes:
        server (InstrumentFileServer): The stand-in for IND Money's endpoint.
        instruments (IndMoneyInstruments): The ingester being shown.
    """

    def __init__(self):
        """Builds the stand-in server and the ingester with a token.

        Returns:
            None: This method returns nothing.
        """
        self.server = InstrumentFileServer(SOURCE_FILES)
        self.instruments = IndMoneyInstruments(access_token='indmoney-token-issued-today')

    def run(self):
        """Downloads the master and prints what each request carried and the rows returned.

        Returns:
            None: This method returns nothing.
        """
        with unittest.mock.patch.object(requests, 'get', self.server.get):
            frame = self.instruments.download()
        for source, authorization in self.server.requests_made:
            print(f'Requested source={source} with Authorization: {authorization}')
        print(f'Rows: {len(frame)}')
        print(frame.to_string())


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

It prints:

Requested source=equity with Authorization: indmoney-token-issued-today
Requested source=fno with Authorization: indmoney-token-issued-today
Requested source=index with Authorization: indmoney-token-issued-today
Rows: 4
  exch segment security_id instrument_name     trading_symbol lot_units tick_size series      symbol_name          isin
0  NSE       E        1594          EQUITY               INFY         1       0.1     EQ  INFOSYS LIMITED  INE009A01021
1  BSE       E      500209          EQUITY               INFY         1      0.05      A      INFOSYS LTD  INE009A01021
2  NSE       D       52175          FUTIDX  NIFTY-Oct2026-FUT        75       0.1    NaN            NIFTY           NaN
3  NSE       I          13           INDEX              NIFTY         1      0.05    NaN         NIFTY 50           NaN

examples/stock_brokers/instruments/indmoney/IndMoneyInstruments/example_2_logging_in_when_no_token_is_stored.py

"""Logs IND Money in with `establish_session` when the only stored token is from an earlier day, then downloads with the new one.

IND Money's tokens last twenty-four hours, so when the ingester is built without a token it reads the stored login from MongoDB and keeps the token only if it was issued today. The daily download runs before the morning login refresh, so on most weekdays the stored token is yesterday's and nothing is kept. `establish_session` then logs IND Money in through `ensure_session`, which shares a Redis lock and a retry limit with every other process logging brokers in, and returns the token the login stored. `download` does this itself whenever it has no token; the program calls it directly to show its result.

Nothing here may reach MongoDB, Redis or IND Money, so for the length of the program the MongoDB and Redis clients of the session module and the instruments module are replaced by small stand-ins, the session module's `api_class_for` is replaced so that constructing "the IND Money API class" calls `StandInLogin.log_in`, which stores a new login dated today, and `requests.get` is replaced by a stand-in that answers each source with a canned CSV file.

Notice that the ingester starts with no token although one is stored, that one login produces the token, and that all three downloads carry it.

Run it from the project root:

    python examples/stock_brokers/instruments/indmoney/IndMoneyInstruments/example_2_logging_in_when_no_token_is_stored.py
"""

import datetime
import unittest.mock

import requests

from stock_brokers.api.utilities import session as session_module
from stock_brokers.instruments import indmoney as indmoney_instruments
from stock_brokers.instruments.indmoney import (
    IndMoneyInstruments,
)

CSV_FILE = (
    'exch,segment,security_id,instrument_name,trading_symbol,lot_units,tick_size,series,symbol_name,isin\n'
    'NSE,E,1594,EQUITY,INFY,1,0.1,EQ,INFOSYS LIMITED,INE009A01021\n'
)


class LoginStore:
    """A stand-in for MongoDB and its `last_login` collection, holding one login per broker.

    Attributes:
        logins (dict): Each broker name mapped to its stored login document.
    """

    def __init__(self, logins):
        """Holds the stored logins.

        Args:
            logins (dict): Each broker name mapped to its stored login document.

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

    def __getitem__(self, name):
        """Returns the store itself as the `last_login` collection, as `database['last_login']` does.

        Args:
            name (str): The collection name, always `last_login` here.

        Returns:
            LoginStore: This store.
        """
        return self

    def find_one(self, query, projection=None):
        """Finds the stored login of the broker the query names.

        Args:
            query (dict): The fields to match, here `broker_name`.
            projection (dict | None): The fields to return, which the stand-in ignores.

        Returns:
            dict | None: A copy of the stored login, or None when there is none.
        """
        login = self.logins.get(query['broker_name'])
        if login is None:
            return None
        return dict(login)


class StandInRedis:
    """A dictionary-backed stand-in for the Redis commands `ensure_session` uses.

    Attributes:
        keys (dict): Each key mapped to its value.
    """

    def __init__(self):
        """Starts with nothing stored.

        Returns:
            None: This method returns nothing.
        """
        self.keys = {}

    def get(self, name):
        """Reads a key.

        Args:
            name (str): The key.

        Returns:
            str | None: The value, or None when the key is absent.
        """
        return self.keys.get(name)

    def set(self, name, value, nx=False, ex=None):
        """Sets a key, only when it is absent if `nx` is given.

        Args:
            name (str): The key.
            value (str): The value to store.
            nx (bool): True to set the key only when it does not exist.
            ex (int | None): The expiry in seconds, which the stand-in ignores.

        Returns:
            bool: True when the key was set.
        """
        if nx and name in self.keys:
            return False
        self.keys[name] = value
        return True

    def delete(self, name):
        """Removes a key.

        Args:
            name (str): The key.

        Returns:
            int: 1 when the key existed, otherwise 0.
        """
        if name in self.keys:
            del self.keys[name]
            return 1
        return 0


class StandInLogin:
    """A stand-in for constructing `INDMoneyAPI`, whose construction is the login.

    Attributes:
        store (LoginStore): Where the login is stored, as the real login writes MongoDB.
        logins_made (int): How many logins have been made.
    """

    def __init__(self, store):
        """Holds the store the login writes to.

        Args:
            store (LoginStore): Where the login is stored.

        Returns:
            None: This method returns nothing.
        """
        self.store = store
        self.logins_made = 0

    def log_in(self):
        """Stores a new IND Money login dated now, as a successful real login does.

        Returns:
            None: This method returns nothing.
        """
        self.logins_made = self.logins_made + 1
        self.store.logins['indmoney'] = {
            'broker_name': 'indmoney',
            'access_token': 'indmoney-token-from-login',
            'last_login': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        }


class CsvServer:
    """A stand-in for `requests.get` that answers every source with the same canned CSV file.

    Attributes:
        authorizations (list): The `Authorization` header of every request, in order.
    """

    def __init__(self):
        """Starts with no requests seen.

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

    def get(self, url, params=None, headers=None, timeout=None):
        """Answers one source's request, in place of `requests.get`.

        Args:
            url (str): The instruments endpoint.
            params (dict | None): The query parameters, naming the source.
            headers (dict | None): The request headers, carrying the token.
            timeout (float | None): The timeout, which the stand-in ignores.

        Returns:
            requests.Response: A successful answer carrying the canned file.
        """
        self.authorizations.append(headers['Authorization'])
        response = requests.Response()
        response.status_code = 200
        response.headers['Content-Type'] = 'text/csv; charset=utf-8'
        response.encoding = 'utf-8'
        response._content = CSV_FILE.encode('utf-8')
        response.url = url
        return response


class LoggingInWhenNoTokenIsStoredExample:
    """Builds the ingester with only yesterday's token stored, logs in, and downloads.

    Attributes:
        store (LoginStore): The stand-in for MongoDB, holding yesterday's login.
        login (StandInLogin): The stand-in for the IND Money login.
        server (CsvServer): The stand-in for IND Money's endpoint.
    """

    def __init__(self):
        """Builds the stand-ins.

        Returns:
            None: This method returns nothing.
        """
        self.store = LoginStore({
            'indmoney': {
                'broker_name': 'indmoney',
                'access_token': 'indmoney-token-from-yesterday',
                'last_login': '2026-09-29 07:00:04',
            },
        })
        self.login = StandInLogin(self.store)
        self.server = CsvServer()

    def run(self):
        """Builds the ingester, establishes a session, downloads, and prints each step's result.

        Returns:
            None: This method returns nothing.
        """
        with (
            unittest.mock.patch.object(indmoney_instruments, 'get_mongo_db', return_value=self.store),
            unittest.mock.patch.object(session_module, 'get_mongo_db', return_value=self.store),
            unittest.mock.patch.object(session_module, 'get_cache', return_value=StandInRedis()),
            unittest.mock.patch.object(session_module, 'api_class_for', return_value=self.login.log_in),
            unittest.mock.patch.object(requests, 'get', self.server.get),
        ):
            instruments = IndMoneyInstruments()
            print(f'Token kept at construction: {instruments.access_token}')
            instruments.access_token = instruments.establish_session()
            print(f'Token after establish_session: {instruments.access_token}')
            print(f'Logins made: {self.login.logins_made}')
            frame = instruments.download()
        print(f'Download requests carried: {self.server.authorizations}')
        print(f'Rows: {len(frame)}')


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

It prints:

Token kept at construction: None
Token after establish_session: indmoney-token-from-login
Logins made: 1
Download requests carried: ['indmoney-token-from-login', 'indmoney-token-from-login', 'indmoney-token-from-login']
Rows: 3

examples/stock_brokers/instruments/indmoney/IndMoneyInstruments/example_3_when_no_token_can_be_had.py

"""Shows `IndMoneyInstruments.download` refusing to run when no token is stored and logging in fails.

When the ingester has no token from today, `download` calls `establish_session` to log IND Money in. If that login fails, `establish_session` prints why and returns None instead of raising, and `download` then raises a `ValueError` that says where to look, before sending any request, because IND Money would only answer an unauthenticated request with an error that says nothing about the cause.

Nothing here may reach MongoDB, Redis or IND Money, so for the length of the program the MongoDB and Redis clients of the session module and the instruments module are replaced by small stand-ins, and the session module's `api_class_for` is replaced so that constructing "the IND Money API class" calls `RefusedLogin.log_in`, which fails the way a rejected TOTP does. `requests.get` is replaced by a stand-in that counts requests, to show that none is sent.

Notice the line `establish_session` prints about the failed login, the `ValueError` that follows it, and that no download request was made.

Run it from the project root:

    python examples/stock_brokers/instruments/indmoney/IndMoneyInstruments/example_3_when_no_token_can_be_had.py
"""

import unittest.mock

import requests

from stock_brokers.api.indmoney import (
    INDMoneyAPIException,
)
from stock_brokers.api.utilities import session as session_module
from stock_brokers.instruments import indmoney as indmoney_instruments
from stock_brokers.instruments.indmoney import (
    IndMoneyInstruments,
)


class LoginStore:
    """A stand-in for MongoDB and its `last_login` collection, holding one login per broker.

    Attributes:
        logins (dict): Each broker name mapped to its stored login document.
    """

    def __init__(self, logins):
        """Holds the stored logins.

        Args:
            logins (dict): Each broker name mapped to its stored login document.

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

    def __getitem__(self, name):
        """Returns the store itself as the `last_login` collection, as `database['last_login']` does.

        Args:
            name (str): The collection name, always `last_login` here.

        Returns:
            LoginStore: This store.
        """
        return self

    def find_one(self, query, projection=None):
        """Finds the stored login of the broker the query names.

        Args:
            query (dict): The fields to match, here `broker_name`.
            projection (dict | None): The fields to return, which the stand-in ignores.

        Returns:
            dict | None: A copy of the stored login, or None when there is none.
        """
        login = self.logins.get(query['broker_name'])
        if login is None:
            return None
        return dict(login)


class StandInRedis:
    """A dictionary-backed stand-in for the Redis commands `ensure_session` uses.

    Attributes:
        keys (dict): Each key mapped to its value.
    """

    def __init__(self):
        """Starts with nothing stored.

        Returns:
            None: This method returns nothing.
        """
        self.keys = {}

    def get(self, name):
        """Reads a key.

        Args:
            name (str): The key.

        Returns:
            str | None: The value, or None when the key is absent.
        """
        return self.keys.get(name)

    def set(self, name, value, nx=False, ex=None):
        """Sets a key, only when it is absent if `nx` is given.

        Args:
            name (str): The key.
            value (str): The value to store.
            nx (bool): True to set the key only when it does not exist.
            ex (int | None): The expiry in seconds, which the stand-in ignores.

        Returns:
            bool: True when the key was set.
        """
        if nx and name in self.keys:
            return False
        self.keys[name] = value
        return True

    def delete(self, name):
        """Removes a key.

        Args:
            name (str): The key.

        Returns:
            int: 1 when the key existed, otherwise 0.
        """
        if name in self.keys:
            del self.keys[name]
            return 1
        return 0


class RefusedLogin:
    """A stand-in for constructing `INDMoneyAPI`, whose login IND Money refuses."""

    def log_in(self):
        """Fails the login as IND Money does for a TOTP it rejects.

        Returns:
            None: This method never returns.

        Raises:
            INDMoneyAPIException: Always.
        """
        raise INDMoneyAPIException(code=401, message='Cannot generate access token from INDstocks API: invalid TOTP')


class CountingServer:
    """A stand-in for `requests.get` that only counts the requests it is sent.

    Attributes:
        requests_made (int): How many requests were sent.
    """

    def __init__(self):
        """Starts with no requests seen.

        Returns:
            None: This method returns nothing.
        """
        self.requests_made = 0

    def get(self, url, params=None, headers=None, timeout=None):
        """Counts one request and refuses it, in place of `requests.get`.

        Args:
            url (str): The instruments endpoint.
            params (dict | None): The query parameters.
            headers (dict | None): The request headers.
            timeout (float | None): The timeout, which the stand-in ignores.

        Returns:
            requests.Response: Never returns.

        Raises:
            requests.ConnectionError: Always, since no request should be sent.
        """
        self.requests_made = self.requests_made + 1
        raise requests.ConnectionError('No request should have been sent.')


class WhenNoTokenCanBeHadExample:
    """Runs `download` with only yesterday's token stored and a login that fails.

    Attributes:
        store (LoginStore): The stand-in for MongoDB, holding yesterday's login.
        server (CountingServer): The stand-in for IND Money's endpoint.
    """

    def __init__(self):
        """Builds the stand-ins.

        Returns:
            None: This method returns nothing.
        """
        self.store = LoginStore({
            'indmoney': {
                'broker_name': 'indmoney',
                'access_token': 'indmoney-token-from-yesterday',
                'last_login': '2026-09-29 07:00:04',
            },
        })
        self.server = CountingServer()

    def run(self):
        """Builds the ingester, runs `download` and prints what it raised.

        Returns:
            None: This method returns nothing.
        """
        with (
            unittest.mock.patch.object(indmoney_instruments, 'get_mongo_db', return_value=self.store),
            unittest.mock.patch.object(session_module, 'get_mongo_db', return_value=self.store),
            unittest.mock.patch.object(session_module, 'get_cache', return_value=StandInRedis()),
            unittest.mock.patch.object(session_module, 'api_class_for', return_value=RefusedLogin().log_in),
            unittest.mock.patch.object(requests, 'get', self.server.get),
        ):
            instruments = IndMoneyInstruments()
            try:
                instruments.download()
            except ValueError as error:
                print(f'Raised ValueError: {error}')
        print(f'Download requests sent: {self.server.requests_made}')


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

It prints:

indmoney: could not establish a session for the instrument download: Could not establish a indmoney session: INDMoneyAPIException: (401, 'Cannot generate access token from INDstocks API: invalid TOTP')
Raised ValueError: No IND Money access token available, and logging in did not produce one. Try 'bin/indmoney/session/connect' and read the error it reports. The token lasts twenty-four hours, and one issued on an earlier day is not reused.
Download requests sent: 0

Methods:

Name	Description
establish_session	Log IND Money in and return the token that produces, or None.
download	Fetch IND Money's three instrument master CSV files.
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/indmoney.py

def __init__(self, access_token=None):
    """
    Build the IND Money ingester.

    Args:
        access_token (str | None): Token to authenticate with. When omitted, today's token is read from the MongoDB last_login collection.
    """
    super().__init__()
    self.access_token = access_token or stored_access_token()

establish_session

establish_session()

Log IND Money in and return the token that produces, or None.

Called only when no usable token is already stored. Every other broker publishes its instrument master as a public file, so this is the one ingester whose success depends on a session, and the daily download runs before the morning login refresh - without this it would fail every weekday on a stale token and wait for somebody to notice.

Goes through ensure_session rather than constructing the API class directly, so it shares the Redis lock and the rate limiter with any other process using ensure_session. IND Money issues one token per session, so a download logging in while another process does the same would otherwise leave one of them holding a replaced token.

Returns:

Type	Description
str | None	Today's token, or None when the login did not produce one.

Example usage

From example_2_logging_in_when_no_token_is_stored.py

instruments.access_token = instruments.establish_session()

Source code in stock_brokers/instruments/indmoney.py

def establish_session(self):
    """
    Log IND Money in and return the token that produces, or None.

    Called only when no usable token is already stored. Every other broker publishes its
    instrument master as a public file, so this is the one ingester whose success depends on
    a session, and the daily download runs before the morning login refresh - without this it
    would fail every weekday on a stale token and wait for somebody to notice.

    Goes through `ensure_session` rather than constructing the API class directly, so it
    shares the Redis lock and the rate limiter with any other process using `ensure_session`.
    IND Money issues one token per session, so a download logging in while another process does
    the same would otherwise leave one of them holding a replaced token.

    Returns:
        str | None: Today's token, or None when the login did not produce one.
    """
    from stock_brokers.api.utilities.session import SessionException, ensure_session

    try:
        ensure_session("indmoney")
    except SessionException as exception:
        print(f"indmoney: could not establish a session for the instrument download: "
              f"{exception}")
        return None
    return stored_access_token()

download

download()

Fetch IND Money's three instrument master CSV files.

Logs in first when no usable token is stored, since the endpoint rejects an unauthenticated request and a token issued on an earlier day is treated as absent.

Returns:

Type	Description
DataFrame	Every row IND Money published, read as text.

Raises:

Type	Description
ValueError	If no access token could be obtained.
HTTPError	If any of the three requests fails.

Example usage

From example_1_downloading_with_a_stored_token.py

frame = self.instruments.download()

From example_2_logging_in_when_no_token_is_stored.py

frame = instruments.download()

From example_3_when_no_token_can_be_had.py

instruments.download()

Source code in stock_brokers/instruments/indmoney.py

def download(self):
    """
    Fetch IND Money's three instrument master CSV files.

    Logs in first when no usable token is stored, since the endpoint rejects an
    unauthenticated request and a token issued on an earlier day is treated as absent.

    Returns:
        pandas.DataFrame: Every row IND Money published, read as text.

    Raises:
        ValueError: If no access token could be obtained.
        requests.HTTPError: If any of the three requests fails.
    """
    if not self.access_token:
        self.access_token = self.establish_session()

    if not self.access_token:
        raise ValueError(
            "No IND Money access token available, and logging in did not produce one. "
            "Try 'bin/indmoney/session/connect' and read the error it reports. The token lasts "
            "twenty-four hours, and one issued on an earlier day is not reused."
        )

    frames = []
    for source in SOURCES:
        response = requests.get(
            INSTRUMENTS_URL,
            params={"source": source},
            headers={
                "Content-Type": "application/json",
                "Authorization": self.access_token,
            },
            timeout=DOWNLOAD_TIMEOUT_SECONDS,
        )
        response.raise_for_status()
        frames.append(pandas.read_csv(StringIO(response.text), dtype=str))
    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

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

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

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

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

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

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

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

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

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}

stored_access_token

stored_access_token()

Read today's IND Money access token from the MongoDB last_login collection.

The daily broker login job writes one document per broker there, holding the token it received and the moment it received it. A token from an earlier day is not returned: IND Money's tokens last twenty-four hours, and sending a stale one produces an authentication error that says nothing about why.

Returns:

Type	Description
str | None	The token issued today, or None when there is no document, no token in it, or the token was issued on an earlier date.

Source code in stock_brokers/instruments/indmoney.py

def stored_access_token():
    """
    Read today's IND Money access token from the MongoDB last_login collection.

    The daily broker login job writes one document per broker there, holding the token it received and the moment it received it. A token from an earlier day is not returned: IND Money's tokens last twenty-four hours, and sending a stale one produces an authentication error that says nothing about why.

    Returns:
        str | None: The token issued today, or None when there is no document, no token in it, or the token was issued on an earlier date.
    """
    document = get_mongo_db()["last_login"].find_one({"broker_name": "indmoney"})
    if document is None:
        return None
    last_login = document.get("last_login")
    if not last_login or last_login[:10] != datetime.now().strftime("%Y-%m-%d"):
        return None
    return document.get("access_token")