Initial work on quandl bundle

catalyst/data/bundles/base.py

@@ -18,7 +18,7 @@ from itertools import count
 import tarfile
 from time import time, sleep
 
-from abc import abstractmethod
+from abc import abstractmethod, abstractproperty
 import logbook
 import pandas as pd
 
@@ -78,6 +78,14 @@ class BaseBundle(object):
     def wait_time(self):
         raise NotImplementedError()
 
+    @abstractproperty
+    def splits(self):
+        raise NotImplementedError()
+
+    @abstractproperty
+    def dividends(self):
+        raise NotImplementedError()
+
     @abstractmethod
     def fetch_raw_metadata_frame(self, api_key, page_number):
         raise NotImplementedError()
@@ -185,7 +193,21 @@ class BaseBundle(object):
             # contains an appropriately initialized file structure.  We don't
             # forsee a usecase for adjustments at this time, but may later
             # choose to expose this functionality in the future.
-            adjustment_writer.write()
+            if len(self.splits) > 0 or len(self.dividends) > 0:
+                adjustment_writer.write(
+                    splits=(
+                        pd.concat(self.splits, ignore_index=True)
+                        #if self.splits is not None \
+                        #and len(self.splits) > 0 else
+                        #None
+                    ),
+                    dividends=(
+                        pd.concat(self.dividends, ignore_index=True)
+                        #if self.dividends is not None \
+                        #and len(dividends) > 0 else
+                        #None
+                    ),
+                )
         else:
             # Otherwise, user has instructed to download and untar bundle
             # directly from the bundles `tar_url`.
@@ -246,9 +268,16 @@ class BaseBundle(object):
                             page_number,
                         )
                         break
-                    except ValueError:
+                    except ValueError as e:
                         raw = pd.DataFrame([])
                         break
+                    except Exception as e:
+                        log.exception(
+                            'Failed to load metadata from {}. '
+                            'Retrying.'.format(
+                                name=self.name,
+                            )
+                        ) 
                 else:
                     raise ValueError(
                         'Failed to download metadata page %d after %d '
@@ -297,6 +326,7 @@ class BaseBundle(object):
 
                 # Perform and require post-processing of metadata.
                 final_symbol_metadata = self.post_process_symbol_metadata(
+                    asset_id,
                     metadata.iloc[asset_id],
                     raw_data,        
                 )
@@ -335,9 +365,11 @@ class BaseBundle(object):
                 api_key,
                 cache,
                 symbol,
+                calendar,
                 start_session,
                 end_session,
                 data_frequency,
+                retries,
             )
 
             # TODO(cfromknecht) further data validation?
@@ -359,9 +391,11 @@ class BaseBundle(object):
                                    api_key,
                                    cache,
                                    symbol,
+                                   calendar,
                                    start_session,
                                    end_session,
-                                   data_frequency):
+                                   data_frequency,
+                                   retries):
 
         # Attempt to load pre-existing symbol data from cache.
         try:
@@ -371,54 +405,68 @@ class BaseBundle(object):
 
         # Select the most recent date in cached dataset if it exists,
         # otherwise use the provided `start_session`.
-        last = (
-            raw_data.index[-1].tz_localize('UTC')
-            if raw_data is not None and not raw_data.empty else
-            start_session
-        )
+        last = start_session
+        if raw_data is not None and len(raw_data) > 0:
+            last = raw_data.index[-1].tz_localize('UTC')
+        
+        should_sleep = False
 
         # Determine time at which cached data will be considered stale.
-        cache_expiration = last + pd.Timedelta(minutes=5)
-        if start_time <= cache_expiration:
+        cache_expiration = last + pd.Timedelta(days=2)
+        if start_time <= cache_expiration and raw_data is not None:
             # Data is fresh enough to reuse, no need to update. Iterator can
             # proceed to next symbol directly since no API call was required.
