One broker tick in, at most one unified quote and one table row out.
The steps, cheapest refusal first so that ticks which will be dropped cost as little as possible:
- resolve the tick's token to a compiled plan - one dictionary look-up once the plan exists;
- drop it if it arrived outside its instrument's session window;
- ask the ownership engine whether this broker owns the instrument - a backup's tick stops here;
- normalize the values;
- set the previous close, from the broker where its
closemeans that and otherwise from what this instrument's owner reported earlier the same day, and recompute the change from it; - drop it if nothing but the clock has changed since the last tick written for the instrument.
The survivor becomes a quote - the JSON the unified cache stores and publishes - and a row for
unified.ticks. The quote splices in the instrument's identity, serialized once when its plan was
compiled, so building one costs a single json.dumps of the changing fields.
Nothing here touches Redis or the database: the live service drives this class.
Classes:
| Name | Description |
|---|---|
TickPipeline |
Turns broker ticks into unified quotes and rows. |
Functions:
| Name | Description |
|---|---|
change_percent_from |
The change since the previous close, as a percentage rounded to four places. |
quote_document |
The unified quote JSON for one normalized tick. |
Turns broker ticks into unified quotes and rows.
Attributes:
| Name | Type | Description |
|---|---|---|
counts |
Counter
|
How many ticks ended at each step - received, unresolved, out_of_session, not_owner, no_price, duplicate and written. |
Build the pipeline.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
resolver
|
TickResolver
|
Resolves tokens to plans. |
required |
ownership
|
OwnershipEngine
|
Decides which broker's ticks are written. |
required |
normalizers
|
dict
|
Broker name to TickNormalizer. |
required |
gate
|
SessionGate | None
|
Session windows, defaulting to one with the standard tables. |
None
|
clock
|
callable
|
Returns the current epoch, stamped on quotes as unified_at. |
time
|
Returns:
| Type | Description |
|---|---|
None
|
This function returns nothing. |
Example programs
examples/stock_brokers/instruments/ticks/utilities/pipeline/TickPipeline/example_1_ticks_into_quotes.py
"""Takes three Zerodha ticks for RELIANCE through the tick pipeline, and flags the last quote stale.
`TickPipeline.process` takes one broker tick through every step the unified tick service applies: resolve the token to a plan, drop it outside the session window, ask the ownership engine whether this broker's ticks are written, normalize it, set the previous close and the change from it, and drop it if nothing but the clock changed. A survivor becomes a quote (the JSON the unified cache stores and publishes) and a row for `unified.ticks`.
The pipeline is built from parts that the live service backs with Redis and the database, so this program passes small stand-ins: a resolver that hands out one hand-built plan, an ownership engine under which Zerodha owns every instrument, and a clock fixed at one instant for the quote's `unified_at`. The session gate uses an empty calendar, so the Tuesday used here is an ordinary trading day. The second tick repeats the first one's values a second later, so it is dropped as a duplicate; the third has a new price and is written. `stale_quote` then marks the last quote stale, which the service does when an instrument's owner stops sending ticks and there is no backup.
Run it from the project root:
python examples/stock_brokers/instruments/ticks/utilities/pipeline/TickPipeline/example_1_ticks_into_quotes.py
"""
import datetime
import zoneinfo
from stock_brokers.instruments.ticks import base
from stock_brokers.instruments.ticks.base import (
InstrumentPlan,
)
from stock_brokers.instruments.ticks.utilities import pipeline
from stock_brokers.instruments.ticks.utilities import sessions
from stock_brokers.instruments.ticks.utilities.pipeline import (
TickPipeline,
)
from stock_brokers.instruments.ticks.utilities.sessions import (
SessionGate,
TradingCalendar,
)
from stock_brokers.instruments.ticks.zerodha import (
ZerodhaTickNormalizer,
)
INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
RELIANCE_ID = '11111111-1111-5111-8111-000000000102'
class StandInResolver:
"""A stand-in for `TickResolver` that knows one Zerodha token.
Attributes:
plan (InstrumentPlan): The plan for RELIANCE at Zerodha.
"""
def __init__(self):
"""Builds the plan for RELIANCE on NSE.
Returns:
None: This method returns nothing.
"""
identity = {
'exchange': 'nse',
'segment': 'nse_equities',
'shape': 'security',
'symbol': 'RELIANCE',
}
multipliers = {}
for field in base.QUANTITY_FIELDS:
multipliers[field] = 1
self.plan = InstrumentPlan(
instrument_id=RELIANCE_ID,
broker='zerodha',
broker_token='738561',
exchange='nse',
segment='nse_equities',
shape='security',
session=sessions.session_for('nse_equities'),
lot_size=1,
multipliers=multipliers,
price_decimals=2,
negative_prices=False,
close_policy=base.CLOSE_ALWAYS,
has_open_interest=False,
trusts_last_trade_time=True,
trusts_exchange_time=True,
identity_json=base.identity_json(RELIANCE_ID, 'zerodha', '738561', identity, 1),
)
def plan_for(self, broker, instrument_token, now):
"""Hands out the plan for Zerodha token 738561.
Args:
broker (str): The broker the tick came from.
instrument_token (int | str): The tick's token.
now (float): The current instant.
Returns:
InstrumentPlan | None: The plan, or None for any other token.
"""
if broker == 'zerodha' and instrument_token == 738561:
return self.plan
return None
class ZerodhaOwnsEverything:
"""A stand-in for the ownership engine under which Zerodha owns every instrument."""
def observe(self, broker, socket, instrument_id, exchange, received_at, window_end):
"""Says whether a broker's tick is written.
Args:
broker (str): The broker whose tick arrived.
socket (str): The socket it arrived on.
instrument_id (str): The instrument.
exchange (str): The canonical exchange.
received_at (float): When it arrived.
window_end (float): When the session window around it closes.
Returns:
tuple: (whether to write the tick, a list of ownership changes).
"""
return broker == 'zerodha', []
class FixedClock:
"""A clock that always answers the same instant.
Attributes:
instant (float): The instant, in epoch seconds.
"""
def __init__(self, instant):
"""Remembers the instant.
Args:
instant (float): The instant, in epoch seconds.
Returns:
None: This method returns nothing.
"""
self.instant = instant
def now(self):
"""Answers the instant.
Returns:
float: The instant, in epoch seconds.
"""
return self.instant
class TicksIntoQuotesExample:
"""Sends three ticks through a pipeline and prints what comes out.
Attributes:
start (float): When the first tick arrives, in epoch seconds.
pipeline (TickPipeline): The pipeline being shown.
"""
def __init__(self):
"""Builds the pipeline from its stand-in parts.
Returns:
None: This method returns nothing.
"""
moment = datetime.datetime(2026, 9, 15, 10, 0, tzinfo=INDIA)
self.start = moment.timestamp()
normalizers = {
'zerodha': ZerodhaTickNormalizer(),
}
self.pipeline = TickPipeline(
StandInResolver(),
ZerodhaOwnsEverything(),
normalizers,
gate=SessionGate(TradingCalendar.empty()),
clock=FixedClock(self.start + 0.25).now,
)
def tick(self, seconds_later, last_price):
"""Builds a Kite tick for RELIANCE.
Args:
seconds_later (int): Seconds after the first tick that this one arrives.
last_price (float): The last traded price.
Returns:
dict: The tick.
"""
received_at = self.start + seconds_later
return {
'instrument_token': 738561,
'last_price': last_price,
'last_quantity': 15,
'average_price': 1263.84,
'volume': 1885670,
'ohlc': {
'open': 1266.4,
'high': 1267.4,
'low': 1261.5,
'close': 1274.0,
},
'last_trade_time': received_at - 1.0,
'exchange_timestamp': received_at - 0.5,
'received_at': received_at,
'depth': {
'buy': [
{
'price': 1262.6,
'quantity': 120,
'orders': 4,
},
],
'sell': [
{
'price': 1262.7,
'quantity': 95,
'orders': 3,
},
],
},
}
def run(self):
"""Prints the outcome of each tick, the stale quote and the step counts.
Returns:
None: This method returns nothing.
"""
first = self.pipeline.process('zerodha', 'quotes', self.tick(0, 1262.7))
print(f'First tick written for {first.instrument_id}, ownership changes {first.transitions}')
print(f'Quote: {first.quote}')
print('Row, first twenty columns:')
for index in range(20):
print(f' {pipeline.TICK_COLUMNS[index]}: {first.row[index]}')
repeat = self.pipeline.process('zerodha', 'quotes', self.tick(1, 1262.7))
print(f'Same values a second later: {repeat}')
moved = self.pipeline.process('zerodha', 'quotes', self.tick(2, 1263.5))
print(f'New price two seconds later: row last_price {moved.row[5]}, previous_close {moved.row[14]}, change_percent {moved.row[15]}')
stale = self.pipeline.stale_quote(RELIANCE_ID, self.start + 120)
print(f'Stale quote ends with: {stale[-60:]}')
print(f"Stale quote for an instrument never written: {self.pipeline.stale_quote('11111111-1111-5111-8111-000000000999', self.start)}")
print(f'Counts: {dict(self.pipeline.counts)}')
if __name__ == '__main__':
TicksIntoQuotesExample().run()
It prints:
First tick written for 11111111-1111-5111-8111-000000000102, ownership changes []
Quote: {"instrument_id":"11111111-1111-5111-8111-000000000102","broker":"zerodha","broker_token":"738561","exchange":"nse","segment":"nse_equities","shape":"security","symbol":"RELIANCE","underlying_symbol":null,"expiry_date":null,"strike_price":null,"option_type":null,"lot_size":1,"last_price":1262.7,"average_price":1263.84,"ohlc":{"open":1266.4,"high":1267.4,"low":1261.5},"previous_close":1274.0,"change_percent":-0.887,"last_quantity":15,"volume":1885670,"buy_quantity":null,"sell_quantity":null,"oi":null,"oi_day_high":null,"oi_day_low":null,"depth":{"buy":[{"price":1262.6,"quantity":120,"orders":4}],"sell":[{"price":1262.7,"quantity":95,"orders":3}]},"last_trade_time":1789446599.0,"exchange_time":1789446599.5,"received_at":1789446600.0,"unified_at":1789446600.25,"stale":false,"stale_since":null}
Row, first twenty columns:
time: 1789446600.0
instrument_id: 11111111-1111-5111-8111-000000000102
broker: zerodha
exchange_time: 1789446599.5
last_trade_time: 1789446599.0
last_price: 1262.7
last_quantity: 15
average_price: 1263.84
volume: 1885670
buy_quantity: None
sell_quantity: None
open: 1266.4
high: 1267.4
low: 1261.5
previous_close: 1274.0
change_percent: -0.887
oi: None
oi_day_high: None
oi_day_low: None
lot_size: 1
Same values a second later: None
New price two seconds later: row last_price 1263.5, previous_close 1274.0, change_percent -0.8242
Stale quote ends with: d_at":1789446600.25,"stale":true,"stale_since":1789446720.0}
Stale quote for an instrument never written: None
Counts: {'received': 3, 'written': 2, 'duplicate': 1}
examples/stock_brokers/instruments/ticks/utilities/pipeline/TickPipeline/example_2_ticks_that_are_dropped.py
"""Shows every step at which the tick pipeline drops a tick, and a previous close seeded from before the process started.
The pipeline refuses the cheapest things first. A tick with no receipt time, or whose token resolves to no plan, is counted as `unresolved`. One received outside its instrument's session window, such as a Sunday reconnect replaying Friday's prices, is `out_of_session`. One from a broker that does not own the instrument is `not_owner`, unless it caused an ownership change, which is still returned so it can be recorded. One without a usable last price is `no_price`.
Groww's `close` is never the previous session's close, so a Groww tick has no previous close of its own. The live service remembers the owner's previous close from earlier in the day and, after a restart, seeds it from storage with `seed_previous_close`; the program seeds one so the change percentage can be computed. A seeded close only counts on the India day it was recorded for.
The resolver, the ownership engine and the clock are small stand-ins defined here: the resolver hands out hand-built plans for one Zerodha and one Groww token for the same instrument, and the ownership engine makes Groww the owner, reporting a change of owner on Groww's first tick. That first tick has a zero price, so it is not written, but the change of owner it caused is still returned. The session gate uses an empty calendar. All times are fixed, and no data store is involved.
Run it from the project root:
python examples/stock_brokers/instruments/ticks/utilities/pipeline/TickPipeline/example_2_ticks_that_are_dropped.py
"""
import datetime
import zoneinfo
from stock_brokers.instruments.ticks import base
from stock_brokers.instruments.ticks.base import (
InstrumentPlan,
)
from stock_brokers.instruments.ticks.groww import (
GrowwTickNormalizer,
)
from stock_brokers.instruments.ticks.utilities import sessions
from stock_brokers.instruments.ticks.utilities.pipeline import (
TickPipeline,
)
from stock_brokers.instruments.ticks.utilities.sessions import (
SessionGate,
TradingCalendar,
)
from stock_brokers.instruments.ticks.zerodha import (
ZerodhaTickNormalizer,
)
INDIA = zoneinfo.ZoneInfo('Asia/Kolkata')
RELIANCE_ID = '11111111-1111-5111-8111-000000000102'
class StandInResolver:
"""A stand-in for `TickResolver` that knows RELIANCE at Zerodha and at Groww.
Attributes:
plans (dict): (broker, token) to the plan for it.
"""
def __init__(self):
"""Builds the two plans.
Returns:
None: This method returns nothing.
"""
self.plans = {
('zerodha', 738561): self.plan(ZerodhaTickNormalizer(), 'zerodha', '738561'),
('groww', 'NSE|CASH|2885'): self.plan(GrowwTickNormalizer(), 'groww', '2885'),
}
def plan(self, normalizer, broker, broker_token):
"""Builds the plan for RELIANCE at one broker.
Args:
normalizer (TickNormalizer): The broker's normalizer, which decides the close policy and trusted times.
broker (str): The broker name.
broker_token (str): The broker's token.
Returns:
InstrumentPlan: The plan.
"""
identity = {
'exchange': 'nse',
'segment': 'nse_equities',
'shape': 'security',
'symbol': 'RELIANCE',
}
multipliers = {}
for field in base.QUANTITY_FIELDS:
multipliers[field] = 1
return InstrumentPlan(
instrument_id=RELIANCE_ID,
broker=broker,
broker_token=broker_token,
exchange='nse',
segment='nse_equities',
shape='security',
session=sessions.session_for('nse_equities'),
lot_size=1,
multipliers=multipliers,
price_decimals=2,
negative_prices=False,
close_policy=normalizer.close_policy('nse'),
has_open_interest=False,
trusts_last_trade_time=normalizer.TRUSTS_LAST_TRADE_TIME,
trusts_exchange_time=normalizer.TRUSTS_EXCHANGE_TIME,
identity_json=base.identity_json(RELIANCE_ID, broker, broker_token, identity, 1),
)
def plan_for(self, broker, instrument_token, now):
"""Hands out a plan by broker and token.
Args:
broker (str): The broker the tick came from.
instrument_token (int | str): The tick's token.
now (float): The current instant.
Returns:
InstrumentPlan | None: The plan, or None for a token it does not know.
"""
return self.plans.get((broker, instrument_token))
class GrowwOwnsReliance:
"""A stand-in ownership engine under which Groww owns every instrument.
Attributes:
claimed (set): Instruments Groww has already been recorded as owning.
"""
def __init__(self):
"""Starts with nothing claimed.
Returns:
None: This method returns nothing.
"""
self.claimed = set()
def observe(self, broker, socket, instrument_id, exchange, received_at, window_end):
"""Writes Groww's ticks and reports the change of owner on its first one.
Args:
broker (str): The broker whose tick arrived.
socket (str): The socket it arrived on.
instrument_id (str): The instrument.
exchange (str): The canonical exchange.
received_at (float): When it arrived.
window_end (float): When the session window around it closes.
Returns:
tuple: (whether to write the tick, a list of ownership changes).
"""
if broker != 'groww':
return False, []
if instrument_id in self.claimed:
return True, []
self.claimed.add(instrument_id)
change = {
'instrument_id': instrument_id,
'owner': 'groww',
}
return True, [
change,
]
class FixedClock:
"""A clock that always answers the same instant.
Attributes:
instant (float): The instant, in epoch seconds.
"""
def __init__(self, instant):
"""Remembers the instant.
Args:
instant (float): The instant, in epoch seconds.
Returns:
None: This method returns nothing.
"""
self.instant = instant
def now(self):
"""Answers the instant.
Returns:
float: The instant, in epoch seconds.
"""
return self.instant
class TicksThatAreDroppedExample:
"""Sends ticks that are dropped at each step, then two Groww ticks, and prints the counts.
Attributes:
tuesday (float): 10:00 India time on Tuesday 2026-09-15, in epoch seconds.
pipeline (TickPipeline): The pipeline being shown.
"""
def __init__(self):
"""Builds the pipeline from its stand-in parts.
Returns:
None: This method returns nothing.
"""
moment = datetime.datetime(2026, 9, 15, 10, 0, tzinfo=INDIA)
self.tuesday = moment.timestamp()
normalizers = {
'zerodha': ZerodhaTickNormalizer(),
'groww': GrowwTickNormalizer(),
}
self.pipeline = TickPipeline(
StandInResolver(),
GrowwOwnsReliance(),
normalizers,
gate=SessionGate(TradingCalendar.empty()),
clock=FixedClock(self.tuesday).now,
)
def tick(self, token, received_at, last_price):
"""Builds a tick for RELIANCE.
Args:
token (int | str): The broker's token as the feed spells it.
received_at (float | None): When it arrived, or None for a tick missing its receipt time.
last_price (float | None): The last traded price.
Returns:
dict: The tick.
"""
return {
'instrument_token': token,
'last_price': last_price,
'volume': 1885670,
'ohlc': {
'open': 1266.4,
'close': 1262.7,
},
'received_at': received_at,
}
def send(self, label, broker, tick):
"""Sends one tick and prints the outcome.
Args:
label (str): What the tick is meant to show.
broker (str): The broker whose feed published it.
tick (dict): The tick.
Returns:
None: This method returns nothing.
"""
output = self.pipeline.process(broker, 'quotes', tick)
if output is None:
print(f'{label}: dropped')
return
if output.row is None:
print(f'{label}: not written, ownership changes {output.transitions}')
return
print(f'{label}: written, last_price {output.row[5]}, previous_close {output.row[14]}, change_percent {output.row[15]}, ownership changes {output.transitions}')
def run(self):
"""Prints each outcome and the counts per step.
Returns:
None: This method returns nothing.
"""
sunday = datetime.datetime(2026, 9, 13, 10, 0, tzinfo=INDIA).timestamp()
monday = datetime.datetime(2026, 9, 14, 15, 30, tzinfo=INDIA).timestamp()
self.send('No receipt time', 'zerodha', self.tick(738561, None, 1262.7))
self.send('Unknown token', 'zerodha', self.tick(999, self.tuesday, 1262.7))
self.send('Sunday replay', 'zerodha', self.tick(738561, sunday, 1262.7))
self.send('Zerodha, not the owner', 'zerodha', self.tick(738561, self.tuesday, 1262.7))
self.send('Groww, zero price', 'groww', self.tick('NSE|CASH|2885', self.tuesday, 0))
self.send('Groww, no close yet', 'groww', self.tick('NSE|CASH|2885', self.tuesday + 1, 1262.7))
self.pipeline.seed_previous_close(RELIANCE_ID, monday, 1257.5)
self.send('Groww, after seeding Monday close', 'groww', self.tick('NSE|CASH|2885', self.tuesday + 2, 1262.9))
self.pipeline.seed_previous_close(RELIANCE_ID, self.tuesday, 1257.5)
self.send('Groww, after seeding it for today', 'groww', self.tick('NSE|CASH|2885', self.tuesday + 3, 1263.1))
for step, count in sorted(self.pipeline.counts.items()):
print(f'{step}: {count}')
if __name__ == '__main__':
TicksThatAreDroppedExample().run()
It prints:
No receipt time: dropped
Unknown token: dropped
Sunday replay: dropped
Zerodha, not the owner: dropped
Groww, zero price: not written, ownership changes [{'instrument_id': '11111111-1111-5111-8111-000000000102', 'owner': 'groww'}]
Groww, no close yet: written, last_price 1262.7, previous_close None, change_percent None, ownership changes []
Groww, after seeding Monday close: written, last_price 1262.9, previous_close None, change_percent None, ownership changes []
Groww, after seeding it for today: written, last_price 1263.1, previous_close 1257.5, change_percent 0.4453, ownership changes []
no_price: 1
not_owner: 1
out_of_session: 1
received: 8
unresolved: 2
written: 3
Methods:
| Name | Description |
|---|---|
process |
Take one tick through every step. |
stale_quote |
The last quote written for an instrument, flagged stale. |
seed_previous_close |
Remember an instrument's previous close from before this process started. |
Source code in stock_brokers/instruments/ticks/utilities/pipeline.py
Take one tick through every step.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
broker
|
str
|
The broker whose feed published the tick. |
required |
socket
|
str
|
The socket it arrived on. |
required |
tick
|
dict
|
The tick as the broker's market feed published it. |
required |
Returns:
| Type | Description |
|---|---|
Output | None
|
The outcome, or None when the tick was dropped without changing ownership. |
Example usage
From example_1_ticks_into_quotes.py
first = self.pipeline.process('zerodha', 'quotes', self.tick(0, 1262.7))
From example_1_ticks_into_quotes.py
repeat = self.pipeline.process('zerodha', 'quotes', self.tick(1, 1262.7))
From example_1_ticks_into_quotes.py
moved = self.pipeline.process('zerodha', 'quotes', self.tick(2, 1263.5))
Source code in stock_brokers/instruments/ticks/utilities/pipeline.py
143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 | |
The last quote written for an instrument, flagged stale.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instrument_id
|
str
|
The instrument id. |
required |
stale_since
|
float
|
When its owner went unhealthy with no backup. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The quote JSON, or None when nothing has been written for the instrument. |
Example usage
From example_1_ticks_into_quotes.py
print(f"Stale quote for an instrument never written: {self.pipeline.stale_quote('11111111-1111-5111-8111-000000000999', self.start)}")
Source code in stock_brokers/instruments/ticks/utilities/pipeline.py
Remember an instrument's previous close from before this process started.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instrument_id
|
str
|
The instrument id. |
required |
epoch
|
float
|
When it was recorded, which decides the day it belongs to. |
required |
previous_close
|
float
|
The previous close. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This function returns nothing. |
Example usage
From example_2_ticks_that_are_dropped.py
self.pipeline.seed_previous_close(RELIANCE_ID, monday, 1257.5)
From example_2_ticks_that_are_dropped.py
self.pipeline.seed_previous_close(RELIANCE_ID, self.tuesday, 1257.5)
Source code in stock_brokers/instruments/ticks/utilities/pipeline.py
The change since the previous close, as a percentage rounded to four places.
Recomputed rather than taken from the broker, because brokers disagree on what their own change field means - Kite's index packets send points where everything else sends a percentage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
last_price
|
float
|
The last traded price. |
required |
previous_close
|
float | None
|
The previous session's close. |
required |
Returns:
| Type | Description |
|---|---|
float | None
|
The change, or None without a previous close. |
Source code in stock_brokers/instruments/ticks/utilities/pipeline.py
The unified quote JSON for one normalized tick.
The one place the quote's layout is written, so the unified tick service and anything else that produces a quote - the REST API's broker fallback - hand out documents of exactly the same shape.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
plan
|
InstrumentPlan
|
The instrument's compiled plan, whose serialized identity is spliced in. |
required |
values
|
dict
|
What the broker's normalizer returned. |
required |
previous_close
|
float | None
|
The previous session's close. |
required |
change_percent
|
float | None
|
The change since that close. |
required |
received_at
|
float
|
When the tick was received, in epoch seconds. |
required |
unified_at
|
float
|
When the quote was built, in epoch seconds. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The quote as compact JSON. |