如何在自定义zipline包中使用自定义日历？

Question

我的viacsv.py文件中有以下代码,旨在允许自定义捆绑包被摄取:

#
# Ingest stock csv files to create a zipline data bundle

import os

import numpy  as np
import pandas as pd
import datetime

boDebug=True # Set True to get trace messages

from zipline.utils.cli import maybe_show_progress

def viacsv(symbols,start=None,end=None):

    # strict this in memory so that we can reiterate over it.
    # (Because it could be a generator and they live only once)
    tuSymbols = tuple(symbols)

    if boDebug:
        print "entering viacsv.  tuSymbols=",tuSymbols

    # Define our custom ingest function
    def ingest(environ,
               asset_db_writer,
               minute_bar_writer,  # unused
               daily_bar_writer,
               adjustment_writer,
               calendar,
               cache,
               show_progress,
               output_dir,
               # pass these as defaults to make them 'nonlocal' in py2
               start=start,
               end=end):

        if boDebug:
            print "entering ingest and creating blank dfMetadata"

        dfMetadata = pd.DataFrame(np.empty(len(tuSymbols), dtype=[
            ('start_date', 'datetime64[ns]'),
            ('end_date', 'datetime64[ns]'),
            ('auto_close_date', 'datetime64[ns]'),
            ('symbol', 'object'),
        ]))

        if boDebug:
            print "dfMetadata",type(dfMetadata)
            print dfMetadata.describe
            print

        # We need to feed something that is iterable - like a list or a generator -
        # that is a tuple with an integer for sid and a DataFrame for the data to
        # daily_bar_writer

        liData=[]
        iSid=0
        for S in tuSymbols:
            IFIL="~/notebooks/csv/"+S+".csv"
            if boDebug:
               print "S=",S,"IFIL=",IFIL
            dfData=pd.read_csv(IFIL,index_col='Date',parse_dates=True).sort_index()
            if boDebug:
               print "read_csv dfData",type(dfData),"length",len(dfData)
               print
            dfData.rename(
                columns={
                    'Open': 'open',
                    'High': 'high',
                    'Low': 'low',
                    'Close': 'close',
                    'Volume': 'volume',
                    'Adj Close': 'price',
                },
                inplace=True,
            )
            dfData['volume']=dfData['volume']/1000
            liData.append((iSid,dfData))

            # the start date is the date of the first trade and
            start_date = dfData.index[0]
            if boDebug:
                print "start_date",type(start_date),start_date

            # the end date is the date of the last trade
            end_date = dfData.index[-1]
            if boDebug:
                print "end_date",type(end_date),end_date

            # The auto_close date is the day after the last trade.
            ac_date = end_date + pd.Timedelta(days=1)
            if boDebug:
                print "ac_date",type(ac_date),ac_date

            # Update our meta data
            dfMetadata.iloc[iSid] = start_date, end_date, ac_date, S

            iSid += 1

        if boDebug:
            print "liData",type(liData),"length",len(liData)
            print liData
            print
            print "Now calling daily_bar_writer"

        daily_bar_writer.write(liData, show_progress=False)

        # Hardcode the exchange to "YAHOO" for all assets and (elsewhere)
        # register "YAHOO" to resolve to the NYSE calendar, because these are
        # all equities and thus can use the NYSE calendar.
        dfMetadata['exchange'] = "YAHOO"

        if boDebug:
            print "returned from daily_bar_writer"
            print "calling asset_db_writer"
            print "dfMetadata",type(dfMetadata)
            print dfMetadata
            print

        # Not sure why symbol_map is needed
        symbol_map = pd.Series(dfMetadata.symbol.index, dfMetadata.symbol)
        if boDebug:
            print "symbol_map",type(symbol_map)
            print symbol_map
            print

        asset_db_writer.write(equities=dfMetadata)

        if boDebug:
            print "returned from asset_db_writer"
            print "calling adjustment_writer"

        adjustment_writer.write()

        if boDebug:
            print "returned from adjustment_writer"
            print "now leaving ingest function"

    if boDebug:
       print "about to return ingest function"
    return ingest

我的问题是我输入的数据不是美国数据,而是澳大利亚股票数据.因此,它遵守澳大利亚假期,而不是美国假期.似乎以下代码默认使用美国交易日历并告诉我,我无法传递美国市场意图关闭的数据,反之亦然.我如何调整上述代码以获取自定义日历？要摄取捆绑包,我在终端上运行以下命令:

zipline ingest -b CBA.csv

想法？

Answer 1

您需要在zipline/utils/calendars中定义自己的日历:只需创建一个现有文件(例如,exchange_calendar_nyse.py)并使用所需的假期进行编辑.假设您将此文件称为my_own_calendar.py和MyOwnCalendar类.

请注意,您还需要采取其他2个(或3个)步骤:

1)在zipline/util/calendars/calendar_utils.py中注册您的日历:您可以在_default_calendar_factories中添加条目,如果需要别名,则可以添加_default_calendar_aliases.例如,要将my_own_calendar.py映射到"OWN"并使用别名"MY_CALENDAR":

_default_calendar_factories = {
    'NYSE': NYSEExchangeCalendar,
    'CME': CMEExchangeCalendar,
    ...
    'OWN': MyOwnCalendar
}

_default_calendar_aliases = {
    'NASDAQ': 'NYSE',
    ...
    'MY_CALENDAR': 'OWN'
 }

2)您需要编辑.zipline/extension.py(您将在主目录中找到.zipline - 在Windows下查看您的主页,打开dos shell并键入echo%USERPROFILE%

# List the tickers of the market you defined
tickers_of_interest = {'TICKER1', 'TICKER2', ...}

register('my_market', viacsv(tickers_of_interest), calendar_name="OWN")

通过这些步骤,您只需输入zipline ingest -b my_market就可以摄取您的捆绑包.

3)我个人遇到的问题是,我需要更多地控制交易日历,因为超级TradingCalendar假设星期六/星期日是非交易日,而且每个市场/资产类别都不是这样.具有错误的日历定义将导致摄取时间异常.例如,要为24/24交易7/7的市场制作日历,我按如下方式攻击日历:

from datetime import time
from pytz import timezone
from pandas import date_range
from .trading_calendar import TradingCalendar, HolidayCalendar

from zipline.utils.memoize import lazyval

from pandas.tseries.offsets import CustomBusinessDay

class MyOwnCalendar(TradingCalendar):
    """
    Round the clock calendar: 7/7, 24/24
    """

    @property
    def name(self):
        return "OWN"

    @property
    def tz(self):
        return timezone("Europe/London")

    @property
    def open_time(self):
        return time(0)

    @property
    def close_time(self):
        return time(23, 59)

    @property
    def regular_holidays(self):
        return []

    @property
    def special_opens(self):
        return []

    def sessions_in_range(self, start_session, last_session):
        return date_range(start_session, last_session)

    @lazyval
    def day(self):
        return CustomBusinessDay(holidays=self.adhoc_holidays,
        calendar=self.regular_holidays,weekmask="Mon Tue Wed Thu Fri Sat Sun")