Merge pull request #1430 from quantopian/ingest-versions

Ingest versions

conda/alembic/bld.bat

@@ -0,0 +1,8 @@
+"%PYTHON%" setup.py install
+if errorlevel 1 exit 1
+
+:: Add more build steps here, if they are necessary.
+
+:: See
+:: http://docs.continuum.io/conda/build.html
+:: for a list of environment variables that are set during the build process.

conda/alembic/build.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+$PYTHON setup.py install
+
+# Add more build steps here, if they are necessary.
+
+# See
+# http://docs.continuum.io/conda/build.html
+# for a list of environment variables that are set during the build process.

conda/alembic/meta.yaml

@@ -0,0 +1,73 @@
+package:
+  name: alembic
+  version: "0.7.7"
+
+source:
+  fn: alembic-0.7.7.tar.gz
+  url: https://pypi.python.org/packages/41/80/58c25d9a4e8321376a0231ef4cedd22f93c899c28c93bc236ca991a0f291/alembic-0.7.7.tar.gz
+  md5: 8bd77f40857100da2cdcb6f5da9a7f1c
+#  patches:
+   # List any patch files here
+   # - fix.patch
+
+build:
+  # noarch_python: True
+  # preserve_egg_dir: True
+  entry_points:
+    # Put any entry points (scripts to be generated automatically) here. The
+    # syntax is module:function.  For example
+    #
+    # - alembic = alembic:main
+    #
+    # Would create an entry point called alembic that calls alembic.main()
+
+    - alembic = alembic.config:main
+
+  # If this is a new build for the same version, increment the build
+  # number. If you do not include this key, it defaults to 0.
+  # number: 1
+
+requirements:
+  build:
+    - python
+    - setuptools
+    - sqlalchemy >=0.7.6
+    - mako
+
+  run:
+    - python
+    - sqlalchemy >=0.7.6
+    - mako
+
+test:
+  # Python imports
+  imports:
+    - alembic
+    - alembic.autogenerate
+    - alembic.ddl
+    - alembic.testing
+    - alembic.testing.plugin
+
+  commands:
+    # You can put test commands to be run here.  Use this to test that the
+    # entry points work.
+
+    - alembic --help
+
+  # You can also put a file called run_test.py in the recipe that will be run
+  # at test time.
+
+  requires:
+    - mock
+    - nose >=0.11
+    # Put any additional test requirements here.  For example
+    # - nose
+
+about:
+  home: http://bitbucket.org/zzzeek/alembic
+  license: MIT
+  summary: 'A database migration tool for SQLAlchemy.'
+
+# See
+# http://docs.continuum.io/conda/build.html for
+# more information about meta.yaml

etc/requirements.txt

@@ -56,6 +56,8 @@ multipledispatch==0.4.8
 
 # Asset writer and finder
 sqlalchemy==1.0.8
+# For asset db management
+alembic==0.7.7
 
 # for intervaltree
 sortedcontainers==1.4.4

etc/requirements_dev.txt

@@ -61,6 +61,3 @@ piprot==0.9.6
 
 # For mocking out requests fetches
 responses==0.4.0
-
-# For asset db management
-alembic==0.7.7

tests/data/bundles/test_core.py

@@ -2,12 +2,17 @@ import os
 
 from nose_parameterized import parameterized
 import pandas as pd
+import sqlalchemy as sa
 from toolz import valmap
 import toolz.curried.operator as op
+from zipline.assets import ASSET_DB_VERSION
 
+from zipline.assets.asset_writer import check_version_info
 from zipline.assets.synthetic import make_simple_equity_info
