underlyings

Decide, once per mapping date, which instrument every live future and option is written on, from the brokers' own records.

python -m stock_brokers.instruments.mapping.utilities.underlyings
python -m stock_brokers.instruments.mapping.utilities.underlyings --date 2026-09-28
python -m stock_brokers.instruments.mapping.utilities.underlyings --date 2026-09-28 --dry-run

A derivative names its underlying only by underlying_symbol, and that name does not always match the underlying's own symbol: the nse's NIFTYFPI contracts are written on the index stored as "Nifty FPI 150", and the bse's SENSEX50 contracts on the one stored as "SNSX50". Three brokers' instrument files say which instrument each derivative is written on, by the exchange's own code for it, and raw_attributes.py keeps that code as the underlying_token attribute: Dhan's underlying_security_id, Groww's underlying_exchange_token and Fyers' underlying_scrip_code. Wisdom Capital's underlyinginstrumentid is left out, because it is prefixed with a segment code and is -1 for every index.

The same exchange codes identify the candidates. Dhan's and Groww's broker token is the exchange's code for a share or a future, and Fyers keeps an index's and a share's own exchange code as their underlying_token, which covers the indices, whose Dhan and Groww tokens are not the exchange's code. A code is looked up only among the candidates on the derivative's exchange and in the segments its underlying can be in: the family's cash or index segment, and for an option also the family's futures segment, because an MCX option is written on a future. That restriction matters, because the same code names unrelated instruments in other segments: an nse commodity future's code also names an equity index.

A contract is resolved when every code any broker gives leads to one and the same instrument, and that instrument is its underlying_instrument_id. It is ambiguous when the codes lead to more than one, unresolved when they lead to none, and no_code when no broker gives a code. The decision and every broker's code are written to unified.underlyings for the date, and warm_cache.py copies the resolved ones into Redis for the REST API's /details and /master.

Classes:

Name	Description
UnderlyingResolver	Decides and records which instrument every live derivative of one mapping date is written on.
UnderlyingCommand	The command line: decide a date's underlyings and print how many derivatives each segment has in each status.

UnderlyingResolver

UnderlyingResolver(engine=None)

Decides and records which instrument every live derivative of one mapping date is written on.

Attributes:

Name	Type	Description
WRITE_BATCH_ROWS	int	How many rows one insert sends.
engine	Engine	The database the unified tables are read from and written to.

Builds the resolver around a database engine.

Parameters:

Name	Type	Description	Default
engine	Engine | None	The engine to use, or None for the configured one.	None

Returns:

Type	Description
None	This method returns nothing.

Example programs

Deciding and writingA dry run

examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingResolver/example_1_deciding_and_writing.py

"""Decides which instrument four live derivatives are written on, writes the decisions and counts them.

A derivative names its underlying only by a symbol that does not always match the underlying's own symbol: NSE's NIFTYFPI contracts are written on the index stored as "Nifty FPI 150". Dhan, Groww and Fyers each give the exchange's code for a derivative's underlying, and the resolver looks that code up among the shares, indices and futures on the same exchange in the segments the underlying can be in.

This program takes the three steps `run` would take one at a time: `decisions` reads the derivatives, the codes and the candidates and decides each derivative, `write` replaces the date's stored decisions in one transaction, and `summarise` counts them by segment and status.

The engine is a stand-in that answers from rows held in memory and records what is written, so no PostgreSQL is needed. The codes are made up. Notice that the MCX GOLD option resolves to the GOLD future, because an MCX option is written on a future, that the INFY option resolves to the share even though Dhan and Groww both give the code, and that the RELIANCE future, for which no broker gives a code, is `no_code`.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingResolver/example_1_deciding_and_writing.py
"""

import datetime
import types

from stock_brokers.instruments.mapping.utilities.underlyings import (
    UnderlyingResolver,
)


class StandInConnection:
    """A stand-in database connection that answers the resolver's three queries from rows in memory and records writes.

    Attributes:
        database (StandInDatabase): The rows and the record of writes.
    """

    def __init__(self, database):
        """Holds the stand-in database.

        Args:
            database (StandInDatabase): The rows and the record of writes.

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

    def __enter__(self):
        """Opens the connection.

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

    def __exit__(self, error_type, error, traceback):
        """Closes the connection.

        Args:
            error_type (type | None): The exception type raised inside the block, if any.
            error (BaseException | None): The exception raised inside the block, if any.
            traceback (types.TracebackType | None): The traceback, if any.

        Returns:
            bool: False, so an exception is never swallowed.
        """
        return False

    def execute(self, statement, parameters):
        """Answers one query from the rows in memory, or records one write.

        Args:
            statement (sqlalchemy.sql.elements.TextClause): The statement.
            parameters (dict | list): The statement's parameters, or a list of them for a batch insert.

        Returns:
            list: The rows the query returns, empty for a write.
        """
        sql = str(statement)
        if sql.startswith('DELETE'):
            self.database.deleted_dates.append(parameters['mapping_date'])
            return []
        if sql.startswith('INSERT'):
            for row in parameters:
                self.database.written_rows.append(row)
            return []
        if sql.startswith('SELECT m.instrument_id, m.broker'):
            return self.database.code_rows
        if 'CASE WHEN' in sql:
            return self.database.candidate_rows
        return self.database.derivative_rows