-            should_sleep = False
+            return raw_data, should_sleep
+
+        # Data for symbol is old enough to attempt an update or is not
+        # present in the cache.  Fetch raw data for a single symbol 
+        # with requested intervals and frequency. Retry as necessary.
+        for _ in range(retries):
+            try:
+                raw_data = self.fetch_raw_symbol_frame(
+                    api_key,
+                    symbol,
+                    calendar,
+                    start_session,
+                    end_session,
+                    data_frequency,
+                )
+                raw_data.index = pd.to_datetime(raw_data.index, utc=True)
+
+                # Filter incoming data to fit start and end sessions.
+                raw_data = raw_data[
+                    (raw_data.index >= start_session) &
+                    (raw_data.index <= end_session)
+                ]
+
+                # Filter out any duplicates entries, keep last one, since
+                # previous frame is probably an incomplete.
+                raw_data = raw_data[~raw_data.index.duplicated(keep='last')]
+
+                # Cache latest symbol data.
+                cache[symbol] = raw_data
+
+                # If we arrive here, we must have attempted an API call.
+                # This flag tells the iterator to pause before starting the next
+                # asset, that we don't exceed the data source's rate limit.
+                should_sleep = True
+
+                return raw_data, should_sleep
+
+            except Exception as e:
+                log.exception(
+                    'Exception raised fetching {name} data. Retrying.'
+                    .format(name=self.name)
+                )
         else:
-            # Data for symbol is old enough to attempt an update or is not
-            # present in the cache.  Fetch raw data for a single symbol 
-            # with requested intervals and frequency.
-            raw_diff = self.fetch_raw_symbol_frame(
-                api_key,
-                symbol,
-                last,
-                end_session,
-                data_frequency,
+            raise ValueError(
+                'Failed to download data for symbol {sym} '
+                'after {n} attempts.'.format(
+                    sym=symbol,
+                    n=retries,
+                )
             )
 
-            # Filter incoming data to minimize overlap.
-            raw_diff = raw_diff[
-                (raw_diff.index >= last) &
-                (raw_diff.index <= end_session)
-            ]
-
-            # Append incoming data to cached data if it exists,
-            # otherwise treat incoming data as the entire raw dataset.
-            raw_data = cache[symbol] = (
-                raw_data.append(raw_diff)
-                if raw_data is not None else
-                raw_diff
-            )
-
-            # Filter out any duplicates entries, keep last one as previous
-            # one was probably an incomplete frame.
-            raw_data = raw_data[~raw_data.index.duplicated(keep='last')]
-
-            # If we arrive here, we must have attempted an API call.
-            # This flag tells the iterator to pause before starting the next
-            # asset, that we don't exceed the data source's rate limit.
-            should_sleep = True
-
-        return raw_data, should_sleep
 
     def _write_symbol_for_freq(self,
                                pricing_iter,

catalyst/data/bundles/base_pricing.py

@@ -38,6 +38,9 @@ class BasePricingBundle(BaseBundle):
         ]
 
 class BaseCryptoPricingBundle(BasePricingBundle):
+    def __init__(self):
+        super(BasePricingBundle, self).__init__()
+
     @lazyval
     def calendar_name(self):
         return 'OPEN'
@@ -46,7 +49,20 @@ class BaseCryptoPricingBundle(BasePricingBundle):
     def minutes_per_day(self):
         return 1440
 
+    @property
+    def splits(self):
+        return []
+
+    @property
+    def dividends(self):
+        return []
+
 class BaseEquityPricingBundle(BasePricingBundle):
+    def __init__(self):
+        super(BasePricingBundle, self).__init__()
+        self._splits = []
+        self._dividends = []
+
     @lazyval
     def calendar_name(self):
         return 'NYSE'
@@ -54,3 +70,12 @@ class BaseEquityPricingBundle(BasePricingBundle):
     @lazyval
     def minutes_per_day(self):
         return 390
+
+
+    @property
+    def splits(self):
+        return self._splits
+
+    @property
+    def dividends(self):
+        return self._dividends

catalyst/data/bundles/poloniex.py

@@ -67,15 +67,17 @@ class PoloniexBundle(BaseCryptoPricingBundle):
             inplace=True,
         )
 
+        raw = raw[raw['isFrozen'] == 0]
+
         return raw
 
