loader

Copy the brokers' stored bars into unified.price_history.

One run, for one interval, goes:

1. Discover. Every series with bars in each source broker's price_history_progress.
2. Resolve. Each series to its instrument, through resolution, and each (series, instrument) to a row in unified.price_history_sources, kept when the source policy allows that broker for that instrument's segment. A row whose series now resolves elsewhere is marked superseded.
3. Choose work. An instrument is rebuilt when one of its sources is new, superseded, or has bars its source row has not seen: the broker's earliest bar moved earlier, or its latest later.
4. Stitch. For each instrument, the primary broker's series are laid over each other by time. Where two have the same bar, the series with the most recent bars wins, since that is the token the instrument trades under now. INDIAGLYCO's EQ series owns its days up to 2026-09-01 and its BE series from 2026-09-02; ORICONENT's BE series fills the six weeks of 2026 when EQ had no bars.
5. Gap-fill. A gap-fill broker supplies only the trading days the primary has no bar for, and only when its bars agree with the primary's on the days both have.
6. Filter. Bars on days the exchange did not trade, off the interval's grid, or with a high below the open or close or a low above them, are dropped and counted.
7. Write. Upserted, with a row only rewritten when a value actually differs so that a re-run does not decompress chunks for nothing; bars no source provides any more are deleted; and the source rows' watermarks moved - all in one transaction per instrument.

An instrument is rebuilt over its whole history for daily bars, which is at most a few thousand rows. Nothing is ever averaged between brokers.

Classes:

Name	Description
UnifiedLoader	Loads one interval of bars into unified.price_history.
IntradayLoader	Loads intraday bars, set-based in SQL.

Functions:

Name	Description
corrected_bar	A bar with a correction applied.

UnifiedLoader

UnifiedLoader(interval, instrument_ids=None, resolver=None, calendar=None)

Loads one interval of bars into unified.price_history.

Attributes:

Name	Type	Description
interval	str	The stored interval name being loaded.
connection		The psycopg2 connection everything is read and written through.
resolver	SeriesResolver	Finds each series' instrument.
calendar	TradingCalendar	Trading days per exchange.
counts	Counter	What the run did, for its summary.

Prepare a load.

Parameters:

Name	Type	Description	Default
interval	str	The stored interval name, for example "day".	required
instrument_ids	set[str] | None	Restrict the load to these instruments, or None for all.	None
resolver	SeriesResolver | None	A resolver to share, or None to build one.	None
calendar	TradingCalendar | None	A calendar to share, or None to build one.	None

Returns:

Type	Description
None	This function returns nothing.

Example programs

Loading daily barsEach step by hand

examples/stock_brokers/instruments/historical/utilities/unified/loader/UnifiedLoader/example_1_loading_daily_bars.py

"""Loads INDIAGLYCO's daily bars into the unified price history from Flattrade's EQ and BE series, filtering the bars that cannot be right.

`UnifiedLoader.run` does one interval's whole load: it finds every series with bars in each source broker's progress table, resolves each series to its instrument, records the pairing in `unified.price_history_sources`, and rebuilds every instrument one of whose sources is new or has grown. A rebuild lays the primary broker's series over each other with the most recent series on top: INDIAGLYCO traded on its EQ token until 2026-09-01 and on its BE token from that day, so both have a bar on 2026-09-01 and the BE series wins it. Bars on a day the exchange did not trade, off the day grid, or with a high below the open or close are dropped and counted, a bar already stored that no source provides any more is deleted, and each source row's watermarks are moved, all in one transaction.

The loader opens PostgreSQL through `get_postgres` in its constructor, and the example runner points PostgreSQL at a closed port, so the program replaces `get_postgres` in the loader module with a stand-in returning an in-memory connection. It answers the loader's queries with recorded-looking rows for two weeks of Flattrade bars and records every row written. The resolver and the trading calendar are passed in, as the constructor allows: a small stand-in resolver that knows both series, and a real `TradingCalendar` whose days are filled in by hand so it reads nothing. What to notice: the Saturday bar, the bar stamped at 09:15 and the bar with an impossible range are each dropped; the BE series' close of 2026-09-01 is the one kept; and the bar of 2026-08-21, which no source provides, is deleted.

Run it from the project root:

    python examples/stock_brokers/instruments/historical/utilities/unified/loader/UnifiedLoader/example_1_loading_daily_bars.py
"""

import datetime
import decimal
import zoneinfo

from stock_brokers.instruments.historical.utilities.unified import loader
from stock_brokers.instruments.historical.utilities.unified.calendar import (
    TradingCalendar,
)
from stock_brokers.instruments.historical.utilities.unified.loader import (
    UnifiedLoader,
)
from stock_brokers.instruments.historical.utilities.unified.resolution import (
    Resolution,
)

INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
EQ_SERIES = 'NSE|7000|INDIAGLYCO-EQ'
BE_SERIES = 'NSE|7001|INDIAGLYCO-BE'


class StandInSeriesResolver:
    """A stand-in for `SeriesResolver` that resolves INDIAGLYCO's two Flattrade series."""

    def resolve(self, broker, identifiers):
        """Resolves each identifier to INDIAGLYCO on NSE.

        Args:
            broker (str): The broker.
            identifiers (list): The broker's series identifiers.

        Returns:
            dict: Identifier to a list holding one Resolution.
        """
        resolved = {}
        for identifier in identifiers:
            resolution = Resolution(
                identifier,
                'INDIAGLYCO-NSE',
                'nse',
                'nse_equities',
                'INDIAGLYCO',
                'broker_mapping',
                'active',
                None,
                datetime.date(2026, 8, 12),
                datetime.date(2026, 9, 29),
                None,
                None,
            )
            resolved[identifier] = [
                resolution,
            ]
        return resolved


class StandInCursor:
    """A stand-in for a psycopg2 cursor that hands every statement to its connection.

    Attributes:
        connection (StandInConnection): The connection that answers the statements.
        result (list): The rows the last statement produced.
        rowcount (int): How many rows the last statement changed.
    """

    def __init__(self, connection):
        """Builds a cursor over the stand-in connection.

        Args:
            connection (StandInConnection): The connection that answers the statements.

        Returns:
            None: This method returns nothing.
        """
        self.connection = connection
        self.result = []
        self.rowcount = 0

    def __enter__(self):
        """Opens the cursor as a context manager.

        Returns:
            StandInCursor: This cursor.
        """
        return self

    def __exit__(self, exception_type, exception, traceback):
        """Closes the cursor, letting any exception carry on.

        Args:
            exception_type (type | None): The class of an exception raised inside the block.
            exception (BaseException | None): The exception raised inside the block.
            traceback (types.TracebackType | None): Where it was raised.

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

    def mogrify(self, template, arguments):
        """Stages one row of a multi-row insert, as `execute_values` asks a cursor to.

        Args:
            template (bytes): The row template.
            arguments (tuple): The row's values.

        Returns:
            bytes: A placeholder for the rendered row.
        """
        self.connection.staged_rows.append(arguments)
        return b'(...)'

    def execute(self, statement, parameters=None):
        """Runs one statement against the stand-in connection.

        Args:
            statement (str | bytes): The SQL text.
            parameters (tuple | list | None): The statement's parameters.

        Returns:
            None: This method returns nothing.
        """
        if isinstance(statement, bytes):
            statement = statement.decode()
        self.result, self.rowcount = self.connection.answer(statement, parameters)

    def fetchone(self):
        """The first row the last statement produced.

        Returns:
            tuple | None: The row, or None when there was none.
        """
        if not self.result:
            return None
        return self.result[0]

    def fetchall(self):
        """Every row the last statement produced.

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


class StandInConnection:
    """A stand-in for a psycopg2 connection over Flattrade's INDIAGLYCO bars.

    Attributes:
        encoding (str): The client encoding, which `execute_values` reads.
        bars (dict): Series identifier to its (time, open, high, low, close, volume, oi) rows.
        staged_rows (list): Rows staged by `mogrify` for the next insert.
        written (list): Every row inserted into the unified price history.
        deleted (list): The bar times deleted from it.
        source_updates (list): (source id, owned from, owned to) for each source row updated.
        next_source_id (int): The id the next new source row gets.
    """

    encoding = 'UTF8'

    def __init__(self):
        """Fills in the two series' bars.

        Returns:
            None: This method returns nothing.
        """
        self.bars = {
            EQ_SERIES: [],
            BE_SERIES: [],
        }
        self.staged_rows = []
        self.written = []
        self.deleted = []
        self.source_updates = []
        self.next_source_id = 11
        self.add_bar(EQ_SERIES, datetime.datetime(2026, 8, 27, tzinfo=INDIA), '480.00')
        self.add_bar(EQ_SERIES, datetime.datetime(2026, 8, 28, tzinfo=INDIA), '482.50')
        self.add_bar(EQ_SERIES, datetime.datetime(2026, 8, 29, tzinfo=INDIA), '482.50')
        self.add_bar(EQ_SERIES, datetime.datetime(2026, 8, 31, tzinfo=INDIA), '485.00')
        self.add_bar(EQ_SERIES, datetime.datetime(2026, 9, 1, tzinfo=INDIA), '484.00')
        self.add_bar(BE_SERIES, datetime.datetime(2026, 9, 1, tzinfo=INDIA), '484.20')
        self.add_bar(BE_SERIES, datetime.datetime(2026, 9, 2, tzinfo=INDIA), '242.50')
        self.add_bar(BE_SERIES, datetime.datetime(2026, 9, 3, 9, 15, tzinfo=INDIA), '243.00')
        self.add_bar(BE_SERIES, datetime.datetime(2026, 9, 3, tzinfo=INDIA), '244.00')
        self.add_bar(BE_SERIES, datetime.datetime(2026, 9, 4, tzinfo=INDIA), '0')

    def add_bar(self, series, moment, close_text):
        """Adds one daily bar whose open, high and low sit around its close.

        A close of 0 stands for a bar whose high was served below its open.

        Args:
            series (str): The series identifier.
            moment (datetime.datetime): The bar time.
            close_text (str): The close, as text so it becomes an exact Decimal.

        Returns:
            None: This method returns nothing.
        """
        close = decimal.Decimal(close_text)
        if close == 0:
            row = self.row(moment, decimal.Decimal('245.00'), decimal.Decimal('244.00'), decimal.Decimal('243.00'), decimal.Decimal('244.50'), 90000, None)
        else:
            row = self.row(moment, close - 1, close + 2, close - 3, close, 120000, None)
        self.bars[series].append(row)

    def row(self, *columns):
        """Builds one result row.

        Args:
            *columns (object): The row's columns, in order.

        Returns:
            tuple: The row.
        """
        return columns

    def cursor(self):
        """Opens a stand-in cursor.

        Returns:
            StandInCursor: The cursor.
        """
        return StandInCursor(self)

    def answer(self, statement, parameters):
        """Answers one of the loader's statements.

        Args:
            statement (str): The SQL text.
            parameters (tuple | list | None): The statement's parameters.

        Returns:
            tuple: (rows, rows changed).
        """
        if 'from unified.price_history_sources where' in statement:
            return [], 0
        if 'price_history_progress' in statement:
            if 'flattrade.' not in statement:
                return [], 0
            return [
                self.row(EQ_SERIES, self.bars[EQ_SERIES][0][0], self.bars[EQ_SERIES][-1][0]),
                self.row(BE_SERIES, self.bars[BE_SERIES][0][0], self.bars[BE_SERIES][-1][0]),
            ], 0
        if 'insert into unified.price_history_sources' in statement:
            source_id = self.next_source_id
            self.next_source_id += 1
            return [
                self.row(source_id),
            ], 1
        if 'from flattrade.price_history' in statement:
            return self.bars[parameters[0]], 0
        if 'from unified.correction_ranges' in statement:
            return [], 0
        if 'select "time" from unified.price_history' in statement:
            return [
                self.row(datetime.datetime(2026, 8, 21, tzinfo=INDIA)),
                self.row(datetime.datetime(2026, 8, 27, tzinfo=INDIA)),
            ], 0
        if 'delete from unified.price_history' in statement:
            self.deleted.extend(parameters[2])
            return [], len(parameters[2])
        if 'insert into unified.price_history' in statement:
            self.written.extend(self.staged_rows)
            changed = len(self.staged_rows)
            self.staged_rows = []
            return [], changed
        if 'update unified.price_history_sources set' in statement:
            self.source_updates.append(self.row(parameters[6], parameters[2], parameters[3]))
            return [], 1
        raise ValueError(f'The stand-in does not know this statement: {statement[:60]}')

    def commit(self):
        """Accepts a commit.

        Returns:
            None: This method returns nothing.
        """

    def rollback(self):
        """Accepts a rollback.

        Returns:
            None: This method returns nothing.
        """


class StandInPostgres:
    """Hands out the stand-in connection in the place of `get_postgres`.

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

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

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

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

    def get_postgres(self):
        """Returns the stand-in connection.

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