class StandInDatabase:
    """The rows the stand-in answers with, and the writes it has received.

    Attributes:
        derivative_rows (list): The live derivatives, as rows with `instrument_id`, `exchange` and `segment`.
        code_rows (list): The codes brokers give, as rows with `instrument_id`, `broker` and `code`.
        candidate_rows (list): The candidates, as rows with `instrument_id`, `exchange`, `segment` and `code`.
        deleted_dates (list): The mapping dates whose decisions were deleted.
        written_rows (list): The decision rows inserted.
    """

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

        Args:
            derivative_rows (list): The live derivatives.
            code_rows (list): The codes brokers give.
            candidate_rows (list): The candidates.

        Returns:
            None: This method returns nothing.
        """
        self.derivative_rows = derivative_rows
        self.code_rows = code_rows
        self.candidate_rows = candidate_rows
        self.deleted_dates = []
        self.written_rows = []


class StandInEngine:
    """A stand-in SQLAlchemy engine whose connections all share one stand-in database.

    Attributes:
        database (StandInDatabase): The shared rows.
    """

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

        Args:
            database (StandInDatabase): The shared rows.

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

    def connect(self):
        """Opens a connection for reading.

        Returns:
            StandInConnection: A connection over the shared rows.
        """
        return StandInConnection(self.database)

    def begin(self):
        """Opens a connection inside a transaction.

        Returns:
            StandInConnection: A connection over the shared rows.
        """
        return StandInConnection(self.database)


class Rows:
    """Builds the three kinds of stand-in row."""

    def derivative(self, instrument_id, exchange, segment):
        """Builds one live derivative row.

        Args:
            instrument_id (str): The derivative's instrument id.
            exchange (str): Its exchange.
            segment (str): Its exchange-prefixed segment.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            exchange=exchange,
            segment=segment,
        )

    def code(self, instrument_id, broker, code):
        """Builds one row of the code a broker gives for a derivative's underlying.

        Args:
            instrument_id (str): The derivative's instrument id.
            broker (str): The broker.
            code (str): The exchange code it gives.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            broker=broker,
            code=code,
        )

    def candidate(self, instrument_id, exchange, segment, code):
        """Builds one candidate row.

        Args:
            instrument_id (str): The candidate's instrument id.
            exchange (str): Its exchange.
            segment (str): Its exchange-prefixed segment.
            code (str): Its exchange code.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            exchange=exchange,
            segment=segment,
            code=code,
        )


class DecidingAndWritingExample:
    """Decides, writes and summarises the underlyings of four derivatives on 2026-09-28.

    Attributes:
        database (StandInDatabase): The stand-in rows.
        resolver (UnderlyingResolver): The resolver being shown.
    """

    def __init__(self):
        """Builds the stand-in rows and the resolver over them.

        Returns:
            None: This method returns nothing.
        """
        rows = Rows()
        self.database = StandInDatabase(
            [
                rows.derivative('niftyfpi-future-oct', 'nse', 'nse_equity_index_futures'),
                rows.derivative('infy-option-1600-ce', 'nse', 'nse_equity_options'),
                rows.derivative('gold-option-120000-ce', 'mcx', 'mcx_commodity_options'),
                rows.derivative('reliance-future-oct', 'nse', 'nse_equity_futures'),
            ],
            [
                rows.code('niftyfpi-future-oct', 'fyers', '26074'),
                rows.code('infy-option-1600-ce', 'dhan', '1594'),
                rows.code('infy-option-1600-ce', 'groww', '1594'),
                rows.code('gold-option-120000-ce', 'dhan', '454818'),
            ],
            [
                rows.candidate('nifty-fpi-150-index', 'nse', 'nse_equity_indices', '26074'),
                rows.candidate('infy-share', 'nse', 'nse_equities', '1594'),
                rows.candidate('infy-future-oct', 'nse', 'nse_equity_futures', '53001'),
                rows.candidate('gold-future-dec', 'mcx', 'mcx_commodity_futures', '454818'),
            ],
        )
        self.resolver = UnderlyingResolver(StandInEngine(self.database))

    def run(self):
        """Decides, writes and summarises, printing each step.

        Returns:
            None: This method returns nothing.
        """
        mapping_date = datetime.date(2026, 9, 28)
        decided = self.resolver.decisions(mapping_date)
        for row in decided:
            print(f'{row["instrument_id"]}: {row["status"]}, underlying {row["underlying_instrument_id"]}, codes {row["sources"]}')
        written = self.resolver.write(mapping_date, decided)
        print(f'Rows written: {written}, after deleting {len(self.database.deleted_dates)} earlier set for the date')
        summary = self.resolver.summarise(decided)
        for segment, status in sorted(summary):
            print(f'  {segment} {status}: {summary[(segment, status)]}')


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

It prints:

niftyfpi-future-oct: resolved, underlying nifty-fpi-150-index, codes {"fyers": "26074"}
infy-option-1600-ce: resolved, underlying infy-share, codes {"dhan": "1594", "groww": "1594"}
gold-option-120000-ce: resolved, underlying gold-future-dec, codes {"dhan": "454818"}
reliance-future-oct: no_code, underlying None, codes {}
Rows written: 4, after deleting 1 earlier set for the date
  mcx_commodity_options resolved: 1
  nse_equity_futures no_code: 1
  nse_equity_index_futures resolved: 1
  nse_equity_options resolved: 1

examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingResolver/example_2_a_dry_run.py

"""Tries the underlying decision without writing anything, and shows why a code is looked up only in the right segments.

`run` with `dry_run=True` decides every live derivative and returns the count per segment and status, but never touches `unified.underlyings`, which is what `--dry-run` on the command line does. The three reading steps, `live_derivatives`, `underlying_codes` and `candidates`, can also be called on an open connection to look at what the decision is made from.

`decide` then shows the three outcomes other than `resolved`. An NSE commodity future whose code also names an NSE equity index stays `unresolved`, because a commodity future's underlying can only be in the commodities segment and the index is never considered; this is the reason the lookup is restricted by segment. An option whose code leads to two different instruments is `ambiguous`. A derivative no broker gives a code for is `no_code`.

The engine is a stand-in answering from rows in memory, so no PostgreSQL is needed; the codes are made up.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingResolver/example_2_a_dry_run.py
"""

import datetime
import types

from stock_brokers.instruments.mapping.utilities.underlyings import (
    UnderlyingResolver,
)


class StandInConnection:
    """A stand-in database connection that answers the resolver's three queries from rows in memory and records writes.

    Attributes:
        database (StandInDatabase): The rows and the record of writes.
    """

    def __init__(self, database):
        """Holds the stand-in database.

        Args:
            database (StandInDatabase): The rows and the record of writes.

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

    def __enter__(self):
        """Opens the connection.

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

    def __exit__(self, error_type, error, traceback):
        """Closes the connection.

        Args:
            error_type (type | None): The exception type raised inside the block, if any.
            error (BaseException | None): The exception raised inside the block, if any.
            traceback (types.TracebackType | None): The traceback, if any.

        Returns:
            bool: False, so an exception is never swallowed.
        """
        return False

    def execute(self, statement, parameters):
        """Answers one query from the rows in memory, or records one write.

        Args:
            statement (sqlalchemy.sql.elements.TextClause): The statement.
            parameters (dict | list): The statement's parameters, or a list of them for a batch insert.

        Returns:
            list: The rows the query returns, empty for a write.
        """
        sql = str(statement)
        if sql.startswith('DELETE'):
            self.database.deleted_dates.append(parameters['mapping_date'])
            return []
        if sql.startswith('INSERT'):
            for row in parameters:
                self.database.written_rows.append(row)
            return []
        if sql.startswith('SELECT m.instrument_id, m.broker'):
            return self.database.code_rows
        if 'CASE WHEN' in sql:
            return self.database.candidate_rows
        return self.database.derivative_rows


class StandInDatabase:
    """The rows the stand-in answers with, and the writes it has received.

    Attributes:
        derivative_rows (list): The live derivatives, as rows with `instrument_id`, `exchange` and `segment`.
        code_rows (list): The codes brokers give, as rows with `instrument_id`, `broker` and `code`.
        candidate_rows (list): The candidates, as rows with `instrument_id`, `exchange`, `segment` and `code`.
        deleted_dates (list): The mapping dates whose decisions were deleted.
        written_rows (list): The decision rows inserted.
    """

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

        Args:
            derivative_rows (list): The live derivatives.
            code_rows (list): The codes brokers give.
            candidate_rows (list): The candidates.

        Returns:
            None: This method returns nothing.
        """
        self.derivative_rows = derivative_rows
        self.code_rows = code_rows
        self.candidate_rows = candidate_rows
        self.deleted_dates = []
        self.written_rows = []


class StandInEngine:
    """A stand-in SQLAlchemy engine whose connections all share one stand-in database.

    Attributes:
        database (StandInDatabase): The shared rows.
    """

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

        Args:
            database (StandInDatabase): The shared rows.

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

    def connect(self):
        """Opens a connection for reading.

        Returns:
            StandInConnection: A connection over the shared rows.
        """
        return StandInConnection(self.database)

    def begin(self):
        """Opens a connection inside a transaction.

        Returns:
            StandInConnection: A connection over the shared rows.
        """
        return StandInConnection(self.database)


class Rows:
    """Builds the three kinds of stand-in row."""

    def derivative(self, instrument_id, exchange, segment):
        """Builds one live derivative row.

        Args:
            instrument_id (str): The derivative's instrument id.
            exchange (str): Its exchange.
            segment (str): Its exchange-prefixed segment.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            exchange=exchange,
            segment=segment,
        )

    def code(self, instrument_id, broker, code):
        """Builds one row of the code a broker gives for a derivative's underlying.

        Args:
            instrument_id (str): The derivative's instrument id.
            broker (str): The broker.
            code (str): The exchange code it gives.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            broker=broker,
            code=code,
        )

    def candidate(self, instrument_id, exchange, segment, code):
        """Builds one candidate row.

        Args:
            instrument_id (str): The candidate's instrument id.
            exchange (str): Its exchange.
            segment (str): Its exchange-prefixed segment.
            code (str): Its exchange code.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            exchange=exchange,
            segment=segment,
            code=code,
        )


class DryRunExample:
    """Runs a dry run over three derivatives, reads the inputs and decides three edge cases.

    Attributes:
        database (StandInDatabase): The stand-in rows.
        engine (StandInEngine): The stand-in engine.
        resolver (UnderlyingResolver): The resolver being shown.
        mapping_date (datetime.date): The date being decided.
    """

    def __init__(self):
        """Builds the stand-in rows and the resolver over them.

        Returns:
            None: This method returns nothing.
        """
        rows = Rows()
        self.database = StandInDatabase(
            [
                rows.derivative('crude-future-nse-oct', 'nse', 'nse_commodity_futures'),
                rows.derivative('sbin-option-800-pe', 'nse', 'nse_equity_options'),
                rows.derivative('usdinr-future-oct', 'nse', 'nse_currency_futures'),
            ],
            [
                rows.code('crude-future-nse-oct', 'dhan', '26000'),
                rows.code('sbin-option-800-pe', 'dhan', '3045'),
                rows.code('sbin-option-800-pe', 'groww', '3045'),
            ],
            [
                rows.candidate('nifty-50-index', 'nse', 'nse_equity_indices', '26000'),
                rows.candidate('sbin-share', 'nse', 'nse_equities', '3045'),
                rows.candidate('sbin-future-oct', 'nse', 'nse_equity_futures', '3045'),
            ],
        )
        self.engine = StandInEngine(self.database)
        self.resolver = UnderlyingResolver(self.engine)
        self.mapping_date = datetime.date(2026, 9, 28)

    def run(self):
        """Prints the dry run's counts, the inputs and three direct decisions.

        Returns:
            None: This method returns nothing.
        """
        summary = self.resolver.run(self.mapping_date, dry_run=True)
        for segment, status in sorted(summary):
            print(f'{segment} {status}: {summary[(segment, status)]}')
        print(f'Rows written by the dry run: {len(self.database.written_rows)}')
        with self.engine.connect() as connection:
            derivatives = self.resolver.live_derivatives(connection, self.mapping_date)
            codes = self.resolver.underlying_codes(connection, self.mapping_date)
            candidates = self.resolver.candidates(connection, self.mapping_date)
        print(f'Derivatives: {derivatives}')
        print(f'Codes: {dict(codes)}')
        for key in sorted(candidates):
            print(f'Candidate {key}: {sorted(candidates[key])}')
        underlying, status = self.resolver.decide(
            'nse',
            'nse_commodity_futures',
            {
                'dhan': '26000',
            },
            candidates,
        )
        print(f'NSE commodity future with code 26000: {status}, {underlying}')
        underlying, status = self.resolver.decide(
            'nse',
            'nse_equity_options',
            {
                'dhan': '3045',
            },
            candidates,
        )
        print(f'NSE equity option with code 3045: {status}, {underlying}')
        underlying, status = self.resolver.decide(
            'nse',
            'nse_currency_futures',
            {},
            candidates,
        )
        print(f'NSE currency future with no code: {status}, {underlying}')


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

It prints:

nse_commodity_futures unresolved: 1
nse_currency_futures no_code: 1
nse_equity_options ambiguous: 1
Rows written by the dry run: 0
Derivatives: {'crude-future-nse-oct': ('nse', 'nse_commodity_futures'), 'sbin-option-800-pe': ('nse', 'nse_equity_options'), 'usdinr-future-oct': ('nse', 'nse_currency_futures')}
Codes: {'crude-future-nse-oct': {'dhan': '26000'}, 'sbin-option-800-pe': {'dhan': '3045', 'groww': '3045'}}
Candidate ('nse', 'nse_equities', '3045'): ['sbin-share']
Candidate ('nse', 'nse_equity_futures', '3045'): ['sbin-future-oct']
Candidate ('nse', 'nse_equity_indices', '26000'): ['nifty-50-index']
NSE commodity future with code 26000: unresolved, None
NSE equity option with code 3045: ambiguous, None
NSE currency future with no code: no_code, None

Methods:

Name	Description
live_derivatives	Reads every future and option mapped on the date that has not expired.
underlying_codes	Reads the exchange code each broker gives for each derivative's underlying.
candidates	Reads the exchange code of every share, index and future mapped on the date.
decide	Decides one derivative's underlying from the codes the brokers give for it.
decisions	Decides every live derivative's underlying for the date.
write	Replaces the date's decisions with the given rows, in one transaction.
summarise	Counts the decisions by segment and status.
run	Decides every live derivative's underlying for the date, and writes the decisions unless this is a dry run.

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def __init__(self, engine=None):
    """
    Builds the resolver around a database engine.

    Args:
        engine (sqlalchemy.engine.Engine | None): The engine to use, or None for the configured one.

    Returns:
        None: This method returns nothing.
    """
    self.engine = engine or get_postgres_engine()

live_derivatives

live_derivatives(connection, mapping_date)

Reads every future and option mapped on the date that has not expired.

Parameters:

Name	Type	Description	Default
connection	Connection	The open connection.	required
mapping_date	date	The mapping date.	required

Returns:

Type	Description
dict	Instrument id as text to a pair of its exchange and its exchange-prefixed segment.

Example usage

From example_2_a_dry_run.py

