Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def avg_win(returns, aggregate=None, compounded=True):
"""
calculates the average winning
return/trade return for a period
"""
returns = _utils._prepare_returns(returns)
if aggregate:
returns = _utils.aggregate_returns(returns, aggregate, compounded)
return returns[returns > 0].dropna().mean()
def r_squared(returns, benchmark):
""" measures the straight line fit of the equity curve """
# slope, intercept, r_val, p_val, std_err = _linregress(
_, _, r_val, _, _ = _linregress(
_utils._prepare_returns(returns),
_utils._prepare_benchmark(benchmark, returns.index))
return r_val**2
def pct_rank(prices, window=60):
""" rank prices by window """
rank = _utils.multi_shift(prices, window).T.rank(pct=True).T
return rank.iloc[:, 0] * 100.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
__version__ = "0.0.25"
__author__ = "Ran Aroussi"
from . import stats, utils, plots, reports
__all__ = ['stats', 'plots', 'reports', 'utils', 'extend_pandas']
# try automatic matplotlib inline
utils._in_notebook(matplotlib_inline=True)
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
tpl = tpl.replace('{{metrics}}', _html_table(mtrx))
tpl = tpl.replace('',
'<hr>')
tpl = tpl.replace('',
'<hr>')
if benchmark is not None:
yoy = _stats.compare(returns, benchmark, "A", compounded=compounded)
yoy.columns = ['Benchmark', 'Strategy', 'Multiplier', 'Won']
yoy.index.name = 'Year'
tpl = tpl.replace('{{eoy_title}}', '<h3>EOY Returns vs Benchmark</h3>')
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']
def ulcer_performance_index(returns, rf=0):
"""
calculates the ulcer index score
(downside risk measurment)
"""
returns = _utils._prepare_returns(returns, rf)
dd = 1. - returns/returns.cummax()
ulcer = _np.sqrt(_np.divide((dd**2).sum(), returns.shape[0] - 1))
return returns.mean() / ulcer
def html(returns, benchmark=None, rf=0.,
grayscale=False, title='Strategy Tearsheet',
output=None, compounded=True):
if output is None and not _utils._in_notebook():
raise ValueError("`file` must be specified")
tpl = ""
with open(__file__[:-4] + '.html') as f:
tpl = f.read()
f.close()
date_range = returns.index.strftime('%e %b, %Y')
tpl = tpl.replace('{{date_range}}', date_range[0] + ' - ' + date_range[-1])
tpl = tpl.replace('{{title}}', title)
tpl = tpl.replace('{{v}}', __version__)
mtrx = metrics(returns=returns, benchmark=benchmark,
rf=rf, display=False, mode='full',
sep=True, internal="True",
compounded=compounded)[2:]
def max_drawdown(prices):
""" calculates the maximum drawdown """
prices = _utils._prepare_prices(prices)
return (prices / prices.expanding(min_periods=0).max()).min() - 1
def consecutive_wins(returns, aggregate=None, compounded=True):
""" returns the maximum consecutive wins by day/month/week/quarter/year """
returns = _utils._prepare_returns(returns)
returns = _utils.aggregate_returns(returns, aggregate, compounded) > 0
return _utils.count_consecutive(returns).max()
_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
_po.mtd = utils._mtd
_po.qtd = utils._qtd
_po.ytd = utils._ytd
# methods that requires benchmark stats
_po.r_squared = stats.r_squared
_po.r2 = stats.r2
_po.information_ratio = stats.information_ratio
_po.greeks = stats.greeks
_po.rolling_greeks = stats.rolling_greeks
_po.compare = stats.compare
# plotting methods
_po.plot_snapshot = plots.snapshot
_po.plot_earnings = plots.earnings
_po.plot_daily_returns = plots.daily_returns
_po.plot_distribution = plots.distribution