1. 等自然时长机制的K线生成器说明

交易日内等自然时长K线生成器，可以实现周以下的周期单位的时间就间隔：
x=window，interval的取：
Interval.MINUTE ：x分钟宽度的K线，对齐交易日开始时间，不可以超过日交易时长
Interval.HOUR   : x小时宽度的K线，对齐交易日开始时间，不可以超过日交易时长
Interval.DAILY  : x日宽度的K线，对齐上市交易日开始时间开始的整x日，x可以任意值
Interval.WEEKLY  : x周宽度的K线，对齐上市交易日开始时间整x周，x可以任意值

再策略中的应用举例：

• 等自然时长机制的30分钟K线创建：
  self.bg = ENTBarGenerator(self.on_bar,30,self.on_30m_bar,interval=Interval.MINUTE，self.vt_symbol)
• 等自然时长机制的2小时K线创建：
  self.bg = ENTBarGenerator(self.on_bar,120,self.on_2h_bar,interval=Interval.MINUTE，self.vt_symbol)
  或者：
  self.bg = ENTBarGenerator(self.on_bar,2,self.on_2h_bar,interval=Interval.HOUR，self.vt_symbol)
• 等自然时长机制的1日K线创建：
  self.bg = ENTBarGenerator(self.on_bar,1,self.on_1d_bar,interval=Interval.DAILY，self.vt_symbol)
• 等自然时长机制的1周K线创建：
  self.bg = ENTBarGenerator(self.on_bar,5,self.on_1w_bar,interval=Interval.DAILY，self.vt_symbol)
  或者：
  self.bg = ENTBarGenerator(self.on_bar,1,self.on_1w_bar,interval=Interval.WEEKLY，self.vt_symbol)

2. 先实现等自然时长机制的K线生成器

保存文件：vnpy\usertools\equal_nature_time.py
内容如下：

"""
实现一个等自然时长机制的K线生成器

作者：hxxjava 
日期：2020-9-2
"""

from typing import Callable,List,Dict, Tuple, Union
from vnpy.app.cta_strategy import BarGenerator
from vnpy.trader.constant import Interval
from vnpy.trader.utility import extract_vt_symbol
from vnpy.trader.object import BarData

import pytz
CHINA_TZ = pytz.timezone("Asia/Shanghai")

from vnpy.usertools.trade_hour import (
    TradeHours,
    Timeunit,
    get_listed_date,
    get_de_listed_date,
)

import datetime 
import rqdatac as rq


class MyError(Exception):
    def __init__(self,ErrorInfo):
        super().__init__(self) #初始化父类
        self.errorinfo=ErrorInfo
    def __str__(self):
        return self.errorinfo       

