Skip to content

orchestrator

Runs instrument master ingestion across every broker.

Each broker is downloaded in its own try/except, so one broker's endpoint being down, or its token having expired, never costs the others their snapshot for the day. That matters more here than it would elsewhere: nine of the ten brokers publish only today's file, and Kotak's URL is stamped with today's date, so a snapshot missed is a snapshot that can never be fetched later.

Functions:

Name Description
ingest_one

Ingest one broker's instrument master, catching any failure rather than raising.
ingest_all

Ingest every named broker in turn, isolating each one's failure from the rest.

ingest_one 

ingest_one(broker_name, download_date=None, bootstrap=False)

Ingest one broker's instrument master, catching any failure rather than raising.

  • broker_name is the name of the broker to ingest.
  • download_date is the snapshot date to record, defaulting to today.
  • bootstrap replaces that date's stored rows rather than skipping when already ingested.

Returns a (rows, deviation, error) triple, where deviation is the row count canary's verdict and error is None on success.

Source code in stock_brokers/instruments/orchestrator.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def ingest_one(broker_name, download_date=None, bootstrap=False):
    """
    Ingest one broker's instrument master, catching any failure rather than raising.

    - `broker_name` is the name of the broker to ingest.
    - `download_date` is the snapshot date to record, defaulting to today.
    - `bootstrap` replaces that date's stored rows rather than skipping when already ingested.

    Returns a `(rows, deviation, error)` triple, where `deviation` is the row count canary's
    verdict and `error` is None on success.
    """
    try:
        ingester = INGESTERS[broker_name]()
        rows = ingester.ingest(download_date=download_date, bootstrap=bootstrap)
        deviation = ingester.check_row_count_deviation(download_date or date.today()) if rows else None
        return rows, deviation, None
    except Exception as exception:
        return 0, None, f"{type(exception).__name__}: {str(exception).splitlines()[0]}"

ingest_all 

ingest_all(brokers=INSTRUMENT_BROKERS, download_date=None, bootstrap=False)

Ingest every named broker in turn, isolating each one's failure from the rest.

  • brokers is the list of broker names to ingest.
  • download_date is the snapshot date to record, defaulting to today.
  • bootstrap replaces that date's stored rows rather than skipping when already ingested.
Source code in stock_brokers/instruments/orchestrator.py 
57
58
59
60
61
62
63
64
65
66
def ingest_all(brokers=INSTRUMENT_BROKERS, download_date=None, bootstrap=False):
    """
    Ingest every named broker in turn, isolating each one's failure from the rest.

    - `brokers` is the list of broker names to ingest.
    - `download_date` is the snapshot date to record, defaulting to today.
    - `bootstrap` replaces that date's stored rows rather than skipping when already ingested.
    """
    return {broker_name: ingest_one(broker_name, download_date, bootstrap)
            for broker_name in brokers}