custom_mas.py

"""
MAs logic generated by LLM.

This file will be refactored/modified when the real-time(...WS) version of srl-python-indicators comes out.
"""
import numpy as np
import pandas as pd

class MAType:
    Simple = 0
    Exponential = 1
    Weighted = 2
    Triangular = 3
    Hull = 4
    VIDYA = 5
    WilderSmoothing = 6
    KaufmanAdaptive = 7

def get_ma(arr: np.array, ma_type: MAType, ma_period: int):
    match ma_type:
        case MAType.Simple:
            return sma(arr, ma_period)
        case MAType.Exponential:
            return ema(arr, ma_period)
        case MAType.Weighted:
            return wma(arr, ma_period)
        case MAType.Triangular:
            return tma(arr, ma_period)
        case MAType.Hull:
            return hma(arr, ma_period)
        case MAType.VIDYA:
            return vidya(arr, ma_period)
        case MAType.WilderSmoothing:
            return wilder(arr, ma_period)
        case MAType.KaufmanAdaptive:
            return kama(arr, ma_period)

def get_stddev(values: pd.Series, ma_values: pd.Series, ma_period: int):
    return rolling_std_pandas(values, ma_values, ma_period)

def rolling_std_pandas(prices, ma_values, period):
    # squared diff from precomputed MA
    diff2 = (prices - ma_values) ** 2

    # rolling sum of squared diffs
    rolled = diff2.rolling(period).sum()

    # sample std
    std = np.sqrt(rolled / (period - 1))

    return std
def rolling_std_numpy(prices, ma_values, period):
    arr = np.array(prices, dtype=float)
    ma = np.array(ma_values, dtype=float)

    # sliding windows for prices
    windows = np.lib.stride_tricks.sliding_window_view(arr, period)

    std_list = [None] * (period - 1)

    # compute sample std using precomputed MA
    for i in range(len(windows)):
        mean = ma[i + period - 1]  # aligned with same index as MA
        diffs = windows[i] - mean
        std = np.sqrt(np.sum(diffs * diffs) / (period - 1))
        std_list.append(std)

    return std_list

def sma(arr, period):
    if period <= 1:
        return arr.copy()

    kernel = np.ones(period) / period
    out = np.convolve(arr, kernel, mode="full")
    out = out[period - 1:period - 1 + len(arr)]

    return out

def ema(arr, period):
    alpha = 2 / (period + 1)
    out = np.zeros_like(arr, dtype=float)
    out[0] = arr[0]

    for i in range(1, len(arr)):
        out[i] = alpha * arr[i] + (1 - alpha) * out[i-1]

    return out

def wma(arr, period):
    weights = np.arange(1, period + 1)
    kernel = weights / weights.sum()

    out = np.convolve(arr, kernel, mode="full")
    out = out[period-1:period-1+len(arr)]

    return out

def tma(arr, period):
    return sma(sma(arr, period), period)