class LoadingDailyBarsExample:
    """Runs a daily load for INDIAGLYCO and prints what was written, deleted and counted.

    Attributes:
        connection (StandInConnection): The in-memory connection.
        loader (UnifiedLoader): The loader being shown.
    """

    def __init__(self):
        """Installs the stand-in connection and builds the loader with a hand-filled calendar.

        Returns:
            None: This method returns nothing.
        """
        self.connection = StandInConnection()
        loader.get_postgres = StandInPostgres(self.connection).get_postgres
        calendar = TradingCalendar()
        calendar.days['nse'] = set()
        day = datetime.date(2026, 8, 17)
        while day <= datetime.date(2026, 9, 11):
            if day.weekday() < 5:
                calendar.days['nse'].add(day)
            day += datetime.timedelta(days=1)
        self.loader = UnifiedLoader('day', resolver=StandInSeriesResolver(), calendar=calendar)

    def run(self):
        """Runs the load and prints the outcome.

        Returns:
            None: This method returns nothing.
        """
        counts = self.loader.run()
        for name in sorted(counts):
            print(f'{name}: {counts[name]}')
        print('Rows written (time, close, source id):')
        for row in self.connection.written:
            print(f'  {row[0].date()} {row[6]} {row[9]}')
        deleted = []
        for moment in self.connection.deleted:
            deleted.append(moment.date().isoformat())
        print(f'Deleted: {deleted}')
        for source_id, owned_from, owned_to in self.connection.source_updates:
            print(f'Source {source_id} owns {owned_from.date()} to {owned_to.date()}')


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

It prints:

bars_deleted: 1
bars_kept: 6
bars_written: 6
filtered_bad_range: 1
filtered_non_trading_day: 1
filtered_off_grid: 1
instruments_rebuilt: 1
series_dhan: 0
series_flattrade: 2
series_zerodha: 0
sources_superseded: 0
Rows written (time, close, source id):
  2026-08-27 480.00 11
  2026-08-28 482.50 11
  2026-08-31 485.00 11
  2026-09-01 484.20 12
  2026-09-02 242.50 12
  2026-09-03 244.00 12
Deleted: ['2026-08-21']
Source 11 owns 2026-08-27 to 2026-08-31
Source 12 owns 2026-09-01 to 2026-09-03

examples/stock_brokers/instruments/historical/utilities/unified/loader/UnifiedLoader/example_2_each_step_by_hand.py

"""Calls each step of `UnifiedLoader` by hand: refreshing the sources table, reading a corrected series, checking bars, and gap-filling an index from a second broker.

`run` strings the loader's steps together; this program calls them one at a time so each answer is visible.

- `refresh_sources` resolves every series in the source brokers' progress tables. A series that now resolves to a different instrument than its active source row says gets a new row, the old row is marked superseded, and the instrument it no longer feeds is returned as a placeholder so its bars are rebuilt; a series that resolves to nothing is kept in `unresolved`. `upsert_source` writes one source row and returns its id.
- `read_series` reads one source's bars and applies any confirmed correction: PIDILITIND's NSE bars before 2025-09-23 were served at half their price, and come back doubled, rounded to the paisa, with the volume untouched.
- `valid` passes or drops single bars, counting why.
- `stitch` lays one broker's series over each other, and `agreement` measures how often two brokers' closes agree on the days both have.
- `rebuild` builds NIFTY 50's bars: Zerodha is the primary and lacks 2026-09-10; Dhan agrees with Zerodha on all 22 shared days and fills that day; Flattrade's closes are 1% away and it is refused as a gap-fill broker.

The loader opens PostgreSQL through `get_postgres` in its constructor, which the example runner points at a closed port, so the program replaces `get_postgres` in the loader module with a stand-in returning an in-memory connection that answers the loader's queries with canned rows and records what is written. The resolver is a small stand-in passed to the constructor, and the calendar is a real `TradingCalendar` whose days are filled in by hand.

Run it from the project root:

    python examples/stock_brokers/instruments/historical/utilities/unified/loader/UnifiedLoader/example_2_each_step_by_hand.py
"""

import datetime
import decimal
import zoneinfo

from stock_brokers.instruments.historical.utilities.unified import loader
from stock_brokers.instruments.historical.utilities.unified.calendar import (
    TradingCalendar,
)
from stock_brokers.instruments.historical.utilities.unified.loader import (
    Bar,
    Source,
    UnifiedLoader,
)
from stock_brokers.instruments.historical.utilities.unified.resolution import (
    Resolution,
)

INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
MISSING_DAY = datetime.date(2026, 9, 10)
CORRECTION_DATE = datetime.datetime(2025, 9, 23, tzinfo=INDIA)


class StandInSeriesResolver:
    """A stand-in for `SeriesResolver` that resolves a renamed series to its new instrument and cannot place another."""

    def resolve(self, broker, identifiers):
        """Resolves each identifier, or reports it rejected.

        Args:
            broker (str): The broker.
            identifiers (list): The broker's series identifiers.

        Returns:
            dict: Identifier to a list holding one Resolution.
        """
        resolved = {}
        for identifier in identifiers:
            if identifier == 'NSE|9999|NEWCO-EQ':
                resolution = Resolution(
                    identifier,
                    'NEWCO-NSE',
                    'nse',
                    'nse_equities',
                    'NEWCO',
                    'broker_mapping',
                    'active',
                    None,
                    datetime.date(2026, 9, 3),
                    datetime.date(2026, 9, 29),
                    None,
                    None,
                )
            else:
                resolution = Resolution(
                    identifier,
                    None,
                    None,
                    None,
                    None,
                    None,
                    'rejected',
                    'no mapping, exchange token or symbol match',
                    None,
                    None,
                    None,
                    None,
                )
            resolved[identifier] = [
                resolution,
            ]
        return resolved


class StandInCursor:
    """A stand-in for a psycopg2 cursor that hands every statement to its connection.

    Attributes:
        connection (StandInConnection): The connection that answers the statements.
        result (list): The rows the last statement produced.
        rowcount (int): How many rows the last statement changed.
    """

    def __init__(self, connection):
        """Builds a cursor over the stand-in connection.

        Args:
            connection (StandInConnection): The connection that answers the statements.

        Returns:
            None: This method returns nothing.
        """
        self.connection = connection
        self.result = []
        self.rowcount = 0

    def __enter__(self):
        """Opens the cursor as a context manager.

        Returns:
            StandInCursor: This cursor.
        """
        return self

    def __exit__(self, exception_type, exception, traceback):
        """Closes the cursor, letting any exception carry on.

        Args:
            exception_type (type | None): The class of an exception raised inside the block.
            exception (BaseException | None): The exception raised inside the block.
            traceback (types.TracebackType | None): Where it was raised.

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

    def mogrify(self, template, arguments):
        """Stages one row of a multi-row insert, as `execute_values` asks a cursor to.

        Args:
            template (bytes): The row template.
            arguments (tuple): The row's values.

        Returns:
            bytes: A placeholder for the rendered row.
        """
        self.connection.staged_rows.append(arguments)
        return b'(...)'

    def execute(self, statement, parameters=None):
        """Runs one statement against the stand-in connection.

        Args:
            statement (str | bytes): The SQL text.
            parameters (tuple | list | None): The statement's parameters.

        Returns:
            None: This method returns nothing.
        """
        if isinstance(statement, bytes):
            statement = statement.decode()
        self.result, self.rowcount = self.connection.answer(statement, parameters)

    def fetchone(self):
        """The first row the last statement produced.

        Returns:
            tuple | None: The row, or None when there was none.
        """
        if not self.result:
            return None
        return self.result[0]

    def fetchall(self):
        """Every row the last statement produced.

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


class StandInConnection:
    """A stand-in for a psycopg2 connection over a few brokers' stored bars.

    Attributes:
        encoding (str): The client encoding, which `execute_values` reads.
        bars (dict): (broker, series identifier) to its (time, open, high, low, close, volume, oi) rows.
        staged_rows (list): Rows staged by `mogrify` for the next insert.
        written (list): Every row inserted into the unified price history.
        superseded (list): The source ids marked superseded.
    """

    encoding = 'UTF8'

    def __init__(self):
        """Fills in NIFTY 50's bars at three brokers and PIDILITIND's at Flattrade.

        Returns:
            None: This method returns nothing.
        """
        self.bars = {
            ('zerodha', '256265'): [],
            ('dhan', '13|IDX_I|NIFTY'): [],
            ('flattrade', 'NSE|26000|Nifty 50'): [],
            ('flattrade', 'NSE|10575|PIDILITIND-EQ'): [],
        }
        self.staged_rows = []
        self.written = []
        self.superseded = []
        day = datetime.date(2026, 8, 31)
        position = 0
        while day <= datetime.date(2026, 9, 30):
            if day.weekday() < 5:
                moment = datetime.datetime(day.year, day.month, day.day, tzinfo=INDIA)
                close = decimal.Decimal(25000 + position * 10)
                if day != MISSING_DAY:
                    self.bars[('zerodha', '256265')].append(self.bar(moment, close))
                self.bars[('dhan', '13|IDX_I|NIFTY')].append(self.bar(moment, close))
                self.bars[('flattrade', 'NSE|26000|Nifty 50')].append(self.bar(moment, close * decimal.Decimal('1.01')))
                position += 1
            day += datetime.timedelta(days=1)
        self.bars[('flattrade', 'NSE|10575|PIDILITIND-EQ')].append(self.bar(datetime.datetime(2025, 9, 22, tzinfo=INDIA), decimal.Decimal('1519.05')))
        self.bars[('flattrade', 'NSE|10575|PIDILITIND-EQ')].append(self.bar(CORRECTION_DATE, decimal.Decimal('1523.00')))

    def bar(self, moment, close):
        """Builds one daily bar whose open, high and low sit around its close.

        Args:
            moment (datetime.datetime): The bar time.
            close (decimal.Decimal): The close.

        Returns:
            tuple: The bar's columns.
        """
        return self.row(moment, close - 5, close + 10, close - 12, close, 250000, None)

    def row(self, *columns):
        """Builds one result row.

        Args:
            *columns (object): The row's columns, in order.

        Returns:
            tuple: The row.
        """
        return columns

    def cursor(self):
        """Opens a stand-in cursor.

        Returns:
            StandInCursor: The cursor.
        """
        return StandInCursor(self)

    def answer(self, statement, parameters):
        """Answers one of the loader's statements.

        Args:
            statement (str): The SQL text.
            parameters (tuple | list | None): The statement's parameters.

        Returns:
            tuple: (rows, rows changed).
        """
        if 'from unified.price_history_sources where' in statement:
            return [
                self.row(7, 'flattrade', 'NSE|9999|NEWCO-EQ', 'OLDCO-NSE', 'active', None, None, None, None),
            ], 0
        if 'price_history_progress' in statement:
            if 'flattrade.' not in statement:
                return [], 0
            return [
                self.row('NSE|9999|NEWCO-EQ', datetime.datetime(2026, 9, 3, tzinfo=INDIA), datetime.datetime(2026, 9, 29, tzinfo=INDIA)),
                self.row('NSE|8888|MYSTERY-EQ', datetime.datetime(2026, 9, 3, tzinfo=INDIA), datetime.datetime(2026, 9, 29, tzinfo=INDIA)),
            ], 0
        if 'insert into unified.price_history_sources' in statement:
            return [
                self.row(31),
            ], 1
        if "set status = 'superseded'" in statement:
            self.superseded.append(parameters[0])
            return [], 1
        brokers = [
            'zerodha',
            'dhan',
            'flattrade',
        ]
        for broker in brokers:
            if f'from {broker}.price_history' in statement:
                return self.bars[(broker, parameters[0])], 0
        if 'from unified.correction_ranges' in statement:
            if parameters[1] != 'NSE|10575|PIDILITIND-EQ':
                return [], 0
            return [
                self.row(datetime.datetime(2020, 1, 1, tzinfo=INDIA), CORRECTION_DATE, decimal.Decimal('2'), decimal.Decimal('1')),
            ], 0
        if 'select "time" from unified.price_history' in statement:
            return [], 0
        if 'insert into unified.price_history' in statement:
            self.written.extend(self.staged_rows)
            changed = len(self.staged_rows)
            self.staged_rows = []
            return [], changed
        if 'update unified.price_history_sources set' in statement:
            return [], 1
        raise ValueError(f'The stand-in does not know this statement: {statement[:60]}')

    def commit(self):
        """Accepts a commit.

        Returns:
            None: This method returns nothing.
        """

    def rollback(self):
        """Accepts a rollback.

        Returns:
            None: This method returns nothing.
        """


class StandInPostgres:
    """Hands out the stand-in connection in the place of `get_postgres`.

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

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

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

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

    def get_postgres(self):
        """Returns the stand-in connection.

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


