#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ================================================================================ ADVANCED FOREX & CRYPTO TRADING INDICATOR ========================================== Platform : Hugging Face Free Tier (16 GB RAM | 2 CPU Cores | 50 GB Storage) Version : 3.0.0 Timezone : UTC+5:30 (Asia/Kolkata — IST) Type : Indicator ONLY · No trade execution · No automation Markets : All Forex Pairs + All Major Crypto Pairs Timeframes: 30s · 1m · 2m · 3m · 5m · 10m · 15m · 30m · 1h · 2h File : Single Python file — No external project modules ================================================================================ SIGNAL TYPES 🟢 BUY — Strong bullish confluence across multiple methods 🔴 SELL — Strong bearish confluence across multiple methods ⚪ NEUTRAL — No clear direction / conflicting / low-confidence signals ANALYTICAL ENGINE (10 layers) 1. Trend Analysis — EMA/SMA/HMA/DEMA/TEMA/FRAMA stacks 2. Momentum Analysis — RSI · StochRSI · MACD · CCI · WillR · UO · TSI 3. Volatility Analysis — BB · Keltner · Donchian · ATR · HV 4. Volume Analysis — OBV · VWAP · CMF · MFI · Force · EOM 5. Candlestick Patterns — 20+ single / 2-bar / 3-bar patterns 6. Support & Resistance — Pivot points · Fibonacci · Dynamic S/R 7. Wavelet Decomposition — PyWavelets noise filtering + trend extraction 8. Fourier Cycle Analysis— FFT dominant cycle + phase detection 9. Market Regime — Trending / Ranging / Breakout / Volatile 10. ML Ensemble — RF + ET + XGB + LGB + MLP + LSTM (PyTorch/TF) ================================================================================ INSTALLATION (run once) -------------------------------------------------------------------------------- pip install numpy pandas scipy scikit-learn joblib statsmodels pip install xgboost lightgbm pip install pandas-ta yfinance ccxt pip install PyWavelets numba pip install matplotlib plotly pip install rich colorama tabulate pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu pip install tensorflow-cpu pip install pyarrow optuna shap pip install psutil requests websocket-client TOTAL ≈ 10–12 GB (PyTorch CPU + TensorFlow CPU dominate) -------------------------------------------------------------------------------- HUGGING FACE FREE TIER DEPLOYMENT 1. Create a new Space → SDK: Gradio or Streamlit 2. Upload this single file (advanced_indicator.py) 3. Create requirements.txt with the packages above 4. Set Python 3.10+ runtime 5. Run: python advanced_indicator.py --demo Or: python advanced_indicator.py --interactive Or: python advanced_indicator.py --symbol EURUSD=X --tf 15m PERFORMANCE NOTES FOR HUGGING FACE FREE TIER • ML training uses max 2 CPU cores (n_jobs=2) • LSTM uses CPU-only PyTorch / TensorFlow-CPU • Data cache avoids repeated API calls (60 s TTL) • gc.collect() after each analysis to free RAM • Numba JIT compiles hot loops on first call (warm-up ~2 s) • Use --lookback 300 instead of 500 if RAM is tight • XGBoost hist method is fastest on CPU ================================================================================ USAGE EXAMPLES python advanced_indicator.py # demo mode python advanced_indicator.py --interactive # interactive CLI python advanced_indicator.py --symbol EURUSD=X --tf 15m python advanced_indicator.py --symbol BTC-USD --tf 5m python advanced_indicator.py --scan forex --tf 15m --top 5 python advanced_indicator.py --scan crypto --tf 5m --top 5 python advanced_indicator.py --watch EURUSD=X --tf 1m --interval 60 python advanced_indicator.py --list ================================================================================ """ # ============================================================================== # ░░ SECTION 0 — STANDARD LIBRARY IMPORTS ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== import os import sys import gc import time import json import math import warnings import logging import traceback import threading import concurrent.futures from collections import defaultdict, deque from datetime import datetime, timedelta, timezone as _tz from pathlib import Path from typing import Dict, List, Optional, Tuple, Union, Any # Silence noisy C++ / TF logs before any import warnings.filterwarnings("ignore") os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3" os.environ["PYTHONWARNINGS"] = "ignore" os.environ["TOKENIZERS_PARALLELISM"] = "false" os.environ["OMP_NUM_THREADS"] = "2" os.environ["MKL_NUM_THREADS"] = "2" os.environ["OPENBLAS_NUM_THREADS"] = "2" os.environ["TF_ENABLE_ONEDNN_OPTS"] = "0" # ============================================================================== # ░░ SECTION 1 — THIRD-PARTY IMPORTS WITH GRACEFUL FALLBACKS ░░░░░░░░░░░░░░░ # ============================================================================== # ─── Core numerical / data ──────────────────────────────────────────────────── import numpy as np import pandas as pd from scipy import stats from scipy.signal import butter, filtfilt, find_peaks, detrend as sp_detrend from scipy.fft import fft as sp_fft, fftfreq as sp_fftfreq # ─── Statistics ─────────────────────────────────────────────────────────────── try: import statsmodels.api as sm from statsmodels.tsa.stattools import adfuller STATSMODELS_OK = True except ImportError: STATSMODELS_OK = False # ─── Wavelet ───────────────────────────────────────────────────────────────── try: import pywt PYWT_OK = True except ImportError: PYWT_OK = False # ─── Machine learning ──────────────────────────────────────────────────────── from sklearn.ensemble import (RandomForestClassifier, ExtraTreesClassifier, GradientBoostingClassifier, IsolationForest) from sklearn.neural_network import MLPClassifier from sklearn.preprocessing import RobustScaler, MinMaxScaler, LabelEncoder from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import accuracy_score from sklearn.decomposition import PCA from sklearn.feature_selection import SelectFromModel import joblib # ─── Gradient boosting ─────────────────────────────────────────────────────── try: import xgboost as xgb XGB_OK = True except ImportError: XGB_OK = False try: import lightgbm as lgb LGB_OK = True except ImportError: LGB_OK = False # ─── PyTorch (CPU) ─────────────────────────────────────────────────────────── try: import torch import torch.nn as nn import torch.optim as torch_optim from torch.utils.data import DataLoader, TensorDataset TORCH_OK = True except ImportError: TORCH_OK = False # ─── TensorFlow / Keras (CPU fallback if torch unavailable) ────────────────── try: import tensorflow as tf tf.get_logger().setLevel("ERROR") from tensorflow.keras.models import Sequential from tensorflow.keras.layers import (LSTM, GRU, Dense, Dropout, BatchNormalization, Conv1D, GlobalAveragePooling1D) from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau from tensorflow.keras.optimizers import Adam as KerasAdam TF_OK = True except Exception: TF_OK = False # ─── Data sources ───────────────────────────────────────────────────────────── try: import yfinance as yf YF_OK = True except ImportError: YF_OK = False try: import ccxt CCXT_OK = True except ImportError: CCXT_OK = False # ─── Performance / JIT ──────────────────────────────────────────────────────── try: from numba import njit NUMBA_OK = True except ImportError: NUMBA_OK = False def njit(fn=None, **kw): # no-op decorator return (lambda f: f)(fn) if fn else (lambda f: f) # ─── Visualisation ─────────────────────────────────────────────────────────── import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from matplotlib.gridspec import GridSpec try: import plotly.graph_objects as go from plotly.subplots import make_subplots PLOTLY_OK = True except ImportError: PLOTLY_OK = False # ─── Rich terminal UI ──────────────────────────────────────────────────────── try: from rich.console import Console from rich.table import Table from rich.panel import Panel from rich.text import Text from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn from rich import box from rich.live import Live RICH_OK = True console = Console() except ImportError: RICH_OK = False console = None # ─── Colour fallback ───────────────────────────────────────────────────────── try: from colorama import Fore, Style as CStyle, init as _cinit _cinit(autoreset=True) CYAN = Fore.CYAN; GREEN = Fore.GREEN RED = Fore.RED; YELLOW = Fore.YELLOW WHITE = Fore.WHITE; RESET = CStyle.RESET_ALL BOLD = CStyle.BRIGHT except ImportError: CYAN = GREEN = RED = YELLOW = WHITE = RESET = BOLD = "" # ─── Arrow / parquet (optional fast I/O) ───────────────────────────────────── try: import pyarrow as pa ARROW_OK = True except ImportError: ARROW_OK = False # ─── SHAP (model explainability) ───────────────────────────────────────────── try: import shap SHAP_OK = True except ImportError: SHAP_OK = False # ─── Optuna (hyperparameter search, optional) ───────────────────────────────── try: import optuna optuna.logging.set_verbosity(optuna.logging.WARNING) OPTUNA_OK = True except ImportError: OPTUNA_OK = False # ─── Tabulate ──────────────────────────────────────────────────────────────── try: from tabulate import tabulate TABULATE_OK = True except ImportError: TABULATE_OK = False # ============================================================================== # ░░ SECTION 2 — GLOBAL CONFIGURATION ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== TIMEZONE = "Asia/Kolkata" # UTC +5:30 # ─── Signal constants ───────────────────────────────────────────────────────── BUY = "BUY" SELL = "SELL" NEUTRAL = "NEUTRAL" # ─── Timeframe definitions ──────────────────────────────────────────────────── # (label → minutes) TF_MINUTES: Dict[str, float] = { "30s": 0.5, "1m": 1, "2m": 2, "3m": 3, "5m": 5, "10m": 10, "15m": 15, "30m": 30, "1h": 60, "2h": 120, } # yfinance only supports certain intervals; map unsupported → closest supported YF_INTERVAL_MAP: Dict[str, str] = { "30s": "1m", # resample 1 m → 30 s (best-effort) "1m": "1m", "2m": "2m", "3m": "5m", # resample 5 m → 3 m "5m": "5m", "10m": "5m", # resample 5 m → 10 m "15m": "15m", "30m": "30m", "1h": "60m", "2h": "60m", # resample 1 h → 2 h } # Pandas resample rule strings (pandas ≥ 2.2 → "min" not "T", "s" not "S") TF_RESAMPLE: Dict[str, str] = { "30s": "30s", "1m": "1min", "2m": "2min", "3m": "3min", "5m": "5min", "10m": "10min","15m": "15min", "30m": "30min", "1h": "60min", "2h": "120min", } # ─── Market universe ────────────────────────────────────────────────────────── FOREX_PAIRS: List[str] = [ "EURUSD=X","GBPUSD=X","USDJPY=X","USDCHF=X","AUDUSD=X", "USDCAD=X","NZDUSD=X","EURGBP=X","EURJPY=X","GBPJPY=X", "AUDJPY=X","CADJPY=X","CHFJPY=X","NZDJPY=X","EURAUD=X", "EURCAD=X","EURCHF=X","EURNZD=X","GBPAUD=X","GBPCAD=X", "GBPCHF=X","GBPNZD=X","AUDCAD=X","AUDCHF=X","AUDNZD=X", "CADCHF=X","NZDCAD=X","NZDCHF=X","SGDJPY=X","USDSGD=X", ] CRYPTO_PAIRS: List[str] = [ "BTC-USD","ETH-USD","BNB-USD","XRP-USD","SOL-USD", "ADA-USD","DOGE-USD","DOT-USD","LTC-USD","AVAX-USD", "LINK-USD","UNI-USD","ATOM-USD","TRX-USD","ETC-USD", "XLM-USD","NEAR-USD","ALGO-USD","VET-USD","MATIC-USD", "ETH-BTC","BNB-BTC","SOL-BTC", ] # ─── Indicator weights for final scoring ────────────────────────────────────── WEIGHTS: Dict[str, float] = { "trend": 0.20, "momentum": 0.18, "volatility": 0.08, "volume": 0.10, "ml": 0.26, "pattern": 0.08, "regime": 0.05, "wavelet": 0.05, } # ─── Signal thresholds ──────────────────────────────────────────────────────── STRONG_BUY = 0.50 MODERATE_BUY = 0.25 MODERATE_SELL= -0.25 STRONG_SELL = -0.50 # ─── Logging ────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.WARNING, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger("INDICATOR") pd.set_option("display.float_format", lambda x: f"{x:.6f}") pd.set_option("display.max_columns", None) # ============================================================================== # ░░ SECTION 3 — UTILITY HELPERS ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== def ist_now() -> str: """Current timestamp in IST (UTC+5:30).""" ist = _tz(timedelta(hours=5, minutes=30)) return datetime.now(tz=ist).strftime("%Y-%m-%d %H:%M:%S IST") def tf_minutes(tf: str) -> float: """Timeframe string → float minutes.""" return TF_MINUTES.get(tf, 5.0) def clip1(x: float) -> float: """Clamp a float to [-1, 1].""" return max(-1.0, min(1.0, float(x))) def safe_last(arr, default: float = 0.0) -> float: """Return last finite value from array-like, or default.""" if arr is None: return default a = np.asarray(arr, dtype=np.float64).ravel() finite = a[np.isfinite(a)] return float(finite[-1]) if len(finite) else default def pct_rank(series: pd.Series, window: int = 50) -> pd.Series: """Rolling percentile rank normalised to [0, 1].""" return series.rolling(window, min_periods=5).rank(pct=True) def norm_series(series: pd.Series, window: int = 50) -> pd.Series: """ Normalise series to [-1, +1] using rolling min/max. Values outside the window clamp to ±1. """ lo = series.rolling(window, min_periods=5).min() hi = series.rolling(window, min_periods=5).max() rng = (hi - lo).replace(0, np.nan) return ((2 * (series - lo) / rng) - 1).clip(-1, 1).fillna(0) def free_memory() -> None: """Release unused memory (important on 16 GB HF tier).""" gc.collect() if TORCH_OK and torch.cuda.is_available(): torch.cuda.empty_cache() def mem_mb() -> float: """Current process RSS in MB.""" try: import psutil return psutil.Process(os.getpid()).memory_info().rss / 1_048_576 except Exception: return 0.0 # ============================================================================== # ░░ SECTION 4 — DATA FETCHER ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class DataFetcher: """ Fetch OHLCV candles from multiple sources. Priority -------- 1. yfinance — supports Forex (via Yahoo) and Crypto 2. ccxt — Binance fallback for Crypto All returned DataFrames: - columns : Open · High · Low · Close · Volume (float64) - index : DatetimeTZAware (Asia/Kolkata = UTC+5:30) - sorted : ascending (oldest → newest) """ _CACHE: Dict[str, Tuple[pd.DataFrame, float]] = {} _TTL = 60 # seconds before cache expires _LOCK = threading.Lock() def fetch( self, symbol: str, tf: str = "15m", lookback: int = 500, refresh: bool = False, ) -> Optional[pd.DataFrame]: """Main entry — returns OHLCV DataFrame or None.""" key = f"{symbol}|{tf}" now = time.time() with self._LOCK: if not refresh and key in self._CACHE: df_cached, ts = self._CACHE[key] if now - ts < self._TTL: return df_cached df = None if YF_OK: df = self._yfinance(symbol, tf, lookback) if df is None and CCXT_OK and self._is_crypto(symbol): df = self._ccxt(symbol, tf, lookback) if df is None: log.warning("No data for %s [%s]", symbol, tf) return None df = self._to_ist(df).sort_index() df = df.iloc[-lookback:] # keep last N bars if len(df) < 50: return None with self._LOCK: self._CACHE[key] = (df, now) return df # ── private ────────────────────────────────────────────────────────────── def _yfinance(self, symbol: str, tf: str, lookback: int) -> Optional[pd.DataFrame]: try: yf_iv = YF_INTERVAL_MAP.get(tf, "5m") tf_min = tf_minutes(tf) days = max(1, int(tf_min * lookback * 1.3 / (60 * 24))) # yfinance caps: < 1h → 7 d, < 4h → 60 d if tf_min < 60: days = min(days, 7) elif tf_min < 240: days = min(days, 59) raw = yf.Ticker(symbol).history( period=f"{days}d", interval=yf_iv, auto_adjust=True ) if raw is None or raw.empty: return None df = raw[["Open","High","Low","Close","Volume"]].copy() df.index = pd.to_datetime(df.index) return self._resample(df, tf, yf_iv) except Exception as e: log.debug("yfinance error [%s]: %s", symbol, e) return None def _ccxt(self, symbol: str, tf: str, lookback: int) -> Optional[pd.DataFrame]: try: ccxt_sym = symbol.replace("-", "/").replace("=X", "") ccxt_tf = {"30s":"1m","3m":"3m","10m":"15m"}.get(tf, tf) since = int( (datetime.utcnow() - timedelta( minutes=tf_minutes(tf) * lookback * 1.5 )).timestamp() * 1000 ) ex = ccxt.binance({"enableRateLimit": True}) ohlcv = ex.fetch_ohlcv(ccxt_sym, ccxt_tf, since=since, limit=lookback) if not ohlcv: return None df = pd.DataFrame( ohlcv, columns=["ts","Open","High","Low","Close","Volume"] ) df.index = pd.to_datetime(df["ts"], unit="ms", utc=True) df = df[["Open","High","Low","Close","Volume"]] return self._resample(df, tf, ccxt_tf) except Exception as e: log.debug("ccxt error: %s", e) return None @staticmethod def _resample(df: pd.DataFrame, target_tf: str, source_tf: str) -> pd.DataFrame: src_min = tf_minutes(source_tf) tgt_min = tf_minutes(target_tf) if tgt_min <= src_min: return df rule = TF_RESAMPLE.get(target_tf, f"{int(tgt_min)}T") return ( df.resample(rule) .agg({"Open":"first","High":"max","Low":"min", "Close":"last","Volume":"sum"}) .dropna(subset=["Open","Close"]) ) @staticmethod def _to_ist(df: pd.DataFrame) -> pd.DataFrame: try: if df.index.tz is None: df.index = df.index.tz_localize("UTC") df.index = df.index.tz_convert(TIMEZONE) except Exception: pass return df @staticmethod def _is_crypto(sym: str) -> bool: return any(x in sym for x in ["-USD","-USDT","-BTC","-ETH"]) # ============================================================================== # ░░ SECTION 5 — TECHNICAL INDICATORS ENGINE ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class Indicators: """ Compute 35+ technical indicators. All methods accept optional `src` override and return numpy float64 arrays of the same length as the input DataFrame. Nans fill the warm-up period — never raise exceptions. """ def __init__(self, df: pd.DataFrame): self.df = df self.o = df["Open"].to_numpy(dtype=np.float64) self.h = df["High"].to_numpy(dtype=np.float64) self.l = df["Low"].to_numpy(dtype=np.float64) self.c = df["Close"].to_numpy(dtype=np.float64) self.v = df["Volume"].to_numpy(dtype=np.float64) self.n = len(df) # ── Moving averages ─────────────────────────────────────────────────────── def ema(self, p: int, src: Optional[np.ndarray] = None) -> np.ndarray: s = pd.Series(self.c if src is None else src) return s.ewm(span=p, adjust=False).mean().to_numpy(dtype=np.float64) def sma(self, p: int, src: Optional[np.ndarray] = None) -> np.ndarray: s = pd.Series(self.c if src is None else src) return s.rolling(p).mean().to_numpy(dtype=np.float64) def wma(self, p: int, src: Optional[np.ndarray] = None) -> np.ndarray: s = pd.Series(self.c if src is None else src) w = np.arange(1, p + 1, dtype=np.float64) return s.rolling(p).apply(lambda x: (x * w).sum() / w.sum(), raw=True)\ .to_numpy(dtype=np.float64) def hma(self, p: int) -> np.ndarray: """Hull Moving Average — reactive + smooth.""" hp = max(2, p // 2) sqrp = max(2, int(np.sqrt(p))) return self.wma(sqrp, src=2 * self.wma(hp) - self.wma(p)) def dema(self, p: int) -> np.ndarray: """Double EMA — reduces lag.""" e = self.ema(p) return 2 * e - pd.Series(e).ewm(span=p, adjust=False).mean().to_numpy() def tema(self, p: int) -> np.ndarray: """Triple EMA — minimal lag.""" e1 = self.ema(p) e2 = pd.Series(e1).ewm(span=p, adjust=False).mean().to_numpy() e3 = pd.Series(e2).ewm(span=p, adjust=False).mean().to_numpy() return 3 * e1 - 3 * e2 + e3 def vwma(self, p: int) -> np.ndarray: """Volume Weighted MA.""" cv = self.c * self.v scv = pd.Series(cv).rolling(p).sum().to_numpy() sv = pd.Series(self.v).rolling(p).sum().to_numpy() return np.where(sv > 0, scv / sv, np.nan) def frama(self, p: int = 16) -> np.ndarray: """ Fractal Adaptive MA (Ehlers). The EMA coefficient α adapts to the market's fractal dimension D. D near 1 → trending → fast α; D near 2 → noisy → slow α. """ c = self.c n = self.n out = np.full(n, np.nan) if n < p * 2: return out for i in range(p * 2, n): w = c[i - p * 2 : i] h1 = w[:p].max(); l1 = w[:p].min() h2 = w[p:].max(); l2 = w[p:].min() h = w.max(); l = w.min() n1 = (h1 - l1) / p n2 = (h2 - l2) / p n3 = (h - l ) / (p * 2) if n1 > 0 and n2 > 0 and n3 > 0: D = (np.log(n1 + n2) - np.log(n3)) / np.log(2) alpha = np.clip(np.exp(-4.6 * (D - 1)), 0.01, 1.0) else: alpha = 0.1 out[i] = alpha * c[i] + (1 - alpha) * (out[i-1] if np.isfinite(out[i-1]) else c[i]) return out # ── Momentum indicators ─────────────────────────────────────────────────── def rsi(self, p: int = 14) -> np.ndarray: """Wilder RSI.""" d = pd.Series(self.c).diff() gain = d.clip(lower=0).ewm(span=p, adjust=False).mean() loss = (-d.clip(upper=0)).ewm(span=p, adjust=False).mean() rs = gain / loss.replace(0, np.nan) return (100 - 100 / (1 + rs)).to_numpy(dtype=np.float64) def stoch_rsi( self, rsi_p: int = 14, stoch_p: int = 14, k_p: int = 3, d_p: int = 3 ) -> Tuple[np.ndarray, np.ndarray]: """Stochastic RSI — K and D lines.""" r = pd.Series(self.rsi(rsi_p)) lo = r.rolling(stoch_p).min() hi = r.rolling(stoch_p).max() rng = (hi - lo).replace(0, np.nan) stk = (100 * (r - lo) / rng).rolling(k_p).mean() std_ = stk.rolling(d_p).mean() return stk.to_numpy(dtype=np.float64), std_.to_numpy(dtype=np.float64) def macd( self, fast: int = 12, slow: int = 26, sig: int = 9 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """MACD line · signal line · histogram.""" c = pd.Series(self.c) ml = c.ewm(span=fast, adjust=False).mean() - \ c.ewm(span=slow, adjust=False).mean() sl = ml.ewm(span=sig, adjust=False).mean() return ( ml.to_numpy(dtype=np.float64), sl.to_numpy(dtype=np.float64), (ml - sl).to_numpy(dtype=np.float64), ) def cci(self, p: int = 20) -> np.ndarray: """Commodity Channel Index.""" tp = pd.Series((self.h + self.l + self.c) / 3) mean = tp.rolling(p).mean() mad = tp.rolling(p).apply(lambda x: np.abs(x - x.mean()).mean(), raw=True) return ((tp - mean) / (0.015 * mad.replace(0, np.nan))).to_numpy(dtype=np.float64) def williams_r(self, p: int = 14) -> np.ndarray: """Williams %R.""" hi = pd.Series(self.h).rolling(p).max() lo = pd.Series(self.l).rolling(p).min() rng = (hi - lo).replace(0, np.nan) return (-100 * (hi - pd.Series(self.c)) / rng).to_numpy(dtype=np.float64) def roc(self, p: int = 12) -> np.ndarray: """Rate of Change (%).""" c = pd.Series(self.c) return ((c - c.shift(p)) / c.shift(p) * 100).to_numpy(dtype=np.float64) def mfi(self, p: int = 14) -> np.ndarray: """Money Flow Index.""" tp = (self.h + self.l + self.c) / 3 rmf = tp * self.v dir_ = np.sign(np.diff(tp, prepend=tp[0])) pos = pd.Series(np.where(dir_ > 0, rmf, 0.0)).rolling(p).sum() neg = pd.Series(np.where(dir_ <= 0, rmf, 0.0)).rolling(p).sum() mfr = pos / neg.replace(0, np.nan) return (100 - 100 / (1 + mfr)).to_numpy(dtype=np.float64) def uo(self, p1: int = 7, p2: int = 14, p3: int = 28) -> np.ndarray: """Ultimate Oscillator.""" pc = np.roll(self.c, 1); pc[0] = self.c[0] tr = np.maximum(self.h - self.l, np.maximum(np.abs(self.h - pc), np.abs(self.l - pc))) bp = self.c - np.minimum(self.l, pc) def avg(p): return pd.Series(bp).rolling(p).sum() / \ pd.Series(tr).rolling(p).sum() return ((100 * (4*avg(p1) + 2*avg(p2) + avg(p3)) / 7) .to_numpy(dtype=np.float64)) def tsi(self, r: int = 25, s: int = 13) -> np.ndarray: """True Strength Index.""" d = pd.Series(self.c).diff() sm = d.ewm(span=r).mean().ewm(span=s).mean() ab = d.abs().ewm(span=r).mean().ewm(span=s).mean() return (100 * sm / ab.replace(0, np.nan)).to_numpy(dtype=np.float64) # ── Volatility indicators ───────────────────────────────────────────────── def atr(self, p: int = 14) -> np.ndarray: """Average True Range (Wilder smoothed).""" pc = np.roll(self.c, 1); pc[0] = self.c[0] tr = np.maximum(self.h - self.l, np.maximum(np.abs(self.h - pc), np.abs(self.l - pc))) return pd.Series(tr).ewm(span=p, adjust=False).mean().to_numpy(dtype=np.float64) def bollinger( self, p: int = 20, k: float = 2.0 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """Bollinger Bands → upper, mid, lower.""" mid = pd.Series(self.c).rolling(p).mean() std = pd.Series(self.c).rolling(p).std() return (mid + k*std).to_numpy(), mid.to_numpy(), (mid - k*std).to_numpy() def keltner( self, p: int = 20, mult: float = 2.0 ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """Keltner Channels.""" mid = self.ema(p) atr_ = self.atr(p) return mid + mult*atr_, mid, mid - mult*atr_ def donchian(self, p: int = 20) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """Donchian Channels.""" hi = pd.Series(self.h).rolling(p).max().to_numpy() lo = pd.Series(self.l).rolling(p).min().to_numpy() return hi, (hi + lo) / 2, lo def bb_pct_b(self, p: int = 20, k: float = 2.0) -> np.ndarray: """Bollinger %B — position within bands [0..1].""" u, m, lo = self.bollinger(p, k) rng = u - lo return np.where(rng > 0, (self.c - lo) / rng, 0.5) def bb_width(self, p: int = 20, k: float = 2.0) -> np.ndarray: """Bollinger Band Width — volatility expansion proxy.""" u, m, lo = self.bollinger(p, k) return np.where(m > 0, (u - lo) / m, np.nan) def hist_vol(self, p: int = 20) -> np.ndarray: """Annualised historical (close-to-close) volatility.""" lr = np.log(pd.Series(self.c) / pd.Series(self.c).shift(1)) hv = lr.rolling(p).std() * np.sqrt(252 * 390) # ≈ intraday annualised return hv.to_numpy(dtype=np.float64) # ── Trend / directional ─────────────────────────────────────────────────── def adx(self, p: int = 14) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """ADX, +DI, −DI.""" pc = np.roll(self.c, 1); pc[0] = self.c[0] ph = np.roll(self.h, 1); ph[0] = self.h[0] pl = np.roll(self.l, 1); pl[0] = self.l[0] tr = np.maximum(self.h - self.l, np.maximum(np.abs(self.h - pc), np.abs(self.l - pc))) pdm = np.where((self.h - ph) > (pl - self.l), np.maximum(self.h - ph, 0), 0) ndm = np.where((pl - self.l) > (self.h - ph), np.maximum(pl - self.l, 0), 0) atr_s = pd.Series(tr).ewm(span=p, adjust=False).mean() pdi = 100 * pd.Series(pdm).ewm(span=p, adjust=False).mean() / atr_s.replace(0, np.nan) ndi = 100 * pd.Series(ndm).ewm(span=p, adjust=False).mean() / atr_s.replace(0, np.nan) dx = 100 * (pdi - ndi).abs() / (pdi + ndi).replace(0, np.nan) adx_ = dx.ewm(span=p, adjust=False).mean() return (adx_.to_numpy(dtype=np.float64), pdi.to_numpy(dtype=np.float64), ndi.to_numpy(dtype=np.float64)) def aroon(self, p: int = 25) -> Tuple[np.ndarray, np.ndarray]: """Aroon Up & Down.""" up = pd.Series(self.h).rolling(p+1).apply( lambda x: (np.argmax(x) / p) * 100, raw=True) down = pd.Series(self.l).rolling(p+1).apply( lambda x: (np.argmin(x) / p) * 100, raw=True) return up.to_numpy(dtype=np.float64), down.to_numpy(dtype=np.float64) def supertrend( self, p: int = 7, mult: float = 3.0 ) -> Tuple[np.ndarray, np.ndarray]: """ Supertrend indicator. Returns (supertrend_line, direction): +1 = bullish, −1 = bearish. """ atr_ = self.atr(p) hl2 = (self.h + self.l) / 2 ub = hl2 + mult * atr_ lb = hl2 - mult * atr_ st = np.full(self.n, np.nan) dr = np.ones(self.n) for i in range(1, self.n): if not np.isfinite(atr_[i]): continue lb[i] = max(lb[i], lb[i-1]) if self.c[i-1] > lb[i-1] else lb[i] ub[i] = min(ub[i], ub[i-1]) if self.c[i-1] < ub[i-1] else ub[i] prev = st[i-1] if not np.isfinite(prev): st[i] = lb[i]; continue if prev == ub[i-1]: st[i] = lb[i] if self.c[i] > ub[i] else ub[i] else: st[i] = ub[i] if self.c[i] < lb[i] else lb[i] dr[i] = 1.0 if st[i] < self.c[i] else -1.0 return st, dr def parabolic_sar( self, step: float = 0.02, max_af: float = 0.2 ) -> Tuple[np.ndarray, np.ndarray]: """Parabolic SAR. Returns (sar, direction) +1/−1.""" sar = np.zeros(self.n); dr = np.ones(self.n) bull = True; af = step; ep = self.l[0] sar[0] = self.h[0] for i in range(1, self.n): ps = sar[i-1] if bull: sar[i] = ps + af * (ep - ps) sar[i] = min(sar[i], self.l[i-1], self.l[i-2] if i > 1 else self.l[0]) if self.l[i] < sar[i]: bull = False; sar[i] = ep; ep = self.l[i]; af = step else: if self.h[i] > ep: ep = self.h[i]; af = min(af + step, max_af) else: sar[i] = ps + af * (ep - ps) sar[i] = max(sar[i], self.h[i-1], self.h[i-2] if i > 1 else self.h[0]) if self.h[i] > sar[i]: bull = True; sar[i] = ep; ep = self.h[i]; af = step else: if self.l[i] < ep: ep = self.l[i]; af = min(af + step, max_af) dr[i] = 1.0 if bull else -1.0 return sar, dr def ichimoku(self) -> Dict[str, np.ndarray]: """Full Ichimoku Cloud components.""" def dc(p): hi = pd.Series(self.h).rolling(p).max() lo = pd.Series(self.l).rolling(p).min() return ((hi + lo) / 2).to_numpy(dtype=np.float64) t = dc(9); k = dc(26) sa = np.roll((t + k) / 2, 26) sb = np.roll(dc(52), 26) ch = np.roll(self.c, -26) sa[:26] = np.nan; sb[:26] = np.nan; ch[-26:] = np.nan return {"tenkan": t, "kijun": k, "senkou_a": sa, "senkou_b": sb, "chikou": ch} # ── Volume indicators ───────────────────────────────────────────────────── def obv(self) -> np.ndarray: """On-Balance Volume.""" d = np.sign(np.diff(self.c, prepend=self.c[0])) return np.cumsum(d * self.v) def vwap(self) -> np.ndarray: """Cumulative intraday VWAP.""" tp = (self.h + self.l + self.c) / 3 ctv = np.cumsum(tp * self.v) cv = np.cumsum(self.v) return np.where(cv > 0, ctv / cv, tp) def cmf(self, p: int = 20) -> np.ndarray: """Chaikin Money Flow.""" rng = np.where(self.h - self.l > 0, self.h - self.l, 1e-10) mfv = ((self.c - self.l) - (self.h - self.c)) / rng * self.v sums = pd.Series(mfv).rolling(p).sum() sumv = pd.Series(self.v).rolling(p).sum().replace(0, np.nan) return (sums / sumv).to_numpy(dtype=np.float64) def force_index(self, p: int = 13) -> np.ndarray: """Elder's Force Index.""" fi = np.diff(self.c, prepend=self.c[0]) * self.v return pd.Series(fi).ewm(span=p, adjust=False).mean().to_numpy(dtype=np.float64) def vol_osc(self, fast: int = 5, slow: int = 10) -> np.ndarray: """Volume Oscillator (fast EMA − slow EMA) / slow EMA × 100.""" v = pd.Series(self.v) fv = v.ewm(span=fast).mean() sv = v.ewm(span=slow).mean() return ((fv - sv) / sv.replace(0, np.nan) * 100).to_numpy(dtype=np.float64) def eom(self, p: int = 14) -> np.ndarray: """Ease of Movement.""" ph = np.roll(self.h, 1); ph[0] = self.h[0] pl = np.roll(self.l, 1); pl[0] = self.l[0] dm = (self.h + self.l) / 2 - (ph + pl) / 2 br = (self.v / 1e6) / (self.h - self.l + 1e-10) return pd.Series(dm / (br + 1e-10)).rolling(p).mean().to_numpy(dtype=np.float64) # ── Divergence detection ────────────────────────────────────────────────── def rsi_divergence(self, p: int = 14, lookback: int = 30) -> float: """ Detect RSI divergence over the last `lookback` bars. Returns: +1.0 = bullish divergence, −1.0 = bearish, 0.0 = none. """ if self.n < lookback: return 0.0 c_w = self.c[-lookback:] rsi_w = self.rsi(p)[-lookback:] # Price peaks / troughs price_peaks, _ = find_peaks( c_w, distance=5) price_trough, _ = find_peaks(-c_w, distance=5) rsi_peaks, _ = find_peaks( rsi_w, distance=5) rsi_trough, _ = find_peaks(-rsi_w, distance=5) score = 0.0 # Bearish divergence: price higher high, RSI lower high if len(price_peaks) >= 2 and len(rsi_peaks) >= 2: if (c_w[price_peaks[-1]] > c_w[price_peaks[-2]] and rsi_w[rsi_peaks[-1]] < rsi_w[rsi_peaks[-2]]): score = -1.0 # Bullish divergence: price lower low, RSI higher low if len(price_trough) >= 2 and len(rsi_trough) >= 2: if (c_w[price_trough[-1]] < c_w[price_trough[-2]] and rsi_w[rsi_trough[-1]] > rsi_w[rsi_trough[-2]]): score = 1.0 return score # ── Support / resistance (pivot-based) ─────────────────────────────────── def pivot_points(self) -> Dict[str, float]: """Classic floor pivot points from previous candle.""" i = max(-2, -self.n) h, l, c = self.h[i], self.l[i], self.c[i] pp = (h + l + c) / 3 return { "PP": pp, "R1": 2*pp - l, "R2": pp + (h-l), "R3": h + 2*(pp-l), "S1": 2*pp - h, "S2": pp - (h-l), "S3": l - 2*(h-pp), } def fibonacci_levels(self, lookback: int = 50) -> Dict[str, float]: """Fibonacci retracement levels from recent swing H/L.""" hw = self.h[-lookback:].max() lw = self.l[-lookback:].min() d = hw - lw return { "0.0": lw, "0.236": lw + .236*d, "0.382": lw + .382*d,"0.5": lw + .5*d, "0.618": lw + .618*d,"0.786": lw + .786*d, "1.0": hw, } # ── Master compute ──────────────────────────────────────────────────────── def compute_all(self) -> Dict[str, np.ndarray]: """ Compute every indicator and return a flat dict of arrays. This dict is the single source of truth for all downstream components. """ d: Dict[str, np.ndarray] = {} # Moving averages for p in [8, 13, 21, 34, 55, 89, 200]: d[f"ema_{p}"] = self.ema(p) d[f"sma_{p}"] = self.sma(p) d["hma_21"] = self.hma(21) d["dema_21"] = self.dema(21) d["tema_21"] = self.tema(21) d["vwma_20"] = self.vwma(20) d["frama"] = self.frama(16) # Momentum d["rsi_14"] = self.rsi(14) d["rsi_7"] = self.rsi(7) d["rsi_21"] = self.rsi(21) sk, sd = self.stoch_rsi() d["stoch_k"], d["stoch_d"] = sk, sd ml, ms, mh = self.macd() d["macd"] = ml; d["macd_sig"] = ms; d["macd_hist"] = mh d["cci_20"] = self.cci(20) d["wr"] = self.williams_r(14) d["roc_12"] = self.roc(12) d["mfi_14"] = self.mfi(14) d["uo"] = self.uo() d["tsi"] = self.tsi() # Volatility d["atr_14"] = self.atr(14) bu, bm, bl = self.bollinger() d["bb_u"] = bu; d["bb_m"] = bm; d["bb_l"] = bl d["bb_pct"] = self.bb_pct_b() d["bb_w"] = self.bb_width() ku, km, kl = self.keltner() d["kc_u"] = ku; d["kc_m"] = km; d["kc_l"] = kl du, dm, dl = self.donchian() d["dc_u"] = du; d["dc_m"] = dm; d["dc_l"] = dl d["hv"] = self.hist_vol() # Trend adx_, pdi, ndi = self.adx() d["adx"] = adx_; d["pdi"] = pdi; d["ndi"] = ndi ar_u, ar_d = self.aroon() d["aroon_u"] = ar_u; d["aroon_d"] = ar_d st, st_dir = self.supertrend() d["supertrend"] = st; d["st_dir"] = st_dir sar_, sar_dir = self.parabolic_sar() d["sar"] = sar_; d["sar_dir"] = sar_dir ich = self.ichimoku() for k, v in ich.items(): d[f"ichi_{k}"] = v # Volume d["obv"] = self.obv() d["vwap"] = self.vwap() d["cmf"] = self.cmf() d["force"] = self.force_index() d["vol_osc"] = self.vol_osc() d["eom"] = self.eom() return d # ============================================================================== # ░░ SECTION 6 — CANDLESTICK PATTERN DETECTOR ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class CandlePatterns: """ Detect 22 classic candlestick formations on the last 3 bars. Pattern score convention ───────────────────────── +1 → bullish reversal / continuation −1 → bearish reversal / continuation 0 → indecision / neutral """ def __init__(self, df: pd.DataFrame): self.o = df["Open"].to_numpy(np.float64) self.h = df["High"].to_numpy(np.float64) self.l = df["Low"].to_numpy(np.float64) self.c = df["Close"].to_numpy(np.float64) self.n = len(df) # helpers def _body(self, i): return abs(self.c[i] - self.o[i]) def _range(self, i): return self.h[i] - self.l[i] def _up_sh(self, i): return self.h[i] - max(self.c[i], self.o[i]) def _lo_sh(self, i): return min(self.c[i], self.o[i]) - self.l[i] def _bull(self, i): return self.c[i] > self.o[i] def _bear(self, i): return self.c[i] < self.o[i] def detect(self) -> Dict[str, int]: """Run all detectors; return {pattern_name: score}.""" i = self.n - 1 if i < 2: return {} j = i - 1; k = i - 2 out: Dict[str, int] = {} body_i = self._body(i); rng_i = self._range(i) or 1e-10 up_sh_i = self._up_sh(i); lo_sh_i = self._lo_sh(i) # ── 1-bar patterns ─────────────────────────────────────────────────── # Doji if body_i < 0.08 * rng_i: out["doji"] = 0 # Dragonfly doji (lo shadow ≥ 2× body, tiny up shadow) — bullish doji if lo_sh_i > 2 * body_i and up_sh_i < 0.3 * body_i + 1e-10: out["dragonfly_doji"] = 1 # Gravestone doji (up shadow ≥ 2× body, tiny lo shadow) — bearish doji if up_sh_i > 2 * body_i and lo_sh_i < 0.3 * body_i + 1e-10: out["gravestone_doji"] = -1 # Hammer (bullish) if lo_sh_i > 2 * body_i and up_sh_i < body_i and self._bull(i): out["hammer"] = 1 # Hanging man (bearish version of hammer) if lo_sh_i > 2 * body_i and up_sh_i < body_i and self._bear(i): out["hanging_man"] = -1 # Shooting star (bearish) if up_sh_i > 2 * body_i and lo_sh_i < body_i and self._bear(i): out["shooting_star"] = -1 # Inverted hammer (bullish) if up_sh_i > 2 * body_i and lo_sh_i < body_i and self._bull(i): out["inverted_hammer"] = 1 # Bull marubozu if body_i > 0.85 * rng_i and self._bull(i): out["bull_marubozu"] = 1 # Bear marubozu if body_i > 0.85 * rng_i and self._bear(i): out["bear_marubozu"] = -1 # Spinning top (indecision) if body_i < 0.3 * rng_i and up_sh_i > 0.2*rng_i and lo_sh_i > 0.2*rng_i: out["spinning_top"] = 0 # ── 2-bar patterns ─────────────────────────────────────────────────── # Bullish engulfing if (self._bear(j) and self._bull(i) and self.o[i] <= self.c[j] and self.c[i] >= self.o[j]): out["bull_engulfing"] = 1 # Bearish engulfing if (self._bull(j) and self._bear(i) and self.o[i] >= self.c[j] and self.c[i] <= self.o[j]): out["bear_engulfing"] = -1 # Bullish harami if (self._bear(j) and self._bull(i) and self.c[i] < self.o[j] and self.o[i] > self.c[j]): out["bull_harami"] = 1 # Bearish harami if (self._bull(j) and self._bear(i) and self.c[i] > self.o[j] and self.o[i] < self.c[j]): out["bear_harami"] = -1 # Piercing line if (self._bear(j) and self._bull(i) and self.o[i] < self.l[j] and self.c[i] > (self.o[j] + self.c[j]) / 2): out["piercing_line"] = 1 # Dark cloud cover if (self._bull(j) and self._bear(i) and self.o[i] > self.h[j] and self.c[i] < (self.o[j] + self.c[j]) / 2): out["dark_cloud_cover"] = -1 # Tweezer bottom tol = self.c[i] * 0.001 if (abs(self.l[i] - self.l[j]) < tol and self._bear(j) and self._bull(i)): out["tweezer_bottom"] = 1 # Tweezer top if (abs(self.h[i] - self.h[j]) < tol and self._bull(j) and self._bear(i)): out["tweezer_top"] = -1 # ── 3-bar patterns ─────────────────────────────────────────────────── # Morning star if (self._bear(k) and self._body(j) < 0.35 * self._body(k) and self._bull(i) and self.c[i] > (self.o[k] + self.c[k]) / 2): out["morning_star"] = 1 # Evening star if (self._bull(k) and self._body(j) < 0.35 * self._body(k) and self._bear(i) and self.c[i] < (self.o[k] + self.c[k]) / 2): out["evening_star"] = -1 # Three white soldiers if (all(self._bull(x) for x in [k,j,i]) and self.o[j] > self.o[k] and self.o[i] > self.o[j] and self.c[j] > self.c[k] and self.c[i] > self.c[j]): out["three_white_soldiers"] = 1 # Three black crows if (all(self._bear(x) for x in [k,j,i]) and self.o[j] < self.o[k] and self.o[i] < self.o[j] and self.c[j] < self.c[k] and self.c[i] < self.c[j]): out["three_black_crows"] = -1 return out def score(self) -> float: """Aggregate all pattern scores → single float in [−1, 1].""" pts = self.detect() if not pts: return 0.0 vals = [v for v in pts.values() if v != 0] return float(np.tanh(np.mean(vals))) if vals else 0.0 # ============================================================================== # ░░ SECTION 7 — ADVANCED ANALYSIS (WAVELET · FOURIER · HURST · REGIME) ░░░░ # ============================================================================== class AdvancedAnalysis: """ Layer 7–9 of the analytical engine. ┌─────────────────────────────────────────────────┐ │ Wavelet Denoising → clean price series │ │ Wavelet Trend → low-frequency direction │ │ Fourier Cycles → dominant period + phase │ │ Hurst Exponent → trend persistence H │ │ Regime Detection → trending/ranging/volatile │ │ ADF Stationarity → for regime confirmation │ └─────────────────────────────────────────────────┘ """ def __init__(self, df: pd.DataFrame): self.c = df["Close"].to_numpy(dtype=np.float64) self.h = df["High"].to_numpy(dtype=np.float64) self.l = df["Low"].to_numpy(dtype=np.float64) self.v = df["Volume"].to_numpy(dtype=np.float64) self.n = len(df) # ── Wavelet ─────────────────────────────────────────────────────────────── def wavelet_denoise(self, wavelet: str = "db4", level: int = 3) -> np.ndarray: """ Denoise price using Donoho–Johnstone soft-thresholding. Returns cleaned close array (same length). """ if not PYWT_OK or self.n < 32: return self.c try: coeffs = pywt.wavedec(self.c, wavelet, level=level) sigma = np.median(np.abs(coeffs[-1])) / 0.6745 thr = sigma * np.sqrt(2 * np.log(max(self.n, 2))) new_c = [coeffs[0]] + [pywt.threshold(c, thr, mode="soft") for c in coeffs[1:]] return pywt.waverec(new_c, wavelet)[:self.n] except Exception: return self.c def wavelet_trend_score(self) -> float: """ Reconstruct only the low-frequency (approximation) component; measure its recent slope to derive a trend score in [−1, 1]. """ if not PYWT_OK or self.n < 32: return 0.0 try: level = min(5, pywt.dwt_max_level(self.n, "haar")) coeffs = pywt.wavedec(self.c, "haar", level=level) zeros = [np.zeros_like(c) for c in coeffs[1:]] low_f = pywt.waverec([coeffs[0]] + zeros, "haar")[:self.n] slope = np.polyfit(np.arange(min(10, self.n)), low_f[-10:], 1)[0] prange = np.ptp(low_f) or 1e-10 return float(np.tanh(slope / prange * 100)) except Exception: return 0.0 # ── Fourier ─────────────────────────────────────────────────────────────── def fourier_dominant_cycle(self) -> Tuple[float, float]: """ FFT on Hanning-windowed, linearly detrended price. Returns (dominant_period_bars, power_strength [0..1]). """ if self.n < 32: return 0.0, 0.0 try: det = sp_detrend(self.c) win = np.hanning(len(det)) fv = sp_fft(det * win) pwr = np.abs(fv[:self.n // 2]) ** 2 freq = sp_fftfreq(self.n)[:self.n // 2] idx = np.argmax(pwr[1:]) + 1 dom = 1.0 / (freq[idx] + 1e-10) str_ = pwr[idx] / (pwr.sum() + 1e-10) return float(dom), float(str_) except Exception: return 0.0, 0.0 def fourier_phase_score(self) -> float: """ Reconstruct the dominant Fourier cycle and read its current phase to generate a directional score in [−1, 1]. """ period, strength = self.fourier_dominant_cycle() if period < 4 or period > self.n * 0.5 or strength < 0.05: return 0.0 try: det = sp_detrend(self.c) fv = sp_fft(det) freq = sp_fftfreq(self.n) dom = 1.0 / period mask = np.abs(np.abs(freq) - dom) < (1.0 / self.n) cyc = np.real(np.fft.ifft(fv * mask))[:self.n] prange = np.ptp(cyc) or 1e-10 return float(np.tanh(cyc[-1] / prange * 3)) except Exception: return 0.0 # ── Hurst exponent ──────────────────────────────────────────────────────── def hurst(self, max_lag: int = 20) -> float: """ R/S Hurst exponent. H > 0.55 → persistent trend H ≈ 0.50 → random walk H < 0.45 → mean reverting """ if self.n < max_lag * 3: return 0.5 try: lags = range(2, min(max_lag, self.n // 4)) rs_arr = [] for lag in lags: s = self.c[-lag*4:] m = s.mean() z = np.cumsum(s - m) R = z.max() - z.min() S = s.std(ddof=1) or 1e-10 rs_arr.append(R / S) H, _ = np.polyfit(np.log(list(lags)), np.log(rs_arr), 1) return float(np.clip(H, 0.0, 1.0)) except Exception: return 0.5 # ── Regime detection ────────────────────────────────────────────────────── def detect_regime(self) -> Dict[str, Any]: """ Classify the current market regime: trending_up | trending_down | ranging | volatile | breakout_up | breakout_down Uses linear regression R² + slope + short/long vol ratio. """ n = min(60, self.n) y = self.c[-n:] x = np.arange(n, dtype=float) slope, _, r, *_ = stats.linregress(x, y) r2 = r ** 2 norm_slope = slope / (y.mean() + 1e-10) * 100 # %/bar ret = pd.Series(y).pct_change().dropna() vol_s = ret.tail(10).std() vol_l = ret.tail(n).std() vol_ratio = vol_s / (vol_l + 1e-10) # Regime classification if r2 > 0.65 and abs(norm_slope) > 0.05: if norm_slope > 0: regime = "trending_up"; score = min(1.0, r2) else: regime = "trending_down"; score = -min(1.0, r2) elif vol_ratio > 1.6: if norm_slope > 0: regime = "breakout_up"; score = 0.65 else: regime = "breakout_down"; score = -0.65 elif vol_ratio < 0.75: regime = "ranging"; score = 0.0 else: regime = "volatile"; score = 0.0 return { "regime": regime, "r2": float(r2), "norm_slope": float(norm_slope), "vol_ratio": float(vol_ratio), "score": float(np.tanh(score * 2)), } # ── Kalman filter (price smoother) ─────────────────────────────────────── def kalman_smooth(self) -> np.ndarray: """ 1-D Kalman filter for price. Returns smoothed close series — useful to distinguish noise from trend. """ obs_var = 1.0 proc_var = 1e-4 x_est = self.c[0] p_est = 1.0 result = np.zeros(self.n) for i, obs in enumerate(self.c): # Predict x_pred = x_est p_pred = p_est + proc_var # Update K = p_pred / (p_pred + obs_var) x_est = x_pred + K * (obs - x_pred) p_est = (1 - K) * p_pred result[i] = x_est return result # ── ADF stationarity ────────────────────────────────────────────────────── def adf_stat(self) -> float: """ Augmented Dickey-Fuller p-value. p < 0.05 → stationary (mean-reverting tendency). Returns p-value ∈ [0, 1]. """ if not STATSMODELS_OK or self.n < 30: return 0.5 try: return float(adfuller(self.c, autolag="AIC")[1]) except Exception: return 0.5 # ── Master compute ──────────────────────────────────────────────────────── def compute_all(self) -> Dict[str, Any]: """Run all advanced methods and return a summary dict.""" reg = self.detect_regime() return { "wavelet_score": self.wavelet_trend_score(), "fourier_score": self.fourier_phase_score(), "hurst": self.hurst(), "regime": reg["regime"], "regime_score": reg["score"], "regime_r2": reg["r2"], "vol_ratio": reg["vol_ratio"], "adf_pval": self.adf_stat(), } # ============================================================================== # ░░ SECTION 8 — FEATURE ENGINEERING ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class FeatureEngineer: """ Transform raw indicators into a normalised ML feature matrix. Features (≈ 80 columns) ────────────────────────────────────────────── • Normalised indicator values (rolling min-max) • Directional crossing signals (binary / ternary) • Lag features t−1, t−2, t−3 for key oscillators • Percentile ranks of volatility and volume • Cyclical time encoding (hour, day-of-week) • Price distance from key MAs ────────────────────────────────────────────── Labels: +1 = BUY | −1 = SELL | 0 = NEUTRAL (based on forward n-bar return vs threshold) """ def __init__(self, df: pd.DataFrame, ind: Dict[str, np.ndarray]): self.df = df self.ind = ind self.c = df["Close"].to_numpy(dtype=np.float64) self.n = len(df) def build(self) -> pd.DataFrame: f = pd.DataFrame(index=self.df.index) c = pd.Series(self.c, index=self.df.index) nd = self.nd # ── Price returns ───────────────────────────────────────────────────── for lag in [1, 3, 5, 10, 20]: f[f"ret_{lag}"] = c.pct_change(lag) f["log_ret"] = np.log(c / c.shift(1)) # ── Normalised oscillators ──────────────────────────────────────────── osc_keys = [ "rsi_14","rsi_7","stoch_k","stoch_d","macd_hist", "cci_20","wr","mfi_14","uo","tsi","cmf","vol_osc", "bb_pct","eom","force" ] for k in osc_keys: s = nd(k) f[k] = s for lag in [1, 2, 3]: f[f"{k}_l{lag}"] = s.shift(lag) # ── ADX / directional ───────────────────────────────────────────────── f["adx"] = nd("adx") f["di_diff"] = nd("pdi") - nd("ndi") # ── MA crossovers (ternary: +1/0/−1) ───────────────────────────────── def sign_cross(a, b): return np.sign(self._arr(a) - self._arr(b)) f["ema8_21"] = sign_cross("ema_8", "ema_21") f["ema21_55"] = sign_cross("ema_21", "ema_55") f["ema55_200"] = sign_cross("ema_55", "ema_200") f["price_ema200"] = np.sign(c - pd.Series(self._arr("ema_200"), index=self.df.index)) # ── Supertrend / SAR ────────────────────────────────────────────────── f["st_dir"] = pd.Series(self._arr("st_dir"), index=self.df.index) f["sar_dir"] = pd.Series(self._arr("sar_dir"), index=self.df.index) # ── Ichimoku ───────────────────────────────────────────────────────── tk = self._arr("ichi_tenkan"); kj = self._arr("ichi_kijun") sa = self._arr("ichi_senkou_a"); sb = self._arr("ichi_senkou_b") f["ichi_tk"] = np.sign(tk - kj) f["ichi_cloud"] = np.sign(sa - sb) f["price_cloud"] = np.sign( c.to_numpy() - (sa + sb) / 2 ) # ── Bollinger squeeze ───────────────────────────────────────────────── f["bb_squeeze"] = pct_rank(pd.Series(self._arr("bb_w"), index=self.df.index)).fillna(0.5) # ── Volume signals ──────────────────────────────────────────────────── obv = pd.Series(self._arr("obv"), index=self.df.index) f["obv_slope"] = np.sign(obv - obv.shift(5)) f["vwap_dist"] = np.sign(c - pd.Series(self._arr("vwap"), index=self.df.index)) # ── MACD histogram momentum ─────────────────────────────────────────── mh = pd.Series(self._arr("macd_hist"), index=self.df.index) f["macd_accel"] = np.sign(mh - mh.shift(1)) f["macd_sign"] = np.sign(pd.Series(self._arr("macd"), index=self.df.index)) # ── Volatility percentile ───────────────────────────────────────────── f["atr_pct"] = pct_rank(pd.Series(self._arr("atr_14"), index=self.df.index) ).fillna(0.5) f["adx_strong"] = (pd.Series(self._arr("adx"), index=self.df.index) > 25).astype(float) # ── Time features (cyclical) ────────────────────────────────────────── idx = self.df.index if hasattr(idx, "hour"): f["hr_sin"] = np.sin(2 * np.pi * idx.hour / 24) f["hr_cos"] = np.cos(2 * np.pi * idx.hour / 24) if hasattr(idx, "dayofweek"): f["dow_sin"] = np.sin(2 * np.pi * idx.dayofweek / 7) f["dow_cos"] = np.cos(2 * np.pi * idx.dayofweek / 7) return f.replace([np.inf, -np.inf], 0).fillna(0) def labels(self, fwd: int = 5, thr: float = 0.001) -> pd.Series: """ Forward return classification labels. thr = minimum % move to label as BUY/SELL; else NEUTRAL. """ c = pd.Series(self.c, index=self.df.index) fr = c.shift(-fwd) / c - 1 y = np.where(fr > thr, 1, np.where(fr < -thr, -1, 0)) return pd.Series(y, index=self.df.index) # helpers def _arr(self, key: str) -> np.ndarray: a = self.ind.get(key, np.zeros(self.n)) return np.where(np.isfinite(a), a, 0.0) def nd(self, key: str) -> pd.Series: return norm_series( pd.Series(self._arr(key), index=self.df.index) ) # ============================================================================== # ░░ SECTION 9 — ML ENSEMBLE SIGNAL GENERATOR ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class MLEngine: """ Ensemble of 5–7 models with majority-vote aggregation. Models ────────────────────────────────────────────────────────────── RandomForestClassifier — decision trees, handles non-linearity ExtraTreesClassifier — faster, more random splits XGBClassifier — gradient boosting (if xgboost installed) LGBMClassifier — fast gradient boosting (if lightgbm installed) MLPClassifier — 3-layer neural net LSTM (PyTorch or Keras) — sequence model for temporal patterns ────────────────────────────────────────────────────────────── Final signal = weighted majority vote across all models. Confidence = fraction of models agreeing on predicted class. """ def __init__(self, n_jobs: int = 2): self.n_jobs = n_jobs self.scaler = RobustScaler() self.seq_scaler = MinMaxScaler() self.label_enc = LabelEncoder() self.models: Dict[str, Any] = {} self.trained = False self.classes_ : Optional[np.ndarray] = None self.lstm_model = None self.seq_len = 20 # ── Model factory ───────────────────────────────────────────────────────── def _build_models(self) -> Dict[str, Any]: m: Dict[str, Any] = { "rf": RandomForestClassifier( n_estimators=150, max_depth=7, min_samples_leaf=5, class_weight="balanced", n_jobs=self.n_jobs, random_state=42, max_features="sqrt", ), "et": ExtraTreesClassifier( n_estimators=150, max_depth=7, min_samples_leaf=5, class_weight="balanced", n_jobs=self.n_jobs, random_state=42, max_features="sqrt", ), "mlp": MLPClassifier( hidden_layer_sizes=(128, 64, 32), activation="relu", solver="adam", alpha=0.005, max_iter=300, early_stopping=True, validation_fraction=0.1, random_state=42, ), } if XGB_OK: m["xgb"] = xgb.XGBClassifier( n_estimators=150, max_depth=5, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, tree_method="hist", n_jobs=self.n_jobs, use_label_encoder=False, eval_metric="mlogloss", random_state=42, verbosity=0, ) if LGB_OK: m["lgb"] = lgb.LGBMClassifier( n_estimators=150, max_depth=5, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, class_weight="balanced", n_jobs=self.n_jobs, random_state=42, verbose=-1, ) return m # ── Training ────────────────────────────────────────────────────────────── def fit(self, X: pd.DataFrame, y: pd.Series, min_rows: int = 120) -> bool: """ Train all models. Uses all data except the last 15 bars (reserved for live prediction). Returns True on success. """ # Exclude last 15 bars + rows with NaN labels X_tr = X.iloc[:-15] y_tr = y.iloc[:-15].dropna() X_tr = X_tr.loc[y_tr.index] if len(X_tr) < min_rows: log.warning("ML: insufficient training rows (%d)", len(X_tr)) return False # Encode labels → integers 0,1,2 y_enc = self.label_enc.fit_transform(y_tr) self.classes_ = self.label_enc.classes_ Xs = self.scaler.fit_transform(X_tr) self.models = self._build_models() for name, mdl in self.models.items(): try: mdl.fit(Xs, y_enc) except Exception as e: log.warning("ML fit error [%s]: %s", name, e) del self.models[name] # LSTM if TORCH_OK: self._fit_lstm_torch(X_tr, y_enc) elif TF_OK: self._fit_lstm_tf(X_tr, y_enc) self.trained = True return True # ── LSTM – PyTorch ─────────────────────────────────────────────────────── def _fit_lstm_torch(self, X: pd.DataFrame, y_enc: np.ndarray, epochs: int = 40) -> None: try: sl = self.seq_len Xs = self.seq_scaler.fit_transform(X.values) seqs = [Xs[i-sl:i] for i in range(sl, len(Xs))] lbls = y_enc[sl:] if len(seqs) < 60: return Xt = torch.tensor(np.array(seqs), dtype=torch.float32) yt = torch.tensor(lbls, dtype=torch.long) dl = DataLoader(TensorDataset(Xt, yt), batch_size=32, shuffle=True) n_feat = X.shape[1] n_cls = len(np.unique(y_enc)) class _LSTM(nn.Module): def __init__(self): super().__init__() self.lstm = nn.LSTM(n_feat, 64, num_layers=2, batch_first=True, dropout=0.2) self.bn = nn.BatchNorm1d(64) self.fc1 = nn.Linear(64, 32) self.drop = nn.Dropout(0.2) self.out = nn.Linear(32, n_cls) def forward(self, x): h, _ = self.lstm(x) h = self.bn(h[:, -1, :]) return self.out(self.drop(torch.relu(self.fc1(h)))) model = _LSTM() opt = torch_optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-4) crit = nn.CrossEntropyLoss() model.train() for _ in range(epochs): for xb, yb in dl: opt.zero_grad() crit(model(xb), yb).backward() opt.step() model.eval() self.lstm_model = model except Exception as e: log.debug("LSTM (torch) fit error: %s", e) # ── LSTM – TensorFlow/Keras ─────────────────────────────────────────────── def _fit_lstm_tf(self, X: pd.DataFrame, y_enc: np.ndarray, epochs: int = 40) -> None: try: sl = self.seq_len Xs = self.seq_scaler.fit_transform(X.values) seqs = [Xs[i-sl:i] for i in range(sl, len(Xs))] lbls = y_enc[sl:] if len(seqs) < 60: return n_feat = X.shape[1] n_cls = len(np.unique(y_enc)) Xt = np.array(seqs) yt = tf.keras.utils.to_categorical(lbls, n_cls) model = Sequential([ LSTM(64, return_sequences=True, input_shape=(sl, n_feat)), Dropout(0.2), LSTM(32), BatchNormalization(), Dense(32, activation="relu"), Dropout(0.2), Dense(n_cls, activation="softmax"), ]) model.compile(optimizer=KerasAdam(1e-3), loss="categorical_crossentropy") model.fit( Xt, yt, epochs=epochs, batch_size=32, validation_split=0.1, verbose=0, callbacks=[EarlyStopping(patience=5, restore_best_weights=True), ReduceLROnPlateau(patience=3, factor=0.5)], ) self.lstm_model = model except Exception as e: log.debug("LSTM (TF) fit error: %s", e) # ── Prediction ──────────────────────────────────────────────────────────── def predict(self, X: pd.DataFrame) -> Dict[str, Any]: """ Generate ensemble prediction for the latest bar. Returns ─────── signal : BUY | SELL | NEUTRAL ml_score : float ∈ [−1, 1] confidence : fraction of models agreeing [0, 1] """ if not self.trained or not self.models: return {"signal": NEUTRAL, "ml_score": 0.0, "confidence": 0.0} try: row = X.tail(1).fillna(0).replace([np.inf, -np.inf], 0) Xs = self.scaler.transform(row) votes = []; probs = [] for mdl in self.models.values(): try: votes.append(int(mdl.predict(Xs)[0])) probs.append(mdl.predict_proba(Xs)[0]) except Exception: pass # LSTM vote if self.lstm_model is not None: lp = self._lstm_predict(X) if lp is not None: votes.append(lp["pred"]) probs.append(lp["proba"]) if not votes: return {"signal": NEUTRAL, "ml_score": 0.0, "confidence": 0.0} va = np.array(votes) maj = int(np.bincount(va).argmax()) conf = float(np.mean(va == maj)) avg_p = np.mean(probs, axis=0) if probs else None cls = list(self.classes_) if self.classes_ is not None else [] predicted_label = cls[maj] if maj < len(cls) else 0 # Build directional score from probabilities ml_score = 0.0 if avg_p is not None and 1 in cls and -1 in cls: bi = cls.index(1); si = cls.index(-1) ml_score = float(np.tanh((avg_p[bi] - avg_p[si]) * 4)) elif predicted_label == 1: ml_score = conf elif predicted_label == -1: ml_score = -conf if predicted_label == 1: signal = BUY elif predicted_label == -1: signal = SELL else: signal = NEUTRAL return {"signal": signal, "ml_score": clip1(ml_score), "confidence": conf} except Exception as e: log.debug("ML predict error: %s", e) return {"signal": NEUTRAL, "ml_score": 0.0, "confidence": 0.0} def _lstm_predict(self, X: pd.DataFrame) -> Optional[Dict]: try: sl = self.seq_len if len(X) < sl: return None Xs = self.seq_scaler.transform(X.values) seq = Xs[-sl:][np.newaxis] # (1, seq_len, n_feat) if TORCH_OK and isinstance(self.lstm_model, nn.Module): self.lstm_model.eval() with torch.no_grad(): logits = self.lstm_model( torch.tensor(seq, dtype=torch.float32) ).numpy()[0] proba = np.exp(logits) / (np.exp(logits).sum() + 1e-10) else: proba = self.lstm_model.predict(seq, verbose=0)[0] return {"pred": int(np.argmax(proba)), "proba": proba} except Exception: return None # ============================================================================== # ░░ SECTION 10 — SUPPORT & RESISTANCE ENGINE ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class SREngine: """ Dynamic S/R levels based on local peaks/troughs. Generates a proximity signal: +0.8 → at support (bullish bias) −0.8 → at resistance (bearish bias) 0.0 → no meaningful proximity """ def __init__(self, df: pd.DataFrame, tol: float = 0.002): self.h = df["High"].to_numpy(dtype=np.float64) self.l = df["Low"].to_numpy(dtype=np.float64) self.c = df["Close"].to_numpy(dtype=np.float64) self.tol = tol def levels(self, lookback: int = 100) -> Dict[str, List[float]]: h = self.h[-lookback:]; l = self.l[-lookback:] res_idx, _ = find_peaks( h, distance=5, prominence=h.std() * 0.5) sup_idx, _ = find_peaks(-l, distance=5, prominence=l.std() * 0.5) return { "resistance": sorted(set(h[res_idx]), reverse=True)[:6], "support": sorted(set(l[sup_idx]))[:6], } def score(self) -> float: price = self.c[-1] lv = self.levels() near_sup = min(lv["support"], key=lambda x: abs(x - price), default=None) near_res = min(lv["resistance"], key=lambda x: abs(x - price), default=None) dist_sup = (price - near_sup) / price if near_sup else 1.0 dist_res = (near_res - price) / price if near_res else 1.0 if near_sup and 0 <= dist_sup < self.tol: return 0.80 # bouncing off support if near_res and 0 <= dist_res < self.tol: return -0.80 # hitting resistance if near_sup and near_res: return 0.20 if dist_sup < dist_res else -0.20 return 0.0 # ============================================================================== # ░░ SECTION 11 — SIGNAL AGGREGATOR (CORE SCORING ENGINE) ░░░░░░░░░░░░░░░░░░ # ============================================================================== class SignalAggregator: """ Combine all indicator-layer scores into one final signal. Each layer returns a float ∈ [−1, +1]. The final_score is a weighted sum → clipped to [−1, +1]. Thresholds ────────────────────────────────────────────── ≥ +0.50 → STRONG BUY ≥ +0.25 → MODERATE BUY ≤ −0.50 → STRONG SELL ≤ −0.25 → MODERATE SELL else → NEUTRAL ────────────────────────────────────────────── Hurst modulation: H > 0.55 → trending regime → amplify trend signals × 1.2 H < 0.45 → mean-reverting → dampen trend signals × 0.7 """ def __init__(self, df: pd.DataFrame): self.c = df["Close"].to_numpy(dtype=np.float64) self.n = len(df) # ── Layer 1 — Trend score ───────────────────────────────────────────────── def _trend(self, ind: Dict) -> float: scores = [] c = self.c[-1] # EMA stack alignment: 8 > 21 > 55 > 200 → full bullish vals = {p: safe_last(ind.get(f"ema_{p}"), c) for p in [8,21,55,200]} if vals[8] > vals[21] > vals[55] > vals[200]: scores.append( 1.0) elif vals[8] < vals[21] < vals[55] < vals[200]: scores.append(-1.0) elif vals[8] > vals[21]: scores.append( 0.5) elif vals[8] < vals[21]: scores.append(-0.5) # Price vs EMA 200 e200 = safe_last(ind.get("ema_200"), c) scores.append(np.tanh((c - e200) / (e200 * 0.01 + 1e-10))) # Supertrend & SAR direction scores.append(safe_last(ind.get("st_dir"), 0.0)) scores.append(safe_last(ind.get("sar_dir"), 0.0)) # ADX + DI crossover adx_ = safe_last(ind.get("adx"), 20) pdi = safe_last(ind.get("pdi"), 25) ndi = safe_last(ind.get("ndi"), 25) if adx_ > 20: scores.append(np.tanh((pdi - ndi) / 10)) else: scores.append(0.0) # Ichimoku cloud position sa = safe_last(ind.get("ichi_senkou_a"), c) sb = safe_last(ind.get("ichi_senkou_b"), c) top = max(sa, sb); bot = min(sa, sb) if c > top: scores.append( 0.7) elif c < bot: scores.append(-0.7) else: scores.append( 0.0) # Tenkan / Kijun cross tk = safe_last(ind.get("ichi_tenkan"), c) kj = safe_last(ind.get("ichi_kijun"), c) scores.append(np.tanh((tk - kj) / (c * 0.001 + 1e-10))) # FRAMA fr = safe_last(ind.get("frama"), c) scores.append(np.tanh((c - fr) / (c * 0.005 + 1e-10))) # Aroon oscillator ar_u = safe_last(ind.get("aroon_u"), 50) ar_d = safe_last(ind.get("aroon_d"), 50) scores.append(np.tanh((ar_u - ar_d) / 50)) return float(np.tanh(np.mean(scores))) # ── Layer 2 — Momentum score ────────────────────────────────────────────── def _momentum(self, ind: Dict) -> float: scores = [] # RSI 14 — standard zones + divergence boost r14 = safe_last(ind.get("rsi_14"), 50) if r14 > 70: scores.append(-0.8) elif r14 > 60: scores.append( 0.4) elif r14 < 30: scores.append( 0.8) elif r14 < 40: scores.append(-0.4) else: scores.append(np.tanh((r14 - 50) / 20)) # RSI 7 (faster) r7 = safe_last(ind.get("rsi_7"), 50) scores.append(np.tanh((r7 - 50) / 25)) # StochRSI sk = safe_last(ind.get("stoch_k"), 50) sd = safe_last(ind.get("stoch_d"), 50) if sk > 80: scores.append(-0.7) elif sk < 20: scores.append( 0.7) else: scores.append(np.tanh((sk - sd) / 15)) # MACD histogram + acceleration mh = self._arr(ind, "macd_hist") if len(mh) > 1: scores.append(np.tanh(mh[-1] * 50)) scores.append(0.3 * np.sign(mh[-1] - mh[-2])) ml = safe_last(ind.get("macd"), 0) scores.append(0.3 * np.sign(ml)) # CCI cci = safe_last(ind.get("cci_20"), 0) if cci > 150: scores.append(-0.7) elif cci < -150: scores.append( 0.7) else: scores.append(np.tanh(cci / 100)) # Williams %R wr = safe_last(ind.get("wr"), -50) if wr > -20: scores.append(-0.7) elif wr < -80: scores.append( 0.7) else: scores.append(np.tanh((-wr - 50) / 30)) # Ultimate oscillator uo = safe_last(ind.get("uo"), 50) scores.append(np.tanh((uo - 50) / 25)) # TSI tsi = safe_last(ind.get("tsi"), 0) scores.append(np.tanh(tsi / 25)) # ROC roc = safe_last(ind.get("roc_12"), 0) scores.append(np.tanh(roc / 1.5)) return float(np.tanh(np.mean(scores))) # ── Layer 3 — Volatility score ──────────────────────────────────────────── def _volatility(self, ind: Dict) -> float: scores = [] c = self.c[-1] # Bollinger %B bb_pct = safe_last(ind.get("bb_pct"), 0.5) if bb_pct < 0.05: scores.append( 0.8) # near/below lower band elif bb_pct > 0.95: scores.append(-0.8) # near/above upper band else: scores.append(np.tanh((0.5 - bb_pct) * 4)) # Keltner channel position kc_u = safe_last(ind.get("kc_u"), c) kc_l = safe_last(ind.get("kc_l"), c) kc_m = safe_last(ind.get("kc_m"), c) kc_r = kc_u - kc_l or 1e-10 scores.append(np.tanh((kc_m - c) / kc_r * 4)) # Donchian midline dc_m = safe_last(ind.get("dc_m"), c) scores.append(np.tanh((c - dc_m) / (dc_m * 0.005 + 1e-10))) return float(np.tanh(np.mean(scores))) # ── Layer 4 — Volume score ──────────────────────────────────────────────── def _volume(self, ind: Dict) -> float: scores = [] c = self.c[-1] # OBV slope (last 10 bars) obv = self._arr(ind, "obv") if len(obv) > 10: sl = np.polyfit(np.arange(10), obv[-10:], 1)[0] om = abs(obv[-10:].mean()) or 1e-10 scores.append(np.tanh(sl / om * 5)) # MFI mfi = safe_last(ind.get("mfi_14"), 50) if mfi > 80: scores.append(-0.7) elif mfi < 20: scores.append( 0.7) else: scores.append(np.tanh((mfi - 50) / 25)) # CMF cmf = safe_last(ind.get("cmf"), 0) scores.append(np.tanh(cmf * 3)) # Force index fi_arr = self._arr(ind, "force") if len(fi_arr) > 1: fi_std = fi_arr.std() or 1e-10 scores.append(np.tanh(fi_arr[-1] / fi_std)) # Volume oscillator vo = safe_last(ind.get("vol_osc"), 0) scores.append(np.tanh(vo / 20)) # VWAP distance vwap = safe_last(ind.get("vwap"), c) scores.append(np.tanh((c - vwap) / (vwap * 0.005 + 1e-10))) # EOM eom = safe_last(ind.get("eom"), 0) eom_arr = self._arr(ind, "eom") std_ = eom_arr.std() or 1e-10 scores.append(np.tanh(eom / std_)) return float(np.tanh(np.mean(scores))) # ── Aggregation entry ───────────────────────────────────────────────────── def aggregate( self, ind: Dict[str, np.ndarray], ml: Dict, pat_score: float, adv: Dict, sr_score: float, divergence: float, ) -> Dict[str, Any]: """ Weighted combination of all component scores. Returns the full result dict consumed by the display layer. """ trend_s = self._trend(ind) mom_s = self._momentum(ind) vol_s = self._volatility(ind) volume_s = self._volume(ind) ml_s = ml.get("ml_score", 0.0) wav_s = adv.get("wavelet_score", 0.0) reg_s = adv.get("regime_score", 0.0) # Incorporate RSI divergence into momentum if divergence != 0: mom_s = clip1(mom_s * 0.7 + divergence * 0.3) # Hurst modulation H = adv.get("hurst", 0.5) if H > 0.55: trend_s = clip1(trend_s * 1.2) mom_s = clip1(mom_s * 1.1) elif H < 0.45: trend_s = clip1(trend_s * 0.7) # Weighted sum W = WEIGHTS final = ( W["trend"] * trend_s + W["momentum"] * mom_s + W["volatility"] * vol_s + W["volume"] * volume_s + W["ml"] * ml_s + W["pattern"] * pat_score + W["regime"] * reg_s + W["wavelet"] * wav_s ) + 0.04 * sr_score final = clip1(final) # Signal label if final >= STRONG_BUY: signal, strength = BUY, "STRONG" elif final >= MODERATE_BUY: signal, strength = BUY, "MODERATE" elif final <= STRONG_SELL: signal, strength = SELL, "STRONG" elif final <= MODERATE_SELL: signal, strength = SELL, "MODERATE" else: signal, strength = NEUTRAL, "WEAK" # Confidence: agreement ratio + ML confidence comp_scores = [trend_s, mom_s, volume_s, ml_s, pat_score] if final > 0: agree = np.mean([s > 0 for s in comp_scores]) elif final < 0: agree = np.mean([s < 0 for s in comp_scores]) else: agree = 0.5 conf = min(0.97, abs(final) * 0.5 + ml.get("confidence", 0.5) * 0.3 + agree * 0.2) breakdown = { "Trend": trend_s, "Momentum": mom_s, "Volatility": vol_s, "Volume": volume_s, "ML Signal": ml_s, "Pattern": pat_score, "Regime": reg_s, "Wavelet": wav_s, "SR Adjust": sr_score, "Divergence": divergence, } return { "signal": signal, "strength": strength, "score": float(final), "confidence": float(conf), "breakdown": breakdown, "hurst": float(H), "regime": adv.get("regime", "unknown"), "ml_signal": ml.get("signal", NEUTRAL), "ml_conf": float(ml.get("confidence", 0.0)), } # ── helpers ─────────────────────────────────────────────────────────────── @staticmethod def _arr(ind: Dict, key: str) -> np.ndarray: a = ind.get(key, np.array([0.0])) a = np.asarray(a, dtype=np.float64) return a[np.isfinite(a)] # ============================================================================== # ░░ SECTION 12 — MAIN INDICATOR ORCHESTRATOR ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class AdvancedIndicator: """ Top-level class that orchestrates every layer. Public API ────────── .analyze(symbol, tf) → Dict (full result) .scan_market(market, tf) → List[Dict] (sorted by |score|) .analyze_multiple(syms, tf) → List[Dict] """ def __init__(self, retrain_every: int = 50): self.fetcher = DataFetcher() self.retrain_every = retrain_every self._ml: Dict[str, MLEngine] = {} self._calls:Dict[str, int] = defaultdict(int) # ── Main pipeline ───────────────────────────────────────────────────────── def analyze( self, symbol: str, tf: str = "15m", lookback: int = 500, retrain: bool = False, ) -> Optional[Dict]: """ Full 10-layer analysis for one symbol / timeframe. Pipeline ──────── 1. Fetch OHLCV 2. Technical indicators (35+) 3. Advanced analysis (wavelet, Fourier, Hurst, regime) 4. Candlestick patterns 5. Support & resistance 6. Feature engineering 7. ML ensemble (train if needed) 8. Signal aggregation 9. Build output dict """ t0 = time.time() # 1. Data df = self.fetcher.fetch(symbol, tf, lookback) if df is None: return None # 2. Indicators ind_eng = Indicators(df) ind = ind_eng.compute_all() # 3. Advanced adv_eng = AdvancedAnalysis(df) adv = adv_eng.compute_all() # 4. Patterns cp = CandlePatterns(df) pat_score = cp.score() pat_detail = cp.detect() # 5. S/R sr = SREngine(df) sr_score = sr.score() sr_levels = sr.levels() # RSI divergence divergence = ind_eng.rsi_divergence() # 6. Features fe = FeatureEngineer(df, ind) X = fe.build() y = fe.labels() # 7. ML key = f"{symbol}|{tf}" cnt = self._calls[key] ml_eng = self._ml.get(key) if ml_eng is None or retrain or cnt % self.retrain_every == 0: ml_eng = MLEngine(n_jobs=2) if ml_eng.fit(X, y): self._ml[key] = ml_eng self._calls[key] += 1 ml_result = ml_eng.predict(X) if (ml_eng and ml_eng.trained) else \ {"signal": NEUTRAL, "ml_score": 0.0, "confidence": 0.0} # 8. Aggregate agg = SignalAggregator(df) result = agg.aggregate(ind, ml_result, pat_score, adv, sr_score, divergence) # 9. Build output pv = ind_eng.pivot_points() fib = ind_eng.fibonacci_levels() disp = self._display_values(ind, df) out = { "symbol": symbol, "tf": tf, "signal": result["signal"], "strength": result["strength"], "score": result["score"], "confidence": result["confidence"], "price": float(df["Close"].iloc[-1]), "timestamp": ist_now(), "elapsed_ms": int((time.time() - t0) * 1000), "breakdown": result["breakdown"], "hurst": result["hurst"], "regime": result["regime"], "ml": ml_result, "patterns": pat_detail, "sr": sr_levels, "indicators": disp, "pivot": pv, "fib": fib, "adv": adv, } free_memory() return out # ── Multi-symbol ───────────────────────────────────────────────────────── def analyze_multiple( self, symbols: List[str], tf: str = "15m", workers: int = 2, ) -> List[Dict]: """Analyse multiple symbols in parallel; return sorted by |score|.""" results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex: futs = {ex.submit(self.analyze, s, tf): s for s in symbols} for f in concurrent.futures.as_completed(futs): try: r = f.result() if r: results.append(r) except Exception as e: log.debug("Multi-analyze error: %s", e) results.sort(key=lambda x: abs(x.get("score", 0)), reverse=True) return results def scan_market( self, market: str = "forex", tf: str = "15m", top_n: int = 5, ) -> List[Dict]: """Scan entire market universe; return top N non-neutral signals.""" syms = FOREX_PAIRS if market.lower() == "forex" else CRYPTO_PAIRS all_ = self.analyze_multiple(syms, tf, workers=2) non_neutral = [r for r in all_ if r["signal"] != NEUTRAL] return non_neutral[:top_n] # ── Helper: display values ──────────────────────────────────────────────── @staticmethod def _display_values(ind: Dict, df: pd.DataFrame) -> Dict[str, Any]: """Extract latest scalar values from indicator arrays for display.""" SL = safe_last return { "RSI(14)": SL(ind.get("rsi_14")), "RSI(7)": SL(ind.get("rsi_7")), "Stoch K": SL(ind.get("stoch_k")), "Stoch D": SL(ind.get("stoch_d")), "MACD Line": SL(ind.get("macd")), "MACD Signal": SL(ind.get("macd_sig")), "MACD Hist": SL(ind.get("macd_hist")), "CCI(20)": SL(ind.get("cci_20")), "Williams %R": SL(ind.get("wr")), "MFI(14)": SL(ind.get("mfi_14")), "ADX": SL(ind.get("adx")), "+DI": SL(ind.get("pdi")), "-DI": SL(ind.get("ndi")), "ATR(14)": SL(ind.get("atr_14")), "BB Upper": SL(ind.get("bb_u")), "BB Middle": SL(ind.get("bb_m")), "BB Lower": SL(ind.get("bb_l")), "BB %B": SL(ind.get("bb_pct")), "SuperTr Dir": SL(ind.get("st_dir")), "SAR Dir": SL(ind.get("sar_dir")), "EMA(8)": SL(ind.get("ema_8")), "EMA(21)": SL(ind.get("ema_21")), "EMA(55)": SL(ind.get("ema_55")), "EMA(200)": SL(ind.get("ema_200")), "HMA(21)": SL(ind.get("hma_21")), "FRAMA": SL(ind.get("frama")), "VWAP": SL(ind.get("vwap")), "OBV": SL(ind.get("obv")), "CMF": SL(ind.get("cmf")), "UO": SL(ind.get("uo")), "TSI": SL(ind.get("tsi")), "Price": float(df["Close"].iloc[-1]), } # ============================================================================== # ░░ SECTION 13 — DISPLAY ENGINE ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== class Display: """ Rich terminal display with plain-text fallback. Every public method is safe to call regardless of whether Rich is installed. """ # ── Banner ──────────────────────────────────────────────────────────────── @staticmethod def banner() -> None: lines = [ "╔══════════════════════════════════════════════════════════════════════╗", "║ ADVANCED FOREX & CRYPTO TRADING INDICATOR v3.0 ║", "║ Platform : Hugging Face Free Tier (16 GB RAM · 2 CPU) ║", "║ Timezone : UTC+5:30 (Asia/Kolkata — IST) ║", "║ Signals : BUY 🟢 | SELL 🔴 | NEUTRAL ⚪ ║", "║ Analysis : 35+ TA · Wavelet · Fourier · Hurst · ML Ensemble ║", "║ Markets : All Forex + All Major Crypto ║", "╚══════════════════════════════════════════════════════════════════════╝", ] if RICH_OK: console.print("\n".join(lines), style="bold bright_blue") else: print(BOLD + CYAN + "\n".join(lines) + RESET) # ── Library status ──────────────────────────────────────────────────────── @staticmethod def lib_status() -> None: libs = [ ("numpy/pandas/scipy", True), ("scikit-learn", True), ("xgboost", XGB_OK), ("lightgbm", LGB_OK), ("pytorch (CPU)", TORCH_OK), ("tensorflow-cpu", TF_OK), ("PyWavelets", PYWT_OK), ("statsmodels", STATSMODELS_OK), ("yfinance", YF_OK), ("ccxt", CCXT_OK), ("numba", NUMBA_OK), ("plotly", PLOTLY_OK), ("rich", RICH_OK), ("optuna", OPTUNA_OK), ("shap", SHAP_OK), ("pyarrow", ARROW_OK), ] if RICH_OK: t = Table(title="Library Status", box=box.SIMPLE_HEAD) t.add_column("Package", style="cyan", width=22) t.add_column("Status", width=10) for name, ok in libs: t.add_row(name, "[bright_green]✓ OK[/]" if ok else "[dim red]✗ Missing[/]") console.print(t) else: print(f"\n{CYAN}LIBRARY STATUS:{RESET}") for name, ok in libs: mark = GREEN + "✓" + RESET if ok else RED + "✗" + RESET print(f" {mark} {name}") # ── Single result ───────────────────────────────────────────────────────── @staticmethod def result(r: Dict, verbose: bool = True) -> None: if RICH_OK: Display._result_rich(r, verbose) else: Display._result_plain(r, verbose) @staticmethod def _result_rich(r: Dict, verbose: bool) -> None: sig = r["signal"] col = ("bright_green" if sig == BUY else "bright_red" if sig == SELL else "yellow") emo = ("🟢" if sig == BUY else "🔴" if sig == SELL else "⚪") # Header hdr = Text() hdr.append(f"\n {emo} ", style="bold") hdr.append(f"{sig} — {r['strength']}", style=f"bold {col}") hdr.append(f" │ Score: {r['score']:+.4f} │ Conf: {r['confidence']:.1%}", style="cyan") hdr.append(f"\n 📊 {r['symbol']} ⏱ {r['tf'].upper()} 💰 {r['price']:.6f}", style="white") hdr.append(f"\n 🕒 {r['timestamp']} ⚡ {r['elapsed_ms']} ms " f"📈 Regime: {r['regime']} 🌊 Hurst: {r['hurst']:.3f}", style="dim") console.print(Panel(hdr, border_style=col, padding=(0,1))) if not verbose: return # ── Signal breakdown ────────────────────────────────────────────────── t = Table(title="Signal Breakdown", box=box.ROUNDED, border_style="blue") t.add_column("Layer", style="cyan", width=14) t.add_column("Score", justify="right",width=9) t.add_column("Direction", width=13) t.add_column("Bar", width=18) for layer, sc in r["breakdown"].items(): c2 = "bright_green" if sc > 0 else "bright_red" if sc < 0 else "yellow" dir_str = ("▲ Bullish" if sc > 0.15 else "▼ Bearish" if sc < -0.15 else "◆ Neutral") bar = Display._bar(sc) t.add_row(layer, f"[{c2}]{sc:+.3f}[/]", f"[{c2}]{dir_str}[/]", bar) console.print(t) # ── Indicator values (2-column) ─────────────────────────────────────── iv = t2 = Table(title="Key Indicator Values", box=box.SIMPLE, border_style="dim") t2.add_column("Indicator", style="dim cyan", width=14) t2.add_column("Value", justify="right", width=13) t2.add_column("Indicator", style="dim cyan", width=14) t2.add_column("Value", justify="right", width=13) items = list(r["indicators"].items()) for i in range(0, len(items), 2): k1, v1 = items[i] k2, v2 = items[i+1] if i+1 < len(items) else ("","") s1 = f"{v1:.5f}" if isinstance(v1, float) else str(v1) s2 = f"{v2:.5f}" if isinstance(v2, float) else str(v2) t2.add_row(k1, s1, str(k2), s2) console.print(t2) # ── Patterns ───────────────────────────────────────────────────────── if r["patterns"]: parts = [] for p, v in r["patterns"].items(): c3 = "bright_green" if v > 0 else "bright_red" if v < 0 else "yellow" parts.append(f"[{c3}]{p}[/]") console.print(Panel(Text.from_markup(" ".join(parts)), title="🕯 Candlestick Patterns", border_style="dim")) # ── Pivot & Fibonacci side-by-side ──────────────────────────────────── pv = r.get("pivot", {}); fb = r.get("fib", {}) price = r["price"] t3 = Table(title="Pivot Points & Fibonacci", box=box.MINIMAL, border_style="dim") t3.add_column("Pivot", style="cyan", width=8) t3.add_column("Level", justify="right", width=12) t3.add_column("Fib", style="magenta", width=8) t3.add_column("Level", justify="right", width=12) pi = list(pv.items()); fi = list(fb.items()) for i in range(max(len(pi), len(fi))): pk, pval = pi[i] if i < len(pi) else ("","") fk, fval = fi[i] if i < len(fi) else ("","") ps = (f"[bright_white]{pval:.6f}[/]" if isinstance(pval, float) and abs(pval - price) < price*0.002 else f"{pval:.6f}" if isinstance(pval,float) else str(pval)) fs = f"{fval:.6f}" if isinstance(fval, float) else str(fval) t3.add_row(pk, ps, fk, fs) console.print(t3) # ── ML summary ──────────────────────────────────────────────────────── ml = r.get("ml", {}) msc = ml.get("ml_score", 0.0) mc = ml.get("confidence", 0.0) mc2 = "bright_green" if ml.get("signal") == BUY else \ "bright_red" if ml.get("signal") == SELL else "yellow" console.print(Panel( Text.from_markup( f"ML Signal: [{mc2}]{ml.get('signal','N/A')}[/] " f"ML Score: [cyan]{msc:+.4f}[/] " f"ML Confidence: [cyan]{mc:.1%}[/]" ), title="🤖 ML Ensemble Prediction", border_style="magenta", )) console.print() @staticmethod def _result_plain(r: Dict, verbose: bool) -> None: sig = r["signal"] sc = r["score"] cv = {BUY: GREEN, SELL: RED, NEUTRAL: YELLOW}.get(sig, "") sep = "=" * 72 print(f"\n{sep}") print(f"{BOLD}{CYAN}ADVANCED FOREX & CRYPTO INDICATOR{RESET}") print(sep) print(f"Symbol : {r['symbol']} | TF: {r['tf'].upper()}") print(f"Price : {r['price']:.6f}") print(f"Time : {r['timestamp']}") print(f"Regime : {r['regime']} | Hurst: {r['hurst']:.3f}") print(sep) print(f"SIGNAL : {cv}{BOLD}{sig} — {r['strength']}{RESET}") print(f"SCORE : {sc:+.4f} | CONFIDENCE: {r['confidence']:.1%}") print(sep) if verbose: print(f"\n{CYAN}SIGNAL BREAKDOWN:{RESET}") for layer, ls in r["breakdown"].items(): bar = Display._bar(ls, 16) arrow = "▲" if ls > 0 else "▼" if ls < 0 else "—" c2 = GREEN if ls > 0 else RED if ls < 0 else YELLOW print(f" {layer:<14} {c2}{arrow}{RESET} {bar} {ls:+.3f}") print(f"\n{CYAN}KEY INDICATORS:{RESET}") items = list(r["indicators"].items()) for i in range(0, len(items), 2): k1, v1 = items[i] k2, v2 = items[i+1] if i+1 < len(items) else ("","") s1 = f"{v1:.5f}" if isinstance(v1, float) else str(v1) s2 = f"{v2:.5f}" if isinstance(v2, float) else str(v2) print(f" {k1:<16} {s1:<13} {k2:<16} {s2}") if r["patterns"]: print(f"\n{CYAN}PATTERNS:{RESET}", " | ".join(r["patterns"].keys())) ml = r.get("ml", {}) msc = ml.get("signal", NEUTRAL) mc = {BUY: GREEN, SELL: RED, NEUTRAL: YELLOW}.get(msc, "") print(f"\n{CYAN}ML SIGNAL:{RESET} {mc}{msc}{RESET} " f"Score: {ml.get('ml_score',0):+.4f} " f"Conf: {ml.get('confidence',0):.1%}") print(sep) # ── Scan summary table ──────────────────────────────────────────────────── @staticmethod def scan_table(results: List[Dict]) -> None: if not results: print("No signals found in scan.") return if RICH_OK: t = Table(title="🔍 Market Scan Results", box=box.DOUBLE_EDGE, border_style="bright_blue") t.add_column("Symbol", style="bright_white", width=12) t.add_column("TF", width=5) t.add_column("Signal", width=10) t.add_column("Strength", width=10) t.add_column("Score", justify="right", width=8) t.add_column("Conf", justify="right", width=6) t.add_column("Price", justify="right", width=13) t.add_column("Regime", width=14) t.add_column("Hurst", justify="right", width=6) for r in results: sig = r["signal"] col = ("bright_green" if sig == BUY else "bright_red" if sig == SELL else "yellow") t.add_row( r["symbol"].replace("=X",""), r["tf"], f"[{col}]{sig}[/]", r["strength"], f"[{col}]{r['score']:+.3f}[/]", f"{r['confidence']:.0%}", f"{r['price']:.6f}", r["regime"], f"{r['hurst']:.2f}", ) console.print(t) else: hdr = (f"{'SYMBOL':<12} {'TF':<5} {'SIGNAL':<7} " f"{'SCORE':>7} {'CONF':>5} {'PRICE':>13} {'REGIME':<14} HURST") print(f"\n{CYAN}{hdr}{RESET}") print("─" * 75) for r in results: sig = r["signal"] cc = GREEN if sig == BUY else RED if sig == SELL else YELLOW print(f"{cc}{r['symbol'].replace('=X',''):<12}{RESET} " f"{r['tf']:<5} {cc}{sig:<7}{RESET} " f"{r['score']:>7.3f} {r['confidence']:>5.0%} " f"{r['price']:>13.6f} {r['regime']:<14} " f"{r['hurst']:.2f}") # ── ASCII score bar ─────────────────────────────────────────────────────── @staticmethod def _bar(score: float, width: int = 14) -> str: half = width // 2 filled = int(abs(score) * half) filled = min(filled, half) if score >= 0: return "░" * half + "█" * filled + "·" * (half - filled) else: return "·" * (half - filled) + "█" * filled + "░" * half # ============================================================================== # ░░ SECTION 14 — INTERACTIVE CLI ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== def interactive(indicator: AdvancedIndicator) -> None: """ Simple REPL for manual exploration. Commands ──────────────────────────────────────────────────── analyze [TF] analyse one pair scan [TF] [N] scan market watch [TF] [SEC] live refresh every N seconds list show all available symbols & TFs status show library status quit / exit leave ──────────────────────────────────────────────────── """ disp = Display() disp.banner() disp.lib_status() help_text = ( "\nCommands:\n" f" {CYAN}analyze{RESET} [TF] " "— e.g. analyze EURUSD=X 15m\n" f" {CYAN}scan{RESET} [TF] [N] " "— e.g. scan forex 15m 5\n" f" {CYAN}watch{RESET} [TF] [SEC] " "— e.g. watch BTC-USD 1m 60\n" f" {CYAN}list{RESET} " "— available symbols & timeframes\n" f" {CYAN}status{RESET} " "— library status\n" f" {CYAN}quit{RESET} — exit\n" ) print(help_text) while True: try: raw = input(f"{CYAN}INDICATOR>{RESET} ").strip() if not raw: continue parts = raw.split() cmd = parts[0].lower() if cmd in ("quit", "exit", "q"): print("Goodbye! 👋") break elif cmd == "analyze" and len(parts) >= 2: sym = parts[1].upper() tf_ = parts[2] if len(parts) > 2 else "15m" print(f" Analysing {sym} [{tf_}] …") r = indicator.analyze(sym, tf_) if r: disp.result(r) else: print(f" [!] Could not fetch data for {sym}") elif cmd == "scan": market = parts[1].lower() if len(parts) > 1 else "forex" tf_ = parts[2] if len(parts) > 2 else "15m" top = int(parts[3]) if len(parts) > 3 else 5 print(f" Scanning {market} market [{tf_}] …") results = indicator.scan_market(market, tf_, top) disp.scan_table(results) if results: print(f"\n Detailed view of top signal:") disp.result(results[0]) elif cmd == "watch" and len(parts) >= 2: sym = parts[1].upper() tf_ = parts[2] if len(parts) > 2 else "15m" interval = int(parts[3]) if len(parts) > 3 else 60 print(f" Watching {sym} [{tf_}] every {interval}s (Ctrl+C to stop)") try: while True: r = indicator.analyze(sym, tf_) if r: disp.result(r, verbose=False) time.sleep(interval) except KeyboardInterrupt: print("\n Watch stopped.") elif cmd == "list": print(f"\n{CYAN}FOREX PAIRS ({len(FOREX_PAIRS)}):{RESET}") print(" " + " ".join(p.replace("=X","") for p in FOREX_PAIRS)) print(f"\n{CYAN}CRYPTO PAIRS ({len(CRYPTO_PAIRS)}):{RESET}") print(" " + " ".join(CRYPTO_PAIRS)) print(f"\n{CYAN}TIMEFRAMES:{RESET}") print(" " + " ".join(TF_MINUTES.keys())) elif cmd == "status": disp.lib_status() else: print(f" Unknown command '{cmd}'. {help_text}") except KeyboardInterrupt: print("\n Interrupted (type 'quit' to exit)") except Exception as e: print(f" Error: {e}") if "--debug" in sys.argv: traceback.print_exc() # ============================================================================== # ░░ SECTION 15 — DEMO RUN ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== def demo() -> None: """ Quick demonstration: 3 individual analyses + Forex market scan. Designed to run successfully on the Hugging Face Free Tier. """ disp = Display() disp.banner() disp.lib_status() ind = AdvancedIndicator() demo_pairs = [ ("EURUSD=X", "15m"), ("BTC-USD", "15m"), ("GBPUSD=X", "5m"), ] print(f"\n{CYAN}{'═'*60}{RESET}") print(f"{CYAN} DEMO — Individual Analysis{RESET}") print(f"{CYAN}{'═'*60}{RESET}\n") for sym, tf_ in demo_pairs: print(f" Analysing {sym} [{tf_}] …") r = ind.analyze(sym, tf_) if r: disp.result(r, verbose=True) else: print(f" [!] No data available for {sym}\n") print(f"\n{CYAN}{'═'*60}{RESET}") print(f"{CYAN} DEMO — Forex Market Scan (15m, Top 5){RESET}") print(f"{CYAN}{'═'*60}{RESET}\n") scan_res = ind.scan_market("forex", "15m", top_n=5) disp.scan_table(scan_res) print(f"\n{CYAN}{'═'*60}{RESET}") print(f"{CYAN} DEMO — Crypto Market Scan (15m, Top 5){RESET}") print(f"{CYAN}{'═'*60}{RESET}\n") scan_c = ind.scan_market("crypto", "15m", top_n=5) disp.scan_table(scan_c) # ============================================================================== # ░░ SECTION 16 — ENTRY POINT ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ # ============================================================================== def _parse_args(): import argparse p = argparse.ArgumentParser( prog="advanced_indicator", description="Advanced Forex & Crypto Trading Indicator — v3.0", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) p.add_argument("--symbol", "-s", type=str, help="Symbol e.g. EURUSD=X BTC-USD") p.add_argument("--tf", "-t", type=str, default="15m", help="Timeframe (30s 1m 5m 15m 1h …)") p.add_argument("--scan", "-m", type=str, help="Market scan: forex | crypto") p.add_argument("--top", "-n", type=int, default=5, help="Top N results from scan") p.add_argument("--watch", "-w", type=str, help="Watch symbol — live refresh") p.add_argument("--interval", "-i", type=int, default=60,help="Watch refresh interval (seconds)") p.add_argument("--lookback", "-l", type=int, default=500,help="Bars to load (300 for low RAM)") p.add_argument("--interactive", action="store_true", help="Interactive CLI mode") p.add_argument("--demo", action="store_true", help="Run demonstration") p.add_argument("--list", action="store_true", help="List all symbols & TFs") p.add_argument("--debug", action="store_true", help="Verbose logging") return p.parse_args() if __name__ == "__main__": args = _parse_args() if args.debug: logging.getLogger().setLevel(logging.DEBUG) log.setLevel(logging.DEBUG) disp = Display() ind = AdvancedIndicator() # ── list ────────────────────────────────────────────────────────────────── if args.list: disp.banner() print(f"\n{CYAN}FOREX PAIRS:{RESET}") print(" " + " ".join(p.replace("=X","") for p in FOREX_PAIRS)) print(f"\n{CYAN}CRYPTO PAIRS:{RESET}") print(" " + " ".join(CRYPTO_PAIRS)) print(f"\n{CYAN}TIMEFRAMES:{RESET}") print(" " + " ".join(TF_MINUTES.keys())) sys.exit(0) # ── interactive ─────────────────────────────────────────────────────────── elif args.interactive: interactive(ind) # ── demo ────────────────────────────────────────────────────────────────── elif args.demo: demo() # ── single symbol ───────────────────────────────────────────────────────── elif args.symbol: disp.banner() sym = args.symbol.upper() print(f"\n Analysing {sym} [{args.tf}] …") r = ind.analyze(sym, args.tf, args.lookback) if r: disp.result(r, verbose=True) else: print(f" [!] Failed to fetch data for {sym}. " f"Check symbol format and internet connection.") # ── market scan ─────────────────────────────────────────────────────────── elif args.scan: disp.banner() market = args.scan.lower() print(f"\n Scanning {market} market [{args.tf}] …") results = ind.scan_market(market, args.tf, args.top) disp.scan_table(results) if results: print(f"\n Detailed view of top signal:") disp.result(results[0]) # ── watch mode ──────────────────────────────────────────────────────────── elif args.watch: disp.banner() sym = args.watch.upper() print(f"\n Watching {sym} [{args.tf}] every {args.interval}s " f"(Ctrl+C to stop)") try: while True: r = ind.analyze(sym, args.tf, args.lookback) if r: disp.result(r, verbose=False) time.sleep(args.interval) except KeyboardInterrupt: print("\n Watch stopped.") # ── default: demo ───────────────────────────────────────────────────────── else: demo()