Momentum & Volatility Trading

# Build a small universe of 8 NSE stocks and download their daily closes.
import os
import time
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=400)).strftime("%Y-%m-%d")

universe = ["RELIANCE", "TCS", "INFY", "HDFCBANK", "ICICIBANK", "SBIN", "HINDUNILVR", "ITC"]


def closes(symbol):
    # Retry -- fetching many symbols in a loop can hit a busy data server.
    for _ in range(3):
        df = client.history(symbol=symbol, exchange="NSE", interval="D",
                            start_date=start, end_date=end)
        if isinstance(df, pd.DataFrame) and "close" in df:
            return df["close"]
        time.sleep(1)
    raise RuntimeError(f"Could not fetch history for {symbol}: {df}")


# One tidy table: dates down the rows, stocks across the columns.
panel = pd.DataFrame({symbol: closes(symbol) for symbol in universe}).dropna()
print("Universe:", len(universe), "stocks,", len(panel), "common trading days")
print(panel.tail(3).round(1))

# Cross-sectional momentum: rank the universe by 60-day return, buy the leaders.
import os
import time
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=400)).strftime("%Y-%m-%d")

universe = ["RELIANCE", "TCS", "INFY", "HDFCBANK", "ICICIBANK", "SBIN", "HINDUNILVR", "ITC"]


def closes(symbol):
    for _ in range(3):
        df = client.history(symbol=symbol, exchange="NSE", interval="D",
                            start_date=start, end_date=end)
        if isinstance(df, pd.DataFrame) and "close" in df:
            return df["close"]
        time.sleep(1)
    raise RuntimeError(f"Could not fetch history for {symbol}: {df}")


panel = pd.DataFrame({s: closes(s) for s in universe}).dropna()

# 60-day return for every stock, then sort high to low.
momentum = panel.pct_change(60).iloc[-1].sort_values(ascending=False) * 100

print("60-day return ranking (%):")
print(momentum.round(2).to_string())

top_n = 3
winners = list(momentum.head(top_n).index)
print(f"\nGo long the top {top_n}: {winners}")
print("'Cross-sectional' means we compare stocks AGAINST EACH OTHER, then hold the strongest.")

# ATR (Average True Range): a clean dollar measure of how much a stock moves per day.
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)

# ATR averages the true range (today's range, accounting for gaps) over 14 days.
df["atr"] = ta.atr(df["high"], df["low"], df["close"], period=14)
df["atr_pct"] = df["atr"] / df["close"] * 100

print(df[["close", "atr", "atr_pct"]].tail(5).round(2))
last = df.iloc[-1]
print(f"\nRELIANCE typically moves about Rs {last['atr']:.1f} a day ({last['atr_pct']:.2f}% of price).")
print("ATR is the building block for both breakout levels and position sizing.")

# ATR breakout entry: buy when price closes above yesterday's close + N x ATR.
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="ICICIBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

df["atr"] = ta.atr(df["high"], df["low"], df["close"], period=14)

# The breakout level sits one ATR above the PREVIOUS close. .shift(1) avoids look-ahead:
# we only use information that was known before today's bar formed.
mult = 1.0
df["trigger"] = df["close"].shift(1) + mult * df["atr"].shift(1)
df["breakout"] = df["close"] > df["trigger"]

signals = df[df["breakout"]][["close", "trigger"]].dropna()
print(f"Breakout days (close > prev close + {mult} ATR):", len(signals))
print(signals.tail(5).round(2))
print("\nSizing the trigger in ATRs adapts the strategy to each stock's own volatility.")

# Bollinger squeeze: spot low-volatility coils, then trade the expansion.
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)

# Bandwidth = how wide the Bollinger Bands are. Low bandwidth = a quiet "squeeze";
# the band then expands when a move finally breaks out.
df["bbwidth"] = ta.bbwidth(df["close"], period=20, std_dev=2.0)

# A squeeze: bandwidth in the bottom 20% of its recent range.
floor = df["bbwidth"].rolling(60).quantile(0.20)
df["squeeze"] = df["bbwidth"] < floor

print(df[["close", "bbwidth", "squeeze"]].dropna().tail(6).round(4))
print("\nSqueeze days in the window:", int(df["squeeze"].sum()))
print("Quiet bands tend to precede big moves -- arm a breakout order when a squeeze appears.")