class EachStepByHandExample:
    """Calls the loader's steps one at a time and prints each answer.

    Attributes:
        connection (StandInConnection): The in-memory connection.
        loader (UnifiedLoader): The loader being shown.
    """

    def __init__(self):
        """Installs the stand-in connection and builds the loader with a hand-filled calendar.

        Returns:
            None: This method returns nothing.
        """
        self.connection = StandInConnection()
        loader.get_postgres = StandInPostgres(self.connection).get_postgres
        calendar = TradingCalendar()
        calendar.days['nse'] = set()
        day = datetime.date(2025, 9, 1)
        while day <= datetime.date(2026, 9, 30):
            if day.weekday() < 5:
                calendar.days['nse'].add(day)
            day += datetime.timedelta(days=1)
        self.loader = UnifiedLoader('day', resolver=StandInSeriesResolver(), calendar=calendar)

    def source(self, source_id, broker, series, role):
        """Builds one of NIFTY 50's source rows.

        Args:
            source_id (int): The source row's id.
            broker (str): The broker.
            series (str): The broker's series identifier.
            role (str): "primary" or "gap_fill".

        Returns:
            Source: The source.
        """
        return Source(
            source_id,
            broker,
            series,
            'NIFTY50-NSE',
            role,
            'nse',
            None,
            None,
            datetime.datetime(2026, 8, 31, tzinfo=INDIA),
            datetime.datetime(2026, 9, 30, tzinfo=INDIA),
            True,
        )

    def check(self, label, bar):
        """Prints whether one bar passes the loader's checks.

        Args:
            label (str): What is odd about the bar.
            bar (Bar): The bar.

        Returns:
            None: This method returns nothing.
        """
        print(f'valid, {label}: {self.loader.valid(bar, "nse")}')

    def run(self):
        """Walks through each step and prints what it did.

        Returns:
            None: This method returns nothing.
        """
        sources = self.loader.refresh_sources()
        for source in sources:
            print(f'refresh_sources: {source.instrument_id} from {source.broker} {source.broker_series}, source {source.source_id}, changed {source.changed}')
        print(f'Superseded source rows: {self.connection.superseded}; unresolved: {self.loader.unresolved}')
        identifiers = [
            'NSE|9999|NEWCO-EQ',
        ]
        resolution = StandInSeriesResolver().resolve('flattrade', identifiers)['NSE|9999|NEWCO-EQ'][0]
        print(f"upsert_source returns id {self.loader.upsert_source('flattrade', 'NSE|9999|NEWCO-EQ', resolution, 'primary')}")
        corrected = Source(41, 'flattrade', 'NSE|10575|PIDILITIND-EQ', 'PIDILITIND-NSE', 'primary', 'nse', None, None, None, None, True)
        for bar in self.loader.read_series(corrected):
            print(f'read_series PIDILITIND {bar.time.date()}: close {bar.close}, high {bar.high}, volume {bar.volume}')
        moment = datetime.datetime(2026, 9, 15, tzinfo=INDIA)
        good = Bar(moment, decimal.Decimal('100'), decimal.Decimal('104'), decimal.Decimal('99'), decimal.Decimal('102'), 10, None)
        self.check('an ordinary bar', good)
        self.check('a zero close', good._replace(close=decimal.Decimal('0')))
        self.check('on a Sunday', good._replace(time=datetime.datetime(2026, 9, 13, tzinfo=INDIA)))
        self.check('stamped 09:15', good._replace(time=datetime.datetime(2026, 9, 15, 9, 15, tzinfo=INDIA)))
        self.check('high below the close', good._replace(high=decimal.Decimal('101')))
        zerodha = self.source(21, 'zerodha', '256265', 'primary')
        dhan = self.source(22, 'dhan', '13|IDX_I|NIFTY', 'gap_fill')
        flattrade = self.source(23, 'flattrade', 'NSE|26000|Nifty 50', 'gap_fill')
        primary_bars = self.loader.stitch([zerodha])
        print(f'stitch Zerodha: {len(primary_bars)} bars')
        print(f'agreement with Dhan: {self.loader.agreement(primary_bars, self.loader.stitch([dhan]))}')
        print(f'agreement with Flattrade: {self.loader.agreement(primary_bars, self.loader.stitch([flattrade]))}')
        members = [
            zerodha,
            dhan,
            flattrade,
        ]
        self.loader.rebuild('NIFTY50-NSE', members)
        filled = []
        for row in self.connection.written:
            if row[9] != 21:
                filled.append(f'{row[0].date()} from source {row[9]}')
        print(f'rebuild wrote {len(self.connection.written)} bars; filled from another broker: {filled}')
        for name in sorted(self.loader.counts):
            print(f'{name}: {self.loader.counts[name]}')

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

It prints:

refresh_sources: NEWCO-NSE from flattrade NSE|9999|NEWCO-EQ, source 31, changed True
refresh_sources: OLDCO-NSE from None None, source None, changed True
Superseded source rows: [7]; unresolved: [('flattrade', 'NSE|8888|MYSTERY-EQ', 'rejected', 'no mapping, exchange token or symbol match')]
upsert_source returns id 31
read_series PIDILITIND 2025-09-22: close 3038.10, high 3058.10, volume 250000
read_series PIDILITIND 2025-09-23: close 1523.00, high 1533.00, volume 250000
valid, an ordinary bar: True
valid, a zero close: False
valid, on a Sunday: False
valid, stamped 09:15: False
valid, high below the close: False
stitch Zerodha: 22 bars
agreement with Dhan: (1.0, 22)
agreement with Flattrade: (0.0, 22)
rebuild wrote 23 bars; filled from another broker: ['2026-09-10 from source 22']
bars_corrected: 1
bars_deleted: 0
bars_gap_filled: 1
bars_kept: 23
bars_written: 23
filtered_bad_range: 1
filtered_missing_price: 1
filtered_non_trading_day: 1
filtered_off_grid: 1
gap_fill_refused: 1
series_dhan: 0
series_flattrade: 2
series_zerodha: 0
sources_superseded: 1
unresolved_flattrade: 1

Methods:

Name	Description
run	Discover, resolve and load.
refresh_sources	Resolve every series of every source broker and bring the sources table up to date.
upsert_source	Insert or refresh one source row, returning its id.
read_series	One source's bars, restricted to the part of the series that belongs to its instrument, with
stitch	Lay one broker's series for an instrument over each other, the most recent series on top.
valid	Whether a bar passes the calendar, grid and range checks, counting why when it does not.
agreement	The share of shared bar times on which two brokers' closes agree.
rebuild	Rebuild one instrument's bars from its sources, in one transaction.

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def __init__(self, interval, instrument_ids=None, resolver=None, calendar=None):
    """
    Prepare a load.

    Args:
        interval (str): The stored interval name, for example "day".
        instrument_ids (set[str] | None): Restrict the load to these instruments, or None for all.
        resolver (SeriesResolver | None): A resolver to share, or None to build one.
        calendar (TradingCalendar | None): A calendar to share, or None to build one.

    Returns:
        None: This function returns nothing.
    """
    self.interval = interval
    self.instrument_ids = set(instrument_ids) if instrument_ids else None
    self.connection = get_postgres()
    self.resolver = resolver or SeriesResolver()
    self.calendar = calendar or TradingCalendar()
    self.counts = Counter()
    self.unresolved = []

run

run()

Discover, resolve and load.

Returns:

Type	Description
Counter	What was done - series seen, instruments rebuilt, bars written, deleted and filtered.

Example usage

From example_1_loading_daily_bars.py

counts = self.loader.run()

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def run(self):
    """
    Discover, resolve and load.

    Returns:
        Counter: What was done - series seen, instruments rebuilt, bars written, deleted and filtered.
    """
    sources = self.refresh_sources()
    by_instrument = {}
    for source in sources:
        by_instrument.setdefault(source.instrument_id, []).append(source)

    work = [instrument_id for instrument_id, members in by_instrument.items()
            if any(source.changed for source in members)]
    LOGGER.info("%s: %d instruments with sources, %d to rebuild",
                self.interval, len(by_instrument), len(work))
    for position, instrument_id in enumerate(sorted(work), 1):
        self.rebuild(instrument_id, by_instrument[instrument_id])
        if position % 500 == 0:
            LOGGER.info("%s: rebuilt %d of %d instruments", self.interval, position, len(work))
    self.counts["instruments_rebuilt"] += len(work)
    return self.counts

refresh_sources

refresh_sources()

Resolve every series of every source broker and bring the sources table up to date.

Returns:

Type	Description
list[Source]	The active sources, restricted to instrument_ids when given.

Example usage

From example_2_each_step_by_hand.py

sources = self.loader.refresh_sources()

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def refresh_sources(self):
    """
    Resolve every series of every source broker and bring the sources table up to date.

    Returns:
        list[Source]: The active sources, restricted to `instrument_ids` when given.
    """
    existing = {}
    with self.connection.cursor() as cursor:
        cursor.execute(f"""
            select source_id, broker, broker_series, instrument_id::text, status,
                   broker_earliest_seen, broker_latest_seen, owned_from, owned_to
            from {tables.PRICE_HISTORY_SOURCES} where "interval" = %s
        """, (self.interval,))
        for row in cursor.fetchall():
            existing[(row[1], row[2], row[3])] = row

    sources = []
    superseded = set()
    seen_keys = set()
    for broker in brokers_for_interval(self.interval):
        with self.connection.cursor() as cursor:
            cursor.execute(f"""
                select instrument_token, earliest_bar_time, latest_bar_time
                from {broker}.price_history_progress
                where "interval" = %s and bar_count > 0 and earliest_bar_time is not null
            """, (self.interval,))
            progress = {row[0]: (row[1], row[2]) for row in cursor.fetchall()}
        self.counts[f"series_{broker}"] = len(progress)

        resolutions = self.resolver.resolve(broker, list(progress))
        for identifier, found in resolutions.items():
            earliest, latest = progress[identifier]
            for resolution in found:
                if resolution.instrument_id is None:
                    self.counts[f"unresolved_{broker}"] += 1
                    self.unresolved.append((broker, identifier, resolution.status, resolution.status_reason))
                    continue
                role = dict(sources_for(resolution.exchange, resolution.segment, self.interval)).get(broker)
                if role is None:
                    continue
                key = (broker, identifier, resolution.instrument_id)
                seen_keys.add(key)
                if self.instrument_ids is not None and resolution.instrument_id not in self.instrument_ids:
                    continue
                previous = existing.get(key)
                changed = (previous is None or previous[4] != "active"
                           or previous[5] is None or earliest < previous[5]
                           or previous[6] is None or latest > previous[6])
                source_id = self.upsert_source(broker, identifier, resolution, role)
                sources.append(Source(source_id, broker, identifier, resolution.instrument_id, role,
                                      resolution.exchange, resolution.valid_from, resolution.valid_to,
                                      earliest, latest, changed))

    # A series that no longer resolves to an instrument it used to feed.
    for key, row in existing.items():
        if key in seen_keys or row[4] != "active":
            continue
        if self.instrument_ids is not None and key[2] not in self.instrument_ids:
            continue
        with self.connection.cursor() as cursor:
            cursor.execute(f"""
                update {tables.PRICE_HISTORY_SOURCES}
                set status = 'superseded', status_reason = 'series no longer resolves to this instrument'
                where source_id = %s
            """, (row[0],))
        superseded.add(key[2])
    self.connection.commit()
    self.counts["sources_superseded"] += len(superseded)

    # An instrument that lost a source is rebuilt from what it has left.
    if superseded:
        instruments_with_sources = {source.instrument_id for source in sources}
        for instrument_id in superseded:
            if instrument_id in instruments_with_sources:
                sources = [source._replace(changed=True) if source.instrument_id == instrument_id else source
                           for source in sources]
            else:
                sources.append(Source(None, None, None, instrument_id, None, None, None, None,
                                      None, None, True))
    return sources

upsert_source

upsert_source(broker, identifier, resolution, role)

Insert or refresh one source row, returning its id.

Parameters:

Name	Type	Description	Default
broker	str	The broker.	required
identifier	str	The broker's series identifier.	required
resolution	Resolution	The instrument it resolved to.	required
role	str	"primary" or "gap_fill".	required

Returns:

Type	Description
int	The source_id.

Example usage

From example_2_each_step_by_hand.py

print(f"upsert_source returns id {self.loader.upsert_source('flattrade', 'NSE|9999|NEWCO-EQ', resolution, 'primary')}")

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def upsert_source(self, broker, identifier, resolution, role):
    """
    Insert or refresh one source row, returning its id.

    Args:
        broker (str): The broker.
        identifier (str): The broker's series identifier.
        resolution (Resolution): The instrument it resolved to.
        role (str): "primary" or "gap_fill".

    Returns:
        int: The source_id.
    """
    with self.connection.cursor() as cursor:
        cursor.execute(f"""
            insert into {tables.PRICE_HISTORY_SOURCES}
                (broker, broker_series, "interval", instrument_id, resolved_by, mapping_first_date,
                 mapping_last_date, price_basis, role, status, status_reason)
            values (%s, %s, %s, %s, %s, %s, %s, %s, %s, 'active', %s)
            on conflict (broker, broker_series, "interval", instrument_id) do update set
                resolved_by = excluded.resolved_by,
                mapping_first_date = excluded.mapping_first_date,
                mapping_last_date = excluded.mapping_last_date,
                price_basis = excluded.price_basis,
                role = excluded.role,
                status = case when {tables.PRICE_HISTORY_SOURCES}.resolved_by = 'manual'
                              then {tables.PRICE_HISTORY_SOURCES}.status else 'active' end,
                status_reason = excluded.status_reason
            returning source_id
        """, (broker, identifier, self.interval, resolution.instrument_id, resolution.resolved_by,
              resolution.mapping_first_date, resolution.mapping_last_date,
              price_basis(resolution.segment), role, resolution.status_reason))
        return cursor.fetchone()[0]