derivatives = self.resolver.live_derivatives(connection, self.mapping_date)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def live_derivatives(self, connection, mapping_date):
    """
    Reads every future and option mapped on the date that has not expired.

    Args:
        connection (sqlalchemy.engine.Connection): The open connection.
        mapping_date (datetime.date): The mapping date.

    Returns:
        dict: Instrument id as text to a pair of its exchange and its exchange-prefixed segment.
    """
    statement = text(
        f"SELECT DISTINCT i.instrument_id, i.exchange, i.segment FROM {tables.MASTER} i "
        f"JOIN {tables.BROKER_MAPPINGS} m ON m.instrument_id = i.instrument_id AND m.mapping_date = :mapping_date "
        "WHERE i.shape IN ('future', 'option') AND i.expiry_date >= :mapping_date"
    )
    derivatives = {}
    for row in connection.execute(statement, {"mapping_date": mapping_date}):
        derivatives[str(row.instrument_id)] = (row.exchange, row.segment)
    return derivatives

underlying_codes

underlying_codes(connection, mapping_date)

Reads the exchange code each broker gives for each derivative's underlying.

Parameters:

Name	Type	Description	Default
connection	Connection	The open connection.	required
mapping_date	date	The mapping date.	required

Returns:

Type	Description
dict	Instrument id as text to a dict of broker name to the code that broker gives.

Example usage

From example_2_a_dry_run.py

codes = self.resolver.underlying_codes(connection, self.mapping_date)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def underlying_codes(self, connection, mapping_date):
    """
    Reads the exchange code each broker gives for each derivative's underlying.

    Args:
        connection (sqlalchemy.engine.Connection): The open connection.
        mapping_date (datetime.date): The mapping date.

    Returns:
        dict: Instrument id as text to a dict of broker name to the code that broker gives.
    """
    statement = text(
        "SELECT m.instrument_id, m.broker, m.attributes->>'underlying_token' AS code "
        f"FROM {tables.BROKER_MAPPINGS} m JOIN {tables.MASTER} i ON i.instrument_id = m.instrument_id "
        "WHERE m.mapping_date = :mapping_date AND m.broker = ANY(:brokers) "
        "AND i.shape IN ('future', 'option') AND i.expiry_date >= :mapping_date "
        "AND nullif(m.attributes->>'underlying_token', '') IS NOT NULL"
    )
    codes = collections.defaultdict(dict)
    parameters = {
        "mapping_date": mapping_date,
        "brokers": CODE_BROKERS,
    }
    for row in connection.execute(statement, parameters):
        codes[str(row.instrument_id)][row.broker] = row.code
    return codes

candidates

candidates(connection, mapping_date)

Reads the exchange code of every share, index and future mapped on the date.

Parameters:

Name	Type	Description	Default
connection	Connection	The open connection.	required
mapping_date	date	The mapping date.	required

Returns:

Type	Description
dict	A triple of exchange, exchange-prefixed segment and exchange code to the set of instrument ids, as text, that carry that code there.

Example usage

From example_2_a_dry_run.py

candidates = self.resolver.candidates(connection, self.mapping_date)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def candidates(self, connection, mapping_date):
    """
    Reads the exchange code of every share, index and future mapped on the date.

    Args:
        connection (sqlalchemy.engine.Connection): The open connection.
        mapping_date (datetime.date): The mapping date.

    Returns:
        dict: A triple of exchange, exchange-prefixed segment and exchange code to the set of instrument ids, as text, that carry that code there.
    """
    statement = text(
        "SELECT DISTINCT i.instrument_id, i.exchange, i.segment, "
        "CASE WHEN m.broker = ANY(:token_brokers) THEN m.broker_token "
        "ELSE m.attributes->>'underlying_token' END AS code "
        f"FROM {tables.BROKER_MAPPINGS} m JOIN {tables.MASTER} i ON i.instrument_id = m.instrument_id "
        "WHERE m.mapping_date = :mapping_date AND ("
        "(m.broker = ANY(:token_brokers) AND i.shape IN ('security', 'future')) "
        "OR (m.broker = ANY(:attribute_brokers) AND i.shape = 'security'))"
    )
    parameters = {
        "mapping_date": mapping_date,
        "token_brokers": TOKEN_CODE_BROKERS,
        "attribute_brokers": ATTRIBUTE_CODE_BROKERS,
    }
    found = collections.defaultdict(set)
    for row in connection.execute(statement, parameters):
        if row.code:
            found[(row.exchange, row.segment, row.code)].add(str(row.instrument_id))
    return found

decide

decide(exchange, segment, codes, candidates)

Decides one derivative's underlying from the codes the brokers give for it.

Parameters:

Name	Type	Description	Default
exchange	str	The derivative's exchange.	required
segment	str	The derivative's exchange-prefixed segment.	required
codes	dict	Broker name to the code that broker gives, which may be empty.	required
candidates	dict	The candidates, as candidates returns them.	required

Returns:

Type	Description
tuple	The underlying's instrument id as text, or None, and the status.

Example usage

From example_2_a_dry_run.py

underlying, status = self.resolver.decide(
    'nse',
    'nse_commodity_futures',
    {
        'dhan': '26000',
    },
    candidates,
)

From example_2_a_dry_run.py

underlying, status = self.resolver.decide(
    'nse',
    'nse_equity_options',
    {
        'dhan': '3045',
    },
    candidates,
)

From example_2_a_dry_run.py

