Spaces:
Running
Running
File size: 4,931 Bytes
7d10940 8985e43 7d10940 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """Deterministic indicator math — no AI, no guessing.
All formulas match TradingView built-ins exactly.
Indicators run on CLOSED candles only (caller must drop the forming bar).
"""
from __future__ import annotations
import pandas as pd
def ema(series: pd.Series, length: int) -> pd.Series:
return series.ewm(span=length, adjust=False).mean()
def rma(series: pd.Series, length: int) -> pd.Series:
return series.ewm(alpha=1.0 / length, adjust=False).mean()
def rsi(close: pd.Series, length: int = 14) -> pd.Series:
delta = close.diff()
gain = rma(delta.clip(lower=0.0), length)
loss = rma(-delta.clip(upper=0.0), length)
rs = gain / loss.replace(0.0, 1e-12)
return 100.0 - (100.0 / (1.0 + rs))
def true_range(df: pd.DataFrame) -> pd.Series:
prev = df["close"].shift(1)
return pd.concat([
df["high"] - df["low"],
(df["high"] - prev).abs(),
(df["low"] - prev).abs(),
], axis=1).max(axis=1)
def atr(df: pd.DataFrame, length: int = 14) -> pd.Series:
return rma(true_range(df), length)
def swing_pivots(df: pd.DataFrame, k: int = 3):
highs, lows = [], []
h, l = df["high"].values, df["low"].values
for i in range(k, len(df) - k):
wh = h[i - k: i + k + 1]
wl = l[i - k: i + k + 1]
if h[i] == wh.max() and (wh == h[i]).sum() == 1:
highs.append(float(h[i]))
if l[i] == wl.min() and (wl == l[i]).sum() == 1:
lows.append(float(l[i]))
return highs, lows
def nearest_levels(price, pivot_highs, pivot_lows, lookback=20):
pts = pivot_highs[-lookback:] + pivot_lows[-lookback:]
supports = [p for p in pts if p < price]
resistances = [p for p in pts if p > price]
return (max(supports) if supports else None,
min(resistances) if resistances else None)
def structure_tag(df: pd.DataFrame) -> str:
e20 = ema(df["close"], 20)
e50 = ema(df["close"], 50)
close = df["close"].iloc[-1]
e20_now, e20_prev = e20.iloc[-1], e20.iloc[-6]
e50_now = e50.iloc[-1]
if e20_now > e50_now and close > e20_now and e20_now > e20_prev:
return "uptrend"
if e20_now < e50_now and close < e20_now and e20_now < e20_prev:
return "downtrend"
return "range"
def compute_taker_ratio(df: pd.DataFrame, lookback: int = 10) -> float | None:
"""Return taker-buy ratio over the last `lookback` CLOSED candles.
Formula: sum(taker_buy_vol[-lookback:]) / sum(volume[-lookback:])
Closed candles = df[:-1] (the last row is the forming/live bar).
Returns None if data is missing, insufficient, or volume is zero.
Clamps result to [0.0, 1.0] to guard against exchange data errors.
Window choice — lookback=10 on 15m = 150 min (2.5 h):
• Too short (≤3): single-candle spikes dominate; ratio is noisy.
• Too long (≥20): captures prior sessions; signal becomes stale.
• 10 candles smooths intra-hour noise while staying within the same
trading session, making it actionable for the scorer's 15m signals.
"""
if "taker_buy_vol" not in df.columns:
return None
closed = df.iloc[:-1] # drop the live/forming candle
if len(closed) < lookback: # insufficient history
return None
window = closed.iloc[-lookback:]
tbv = pd.to_numeric(window["taker_buy_vol"], errors="coerce")
vol = pd.to_numeric(window["volume"], errors="coerce")
# Drop rows where either column is NaN so they cancel symmetrically
mask = tbv.notna() & vol.notna()
tbv, vol = tbv[mask], vol[mask]
if len(tbv) == 0: # all NaN after cleaning
return None
total_vol = vol.sum()
if total_vol == 0.0: # all-zero volume (halted / bad data)
return None
ratio = float(tbv.sum() / total_vol)
return max(0.0, min(1.0, ratio)) # clamp: handles taker_buy_vol > volume
def analyze_timeframe(df: pd.DataFrame) -> dict:
if len(df) < 60:
return {"error": f"insufficient history ({len(df)} candles)"}
close = df["close"]
e20 = ema(close, 20)
e50 = ema(close, 50)
r = rsi(close, 14)
a = atr(df, 14)
vol = df["volume"]
vol_avg20 = vol.rolling(20).mean()
ph, pl = swing_pivots(df, k=3)
last_close = float(close.iloc[-1])
sup, res = nearest_levels(last_close, ph, pl)
return {
"close": round(last_close, 8),
"ema20": round(float(e20.iloc[-1]), 6),
"ema50": round(float(e50.iloc[-1]), 6),
"rsi14": round(float(r.iloc[-1]), 2),
"atr14": round(float(a.iloc[-1]), 6),
"atr_pct": round(float(a.iloc[-1]) / last_close * 100, 3),
"vol_ratio": round(float(vol.iloc[-1] / vol_avg20.iloc[-1]), 2)
if vol_avg20.iloc[-1] > 0 else None,
"support": round(sup, 6) if sup else None,
"resistance": round(res, 6) if res else None,
"structure": structure_tag(df),
}
|