89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
"""Common technical indicators as pure numpy functions.
|
|
|
|
All functions take a 1-D price array (or H/L/C arrays) and return an array of
|
|
the same length, with leading ``NaN`` where the lookback window is not yet
|
|
full. Add strategy-specific indicators (your custom range filter, regime
|
|
detector, …) alongside these as you need them.
|
|
|
|
These are vectorized with numpy; for the hot path compile them with numba if
|
|
profiling demands it (doc 01 — numba is in the stack for that reason).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
|
|
|
|
def sma(prices: np.ndarray, period: int) -> np.ndarray:
|
|
"""Simple moving average. ``NaN`` until ``period`` values are seen."""
|
|
if period <= 0:
|
|
raise ValueError("period must be > 0")
|
|
out = np.full(prices.shape, np.nan, dtype=float)
|
|
if len(prices) < period:
|
|
return out
|
|
csum = np.cumsum(prices, dtype=float)
|
|
csum[period:] = csum[period:] - csum[:-period]
|
|
out[period - 1:] = csum[period - 1:] / period
|
|
return out
|
|
|
|
|
|
def ema(prices: np.ndarray, period: int) -> np.ndarray:
|
|
"""Exponential moving average (seeded with the first ``period`` SMA)."""
|
|
if period <= 0:
|
|
raise ValueError("period must be > 0")
|
|
out = np.full(prices.shape, np.nan, dtype=float)
|
|
if len(prices) < period:
|
|
return out
|
|
alpha = 2.0 / (period + 1.0)
|
|
out[period - 1] = prices[:period].mean()
|
|
for i in range(period, len(prices)):
|
|
out[i] = alpha * prices[i] + (1.0 - alpha) * out[i - 1]
|
|
return out
|
|
|
|
|
|
def rsi(prices: np.ndarray, period: int = 14) -> np.ndarray:
|
|
"""Relative Strength Index (Wilder's smoothing). Range ``[0, 100]``."""
|
|
if period <= 0:
|
|
raise ValueError("period must be > 0")
|
|
out = np.full(prices.shape, np.nan, dtype=float)
|
|
if len(prices) <= period:
|
|
return out
|
|
deltas = np.diff(prices, prepend=prices[0])
|
|
gains = np.where(deltas > 0, deltas, 0.0)
|
|
losses = np.where(deltas < 0, -deltas, 0.0)
|
|
avg_gain = gains[1:period + 1].mean()
|
|
avg_loss = losses[1:period + 1].mean()
|
|
for i in range(period, len(prices)):
|
|
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
|
|
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
|
|
rs = avg_gain / avg_loss if avg_loss != 0 else np.inf
|
|
out[i] = 100.0 - 100.0 / (1.0 + rs)
|
|
return out
|
|
|
|
|
|
def atr(
|
|
high: np.ndarray,
|
|
low: np.ndarray,
|
|
close: np.ndarray,
|
|
period: int = 14,
|
|
) -> np.ndarray:
|
|
"""Average True Range (Wilder's smoothing)."""
|
|
if not (len(high) == len(low) == len(close)):
|
|
raise ValueError("high/low/close must have equal length")
|
|
if period <= 0:
|
|
raise ValueError("period must be > 0")
|
|
out = np.full(close.shape, np.nan, dtype=float)
|
|
if len(close) <= period:
|
|
return out
|
|
prev_close = np.concatenate(([close[0]], close[:-1]))
|
|
tr = np.maximum.reduce([
|
|
high - low,
|
|
np.abs(high - prev_close),
|
|
np.abs(low - prev_close),
|
|
])
|
|
atr_val = tr[1:period + 1].mean()
|
|
out[period] = atr_val
|
|
for i in range(period + 1, len(close)):
|
|
atr_val = (atr_val * (period - 1) + tr[i]) / period
|
|
out[i] = atr_val
|
|
return out
|