Skip to content

flattrade

Flattrade mapping adapter.

Flattrade's file needs work at every stage of the pipeline, most of it about rows rather than columns.

The fetch itself is part of the classification. Around 884 NSE symbols are listed twice, once under the EQ series and once under BE, and the EQ row should win; rows are therefore read with the EQ rows last, so the run's own last-write-wins collapse prefers them. A further hundred or so rows ship with both exchange and instrument blank, a gap in Flattrade's own file. Their tokens are matched against the BSE equity token space of the brokers whose files are complete, which recovers almost all of them as real BSE equities. Those recovered rows are read separately, tagged, and placed first, so a symbol present in both batches keeps its real row rather than the recovered one.

Classification then decides three things no rule can: whether an ambiguous NSE EQ or BE row, or BSE A or B row, is a plain equity, a real exchange traded fund, or a fund to leave alone, which is a priority decision between two segments matching the same predicate; and which NSE index rows are bond-shaped and do not belong in the equity index segment.

Two data faults are corrected on the way out. BSE option rows mis-round half-point strikes, reporting a real 102.5 as 103.0, so the true strike is re-extracted from the trading symbol. And the file carries no tick size column at all, so every segment's tick size resolves to nothing.

Classes:

Name Description
FlattradeMappingAdapter

Mapping adapter for Flattrade, covering 21 real segments plus the uncategorised buckets.

Functions:

Name Description
normalize_index_name

Reduce an index name to uppercase alphanumerics so that differently-spelled broker names collide.
bse_equity_tokens

Collect the BSE equity token space from the brokers whose files carry a complete exchange column.

FlattradeMappingAdapter 

FlattradeMappingAdapter()

Bases: BrokerMappingAdapter

Mapping adapter for Flattrade, covering 21 real segments plus the uncategorised buckets.

Example programs

examples/stock_brokers/instruments/mapping/flattrade/FlattradeMappingAdapter/example_1_funds_indices_and_strikes.py
"""Classifies Flattrade rows whose segment depends on other brokers' lists, and corrects a mis-rounded BSE strike.

Flattrade's file cannot tell an NSE equity from an exchange traded fund or a mutual fund by its own columns, so `FlattradeMappingAdapter.classify` checks the symbol against symbol sets gathered from the other brokers. It also drops the bond-shaped NSE index rows, and it files a blank-exchange row that `read_raw_rows` recovered as a BSE equity. `to_identity` then fixes two things: BSE option strikes, which Flattrade rounds (a real 102.5 arrives as 103.0), are read back from the trading symbol, and NSE index names are brought to one spelling shared with the other brokers.

`run` normally fills the symbol sets and the index name lookup from the database before it classifies anything. This program sets them by hand, so no database is needed. `uncategorised_exchange` shows which exchange's uncategorised bucket a dropped row goes to.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/flattrade/FlattradeMappingAdapter/example_1_funds_indices_and_strikes.py
"""

from stock_brokers.instruments.mapping import flattrade
from stock_brokers.instruments.mapping.flattrade import (
    FlattradeMappingAdapter,
)


class FundsIndicesAndStrikesExample:
    """Classifies eight hand-written Flattrade rows and prints each one's segment and identity.

    Attributes:
        adapter (FlattradeMappingAdapter): The adapter being shown.
        raw_rows (list): The raw rows, as dictionaries of text columns.
    """

    def __init__(self):
        """Builds the adapter, sets the lists `run` would gather, and writes the rows.

        Returns:
            None: This method returns nothing.
        """
        self.adapter = FlattradeMappingAdapter()
        self.adapter.nse_etf_symbols = {
            'NIFTYBEES',
        }
        self.adapter.nse_fund_symbols = {
            'LIQUIDCASE',
        }
        self.adapter.bse_etf_symbols = set()
        self.adapter.bse_fund_symbols = set()
        self.adapter.nse_index_master_lookup = {
            'NIFTYBANK': 'NIFTY BANK',
        }
        recovered_row = self.raw_row('', '500325', 'RELIANCE', 'RELIANCE', '', None, None, None)
        recovered_row[flattrade.RECOVERED_BSE_EQUITY_FLAG] = True
        self.raw_rows = [
            self.raw_row('NSE', '1594', 'INFY', 'INFY-EQ', 'EQ', None, None, None),
            self.raw_row('NSE', '10576', 'NIFTYBEES', 'NIFTYBEES-EQ', 'EQ', None, None, None),
            self.raw_row('NSE', '26401', 'LIQUIDCASE', 'LIQUIDCASE-BE', 'BE', None, None, None),
            self.raw_row('NSE', '26009', 'Nifty Bank', 'Nifty Bank', 'INDEX', None, None, None),
            self.raw_row('NSE', '26071', 'NIFTY SMLCAP 100', 'NIFTY SMLCAP 100', 'INDEX', None, None, None),
            self.raw_row('NSE', '26047', 'Nifty GS 10Yr', 'Nifty GS 10Yr', 'INDEX', None, None, None),
            self.raw_row('BFO', '1165432', 'IRFC', 'IRFC26OCT102.5CE', 'OPTSTK', '27-OCT-2026', '103.00', 'CE'),
            recovered_row,
        ]

    def raw_row(self, exchange, token, symbol, trading_symbol, instrument, expiry, strike, option_type):
        """Builds one raw row in the shape of `flattrade.instruments`.

        Args:
            exchange (str): The exchange code, blank where Flattrade's file leaves it blank.
            token (str): Flattrade's token for the row.
            symbol (str): The `symbol` column.
            trading_symbol (str): The `tradingsymbol` column.
            instrument (str): The `instrument` column.
            expiry (str | None): The expiry, as `DD-MON-YYYY` text.
            strike (str | None): The strike price as Flattrade reports it.
            option_type (str | None): CE or PE for an option.

        Returns:
            dict: The raw row, every column as text as the raw table stores it.
        """
        return {
            'exchange': exchange,
            'token': token,
            'lotsize': '1',
            'symbol': symbol,
            'tradingsymbol': trading_symbol,
            'instrument': instrument,
            'expiry': expiry,
            'strike': strike,
            'optiontype': option_type,
        }

    def run(self):
        """Prints the segment and identity of each row, or where an unmatched row is filed.

        Returns:
            None: This method returns nothing.
        """
        for raw_row in self.raw_rows:
            label = f'{raw_row["exchange"] or "(blank)"} {raw_row["tradingsymbol"]}'
            segment_configuration = self.adapter.classify(raw_row)
            if segment_configuration is None:
                exchange = self.adapter.uncategorised_exchange(raw_row)
                print(f'{label} -> no segment, filed under {exchange}_uncategorised')
                continue
            identity = self.adapter.to_identity(raw_row, segment_configuration)
            print(f'{label} -> {segment_configuration["segment"]}')
            print(f'    identity {identity}')


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

It prints:

NSE INFY-EQ -> nse_equities
    identity {'symbol': 'INFY'}
NSE NIFTYBEES-EQ -> nse_exchange_traded_funds
    identity {'symbol': 'NIFTYBEES'}
NSE LIQUIDCASE-BE -> no segment, filed under nse_uncategorised
NSE Nifty Bank -> nse_equity_indices
    identity {'symbol': 'NIFTY BANK'}
NSE NIFTY SMLCAP 100 -> nse_equity_indices
    identity {'symbol': 'NIFTY SMALLCAP 100'}
NSE Nifty GS 10Yr -> no segment, filed under nse_uncategorised
BFO IRFC26OCT102.5CE -> bse_equity_options
    identity {'underlying_symbol': 'IRFC', 'expiry_date': datetime.date(2026, 10, 27), 'strike_price': 102.5, 'option_type': 'CE'}
(blank) RELIANCE -> bse_equities
    identity {'symbol': 'RELIANCE'}
examples/stock_brokers/instruments/mapping/flattrade/FlattradeMappingAdapter/example_2_mapping_one_day.py
"""Maps one day of Flattrade rows, recovering blank-exchange rows and letting the EQ series win over BE.

`FlattradeMappingAdapter.read_raw_rows` does more than fetch. It asks the database to put the rows with a blank exchange first and the EQ-series rows last, and it marks a blank-exchange row as a recovered BSE equity when Dhan or Stoxkart lists its token as a BSE equity. `run` gathers the fund, exchange traded fund and index lists from the other brokers and then hands over to the shared mapping run, which keeps the last row written for each instrument. So when a symbol is listed under both BE and EQ, the EQ row, read last, is the one stored.

The adapter normally talks to PostgreSQL, so this program replaces its `engine` with a stand-in. The stand-in connection answers queries by matching a piece of their text: the raw-row query gets four hand-written rows, already in the order the real query's `ORDER BY` would give, and Dhan's BSE equity token query gets one token, 500325. Every other cross-broker query gets no rows, and every write is recorded instead of run. pandas only reads through something it recognises as an SQLAlchemy connectable, and it recognises one by the `ConnectionEventsTarget` base class, so the stand-in connection inherits from that class.

Notice that the summary counts four segment memberships but only three distinct instruments, and that the INFY instrument is stored with the EQ row's token, 1594. LIQUIDCASE is filed as a plain equity only because the stand-in's fund lists are empty; on a real day another broker's ISIN would mark it as a fund and leave it uncategorised.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/flattrade/FlattradeMappingAdapter/example_2_mapping_one_day.py
"""

import collections
import datetime

from sqlalchemy.engine import interfaces

from stock_brokers.instruments.mapping.flattrade import (
    FlattradeMappingAdapter,
)


class StandInResult:
    """A stand-in for an SQLAlchemy result holding a fixed set of rows.

    Attributes:
        columns (list): The column names.
        rows (list): The rows, as dictionaries.
    """

    def __init__(self, rows):
        """Holds the rows.

        Args:
            rows (list): The rows, as dictionaries that all share the same keys.

        Returns:
            None: This method returns nothing.
        """
        self.rows = rows
        self.columns = []
        if rows:
            self.columns = list(rows[0])

    def keys(self):
        """The column names, as pandas asks for them.

        Returns:
            list: The column names.
        """
        return self.columns

    def fetchall(self):
        """Every row as a plain tuple, as pandas asks for them.

        Returns:
            list: The rows as tuples in column order.
        """
        tuples = []
        for row in self.rows:
            tuples.append(tuple(row.values()))
        return tuples

    def all(self):
        """Every row as a named tuple, so a column can be read as an attribute.

        Returns:
            list: The rows as named tuples.
        """
        row_type = collections.namedtuple('StandInRow', self.columns)
        named_rows = []
        for row in self.rows:
            named_rows.append(row_type(**row))
        return named_rows


class StandInConnection(interfaces.ConnectionEventsTarget):
    """A stand-in for a PostgreSQL connection that answers queries from prepared rows and records writes.

    Attributes:
        answers (list): Pairs of (a piece of query text, the rows to answer with).
        unanswered_queries (int): How many queries matched no prepared answer and got no rows.
        writes (list): Each write statement's first words and its parameters.
    """

    def __init__(self, answers):
        """Holds the prepared answers and starts with nothing recorded.

        Args:
            answers (list): Pairs of (a piece of query text, the rows to answer with).

        Returns:
            None: This method returns nothing.
        """
        self.answers = answers
        self.unanswered_queries = 0
        self.writes = []

    def __enter__(self):
        """Opens the connection for a `with` block.

        Returns:
            StandInConnection: This connection.
        """
        return self

    def __exit__(self, exception_type, exception, traceback):
        """Closes the connection at the end of a `with` block.

        Args:
            exception_type (type | None): The type of any exception raised in the block.
            exception (BaseException | None): Any exception raised in the block.
            traceback (object | None): The traceback of any exception raised in the block.

        Returns:
            bool: False, so any exception carries on.
        """
        return False

    def execute(self, statement, parameters=None):
        """Answers a query from the prepared rows, or records a write.

        Args:
            statement (sqlalchemy.sql.elements.TextClause): The statement to run.
            parameters (dict | list | None): Its bound parameters.

        Returns:
            StandInResult: The prepared rows whose text matches, otherwise an empty result.
        """
        sql = ' '.join(str(statement).split())
        if not sql.startswith('SELECT'):
            self.writes.append((' '.join(sql.split()[:3]), parameters))
            return StandInResult([])
        for query_text, rows in self.answers:
            if query_text in sql:
                return StandInResult(rows)
        self.unanswered_queries += 1
        return StandInResult([])


class StandInEngine:
    """A stand-in for an SQLAlchemy engine that always hands out the same stand-in connection.

    Attributes:
        connection (StandInConnection): The connection handed out.
    """

    def __init__(self, connection):
        """Holds the connection to hand out.

        Args:
            connection (StandInConnection): The connection handed out.

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

    def connect(self):
        """Hands out the connection for reading.

        Returns:
            StandInConnection: The connection.
        """
        return self.connection

    def begin(self):
        """Hands out the connection for one transaction.

        Returns:
            StandInConnection: The connection.
        """
        return self.connection


class MappingOneDayExample:
    """Maps four Flattrade rows for one date and prints what would be written.

    Attributes:
        mapping_date (datetime.date): The snapshot date being mapped.
        connection (StandInConnection): The stand-in connection.
        adapter (FlattradeMappingAdapter): The adapter being shown, using the stand-in engine.
    """

    def __init__(self):
        """Builds the adapter and swaps its engine for the stand-in.

        Returns:
            None: This method returns nothing.
        """
        self.mapping_date = datetime.date(2026, 9, 28)
        raw_rows = [
            self.raw_row('', '500325', 'RELIANCE', 'RELIANCE', ''),
            self.raw_row('NSE', '26401', 'LIQUIDCASE', 'LIQUIDCASE-BE', 'BE'),
            self.raw_row('NSE', '21594', 'INFY', 'INFY-BE', 'BE'),
            self.raw_row('NSE', '1594', 'INFY', 'INFY-EQ', 'EQ'),
        ]
        dhan_bse_equity_tokens = [
            {
                'security_id': '500325',
            },
        ]
        answers = [
            (
                'SELECT * FROM flattrade.instruments',
                raw_rows,
            ),
            (
                'SELECT security_id FROM dhan.instruments',
                dhan_bse_equity_tokens,
            ),
        ]
        self.connection = StandInConnection(answers)
        self.adapter = FlattradeMappingAdapter()
        self.adapter.engine = StandInEngine(self.connection)

    def raw_row(self, exchange, token, symbol, trading_symbol, instrument):
        """Builds one cash-market raw row in the shape of `flattrade.instruments`.

        Args:
            exchange (str): The exchange code, blank where Flattrade's file leaves it blank.
            token (str): Flattrade's token for the row.
            symbol (str): The `symbol` column.
            trading_symbol (str): The `tradingsymbol` column.
            instrument (str): The `instrument` column, which holds the series for cash rows.

        Returns:
            dict: The raw row, every column as text as the raw table stores it.
        """
        return {
            'exchange': exchange,
            'token': token,
            'lotsize': '1',
            'symbol': symbol,
            'tradingsymbol': trading_symbol,
            'instrument': instrument,
            'expiry': None,
            'strike': None,
            'optiontype': None,
            'download_date': self.mapping_date,
        }

    def run(self):
        """Reads the rows, maps the day, and prints the summary and the rows written.

        Returns:
            None: This method returns nothing.
        """
        raw = self.adapter.read_raw_rows(self.connection, self.mapping_date)
        print('Rows as read_raw_rows returns them:')
        for raw_row in raw.to_dict('records'):
            print(f'  {raw_row["exchange"] or "(blank)"} {raw_row["tradingsymbol"]}: recovered BSE equity {raw_row["recovered_bse_equity"]}')
        summary = self.adapter.run(self.mapping_date)
        print(f'Cross-broker queries that found nothing: {self.connection.unanswered_queries}')
        print(f'Summary: matched {summary["matched"]}, uncategorised {summary["uncategorised"]}, memberships {summary["memberships"]}, instruments {summary["instruments_upserted"]}')
        for statement, parameters in self.connection.writes:
            print(statement)
            if not isinstance(parameters, list):
                continue
            for row in parameters:
                if 'segment' in row:
                    print(f'  {row["segment"]}: {row["symbol"]}')
                else:
                    print(f'  token {row["broker_token"]}: {row["broker_symbol"]}')


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

It prints:

Rows as read_raw_rows returns them:
  (blank) RELIANCE: recovered BSE equity True
  NSE LIQUIDCASE-BE: recovered BSE equity False
  NSE INFY-BE: recovered BSE equity False
  NSE INFY-EQ: recovered BSE equity False
flattrade: 4/4 raw row(s) classified, 0 uncategorised, 4 segment membership(s), 3 distinct instrument(s), 0 error(s).
Cross-broker queries that found nothing: 31
Summary: matched 4, uncategorised 0, memberships 4, instruments 3
DELETE FROM unified.broker_mappings
INSERT INTO unified.instruments
  nse_equities: LIQUIDCASE
  bse_equities: RELIANCE
  nse_equities: INFY
INSERT INTO unified.broker_mappings
  token 26401: LIQUIDCASE-BE
  token 500325: RELIANCE
  token 1594: INFY-EQ

Load and validate the broker's rules file and create the database engine.

Raises:

Type Description
NotImplementedError

If the subclass did not set BROKER_NAME.
ValueError

If the rules file fails validation.
FileNotFoundError

If the broker has no rules file.

Example programs

examples/stock_brokers/instruments/mapping/base/BrokerMappingAdapter/example_1_classifying_rows_by_rules.py
"""Classifies three raw instrument rows using nothing but a broker's rules file.

`BrokerMappingAdapter` is the base class every broker's mapping adapter subclasses. A subclass only has to name its broker in `BROKER_NAME`, and the base class loads the matching rules file from `stock_brokers/instruments/mapping/utilities/rules/`, checks it against the canonical segment vocabulary, and classifies rows by plain equality rules. This program's subclass sets `BROKER_NAME` to Dhan and overrides nothing, so every answer below comes from the base class and Dhan's rules file alone, without the extra filters the real `DhanMappingAdapter` adds.

The rows are written out by hand in the shape of `dhan.instruments`, where every column is text. The adapter builds a database engine when it is constructed, but an engine does not connect until it is used, and this program never asks it to, so nothing touches PostgreSQL.

For each row the program prints the segment the rules chose, any extra segments (the base class never adds any), the identity fields that go into the instrument id, the broker's own fields, and the deterministic instrument id. The third row matches no rule. The base class cannot tell which exchange it belongs to, so it falls into the unprefixed `uncategorised` segment, and the run would store its token as its symbol.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/base/BrokerMappingAdapter/example_1_classifying_rows_by_rules.py
"""

from stock_brokers.instruments.mapping import base
from stock_brokers.instruments.mapping.base import (
    BrokerMappingAdapter,
)


class DhanRulesOnlyAdapter(BrokerMappingAdapter):
    """A mapping adapter that uses Dhan's rules file and nothing but the base class's behaviour."""

    BROKER_NAME = 'dhan'


class ClassifyingRowsByRulesExample:
    """Runs three hand-written Dhan rows through the base class's rules engine and prints the result.

    Attributes:
        adapter (DhanRulesOnlyAdapter): The adapter being shown.
        raw_rows (list): The raw rows to classify, as dictionaries of text columns.
    """

    def __init__(self):
        """Builds the adapter and the three raw rows.

        Returns:
            None: This method returns nothing.
        """
        self.adapter = DhanRulesOnlyAdapter()
        self.raw_rows = [
            {
                'exch_id': 'NSE',
                'segment': 'E',
                'security_id': '1594',
                'isin': 'INE009A01021',
                'instrument': 'EQUITY',
                'underlying_symbol': 'INFY',
                'symbol_name': 'INFOSYS LIMITED',
                'display_name': 'Infosys',
                'instrument_type': 'ES',
                'series': 'EQ',
                'lot_size': '1.0',
                'sm_expiry_date': None,
                'strike_price': None,
                'option_type': None,
                'tick_size': '10.0',
            },
            {
                'exch_id': 'NSE',
                'segment': 'D',
                'security_id': '40215',
                'isin': None,
                'instrument': 'OPTIDX',
                'underlying_symbol': 'NIFTY',
                'symbol_name': 'NIFTY-Oct2026-25000-CE',
                'display_name': 'NIFTY 27 OCT 25000 CALL',
                'instrument_type': 'OP',
                'series': None,
                'lot_size': '75.0',
                'sm_expiry_date': '2026-10-27',
                'strike_price': '25000.00000',
                'option_type': 'CE',
                'tick_size': '5.0',
            },
            {
                'exch_id': 'NSE',
                'segment': 'X',
                'security_id': '990001',
                'isin': None,
                'instrument': 'SPREAD',
                'underlying_symbol': '',
                'symbol_name': 'TEST SPREAD',
                'display_name': 'Test spread',
                'instrument_type': 'SP',
                'series': None,
                'lot_size': '1.0',
                'sm_expiry_date': None,
                'strike_price': None,
                'option_type': None,
                'tick_size': '0',
            },
        ]

    def run(self):
        """Prints how the rules file classifies each row.

        Returns:
            None: This method returns nothing.
        """
        equities = self.adapter.segment_config('nse_equities')
        print(f'Rules file: {self.adapter.config["raw_table"]}, {len(self.adapter.config["segments"])} segments')
        print(f'nse_equities is a {equities["shape"]} on {equities["exchange"]} with {len(equities["rules"])} rules')
        print(f'Unconfigured segment: {self.adapter.segment_config("nse_bonds")}')
        for raw_row in self.raw_rows:
            print()
            print(f'Row {raw_row["security_id"]} ({raw_row["symbol_name"]})')
            segment_configuration = self.adapter.classify(raw_row)
            if segment_configuration is None:
                exchange = self.adapter.uncategorised_exchange(raw_row)
                print(f'  No rule matched, and the base class finds no exchange: {exchange}')
                segment_configuration = self.adapter.segment_config('uncategorised')
            extra_segments = self.adapter.classify_extra(raw_row)
            identity = self.adapter.to_identity(raw_row, segment_configuration)
            broker_fields = self.adapter.to_broker_fields(raw_row, segment_configuration)
            identifier = base.instrument_id(
                segment_configuration['exchange'],
                segment_configuration['segment'],
                segment_configuration['shape'],
                identity,
            )
            print(f'  Segment: {segment_configuration["segment"]} ({segment_configuration["shape"]})')
            print(f'  Extra segments: {extra_segments}')
            print(f'  Identity: {identity}')
            print(f'  Broker fields: {broker_fields}')
            print(f'  Instrument id: {identifier}')


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

It prints:

Rules file: dhan.instruments, 32 segments
nse_equities is a security on nse with 2 rules
Unconfigured segment: None

Row 1594 (INFOSYS LIMITED)
  Segment: nse_equities (security)
  Extra segments: []
  Identity: {'symbol': 'INFY'}
  Broker fields: {'broker_token': '1594', 'broker_symbol': 'Infosys', 'order_symbol': None, 'lot_size': 1.0, 'tick_size': 0.1}
  Instrument id: f310e1b0-e082-5b16-808a-25f741abb9de

Row 40215 (NIFTY-Oct2026-25000-CE)
  Segment: nse_equity_index_options (option)
  Extra segments: []
  Identity: {'underlying_symbol': 'NIFTY', 'expiry_date': datetime.date(2026, 10, 27), 'strike_price': 25000.0, 'option_type': 'CE'}
  Broker fields: {'broker_token': '40215', 'broker_symbol': 'NIFTY-Oct2026-25000-CE', 'order_symbol': None, 'lot_size': 75.0, 'tick_size': 0.05}
  Instrument id: fe85c704-ce49-5190-b068-d4d3646ed162

Row 990001 (TEST SPREAD)
  No rule matched, and the base class finds no exchange: None
  Segment: uncategorised (security)
  Extra segments: []
  Identity: {'symbol': ''}
  Broker fields: {'broker_token': '990001', 'broker_symbol': 'Test spread', 'order_symbol': None, 'lot_size': 1.0, 'tick_size': None}
  Instrument id: 55a8952d-9071-5670-a4fe-a5384d3876bd
examples/stock_brokers/instruments/mapping/base/BrokerMappingAdapter/example_2_mapping_one_day.py
"""Maps one day's raw rows into the two unified tables, against a stand-in database.

