How to access algorithm time for intraday strategies?

[BlackArbsCEO]

Is there a function/method that allows us to access the algorithm time? For example if you only want to accept trades during certain periods during the market session.

[polakowo]

I don't fully understand what you mean by accessing algorithm time. If you want to avoid vectorbt to execute some signals within a specific period, you should manually disable any signal that comes within this period. vectorbt consumes everything the user gives him. If you're using from_signals, you should set entries and exits to False. If from_orders, you should set the size to np.nan. If from_order_func, there is activity_mask that you can use to disable some time intervals.

[BlackArbsCEO]

I'm using the event driven mode using the framework you demonstrate here as a model.

I'm not sure how to use active_mask for this. To clarify I only want to execute entries from 7am-3pm GMT and 5pm-8pm GMT. So basically if I could access the current timestamp of the algorithm I could run a check like:

if '7 am UTC' < algorithm.time < `3 pm UTC`:
  do trade functions
else:
  exit 

[polakowo]

@BlackArbsCEO you need to create a pd.Series that has the same number of data points as your price, and where each element is a boolean that says whether this time step should be executed = a mask. You can achieve this with pandas (see https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.between_time.html).

[BlackArbsCEO]

ok thanks. what about the situations where you want to filter entries by timestamp but you want to be able to exit on any timestamp once your trade is open?

For example you only want to enter during the London Session (7am GMT to 3pm GMT) but you want to be able to exit at any time when your exit criteria is met (example a stop loss or profit target is hit during the Tokyo Session).

[polakowo]

Then instead of passing the activity mask directly to Portfolio.from_order_func, pass it to the function that decides when to buy/sell (segment_prep_func_nb, for example) and define your logic depending upon the mask you provided. There isn't one way of doing it, you need to figure it out yourself and be creative.

[BlackArbsCEO]

ok thanks for the insight. I'm just unfamiliar with this methodology within the event driven framework, that's why I asked about the algorithm time which is what I'm most used to.

[forLinDre]

Hello @polakowo and @BlackArbsCEO, Not sure if @BlackArbsCEO got anywhere with this issue? I am attempting to do something similar but implement start time for trades, end time for trades, and time to exit all trades into my indicator factory apply function. I would eventually like these times to be parameters so that I may optimize my indicator on top of intraday trading hours. For example, my bot wouldn't start trading until 9:30, 9:35, or 9:45 am. It would be able to put in new trades all the way up to 3:30, 3:35, or 3:40 pm. If any positions still exist at 3:55 pm, it would exit all positions.

The way I am approaching this is by creating datetime64 arrays which match the row number of my input price data and have the number of columns that would meet my sweep of times for each parameter as shown below.

My apply function and indicator factory then look like this:

#Create custom combined function for RMA indicator
@njit
def RMA_strat_apply_func_nb(high, low, close, max_pos, max_value, value_change, EWM_window1, EWM_window2, ewm1, ewm2, change_window, change_thres, entry_st, entry_et, exit_st):
    # Calculate RMAs
    RMA1 = vbt.indicators.nb.ma_nb(a=close, window=EWM_window1, ewm=ewm1)
    RMA2 = vbt.indicators.nb.ma_nb(a=close, window=EWM_window2, ewm=ewm2)
    
    # Create outputs from time parameters
    entry_st = entry_st
    entry_et = entry_et
    exit_st = exit_st
    
    # Set up empty arrays for iterative indicators
    pct_change = np.full_like(low, np.nan, dtype=np.float_)
    change_thres = np.full_like(low, change_thres, np.float_)
    
    buf = max(EWM_window1, EWM_window2, change_window)
    for col in range(close.shape[1]):
        for i in range(buf, close.shape[0]):
            max_pos[i, col] = vbt.nb.argmax_reduce_nb(0, high[i - change_window:i, col])
            max_value[i, col] = high[i - change_window + max_pos[i, col], col]
            value_change[i, col] = low[i - 1, col] - max_value[i, col]
            pct_change[i, col] = (value_change[i, col]/max_value[i, col])*100
    return RMA1, RMA2, pct_change, change_thres, entry_st, entry_et, exit_st

