第一次发帖，因为要用tick合成的秒级别数据，折腾了几天，终于完成，功能如下：
通过tick合成1秒数据，通过1秒合成能整除60秒或者被60秒整除的bar组合，欢迎大家测试看是否还有bug
替换原有BarGenerator，不影响原有功能
class BarGenerator:
"""
For:
1. generating 1 minute bar data from tick data
2. generating x minute bar/x hour bar data from 1 minute data
Notice:
1. for x minute bar, x must be able to divide 60: 2, 3, 5, 6, 10, 15, 20, 30
2. for x hour bar, x can be any number
"""
def __init__(
self,
on_bar: Callable,
window: int = 0,
on_window_bar: Callable = None,
interval: Interval = Interval.MINUTE,
bar_size: list = ['15', '60', '300'],
on_build_bars: Callable = None,
):
"""Constructor"""
self.bar_s: BarData = None
self.bar: BarData = None
self.on_bar: Callable = on_bar
self.interval: Interval = interval
self.interval_count: int = 0
self.hour_bar: BarData = None
self.window: int = window
self.window_bar: BarData = None
self.on_window_bar: Callable = on_window_bar
# 设置所需要的bar周期，以秒为单位
self.bar_size = bar_size
self.bar_list: dict = {x: None for x in self.bar_size}
self.on_build_bars: Callable = on_build_bars
self.last_tick: TickData = None
self.last_bar: BarData = None
def update_tick(self, tick: TickData) -> None:
if self.on_build_bars:
self.update_second(tick)
self.update_minute(tick)
self.last_tick = tick
def update_second(self, tick: TickData) -> None:
"""根据 tick 数据，生成 1秒的 bar线"""
# print('-----------', tick.datetime)
# if tick.datetime.microsecond == 0:
# tick.datetime += datetime.timedelta(microseconds=500000)
# if 15 < tick.datetime.hour < 21:
# print('是')
# return
# print(tick.datetime)
# 过滤掉 last_price 为0的情况
if not tick.last_price:
return
# 过滤掉旧日期的 tick 数据
if self.last_tick and tick.datetime < self.last_tick.datetime:
return
if self.bar_s:
self.bar_s.high_price = max(self.bar_s.high_price, tick.last_price)
if tick.high_price > self.last_tick.high_price:
self.bar_s.high_price = max(self.bar_s.high_price, tick.high_price)
self.bar_s.low_price = min(self.bar_s.low_price, tick.last_price)
if tick.low_price < self.last_tick.low_price:
self.bar_s.low_price = min(self.bar_s.low_price, tick.low_price)
self.bar_s.close_price = tick.last_price
self.bar_s.open_interest = tick.open_interest
self.bar_s.datetime = tick.datetime
else:
self.bar_s = BarData(
symbol=tick.symbol,
exchange=tick.exchange,
interval=Interval.SECOND,
datetime=tick.datetime,
gateway_name=tick.gateway_name,
open_price=tick.last_price,
high_price=tick.last_price,
low_price=tick.last_price,
close_price=tick.last_price,
open_interest=tick.open_interest
)
if self.last_tick:
volume_change = tick.volume - self.last_tick.volume
self.bar_s.volume += max(volume_change, 0)
else:
self.bar_s.volume += max(tick.last_volume, 0)
if self.last_tick and not (tick.datetime.replace(microsecond=0).second == 0 and self.last_tick.datetime.replace(microsecond=0).second == 0):
if self.last_tick and (
(self.last_tick.datetime.minute != tick.datetime.minute)
or (self.last_tick.datetime.hour != tick.datetime.hour)
or (self.last_tick.datetime.second != tick.datetime.second)
):
self.bar_s.datetime = self.bar_s.datetime.replace(
microsecond=0
)
[self.update_bar_second_window(self.bar_s, size=x) for x in self.bar_size]
self.last_bar = self.bar_s
self.bar_s = None
def update_minute(self, tick: TickData) -> None:
"""
Update new tick data into generator.
"""
# Filter tick data with 0 last price
if not tick.last_price:
return
# Filter tick data with older timestamp
if self.last_tick and tick.datetime < self.last_tick.datetime:
return
if self.bar:
self.bar.high_price = max(self.bar.high_price, tick.last_price)
if tick.high_price > self.last_tick.high_price:
self.bar.high_price = max(self.bar.high_price, tick.high_price)
self.bar.low_price = min(self.bar.low_price, tick.last_price)
if tick.low_price < self.last_tick.low_price:
self.bar.low_price = min(self.bar.low_price, tick.low_price)
self.bar.close_price = tick.last_price
self.bar.open_interest = tick.open_interest
self.bar.datetime = tick.datetime
else:
self.bar = BarData(
symbol=tick.symbol,
exchange=tick.exchange,
interval=Interval.MINUTE,
datetime=tick.datetime,
gateway_name=tick.gateway_name,
open_price=tick.last_price,
high_price=tick.last_price,
low_price=tick.last_price,
close_price=tick.last_price,
open_interest=tick.open_interest
)
if self.last_tick:
volume_change = tick.volume - self.last_tick.volume
self.bar.volume += max(volume_change, 0)
turnover_change = tick.turnover - self.last_tick.turnover
self.bar.turnover += max(turnover_change, 0)
# if self.last_tick and self.last_tick.datetime + datetime.timedelta(minutes=10) > tick.datetime:
if self.last_tick and not (tick.datetime.replace(microsecond=0).second == 0 and self.last_tick.datetime.replace(microsecond=0).second == 0):
if self.last_tick and (
(self.last_tick.datetime.minute != tick.datetime.minute)
or (self.last_tick.datetime.hour != tick.datetime.hour)
):
self.bar.datetime = self.bar.datetime.replace(
second=0, microsecond=0
)
self.on_bar(self.bar)
self.bar = None
def update_bar(self, bar: BarData) -> None:
"""
Update 1 minute bar into generator
"""
# if self.interval == Interval.SECOND:
# [self.update_bar_second_window(self.bar_s, size=x) for x in self.bar_size]
if self.interval == Interval.MINUTE:
self.update_bar_minute_window(bar)
else:
self.update_bar_hour_window(bar)
def update_bar_minute_window(self, bar: BarData) -> None:
""""""
# If not inited, create window bar object
if not self.window_bar:
dt = bar.datetime.replace(second=0, microsecond=0)
self.window_bar = BarData(
symbol=bar.symbol,
exchange=bar.exchange,
datetime=dt,
gateway_name=bar.gateway_name,
open_price=bar.open_price,
high_price=bar.high_price,
low_price=bar.low_price
)
# Otherwise, update high/low price into window bar
else:
self.window_bar.high_price = max(
self.window_bar.high_price,
bar.high_price
)
self.window_bar.low_price = min(
self.window_bar.low_price,
bar.low_price
)
# Update close price/volume/turnover into window bar
self.window_bar.close_price = bar.close_price
self.window_bar.volume += bar.volume
self.window_bar.turnover += bar.turnover
self.window_bar.open_interest = bar.open_interest
# Check if window bar completed
if not (bar.datetime.minute + 1) % self.window:
self.on_window_bar(self.window_bar)
self.window_bar = None
def update_bar_hour_window(self, bar: BarData) -> None:
""""""
# If not inited, create window bar object
if not self.hour_bar:
dt = bar.datetime.replace(minute=0, second=0, microsecond=0)
self.hour_bar = BarData(
symbol=bar.symbol,
exchange=bar.exchange,
datetime=dt,
gateway_name=bar.gateway_name,
open_price=bar.open_price,
high_price=bar.high_price,
low_price=bar.low_price,
close_price=bar.close_price,
volume=bar.volume,
turnover=bar.turnover,
open_interest=bar.open_interest
)
return
finished_bar = None
# If minute is 59, update minute bar into window bar and push
if bar.datetime.minute == 59:
self.hour_bar.high_price = max(
self.hour_bar.high_price,
bar.high_price
)
self.hour_bar.low_price = min(
self.hour_bar.low_price,
bar.low_price
)
self.hour_bar.close_price = bar.close_price
self.hour_bar.volume += bar.volume
self.hour_bar.turnover += bar.turnover
self.hour_bar.open_interest = bar.open_interest
finished_bar = self.hour_bar
self.hour_bar = None
# If minute bar of new hour, then push existing window bar
elif bar.datetime.hour != self.hour_bar.datetime.hour:
finished_bar = self.hour_bar
dt = bar.datetime.replace(minute=0, second=0, microsecond=0)
self.hour_bar = BarData(
symbol=bar.symbol,
exchange=bar.exchange,
datetime=dt,
gateway_name=bar.gateway_name,
open_price=bar.open_price,
high_price=bar.high_price,
low_price=bar.low_price,
close_price=bar.close_price,
volume=bar.volume,
turnover=bar.turnover,
open_interest=bar.open_interest
)
# Otherwise only update minute bar
else:
self.hour_bar.high_price = max(
self.hour_bar.high_price,
bar.high_price
)
self.hour_bar.low_price = min(
self.hour_bar.low_price,
bar.low_price
)
self.hour_bar.close_price = bar.close_price
self.hour_bar.volume += bar.volume
self.hour_bar.turnover += bar.turnover
self.hour_bar.open_interest = bar.open_interest
# Push finished window bar
if finished_bar:
self.on_hour_bar(finished_bar)
def on_hour_bar(self, bar: BarData) -> None:
""""""
if self.window == 1:
self.on_window_bar(bar)
else:
if not self.window_bar:
self.window_bar = BarData(
symbol=bar.symbol,
exchange=bar.exchange,
datetime=bar.datetime,
gateway_name=bar.gateway_name,
open_price=bar.open_price,
high_price=bar.high_price,
low_price=bar.low_price
)
else:
self.window_bar.high_price = max(
self.window_bar.high_price,
bar.high_price
)
self.window_bar.low_price = min(
self.window_bar.low_price,
bar.low_price
)
self.window_bar.close_price = bar.close_price
self.window_bar.volume += bar.volume
self.window_bar.turnover += bar.turnover
self.window_bar.open_interest = bar.open_interest
self.interval_count += 1
if not self.interval_count % self.window:
self.interval_count = 0
self.on_window_bar(self.window_bar)
self.window_bar = None
def update_bar_second_window(self, bar: BarData, size) -> None:
""""""
# 如果已存在bar，但是当前时间已超过上根bar合成时间，则将上根bar先推出去
if self.last_bar and self.bar_list[size]:
while self.last_bar.datetime < bar.datetime:
if (int(size) < 60 and not (self.last_bar.datetime.second) % int(size)) \
or (self.last_bar.datetime.second == 0 and int(size) >= 60 and not self.last_bar.datetime.minute % (int(size)/60)):
self.bar_list[size].datetime = self.last_bar.datetime.replace(microsecond=0)
self.on_build_bars({size:self.bar_list[size]})
self.bar_list[size] = None
break
else:
self.last_bar.datetime+=datetime.timedelta(seconds=1)
dt = bar.datetime.replace(microsecond=0)
# If not inited, create window bar object
if not self.bar_list[size]:
# dt = bar.datetime.replace(microsecond=0)
self.bar_list[size] = BarData(
symbol=bar.symbol,
exchange=bar.exchange,
datetime=dt,
gateway_name=bar.gateway_name,
open_price=bar.open_price,
high_price=bar.high_price,
low_price=bar.low_price
)
# Otherwise, update high/low price into window bar
else:
self.bar_list[size].high_price = max(
self.bar_list[size].high_price,
bar.high_price
)
self.bar_list[size].low_price = min(
self.bar_list[size].low_price,
bar.low_price
)
# Update close price/volume/turnover into window bar
self.bar_list[size].close_price = bar.close_price
self.bar_list[size].volume += bar.volume
self.bar_list[size].turnover += bar.turnover
self.bar_list[size].open_interest = bar.open_interest
self.bar_list[size].datetime = dt
if (int(size) < 60 and not (bar.datetime.second) % int(size)) \
or (bar.datetime.second == 0 and int(size) >= 60 and not bar.datetime.minute % (int(size)/60)):
self.bar_list[size].datetime = bar.datetime.replace(microsecond=0)
self.on_build_bars({size:self.bar_list[size]})
self.bar_list[size] = None
def generate(self) -> Optional[BarData]:
"""
Generate the bar data and call callback immediately.
"""
bar = self.bar
if self.bar:
bar.datetime = bar.datetime.replace(second=0, microsecond=0)
self.on_bar(bar)
self.bar = None
return bar使用方法:
self.bg = BarGenerator(self.on_bar, 1, interval=Interval.MINUTE, on_window_bar=self.on_window_bar, bar_size=['30', '60', '120'], on_build_bars=self.on_build_bars)def on_build_bars(self, bar_data):
print(['------------%s:%s'%(x,bar_data[x].datetime) for x in bar_data.keys()])
