本文整理汇总了Python中zipline.testing.predicates.assert_equal函数的典型用法代码示例。如果您正苦于以下问题:Python assert_equal函数的具体用法?Python assert_equal怎么用?Python assert_equal使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了assert_equal函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_reversability
def test_reversability(self):
class F(Filter):
inputs = ()
window_length = 0
missing_value = False
f = F()
column_data = array(
[[True, f.missing_value],
[True, f.missing_value],
[True, True]],
dtype=bool,
)
assert_equal(f.postprocess(column_data.ravel()), column_data.ravel())
# only include the non-missing data
pipeline_output = pd.Series(
data=True,
index=pd.MultiIndex.from_arrays([
[pd.Timestamp('2014-01-01'),
pd.Timestamp('2014-01-02'),
pd.Timestamp('2014-01-03'),
pd.Timestamp('2014-01-03')],
[0, 0, 0, 1],
]),
)
assert_equal(
f.to_workspace_value(pipeline_output, pd.Index([0, 1])),
column_data,
)
开发者ID:mannau,项目名称:zipline,代码行数:32,代码来源:test_filter.py
示例2: test_bundle
def test_bundle(self):
environ = {
'CSVDIR': test_resource_path('csvdir_samples', 'csvdir')
}
ingest('csvdir', environ=environ)
bundle = load('csvdir', environ=environ)
sids = 0, 1, 2, 3
assert_equal(set(bundle.asset_finder.sids), set(sids))
for equity in bundle.asset_finder.retrieve_all(sids):
assert_equal(equity.start_date, self.asset_start, msg=equity)
assert_equal(equity.end_date, self.asset_end, msg=equity)
sessions = self.calendar.all_sessions
actual = bundle.equity_daily_bar_reader.load_raw_arrays(
self.columns,
sessions[sessions.get_loc(self.asset_start, 'bfill')],
sessions[sessions.get_loc(self.asset_end, 'ffill')],
sids,
)
expected_pricing, expected_adjustments = self._expected_data(
bundle.asset_finder,
)
assert_equal(actual, expected_pricing, array_decimal=2)
adjs_for_cols = bundle.adjustment_reader.load_pricing_adjustments(
self.columns,
sessions,
pd.Index(sids),
)
assert_equal([sorted(adj.keys()) for adj in adjs_for_cols],
expected_adjustments)
开发者ID:quantopian,项目名称:zipline,代码行数:34,代码来源:test_csvdir.py
示例3: do_checks
def do_checks(cls, colnames):
specialized = cls.specialize(domain)
# Specializations should be memoized.
self.assertIs(specialized, cls.specialize(domain))
# Specializations should have the same name.
assert_equal(specialized.__name__, cls.__name__)
self.assertIs(specialized.domain, domain)
for attr in colnames:
original = getattr(cls, attr)
new = getattr(specialized, attr)
# We should get a new column from the specialization, which
# should be the same object that we would get from specializing
# the original column.
self.assertIsNot(original, new)
self.assertIs(new, original.specialize(domain))
# Columns should be bound to their respective datasets.
self.assertIs(original.dataset, cls)
self.assertIs(new.dataset, specialized)
# The new column should have the domain of the specialization.
assert_equal(new.domain, domain)
# Names, dtypes, and missing_values should match.
assert_equal(original.name, new.name)
assert_equal(original.dtype, new.dtype)
assert_equal(original.missing_value, new.missing_value)
开发者ID:jiapei100,项目名称:zipline,代码行数:32,代码来源:test_domain.py
示例4: test_mixed_generics
def test_mixed_generics(self):
"""
Test that we can run pipelines with mixed generic/non-generic terms.
This test is a regression test for failures encountered during
development where having a mix of generic and non-generic columns in
the term graph caused bugs in our extra row accounting.
"""
USTestingDataSet = TestingDataSet.specialize(US_EQUITIES)
base_terms = {
'sum3_generic': Sum.create(TestingDataSet.float_col, 3),
'sum3_special': Sum.create(USTestingDataSet.float_col, 3),
'sum10_generic': Sum.create(TestingDataSet.float_col, 10),
'sum10_special': Sum.create(USTestingDataSet.float_col, 10),
}
def run(ts):
pipe = Pipeline(ts, domain=US_EQUITIES)
start = self.trading_days[-5]
end = self.trading_days[-1]
return self.run_pipeline(pipe, start, end)
base_result = run(base_terms)
for subset in powerset(base_terms):
subset_terms = {t: base_terms[t] for t in subset}
result = run(subset_terms).sort_index(axis=1)
expected = base_result[list(subset)].sort_index(axis=1)
assert_equal(result, expected)
开发者ID:jiapei100,项目名称:zipline,代码行数:29,代码来源:test_domain.py
示例5: test_stock_dividends
def test_stock_dividends(self):
sids = np.arange(5)
dates = self.trading_calendar.all_sessions.tz_convert(None)
def T(n):
return dates[n]
sort_key = ['sid', 'ex_date', 'payment_sid', 'ratio']
input_ = pd.DataFrame(
[[0, T(0), 1.5, 1],
[0, T(1), 0.5, 2],
# the same asset has two stock dividends for different assets on
# the same day
[1, T(0), 1, 2],
[1, T(0), 1.2, 3]],
columns=['sid', 'ex_date', 'ratio', 'payment_sid'],
).sort_values(sort_key)
# give every extra date field a unique date so that we can make sure
# they appear unchanged in the dividends payouts
ix = 0
for col in 'declared_date', 'record_date', 'pay_date':
extra_dates = dates[ix:ix + len(input_)]
ix += len(input_)
input_[col] = extra_dates
self.writer_without_pricing(dates, sids).write(stock_dividends=input_)
dfs = self.component_dataframes()
output = dfs.pop('stock_dividend_payouts').sort_values(sort_key)
self.assert_all_empty(dfs)
assert_equal(output, input_)
开发者ID:barrygolden,项目名称:zipline,代码行数:34,代码来源:test_adjustments.py
示例6: _empty_ingest
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]
开发者ID:JasonGiedymin,项目名称:zipline,代码行数:32,代码来源:test_core.py
示例7: test_session_closes_in_range
def test_session_closes_in_range(self):
found_closes = self.calendar.session_closes_in_range(
self.answers.index[0],
self.answers.index[-1],
)
assert_equal(found_closes, self.answers['market_close'])
开发者ID:zhou,项目名称:zipline,代码行数:7,代码来源:test_trading_calendar.py
示例8: test_repr
def test_repr(self):
assert_equal(
repr(self.Term().alias('ayy lmao')),
"Aliased%s(Term(...), name='ayy lmao')" % (
self.Term.__base__.__name__,
),
)
开发者ID:FranSal,项目名称:zipline,代码行数:7,代码来源:test_alias.py
示例9: test_reversability
def test_reversability(self, dtype_):
class F(Factor):
inputs = ()
dtype = dtype_
window_length = 0
f = F()
column_data = array(
[[0, f.missing_value],
[1, f.missing_value],
[2, 3]],
dtype=dtype_,
)
assert_equal(f.postprocess(column_data.ravel()), column_data.ravel())
# only include the non-missing data
pipeline_output = pd.Series(
data=array([0, 1, 2, 3], dtype=dtype_),
index=pd.MultiIndex.from_arrays([
[pd.Timestamp('2014-01-01'),
pd.Timestamp('2014-01-02'),
pd.Timestamp('2014-01-03'),
pd.Timestamp('2014-01-03')],
[0, 0, 0, 1],
]),
)
assert_equal(
f.to_workspace_value(pipeline_output, pd.Index([0, 1])),
column_data,
)
开发者ID:SJCosgrove,项目名称:quantopianresearch,代码行数:32,代码来源:test_factor.py
示例10: test_multiple_qtrs_requested
def test_multiple_qtrs_requested(self):
dataset1 = QuartersEstimates(1)
dataset2 = QuartersEstimates(2)
engine = SimplePipelineEngine(
lambda x: self.loader,
self.trading_days,
self.asset_finder,
)
results = engine.run_pipeline(
Pipeline(
merge([{c.name + '1': c.latest for c in dataset1.columns},
{c.name + '2': c.latest for c in dataset2.columns}])
),
start_date=self.trading_days[0],
end_date=self.trading_days[-1],
)
q1_columns = [col.name + '1' for col in self.columns]
q2_columns = [col.name + '2' for col in self.columns]
# We now expect a column for 1 quarter out and a column for 2
# quarters out for each of the dataset columns.
assert_equal(sorted(np.array(q1_columns + q2_columns)),
sorted(results.columns.values))
assert_equal(self.expected_out.sort(axis=1),
results.xs(0, level=1).sort(axis=1))
开发者ID:kongscn,项目名称:zipline,代码行数:26,代码来源:test_quarters_estimates.py
示例11: test_v5_to_v4_selects_most_recent_ticker
def test_v5_to_v4_selects_most_recent_ticker(self):
T = pd.Timestamp
AssetDBWriter(self.engine).write(
equities=pd.DataFrame(
[['A', 'A', T('2014-01-01'), T('2014-01-02')],
['B', 'B', T('2014-01-01'), T('2014-01-02')],
# these two are both ticker sid 2
['B', 'C', T('2014-01-03'), T('2014-01-04')],
['C', 'C', T('2014-01-01'), T('2014-01-02')]],
index=[0, 1, 2, 2],
columns=['symbol', 'asset_name', 'start_date', 'end_date'],
),
)
downgrade(self.engine, 4)
metadata = sa.MetaData(self.engine)
metadata.reflect()
def select_fields(r):
return r.sid, r.symbol, r.asset_name, r.start_date, r.end_date
expected_data = {
(0, 'A', 'A', T('2014-01-01').value, T('2014-01-02').value),
(1, 'B', 'B', T('2014-01-01').value, T('2014-01-02').value),
(2, 'B', 'C', T('2014-01-01').value, T('2014-01-04').value),
}
actual_data = set(map(
select_fields,
sa.select(metadata.tables['equities'].c).execute(),
))
assert_equal(expected_data, actual_data)
开发者ID:jeyoor,项目名称:zipline,代码行数:32,代码来源:test_assets.py
示例12: _check_bundles
def _check_bundles(self, names):
assert_equal(set(self.bundles.keys()), names)
for name in names:
self.unregister(name)
assert_false(self.bundles)
开发者ID:nathanwolfe,项目名称:zipline-minute-bars,代码行数:7,代码来源:test_core.py
示例13: test_parse_namespaces
def test_parse_namespaces(self):
n = Namespace()
create_args(
[
"first.second.a=blah1",
"first.second.b=blah2",
"first.third=blah3",
"second.a=blah4",
"second.b=blah5",
],
n
)
assert_equal(n.first.second.a, 'blah1')
assert_equal(n.first.second.b, 'blah2')
assert_equal(n.first.third, 'blah3')
assert_equal(n.second.a, 'blah4')
assert_equal(n.second.b, 'blah5')
n = Namespace()
msg = "Conflicting assignments at namespace level 'second'"
with assert_raises_str(ValueError, msg):
create_args(
[
"first.second.a=blah1",
"first.second.b=blah2",
"first.second=blah3",
],
n
)
开发者ID:barrygolden,项目名称:zipline,代码行数:32,代码来源:test_cmdline.py
示例14: test_fso_expected_with_talib
def test_fso_expected_with_talib(self):
"""
Test the output that is returned from the fast stochastic oscillator
is the same as that from the ta-lib STOCHF function.
"""
window_length = 14
nassets = 6
closes = np.random.random_integers(1, 6, size=(50, nassets))*1.0
highs = np.random.random_integers(4, 6, size=(50, nassets))*1.0
lows = np.random.random_integers(1, 3, size=(50, nassets))*1.0
expected_out_k = []
for i in range(nassets):
e = talib.STOCHF(
high=highs[:, i],
low=lows[:, i],
close=closes[:, i],
fastk_period=window_length,
)
expected_out_k.append(e[0][-1])
expected_out_k = np.array(expected_out_k)
today = pd.Timestamp('2015')
out = np.empty(shape=(nassets,), dtype=np.float)
assets = np.arange(nassets, dtype=np.float)
fso = FastStochasticOscillator()
fso.compute(
today, assets, out, closes, lows, highs
)
assert_equal(out, expected_out_k)
开发者ID:4ever911,项目名称:zipline,代码行数:33,代码来源:test_technical.py
示例15: test_reversability_int64
def test_reversability_int64(self):
class F(Classifier):
inputs = ()
window_length = 0
dtype = int64_dtype
missing_value = -1
f = F()
column_data = np.array(
[[0, f.missing_value],
[1, f.missing_value],
[2, 3]],
)
assert_equal(f.postprocess(column_data.ravel()), column_data.ravel())
# only include the non-missing data
pipeline_output = pd.Series(
data=[0, 1, 2, 3],
index=pd.MultiIndex.from_arrays([
[pd.Timestamp('2014-01-01'),
pd.Timestamp('2014-01-02'),
pd.Timestamp('2014-01-03'),
pd.Timestamp('2014-01-03')],
[0, 0, 0, 1],
]),
dtype=int64_dtype,
)
assert_equal(
f.to_workspace_value(pipeline_output, pd.Index([0, 1])),
column_data,
)
开发者ID:barrygolden,项目名称:zipline,代码行数:33,代码来源:test_classifier.py
示例16: test_inheritance
def test_inheritance(self):
class Parent(DataSetFamily):
extra_dims = [
('dim_0', {'a', 'b', 'c'}),
('dim_1', {'d', 'e', 'f'}),
]
column_0 = Column('f8')
column_1 = Column('?')
class Child(Parent):
column_2 = Column('O')
column_3 = Column('i8', -1)
assert_is_subclass(Child, Parent)
assert_equal(Child.extra_dims, Parent.extra_dims)
ChildSlice = Child.slice(dim_0='a', dim_1='d')
expected_child_slice_columns = frozenset({
ChildSlice.column_0,
ChildSlice.column_1,
ChildSlice.column_2,
ChildSlice.column_3,
})
assert_equal(ChildSlice.columns, expected_child_slice_columns)
开发者ID:quantopian,项目名称:zipline,代码行数:26,代码来源:test_multidimensional_dataset.py
示例17: make_equity_info
def make_equity_info(cls):
out = pd.concat(
[
# 15 assets on each exchange. Each asset lives for 5 days.
# A new asset starts each day.
make_rotating_equity_info(
num_assets=20,
first_start=cls.START_DATE,
frequency=get_calendar(exchange).day,
periods_between_starts=1,
# NOTE: The asset_lifetime parameter name is a bit
# misleading. It determines the number of trading
# days between each asset's start_date and end_date,
# so assets created with this method actual "live"
# for (asset_lifetime + 1) days. But, since pipeline
# doesn't show you an asset the day it IPOs, this
# number matches the number of days that each asset
# should appear in a pipeline output.
asset_lifetime=5,
exchange=exchange,
)
for exchange in cls.EXCHANGE_INFO.exchange
],
ignore_index=True,
)
assert_equal(out.end_date.max(), cls.END_DATE)
return out
开发者ID:barrygolden,项目名称:zipline,代码行数:27,代码来源:test_international_markets.py
示例18: check_roundtrip
def check_roundtrip(arr):
assert_equal(
arr.as_string_array(),
LabelArray(
arr.as_string_array(),
arr.missing_value,
).as_string_array(),
)
开发者ID:FranSal,项目名称:zipline,代码行数:8,代码来源:test_labelarray.py
示例19: check_equivalent_terms
def check_equivalent_terms(self, terms):
self.assertTrue(len(terms) > 1, "Need at least two terms to compare")
pipe = Pipeline(terms)
start, end = self.trading_days[[-10, -1]]
results = self.pipeline_engine.run_pipeline(pipe, start, end)
first_column = results.iloc[:, 0]
for name in terms:
assert_equal(results.loc[:, name], first_column, check_names=False)
开发者ID:SJCosgrove,项目名称:quantopianresearch,代码行数:9,代码来源:test_factor.py
示例20: init_instance_fixtures
def init_instance_fixtures(self):
super(MetricsSetCoreTestCase, self).init_instance_fixtures()
self.metrics_sets, self.register, self.unregister, self.load = (
_make_metrics_set_core()
)
# make sure this starts empty
assert_equal(self.metrics_sets, mappingproxy({}))
开发者ID:barrygolden,项目名称:zipline,代码行数:9,代码来源:test_core.py
注:本文中的zipline.testing.predicates.assert_equal函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论