!pip install pandas numpy plotly

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

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 breakout_detection(
    df: pd.DataFrame,
    window: int = 20,
    volume_factor: float = 1.5,
) -> pd.DataFrame:
    """
    Detect price breakouts from a rolling Donchian Channel.

    Core logic
    ----------
    1. Compute the rolling maximum of highs (upper channel) and rolling minimum
       of lows (lower channel) over window bars, shifted by one to avoid look-ahead.
    2. Compare the current close against the prior bar's channel boundaries.
    3. Optionally flag breakouts where volume exceeds volume_factor × rolling average.

    Parameters
    ----------
    df : pd.DataFrame
        OHLCV DataFrame with columns: open, high, low, close, volume, datetime.
    window : int
        Lookback period for the Donchian Channel.
    volume_factor : float
        Multiplier; breakouts with volume > volume_factor × avg_volume are 'confirmed'.

    Returns
    -------
    pd.DataFrame
        Original DataFrame extended with: channel_high, channel_low,
        avg_volume, signal, confirmed_breakout.
    """
    df = df.copy().sort_values("datetime", ignore_index=True)

    # ── Donchian Channel (lagged by 1 bar to prevent look-ahead bias) ────────
    df["channel_high"] = df["high"].rolling(window).max().shift(1)
    df["channel_low"]  = df["low"].rolling(window).min().shift(1)

    # ── Rolling average volume ───────────────────────────────────────────────
    df["avg_volume"] = df["volume"].rolling(window).mean().shift(1)

    # ── Breakout signals ─────────────────────────────────────────────────────
    df["signal"] = np.where(
        df["close"] > df["channel_high"], 1,
        np.where(df["close"] < df["channel_low"], -1, 0)
    )

    # ── Volume confirmation flag ─────────────────────────────────────────────
    df["confirmed_breakout"] = (
        (df["signal"] != 0) &
        (df["volume"] > volume_factor * df["avg_volume"])
    )

    return df

df_signals = breakout_detection(df, window=20, volume_factor=1.5)

print("--- Signal Distribution ---")
print(df_signals["signal"].value_counts())
print("\n--- Confirmed Breakouts ---")
print(df_signals["confirmed_breakout"].value_counts())

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 + Breakout Signals + Donchian Channel", "Volume"],
    row_heights=[0.65, 0.35])

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=df_signals["datetime"], y=df_signals["channel_high"],
    mode="lines", name="Channel High", line=dict(color="red", width=1, dash="dash")),
    row=1, col=1)
fig.add_trace(go.Scatter(
    x=df_signals["datetime"], y=df_signals["channel_low"],
    mode="lines", name="Channel Low", line=dict(color="green", width=1, dash="dash")),
    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="Breakout Up (+1)"), 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="Breakout Down (-1)"), row=1, col=1)

colors = ["green" if s == 1 else "red" if s == -1 else "gray" for s in df_signals["signal"]]
fig.add_trace(go.Bar(
    x=df_signals["datetime"], y=df_signals["volume"],
    marker_color=colors, name="Volume"), row=2, col=1)

fig.update_layout(
    title_text="Price Breakout Detection",
    xaxis_rangeslider_visible=False,
    height=700, xaxis2_title="Datetime",
)
fig.show()