| def calculate_signal_vs_profit( | |
| df, | |
| signal_window: pd.Timedelta, | |
| profit_window: pd.Timedelta, | |
| data_time_bucket: pd.Timedelta, | |
| ) -> pd.DataFrame: | |
| """Calculate signals and profits for all incoming candles.""" | |
| # Create entries for past price to be used for signal | |
| # and future price (used for the price correlation) | |
| signal_offset = to_offset(signal_window) | |
| profit_offset = to_offset(profit_window) | |
| df["prev"] = df["close"].shift(freq=signal_offset) | |
| df["next"] = df["open"].shift(freq=-profit_offset) | |
| # Calculate signal from the past and price difference to the future | |
| df["signal"] = (df["prev"] - df["open"]) / df["open"] | |
| df["price_diff"] = (df["next"] - df["open"]) / df["open"] # Get the profit on the final day of profit window | |
| # On negative signals, we go short. | |
| # On zero signal and lack of data set side to NA | |
| df["side"] = pd.NA | |
| df.loc[df["signal"] > 0, "side"] = "long" | |
| df.loc[df["signal"] < 0, "side"] = "short" | |
| number_of_candles = profit_window / data_time_bucket | |
| assert number_of_candles > 0 and number_of_candles.is_integer(), f"Could not calculate candle count that fits into profit window {profit_window} for data time frame {data_time_bucket}" | |
| number_of_candles = int(number_of_candles) | |
| # Max and min price wihtin the profit window will determine the profit for longs and shorts respective | |
| df["max_future_price"] = df["close"].rolling(number_of_candles).max().shift(-number_of_candles) # Get the max profit on the profit window, assuming take profit % | |
| df["min_future_price"] = df["close"].rolling(number_of_candles).min().shift(-number_of_candles) # Get the max profit on the profit window, assuming take profit % | |
| # Calculate profit separately for longs and shorts | |
| # using Pandas Mask | |
| # https://stackoverflow.com/a/33770421/315168 | |
| # | |
| # We calculate both profit after X time, | |
| # and also max take profit, assuming | |
| # we could do a perfect trailing stop loss | |
| # | |
| longs = (df["side"] == "long") | |
| shorts = (df["side"] == "short") | |
| df.loc[longs, "profit"] = df["price_diff"] | |
| df.loc[shorts, "profit"] = -df["price_diff"] | |
| df.loc[longs, "profit_max"] = (df["max_future_price"] - df["open"]) / df["open"] # Get the profit based on max price | |
| df.loc[shorts, "profit_max"] = -(df["min_future_price"] - df["open"]) / df["open"] # Get the profit based on max price | |
| df.loc[longs, "desc"] = df.agg('{0[pair]} long'.format, axis=1) | |
| df.loc[shorts, "desc"] = df.agg('{0[pair]} short'.format, axis=1) | |
| return df |