Oscillators

# ROC vs CMO: an UNBOUNDED oscillator next to a BOUNDED one.
import os
from datetime import datetime, timedelta

import numpy as np
from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=120)).strftime("%Y-%m-%d")
df = client.history(symbol="RELIANCE", exchange="NSE", interval="D", start_date=start, end_date=end)

# ROC (Rate of Change) returns a NUMPY ARRAY - it has no fixed ceiling.
roc = ta.roc(df["close"], 12)
print("ta.roc returns a", type(roc).__name__)

# CMO (Chande Momentum Oscillator) returns a SERIES bounded between -100 and +100.
df["CMO"] = ta.cmo(df["close"], 14)
print("ta.cmo returns a", type(df["CMO"]).__name__)

print(f"\nLatest ROC(12): {float(np.asarray(roc)[-1]):+.2f}%   (unbounded - just a % change)")
print(f"Latest CMO(14): {df['CMO'].iloc[-1]:+.2f}     (bounded -100..+100)")
print("CMO zone:", "overbought (>50)" if df["CMO"].iloc[-1] > 50
      else "oversold (<-50)" if df["CMO"].iloc[-1] < -50 else "neutral")

# TRIX: a triple-smoothed momentum oscillator that filters out noise.
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=200)).strftime("%Y-%m-%d")
df = client.history(symbol="TCS", exchange="NSE", interval="D", start_date=start, end_date=end)

# TRIX = rate of change of a triple-EMA-smoothed price. Heavy smoothing = few false signals.
df["TRIX"] = ta.trix(df["close"], 15)

trix = df["TRIX"].iloc[-1]
prev = df["TRIX"].iloc[-2]
print(df[["close", "TRIX"]].tail(5).round(4))
print(f"\nLatest TRIX(15): {trix:+.4f}")
print("Momentum:", "positive (above zero)" if trix > 0 else "negative (below zero)")
# A zero-line cross is the classic TRIX signal.
print("Crossed zero this bar:", (prev <= 0 < trix) or (prev >= 0 > trix))

# Ultimate Oscillator: blends three timeframes to cut down on false divergences.
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=150)).strftime("%Y-%m-%d")
df = client.history(symbol="SILVERM30JUN26FUT", exchange="MCX", interval="D", start_date=start, end_date=end)

# UO weights short (7), medium (14) and long (28) periods - bounded 0..100.
df["UO"] = ta.ultimate_oscillator(df["high"], df["low"], df["close"])

uo = df["UO"].iloc[-1]
print(df[["close", "UO"]].tail(5).round(2))
print(f"\nLatest Ultimate Oscillator: {uo:.1f}")
print("Zone:", "overbought (>70)" if uo > 70 else "oversold (<30)" if uo < 30 else "neutral (30-70)")

# Awesome Oscillator (AO): momentum as the gap between two midpoint averages.
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=120)).strftime("%Y-%m-%d")
df = client.history(symbol="HDFCBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

# AO = SMA(5) of the bar midpoint minus SMA(34). Uses high & low only. Unbounded, oscillates around 0.
df["AO"] = ta.awesome_oscillator(df["high"], df["low"])

ao = df["AO"].iloc[-1]
prev = df["AO"].iloc[-2]
print(df[["close", "AO"]].tail(5).round(2))
print(f"\nLatest AO: {ao:+.2f}")
print("Side of zero:", "bullish (above)" if ao > 0 else "bearish (below)")
print("Histogram bar color:", "green (rising)" if ao > prev else "red (falling)")

# PPO: MACD expressed as a PERCENTAGE, so you can compare across symbols.
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=150)).strftime("%Y-%m-%d")
df = client.history(symbol="BANKNIFTY30JUN26FUT", exchange="NFO", interval="D", start_date=start, end_date=end)

# PPO returns a TUPLE: (ppo_line, signal_line, histogram). Values are percentages.
ppo_line, signal_line, histogram = ta.ppo(df["close"])
df["PPO"] = ppo_line
df["PPO_Signal"] = signal_line
df["PPO_Hist"] = histogram

print(df[["close", "PPO", "PPO_Signal", "PPO_Hist"]].tail(5).round(3))
print(f"\nLatest PPO: {df['PPO'].iloc[-1]:+.3f}%   Signal: {df['PPO_Signal'].iloc[-1]:+.3f}%")
print("Momentum:", "bullish (PPO above signal)" if df["PPO"].iloc[-1] > df["PPO_Signal"].iloc[-1]
      else "bearish (PPO below signal)")

# DPO: removes the trend so you can see the price CYCLE underneath.
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=150)).strftime("%Y-%m-%d")
df = client.history(symbol="INFY", exchange="NSE", interval="D", start_date=start, end_date=end)

# DPO (Detrended Price Oscillator) subtracts a shifted moving average. It is NOT a momentum
# indicator - it strips out the trend to expose short cycles (peaks and troughs).
df["DPO"] = ta.dpo(df["close"], period=20)

dpo = df["DPO"].iloc[-1]
print(df[["close", "DPO"]].tail(5).round(2))
print(f"\nLatest DPO(20): {dpo:+.2f}")
print("Cycle position:", "above its detrended average (toward a peak)" if dpo > 0
      else "below its detrended average (toward a trough)")

# Aroon Oscillator: how recently did we make a new high vs a new low?
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=120)).strftime("%Y-%m-%d")
df = client.history(symbol="GOLDM03JUL26FUT", exchange="MCX", interval="D", start_date=start, end_date=end)