class ENTBarGenerator(BarGenerator):
    """ 
    交易日内等自然时长K线生成器，可以实现周以下的周期单位的时间就间隔：
    x=window，interval的取：
    Interval.MINUTE ：x分钟宽度的K线，对齐交易日开始时间，不可以超过日交易时长
    Interval.HOUR   : x小时宽度的K线，对齐交易日开始时间，不可以超过日交易时长
    Interval.DAILY  : x日宽度的K线，对齐上市交易日开始时间开始的整x日，x可以任意值
    Interval.WEEKLY  : x周宽度的K线，对齐上市交易日开始时间整x周，x可以任意值

    ENT：equal nature time ———— 等自然时长的缩写
    """ 
    def __init__(
        self,
        on_bar: Callable,
        window: int = 0,
        on_window_bar: Callable = None,
        interval: Interval = Interval.MINUTE,
        vt_symbol:str=''
    ):
        super().__init__(on_bar,window,on_window_bar,interval)

        symbol,exchange = extract_vt_symbol(vt_symbol)
        self.trade_hours = TradeHours(symbol.upper())

        self.trade_start,self.trade_stop = None,None

        self.bar_width = self.get_bar_witdh()
        self.bar_index = None

        if self.interval in [Interval.DAILY,Interval.WEEKLY]:
            self.kx_start,self.kx_stop = (None,None)

    def get_bar_witdh(self):
        """
        求window_bar的宽度
        """
        # 求每日的交易时长
        TTPD = self.trade_hours.get_trade_time_perday()

        try:
            if self.interval == Interval.MINUTE:
                # 以分钟作为单位
                bar_width = self.window
                if bar_width>TTPD:
                    raise MyError(f'window_bar宽度不可以大于1日的交易总时长')

            elif self.interval == Interval.HOUR:
                # 以小时作为单位
                bar_width = self.window*60
                if bar_width>TTPD:
                    raise MyError(f'window_bar宽度不可以大于1日的交易总时长')

            elif self.interval == Interval.DAILY:
                # 以日作为单位
                bar_width = self.window * TTPD

            elif self.interval == Interval.WEEKLY:
                # 以周作为单位
                bar_width = self.window * TTPD * 5

            else:
                raise MyError(f'不可以使用{self.interval}作为interval参数')

            return bar_width

        except MyError as e:
            print(e)


    def get_bar_index(self,bar:BarData):
        """
        计算当前分钟bar所属window_bar的日内索引
        """
        time_diff = bar.datetime - self.trade_start
        days = time_diff.days 
        sec = time_diff.seconds
        us = time_diff.microseconds
        minutes = (days*24*3600 +sec+us*0.000001)/60.0 
        index = int(minutes/self.bar_width)
        return index 

    def update_bar(self,bar:BarData) -> None:
        if self.interval in [Interval.MINUTE,Interval.HOUR]:
            self.update_bar1(bar)
        elif self.interval in [Interval.DAILY,Interval.WEEKLY]:
            self.update_bar2(bar)

    def update_bar1(self,bar:BarData) -> None:
        """
        Update 1 minute bar into generator
        """

        # 如果bar的时间戳笔windows的时间戳还早，丢弃bar
        if self.window_bar and bar.datetime < self.window_bar.datetime:
            return

        in_trade,(trade_start,trade_stop) = self.trade_hours.get_date_start_stop(bar.datetime)
        if not in_trade:
            # 无效K线，不可以处理
            # print(f"bar.datetime= {bar.datetime} 无效K线，不可以处理")
            return

        if (self.trade_start,self.trade_stop) == (None,None):
            # 计算当日的开始和停止交易时间
            self.trade_start,self.trade_stop = trade_start,trade_stop
        elif trade_start != self.trade_start:
            # 判断是否跨日,更新当日的开始和停止交易时间
            self.trade_start,self.trade_stop = trade_start,trade_stop

        # print(f"!1 {self.trade_start,self.trade_stop}")

        # 计算当前window_bar的日内索引
        bar_index = self.get_bar_index(bar)

        if self.bar_index is None or bar_index != self.bar_index:
            # 产生新window_barK线
            self.bar_index = bar_index

            if self.window_bar:
                # 推送已经走完的window_bar
                self.on_window_bar(self.window_bar)
                self.window_bar = None

            # 计算K线的交易起止时间
            bar_datetime = self.trade_start + datetime.timedelta(minutes=self.bar_index*self.bar_width)

            # 生成新的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,
            )         

        # 更新window_bar的high和low
        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)

        # 更新 close price/volume到window bar
        self.window_bar.close_price = bar.close_price
        self.window_bar.volume += int(bar.volume)
        self.window_bar.open_interest = bar.open_interest 

    def update_bar2(self,bar:BarData) -> None:
        """
        Update 1 minute bar into generator
        """

        # 如果bar的时间戳笔windows的时间戳还早，丢弃bar
        if self.window_bar and bar.datetime < self.window_bar.datetime:
            return

        if (self.kx_start,self.kx_stop) == (None,None):
            self.kx_start,self.kx_stop = self.trade_hours.get_bar_window(bar.datetime,self.window,self.interval)
            if (self.kx_start,self.kx_stop) == (None,None):
                return

        # If not inited, creaate window bar object
        if (not self.window_bar):
            # 获得K线的交易起止时间
            self.window_bar = BarData(
                symbol=bar.symbol,
                exchange=bar.exchange,
                datetime=self.kx_start,
                gateway_name=bar.gateway_name,
                open_price=bar.open_price,
                high_price=bar.high_price,
                low_price=bar.low_price,
            )         

        elif self.kx_start <= bar.datetime and bar.datetime < self.kx_stop:
            # 1分钟K线属于当前K线
            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)

        elif bar.datetime >= self.kx_stop:       # Check if window bar completed
            self.on_window_bar(self.window_bar)
            self.window_bar = None

            self.kx_start,self.kx_stop = self.trade_hours.get_bar_window(bar.datetime,self.window,self.interval)
            if (self.kx_start,self.kx_stop) == (None,None): 
                # 不在交易时段
                return

            self.window_bar = BarData(
                symbol=bar.symbol,
                exchange=bar.exchange,
                datetime=self.kx_start,
                gateway_name=bar.gateway_name,
                open_price=bar.open_price,
                high_price=bar.high_price,
                low_price=bar.low_price,
            )       

        # Update close price/volume into window bar
        self.window_bar.close_price = bar.close_price
        self.window_bar.volume += int(bar.volume)
        self.window_bar.open_interest = bar.open_interest 