#Use indicator factory to apply RMA indicator function
RMA_Strat_Indicator = vbt.IndicatorFactory(
    input_names=['high', 'low', 'close'],
    param_names=['EWM_window1', 'EWM_window2', 'ewm1', 'ewm2', 'change_window', 'change_thres', 'entry_st', 'entry_et', 'exit_st'],
    output_names=['RMA1', 'RMA2', 'pct_change', 'change_thres', 'entry_st', 'entry_et', 'exit_st'],
    in_output_names=['max_pos', 'max_value', 'value_change'],
    attr_settings=dict(
        close=dict(dtype=np.float_),
        high=dict(dtype=np.float_),
        low=dict(dtype=np.float_),
        max_pos=dict(dtype=np.int_),
        max_value=dict(dtype=np.float_),
        value_change=dict(dtype=np.float_),
        RMA1=dict(dtype=np.float_),
        RMA2=dict(dtype=np.float_),
        pct_change=dict(dtype=np.float_),
        change_thres=dict(dtype=np.float_),
        entry_st=dict(dtype=np.datetime64),
        entry_et=dict(dtype=np.datetime64),
        exit_st=dict(dtype=np.datetime64)
    )
).from_apply_func(
    RMA_strat_apply_func_nb,
    high=High_prefetch, low=Low_prefetch, close=Close_prefetch_shift,
    max_pos=1,
    max_value=1.00,
    value_change=1.00,
    EWM_window1=EMA3_span,
    EWM_window2=EMA4_span,
    ewm1=EMA3_exp,
    ewm2=EMA4_exp,
    change_window=pct_change_span,
    change_thres=pct_change_thres_span,
    entry_st=entry_st,
    entry_et=entry_et,
    exit_st=exit_st,
    param_settings=dict(
        entry_st=dict(is_array_like=True, bc_to_input=True),
        entry_et=dict(is_array_like=True, bc_to_input=True),
        exit_st=dict(is_array_like=True, bc_to_input=True)
    ),
    param_product=True,
    run_unique=True

# Run RMA strategy entries with desired parameter sweeps
RMA_strat_indicator = RMA_Strat_Indicator.run(
    EWM_window1=65,
    EWM_window2=[200, 210],
    ewm1=True,
    ewm2=True,
    change_window=6,
    change_thres=-1.0,
    entry_st=entry_st,
    entry_et=entry_et,
    exit_st=exit_st,
)

RMA_strat_indicator = RMA_strat_indicator[indicator_mask]
)

You can see that I am setting my entry/exit time parameter arrays to is_array_like and bc_to_input as true. I am doing this simply so that each array is seen as an array rather than a huge number of parameters I need to run through my indicator factory. When I run my indicator however, I get an error that reads:

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
C:\Users\ANDRE~1.DES\AppData\Local\Temp/ipykernel_11256/1241169905.py in <module>
      1 # Run RMA strategy entries with desired parameter sweeps