`run` is what the daily mapping job calls for each broker. It reads the broker's raw rows for one date with `read_raw_rows`, classifies each row, and then, inside one transaction, deletes that broker's mappings for the date and upserts one row per instrument into `unified.instruments` and one per broker mapping into `unified.broker_mappings`. It returns a summary and prints a one-line report.

The adapter normally reads and writes PostgreSQL, so this program replaces its `engine` with a small stand-in. The stand-in connection answers the raw-row query with four hand-written rows in the shape of `dhan.instruments`, and records every other statement it is sent instead of running it. pandas only reads through something it recognises as an SQLAlchemy connectable, and it recognises one by the `ConnectionEventsTarget` base class, so the stand-in connection inherits from that class and nothing else.

The subclass uses Dhan's rules file and overrides nothing. The fourth row has a strike price that is not a number, so building its identity fails; `run` records that as an error and carries on with the other rows rather than stopping. The unmatched row is stored under its token as its symbol, and its tick size of 5.0 is kept as written, because Dhan's rules divide the tick size by 100 only in the real segments and not in the uncategorised one. The last lines show that a subclass which forgets to set `BROKER_NAME` cannot be built.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/base/BrokerMappingAdapter/example_2_mapping_one_day.py
"""

import datetime

from sqlalchemy.engine import interfaces

from stock_brokers.instruments.mapping.base import (
    BrokerMappingAdapter,
)


class DhanRulesOnlyAdapter(BrokerMappingAdapter):
    """A mapping adapter that uses Dhan's rules file and nothing but the base class's behaviour."""

    BROKER_NAME = 'dhan'


class UnnamedAdapter(BrokerMappingAdapter):
    """A mapping adapter that forgets to name its broker."""


class StandInResult:
    """A stand-in for an SQLAlchemy result holding a fixed set of rows.

    Attributes:
        columns (list): The column names.
        rows (list): The rows, as tuples in column order.
    """

    def __init__(self, columns, rows):
        """Holds the columns and rows.

        Args:
            columns (list): The column names.
            rows (list): The rows, as tuples in column order.

        Returns:
            None: This method returns nothing.
        """
        self.columns = columns
        self.rows = rows

    def keys(self):
        """The column names, as pandas asks for them.

        Returns:
            list: The column names.
        """
        return self.columns

    def fetchall(self):
        """Every row, as pandas asks for them.

        Returns:
            list: The rows.
        """
        return self.rows

    def all(self):
        """Every row, as the adapter's own queries ask for them.

        Returns:
            list: The rows.
        """
        return self.rows


class StandInConnection(interfaces.ConnectionEventsTarget):
    """A stand-in for a PostgreSQL connection that serves one raw table and records every other statement.

    Attributes:
        raw_table (str): The raw table whose rows this connection serves.
        raw_rows (list): The raw rows, as dictionaries of column values.
        statements (list): Each recorded statement's first line and its parameters.
    """

    def __init__(self, raw_table, raw_rows):
        """Holds the raw rows and starts with no statements recorded.

        Args:
            raw_table (str): The raw table whose rows this connection serves.
            raw_rows (list): The raw rows, as dictionaries of column values.

        Returns:
            None: This method returns nothing.
        """
        self.raw_table = raw_table
        self.raw_rows = raw_rows
        self.statements = []

    def __enter__(self):
        """Opens the connection for a `with` block.

        Returns:
            StandInConnection: This connection.
        """
        return self

    def __exit__(self, exception_type, exception, traceback):
        """Closes the connection at the end of a `with` block.

        Args:
            exception_type (type | None): The type of any exception raised in the block.
            exception (BaseException | None): Any exception raised in the block.
            traceback (object | None): The traceback of any exception raised in the block.

        Returns:
            bool: False, so any exception carries on.
        """
        return False

    def execute(self, statement, parameters=None):
        """Answers the raw-row query from the held rows and records anything else.

        Args:
            statement (sqlalchemy.sql.elements.TextClause): The statement to run.
            parameters (dict | list | None): Its bound parameters.

        Returns:
            StandInResult: The raw rows for the raw-row query, otherwise an empty result.
        """
        sql = ' '.join(str(statement).split())
        if sql.startswith(f'SELECT * FROM {self.raw_table} '):
            columns = list(self.raw_rows[0])
            rows = []
            for raw_row in self.raw_rows:
                values = []
                for column in columns:
                    values.append(raw_row[column])
                rows.append(tuple(values))
            return StandInResult(columns, rows)
        self.statements.append((sql, parameters))
        return StandInResult([], [])


class StandInEngine:
    """A stand-in for an SQLAlchemy engine that always hands out the same stand-in connection.

    Attributes:
        connection (StandInConnection): The connection handed out.
    """

    def __init__(self, connection):
        """Holds the connection to hand out.

        Args:
            connection (StandInConnection): The connection handed out.

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

    def connect(self):
        """Hands out the connection for reading.

        Returns:
            StandInConnection: The connection.
        """
        return self.connection

    def begin(self):
        """Hands out the connection for one transaction.

        Returns:
            StandInConnection: The connection.
        """
        return self.connection


class MappingOneDayExample:
    """Maps four raw rows for one date and prints what would be written.

    Attributes:
        mapping_date (datetime.date): The snapshot date being mapped.
        connection (StandInConnection): The stand-in connection holding the raw rows.
        adapter (DhanRulesOnlyAdapter): The adapter being shown, using the stand-in engine.
    """

    def __init__(self):
        """Builds the adapter and swaps its engine for the stand-in.

        Returns:
            None: This method returns nothing.
        """
        self.mapping_date = datetime.date(2026, 9, 28)
        raw_rows = [
            self.raw_row('1594', 'NSE', 'E', 'EQUITY', 'ES', 'INFY', 'INFOSYS LIMITED', None, None, None),
            self.raw_row('40215', 'NSE', 'D', 'OPTIDX', 'OP', 'NIFTY', 'NIFTY-Oct2026-25000-CE', '2026-10-27', '25000.00000', 'CE'),
            self.raw_row('990001', 'NSE', 'X', 'SPREAD', 'SP', '', 'TEST SPREAD', None, None, None),
            self.raw_row('40216', 'NSE', 'D', 'OPTIDX', 'OP', 'NIFTY', 'NIFTY-Oct2026-XX-PE', '2026-10-27', 'not a number', 'PE'),
        ]
        self.connection = StandInConnection('dhan.instruments', raw_rows)
        self.adapter = DhanRulesOnlyAdapter()
        self.adapter.engine = StandInEngine(self.connection)

    def raw_row(self, security_id, exchange, segment, instrument, instrument_type, underlying_symbol, symbol_name, expiry, strike_price, option_type):
        """Builds one raw row in the shape of `dhan.instruments`.

        Args:
            security_id (str): Dhan's token for the row.
            exchange (str): The `exch_id` column.
            segment (str): Dhan's one-letter segment code.
            instrument (str): The `instrument` column.
            instrument_type (str): The `instrument_type` column.
            underlying_symbol (str): The `underlying_symbol` column.
            symbol_name (str): The `symbol_name` column.
            expiry (str | None): The `sm_expiry_date` column.
            strike_price (str | None): The `strike_price` column.
            option_type (str | None): The `option_type` column.

        Returns:
            dict: The raw row, every column as text as the raw table stores it.
        """
        return {
            'exch_id': exchange,
            'segment': segment,
            'security_id': security_id,
            'isin': None,
            'instrument': instrument,
            'underlying_security_id': None,
            'underlying_symbol': underlying_symbol,
            'symbol_name': symbol_name,
            'display_name': symbol_name,
            'instrument_type': instrument_type,
            'series': None,
            'lot_size': '1.0',
            'sm_expiry_date': expiry,
            'strike_price': strike_price,
            'option_type': option_type,
            'tick_size': '5.0',
            'download_date': self.mapping_date,
        }

    def short_row(self, row):
        """Shortens one written row to its filled-in columns, leaving out the dates the example already names.

        Args:
            row (dict): One row the adapter asked to insert.

        Returns:
            dict: The row's columns that hold a value, with the instrument id cut to its first eight characters.
        """
        shortened = {}
        for column, value in row.items():
            if value is None or isinstance(value, datetime.date):
                continue
            if column == 'instrument_id':
                value = value[:8]
            shortened[column] = value
        return shortened

    def run(self):
        """Reads the raw rows, maps them, and prints the summary and the statements sent.

        Returns:
            None: This method returns nothing.
        """
        raw = self.adapter.read_raw_rows(self.connection, self.mapping_date)
        print(f'read_raw_rows returned {len(raw)} rows with {len(raw.columns)} columns')
        print()
        summary = self.adapter.run(self.mapping_date)
        print()
        for key in summary:
            print(f'{key}: {summary[key]}')
        print()
        print('Statements sent inside the write transaction:')
        for sql, parameters in self.connection.statements:
            print(f'  {sql[:60]}...')
            if isinstance(parameters, list):
                for row in parameters:
                    print(f'    {self.short_row(row)}')
            else:
                print(f'    {parameters}')
        print()
        try:
            UnnamedAdapter()
        except NotImplementedError as error:
            print(f'NotImplementedError: {error}')


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

