tvsignals-to-tg/app/indicators/common.py

from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from typing import Literal

import numpy as np
import pandas as pd

StrategyId = Literal["ltf", "fvg"]
Side = Literal["long", "short"]


@dataclass(frozen=True)
class IndicatorSignal:
    strategy_id: StrategyId
    side: Side
    entry: float
    close: float
    sl: float
    tp1: float
    tp2: float | None
    tp3: float | None
    size_usd: float
    bar_open_ts: int
    visual_timeframe: str


def ohlc_frame(df: pd.DataFrame) -> pd.DataFrame:
    """Normalize OHLC frame to lowercase columns and UTC DatetimeIndex."""
    out = df.copy()
    out.columns = [str(c).lower() for c in out.columns]
    required = {"open", "high", "low", "close"}
    missing = required - set(out.columns)
    if missing:
        raise ValueError(f"OHLC frame missing columns: {sorted(missing)}")
    if not isinstance(out.index, pd.DatetimeIndex):
        raise ValueError("OHLC frame must be indexed by timestamp")
    if out.index.tz is None:
        out.index = out.index.tz_localize("UTC")
    else:
        out.index = out.index.tz_convert("UTC")
    cols = ["open", "high", "low", "close"]
    if "volume" in out.columns:
        cols.append("volume")
    return out[cols]


def epoch_ms(index: pd.DatetimeIndex) -> np.ndarray:
    utc = index.tz_convert("UTC") if index.tz is not None else index.tz_localize("UTC")
    return utc.asi8.astype(np.float64) / 1_000_000.0


def alma(
    series: pd.Series, length: int = 500, offset: float = 0.85, sigma: float = 5.0
) -> pd.Series:
    """Arnaud Legoux Moving Average (Pine ta.alma compatible)."""
    if length < 1:
        raise ValueError("ALMA length must be >= 1")
    m = offset * (length - 1)
    s = length / sigma
    idx = np.arange(length, dtype=float)
    weights = np.exp(-((idx - m) ** 2) / (2 * s * s))
    weights /= weights.sum()

    values = series.to_numpy(dtype=float)
    out = np.full(len(values), np.nan, dtype=float)
    if len(values) < length:
        return pd.Series(out, index=series.index)

    valid = np.convolve(values, weights[::-1], mode="valid")
    out[length - 1 :] = valid
    nan_in_window = np.convolve(np.isnan(values).astype(float), np.ones(length), mode="valid") > 0
    out[length - 1 :][nan_in_window] = np.nan
    return pd.Series(out, index=series.index)


def rma(series: pd.Series, length: int) -> pd.Series:
    """Wilder's RMA (Pine ta.rma)."""
    return series.ewm(alpha=1 / length, adjust=False, min_periods=length).mean()


def ha_rsi(
    open_: pd.Series,
    high: pd.Series,
    low: pd.Series,
    close: pd.Series,
    length: int = 14,
) -> pd.Series:
    """RSI on Heikin-Ashi close (Pine f_ha_rsi)."""
    ha_close = (open_ + high + low + close) / 4.0
    delta = ha_close.diff()
    up = rma(delta.clip(lower=0), length)
    down = rma((-delta).clip(lower=0), length)
    rs = up / down.replace(0, np.nan)
    rsi = 100 - (100 / (1 + rs))
    rsi = rsi.where(down != 0, 100.0)
    rsi = rsi.where(up != 0, 0.0)
    both_zero = (up == 0) & (down == 0)
    return rsi.where(~both_zero, 100.0)


def true_range(high: pd.Series, low: pd.Series, close: pd.Series) -> pd.Series:
    prev_close = close.shift(1)
    return pd.concat(
        [high - low, (high - prev_close).abs(), (low - prev_close).abs()],
        axis=1,
    ).max(axis=1)


def atr(high: pd.Series, low: pd.Series, close: pd.Series, length: int) -> pd.Series:
    """Pine ta.atr(length)."""
    return rma(true_range(high, low, close), length)


