Spaces:
Paused
Paused
| """Indicator ports (Order Block MTF) + the legacy strategy-indicator helper. | |
| This package is imported TWO ways across the codebase, and must serve both: | |
| * ``from indicators import add_strategy_indicators`` — the NN / SMC / jfghla / | |
| retracement / trending / family LIVE bots (19+ modules); | |
| * ``from indicators import luxalgo_smc, bigbeluga_pa`` — the Order-Block-MTF | |
| backtest (``strategy/orderblock_mtf.py``). | |
| A package SHADOWS a same-named module, so once this ``indicators/`` package exists | |
| ``indicators.py`` is never importable — its ``add_strategy_indicators`` therefore | |
| lives here. (An untracked copy of this package, committed 2026-07-25, once broke | |
| every bot doing ``from indicators import add_strategy_indicators`` for exactly this | |
| reason.) | |
| Ports: | |
| * ``luxalgo_smc`` -> Smart Money Concepts [LuxAlgo] (CC BY-NC-SA 4.0) | |
| * ``bigbeluga_pa`` -> BigBeluga - Smart Money Concepts (MPL 2.0) | |
| """ | |
| import numpy as np | |
| import pandas as pd | |
| from . import luxalgo_smc, bigbeluga_pa # noqa: F401 (submodules import only numpy/pandas) | |
| __all__ = ["add_strategy_indicators", "luxalgo_smc", "bigbeluga_pa"] | |
| 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, ema_fast: int = 5, ema_slow: int = 9) -> pd.DataFrame: | |
| out = df.copy().sort_values("timestamp").reset_index(drop=True) | |
| # Indicator 2. Column names stay ema5/ema9 (fast/slow roles) regardless of the | |
| # configured span so existing NN feature names keep working when the period changes. | |
| out["ema5"] = out["close"].ewm(span=ema_fast, adjust=False).mean() | |
| out["ema9"] = out["close"].ewm(span=ema_slow, 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 | |