It prints:

read_raw_rows returned 4 rows with 17 columns

dhan: 3/4 raw row(s) classified, 1 uncategorised, 3 segment membership(s), 3 distinct instrument(s), 1 error(s).

broker: dhan
mapping_date: 2026-09-28
raw_rows: 4
matched: 3
uncategorised: 1
memberships: 3
instruments_upserted: 3
errors: [('nse_equity_index_options', "could not convert string to float: 'not a number'")]

Statements sent inside the write transaction:
  DELETE FROM unified.broker_mappings WHERE broker = :b AND ma...
    {'b': 'dhan', 'd': datetime.date(2026, 9, 28)}
  INSERT INTO unified.instruments AS master (instrument_id, ex...
    {'instrument_id': '8fdcd38e', 'exchange': 'unknown', 'segment': 'uncategorised', 'shape': 'security', 'symbol': '990001'}
    {'instrument_id': 'f310e1b0', 'exchange': 'nse', 'segment': 'nse_equities', 'shape': 'security', 'symbol': 'INFY'}
    {'instrument_id': 'fe85c704', 'exchange': 'nse', 'segment': 'nse_equity_index_options', 'shape': 'option', 'underlying_symbol': 'NIFTY', 'strike_price': 25000.0, 'option_type': 'CE'}
  INSERT INTO unified.broker_mappings (instrument_id, broker, ...
    {'instrument_id': '8fdcd38e', 'broker': 'dhan', 'broker_token': '990001', 'broker_symbol': 'TEST SPREAD', 'lot_size': 1.0, 'tick_size': 5.0, 'attributes': '{"display_name": "TEST SPREAD", "instrument_type": "SP"}'}
    {'instrument_id': 'f310e1b0', 'broker': 'dhan', 'broker_token': '1594', 'broker_symbol': 'INFOSYS LIMITED', 'lot_size': 1.0, 'tick_size': 0.05, 'attributes': '{"display_name": "INFOSYS LIMITED", "instrument_type": "ES"}'}
    {'instrument_id': 'fe85c704', 'broker': 'dhan', 'broker_token': '40215', 'broker_symbol': 'NIFTY-Oct2026-25000-CE', 'lot_size': 1.0, 'tick_size': 0.05, 'attributes': '{"display_name": "NIFTY-Oct2026-25000-CE", "instrument_type": "OP"}'}

NotImplementedError: UnnamedAdapter must set BROKER_NAME

Methods:

Name Description
run

Map Flattrade's raw rows for one date, precomputing the four allowlists and the index lookup first.
read_raw_rows

Read Flattrade's rows in the order the classification depends on, marking the blank-exchange rows that can be recovered.
classify

Classify a raw row, deciding the ambiguous equity, fund, and index cases before the rules engine.
to_identity

Build the unified identity fields, correcting the BSE strike rounding and resolving index names.
uncategorised_exchange

Determine the canonical exchange for an unmatched Flattrade row from its exchange column.
segment_config

Look up one segment configuration by its segment value.
classify_extra

Return additional segments this raw row also belongs to beyond the primary match.
to_broker_fields

Build the per-broker mapping fields for a raw row under one segment.
Source code in stock_brokers/instruments/mapping/base.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def __init__(self):
    """
    Load and validate the broker's rules file and create the database engine.

    Raises:
        NotImplementedError: If the subclass did not set BROKER_NAME.
        ValueError: If the rules file fails validation.
        FileNotFoundError: If the broker has no rules file.
    """
    if not self.BROKER_NAME:
        raise NotImplementedError(f"{type(self).__name__} must set BROKER_NAME")
    with open(RULES_DIRECTORY / f"{self.BROKER_NAME}.yaml") as rules_file:
        self.config = yaml.safe_load(rules_file)
    self._validate_config()
    self.engine = get_postgres_engine()
    self.raw_attributes = RawAttributes()

run 

run(mapping_date)

Map Flattrade's raw rows for one date, precomputing the four allowlists and the index lookup first.

Parameters:

Name Type Description Default
mapping_date date

The raw snapshot date to map.

 required

Returns:

Type Description
dict

The run summary, as described in BrokerMappingAdapter.run.

Example usage

From example_2_mapping_one_day.py
summary = self.adapter.run(self.mapping_date)
Source code in stock_brokers/instruments/mapping/flattrade.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def run(self, mapping_date):
    """
    Map Flattrade's raw rows for one date, precomputing the four allowlists and the index lookup first.

    Args:
        mapping_date (datetime.date): The raw snapshot date to map.

    Returns:
        dict: The run summary, as described in BrokerMappingAdapter.run.
    """
    with self.engine.connect() as connection:
        self.nse_fund_symbols = known_nse_fund_symbols(connection, mapping_date)
        self.nse_etf_symbols = known_nse_etf_symbols(connection, mapping_date)
        self.bse_fund_symbols = known_bse_fund_symbols(connection, mapping_date)
        self.bse_etf_symbols = known_bse_etf_symbols(connection, mapping_date)
        self.nse_index_master_lookup = equity_index_lookup(connection, mapping_date)


    return super().run(mapping_date)

read_raw_rows 

read_raw_rows(connection, mapping_date)

Read Flattrade's rows in the order the classification depends on, marking the blank-exchange rows that can be recovered.

Two orderings are imposed at once. The EQ rows come last, so they win the collapse over a same-symbol BE row. The blank-exchange rows come first, so a real row for the same symbol wins over a recovered one.

The recovery marks rows in place rather than reading them a second time. Reading them twice puts each blank-exchange row through the classification twice, once recovered and once not, and the second pass files a duplicate into the uncategorised bucket.

Parameters:

Name Type Description Default
connection Connection

An open SQLAlchemy connection.

 required
mapping_date date

The raw snapshot date to read.

 required

Returns:

Type Description
DataFrame

The raw rows, ordered, with a flag column marking the recovered ones.

Example usage

From example_2_mapping_one_day.py
raw = self.adapter.read_raw_rows(self.connection, self.mapping_date)
Source code in stock_brokers/instruments/mapping/flattrade.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def read_raw_rows(self, connection, mapping_date):
    """
    Read Flattrade's rows in the order the classification depends on, marking the blank-exchange rows that can be recovered.

    Two orderings are imposed at once. The EQ rows come last, so they win the collapse over a same-symbol BE row. The blank-exchange rows come first, so a real row for the same symbol wins over a recovered one.

    The recovery marks rows in place rather than reading them a second time. Reading them twice puts each blank-exchange row through the classification twice, once recovered and once not, and the second pass files a duplicate into the uncategorised bucket.

    Args:
        connection (sqlalchemy.engine.Connection): An open SQLAlchemy connection.
        mapping_date (datetime.date): The raw snapshot date to read.

    Returns:
        pandas.DataFrame: The raw rows, ordered, with a flag column marking the recovered ones.
    """
    raw = pd.read_sql(
        text(
            "SELECT * FROM flattrade.instruments WHERE download_date = :d "
            "ORDER BY (exchange IS NULL OR exchange = '') DESC, (instrument = 'EQ') ASC"
        ),
        connection,
        params={
            "d": mapping_date,
        },
    )
    if raw.empty:
        return raw

    tokens = bse_equity_tokens(connection, mapping_date)
    recovered_flags = []
    for exchange, token in zip(raw["exchange"], raw["token"]):
        is_blank = exchange is None or pd.isna(exchange) or str(exchange).strip() == ""
        recovered_flags.append(is_blank and str(token) in tokens)
    return raw.assign(**{RECOVERED_BSE_EQUITY_FLAG: recovered_flags})

classify 

classify(raw_row)

Classify a raw row, deciding the ambiguous equity, fund, and index cases before the rules engine.

Parameters:

Name Type Description Default
raw_row dict

One raw row from flattrade.instruments.

 required

Returns:

Type Description
dict | None

The matched segment configuration, or None when no segment matches.

Example usage

From example_1_funds_indices_and_strikes.py
segment_configuration = self.adapter.classify(raw_row)
Source code in stock_brokers/instruments/mapping/flattrade.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def classify(self, raw_row):
    """
    Classify a raw row, deciding the ambiguous equity, fund, and index cases before the rules engine.

    Args:
        raw_row (dict): One raw row from flattrade.instruments.

    Returns:
        dict | None: The matched segment configuration, or None when no segment matches.
    """
    exchange = raw_row.get("exchange")
    instrument = raw_row.get("instrument")

    if raw_row.get(RECOVERED_BSE_EQUITY_FLAG) is True:
        if raw_row.get("tradingsymbol") in self.bse_fund_symbols:
            return None
        return self.segment_config("bse_equities")

    if exchange == "NSE" and instrument in ("EQ", "BE"):
        symbol = raw_row.get("symbol")
        if symbol in self.nse_etf_symbols:
            return self.segment_config("nse_exchange_traded_funds")
        if symbol in self.nse_fund_symbols:
            return None
        return self.segment_config("nse_equities")

    if exchange == "BSE" and instrument in ("A", "B"):
        trading_symbol = raw_row.get("tradingsymbol")
        if trading_symbol in self.bse_etf_symbols:
            return self.segment_config("bse_exchange_traded_funds")
        if trading_symbol in self.bse_fund_symbols:
            return None
        return self.segment_config("bse_equities")

    if exchange == "NSE" and instrument == "INDEX":
        trading_symbol = str(raw_row.get("tradingsymbol") or "")
        for excluded_prefix in EXCLUDED_NSE_INDEX_PREFIXES:
            if trading_symbol.startswith(excluded_prefix):
                return None

    return super().classify(raw_row)

to_identity 

to_identity(raw_row, segment_configuration)

Build the unified identity fields, correcting the BSE strike rounding and resolving index names.

Parameters:

Name Type Description Default
raw_row dict

One raw row from flattrade.instruments.

 required
segment_configuration dict

The segment configuration whose identity mapping applies.

 required

Returns:

Type Description
dict

Identity fields, as described in BrokerMappingAdapter.to_identity.

Raises:

Type Description
ValueError

If an expiry value is numeric with no explicit transform.

Example usage

From example_1_funds_indices_and_strikes.py
identity = self.adapter.to_identity(raw_row, segment_configuration)
Source code in stock_brokers/instruments/mapping/flattrade.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def to_identity(self, raw_row, segment_configuration):
    """
    Build the unified identity fields, correcting the BSE strike rounding and resolving index names.

    Args:
        raw_row (dict): One raw row from flattrade.instruments.
        segment_configuration (dict): The segment configuration whose identity mapping applies.

    Returns:
        dict: Identity fields, as described in BrokerMappingAdapter.to_identity.

    Raises:
        ValueError: If an expiry value is numeric with no explicit transform.
    """
    identity = super().to_identity(raw_row, segment_configuration)
    segment = segment_configuration["segment"]

    if segment in BSE_OPTION_SEGMENTS:
        match = STRIKE_PATTERN.search(str(raw_row.get("tradingsymbol") or ""))
        if match:
            identity["strike_price"] = float(match.group(1))
    elif segment == "bse_equity_indices":
        identity["symbol"] = BSE_INDEX_ALIASES.get(identity["symbol"], identity["symbol"])
    elif segment == "nse_equity_indices":
        name = raw_row.get("symbol")
        if name in NSE_INDEX_ALIASES:
            identity["symbol"] = NSE_INDEX_ALIASES[name]
        else:
            normalized = normalize_index_name(name)
            if normalized in self.nse_index_master_lookup:
                identity["symbol"] = self.nse_index_master_lookup[normalized]
    return identity

uncategorised_exchange 

uncategorised_exchange(raw_row)

Determine the canonical exchange for an unmatched Flattrade row from its exchange column.

Parameters:

Name Type Description Default
raw_row dict

One raw row from flattrade.instruments.

 required

Returns:

Type Description
str | None

"nse", "bse", or "mcx", or None when the exchange is blank or unknown.

Example usage

From example_1_funds_indices_and_strikes.py
exchange = self.adapter.uncategorised_exchange(raw_row)
Source code in stock_brokers/instruments/mapping/flattrade.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def uncategorised_exchange(self, raw_row):
    """
    Determine the canonical exchange for an unmatched Flattrade row from its exchange column.

    Args:
        raw_row (dict): One raw row from flattrade.instruments.

    Returns:
        str | None: "nse", "bse", or "mcx", or None when the exchange is blank or unknown.
    """
    exchange_identifier = raw_row.get("exchange")
    if exchange_identifier in ("NSE", "NFO", "CDS"):
        return "nse"
    if exchange_identifier in ("BSE", "BFO"):
        return "bse"
    if exchange_identifier == "MCX":
        return "mcx"
    return None

_validate_config 

_validate_config()

Check the loaded rules file against the canonical vocabulary.

Every configured segment must be a real canonical segment for its exchange with the matching shape, and the segments must appear in the canonical asset-class order. The file must end with the unprefixed uncategorised entry on exchange unknown, which is the guaranteed fallback for rows whose exchange cannot be determined; unknown is allowed for that one entry only, since it is the value stored in the exchange column of the mapped table for those rows.

Raises:

Type Description
ValueError

On any invalid exchange, segment, shape, ordering, or missing or misplaced uncategorised entry.
Source code in stock_brokers/instruments/mapping/base.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def _validate_config(self):
    """
    Check the loaded rules file against the canonical vocabulary.

    Every configured segment must be a real canonical segment for its exchange with the matching shape, and the segments must appear in the canonical asset-class order. The file must end with the unprefixed ``uncategorised`` entry on exchange ``unknown``, which is the guaranteed fallback for rows whose exchange cannot be determined; ``unknown`` is allowed for that one entry only, since it is the value stored in the exchange column of the mapped table for those rows.

    Raises:
        ValueError: On any invalid exchange, segment, shape, ordering, or missing or misplaced uncategorised entry.
    """
    previous_rank = None
    has_plain_uncategorised = False
    for segment_configuration in self.config["segments"]:
        exchange = segment_configuration["exchange"]
        segment = segment_configuration["segment"]
        shape = segment_configuration["shape"]
        is_plain_uncategorised = segment == "uncategorised" and exchange == "unknown"
        if is_plain_uncategorised:
            has_plain_uncategorised = True
            previous_rank = (1000000000, 1000000000)
            continue
        if exchange not in CANONICAL_EXCHANGES:
            raise ValueError(
                f"{self.BROKER_NAME}: {exchange!r} is not a canonical exchange, "
                f"expected one of {CANONICAL_EXCHANGES} or 'unknown' on the plain "
                f"'uncategorised' entry"
            )
        bare_segment = segment
        if segment.startswith(f"{exchange}_"):
            bare_segment = segment[len(exchange) + 1:]
        try:
            declared_shape = segment_shape(bare_segment)
        except ValueError:
            raise ValueError(
                f"{self.BROKER_NAME}: {segment!r} is not a canonical segment for exchange "
                f"{exchange!r}; use the canonical segment value instead"
            )
        if declared_shape != shape:
            raise ValueError(
                f"{self.BROKER_NAME}: segment {segment!r} has shape {declared_shape!r} "
                f"in the canonical vocabulary, not {shape!r} as configured here"
            )
        rank = segment_rank(exchange, bare_segment)
        if previous_rank is not None and rank < previous_rank:
            raise ValueError(
                f"{self.BROKER_NAME}: segment {segment!r} appears out of canonical order; "
                f"segments must be listed fixed income, equities, currencies, commodities, "
                f"then the catch-alls, then the uncategorised buckets, and within a class "
                f"simple instruments, futures, options, indices, index futures, index options"
            )
        previous_rank = rank
    if not has_plain_uncategorised:
        raise ValueError(
            f"{self.BROKER_NAME}: the rules file must end with the unprefixed "
            f"'uncategorised' entry on exchange 'unknown'"
        )

segment_config 

segment_config(segment)

Look up one segment configuration by its segment value.

Parameters:

Name Type Description Default
segment str

The exchange-prefixed segment value, for example "nse_equities".

 required

Returns:

Type Description
dict | None

The matching segment configuration, or None when the rules file does not configure that segment.

Example usage

From example_1_classifying_rows_by_rules.py
equities = self.adapter.segment_config('nse_equities')
From example_1_classifying_rows_by_rules.py
segment_configuration = self.adapter.segment_config('uncategorised')
From example_1_classifying_rows_by_rules.py
print(f'Unconfigured segment: {self.adapter.segment_config("nse_bonds")}')
Source code in stock_brokers/instruments/mapping/base.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def segment_config(self, segment):
    """
    Look up one segment configuration by its segment value.

    Args:
        segment (str): The exchange-prefixed segment value, for example "nse_equities".

    Returns:
        dict | None: The matching segment configuration, or None when the rules file does not configure that segment.
    """
    for segment_configuration in self.config["segments"]:
        if segment_configuration["segment"] == segment:
            return segment_configuration
    return None

_field 

_field(raw_row, specification)

Read one raw column value, applying a named transform when the specification asks for one.

Parameters:

Name Type Description Default
raw_row dict

One raw row from the broker's instrument table.

 required
specification str | dict

Either a bare column name, or a dict with "column" and optionally "transform" keys.

 required

Returns:

Type Description
Any

The column value, transformed if requested.
Source code in stock_brokers/instruments/mapping/base.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def _field(self, raw_row, specification):
    """
    Read one raw column value, applying a named transform when the specification asks for one.

    Args:
        raw_row (dict): One raw row from the broker's instrument table.
        specification (str | dict): Either a bare column name, or a dict with "column" and optionally "transform" keys.

    Returns:
        Any: The column value, transformed if requested.
    """
    if isinstance(specification, dict):
        value = raw_row.get(specification["column"])
        transform = specification.get("transform")
        if transform:
            return TRANSFORMS[transform](value)
        return value
    return raw_row.get(specification)

_row_matches 

_row_matches(raw_row, match)

Check one equality rule against a raw row.

Parameters:

Name Type Description Default
raw_row dict

One raw row from the broker's instrument table.

 required
match dict

Mapping of raw column name to expected value, or to a list of acceptable values.

 required

Returns:

Type Description
bool

True when every column in the rule matches.
Source code in stock_brokers/instruments/mapping/base.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def _row_matches(self, raw_row, match):
    """
    Check one equality rule against a raw row.

    Args:
        raw_row (dict): One raw row from the broker's instrument table.
        match (dict): Mapping of raw column name to expected value, or to a list of acceptable values.

    Returns:
        bool: True when every column in the rule matches.
    """
    for column_name, expected in match.items():
        actual = raw_row.get(column_name)
        if isinstance(expected, list):
            if actual not in expected:
                return False
        elif actual != expected:
            return False
    return True

classify_extra 

classify_extra(raw_row)

Return additional segments this raw row also belongs to beyond the primary match.

A raw row can genuinely satisfy more than one segment's criteria at once — a BSE row can be both a real equity and a crossref-confirmed exchange traded fund. The base implementation returns an empty list; the brokers where dual membership was confirmed override it.

Parameters:

Name Type Description Default
raw_row dict

One raw row from the broker's instrument table.

 required

Returns:

Type Description
list[dict]

Additional segment configurations, empty by default.

Example usage

From example_1_classifying_rows_by_rules.py
extra_segments = self.adapter.classify_extra(raw_row)
Source code in stock_brokers/instruments/mapping/base.py 
371
372
373
374
375
376
377
378
379
380
381
382
383
def classify_extra(self, raw_row):
    """
    Return additional segments this raw row also belongs to beyond the primary match.

    A raw row can genuinely satisfy more than one segment's criteria at once — a BSE row can be both a real equity and a crossref-confirmed exchange traded fund. The base implementation returns an empty list; the brokers where dual membership was confirmed override it.

    Args:
        raw_row (dict): One raw row from the broker's instrument table.

    Returns:
        list[dict]: Additional segment configurations, empty by default.
    """
    return []

to_broker_fields 

to_broker_fields(raw_row, segment_configuration)

Build the per-broker mapping fields for a raw row under one segment.

A tick size of zero or below is stored as None, because no instrument moves in steps of nothing, and brokers send 0 or -1 where they have no tick for a row.

Parameters:

Name Type Description Default
raw_row dict

One raw row from the broker's instrument table.

 required
segment_configuration dict

The segment configuration whose broker field mapping applies.

 required

Returns:

Type Description
dict

Keys "broker_token" (str), "broker_symbol", "order_symbol", "lot_size" (float | None), and "tick_size" (float | None).

Example usage

From example_1_classifying_rows_by_rules.py
broker_fields = self.adapter.to_broker_fields(raw_row, segment_configuration)
Source code in stock_brokers/instruments/mapping/base.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
def to_broker_fields(self, raw_row, segment_configuration):
    """
    Build the per-broker mapping fields for a raw row under one segment.

    A tick size of zero or below is stored as None, because no instrument moves in steps of nothing, and brokers send 0 or -1 where they have no tick for a row.

    Args:
        raw_row (dict): One raw row from the broker's instrument table.
        segment_configuration (dict): The segment configuration whose broker field mapping applies.

    Returns:
        dict: Keys "broker_token" (str), "broker_symbol", "order_symbol", "lot_size" (float | None), and "tick_size" (float | None).
    """
    broker_field_specifications = segment_configuration["broker_fields"]
    lot_size = self._field(raw_row, broker_field_specifications["lot_size"])
    tick_size = self._field(raw_row, broker_field_specifications["tick_size"])
    if tick_size is None or pd.isna(tick_size) or float(tick_size) <= 0:
        tick_size = None
    else:
        tick_size = float(tick_size)
    order_symbol_specification = broker_field_specifications.get("order_symbol")
    if order_symbol_specification is None:
        order_symbol = None
    else:
        order_symbol = self._field(raw_row, order_symbol_specification)
    if order_symbol is not None and pd.isna(order_symbol):
        order_symbol = None
    return {
        "broker_token": str(self._field(raw_row, broker_field_specifications["token"])),
        "broker_symbol": self._field(raw_row, broker_field_specifications["broker_symbol"]),
        "order_symbol": None if order_symbol is None else str(order_symbol).strip(),
        "lot_size": None if lot_size is None or pd.isna(lot_size) else float(lot_size),
        "tick_size": tick_size,
    }

_uncategorised_config 

_uncategorised_config(raw_row)

Pick the uncategorised segment configuration a row falls back into.

The per-exchange bucket is used when uncategorised_exchange resolves an exchange and the rules file configures that bucket; otherwise the unprefixed uncategorised entry is used.

Parameters:

Name Type Description Default
raw_row dict

One raw row from the broker's instrument table.

 required

Returns:

Type Description
dict

The uncategorised segment configuration.

Raises:

Type Description
ValueError

If no uncategorised configuration exists, which validation should have made impossible.
Source code in stock_brokers/instruments/mapping/base.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
def _uncategorised_config(self, raw_row):
    """
    Pick the uncategorised segment configuration a row falls back into.

    The per-exchange bucket is used when ``uncategorised_exchange`` resolves an exchange and the rules file configures that bucket; otherwise the unprefixed ``uncategorised`` entry is used.

    Args:
        raw_row (dict): One raw row from the broker's instrument table.

    Returns:
        dict: The uncategorised segment configuration.

    Raises:
        ValueError: If no uncategorised configuration exists, which validation should have made impossible.
    """
    exchange = self.uncategorised_exchange(raw_row)
    if exchange is not None:
        preferred = self.segment_config(segment_value(exchange, "uncategorised"))
        if preferred is not None:
            return preferred
    fallback = self.segment_config("uncategorised")
    if fallback is None:
        raise ValueError(
            f"{self.BROKER_NAME}: no uncategorised segment configuration found, "
            f"which validation should have made impossible"
        )
    return fallback

_write_results 

_write_results(instrument_rows, broker_rows, mapping_date)

Write one run's rows into the mapped tables inside a single transaction.

Both row lists are sorted by instrument id before the insert so concurrent writers acquire locks in the same order, which is the standard fix for the multi-writer deadlock this design was observed hitting when adapters ran in parallel.

The seen dates widen rather than overwrite, so they do not depend on the order dates happen to be mapped in. Assigning the run's date to the last seen date directly looks equivalent while a backfill walks forward, and is wrong the moment any older date is mapped afterwards: it drags the last seen date backwards, and leaves the first seen date frozen at whichever date was written first rather than the earliest the instrument was actually seen.

Parameters:

Name Type Description Default
instrument_rows dict

Mapping of instrument id to the unified.instruments row.

 required
broker_rows dict

Mapping of instrument id to the unified.broker_mappings row.

 required
mapping_date date

The mapping date the run covers.

 required
Source code in stock_brokers/instruments/mapping/base.py 
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
def _write_results(self, instrument_rows, broker_rows, mapping_date):
    """
    Write one run's rows into the mapped tables inside a single transaction.

    Both row lists are sorted by instrument id before the insert so concurrent writers acquire locks in the same order, which is the standard fix for the multi-writer deadlock this design was observed hitting when adapters ran in parallel.

    The seen dates widen rather than overwrite, so they do not depend on the order dates happen to be mapped in. Assigning the run's date to the last seen date directly looks equivalent while a backfill walks forward, and is wrong the moment any older date is mapped afterwards: it drags the last seen date backwards, and leaves the first seen date frozen at whichever date was written first rather than the earliest the instrument was actually seen.

    Args:
        instrument_rows (dict): Mapping of instrument id to the unified.instruments row.
        broker_rows (dict): Mapping of instrument id to the unified.broker_mappings row.
        mapping_date (datetime.date): The mapping date the run covers.
    """
    sorted_instruments = sorted(instrument_rows.values(), key=lambda row: row["instrument_id"])
    sorted_brokers = sorted(broker_rows.values(), key=lambda row: row["instrument_id"])

    with self.engine.begin() as connection:
        connection.execute(
            text(
                f"DELETE FROM {tables.BROKER_MAPPINGS} "
                "WHERE broker = :b AND mapping_date = :d"
            ),
            {
                "b": self.BROKER_NAME,
                "d": mapping_date,
            },
        )
        if sorted_instruments:
            connection.execute(
                text(
                    f"""
                    INSERT INTO {tables.MASTER} AS master
                        (instrument_id, exchange, segment, shape, symbol, underlying_symbol,
                         expiry_date, strike_price, option_type, first_seen_date, last_seen_date)
                    VALUES
                        (:instrument_id, :exchange, :segment, :shape, :symbol, :underlying_symbol,
                         :expiry_date, :strike_price, :option_type, :first_seen_date, :last_seen_date)
                    ON CONFLICT (instrument_id) DO UPDATE SET
                        first_seen_date = LEAST(master.first_seen_date, EXCLUDED.first_seen_date),
                        last_seen_date = GREATEST(master.last_seen_date, EXCLUDED.last_seen_date)
                    """
                ),
                sorted_instruments,
            )
        if sorted_brokers:
            connection.execute(
                text(
                    f"""
                    INSERT INTO {tables.BROKER_MAPPINGS}
                        (instrument_id, broker, broker_token, broker_symbol, order_symbol, lot_size, tick_size, mapping_date, attributes)
                    VALUES
                        (:instrument_id, :broker, :broker_token, :broker_symbol, :order_symbol, :lot_size, :tick_size, :mapping_date, CAST(:attributes AS jsonb))
                    ON CONFLICT (instrument_id, broker, mapping_date) DO UPDATE SET
                        broker_token = EXCLUDED.broker_token, broker_symbol = EXCLUDED.broker_symbol,
                        order_symbol = EXCLUDED.order_symbol,
                        lot_size = EXCLUDED.lot_size, tick_size = EXCLUDED.tick_size,
                        attributes = EXCLUDED.attributes
                    """
                ),
                sorted_brokers,
            )

normalize_index_name 

normalize_index_name(name)

Reduce an index name to uppercase alphanumerics so that differently-spelled broker names collide.

Parameters:

Name Type Description Default
name str

The raw index name.

 required

Returns:

Type Description
str

The name with every non-alphanumeric character removed and uppercased.
Source code in stock_brokers/instruments/mapping/flattrade.py 
59
60
61
62
63
64
65
66
67
68
69
def normalize_index_name(name):
    """
    Reduce an index name to uppercase alphanumerics so that differently-spelled broker names collide.

    Args:
        name (str): The raw index name.

    Returns:
        str: The name with every non-alphanumeric character removed and uppercased.
    """
    return re.sub(r"[^A-Z0-9]", "", str(name).upper())

bse_equity_tokens 

bse_equity_tokens(connection, mapping_date)

Collect the BSE equity token space from the brokers whose files carry a complete exchange column.

Flattrade shares the exchange-assigned token numbering with Dhan and Stoxkart, so a blank-exchange Flattrade row whose token appears in their BSE equity lists is itself a BSE equity.

Parameters:

Name Type Description Default
connection Connection

An open SQLAlchemy connection.

 required
mapping_date date

The snapshot date to read.

 required

Returns:

Type Description
set

BSE equity tokens, as strings.
Source code in stock_brokers/instruments/mapping/flattrade.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def bse_equity_tokens(connection, mapping_date):
    """
    Collect the BSE equity token space from the brokers whose files carry a complete exchange column.

    Flattrade shares the exchange-assigned token numbering with Dhan and Stoxkart, so a blank-exchange Flattrade row whose token appears in their BSE equity lists is itself a BSE equity.

    Args:
        connection (sqlalchemy.engine.Connection): An open SQLAlchemy connection.
        mapping_date (datetime.date): The snapshot date to read.

    Returns:
        set: BSE equity tokens, as strings.
    """
    tokens = set()
    dhan_rows = connection.execute(
        text(
            "SELECT security_id FROM dhan.instruments WHERE download_date = :d "
            "AND exch_id = 'BSE' AND segment = 'E' AND instrument_type = 'ES'"
        ),
        {
            "d": mapping_date,
        },
    ).all()
    for row in dhan_rows:
        tokens.add(str(row.security_id))

    stoxkart_rows = connection.execute(
        text(
            "SELECT token FROM stoxkart.instruments WHERE download_date = :d "
            "AND exchange = 'BSE' AND instrument_type = 'EQUITIES'"
        ),
        {
            "d": mapping_date,
        },
    ).all()
    for row in stoxkart_rows:
        tokens.add(str(row.token))
    return tokens