underlying, status = self.resolver.decide(
    'nse',
    'nse_currency_futures',
    {},
    candidates,
)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def decide(self, exchange, segment, codes, candidates):
    """
    Decides one derivative's underlying from the codes the brokers give for it.

    Args:
        exchange (str): The derivative's exchange.
        segment (str): The derivative's exchange-prefixed segment.
        codes (dict): Broker name to the code that broker gives, which may be empty.
        candidates (dict): The candidates, as `candidates` returns them.

    Returns:
        tuple: The underlying's instrument id as text, or None, and the status.
    """
    if not codes:
        return None, NO_CODE
    bare_segment = segment.removeprefix(f"{exchange}_")
    found = set()
    for candidate_bare in CANDIDATE_SEGMENTS.get(bare_segment, []):
        candidate_segment = f"{exchange}_{candidate_bare}"
        for code in codes.values():
            found.update(candidates.get((exchange, candidate_segment, code), set()))
    if len(found) == 1:
        return next(iter(found)), RESOLVED
    if found:
        return None, AMBIGUOUS
    return None, UNRESOLVED

decisions

decisions(mapping_date)

Decides every live derivative's underlying for the date.

Parameters:

Name	Type	Description	Default
mapping_date	date	The mapping date.	required

Returns:

Type	Description
list	One dictionary per derivative with instrument_id, mapping_date, segment, underlying_instrument_id, status and sources.

Example usage

From example_1_deciding_and_writing.py

decided = self.resolver.decisions(mapping_date)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def decisions(self, mapping_date):
    """
    Decides every live derivative's underlying for the date.

    Args:
        mapping_date (datetime.date): The mapping date.

    Returns:
        list: One dictionary per derivative with `instrument_id`, `mapping_date`, `segment`, `underlying_instrument_id`, `status` and `sources`.
    """
    with self.engine.connect() as connection:
        derivatives = self.live_derivatives(connection, mapping_date)
        codes_by_derivative = self.underlying_codes(connection, mapping_date)
        candidates = self.candidates(connection, mapping_date)
    rows = []
    for instrument_id, (exchange, segment) in derivatives.items():
        codes = codes_by_derivative.get(instrument_id, {})
        underlying, status = self.decide(exchange, segment, codes, candidates)
        rows.append({
            "instrument_id": instrument_id,
            "mapping_date": mapping_date,
            "segment": segment,
            "underlying_instrument_id": underlying,
            "status": status,
            "sources": json.dumps(codes, sort_keys=True),
        })
    return rows

write

write(mapping_date, rows)

Replaces the date's decisions with the given rows, in one transaction.

Parameters:

Name	Type	Description	Default
mapping_date	date	The mapping date.	required
rows	list	The decisions, as decisions returns them.	required

Returns:

Type	Description
int	The number of rows written.

Example usage

From example_1_deciding_and_writing.py

written = self.resolver.write(mapping_date, decided)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def write(self, mapping_date, rows):
    """
    Replaces the date's decisions with the given rows, in one transaction.

    Args:
        mapping_date (datetime.date): The mapping date.
        rows (list): The decisions, as `decisions` returns them.

    Returns:
        int: The number of rows written.
    """
    delete_statement = text(
        f"DELETE FROM {tables.UNDERLYINGS} "
        "WHERE mapping_date = :mapping_date"
    )
    insert_statement = text(
        f"INSERT INTO {tables.UNDERLYINGS} "
        "(instrument_id, mapping_date, segment, underlying_instrument_id, status, sources) "
        "VALUES (CAST(:instrument_id AS uuid), :mapping_date, :segment, "
        "CAST(:underlying_instrument_id AS uuid), :status, CAST(:sources AS jsonb))"
    )
    with self.engine.begin() as connection:
        connection.execute(
            delete_statement,
            {
                "mapping_date": mapping_date,
            },
        )
        for start in range(0, len(rows), self.WRITE_BATCH_ROWS):
            batch = rows[start:start + self.WRITE_BATCH_ROWS]
            connection.execute(insert_statement, batch)
    return len(rows)

summarise

summarise(rows)

Counts the decisions by segment and status.

Parameters:

Name	Type	Description	Default
rows	list	The decisions, as decisions returns them.	required

Returns:

Type	Description
Counter	(segment, status) to the number of derivatives.

Example usage

From example_1_deciding_and_writing.py

summary = self.resolver.summarise(decided)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def summarise(self, rows):
    """
    Counts the decisions by segment and status.

    Args:
        rows (list): The decisions, as `decisions` returns them.

    Returns:
        collections.Counter: `(segment, status)` to the number of derivatives.
    """
    summary = collections.Counter()
    for row in rows:
        summary[(row["segment"], row["status"])] += 1
    return summary

run

run(mapping_date, dry_run=False)

Decides every live derivative's underlying for the date, and writes the decisions unless this is a dry run.

Parameters:

Name	Type	Description	Default
mapping_date	date	The mapping date.	required
dry_run	bool	True to decide without writing anything.	False

Returns:

Type	Description
Counter	(segment, status) to the number of derivatives.

Example usage

From example_2_a_dry_run.py

summary = self.resolver.run(self.mapping_date, dry_run=True)

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def run(self, mapping_date, dry_run=False):
    """
    Decides every live derivative's underlying for the date, and writes the decisions unless this is a dry run.

    Args:
        mapping_date (datetime.date): The mapping date.
        dry_run (bool): True to decide without writing anything.

    Returns:
        collections.Counter: `(segment, status)` to the number of derivatives.
    """
    rows = self.decisions(mapping_date)
    if not dry_run:
        self.write(mapping_date, rows)
    return self.summarise(rows)

UnderlyingCommand

The command line: decide a date's underlyings and print how many derivatives each segment has in each status.

Example programs

A malformed dateA dry run for a date

examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingCommand/example_1_a_malformed_date.py

"""Runs the underlyings command line with a date written the wrong way round, and shows that it stops before touching the database.

`UnderlyingCommand.run` reads its arguments from `sys.argv`, as `python -m stock_brokers.instruments.mapping.utilities.underlyings --date 2026-09-28` would pass them. A `--date` that is not `YYYY-MM-DD` is refused with a message and exit code 2, which the project uses for a bad argument, and no resolver or database engine is built.

The program sets `sys.argv` itself to the date `28-09-2026`, so it needs no database and no shell arguments.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingCommand/example_1_a_malformed_date.py
"""

import sys

from stock_brokers.instruments.mapping.utilities.underlyings import (
    UnderlyingCommand,
)


class MalformedDateExample:
    """Runs the command with a day-first date and prints its exit code.

    Attributes:
        command (UnderlyingCommand): The command being shown.
    """

    def __init__(self):
        """Builds the command and sets the arguments it will read.

        Returns:
            None: This method returns nothing.
        """
        sys.argv = [
            'underlyings',
            '--date',
            '28-09-2026',
            '--dry-run',
        ]
        self.command = UnderlyingCommand()

    def run(self):
        """Runs the command and prints the exit code it returns.

        Returns:
            None: This method returns nothing.
        """
        exit_code = self.command.run()
        print(f'Exit code: {exit_code}')


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

It prints:

--date needs a date as YYYY-MM-DD, not '28-09-2026'.
Exit code: 2

examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingCommand/example_2_a_dry_run_for_a_date.py

"""Runs the underlyings command line as a dry run for one date, against a stand-in database.

`UnderlyingCommand.run` reads `--date` and `--dry-run` from `sys.argv`, builds an `UnderlyingResolver` over the configured PostgreSQL engine, decides every live derivative's underlying and prints how many derivatives each segment has in each status. With `--dry-run` nothing is written to `unified.underlyings`, which makes it the safe way to look at a date's decisions first.

The command builds its engine through `get_postgres_engine`, so this program replaces that function in the `underlyings` module with one returning an in-memory stand-in engine. That keeps the real command and the real resolver running while no database is reached. The stand-in holds a BSE SENSEX50 future whose code leads to the index stored as "SNSX50", a BANKNIFTY option that resolves to the index, and a TCS future no broker gives a code for; the codes are made up.

Run it from the project root:

    python examples/stock_brokers/instruments/mapping/utilities/underlyings/UnderlyingCommand/example_2_a_dry_run_for_a_date.py
"""

import sys
import types

from stock_brokers.instruments.mapping.utilities import underlyings
from stock_brokers.instruments.mapping.utilities.underlyings import (
    UnderlyingCommand,
)


class StandInConnection:
    """A stand-in database connection that answers the resolver's three queries from rows in memory and records writes.

    Attributes:
        database (StandInDatabase): The rows and the record of writes.
    """

    def __init__(self, database):
        """Holds the stand-in database.

        Args:
            database (StandInDatabase): The rows and the record of writes.

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

    def __enter__(self):
        """Opens the connection.

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

    def __exit__(self, error_type, error, traceback):
        """Closes the connection.

        Args:
            error_type (type | None): The exception type raised inside the block, if any.
            error (BaseException | None): The exception raised inside the block, if any.
            traceback (types.TracebackType | None): The traceback, if any.

        Returns:
            bool: False, so an exception is never swallowed.
        """
        return False

    def execute(self, statement, parameters):
        """Answers one query from the rows in memory, or records one write.

        Args:
            statement (sqlalchemy.sql.elements.TextClause): The statement.
            parameters (dict | list): The statement's parameters, or a list of them for a batch insert.

        Returns:
            list: The rows the query returns, empty for a write.
        """
        sql = str(statement)
        if sql.startswith('DELETE'):
            self.database.deleted_dates.append(parameters['mapping_date'])
            return []
        if sql.startswith('INSERT'):
            for row in parameters:
                self.database.written_rows.append(row)
            return []
        if sql.startswith('SELECT m.instrument_id, m.broker'):
            return self.database.code_rows
        if 'CASE WHEN' in sql:
            return self.database.candidate_rows
        return self.database.derivative_rows


