"""
Strategy and backtesting methods for Symbol class.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
from .analysis.tags import (tag_column_to_strategy_group, backtesting as run_backtesting,
tag_comparison, tag_cross, merge_series,
clean_in_out as clean_in_out_func)
from .analysis.indicators import ffill_indicator, shift_indicator
from .core.exceptions import BinPanException
[docs]
class SymbolStrategy:
"""Strategy, tagging and backtesting methods for Symbol."""
###############
# Backtesting #
###############
[docs]
def backtesting(self,
actions_col: str | int,
target_column: str | pd.Series = None,
stop_loss_column: str | pd.Series = None,
entry_filter_column: str | pd.Series = None,
fixed_target: bool = True,
fixed_stop_loss: bool = True,
base: float = 0,
quote: float = 1000,
priced_actions_col: str = 'Open',
label_in=1, label_out=-1,
fee: float = 0.001,
evaluating_quote: str = None,
short: bool = False,
inplace=True,
suffix: str = None,
colors: list = None) -> pd.DataFrame | pd.Series:
"""
Simulates buys and sells using labels in a tagged column with actions. Actions are considered before the tag, in the next
candle using priced_actions_col price of that candle before.
:param str | int actions_col: A column name or index.
:param target_column: Column with data for operation target values.
:param stop_loss_column: Column with data for operation stop loss values.
:param pd.Series | str entry_filter_column: A serie or colum with ones or zeros to allow or avoid entries.
:param bool fixed_target: Target for any operation will be calculated and fixed at the beginning of the operation.
:param bool fixed_stop_loss: Stop loss for any operation will be calculated and fixed at the beginning of the operation.
:param float base: Base inverted quantity.
:param float quote: Quote inverted quantity.
:param str | int priced_actions_col: Columna name or index with prices to use when action label in a row.
:param str | int label_in: A label consider as trade in trigger.
:param str | int label_out: A label consider as trade out trigger.
:param float fee: Fees applied to the simulation.
:param str evaluating_quote: A quote used to convert value of the backtesting line for better reference.
:param bool short: Backtest in short mode, with in as shorts and outs as repays.
:param bool inplace: Make it permanent in the instance or not.
:param str suffix: A decorative suffix for the name of the column created.
:param list colors: Defaults to red and green.
:return pd.DataFrame | pd.Series:
.. image:: images/plot_tagged.png
:width: 800
:alt: Backtesting results with buy/sell markers
"""
if type(actions_col) == int:
actions = self.df.iloc[:, actions_col]
else:
actions = self.df[actions_col]
if suffix:
suffix = '_' + suffix
if not short:
wallet_df = run_backtesting(df=self.df, actions_column=actions, target_column=target_column, stop_loss_column=stop_loss_column,
entry_filter_column=entry_filter_column, priced_actions_col=priced_actions_col,
fixed_target=fixed_target,
fixed_stop_loss=fixed_stop_loss, base=base, quote=quote, fee=fee, label_in=label_in,
label_out=label_out, suffix=suffix,
evaluating_quote=evaluating_quote, info_dic=self.info_dic)
else:
wallet_df = run_backtesting(df=self.df, actions_column=actions, target_column=target_column,
stop_loss_column=stop_loss_column, entry_filter_column=entry_filter_column,
priced_actions_col=priced_actions_col,
fixed_target=fixed_target, fixed_stop_loss=fixed_stop_loss, base=base, quote=quote, fee=fee,
label_in=label_in,
label_out=label_out, suffix=suffix, evaluating_quote=evaluating_quote, info_dic=self.info_dic,
direction='short')
if inplace and self.is_new(wallet_df):
column_names = wallet_df.columns
self.row_counter += 1
if not colors:
colors = ['cornflowerblue', 'blue', 'black', 'grey', 'green']
for i, col in enumerate(column_names):
self.set_plot_color(indicator_column=col, color=colors[i])
self.set_plot_color_fill(indicator_column=col, color_fill=False)
self.set_plot_row(indicator_column=col, row_position=self.row_counter + i)
# second row added in loop, need to sync row counter with las added row
self.row_counter += 1
self.df = pd.concat([self.df, wallet_df], axis=1)
return wallet_df
[docs]
def roi(self, column: str = None) -> float:
"""
It returns win or loos percent for a evaluation column. Just compares first and last value increment by the first price in percent.
If not column passed, it will search for an Evaluation column.
:param str column: A column in the BinPan's DataFrame with values to check ROI (return of inversion).
:return float: Resulting return of inversion.
"""
if not column:
column = [i for i in self.df.columns if i.startswith('Eval')][-1]
print(f"Auto selected column {column}")
my_column = self.df[column].dropna()
first = my_column.iloc[0]
last = my_column.iloc[-1]
return 100 * (last - first) / first
[docs]
def profit_hour(self, column: str = None) -> float:
"""
It returns win or loos quantity per hour. Just compares first and last value. Expected datetime index. If not column passed, it
will search for an Evaluation column.
:param str column: A column in the BinPan's DataFrame with values to check profit with expected datetime index.
:return float: Resulting return of inversion.
"""
if not column:
column = [i for i in self.df.columns if i.startswith('Eval')]
if not column:
column = "Close"
else:
column = column[-1]
print(f"Auto selected column {column}")
my_column = self.df[column].dropna()
first = my_column.iloc[0]
last = my_column.iloc[-1]
profit = last - first
ms = self.df['Close timestamp'].dropna().iloc[-1] - self.df['Open timestamp'].dropna().iloc[0]
hours = ms / (1000 * 60 * 60)
print(f"Total profit for {column}: {profit} with ratio {profit / hours} per hour.")
return profit / hours
#############
# Relations #
#############
[docs]
def tag(self,
column: str | int | pd.Series,
reference: str | int | float | pd.Series,
relation: str = 'gt',
match_tag: str | int = 1,
mismatch_tag: str | int = 0,
strategy_group: str = '',
inplace=True, suffix: str = '',
color: str | int = 'green') -> pd.Series:
"""
It tags values of a column/serie compared to other serie or value by methods gt,ge,eq,le,lt as condition.
:param pd.Series | str column: A numeric serie or column name or column index. Default is Close price.
:param pd.Series | str | int | float reference: A number or numeric serie or column name.
:param str relation: The condition to apply comparing column to reference (default is greater than):
eq (equivalent to ==) - equals to
ne (equivalent to !=) - not equals to
le (equivalent to <=) - less than or equals to
lt (equivalent to <) - less than
ge (equivalent to >=) - greater than or equals to
gt (equivalent to >) - greater than
:param int | str match_tag: Value or string to tag matched relation.
:param int | str mismatch_tag: Value or string to tag mismatched relation.
:param str strategy_group: A name for a group of columns to assign to a strategy.
:param bool inplace: Permanent or not. Default is false, because of some testing required sometimes.
:param str suffix: A string to decorate resulting Pandas series name.
:param str | int color: A color from plotly list of colors or its index in that list.
:return pd.Series: A serie with tags as values.
.. code-block::
import binpan
sym = binpan.Symbol('btcbusd', '1m')
sym.ema(window=200, color='darkgrey')
# comparing close price (default) greater or equal, than exponential moving average of 200 ticks window previously added.
sym.tag(reference='EMA_200', relation='ge')
sym.plot()
.. image:: images/relations/tag.png
:width: 1000
"""
if relation not in ('gt', 'ge', 'eq', 'le', 'lt'):
raise BinPanException("relation must be 'gt','ge','eq','le' or 'lt'")
# parse params
if type(column) == str:
data_a = self.df[column]
elif type(column) == int:
data_a = self.df.iloc[:, column]
else:
data_a = column.copy(deep=True)
if type(reference) == str:
data_b = self.df[reference]
elif type(reference) == int or type(reference) == float:
data_b = pd.Series(data=reference, index=data_a.index)
else:
data_b = reference.copy(deep=True)
compared = tag_comparison(serie_a=data_a, serie_b=data_b, **{relation: True}, match_tag=match_tag, mismatch_tag=mismatch_tag)
if not data_b.name:
data_b.name = reference
if suffix:
suffix = '_' + suffix
column_name = f"Tag_{data_a.name}_{relation}_{data_b.name}" + suffix
compared.name = column_name
if inplace and self.is_new(compared):
self.row_counter += 1
self.set_plot_color(indicator_column=column_name, color=color)
self.set_plot_color_fill(indicator_column=column_name, color_fill=None)
self.set_plot_row(indicator_column=column_name, row_position=self.row_counter) # overlaps are one
self.df.loc[:, column_name] = compared
if strategy_group:
self.strategy_groups = tag_column_to_strategy_group(column=column_name, group=strategy_group,
strategy_groups=self.strategy_groups)
return compared
[docs]
def cross(self,
slow: str | int | float | pd.Series,
fast: str | int | pd.Series = 'Close',
cross_over_tag: str | int = 1,
cross_below_tag: str | int = -1, echo=0,
non_zeros: bool = True,
strategy_group: str = None,
inplace=True,
suffix: str = '',
color: str | int = 'green') -> pd.Series:
"""
It tags crossing values from a column/serie (fast) over a serie or value (slow).
:param pd.Series | str | int | float slow: A number or numeric serie or column name.
:param pd.Series | str fast: A numeric serie or column name or column index. Default is Close price.
:param int | str cross_over_tag: Value or string to tag matched crossing fast over slow.
:param int | str cross_below_tag: Value or string to tag crossing slow over fast.
:param bool non_zeros: Result will not contain zeros as non tagged values, instead will be nans.
:param int echo: It tags a fixed amount of candles forward the crossed point not including cross candle. If echo want to be used,
must be used non_zeros.
:param str strategy_group: A name for a group of columns to assign to a strategy.
:param bool inplace: Permanent or not. Default is false, because of some testing required sometimes.
:param str suffix: A string to decorate resulting Pandas series name.
:param str | int color: A color from plotly list of colors or its index in that list.
:return pd.Series: A serie with tags as values. 1 and -1 for both crosses.
.. code-block::
import binpan
sym = binpan.Symbol(symbol='ethbusd', tick_interval='1m', limit=300, time_zone='Europe/Madrid')
sym.ema(window=10, color='darkgrey')
sym.cross(slow='Close', fast='EMA_10')
sym.plot(actions_col='Cross_EMA_10_Close', priced_actions_col='EMA_10',
labels=['over', 'below'],
markers=['arrow-bar-left', 'arrow-bar-right'],
marker_colors=['orange', 'blue'])
.. image:: images/relations/cross.png
:width: 1000
"""
# parse params
if type(fast) == str:
data_a = self.df[fast]
elif type(fast) == int:
data_a = self.df.iloc[:, fast]
else:
data_a = fast.copy(deep=True)
if type(slow) == str:
data_b = self.df[slow]
elif type(slow) == int or type(slow) == float:
data_b = pd.Series(data=slow, index=data_a.index)
else:
data_b = slow.copy(deep=True)
if not data_a.name:
data_a.name = fast
if not data_b.name:
data_b.name = slow
if suffix:
suffix = '_' + suffix
column_name = f"Cross_{data_b.name}_{data_a.name}" + suffix
cross = tag_cross(serie_a=data_a, serie_b=data_b, echo=echo, cross_over_tag=cross_over_tag, cross_below_tag=cross_below_tag,
name=column_name, non_zeros=non_zeros)
if inplace and self.is_new(cross):
self.row_counter += 1
self.set_plot_color(indicator_column=column_name, color=color)
self.set_plot_color_fill(indicator_column=column_name, color_fill=None)
self.set_plot_row(indicator_column=str(column_name), row_position=self.row_counter) # overlaps are one
self.df.loc[:, column_name] = cross
if strategy_group:
self.strategy_groups = tag_column_to_strategy_group(column=column_name, group=strategy_group,
strategy_groups=self.strategy_groups)
return cross
[docs]
def shift(self,
column: str | int | pd.Series, window=1,
strategy_group: str = '',
inplace=True, suffix: str = '',
color: str | int = 'grey') -> pd.Series:
"""
It shifts a candle ahead by the window argument value (or backwards if negative)
:param str | int | pd.Series column: Column to shift values.
:param int window: Number of candles moved ahead.
:param str strategy_group: A name for a group of columns to assign to a strategy.
:param bool inplace: Permanent or not. Default is false, because of some testing required sometimes.
:param str suffix: A string to decorate resulting Pandas series name.
:param str | int color: A color from plotly list of colors or its index in that list.
:return pd.Series: A serie with tags as values.
"""
if type(column) == str:
data_a = self.df[column]
elif type(column) == int:
data_a = self.df.iloc[:, column]
else:
data_a = column.copy(deep=True)
if suffix:
suffix = '_' + suffix
column_name = f"Shift_{data_a.name}_{window}" + suffix
shift = shift_indicator(serie=data_a, window=window)
shift.name = column_name
if inplace and self.is_new(shift):
if data_a.name in self.row_control.keys():
row_pos = self.row_control[data_a.name]
elif data_a.name in ['High', 'Low', 'Close', 'Open']:
row_pos = 1
else:
self.row_counter += 1
row_pos = self.row_counter
self.set_plot_color(indicator_column=column_name, color=color)
self.set_plot_color_fill(indicator_column=column_name, color_fill=None)
self.set_plot_row(indicator_column=column_name, row_position=row_pos)
self.df.loc[:, column_name] = shift
if strategy_group:
self.strategy_groups = tag_column_to_strategy_group(column=column_name, group=strategy_group,
strategy_groups=self.strategy_groups)
return shift
[docs]
def merge_columns(self,
main_column: str | int | pd.Series,
other_column: str | int | pd.Series,
sign_other: dict = None,
strategy_group: str = '',
inplace=True,
suffix: str = '',
color: str | int = 'grey') -> pd.Series:
"""
Predominant serie will be filled nans with values, if existing, from the other serie.
Same kind of index needed.
:param pd.Series main_column: A serie with nans to fill from other serie.
:param pd.Series other_column: A serie to pick values for the nans.
:param dict sign_other: Replace values by a dict for the "other column". Default is: {1: -1}
:param str strategy_group: A name for a group of columns to assign to a strategy.
:param bool inplace: Permanent or not. Default is false, because of some testing required sometimes.
:param str suffix: A string to decorate resulting Pandas series name.
:param str | int color: A color from plotly list of colors or its index in that list.
:return pd.Series: A merged serie.
"""
if not sign_other:
sign_other = {1: -1}
if type(main_column) == str:
data_a = self.df[main_column]
elif type(main_column) == int:
data_a = self.df.iloc[:, main_column]
else:
data_a = main_column.copy(deep=True)
if type(other_column) == str:
data_b = self.df[other_column]
elif type(other_column) == int:
data_b = self.df.iloc[:, other_column]
else:
data_b = other_column.copy(deep=True)
if sign_other:
data_b = data_b.replace(sign_other)
merged = merge_series(predominant=data_a, other=data_b)
if suffix:
suffix = '_' + suffix
column_name = f"Merged_{data_a.name}_{data_b.name}" + suffix
merged.name = column_name
if inplace and self.is_new(merged):
if data_a.name in self.row_control.keys():
row_pos = self.row_control[data_a.name]
elif data_a.name in ['High', 'Low', 'Close', 'Open']:
row_pos = 1
else:
self.row_counter += 1
row_pos = self.row_counter
self.set_plot_color(indicator_column=column_name, color=color)
self.set_plot_color_fill(indicator_column=column_name, color_fill=None)
self.set_plot_row(indicator_column=column_name, row_position=row_pos)
self.df.loc[:, column_name] = merged
if strategy_group:
self.strategy_groups = tag_column_to_strategy_group(column=column_name, group=strategy_group,
strategy_groups=self.strategy_groups)
return merged
[docs]
def clean_in_out(self,
column: str | int | pd.Series,
in_tag=1,
out_tag=-1,
strategy_group: str = '',
inplace=True, suffix: str = '',
color: str | int = 'grey') -> pd.Series:
"""
It cleans a serie with in and out tags by eliminating in streaks and out streaks.
Same kind of index needed.
:param pd.Series column: A column to clean in and out values.
:param in_tag: Tag for in tags. Default is 1.
:param out_tag: Tag for out tags. Default is -1.
:param str strategy_group: A name for a group of columns to assign to a strategy.
:param bool inplace: Permanent or not. Default is false, because of some testing required sometimes.
:param str suffix: A string to decorate resulting Pandas series name.
:param str | int color: A color from plotly list of colors or its index in that list.
:return pd.Series: A merged serie.
"""
if type(column) == str:
data_a = self.df[column]
elif type(column) == int:
data_a = self.df.iloc[:, column].copy(deep=True)
else:
data_a = column.copy(deep=True)
clean = clean_in_out_func(serie=data_a, in_tag=in_tag, out_tag=out_tag)
if suffix:
suffix = '_' + suffix
column_name = f"Clean_{data_a.name}" + suffix
clean.name = column_name
if inplace and self.is_new(clean):
if data_a.name in self.row_control.keys():
row_pos = self.row_control[data_a.name]
elif data_a.name in ['High', 'Low', 'Close', 'Open']:
row_pos = 1
else:
self.row_counter += 1
row_pos = self.row_counter
self.set_plot_color(indicator_column=column_name, color=color)
self.set_plot_color_fill(indicator_column=column_name, color_fill=None)
self.set_plot_row(indicator_column=column_name, row_position=row_pos)
self.df.loc[:, column_name] = clean
if strategy_group:
self.strategy_groups = tag_column_to_strategy_group(column=column_name, group=strategy_group,
strategy_groups=self.strategy_groups)
return clean
[docs]
def set_strategy_groups(self, column: str, group: str, strategy_groups: dict = None) -> dict:
"""
Returns strategy_groups for BinPan DataFrame.
:param str column: A column to tag with a strategy group.
:param str group: Name of the group.
:param str strategy_groups: The existing strategy groups.
:return dict: Updated strategy groups of columns.
"""
if not strategy_groups:
strategy_groups = self.strategy_groups
if column and group:
self.strategy_groups = tag_column_to_strategy_group(column=column, group=group, strategy_groups=strategy_groups)
return self.strategy_groups
[docs]
def get_strategy_columns(self) -> list:
"""
Returns column names starting with "Strategy".
:return dict: Updated strategy groups of columns.
"""
return [i for i in self.df.columns if i.lower().startswith('strategy')]
[docs]
def ffill_window(self,
column: str | int | pd.Series,
window: int = 1,
inplace=True,
replace=False,
suffix: str = '',
color: str | int = 'blue'):
"""
It forward fills a value through nans a window ahead.
:param str | int | pd.Series column: A pandas Series.
:param int window: Times values are shifted ahead. Default is 1.
:param bool replace: Permanent replace for a column with results.
:param bool inplace: Permanent or not. Default is false, because of some testing required sometimes.
:param str suffix: A string to decorate resulting Pandas series name.
:param str | int color: A color from plotly list of colors or its index in that list.
:return pd.Series: A series with index adjusted to the new shifted positions of values.
"""
if type(column) == str:
serie = self.df[column]
elif type(column) == int:
serie = self.df.iloc[:, column]
else:
serie = column.copy(deep=True)
my_ffill = ffill_indicator(serie=serie, window=window)
if suffix:
suffix = '_' + suffix
self.strategies += 1
column_name = f"Ffill_{serie.name}_{self.strategies}" + suffix
my_ffill.name = column_name
if replace:
self.df.loc[:, serie.name] = my_ffill
if inplace and self.is_new(my_ffill):
self.row_counter += 1
row_pos = self.row_counter
self.set_plot_color(indicator_column=column_name, color=color)
self.set_plot_color_fill(indicator_column=column_name, color_fill=None)
self.set_plot_row(indicator_column=column_name, row_position=row_pos)
self.df.loc[:, column_name] = my_ffill
return my_ffill