# 下面是ENTBarGenerator类的测试代码
def test():
    from vnpy.trader.constant import Exchange
    import pytz
    CHINA_TZ = pytz.timezone("Asia/Shanghai")

    def on_bar(bar: BarData):
        pass

    def on_xbar_bar(bar: BarData):
        print(bar)

    bar_gen = ENTBarGenerator(on_bar,30,on_xbar_bar,Interval.MINUTE,vt_symbol='rb2010.SHFE')
    dt0 = CHINA_TZ.localize(datetime.datetime(2020,8,31,9,20,1))

    for i in range(60*24):
        # print(f"{[dt0 + datetime.timedelta(minutes=i)]}")
        bar = BarData(
            gateway_name = 'CTP', 
            symbol = 'rb2010', 
            exchange = Exchange.SHFE, 
            datetime = dt0 + datetime.timedelta(minutes=i),
            interval=Interval.MINUTE,
            volume=2,
            open_price=3560.0,
            high_price= 3565.0,
            close_price=3558.0,
            low_price=3555.0,
            open_interest=10,
        )

        bar_gen.update_bar(bar)

if __name__ == "__main__":
    # RQDatac初始化
    rq.init('xxxxxx','******',("rqdatad-pro.ricequant.com",16011))

    test()

3. 必须的合约交易时间段类

创建vnpy\usertools\trade_hour.py
内容如下：

"""
本文件主要实现合约的交易时间段
"""
from typing import Callable,List,Dict, Tuple, Union
from enum import Enum

import datetime
import pytz
CHINA_TZ = pytz.timezone("Asia/Shanghai")

from vnpy.trader.utility import extract_vt_symbol
from vnpy.trader.constant import Interval

from rqdatac.utils import to_date
import rqdatac as rq


def get_listed_date(symbol:str):
    ''' 
    获得上市日期 
    '''
    info = rq.instruments(symbol)
    return to_date(info.listed_date)

def get_de_listed_date(symbol:str):
    ''' 
    获得交割日期 
    '''
    info = rq.instruments(symbol)
    return to_date(info.de_listed_date)

class Timeunit(Enum):
    """ 
    时间单位 
    """
    SECOND = '1s'
    MINUTE = '1m'
    HOUR = '1h'

