!pip install pandas numpy plotly scipy

import warnings; warnings.filterwarnings("ignore")
import pandas as pd
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from scipy.signal import argrelextrema

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 swing_high_low_detector(
    df: pd.DataFrame,
    order: int = 10,
) -> pd.DataFrame:
    """
    Detect swing highs and swing lows using symmetric rolling extrema.

    Core logic
    ----------
    1. Apply scipy.signal.argrelextrema to the high series to locate local maxima
       (swing highs) and to the low series to locate local minima (swing lows).
    2. Label each detected extremum in the DataFrame.
    3. Assign a signal: +1 at swing lows (bullish reversal candidate),
       -1 at swing highs (bearish reversal candidate).

    Parameters
    ----------
    df : pd.DataFrame
        OHLCV DataFrame with columns: open, high, low, close, volume, datetime.
    order : int
        Minimum number of bars on each side that must be lower (for highs)
        or higher (for lows) to qualify as a swing point.

    Returns
    -------
    pd.DataFrame
        Original DataFrame extended with: swing_type, signal.
    """
    df = df.copy().sort_values("datetime", ignore_index=True)
    df["swing_type"] = "none"
    df["signal"]     = 0

    highs = df["high"].values
    lows  = df["low"].values

    # ── Locate swing extrema ─────────────────────────────────────────────────
    peak_idx   = argrelextrema(highs, np.greater, order=order)[0]  # swing highs
    trough_idx = argrelextrema(lows,  np.less,    order=order)[0]  # swing lows

    # ── Assign labels and signals ────────────────────────────────────────────
    df.loc[peak_idx,   "swing_type"] = "swing_high"
    df.loc[trough_idx, "swing_type"] = "swing_low"

    df.loc[peak_idx,   "signal"] = -1  # sell at swing high
    df.loc[trough_idx, "signal"] = 1   # buy at swing low

    return df

df_signals = swing_high_low_detector(df, order=10)

print("--- Swing Type Distribution ---")
print(df_signals["swing_type"].value_counts())
print("\n--- Signal Distribution ---")
print(df_signals["signal"].value_counts())

swing_highs = df_signals[df_signals["swing_type"] == "swing_high"]
swing_lows  = df_signals[df_signals["swing_type"] == "swing_low"]

fig = make_subplots(rows=2, cols=1, shared_xaxes=True,
    subplot_titles=["Price + Swing Points", "Signal"],
    row_heights=[0.7, 0.3])

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=swing_lows["datetime"], y=swing_lows["low"] * 0.999,
    mode="markers", marker=dict(symbol="triangle-up", size=10, color="green"),
    name="Swing Low (+1)"), row=1, col=1)

fig.add_trace(go.Scatter(
    x=swing_highs["datetime"], y=swing_highs["high"] * 1.001,
    mode="markers", marker=dict(symbol="triangle-down", size=10, color="red"),
    name="Swing High (-1)"), row=1, col=1)

fig.add_trace(go.Scatter(
    x=df_signals["datetime"], y=df_signals["signal"],
    mode="lines", name="Signal", line=dict(color="purple", 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="Swing High / Low Detector",
    xaxis_rangeslider_visible=False,
    height=700, xaxis2_title="Datetime",
)
fig.show()