From 6d473ce23fb011122888bf76d595532f6b222798 Mon Sep 17 00:00:00 2001 From: Richard Frank Date: Tue, 23 Aug 2016 10:07:58 -0400 Subject: [PATCH 1/6] MAINT: Clean up usage of engine versus connection --- tests/test_assets.py | 6 +-- zipline/assets/asset_db_migrations.py | 56 +++++++++++++-------------- 2 files changed, 31 insertions(+), 31 deletions(-) diff --git a/tests/test_assets.py b/tests/test_assets.py index 3668a0cc..50842a06 100644 --- a/tests/test_assets.py +++ b/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) diff --git a/zipline/assets/asset_db_migrations.py b/zipline/assets/asset_db_migrations.py index 72d24b78..550e096c 100644 --- a/zipline/assets/asset_db_migrations.py +++ b/zipline/assets/asset_db_migrations.py @@ -21,39 +21,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.connect() 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 +96,10 @@ def downgrades(src): @do(op.setitem(_downgrade_methods, destination)) @wraps(f) - def wrapper(op, engine, version_info_table): + def wrapper(op, conn, version_info_table): version_info_table.delete().execute() # clear the version f(op) - write_version_info(engine, version_info_table, destination) + write_version_info(conn, version_info_table, destination) return wrapper return _ From bc87ea4efbfe6ff3c0f38e1c0a32ef0bb9df10dd Mon Sep 17 00:00:00 2001 From: Richard Frank Date: Tue, 23 Aug 2016 10:17:33 -0400 Subject: [PATCH 2/6] MAINT: Consolidate coercion to sqlite conn/eng --- zipline/assets/asset_db_migrations.py | 3 +++ zipline/assets/asset_writer.py | 5 +++-- zipline/assets/assets.py | 10 ++++------ zipline/data/us_equity_pricing.py | 5 ++--- zipline/finance/trading.py | 2 +- zipline/pipeline/__init__.py | 3 --- zipline/utils/sqlite_utils.py | 10 ++++++++++ 7 files changed, 23 insertions(+), 15 deletions(-) diff --git a/zipline/assets/asset_db_migrations.py b/zipline/assets/asset_db_migrations.py index 550e096c..82f9d147 100644 --- a/zipline/assets/asset_db_migrations.py +++ b/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. diff --git a/zipline/assets/asset_writer.py b/zipline/assets/asset_writer.py index 17d61b2f..50e9070d 100644 --- a/zipline/assets/asset_writer.py +++ b/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, diff --git a/zipline/assets/assets.py b/zipline/assets/assets.py index 8ca34629..caf6727c 100644 --- a/zipline/assets/assets.py +++ b/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 = "" + @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: diff --git a/zipline/data/us_equity_pricing.py b/zipline/data/us_equity_pricing.py index 71579ca5..29acdf0d 100644 --- a/zipline/data/us_equity_pricing.py +++ b/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 diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 79fe5b75..4b13b8e1 100644 --- a/zipline/finance/trading.py +++ b/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 diff --git a/zipline/pipeline/__init__.py b/zipline/pipeline/__init__.py index 8d2a412f..a169256b 100644 --- a/zipline/pipeline/__init__.py +++ b/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) diff --git a/zipline/utils/sqlite_utils.py b/zipline/utils/sqlite_utils.py index 439db2f2..17bfc9a5 100644 --- a/zipline/utils/sqlite_utils.py +++ b/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) +) From 418e8901f4baccb3ad72b9066b18ea5a9a5e0d45 Mon Sep 17 00:00:00 2001 From: Richard Frank Date: Tue, 23 Aug 2016 11:19:42 -0400 Subject: [PATCH 3/6] MAINT: Put downgrade in transaction --- zipline/assets/asset_db_migrations.py | 4 ++-- zipline/assets/asset_writer.py | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/zipline/assets/asset_db_migrations.py b/zipline/assets/asset_db_migrations.py index 82f9d147..2b1431c5 100644 --- a/zipline/assets/asset_db_migrations.py +++ b/zipline/assets/asset_db_migrations.py @@ -24,7 +24,7 @@ def downgrade(engine, desired_version): """ # Check the version of the db at the engine - with engine.connect() as conn: + with engine.begin() as conn: metadata = sa.MetaData(conn) metadata.reflect() version_info_table = metadata.tables['version_info'] @@ -100,7 +100,7 @@ def downgrades(src): @do(op.setitem(_downgrade_methods, destination)) @wraps(f) def wrapper(op, conn, version_info_table): - version_info_table.delete().execute() # clear the version + conn.execute(version_info_table.delete()) # clear the version f(op) write_version_info(conn, version_info_table, destination) diff --git a/zipline/assets/asset_writer.py b/zipline/assets/asset_writer.py index 50e9070d..050904ee 100644 --- a/zipline/assets/asset_writer.py +++ b/zipline/assets/asset_writer.py @@ -442,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( @@ -457,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, ) From 419bd1e3b52e5c7d31d2af75bccc35f37323e804 Mon Sep 17 00:00:00 2001 From: Richard Frank Date: Tue, 23 Aug 2016 12:16:50 -0400 Subject: [PATCH 4/6] DEV: zipline ingest can downgrade the assets db This lets us publish an "old" db version for the most recent release of zipline, using the latest code base. --- tests/data/bundles/test_core.py | 107 ++++++++++++++++++++++++++----- zipline/__main__.py | 21 +++--- zipline/data/bundles/__init__.py | 2 + zipline/data/bundles/core.py | 45 ++++++++++--- zipline/utils/cache.py | 6 +- 5 files changed, 143 insertions(+), 38 deletions(-) diff --git a/tests/data/bundles/test_core.py b/tests/data/bundles/test_core.py index 870945b4..e6cf65ad 100644 --- a/tests/data/bundles/test_core.py +++ b/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: diff --git a/zipline/__main__.py b/zipline/__main__.py index 3dde2b59..93c26bf4 100644 --- a/zipline/__main__.py +++ b/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 [""]: - print("%s %s" % (bundle, timestamp)) + click.echo("%s %s" % (bundle, timestamp)) if __name__ == '__main__': diff --git a/zipline/data/bundles/__init__.py b/zipline/data/bundles/__init__.py index beea33bf..40746c6c 100644 --- a/zipline/data/bundles/__init__.py +++ b/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', diff --git a/zipline/data/bundles/core.py b/zipline/data/bundles/core.py index 33b1acd0..00c30a4e 100644 --- a/zipline/data/bundles/core.py +++ b/zipline/data/bundles/core.py @@ -20,9 +20,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 +33,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 +84,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 +123,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', @@ -328,6 +341,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 +354,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 +405,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 +424,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 +442,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. diff --git a/zipline/utils/cache.py b/zipline/utils/cache.py index 21aee7d4..427a67d6 100644 --- a/zipline/utils/cache.py +++ b/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__() From cf6ec9fd73e69fef61d1d7df29ed7f5a9071efe4 Mon Sep 17 00:00:00 2001 From: Richard Frank Date: Tue, 23 Aug 2016 16:21:20 -0400 Subject: [PATCH 5/6] BUG: More python3 compat --- zipline/data/bundles/core.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/zipline/data/bundles/core.py b/zipline/data/bundles/core.py index 00c30a4e..7bb1b2f8 100644 --- a/zipline/data/bundles/core.py +++ b/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 ( @@ -295,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 From abdfe0265d6e352fc679793bae67745d9b4d425a Mon Sep 17 00:00:00 2001 From: Richard Frank Date: Wed, 24 Aug 2016 09:59:48 -0400 Subject: [PATCH 6/6] BLD: alembic is now a production requirement --- conda/alembic/bld.bat | 8 +++++ conda/alembic/build.sh | 9 +++++ conda/alembic/meta.yaml | 73 ++++++++++++++++++++++++++++++++++++++++ etc/requirements.txt | 2 ++ etc/requirements_dev.txt | 3 -- 5 files changed, 92 insertions(+), 3 deletions(-) create mode 100644 conda/alembic/bld.bat create mode 100644 conda/alembic/build.sh create mode 100644 conda/alembic/meta.yaml diff --git a/conda/alembic/bld.bat b/conda/alembic/bld.bat new file mode 100644 index 00000000..87b1481d --- /dev/null +++ b/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. diff --git a/conda/alembic/build.sh b/conda/alembic/build.sh new file mode 100644 index 00000000..4d7fc032 --- /dev/null +++ b/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. diff --git a/conda/alembic/meta.yaml b/conda/alembic/meta.yaml new file mode 100644 index 00000000..f0fa7254 --- /dev/null +++ b/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 diff --git a/etc/requirements.txt b/etc/requirements.txt index 6a84565e..6e672d39 100644 --- a/etc/requirements.txt +++ b/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 diff --git a/etc/requirements_dev.txt b/etc/requirements_dev.txt index b88bd61b..38a44ae2 100644 --- a/etc/requirements_dev.txt +++ b/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