Kotak instrument master ingestion.
Kotak publishes seven public CSV files under a URL path stamped with the day's date. Two of them sit under a different path prefix from the other five and carry one extra column.
Classes:
| Name | Description |
|---|---|
KotakInstruments |
Downloads Kotak's daily instrument master. |
Bases: BrokerInstruments
Downloads Kotak's daily instrument master.
Attributes:
| Name | Type | Description |
|---|---|---|
BROKER_NAME |
str
|
Always "kotak". |
DEDUPE_KEY_COLUMNS |
list[str]
|
Exchange segment and trading symbol together. |
DEDUPE_SORT_COLUMN |
str | None
|
Instrument type, so the kept row is predictable. |
Example programs
examples/stock_brokers/instruments/kotak/KotakInstruments/example_1_downloading_the_master.py
"""Downloads Kotak's instrument master with `KotakInstruments.download`, from canned files instead of the network.
Kotak publishes seven CSV files, five under a `transformed` folder and two under `transformed-v1`, all inside a folder named after today's date. `download` reads each file with `pandas.read_csv`, every value as text so that nothing is coerced on the way in, and returns one frame, stacking the files in the order listed in the module. `ingest` then cleans that frame and stores it in `kotak.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 every URL carries today's date, which is why the output shows only the last two parts of each, that the `dStrikePrice;` header of this canned file arrives with a stray semicolon, the kind of punctuation `normalize_columns` later removes, and that the `nse_com` file holds no rows that day.
Run it from the project root:
python examples/stock_brokers/instruments/kotak/KotakInstruments/example_1_downloading_the_master.py
"""
import io
import unittest.mock
import pandas
from stock_brokers.instruments.kotak import (
KotakInstruments,
)
PUBLISHED_FILES = {
'transformed/cde_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'1201,,cde_fo,FUTCUR,USDINR,USDINR26OCTFUT,XX,,25.000000,1,1477328400,-1.000000,CDS\n'
),
'transformed/mcx_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'451669,,mcx_fo,FUTCOM,CRUDEOIL,CRUDEOIL26OCTFUT,XX,,100.000000,1,1476921600,-1.000000,MCX\n'
),
'transformed/nse_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'52175,,nse_fo,FUTIDX,NIFTY,NIFTY26OCTFUT,XX,,10.000000,75,1477594800,-1.000000,NSE\n'
),
'transformed/bse_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'1135463,,bse_fo,FUTIDX,SENSEX,SENSEX26OCTFUT,XX,,5.000000,20,1477767600,-1.000000,BSE\n'
),
'transformed/nse_com.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
),
'transformed-v1/bse_cm-v1.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'500209,A,bse_cm,,INFY,INFY,,INE009A01021,5.000000,1,,,BSE\n'
),
'transformed-v1/nse_cm-v1.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'1594,EQ,nse_cm,,INFY,INFY-EQ,,INE009A01021,10.000000,1,,,NSE\n'
'11536,EQ,nse_cm,,TCS,TCS-EQ,,INE467B01029,10.000000,1,,,NSE\n'
),
}
class PublishedFiles:
"""A stand-in for `pandas.read_csv` that answers each of Kotak'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 Kotak's published files.
instruments (KotakInstruments): 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 = KotakInstruments()
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' ... {"/".join(source.split("/")[-2:])}')
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:
... transformed/cde_fo.csv
... transformed/mcx_fo.csv
... transformed/nse_fo.csv
... transformed/bse_fo.csv
... transformed/nse_com.csv
... transformed-v1/bse_cm-v1.csv
... transformed-v1/nse_cm-v1.csv
Rows: 7
Columns: ['pSymbol', 'pGroup', 'pExchSeg', 'pInstType', 'pSymbolName', 'pTrdSymbol', 'pOptionType', 'pISIN', 'dTickSize', 'lLotSize', 'lExpiryDate', 'dStrikePrice;', 'pExchange']
First six columns:
pSymbol pGroup pExchSeg pInstType pSymbolName pTrdSymbol
0 1201 NaN cde_fo FUTCUR USDINR USDINR26OCTFUT
1 451669 NaN mcx_fo FUTCOM CRUDEOIL CRUDEOIL26OCTFUT
2 52175 NaN nse_fo FUTIDX NIFTY NIFTY26OCTFUT
3 1135463 NaN bse_fo FUTIDX SENSEX SENSEX26OCTFUT
4 500209 A bse_cm NaN INFY INFY
5 1594 EQ nse_cm NaN INFY INFY-EQ
6 11536 EQ nse_cm NaN TCS TCS-EQ
examples/stock_brokers/instruments/kotak/KotakInstruments/example_2_keeping_one_row_per_instrument.py
"""Cleans Kotak'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`. `KotakInstruments` itself decides only what `dedupe` treats as one instrument, through `DEDUPE_KEY_COLUMNS`, and which row it keeps, through `DEDUPE_SORT_COLUMN`: the key is Kotak's exchange segment and trading symbol together, sorted on the instrument type first so that the row kept is predictable.
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/kotak/KotakInstruments/example_2_keeping_one_row_per_instrument.py
"""
import io
import unittest.mock
import pandas
from stock_brokers.instruments.kotak import (
KotakInstruments,
)
PUBLISHED_FILES = {
'transformed/cde_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'1201,,cde_fo,FUTCUR,USDINR,USDINR26OCTFUT,XX,,25.000000,1,1477328400,-1.000000,CDS\n'
),
'transformed/mcx_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'451669,,mcx_fo,FUTCOM,CRUDEOIL,CRUDEOIL26OCTFUT,XX,,100.000000,1,1476921600,-1.000000,MCX\n'
),
'transformed/nse_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'52175,,nse_fo,FUTIDX,NIFTY,NIFTY26OCTFUT,XX,,10.000000,75,1477594800,-1.000000,NSE\n'
),
'transformed/bse_fo.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'1135463,,bse_fo,FUTIDX,SENSEX,SENSEX26OCTFUT,XX,,5.000000,20,1477767600,-1.000000,BSE\n'
),
'transformed/nse_com.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
),
'transformed-v1/bse_cm-v1.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'500209,A,bse_cm,,INFY,INFY,,INE009A01021,5.000000,1,,,BSE\n'
),
'transformed-v1/nse_cm-v1.csv': (
'pSymbol,pGroup,pExchSeg,pInstType,pSymbolName,pTrdSymbol,pOptionType,pISIN,dTickSize,lLotSize,lExpiryDate,dStrikePrice;,pExchange\n'
'1594,EQ,nse_cm,,INFY,INFY-EQ,,INE009A01021,10.000000,1,,,NSE\n'
'11536,EQ,nse_cm,,TCS,TCS-EQ,,INE467B01029,10.000000,1,,,NSE\n'
'1594,EQ,nse_cm,,INFY,INFY-EQ,,INE009A01021,10.000000,1,,,NSE\n'
),
}
SHOWN_COLUMNS = [
'psymbol',
'pexchseg',
'ptrdsymbol',
'pinsttype',
'llotsize',
'dstrikeprice',
]
class PublishedFiles:
"""A stand-in for `pandas.read_csv` that answers each of Kotak'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 Kotak's published files.
instruments (KotakInstruments): 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 = KotakInstruments()
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: {KotakInstruments.DEDUPE_KEY_COLUMNS}, sorted first on: {KotakInstruments.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: 8
Key columns: ['pexchseg', 'ptrdsymbol'], sorted first on: pinsttype
Columns after normalize_columns: ['psymbol', 'pgroup', 'pexchseg', 'pinsttype', 'psymbolname', 'ptrdsymbol', 'poptiontype', 'pisin', 'dticksize', 'llotsize', 'lexpirydate', 'dstrikeprice', 'pexchange']
kotak: dropped 1 duplicate row(s) on ['pexchseg', 'ptrdsymbol'].
Rows to store: 7
psymbol pexchseg ptrdsymbol pinsttype llotsize dstrikeprice
0 451669 mcx_fo CRUDEOIL26OCTFUT FUTCOM 1 -1.000000
1 1201 cde_fo USDINR26OCTFUT FUTCUR 1 -1.000000
2 52175 nse_fo NIFTY26OCTFUT FUTIDX 75 -1.000000
3 1135463 bse_fo SENSEX26OCTFUT FUTIDX 20 -1.000000
4 500209 bse_cm INFY NaN 1 NaN
5 1594 nse_cm INFY-EQ NaN 1 NaN
6 11536 nse_cm TCS-EQ NaN 1 NaN
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:
Methods:
| Name | Description |
|---|---|
download |
Fetch and stack Kotak's seven instrument master CSV files for today. |
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
Fetch and stack Kotak's seven instrument master CSV files for today.
Kotak serves these files under today's date only, so a backfill of an earlier date is not possible from this source.
Returns:
| Type | Description |
|---|---|
DataFrame
|
Every row Kotak published, read as text. |
Example usage
Source code in stock_brokers/instruments/kotak.py
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.
frameis the frame whose column names are to be normalized.
Example usage
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
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.
frameis the frame to strip.
Example usage
Source code in stock_brokers/instruments/base.py
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.
frameis the frame to examine.
Example usage
Source code in stock_brokers/instruments/base.py
Drop the placeholder instruments some brokers ship in their master files.
frameis the frame to filter.
Example usage
Source code in stock_brokers/instruments/base.py
Drop rows repeating this broker's natural key.
frameis the frame to de-duplicate.
Example usage
Source code in stock_brokers/instruments/base.py
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
Whether this broker's table already holds a snapshot for a given date.
download_dateis 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
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_dateis the snapshot date to record, defaulting to today.bootstrapreplaces 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
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_dateis the snapshot date to check.thresholdis 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))