indicators.py

import numpy as np
import pandas as pd


def _wma(series: pd.Series, length: int) -> pd.Series:
    weights = np.arange(1, length + 1)
    return series.rolling(length).apply(
        lambda x: np.dot(x, weights) / weights.sum(),
        raw=True
    )


def add_strategy_indicators(df: pd.DataFrame) -> pd.DataFrame:
    out = df.copy().sort_values("timestamp").reset_index(drop=True)

    # Indicator 2
    out["ema9"] = out["close"].ewm(span=9, adjust=False).mean()
    out["ema21"] = out["close"].ewm(span=21, adjust=False).mean()

    # Indicator 1: BB Stops using WMA(20) on OHLC4, mult=1.0
    out["ohlc4"] = (out["open"] + out["high"] + out["low"] + out["close"]) / 4.0
    out["bb_basis"] = _wma(out["ohlc4"], 20)
    out["bb_dev"] = out["ohlc4"].rolling(20).std(ddof=0) * 1.0
    out["bb_upper"] = out["bb_basis"] + out["bb_dev"]
    out["bb_lower"] = out["bb_basis"] - out["bb_dev"]

    up_flags = []
    down_flags = []
    phases = []
    change_up = []
    change_down = []

    prev_up = False
    prev_down = False

    for i in range(len(out)):
        curr_up = prev_up
        curr_down = prev_down

        if i > 0:
            prev_src = out.loc[i - 1, "ohlc4"]
            curr_src = out.loc[i, "ohlc4"]
            prev_upper = out.loc[i - 1, "bb_upper"]
            curr_upper = out.loc[i, "bb_upper"]
            prev_lower = out.loc[i - 1, "bb_lower"]
            curr_lower = out.loc[i, "bb_lower"]

            if pd.notna(prev_upper) and pd.notna(curr_upper) and pd.notna(prev_lower) and pd.notna(curr_lower):
                crossover = (prev_src <= prev_upper) and (curr_src > curr_upper)
                crossunder = (prev_src >= prev_lower) and (curr_src < curr_lower)

                if crossover:
                    curr_up = True
                    curr_down = False

                if crossunder:
                    curr_up = False
                    curr_down = True

        up_flags.append(curr_up)
        down_flags.append(curr_down)
        phases.append("green" if curr_up else "red" if curr_down else None)
        change_up.append(prev_down and curr_up)
        change_down.append(prev_up and curr_down)

        prev_up = curr_up
        prev_down = curr_down

    out["bb_up"] = up_flags
    out["bb_down"] = down_flags
    out["bb_phase"] = phases
    out["bb_change_up"] = change_up
    out["bb_change_down"] = change_down

    return out