Spaces:
Running on Zero
Running on Zero
| """Technical indicators computed from a candle DataFrame.""" | |
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| import numpy as np | |
| import pandas as pd | |
| # --------------------------------------------------------------------------- | |
| # Core indicator helpers | |
| # --------------------------------------------------------------------------- | |
| def _ema(series: pd.Series, period: int) -> pd.Series: | |
| return series.ewm(span=period, adjust=False).mean() | |
| def _rsi(close: pd.Series, period: int = 14) -> pd.Series: | |
| delta = close.diff() | |
| gain = delta.clip(lower=0) | |
| loss = -delta.clip(upper=0) | |
| avg_gain = gain.ewm(com=period - 1, adjust=False).mean() | |
| avg_loss = loss.ewm(com=period - 1, adjust=False).mean() | |
| rs = avg_gain / avg_loss.replace(0, np.nan) | |
| return 100 - (100 / (1 + rs)) | |
| def _atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series: | |
| prev_close = close.shift(1) | |
| tr = pd.concat( | |
| [high - low, (high - prev_close).abs(), (low - prev_close).abs()], axis=1 | |
| ).max(axis=1) | |
| return tr.ewm(com=period - 1, adjust=False).mean() | |
| def _macd(close: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9): | |
| ema_fast = _ema(close, fast) | |
| ema_slow = _ema(close, slow) | |
| macd_line = ema_fast - ema_slow | |
| signal_line = _ema(macd_line, signal) | |
| histogram = macd_line - signal_line | |
| return macd_line, signal_line, histogram | |
| def _bollinger(close: pd.Series, period: int = 20, std_dev: float = 2.0): | |
| mid = close.rolling(period).mean() | |
| std = close.rolling(period).std(ddof=0) | |
| upper = mid + std_dev * std | |
| lower = mid - std_dev * std | |
| return upper, mid, lower | |
| # --------------------------------------------------------------------------- | |
| # Public surface | |
| # --------------------------------------------------------------------------- | |
| class IndicatorSnapshot: | |
| """Latest values for all indicators, ready for the LLM prompt.""" | |
| price: float | |
| rsi_14: float | |
| ema_20: float | |
| ema_50: float | |
| ema_200: float | |
| macd_line: float | |
| macd_signal: float | |
| macd_hist: float | |
| bb_upper: float | |
| bb_mid: float | |
| bb_lower: float | |
| atr_14: float | |
| # Derived context | |
| ema_trend: str # "bullish" | "bearish" | "mixed" | |
| bb_position: str # "above_upper" | "near_upper" | "mid" | "near_lower" | "below_lower" | |
| rsi_zone: str # "overbought" | "neutral" | "oversold" | |
| volume_vs_avg: float # current candle volume / 20-bar avg volume | |
| # Recent price structure (last N candles) | |
| recent_highs: list[float] = field(default_factory=list) | |
| recent_lows: list[float] = field(default_factory=list) | |
| # ATR-based TP/SL suggestions (2:1 R/R) | |
| suggested_long_tp: float = 0.0 | |
| suggested_long_sl: float = 0.0 | |
| suggested_short_tp: float = 0.0 | |
| suggested_short_sl: float = 0.0 | |
| def to_prompt_block(self) -> str: | |
| lines = [ | |
| f"Current price : {self.price:.4f}", | |
| "", | |
| "--- Trend ---", | |
| f"EMA 20 : {self.ema_20:.4f}", | |
| f"EMA 50 : {self.ema_50:.4f}", | |
| f"EMA 200 : {self.ema_200:.4f}", | |
| f"EMA trend : {self.ema_trend}", | |
| "", | |
| "--- Momentum ---", | |
| f"RSI(14) : {self.rsi_14:.2f} [{self.rsi_zone}]", | |
| f"MACD line : {self.macd_line:.4f}", | |
| f"MACD signal : {self.macd_signal:.4f}", | |
| f"MACD hist : {self.macd_hist:.4f} ({'positive' if self.macd_hist > 0 else 'negative'})", | |
| "", | |
| "--- Volatility ---", | |
| f"BB upper : {self.bb_upper:.4f}", | |
| f"BB mid : {self.bb_mid:.4f}", | |
| f"BB lower : {self.bb_lower:.4f}", | |
| f"Price vs BB : {self.bb_position}", | |
| f"ATR(14) : {self.atr_14:.4f}", | |
| "", | |
| "--- Volume ---", | |
| f"Vol / 20-avg : {self.volume_vs_avg:.2f}x", | |
| "", | |
| "--- Recent swing highs (last 5) ---", | |
| ", ".join(f"{h:.4f}" for h in self.recent_highs[-5:]), | |
| "--- Recent swing lows (last 5) ---", | |
| ", ".join(f"{l:.4f}" for l in self.recent_lows[-5:]), | |
| "", | |
| "--- ATR-based TP/SL (2:1 R/R) ---", | |
| f"LONG TP : {self.suggested_long_tp:.4f} | SL : {self.suggested_long_sl:.4f}", | |
| f"SHORT TP : {self.suggested_short_tp:.4f} | SL : {self.suggested_short_sl:.4f}", | |
| ] | |
| return "\n".join(lines) | |
| def compute_indicators(df: pd.DataFrame, atr_multiplier_tp: float = 2.0, atr_multiplier_sl: float = 1.0) -> IndicatorSnapshot: | |
| """Compute all indicators from a candle DataFrame and return the latest snapshot.""" | |
| close = df["close"] | |
| high = df["high"] | |
| low = df["low"] | |
| volume = df["volume"] | |
| rsi = _rsi(close) | |
| ema20 = _ema(close, 20) | |
| ema50 = _ema(close, 50) | |
| ema200 = _ema(close, 200) | |
| macd_line, macd_sig, macd_hist = _macd(close) | |
| bb_upper, bb_mid, bb_lower = _bollinger(close) | |
| atr = _atr(high, low, close) | |
| vol_avg = volume.rolling(20).mean() | |
| price = float(close.iloc[-1]) | |
| atr_val = float(atr.iloc[-1]) | |
| # EMA trend | |
| e20, e50, e200 = float(ema20.iloc[-1]), float(ema50.iloc[-1]), float(ema200.iloc[-1]) | |
| if price > e20 > e50 > e200: | |
| ema_trend = "strongly_bullish" | |
| elif price > e20 and e20 > e50: | |
| ema_trend = "bullish" | |
| elif price < e20 < e50 < e200: | |
| ema_trend = "strongly_bearish" | |
| elif price < e20 and e20 < e50: | |
| ema_trend = "bearish" | |
| else: | |
| ema_trend = "mixed" | |
| # BB position | |
| bbu, bbm, bbl = float(bb_upper.iloc[-1]), float(bb_mid.iloc[-1]), float(bb_lower.iloc[-1]) | |
| band_width = bbu - bbl | |
| if price > bbu: | |
| bb_pos = "above_upper" | |
| elif price > bbu - 0.1 * band_width: | |
| bb_pos = "near_upper" | |
| elif price < bbl: | |
| bb_pos = "below_lower" | |
| elif price < bbl + 0.1 * band_width: | |
| bb_pos = "near_lower" | |
| else: | |
| bb_pos = "mid" | |
| # RSI zone | |
| rsi_val = float(rsi.iloc[-1]) | |
| if rsi_val >= 70: | |
| rsi_zone = "overbought" | |
| elif rsi_val <= 30: | |
| rsi_zone = "oversold" | |
| else: | |
| rsi_zone = "neutral" | |
| # Volume ratio | |
| vol_ratio = float(volume.iloc[-1] / vol_avg.iloc[-1]) if float(vol_avg.iloc[-1]) > 0 else 1.0 | |
| # Swing highs / lows (simple local extrema over last 20 candles) | |
| window = df.tail(20) | |
| recent_highs = list(window["high"].nlargest(5).sort_index().values) | |
| recent_lows = list(window["low"].nsmallest(5).sort_index().values) | |
| # TP / SL | |
| long_tp = price + atr_multiplier_tp * atr_val | |
| long_sl = price - atr_multiplier_sl * atr_val | |
| short_tp = price - atr_multiplier_tp * atr_val | |
| short_sl = price + atr_multiplier_sl * atr_val | |
| return IndicatorSnapshot( | |
| price=price, | |
| rsi_14=rsi_val, | |
| ema_20=e20, | |
| ema_50=e50, | |
| ema_200=e200, | |
| macd_line=float(macd_line.iloc[-1]), | |
| macd_signal=float(macd_sig.iloc[-1]), | |
| macd_hist=float(macd_hist.iloc[-1]), | |
| bb_upper=bbu, | |
| bb_mid=bbm, | |
| bb_lower=bbl, | |
| atr_14=atr_val, | |
| ema_trend=ema_trend, | |
| bb_position=bb_pos, | |
| rsi_zone=rsi_zone, | |
| volume_vs_avg=vol_ratio, | |
| recent_highs=recent_highs, | |
| recent_lows=recent_lows, | |
| suggested_long_tp=round(long_tp, 4), | |
| suggested_long_sl=round(long_sl, 4), | |
| suggested_short_tp=round(short_tp, 4), | |
| suggested_short_sl=round(short_sl, 4), | |
| ) | |