class TradeHours(object):
    """ 合约交易时间段 """
    def __init__(self,symbol:str):
        self.symbol = symbol.upper()
        self.init()

    def init(self):
        """ 
        初始化交易日字典及交易时间段数据列表 
        """
        self.listed_date = get_listed_date(self.symbol)
        self.de_listed_date = get_de_listed_date(self.symbol)

        self.trade_date_index = {}   # 合约的交易日索引字典
        self.trade_index_date = {}   # 交易天数与交易日字典

        trade_dates = rq.get_trading_dates(self.listed_date,self.de_listed_date) # 合约的所有的交易日
        days = 0
        for td in trade_dates:
            self.trade_date_index[td] = days
            self.trade_index_date[days] = td
            days += 1

        trading_hours = rq.get_trading_hours(self.symbol,date=self.listed_date,frequency='tick',expected_fmt='datetime')

        self.time_dn_pairs = self._get_trading_times_dn(trading_hours)

        trading_hours0 = [(CHINA_TZ.localize(start),CHINA_TZ.localize(stop)) for start,stop in trading_hours]
        self.trade_date_index[self.listed_date] = (0,trading_hours0)
        for day in range(1,days):
            td = self.trade_index_date[day]
            trade_datetimes = []
            for (start,dn1),(stop,dn2) in self.time_dn_pairs:
                #start:开始时间,dn1：相对交易日前推天数,
                #stop :开始时间,dn2：相对开始时间后推天数     
                d = self.trade_index_date[day+dn1]
                start_dt = CHINA_TZ.localize(datetime.datetime.combine(d,start))
                stop_dt = CHINA_TZ.localize(datetime.datetime.combine(d,stop))
                trade_datetimes.append((start_dt,stop_dt+datetime.timedelta(days=dn2)))
            self.trade_date_index[td] = (day,trade_datetimes)

    def _get_trading_times_dn(self,trading_hours:List[Tuple[datetime.datetime,datetime.datetime]]): 
        """ 
        交易时间跨天处理，不推荐外部使用 。
        产生的结果：[((start1,dn11),(stop1,dn21)),((start2,dn12),(stop2,dn22)),...,((startN,dn1N),(stopN,dn2N))]
        其中：
            startN:开始时间,dn1N：相对交易日前推天数,
            stopN：开始时间,dn2N：相对开始时间后推天数      
        """
        ilen = len(trading_hours)
        if ilen == 0:
            return []
        start_stops = []
        for start,stop in trading_hours:
            start_stops.insert(0,(start.time(),stop.time()))

        pre_start,pre_stop = start_stops[0]
        dn1 = 0
        dn2 = 1 if pre_start > pre_stop else 0
        time_dn_pairs = [((pre_start,dn1),(pre_stop,dn2))]
        for start,stop in start_stops[1:]:
            if start > pre_start:
                dn1 -= 1
            dn2 = 1 if start > stop else 0
            time_dn_pairs.insert(0,((start,dn1),(stop,dn2)))
            pre_start,pre_stop = start,stop

        return time_dn_pairs

    def get_date_tradetimes(self,date:datetime.date):
        """ 
        得到合约date日期的交易时间段 
        """
        idx,trade_times = self.trade_date_index.get(date,(None,[]))
        return idx,trade_times

    def get_trade_datetimes(self,dt:datetime,allday:bool=False):
        """ 
        得到合约date日期的交易时间段 
        """
        # 得到最早的交易时间
        idx0,trade_times0 = self.get_date_tradetimes(self.listed_date)
        start0,stop0 = trade_times0[0]
        if dt < start0:
            return None,[]

        # 首先找到dt日期自上市以来的交易天数
        date,dn = dt.date(),0
        days = None
        while date < self.de_listed_date:
            days,ths = self.trade_date_index.get(date,(None,[]))
            if not days:
                dn += 1
                date = (dt+datetime.timedelta(days=dn)).date()
            else:
                break
        # 如果超出交割日也没有找到，那这就不是一个有效的交易时间
        if days is None:
            return (None,[])

        index_3 = [days,days+1,days-1]  # 前后三天的

        date_3d = []
        for day in index_3: 
            date = self.trade_index_date.get(day,None)
            date_3d.append(date)

        # print(date_3d)

        for date in date_3d:
            if not date:
                # print(f"{date} is not trade date")
                continue

            idx,trade_dts = self.get_date_tradetimes(date)
            # print(f"{date} tradetimes {trade_dts}")
            ilen = len(trade_dts)
            if ilen > 0:
                start0,stop = trade_dts[0]      # start0 是date交易日的开始时间
                start,stop0 = trade_dts[-1]
            if dt<start0 or dt>stop0:
                continue

            for start,stop in trade_dts:
                if dt>=start and dt < stop:
                    if allday:
                        return idx,trade_dts
                    else:
                        return idx,[(start,stop)]

        return None,[]

    def get_trade_time_perday(self):
        """ 
        计算每日的交易总时长（单位：分钟） 
        """
        TTPD = datetime.timedelta(0,0,0)

        datetimes = []
        today = datetime.datetime.now().date()

        for (start,dn1),(stop,dn2) in self.time_dn_pairs:
            start_dt = CHINA_TZ.localize(datetime.datetime.combine(today,start)) + datetime.timedelta(days=dn1)
            stop_dt = CHINA_TZ.localize(datetime.datetime.combine(today,stop)) + datetime.timedelta(days=dn2)
            time_delta = stop_dt - start_dt
            TTPD = TTPD + time_delta
        return int(TTPD.seconds/60)

    def get_trade_time_inday(self,dt:datetime,unit:Timeunit=Timeunit.MINUTE):
        """ 
        计算dt在交易日内的分钟数 
        unit: '1s':second;'1m':minute;'1h';1h
        """
        TTID = datetime.timedelta(0,0,0)

        day,trade_times = self.get_trade_datetimes(dt,allday=True)
        if not trade_times:
            return None

        for start,stop in trade_times:
            if dt > stop:
                time_delta = stop - start
                TTID += time_delta
            elif dt > start:
                time_delta = dt - start
                TTID += time_delta     
                break
            else:
                break          

        if unit == Timeunit.SECOND:
            return TTID.seconds
        elif unit == Timeunit.MINUTE:
            return int(TTID.seconds/60) 
        elif unit == Timeunit.HOUR:
            return int(TTID.seconds/3600) 
        else:
            return TTID

    def get_day_tradetimes(self,dt:datetime):
        """ 
        得到合约日盘的交易时间段 
        """
        index,trade_times = self.get_trade_datetimes(dt,allday=True)
        trade_times1 = []
        if trade_times:
            for start_dt,stop_dt in trade_times:
                if start_dt.time() < datetime.time(18,0,0):
                    trade_times1.append((start_dt,stop_dt))
            return index,trade_times1
        return (index,trade_times1)

    def get_night_tradetimes(self,dt:datetime):
        """ 
        得到合约夜盘的交易时间段 
        """
        index,trade_times = self.get_trade_datetimes(dt,allday=True)
        trade_times1 = []
        if trade_times:
            for start_dt,stop_dt in trade_times:
                if start_dt.time() > datetime.time(18,0,0):
                    trade_times1.append((start_dt,stop_dt))
            return index,trade_times1
        return (index,trade_times1)

    def convet_to_datetime(self,day:int,minutes:int):
        """ 
        计算minutes在第day交易日内的datetime形式的时间 
        """
        date = self.trade_index_date.get(day,None)
        if date is None:
            return None
        idx,trade_times = self.trade_date_index.get(date,(None,[]))
        if not trade_times:     # 不一定必要
            return None
        for (start,stop) in trade_times:
            timedelta = stop - start 
            if minutes < int(timedelta.seconds/60):
                return start + datetime.timedelta(minutes=minutes)
            else:
                minutes -= int(timedelta.seconds/60)
        return None

    def get_bar_window(self,dt:datetime,window:int,interval:Interval=Interval.MINUTE):
        """ 
        计算dt所在K线的起止时间 
        """
        bar_windows = (None,None)

        day,trade_times = self.get_trade_datetimes(dt,allday=True)
        if not trade_times:
            # print(f"day={day} trade_times={trade_times}")
            return bar_windows

        # 求每个交易日的交易时间分钟数
        TTPD = self.get_trade_time_perday()

        # 求dt在交易日内的分钟数
        TTID = self.get_trade_time_inday(dt,unit=Timeunit.MINUTE)

        # 得到dt时刻K线的起止时间 
        total_minites = day*TTPD + TTID

        # 计算K线宽度(分钟数)
        if interval == Interval.MINUTE:
            bar_width = window
        elif interval == Interval.HOUR:
            bar_width = 60*window
        elif interval == Interval.DAILY:
            bar_width = TTPD*window
        elif interval == Interval.WEEKLY:
            bar_width = TTPD*window*5
        else:
            return bar_windows

        # 求K线的开始时间的和结束的分钟形式
        start_m = int(total_minites/bar_width)*bar_width
        stop_m = start_m + bar_width

        # 计算K开始时间的datetime形式
        start_d = int(start_m / TTPD)
        minites = start_m % TTPD
        start_dt = self.convet_to_datetime(start_d,minites)
        # print(f"start_d={start_d} minites={minites}---->{start_dt}")

        # 计算K结束时间的datetime形式
        stop_d = int(stop_m / TTPD)
        minites = stop_m % TTPD
        stop_dt = self.convet_to_datetime(stop_d,minites)
        # print(f"stop_d={stop_d} minites={minites}---->{stop_dt}")

        return start_dt,stop_dt

    def get_date_start_stop(self,dt:datetime):
        """
        获得dt所在交易日的开始和停止时间
        """
        index,trade_times = self.get_trade_datetimes(dt,allday=True)
        if trade_times:
            valid_dt = False
            for t1,t2 in trade_times:
                if t1 < dt and dt < t2:
                    valid_dt = True
                    break
            if valid_dt:
                start_dt = trade_times[0][0]
                stop_dt = trade_times[-1][1]
                return True,(start_dt,stop_dt)
        return False,(None,None)

    def get_day_start_stop(self,dt:datetime):
        """
        获得dt所在交易日日盘的开始和停止时间
        """
        index,trade_times = self.get_day_tradetimes(dt)
        if trade_times:
            valid_dt = False
            for t1,t2 in trade_times:
                if t1 < dt and dt < t2:
                    valid_dt = True
                    break
            if valid_dt:
                start_dt = trade_times[0][0]
                stop_dt = trade_times[-1][1]
                return True,(start_dt,stop_dt)
        return False,(None,None)

    def get_night_start_stop(self,dt:datetime):
        """
        获得dt所在交易日夜盘的开始和停止时间
        """
        index,trade_times = self.get_night_tradetimes(dt)
        if trade_times:
            valid_dt = False
            for t1,t2 in trade_times:
                if t1 < dt and dt < t2:
                    valid_dt = True
                    break
            if valid_dt:
                start_dt = trade_times[0][0]
                stop_dt = trade_times[-1][1]
                return True,(start_dt,stop_dt)
        return False,(None,None)

 # 以下为TradeHours类的测试代码