-    def post_process_symbol_metadata(self, metadata, data):
-        start_date = data.index[0].tz_localize(None)
-        end_date = data.index[-1].tz_localize(None)
+    def post_process_symbol_metadata(self, asset_id, sym_md, sym_data):
+        start_date = sym_data.index[0].tz_localize(None)
+        end_date = sym_data.index[-1].tz_localize(None)
         ac_date = end_date + pd.Timedelta(days=1)
 
         return (
-            metadata.symbol,
+            sym_md.symbol,
             start_date,
             end_date,
             ac_date,
@@ -84,6 +86,7 @@ class PoloniexBundle(BaseCryptoPricingBundle):
     def fetch_raw_symbol_frame(self,
                                api_key,
                                symbol,
+                               calendar,
                                start_date,
                                end_date,
                                frequency):
@@ -130,7 +133,6 @@ class PoloniexBundle(BaseCryptoPricingBundle):
         period_map = {
             'daily': 86400,
             '5-minute': 300,
-            'minute': 60,
         }
 
         try:

catalyst/data/bundles/quandl.py

@@ -1,3 +1,28 @@
+#
+# Copyright 2017 Enigma MPC, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+
+from datetime import datetime
+
+import pandas as pd
+
+from six.moves.urllib.parse import urlencode
+
+from catalyst.data.bundles.core import register_bundle
+from catalyst.data.bundles.base_pricing import BaseEquityPricingBundle
+from catalyst.utils.memoize import lazyval
+
 """
 Module for building a complete daily dataset from Quandl's WIKI dataset.
 """
@@ -17,350 +42,190 @@ from . import core as bundles
 
 log = Logger(__name__)
 seconds_per_call = (pd.Timedelta('10 minutes') / 2000).total_seconds()
-# Invalid symbols that quandl has had in its metadata:
-excluded_symbols = frozenset({'TEST123456789'})
 
+class QuandlBundle(BaseEquityPricingBundle):
+    @lazyval
+    def name(self):
+        return 'quandl'
 
-def _fetch_raw_metadata(api_key, cache, retries, environ):
-    """Generator that yields each page of data from the metadata endpoint
-    as a dataframe.
-    """
-    for page_number in count(1):
-        key = 'metadata-page-%d' % page_number
-        try:
-            raw = cache[key]
-        except KeyError:
-            for _ in range(retries):
-                try:
-                    raw = pd.read_csv(
-                        format_metadata_url(api_key, page_number),
-                        date_parser=pd.tseries.tools.to_datetime,
-                        parse_dates=[
-                            'oldest_available_date',
-                            'newest_available_date',
-                        ],
-                        dtypes={
-                            'dataset_code': 'int',
-                            'name': 'str',
-                            'oldest_available_date': 'str',
-                            'newest_available_date': 'str',
-                        },
-                        usecols=[
-                            'dataset_code',
-                            'name',
-                            'oldest_available_date',
-                            'newest_available_date',
-                        ],
-                    )
-                    break
-                except ValueError:
-                    # when we are past the last page we will get a value
-                    # error because there will be no columns
-                    raw = pd.DataFrame([])
-                    break
-                except Exception:
-                    pass
-            else:
-                raise ValueError(
-                    'Failed to download metadata page %d after %d'
-                    ' attempts.' % (page_number, retries),
-                )
+    @lazyval
+    def exchange(self):
+        return 'QUANDL'
 
-            cache[key] = raw
+    @lazyval
+    def frequencies(self):
+        return set(('daily',))
 
-        if raw.empty:
-            # use the empty dataframe to signal completion
-            break
-        yield raw
+    @lazyval
+    def tar_url(self):
+        return 'https://s3.amazonaws.com/quantopian-public-zipline-data/quandl'
 
+    @lazyval
+    def wait_time(self):
+        return pd.Timedelta(milliseconds=300)
 
-def fetch_symbol_metadata_frame(api_key,
-                                cache,
-                                retries=5,
-                                environ=None,
-                                show_progress=False):
-    """
-    Download Quandl symbol metadata.
+    @lazyval
+    def _excluded_symbols(self):
+        """
+        Invalid symbols that quandl has had in its metadata:
+        """
+        return frozenset({'TEST123456789'})
 
-    Parameters
-    ----------
-    api_key : str
-        The quandl api key to use. If this is None then no api key will be
-        sent.
-    cache : DataFrameCache
-        The cache to use for persisting the intermediate data.
-    retries : int, optional
-        The number of times to retry each request before failing.
-    environ : mapping[str -> str], optional
-        The environment to use to find the catalyst home. By default this
-        is ``os.environ``.
-    show_progress : bool, optional
-        Show a progress bar for the download of this data.
-
-    Returns
-    -------
-    metadata_frame : pd.DataFrame
-        A dataframe with the following columns:
-          symbol: the asset's symbol
-          name: the full name of the asset
-          start_date: the first date of data for this asset
-          end_date: the last date of data for this asset
-          auto_close_date: end_date + one day
-          exchange: the exchange for the asset; this is always 'quandl'
-        The index of the dataframe will be used for symbol->sid mappings but
-        otherwise does not have specific meaning.
-    """
-    raw_iter = _fetch_raw_metadata(api_key, cache, retries, environ)
-
-    def item_show_func(_, _it=iter(count())):
-        'Downloading page: %d' % next(_it)
-
-    with maybe_show_progress(raw_iter,
-                             show_progress,
-                             item_show_func=item_show_func,
-                             label='Downloading WIKI metadata: ') as blocks:
-        data = pd.concat(blocks, ignore_index=True).rename(columns={
-            'dataset_code': 'symbol',
-            'name': 'asset_name',
-            'oldest_available_date': 'start_date',
-            'newest_available_date': 'end_date',
-        }).sort_values('symbol')
-
-    data = data[~data.symbol.isin(excluded_symbols)]
-    # cut out all the other stuff in the name column
-    # we need to escape the paren because it is actually splitting on a regex
-    data.asset_name = data.asset_name.str.split(r' \(', 1).str.get(0)
-    data['exchange'] = 'QUANDL'
-
-    data['start_date'] = data['start_date'].astype(datetime)
-    data['end_date'] = data['end_date'].astype(datetime)
-
-    data['auto_close_date'] = data['end_date'] + pd.Timedelta(days=1)
-    return data
-
-
-def format_metadata_url(api_key, page_number):
-    """Build the query RL for the quandl WIKI metadata.
-    """
-    query_params = [
-        ('per_page', '100'),
-        ('sort_by', 'id'),
-        ('page', str(page_number)),
-        ('database_code', 'WIKI'),
-    ]
-    if api_key is not None:
-        query_params = [('api_key', api_key)] + query_params
-    return (
-        'https://www.quandl.com/api/v3/datasets.csv?' + urlencode(query_params)
-    )
-
-
-def format_wiki_url(api_key, symbol, start_date, end_date):
-    """
-    Build a query URL for a quandl WIKI dataset.
-    """
-    query_params = [
-        ('start_date', start_date.strftime('%Y-%m-%d')),
-        ('end_date', end_date.strftime('%Y-%m-%d')),
-        ('order', 'asc'),
-    ]
-    if api_key is not None:
-        query_params = [('api_key', api_key)] + query_params
-
-    return (
-        "https://www.quandl.com/api/v3/datasets/WIKI/"
-        "{symbol}.csv?{query}".format(
-            symbol=symbol,
-            query=urlencode(query_params),
-        )
-    )
-
-
-def fetch_single_equity(api_key,
-                        symbol,
-                        start_date,
-                        end_date,
-                        retries=5):
-    """
-    Download data for a single equity.
-    """
-    for _ in range(retries):
-        try:
-            return pd.read_csv(
-                format_wiki_url(api_key, symbol, start_date, end_date),
-                parse_dates=['Date'],
-                index_col='Date',
-                usecols=[
-                    'Open',
-                    'High',
-                    'Low',
-                    'Close',
-                    'Volume',
-                    'Date',
-                    'Ex-Dividend',
-                    'Split Ratio',
-                ],
-                na_values=['NA'],
-            ).rename(columns={
-                'Open': 'open',
-                'High': 'high',
-                'Low': 'low',
-                'Close': 'close',
-                'Volume': 'volume',
-                'Date': 'date',
-                'Ex-Dividend': 'ex_dividend',
-                'Split Ratio': 'split_ratio',
-            })
-        except Exception:
-            log.exception("Exception raised reading Quandl data. Retrying.")
-    else:
-        raise ValueError(
-            "Failed to download data for %r after %d attempts." % (
-                symbol, retries
-            )
+    def fetch_raw_metadata_frame(self, api_key, page_number):
+        raw = pd.read_csv(
+            self._format_metadata_url(api_key, page_number),
+            date_parser=pd.tseries.tools.to_datetime,
+            parse_dates=[
+                'oldest_available_date',
+                'newest_available_date',
+            ],
+            dtype={
+                'dataset_code': 'str',
+                'name': 'str',
+                'oldest_available_date': 'str',
+                'newest_available_date': 'str',
+            },
+            usecols=[
+                'dataset_code',
+                'name',
+                'oldest_available_date',
+                'newest_available_date',
+            ],
+        ).rename(
+            columns={
+                'dataset_code': 'symbol',
+                'name': 'asset_name',
+                'oldest_available_date': 'start_date',
+                'newest_available_date': 'end_date',
+            },
         )
 
+        raw['start_date'] = raw['start_date'].astype(datetime)
+        raw['end_date'] = raw['end_date'].astype(datetime)
+        raw['ac_date'] = raw['end_date'] + pd.Timedelta(days=1)
 
-def _update_splits(splits, asset_id, raw_data):
-    split_ratios = raw_data.split_ratio
-    df = pd.DataFrame({'ratio': 1 / split_ratios[split_ratios != 1]})
-    df.index.name = 'effective_date'
-    df.reset_index(inplace=True)
-    df['sid'] = asset_id
-    splits.append(df)
+        # Filter out invalid symbols
+        raw = raw[~raw.symbol.isin(self._excluded_symbols)]
 
+        # cut out all the other stuff in the name column
+        # we need to escape the paren because it is actually splitting on a regex
+        raw.asset_name = raw.asset_name.str.split(r' \(', 1).str.get(0)
 
-def _update_dividends(dividends, asset_id, raw_data):
-    divs = raw_data.ex_dividend
-    df = pd.DataFrame({'amount': divs[divs != 0]})
-    df.index.name = 'ex_date'
-    df.reset_index(inplace=True)
-    df['sid'] = asset_id
-    # we do not have this data in the WIKI dataset
-    df['record_date'] = df['declared_date'] = df['pay_date'] = pd.NaT
-    dividends.append(df)
+        return raw
 
-
-def gen_symbol_data(api_key,
-                    cache,
-                    symbol_map,
-                    calendar,
-                    start_session,
-                    end_session,
-                    splits,
-                    dividends,
-                    retries):
-    for asset_id, symbol in symbol_map.iteritems():
-        start_time = time()
-        try:
-            # see if we have this data cached.
-            raw_data = cache[symbol]
-            should_sleep = False
-        except KeyError:
-            # we need to fetch the data and then write it to our cache
-            raw_data = cache[symbol] = fetch_single_equity(
+    def fetch_raw_symbol_frame(self,
+                               api_key,
+                               symbol,
+                               calendar,
+                               start_session,
+                               end_session,
+                               data_frequency):
+        raw_data = pd.read_csv(
+            self._format_wiki_url(
                 api_key,
                 symbol,
-                start_date=start_session,
-                end_date=end_session,
-            )
-            should_sleep = True
-
-        _update_splits(splits, asset_id, raw_data)
-        _update_dividends(dividends, asset_id, raw_data)
+                start_session,
+                end_session,
+                data_frequency,
+            ),
+            parse_dates=['Date'],
+            index_col='Date',
+            usecols=[
+                'Open',
+                'High',
+                'Low',
+                'Close',
+                'Volume',
+                'Date',
+                'Ex-Dividend',
+                'Split Ratio',
+            ],
+            na_values=['NA'],
+        ).rename(columns={
+            'Open': 'open',
+            'High': 'high',
+            'Low': 'low',
+            'Close': 'close',
+            'Volume': 'volume',
+            'Date': 'date',
+            'Ex-Dividend': 'ex_dividend',
+            'Split Ratio': 'split_ratio',
+        })
 
         sessions = calendar.sessions_in_range(start_session, end_session)
 
-        raw_data = raw_data.reindex(
+        return raw_data.reindex(
             sessions.tz_localize(None),
             copy=False,
         ).fillna(0.0)
-        yield asset_id, raw_data
 
-        if should_sleep:
-            remaining = seconds_per_call - time() - start_time
-            if remaining > 0:
-                sleep(remaining)
+    def post_process_symbol_metadata(self, asset_id, sym_md, sym_data):
+        self._update_splits(asset_id, sym_data)
+        self._update_dividends(asset_id, sym_data)
+
+        return sym_md
+
+    def _update_splits(self, asset_id, raw_data):
+        split_ratios = raw_data.split_ratio
+        df = pd.DataFrame({'ratio': 1 / split_ratios[split_ratios != 1]})
+        df.index.name = 'effective_date'
+        df.reset_index(inplace=True)
+        df['sid'] = asset_id
+        self.splits.append(df)
 
 
-@bundles.register('quandl')
-def quandl_bundle(environ,
-                  asset_db_writer,
-                  minute_bar_writer,
-                  daily_bar_writer,
-                  adjustment_writer,
-                  calendar,
-                  start_session,
-                  end_session,
-                  cache,
-                  show_progress,
-                  output_dir):
-    """Build a catalyst data bundle from the Quandl WIKI dataset.
-    """
-    api_key = environ.get('QUANDL_API_KEY')
-    metadata = fetch_symbol_metadata_frame(
-        api_key,
-        cache=cache,
-        show_progress=show_progress,
-    )
-    symbol_map = metadata.symbol
-
-    # data we will collect in `gen_symbol_data`
-    splits = []
-    dividends = []
-
-    asset_db_writer.write(metadata)
-    daily_bar_writer.write(
-        gen_symbol_data(
-            api_key,
-            cache,
-            symbol_map,
-            calendar,
-            start_session,
-            end_session,
-            splits,
-            dividends,
-            environ.get('QUANDL_DOWNLOAD_ATTEMPTS', 5),
-        ),
-        assets=metadata.index,
-        show_progress=show_progress,
-    )
-    adjustment_writer.write(
-        splits=pd.concat(splits, ignore_index=True),
-        dividends=pd.concat(dividends, ignore_index=True),
-    )
+    def _update_dividends(self, asset_id, raw_data):
+        divs = raw_data.ex_dividend
+        df = pd.DataFrame({'amount': divs[divs != 0]})
+        df.index.name = 'ex_date'
+        df.reset_index(inplace=True)
+        df['sid'] = asset_id
+        # we do not have this data in the WIKI dataset
+        df['record_date'] = df['declared_date'] = df['pay_date'] = pd.NaT
+        self.dividends.append(df)
 
 
-QUANTOPIAN_QUANDL_URL = (
-    'https://s3.amazonaws.com/quantopian-public-zipline-data/quandl'
-)
+    def _format_metadata_url(self, api_key, page_number):
+        """Build the query RL for the quandl WIKI metadata.
+        """
+        query_params = [
+            ('per_page', '100'),
+            ('sort_by', 'id'),
+            ('page', str(page_number)),
+            ('database_code', 'WIKI'),
+        ]
+        if api_key is not None:
+            query_params = [('api_key', api_key)] + query_params
 
-
-@bundles.register('quantopian-quandl', create_writers=False)
-def quantopian_quandl_bundle(environ,
-                             asset_db_writer,
-                             minute_bar_writer,
-                             daily_bar_writer,
-                             adjustment_writer,
-                             calendar,
-                             start_session,
-                             end_session,
-                             cache,
-                             show_progress,
-                             output_dir):
-    if show_progress:
-        data = bundles.download_with_progress(
-            QUANTOPIAN_QUANDL_URL,
-            chunk_size=bundles.ONE_MEGABYTE,
-            label="Downloading Bundle: quantopian-quandl",
+        return (
+            'https://www.quandl.com/api/v3/datasets.csv?' + urlencode(query_params)
         )
-    else:
-        data = bundles.download_without_progress(QUANTOPIAN_QUANDL_URL)
-
-    with tarfile.open('r', fileobj=data) as tar:
-        if show_progress:
-            print("Writing data to %s." % output_dir)
-        tar.extractall(output_dir)
 
 
-register_calendar_alias("QUANDL", "NYSE")
+    def _format_wiki_url(self,
+                         api_key,
+                         symbol,
+                         start_date,
+                         end_date,
+                         data_frequency):
+        """
+        Build a query URL for a quandl WIKI dataset.
+        """
+        query_params = [
+            ('start_date', start_date.strftime('%Y-%m-%d')),
+            ('end_date', end_date.strftime('%Y-%m-%d')),
+            ('order', 'asc'),
+        ]
+        if api_key is not None:
+            query_params = [('api_key', api_key)] + query_params
+
+        return (
+            "https://www.quandl.com/api/v3/datasets/WIKI/"
+            "{symbol}.csv?{query}".format(
+                symbol=symbol,
+                query=urlencode(query_params),
+            )
+        )
+
+register_calendar_alias('QUANDL', 'NYSE')
+register_bundle(QuandlBundle)