read_series

read_series(source)

One source's bars, restricted to the part of the series that belongs to its instrument, with any confirmed correction from unified.price_history_corrections applied.

A corrected price is rounded to the paisa: the broker rounded when it divided, so multiplying back restores the raw price to within that rounding and no closer.

Parameters:

Name	Type	Description	Default
source	Source	The source.	required

Returns:

Type	Description
list[Bar]	The bars, oldest first.

Example usage

From example_2_each_step_by_hand.py

self.loader.read_series(corrected)

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def read_series(self, source):
    """
    One source's bars, restricted to the part of the series that belongs to its instrument, with
    any confirmed correction from unified.price_history_corrections applied.

    A corrected price is rounded to the paisa: the broker rounded when it divided, so multiplying
    back restores the raw price to within that rounding and no closer.

    Args:
        source (Source): The source.

    Returns:
        list[Bar]: The bars, oldest first.
    """
    conditions = ['instrument_token = %s', '"interval" = %s']
    parameters = [source.broker_series, self.interval]
    if source.valid_from is not None:
        conditions.append('"time" >= %s')
        parameters.append(source.valid_from)
    if source.valid_to is not None:
        conditions.append('"time" < %s')
        parameters.append(source.valid_to)
    with self.connection.cursor() as cursor:
        cursor.execute(f"""
            select "time", open, high, low, close, volume, oi from {source.broker}.price_history
            where {' and '.join(conditions)} order by "time"
        """, parameters)
        bars = [Bar(*row) for row in cursor.fetchall()]
        cursor.execute(f"""
            select valid_from, valid_to, price_multiplier, volume_multiplier from {tables.CORRECTION_RANGES}
            where broker = %s and broker_series = %s and "interval" = %s
        """, (source.broker, source.broker_series, self.interval))
        ranges = cursor.fetchall()
    if not ranges:
        return bars
    corrected = []
    for bar in bars:
        for valid_from, valid_to, price_multiplier, volume_multiplier in ranges:
            if valid_from <= bar.time < valid_to:
                bar = corrected_bar(bar, price_multiplier, volume_multiplier)
                self.counts["bars_corrected"] += 1
                break
        corrected.append(bar)
    return corrected

stitch

stitch(members)

Lay one broker's series for an instrument over each other, the most recent series on top.

Parameters:

Name	Type	Description	Default
members	list[Source]	The instrument's sources at one broker.	required

Returns:

Type	Description
dict	Bar time to (Bar, source_id).

Example usage

From example_2_each_step_by_hand.py

primary_bars = self.loader.stitch([zerodha])

From example_2_each_step_by_hand.py

print(f'agreement with Dhan: {self.loader.agreement(primary_bars, self.loader.stitch([dhan]))}')

From example_2_each_step_by_hand.py

print(f'agreement with Flattrade: {self.loader.agreement(primary_bars, self.loader.stitch([flattrade]))}')

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def stitch(self, members):
    """
    Lay one broker's series for an instrument over each other, the most recent series on top.

    Args:
        members (list[Source]): The instrument's sources at one broker.

    Returns:
        dict: Bar time to (Bar, source_id).
    """
    ordered = sorted(members, key=lambda source: source.broker_latest)
    bars = {}
    for source in ordered:
        for bar in self.read_series(source):
            bars[bar.time] = (bar, source.source_id)
    return bars

valid

valid(bar, exchange)

Whether a bar passes the calendar, grid and range checks, counting why when it does not.

Parameters:

Name	Type	Description	Default
bar	Bar	The bar.	required
exchange	str	The instrument's exchange.	required

Returns:

Type	Description
bool	True if the bar is kept.

Example usage

From example_2_each_step_by_hand.py

print(f'valid, {label}: {self.loader.valid(bar, "nse")}')

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def valid(self, bar, exchange):
    """
    Whether a bar passes the calendar, grid and range checks, counting why when it does not.

    Args:
        bar (Bar): The bar.
        exchange (str): The instrument's exchange.

    Returns:
        bool: True if the bar is kept.
    """
    if None in (bar.open, bar.high, bar.low, bar.close) or min(bar.open, bar.high, bar.low, bar.close) <= 0:
        self.counts["filtered_missing_price"] += 1
        return False
    if exchange in ("nse", "bse"):
        day = bar.time.astimezone(datetime.timezone(datetime.timedelta(hours=5, minutes=30))).date()
        if not self.calendar.is_trading_day(exchange, day):
            self.counts["filtered_non_trading_day"] += 1
            return False
    if not on_grid(bar.time, self.interval):
        self.counts["filtered_off_grid"] += 1
        return False
    if bar.high < max(bar.open, bar.close) or bar.low > min(bar.open, bar.close):
        self.counts["filtered_bad_range"] += 1
        return False
    return True

agreement staticmethod

agreement(primary, other)

The share of shared bar times on which two brokers' closes agree.

Parameters:

Name	Type	Description	Default
primary	dict	Bar time to (Bar, source_id).	required
other	dict	Bar time to (Bar, source_id).	required

Returns:

Type	Description
tuple[float, int]	The share agreeing, and how many bars were compared.

Example usage

From example_2_each_step_by_hand.py

print(f'agreement with Dhan: {self.loader.agreement(primary_bars, self.loader.stitch([dhan]))}')

From example_2_each_step_by_hand.py

print(f'agreement with Flattrade: {self.loader.agreement(primary_bars, self.loader.stitch([flattrade]))}')

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

@staticmethod
def agreement(primary, other):
    """
    The share of shared bar times on which two brokers' closes agree.

    Args:
        primary (dict): Bar time to (Bar, source_id).
        other (dict): Bar time to (Bar, source_id).

    Returns:
        tuple[float, int]: The share agreeing, and how many bars were compared.
    """
    shared = primary.keys() & other.keys()
    if not shared:
        return 0.0, 0
    agreeing = 0
    for moment in shared:
        first = primary[moment][0].close
        second = other[moment][0].close
        if first is not None and second is not None and abs(first - second) <= AGREEMENT_TOLERANCE * abs(first):
            agreeing += 1
    return agreeing / len(shared), len(shared)

rebuild

rebuild(instrument_id, members)

Rebuild one instrument's bars from its sources, in one transaction.

Parameters:

Name	Type	Description	Default
instrument_id	str	The instrument.	required
members	list[Source]	Its sources; a list holding only a placeholder with no broker means every source is gone and its bars are deleted.	required

Returns:

Type	Description
None	This function returns nothing.

Example usage

From example_2_each_step_by_hand.py

self.loader.rebuild('NIFTY50-NSE', members)

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def rebuild(self, instrument_id, members):
    """
    Rebuild one instrument's bars from its sources, in one transaction.

    Args:
        instrument_id (str): The instrument.
        members (list[Source]): Its sources; a list holding only a placeholder with no broker
            means every source is gone and its bars are deleted.

    Returns:
        None: This function returns nothing.
    """
    real = [source for source in members if source.broker is not None]
    chosen = {}
    contributed = Counter()
    if real:
        exchange = real[0].exchange
        primaries = [source for source in real if source.role == "primary"]
        by_broker = {}
        for source in primaries:
            by_broker.setdefault(source.broker, []).append(source)
        # One primary broker per instrument; the policy never names two.
        for broker, broker_sources in by_broker.items():
            for moment, (bar, source_id) in self.stitch(broker_sources).items():
                chosen[moment] = (bar, source_id)

        fillers = {}
        for source in real:
            if source.role == "gap_fill":
                fillers.setdefault(source.broker, []).append(source)
        for broker, broker_sources in fillers.items():
            offered = self.stitch(broker_sources)
            share, compared = self.agreement(chosen, offered)
            if chosen and (compared < MINIMUM_OVERLAP or share < GAP_FILL_AGREEMENT):
                self.counts["gap_fill_refused"] += 1
                LOGGER.info("%s %s: %s refused as gap-fill, %.4f agreement over %d bars",
                            self.interval, instrument_id, broker, share, compared)
                continue
            for moment, entry in offered.items():
                if moment not in chosen:
                    chosen[moment] = entry
                    self.counts["bars_gap_filled"] += 1

        kept = {}
        for moment, (bar, source_id) in chosen.items():
            if self.valid(bar, exchange):
                kept[moment] = (bar, source_id)
        chosen = kept

    rows = []
    for moment, (bar, source_id) in sorted(chosen.items()):
        rows.append((moment, instrument_id, self.interval, bar.open, bar.high, bar.low, bar.close,
                     bar.volume, bar.oi, source_id))
        contributed[source_id] += 1

    try:
        with self.connection.cursor() as cursor:
            cursor.execute(f"""
                select "time" from {tables.PRICE_HISTORY}
                where instrument_id = %s and "interval" = %s
            """, (instrument_id, self.interval))
            stale = [row[0] for row in cursor.fetchall() if row[0] not in chosen]
            for start in range(0, len(stale), WRITE_BATCH_ROWS):
                cursor.execute(f"""
                    delete from {tables.PRICE_HISTORY}
                    where instrument_id = %s and "interval" = %s and "time" = any(%s)
                """, (instrument_id, self.interval, stale[start:start + WRITE_BATCH_ROWS]))
            self.counts["bars_deleted"] += len(stale)

            for start in range(0, len(rows), WRITE_BATCH_ROWS):
                execute_values(cursor, f"""
                    insert into {tables.PRICE_HISTORY}
                        ("time", instrument_id, "interval", open, high, low, close, volume, oi, source_id)
                    values %s
                    on conflict (instrument_id, "interval", "time") do update set
                        open = excluded.open, high = excluded.high, low = excluded.low,
                        close = excluded.close, volume = excluded.volume, oi = excluded.oi,
                        source_id = excluded.source_id, loaded_at = now()
                    where ({tables.PRICE_HISTORY}.open, {tables.PRICE_HISTORY}.high,
                           {tables.PRICE_HISTORY}.low, {tables.PRICE_HISTORY}.close,
                           {tables.PRICE_HISTORY}.volume, {tables.PRICE_HISTORY}.oi,
                           {tables.PRICE_HISTORY}.source_id)
                        is distinct from
                          (excluded.open, excluded.high, excluded.low, excluded.close,
                           excluded.volume, excluded.oi, excluded.source_id)
                """, rows[start:start + WRITE_BATCH_ROWS], page_size=WRITE_BATCH_ROWS)
                self.counts["bars_written"] += cursor.rowcount

            for source in real:
                times = [moment for moment, (_, source_id) in chosen.items() if source_id == source.source_id]
                cursor.execute(f"""
                    update {tables.PRICE_HISTORY_SOURCES} set
                        broker_earliest_seen = %s,
                        broker_latest_seen = %s,
                        owned_from = %s,
                        owned_to = %s,
                        loaded_earliest = %s,
                        loaded_latest = %s,
                        last_loaded_at = now()
                    where source_id = %s
                """, (source.broker_earliest, source.broker_latest,
                      min(times) if times else None, max(times) if times else None,
                      min(times) if times else None, max(times) if times else None,
                      source.source_id))
        self.connection.commit()
    except Exception:
        self.connection.rollback()
        raise
    self.counts["bars_kept"] += len(rows)

IntradayLoader

IntradayLoader(interval, instrument_ids=None, resolver=None, calendar=None)

Bases: UnifiedLoader

Loads intraday bars, set-based in SQL.

Intraday history is two orders of magnitude larger than daily - flattrade alone stores some 300 million intraday bars - so bars are never brought into Python. Each instrument is rebuilt only from the point its sources changed: from three days before the newest bar its source rows had seen, or from the beginning for a new source or one whose history grew backwards.

The rebuild for one instrument deletes its bars in that window and inserts the stitched ones in a single statement, DISTINCT ON the bar time with the most recent series first; then inserts a gap-fill broker's bars for whole trading days the primary does not have, when the gap-fill broker agrees with the primary on the days both have. Trading days come from a temporary table filled once from the calendar; the grid and range checks are expressions in the same statement.

Prepare an intraday load.

Parameters:

Name	Type	Description	Default
interval	str	The stored intraday interval name, for example "15minute".	required
instrument_ids	set[str] | None	Restrict the load to these instruments, or None for all.	None
resolver	SeriesResolver | None	A resolver to share, or None to build one.	None
calendar	TradingCalendar | None	A calendar to share, or None to build one.	None

