Shoonya instrument master ingestion.
Shoonya publishes seven public ZIP archives, one per exchange, each holding a single comma-separated text file. The per-exchange files do not all carry the same columns, so stacking them produces the union, and every line ends with a trailing comma.
Classes:
| Name | Description |
|---|---|
ShoonyaInstruments |
Downloads Shoonya's daily instrument master. |
Bases: BrokerInstruments
Downloads Shoonya's daily instrument master.
Attributes:
| Name | Type | Description |
|---|---|---|
BROKER_NAME |
str
|
Always "shoonya". |
DEDUPE_KEY_COLUMNS |
list[str]
|
Exchange and trading symbol together. |
DEDUPE_SORT_COLUMN |
str | None
|
Instrument, so the kept row is predictable. |
Example programs
examples/stock_brokers/instruments/shoonya/ShoonyaInstruments/example_1_downloading_the_zip_archives.py
"""Downloads Shoonya's instrument master from its seven ZIP archives, served from memory instead of the network.
Shoonya publishes one ZIP archive per exchange, each holding a single comma-separated text file. `ShoonyaInstruments.download` fetches every archive with `requests.get`, reads each `.txt` or `.csv` file inside as latin-1 text, adds `source_zip_url` and `source_file_name` columns saying where each row came from, and stacks the files. The exchanges' files have different columns, so a column missing from one file is a missing value in its rows.
The program must not download anything, so `requests.get` is replaced, only while `download` runs, by `ArchiveServer`, a stand-in that builds each archive in memory from a few recorded-looking lines 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 the two columns `download` adds, that the NCX archive holds no rows that day, and that the trailing comma on every line of Shoonya's files arrives as an empty column named after its position, `Unnamed: 7`, `Unnamed: 10` or `Unnamed: 11` depending on the file's width, all of which `ingest` later drops because they are empty on every row.
Run it from the project root:
python examples/stock_brokers/instruments/shoonya/ShoonyaInstruments/example_1_downloading_the_zip_archives.py
"""
import io
import unittest.mock
import zipfile
import requests
from stock_brokers.instruments.shoonya import (
ShoonyaInstruments,
)
ARCHIVE_CONTENTS = {
'NSE_symbols.txt.zip': (
'Exchange,Token,LotSize,Symbol,TradingSymbol,Instrument,TickSize,\n'
'NSE,1594,1,INFY,INFY-EQ,EQ,0.10,\n'
'NSE,11536,1,TCS,TCS-EQ,EQ,0.10,\n'
),
'NFO_symbols.txt.zip': (
'Exchange,Token,LotSize,Symbol,TradingSymbol,Expiry,Instrument,OptionType,StrikePrice,TickSize,\n'
'NFO,52175,75,NIFTY,NIFTY27OCT26F,27-OCT-2026,FUTIDX,XX,0,0.10,\n'
),
'CDS_symbols.txt.zip': (
'Exchange,Token,LotSize,Symbol,TradingSymbol,Expiry,Instrument,OptionType,StrikePrice,TickSize,\n'
'CDS,1201,1,USDINR,USDINR29OCT26F,29-OCT-2026,FUTCUR,XX,0,0.0025,\n'
),
'MCX_symbols.txt.zip': (
'Exchange,Token,LotSize,GNGD,Symbol,TradingSymbol,Expiry,Instrument,OptionType,StrikePrice,TickSize,\n'
'MCX,451669,100,1,CRUDEOIL,CRUDEOIL19OCT26,19-OCT-2026,FUTCOM,XX,0,1.00,\n'
),
'BSE_symbols.txt.zip': (
'Exchange,Token,LotSize,Symbol,TradingSymbol,Instrument,TickSize,\n'
'BSE,500209,1,INFY,INFY,A,0.05,\n'
),
'BFO_symbols.txt.zip': (
'Exchange,Token,LotSize,Symbol,TradingSymbol,Expiry,Instrument,OptionType,StrikePrice,TickSize,\n'
'BFO,1135463,20,SENSEX,SENSEX30OCT26F,30-OCT-2026,FUTIDX,XX,0,0.05,\n'
),
'NCX_symbols.txt.zip': (
'Exchange,Token,LotSize,Symbol,TradingSymbol,Expiry,Instrument,OptionType,StrikePrice,TickSize,\n'
),
}
class ArchiveServer:
"""A stand-in for `requests.get` that serves each of Shoonya's archives, built in memory from canned text.
Attributes:
contents (dict): Each archive's file name mapped to the text of the file inside it.
urls_asked (list): Every URL requested, in order.
"""
def __init__(self, contents):
"""Holds the canned file contents.
Args:
contents (dict): Each archive's file name mapped to the text of the file inside it.
Returns:
None: This method returns nothing.
"""
self.contents = contents
self.urls_asked = []
def archive(self, archive_name):
"""Builds one ZIP archive holding the canned text file for that archive.
Args:
archive_name (str): The archive's file name, such as `NSE_symbols.txt.zip`.
Returns:
bytes: The archive.
"""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, 'w') as archive:
archive.writestr(archive_name.removesuffix('.zip'), self.contents[archive_name])
return buffer.getvalue()
def get(self, url, timeout=None):
"""Serves one archive, in place of `requests.get`.
Args:
url (str): The archive's URL.
timeout (float | None): The timeout, which the stand-in ignores.
Returns:
requests.Response: A successful answer carrying the archive.
"""
self.urls_asked.append(url)
response = requests.Response()
response.status_code = 200
response.headers['Content-Type'] = 'application/zip'
response._content = self.archive(url.rsplit('/', 1)[1])
response.url = url
return response
class DownloadingTheZipArchivesExample:
"""Downloads the master from the in-memory archives and prints what came back.
Attributes:
server (ArchiveServer): The stand-in for Shoonya's file server.
instruments (ShoonyaInstruments): The ingester being shown.
"""
def __init__(self):
"""Builds the stand-in server and the ingester.
Returns:
None: This method returns nothing.
"""
self.server = ArchiveServer(ARCHIVE_CONTENTS)
self.instruments = ShoonyaInstruments()
def run(self):
"""Downloads the master and prints the archives asked for and the rows returned.
Returns:
None: This method returns nothing.
"""
with unittest.mock.patch.object(requests, 'get', self.server.get):
frame = self.instruments.download()
print('Archives asked for:')
for url in self.server.urls_asked:
print(f' {url}')
print(f'Rows: {len(frame)}')
print(f'Columns: {list(frame.columns)}')
shown_columns = [
'Exchange',
'Token',
'TradingSymbol',
'Instrument',
'source_file_name',
]
print(frame[shown_columns].to_string())
if __name__ == '__main__':
DownloadingTheZipArchivesExample().run()
It prints:
Archives asked for:
https://api.shoonya.com/NSE_symbols.txt.zip
https://api.shoonya.com/NFO_symbols.txt.zip
https://api.shoonya.com/CDS_symbols.txt.zip
https://api.shoonya.com/MCX_symbols.txt.zip
https://api.shoonya.com/BSE_symbols.txt.zip
https://api.shoonya.com/BFO_symbols.txt.zip
https://api.shoonya.com/NCX_symbols.txt.zip
Rows: 7
Columns: ['Exchange', 'Token', 'LotSize', 'Symbol', 'TradingSymbol', 'Instrument', 'TickSize', 'Unnamed: 7', 'source_zip_url', 'source_file_name', 'Expiry', 'OptionType', 'StrikePrice', 'Unnamed: 10', 'GNGD', 'Unnamed: 11']
Exchange Token TradingSymbol Instrument source_file_name
0 NSE 1594 INFY-EQ EQ NSE_symbols.txt
1 NSE 11536 TCS-EQ EQ NSE_symbols.txt
2 NFO 52175 NIFTY27OCT26F FUTIDX NFO_symbols.txt
3 CDS 1201 USDINR29OCT26F FUTCUR CDS_symbols.txt
4 MCX 451669 CRUDEOIL19OCT26 FUTCOM MCX_symbols.txt
5 BSE 500209 INFY A BSE_symbols.txt
6 BFO 1135463 SENSEX30OCT26F FUTIDX BFO_symbols.txt
examples/stock_brokers/instruments/shoonya/ShoonyaInstruments/example_2_when_an_archive_is_missing.py
"""Shows the two ways `ShoonyaInstruments.download` fails: an archive that cannot be fetched, and archives that hold nothing readable.
`download` checks every archive's HTTP status and raises `requests.HTTPError` as soon as one fails, so a day's master is never stored with an exchange silently missing. When every archive downloads but none holds a `.txt` or `.csv` file, it raises `ValueError` instead, because an empty master is not a result worth storing.
The program must not download anything, so `requests.get` is replaced, only while `download` runs, by `FailingArchiveServer`, a stand-in built for each case. In the first case it answers the MCX archive with a 404, and in the second it serves every archive with only a `README.md` inside. Building the ingester creates a database engine, but the engine opens no connection until a query runs, and this program runs none.
Notice that the first case stops at the fourth archive, the MCX one, without asking for the other three.
Run it from the project root:
python examples/stock_brokers/instruments/shoonya/ShoonyaInstruments/example_2_when_an_archive_is_missing.py
"""
import io
import unittest.mock
import zipfile
import requests
from stock_brokers.instruments.shoonya import (
ShoonyaInstruments,
)
class FailingArchiveServer:
"""A stand-in for `requests.get` that serves archives with nothing readable in them, and optionally fails one URL.
Attributes:
missing_archive (str | None): The file name of the archive to answer with a 404, or None to answer every one.
urls_asked (list): Every URL requested, in order.
"""
def __init__(self, missing_archive):
"""Holds the name of the archive to fail.
Args:
missing_archive (str | None): The file name of the archive to answer with a 404, or None to answer every one.
Returns:
None: This method returns nothing.
"""
self.missing_archive = missing_archive
self.urls_asked = []
def archive_without_data(self):
"""Builds a ZIP archive holding only a README file.
Returns:
bytes: The archive.
"""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, 'w') as archive:
archive.writestr('README.md', 'Symbols are published after 08:00 IST.\n')
return buffer.getvalue()
def get(self, url, timeout=None):
"""Serves one archive, or a 404 for the missing one, in place of `requests.get`.
Args:
url (str): The archive's URL.
timeout (float | None): The timeout, which the stand-in ignores.
Returns:
requests.Response: A 404 for the missing archive, otherwise an archive with no data file.
"""
self.urls_asked.append(url)
response = requests.Response()
response.url = url
if url.endswith('/' + str(self.missing_archive)):
response.status_code = 404
response.reason = 'Not Found'
response.headers['Content-Type'] = 'text/html'
response._content = b'<html><body>Not Found</body></html>'
return response
response.status_code = 200
response.headers['Content-Type'] = 'application/zip'
response._content = self.archive_without_data()
return response
class WhenAnArchiveIsMissingExample:
"""Runs `download` against a server missing one archive and against one serving no data.
Attributes:
instruments (ShoonyaInstruments): The ingester being shown.
"""
def __init__(self):
"""Builds the ingester.
Returns:
None: This method returns nothing.
"""
self.instruments = ShoonyaInstruments()
def run(self):
"""Runs both cases and prints what each raised.
Returns:
None: This method returns nothing.
"""
server = FailingArchiveServer('MCX_symbols.txt.zip')
with unittest.mock.patch.object(requests, 'get', server.get):
try:
self.instruments.download()
except requests.HTTPError as error:
print(f'Raised {type(error).__name__}: {error}')
print(f'Archives asked for: {len(server.urls_asked)}')
server = FailingArchiveServer(None)
with unittest.mock.patch.object(requests, 'get', server.get):
try:
self.instruments.download()
except ValueError as error:
print(f'Raised {type(error).__name__}: {error}')
print(f'Archives asked for: {len(server.urls_asked)}')
if __name__ == '__main__':
WhenAnArchiveIsMissingExample().run()
It prints:
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 Shoonya's seven ZIP archives and stack the text file inside each. |
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 Shoonya's seven ZIP archives and stack the text file inside each.
The files are decoded as latin-1 because that decoding never fails on a byte sequence, and the content is plain ASCII in practice.
Returns:
| Type | Description |
|---|---|
DataFrame
|
Every row Shoonya published, read as text. |
Raises:
| Type | Description |
|---|---|
HTTPError
|
If any archive cannot be downloaded. |
ValueError
|
If the archives download but hold no readable instrument data. |
Example usage
Source code in stock_brokers/instruments/shoonya.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))