wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-14 12:10:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1178 from quantopian/quantopian-quandl

Browse Source 
ENH: Adds quantopian-quandl bundle as new default.
This commit is contained in:
Joe Jevnik
2016-05-06 12:53:07 -04:00
parent f618cc94a2 120d60fe27
commit 0562179060
15 changed files with 808 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/source/beginner-tutorial.rst
+3 -3
View File
@@ -127,7 +127,7 @@ on OSX):
--capital-base FLOAT The starting capital for the simulation.
[default: 10000000.0]
-b, --bundle BUNDLE-NAME The data bundle to use for the simulation.
[default: quandl]
[default: quantopian-quandl]
--bundle-timestamp TIMESTAMP The date to lookup data on or before.
[default: <current-time>]
-s, --start DATE The start date of the simulation.
@@ -140,8 +140,8 @@ on OSX):
As you can see there are a couple of flags that specify where to find your
algorithm (``-f``) as well as parameters specifying which data to use,
defaulting to the :ref:`quandl-data-bundle`. There are also arguments for the
date range to run the algorithm over (``--start`` and ``--end``). Finally,
defaulting to the :ref:`quantopian-quandl-mirror`. There are also arguments for
the date range to run the algorithm over (``--start`` and ``--end``). Finally,
you'll want to save the performance metrics of your algorithm so that you can
analyze how it performed. This is done via the ``--output`` flag and will cause
it to write the performance ``DataFrame`` in the pickle Python file format.
docs/source/bundles.rst
+113 -52
View File
@@ -1,3 +1,5 @@
.. _data-bundles:
Data Bundles
------------
@@ -5,70 +7,107 @@ A data bundle is a collection of pricing data, adjustment data, and an asset
database. Bundles allow us to preload all of the data we will need to run
backtests and store the data for future runs.
Ingesting Data
~~~~~~~~~~~~~~
.. _bundles-command:
The first step to using a data bundle is to ingest the data. This will invoke
some custom bundle command and then write the data to a standard location that
zipline can find. By default this location is ``$ZIPLINE_ROOT/data/<bundle>``
where by default ``ZIPLINE_ROOT=~/.zipline``. This step may take some time as it
could involve downloading and processing a lot of data. This can be run with:
Discovering Available Bundles
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Zipline comes with a few bundles by default as well as the ability to register
new bundles. To see which bundles we have have available, we may run the
``bundles`` command, for example:
.. code-block:: bash
$ python -m zipline ingest <bundle>
$ zipline bundles
my-custom-bundle 2016-05-05 20:35:19.809398
my-custom-bundle 2016-05-05 20:34:53.654082
my-custom-bundle 2016-05-05 20:34:48.401767
quandl <no ingestions>
quantopian-quandl 2016-05-05 20:06:40.894956
The output here shows that there are 3 bundles available:
- ``my-custom-bundle`` (added by the user)
- ``quandl`` (provided by zipline)
- ``quantopian-quandl`` (provided by zipline)
The dates and times next to the name show the times when the data for this
bundle was ingested. We have run three different ingestions for
``my-custom-bundle``. We have never ingested any data for the ``quandl`` bundle
so it just shows ``<no ingestions>`` instead. Finally, there is only one
ingestion for ``quantopian-quandl``.
Ingesting Data
~~~~~~~~~~~~~~
The first step to using a data bundle is to ingest the data. The ingestion
process will invoke some custom bundle command and then write the data to a
standard location that zipline can find. By default the location where ingested
data will be written is ``$ZIPLINE_ROOT/data/<bundle>`` where by default
``ZIPLINE_ROOT=~/.zipline``. The ingestion step may take some time as it could
involve downloading and processing a lot of data. This can be run with:
.. code-block:: bash
$ zipline ingest [-b <bundle>]
where ``<bundle>`` is the name of the bundle to ingest.
where ``<bundle>`` is the name of the bundle to ingest, defaulting to
:ref:`quantopian-quandl <quantopian-quandl-mirror>`.
Old Data
~~~~~~~~
When the ``ingest`` command is used it will write the new data to a subdirectory
of ``$ZIPLINE_ROOT/data/<bundle>`` which is named with the current date. This
makes it possible to look at older data or even run backtests with this older
copy. This makers it easier to reproduce backtest results later.
makes it possible to look at older data or even run backtests with the older
copies. Running a backtest with an old ingestion makes it easier to reproduce
backtest results later.
One drawback of saving all of this data by default is that the data directory
may grow quite large even if you do not want to use the data. To solve this
problem there is another command ``clean`` which will clear data bundles based
on some time constraints.
One drawback of saving all of the data by default is that the data directory
may grow quite large even if you do not want to use the data. As shown earlier,
we can list all of the ingestions with the :ref:`bundles command
<bundles-command>`. To solve the problem of leaking old data there is another
command: ``clean``, which will clear data bundles based on some time
constraints.
For example:
.. code-block:: bash
# clean everything older than <date>
$ python -m zipline clean <bundle> --before <date>
$ zipline clean [-b <bundle>] --before <date>
# clean everything newer than <date>
$ python -m zipline clean <bundle> --after <date>
$ zipline clean [-b <bundle>] --after <date>
# keep everything in the range of [before, after] and delete the rest
$ python -m zipline clean <bundle> --before <date> --after <after>
$ zipline clean [-b <bundle>] --before <date> --after <after>
# clean all but the last <int> runs
$ python -m zipline clean <bundle> --keep-last <int>
$ zipline clean [-b <bundle>] --keep-last <int>
Running Backtests with Data Bundles
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Now that the data has been ingested we can use it to run backtests with the
``run`` command. This can be specified with the ``--bundle`` option like:
``run`` command. The bundle to use can be specified with the ``--bundle`` option
like:
.. code-block:: bash
$ python -m zipline run --bundle <bundle> --algofile algo.py ...
$ zipline run --bundle <bundle> --algofile algo.py ...
We may also specify the date to use to look up the bundle data with the
``--bundle-date`` option. This will cause us to the the most recent bundle
ingestion that is less than or equal to the ``bundle-date``. This is how we can
run backtests with older data. The reason that this uses a less than or equal to
relationship is that we can specify the date that we ran an old backtest and get
the same data that would have been available to us on that date. The
``bundle-date`` defaults to the current day to use the most recent data.
``--bundle-date`` option. Setting the ``--bundle-date`` will cause run to use
the most recent bundle ingestion that is less than or equal to the
``bundle-date``. This is how we can run backtests with older data. The reason
that ``-bundle-date`` uses a less than or equal to relationship is that we can
specify the date that we ran an old backtest and get the same data that would
have been available to us on that date. The ``bundle-date`` defaults to the
current day to use the most recent data.
Default Data Bundles
~~~~~~~~~~~~~~~~~~~~
@@ -80,21 +119,20 @@ Quandl WIKI Bundle
By default zipline comes with the ``quandl`` data bundle which uses quandl's
`WIKI dataset <https://www.quandl.com/data/WIKI>`_. The quandl data bundle
includes daily pricing data, splits, cash dividends, and asset metadata. This is
the bundle that ``run`` will use by default if no other bundle is specified. To
ingest this data bundle we recommend creating an account on quandl.com to get an
API key to be able to make more API requests per day. Once we have an API key we
may run:
includes daily pricing data, splits, cash dividends, and asset metadata. To
ingest the ``quandl`` data bundle we recommend creating an account on quandl.com
to get an API key to be able to make more API requests per day. Once we have an
API key we may run:
.. code-block:: bash
$ QUANDL_API_KEY=<api-key> python -m zipline ingest quandl
$ QUANDL_API_KEY=<api-key> zipline ingest quandl
though we may still run ``ingest`` as an anonymous quandl user (with no API
key). We may also set the ``QUANDL_DOWNLOAD_ATTEMPTS`` environment variable to
an integer which is the number of attempts that should be made to download data
from quandls servers. By default this will be 5, meaning that we will retry each
attempt 5 times.
from quandls servers. By default ``QUANDL_DOWNLOAD_ATTEMPTS`` will be 5, meaning
that we will retry each attempt 5 times.
.. note::
@@ -104,15 +142,25 @@ attempt 5 times.
equity.
.. _quantopian-quandl-mirror:
Quantopian Quandl WIKI Mirror
'''''''''''''''''''''''''''''
Quantopian provides a mirror of the quandl WIKI dataset with the data in the
formats that zipline expects. This is available under the name:
``quantopian-quandl`` and is the default bundle for zipline.
Yahoo Bundle Factories
``````````````````````
Zipline also ships with a factory function for creating a data bundle out of a
set of tickers from yahoo: :func:`~zipline.data.bundles.yahoo_equities`.
This makes it easy to pre-download and cache the data for a set of equities from
yahoo. This includes daily pricing data along with splits, cash dividends, and
inferred asset metadata. To create a bundle from a set of equities, add the
following to your ``~/.zipline/extensions.py`` file:
:func:`~zipline.data.bundles.yahoo_equities` makes it easy to pre-download and
cache the data for a set of equities from yahoo. The yahoo bundles include daily
pricing data along with splits, cash dividends, and inferred asset metadata. To
create a bundle from a set of equities, add the following to your
``~/.zipline/extensions.py`` file:
.. code-block:: python
@@ -134,8 +182,8 @@ This may now be used like:
.. code-block:: bash
$ python -m zipline ingest my-yahoo-equities-bundle
$ python -m zipline run -f algo.py --bundle my-yahoo-equities-bundle
$ zipline ingest my-yahoo-equities-bundle
$ zipline run -f algo.py --bundle my-yahoo-equities-bundle
More than one yahoo equities bundle may be registered as long as they use
@@ -145,15 +193,16 @@ Writing a New Bundle
~~~~~~~~~~~~~~~~~~~~
Data bundles exist to make it easy to use different data sources with
zipline. To add a new bundle, one must implement an ingest function.
zipline. To add a new bundle, one must implement an ``ingest`` function.
This function is responsible for loading the data into memory and passing it to
a set of writer objects provided by zipline to convert the data to zipline's
internal format. The ingest function may work by downloading data from a remote
location like the ``quandl`` bundle or yahoo bundles or it may just load files
that are already on the machine. The function is provided with writers that will
write the data to the correct location transactionally. If an ingestion fails
part way through the bundle will not be written in an incomplete state.
The ``ingest`` function is responsible for loading the data into memory and
passing it to a set of writer objects provided by zipline to convert the data to
zipline's internal format. The ingest function may work by downloading data from
a remote location like the ``quandl`` bundle or yahoo bundles or it may just
load files that are already on the machine. The function is provided with
writers that will write the data to the correct location transactionally. If an
ingestion fails part way through the bundle will not be written in an incomplete
state.
The signature of the ingest function should be:
@@ -166,7 +215,8 @@ The signature of the ingest function should be:
adjustment_writer,
calendar,
cache,
show_progress)
show_progress,
output_dir)
``environ``
```````````
@@ -249,8 +299,8 @@ have.
````````````
``calendar`` is a ``pandas.DatetimeIndex`` object holding all of the trading
days that the bundle should load data for. This is to help some bundles generate
queries for the days needed.
days that the bundle should load data for. The calendar is provided to help some
bundles generate queries for the days needed.
``cache``
`````````
@@ -275,3 +325,14 @@ the total needed, or how far into some data conversion the ingest function
is. One tool that may help with implementing ``show_progress`` for a loop is
:class:`~zipline.utils.cli.maybe_show_progress`. This argument should always be
forwarded to ``minute_bar_writer.write`` and ``daily_bar_writer.write``.
``output_dir``
``````````````
``output_dir`` is a string representing the file path where all the data will be
written. ``output_dir`` will be some subdirectory of ``$ZIPLINE_ROOT`` and will
contain the time of the start of the current ingestion. This can be used to
directly move resources here if for some reason your ingest function can produce
it's own outputs without the writers. For example, the ``quantopian:quandl``
bundle uses this to directly untar the bundle into the ``output_dir``.
docs/source/whatsnew/1.0.0.txt
+12 -11
View File
@@ -12,27 +12,28 @@ Development
Highlights
~~~~~~~~~~
New Entry Points (:issue:`xxxx`)
````````````````````````````````
New Entry Points (:issue:`1173` and :issue:`1178`)
``````````````````````````````````````````````````
In order to make it easier to use zipline we have updated the entry points for
a backtest. The three supported ways to run a backtest are now:
1. :func:`zipline.run_algo`
2. ``$ python -m zipline run``
2. ``$ zipline run``
3. ``%zipline`` (IPython magic)
Data Bundles (:issue:`xxxx`)
````````````````````````````
Data Bundles (:issue:`1173` and :issue:`1178`)
``````````````````````````````````````````````
1.0.0 introduces data bundles. Data bundles are groups of data that should be
preloaded and used to run backtests later. This allows users to not need to to
specify which tickers they are interested in each time they run an
algorithm. This also allows us to cache the data between runs.
By default, the ``quandl`` bundle will be used which pulls data from quandl's
`WIKI dataset <https://www.quandl.com/data/WIKI>`_. New bundles may be
registered with :func:`zipline.data.bundles.register` like:
By default, the ``quantopian-quandl`` bundle will be used which pulls data from
Quantopian's mirror of the quandl `WIKI dataset
<https://www.quandl.com/data/WIKI>`_. New bundles may be registered with
:func:`zipline.data.bundles.register` like:
.. code-block:: python
@@ -53,10 +54,10 @@ have been passed to write that data to disc in a location that zipline can find
later.
This data can be used in backtests by passing the name as the ``-b / --bundle``
argument to ``$ python -m zipline run`` or as the ``bundle`` argument to
:func:`zipline.run_algo`.
argument to ``$ zipline run`` or as the ``bundle`` argument to
:func:`zipline.run_algorithm`.
For more information see `Data Bundles`_ for more information.
For more information see :ref:`data-bundles` for more information.
String Support in Pipeline (:issue:`1174`)
``````````````````````````````````````````
setup.py
+5
View File
@@ -262,6 +262,11 @@ setup(
version=versioneer.get_version(),
cmdclass=LazyBuildExtCommandClass(versioneer.get_cmdclass()),
description='A backtester for financial algorithms.',
entry_points={
'console_scripts': [
'zipline = zipline.__main__:main',
],
},
author='Quantopian Inc.',
author_email='opensource@quantopian.com',
packages=find_packages('.', include=['zipline', 'zipline.*']),
tests/data/bundles/test_core.py
+246 -13
View File
@@ -1,9 +1,12 @@
import os
from nose_parameterized import parameterized
import pandas as pd
from toolz import valmap
import toolz.curried.operator as op
from zipline.assets.synthetic import make_simple_equity_info
from zipline.data.bundles import load
from zipline.data.bundles import UnknownBundle, from_bundle_ingest_dirname
from zipline.data.bundles.core import _make_bundle_core
from zipline.lib.adjustment import Float64Multiply
from zipline.pipeline.loaders.synthetic import (
@@ -12,30 +15,39 @@ from zipline.pipeline.loaders.synthetic import (
)
from zipline.testing import (
subtest,
tmp_dir,
str_to_seconds,
tmp_trading_env,
)
from zipline.testing.fixtures import ZiplineTestCase
from zipline.testing.fixtures import WithInstanceTmpDir, ZiplineTestCase
from zipline.testing.predicates import (
assert_equal,
assert_false,
assert_in,
assert_is,
assert_is_instance,
assert_is_none,
assert_raises,
assert_true,
)
from zipline.utils.cache import dataframe_cache
from zipline.utils.functional import apply
from zipline.utils.tradingcalendar import trading_days
import zipline.utils.paths as pth
class BundleCoreTestCase(ZiplineTestCase):
_1_ns = pd.Timedelta(1, unit='ns')
class BundleCoreTestCase(WithInstanceTmpDir, ZiplineTestCase):
def init_instance_fixtures(self):
super(BundleCoreTestCase, self).init_instance_fixtures()
(self.bundles,
self.register,
self.unregister,
self.ingest) = _make_bundle_core()
self.ingest,
self.load,
self.clean) = _make_bundle_core()
self.environ = {'ZIPLINE_ROOT': self.instance_tmpdir.path}
def test_register_decorator(self):
@apply
@@ -75,17 +87,35 @@ class BundleCoreTestCase(ZiplineTestCase):
assert_false(self.bundles)
def test_register_no_create(self):
called = [False]
@self.register('bundle', create_writers=False)
def bundle_ingest(environ,
asset_db_writer,
minute_bar_writer,
daily_bar_writer,
adjustment_writer,
calendar,
cache,
show_progress,
output_dir):
assert_is_none(asset_db_writer)
assert_is_none(minute_bar_writer)
assert_is_none(daily_bar_writer)
assert_is_none(adjustment_writer)
called[0] = True
self.ingest('bundle', self.environ)
assert_true(called[0])
def test_ingest(self):
zipline_root = self.enter_instance_context(tmp_dir()).path
env = self.enter_instance_context(tmp_trading_env())
start = pd.Timestamp('2014-01-06', tz='utc')
end = pd.Timestamp('2014-01-10', tz='utc')
calendar = trading_days[trading_days.slice_indexer(start, end)]
minutes = env.minutes_for_days_in_range(calendar[0], calendar[-1])
outer_environ = {
'ZIPLINE_ROOT': zipline_root,
}
sids = tuple(range(3))
equities = make_simple_equity_info(
@@ -122,8 +152,9 @@ class BundleCoreTestCase(ZiplineTestCase):
adjustment_writer,
calendar,
cache,
show_progress):
assert_is(environ, outer_environ)
show_progress,
output_dir):
assert_is(environ, self.environ)
asset_db_writer.write(equities=equities)
minute_bar_writer.write(minute_bar_data)
@@ -134,8 +165,8 @@ class BundleCoreTestCase(ZiplineTestCase):
assert_is_instance(cache, dataframe_cache)
assert_is_instance(show_progress, bool)
self.ingest('bundle', environ=outer_environ)
bundle = load('bundle', environ=outer_environ)
self.ingest('bundle', environ=self.environ)
bundle = self.load('bundle', environ=self.environ)
assert_equal(set(bundle.asset_finder.sids), set(sids))
@@ -216,3 +247,205 @@ class BundleCoreTestCase(ZiplineTestCase):
},
msg='volume',
)
@parameterized.expand([('clean',), ('load',)])
def test_bundle_doesnt_exist(self, fnname):
with assert_raises(UnknownBundle) as e:
getattr(self, fnname)('ayy', environ=self.environ)
assert_equal(e.exception.name, 'ayy')
def test_load_no_data(self):
# register but do not ingest data
self.register('bundle', lambda *args: None)
ts = pd.Timestamp('2014')
with assert_raises(ValueError) as e:
self.load('bundle', timestamp=ts, environ=self.environ)
assert_in(
"no data for bundle 'bundle' on or before %s" % ts,
str(e.exception),
)
def _list_bundle(self):
return {
os.path.join(pth.data_path(['bundle', d], environ=self.environ))
for d in os.listdir(
pth.data_path(['bundle'], environ=self.environ),
)
}
def _empty_ingest(self, _wrote_to=[]):
"""Run the nth empty ingest.
Returns
-------
wrote_to : str
The timestr of the bundle written.
"""
if not self.bundles:
@self.register('bundle',
calendar=pd.DatetimeIndex([pd.Timestamp('2014')]))
def _(environ,
asset_db_writer,
minute_bar_writer,
daily_bar_writer,
adjustment_writer,
calendar,
cache,
show_progress,
output_dir):
_wrote_to.append(output_dir)
_wrote_to.clear()
self.ingest('bundle', environ=self.environ)
assert_equal(len(_wrote_to), 1, msg='ingest was called more than once')
ingestions = self._list_bundle()
assert_in(
_wrote_to[0],
ingestions,
msg='output_dir was not in the bundle directory',
)
return _wrote_to[0]
def test_clean_keep_last(self):
first = self._empty_ingest()
assert_equal(
self.clean('bundle', keep_last=1, environ=self.environ),
set(),
)
assert_equal(
self._list_bundle(),
{first},
msg='directory should not have changed',
)
second = self._empty_ingest()
assert_equal(
self._list_bundle(),
{first, second},
msg='two ingestions are not present',
)
assert_equal(
self.clean('bundle', keep_last=1, environ=self.environ),
{first},
)
assert_equal(
self._list_bundle(),
{second},
msg='first ingestion was not removed with keep_last=2',
)
third = self._empty_ingest()
fourth = self._empty_ingest()
fifth = self._empty_ingest()
assert_equal(
self._list_bundle(),
{second, third, fourth, fifth},
msg='larger set of ingestions did not happen correctly',
)
assert_equal(
self.clean('bundle', keep_last=2, environ=self.environ),
{second, third},
)
assert_equal(
self._list_bundle(),
{fourth, fifth},
msg='keep_last=2 did not remove the correct number of ingestions',
)
@staticmethod
def _ts_of_run(run):
return from_bundle_ingest_dirname(run.rsplit(os.path.sep, 1)[-1])
def test_clean_before_after(self):
first = self._empty_ingest()
assert_equal(
self.clean(
'bundle',
before=self._ts_of_run(first),
environ=self.environ,
),
set(),
)
assert_equal(
self._list_bundle(),
{first},
msg='directory should not have changed (before)',
)
assert_equal(
self.clean(
'bundle',
after=self._ts_of_run(first),
environ=self.environ,
),
set(),
)
assert_equal(
self._list_bundle(),
{first},
msg='directory should not have changed (after)',
)
assert_equal(
self.clean(
'bundle',
before=self._ts_of_run(first) + _1_ns,
environ=self.environ,
),
{first},
)
assert_equal(
self._list_bundle(),
set(),
msg='directory now be empty (before)',
)
second = self._empty_ingest()
assert_equal(
self.clean(
'bundle',
after=self._ts_of_run(second) - _1_ns,
environ=self.environ,
),
{second},
)
assert_equal(
self._list_bundle(),
set(),
msg='directory now be empty (after)',
)
third = self._empty_ingest()
fourth = self._empty_ingest()
fifth = self._empty_ingest()
sixth = self._empty_ingest()
assert_equal(
self._list_bundle(),
{third, fourth, fifth, sixth},
msg='larger set of ingestions did no happen correctly',
)
assert_equal(
self.clean(
'bundle',
before=self._ts_of_run(fourth),
after=self._ts_of_run(fifth),
environ=self.environ,
),
{third, sixth},
)
assert_equal(
self._list_bundle(),
{fourth, fifth},
msg='did not strip first and last directories',
)
tests/resources/example_data.tar.gz
BIN
View File
Binary file not shown.
tests/test_examples.py
+5 -3
View File
@@ -12,9 +12,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# This code is based on a unittest written by John Salvatier:
# https://github.com/pymc-devs/pymc/blob/pymc3/tests/test_examples.py
from functools import partial
import tarfile
import matplotlib
@@ -22,6 +20,7 @@ from nose_parameterized import parameterized
import pandas as pd
from zipline import examples, run_algorithm
from zipline.data.bundles import register, unregister
from zipline.testing import test_resource_path
from zipline.testing.fixtures import WithTmpDir, ZiplineTestCase
from zipline.testing.predicates import assert_equal
@@ -76,6 +75,9 @@ class ExamplesTests(WithTmpDir, ZiplineTestCase):
def init_class_fixtures(cls):
super(ExamplesTests, cls).init_class_fixtures()
register('test', lambda *args: None)
cls.add_class_callback(partial(unregister, 'test'))
with tarfile.open(test_resource_path('example_data.tar.gz')) as tar:
tar.extractall(cls.tmpdir.path)
zipline/__main__.py
+59 -17
View File
@@ -1,4 +1,4 @@
import datetime
import errno
import os
from functools import wraps
@@ -6,8 +6,9 @@ import click
import logbook
import pandas as pd
from zipline.data import bundles
from zipline.data import bundles as bundles_module
from zipline.utils.cli import Date, Timestamp
import zipline.utils.paths as pth
from zipline.utils.run_algo import _run, load_extensions
try:
@@ -37,7 +38,7 @@ except NameError:
default=True,
help="Don't load the default zipline extension.py file in $ZIPLINE_HOME.",
)
def cli(extension, strict_extensions, default_extension):
def main(extension, strict_extensions, default_extension):
"""Top level zipline entry point.
"""
# install a logbook handler before performing any other operations
@@ -98,7 +99,7 @@ def ipython_only(option):
return d
@cli.command()
@main.command()
@click.option(
'-f',
'--algofile',
@@ -137,7 +138,7 @@ def ipython_only(option):
@click.option(
'-b',
'--bundle',
default='quandl',
default='quantopian-quandl',
metavar='BUNDLE-NAME',
show_default=True,
help='The data bundle to use for the simulation.',
@@ -274,27 +275,41 @@ def zipline_magic(line, cell=None):
raise ValueError('main returned non-zero status code: %d' % e.code)
@cli.command()
@click.argument('BUNDLE-NAME')
@main.command()
@click.option(
'-b',
'--bundle',
default='quantopian-quandl',
metavar='BUNDLE-NAME',
show_default=True,
help='The data bundle to ingest.',
)
@click.option(
'--show-progress/--no-show-progress',
is_flag=True,
default=True,
help='Print progress information to the terminal.'
)
def ingest(bundle_name, show_progress):
def ingest(bundle, show_progress):
"""Ingest the data for the given bundle.
"""
bundles.ingest(
bundle_name,
bundles_module.ingest(
bundle,
os.environ,
datetime.date.today(),
pd.Timestamp.utcnow(),
show_progress,
)
@cli.command()
@click.argument('BUNDLE-NAME')
@main.command()
@click.option(
'-b',
'--bundle',
default='quantopian-quandl',
metavar='BUNDLE-NAME',
show_default=True,
help='The data bundle to clean.',
)
@click.option(
'-b',
'--before',
@@ -317,16 +332,43 @@ def ingest(bundle_name, show_progress):
help='Clear all but the last N downloads.'
' This may not be passed with -b / --before or -a / --after',
)
def clean(bundle_name, before, after, keep_last):
def clean(bundle, before, after, keep_last):
"""Clean up data downloaded with the ingest command.
"""
bundles.clean(
bundle_name,
bundles_module.clean(
bundle,
before,
after,
keep_last,
)
@main.command()
def bundles():
"""List all of the available data bundles.
"""
for bundle in sorted(bundles_module.bundles.keys()):
try:
ingestions = sorted(
(str(bundles_module.from_bundle_ingest_dirname(ing))
for ing in os.listdir(pth.data_path([bundle]))
if not pth.hidden(ing)),
reverse=True,
)
except IOError as e:
if e.errno != errno.ENOENT:
raise
ingestions = []
print(
'\n'.join(
'%s %s' % (bundle, line)
for line in (
ingestions if ingestions else ('<no ingestions>',)
)
),
)
if __name__ == '__main__':
cli()
main()
zipline/assets/assets.py
+1 -1
View File
@@ -101,7 +101,7 @@ class AssetFinder(object):
PERSISTENT_TOKEN = "<AssetFinder>"
def __init__(self, engine):
if isinstance(engine, str):
if isinstance(engine, string_types):
engine = sa.create_engine('sqlite:///' + engine)
self.engine = engine
zipline/data/bundles/__init__.py
+7
View File
@@ -1,20 +1,27 @@
from . import quandl # noqa
from .core import (
UnknownBundle,
bundles,
clean,
from_bundle_ingest_dirname,
ingest,
load,
register,
to_bundle_ingest_dirname,
unregister,
)
from .yahoo import yahoo_equities
__all__ = [
'UnknownBundle',
'bundles',
'clean',
'from_bundle_ingest_dirname',
'ingest',
'load',
'register',
'to_bundle_ingest_dirname',
'unregister',
'yahoo_equities',
]
zipline/data/bundles/core.py
+282 -216
View File
@@ -4,9 +4,10 @@ import os
import shutil
import warnings
from contextlib2 import ExitStack
import click
import pandas as pd
from toolz import curry, complement, compose
from toolz import curry, complement
from ..us_equity_pricing import (
BcolzDailyBarReader,
@@ -59,16 +60,59 @@ def adjustment_db_path(bundle_name, timestr, environ=None):
)
def cache_path(bundle_name, timestr, environ=None):
def cache_path(bundle_name, environ=None):
return pth.data_path(
[bundle_name, timestr, '.cache'],
[bundle_name, '.cache'],
environ=environ,
)
def to_bundle_ingest_dirname(ts):
"""Convert a pandas Timestamp into the name of the directory for the
ingestion.
Parameters
----------
ts : pandas.Timestamp
The time of the ingestions
Returns
-------
name : str
The name of the directory for this ingestion.
"""
return ts.isoformat().replace(':', ';')
def from_bundle_ingest_dirname(cs):
"""Read a bundle ingestion directory name into a pandas Timestamp.
Parameters
----------
cs : str
The name of the directory.
Returns
-------
ts : pandas.Timestamp
The time when this ingestion happened.
"""
return pd.Timestamp(cs.replace(';', ':'))
_BundlePayload = namedtuple(
'_BundlePayload',
'calendar opens closes minutes_per_day ingest',
'calendar opens closes minutes_per_day ingest create_writers',
)
BundleData = namedtuple(
'BundleData',
'asset_finder minute_bar_reader daily_bar_reader adjustment_reader',
)
BundleCore = namedtuple(
'BundleCore',
'bundles register unregister ingest load clean',
)
@@ -87,6 +131,33 @@ class UnknownBundle(click.ClickException, LookupError):
return self.message
class BadClean(click.ClickException, ValueError):
"""Exception indicating that an invalid argument set was passed to
``clean``.
Parameters
----------
before, after, keep_last : any
The bad arguments to ``clean``.
See Also
--------
clean
"""
def __init__(self, before, after, keep_last):
super(BadClean, self).__init__(
'Cannot pass a combination of `before` and `after` with'
'`keep_last`. Got: before=%r, after=%r, keep_n=%r\n' % (
before,
after,
keep_last,
),
)
def __str__(self):
return self.message
def _make_bundle_core():
"""Create a family of data bundle functions that read from the same
bundle mapping.
@@ -99,8 +170,12 @@ def _make_bundle_core():
The function which registers new bundles in the ``bundles`` mapping.
unregister : callable
The function which deregisters bundles from the ``bundles`` mapping.
ingest_bundle : callable
ingest : callable
The function which downloads and write data for a given data bundle.
load : callable
The function which loads the ingested bundles back into memory.
clean : callable
The function which cleans up data written with ``ingest``.
"""
_bundles = {} # the registered bundles
# Expose _bundles through a proxy so that users cannot mutate this
@@ -114,7 +189,8 @@ def _make_bundle_core():
calendar=trading_days,
opens=open_and_closes['market_open'],
closes=open_and_closes['market_close'],
minutes_per_day=390):
minutes_per_day=390,
create_writers=True):
"""Register a data bundle ingest function.
Parameters
@@ -154,6 +230,10 @@ def _make_bundle_core():
NYSE calendar.
minutes_per_day : int, optional
The number of minutes in each normal trading day.
create_writers : bool, optional
Should the ingest machinery create the writers for the ingest
function. This can be disabled as an optimization for cases where
they are not needed, like the ``quantopian-quandl`` bundle.
Notes
-----
@@ -180,6 +260,7 @@ def _make_bundle_core():
closes,
minutes_per_day,
f,
create_writers,
)
return f
@@ -208,7 +289,7 @@ def _make_bundle_core():
def ingest(name,
environ=os.environ,
timestamp=None,
show_progress=True):
show_progress=False):
"""Ingest data for a given bundle.
Parameters
@@ -231,239 +312,224 @@ def _make_bundle_core():
if timestamp is None:
timestamp = pd.Timestamp.utcnow()
timestamp = timestamp.tz_convert('utc').tz_localize(None)
timestr = str(timestamp.value)
cachepath = cache_path(name, timestr, environ=environ)
timestr = to_bundle_ingest_dirname(timestamp)
cachepath = cache_path(name, environ=environ)
pth.ensure_directory(pth.data_path([name, timestr], environ=environ))
pth.ensure_directory(cachepath)
with dataframe_cache(cachepath, clean_on_failure=False) as cache, \
working_dir(
daily_equity_path(name, timestr, environ=environ),
) as daily_bars_dir, \
working_dir(
minute_equity_path(name, timestr, environ=environ),
) as minute_bars_dir, \
working_file(
asset_db_path(name, timestr, environ=environ),
) as asset_db_file, \
working_file(
adjustment_db_path(name, timestr, environ=environ),
) as adjustment_db_file:
ExitStack() as stack:
# we use `cleanup_on_failure=False` so that we don't purge the
# cache directory if the load fails in the middle
daily_bar_writer = BcolzDailyBarWriter(
daily_bars_dir.name,
bundle.calendar,
)
# Do an empty write to ensure that the daily ctables exist
# when we create the SQLiteAdjustmentWriter below. The
# SQLiteAdjustmentWriter needs to open the daily ctables so that
# it can compute the adjustment ratios for the dividends.
daily_bar_writer.write(())
bundle.ingest(
environ,
AssetDBWriter(asset_db_file.name),
BcolzMinuteBarWriter(
if bundle.create_writers:
daily_bars_path = stack.enter_context(working_dir(
daily_equity_path(name, timestr, environ=environ),
)).path
daily_bar_writer = BcolzDailyBarWriter(
daily_bars_path,
bundle.calendar,
)
# Do an empty write to ensure that the daily ctables exist
# when we create the SQLiteAdjustmentWriter below. The
# SQLiteAdjustmentWriter needs to open the daily ctables so
# that it can compute the adjustment ratios for the dividends.
daily_bar_writer.write(())
minute_bar_writer = BcolzMinuteBarWriter(
bundle.calendar[0],
minute_bars_dir.name,
stack.enter_context(working_dir(
minute_equity_path(name, timestr, environ=environ),
)).path,
bundle.opens,
bundle.closes,
minutes_per_day=bundle.minutes_per_day,
),
daily_bar_writer,
SQLiteAdjustmentWriter(
adjustment_db_file.name,
BcolzDailyBarReader(daily_bars_dir.name),
)
asset_db_writer = AssetDBWriter(
stack.enter_context(working_file(
asset_db_path(name, timestr, environ=environ),
)).path,
)
adjustment_db_writer = SQLiteAdjustmentWriter(
stack.enter_context(working_file(
adjustment_db_path(name, timestr, environ=environ),
)).path,
BcolzDailyBarReader(daily_bars_path),
bundle.calendar,
overwrite=True,
),
)
else:
daily_bar_writer = None
minute_bar_writer = None
asset_db_writer = None
adjustment_db_writer = None
bundle.ingest(
environ,
asset_db_writer,
minute_bar_writer,
daily_bar_writer,
adjustment_db_writer,
bundle.calendar,
cache,
show_progress,
pth.data_path([name, timestr], environ=environ),
)
return bundles, register, unregister, ingest
def most_recent_data(bundle_name, timestamp, environ=None):
"""Get the path to the most recent data after ``date``for the
given bundle.
Parameters
----------
bundle_name : str
The name of the bundle to lookup.
timestamp : datetime
The timestamp to begin searching on or before.
environ : dict, optional
An environment dict to forward to zipline_root.
"""
if bundle_name not in bundles:
raise UnknownBundle(bundle_name)
bundles, register, unregister, ingest = _make_bundle_core()
try:
candidates = os.listdir(
pth.data_path([bundle_name], environ=environ),
)
return pth.data_path(
[bundle_name,
max(
filter(complement(pth.hidden), candidates),
key=from_bundle_ingest_dirname,
)],
environ=environ,
)
except (ValueError, OSError) as e:
if getattr(e, 'errno', ~errno.ENOENT) != errno.ENOENT:
raise
raise ValueError(
'no data for bundle %r on or before %s' % (
bundle_name,
timestamp,
),
)
BundleData = namedtuple(
'BundleData',
'asset_finder minute_bar_reader daily_bar_reader adjustment_reader',
)
def load(name, environ=os.environ, timestamp=None):
"""Loads a previously ingested bundle.
Parameters
----------
name : str
The name of the bundle.
environ : mapping, optional
The environment variables. Defaults of os.environ.
timestamp : datetime, optional
The timestamp of the data to lookup.
Defaults to the current time.
def most_recent_data(bundle_name, timestamp, environ=None):
"""Get the path to the most recent data after ``date``for the given bundle.
Parameters
----------
bundle_name : str
The name of the bundle to lookup.
timestamp : datetime
The timestamp to begin searching on or before.
environ : dict, optional
An environment dict to forward to zipline_root.
"""
try:
candidates = os.listdir(pth.data_path([bundle_name], environ=environ))
return pth.data_path(
[bundle_name,
max(
filter(complement(pth.hidden), candidates),
key=compose(pd.Timestamp, int),
)],
environ=environ,
)
except ValueError:
raise ValueError(
'no data for bundle %r on or before %s' % (
bundle_name,
timestamp,
Returns
-------
bundle_data : BundleData
The raw data readers for this bundle.
"""
if timestamp is None:
timestamp = pd.Timestamp.utcnow()
timestr = most_recent_data(name, timestamp, environ=environ)
return BundleData(
asset_finder=AssetFinder(
asset_db_path(name, timestr, environ=environ),
),
minute_bar_reader=BcolzMinuteBarReader(
minute_equity_path(name, timestr, environ=environ),
),
daily_bar_reader=BcolzDailyBarReader(
daily_equity_path(name, timestr, environ=environ),
),
adjustment_reader=SQLiteAdjustmentReader(
adjustment_db_path(name, timestr, environ=environ),
),
)
except OSError as e:
if e.errno != errno.ENOENT:
raise
raise UnknownBundle(bundle_name)
def load(name, environ=os.environ, timestamp=None):
"""Loads a previously ingested bundle.
Parameters
----------
name : str
The name of the bundle.
environ : mapping, optional
The environment variables. Defaults of os.environ.
timestamp : datetime, optional
The timestamp of the data to lookup.
Defaults to the current time.
Returns
-------
bundle_data : BundleData
The raw data readers for this bundle.
"""
if timestamp is None:
timestamp = pd.Timestamp.utcnow()
timestr = most_recent_data(name, timestamp, environ=environ)
return BundleData(
asset_finder=AssetFinder(
asset_db_path(name, timestr, environ=environ),
),
minute_bar_reader=BcolzMinuteBarReader(
minute_equity_path(name, timestr, environ=environ),
),
daily_bar_reader=BcolzDailyBarReader(
daily_equity_path(name, timestr, environ=environ),
),
adjustment_reader=SQLiteAdjustmentReader(
adjustment_db_path(name, timestr, environ=environ),
),
@preprocess(
before=optionally(ensure_timestamp),
after=optionally(ensure_timestamp),
)
def clean(name,
before=None,
after=None,
keep_last=None,
environ=os.environ):
"""Clean up data that was created with ``ingest`` or
``$ python -m zipline ingest``
Parameters
----------
name : str
The name of the bundle to remove data for.
before : datetime, optional
Remove data ingested before this date.
This argument is mutually exclusive with: keep_last
after : datetime, optional
Remove data ingested after this date.
This argument is mutually exclusive with: keep_last
keep_last : int, optional
Remove all but the last ``keep_last`` ingestions.
This argument is mutually exclusive with:
before
after
environ : mapping, optional
The environment variables. Defaults of os.environ.
Returns
-------
cleaned : set[str]
The names of the runs that were removed.
Raises
------
BadClean
Raised when ``before`` and or ``after`` are passed with
``keep_last``. This is a subclass of ``ValueError``.
"""
try:
all_runs = sorted(
filter(
complement(pth.hidden),
os.listdir(pth.data_path([name], environ=environ)),
),
key=from_bundle_ingest_dirname,
)
except OSError as e:
if e.errno != errno.ENOENT:
raise
raise UnknownBundle(name)
if ((before is not None or after is not None) and
keep_last is not None):
raise BadClean(before, after, keep_last)
if keep_last is None:
def should_clean(name):
dt = from_bundle_ingest_dirname(name)
return (
(before is not None and dt < before) or
(after is not None and dt > after)
)
else:
last_n_dts = set(all_runs[-keep_last:])
def should_clean(name):
return name not in last_n_dts
cleaned = set()
for run in all_runs:
if should_clean(run):
path = pth.data_path([name, run], environ=environ)
shutil.rmtree(path)
cleaned.add(path)
return cleaned
return BundleCore(bundles, register, unregister, ingest, load, clean)
class BadClean(click.ClickException, ValueError):
"""Exception indicating that an invalid argument set was passed to
``clean``.
Parameters
----------
before, after, keep_last : any
The bad arguments to ``clean``.
See Also
--------
clean
"""
def __init__(self, before, after, keep_last):
super(BadClean, self).__init__(
'Cannot pass a combination of `before` and `after` with'
'`keep_last`. Got: before=%r, after=%r, keep_n=%r\n' % (
before,
after,
keep_last,
),
)
def __str__(self):
return self.message
@preprocess(
before=optionally(ensure_timestamp),
after=optionally(ensure_timestamp),
)
def clean(name, before=None, after=None, keep_last=None, environ=os.environ):
"""Clean up data that was created with ``ingest`` or
``$ python -m zipline ingest``
Parameters
----------
name : str
The name of the bundle to remove data for.
before : datetime, optional
Remove data ingested before this date.
This argument is mutually exclusive with: keep_last
after : datetime, optional
Remove data ingested after this date.
This argument is mutually exclusive with: keep_last
keep_last : int, optional
Remove all but the last ``keep_last`` ingestions.
This argument is mutually exclusive with:
before
after
Returns
-------
cleaned : set[str]
The names of the runs that were removed.
Raises
------
BadClean
Raised when ``before`` and or ``after`` are passed with ``keep_last``.
This is a subclass of ``ValueError``.
"""
try:
all_runs = sorted(
pd.Timestamp(f)
for f in os.listdir(pth.data_path([name], environ=environ))
if not pth.hidden(f)
)
except OSError as e:
if e.errno != errno.ENOENT:
raise
raise UnknownBundle(name)
if (before is not None or after is not None) and keep_last is not None:
raise BadClean(before, after, keep_last)
if keep_last is None:
def in_last_n(dt):
return False
else:
last_n_dts = set(all_runs[:keep_last])
def in_last_n(dt):
return dt in last_n_dts
def should_clean(name):
dt = pd.Timestamp(name)
return (
(
(before is not None and dt < before) or
(after is not None and dt > after)
) and
not in_last_n(dt)
)
cleaned = set()
for run in all_runs:
if should_clean(run):
shutil.rmdir(run)
cleaned.add(run)
return cleaned
bundles, register, unregister, ingest, load, clean = _make_bundle_core()
zipline/data/bundles/quandl.py
+32 -3
View File
@@ -1,15 +1,19 @@
"""
Module for building a complete daily dataset from Quandl's WIKI dataset.
"""
from contextlib import closing
from io import BytesIO
from itertools import count
import tarfile
from time import time, sleep
from logbook import Logger
import pandas as pd
from six.moves.urllib.parse import urlencode
from six.moves.urllib.request import urlopen
from . import core as bundles
from zipline.utils.cli import maybe_show_progress
from zipline.data import bundles
log = Logger(__name__)
seconds_per_call = (pd.Timedelta('10 minutes') / 2000).total_seconds()
@@ -260,12 +264,13 @@ def gen_symbol_data(api_key,
@bundles.register('quandl')
def quandl_bundle(environ,
asset_db_writer,
minute_bar_writer, # unused
minute_bar_writer,
daily_bar_writer,
adjustment_writer,
calendar,
cache,
show_progress):
show_progress,
output_dir):
"""Build a zipline data bundle from the Quandl WIKI dataset.
"""
api_key = environ.get('QUANDL_API_KEY')
@@ -291,8 +296,32 @@ def quandl_bundle(environ,
dividends,
environ.get('QUANDL_DOWNLOAD_ATTEMPTS', 5),
),
show_progress=show_progress,
)
adjustment_writer.write(
splits=pd.concat(splits, ignore_index=True),
dividends=pd.concat(dividends, ignore_index=True),
)
QUANTOPIAN_QUANDL_URL = (
'https://s3.amazonaws.com/quantopian-public-zipline-data/quandl'
)
@bundles.register('quantopian-quandl', create_writers=False)
def quantopian_quandl_bundle(environ,
asset_db_writer,
minute_bar_writer,
daily_bar_writer,
adjustment_writer,
calendar,
cache,
show_progress,
output_dir):
if show_progress:
print('Downloading quandl data. This can take about 1 minute.')
# use closing for py2 compat
with closing(urlopen(QUANTOPIAN_QUANDL_URL)) as f, \
tarfile.open('r', fileobj=BytesIO(f.read())) as tar:
tar.extractall(output_dir)
zipline/data/bundles/yahoo.py
+1
View File
@@ -62,6 +62,7 @@ def yahoo_equities(symbols, start=None, end=None):
calendar,
cache,
show_progress,
output_dir,
# pass these as defaults to make them 'nonlocal' in py2
start=start,
end=end):
zipline/utils/cache.py
+38 -11
View File
@@ -194,7 +194,7 @@ class dataframe_cache(MutableMapping):
self._protocol = int(s[1]) if len(s) == 2 else None
self.serialize = self._serialize_pickle
self.deserialize = self._deserialize_pickle
self.deserialize = pickle.load
ensure_directory(self.path)
@@ -202,10 +202,6 @@ class dataframe_cache(MutableMapping):
with open(path, 'wb') as f:
pickle.dump(df, f, protocol=self._protocol)
def _deserialize_pickle(self, path):
with open(path, 'rb') as f:
return pickle.load(f)
def _keypath(self, key):
return os.path.join(self.path, key)
@@ -226,9 +222,11 @@ class dataframe_cache(MutableMapping):
with self.lock:
try:
return self.deserialize(self._keypath(key))
except UnboundLocalError:
# This is how pandas fails if the file doesn't exist! #pandas
with open(self._keypath(key), 'rb') as f:
return self.deserialize(f)
except IOError as e:
if e.errno != errno.ENOENT:
raise
raise KeyError(key)
def __setitem__(self, key, value):
@@ -280,6 +278,13 @@ class working_file(object):
self._tmpfile = NamedTemporaryFile(*args, **kwargs)
self._final_path = final_path
@property
def path(self):
"""Alias for ``name`` to be consistent with
:class:`~zipline.utils.cache.working_dir`.
"""
return self._tmpfile.name
def _commit(self):
"""Sync the temporary file to the final path.
"""
@@ -316,13 +321,35 @@ class working_dir(object):
meaning it has as strong of guarantees as :func:`shutil.copytree`.
"""
def __init__(self, final_path, *args, **kwargs):
self.name = mkdtemp()
self.path = mkdtemp()
self._final_path = final_path
def mkdir(self, *path_parts):
"""Create a subdirectory of the working directory.
Parameters
----------
path_parts : iterable[str]
The parts of the path after the working directory.
"""
path = self.getpath(*path_parts)
os.mkdir(path)
return path
def getpath(self, *path_parts):
"""Get a path relative to the working directory.
Parameters
----------
path_parts : iterable[str]
The parts of the path after the working directory.
"""
return os.path.join(self.path, *path_parts)
def _commit(self):
"""Sync the temporary directory to the final path.
"""
copytree(self.name, self._final_path)
copytree(self.path, self._final_path)
def __enter__(self):
return self
@@ -330,4 +357,4 @@ class working_dir(object):
def __exit__(self, *exc_info):
if exc_info[0] is None:
self._commit()
rmtree(self.name)
rmtree(self.path)
zipline/utils/run_algo.py
+4 -4
View File
@@ -145,7 +145,7 @@ def _run(handle_data,
'before_trading_start': before_trading_start,
'analyze': analyze,
} if algotext is None else {
'algo_filename': algofile,
'algo_filename': algofile.name,
'script': algotext,
}
).run(
@@ -263,7 +263,7 @@ def run_algorithm(start,
``bundle_timestamp``
bundle : str, optional
The name of the data bundle to use to load the data to run the backtest
with. This defaults to 'quandl'.
with. This defaults to 'quantopian-quandl'.
This argument is mutually exclusive with ``data``.
bundle_timestamp : datetime, optional
The datetime to lookup the bundle data for. This defaults to the
@@ -299,8 +299,8 @@ def run_algorithm(start,
'bundle': bundle,
})
if not non_none_data:
# if neither data nor bundle are passed use 'quandl'
bundle = 'quandl'
# if neither data nor bundle are passed use 'quantopian-quandl'
bundle = 'quantopian-quandl'
if len(non_none_data) != 1:
raise ValueError(