# Regime filter: only take long trades when price is above its 200-day SMA.
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=500)).strftime("%Y-%m-%d")
df = client.history(symbol="HDFCBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

# The 200-day SMA is the classic line between a bull and a bear regime.
df["sma200"] = ta.sma(df["close"], 200)
df["bull"] = df["close"] > df["sma200"]

last = df.dropna().iloc[-1]
in_uptrend = bool(last["bull"])
bull_days = int(df["bull"].sum())

print(f"Close {last['close']:.1f}  vs 200-SMA {last['sma200']:.1f}")
print("Regime now:", "UPTREND -- longs allowed" if in_uptrend else "DOWNTREND -- stand aside")
print(f"Days in uptrend over the window: {bull_days} of {df['bull'].notna().sum()}")
print("\nA regime filter keeps you from buying momentum into a falling market -- the No.1 way these systems blow up.")

# Volatility targeting: size each position by target_vol / realised_vol.
import os
from datetime import datetime, timedelta

import numpy as np

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=200)).strftime("%Y-%m-%d")
df = client.history(symbol="SBIN", exchange="NSE", interval="D", start_date=start, end_date=end)

# Realised vol = annualised std dev of daily returns over 20 days.
daily_ret = df["close"].pct_change()
realised_vol = daily_ret.rolling(20).std() * np.sqrt(252)

# Aim for a steady 15% portfolio volatility. Quiet stock -> bigger position; wild stock -> smaller.
target_vol = 0.15
raw_size = target_vol / realised_vol
position = raw_size.clip(upper=3.0)  # cap leverage at 3x for safety

capital = 200000
last_price = df["close"].iloc[-1]
shares = int((capital * position.iloc[-1]) / last_price)

print(f"Realised volatility: {realised_vol.iloc[-1] * 100:.1f}%   target: {target_vol * 100:.0f}%")
print(f"Volatility-scaled size factor: {position.iloc[-1]:.2f}x")
print(f"At Rs {last_price:.1f}, that is about {shares} shares on Rs {capital:,} capital.")
print("\nVol targeting keeps risk roughly CONSTANT across calm and stormy markets.")

# The same volatility toolkit on a commodity: an ATR breakout on MCX gold.
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=300)).strftime("%Y-%m-%d")
# GOLDM = the mini gold future on MCX, a popular commodity for volatility strategies.
df = client.history(symbol="GOLDM03JUL26FUT", exchange="MCX", interval="D", start_date=start, end_date=end)

df["atr"] = ta.atr(df["high"], df["low"], df["close"], period=14)
df["trigger"] = df["close"].shift(1) + 1.5 * df["atr"].shift(1)
df["breakout"] = df["close"] > df["trigger"]

last = df.iloc[-1]
print(f"GOLDM close {last['close']:.0f}   ATR {last['atr']:.0f}   ({last['atr'] / last['close'] * 100:.2f}% of price)")
print("Breakout days in window:", int(df["breakout"].sum()))
print(df[df["breakout"]][["close", "trigger"]].dropna().tail(4).round(0))
print("\nCommodities like gold and crude carry their own volatility -- ATR adapts automatically.")

# Backtest a regime-filtered momentum trade on the strongest stock with VectorBT.
import os
from datetime import datetime, timedelta

import pandas as pd

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=500)).strftime("%Y-%m-%d")
# ICICIBANK led the 60-day momentum ranking earlier -- backtest a trend trade on it.
df = client.history(symbol="ICICIBANK", exchange="NSE", interval="D", start_date=start, end_date=end)

close = df["close"]
mom = close.pct_change(60)            # positive momentum = an uptrend
sma200 = ta.sma(close, 200)           # regime filter
bull = close > sma200

# Enter long only when BOTH momentum is positive AND we are in a bull regime.
entries = (mom > 0) & bull & ~((mom.shift(1) > 0) & bull.shift(1))
exits = (mom < 0) | ~bull

import vectorbt as vbt

pf = vbt.Portfolio.from_signals(close, entries.fillna(False), exits.fillna(False),
                                init_cash=200000, fees=0.001, freq="1D")
print("Momentum + 200-SMA regime filter on ICICIBANK")
print("Total return :", round(float(pf.total_return()) * 100, 2), "%")
print("Trades       :", int(pf.trades.count()))
print("Max drawdown :", round(float(pf.max_drawdown()) * 100, 2), "%")
print("\nThe regime filter is what stops momentum from buying into a crash.")