本文整理汇总了Python中toolz.complement函数的典型用法代码示例。如果您正苦于以下问题：Python complement函数的具体用法？Python complement怎么用？Python complement使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了complement函数的10个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: most_recent_data
def most_recent_data(bundle_name, timestamp, environ=None):
"""Get the path to the most recent data after ``date``for the
given bundle.
Parameters
----------
bundle_name : str
The name of the bundle to lookup.
timestamp : datetime
The timestamp to begin searching on or before.
environ : dict, optional
An environment dict to forward to zipline_root.
"""
if bundle_name not in bundles:
raise UnknownBundle(bundle_name)
try:
candidates = os.listdir(pth.data_path([bundle_name], environ=environ))
return pth.data_path(
[bundle_name, max(filter(complement(pth.hidden), candidates), key=from_bundle_ingest_dirname)],
environ=environ,
)
except (ValueError, OSError) as e:
if getattr(e, "errno", ~errno.ENOENT) != errno.ENOENT:
raise
raise ValueError("no data for bundle %r on or before %s" % (bundle_name, timestamp))
开发者ID:umuzungu，项目名称:zipline，代码行数:26，代码来源:core.py
示例2: _expect_element
def _expect_element(collection):
template = (
"%(funcname)s() expected a value in {collection} "
"for argument '%(argname)s', but got %(actual)s instead."
).format(collection=collection)
return make_check(
ValueError,
template,
complement(op.contains(collection)),
repr,
)
开发者ID:louiekang，项目名称:zipline，代码行数:11，代码来源:input_validation.py
示例3: _expect_element
def _expect_element(collection):
if isinstance(collection, (set, frozenset)):
# Special case the error message for set and frozen set to make it
# less verbose.
collection_for_error_message = tuple(sorted(collection))
else:
collection_for_error_message = collection
template = (
"%(funcname)s() expected a value in {collection} "
"for argument '%(argname)s', but got %(actual)s instead."
).format(collection=collection_for_error_message)
return make_check(
ValueError,
template,
complement(op.contains(collection)),
repr,
)
开发者ID:kongscn，项目名称:zipline，代码行数:18，代码来源:input_validation.py
示例4: make_while_loop_test_expr
def make_while_loop_test_expr(loop_body_instrs):
"""
Make an expression in the context of a while-loop test.
Code of the form::
while <expr>:
<body>
generates a POP_JUMP_IF_FALSE for the loop test, while code of the form::
while not <expr>:
<body>
generates a POP_JUMP_IF_TRUE for the loop test.
Code of the form::
while True:
<body>
generates no jumps at all.
"""
bottom_of_loop = loop_body_instrs[-1]
is_jump_to_bottom = compose(op.is_(bottom_of_loop), op.attrgetter('arg'))
# Consume instructions until we find a jump to the bottom of the loop.
test_builders = deque(
popwhile(complement(is_jump_to_bottom), loop_body_instrs, side='left')
)
# If we consumed the entire loop body without finding a jump, assume this
# is a while True loop. Return the rest of the instructions as the loop
# body.
if not loop_body_instrs:
return ast.NameConstant(value=True), test_builders
# Top of the body is either a POP_JUMP_IF_TRUE or POP_JUMP_IF_FALSE.
jump = loop_body_instrs.popleft()
expr = make_expr(test_builders)
if isinstance(jump, instrs.POP_JUMP_IF_TRUE):
return ast.UnaryOp(op=ast.Not(), operand=expr), loop_body_instrs
else:
return expr, loop_body_instrs
开发者ID:jcrist，项目名称:codetransformer，代码行数:43，代码来源:decompiler.py
示例5: most_recent_data
def most_recent_data(bundle_name, timestamp, environ=None):
"""Get the path to the most recent data after ``date``for the
given bundle.
Parameters
----------
bundle_name : str
The name of the bundle to lookup.
timestamp : datetime
The timestamp to begin searching on or before.
environ : dict, optional
An environment dict to forward to catalyst_root.
"""
if bundle_name not in bundles:
raise UnknownBundle(bundle_name)
try:
candidates = os.listdir(
pth.data_path([bundle_name], environ=environ),
)
return pth.data_path(
[bundle_name,
max(
filter(complement(pth.hidden), candidates),
key=from_bundle_ingest_dirname,
)],
environ=environ,
)
except (ValueError, OSError) as e:
if getattr(e, 'errno', errno.ENOENT) != errno.ENOENT:
raise
raise ValueError(
'no data for bundle {bundle!r} on or before {timestamp}\n'
'maybe you need to run: $ catalyst ingest -b {bundle}'.format(
bundle=bundle_name,
timestamp=timestamp,
),
)
开发者ID:zhoukalex，项目名称:catalyst，代码行数:38，代码来源:core.py
示例6: get_dirs_and_files_in_path
def get_dirs_and_files_in_path(path):
# filter function
def isdir(a): return os.path.isdir(a)
# gives the opposite results as above
not_isdir = toolz.complement(isdir)
if not path and platform.system() == 'Windows':
import win32api
drives = win32api.GetLogicalDriveStrings()
drives = [d for d in drives.split('\000') if d]
return drives
elif os.path.exists(path):
r = os.listdir(path)
# 2x acccess means I have to remove the generator
f = [os.path.join(path, a) for a in r]
dirs = filter(isdir, f)
files = filter(not_isdir, f)
else:
try:
head, tail = os.path.split(path)
r = os.listdir(head)
filtered_everything = filter(lambda a: a.startswith(tail), r)
# because this was accesssed twice, I needed to remove the generator
filtered_everything = [os.path.join(head, a) for a in filtered_everything]
dirs = filter(isdir, filtered_everything)
files = filter(not_isdir, filtered_everything)
except Exception as e:
print('{0} doesn\'t even exist you stupid'.format(head))
return None
result = (sorted(list(toolz.take(100, dirs))),
sorted(list(toolz.take(100, files))))
return result
开发者ID:wingillis，项目名称:sarnaandra，代码行数:36，代码来源:helpers.py
示例7: complement
from zipline.utils.numpy_utils import repeat_last_axis
AD_FIELD_NAME = 'asof_date'
TS_FIELD_NAME = 'timestamp'
SID_FIELD_NAME = 'sid'
valid_deltas_node_types = (
bz.expr.Field,
bz.expr.ReLabel,
bz.expr.Symbol,
)
traversable_nodes = (
bz.expr.Field,
bz.expr.Label,
)
is_invalid_deltas_node = complement(flip(isinstance, valid_deltas_node_types))
getname = op.attrgetter('__name__')
class _ExprRepr(object):
"""Box for repring expressions with the str of the expression.
Parameters
----------
expr : Expr
The expression to box for repring.
"""
__slots__ = 'expr',
def __init__(self, expr):
self.expr = expr
开发者ID:ChinaQuants，项目名称:zipline，代码行数:31，代码来源:blaze.py
示例8: clean
def clean(name,
before=None,
after=None,
keep_last=None,
environ=os.environ):
"""Clean up data that was created with ``ingest`` or
``$ python -m zipline ingest``
Parameters
----------
name : str
The name of the bundle to remove data for.
before : datetime, optional
Remove data ingested before this date.
This argument is mutually exclusive with: keep_last
after : datetime, optional
Remove data ingested after this date.
This argument is mutually exclusive with: keep_last
keep_last : int, optional
Remove all but the last ``keep_last`` ingestions.
This argument is mutually exclusive with:
before
after
environ : mapping, optional
The environment variables. Defaults of os.environ.
Returns
-------
cleaned : set[str]
The names of the runs that were removed.
Raises
------
BadClean
Raised when ``before`` and or ``after`` are passed with
``keep_last``. This is a subclass of ``ValueError``.
"""
try:
all_runs = sorted(
filter(
complement(pth.hidden),
os.listdir(pth.data_path([name], environ=environ)),
),
key=from_bundle_ingest_dirname,
)
except OSError as e:
if e.errno != errno.ENOENT:
raise
raise UnknownBundle(name)
if ((before is not None or after is not None) and
keep_last is not None):
raise BadClean(before, after, keep_last)
if keep_last is None:
def should_clean(name):
dt = from_bundle_ingest_dirname(name)
return (
(before is not None and dt < before) or
(after is not None and dt > after)
)
else:
last_n_dts = set(all_runs[-keep_last:])
def should_clean(name):
return name not in last_n_dts
cleaned = set()
for run in all_runs:
if should_clean(run):
path = pth.data_path([name, run], environ=environ)
shutil.rmtree(path)
cleaned.add(path)
return cleaned
开发者ID:RoyHsiao，项目名称:zipline，代码行数:75，代码来源:core.py
示例9: ingest
def ingest(name,
environ=os.environ,
timestamp=None,
assets_versions=(),
show_progress=False,
writer="bcolz"):
"""Ingest data for a given bundle.
Parameters
----------
name : str
The name of the bundle.
environ : mapping, optional
The environment variables. By default this is os.environ.
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.
incremental : bool, optional
Tell the ingest function to incremental ingest
"""
try:
bundle = bundles[name]
except KeyError:
raise UnknownBundle(name)
calendar = get_calendar(bundle.calendar_name)
start_session = bundle.start_session
end_session = bundle.end_session
if start_session is None or start_session < calendar.first_session:
start_session = calendar.first_session
if end_session is None or end_session > calendar.last_session:
end_session = calendar.last_session
try:
candidates = os.listdir(
pth.data_path([name], environ=environ),
)
timestr = max(
filter(complement(pth.hidden), candidates),
key=from_bundle_ingest_dirname,
)
except Exception:
if timestamp is None:
timestamp = pd.Timestamp.utcnow()
timestamp = timestamp.tz_convert('utc').tz_localize(None)
timestr = to_bundle_ingest_dirname(timestamp)
cachepath = cache_path(name, environ=environ)
pth.ensure_directory(pth.data_path([name, timestr], environ=environ))
pth.ensure_directory(cachepath)
with dataframe_cache(cachepath, clean_on_failure=False) as cache, \
ExitStack() as stack:
# we use `cleanup_on_failure=False` so that we don't purge the
# cache directory if the load fails in the middle
if bundle.create_writers:
wd = stack.enter_context(working_dir(
pth.data_path([], environ=environ))
)
daily_bars_path = daily_equity_path(name, timestr, environ=environ)
pth.ensure_directory(
daily_bars_path
)
daily_bar_writer = BcolzDailyBarWriter(
daily_bars_path,
calendar,
start_session,
end_session,
)
# Do an empty write to ensure that the daily ctables exist
# when we create the SQLiteAdjustmentWriter below. The
# SQLiteAdjustmentWriter needs to open the daily ctables so
# that it can compute the adjustment ratios for the dividends.
daily_bar_writer.write(())
minute_bars_path = minute_equity_path(name, timestr, environ=environ)
pth.ensure_directory(
minute_bars_path
)
if writer == "rocksdb":
minute_bar_writer = RocksdbMinuteBarWriter(
minute_bars_path,
calendar,
start_session,
end_session,
minutes_per_day=bundle.minutes_per_day,
)
else:
minute_bar_writer = BcolzMinuteBarWriter(
minute_bars_path,
calendar,
start_session,
end_session,
minutes_per_day=bundle.minutes_per_day,
#.........这里部分代码省略.........
开发者ID:huangzhengyong，项目名称:zipline，代码行数:101，代码来源:core.py
示例10: not_a
def not_a(type_):
"""More curryable version of not isinstance."""
return complement(is_a(type_))
开发者ID:thedrow，项目名称:codetransformer，代码行数:3，代码来源:functional.py
注：本文中的toolz.complement函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论