Zipline 1.4.1 py 3.9

pylivetrader/algorithm.py

@@ -72,9 +72,8 @@
     require_initialized,
     disallowed_in_before_trading_start,
 )
-from pylivetrader.misc.pd_utils import normalize_date
-from pylivetrader.misc.preprocess import preprocess
-from pylivetrader.misc.input_validation import (
+from zipline.utils.preprocess import preprocess
+from zipline.utils.input_validation import (
     coerce_string,
     ensure_upper_case,
     expect_types,
@@ -876,7 +875,7 @@ def _can_order_asset(self, asset):
             )
 
         if asset.auto_close_date:
-            day = normalize_date(self.get_datetime())
+            day = self.get_datetime().normalize()
 
             if day > min(asset.end_date, asset.auto_close_date):
                 # If we are after the asset's end date or auto close date, warn

pylivetrader/backend/alpaca.py

@@ -46,7 +46,6 @@
     StopOrder,
     StopLimitOrder,
 )
-from pylivetrader.misc.pd_utils import normalize_date
 from pylivetrader.misc.parallel_utils import parallelize, \
     parallelize_with_multi_process
 from pylivetrader.errors import SymbolNotFound
@@ -186,7 +185,7 @@ def _symbols2assets(self, symbols):
 
     def get_equities(self):
         assets = []
-        t = normalize_date(pd.Timestamp('now', tz=NY))
+        t = pd.Timestamp('now', tz=NY).normalize()
         raw_assets = self._api.list_assets(asset_class='us_equity')
         for raw_asset in raw_assets:
 

pylivetrader/data/bardata.py

@@ -19,7 +19,6 @@
 from contextlib import contextmanager
 from collections import Iterable
 
-from pylivetrader.misc.pd_utils import normalize_date
 from pylivetrader.assets import Asset
 from pylivetrader.misc.parallel_utils import parallelize
 
@@ -123,21 +122,62 @@ def fetch(asset, field):
             )
 
     def history(self, assets, fields, bar_count, frequency):
+        """
+        Returns a trailing window of length ``bar_count`` with data for
+        the given assets, fields, and frequency, adjusted for splits, dividends,
+        and mergers as of the current simulation time.
+        The semantics for missing data are identical to the ones described in
+        the notes for :meth:`current`.
+        Parameters
+        ----------
+        assets: zipline.assets.Asset or iterable of zipline.assets.Asset
+            The asset(s) for which data is requested.
+        fields: string or iterable of string.
+            Requested data field(s). Valid field names are: "price",
+            "last_traded", "open", "high", "low", "close", and "volume".
+        bar_count: int
+            Number of data observations requested.
+        frequency: str
+            String indicating whether to load daily or minutely data
+            observations. Pass '1m' for minutely data, '1d' for daily data.
+        Returns
+        -------
+        history : pd.Series or pd.DataFrame or pd.Panel
+            See notes below.
+        Notes
+        -----
+        The return type of this function depends on the types of ``assets`` and
+        ``fields``:
+        - If a single asset and a single field are requested, the returned
+          value is a :class:`pd.Series` of length ``bar_count`` whose index is
+          :class:`pd.DatetimeIndex`.
+        - If a single asset and multiple fields are requested, the returned
+          value is a :class:`pd.DataFrame` with shape
+          ``(bar_count, len(fields))``. The frame's index will be a
+          :class:`pd.DatetimeIndex`, and its columns will be ``fields``.
+        - If multiple assets and a single field are requested, the returned
+          value is a :class:`pd.DataFrame` with shape
+          ``(bar_count, len(assets))``. The frame's index will be a
+          :class:`pd.DatetimeIndex`, and its columns will be ``assets``.
+        - If multiple assets and multiple fields are requested, the returned
+          value is a :class:`pd.DataFrame` with a pd.MultiIndex containing pairs of
+           :class:`pd.DatetimeIndex`, and ``assets``, while the columns while contain the field(s).
+           It has shape``(bar_count * len(assets), len(fields))``. The names of the pd.MultiIndex are
+            - ``date`` if frequency == '1d'`` or ``date_time`` if frequency == '1m``, and
+            - ``asset``
+        If the current simulation time is not a valid market time, we use the last market close instead.
+        """
 
-        if isinstance(assets, pandas.core.indexes.base.Index):
-            assets = list(assets)
-
-        if not (assets and fields):
-            return None
+        single_field = isinstance(fields, str)
 
-        if isinstance(fields, str):
-            single_asset = isinstance(assets, Asset)
+        single_asset = isinstance(assets, Asset)
 
-            if single_asset:
-                asset_list = [assets]
-            else:
-                asset_list = assets
+        if single_asset:
+            asset_list = [assets]
+        else:
+            asset_list = assets
 
+        if single_field:  # for one or more assets:
             df = self.data_portal.get_history_window(
                 asset_list,
                 self._get_current_minute(),
@@ -147,42 +187,60 @@ def history(self, assets, fields, bar_count, frequency):
                 self.data_frequency,
             )
 