Returns:

Type	Description
None	This function returns nothing.

Example programs

An incremental 15 minute loadWindow start and a refused gap fill

examples/stock_brokers/instruments/historical/utilities/unified/loader/IntradayLoader/example_1_an_incremental_15_minute_load.py

"""Runs an incremental 15 minute load for a BSE share, with Flattrade as the primary broker and Wisdom Capital filling whole missing days.

Intraday history is far too large to bring into Python, so `IntradayLoader` does the stitching in SQL. When it is built it copies the calendar's trading days into a temporary table and reads how far each source row had been seen. `run` then refreshes the sources as the daily loader does, and rebuilds each changed instrument only from where its sources changed: three days before the newest bar a changed source had already been seen, so a window the downloader re-fetched and corrected is picked up. The rebuild deletes the instrument's bars from that point, inserts the primary broker's bars in one statement, checks that the gap-fill broker's closes agree with them within 1% on at least 95% of the bars both have, and only then inserts the gap-fill broker's bars for whole days the primary lacks.

The loader opens PostgreSQL through `get_postgres` in its constructor, which the example runner points at a closed port, so the program replaces `get_postgres` in the loader module with a stand-in returning an in-memory connection. Because the real work happens inside PostgreSQL, the stand-in cannot compute it: it answers each statement with recorded-looking row counts, 118 bars deleted, 125 inserted from Flattrade, 121 of 125 shared bars agreeing and 25 filled from Wisdom Capital, and records the parameters each statement was sent with. What the program shows is therefore the loader's own decisions: the window start, the statements' parameters, the agreement test and the counts. The resolver is a stand-in passed to the constructor and the calendar is a real `TradingCalendar` with its days filled in by hand.

Run it from the project root:

    python examples/stock_brokers/instruments/historical/utilities/unified/loader/IntradayLoader/example_1_an_incremental_15_minute_load.py
"""

import datetime
import zoneinfo

from stock_brokers.instruments.historical.utilities.unified import loader
from stock_brokers.instruments.historical.utilities.unified.calendar import (
    TradingCalendar,
)
from stock_brokers.instruments.historical.utilities.unified.loader import (
    IntradayLoader,
)
from stock_brokers.instruments.historical.utilities.unified.resolution import (
    Resolution,
)

INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
FLATTRADE_SERIES = 'BSE|500331|PIDILITIND'
WISDOM_SERIES = '12|500331'
LAST_SEEN = datetime.datetime(2026, 9, 25, 15, 15, tzinfo=INDIA)
NEWEST = datetime.datetime(2026, 9, 29, 15, 15, tzinfo=INDIA)
EARLIEST = datetime.datetime(2025, 9, 1, 9, 15, tzinfo=INDIA)


class StandInSeriesResolver:
    """A stand-in for `SeriesResolver` that resolves both brokers' series to PIDILITIND on BSE."""

    def resolve(self, broker, identifiers):
        """Resolves each identifier to PIDILITIND on BSE.

        Args:
            broker (str): The broker.
            identifiers (list): The broker's series identifiers.

        Returns:
            dict: Identifier to a list holding one Resolution.
        """
        resolved = {}
        for identifier in identifiers:
            resolution = Resolution(
                identifier,
                'PIDILITIND-BSE',
                'bse',
                'bse_equities',
                'PIDILITIND',
                'broker_mapping',
                'active',
                None,
                datetime.date(2026, 8, 12),
                datetime.date(2026, 9, 29),
                None,
                None,
            )
            resolved[identifier] = [
                resolution,
            ]
        return resolved


class StandInCursor:
    """A stand-in for a psycopg2 cursor that hands every statement to its connection.

    Attributes:
        connection (StandInConnection): The connection that answers the statements.
        result (list): The rows the last statement produced.
        rowcount (int): How many rows the last statement changed.
    """

    def __init__(self, connection):
        """Builds a cursor over the stand-in connection.

        Args:
            connection (StandInConnection): The connection that answers the statements.

        Returns:
            None: This method returns nothing.
        """
        self.connection = connection
        self.result = []
        self.rowcount = 0

    def __enter__(self):
        """Opens the cursor as a context manager.

        Returns:
            StandInCursor: This cursor.
        """
        return self

    def __exit__(self, exception_type, exception, traceback):
        """Closes the cursor, letting any exception carry on.

        Args:
            exception_type (type | None): The class of an exception raised inside the block.
            exception (BaseException | None): The exception raised inside the block.
            traceback (types.TracebackType | None): Where it was raised.

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

    def mogrify(self, template, arguments):
        """Renders one row, as `execute_values` and the loader's series list ask a cursor to.

        Args:
            template (bytes): The row template.
            arguments (tuple): The row's values.

        Returns:
            bytes: The row's values as text, standing in for the rendered row.
        """
        self.connection.staged_rows.append(arguments)
        return str(arguments).encode()

    def execute(self, statement, parameters=None):
        """Runs one statement against the stand-in connection.

        Args:
            statement (str | bytes): The SQL text.
            parameters (tuple | list | None): The statement's parameters.

        Returns:
            None: This method returns nothing.
        """
        if isinstance(statement, bytes):
            statement = statement.decode()
        self.result, self.rowcount = self.connection.answer(statement, parameters)

    def fetchone(self):
        """The first row the last statement produced.

        Returns:
            tuple | None: The row, or None when there was none.
        """
        if not self.result:
            return None
        return self.result[0]

    def fetchall(self):
        """Every row the last statement produced.

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


class StandInConnection:
    """A stand-in for a psycopg2 connection that answers the intraday loader's statements with row counts.

    Attributes:
        encoding (str): The client encoding, which `execute_values` reads.
        staged_rows (list): Rows rendered by `mogrify` since the last statement.
        trading_days_copied (int): How many trading days were copied into the temporary table.
        agreement (tuple): The (bars compared, bars agreeing) the agreement query answers.
        statements (list): A short description of each statement that changes bars.
    """

    encoding = 'UTF8'

    def __init__(self, agreement):
        """Starts with nothing executed.

        Args:
            agreement (tuple): The (bars compared, bars agreeing) the agreement query answers.

        Returns:
            None: This method returns nothing.
        """
        self.staged_rows = []
        self.trading_days_copied = 0
        self.agreement = agreement
        self.statements = []

    def row(self, *columns):
        """Builds one result row.

        Args:
            *columns (object): The row's columns, in order.

        Returns:
            tuple: The row.
        """
        return columns

    def cursor(self):
        """Opens a stand-in cursor.

        Returns:
            StandInCursor: The cursor.
        """
        return StandInCursor(self)

    def answer(self, statement, parameters):
        """Answers one of the loader's statements.

        Args:
            statement (str): The SQL text.
            parameters (tuple | list | dict | None): The statement's parameters.

        Returns:
            tuple: (rows, rows changed).
        """
        if 'unified_trading_days values' in statement:
            self.trading_days_copied += len(self.staged_rows)
            self.staged_rows = []
            return [], 0
        if 'temporary table' in statement or 'truncate' in statement:
            return [], 0
        if 'select source_id, broker_earliest_seen, broker_latest_seen' in statement:
            return [
                self.row(51, EARLIEST, LAST_SEEN),
                self.row(52, EARLIEST, LAST_SEEN),
            ], 0
        if 'from unified.price_history_sources where' in statement:
            return [
                self.row(51, 'flattrade', FLATTRADE_SERIES, 'PIDILITIND-BSE', 'active', EARLIEST, LAST_SEEN, EARLIEST, LAST_SEEN),
                self.row(52, 'wisdom_capital', WISDOM_SERIES, 'PIDILITIND-BSE', 'active', EARLIEST, LAST_SEEN, EARLIEST, LAST_SEEN),
            ], 0
        if 'price_history_progress' in statement:
            if 'flattrade.' in statement:
                return [
                    self.row(FLATTRADE_SERIES, EARLIEST, NEWEST),
                ], 0
            if 'wisdom_capital.' in statement:
                return [
                    self.row(WISDOM_SERIES, EARLIEST, NEWEST),
                ], 0
            return [], 0
        if 'insert into unified.price_history_sources' in statement:
            source_id = 52
            if parameters[0] == 'flattrade':
                source_id = 51
            return [
                self.row(source_id),
            ], 1
        if 'delete from unified.price_history' in statement:
            self.statements.append(f'delete bars of {parameters[0]} from {parameters[2].isoformat()}')
            return [], 118
        if 'select count(*), count(*) filter' in statement:
            return [
                self.agreement,
            ], 0
        if 'insert into unified.price_history' in statement:
            self.staged_rows = []
            broker = statement.split(' b\n')[0].split('from ')[-1].split('.')[0]
            self.statements.append(f"insert from {broker}: start {parameters['start'].isoformat()}, every {parameters['minutes']} minutes, exchange {parameters['exchange']}, series {parameters['series']}")
            if broker == 'flattrade':
                return [], 125
            return [], 25
        if 'update unified.price_history_sources' in statement:
            return [], 1
        raise ValueError(f'The stand-in does not know this statement: {statement[:60]}')

    def commit(self):
        """Accepts a commit.

        Returns:
            None: This method returns nothing.
        """

    def rollback(self):
        """Accepts a rollback.

        Returns:
            None: This method returns nothing.
        """


class StandInPostgres:
    """Hands out the stand-in connection in the place of `get_postgres`.

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

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

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

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

    def get_postgres(self):
        """Returns the stand-in connection.

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


class IncrementalLoadExample:
    """Runs a 15 minute load for PIDILITIND on BSE and prints the loader's decisions.

    Attributes:
        connection (StandInConnection): The in-memory connection.
        loader (IntradayLoader): The loader being shown.
    """

    def __init__(self):
        """Installs the stand-in connection and builds the loader with a hand-filled calendar.

        Returns:
            None: This method returns nothing.
        """
        agreement = (
            125,
            121,
        )
        self.connection = StandInConnection(agreement)
        loader.get_postgres = StandInPostgres(self.connection).get_postgres
        calendar = TradingCalendar()
        calendar.days['nse'] = set()
        calendar.days['bse'] = set()
        day = datetime.date(2026, 9, 1)
        while day <= datetime.date(2026, 9, 30):
            if day.weekday() < 5:
                calendar.days['nse'].add(day)
                calendar.days['bse'].add(day)
            day += datetime.timedelta(days=1)
        self.loader = IntradayLoader('15minute', resolver=StandInSeriesResolver(), calendar=calendar)

    def run(self):
        """Runs the load and prints what was decided and sent.

        Returns:
            None: This method returns nothing.
        """
        print(f'Minutes per bar: {self.loader.minutes}; trading days copied: {self.connection.trading_days_copied}')
        for source_id, seen in self.loader.previously_seen.items():
            print(f'Source {source_id} previously seen from {seen[0].isoformat()} to {seen[1].isoformat()}')
        counts = self.loader.run()
        for statement in self.connection.statements:
            print(statement)
        for name in sorted(counts):
            print(f'{name}: {counts[name]}')

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

It prints:

Minutes per bar: 15; trading days copied: 44
Source 51 previously seen from 2025-09-01T09:15:00+05:30 to 2026-09-25T15:15:00+05:30
Source 52 previously seen from 2025-09-01T09:15:00+05:30 to 2026-09-25T15:15:00+05:30
delete bars of PIDILITIND-BSE from 2026-09-22T15:15:00+05:30
insert from flattrade: start 2026-09-22T15:15:00+05:30, every 15 minutes, exchange bse, series ('BSE|500331|PIDILITIND', 51, 0, None, None)
insert from wisdom_capital: start 2026-09-22T15:15:00+05:30, every 15 minutes, exchange bse, series ('12|500331', 52, 0, None, None)
bars_deleted: 118
bars_gap_filled: 25
bars_written: 125
instruments_rebuilt: 1
series_dhan: 0
series_flattrade: 1
series_wisdom_capital: 1
series_zerodha: 0
sources_superseded: 0

examples/stock_brokers/instruments/historical/utilities/unified/loader/IntradayLoader/example_2_window_start_and_a_refused_gap_fill.py

"""Calls `IntradayLoader`'s steps by hand: choosing where a rebuild starts, testing a gap-fill broker, and rebuilding with that broker refused.

`window_start` decides how much of an instrument's intraday history is rebuilt. It starts three days before the newest bar a changed source had already been seen, so a recently corrected window is picked up; but it rebuilds from the beginning, 1990-01-01, when any source is new, when a source's history grew backwards, or when the list holds only a placeholder for an instrument that lost its sources. The program asks it about each of those cases.

`intraday_agreement` counts, in one query, the bars on which a gap-fill broker's close is within 1% of the bars already inserted, and `insert_bars` inserts one broker's stitched bars from a start time, restricted to whole missing days for a gap-fill broker. `rebuild` combines them: here Wisdom Capital agrees on only 20 of 30 shared bars, below the 95% the loader requires, so it is refused and only Flattrade's bars are inserted.

The loader opens PostgreSQL through `get_postgres` in its constructor, which the example runner points at a closed port, so the program replaces `get_postgres` in the loader module with a stand-in returning an in-memory connection. The real stitching runs inside PostgreSQL, so the stand-in answers with recorded-looking row counts and records the parameters each statement was sent with; what the program shows is the loader's own decisions around those statements. The resolver is a stand-in and the calendar is a real `TradingCalendar` with its days filled in by hand.

Run it from the project root:

    python examples/stock_brokers/instruments/historical/utilities/unified/loader/IntradayLoader/example_2_window_start_and_a_refused_gap_fill.py
"""

