Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _calc_dd(df, display=True):
dd = _stats.to_drawdown_series(df)
dd_info = _stats.drawdown_details(dd)
if dd_info.empty:
return _pd.DataFrame()
if "returns" in dd_info:
ret_dd = dd_info['returns']
else:
ret_dd = dd_info
# pct multiplier
pct = 1 if display else 100
dd_stats = {
'returns': {
'Max Drawdown %': ret_dd.sort_values(
def drawdown(returns, grayscale=False, figsize=(10, 5),
fontname='Arial', lw=1, log_scale=False,
match_volatility=False, compound=False, ylabel="Drawdown",
resample=None, subtitle=True, savefig=None, show=True):
dd = _stats.to_drawdown_series(returns)
_core.plot_timeseries(dd, title='Underwater Plot',
hline=dd.mean(), hlw=2, hllabel="Average",
returns_label="Drawdown",
compound=compound, match_volatility=match_volatility,
log_scale=log_scale, resample=resample,
fill=True, lw=lw, figsize=figsize,
ylabel=ylabel,
fontname=fontname, grayscale=grayscale,
subtitle=subtitle,
savefig=savefig, show=show)
fig.set_facecolor('white')
if subtitle:
axes[0].set_title("\n%s - %s ; Sharpe: %.2f " % (
returns.index.date[:1][0].strftime('%e %b \'%y'),
returns.index.date[-1:][0].strftime('%e %b \'%y'),
_stats.sharpe(returns)
), fontsize=12, color='gray')
axes[0].set_ylabel('Cumulative Return', fontname=fontname,
fontweight='bold', fontsize=12)
axes[0].plot(_stats.compsum(returns) * 100, color=colors[1],
lw=1 if grayscale else lw, zorder=1)
axes[0].axhline(0, color='silver', lw=1, zorder=0)
dd = _stats.to_drawdown_series(returns) * 100
ddmin = _utils._round_to_closest(abs(dd.min()), 5)
ddmin_ticks = 5
if ddmin > 50:
ddmin_ticks = ddmin / 4
elif ddmin > 20:
ddmin_ticks = ddmin / 3
ddmin_ticks = int(_utils._round_to_closest(ddmin_ticks, 5))
# ddmin_ticks = int(_utils._round_to_closest(ddmin, 5))
axes[1].set_ylabel('Drawdown', fontname=fontname,
fontweight='bold', fontsize=12)
axes[1].set_yticks(_np.arange(-ddmin, 0, step=ddmin_ticks))
axes[1].plot(dd, color=colors[2], lw=1 if grayscale else 1, zorder=1)
axes[1].axhline(0, color='silver', lw=1, zorder=0)
if not grayscale:
axes[1].fill_between(dd.index, 0, dd, color=colors[2], alpha=.1)
_po.cvar = stats.cvar
_po.expected_shortfall = stats.expected_shortfall
_po.tail_ratio = stats.tail_ratio
_po.payoff_ratio = stats.payoff_ratio
_po.win_loss_ratio = stats.win_loss_ratio
_po.profit_ratio = stats.profit_ratio
_po.profit_factor = stats.profit_factor
_po.gain_to_pain_ratio = stats.gain_to_pain_ratio
_po.cpc_index = stats.cpc_index
_po.common_sense_ratio = stats.common_sense_ratio
_po.outlier_win_ratio = stats.outlier_win_ratio
_po.outlier_loss_ratio = stats.outlier_loss_ratio
_po.recovery_factor = stats.recovery_factor
_po.risk_return_ratio = stats.risk_return_ratio
_po.max_drawdown = stats.max_drawdown
_po.to_drawdown_series = stats.to_drawdown_series
_po.kelly_criterion = stats.kelly_criterion
_po.monthly_returns = stats.monthly_returns
_po.pct_rank = stats.pct_rank
# methods from utils
_po.to_returns = utils.to_returns
_po.to_prices = utils.to_prices
_po.to_log_returns = utils.to_log_returns
_po.log_returns = utils.log_returns
_po.exponential_stdev = utils.exponential_stdev
_po.rebase = utils.rebase
_po.aggregate_returns = utils.aggregate_returns
_po.to_excess_returns = utils.to_excess_returns
_po.multi_shift = utils.multi_shift
_po.curr_month = utils._pandas_current_month
_po.date = utils._pandas_date
tpl = tpl.replace('{{eoy_table}}', _html_table(yoy))
else:
# pct multiplier
yoy = _pd.DataFrame(
_utils.group_returns(returns, returns.index.year) * 100)
yoy.columns = ['Return']
yoy['Cumulative'] = _utils.group_returns(
returns, returns.index.year, True)
yoy['Return'] = yoy['Return'].round(2).astype(str) + '%'
yoy['Cumulative'] = (yoy['Cumulative'] *
100).round(2).astype(str) + '%'
yoy.index.name = 'Year'
tpl = tpl.replace('{{eoy_title}}', '<h3>EOY Returns</h3>')
tpl = tpl.replace('{{eoy_table}}', _html_table(yoy))
dd = _stats.to_drawdown_series(returns)
dd_info = _stats.drawdown_details(dd).sort_values(
by='max drawdown', ascending=True)[:10]
dd_info = dd_info[['start', 'end', 'max drawdown', 'days']]
dd_info.columns = ['Started', 'Recovered', 'Drawdown', 'Days']
tpl = tpl.replace('{{dd_info}}', _html_table(dd_info, False))
# plots
figfile = _utils._file_stream()
_plots.returns(returns, benchmark, grayscale=grayscale,
figsize=(8, 5), subtitle=False,
savefig={'fname': figfile, 'format': 'svg'},
show=False, ylabel=False, cumulative=compounded)
tpl = tpl.replace('{{returns}}', figfile.getvalue().decode())
figfile = _utils._file_stream()
def plot_longest_drawdowns(returns, periods=5, lw=1.5,
fontname='Arial', grayscale=False,
log_scale=False, figsize=(10, 6), ylabel=True,
subtitle=True, compounded=True,
savefig=None, show=True):
colors = ['#348dc1', '#003366', 'red']
if grayscale:
colors = ['#000000'] * 3
dd = _stats.to_drawdown_series(returns.fillna(0))
dddf = _stats.drawdown_details(dd)
longest_dd = dddf.sort_values(
by='days', ascending=False, kind='mergesort')[:periods]
fig, ax = _plt.subplots(figsize=figsize)
ax.spines['top'].set_visible(False)
ax.spines['right'].set_visible(False)
ax.spines['bottom'].set_visible(False)
ax.spines['left'].set_visible(False)
fig.suptitle("Top %.0f Drawdown Periods\n" %
periods, y=.99, fontweight="bold", fontname=fontname,
fontsize=14, color="black")
if subtitle:
ax.set_title("\n%s - %s " % (
returns.index.date[:1][0].strftime('%e %b \'%y'),
def full(returns, benchmark=None, rf=0., grayscale=False,
figsize=(8, 5), display=True, compounded=True):
dd = _stats.to_drawdown_series(returns)
dd_info = _stats.drawdown_details(dd).sort_values(
by='max drawdown', ascending=True)[:5]
if not dd_info.empty:
dd_info.index = range(1, min(6, len(dd_info)+1))
dd_info.columns = map(lambda x: str(x).title(), dd_info.columns)
if _utils._in_notebook():
iDisplay(iHTML('<h4>Performance Metrics</h4>'))
iDisplay(metrics(returns=returns, benchmark=benchmark,
rf=rf, display=display, mode='full',
compounded=compounded))
iDisplay(iHTML('<h4>5 Worst Drawdowns</h4>'))
if dd_info.empty:
iDisplay(iHTML("<p>(no drawdowns)</p>"))
else: