def analyze(context=None, results=None):
import matplotlib.pyplot as plt
import logbook
logbook.StderrHandler().push_application()
log = logbook.Logger('Algorithm')
fig = plt.figure()
ax1 = fig.add_subplot(211)
results.algorithm_period_return.plot(ax=ax1,color='blue',legend=u'????')
ax1.set_ylabel(u'??')
results.benchmark_period_return.plot(ax=ax1,color='red',legend=u'????')
plt.show()
# capital_base is the base value of capital
#
python类StderrHandler()的实例源码
def main():
"""Plugin entrypoint."""
with logbook.StderrHandler().applicationbound():
# Read request message from stdin
data = sys.stdin.buffer.read()
# Parse request
request = plugin.CodeGeneratorRequest()
request.ParseFromString(data)
# Create response
response = plugin.CodeGeneratorResponse()
# Generate code
generate(request, response)
# Serialise response message
output = response.SerializeToString()
# Write to stdout
sys.stdout.buffer.write(output)
def initialize_cli(options):
from ._logger import set_log_level
debug_format_str = (
"[{record.level_name}] {record.channel} {record.func_name} "
"({record.lineno}): {record.message}")
if options.log_level == logbook.DEBUG:
info_format_str = debug_format_str
else:
info_format_str = (
"[{record.level_name}] {record.channel}: {record.message}")
logbook.StderrHandler(
level=logbook.DEBUG, format_string=debug_format_str
).push_application()
logbook.StderrHandler(
level=logbook.INFO, format_string=info_format_str
).push_application()
set_log_level(options.log_level)
spr.SubprocessRunner.is_save_history = True
if options.is_output_stacktrace:
spr.SubprocessRunner.is_output_stacktrace = (
options.is_output_stacktrace)
def test_stderr(
self, capsys, command, ignore_stderr_regexp, out_regexp, expected):
import logbook
import subprocrunner
logbook.StderrHandler(
level=logbook.DEBUG).push_application()
subprocrunner.set_log_level(logbook.INFO)
runner = SubprocessRunner(
command, ignore_stderr_regexp=ignore_stderr_regexp)
runner.run()
assert is_null_string(runner.stdout.strip())
assert is_not_null_string(runner.stderr.strip())
out, err = capsys.readouterr()
print("[sys stdout]\n{}\n".format(out))
print("[sys stderr]\n{}\n".format(err))
print("[proc stdout]\n{}\n".format(runner.stdout))
print("[proc stderr]\n{}\n".format(runner.stderr))
actual = out_regexp.search(err) is not None
assert actual == expected
def initialize_log_handler(log_level):
debug_format_str = (
"[{record.level_name}] {record.channel} {record.func_name} "
"({record.lineno}): {record.message}")
if log_level == logbook.DEBUG:
info_format_str = debug_format_str
else:
info_format_str = (
"[{record.level_name}] {record.channel}: {record.message}")
logbook.StderrHandler(
level=logbook.DEBUG, format_string=debug_format_str
).push_application()
logbook.StderrHandler(
level=logbook.INFO, format_string=info_format_str
).push_application()
def logging_context(path=None, level=None):
from logbook import StderrHandler, FileHandler
from logbook.compat import redirected_logging
with StderrHandler(level=level or 'INFO').applicationbound():
if path:
if not os.path.isdir(os.path.dirname(path)):
os.makedirs(os.path.dirname(path))
with FileHandler(path, bubble=True).applicationbound():
with redirected_logging():
yield
else:
with redirected_logging():
yield
def main(extension, strict_extensions, default_extension):
"""Top level catalyst entry point.
"""
# install a logbook handler before performing any other operations
logbook.StderrHandler().push_application()
load_extensions(
default_extension,
extension,
strict_extensions,
os.environ,
)
def cli(ctx, verbosity, colored):
"""
Command Line Interface. Use --help for details.
"""
if verbosity > 0:
try:
# noinspection PyPackageRequirements,PyUnresolvedReferences
import logbook
# noinspection PyUnresolvedReferences,PyPackageRequirements
import logbook.more
except ImportError:
click.echo('Please install saltyrtc.server[logging] for logging support.',
err=True)
ctx.exit(code=_ErrorCode.import_error)
# Translate logging level
level = _get_logging_level(verbosity)
# Enable asyncio debug logging if verbosity is high enough
# noinspection PyUnboundLocalVariable
if level <= logbook.DEBUG:
os.environ['PYTHONASYNCIODEBUG'] = '1'
# Enable logging
util.enable_logging(level=level, redirect_loggers={
'asyncio': level,
'websockets': level,
})
# Get handler class
if colored:
handler_class = logbook.more.ColorizedStderrHandler
else:
handler_class = logbook.StderrHandler
# Set up logging handler
handler = handler_class(level=level)
handler.push_application()
ctx.obj['logging_handler'] = handler
def server_factory(request, event_loop, server_permanent_keys):
"""
Return a factory to create :class:`saltyrtc.Server` instances.
"""
# Enable asyncio debug logging
os.environ['PYTHONASYNCIODEBUG'] = '1'
# Enable logging
util.enable_logging(level=logbook.NOTICE, redirect_loggers={
'asyncio': logbook.WARNING,
'websockets': logbook.WARNING,
})
# Push handler
logging_handler = logbook.StderrHandler()
logging_handler.push_application()
_server_instances = []
def _server_factory(permanent_keys=None):
if permanent_keys is None:
permanent_keys = server_permanent_keys
# Setup server
port = unused_tcp_port()
coroutine = serve(
util.create_ssl_context(
pytest.saltyrtc.cert, dh_params_file=pytest.saltyrtc.dh_params),
permanent_keys,
host=pytest.saltyrtc.ip,
port=port,
loop=event_loop,
server_class=TestServer,
)
server_ = event_loop.run_until_complete(coroutine)
# Inject timeout and address (little bit of a hack but meh...)
server_.timeout = _get_timeout(request=request)
server_.address = (pytest.saltyrtc.ip, port)
_server_instances.append(server_)
def fin():
server_.close()
event_loop.run_until_complete(server_.wait_closed())
_server_instances.remove(server_)
if len(_server_instances) == 0:
logging_handler.pop_application()
request.addfinalizer(fin)
return server_
return _server_factory