import datetime
import zoneinfo

from stock_brokers.instruments.historical.utilities.unified import loader
from stock_brokers.instruments.historical.utilities.unified.calendar import (
    TradingCalendar,
)
from stock_brokers.instruments.historical.utilities.unified.loader import (
    IntradayLoader,
    Source,
)
from stock_brokers.instruments.historical.utilities.unified.resolution import (
    Resolution,
)

INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
FLATTRADE_SERIES = 'BSE|500331|PIDILITIND'
WISDOM_SERIES = '12|500331'
LAST_SEEN = datetime.datetime(2026, 9, 25, 15, 15, tzinfo=INDIA)
NEWEST = datetime.datetime(2026, 9, 29, 15, 15, tzinfo=INDIA)
EARLIEST = datetime.datetime(2025, 9, 1, 9, 15, tzinfo=INDIA)


class StandInSeriesResolver:
    """A stand-in for `SeriesResolver` that resolves both brokers' series to PIDILITIND on BSE."""

    def resolve(self, broker, identifiers):
        """Resolves each identifier to PIDILITIND on BSE.

        Args:
            broker (str): The broker.
            identifiers (list): The broker's series identifiers.

        Returns:
            dict: Identifier to a list holding one Resolution.
        """
        resolved = {}
        for identifier in identifiers:
            resolution = Resolution(
                identifier,
                'PIDILITIND-BSE',
                'bse',
                'bse_equities',
                'PIDILITIND',
                'broker_mapping',
                'active',
                None,
                datetime.date(2026, 8, 12),
                datetime.date(2026, 9, 29),
                None,
                None,
            )
            resolved[identifier] = [
                resolution,
            ]
        return resolved


class StandInCursor:
    """A stand-in for a psycopg2 cursor that hands every statement to its connection.

    Attributes:
        connection (StandInConnection): The connection that answers the statements.
        result (list): The rows the last statement produced.
        rowcount (int): How many rows the last statement changed.
    """

    def __init__(self, connection):
        """Builds a cursor over the stand-in connection.

        Args:
            connection (StandInConnection): The connection that answers the statements.

        Returns:
            None: This method returns nothing.
        """
        self.connection = connection
        self.result = []
        self.rowcount = 0

    def __enter__(self):
        """Opens the cursor as a context manager.

        Returns:
            StandInCursor: This cursor.
        """
        return self

    def __exit__(self, exception_type, exception, traceback):
        """Closes the cursor, letting any exception carry on.

        Args:
            exception_type (type | None): The class of an exception raised inside the block.
            exception (BaseException | None): The exception raised inside the block.
            traceback (types.TracebackType | None): Where it was raised.

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

    def mogrify(self, template, arguments):
        """Renders one row, as `execute_values` and the loader's series list ask a cursor to.

        Args:
            template (bytes): The row template.
            arguments (tuple): The row's values.

        Returns:
            bytes: The row's values as text, standing in for the rendered row.
        """
        self.connection.staged_rows.append(arguments)
        return str(arguments).encode()

    def execute(self, statement, parameters=None):
        """Runs one statement against the stand-in connection.

        Args:
            statement (str | bytes): The SQL text.
            parameters (tuple | list | None): The statement's parameters.

        Returns:
            None: This method returns nothing.
        """
        if isinstance(statement, bytes):
            statement = statement.decode()
        self.result, self.rowcount = self.connection.answer(statement, parameters)

    def fetchone(self):
        """The first row the last statement produced.

        Returns:
            tuple | None: The row, or None when there was none.
        """
        if not self.result:
            return None
        return self.result[0]

    def fetchall(self):
        """Every row the last statement produced.

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


class StandInConnection:
    """A stand-in for a psycopg2 connection that answers the intraday loader's statements with row counts.

    Attributes:
        encoding (str): The client encoding, which `execute_values` reads.
        staged_rows (list): Rows rendered by `mogrify` since the last statement.
        trading_days_copied (int): How many trading days were copied into the temporary table.
        agreement (tuple): The (bars compared, bars agreeing) the agreement query answers.
        statements (list): A short description of each statement that changes bars.
    """

    encoding = 'UTF8'

    def __init__(self, agreement):
        """Starts with nothing executed.

        Args:
            agreement (tuple): The (bars compared, bars agreeing) the agreement query answers.

        Returns:
            None: This method returns nothing.
        """
        self.staged_rows = []
        self.trading_days_copied = 0
        self.agreement = agreement
        self.statements = []

    def row(self, *columns):
        """Builds one result row.

        Args:
            *columns (object): The row's columns, in order.

        Returns:
            tuple: The row.
        """
        return columns

    def cursor(self):
        """Opens a stand-in cursor.

        Returns:
            StandInCursor: The cursor.
        """
        return StandInCursor(self)

    def answer(self, statement, parameters):
        """Answers one of the loader's statements.

        Args:
            statement (str): The SQL text.
            parameters (tuple | list | dict | None): The statement's parameters.

        Returns:
            tuple: (rows, rows changed).
        """
        if 'unified_trading_days values' in statement:
            self.trading_days_copied += len(self.staged_rows)
            self.staged_rows = []
            return [], 0
        if 'temporary table' in statement or 'truncate' in statement:
            return [], 0
        if 'select source_id, broker_earliest_seen, broker_latest_seen' in statement:
            return [
                self.row(51, EARLIEST, LAST_SEEN),
                self.row(52, EARLIEST, LAST_SEEN),
            ], 0
        if 'from unified.price_history_sources where' in statement:
            return [
                self.row(51, 'flattrade', FLATTRADE_SERIES, 'PIDILITIND-BSE', 'active', EARLIEST, LAST_SEEN, EARLIEST, LAST_SEEN),
                self.row(52, 'wisdom_capital', WISDOM_SERIES, 'PIDILITIND-BSE', 'active', EARLIEST, LAST_SEEN, EARLIEST, LAST_SEEN),
            ], 0
        if 'price_history_progress' in statement:
            if 'flattrade.' in statement:
                return [
                    self.row(FLATTRADE_SERIES, EARLIEST, NEWEST),
                ], 0
            if 'wisdom_capital.' in statement:
                return [
                    self.row(WISDOM_SERIES, EARLIEST, NEWEST),
                ], 0
            return [], 0
        if 'insert into unified.price_history_sources' in statement:
            source_id = 52
            if parameters[0] == 'flattrade':
                source_id = 51
            return [
                self.row(source_id),
            ], 1
        if 'delete from unified.price_history' in statement:
            self.statements.append(f'delete bars of {parameters[0]} from {parameters[2].isoformat()}')
            return [], 118
        if 'select count(*), count(*) filter' in statement:
            return [
                self.agreement,
            ], 0
        if 'insert into unified.price_history' in statement:
            self.staged_rows = []
            broker = statement.split(' b\n')[0].split('from ')[-1].split('.')[0]
            self.statements.append(f"insert from {broker}: start {parameters['start'].isoformat()}, every {parameters['minutes']} minutes, exchange {parameters['exchange']}, series {parameters['series']}")
            if broker == 'flattrade':
                return [], 125
            return [], 25
        if 'update unified.price_history_sources' in statement:
            return [], 1
        raise ValueError(f'The stand-in does not know this statement: {statement[:60]}')

    def commit(self):
        """Accepts a commit.

        Returns:
            None: This method returns nothing.
        """

    def rollback(self):
        """Accepts a rollback.

        Returns:
            None: This method returns nothing.
        """


class StandInPostgres:
    """Hands out the stand-in connection in the place of `get_postgres`.

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

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

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

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

    def get_postgres(self):
        """Returns the stand-in connection.

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


class WindowStartExample:
    """Asks the intraday loader where rebuilds start, then rebuilds with a gap-fill broker it refuses.

    Attributes:
        connection (StandInConnection): The in-memory connection.
        loader (IntradayLoader): The loader being shown.
    """

    def __init__(self):
        """Installs the stand-in connection and builds the loader with a hand-filled calendar.

        Returns:
            None: This method returns nothing.
        """
        agreement = (
            30,
            20,
        )
        self.connection = StandInConnection(agreement)
        loader.get_postgres = StandInPostgres(self.connection).get_postgres
        calendar = TradingCalendar()
        calendar.days['nse'] = set()
        calendar.days['bse'] = set()
        day = datetime.date(2026, 9, 1)
        while day <= datetime.date(2026, 9, 30):
            if day.weekday() < 5:
                calendar.days['bse'].add(day)
            day += datetime.timedelta(days=1)
        self.loader = IntradayLoader('15minute', resolver=StandInSeriesResolver(), calendar=calendar)

    def source(self, source_id, broker, series, role, earliest, changed):
        """Builds one of PIDILITIND's source rows.

        Args:
            source_id (int | None): The source row's id, or None for a placeholder.
            broker (str | None): The broker, or None for a placeholder.
            series (str | None): The broker's series identifier.
            role (str | None): "primary" or "gap_fill".
            earliest (datetime.datetime | None): The broker's earliest bar.
            changed (bool): Whether the source has changed since it was last loaded.

        Returns:
            Source: The source.
        """
        return Source(
            source_id,
            broker,
            series,
            'PIDILITIND-BSE',
            role,
            'bse',
            None,
            None,
            earliest,
            NEWEST,
            changed,
        )

    def ask(self, label, *members):
        """Prints where a rebuild of these sources would start.

        Args:
            label (str): What is special about the sources.
            *members (Source): The instrument's sources.

        Returns:
            None: This method returns nothing.
        """
        print(f'window_start, {label}: {self.loader.window_start(list(members)).isoformat()}')

    def run(self):
        """Asks about each case, tests the gap-fill broker, and rebuilds.

        Returns:
            None: This method returns nothing.
        """
        flattrade = self.source(51, 'flattrade', FLATTRADE_SERIES, 'primary', EARLIEST, True)
        wisdom = self.source(52, 'wisdom_capital', WISDOM_SERIES, 'gap_fill', EARLIEST, False)
        self.ask('one source grew forwards', flattrade, wisdom)
        self.ask('a new source', flattrade, self.source(99, 'wisdom_capital', WISDOM_SERIES, 'gap_fill', EARLIEST, True))
        grew_backwards = self.source(51, 'flattrade', FLATTRADE_SERIES, 'primary', datetime.datetime(2025, 6, 2, 9, 15, tzinfo=INDIA), True)
        self.ask('a source grew backwards', grew_backwards, wisdom)
        self.ask('only a placeholder', self.source(None, None, None, None, None, True))
        cursor = self.connection.cursor()
        gap_fill_sources = [
            wisdom,
        ]
        share, compared = self.loader.intraday_agreement(cursor, 'PIDILITIND-BSE', gap_fill_sources)
        print(f'intraday_agreement with Wisdom Capital: {share:.3f} over {compared} bars')
        start = datetime.datetime(2026, 9, 22, 15, 15, tzinfo=INDIA)
        primary_sources = [
            flattrade,
        ]
        self.loader.insert_bars(cursor, 'PIDILITIND-BSE', 'bse', primary_sources, start, whole_days_only=False)
        print(f'insert_bars by hand: {self.connection.statements[-1]}')
        self.connection.statements = []
        self.loader.counts.clear()
        members = [
            flattrade,
            wisdom,
        ]
        self.loader.rebuild('PIDILITIND-BSE', members)
        print('rebuild sent:')
        for statement in self.connection.statements:
            print(f'  {statement}')
        for name in sorted(self.loader.counts):
            print(f'{name}: {self.loader.counts[name]}')

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

It prints:

window_start, one source grew forwards: 2026-09-22T15:15:00+05:30
window_start, a new source: 1990-01-01T00:00:00+00:00
window_start, a source grew backwards: 1990-01-01T00:00:00+00:00
window_start, only a placeholder: 1990-01-01T00:00:00+00:00
intraday_agreement with Wisdom Capital: 0.667 over 30 bars
insert_bars by hand: insert from flattrade: start 2026-09-22T15:15:00+05:30, every 15 minutes, exchange bse, series ('BSE|500331|PIDILITIND', 51, 0, None, None)
rebuild sent:
  delete bars of PIDILITIND-BSE from 2026-09-22T15:15:00+05:30
  insert from flattrade: start 2026-09-22T15:15:00+05:30, every 15 minutes, exchange bse, series ('BSE|500331|PIDILITIND', 51, 0, None, None)
bars_deleted: 118
bars_written: 125
gap_fill_refused: 1

Methods:

