Statistical, Hybrid & Utility Functions

# Linear Regression line + slope: fit a straight line to price and read its tilt.
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="RELIANCE", exchange="NSE", interval="D", start_date=start, end_date=end)

# linreg fits a least-squares line; its endpoint is the "fair value" for that bar.
df["LINREG"] = ta.linreg(df["close"], 20)
# lrslope is the tilt of that line: positive = rising trend, negative = falling.
df["SLOPE"] = ta.lrslope(df["close"], period=20)

print(df[["close", "LINREG", "SLOPE"]].tail(5).round(2))
slope = df["SLOPE"].iloc[-1]
print(f"\nLatest slope: {slope:+.3f}")
print("Trend:", "up" if slope > 0 else "down" if slope < 0 else "flat")
gap = df["close"].iloc[-1] - df["LINREG"].iloc[-1]
print(f"Price is {gap:+.2f} vs its regression line ({'above' if gap > 0 else 'below'}).")

# Pearson Correlation: do two instruments move together?
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")
a = client.history(symbol="HDFCBANK", exchange="NSE", interval="D", start_date=start, end_date=end)
b = client.history(symbol="ICICIBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

# correlation runs from -1 (move opposite) through 0 (unrelated) to +1 (move together).
corr = ta.correlation(a["close"], b["close"], 20)
a["CORR"] = corr

c = a["CORR"].iloc[-1]
print(a[["close", "CORR"]].tail(5).round(3))
print(f"\nLatest 20-day correlation HDFCBANK vs ICICIBANK: {c:+.3f}")
print("Reading:", "strongly linked" if c > 0.7 else "loosely linked" if c > 0.3
      else "barely related" if c > -0.3 else "moving opposite")
print("Pairs traders want a HIGH, stable correlation between the two legs.")

# Beta: how much a stock amplifies the market's moves.
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")
stock = client.history(symbol="HDFCBANK", exchange="NSE", interval="D", start_date=start, end_date=end)
mkt = client.history(symbol="NIFTY", exchange="NSE_INDEX", interval="D", start_date=start, end_date=end)

# Beta measures the stock's volatility relative to the market (here NIFTY).
# Beta 1 = moves with the index; >1 = amplifies it; <1 = calmer than the index.
stock["BETA"] = ta.beta(stock["close"], mkt["close"], period=60)

b = stock["BETA"].iloc[-1]
print(stock[["close", "BETA"]].tail(5).round(3))
print(f"\nLatest 60-day Beta vs NIFTY: {b:.2f}")
print("Behaviour:", "amplifies the market (aggressive)" if b > 1.1
      else "tracks the market" if b > 0.9 else "calmer than the market (defensive)")

# Variance and Time Series Forecast: dispersion now, projected value next.
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)

# Variance measures how spread out prices are (the square of standard deviation).
df["VAR"] = ta.variance(df["close"], lookback=20)
# TSF projects the regression line one step ahead - a simple forecast of next value.
df["TSF"] = ta.tsf(df["close"], 14)

print(df[["close", "VAR", "TSF"]].tail(5).round(2))
print(f"\nLatest variance(20): {df['VAR'].iloc[-1]:,.2f}  (higher = wilder swings)")
print(f"Forecast for next bar (TSF): {df['TSF'].iloc[-1]:,.2f}")
print(f"Last actual close          : {df['close'].iloc[-1]:,.2f}")

# Rolling Median: a "middle" price that ignores spikes and outliers.
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=90)).strftime("%Y-%m-%d")
df = client.history(symbol="SBIN", exchange="NSE", interval="D", start_date=start, end_date=end)

# The median is the middle value of a window. Unlike the average, one freak
# spike can't drag it around - useful when data is noisy.
df["MEDIAN"] = ta.median(df["close"], 5)
df["MEAN"] = df["close"].rolling(5).mean()

print(df[["close", "MEDIAN", "MEAN"]].tail(5).round(2))
print(f"\nLatest 5-day median: {df['MEDIAN'].iloc[-1]:.2f}")
print(f"Latest 5-day mean  : {df['MEAN'].iloc[-1]:.2f}")
print("The median sits near the typical price even when a single day spikes.")

# ADX: how STRONG is the trend (regardless of direction)?
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="TCS", exchange="NSE", interval="D", start_date=start, end_date=end)

# ADX returns a TUPLE: (+DI, -DI, ADX). ADX measures strength; +DI/-DI give direction.
di_plus, di_minus, adx = ta.adx(df["high"], df["low"], df["close"], period=14)
df["ADX"] = adx
df["DI+"] = di_plus
df["DI-"] = di_minus

a = df["ADX"].iloc[-1]
print(df[["close", "DI+", "DI-", "ADX"]].tail(5).round(1))
print(f"\nLatest ADX: {a:.1f}")
print("Trend strength:", "strong (>25)" if a > 25 else "weak / ranging (<20)" if a < 20 else "developing")
print("Direction:", "up (+DI leads)" if df["DI+"].iloc[-1] > df["DI-"].iloc[-1] else "down (-DI leads)")

# Aroon Up / Aroon Down: time since the last new high vs 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="NIFTY30JUN26FUT", exchange="NFO", interval="D", start_date=start, end_date=end)

# aroon returns a TUPLE (up, down). Each runs 0..100. High Up = recent new high.
aroon_up, aroon_down = ta.aroon(df["high"], df["low"], period=14)
df["Aroon_Up"] = aroon_up
df["Aroon_Down"] = aroon_down

up, dn = df["Aroon_Up"].iloc[-1], df["Aroon_Down"].iloc[-1]
print(df[["close", "Aroon_Up", "Aroon_Down"]].tail(5).round(0))
print(f"\nAroon Up: {up:.0f}   Aroon Down: {dn:.0f}")
print("Signal:", "strong uptrend" if up > 70 and dn < 30
      else "strong downtrend" if dn > 70 and up < 30 else "no dominant trend")

# Pivot Points: classic support/resistance levels from the prior bar.
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=60)).strftime("%Y-%m-%d")
df = client.history(symbol="CRUDEOIL20JUL26FUT", exchange="MCX", interval="D", start_date=start, end_date=end)

# pivot_points returns a 7-TUPLE: pivot, r1, s1, r2, s2, r3, s3.
pivot, r1, s1, r2, s2, r3, s3 = ta.pivot_points(df["high"], df["low"], df["close"])
df["Pivot"], df["R1"], df["S1"] = pivot, r1, s1

last = df.iloc[-1]
print(df[["close", "S1", "Pivot", "R1"]].tail(3).round(1))
print(f"\nLatest close: {last['close']:.1f}")
print(f"  Resistance 1: {last['R1']:.1f}")
print(f"  Pivot       : {last['Pivot']:.1f}")
print(f"  Support 1   : {last['S1']:.1f}")
print("Bias:", "above pivot - bullish" if last["close"] > last["Pivot"] else "below pivot - bearish")

# Parabolic SAR: a trailing stop that flips with the trend.
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="RELIANCE", exchange="NSE", interval="D", start_date=start, end_date=end)

# Despite the docs hinting at a tuple, ta.psar returns a SINGLE Series of stop levels.
# When price is above the SAR dot you're long; below, you're short.
df["SAR"] = ta.psar(df["high"], df["low"], acceleration=0.02, maximum=0.2)
print("ta.psar returns a", type(df["SAR"]).__name__)

last = df.iloc[-1]
print(df[["close", "SAR"]].tail(5).round(2))
print(f"\nClose: {last['close']:.2f}   SAR: {last['SAR']:.2f}")
print("Position:", "LONG (price above SAR)" if last["close"] > last["SAR"] else "SHORT (price below SAR)")
print(f"Trailing stop sits at {last['SAR']:.2f}.")

# DMI directional lines + Williams Fractals turning points.
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="INFY", exchange="NSE", interval="D", start_date=start, end_date=end)

# DMI returns a TUPLE (+DI, -DI) - the directional half of the ADX system.
di_plus, di_minus = ta.dmi(df["high"], df["low"], df["close"], period=14)
df["DI+"], df["DI-"] = di_plus, di_minus

# Fractals return a TUPLE of two BOOLEAN Series marking local tops and bottoms.
frac_up, frac_down = ta.fractals(df["high"], df["low"], periods=2)
df["FracUp"], df["FracDown"] = frac_up, frac_down

print(df[["close", "DI+", "DI-"]].tail(3).round(1))
print("Direction:", "bullish (+DI > -DI)" if df["DI+"].iloc[-1] > df["DI-"].iloc[-1] else "bearish (-DI > +DI)")
print(f"\nUp-fractals (swing highs) found: {int(df['FracUp'].sum())}")
print(f"Down-fractals (swing lows) found: {int(df['FracDown'].sum())}")
print("Last swing-high price:", round(df.loc[df["FracUp"], "high"].iloc[-1], 2))

# crossover / crossunder: the moment one line crosses another.
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=200)).strftime("%Y-%m-%d")
df = client.history(symbol="RELIANCE", exchange="NSE", interval="D", start_date=start, end_date=end)

fast = ta.ema(df["close"], 10)
slow = ta.ema(df["close"], 20)

# crossover/crossunder return NUMPY BOOLEAN ARRAYS: True only on the bar of the cross.
buy = ta.crossover(fast, slow)
sell = ta.crossunder(fast, slow)
print("ta.crossover returns a", type(buy).__name__, "of dtype", buy.dtype)

print(f"\nBuy crosses (10 EMA over 20 EMA): {int(np.sum(buy))}")
print(f"Sell crosses (10 EMA under 20 EMA): {int(np.sum(sell))}")

# Use the boolean array to pull the dates/prices where a cross happened.
buy_dates = df.index[buy]
print("\nMost recent BUY cross dates:")
for d in buy_dates[-3:]:
    print(f"  {d.date()}  close {df.loc[d, 'close']:.2f}")

