Last active
January 21, 2020 13:32
-
-
Save lionelyoung/041d8720e60ffe4df835c1858518dd78 to your computer and use it in GitHub Desktop.
update mlfinlab.labeling to use indexes profit taking and stop loss horizontal barriers
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def apply_pt_sl_on_t1(close, events, pt_sl, molecule): # pragma: no cover | |
""" | |
Snippet 3.2, page 45, Triple Barrier Labeling Method | |
This function applies the triple-barrier labeling method. It works on a set of | |
datetime index values (molecule). This allows the program to parallelize the processing. | |
Mainly it returns a DataFrame of timestamps regarding the time when the first barriers were reached. | |
:param close: (series) close prices | |
:param events: (series) of indices that signify "events" (see cusum_filter function | |
for more details) | |
:param pt_sl: (array) element 0, indicates the profit taking level; element 1 is stop loss level | |
:param molecule: (an array) a set of datetime index values for processing | |
:return: DataFrame of timestamps of when first barrier was touched | |
""" | |
# Apply stop loss/profit taking, if it takes place before t1 (end of event) | |
events_ = events.loc[molecule] | |
out = events_[['t1']].copy(deep=True) | |
profit_taking_multiple = pt_sl[0] | |
stop_loss_multiple = pt_sl[1] | |
# Profit taking active | |
if profit_taking_multiple > 0: | |
profit_taking = profit_taking_multiple * events_['trgt'] | |
else: | |
profit_taking = pd.Series(index=events.index) # NaNs | |
# Stop loss active | |
if stop_loss_multiple > 0: | |
stop_loss = -stop_loss_multiple * events_['trgt'] | |
else: | |
stop_loss = pd.Series(index=events.index) # NaNs | |
# Get events | |
for loc, vertical_barrier in events_['t1'].fillna(close.index[-1]).iteritems(): | |
closing_prices = close[loc: vertical_barrier] # Path prices for a given trade | |
cum_returns = (closing_prices / close[loc] - 1) * events_.at[loc, 'side'] # Path returns | |
out.loc[loc, 'sl'] = cum_returns[cum_returns < stop_loss[loc]].index.min() # Earliest stop loss date | |
out.loc[loc, 'pt'] = cum_returns[cum_returns > profit_taking[loc]].index.min() # Earliest profit taking date | |
return out | |
def get_events(close, t_events, pt_sl, target, min_ret, num_threads, vertical_barrier_times=False, | |
side_prediction=None): | |
""" | |
Snippet 3.6 page 50, Getting the Time of the First Touch, with Meta Labels | |
This function is orchestrator to meta-label the data, in conjunction with the Triple Barrier Method. | |
:param close: (series) Close prices | |
:param t_events: (series) of t_events. These are timestamps that will seed every triple barrier. | |
These are the timestamps selected by the sampling procedures discussed in Chapter 2, Section 2.5. | |
Eg: CUSUM Filter | |
:param pt_sl: (2 element array) element 0, indicates the profit taking level; element 1 is stop loss level. | |
A non-negative float that sets the width of the two barriers. A 0 value means that the respective | |
horizontal barrier (profit taking and/or stop loss) will be disabled. | |
:param target: (series) of values that are used (in conjunction with pt_sl) to determine the width | |
of the barrier. In this program this is daily volatility series. | |
:param min_ret: (float) The minimum target return required for running a triple barrier search. | |
:param num_threads: (int) The number of threads concurrently used by the function. | |
:param vertical_barrier_times: (series) A pandas series with the timestamps of the vertical barriers. | |
We pass a False when we want to disable vertical barriers. | |
:param side_prediction: (series) Side of the bet (long/short) as decided by the primary model | |
:return: (data frame) of events | |
-events.index is event's starttime | |
-events['t1'] is event's endtime | |
-events['trgt'] is event's target | |
-events['side'] (optional) implies the algo's position side | |
-events['pt'] Profit taking multiple | |
-events['sl'] Stop loss multiple | |
""" | |
# 1) Get target | |
target = target.loc[t_events] | |
target = target[target > min_ret] # min_ret | |
# 2) Get vertical barrier (max holding period) | |
if vertical_barrier_times is False: | |
vertical_barrier_times = pd.Series(pd.NaT, index=t_events) | |
# 3) Form events object, apply stop loss on vertical barrier | |
if side_prediction is None: | |
side_ = pd.Series(1.0, index=target.index) | |
pt_sl_ = [pt_sl[0], pt_sl[0]] | |
else: | |
side_ = side_prediction.loc[target.index] # Subset side_prediction on target index. | |
pt_sl_ = pt_sl[:2] | |
# Create a new df with [v_barrier, target, side] and drop rows that are NA in target | |
events = pd.concat({'t1': vertical_barrier_times, 'trgt': target, 'side': side_}, axis=1) | |
events = events.dropna(subset=['trgt']) | |
# Apply Triple Barrier | |
first_touch_dates = mp_pandas_obj(func=apply_pt_sl_on_t1, | |
pd_obj=('molecule', events.index), | |
num_threads=num_threads, | |
close=close, | |
events=events, | |
pt_sl=pt_sl_) | |
for ind in events.index: | |
events.loc[ind, 't1'] = first_touch_dates.loc[ind, :].dropna().min() | |
if side_prediction is None: | |
events = events.drop('side', axis=1) | |
# Add profit taking and stop loss multiples for vertical barrier calculations | |
events['pt'] = pt_sl[0] | |
events['sl'] = pt_sl[1] | |
return events |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment