Skip to content

stoxkart

Stoxkart instrument master ingestion.

Stoxkart publishes one public CSV of roughly thirty-six megabytes covering every exchange it supports.

Classes:

Name Description
StoxkartInstruments

Downloads Stoxkart's daily instrument master.

StoxkartInstruments

StoxkartInstruments()

Bases: BrokerInstruments

Downloads Stoxkart's daily instrument master.

Attributes:

Name Type Description
BROKER_NAME str

Always "stoxkart".

DEDUPE_KEY_COLUMNS list[str]

Exchange and token together.

DEDUPE_SORT_COLUMN str | None

Series, so the kept row is predictable.

Example programs

examples/stock_brokers/instruments/stoxkart/StoxkartInstruments/example_1_downloading_the_master.py
"""Downloads Stoxkart's instrument master with `StoxkartInstruments.download`, from canned files instead of the network.

Stoxkart publishes one public CSV file. `download` reads it with `pandas.read_csv`, every value as text so that nothing is coerced on the way in, and returns one frame. `ingest` then cleans that frame and stores it in `stoxkart.instruments`.

The program must not download anything, so `pandas.read_csv` is replaced, only while `download` runs, by `PublishedFiles`, a stand-in that answers each URL with a few recorded-looking lines of the file served there and remembers which URLs 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 which files were read, and that the headers arrive exactly as Stoxkart wrote them, with capitals and spaces, and `ingest` normalizes them only afterwards.

Run it from the project root:

    python examples/stock_brokers/instruments/stoxkart/StoxkartInstruments/example_1_downloading_the_master.py
"""

import io
import unittest.mock

import pandas

from stock_brokers.instruments.stoxkart import (
    StoxkartInstruments,
)

PUBLISHED_FILES = {
    'https://openapi.stoxkart.com/scrip-master/csv': (
        'Exchange,Market Segment Id,Token,Symbol,Symbol Description,Series,Instrument Type,Option Type,Expiry Date,Lot Size,Strike Price,ISIN Code,Tick Size\n'
        'NSE,1,1594,INFY,INFOSYS LIMITED,EQ,,,,1,,INE009A01021,0.10\n'
        'NSE,1,11536,TCS,TATA CONSULTANCY SERV LT,EQ,,,,1,,INE467B01029,0.10\n'
        'BSE,3,500209,INFY,INFOSYS LTD,A,,,,1,,INE009A01021,0.05\n'
        'NFO,2,52175,NIFTY,NIFTY 27OCT2026 FUT,,FUTIDX,XX,27-Oct-2026,75,,,0.10\n'
    ),
}


class PublishedFiles:
    """A stand-in for `pandas.read_csv` that answers each of Stoxkart's URLs with a canned file.

    Attributes:
        files (dict): The end of each URL mapped to the text of the file served there.
        real_read_csv (Callable): The real `pandas.read_csv`, which parses the canned text.
        sources_read (list): Every URL asked for, in order.
    """

    def __init__(self, files):
        """Holds the canned files and keeps the real reader for parsing them.

        Args:
            files (dict): The end of each URL mapped to the text of the file served there.

        Returns:
            None: This method returns nothing.
        """
        self.files = files
        self.real_read_csv = pandas.read_csv
        self.sources_read = []

    def read_csv(self, source, **options):
        """Parses the canned file for a URL, in place of `pandas.read_csv`.

        Args:
            source (str): The URL `download` asked for.
            **options (Any): The parsing options `download` passed, such as `dtype`.

        Returns:
            pandas.DataFrame: The canned file, parsed with those options.

        Raises:
            KeyError: When no canned file matches the URL.
        """
        self.sources_read.append(source)
        for ending, text in self.files.items():
            if source.endswith(ending):
                return self.real_read_csv(io.StringIO(text), **options)
        raise KeyError(f'No canned file for {source}')