def hma(arr, period):
    half = max(1, period // 2)
    sqrt_p = max(1, int(np.sqrt(period)))

    wma_half = wma(arr, half)
    wma_full = wma(arr, period)

    raw = 2 * wma_half - wma_full

    return wma(raw, sqrt_p)

def cmo(arr, period):
    diff = np.diff(arr, prepend=arr[0])

    up = np.where(diff > 0, diff, 0)
    down = np.where(diff < 0, -diff, 0)

    up_sum = np.convolve(up, np.ones(period), mode="full")[period-1:period-1+len(arr)]
    down_sum = np.convolve(down, np.ones(period), mode="full")[period-1:period-1+len(arr)]

    denom = up_sum + down_sum

    cmo_val = np.where(denom == 0, 0, 100 * (up_sum - down_sum) / denom)

    return cmo_val

def vidya(arr, period, alpha_base=0.65):
    cmo_val = cmo(arr, period) / 100.0
    k = alpha_base * np.abs(cmo_val)

    out = np.zeros_like(arr, dtype=float)
    out[0] = arr[0]

    for i in range(1, len(arr)):
        out[i] = k[i] * arr[i] + (1 - k[i]) * out[i-1]

    return out

def wilder(arr, period):
    alpha = 1 / period
    out = np.zeros_like(arr, dtype=float)
    out[0] = arr[0]

    for i in range(1, len(arr)):
        out[i] = out[i-1] + alpha * (arr[i] - out[i-1])

    return out

def kama(arr, period, fast=2, slow=30):
    change = np.abs(arr - np.concatenate(([arr[0]], arr[:-period])))
    change[:period] = 0

    volatility = np.convolve(np.abs(np.diff(arr, prepend=arr[0])),
                             np.ones(period), mode="full")[period-1:period-1+len(arr)]

    er = np.where(volatility == 0, 0, change / volatility)

    fast_sc = 2 / (fast + 1)
    slow_sc = 2 / (slow + 1)
    sc = (er * (fast_sc - slow_sc) + slow_sc) ** 2

    out = np.zeros_like(arr, dtype=float)
    out[0] = arr[0]

    for i in range(1, len(arr)):
        out[i] = out[i-1] + sc[i] * (arr[i] - out[i-1])

    return out

"""
# Direct conversion

import math

class MAType:
    Simple = 0
    Exponential = 1
    Weighted = 2
    Triangular = 3
    Hull = 4
    VIDYA = 5
    WilderSmoothing = 6
    KaufmanAdaptive = 7

def std_dev(index, period, ma_value, buffer):
    mean = ma_value
    sum_sq = 0.0
    for i in range(index - period + 1, index + 1):
        try:
            diff = buffer[i] - mean
            sum_sq += diff * diff
        except:
            pass
    return math.sqrt(sum_sq / (period - 1)) if period > 1 else 0.0


def sma(index, period, buffer):
    if len(buffer) < period:
        return float("nan")

    total = 0.0
    for i in range(index, index - period, -1):
        try:
            total += buffer[i]
        except:
            pass

    return total / period


def ema(index, period, buffer, ema_dict):
    if len(ema_dict) == 0:
        ema_dict[0] = buffer[index]
        ema_dict[1] = buffer[index]
        ema_dict[index] = buffer[index]
        return buffer[index]

    k = 2.0 / (period + 1)
    prev = ema_dict[0]
    value = buffer[index] * k + prev * (1 - k)

    # detect index jumps (like Sunday bar gaps)
    if index != max(ema_dict.keys()):
        prev_last = ema_dict[1]
        ema_dict.clear()
        ema_dict[0] = prev_last
        ema_dict[1] = value
        ema_dict[index] = value
    else:
        ema_dict[1] = value
        ema_dict[index] = value

    return value


def wma(index, period, buffer, override_last=None):
    if len(buffer) < period:
        return float("nan")

    numerator = 0.0
    denominator = 0.0
    w = 1

    start = index - period + 1
    for i in range(start, index + 1):
        try:
            if i == index and override_last is not None:
                v = override_last
            else:
                v = buffer[i]
        except:
            v = 0
        numerator += v * w
        denominator += w
        w += 1

    return numerator / denominator


def tma(index, period, buffer):
    if period <= 1:
        return buffer[index]

    if len(buffer) < 2 * period - 2:
        return float("nan")

    sum_sma = 0.0
    for k in range(index - period + 1, index + 1):
        sum_sma += sma(k, period, buffer)

    return sum_sma / period


def hull(index, period, buffer):
    if period < 2:
        return buffer[index]

    half = max(1, period // 2)
    sqrt_p = max(1, int(round(math.sqrt(period))))

    wma_half = wma(index, half, buffer)
    wma_full = wma(index, period, buffer)

    raw = 2 * wma_half - wma_full
    return wma(index, sqrt_p, buffer, override_last=raw)


def wilder(index, period, buffer, wilder_dict):
    if len(wilder_dict) == 0:
        wilder_dict[0] = buffer[index]
        wilder_dict[1] = buffer[index]
        wilder_dict[index] = buffer[index]
        return buffer[index]

    prev = wilder_dict[0]
    value = (prev * (period - 1) + buffer[index]) / period

    if index != max(wilder_dict.keys()):
        prev_last = wilder_dict[1]
        wilder_dict.clear()
        wilder_dict[0] = prev_last
        wilder_dict[1] = value
        wilder_dict[index] = value
    else:
        wilder_dict[1] = value
        wilder_dict[index] = value

    return value


def kama(index, period, fast, slow, buffer, kama_dict):
    if len(kama_dict) == 0:
        kama_dict[0] = buffer[index]
        kama_dict[1] = buffer[index]
        kama_dict[index] = buffer[index]
        return buffer[index]

    if len(buffer) < period:
        return sma(index, period, buffer)

    # Change
    try:
        change = abs(buffer[index] - buffer[index - period])
    except:
        # search backwards for existing index
        missing = index - period
        for i in range(missing, index + 1):
            if i in buffer:
                missing = i
                break
        change = abs(buffer[index] - buffer[missing])

    # Volatility
    vol = 0.0
    for i in range(index - period + 1, index + 1):
        try:
            vol += abs(buffer[i] - buffer[i - 1])
        except:
            pass

    er = 0 if vol == 0 else (change / vol)

    fast_sc = 2.0 / (fast + 1)
    slow_sc = 2.0 / (slow + 1)
    sc = (er * (fast_sc - slow_sc) + slow_sc) ** 2

    prev = kama_dict[0]
    value = prev + sc * (buffer[index] - prev)

    if index != max(kama_dict.keys()):
        prev_last = kama_dict[1]
        kama_dict.clear()
        kama_dict[0] = prev_last
        kama_dict[1] = value
        kama_dict[index] = value
    else:
        kama_dict[1] = value
        kama_dict[index] = value

    return value


def cmo(index, length, buffer):
    if index < 1 or length < 1:
        return 0.0

    start = max(1, index - length + 1)
    up = 0.0
    down = 0.0

    for i in range(start, index + 1):
        try:
            diff = buffer[i] - buffer[i - 1]
            if diff > 0:
                up += diff
            else:
                down += -diff
        except:
            pass

    denom = up + down
    return 0.0 if denom == 0 else 100.0 * (up - down) / denom


def vidya(index, period, buffer, vidya_dict):
    if len(vidya_dict) == 0:
        vidya_dict[0] = buffer[index]
        vidya_dict[1] = buffer[index]
        vidya_dict[index] = buffer[index]
        return buffer[index]

    cmo_val = cmo(index, period, buffer)
    alpha_base = 0.65
    k = alpha_base * abs(cmo_val / 100.0)

    prev = vidya_dict[0]
    value = k * buffer[index] + (1 - k) * prev

    if index != max(vidya_dict.keys()):
        prev_last = vidya_dict[1]
        vidya_dict.clear()
        vidya_dict[0] = prev_last
        vidya_dict[1] = value
        vidya_dict[index] = value
    else:
        vidya_dict[1] = value
        vidya_dict[index] = value

    return value
"""