python - 如何在自定义 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

我的问题是我输入的数据不是美国数据，而是澳大利亚股票数据。因此，它遵守澳大利亚假期，而不是美国假期。不知何故，下面的代码似乎默认使用美国交易日历，并告诉我我无法在美国市场本应休市的日子里传递数据，反之亦然。我怎样才能调整上面的代码来接受自定义日历？为了获取 bundle ，我在终端运行以下命令:

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 来提取您的包。

1. 我个人遇到的问题是，鉴于父类(super class) TradingCalendar 假定星期六/星期日为非交易日，我需要对交易日历有更多的控制，但并非每个市场都如此/ Assets 类别。日历定义错误会导致摄取时出现异常。例如，要为 7/7 24/24 交易的市场设置日历，我按如下方式破解了日历:

   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")