# highest / lowest: rolling extremes that define breakout levels.
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="GOLDM03JUL26FUT", exchange="MCX", interval="D", start_date=start, end_date=end)

# highest/lowest return NUMPY ARRAYS of the rolling max/min over the window.
hh = ta.highest(df["high"], 20)
ll = ta.lowest(df["low"], 20)
print("ta.highest returns a", type(hh).__name__)

# A 20-day breakout: today's close pushes past the prior 20-day high.
prior_high = np.asarray(hh)[-2]   # the highest high as of yesterday
last_close = df["close"].iloc[-1]
print(f"\n20-day high (as of yesterday): {prior_high:,.0f}")
print(f"20-day low  (latest)         : {np.asarray(ll)[-1]:,.0f}")
print(f"Today's close                : {last_close:,.0f}")
print("Breakout:", "YES - new 20-day high" if last_close > prior_high else "no breakout yet")

# exrem: remove repeated signals so you act only on the FIRST 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=250)).strftime("%Y-%m-%d")
df = client.history(symbol="SBIN", exchange="NSE", interval="D", start_date=start, end_date=end)

rsi = ta.rsi(df["close"], 14)
# Raw conditions fire on EVERY bar the rule is true - lots of duplicates.
raw_buy = ta.crossover(rsi, np.full(len(rsi), 30.0))   # RSI crossing up through 30
raw_sell = ta.crossunder(rsi, np.full(len(rsi), 70.0))  # RSI crossing down through 70

# exrem keeps the first buy, ignores further buys until a sell has occurred (and vice versa).
clean_buy = ta.exrem(raw_buy, raw_sell)
clean_sell = ta.exrem(raw_sell, raw_buy)

print(f"Raw buy signals   : {int(np.sum(raw_buy))}")
print(f"Cleaned buy signals: {int(np.sum(clean_buy))}  (duplicates removed)")
print(f"Raw sell signals   : {int(np.sum(raw_sell))}")
print(f"Cleaned sell signals: {int(np.sum(clean_sell))}")
print("\nexrem stops you from buying again while you are already 'in'.")

# flip: turn entry/exit signals into a continuous "are we in?" state.
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=250)).strftime("%Y-%m-%d")
df = client.history(symbol="RELIANCE", exchange="NSE", interval="D", start_date=start, end_date=end)

fast, slow = ta.ema(df["close"], 10), ta.ema(df["close"], 20)
buy = ta.crossover(fast, slow)
sell = ta.crossunder(fast, slow)

# flip turns ON at the first buy and stays ON until a sell flips it OFF.
# It converts two momentary signals into a holding state (1 = in position, 0 = flat).
in_position = ta.flip(buy, sell)
print("ta.flip returns a", type(in_position).__name__)

df["InPos"] = np.asarray(in_position).astype(int)
days_in = int(df["InPos"].sum())
print(f"\nBars held long: {days_in} out of {len(df)}")
print("Currently:", "HOLDING long" if df["InPos"].iloc[-1] else "FLAT")
print("\nLast 8 bars (1 = in position):")
print(df[["close", "InPos"]].tail(8).round(2))

# valuewhen: recall the price at the last time a condition was true.
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=250)).strftime("%Y-%m-%d")
df = client.history(symbol="HDFCBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

fast, slow = ta.ema(df["close"], 10), ta.ema(df["close"], 20)
buy = ta.crossover(fast, slow)

# valuewhen pulls the value of a series at the bar a condition last fired.
# n=1 = most recent buy, n=2 = the buy before that.
entry_price = ta.valuewhen(buy, df["close"], 1)
print("ta.valuewhen returns a", type(entry_price).__name__)

last_entry = float(np.asarray(entry_price)[-1])
now = df["close"].iloc[-1]
print(f"\nPrice at the last EMA buy cross: {last_entry:.2f}")
print(f"Current price                  : {now:.2f}")
print(f"Open trade P&L                 : {(now - last_entry) / last_entry * 100:+.2f}%")
print("\nvaluewhen is how a strategy remembers its entry to size stops and targets.")

# rising / falling: simple trend filters you can stack onto any signal.
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="ICICIBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

# rising/falling return NUMPY BOOLEAN ARRAYS: True if a value is higher/lower than n bars ago.
price_rising = ta.rising(df["close"], 3)
vol_rising = ta.rising(df["volume"], 3)
print("ta.rising returns a", type(price_rising).__name__, "of dtype", price_rising.dtype)

# Combine filters with & to demand both at once: a price up-move backed by rising volume.
confirmed = price_rising & vol_rising

print(f"\nBars where price has risen 3 sessions : {int(np.sum(price_rising))}")
print(f"Bars where BOTH price AND volume rising: {int(np.sum(confirmed))}")
print("Latest bar:", "confirmed up-move (price + volume)" if confirmed[-1]
      else "no confirmed up-move")
print(f"Price falling now? {bool(ta.falling(df['close'], 3)[-1])}")