Multi Timeframe Signal Alignment

import pandas as pd
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def generate_data(periods: int) -> pd.DataFrame:
    """
    Generate synthetic OHLCV price data using a geometric random walk.

    Parameters
    ----------
    periods : int
        Number of 1-minute bars to generate.

    Returns
    -------
    pd.DataFrame
        DataFrame with columns: open, high, low, close, volume, datetime.
    """
    start_date     = pd.to_datetime("2024-01-01 00:00:00+00:00")
    datetime_index = pd.date_range(start_date, periods=periods, freq="1min", tz="UTC")
    price_data = []
    last_close = 42000
    for i in range(periods):
        open_price  = last_close + np.random.normal(0, last_close * 0.0005)
        close_price = open_price + np.random.normal(0, last_close * 0.005)
        body_high   = max(open_price, close_price)
        body_low    = min(open_price, close_price)
        high_price  = max(body_high + abs(np.random.normal(0, last_close * 0.002)), open_price, close_price)
        low_price   = min(body_low  - abs(np.random.normal(0, last_close * 0.002)), open_price, close_price)
        if high_price < low_price:
            high_price, low_price = low_price, high_price
        price_data.append({
            "open":  max(1, int(open_price)),
            "high":  max(1, int(high_price)),
            "low":   max(1, int(low_price)),
            "close": max(1, int(close_price)),
        })
        last_close = close_price
    df = pd.DataFrame(price_data, index=datetime_index)
    df.index.name = "datetime"
    df["volume"]   = np.random.uniform(100.0, 500.0, periods)
    df["datetime"] = df.index.to_series()
    return df.reset_index(drop=True)

df = generate_data(500)
display(df.head())

def multi_timeframe_signal_alignment(
    df: pd.DataFrame,
    htf_resample: str = "60min",
    htf_ema_period: int = 20,
    ltf_ema_fast: int = 5,
    ltf_ema_slow: int = 20,
) -> pd.DataFrame:
    """
    Align LTF entry signals with HTF trend direction.

    Core logic
    ----------
    1. Resample the 1-minute DataFrame to the HTF bar frequency.
    2. Compute an EMA on HTF close prices; derive HTF trend bias from EMA slope.
    3. Forward-fill the HTF bias to every LTF bar via merge_asof.
    4. Compute an EMA cross signal on the LTF bars.
    5. Emit the LTF signal only when it agrees with the HTF bias.

    Parameters
    ----------
    df : pd.DataFrame     1-minute OHLCV DataFrame.
    htf_resample : str    Pandas resample rule for the higher timeframe.
    htf_ema_period : int  EMA period on HTF bars.
    ltf_ema_fast : int    Fast EMA period on LTF bars.
    ltf_ema_slow : int    Slow EMA period on LTF bars.

    Returns
    -------
    pd.DataFrame with: htf_bias, ltf_signal, signal.
    """
    df = df.copy().sort_values("datetime", ignore_index=True)
    df["datetime"] = pd.to_datetime(df["datetime"], utc=True)

    # ── HTF resampling ────────────────────────────────────────────────────────
    df_htf = df.set_index("datetime").resample(htf_resample).agg({
        "open": "first", "high": "max", "low": "min",
        "close": "last", "volume": "sum",
    }).dropna().reset_index()

    df_htf["htf_ema"]  = df_htf["close"].ewm(span=htf_ema_period, adjust=False).mean()
    df_htf["htf_bias"] = np.where(df_htf["htf_ema"] > df_htf["htf_ema"].shift(1), 1, -1)

    # ── Forward-fill HTF bias onto LTF bars ───────────────────────────────────
    df = pd.merge_asof(df, df_htf[["datetime", "htf_bias"]],
                       on="datetime", direction="backward")

    # ── LTF EMA cross signal ──────────────────────────────────────────────────
    df["ema_fast"] = df["close"].ewm(span=ltf_ema_fast, adjust=False).mean()
    df["ema_slow"] = df["close"].ewm(span=ltf_ema_slow, adjust=False).mean()
    df["ltf_signal"] = np.where(df["ema_fast"] > df["ema_slow"], 1, -1)

    # ── MTF-aligned signal ─────────────────────────────────────────────────────
    df["signal"] = np.where(df["ltf_signal"] == df["htf_bias"], df["ltf_signal"], 0)

    return df

df_signals = multi_timeframe_signal_alignment(df, htf_resample="60min")
print(df_signals[["htf_bias", "ltf_signal", "signal"]].value_counts().head(10))

buy_signals  = df_signals[df_signals["signal"] ==  1]
sell_signals = df_signals[df_signals["signal"] == -1]
fig = make_subplots(rows=2, cols=1, shared_xaxes=True,
    subplot_titles=["Price + MTF-Aligned Signals", "HTF Bias"],
    row_heights=[0.8, 0.2]) # Adjusted row_heights
fig.add_trace(go.Candlestick(x=df_signals["datetime"],
    open=df_signals["open"], high=df_signals["high"],
    low=df_signals["low"], close=df_signals["close"], name="Price"), row=1, col=1)
fig.add_trace(go.Scatter(x=buy_signals["datetime"],  y=buy_signals["low"]  * 0.999,
    mode="markers", marker=dict(symbol="triangle-up",   size=10, color="green"), name="Buy"),  row=1, col=1)
fig.add_trace(go.Scatter(x=sell_signals["datetime"], y=sell_signals["high"] * 1.001,
    mode="markers", marker=dict(symbol="triangle-down", size=10, color="red"),   name="Sell"), row=1, col=1)
fig.add_trace(go.Scatter(x=df_signals["datetime"], y=df_signals["htf_bias"],
    mode="lines", name="HTF Bias", line=dict(color="blue", width=1)), row=2, col=1)
fig.add_hline(y=0, line_dash="dot", line_color="gray", row=2, col=1)
fig.update_layout(title_text="Multi-Timeframe Signal Alignment",
    xaxis_rangeslider_visible=False, height=700, xaxis2_title="Datetime")
fig.update_yaxes(range=[-1.2, 1.2], row=2, col=1) # Fixed y-axis range for HTF Bias
fig.show()