class DownloadingTheMasterExample:
    """Downloads the master from the canned files and prints what came back.

    Attributes:
        published_files (PublishedFiles): The stand-in for Stoxkart's published files.
        instruments (StoxkartInstruments): The ingester being shown.
    """

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

        Returns:
            None: This method returns nothing.
        """
        self.published_files = PublishedFiles(PUBLISHED_FILES)
        self.instruments = StoxkartInstruments()

    def run(self):
        """Downloads the master and prints the files read and the frame returned.

        Returns:
            None: This method returns nothing.
        """
        with unittest.mock.patch.object(pandas, 'read_csv', self.published_files.read_csv):
            frame = self.instruments.download()
        print('Files read:')
        for source in self.published_files.sources_read:
            print(f'    {source}')
        print(f'Rows: {len(frame)}')
        print(f'Columns: {list(frame.columns)}')
        print('First six columns:')
        print(frame.iloc[:, :6].to_string())


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

It prints:

Files read:
    https://openapi.stoxkart.com/scrip-master/csv
Rows: 4
Columns: ['Exchange', 'Market Segment Id', 'Token', 'Symbol', 'Symbol Description', 'Series', 'Instrument Type', 'Option Type', 'Expiry Date', 'Lot Size', 'Strike Price', 'ISIN Code', 'Tick Size']
First six columns:
  Exchange Market Segment Id   Token Symbol        Symbol Description Series
0      NSE                 1    1594   INFY           INFOSYS LIMITED     EQ
1      NSE                 1   11536    TCS  TATA CONSULTANCY SERV LT     EQ
2      BSE                 3  500209   INFY               INFOSYS LTD      A
3      NFO                 2   52175  NIFTY       NIFTY 27OCT2026 FUT    NaN
examples/stock_brokers/instruments/stoxkart/StoxkartInstruments/example_2_keeping_one_row_per_instrument.py
"""Cleans Stoxkart's downloaded master the way `ingest` does before storing it, keeping one row per instrument.

Before it writes anything, `ingest` runs five steps from `BrokerInstruments` on the downloaded frame: `normalize_columns`, `strip_whitespace`, `drop_unnamed_columns`, `drop_garbage_rows` and `dedupe`. `StoxkartInstruments` itself decides only what `dedupe` treats as one instrument, through `DEDUPE_KEY_COLUMNS`, and which row it keeps, through `DEDUPE_SORT_COLUMN`: a Stoxkart token is unique only within its exchange, so the key is exchange and token together, and rows are sorted on `series` first so that the row kept does not depend on the order Stoxkart listed them in.

The program must not download anything, so `pandas.read_csv` is replaced, only while `download` runs, by `PublishedFiles`, a stand-in that answers each URL with a few recorded-looking lines of the file served there. The canned file repeats one instrument, so `dedupe` has a row to drop. Building the ingester creates a database engine, but the engine opens no connection until a query runs, and this program runs none.

Notice the column names after `normalize_columns`, which are the names of the table's columns, and the line `dedupe` prints about the row it dropped.

Run it from the project root:

    python examples/stock_brokers/instruments/stoxkart/StoxkartInstruments/example_2_keeping_one_row_per_instrument.py
"""

import io
import unittest.mock

import pandas

from stock_brokers.instruments.stoxkart import (
    StoxkartInstruments,
)

PUBLISHED_FILES = {
    'https://openapi.stoxkart.com/scrip-master/csv': (
        'Exchange,Market Segment Id,Token,Symbol,Symbol Description,Series,Instrument Type,Option Type,Expiry Date,Lot Size,Strike Price,ISIN Code,Tick Size\n'
        'NSE,1,1594,INFY,INFOSYS LIMITED,EQ,,,,1,,INE009A01021,0.10\n'
        'NSE,1,11536,TCS,TATA CONSULTANCY SERV LT,EQ,,,,1,,INE467B01029,0.10\n'
        'BSE,3,500209,INFY,INFOSYS LTD,A,,,,1,,INE009A01021,0.05\n'
        'NFO,2,52175,NIFTY,NIFTY 27OCT2026 FUT,,FUTIDX,XX,27-Oct-2026,75,,,0.10\n'
        'NSE,1,1594,INFY,INFOSYS LIMITED,EQ,,,,1,,INE009A01021,0.10\n'
    ),
}

SHOWN_COLUMNS = [
    'exchange',
    'token',
    'symbol',
    'series',
    'instrument_type',
    'lot_size',
]


class PublishedFiles:
    """A stand-in for `pandas.read_csv` that answers each of Stoxkart's URLs with a canned file.

    Attributes:
        files (dict): The end of each URL mapped to the text of the file served there.
        real_read_csv (Callable): The real `pandas.read_csv`, which parses the canned text.
        sources_read (list): Every URL asked for, in order.
    """

    def __init__(self, files):
        """Holds the canned files and keeps the real reader for parsing them.

        Args:
            files (dict): The end of each URL mapped to the text of the file served there.

        Returns:
            None: This method returns nothing.
        """
        self.files = files
        self.real_read_csv = pandas.read_csv
        self.sources_read = []

    def read_csv(self, source, **options):
        """Parses the canned file for a URL, in place of `pandas.read_csv`.

        Args:
            source (str): The URL `download` asked for.
            **options (Any): The parsing options `download` passed, such as `dtype`.

        Returns:
            pandas.DataFrame: The canned file, parsed with those options.

        Raises:
            KeyError: When no canned file matches the URL.
        """
        self.sources_read.append(source)
        for ending, text in self.files.items():
            if source.endswith(ending):
                return self.real_read_csv(io.StringIO(text), **options)
        raise KeyError(f'No canned file for {source}')


class KeepingOneRowPerInstrumentExample:
    """Downloads the master from the canned files and runs the cleaning steps `ingest` runs.

    Attributes:
        published_files (PublishedFiles): The stand-in for Stoxkart's published files.
        instruments (StoxkartInstruments): The ingester being shown.
    """

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

        Returns:
            None: This method returns nothing.
        """
        self.published_files = PublishedFiles(PUBLISHED_FILES)
        self.instruments = StoxkartInstruments()

    def run(self):
        """Downloads the master, cleans it and prints the rows that would be stored.

        Returns:
            None: This method returns nothing.
        """
        with unittest.mock.patch.object(pandas, 'read_csv', self.published_files.read_csv):
            frame = self.instruments.download()
        print(f'Rows downloaded: {len(frame)}')
        print(f'Key columns: {StoxkartInstruments.DEDUPE_KEY_COLUMNS}, sorted first on: {StoxkartInstruments.DEDUPE_SORT_COLUMN}')
        frame = self.instruments.normalize_columns(frame)
        print(f'Columns after normalize_columns: {list(frame.columns)}')
        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)
        print(f'Rows to store: {len(frame)}')
        print(frame[SHOWN_COLUMNS].to_string())


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

It prints:

Rows downloaded: 5
Key columns: ['exchange', 'token'], sorted first on: series
Columns after normalize_columns: ['exchange', 'market_segment_id', 'token', 'symbol', 'symbol_description', 'series', 'instrument_type', 'option_type', 'expiry_date', 'lot_size', 'strike_price', 'isin_code', 'tick_size']
stoxkart: dropped 1 duplicate row(s) on ['exchange', 'token'].
Rows to store: 4
  exchange   token symbol series instrument_type lot_size
0      BSE  500209   INFY      A             NaN        1
1      NSE    1594   INFY     EQ             NaN        1
2      NSE   11536    TCS     EQ             NaN        1
3      NFO   52175  NIFTY    NaN          FUTIDX       75

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
download

Fetch Stoxkart's single instrument master CSV.

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
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)

download

download()

Fetch Stoxkart's single instrument master CSV.

Returns:

Type Description
DataFrame

Every row Stoxkart published, read as text.

Example usage

From example_1_downloading_the_master.py
frame = self.instruments.download()
From example_2_keeping_one_row_per_instrument.py
frame = self.instruments.download()
Source code in stock_brokers/instruments/stoxkart.py
def download(self):
    """
    Fetch Stoxkart's single instrument master CSV.

    Returns:
        pandas.DataFrame: Every row Stoxkart published, read as text.
    """
    return pandas.read_csv(INSTRUMENTS_URL, dtype=str)

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}