-from zipline.data.bundles import UnknownBundle, from_bundle_ingest_dirname
-from zipline.data.bundles.core import _make_bundle_core, BadClean
+from zipline.data.bundles import UnknownBundle, from_bundle_ingest_dirname, \
+    ingestions_for_bundle
+from zipline.data.bundles.core import _make_bundle_core, BadClean, \
+    to_bundle_ingest_dirname, asset_db_path
 from zipline.lib.adjustment import Float64Multiply
 from zipline.pipeline.loaders.synthetic import (
     make_bar_data,
@@ -17,7 +22,8 @@ from zipline.testing import (
     subtest,
     str_to_seconds,
 )
-from zipline.testing.fixtures import WithInstanceTmpDir, ZiplineTestCase
+from zipline.testing.fixtures import WithInstanceTmpDir, ZiplineTestCase, \
+    WithDefaultDateBounds
 from zipline.testing.predicates import (
     assert_equal,
     assert_false,
@@ -37,7 +43,13 @@ import zipline.utils.paths as pth
 _1_ns = pd.Timedelta(1, unit='ns')
 
 
-class BundleCoreTestCase(WithInstanceTmpDir, ZiplineTestCase):
+class BundleCoreTestCase(WithInstanceTmpDir,
+                         WithDefaultDateBounds,
+                         ZiplineTestCase):
+
+    START_DATE = pd.Timestamp('2014-01-06', tz='utc')
+    END_DATE = pd.Timestamp('2014-01-10', tz='utc')
+
     def init_instance_fixtures(self):
         super(BundleCoreTestCase, self).init_instance_fixtures()
         (self.bundles,
@@ -111,18 +123,17 @@ class BundleCoreTestCase(WithInstanceTmpDir, ZiplineTestCase):
         assert_true(called[0])
 
     def test_ingest(self):
-        start = pd.Timestamp('2014-01-06', tz='utc')
-        end = pd.Timestamp('2014-01-10', tz='utc')
         calendar = get_calendar('NYSE')
-
-        sessions = calendar.sessions_in_range(start, end)
-        minutes = calendar.minutes_for_sessions_in_range(start, end)
+        sessions = calendar.sessions_in_range(self.START_DATE, self.END_DATE)
+        minutes = calendar.minutes_for_sessions_in_range(
+            self.START_DATE, self.END_DATE,
+        )
 
         sids = tuple(range(3))
         equities = make_simple_equity_info(
             sids,
-            start,
-            end,
+            self.START_DATE,
+            self.END_DATE,
         )
 
         daily_bar_data = make_bar_data(equities, sessions)
@@ -145,8 +156,8 @@ class BundleCoreTestCase(WithInstanceTmpDir, ZiplineTestCase):
         @self.register(
             'bundle',
             calendar=calendar,
-            start_session=start,
-            end_session=end,
+            start_session=self.START_DATE,
+            end_session=self.END_DATE,
         )
         def bundle_ingest(environ,
                           asset_db_writer,
@@ -193,8 +204,8 @@ class BundleCoreTestCase(WithInstanceTmpDir, ZiplineTestCase):
 
         actual = bundle.equity_daily_bar_reader.load_raw_arrays(
             columns,
-            start,
-            end,
+            self.START_DATE,
+            self.END_DATE,
             sids,
         )
         for actual_column, colname in zip(actual, columns):
@@ -253,6 +264,72 @@ class BundleCoreTestCase(WithInstanceTmpDir, ZiplineTestCase):
             msg='volume',
         )
 
+    def test_ingest_assets_versions(self):
+        versions = (1, 2)
+
+        called = [False]
+
+        @self.register('bundle', create_writers=False)
+        def bundle_ingest_no_create_writers(*args, **kwargs):
+            called[0] = True
+
+        now = pd.Timestamp.utcnow()
+        with self.assertRaisesRegexp(ValueError,
+                                     "ingest .* creates writers .* downgrade"):
+            self.ingest('bundle', self.environ, assets_versions=versions,
+                        timestamp=now - pd.Timedelta(seconds=1))
+        assert_false(called[0])
+        assert_equal(len(ingestions_for_bundle('bundle', self.environ)), 1)
+
+        @self.register('bundle', create_writers=True)
+        def bundle_ingest_create_writers(
+                environ,
+                asset_db_writer,
+                minute_bar_writer,
+                daily_bar_writer,
+                adjustment_writer,
+                calendar,
+                start_session,
+                end_session,
+                cache,
+                show_progress,
+                output_dir):
+            self.assertIsNotNone(asset_db_writer)
+            self.assertIsNotNone(minute_bar_writer)
+            self.assertIsNotNone(daily_bar_writer)
+            self.assertIsNotNone(adjustment_writer)
+
+            equities = make_simple_equity_info(
+                tuple(range(3)),
+                self.START_DATE,
+                self.END_DATE,
+            )
+            asset_db_writer.write(equities=equities)
+            called[0] = True
+
+        # Explicitly use different timestamp; otherwise, test could run so fast
+        # that first ingestion is re-used.
+        self.ingest('bundle', self.environ, assets_versions=versions,
+                    timestamp=now)
+        assert_true(called[0])
+
+        ingestions = ingestions_for_bundle('bundle', self.environ)
+        assert_equal(len(ingestions), 2)
+        for version in sorted(set(versions) | {ASSET_DB_VERSION}):
+            eng = sa.create_engine(
+                'sqlite:///' +
+                asset_db_path(
+                    'bundle',
+                    to_bundle_ingest_dirname(ingestions[0]),  # most recent
+                    self.environ,
+                    version,
+                )
+            )
+            metadata = sa.MetaData()
+            metadata.reflect(eng)
+            version_table = metadata.tables['version_info']
+            check_version_info(eng, version_table, version)
+
     @parameterized.expand([('clean',), ('load',)])
     def test_bundle_doesnt_exist(self, fnname):
         with assert_raises(UnknownBundle) as e:

tests/test_assets.py

@@ -1452,7 +1452,7 @@ class TestAssetDBVersioning(ZiplineTestCase):
         # first downgrade to v3
         downgrade(self.engine, 3)
         metadata = sa.MetaData(conn)
-        metadata.reflect(bind=self.engine)
+        metadata.reflect()
         check_version_info(conn, metadata.tables['version_info'], 3)
         self.assertFalse('exchange_full' in metadata.tables)
 
@@ -1461,9 +1461,9 @@ class TestAssetDBVersioning(ZiplineTestCase):
 
         # Verify that the db version is now 0
         metadata = sa.MetaData(conn)
-        metadata.reflect(bind=self.engine)
+        metadata.reflect()
         version_table = metadata.tables['version_info']
-        check_version_info(self.engine, version_table, 0)
+        check_version_info(conn, version_table, 0)
 
         # Check some of the v1-to-v0 downgrades
         self.assertTrue('futures_contracts' in metadata.tables)

zipline/__main__.py

@@ -5,10 +5,10 @@ from functools import wraps
 import click
 import logbook
 import pandas as pd
+from six import text_type
 
 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:
@@ -284,19 +284,25 @@ def zipline_magic(line, cell=None):
     show_default=True,
     help='The data bundle to ingest.',
 )
+@click.option(
+    '--assets-version',
+    type=int,
+    multiple=True,
+    help='Version of the assets db to which to downgrade.',
+)
 @click.option(
     '--show-progress/--no-show-progress',
-    is_flag=True,
     default=True,
     help='Print progress information to the terminal.'
 )
-def ingest(bundle, show_progress):
+def ingest(bundle, assets_version, show_progress):
     """Ingest the data for the given bundle.
     """
     bundles_module.ingest(
         bundle,
         os.environ,
         pd.Timestamp.utcnow(),
+        assets_version,
         show_progress,
     )
 
@@ -352,11 +358,8 @@ def bundles():
             # hide the test data
             continue
         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,
+            ingestions = list(
+                map(text_type, bundles_module.ingestions_for_bundle(bundle))
             )
         except OSError as e:
             if e.errno != errno.ENOENT:
@@ -367,7 +370,7 @@ def bundles():
         # because there were no entries, print a single message indicating that
         # no ingestions have yet been made.
         for timestamp in ingestions or ["<no ingestions>"]:
-            print("%s %s" % (bundle, timestamp))
+            click.echo("%s %s" % (bundle, timestamp))
 
 
 if __name__ == '__main__':

zipline/assets/asset_db_migrations.py

@@ -7,8 +7,11 @@ from toolz.curried import do, operator as op
 
 from zipline.assets.asset_writer import write_version_info
 from zipline.errors import AssetDBImpossibleDowngrade
+from zipline.utils.preprocess import preprocess
+from zipline.utils.sqlite_utils import coerce_string_to_eng
 
 
+@preprocess(engine=coerce_string_to_eng)
 def downgrade(engine, desired_version):
     """Downgrades the assets db at the given engine to the desired version.
 
@@ -21,39 +24,39 @@ def downgrade(engine, desired_version):
     """
 
     # Check the version of the db at the engine
-    conn = engine.connect()
-    metadata = sa.MetaData(conn)
-    metadata.reflect(bind=engine)
-    version_info_table = metadata.tables['version_info']
-    starting_version = sa.select((version_info_table.c.version,)).scalar()
+    with engine.begin() as conn:
+        metadata = sa.MetaData(conn)
+        metadata.reflect()
+        version_info_table = metadata.tables['version_info']
+        starting_version = sa.select((version_info_table.c.version,)).scalar()
 
-    # Check for accidental upgrade
-    if starting_version < desired_version:
-        raise AssetDBImpossibleDowngrade(db_version=starting_version,
-                                         desired_version=desired_version)
+        # Check for accidental upgrade
+        if starting_version < desired_version:
+            raise AssetDBImpossibleDowngrade(db_version=starting_version,
+                                             desired_version=desired_version)
 
-    # Check if the desired version is already the db version
-    if starting_version == desired_version:
-        # No downgrade needed
-        return
+        # Check if the desired version is already the db version
+        if starting_version == desired_version:
+            # No downgrade needed
+            return
 
-    # Create alembic context
-    ctx = MigrationContext.configure(conn)
-    op = Operations(ctx)
+        # Create alembic context
+        ctx = MigrationContext.configure(conn)
+        op = Operations(ctx)
 
-    # Integer keys of downgrades to run
-    # E.g.: [5, 4, 3, 2] would downgrade v6 to v2
-    downgrade_keys = range(desired_version, starting_version)[::-1]
+        # Integer keys of downgrades to run
+        # E.g.: [5, 4, 3, 2] would downgrade v6 to v2
+        downgrade_keys = range(desired_version, starting_version)[::-1]
 
-    # Disable foreign keys until all downgrades are complete
-    _pragma_foreign_keys(conn, False)
+        # Disable foreign keys until all downgrades are complete
+        _pragma_foreign_keys(conn, False)
 
-    # Execute the downgrades in order
-    for downgrade_key in downgrade_keys:
-        _downgrade_methods[downgrade_key](op, engine, version_info_table)
+        # Execute the downgrades in order
+        for downgrade_key in downgrade_keys:
+            _downgrade_methods[downgrade_key](op, conn, version_info_table)
 
-    # Re-enable foreign keys
-    _pragma_foreign_keys(conn, True)
+        # Re-enable foreign keys
+        _pragma_foreign_keys(conn, True)
 
 
 def _pragma_foreign_keys(connection, on):
@@ -96,10 +99,10 @@ def downgrades(src):
 
         @do(op.setitem(_downgrade_methods, destination))
         @wraps(f)
-        def wrapper(op, engine, version_info_table):
-            version_info_table.delete().execute()  # clear the version
+        def wrapper(op, conn, version_info_table):
+            conn.execute(version_info_table.delete())  # clear the version
             f(op)
-            write_version_info(engine, version_info_table, destination)
+            write_version_info(conn, version_info_table, destination)
 
         return wrapper
     return _

zipline/assets/asset_writer.py

@@ -35,7 +35,9 @@ from zipline.assets.asset_db_schema import (
     version_info,
 )
 
+from zipline.utils.preprocess import preprocess
 from zipline.utils.range import from_tuple, intersecting_ranges
+from zipline.utils.sqlite_utils import coerce_string_to_eng
 
 # Define a namedtuple for use with the load_data and _load_data methods
 AssetData = namedtuple(
@@ -344,9 +346,8 @@ class AssetDBWriter(object):
     """
     DEFAULT_CHUNK_SIZE = SQLITE_MAX_VARIABLE_NUMBER
 
+    @preprocess(engine=coerce_string_to_eng)
     def __init__(self, engine):
-        if isinstance(engine, str):
-            engine = sa.create_engine('sqlite:///' + engine)
         self.engine = engine
 
     def write(self,
@@ -441,9 +442,9 @@ class AssetDBWriter(object):
         --------
         zipline.assets.asset_finder
         """
-        with self.engine.begin() as txn:
+        with self.engine.begin() as conn:
             # Create SQL tables if they do not exist.
-            self.init_db(txn)
+            self.init_db(conn)
 
             # Get the data to add to SQL.
             data = self._load_data(
@@ -456,25 +457,25 @@ class AssetDBWriter(object):
             self._write_df_to_table(
                 futures_exchanges,
                 data.exchanges,
-                txn,
+                conn,
                 chunk_size,
             )
             self._write_df_to_table(
                 futures_root_symbols,
                 data.root_symbols,
-                txn,
+                conn,
                 chunk_size,
             )
             self._write_assets(
                 'future',
                 data.futures,
-                txn,
+                conn,
                 chunk_size,
             )
             self._write_assets(
                 'equity',
                 data.equities,
-                txn,
+                conn,
                 chunk_size,
                 mapping_data=data.equities_mappings,
             )

zipline/assets/assets.py

@@ -60,7 +60,8 @@ from .asset_db_schema import (
 from zipline.utils.control_flow import invert
 from zipline.utils.memoize import lazyval
 from zipline.utils.numpy_utils import as_column
-from zipline.utils.sqlite_utils import group_into_chunks
+from zipline.utils.preprocess import preprocess
+from zipline.utils.sqlite_utils import group_into_chunks, coerce_string_to_eng
 
 log = Logger('assets.py')
 
@@ -141,12 +142,9 @@ class AssetFinder(object):
     # reference to an AssetFinder.
     PERSISTENT_TOKEN = "<AssetFinder>"
 
+    @preprocess(engine=coerce_string_to_eng)
     def __init__(self, engine):
-        self.engine = engine = (
-            sa.create_engine('sqlite:///' + engine)
-            if isinstance(engine, string_types) else
-            engine
-        )
+        self.engine = engine
         metadata = sa.MetaData(bind=engine)
         metadata.reflect(only=asset_db_table_names)
         for table_name in asset_db_table_names:

zipline/data/bundles/__init__.py

@@ -6,6 +6,7 @@ from .core import (
     clean,
     from_bundle_ingest_dirname,
     ingest,
+    ingestions_for_bundle,
     load,
     register,
     to_bundle_ingest_dirname,
@@ -20,6 +21,7 @@ __all__ = [
     'clean',
     'from_bundle_ingest_dirname',
     'ingest',
+    'ingestions_for_bundle',
     'load',
     'register',
     'to_bundle_ingest_dirname',

zipline/data/bundles/core.py

@@ -7,6 +7,7 @@ import warnings
 from contextlib2 import ExitStack
 import click
 import pandas as pd
+from six import string_types
 from toolz import curry, complement, take
 
 from ..us_equity_pricing import (
@@ -20,9 +21,11 @@ from ..minute_bars import (
     BcolzMinuteBarWriter,
 )
 from zipline.assets import AssetDBWriter, AssetFinder, ASSET_DB_VERSION
+from zipline.assets.asset_db_migrations import downgrade
 from zipline.utils.cache import (
     dataframe_cache,
     working_dir,
+    working_file,
 )
 from zipline.utils.compat import mappingproxy
 from zipline.utils.input_validation import ensure_timestamp, optionally
@@ -31,9 +34,9 @@ from zipline.utils.preprocess import preprocess
 from zipline.utils.calendars import get_calendar, register_calendar
 
 
-def asset_db_path(bundle_name, timestr, environ=None):
+def asset_db_path(bundle_name, timestr, environ=None, db_version=None):
     return pth.data_path(
-        asset_db_relative(bundle_name, timestr, environ),
+        asset_db_relative(bundle_name, timestr, environ, db_version),
         environ=environ,
     )
 
@@ -82,8 +85,10 @@ def minute_equity_relative(bundle_name, timestr, environ=None):
     return bundle_name, timestr, 'minute_equities.bcolz'
 
 
-def asset_db_relative(bundle_name, timestr, environ=None):
-    return bundle_name, timestr, 'assets-%d.sqlite' % ASSET_DB_VERSION
+def asset_db_relative(bundle_name, timestr, environ=None, db_version=None):
+    db_version = ASSET_DB_VERSION if db_version is None else db_version
+
+    return bundle_name, timestr, 'assets-%d.sqlite' % db_version
 
 
 def to_bundle_ingest_dirname(ts):
@@ -119,6 +124,15 @@ def from_bundle_ingest_dirname(cs):
     return pd.Timestamp(cs.replace(';', ':'))
 
 
+def ingestions_for_bundle(bundle, environ=None):
+    return sorted(
+        (from_bundle_ingest_dirname(ing)
+         for ing in os.listdir(pth.data_path([bundle], environ))
+         if not pth.hidden(ing)),
+        reverse=True,
+    )
+
+
 _BundlePayload = namedtuple(
     '_BundlePayload',
     'calendar start_session end_session minutes_per_day ingest create_writers',
@@ -282,7 +296,7 @@ def _make_bundle_core():
                 stacklevel=3,
             )
 
-        if isinstance(calendar, str):
+        if isinstance(calendar, string_types):
             calendar = get_calendar(calendar)
 
         # If the start and end sessions are not provided or lie outside
@@ -328,6 +342,7 @@ def _make_bundle_core():
     def ingest(name,
                environ=os.environ,
                timestamp=None,
+               assets_versions=(),
                show_progress=False):
         """Ingest data for a given bundle.
 
@@ -340,6 +355,8 @@ def _make_bundle_core():
         timestamp : datetime, optional
             The timestamp to use for the load.
             By default this is the current time.
+        assets_versions : Iterable[int], optional
+            Versions of the assets db to which to downgrade.
         show_progress : bool, optional
             Tell the ingest function to display the progress where possible.
         """
@@ -389,11 +406,10 @@ def _make_bundle_core():
                     bundle.end_session,
                     minutes_per_day=bundle.minutes_per_day,
                 )
-                asset_db_writer = AssetDBWriter(
-                    wd.getpath(*asset_db_relative(
-                        name, timestr, environ=environ,
-                    ))
-                )
+                assets_db_path = wd.getpath(*asset_db_relative(
+                    name, timestr, environ=environ,
+                ))
+                asset_db_writer = AssetDBWriter(assets_db_path)
 
                 adjustment_db_writer = stack.enter_context(
                     SQLiteAdjustmentWriter(
@@ -409,6 +425,10 @@ def _make_bundle_core():
                 minute_bar_writer = None
                 asset_db_writer = None
                 adjustment_db_writer = None
+                if assets_versions:
+                    raise ValueError('Need to ingest a bundle that creates '
+                                     'writers in order to downgrade the assets'
+                                     ' db.')
             bundle.ingest(
                 environ,
                 asset_db_writer,
@@ -423,6 +443,14 @@ def _make_bundle_core():
                 pth.data_path([name, timestr], environ=environ),
             )
 
+            for version in sorted(set(assets_versions), reverse=True):
+                version_path = wd.getpath(*asset_db_relative(
+                    name, timestr, environ=environ, db_version=version,
+                ))
+                with working_file(version_path) as wf:
+                    shutil.copy2(assets_db_path, wf.path)
+                    downgrade(wf.path, version)
+
     def most_recent_data(bundle_name, timestamp, environ=None):
         """Get the path to the most recent data after ``date``for the
         given bundle.

zipline/data/us_equity_pricing.py

@@ -54,12 +54,11 @@ from zipline.utils.calendars import get_calendar
 from zipline.utils.functional import apply
 from zipline.utils.preprocess import call
 from zipline.utils.input_validation import (
-    coerce_string,
     preprocess,
     expect_element,
     verify_indices_all_unique,
 )
-from zipline.utils.sqlite_utils import group_into_chunks
+from zipline.utils.sqlite_utils import group_into_chunks, coerce_string_to_conn
 from zipline.utils.memoize import lazyval
 from zipline.utils.cli import maybe_show_progress
 from ._equities import _compute_row_slices, _read_bcolz_data
@@ -1228,7 +1227,7 @@ class SQLiteAdjustmentReader(object):
     :class:`zipline.data.us_equity_pricing.SQLiteAdjustmentWriter`
     """
 
-    @preprocess(conn=coerce_string(sqlite3.connect))
+    @preprocess(conn=coerce_string_to_conn)
     def __init__(self, conn):
         self.conn = conn
 

zipline/finance/trading.py

@@ -99,7 +99,7 @@ class TradingEnvironment(object):
         self.exchange_tz = exchange_tz
 
         if isinstance(asset_db_path, string_types):
-            asset_db_path = 'sqlite:///%s' % asset_db_path
+            asset_db_path = 'sqlite:///' + asset_db_path
             self.engine = engine = create_engine(asset_db_path)
         else:
             self.engine = engine = asset_db_path

zipline/pipeline/__init__.py

@@ -35,9 +35,6 @@ def engine_from_files(daily_bar_path,
         memory consumption.  Default is False
     """
     loader = USEquityPricingLoader.from_files(daily_bar_path, adjustments_path)
-
-    if not asset_db_path.startswith("sqlite:"):
-        asset_db_path = "sqlite:///" + asset_db_path
     asset_finder = AssetFinder(asset_db_path)
     if warmup_assets:
         results = asset_finder.retrieve_all(asset_finder.sids)

zipline/utils/cache.py

@@ -290,11 +290,7 @@ class working_file(object):
     def _commit(self):
         """Sync the temporary file to the final path.
         """
-        self._tmpfile.close()
-        move(self._name, self._final_path)
-
-    def __getattr__(self, attr):
-        return getattr(self._tmpfile, attr)
+        move(self.path, self._final_path)
 
     def __enter__(self):
         self._tmpfile.__enter__()

zipline/utils/sqlite_utils.py

@@ -12,9 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import sqlite3
 
+import sqlalchemy as sa
 from six.moves import range
 
+from .input_validation import coerce_string
+
 SQLITE_MAX_VARIABLE_NUMBER = 998
 
 
@@ -22,3 +26,9 @@ def group_into_chunks(items, chunk_size=SQLITE_MAX_VARIABLE_NUMBER):
     items = list(items)
     return [items[x:x+chunk_size]
             for x in range(0, len(items), chunk_size)]
+
+
+coerce_string_to_conn = coerce_string(sqlite3.connect)
+coerce_string_to_eng = coerce_string(
+    lambda s: sa.create_engine('sqlite:///' + s)
+)