----> 2 RMA_strat_indicator = RMA_Strat_Indicator.run(
      3     EWM_window1=65,
      4     EWM_window2=[200, 210],
      5     ewm1=True,

~\anaconda3\envs\strategy_lib_vectorbt\lib\site-packages\vectorbt\indicators\factory.py in run(cls, high, low, close, EWM_window1, EWM_window2, ewm1, ewm2, change_window, change_thres, entry_st, entry_et, exit_st, max_pos, max_value, value_change, short_name, hide_params, hide_default, **kwargs)
     12 Each indicator is basically a pipeline that:
     13 
---> 14 * Accepts a list of input arrays (for example, OHLCV data)
     15 * Accepts a list of parameter arrays (for example, window size)
     16 * Accepts other relevant arguments and keyword arguments

~\anaconda3\envs\strategy_lib_vectorbt\lib\site-packages\vectorbt\indicators\factory.py in _run(cls, *args, **kwargs)
   2863 
   2864             # Run the pipeline
-> 2865             results = run_pipeline(
   2866                 len(output_names),  # number of returned outputs
   2867                 custom_func,

~\anaconda3\envs\strategy_lib_vectorbt\lib\site-packages\vectorbt\indicators\factory.py in run_pipeline(num_ret_outputs, custom_func, require_input_shape, input_shape, input_index, input_columns, input_list, in_output_list, in_output_settings, broadcast_kwargs, param_list, param_product, param_settings, run_unique, silence_warnings, per_column, pass_col, keep_pd, to_2d, as_lists, pass_input_shape, pass_flex_2d, level_names, hide_levels, stacking_kwargs, return_raw, use_raw, wrapper_kwargs, seed, *args, **kwargs)
   1975     if len(param_list) > 0:
   1976         # Build new column levels on top of input levels
-> 1977         param_indexes, new_columns = build_columns(
   1978             param_list,
   1979             input_columns,

~\anaconda3\envs\strategy_lib_vectorbt\lib\site-packages\vectorbt\indicators\factory.py in build_columns(param_list, input_columns, level_names, hide_levels, param_settings, per_column, ignore_default, **kwargs)
   1315                     )
   1316             else:
-> 1317                 param_index = index_fns.index_from_values(param_list[i], name=level_name)
   1318                 param_index = index_fns.repeat_index(
   1319                     param_index,

~\anaconda3\envs\strategy_lib_vectorbt\lib\site-packages\vectorbt\base\index_fns.py in index_from_values(values, name)
     52             value_names.append(v)
     53         elif isinstance(v, np.ndarray):
---> 54             if np.isclose(v, v.item(0), equal_nan=True).all():
     55                 value_names.append(v.item(0))
     56             else:

<__array_function__ internals> in isclose(*args, **kwargs)

~\anaconda3\envs\strategy_lib_vectorbt\lib\site-packages\numpy\core\numeric.py in isclose(a, b, rtol, atol, equal_nan)
   2356 
   2357     xfin = isfinite(x)
-> 2358     yfin = isfinite(y)
   2359     if all(xfin) and all(yfin):
   2360         return within_tol(x, y, atol, rtol)

TypeError: ufunc 'isfinite' not supported for the input types, and the inputs could not be safely coerced to any supported types according to the casting rule ''safe''

I cannot get to the bottom of the issue. If I do not set any parameter settings I quickly run out of memory since I'm assuming it's thinking each one of my time arrays is composed of parameters that i am trying to sweep through the indicator, which I'm not. The idea at the end would be that I would create my entry signals by writing some statement similar to: RMA_strat_indicator."some output".index.tz_localize(None) > RMA_strat_indicator.entry_st to try all different combinations of entry times but I cannot get to this point due to the error.

Any help as always would be greatly appreciated as I learn vectorbt. Thank you:)

[polakowo]

@forLinDre I fear the issue is on vectorbt's end. I haven't really tested datetime and timedelta arrays as parameters. I know the exact place where the error comes from and how to solve it, but unfortunately there is no workaround in the current version that you could use to circumvent this. Please provide this array as a regular input before I sort this out.

[forLinDre]

@polakowo I see. Do you possibly know how others are approaching this problem? It's very common for intraday trading to wait for the market to settle prior to initiating trades. Usually the first 15 minutes or so. All other parameters should be optimized around these specific start/end times.

If I were to temporarily feed a single set of times as inputs as you suggest, how would you go about making this work? From my example above, I have switched my numpy time arrays from parameters to inputs. But now when I create my entry signals using "RMA_strat_indicator.RMA1.index.tz_localize(None) >= RMA_strat_indicator.entry_st" I get an error stating "ValueError: Unable to coerce to Series, length must be 2: given 790". My entry start indicator looks like this:

[polakowo]

@forLinDre if your times are the same for each day, you can map each to a timestamp (integer) and then reverse map once you are back in pandas.

I can't tell what the problem with your latest statement is, entry_st looks ok from layout point of view. When comparing pandas objects, make sure that they have the same index and columns. If they have different index/columns but can be broadcast together, append .vbt to the operand on the left.

[forLinDre]

@polakowo, I am comparing the columns pictured with their index. Can one do that? I have 2 columns because I currently have two custom_EWM_window2 parameter values. I will have more columns once start hyper-optimization.

My index:

Columns I'm comparing to the index:

Non-working expression with error (Unable to coerce to Series, length must be 2: given 790): RMA_strat_indicator.entry_st.index.tz_localize(None) >= RMA_strat_indicator.entry_st

So based on your statement, they have the same index but of course they wouldn't have the same number of columns since one is the row index. Can you please specify where you would recommend appending .vbt exactly and how this is going to perform broadcasting to create the comparison? I tried appending it as shown below but received an error stating datetimeindex object has no attribut 'vbt'. RMA_strat_indicator.entry_st.index.tz_localize(None).vbt

[polakowo]

Convert left object to series with the same index as on the right.

[forLinDre]

@polakowo thanks for the help... again. I'm still learning python so some of these do not yet come naturally to me. Appreciate your time!

[A7DC]

@polakowo thanks for the help... again. I'm still learning python so some of these do not yet come naturally to me. Appreciate your time!

Hi mate did you manage to get your code working? If so could you share an example please