class StandInDatabase:
    """The rows the stand-in answers with, and the writes it has received.

    Attributes:
        derivative_rows (list): The live derivatives, as rows with `instrument_id`, `exchange` and `segment`.
        code_rows (list): The codes brokers give, as rows with `instrument_id`, `broker` and `code`.
        candidate_rows (list): The candidates, as rows with `instrument_id`, `exchange`, `segment` and `code`.
        deleted_dates (list): The mapping dates whose decisions were deleted.
        written_rows (list): The decision rows inserted.
    """

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

        Args:
            derivative_rows (list): The live derivatives.
            code_rows (list): The codes brokers give.
            candidate_rows (list): The candidates.

        Returns:
            None: This method returns nothing.
        """
        self.derivative_rows = derivative_rows
        self.code_rows = code_rows
        self.candidate_rows = candidate_rows
        self.deleted_dates = []
        self.written_rows = []


class StandInEngine:
    """A stand-in SQLAlchemy engine whose connections all share one stand-in database.

    Attributes:
        database (StandInDatabase): The shared rows.
    """

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

        Args:
            database (StandInDatabase): The shared rows.

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

    def connect(self):
        """Opens a connection for reading.

        Returns:
            StandInConnection: A connection over the shared rows.
        """
        return StandInConnection(self.database)

    def begin(self):
        """Opens a connection inside a transaction.

        Returns:
            StandInConnection: A connection over the shared rows.
        """
        return StandInConnection(self.database)


class Rows:
    """Builds the three kinds of stand-in row."""

    def derivative(self, instrument_id, exchange, segment):
        """Builds one live derivative row.

        Args:
            instrument_id (str): The derivative's instrument id.
            exchange (str): Its exchange.
            segment (str): Its exchange-prefixed segment.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            exchange=exchange,
            segment=segment,
        )

    def code(self, instrument_id, broker, code):
        """Builds one row of the code a broker gives for a derivative's underlying.

        Args:
            instrument_id (str): The derivative's instrument id.
            broker (str): The broker.
            code (str): The exchange code it gives.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            broker=broker,
            code=code,
        )

    def candidate(self, instrument_id, exchange, segment, code):
        """Builds one candidate row.

        Args:
            instrument_id (str): The candidate's instrument id.
            exchange (str): Its exchange.
            segment (str): Its exchange-prefixed segment.
            code (str): Its exchange code.

        Returns:
            types.SimpleNamespace: The row.
        """
        return types.SimpleNamespace(
            instrument_id=instrument_id,
            exchange=exchange,
            segment=segment,
            code=code,
        )


class DryRunForADateExample:
    """Runs the command as a dry run for 2026-09-28 and prints its exit code and how much it wrote.

    Attributes:
        database (StandInDatabase): The stand-in rows.
        command (UnderlyingCommand): The command being shown.
    """

    def __init__(self):
        """Builds the stand-in rows, points the command's engine at them and sets the arguments.

        Returns:
            None: This method returns nothing.
        """
        rows = Rows()
        self.database = StandInDatabase(
            [
                rows.derivative('sensex50-future-oct', 'bse', 'bse_equity_index_futures'),
                rows.derivative('banknifty-option-55000-ce', 'nse', 'nse_equity_index_options'),
                rows.derivative('tcs-future-oct', 'nse', 'nse_equity_futures'),
            ],
            [
                rows.code('sensex50-future-oct', 'fyers', '12'),
                rows.code('banknifty-option-55000-ce', 'fyers', '26009'),
            ],
            [
                rows.candidate('snsx50-index', 'bse', 'bse_equity_indices', '12'),
                rows.candidate('nifty-bank-index', 'nse', 'nse_equity_indices', '26009'),
            ],
        )
        underlyings.get_postgres_engine = self.stand_in_engine
        sys.argv = [
            'underlyings',
            '--date',
            '2026-09-28',
            '--dry-run',
        ]
        self.command = UnderlyingCommand()

    def stand_in_engine(self):
        """Builds the stand-in engine the command uses in place of the configured one.

        Returns:
            StandInEngine: An engine over the stand-in rows.
        """
        return StandInEngine(self.database)

    def run(self):
        """Runs the command and prints its exit code and how many rows it wrote.

        Returns:
            None: This method returns nothing.
        """
        exit_code = self.command.run()
        print(f'Exit code: {exit_code}')
        print(f'Rows written: {len(self.database.written_rows)}')


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

It prints:

underlyings for 2026-09-28, not written, a dry run
  segment                          status       derivatives
  bse_equity_index_futures         resolved               1
  nse_equity_futures               no_code                1
  nse_equity_index_options         resolved               1
Exit code: 0
Rows written: 0

Methods:

Name	Description
run	Parses the arguments, runs the resolver and prints the summary.

run

run()

Parses the arguments, runs the resolver and prints the summary.

Returns:

Type	Description
int	The exit code: 0 on success, 2 for a malformed date.

Example usage

From example_1_a_malformed_date.py

exit_code = self.command.run()

From example_2_a_dry_run_for_a_date.py

exit_code = self.command.run()

Source code in stock_brokers/instruments/mapping/utilities/underlyings.py

def run(self):
    """
    Parses the arguments, runs the resolver and prints the summary.

    Returns:
        int: The exit code: 0 on success, 2 for a malformed date.
    """
    parser = argparse.ArgumentParser(
        description="Decide which instrument every live future and option is written on, for a mapping date.",
    )
    parser.add_argument(
        "--date",
        help="the mapping date, as YYYY-MM-DD. Today when omitted.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="decide and print the summary without writing unified.underlyings",
    )
    arguments = parser.parse_args()
    mapping_date = datetime.date.today()
    if arguments.date:
        try:
            mapping_date = datetime.date.fromisoformat(arguments.date)
        except ValueError:
            print(f"--date needs a date as YYYY-MM-DD, not {arguments.date!r}.")
            return 2
    summary = UnderlyingResolver().run(mapping_date, dry_run=arguments.dry_run)
    written = "not written, a dry run" if arguments.dry_run else "written to unified.underlyings"
    print(f"underlyings for {mapping_date}, {written}")
    print(f"  {'segment':<32} {'status':<12} {'derivatives':>11}")
    for segment, status in sorted(summary):
        print(f"  {segment:<32} {status:<12} {summary[(segment, status)]:>11}")
    return 0