-            if single_asset:
-                return df[assets]
-            else:
-                return df
-        else:
-            single_asset = isinstance(assets, Asset)
-
-            if single_asset:
-
-                df_dict = {
-                    field: self.data_portal.get_history_window(
-                        [assets],
-                        self._get_current_minute(),
-                        bar_count,
-                        frequency,
-                        field,
-                        self.data_frequency,
-                    )[assets] for field in fields
-                }
+            if self._adjust_minutes:
+                adjs = self.data_portal.get_adjustments(
+                    asset_list,
+                    fields,
+                    self._get_current_minute(),
+                    self.simulation_dt_func()
+                )
 
-                return pd.DataFrame(df_dict)
+                df = df * adjs
 
+            if single_asset:
+                # single asset, single field: return pd.Series with pd.DateTimeIndex
+                return df.loc[:, assets]
             else:
-
-                df_dict = {
-                    field: self.data_portal.get_history_window(
+                # multiple assets, single field: return DataFrame with pd.DateTimeIndex
+                # and assets in columns.
+                return df
+        else:  # multiple fields
+            # if single_asset:
+            # todo: optimize by querying multiple fields
+            # Make multiple history calls, one per field, then combine results
+
+            df_dict = {
+                field: self.data_portal.get_history_window(asset_list,
+                                                           self._get_current_minute(),
+                                                           bar_count,
+                                                           frequency,
+                                                           field,
+                                                           self.data_frequency,
+                                                           ).loc[:, asset_list]
+                for field in fields
+            }
+
+            if self._adjust_minutes:
+                adjs = {
+                    field: self.data_portal.get_adjustments(
                         assets,
-                        self._get_current_minute(),
-                        bar_count,
-                        frequency,
                         field,
-                        self.data_frequency,
-                    ) for field in fields
+                        self._get_current_minute(),
+                        self.simulation_dt_func()
+                    )[0] for field in fields
                 }
 
-                return pd.Panel(df_dict)
+                df_dict = {field: df * adjs[field]
+                           for field, df in df_dict.items()}
+
+            dt_label = 'date' if frequency == '1d' else 'date_time'
+            df = (pd.concat(df_dict,
+                            keys=df_dict.keys(),
+                            names=['fields', dt_label])
+                  .stack(dropna=False)  # ensure we return all fields/assets/dates despite missing values
+                  .unstack(level='fields'))
+            df.index.set_names([dt_label, 'asset'])
+            return df.sort_index()
 
     def can_trade(self, assets):
         """
@@ -237,6 +295,7 @@ def fetch(asset):
             tradeable = parallelize(fetch)(assets)
             return pd.Series(data=tradeable, index=assets, dtype=bool)
 
+
     @property
     def calendar(self):
         return self.data_portal.trading_calendar
@@ -334,7 +393,7 @@ def is_stale(self, assets):
             })
 
     def _is_stale_for_asset(self, asset, dt, adjusted_dt, data_portal):
-        session_label = normalize_date(dt)  # FIXME
+        session_label = dt.normalize()
 
         if not asset.is_alive_for_session(session_label):
             return False

pylivetrader/misc/events.py

@@ -24,7 +24,7 @@
 import pytz
 from toolz import curry
 
-from .preprocess import preprocess
+from zipline.utils.preprocess import preprocess
 from .memorize import lazyval
 from .sentinel import sentinel
 

pylivetrader/testing/fixtures.py

@@ -4,7 +4,6 @@
 from pylivetrader.data.data_portal import DataPortal
 from pylivetrader.assets import AssetFinder
 from pylivetrader.assets import Equity
-from pylivetrader.misc.pd_utils import normalize_date
 from pylivetrader.finance.order import Order as ZPOrder
 from trading_calendars import get_calendar
 
@@ -44,8 +43,8 @@ def get_asset(sym):
 class Backend:
 
     def __init__(self, start=None, end=None, assets=None, exchange='NYSE'):
-        self.start = normalize_date(pd.Timestamp(start or '2018-08-13'))
-        self.end = normalize_date(pd.Timestamp(end or '2018-08-14'))
+        self.start = pd.Timestamp(start or '2018-08-13').normalize()
+        self.end = pd.Timestamp(end or '2018-08-14').normalize()
 
         self._exchange = exchange
         self._calendar = get_calendar(exchange)

requirements/requirements.txt

@@ -1,7 +1,4 @@
-statsmodels==0.11.1  # pyup: ignore # limit to work properly with zipline 1.3.0
-scipy<1.6.0   # pyup: ignore - requires python >= 3.7
-numpy<=1.19.4
-pipeline-live>=0.1.12
+zipline-reloaded>=2
 bottleneck>=1.3
 pytz>=2020.1
 logbook>=1.5
@@ -11,5 +8,3 @@ click>=7,<8
 PyYAML>=5, <6
 ipython>=7
 alpaca-trade-api>=1.2.0
-pandas>=0.18.1, <=0.22.0  # pyup: ignore # limit to work properly with zipline 1.3.0
-pandas-datareader<=0.8.1  # pyup: ignore # higher requires pandas>=0.23, zipline limits to 0.22

tests/test_data/test_bardata.py

@@ -60,9 +60,6 @@ def test_bardata():
     assert len(o.index) == 1
     assert len(o.columns) == 2
 
-    o = data.history([asset0, asset1], ['open', 'close'], 1, 'minute')
-    assert type(o) == pd.Panel
-
     # can_trade
     data.datetime = pd.Timestamp('2018-08-13', tz='UTC')