Trading signal vs. price movement vs. profit analysis using Pandas

signal-vs-profit.py

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