Name	Description
window_start	Where an instrument's rebuild starts.
rebuild	Rebuild one instrument's intraday bars from where its sources changed, in one transaction.
insert_bars	Insert one broker's stitched bars for an instrument from start.
intraday_agreement	The share of bars on which a gap-fill broker's close agrees with the bars already inserted.
run	Discover, resolve and load.
refresh_sources	Resolve every series of every source broker and bring the sources table up to date.
upsert_source	Insert or refresh one source row, returning its id.
read_series	One source's bars, restricted to the part of the series that belongs to its instrument, with
stitch	Lay one broker's series for an instrument over each other, the most recent series on top.
valid	Whether a bar passes the calendar, grid and range checks, counting why when it does not.
agreement	The share of shared bar times on which two brokers' closes agree.

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def __init__(self, interval, instrument_ids=None, resolver=None, calendar=None):
    """
    Prepare an intraday load.

    Args:
        interval (str): The stored intraday interval name, for example "15minute".
        instrument_ids (set[str] | None): Restrict the load to these instruments, or None for all.
        resolver (SeriesResolver | None): A resolver to share, or None to build one.
        calendar (TradingCalendar | None): A calendar to share, or None to build one.

    Returns:
        None: This function returns nothing.
    """
    super().__init__(interval, instrument_ids, resolver, calendar)
    from stock_brokers.instruments.historical.utilities.unified.sources import INTRADAY_MINUTES
    self.minutes = INTRADAY_MINUTES[interval]
    with self.connection.cursor() as cursor:
        cursor.execute("create temporary table if not exists unified_trading_days (exchange text, day date, "
                       "primary key (exchange, day))")
        cursor.execute("truncate unified_trading_days")
        for exchange in ("nse", "bse"):
            execute_values(cursor, "insert into unified_trading_days values %s",
                           [(exchange, day) for day in self.calendar.trading_days(exchange)], page_size=10000)
        cursor.execute(f"""
            select source_id, broker_earliest_seen, broker_latest_seen
            from {tables.PRICE_HISTORY_SOURCES} where "interval" = %s
        """, (interval,))
        self.previously_seen = {row[0]: (row[1], row[2]) for row in cursor.fetchall()}
    self.connection.commit()

window_start

window_start(members)

Where an instrument's rebuild starts.

The whole history is rebuilt when a source is new, when a source lost, or when one's history grew backwards; otherwise from a little before the newest bar any changed source had seen.

Parameters:

Name	Type	Description	Default
members	list[Source]	The instrument's sources.	required

Returns:

Type	Description
datetime	The first bar time to rebuild.

Example usage

From example_2_window_start_and_a_refused_gap_fill.py

print(f'window_start, {label}: {self.loader.window_start(list(members)).isoformat()}')

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def window_start(self, members):
    """
    Where an instrument's rebuild starts.

    The whole history is rebuilt when a source is new, when a source lost, or when one's history
    grew backwards; otherwise from a little before the newest bar any changed source had seen.

    Args:
        members (list[Source]): The instrument's sources.

    Returns:
        datetime.datetime: The first bar time to rebuild.
    """
    beginning = datetime.datetime(1990, 1, 1, tzinfo=datetime.timezone.utc)
    starts = []
    for source in members:
        if source.broker is None:
            return beginning
        previous_earliest, previous_latest = self.previously_seen.get(source.source_id, (None, None))
        if previous_earliest is None or previous_latest is None:
            return beginning
        if source.broker_earliest < previous_earliest:
            return beginning
        if source.changed:
            starts.append(previous_latest - INTRADAY_OVERLAP)
    return min(starts) if starts else beginning

rebuild

rebuild(instrument_id, members)

Rebuild one instrument's intraday bars from where its sources changed, in one transaction.

Parameters:

Name	Type	Description	Default
instrument_id	str	The instrument.	required
members	list[Source]	Its sources.	required

Returns:

Type	Description
None	This function returns nothing.

Example usage

From example_2_window_start_and_a_refused_gap_fill.py

self.loader.rebuild('PIDILITIND-BSE', members)

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def rebuild(self, instrument_id, members):
    """
    Rebuild one instrument's intraday bars from where its sources changed, in one transaction.

    Args:
        instrument_id (str): The instrument.
        members (list[Source]): Its sources.

    Returns:
        None: This function returns nothing.
    """
    real = [source for source in members if source.broker is not None]
    start = self.window_start(members)

    try:
        with self.connection.cursor() as cursor:
            cursor.execute(f"""
                delete from {tables.PRICE_HISTORY}
                where instrument_id = %s and "interval" = %s and "time" >= %s
            """, (instrument_id, self.interval, start))
            self.counts["bars_deleted"] += cursor.rowcount

            exchange = real[0].exchange if real else None
            primaries = [source for source in real if source.role == "primary"]
            fillers = [source for source in real if source.role == "gap_fill"]
            if primaries:
                self.insert_bars(cursor, instrument_id, exchange, primaries, start, whole_days_only=False)
            by_broker = {}
            for source in fillers:
                by_broker.setdefault(source.broker, []).append(source)
            for broker, broker_sources in by_broker.items():
                share, compared = self.intraday_agreement(cursor, instrument_id, broker_sources)
                if primaries and (compared < MINIMUM_OVERLAP or share < INTRADAY_GAP_FILL_AGREEMENT):
                    self.counts["gap_fill_refused"] += 1
                    LOGGER.info("%s %s: %s refused as gap-fill, %.4f agreement over %d bars",
                                self.interval, instrument_id, broker, share, compared)
                    continue
                self.insert_bars(cursor, instrument_id, exchange, broker_sources, start, whole_days_only=True)

            for source in real:
                cursor.execute(f"""
                    update {tables.PRICE_HISTORY_SOURCES} s set
                        broker_earliest_seen = %s, broker_latest_seen = %s,
                        owned_from = b.first_bar, owned_to = b.last_bar,
                        loaded_earliest = b.first_bar, loaded_latest = b.last_bar, last_loaded_at = now()
                    from (select min("time") as first_bar, max("time") as last_bar
                          from {tables.PRICE_HISTORY}
                          where instrument_id = %s and "interval" = %s and source_id = %s) b
                    where s.source_id = %s
                """, (source.broker_earliest, source.broker_latest, instrument_id, self.interval,
                      source.source_id, source.source_id))
        self.connection.commit()
    except Exception:
        self.connection.rollback()
        raise

insert_bars

insert_bars(cursor, instrument_id, exchange, sources, start, whole_days_only)

Insert one broker's stitched bars for an instrument from start.

Parameters:

Name	Type	Description	Default
cursor	cursor	An open cursor inside the instrument's transaction.	required
instrument_id	str	The instrument.	required
exchange	str	Its exchange, for the trading day check.	required
sources	list[Source]	The broker's series for the instrument.	required
start	datetime	The first bar time to insert.	required
whole_days_only	bool	Insert only on India dates the instrument has no bar for yet.	required

Returns:

Type	Description
None	This function returns nothing.

Example usage

From example_2_window_start_and_a_refused_gap_fill.py

self.loader.insert_bars(cursor, 'PIDILITIND-BSE', 'bse', primary_sources, start, whole_days_only=False)

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def insert_bars(self, cursor, instrument_id, exchange, sources, start, whole_days_only):
    """
    Insert one broker's stitched bars for an instrument from `start`.

    Args:
        cursor (psycopg2.extensions.cursor): An open cursor inside the instrument's transaction.
        instrument_id (str): The instrument.
        exchange (str): Its exchange, for the trading day check.
        sources (list[Source]): The broker's series for the instrument.
        start (datetime.datetime): The first bar time to insert.
        whole_days_only (bool): Insert only on India dates the instrument has no bar for yet.

    Returns:
        None: This function returns nothing.
    """
    broker = sources[0].broker
    ranked = sorted(sources, key=lambda source: source.broker_latest, reverse=True)
    series_values = [(source.broker_series, source.source_id, rank, source.valid_from, source.valid_to)
                     for rank, source in enumerate(ranked)]
    day = "((b.\"time\" at time zone 'Asia/Kolkata')::date)"
    since_open = ("(extract(hour from b.\"time\" at time zone 'Asia/Kolkata') * 60 "
                  "+ extract(minute from b.\"time\" at time zone 'Asia/Kolkata') - 555)")
    missing_day = ""
    if whole_days_only:
        missing_day = f"""
            and not exists (select 1 from {tables.PRICE_HISTORY} u
                            where u.instrument_id = %(instrument_id)s and u."interval" = %(interval)s
                              and u."time" >= {day}::timestamp at time zone 'Asia/Kolkata'
                              and u."time" < ({day} + 1)::timestamp at time zone 'Asia/Kolkata')"""
    trading_day = ""
    if exchange in ("nse", "bse"):
        trading_day = f"and exists (select 1 from unified_trading_days t where t.exchange = %(exchange)s and t.day = {day})"
    statement = f"""
        insert into {tables.PRICE_HISTORY}
            ("time", instrument_id, "interval", open, high, low, close, volume, oi, source_id)
        select distinct on (b."time")
            b."time", %(instrument_id)s, b."interval", b.open, b.high, b.low, b.close, b.volume, b.oi, s.source_id
        from {broker}.price_history b
        join (values %(series)s) as s(series, source_id, rank, valid_from, valid_to)
          on b.instrument_token = s.series
        where b."interval" = %(interval)s and b."time" >= %(start)s
          and (s.valid_from is null or b."time" >= s.valid_from::timestamptz)
          and (s.valid_to is null or b."time" < s.valid_to::timestamptz)
          and b.open > 0 and b.high > 0 and b.low > 0 and b.close > 0
          and b.high >= greatest(b.open, b.close) and b.low <= least(b.open, b.close)
          and extract(second from b."time") = 0
          and {since_open} >= 0 and {since_open}::int %% %(minutes)s = 0
          {trading_day}
          {missing_day}
        order by b."time", s.rank
        on conflict (instrument_id, "interval", "time") do nothing
    """
    from psycopg2.extensions import AsIs
    rendered_series = ",".join(cursor.mogrify("(%s, %s, %s, %s::text, %s::text)", values).decode()
                               for values in series_values)
    cursor.execute(statement, {"instrument_id": instrument_id, "interval": self.interval, "start": start,
                               "exchange": exchange, "minutes": self.minutes, "series": AsIs(rendered_series)})
    self.counts["bars_gap_filled" if whole_days_only else "bars_written"] += cursor.rowcount

intraday_agreement

intraday_agreement(cursor, instrument_id, sources)

The share of bars on which a gap-fill broker's close agrees with the bars already inserted.

Parameters:

Name	Type	Description	Default
cursor	cursor	An open cursor.	required
instrument_id	str	The instrument.	required
sources	list[Source]	The gap-fill broker's series.	required

Returns:

Type	Description
tuple[float, int]	The share agreeing, and how many bars were compared.

Example usage

From example_2_window_start_and_a_refused_gap_fill.py

share, compared = self.loader.intraday_agreement(cursor, 'PIDILITIND-BSE', gap_fill_sources)

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def intraday_agreement(self, cursor, instrument_id, sources):
    """
    The share of bars on which a gap-fill broker's close agrees with the bars already inserted.

    Args:
        cursor (psycopg2.extensions.cursor): An open cursor.
        instrument_id (str): The instrument.
        sources (list[Source]): The gap-fill broker's series.

    Returns:
        tuple[float, int]: The share agreeing, and how many bars were compared.
    """
    broker = sources[0].broker
    cursor.execute(f"""
        select count(*), count(*) filter (where abs(b.close - u.close) <= %s * abs(u.close))
        from {tables.PRICE_HISTORY} u
        join {broker}.price_history b
          on b.instrument_token = any(%s) and b."interval" = u."interval" and b."time" = u."time"
        where u.instrument_id = %s and u."interval" = %s
    """, (INTRADAY_AGREEMENT_TOLERANCE, [source.broker_series for source in sources], instrument_id, self.interval))
    compared, agreeing = cursor.fetchone()
    return (agreeing / compared if compared else 0.0), compared

run

run()

Discover, resolve and load.

Returns:

Type	Description
Counter	What was done - series seen, instruments rebuilt, bars written, deleted and filtered.

Example usage

From example_1_loading_daily_bars.py

counts = self.loader.run()

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def run(self):
    """
    Discover, resolve and load.

    Returns:
        Counter: What was done - series seen, instruments rebuilt, bars written, deleted and filtered.
    """
    sources = self.refresh_sources()
    by_instrument = {}
    for source in sources:
        by_instrument.setdefault(source.instrument_id, []).append(source)

    work = [instrument_id for instrument_id, members in by_instrument.items()
            if any(source.changed for source in members)]
    LOGGER.info("%s: %d instruments with sources, %d to rebuild",
                self.interval, len(by_instrument), len(work))
    for position, instrument_id in enumerate(sorted(work), 1):
        self.rebuild(instrument_id, by_instrument[instrument_id])
        if position % 500 == 0:
            LOGGER.info("%s: rebuilt %d of %d instruments", self.interval, position, len(work))
    self.counts["instruments_rebuilt"] += len(work)
    return self.counts

refresh_sources

refresh_sources()

Resolve every series of every source broker and bring the sources table up to date.

Returns:

Type	Description
list[Source]	The active sources, restricted to instrument_ids when given.

Example usage

From example_2_each_step_by_hand.py

sources = self.loader.refresh_sources()

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def refresh_sources(self):
    """
    Resolve every series of every source broker and bring the sources table up to date.

    Returns:
        list[Source]: The active sources, restricted to `instrument_ids` when given.
    """
    existing = {}
    with self.connection.cursor() as cursor:
        cursor.execute(f"""
            select source_id, broker, broker_series, instrument_id::text, status,
                   broker_earliest_seen, broker_latest_seen, owned_from, owned_to
            from {tables.PRICE_HISTORY_SOURCES} where "interval" = %s
        """, (self.interval,))
        for row in cursor.fetchall():
            existing[(row[1], row[2], row[3])] = row

    sources = []
    superseded = set()
    seen_keys = set()
    for broker in brokers_for_interval(self.interval):
        with self.connection.cursor() as cursor:
            cursor.execute(f"""
                select instrument_token, earliest_bar_time, latest_bar_time
                from {broker}.price_history_progress
                where "interval" = %s and bar_count > 0 and earliest_bar_time is not null
            """, (self.interval,))
            progress = {row[0]: (row[1], row[2]) for row in cursor.fetchall()}
        self.counts[f"series_{broker}"] = len(progress)

        resolutions = self.resolver.resolve(broker, list(progress))
        for identifier, found in resolutions.items():
            earliest, latest = progress[identifier]
            for resolution in found:
                if resolution.instrument_id is None:
                    self.counts[f"unresolved_{broker}"] += 1
                    self.unresolved.append((broker, identifier, resolution.status, resolution.status_reason))
                    continue
                role = dict(sources_for(resolution.exchange, resolution.segment, self.interval)).get(broker)
                if role is None:
                    continue
                key = (broker, identifier, resolution.instrument_id)
                seen_keys.add(key)
                if self.instrument_ids is not None and resolution.instrument_id not in self.instrument_ids:
                    continue
                previous = existing.get(key)
                changed = (previous is None or previous[4] != "active"
                           or previous[5] is None or earliest < previous[5]
                           or previous[6] is None or latest > previous[6])
                source_id = self.upsert_source(broker, identifier, resolution, role)
                sources.append(Source(source_id, broker, identifier, resolution.instrument_id, role,
                                      resolution.exchange, resolution.valid_from, resolution.valid_to,
                                      earliest, latest, changed))

    # A series that no longer resolves to an instrument it used to feed.
    for key, row in existing.items():
        if key in seen_keys or row[4] != "active":
            continue
        if self.instrument_ids is not None and key[2] not in self.instrument_ids:
            continue
        with self.connection.cursor() as cursor:
            cursor.execute(f"""
                update {tables.PRICE_HISTORY_SOURCES}
                set status = 'superseded', status_reason = 'series no longer resolves to this instrument'
                where source_id = %s
            """, (row[0],))
        superseded.add(key[2])
    self.connection.commit()
    self.counts["sources_superseded"] += len(superseded)

    # An instrument that lost a source is rebuilt from what it has left.
    if superseded:
        instruments_with_sources = {source.instrument_id for source in sources}
        for instrument_id in superseded:
            if instrument_id in instruments_with_sources:
                sources = [source._replace(changed=True) if source.instrument_id == instrument_id else source
                           for source in sources]
            else:
                sources.append(Source(None, None, None, instrument_id, None, None, None, None,
                                      None, None, True))
    return sources

upsert_source

upsert_source(broker, identifier, resolution, role)

Insert or refresh one source row, returning its id.

Parameters:

Name	Type	Description	Default
broker	str	The broker.	required
identifier	str	The broker's series identifier.	required
resolution	Resolution	The instrument it resolved to.	required
role	str	"primary" or "gap_fill".	required

Returns:

Type	Description
int	The source_id.

Example usage

From example_2_each_step_by_hand.py

print(f"upsert_source returns id {self.loader.upsert_source('flattrade', 'NSE|9999|NEWCO-EQ', resolution, 'primary')}")

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def upsert_source(self, broker, identifier, resolution, role):
    """
    Insert or refresh one source row, returning its id.

    Args:
        broker (str): The broker.
        identifier (str): The broker's series identifier.
        resolution (Resolution): The instrument it resolved to.
        role (str): "primary" or "gap_fill".

    Returns:
        int: The source_id.
    """
    with self.connection.cursor() as cursor:
        cursor.execute(f"""
            insert into {tables.PRICE_HISTORY_SOURCES}
                (broker, broker_series, "interval", instrument_id, resolved_by, mapping_first_date,
                 mapping_last_date, price_basis, role, status, status_reason)
            values (%s, %s, %s, %s, %s, %s, %s, %s, %s, 'active', %s)
            on conflict (broker, broker_series, "interval", instrument_id) do update set
                resolved_by = excluded.resolved_by,
                mapping_first_date = excluded.mapping_first_date,
                mapping_last_date = excluded.mapping_last_date,
                price_basis = excluded.price_basis,
                role = excluded.role,
                status = case when {tables.PRICE_HISTORY_SOURCES}.resolved_by = 'manual'
                              then {tables.PRICE_HISTORY_SOURCES}.status else 'active' end,
                status_reason = excluded.status_reason
            returning source_id
        """, (broker, identifier, self.interval, resolution.instrument_id, resolution.resolved_by,
              resolution.mapping_first_date, resolution.mapping_last_date,
              price_basis(resolution.segment), role, resolution.status_reason))
        return cursor.fetchone()[0]

read_series

read_series(source)

One source's bars, restricted to the part of the series that belongs to its instrument, with any confirmed correction from unified.price_history_corrections applied.

A corrected price is rounded to the paisa: the broker rounded when it divided, so multiplying back restores the raw price to within that rounding and no closer.

Parameters:

Name	Type	Description	Default
source	Source	The source.	required

Returns:

Type	Description
list[Bar]	The bars, oldest first.

Example usage

From example_2_each_step_by_hand.py

self.loader.read_series(corrected)

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def read_series(self, source):
    """
    One source's bars, restricted to the part of the series that belongs to its instrument, with
    any confirmed correction from unified.price_history_corrections applied.

    A corrected price is rounded to the paisa: the broker rounded when it divided, so multiplying
    back restores the raw price to within that rounding and no closer.

    Args:
        source (Source): The source.

    Returns:
        list[Bar]: The bars, oldest first.
    """
    conditions = ['instrument_token = %s', '"interval" = %s']
    parameters = [source.broker_series, self.interval]
    if source.valid_from is not None:
        conditions.append('"time" >= %s')
        parameters.append(source.valid_from)
    if source.valid_to is not None:
        conditions.append('"time" < %s')
        parameters.append(source.valid_to)
    with self.connection.cursor() as cursor:
        cursor.execute(f"""
            select "time", open, high, low, close, volume, oi from {source.broker}.price_history
            where {' and '.join(conditions)} order by "time"
        """, parameters)
        bars = [Bar(*row) for row in cursor.fetchall()]
        cursor.execute(f"""
            select valid_from, valid_to, price_multiplier, volume_multiplier from {tables.CORRECTION_RANGES}
            where broker = %s and broker_series = %s and "interval" = %s
        """, (source.broker, source.broker_series, self.interval))
        ranges = cursor.fetchall()
    if not ranges:
        return bars
    corrected = []
    for bar in bars:
        for valid_from, valid_to, price_multiplier, volume_multiplier in ranges:
            if valid_from <= bar.time < valid_to:
                bar = corrected_bar(bar, price_multiplier, volume_multiplier)
                self.counts["bars_corrected"] += 1
                break
        corrected.append(bar)
    return corrected

stitch

stitch(members)

Lay one broker's series for an instrument over each other, the most recent series on top.

Parameters:

Name	Type	Description	Default
members	list[Source]	The instrument's sources at one broker.	required

Returns:

Type	Description
dict	Bar time to (Bar, source_id).

Example usage

From example_2_each_step_by_hand.py

primary_bars = self.loader.stitch([zerodha])

From example_2_each_step_by_hand.py

print(f'agreement with Dhan: {self.loader.agreement(primary_bars, self.loader.stitch([dhan]))}')

From example_2_each_step_by_hand.py

print(f'agreement with Flattrade: {self.loader.agreement(primary_bars, self.loader.stitch([flattrade]))}')

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def stitch(self, members):
    """
    Lay one broker's series for an instrument over each other, the most recent series on top.

    Args:
        members (list[Source]): The instrument's sources at one broker.

    Returns:
        dict: Bar time to (Bar, source_id).
    """
    ordered = sorted(members, key=lambda source: source.broker_latest)
    bars = {}
    for source in ordered:
        for bar in self.read_series(source):
            bars[bar.time] = (bar, source.source_id)
    return bars

valid

valid(bar, exchange)

Whether a bar passes the calendar, grid and range checks, counting why when it does not.

Parameters:

Name	Type	Description	Default
bar	Bar	The bar.	required
exchange	str	The instrument's exchange.	required

Returns:

Type	Description
bool	True if the bar is kept.

Example usage

From example_2_each_step_by_hand.py

print(f'valid, {label}: {self.loader.valid(bar, "nse")}')

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def valid(self, bar, exchange):
    """
    Whether a bar passes the calendar, grid and range checks, counting why when it does not.

    Args:
        bar (Bar): The bar.
        exchange (str): The instrument's exchange.

    Returns:
        bool: True if the bar is kept.
    """
    if None in (bar.open, bar.high, bar.low, bar.close) or min(bar.open, bar.high, bar.low, bar.close) <= 0:
        self.counts["filtered_missing_price"] += 1
        return False
    if exchange in ("nse", "bse"):
        day = bar.time.astimezone(datetime.timezone(datetime.timedelta(hours=5, minutes=30))).date()
        if not self.calendar.is_trading_day(exchange, day):
            self.counts["filtered_non_trading_day"] += 1
            return False
    if not on_grid(bar.time, self.interval):
        self.counts["filtered_off_grid"] += 1
        return False
    if bar.high < max(bar.open, bar.close) or bar.low > min(bar.open, bar.close):
        self.counts["filtered_bad_range"] += 1
        return False
    return True

agreement staticmethod

agreement(primary, other)

The share of shared bar times on which two brokers' closes agree.

Parameters:

Name	Type	Description	Default
primary	dict	Bar time to (Bar, source_id).	required
other	dict	Bar time to (Bar, source_id).	required

Returns:

Type	Description
tuple[float, int]	The share agreeing, and how many bars were compared.

Example usage

From example_2_each_step_by_hand.py

print(f'agreement with Dhan: {self.loader.agreement(primary_bars, self.loader.stitch([dhan]))}')

From example_2_each_step_by_hand.py

print(f'agreement with Flattrade: {self.loader.agreement(primary_bars, self.loader.stitch([flattrade]))}')

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

@staticmethod
def agreement(primary, other):
    """
    The share of shared bar times on which two brokers' closes agree.

    Args:
        primary (dict): Bar time to (Bar, source_id).
        other (dict): Bar time to (Bar, source_id).

    Returns:
        tuple[float, int]: The share agreeing, and how many bars were compared.
    """
    shared = primary.keys() & other.keys()
    if not shared:
        return 0.0, 0
    agreeing = 0
    for moment in shared:
        first = primary[moment][0].close
        second = other[moment][0].close
        if first is not None and second is not None and abs(first - second) <= AGREEMENT_TOLERANCE * abs(first):
            agreeing += 1
    return agreeing / len(shared), len(shared)

corrected_bar

corrected_bar(bar, price_multiplier, volume_multiplier)

A bar with a correction applied.

Parameters:

Name	Type	Description	Default
bar	Bar	The bar as served.	required
price_multiplier	Decimal	What the prices were divided by.	required
volume_multiplier	Decimal	What the volume was divided by.	required

Returns:

Type	Description
Bar	The corrected bar.

Source code in stock_brokers/instruments/historical/utilities/unified/loader.py

def corrected_bar(bar, price_multiplier, volume_multiplier):
    """
    A bar with a correction applied.

    Args:
        bar (Bar): The bar as served.
        price_multiplier (Decimal): What the prices were divided by.
        volume_multiplier (Decimal): What the volume was divided by.

    Returns:
        Bar: The corrected bar.
    """
    def price(value):
        # Half away from zero, as PostgreSQL's round() does, so the check in `verify` agrees to the paisa.
        return None if value is None else (value * price_multiplier).quantize(PAISA, rounding=ROUND_HALF_UP)
    volume = bar.volume if bar.volume is None or volume_multiplier == 1 else int(round(bar.volume * volume_multiplier))
    return bar._replace(open=price(bar.open), high=price(bar.high), low=price(bar.low), close=price(bar.close),
                        volume=volume)