def compute_fractals(high: np.ndarray, low: np.ndarray, n: int = 5) -> tuple[np.ndarray, np.ndarray]:
    """Pine upFractal / downFractal flags on the confirmation bar (pivot at i - n)."""
    size = len(high)
    up = np.zeros(size, dtype=bool)
    down = np.zeros(size, dtype=bool)
    for c in range(n, size):
        p = c - n
        hp = high[p]
        lp = low[p]
        if np.isnan(hp) or np.isnan(lp):
            continue
        up_prefix1 = (p + 1 < size) and (high[p + 1] <= hp)
        up_prefix2 = up_prefix1 and (p + 2 < size) and (high[p + 2] <= hp)
        up_prefix3 = up_prefix2 and (p + 3 < size) and (high[p + 3] <= hp)
        up_prefix4 = up_prefix3 and (p + 4 < size) and (high[p + 4] <= hp)
        down_prefix1 = (p + 1 < size) and (low[p + 1] >= lp)
        down_prefix2 = down_prefix1 and (p + 2 < size) and (low[p + 2] >= lp)
        down_prefix3 = down_prefix2 and (p + 3 < size) and (low[p + 3] >= lp)
        down_prefix4 = down_prefix3 and (p + 4 < size) and (low[p + 4] >= lp)
        upflag_down = True
        upflag0 = True
        upflag1 = True
        upflag2 = True
        upflag3 = True
        upflag4 = True
        for i in range(1, n + 1):
            if p - i < 0 or not (high[p - i] < hp):
                upflag_down = False
            if p + i >= size or not (high[p + i] < hp):
                upflag0 = False
            if p + i + 1 >= size or not (high[p + i + 1] < hp):
                upflag1 = False
            if p + i + 2 >= size or not (high[p + i + 2] < hp):
                upflag2 = False
            if p + i + 3 >= size or not (high[p + i + 3] < hp):
                upflag3 = False
            if p + i + 4 >= size or not (high[p + i + 4] < hp):
                upflag4 = False
        upflag1 = upflag1 and up_prefix1
        upflag2 = upflag2 and up_prefix2
        upflag3 = upflag3 and up_prefix3
        upflag4 = upflag4 and up_prefix4
        up[c] = upflag_down and (upflag0 or upflag1 or upflag2 or upflag3 or upflag4)
        downflag_down = True
        downflag0 = True
        downflag1 = True
        downflag2 = True
        downflag3 = True
        downflag4 = True
        for i in range(1, n + 1):
            if p - i < 0 or not (low[p - i] > lp):
                downflag_down = False
            if p + i >= size or not (low[p + i] > lp):
                downflag0 = False
            if p + i + 1 >= size or not (low[p + i + 1] > lp):
                downflag1 = False
            if p + i + 2 >= size or not (low[p + i + 2] > lp):
                downflag2 = False
            if p + i + 3 >= size or not (low[p + i + 3] > lp):
                downflag3 = False
            if p + i + 4 >= size or not (low[p + i + 4] > lp):
                downflag4 = False
        downflag1 = downflag1 and down_prefix1
        downflag2 = downflag2 and down_prefix2
        downflag3 = downflag3 and down_prefix3
        downflag4 = downflag4 and down_prefix4
        down[c] = downflag_down and (downflag0 or downflag1 or downflag2 or downflag3 or downflag4)
    return up, down


def tick_decimals(tick: float) -> int:
    exponent = Decimal(str(tick)).normalize().as_tuple().exponent
    if isinstance(exponent, int):
        return max(0, -exponent)
    return 0


def round_to_mintick(price: float, tick: float) -> float:
    """Pine math.round_to_mintick (half-up to exchange tick)."""
    step = Decimal(str(tick))
    if step <= 0:
        return price
    quantized = (Decimal(str(price)) / step).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
    return float(quantized * step)


def realized_pnl_pct(prev_side: str, prev_entry: float, exit_price: float) -> float:
    """Signed percent from previous entry to exit; long/short from the closed side."""
    if prev_entry == 0:
        raise ValueError("entry price is zero")
    if prev_side == "long":
        return (exit_price - prev_entry) / prev_entry * 100
    return (prev_entry - exit_price) / prev_entry * 100


def calc_size(entry: float, sl: float, risk_usd: float) -> float:
    """Pine calc_size: round(risk / abs(entry-sl)/entry)."""
    if entry == 0 or np.isnan(entry) or np.isnan(sl):
        return float("nan")
    stop_pct = abs(entry - sl) / entry
    if stop_pct <= 0:
        return float("nan")
    return float(round(risk_usd / stop_pct))


def format_px(value: float | None, tick: float | None = None) -> str:
    if value is None:
        return ""
    number = float(value)
    if np.isnan(number):
        return ""
    if tick is not None and tick > 0:
        rounded = round_to_mintick(number, tick)
        return f"{rounded:.{tick_decimals(tick)}f}"
    if abs(number) >= 1000:
        text = f"{number:.4f}"
    elif abs(number) >= 1:
        text = f"{number:.6f}"
    else:
        text = f"{number:.8f}"
    return text.rstrip("0").rstrip(".")