Implement dollar bars sampling for financial time series. Dollar bars sample a new bar every time a fixed dollar volume threshold (price * volume) is reached.
def dollar_bars(trades: list[dict], dollar_threshold: float) -> list[dict]:
"""
trades: list of dicts with keys 'price' and 'volume'
dollar_threshold: cumulative dollar volume required to form a bar
"""
bars = []
current_prices = []
current_volume = 0.0
current_dollar_volume = 0.0
for trade in trades:
current_prices.append(trade["price"])
current_volume += trade["volume"]
current_dollar_volume += trade["price"] * trade["volume"]
if current_dollar_volume >= dollar_threshold:
bar = {
"open": current_prices[0],
"high": max(current_prices),
"low": min(current_prices),
"close": current_prices[-1],
"volume": current_volume,
"dollar_volume": current_dollar_volume,
"num_ticks": len(current_prices),
}
bars.append(bar)
current_prices = []
current_volume = 0.0
current_dollar_volume = 0.0
if current_prices:
bar = {
"open": current_prices[0],
"high": max(current_prices),
"low": min(current_prices),
"close": current_prices[-1],
"volume": current_volume,
"dollar_volume": current_dollar_volume,
"num_ticks": len(current_prices),
}
bars.append(bar)
return bars