# Aroon Oscillator = Aroon Up minus Aroon Down. Bounded -100..+100 (uses high & low, no close).
df["AROON_OSC"] = ta.aroon_oscillator(df["high"], df["low"], period=14)

osc = df["AROON_OSC"].iloc[-1]
print(df[["close", "AROON_OSC"]].tail(5).round(1))
print(f"\nLatest Aroon Oscillator: {osc:+.1f}")
print("Trend:", "strong up (recent new highs)" if osc > 50
      else "strong down (recent new lows)" if osc < -50 else "weak / no clear trend")

# Stochastic RSI: an oscillator OF an oscillator - extra sensitive timing.
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=150)).strftime("%Y-%m-%d")
df = client.history(symbol="ICICIBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

# StochRSI returns a TUPLE (%K, %D), both bounded 0..100. It applies the Stochastic
# formula to RSI values, so it moves faster and hits extremes more often than RSI alone.
k, d = ta.stochrsi(df["close"])
df["StochRSI_K"] = k
df["StochRSI_D"] = d

kv, dv = df["StochRSI_K"].iloc[-1], df["StochRSI_D"].iloc[-1]
print(df[["close", "StochRSI_K", "StochRSI_D"]].tail(5).round(1))
print(f"\nLatest %K: {kv:.1f}   %D: {dv:.1f}")
print("Zone:", "overbought (>80)" if kv > 80 else "oversold (<20)" if kv < 20 else "neutral")
print("Short-term:", "%K above %D (bullish)" if kv > dv else "%K below %D (bearish)")

# True Strength Index (TSI): double-smoothed momentum with a signal line.
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=200)).strftime("%Y-%m-%d")
df = client.history(symbol="RELIANCE", exchange="NSE", interval="D", start_date=start, end_date=end)

# TSI returns a TUPLE (tsi_line, signal_line). Double smoothing makes it cleaner than raw momentum.
tsi_line, signal_line = ta.tsi(df["close"])
df["TSI"] = tsi_line
df["TSI_Signal"] = signal_line

t, s = df["TSI"].iloc[-1], df["TSI_Signal"].iloc[-1]
print(df[["close", "TSI", "TSI_Signal"]].tail(5).round(2))
print(f"\nLatest TSI: {t:+.2f}   Signal: {s:+.2f}")
print("Bias:", "bullish (TSI above zero)" if t > 0 else "bearish (TSI below zero)")
print("Trigger:", "TSI above signal (long)" if t > s else "TSI below signal (short)")

# Vortex Indicator (VI): two lines whose crossover marks a trend change.
import os
from datetime import datetime, timedelta

import pandas as pd
from openalgo import api

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=150)).strftime("%Y-%m-%d")
df = client.history(symbol="CRUDEOIL20JUL26FUT", exchange="MCX", interval="D", start_date=start, end_date=end)

# Vortex is two lines built from up-move (VI+) and down-move (VI-) vs the true range.
# We build it directly so you can see the formula: sum the moves over N bars, divide by
# the summed true range. When VI+ is above VI-, an uptrend; below, a downtrend.
period = 14
h, low_, c = df["high"], df["low"], df["close"]
prev_c = c.shift(1)
true_range = pd.concat([h - low_, (h - prev_c).abs(), (low_ - prev_c).abs()], axis=1).max(axis=1)
vm_plus = (h - low_.shift(1)).abs()
vm_minus = (low_ - h.shift(1)).abs()
str_sum = true_range.rolling(period).sum()
df["VI_Plus"] = vm_plus.rolling(period).sum() / str_sum
df["VI_Minus"] = vm_minus.rolling(period).sum() / str_sum

p, m = df["VI_Plus"].iloc[-1], df["VI_Minus"].iloc[-1]
print(df[["close", "VI_Plus", "VI_Minus"]].tail(5).round(3))
print(f"\nLatest VI+: {p:.3f}   VI-: {m:.3f}")
print("Trend:", "up (VI+ leads)" if p > m else "down (VI- leads)")
print("Spread:", f"{abs(p - m):.3f}  (wider = stronger trend)")

# Choppiness Index: is the market TRENDING or RANGING right now?
import os
from datetime import datetime, timedelta

from openalgo import api, ta

client = api(
    api_key=os.getenv("OPENALGO_API_KEY", "your_api_key_here"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)

end = datetime.now().strftime("%Y-%m-%d")
start = (datetime.now() - timedelta(days=120)).strftime("%Y-%m-%d")
df = client.history(symbol="NIFTY30JUN26FUT", exchange="NFO", interval="D", start_date=start, end_date=end)

# CHOP is bounded 0..100 but it measures TREND vs RANGE, not direction.
# High (>61.8) = choppy/sideways; low (<38.2) = strong directional trend.
df["CHOP"] = ta.chop(df["high"], df["low"], df["close"], period=14)

chop = df["CHOP"].iloc[-1]
print(df[["close", "CHOP"]].tail(5).round(1))
print(f"\nLatest Choppiness: {chop:.1f}")
if chop > 61.8:
    print("Regime: CHOPPY / sideways - favour range tools, avoid trend-following")
elif chop < 38.2:
    print("Regime: TRENDING strongly - favour trend-following, avoid mean-reversion")
else:
    print("Regime: in between - no clear edge for either style")