Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def group_returns(returns, groupby, compounded=False):
""" summarize returns
group_returns(df, df.index.year)
group_returns(df, [df.index.year, df.index.month])
"""
if compounded:
return returns.groupby(groupby).apply(_stats.comp)
return returns.groupby(groupby).sum()
if "benchmark" in df:
s_start['benchmark'] = df['benchmark'].index.strftime('%Y-%m-%d')[0]
s_end['benchmark'] = df['benchmark'].index.strftime('%Y-%m-%d')[-1]
s_rf['benchmark'] = rf
metrics['Start Period'] = _pd.Series(s_start)
metrics['End Period'] = _pd.Series(s_end)
metrics['Risk-Free Rate %'] = _pd.Series(s_rf)
metrics['Time in Market %'] = _stats.exposure(df) * pct
metrics['~'] = blank
if compounded:
metrics['Cumulative Return %'] = (
_stats.comp(df) * pct).map('{:,.2f}'.format)
else:
metrics['Total Return %'] = (df.sum() * pct).map('{:,.2f}'.format)
metrics['CAGR%%'] = _stats.cagr(df, rf, compounded) * pct
metrics['Sharpe'] = _stats.sharpe(df, rf)
metrics['Sortino'] = _stats.sortino(df, rf)
metrics['Max Drawdown %'] = blank
metrics['Longest DD Days'] = blank
if mode.lower() == 'full':
ret_vol = _stats.volatility(df['returns']) * pct
if "benchmark" in df:
bench_vol = _stats.volatility(df['benchmark']) * pct
metrics['Volatility (ann.) %'] = [ret_vol, bench_vol]
metrics['R^2'] = _stats.r_squared(df['returns'], df['benchmark'])
else:
def plot_distribution(returns, figsize=(10, 6),
fontname='Arial', grayscale=False, ylabel=True,
subtitle=True, compounded=True,
savefig=None, show=True):
colors = _FLATUI_COLORS
if grayscale:
colors = ['#f9f9f9', '#dddddd', '#bbbbbb', '#999999', '#808080']
# colors, ls, alpha = _get_colors(grayscale)
port = _pd.DataFrame(returns.fillna(0))
port.columns = ['Daily']
apply_fnc = _stats.comp if compounded else _np.sum
port['Weekly'] = port['Daily'].resample(
'W-MON').apply(apply_fnc).resample('W-MON').last()
port['Weekly'].ffill(inplace=True)
port['Monthly'] = port['Daily'].resample(
'M').apply(apply_fnc).resample('M').last()
port['Monthly'].ffill(inplace=True)
port['Quarterly'] = port['Daily'].resample(
'Q').apply(apply_fnc).resample('Q').last()
port['Quarterly'].ffill(inplace=True)
port['Yearly'] = port['Daily'].resample(
'A').apply(apply_fnc).resample('A').last()
port['Yearly'].ffill(inplace=True)
'benchmark.')
if match_volatility and benchmark is not None:
bmark_vol = benchmark.loc[returns.index].std()
returns = (returns / returns.std()) * bmark_vol
# ---------------
colors, _, _ = _get_colors(grayscale)
df = _pd.DataFrame(index=returns.index, data={returns_label: returns})
if isinstance(benchmark, _pd.Series):
df['Benchmark'] = benchmark[benchmark.index.isin(returns.index)]
df = df[['Benchmark', returns_label]]
df = df.dropna()
if resample is not None:
df = df.resample(resample).apply(
_stats.comp).resample(resample).last()
# ---------------
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)
# use a more precise date string for the x axis locations in the toolbar
fig.suptitle(title+"\n", y=.99, fontweight="bold", fontname=fontname,
fontsize=14, color="black")
if subtitle:
ax.set_title("\n%s - %s " % (
df.index.date[:1][0].strftime('%Y'),
df.index.date[-1:][0].strftime('%Y')
def extend_pandas():
"""
extends pandas by exposing methods to be used like:
df.sharpe(), df.best('day'), ...
"""
from pandas.core.base import PandasObject as _po
_po.compsum = stats.compsum
_po.comp = stats.comp
_po.expected_return = stats.expected_return
_po.geometric_mean = stats.geometric_mean
_po.ghpr = stats.ghpr
_po.outliers = stats.outliers
_po.remove_outliers = stats.remove_outliers
_po.best = stats.best
_po.worst = stats.worst
_po.consecutive_wins = stats.consecutive_wins
_po.consecutive_losses = stats.consecutive_losses
_po.exposure = stats.exposure
_po.win_rate = stats.win_rate
_po.avg_return = stats.avg_return
_po.avg_win = stats.avg_win
_po.avg_loss = stats.avg_loss
_po.volatility = stats.volatility
_po.implied_volatility = stats.implied_volatility