Shared machinery for downloading a broker's historical candles into <broker>.price_history.
Each broker subclasses BrokerCandles and supplies four things: which intervals it serves and
what it calls them, how far back a single request may reach, how to fetch one window, and how to
read the response. Everything else - the rate limiter, the work queue, the watermarks, the
backoff, the upsert - is here, so a broker module is short.
Why this is a queue rather than a loop¶
The job is enormous. Zerodha alone publishes over 110,000 instruments and serves eight intervals, which is close to a million series, and the deepest of them reach back to 2005. At the three requests a second Kite allows, a complete backfill is measured in weeks. So this is not a script that runs to completion; it is a worker that makes progress whenever it is running and can be stopped at any moment.
Two consequences shape everything below. Work is claimed one window at a time from a progress table, so the process holds no state that a restart would lose. And the queue is ordered by priority, so that the data most likely to be wanted - cash daily bars - lands on the first night, while the long tail of expired option minute bars drains for as long as it is allowed to.
Nothing is derived¶
A five minute bar is whatever the broker returned for five minutes. It is not five one minute bars added together. That costs roughly three times the requests, and it means the stored data can be compared against the broker's own charts without an argument about aggregation boundaries.
Classes:
| Name | Description |
|---|---|
CandleError |
Base for the failures a candle download distinguishes between. |
CandleThrottled |
The broker asked us to slow down. Back off and retry the same window. |
CandleAuthenticationError |
The session is not usable. Nothing else will work either, so stop this broker. |
CandleBlocked |
The broker is refusing this client outright, not pacing it. Stop this broker. |
CandleInstrumentUnknown |
The broker will not serve this series at all. |
RateLimiter |
Holds a caller to a request rate, and to a daily budget where the broker sets one. |
BrokerCandles |
Base class for one broker's historical candle download. |
Functions:
| Name | Description |
|---|---|
is_intraday |
Whether a stored interval name divides the session rather than spanning whole days. |
daily_bar_time |
Put a daily or coarser bar at midnight India time on the trading date it belongs to. |
Bases: Exception
Base for the failures a candle download distinguishes between.
Example programs
examples/stock_brokers/instruments/historical/base/CandleError/example_1_catching_every_candle_failure.py
"""Catches every kind of candle download failure with the one base class, `CandleError`.
`CandleError` is the parent of the four failures a candle download tells apart: a throttle, a refused session, a block on the whole client and an instrument the broker will not serve. Code that only needs to know that a window failed for a reason the download understands can catch `CandleError`, and code that must react differently to each kind catches the subclasses.
This program raises one of each subclass in turn, catches it as `CandleError`, and prints the class, its message and its place in the hierarchy. It needs no broker and no data store.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleError/example_1_catching_every_candle_failure.py
"""
from stock_brokers.instruments.historical.base import (
CandleAuthenticationError,
CandleBlocked,
CandleError,
CandleInstrumentUnknown,
CandleThrottled,
)
class CatchingEveryCandleFailureExample:
"""Raises each candle failure and catches it as a `CandleError`.
Attributes:
failures (list): The exceptions to raise, one of each subclass.
"""
def __init__(self):
"""Builds one failure of each kind, with the message a broker might have caused.
Returns:
None: This method returns nothing.
"""
self.failures = [
CandleThrottled('Too many requests'),
CandleAuthenticationError('Incorrect `api_key` or `access_token`.'),
CandleBlocked('error code: 1015'),
CandleInstrumentUnknown('invalid token'),
]
def raise_failure(self, failure):
"""Raises one failure, as a broker module's `fetch_candles` would.
Args:
failure (CandleError): The failure to raise.
Returns:
None: This method returns nothing.
Raises:
CandleError: Always, the failure given.
"""
raise failure
def run(self):
"""Raises and catches each failure, then prints the class hierarchy.
Returns:
None: This method returns nothing.
"""
for failure in self.failures:
try:
self.raise_failure(failure)
except CandleError as error:
print(f'Caught {type(error).__name__} as CandleError: {error}')
print(f'CandleError is an Exception: {issubclass(CandleError, Exception)}')
subclass_names = []
for subclass in CandleError.__subclasses__():
subclass_names.append(subclass.__name__)
print(f'Classes that inherit from CandleError directly: {subclass_names}')
if __name__ == '__main__':
CatchingEveryCandleFailureExample().run()
It prints:
Caught CandleThrottled as CandleError: Too many requests
Caught CandleAuthenticationError as CandleError: Incorrect `api_key` or `access_token`.
Caught CandleBlocked as CandleError: error code: 1015
Caught CandleInstrumentUnknown as CandleError: invalid token
CandleError is an Exception: True
Classes that inherit from CandleError directly: ['CandleThrottled', 'CandleAuthenticationError', 'CandleBlocked', 'CandleInstrumentUnknown']
examples/stock_brokers/instruments/historical/base/CandleError/example_2_choosing_a_reaction.py
"""Decides what to do about a failed window from which `CandleError` subclass it raised.
A candle download reacts to each failure differently: it waits and retries after a throttle, stops the broker after a refused session or a block, and retires a series the broker will not serve. Anything that is not a `CandleError` at all is an unexpected failure, which is recorded against the one series. This program writes that decision as a small class, `FailureReaction`, and runs it over a list of failures, including a plain `ValueError` and a bare `CandleError`.
A bare `CandleError` falls through every subclass test, so the program shows that catching the base class is what keeps a new, unclassified failure from being mistaken for a crash. It needs no broker and no data store.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleError/example_2_choosing_a_reaction.py
"""
from stock_brokers.instruments.historical.base import (
CandleAuthenticationError,
CandleBlocked,
CandleError,
CandleInstrumentUnknown,
CandleThrottled,
)
class FailureReaction:
"""Chooses how a candle download reacts to one failure."""
def reaction_to(self, failure):
"""Names the reaction a failure calls for.
Args:
failure (Exception): The failure a window ended with.
Returns:
str: What the download does next.
"""
try:
raise failure
except CandleThrottled:
return 'back off and retry the same window later'
except (CandleAuthenticationError, CandleBlocked):
return 'stop this broker'
except CandleInstrumentUnknown:
return 'retire the series'
except CandleError:
return 'record a candle failure against the series'
except Exception:
return 'record an unexpected failure against the series'
class ChoosingAReactionExample:
"""Prints the reaction to each of a list of failures.
Attributes:
reaction (FailureReaction): The decision being shown.
failures (list): The failures to react to.
"""
def __init__(self):
"""Builds the list of failures.
Returns:
None: This method returns nothing.
"""
self.reaction = FailureReaction()
self.failures = [
CandleThrottled('Too many requests'),
CandleAuthenticationError('Session Expired'),
CandleBlocked('error code: 1015'),
CandleInstrumentUnknown('invalid token'),
CandleError('the broker answered in a shape no parser knows'),
ValueError('could not convert string to float'),
]
def run(self):
"""Prints each failure with its reaction.
Returns:
None: This method returns nothing.
"""
for failure in self.failures:
print(f'{type(failure).__name__} ({failure}): {self.reaction.reaction_to(failure)}')
if __name__ == '__main__':
ChoosingAReactionExample().run()
It prints:
CandleThrottled (Too many requests): back off and retry the same window later
CandleAuthenticationError (Session Expired): stop this broker
CandleBlocked (error code: 1015): stop this broker
CandleInstrumentUnknown (invalid token): retire the series
CandleError (the broker answered in a shape no parser knows): record a candle failure against the series
ValueError (could not convert string to float): record an unexpected failure against the series
Bases: CandleError
The broker asked us to slow down. Back off and retry the same window.
Example programs
examples/stock_brokers/instruments/historical/base/CandleThrottled/example_1_retrying_after_a_throttle.py
"""Raises `CandleThrottled` when a broker asks for fewer requests, then backs off and retries the same window.
A throttle says something about the pace, not about the instrument, so the right reaction is to wait and ask again for the same window. This program has a stand-in broker that refuses the first request with the message Kite sends when it is asked too often, and a small fetcher that turns that message into `CandleThrottled`. The caller catches it, backs the real `RateLimiter` off by a tenth of a second and asks again, and the second answer comes back with candles.
The stand-in stands for the broker so that nothing leaves the machine. The back-off is kept to a tenth of a second so the program finishes quickly; the candle download itself backs off five seconds.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleThrottled/example_1_retrying_after_a_throttle.py
"""
from stock_brokers.instruments.historical.base import (
CandleThrottled,
RateLimiter,
)
class StandInBroker:
"""A stand-in for a broker's history endpoint that refuses its first request.
Attributes:
requests (int): How many requests it has received.
"""
def __init__(self):
"""Builds the stand-in with no requests received.
Returns:
None: This method returns nothing.
"""
self.requests = 0
def candles(self, token):
"""Answers a history request, or refuses the first one.
Args:
token (str): The instrument token.
Returns:
dict: The decoded response.
"""
self.requests += 1
if self.requests == 1:
return {
'status': 'error',
'error_type': 'NetworkException',
'message': 'Too many requests',
}
return {
'status': 'success',
'candles': [
[
'2026-09-29T00:00:00+0530',
1391.0,
1402.5,
1385.2,
1398.4,
6123450,
],
],
'token': token,
}
class ThrottleAwareFetcher:
"""Fetches candles and raises `CandleThrottled` when the broker asks for fewer requests.
Attributes:
broker (StandInBroker): Where candles come from.
"""
def __init__(self, broker):
"""Wraps a broker.
Args:
broker (StandInBroker): Where candles come from.
Returns:
None: This method returns nothing.
"""
self.broker = broker
def fetch(self, token):
"""Fetches one window of candles.
Args:
token (str): The instrument token.
Returns:
list: The candles.
Raises:
CandleThrottled: When the broker refuses the request as too frequent.
"""
answer = self.broker.candles(token)
if answer['status'] == 'error' and answer['error_type'] == 'NetworkException':
raise CandleThrottled(answer['message'])
return answer['candles']
class RetryingAfterAThrottleExample:
"""Fetches one window, backing off and retrying once when throttled.
Attributes:
broker (StandInBroker): The stand-in broker.
fetcher (ThrottleAwareFetcher): The fetcher that raises the throttle.
limiter (RateLimiter): The pace every request is held to.
"""
def __init__(self):
"""Builds the broker, the fetcher and a limiter for ten requests a second.
Returns:
None: This method returns nothing.
"""
self.broker = StandInBroker()
self.fetcher = ThrottleAwareFetcher(self.broker)
self.limiter = RateLimiter(10.0)
def run(self):
"""Fetches the window, retrying after a throttle, and prints each step.
Returns:
None: This method returns nothing.
"""
candles = None
while candles is None:
self.limiter.take()
try:
candles = self.fetcher.fetch('738561')
except CandleThrottled as error:
print(f'Throttled: {error}; backing off and asking again for the same window')
self.limiter.back_off(0.1)
print(f'Received {len(candles)} candle after {self.broker.requests} requests: {candles[0]}')
if __name__ == '__main__':
RetryingAfterAThrottleExample().run()
It prints:
examples/stock_brokers/instruments/historical/base/CandleThrottled/example_2_budget_spent_for_the_day.py
"""Shows `CandleThrottled` raised by a `RateLimiter` whose daily budget is spent, and the loop stopping for the day.
`CandleThrottled` has two sources. A broker raises it through `fetch_candles` when it is asked too often, and the `RateLimiter` raises it itself when a broker's daily request budget is used up. The second kind will not clear by waiting a few seconds, so the candle download's `run` stops when it meets it and leaves the rest of the queue for tomorrow.
This program runs a small loop over five series with a budget of two requests a day. No broker is contacted: each "request" only prints the series it would have fetched. It also shows that `CandleThrottled` is a `CandleError`.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleThrottled/example_2_budget_spent_for_the_day.py
"""
from stock_brokers.instruments.historical.base import (
CandleError,
CandleThrottled,
RateLimiter,
)
class BudgetSpentForTheDayExample:
"""Works through a queue until the daily budget stops it.
Attributes:
limiter (RateLimiter): The limiter with a two-request daily budget.
queue (list): The series tokens waiting to be fetched.
"""
def __init__(self):
"""Builds the limiter and the queue.
Returns:
None: This method returns nothing.
"""
self.limiter = RateLimiter(100.0, requests_per_day=2)
self.queue = [
'738561',
'2953217',
'408065',
'884737',
'341249',
]
def run(self):
"""Fetches series until the budget is spent, then prints what is left.
Returns:
None: This method returns nothing.
"""
fetched = []
for token in self.queue:
try:
self.limiter.take()
except CandleThrottled as error:
print(f'Stopping for the day: {error}')
print(f'It is a CandleError: {isinstance(error, CandleError)}')
break
print(f'Fetching {token}')
fetched.append(token)
left = []
for token in self.queue:
if token not in fetched:
left.append(token)
print(f'Left for tomorrow: {left}')
if __name__ == '__main__':
BudgetSpentForTheDayExample().run()
It prints:
Bases: CandleError
The session is not usable. Nothing else will work either, so stop this broker.
Example programs
examples/stock_brokers/instruments/historical/base/CandleAuthenticationError/example_1_refused_token.py
"""Raises `CandleAuthenticationError` when a broker refuses the session, keeping the broker's own error as its cause.
A refused session is not a fault of the series being fetched, and nothing else will work until the session is fixed, so the candle download treats this failure as a reason to log in again once, and to stop the broker if that does not help. A broker module raises it from `fetch_candles` when the broker's answer says the token is no good.
This program has a stand-in broker client that fails the way Kite's client does, with an exception named `TokenException`. A small fetcher raises `CandleAuthenticationError` from it with `raise ... from`, so the original exception stays attached as `__cause__` for the log. No broker is contacted.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleAuthenticationError/example_1_refused_token.py
"""
from stock_brokers.instruments.historical.base import (
CandleAuthenticationError,
CandleError,
)
class TokenException(Exception):
"""The stand-in broker client's error for a token it does not accept."""
class StandInBrokerClient:
"""A stand-in for a broker client whose stored token has expired."""
def historical_data(self, token):
"""Refuses every request, as a broker does with an expired token.
Args:
token (str): The instrument token.
Returns:
dict: Never returns.
Raises:
TokenException: Always.
"""
raise TokenException(f'Incorrect `api_key` or `access_token` while fetching {token}.')
class SessionAwareFetcher:
"""Fetches candles and raises `CandleAuthenticationError` when the session is refused.
Attributes:
client (StandInBrokerClient): The broker client.
"""
def __init__(self, client):
"""Wraps a broker client.
Args:
client (StandInBrokerClient): The broker client.
Returns:
None: This method returns nothing.
"""
self.client = client
def fetch(self, token):
"""Fetches one window of candles.
Args:
token (str): The instrument token.
Returns:
dict: The broker's answer.
Raises:
CandleAuthenticationError: When the broker refuses the session.
"""
try:
return self.client.historical_data(token)
except TokenException as error:
raise CandleAuthenticationError(str(error)) from error
class RefusedTokenExample:
"""Fetches with an expired token and prints the failure it produces.
Attributes:
fetcher (SessionAwareFetcher): The fetcher being shown.
"""
def __init__(self):
"""Builds the fetcher around the stand-in client.
Returns:
None: This method returns nothing.
"""
self.fetcher = SessionAwareFetcher(StandInBrokerClient())
def run(self):
"""Fetches once and prints the failure, its cause and its place in the hierarchy.
Returns:
None: This method returns nothing.
"""
try:
self.fetcher.fetch('738561')
except CandleAuthenticationError as error:
print(f'CandleAuthenticationError: {error}')
print(f'Caused by: {type(error.__cause__).__name__}')
print(f'It is a CandleError: {isinstance(error, CandleError)}')
if __name__ == '__main__':
RefusedTokenExample().run()
It prints:
examples/stock_brokers/instruments/historical/base/CandleAuthenticationError/example_2_one_login_then_give_up.py
"""Logs in again once after a `CandleAuthenticationError`, and gives up when the fresh session is refused as well.
The candle download's rule for a refused session is one login, then the same window again. If the fresh session is refused too, the error goes back up and the broker is stopped, because logging in over and over is how a broker account gets locked. This program writes that rule as a small class, `OneLoginPolicy`, and runs it against two stand-in brokers: one that accepts the new token and one that refuses every token.
The stand-in login only counts how often it was called; nothing logs in anywhere.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleAuthenticationError/example_2_one_login_then_give_up.py
"""
from stock_brokers.instruments.historical.base import (
CandleAuthenticationError,
)
class StandInBroker:
"""A stand-in broker that refuses the old token, and optionally every token.
Attributes:
refuses_every_token (bool): Whether even a fresh token is refused.
current_token (str): The token requests are sent with.
logins (int): How many logins have been made.
"""
def __init__(self, refuses_every_token):
"""Builds the broker with an expired token in hand.
Args:
refuses_every_token (bool): Whether even a fresh token is refused.
Returns:
None: This method returns nothing.
"""
self.refuses_every_token = refuses_every_token
self.current_token = 'expired'
self.logins = 0
def log_in(self):
"""Pretends to log in, issuing a fresh token.
Returns:
None: This method returns nothing.
"""
self.logins += 1
self.current_token = f'fresh-{self.logins}'
def fetch(self, token):
"""Fetches one window of candles.
Args:
token (str): The instrument token.
Returns:
str: A description of the candles received.
Raises:
CandleAuthenticationError: When the token in hand is refused.
"""
if self.refuses_every_token or self.current_token == 'expired':
raise CandleAuthenticationError(f'token {self.current_token} refused')
return f'375 candles for {token}'
class OneLoginPolicy:
"""Fetches a window, logging in once if the session is refused.
Attributes:
broker (StandInBroker): The broker to fetch from.
"""
def __init__(self, broker):
"""Wraps a broker.
Args:
broker (StandInBroker): The broker to fetch from.
Returns:
None: This method returns nothing.
"""
self.broker = broker
def fetch(self, token):
"""Fetches one window, with at most one login in between.
Args:
token (str): The instrument token.
Returns:
str: A description of the candles received.
Raises:
CandleAuthenticationError: When the session is refused even after one login.
"""
try:
return self.broker.fetch(token)
except CandleAuthenticationError as error:
print(f' refused ({error}); logging in once')
self.broker.log_in()
return self.broker.fetch(token)
class OneLoginThenGiveUpExample:
"""Runs the policy against a broker that recovers and one that does not."""
def run(self):
"""Prints what happens with each broker.
Returns:
None: This method returns nothing.
"""
for refuses_every_token in (
False,
True,
):
broker = StandInBroker(refuses_every_token)
print(f'Broker that refuses every token: {refuses_every_token}')
try:
print(f' {OneLoginPolicy(broker).fetch("738561")}')
except CandleAuthenticationError as error:
print(f' giving up and stopping the broker: {error}')
print(f' logins made: {broker.logins}')
if __name__ == '__main__':
OneLoginThenGiveUpExample().run()
It prints:
Bases: CandleError
The broker is refusing this client outright, not pacing it. Stop this broker.
The difference from CandleThrottled is what retrying does. A throttle clears once the pace
drops, so backing off and moving on to the next series is right. A block - a firewall ban on
the IP address, such as Cloudflare's error 1015 in front of Fyers - is extended by every
further request, and moving on to the next series is simply another request. It is also not a
session problem, so logging in again would only be one more refused request. The one useful
response is to stop and let the process manager bring the download back later.
Example programs
examples/stock_brokers/instruments/historical/base/CandleBlocked/example_1_firewall_ban.py
"""Raises `CandleBlocked` when a firewall in front of the broker bans the client, rather than treating it as a throttle.
A throttle clears once requests slow down, but a ban such as Cloudflare's error 1015 in front of Fyers is extended by every further request. So a broker module that sees one raises `CandleBlocked`, and the candle download stops the broker instead of backing off and trying the next series. This program has a stand-in HTTP answer shaped like Cloudflare's ban page and a small classifier that tells it apart from an ordinary 429 throttle.
No request leaves the machine: the answers are canned.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleBlocked/example_1_firewall_ban.py
"""
from stock_brokers.instruments.historical.base import (
CandleBlocked,
CandleError,
CandleThrottled,
)
class StandInResponse:
"""A stand-in for an HTTP response.
Attributes:
status_code (int): The HTTP status.
text (str): The response body.
"""
def __init__(self, status_code, text):
"""Holds a canned response.
Args:
status_code (int): The HTTP status.
text (str): The response body.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.text = text
class RefusalClassifier:
"""Turns a refused HTTP response into the candle failure it stands for."""
def check(self, response):
"""Raises the failure a refused response stands for.
Args:
response (StandInResponse): The response to check.
Returns:
None: This method returns nothing when the response is not a refusal.
Raises:
CandleBlocked: When a firewall has banned the client.
CandleThrottled: When the broker asks for fewer requests.
"""
if 'error code: 1015' in response.text:
raise CandleBlocked(f'HTTP {response.status_code}: the firewall has banned this address')
if response.status_code == 429:
raise CandleThrottled(f'HTTP {response.status_code}: {response.text}')
class FirewallBanExample:
"""Checks a throttle and a ban and prints what each became.
Attributes:
classifier (RefusalClassifier): The classifier being shown.
responses (list): The canned responses.
"""
def __init__(self):
"""Builds the classifier and two canned refusals.
Returns:
None: This method returns nothing.
"""
self.classifier = RefusalClassifier()
self.responses = [
StandInResponse(429, '{"s":"error","code":429,"message":"request limit reached"}'),
StandInResponse(429, '<html><title>Access denied | api-t1.fyers.in</title>error code: 1015</html>'),
]
def run(self):
"""Checks each response and prints the failure it raised.
Returns:
None: This method returns nothing.
"""
for response in self.responses:
try:
self.classifier.check(response)
except CandleBlocked as error:
print(f'CandleBlocked: {error}; stop this broker and send nothing more')
except CandleThrottled as error:
print(f'CandleThrottled: {error}; back off and carry on')
print(f'CandleBlocked is a CandleError: {issubclass(CandleBlocked, CandleError)}')
print(f'CandleBlocked is a kind of throttle: {issubclass(CandleBlocked, CandleThrottled)}')
if __name__ == '__main__':
FirewallBanExample().run()
It prints:
examples/stock_brokers/instruments/historical/base/CandleBlocked/example_2_stopping_the_whole_queue.py
"""Stops a whole queue of series on `CandleBlocked`, where a throttle only costs the one series a pause.
Moving on to the next series after a ban is simply another request to a firewall that is counting them. This program works a queue of five series against a stand-in broker that throttles the second series and bans the client on the fourth. The loop carries on past the throttle and stops at the ban, leaving the fifth series untouched, which is how the candle download's `run` treats the two failures.
The stand-in broker answers from a script, so no request leaves the machine.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleBlocked/example_2_stopping_the_whole_queue.py
"""
from stock_brokers.instruments.historical.base import (
CandleBlocked,
CandleThrottled,
)
class StandInBroker:
"""A stand-in broker that answers each token from a script.
Attributes:
requests (list): The tokens asked for, in order.
"""
def __init__(self):
"""Builds the broker with no requests received.
Returns:
None: This method returns nothing.
"""
self.requests = []
def fetch(self, token):
"""Fetches one window of candles for a token.
Args:
token (str): The instrument token.
Returns:
int: How many candles came back.
Raises:
CandleThrottled: For the token scripted to be throttled.
CandleBlocked: For the token scripted to meet the ban.
"""
self.requests.append(token)
if token == 'NSE:SBIN-EQ':
raise CandleThrottled('request limit reached')
if token == 'NSE:INFY-EQ':
raise CandleBlocked('error code: 1015')
return 75
class StoppingTheWholeQueueExample:
"""Works a queue until a ban stops it.
Attributes:
broker (StandInBroker): The stand-in broker.
queue (list): The series to fetch.
"""
def __init__(self):
"""Builds the broker and the queue.
Returns:
None: This method returns nothing.
"""
self.broker = StandInBroker()
self.queue = [
'NSE:RELIANCE-EQ',
'NSE:SBIN-EQ',
'NSE:TCS-EQ',
'NSE:INFY-EQ',
'NSE:HDFCBANK-EQ',
]
def run(self):
"""Works the queue and prints each series' outcome.
Returns:
None: This method returns nothing.
"""
for token in self.queue:
try:
candles = self.broker.fetch(token)
except CandleThrottled as error:
print(f'{token}: throttled ({error}), left claimable for later')
continue
except CandleBlocked as error:
print(f'{token}: blocked ({error}), stopping the broker')
break
print(f'{token}: {candles} candles stored')
print(f'Requests sent: {len(self.broker.requests)} of {len(self.queue)}')
if __name__ == '__main__':
StoppingTheWholeQueueExample().run()
It prints:
Bases: CandleError
The broker will not serve this series at all.
Usually an expired contract: Kite answers invalid token for an instrument that has left its
dump, and no amount of retrying changes that. The series is retired rather than retried.
Example programs
examples/stock_brokers/instruments/historical/base/CandleInstrumentUnknown/example_1_retiring_an_expired_contract.py
"""Raises `CandleInstrumentUnknown` for a contract the broker no longer knows, and retires its series instead of retrying it.
Kite answers `invalid token` for an instrument that has left its instrument dump, which is usually an expired contract. No amount of retrying changes that answer, so a broker module raises `CandleInstrumentUnknown` and the candle download marks the series retired. This program has a stand-in broker that knows only live tokens, and a small queue that retires every series the broker does not know.
The stand-in broker answers from memory, so no request leaves the machine.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleInstrumentUnknown/example_1_retiring_an_expired_contract.py
"""
from stock_brokers.instruments.historical.base import (
CandleError,
CandleInstrumentUnknown,
)
class StandInBroker:
"""A stand-in broker that serves only the tokens in its current instrument dump.
Attributes:
live_tokens (set): The tokens the broker still serves.
"""
def __init__(self):
"""Builds the broker with two live tokens.
Returns:
None: This method returns nothing.
"""
self.live_tokens = {
'738561',
'13368834',
}
def fetch(self, token):
"""Fetches one window of candles.
Args:
token (str): The instrument token.
Returns:
int: How many candles came back.
Raises:
CandleInstrumentUnknown: When the token is not in the dump.
"""
if token not in self.live_tokens:
raise CandleInstrumentUnknown(f'invalid token: {token}')
return 250
class RetiringAnExpiredContractExample:
"""Fetches three series and retires the one the broker does not know.
Attributes:
broker (StandInBroker): The stand-in broker.
retired (dict): Token to the reason it was retired.
"""
def __init__(self):
"""Builds the broker and an empty list of retired series.
Returns:
None: This method returns nothing.
"""
self.broker = StandInBroker()
self.retired = {}
def run(self):
"""Fetches each series and prints what became of it.
Returns:
None: This method returns nothing.
"""
tokens = [
'738561',
'12094466',
'13368834',
]
for token in tokens:
try:
candles = self.broker.fetch(token)
except CandleInstrumentUnknown as error:
self.retired[token] = str(error)
print(f'{token}: retired')
continue
print(f'{token}: {candles} candles')
print(f'Retired series and why: {self.retired}')
print(f'CandleInstrumentUnknown is a CandleError: {issubclass(CandleInstrumentUnknown, CandleError)}')
if __name__ == '__main__':
RetiringAnExpiredContractExample().run()
It prints:
examples/stock_brokers/instruments/historical/base/CandleInstrumentUnknown/example_2_telling_a_dead_instrument_from_a_bad_request.py
"""Raises `CandleInstrumentUnknown` only for a refusal that really means the instrument is gone, not for one that means the request was wrong.
Retiring a series is permanent, so a broker module has to be careful about which refusals it turns into `CandleInstrumentUnknown`. Dhan, for example, answers `Input_Exception` both for an instrument it will not serve and for a window that spans too many days; only the first is a dead instrument, and the second must stay an ordinary error so the series is retried with a narrower window. This program writes that distinction as a small classifier and runs it over two Dhan-shaped messages.
The messages are canned copies of what Dhan returns, so no request leaves the machine.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/CandleInstrumentUnknown/example_2_telling_a_dead_instrument_from_a_bad_request.py
"""
from stock_brokers.instruments.historical.base import (
CandleInstrumentUnknown,
)
class DhanStyleClassifier:
"""Decides whether a Dhan-style refusal means the instrument is gone."""
def raise_for(self, message):
"""Raises the failure a refusal stands for.
Args:
message (str): The broker's error text.
Returns:
None: This method returns nothing.
Raises:
CandleInstrumentUnknown: When the refusal means the broker will not serve the instrument.
RuntimeError: For any other refusal, which is retried later.
"""
if 'Input_Exception' in message and 'can be fetched for' not in message:
raise CandleInstrumentUnknown(message)
raise RuntimeError(message)
class TellingADeadInstrumentFromABadRequestExample:
"""Classifies two refusals and prints which would retire a series.
Attributes:
classifier (DhanStyleClassifier): The classifier being shown.
messages (list): The canned refusals.
"""
def __init__(self):
"""Builds the classifier and the refusals.
Returns:
None: This method returns nothing.
"""
self.classifier = DhanStyleClassifier()
self.messages = [
"('Input_Exception', 'Missing required fields')",
"('Input_Exception', 'Data for Intraday Charts can be fetched for 90 days at a time')",
]
def run(self):
"""Classifies each refusal and prints the outcome.
Returns:
None: This method returns nothing.
"""
for message in self.messages:
try:
self.classifier.raise_for(message)
except CandleInstrumentUnknown:
print(f'{message}: CandleInstrumentUnknown, retire the series')
except RuntimeError:
print(f'{message}: an ordinary error, retry the series later')
if __name__ == '__main__':
TellingADeadInstrumentFromABadRequestExample().run()
It prints:
Holds a caller to a request rate, and to a daily budget where the broker sets one.
Deliberately a sleeping limiter with no queue and no threads. One process per broker makes the requests in sequence, which is the simplest thing that respects a published limit, and the limits here are low enough that concurrency would buy little before hitting them.
Rate limiter for one broker.
requests_per_secondis the sustained rate to hold.requests_per_dayis the broker's daily cap, or None when it publishes none.cacheis the Redis client that keeps the day's count, or None to count in this process only.budget_keyis the start of the Redis key for the day's count; the date in India is appended to it.
A count kept only in the process starts again at zero every time the downloader restarts, and the downloaders restart ten minutes after every exit, including the exit that follows a spent budget, so only a count in Redis holds the cap for a whole day.
Example programs
examples/stock_brokers/instruments/historical/base/RateLimiter/example_1_pacing_requests_and_backing_off.py
"""Holds a series of requests to a steady rate, and pushes the next one further out after the broker says no.
A `RateLimiter` is the one thing standing between a candle download and a broker's published request limit. Each call to `take` waits, if it has to, until the next request may be sent, and each call to `back_off` pushes that moment further away, which is what the download does for five seconds after a throttle.
This program holds a limiter to twenty requests a second, so consecutive requests are at least fifty milliseconds apart, and backs off by a fifth of a second. The limiter sleeps rather than queueing, so the program measures how long the calls took with a monotonic clock. It prints only whether each wait was at least as long as the limit requires, never the durations themselves, because those differ a little from run to run.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/RateLimiter/example_1_pacing_requests_and_backing_off.py
"""
import time
from stock_brokers.instruments.historical.base import (
RateLimiter,
)
class PacingRequestsAndBackingOffExample:
"""Takes a few request slots from a limiter and checks how far apart they were.
Attributes:
limiter (RateLimiter): The limiter being shown.
"""
REQUESTS_PER_SECOND = 20.0
BACK_OFF_SECONDS = 0.2
TOLERANCE_SECONDS = 0.005
def __init__(self):
"""Builds a limiter for twenty requests a second with no daily budget.
Returns:
None: This method returns nothing.
"""
self.limiter = RateLimiter(self.REQUESTS_PER_SECOND)
def run(self):
"""Takes four slots, backs off once, takes one more and prints whether the gaps held.
Returns:
None: This method returns nothing.
"""
minimum_gap = 1.0 / self.REQUESTS_PER_SECOND
started = time.monotonic()
for request_number in range(1, 5):
self.limiter.take()
print(f'Request {request_number} may be sent')
four_requests_took = time.monotonic() - started
print(f'Four requests were held at least three gaps apart: {four_requests_took >= 3 * minimum_gap - self.TOLERANCE_SECONDS}')
self.limiter.back_off(self.BACK_OFF_SECONDS)
print(f'The broker said no, so the limiter backs off by {self.BACK_OFF_SECONDS} seconds')
backed_off_at = time.monotonic()
self.limiter.take()
waited = time.monotonic() - backed_off_at
print('Request 5 may be sent')
print(f'Request 5 waited at least the back-off: {waited >= self.BACK_OFF_SECONDS - self.TOLERANCE_SECONDS}')
if __name__ == '__main__':
PacingRequestsAndBackingOffExample().run()
It prints:
examples/stock_brokers/instruments/historical/base/RateLimiter/example_2_daily_budget_spent.py
"""Refuses a request once a broker's daily budget is spent, by raising `CandleThrottled`.
Some brokers publish a daily cap on historical requests as well as a rate. A `RateLimiter` given `requests_per_day` counts every slot `take` hands out, and once the day's budget is used up it raises `CandleThrottled` instead of waiting. The candle download treats that as a reason to stop for the day rather than to retry, because waiting a few seconds would not help.
This program uses a budget of three requests and a rate of a hundred a second, so it finishes in a few hundredths of a second. The budget resets when the date changes, which the program does not wait for.
What to notice in the output: the first three requests are allowed, the fourth raises `CandleThrottled` with a message naming the budget, and a fifth attempt is refused the same way, because nothing about the budget changed.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/RateLimiter/example_2_daily_budget_spent.py
"""
from stock_brokers.instruments.historical.base import (
CandleThrottled,
RateLimiter,
)
class DailyBudgetSpentExample:
"""Takes request slots until the daily budget refuses one.
Attributes:
limiter (RateLimiter): The limiter being shown.
"""
def __init__(self):
"""Builds a limiter for a hundred requests a second and three a day.
Returns:
None: This method returns nothing.
"""
self.limiter = RateLimiter(100.0, requests_per_day=3)
def run(self):
"""Asks for five requests and prints which were allowed.
Returns:
None: This method returns nothing.
"""
for request_number in range(1, 6):
try:
self.limiter.take()
except CandleThrottled as error:
print(f'Request {request_number} refused: {error}')
continue
print(f'Request {request_number} allowed')
if __name__ == '__main__':
DailyBudgetSpentExample().run()
It prints:
examples/stock_brokers/instruments/historical/base/RateLimiter/example_3_budget_kept_in_redis.py
"""Keeps the daily budget in Redis, so a downloader that restarts carries on counting from where the last one stopped.
A candle downloader exits when its daily budget is spent, and systemd starts it again ten minutes later. A count kept only in the process would begin again at zero each time, so the cap would never hold. Given a Redis client and a key, a `RateLimiter` counts every request in `<key>:<date in India>` instead, which `spent_today` reads back and which expires after two days. This program uses a small stand-in for Redis, so it needs nothing running.
`set_requests_per_day` changes the cap while running, which the Fyers downloader uses to raise its cap at weekends; the count carries on either way.
Notice that the second limiter, standing in for the restarted process, starts at 2 rather than 0 and refuses its second request.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/RateLimiter/example_3_budget_kept_in_redis.py
"""
from stock_brokers.instruments.historical.base import (
CandleThrottled,
RateLimiter,
)
class StandInPipeline:
"""A pipeline that applies `INCR` to a stand-in Redis when executed.
Attributes:
cache (StandInRedis): The stand-in Redis.
keys (list): The keys to increment.
"""
def __init__(self, cache):
"""Builds the pipeline.
Args:
cache (StandInRedis): The stand-in Redis.
Returns:
None: This method returns nothing.
"""
self.cache = cache
self.keys = []
def incr(self, key):
"""Queues an `INCR`.
Args:
key (str): The key.
Returns:
StandInPipeline: This pipeline.
"""
self.keys.append(key)
return self
def expire(self, key, seconds):
"""Accepts an `EXPIRE`, which the stand-in does not act on.
Args:
key (str): The key.
seconds (int): The time to live.
Returns:
StandInPipeline: This pipeline.
"""
del key
del seconds
return self
def execute(self):
"""Applies the queued increments.
Returns:
None: This method returns nothing.
"""
for key in self.keys:
self.cache.values[key] = str(int(self.cache.values.get(key, '0')) + 1)
class StandInRedis:
"""A Redis holding string values in a dictionary.
Attributes:
values (dict): The keys and their values.
"""
def __init__(self):
"""Builds an empty Redis.
Returns:
None: This method returns nothing.
"""
self.values = {}
def get(self, key):
"""Reads a value.
Args:
key (str): The key.
Returns:
str | None: The value.
"""
return self.values.get(key)
def pipeline(self, transaction=True):
"""A pipeline.
Args:
transaction (bool): Ignored.
Returns:
StandInPipeline: The pipeline.
"""
del transaction
return StandInPipeline(self)
class BudgetKeptInRedisExample:
"""Spends a budget of three across two limiters that share a stand-in Redis.
Attributes:
cache (StandInRedis): The shared stand-in Redis.
"""
def __init__(self):
"""Builds the stand-in Redis.
Returns:
None: This method returns nothing.
"""
self.cache = StandInRedis()
def run(self):
"""Takes two requests, restarts, and takes two more.
Returns:
None: This method returns nothing.
"""
first = RateLimiter(1000.0, 3, self.cache, 'fyers:price_history:requests')
first.take()
first.take()
print(f'First process spent {first.spent_today()} requests')
second = RateLimiter(1000.0, 3, self.cache, 'fyers:price_history:requests')
print(f'Restarted process starts at {second.spent_today()}')
second.take()
try:
second.take()
except CandleThrottled as error:
print(f'Its second request: {error}')
second.set_requests_per_day(None)
second.take()
print(f'With the cap lifted, one more is allowed; spent today: {second.spent_today()}')
key = f'fyers:price_history:requests:{second.budget_date().isoformat()}'
print(f'Redis holds {self.cache.get(key)} under the key for today')
if __name__ == '__main__':
BudgetKeptInRedisExample().run()
It prints:
Methods:
| Name | Description |
|---|---|
take |
Wait until the next request may be made, and account for it. |
set_requests_per_day |
Change the daily cap from the next request on, for a broker whose cap depends on the day. |
budget_date |
The day the daily budget is counted against: today in India. |
spent_today |
How many requests have been counted against today's budget, in Redis when there is one and in this process otherwise. |
back_off |
Push the next request further out, after the broker said no. |
Source code in stock_brokers/instruments/historical/base.py
Wait until the next request may be made, and account for it.
Raises CandleThrottled when the daily budget is exhausted, which the caller treats as a
reason to stop for the day rather than to retry.
Example usage
Source code in stock_brokers/instruments/historical/base.py
Change the daily cap from the next request on, for a broker whose cap depends on the day.
requests_per_dayis the new cap, or None for no cap; requests already counted today still count.
Example usage
Source code in stock_brokers/instruments/historical/base.py
How many requests have been counted against today's budget, in Redis when there is one and in this process otherwise.
Example usage
From example_3_budget_kept_in_redis.py
print(f'With the cap lifted, one more is allowed; spent today: {second.spent_today()}')
Source code in stock_brokers/instruments/historical/base.py
Count one request against today's budget, in Redis when there is one, keeping the key for two days.
Source code in stock_brokers/instruments/historical/base.py
Push the next request further out, after the broker said no.
secondsis how long to wait beyond the normal gap.
Example usage
Source code in stock_brokers/instruments/historical/base.py
Base class for one broker's historical candle download.
A subclass sets the class attributes below and implements fetch_candles and
parse_response.
BROKER_NAMEis the broker, which is also its PostgreSQL schema.INTERVALSmaps the name stored in the table to the name the broker's API expects.MAXIMUM_WINDOW_DAYSis how many days one request may span, per interval.REQUESTS_PER_SECONDandREQUESTS_PER_DAYare the broker's published limits.EARLIEST_AVAILABLE_DATEis as far back as the broker serves anything.TOKEN_COLUMN,TYPE_COLUMNandEXPIRY_COLUMNname the columns in this broker's instrument master that hold the identifier the API wants, the instrument type, and the expiry - the last two only to decide priority.
Candle downloader for one broker.
Example programs
examples/stock_brokers/instruments/historical/base/BrokerCandles/example_1_one_window_of_daily_bars.py
"""Seeds a small broker's work queue, claims one series and downloads one window of daily bars into it.
`BrokerCandles` is the shared machinery behind every broker's candle download. A broker module only says which intervals it serves, how far one request may reach, how to fetch one window and how to read the answer; the base class decides the order of the work, which window to ask for next, and how to store the bars and move the series' watermark.
This program writes that small broker module itself: `ExampleBrokerCandles` fetches from a stand-in history API that answers with a canned Kite-shaped payload, so no broker is contacted. The base class's constructor opens a PostgreSQL connection through `get_postgres`, and the example runner points PostgreSQL at a closed port, so before building the downloader the program replaces `get_postgres` in the base module with a stand-in that hands out an in-memory connection. That stand-in answers the handful of statements the base class sends, the instrument list, the progress table's row count, the next series to claim and the summary counts, and it records every insert and update so the program can show what would have been written.
What to notice in the output: cash sorts ahead of futures and futures ahead of options, day bars ahead of five-minute bars; seeding a second time adds nothing, because existing series are left as they are; the claimed series has already walked back to 2026-03-01, so the next window ends the day before and spans the 365 days a daily request may cover; and the day bars stamped at midnight India time come back as timezone-aware times. The broker receives two requests, because the program first calls `fetch_candles` and `parse_response` by hand to show them, and `visit` then fetches the same window again to store it. The expiry dates in the priority table are chosen far in the past and far in the future, so whether a contract counts as expired does not depend on the day the program runs.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/BrokerCandles/example_1_one_window_of_daily_bars.py
"""
import datetime
import zoneinfo
from stock_brokers.instruments.historical import base
INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
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.
"""
def __init__(self, connection):
"""Builds a cursor over a stand-in connection.
Args:
connection (StandInConnection): The connection that answers the statements.
Returns:
None: This method returns nothing.
"""
self.connection = connection
self.result = []
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 | None): The statement's parameters.
Returns:
None: This method returns nothing.
"""
if isinstance(statement, bytes):
statement = statement.decode()
self.result = 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 holding a tiny progress table in memory.
Attributes:
encoding (str): The client encoding, which `execute_values` reads.
OUTCOMES (tuple): The outcome names a progress update may write literally.
instruments (list): The rows the instrument master query returns.
pending_series (list): The series `claim` hands out, in order.
progress (list): The (token, interval) pairs registered in the progress table.
staged_rows (list): Rows staged by `mogrify` for the next insert.
stored_bars (list): Every bar row written to the price table.
outcomes (list): One (token, interval, outcome) triple per progress update.
closed (bool): Whether `close` has been called.
"""
encoding = 'UTF8'
OUTCOMES = (
'stored',
'empty',
'backfilled',
'throttled',
'failed',
'retired',
)
def __init__(self, instruments, pending_series):
"""Builds the connection with the rows it will answer with.
Args:
instruments (list): The rows the instrument master query returns.
pending_series (list): The series `claim` hands out, in order.
Returns:
None: This method returns nothing.
"""
self.instruments = instruments
self.pending_series = pending_series
self.progress = []
self.staged_rows = []
self.stored_bars = []
self.outcomes = []
self.closed = False
def cursor(self):
"""Opens a cursor.
Returns:
StandInCursor: A new cursor.
"""
return StandInCursor(self)
def commit(self):
"""Accepts a commit, which the stand-in has nothing to do for.
Returns:
None: This method returns nothing.
"""
return None
def rollback(self):
"""Accepts a rollback, which the stand-in has nothing to do for.
Returns:
None: This method returns nothing.
"""
return None
def close(self):
"""Marks the connection closed.
Returns:
None: This method returns nothing.
"""
self.closed = True
def answer(self, statement, parameters):
"""Produces the rows one statement would return, and records what it would write.
Args:
statement (str): The SQL text.
parameters (tuple | None): The statement's parameters.
Returns:
list: The rows the statement returns.
"""
if 'select distinct on' in statement:
return list(self.instruments)
if 'select count(*),' in statement:
return [
self.summary_row(),
]
if 'select count(*) from' in statement:
return [
(
len(self.progress),
),
]
if 'select instrument_token' in statement:
if not self.pending_series:
return []
return [
self.pending_series.pop(0),
]
if 'insert into' in statement:
rows = self.staged_rows
self.staged_rows = []
if 'price_history_progress' in statement:
for row in rows:
key = (
row[0],
row[1],
)
if key not in self.progress:
self.progress.append(key)
else:
self.stored_bars.extend(rows)
return []
if 'update' in statement:
self.outcomes.append(self.outcome_of(statement, parameters))
return []
def summary_row(self):
"""Counts the progress table the way the summary query does.
Returns:
tuple: The series, retired series, series with bars, bars and stuck series.
"""
retired = 0
series_with_bars = []
for token, interval, outcome in self.outcomes:
if outcome == 'retired':
retired += 1
if outcome == 'stored' and (token, interval) not in series_with_bars:
series_with_bars.append((token, interval))
return (
len(self.progress),
retired,
len(series_with_bars),
len(self.stored_bars),
0,
)
def outcome_of(self, statement, parameters):
"""Reads which outcome a progress update records.
Args:
statement (str): The update's SQL text.
parameters (tuple): The update's parameters, ending with the token and interval.
Returns:
tuple: The (token, interval, outcome) the update records.
"""
token = parameters[-2]
interval = parameters[-1]
outcome = 'unknown'
if 'last_outcome = %s' in statement:
outcome = parameters[0]
for name in self.OUTCOMES:
if f"last_outcome = '{name}'" in statement:
outcome = name
return (
token,
interval,
outcome,
)
class StandInPostgres:
"""Hands the downloader a stand-in connection in place of a real PostgreSQL one.
Attributes:
connection (StandInConnection): The connection handed out.
"""
def __init__(self, connection):
"""Holds the connection to hand out.
Args:
connection (StandInConnection): The connection handed out.
Returns:
None: This method returns nothing.
"""
self.connection = connection
def get_postgres(self):
"""Returns the stand-in connection, in the place of `utilities.configurations.get_postgres`.
Returns:
StandInConnection: The connection.
"""
return self.connection
class StandInHistoryAPI:
"""A stand-in for a broker's history endpoint that answers with a canned Kite-shaped payload.
Attributes:
requests (list): Every request received, as (token, interval, start, end) tuples.
"""
def __init__(self):
"""Builds the stand-in with no requests received.
Returns:
None: This method returns nothing.
"""
self.requests = []
def historical_data(self, token, interval, start_date, end_date):
"""Answers a history request with three daily candles.
Args:
token (str): The broker's instrument token.
interval (str): The broker's interval code.
start_date (datetime.date): The first day of the window.
end_date (datetime.date): The last day of the window.
Returns:
dict: The decoded response, with a `candles` list.
"""
self.requests.append(
(
token,
interval,
start_date,
end_date,
),
)
return {
'candles': [
[
'2026-02-25T00:00:00+0530',
1391.0,
1402.5,
1385.2,
1398.4,
6123450,
],
[
'2026-02-26T00:00:00+0530',
1398.4,
1410.0,
1392.1,
1405.9,
5870012,
],
[
'2026-02-27T00:00:00+0530',
1405.9,
1406.5,
1380.0,
1384.3,
7211904,
],
],
}
class ExampleBrokerCandles(base.BrokerCandles):
"""A made-up broker's candle download, reading Kite-shaped candles from a history API.
Attributes:
history_api (StandInHistoryAPI): Where windows of candles are fetched from.
"""
BROKER_NAME = 'example_broker'
INTERVALS = {
'day': 'day',
'5minute': '5minute',
}
MAXIMUM_WINDOW_DAYS = {
'day': 365,
'5minute': 60,
}
REQUESTS_PER_SECOND = 50.0
EARLIEST_AVAILABLE_DATE = datetime.date(2015, 1, 1)
def __init__(self, history_api):
"""Builds the downloader around a history API.
Args:
history_api (StandInHistoryAPI): Where windows of candles are fetched from.
Returns:
None: This method returns nothing.
"""
super().__init__()
self.history_api = history_api
def fetch_candles(self, token, interval, start_date, end_date):
"""Fetches one window of candles.
Args:
token (str): The broker's instrument token.
interval (str): The stored interval name.
start_date (datetime.date): The first day of the window.
end_date (datetime.date): The last day of the window.
Returns:
dict: The decoded response.
"""
return self.history_api.historical_data(token, self.INTERVALS[interval], start_date, end_date)
def parse_response(self, payload, interval):
"""Turns a Kite-shaped response into bar tuples.
Args:
payload (dict): The decoded response.
interval (str): The stored interval name.
Returns:
list: One (time, open, high, low, close, volume, oi) tuple per candle.
"""
bars = []
for candle in payload.get('candles', []):
moment = datetime.datetime.strptime(candle[0], '%Y-%m-%dT%H:%M:%S%z')
if not base.is_intraday(interval):
moment = base.daily_bar_time(moment)
open_interest = None
if len(candle) > 6:
open_interest = candle[6]
bars.append(
(
moment,
candle[1],
candle[2],
candle[3],
candle[4],
candle[5],
open_interest,
),
)
return bars
class OneWindowOfDailyBarsExample:
"""Seeds the queue, claims a series and stores one window of its daily bars.
Attributes:
connection (StandInConnection): The in-memory database.
history_api (StandInHistoryAPI): The stand-in broker endpoint.
downloader (ExampleBrokerCandles): The downloader being shown.
"""
def __init__(self):
"""Builds the downloader over the stand-in database and history API.
Returns:
None: This method returns nothing.
"""
instruments = [
(
'738561',
'EQ',
None,
),
(
'13368834',
'FUT',
'2099-12-31',
),
(
'13368835',
'CE',
'2020-01-30',
),
]
pending_series = [
(
'738561',
'day',
datetime.datetime(2026, 3, 2, tzinfo=INDIA),
datetime.datetime(2026, 9, 29, tzinfo=INDIA),
datetime.datetime(2026, 3, 1, tzinfo=INDIA),
0,
0,
),
]
self.connection = StandInConnection(instruments, pending_series)
base.get_postgres = StandInPostgres(self.connection).get_postgres
self.history_api = StandInHistoryAPI()
self.downloader = ExampleBrokerCandles(self.history_api)
def run(self):
"""Walks through one window of work and prints each step.
Returns:
None: This method returns nothing.
"""
print(f'Series context for 738561: {ExampleBrokerCandles.series_context("738561")}')
print('Priority of each series (lower is worked first):')
for token, instrument_type, expiry in self.downloader.instruments():
for interval in ExampleBrokerCandles.INTERVALS:
priority = self.downloader.priority_for(instrument_type, expiry, interval)
print(f' {token} {instrument_type} expiry {expiry} {interval}: {priority}')
print(f'Series added by the first seed: {self.downloader.seed()}')
print(f'Series added by a second seed: {self.downloader.seed()}')
series = self.downloader.claim()
print(f'Claimed: {series[0]} {series[1]}, oldest requested {series[4].date()}')
start_date, end_date, direction = self.downloader.next_window(series)
print(f'Next window: {start_date} to {end_date}, walking {direction}')
payload = self.downloader.fetch_candles(series[0], series[1], start_date, end_date)
bars = self.downloader.parse_response(payload, series[1])
print(f'Parsed {len(bars)} bars; the first is {bars[0][0].isoformat()} close {bars[0][4]}')
stored = self.downloader.visit(series)
print(f'Bars stored by visit: {stored}')
for row in self.connection.stored_bars:
print(f' wrote {row[0].isoformat()} {row[1]} {row[2]} close {row[6]} volume {row[7]}')
print(f'Progress updates: {self.connection.outcomes}')
print(f'Requests sent to the broker: {len(self.history_api.requests)}')
print(f'Summary: {self.downloader.summary()}')
self.downloader.close()
print(f'Connection closed: {self.connection.closed}')
if __name__ == '__main__':
OneWindowOfDailyBarsExample().run()
It prints:
Series context for 738561: SeriesContext(broker_token='738561', exchange=None, segments=None, match_on='broker_token', exchange_token=None, symbol=None)
Priority of each series (lower is worked first):
738561 EQ expiry None day: 10
738561 EQ expiry None 5minute: 25
13368834 FUT expiry 2099-12-31 day: 30
13368834 FUT expiry 2099-12-31 5minute: 45
13368835 CE expiry 2020-01-30 day: 90
13368835 CE expiry 2020-01-30 5minute: 105
Series added by the first seed: 6
Series added by a second seed: 0
Claimed: 738561 day, oldest requested 2026-03-01
Next window: 2025-03-01 to 2026-02-28, walking backward
Parsed 3 bars; the first is 2026-02-25T00:00:00+05:30 close 1398.4
Bars stored by visit: 3
wrote 2026-02-25T00:00:00+05:30 738561 day close 1398.4 volume 6123450
wrote 2026-02-26T00:00:00+05:30 738561 day close 1405.9 volume 5870012
wrote 2026-02-27T00:00:00+05:30 738561 day close 1384.3 volume 7211904
Progress updates: [('738561', 'day', 'stored')]
Requests sent to the broker: 2
Summary: {'series': 6, 'retired': 0, 'with_bars': 1, 'bars': 3, 'stuck': 0}
Connection closed: True
examples/stock_brokers/instruments/historical/base/BrokerCandles/example_2_run_until_the_queue_is_drained.py
"""Works a queue of seven series until it is drained, showing how the base class reacts to each kind of answer a broker gives.
`run` claims one series at a time, asks `visit` to fetch and store its next window, and stops when nothing is due, when it is asked to stop, or when the broker refuses the session outright. What `visit` records against a series depends on what happened: bars are stored, an empty window counts towards finishing the backward walk, an unknown instrument is retired, a throttle is noted without being held against the series, and any other error pushes the series back with a growing delay.
The broker here is made up. `ExampleBrokerCandles` fetches from a stand-in history API that answers each token from a script, either with canned Kite-shaped candles or with the error message a real broker sends, and its `fetch_candles` turns those messages into the `CandleError` subclasses the base class understands. The base class's constructor would open a PostgreSQL connection, and the example runner points PostgreSQL at a closed port, so the program replaces `get_postgres` in the base module with a stand-in that hands out an in-memory connection. That connection gives out the queued series in order and records the outcome each progress update writes. The project's configuration module sets up logging to standard error when it is imported, so the program sets it up again, with `force=True`, to write to standard output, and the base class's own messages then appear in order with the program's.
What to notice in the output: the empty window finishes the backfill because it is the third in a row; the series whose backward walk is done and whose last bar is in 2099 is put aside as up to date, whichever day the program runs; the series with no bars anywhere is retired without a request being sent; and the throttled series is left claimable. The throttle is scripted last on purpose, because the base class then holds the next request back five seconds, and the queue is empty by then. A final call with a stop event that is already set shows `run` returning at once.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/BrokerCandles/example_2_run_until_the_queue_is_drained.py
"""
import datetime
import logging
import sys
import threading
import zoneinfo
from stock_brokers.instruments.historical import base
INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
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.
"""
def __init__(self, connection):
"""Builds a cursor over a stand-in connection.
Args:
connection (StandInConnection): The connection that answers the statements.
Returns:
None: This method returns nothing.
"""
self.connection = connection
self.result = []
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 | None): The statement's parameters.
Returns:
None: This method returns nothing.
"""
if isinstance(statement, bytes):
statement = statement.decode()
self.result = 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 holding a tiny progress table in memory.
Attributes:
encoding (str): The client encoding, which `execute_values` reads.
OUTCOMES (tuple): The outcome names a progress update may write literally.
instruments (list): The rows the instrument master query returns.
pending_series (list): The series `claim` hands out, in order.
progress (list): The (token, interval) pairs registered in the progress table.
staged_rows (list): Rows staged by `mogrify` for the next insert.
stored_bars (list): Every bar row written to the price table.
outcomes (list): One (token, interval, outcome) triple per progress update.
closed (bool): Whether `close` has been called.
"""
encoding = 'UTF8'
OUTCOMES = (
'stored',
'empty',
'backfilled',
'throttled',
'failed',
'retired',
)
def __init__(self, instruments, pending_series):
"""Builds the connection with the rows it will answer with.
Args:
instruments (list): The rows the instrument master query returns.
pending_series (list): The series `claim` hands out, in order.
Returns:
None: This method returns nothing.
"""
self.instruments = instruments
self.pending_series = pending_series
self.progress = []
self.staged_rows = []
self.stored_bars = []
self.outcomes = []
self.closed = False
def cursor(self):
"""Opens a cursor.
Returns:
StandInCursor: A new cursor.
"""
return StandInCursor(self)
def commit(self):
"""Accepts a commit, which the stand-in has nothing to do for.
Returns:
None: This method returns nothing.
"""
return None
def rollback(self):
"""Accepts a rollback, which the stand-in has nothing to do for.
Returns:
None: This method returns nothing.
"""
return None
def close(self):
"""Marks the connection closed.
Returns:
None: This method returns nothing.
"""
self.closed = True
def answer(self, statement, parameters):
"""Produces the rows one statement would return, and records what it would write.
Args:
statement (str): The SQL text.
parameters (tuple | None): The statement's parameters.
Returns:
list: The rows the statement returns.
"""
if 'select distinct on' in statement:
return list(self.instruments)
if 'select count(*),' in statement:
return [
self.summary_row(),
]
if 'select count(*) from' in statement:
return [
(
len(self.progress),
),
]
if 'select instrument_token' in statement:
if not self.pending_series:
return []
return [
self.pending_series.pop(0),
]
if 'insert into' in statement:
rows = self.staged_rows
self.staged_rows = []
if 'price_history_progress' in statement:
for row in rows:
key = (
row[0],
row[1],
)
if key not in self.progress:
self.progress.append(key)
else:
self.stored_bars.extend(rows)
return []
if 'update' in statement:
self.outcomes.append(self.outcome_of(statement, parameters))
return []
def summary_row(self):
"""Counts the progress table the way the summary query does.
Returns:
tuple: The series, retired series, series with bars, bars and stuck series.
"""
retired = 0
series_with_bars = []
for token, interval, outcome in self.outcomes:
if outcome == 'retired':
retired += 1
if outcome == 'stored' and (token, interval) not in series_with_bars:
series_with_bars.append((token, interval))
return (
len(self.progress),
retired,
len(series_with_bars),
len(self.stored_bars),
0,
)
def outcome_of(self, statement, parameters):
"""Reads which outcome a progress update records.
Args:
statement (str): The update's SQL text.
parameters (tuple): The update's parameters, ending with the token and interval.
Returns:
tuple: The (token, interval, outcome) the update records.
"""
token = parameters[-2]
interval = parameters[-1]
outcome = 'unknown'
if 'last_outcome = %s' in statement:
outcome = parameters[0]
for name in self.OUTCOMES:
if f"last_outcome = '{name}'" in statement:
outcome = name
return (
token,
interval,
outcome,
)
class StandInPostgres:
"""Hands the downloader a stand-in connection in place of a real PostgreSQL one.
Attributes:
connection (StandInConnection): The connection handed out.
"""
def __init__(self, connection):
"""Holds the connection to hand out.
Args:
connection (StandInConnection): The connection handed out.
Returns:
None: This method returns nothing.
"""
self.connection = connection
def get_postgres(self):
"""Returns the stand-in connection, in the place of `utilities.configurations.get_postgres`.
Returns:
StandInConnection: The connection.
"""
return self.connection
class StandInBrokerError(Exception):
"""An error a stand-in broker answers with, carrying the broker's own message."""
class StandInHistoryAPI:
"""A stand-in for a broker's history endpoint that answers each token from a script.
Attributes:
answers (dict): Token to the payload to return, or to the error message to raise.
requests (list): The tokens requested, in order.
"""
def __init__(self, answers):
"""Builds the stand-in with its scripted answers.
Args:
answers (dict): Token to the payload to return, or to the error message to raise.
Returns:
None: This method returns nothing.
"""
self.answers = answers
self.requests = []
def historical_data(self, token, interval, start_date, end_date):
"""Answers a history request from the script.
Args:
token (str): The broker's instrument token.
interval (str): The broker's interval code.
start_date (datetime.date): The first day of the window.
end_date (datetime.date): The last day of the window.
Returns:
dict: The decoded response, with a `candles` list.
Raises:
StandInBrokerError: When the script gives an error message for the token.
"""
del interval, start_date, end_date
self.requests.append(token)
answer = self.answers[token]
if isinstance(answer, str):
raise StandInBrokerError(answer)
return answer
class ExampleBrokerCandles(base.BrokerCandles):
"""A made-up broker's candle download, reading Kite-shaped candles and Kite-style error messages.
Attributes:
history_api (StandInHistoryAPI): Where windows of candles are fetched from.
"""
BROKER_NAME = 'example_broker'
INTERVALS = {
'day': 'day',
'5minute': '5minute',
}
MAXIMUM_WINDOW_DAYS = {
'day': 365,
'5minute': 60,
}
REQUESTS_PER_SECOND = 50.0
EARLIEST_AVAILABLE_DATE = datetime.date(2015, 1, 1)
def __init__(self, history_api):
"""Builds the downloader around a history API.
Args:
history_api (StandInHistoryAPI): Where windows of candles are fetched from.
Returns:
None: This method returns nothing.
"""
super().__init__()
self.history_api = history_api
def fetch_candles(self, token, interval, start_date, end_date):
"""Fetches one window of candles, turning the broker's refusals into candle errors.
Args:
token (str): The broker's instrument token.
interval (str): The stored interval name.
start_date (datetime.date): The first day of the window.
end_date (datetime.date): The last day of the window.
Returns:
dict: The decoded response.
Raises:
CandleInstrumentUnknown: When the broker does not know the token.
CandleThrottled: When the broker asks for fewer requests.
CandleAuthenticationError: When the broker refuses the session.
StandInBrokerError: For any other refusal.
"""
try:
return self.history_api.historical_data(token, self.INTERVALS[interval], start_date, end_date)
except StandInBrokerError as error:
message = str(error)
if 'invalid token' in message:
raise base.CandleInstrumentUnknown(message) from error
if 'Too many requests' in message:
raise base.CandleThrottled(message) from error
if 'TokenException' in message:
raise base.CandleAuthenticationError(message) from error
raise
def parse_response(self, payload, interval):
"""Turns a Kite-shaped response into bar tuples.
Args:
payload (dict): The decoded response.
interval (str): The stored interval name.
Returns:
list: One (time, open, high, low, close, volume, oi) tuple per candle.
"""
bars = []
for candle in payload.get('candles', []):
moment = datetime.datetime.strptime(candle[0], '%Y-%m-%dT%H:%M:%S%z')
if not base.is_intraday(interval):
moment = base.daily_bar_time(moment)
open_interest = None
if len(candle) > 6:
open_interest = candle[6]
bars.append(
(
moment,
candle[1],
candle[2],
candle[3],
candle[4],
candle[5],
open_interest,
),
)
return bars
class RunUntilTheQueueIsDrainedExample:
"""Runs the download over a scripted queue and prints what happened to each series.
Attributes:
connection (StandInConnection): The in-memory database.
history_api (StandInHistoryAPI): The stand-in broker endpoint.
downloader (ExampleBrokerCandles): The downloader being shown.
"""
def __init__(self):
"""Builds the downloader over a queue of seven series and a scripted broker.
Returns:
None: This method returns nothing.
"""
pending_series = [
self.series('738561', 'day', datetime.date(2026, 3, 1), datetime.date(2026, 9, 29), 0),
self.series('500325', 'day', datetime.date(2018, 1, 1), datetime.date(2026, 9, 29), 2),
self.series('13368834', '5minute', None, None, 0),
self.series('256265', 'day', datetime.date(2015, 1, 1), datetime.date(2099, 1, 1), 0),
self.series('408065', 'day', datetime.date(2015, 1, 1), None, 0),
self.series('884737', 'day', datetime.date(2026, 3, 1), datetime.date(2026, 9, 29), 0),
self.series('341249', 'day', datetime.date(2026, 3, 1), datetime.date(2026, 9, 29), 0),
]
answers = {
'738561': {
'candles': [
[
'2026-02-26T00:00:00+0530',
1398.4,
1410.0,
1392.1,
1405.9,
5870012,
],
[
'2026-02-27T00:00:00+0530',
1405.9,
1406.5,
1380.0,
1384.3,
7211904,
],
],
},
'500325': {
'candles': [],
},
'13368834': 'InputException: invalid token',
'884737': 'GeneralException: Something went wrong, please try again',
'341249': 'NetworkException: Too many requests',
}
self.connection = StandInConnection([], pending_series)
for row in pending_series:
self.connection.progress.append(
(
row[0],
row[1],
),
)
base.get_postgres = StandInPostgres(self.connection).get_postgres
self.history_api = StandInHistoryAPI(answers)
self.downloader = ExampleBrokerCandles(self.history_api)
def series(self, token, interval, oldest_requested, latest, empty_streak):
"""Builds one progress row the way `claim` returns it.
Args:
token (str): The broker's instrument token.
interval (str): The stored interval name.
oldest_requested (datetime.date | None): The earliest day asked for so far, or None before the first window.
latest (datetime.date | None): The day of the newest bar stored, or None when none is.
empty_streak (int): How many empty windows in a row the series has had.
Returns:
tuple: The (token, interval, earliest, latest, oldest requested, empty streak, failures) row.
"""
oldest_requested_time = None
if oldest_requested is not None:
oldest_requested_time = datetime.datetime.combine(oldest_requested, datetime.time(), INDIA)
latest_time = None
if latest is not None:
latest_time = datetime.datetime.combine(latest, datetime.time(), INDIA)
return (
token,
interval,
None,
latest_time,
oldest_requested_time,
empty_streak,
0,
)
def run(self):
"""Drains the queue, prints each series' outcome, then shows a run stopped before it starts.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format='%(levelname)s %(name)s: %(message)s', force=True)
visits, bars = self.downloader.run()
print(f'Windows visited: {visits}, bars stored: {bars}')
print(f'Tokens requested from the broker: {self.history_api.requests}')
print('Outcome recorded for each series:')
for token, interval, outcome in self.connection.outcomes:
print(f' {token} {interval}: {outcome}')
print(f'Summary: {self.downloader.summary()}')
stop_event = threading.Event()
stop_event.set()
print(f'A run asked to stop before it starts: {self.downloader.run(stop_event=stop_event)}')
self.downloader.close()
if __name__ == '__main__':
RunUntilTheQueueIsDrainedExample().run()
It prints:
INFO candles.example_broker: 500325 day: backfill complete (3 empty windows walking back from 2017-12-31).
WARNING candles.example_broker: 884737 day: StandInBrokerError: GeneralException: Something went wrong, please try again
WARNING candles.example_broker: 341249 day: throttled: NetworkException: Too many requests
INFO candles.example_broker: Nothing due. The queue is drained or backing off.
INFO candles.example_broker: Finished: 7 windows, 2 bars.
Windows visited: 7, bars stored: 2
Tokens requested from the broker: ['738561', '500325', '13368834', '884737', '341249']
Outcome recorded for each series:
738561 day: stored
500325 day: backfilled
13368834 5minute: retired
256265 day: up to date
408065 day: retired
884737 day: failed
341249 day: throttled
Summary: {'series': 7, 'retired': 2, 'with_bars': 1, 'bars': 2, 'stuck': 0}
INFO candles.example_broker: Stopping: asked to.
INFO candles.example_broker: Finished: 0 windows, 0 bars.
A run asked to stop before it starts: (0, 0)
examples/stock_brokers/instruments/historical/base/BrokerCandles/example_3_logging_in_again_after_a_refused_session.py
"""Logs in again once when the broker refuses the session in the middle of a download, and stops the broker when that does not help.
A broker session expires every day, and a download that runs for weeks meets that expiry every day. When `fetch_candles` raises `CandleAuthenticationError`, the base class calls `relogin` once and asks for the same window again; `relogin` goes through `ensure_session`, rebuilds the broker's API object around the new session, and calls `after_relogin` so that a downloader can renew anything it keeps on top of the session. Only if the fresh session is refused as well does the error reach `run`, which then stops the broker rather than send thousands of refused requests.
The real `ensure_session` takes a Redis lock and may drive a headless browser through a broker's login, which an example must never do. So this program replaces `ensure_session` in `stock_brokers.api.utilities.session` with a stand-in session store that simply issues a new token, and its made-up broker, `ExampleBrokerCandles`, overrides the private `_build_api` to wrap a stand-in history API around whatever token the store holds. The base class's constructor would open a PostgreSQL connection, and the example runner points PostgreSQL at a closed port, so `get_postgres` in the base module is replaced with a stand-in that hands out an in-memory connection, as in the other examples.
What to notice in the output: the first window is refused with the old token, fetched again after one login, and stored; `after_relogin` renews the downloader's market data token each time a login happens; a login that fails comes back as `CandleAuthenticationError` with the underlying error named in its message; and when the broker refuses even a fresh session, `run` stops on the first series, counting no visit and leaving the second series unclaimed, instead of carrying on through the queue.
Run it from the project root:
python examples/stock_brokers/instruments/historical/base/BrokerCandles/example_3_logging_in_again_after_a_refused_session.py
"""
import datetime
import logging
import sys
import zoneinfo
from stock_brokers.api.utilities import session
from stock_brokers.instruments.historical import base
INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
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.
"""
def __init__(self, connection):
"""Builds a cursor over a stand-in connection.
Args:
connection (StandInConnection): The connection that answers the statements.
Returns:
None: This method returns nothing.
"""
self.connection = connection
self.result = []
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 | None): The statement's parameters.
Returns:
None: This method returns nothing.
"""
if isinstance(statement, bytes):
statement = statement.decode()
self.result = 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 holding a tiny progress table in memory.
Attributes:
encoding (str): The client encoding, which `execute_values` reads.
OUTCOMES (tuple): The outcome names a progress update may write literally.
instruments (list): The rows the instrument master query returns.
pending_series (list): The series `claim` hands out, in order.
progress (list): The (token, interval) pairs registered in the progress table.
staged_rows (list): Rows staged by `mogrify` for the next insert.
stored_bars (list): Every bar row written to the price table.
outcomes (list): One (token, interval, outcome) triple per progress update.
closed (bool): Whether `close` has been called.
"""
encoding = 'UTF8'
OUTCOMES = (
'stored',
'empty',
'backfilled',
'throttled',
'failed',
'retired',
)
def __init__(self, instruments, pending_series):
"""Builds the connection with the rows it will answer with.
Args:
instruments (list): The rows the instrument master query returns.
pending_series (list): The series `claim` hands out, in order.
Returns:
None: This method returns nothing.
"""
self.instruments = instruments
self.pending_series = pending_series
self.progress = []
self.staged_rows = []
self.stored_bars = []
self.outcomes = []
self.closed = False
def cursor(self):
"""Opens a cursor.
Returns:
StandInCursor: A new cursor.
"""
return StandInCursor(self)
def commit(self):
"""Accepts a commit, which the stand-in has nothing to do for.
Returns:
None: This method returns nothing.
"""
return None
def rollback(self):
"""Accepts a rollback, which the stand-in has nothing to do for.
Returns:
None: This method returns nothing.
"""
return None
def close(self):
"""Marks the connection closed.
Returns:
None: This method returns nothing.
"""
self.closed = True
def answer(self, statement, parameters):
"""Produces the rows one statement would return, and records what it would write.
Args:
statement (str): The SQL text.
parameters (tuple | None): The statement's parameters.
Returns:
list: The rows the statement returns.
"""
if 'select distinct on' in statement:
return list(self.instruments)
if 'select count(*),' in statement:
return [
self.summary_row(),
]
if 'select count(*) from' in statement:
return [
(
len(self.progress),
),
]
if 'select instrument_token' in statement:
if not self.pending_series:
return []
return [
self.pending_series.pop(0),
]
if 'insert into' in statement:
rows = self.staged_rows
self.staged_rows = []
if 'price_history_progress' in statement:
for row in rows:
key = (
row[0],
row[1],
)
if key not in self.progress:
self.progress.append(key)
else:
self.stored_bars.extend(rows)
return []
if 'update' in statement:
self.outcomes.append(self.outcome_of(statement, parameters))
return []
def summary_row(self):
"""Counts the progress table the way the summary query does.
Returns:
tuple: The series, retired series, series with bars, bars and stuck series.
"""
retired = 0
series_with_bars = []
for token, interval, outcome in self.outcomes:
if outcome == 'retired':
retired += 1
if outcome == 'stored' and (token, interval) not in series_with_bars:
series_with_bars.append((token, interval))
return (
len(self.progress),
retired,
len(series_with_bars),
len(self.stored_bars),
0,
)
def outcome_of(self, statement, parameters):
"""Reads which outcome a progress update records.
Args:
statement (str): The update's SQL text.
parameters (tuple): The update's parameters, ending with the token and interval.
Returns:
tuple: The (token, interval, outcome) the update records.
"""
token = parameters[-2]
interval = parameters[-1]
outcome = 'unknown'
if 'last_outcome = %s' in statement:
outcome = parameters[0]
for name in self.OUTCOMES:
if f"last_outcome = '{name}'" in statement:
outcome = name
return (
token,
interval,
outcome,
)
class StandInPostgres:
"""Hands the downloader a stand-in connection in place of a real PostgreSQL one.
Attributes:
connection (StandInConnection): The connection handed out.
"""
def __init__(self, connection):
"""Holds the connection to hand out.
Args:
connection (StandInConnection): The connection handed out.
Returns:
None: This method returns nothing.
"""
self.connection = connection
def get_postgres(self):
"""Returns the stand-in connection, in the place of `utilities.configurations.get_postgres`.
Returns:
StandInConnection: The connection.
"""
return self.connection
class StandInSessionStore:
"""A stand-in for the stored broker login, issuing a new token on every login.
Attributes:
valid_token (str): The token the broker currently accepts.
stored_token (str): The token the downloader will be built around.
logins (int): How many logins have been made.
refuse_logins (bool): Whether a login attempt fails.
"""
def __init__(self):
"""Builds the store holding a token that has already expired.
Returns:
None: This method returns nothing.
"""
self.valid_token = 'token-issued-this-morning'
self.stored_token = 'token-from-yesterday'
self.logins = 0
self.refuse_logins = False
def ensure_session(self, broker_name, force=False, logger=None):
"""Logs in, in the place of `stock_brokers.api.utilities.session.ensure_session`.
Args:
broker_name (str): The broker to log in to.
force (bool): Whether to skip the login rate limiter, which the stand-in ignores.
logger (logging.Logger | None): Where progress would be reported.
Returns:
dict: The stored login document after the call.
Raises:
RuntimeError: When the store is set to refuse logins.
"""
del force, logger
if self.refuse_logins:
raise RuntimeError(f'the {broker_name} login page did not accept the TOTP')
self.logins += 1
self.valid_token = f'token-from-login-{self.logins}'
self.stored_token = self.valid_token
return {
'broker_name': broker_name,
'access_token': self.stored_token,
}
class StandInHistoryAPI:
"""A stand-in for a broker's history endpoint that accepts only the current token.
Attributes:
access_token (str): The token this API object was built with.
session_store (StandInSessionStore): Which token the broker accepts.
broker_refuses_everything (bool): Whether even a current token is refused.
"""
def __init__(self, access_token, session_store, broker_refuses_everything):
"""Builds an API object around one token.
Args:
access_token (str): The token this API object sends.
session_store (StandInSessionStore): Which token the broker accepts.
broker_refuses_everything (bool): Whether even a current token is refused.
Returns:
None: This method returns nothing.
"""
self.access_token = access_token
self.session_store = session_store
self.broker_refuses_everything = broker_refuses_everything
def historical_data(self, token, interval, start_date, end_date):
"""Answers with two daily candles, or refuses a token that is not current.
Args:
token (str): The broker's instrument token.
interval (str): The broker's interval code.
start_date (datetime.date): The first day of the window.
end_date (datetime.date): The last day of the window.
Returns:
dict: The decoded response, with a `candles` list.
Raises:
PermissionError: When the token is not the one the broker accepts.
"""
del token, interval, start_date, end_date
print(f' broker asked with {self.access_token}')
if self.broker_refuses_everything or self.access_token != self.session_store.valid_token:
raise PermissionError('TokenException: Incorrect `api_key` or `access_token`.')
return {
'candles': [
[
'2026-02-26T00:00:00+0530',
1398.4,
1410.0,
1392.1,
1405.9,
5870012,
],
[
'2026-02-27T00:00:00+0530',
1405.9,
1406.5,
1380.0,
1384.3,
7211904,
],
],
}
class ExampleBrokerCandles(base.BrokerCandles):
"""A made-up broker's candle download whose history endpoint needs a market data token minted from the session.
Attributes:
session_store (StandInSessionStore): The stand-in stored login.
broker_refuses_everything (bool): Whether the stand-in broker refuses even a fresh session.
market_data_token (str): The second credential, renewed after every login.
"""
BROKER_NAME = 'example_broker'
INTERVALS = {
'day': 'day',
}
MAXIMUM_WINDOW_DAYS = {
'day': 365,
}
REQUESTS_PER_SECOND = 50.0
EARLIEST_AVAILABLE_DATE = datetime.date(2015, 1, 1)
def __init__(self, session_store, broker_refuses_everything):
"""Builds the downloader around the stored login.
Args:
session_store (StandInSessionStore): The stand-in stored login.
broker_refuses_everything (bool): Whether the stand-in broker refuses even a fresh session.
Returns:
None: This method returns nothing.
"""
super().__init__()
self.session_store = session_store
self.broker_refuses_everything = broker_refuses_everything
self._api = self._build_api()
self.market_data_token = f'market-data-for-{session_store.stored_token}'
def _build_api(self):
"""Builds the stand-in history API around whatever token is stored now.
Returns:
StandInHistoryAPI: The API object.
"""
return StandInHistoryAPI(self.session_store.stored_token, self.session_store, self.broker_refuses_everything)
def after_relogin(self):
"""Renews the market data token from the new session.
Returns:
None: This method returns nothing.
"""
self.market_data_token = f'market-data-for-{self.session_store.stored_token}'
print(f' renewed the market data token: {self.market_data_token}')
def fetch_candles(self, token, interval, start_date, end_date):
"""Fetches one window of candles, turning a refused token into a candle error.
Args:
token (str): The broker's instrument token.
interval (str): The stored interval name.
start_date (datetime.date): The first day of the window.
end_date (datetime.date): The last day of the window.
Returns:
dict: The decoded response.
Raises:
CandleAuthenticationError: When the broker refuses the session.
"""
try:
return self._api.historical_data(token, self.INTERVALS[interval], start_date, end_date)
except PermissionError as error:
raise base.CandleAuthenticationError(str(error)) from error
def parse_response(self, payload, interval):
"""Turns a Kite-shaped response into bar tuples.
Args:
payload (dict): The decoded response.
interval (str): The stored interval name.
Returns:
list: One (time, open, high, low, close, volume, oi) tuple per candle.
"""
bars = []
for candle in payload.get('candles', []):
moment = datetime.datetime.strptime(candle[0], '%Y-%m-%dT%H:%M:%S%z')
if not base.is_intraday(interval):
moment = base.daily_bar_time(moment)
bars.append(
(
moment,
candle[1],
candle[2],
candle[3],
candle[4],
candle[5],
None,
),
)
return bars
class LoggingInAgainAfterARefusedSessionExample:
"""Shows a refused session recovered by one login, a failed login, and a broker that refuses every session.
Attributes:
session_store (StandInSessionStore): The stand-in stored login.
"""
def __init__(self):
"""Puts the stand-in login in place of the real one.
Returns:
None: This method returns nothing.
"""
self.session_store = StandInSessionStore()
session.ensure_session = self.session_store.ensure_session
def series(self, token):
"""Builds one daily series' progress row, part way through its backward walk.
Args:
token (str): The broker's instrument token.
Returns:
tuple: The (token, interval, earliest, latest, oldest requested, empty streak, failures) row.
"""
return (
token,
'day',
datetime.datetime(2026, 3, 2, tzinfo=INDIA),
datetime.datetime(2026, 9, 29, tzinfo=INDIA),
datetime.datetime(2026, 3, 1, tzinfo=INDIA),
0,
0,
)
def downloader_over(self, connection, broker_refuses_everything):
"""Builds a downloader whose database connection is the given stand-in.
Args:
connection (StandInConnection): The in-memory database.
broker_refuses_everything (bool): Whether the stand-in broker refuses even a fresh session.
Returns:
ExampleBrokerCandles: The downloader.
"""
base.get_postgres = StandInPostgres(connection).get_postgres
return ExampleBrokerCandles(self.session_store, broker_refuses_everything)
def run(self):
"""Runs the three situations in turn and prints what each did.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format='%(levelname)s %(name)s: %(message)s', force=True)
connection = StandInConnection([], [])
downloader = self.downloader_over(connection, False)
print(f'Built with {downloader.market_data_token}')
print('Visiting one window with yesterday\'s token:')
stored = downloader.visit(self.series('738561'))
print(f'Bars stored: {stored}, logins made: {self.session_store.logins}')
print(f'Progress updates: {connection.outcomes}')
print('Logging in again by hand:')
downloader.relogin()
print('Renewing the market data token by hand, without a login:')
downloader.after_relogin()
self.session_store.refuse_logins = True
print('Logging in again when the login itself fails:')
try:
downloader.relogin()
except base.CandleAuthenticationError as error:
print(f' CandleAuthenticationError: {error}')
downloader.close()
self.session_store.refuse_logins = False
print('Running a queue of two series against a broker that refuses every session:')
connection = StandInConnection(
[],
[
self.series('738561'),
self.series('2953217'),
],
)
downloader = self.downloader_over(connection, True)
visits, bars = downloader.run()
print(f'Windows visited: {visits}, bars stored: {bars}, series left in the queue: {len(connection.pending_series)}')
print(f'Progress updates: {connection.outcomes}')
downloader.close()
if __name__ == '__main__':
LoggingInAgainAfterARefusedSessionExample().run()
It prints:
Built with market-data-for-token-from-yesterday
Visiting one window with yesterday's token:
broker asked with token-from-yesterday
WARNING candles.example_broker: example_broker refused the session (TokenException: Incorrect `api_key` or `access_token`.). Logging in once before continuing.
renewed the market data token: market-data-for-token-from-login-1
INFO candles.example_broker: example_broker logged in again. Continuing.
broker asked with token-from-login-1
Bars stored: 2, logins made: 1
Progress updates: [('738561', 'day', 'stored')]
Logging in again by hand:
renewed the market data token: market-data-for-token-from-login-2
INFO candles.example_broker: example_broker logged in again. Continuing.
Renewing the market data token by hand, without a login:
renewed the market data token: market-data-for-token-from-login-2
Logging in again when the login itself fails:
CandleAuthenticationError: logging in again did not produce a working session: RuntimeError: the example_broker login page did not accept the TOTP
Running a queue of two series against a broker that refuses every session:
broker asked with token-from-login-2
WARNING candles.example_broker: example_broker refused the session (TokenException: Incorrect `api_key` or `access_token`.). Logging in once before continuing.
renewed the market data token: market-data-for-token-from-login-3
INFO candles.example_broker: example_broker logged in again. Continuing.
broker asked with token-from-login-3
ERROR candles.example_broker: Stopping: CandleAuthenticationError: TokenException: Incorrect `api_key` or `access_token`.
INFO candles.example_broker: Finished: 0 windows, 0 bars.
Windows visited: 0, bars stored: 0, series left in the queue: 1
Progress updates: []
Methods:
| Name | Description |
|---|---|
series_context |
Say what instrument one of this broker's stored series can belong to, without logging in. |
fetch_candles |
Fetch one window of candles from the broker. |
parse_response |
Turn a broker response into a list of |
instruments |
Every instrument this broker has ever published, with the fields priority needs. |
priority_for |
Where a series sits in the queue. Lower is worked first. |
seed |
Make sure every instrument and interval this broker serves has a progress row. |
claim |
The next series due for a window, or None when there is nothing to do. |
next_window |
The date range to ask for next, as a |
relogin |
Establish a fresh session and rebuild everything built on the old one. |
after_relogin |
Re-establish anything a downloader keeps on top of the broker session. Nothing by default. |
visit |
Fetch one window for one series and store it, advancing the watermark. |
run |
Work the queue until stopped, out of work, or out of time. |
summary |
How far the download has got, for a status display. |
close |
Close the database connection. |
Source code in stock_brokers/instruments/historical/base.py
classmethod
¶
Say what instrument one of this broker's stored series can belong to, without logging in.
A classmethod on purpose: constructing a downloader establishes a broker session, and the unified price layer reads series for every broker without wanting a session for any. The default treats the identifier as the broker's own token with nothing known about exchange or segment; a broker whose identifier carries more says so by overriding this.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
identifier
|
str
|
The |
required |
Returns:
| Type | Description |
|---|---|
SeriesContext
|
The token to look up, and the exchange and segments to look within. |
Example usage
From example_1_one_window_of_daily_bars.py
print(f'Series context for 738561: {ExampleBrokerCandles.series_context("738561")}')
Source code in stock_brokers/instruments/historical/base.py
Fetch one window of candles from the broker.
Returns whatever the broker's response decodes to; parse_response turns it into bars.
Raises one of the CandleError subclasses to say how the caller should react.
The interval arrives as this project's stored name rather than as the broker's code, and
the module translates it with self.INTERVALS[interval] where it needs the code. Several
brokers need both: the name says whether to call the daily endpoint or the intraday one,
and the code is what goes in the request.
tokenis the broker's own instrument identifier, asinstrumentsproduced it.intervalis the stored interval name, a key ofINTERVALS.start_dateandend_datebound the window, inclusive.
Example usage
From example_1_one_window_of_daily_bars.py
payload = self.downloader.fetch_candles(series[0], series[1], start_date, end_date)
Source code in stock_brokers/instruments/historical/base.py
Turn a broker response into a list of (time, open, high, low, close, volume, oi) tuples.
time must be a timezone aware datetime, not the broker's text. The watermarks come
back from Postgres as datetimes and are compared against these, so a parser that passes
strings through fails the moment a series starts collecting forward. A daily or coarser
bar goes through daily_bar_time, because brokers disagree about what its timestamp
means.
payloadis whateverfetch_candlesreturned.intervalis the stored interval name the window was asked for. Brokers whose bar times depend on the interval - a platform that stamps the end of a bar rather than its start, or one that needs a daily bar normalised - need it; the rest ignore it.
Example usage
Source code in stock_brokers/instruments/historical/base.py
Every instrument this broker has ever published, with the fields priority needs.
The most recent snapshot each token appeared in, not merely today's: a contract that expired last month is still worth asking about, and the answer decides whether it is retired. Today's snapshot alone would silently narrow the universe to whatever is live.
Example usage
Source code in stock_brokers/instruments/historical/base.py
Where a series sits in the queue. Lower is worked first.
The ordering is the difference between useful data in days and useful data in months. Cash instruments come first because they have the deepest history and are wanted most; expired options come last because there are tens of thousands of them, each holding a few weeks of data, and most will be refused outright. Coarse intervals beat fine ones within each tier, since a day bar covers a year in one request and a minute bar covers two months.
instrument_typeis the broker's own type string.expiryis the contract expiry, or None for cash.intervalis the stored interval name.
Example usage
From example_1_one_window_of_daily_bars.py
priority = self.downloader.priority_for(instrument_type, expiry, interval)
Source code in stock_brokers/instruments/historical/base.py
staticmethod
¶
Minutes in an interval name like 15minute, or 0 when it is not a minute interval.
intervalis the stored interval name.
Source code in stock_brokers/instruments/historical/base.py
Make sure every instrument and interval this broker serves has a progress row.
Safe to run again: existing rows are left exactly as they are, so re-seeding after a new instrument master picks up what is new without disturbing what is already downloaded. Returns how many rows were added.
Example usage
From example_1_one_window_of_daily_bars.py
print(f'Series added by the first seed: {self.downloader.seed()}')
From example_1_one_window_of_daily_bars.py
print(f'Series added by a second seed: {self.downloader.seed()}')
Source code in stock_brokers/instruments/historical/base.py
Register one batch of series, leaving any that already exist exactly as they are.
cursoris an open cursor on this broker's connection.batchis a list of(token, interval, priority, seeded_date)tuples.
Source code in stock_brokers/instruments/historical/base.py
The next series due for a window, or None when there is nothing to do.
Ordered by priority and then by how long a series has been waiting. Series that have reached the broker's limit, or that are backing off after failures, are skipped.
Example usage
Source code in stock_brokers/instruments/historical/base.py
The date range to ask for next, as a (start, end, direction) triple.
The backward walk comes first and runs to completion. A series starts at today and steps back a window at a time until it reaches the broker's earliest date or the broker stops returning anything, and only then does it start filling forward with whatever has happened since.
Doing it the other way round - preferring forward whenever the last stored bar is older than today - looks reasonable and does not work: outside market hours the last bar is always older than today, so the series asks for the same empty recent window forever and never walks back at all.
seriesis a row fromclaim.
Example usage
From example_1_one_window_of_daily_bars.py
start_date, end_date, direction = self.downloader.next_window(series)
Source code in stock_brokers/instruments/historical/base.py
Fetch and parse one window, logging in once first if the broker refuses the session.
A token expiring does not mean the download is over, and it does not mean this series is at
fault. The rule is to log in first and then carry on: one login, then the same window
again. Only if that login fails, or the fresh session is refused as well, does the error go
back up - where visit and run stop the broker rather than carry on with a dead token.
Both alternatives are worse. Stopping outright leaves the download waiting for systemd to restart it into a login, and recording the refusal against whichever series it was on and moving to the next sends thousands of refused requests an hour - enough for Cloudflare to ban the machine from Fyers.
The login is re-armed by any window that comes back without a session error, so a backfill that runs for weeks gets its one login at each daily expiry.
token,interval,start_dateandend_dateare as forfetch_candles.
Source code in stock_brokers/instruments/historical/base.py
Establish a fresh session and rebuild everything built on the old one.
Through ensure_session, so it takes the same Redis lock and obeys the same floor on login
attempts as any other process using it - which means it may wait, and may find another
process has already logged in, in which case that session is simply used.
Raises:
| Type | Description |
|---|---|
CandleAuthenticationError
|
If no working session could be established. |
Example usage
Source code in stock_brokers/instruments/historical/base.py
Construct this broker's authenticated API class, around whatever session is now stored.
The broker's own class from api_class_for, which is what every downloader builds. A
downloader that builds its API differently overrides this.
Source code in stock_brokers/instruments/historical/base.py
Re-establish anything a downloader keeps on top of the broker session. Nothing by default.
For a broker whose historical endpoint takes a second credential minted from the session, such as a separate market data token, this is where that credential is renewed.
Example usage
Source code in stock_brokers/instruments/historical/base.py
Fetch one window for one series and store it, advancing the watermark.
Returns the number of bars stored. The bars and the watermark are written in the same transaction, so an interruption costs at most the window in flight - and repeating that window is harmless, because bars upsert on their key.
seriesis a row fromclaim.
Example usage
From example_3_logging_in_again_after_a_refused_session.py
stored = downloader.visit(self.series('738561'))
Source code in stock_brokers/instruments/historical/base.py
Write a window's bars and move the watermark, in one transaction.
tokenis the broker's instrument identifier.intervalis the stored interval name.barsare the parsed(time, open, high, low, close, volume, oi)tuples.requested_fromis the start of the window that was asked for.directionis whether the walk was forward or backward.previous_latestis the newest bar already stored, used to tell a forward window that gained ground from one that returned only what was already known.
Source code in stock_brokers/instruments/historical/base.py
Returns a bar's count, or None when the count cannot fit in a BIGINT column.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The broker's instrument identifier. |
required |
interval
|
str
|
The stored interval name. |
required |
bar_time
|
datetime
|
The time of the bar the value belongs to. |
required |
column
|
str
|
The name of the column the value is written to, either "volume" or "oi". |
required |
value
|
int | float | None
|
The count exactly as the broker sent it. |
required |
Returns:
| Type | Description |
|---|---|
int | float | None
|
The value unchanged when it fits in a BIGINT or is not a number, and None when it does not fit. |
Source code in stock_brokers/instruments/historical/base.py
Note that a window came back with nothing, and step past it.
token,intervalidentify the series.streakis how many empty windows in a row this makes.requested_fromis the start of the window that was asked for.directionis whether the walk was forward or backward.
Source code in stock_brokers/instruments/historical/base.py
Mark the backward walk complete, leaving the series collecting forward.
Distinct from _retire: the broker has no more past for this series, but it will keep
producing new bars. Retiring it here would quietly stop a live instrument updating.
token,intervalidentify the series.reasonis why the walk stopped, kept for review.
Source code in stock_brokers/instruments/historical/base.py
Put a series aside until the next day.
Reached when a forward window brings nothing new, which outside market hours is every forward window. Without it the highest priority series would be re-fetched continuously and spend the whole rate limit re-reading bars already stored.
token,intervalidentify the series.reasonis recorded as the outcome.
Source code in stock_brokers/instruments/historical/base.py
Note that the broker asked us to slow down, without holding it against the series.
The window is left exactly as it was, so the same series is claimed again once the limiter's back off has passed and the work is simply redone.
token,intervalidentify the series.reasonis what the broker said.
Source code in stock_brokers/instruments/historical/base.py
Note a failure and push the next attempt out exponentially.
An hour, then two, four, eight and so on, so a series that is permanently broken stops crowding out the ones that are not.
token,intervalidentify the series.reasonis what went wrong, stored for a human to read.
Source code in stock_brokers/instruments/historical/base.py
Stop asking for a series the broker will not serve.
token,intervalidentify the series.reasonis why, stored so the decision can be reviewed.
Source code in stock_brokers/instruments/historical/base.py
Work the queue until stopped, out of work, or out of time.
Returns a (visits, bars) pair.
stop_eventis athreading.Eventthat ends the loop when set.deadline_secondsends the loop after this long, for a job with a nightly window.
Example usage
From example_2_run_until_the_queue_is_drained.py
print(f'A run asked to stop before it starts: {self.downloader.run(stop_event=stop_event)}')
Source code in stock_brokers/instruments/historical/base.py
How far the download has got, for a status display.
Example usage
Source code in stock_brokers/instruments/historical/base.py
Close the database connection.
Example usage
Whether a stored interval name divides the session rather than spanning whole days.
Several brokers split their candle API in two, one endpoint for intraday bars and another for
daily and coarser ones, so the module has to know which side of that split a name falls on.
Stored interval names are this project's own, and every intraday one ends in minute.
intervalis the stored interval name, such as5minuteorday.
Source code in stock_brokers/instruments/historical/base.py
Put a daily or coarser bar at midnight India time on the trading date it belongs to.
Brokers disagree, silently, about what a daily bar's timestamp means. Measured on the same RELIANCE session: Kite stamps midnight India time, Fyers stamps midnight UTC, and Dhan sends an epoch that decodes to 18:30 UTC - which is the same midnight India time Kite means. Stored as received, the same trading day would sit at three different instants depending on which broker answered, and comparing one broker's day bar against another's would be an argument about time zones rather than about prices.
Every one of those conventions names the same date once read in India time, since no broker stamps a bar before the day it belongs to. So the trading date is recovered by converting to India time, and midnight India time is the value they all converge on.
Intraday bars are left alone: their timestamps are already unambiguous, and moving them would be changing the data.
momentis a timezone aware timestamp as the broker reported it.