89 | @plot_function |
90 | def max_(x: Union[pd.Series, List[pd.Series]], w: Union[Window, int] = Window(None, 0)) -> pd.Series: |
91 | """ |
92 | Maximum value of series over given window |
93 | |
94 | :param x: series: a timeseries or an array of timeseries |
95 | :param w: Window or int: size of window and ramp up to use. e.g. Window(22, 10) where 22 is the window size |
96 | and 10 the ramp up value. Window size defaults to length of series. |
97 | :return: timeseries of maximum value |
98 | |
99 | **Usage** |
100 | |
101 | Returns the maximum value of the series over each window. |
102 | |
103 | If :math:`x` is a series: |
104 | |
105 | :math:`R_t = max(X_{t-w+1}:X_t)` |
106 | |
107 | where :math:`w` is the size of the rolling window. |
108 | |
109 | If :math:`x` is an array of series: |
110 | |
111 | :math:`R_t = max(X_{1, t-w+1}:X_{n, t})` |
112 | |
113 | where :math:`w` is the size of the rolling window, and :math:`n` is the number of series. |
114 | |
115 | If window is not provided, returns the maximum value over the full series. If the window size is greater than the |
116 | available data, will return maximum of available values. |
117 | |
118 | **Examples** |
119 | |
120 | Maximum value of price series over the last :math:`22` observations: |
121 | |
122 | >>> prices = generate_series(100) |
123 | >>> max_(prices, 22) |
124 | |
125 | **See also** |
126 | |
127 | :func:`min_` |
128 | |
129 | """ |
130 | if isinstance(x, list): |
131 | x = pd.concat(x, axis=1).max(axis=1) |
132 | w = normalize_window(x, w) |
133 | assert x.index.is_monotonic_increasing, "series index is monotonic increasing" |
134 | if isinstance(w.w, pd.DateOffset): |
135 | values = [x.loc[(x.index > idx - w.w) & (x.index <= idx)].max() for idx in x.index] |
136 | return apply_ramp(pd.Series(values, index=x.index, dtype=np.dtype(float)), w) |
137 | else: |
138 | return apply_ramp(x.rolling(w.w, 0).max(), w) |