if __name__ == "__main__":
    rq.init('xxxxx','******',("rqdatad-pro.ricequant.com",16011))

    # vt_symbols = ["rb2010.SHFE","ag2012.SHFE","i2010.DCE"]
    vt_symbols = ["ag2012.SHFE"]
    date0 = datetime.date(2020,8,31)
    dt0 = CHINA_TZ.localize(datetime.datetime(2020,8,31,9,20,15))
    for vt_symbol in vt_symbols:
        symbol,exchange = extract_vt_symbol(vt_symbol)
        th = TradeHours(symbol)
        # trade_hours = th.get_date_tradetimes(date0)
        # print(f"\n{vt_symbol} {date0} trade_hours={trade_hours}")

        days,trade_hours = th.get_trade_datetimes(dt0,allday=True)

        print(f"\n{vt_symbol} {dt0} days:{days} trade_hours={trade_hours}")

        if trade_hours:
            day_start = trade_hours[0][0]
            day_end = trade_hours[-1][1]
            print(f"day_start={day_start} day_end={day_end}")
            exit_time = day_end + datetime.timedelta(minutes=-5)
            print(f"exit_time={exit_time}")

        dt1 = CHINA_TZ.localize(datetime.datetime(2020,8,31,9,20,15))
        dt2 = CHINA_TZ.localize(datetime.datetime(2020,9,1,1,1,15))

        for dt in [dt1,dt2]:
            in_trade,(start,stop) = th.get_date_start_stop(dt)
            if (in_trade):
                print(f"\n{vt_symbol} 时间 {dt} 交易日起止:{start,stop}")
            else:
                print(f"\n{vt_symbol} 时间 {dt} 非交易时间")

            in_day,(start,stop) = th.get_day_start_stop(dt)
            if (in_day):
                print(f"\n{vt_symbol} 时间 {dt} 日盘起止:{start,stop}")
            else:
                print(f"\n{vt_symbol} 时间 {dt} 非日盘时间")

            in_night,(start,stop) = th.get_night_start_stop(dt)
            if in_night:
                print(f"\n{vt_symbol} 时间 {dt} 夜盘起止:{start,stop}")
            else:
                print(f"\n{vt_symbol} 时间 {dt} 非夜盘时间")