Technical Indicators in VectorBT

# Custom SuperTrend implementation for Indian markets
def supertrend(high, low, close, period=10, multiplier=3):
    """
    SuperTrend indicator - widely used in Indian markets
    """
    import pandas as pd
    import numpy as np
    
    # Calculate ATR
    hl = high - low
    hc = abs(high - close.shift())
    lc = abs(low - close.shift())
    tr = pd.concat([hl, hc, lc], axis=1).max(axis=1)
    atr = tr.ewm(alpha=1/period, adjust=False).mean()
    
    # Calculate basic bands
    hl_avg = (high + low) / 2
    upper_band = hl_avg + (multiplier * atr)
    lower_band = hl_avg - (multiplier * atr)
    
    # Calculate SuperTrend
    supertrend = pd.Series(index=close.index, dtype='float64')
    direction = pd.Series(index=close.index, dtype='int64')
    
    supertrend.iloc[0] = upper_band.iloc[0]
    direction.iloc[0] = 1
    
    for i in range(1, len(close)):
        if close.iloc[i] <= upper_band.iloc[i]:
            supertrend.iloc[i] = upper_band.iloc[i]
        else:
            supertrend.iloc[i] = lower_band.iloc[i]
            
        if supertrend.iloc[i] == upper_band.iloc[i]:
            if close.iloc[i] <= upper_band.iloc[i]:
                direction.iloc[i] = -1
            else:
                direction.iloc[i] = direction.iloc[i-1]
        else:
            if close.iloc[i] >= lower_band.iloc[i]:
                direction.iloc[i] = 1
            else:
                direction.iloc[i] = direction.iloc[i-1]
    
    return supertrend, direction

# Use on RELIANCE
reliance = vbt.YFData.download("RELIANCE.NS", start="2023-01-01", end="2024-01-01")
high = reliance.get('High')
low = reliance.get('Low')
close = reliance.get('Close')

st, direction = supertrend(high, low, close, period=10, multiplier=3)

# Generate signals
buy_signal = (direction == 1) & (direction.shift() == -1)
sell_signal = (direction == -1) & (direction.shift() == 1)

# Backtest SuperTrend
portfolio = vbt.Portfolio.from_signals(
    close, buy_signal, sell_signal,
    init_cash=10_00_000,
    fees=0.0003
)

print(f"SuperTrend Strategy Return: {portfolio.total_return():.2%}")
print(f"Sharpe Ratio: {portfolio.sharpe_ratio():.2f}")

# MACD - Moving Average Convergence Divergence
macd = vbt.MACD.run(
    close,
    fast_window=12,
    slow_window=26,
    signal_window=9,
    ewm=True
)

# MACD crossover signals
macd_cross_above = macd.macd_crossed_above(macd.signal)
macd_cross_below = macd.macd_crossed_below(macd.signal)

# MACD histogram
histogram = macd.macd - macd.signal

# Zero line crossover
zero_cross_up = macd.macd_crossed_above(0)
zero_cross_down = macd.macd_crossed_below(0)

# Combine MACD with trend filter
sma_200 = vbt.MA.run(close, window=200)
trend_filter = close > sma_200.ma

# Filtered MACD signals
filtered_buy = macd_cross_above & trend_filter
filtered_sell = macd_cross_below | ~trend_filter

portfolio = vbt.Portfolio.from_signals(
    close, filtered_buy, filtered_sell,
    init_cash=10_00_000,
    fees=0.0003
)

print(f"MACD Strategy Return: {portfolio.total_return():.2%}")

# Stochastic Oscillator
stoch = vbt.STOCH.run(
    high, low, close,
    k_window=14,
    d_window=3,
    smooth_k=3
)

# Stochastic signals
stoch_oversold = (stoch.k < 20) & (stoch.k_crossed_above(stoch.d))
stoch_overbought = (stoch.k > 80) & (stoch.k_crossed_below(stoch.d))

# Slow Stochastic
slow_stoch = vbt.STOCH.run(
    high, low, close,
    k_window=14,
    d_window=3,
    smooth_k=3,
    smooth_d=3
)

print(f"Stochastic K: {stoch.k.iloc[-1]:.2f}")
print(f"Stochastic D: {stoch.d.iloc[-1]:.2f}")

# Bollinger Bands
bb = vbt.BBANDS.run(close, window=20, stdev=2)

# Bollinger Band strategies
# 1. Mean Reversion
bb_buy = close <= bb.lower_band
bb_sell = close >= bb.upper_band

# 2. Breakout
bb_breakout_up = close > bb.upper_band
bb_breakout_down = close < bb.lower_band

# 3. Squeeze Detection (low volatility periods)
bandwidth = (bb.upper_band - bb.lower_band) / bb.middle_band
squeeze = bandwidth < bandwidth.rolling(100).quantile(0.2)

# 4. Walking the Bands
walking_upper = (close > bb.middle_band) & (close.shift(1) > bb.middle_band.shift(1))
walking_lower = (close < bb.middle_band) & (close.shift(1) < bb.middle_band.shift(1))

# Backtest BB mean reversion on Bank NIFTY
banknifty = vbt.YFData.download("^NSEBANK", start="2023-01-01", end="2024-01-01")
bn_close = banknifty.get('Close')

bn_bb = vbt.BBANDS.run(bn_close, window=20, stdev=2)
bn_entries = bn_close <= bn_bb.lower_band
bn_exits = bn_close >= bn_bb.upper_band

portfolio = vbt.Portfolio.from_signals(
    bn_close, bn_entries, bn_exits,
    init_cash=10_00_000,
    fees=0.0003
)

print(f"Bank NIFTY BB Strategy Return: {portfolio.total_return():.2%}")

# ATR for volatility-based position sizing
atr = vbt.ATR.run(high, low, close, window=14)

# Volatility-adjusted position sizing
capital = 10_00_000
risk_per_trade = 0.02  # 2% risk per trade
stop_loss_atr = 2  # Stop loss at 2 ATR

position_size = (capital * risk_per_trade) / (atr.atr * stop_loss_atr)

# ATR-based trailing stop
def atr_trailing_stop(close, atr, multiplier=2):
    """Calculate ATR-based trailing stop"""
    stop = pd.Series(index=close.index, dtype='float64')
    stop.iloc[0] = close.iloc[0] - (atr.iloc[0] * multiplier)
    
    for i in range(1, len(close)):
        stop.iloc[i] = max(
            stop.iloc[i-1],
            close.iloc[i] - (atr.iloc[i] * multiplier)
        )
    
    return stop

trailing_stop = atr_trailing_stop(close, atr.atr, multiplier=2)

# Keltner Channels (MA ± ATR)
ma_20 = vbt.MA.run(close, window=20)
kc_upper = ma_20.ma + (atr.atr * 2)
kc_lower = ma_20.ma - (atr.atr * 2)

print(f"Current ATR: {atr.atr.iloc[-1]:.2f}")
print(f"Suggested Position Size: {position_size.iloc[-1]:.0f} shares")

# Get volume data
volume = data.get('Volume')

# On-Balance Volume
obv = vbt.OBV.run(close, volume)

# OBV divergence
obv_ma = vbt.MA.run(obv.obv, window=20)
obv_trend_up = obv.obv > obv_ma.ma
obv_trend_down = obv.obv < obv_ma.ma

# Volume-Price Trend (VPT)
vpt = vbt.indicators.factory(
    class_name='VPT',
    input_names=['close', 'volume'],
    param_names=[],
    output_names=['vpt']
).from_apply_func(
    lambda close, volume: (
        ((close - close.shift(1)) / close.shift(1)) * volume
    ).cumsum()
)

# Money Flow Index (MFI) - Volume-weighted RSI
def calculate_mfi(high, low, close, volume, period=14):
    """Calculate Money Flow Index"""
    typical_price = (high + low + close) / 3
    money_flow = typical_price * volume
    
    positive_flow = money_flow.where(typical_price > typical_price.shift(1), 0)
    negative_flow = money_flow.where(typical_price < typical_price.shift(1), 0)
    
    positive_flow_sum = positive_flow.rolling(window=period).sum()
    negative_flow_sum = negative_flow.rolling(window=period).sum()
    
    money_ratio = positive_flow_sum / negative_flow_sum
    mfi = 100 - (100 / (1 + money_ratio))
    
    return mfi

mfi = calculate_mfi(high, low, close, volume)

# Volume breakout detection
avg_volume = volume.rolling(window=20).mean()
volume_spike = volume > (avg_volume * 2)  # 2x average volume

# Combine price breakout with volume confirmation
price_breakout = close > close.rolling(window=20).max().shift(1)
confirmed_breakout = price_breakout & volume_spike

print(f"Volume spikes detected: {volume_spike.sum()}")
print(f"Confirmed breakouts: {confirmed_breakout.sum()}")

# VWAP - Critical for Indian intraday traders
def calculate_vwap(high, low, close, volume):
    """Calculate VWAP for the day"""
    typical_price = (high + low + close) / 3
    cumulative_tpv = (typical_price * volume).cumsum()
    cumulative_volume = volume.cumsum()
    vwap = cumulative_tpv / cumulative_volume
    return vwap

# For intraday data (reset daily)
def calculate_daily_vwap(high, low, close, volume):
    """Calculate VWAP that resets each day"""
    df = pd.DataFrame({
        'tp': (high + low + close) / 3,
        'volume': volume
    })
    df['date'] = df.index.date
    
    vwap = df.groupby('date').apply(
        lambda x: ((x['tp'] * x['volume']).cumsum() / x['volume'].cumsum())
    ).reset_index(level=0, drop=True)
    
    return vwap

vwap = calculate_vwap(high, low, close, volume)

# VWAP bands
vwap_upper = vwap * 1.01  # 1% above VWAP
vwap_lower = vwap * 0.99  # 1% below VWAP

# VWAP trading strategy
vwap_buy = (close < vwap) & (close.shift(1) >= vwap.shift(1))
vwap_sell = (close > vwap) & (close.shift(1) <= vwap.shift(1))

# Anchored VWAP from specific date
anchor_date = '2023-06-01'
anchored_vwap = calculate_vwap(
    high[anchor_date:],
    low[anchor_date:],
    close[anchor_date:],
    volume[anchor_date:]
)

print(f"Current VWAP: {vwap.iloc[-1]:.2f}")
print(f"Price vs VWAP: {((close.iloc[-1] / vwap.iloc[-1]) - 1) * 100:.2f}%")

# Pivot Points - Widely used in Indian markets
def calculate_pivot_points(high, low, close, method='classic'):
    """
    Calculate various pivot point types
    Methods: classic, fibonacci, camarilla, woodie
    """
    # Previous day's data
    prev_high = high.shift(1)
    prev_low = low.shift(1)
    prev_close = close.shift(1)
    
    if method == 'classic':
        pp = (prev_high + prev_low + prev_close) / 3
        r1 = 2 * pp - prev_low
        r2 = pp + (prev_high - prev_low)
        r3 = prev_high + 2 * (pp - prev_low)
        s1 = 2 * pp - prev_high
        s2 = pp - (prev_high - prev_low)
        s3 = prev_low - 2 * (prev_high - pp)
        
    elif method == 'camarilla':
        pp = (prev_high + prev_low + prev_close) / 3
        range_hl = prev_high - prev_low
        r1 = prev_close + range_hl * 1.1 / 12
        r2 = prev_close + range_hl * 1.1 / 6
        r3 = prev_close + range_hl * 1.1 / 4
        r4 = prev_close + range_hl * 1.1 / 2
        s1 = prev_close - range_hl * 1.1 / 12
        s2 = prev_close - range_hl * 1.1 / 6
        s3 = prev_close - range_hl * 1.1 / 4
        s4 = prev_close - range_hl * 1.1 / 2
        
        return pp, r1, r2, r3, r4, s1, s2, s3, s4
    
    elif method == 'fibonacci':
        pp = (prev_high + prev_low + prev_close) / 3
        range_hl = prev_high - prev_low
        r1 = pp + 0.382 * range_hl
        r2 = pp + 0.618 * range_hl
        r3 = pp + range_hl
        s1 = pp - 0.382 * range_hl
        s2 = pp - 0.618 * range_hl
        s3 = pp - range_hl
    
    return pp, r1, r2, r3, s1, s2, s3

# Calculate for NIFTY
pp, r1, r2, r3, s1, s2, s3 = calculate_pivot_points(high, low, close)

# Pivot point strategy
pivot_buy = (close > pp) & (close.shift(1) <= pp.shift(1))
pivot_sell = (close < pp) & (close.shift(1) >= pp.shift(1))

# Support/Resistance bounce
support_bounce = (low <= s1) & (close > s1)
resistance_reject = (high >= r1) & (close < r1)

print(f"Current Pivot: {pp.iloc[-1]:.2f}")
print(f"R1: {r1.iloc[-1]:.2f}, S1: {s1.iloc[-1]:.2f}")

# Central Pivot Range - Popular among Indian traders
def calculate_cpr(high, low, close):
    """Calculate Central Pivot Range"""
    # Previous day's data
    prev_high = high.shift(1)
    prev_low = low.shift(1)
    prev_close = close.shift(1)
    
    # Pivot calculation
    pivot = (prev_high + prev_low + prev_close) / 3
    
    # Bottom Central Pivot
    bc = (prev_high + prev_low) / 2
    
    # Top Central Pivot
    tc = (pivot - bc) + pivot
    
    # CPR width
    cpr_width = abs(tc - bc)
    
    # Narrow CPR indicates trending day
    avg_width = cpr_width.rolling(window=20).mean()
    narrow_cpr = cpr_width < (avg_width * 0.5)
    
    return pivot, tc, bc, cpr_width, narrow_cpr

pivot, tc, bc, cpr_width, narrow_cpr = calculate_cpr(high, low, close)

# CPR breakout strategy
cpr_breakout_up = (close > tc) & (close.shift(1) <= tc.shift(1))
cpr_breakout_down = (close < bc) & (close.shift(1) >= bc.shift(1))

# Virgin CPR (untested CPR)
virgin_cpr_up = (low > tc) & (low.shift(1) <= tc.shift(1))
virgin_cpr_down = (high < bc) & (high.shift(1) >= bc.shift(1))

print(f"CPR Width: {cpr_width.iloc[-1]:.2f}")
print(f"Narrow CPR Day: {narrow_cpr.iloc[-1]}")

# Open Interest analysis for F&O trading
def analyze_open_interest(symbol, expiry_date):
    """
    Analyze Open Interest for options trading
    Requires NSEPy or custom data source
    """
    from nsepy import get_history
    from datetime import date
    
    # Get F&O data
    fo_data = get_history(
        symbol=symbol,
        start=date(2023, 1, 1),
        end=date(2024, 1, 1),
        futures=True,
        expiry_date=expiry_date
    )
    
    # Calculate OI change
    oi_change = fo_data['Open Interest'].diff()
    price_change = fo_data['Close'].pct_change()
    
    # Interpret OI with price
    long_buildup = (oi_change > 0) & (price_change > 0)  # Bullish
    short_buildup = (oi_change > 0) & (price_change < 0)  # Bearish
    long_unwinding = (oi_change < 0) & (price_change < 0)  # Bearish
    short_covering = (oi_change < 0) & (price_change > 0)  # Bullish
    
    return {
        'long_buildup': long_buildup.sum(),
        'short_buildup': short_buildup.sum(),
        'long_unwinding': long_unwinding.sum(),
        'short_covering': short_covering.sum()
    }

# PCR (Put-Call Ratio) calculation
def calculate_pcr(put_oi, call_oi):
    """Calculate Put-Call Ratio"""
    pcr = put_oi / call_oi
    
    # PCR interpretation
    # > 1.3: Oversold (Bullish)
    # < 0.7: Overbought (Bearish)
    # 0.7-1.3: Neutral
    
    signal = pd.Series(index=pcr.index, dtype='object')
    signal[pcr > 1.3] = 'Bullish'
    signal[pcr < 0.7] = 'Bearish'
    signal[(pcr >= 0.7) & (pcr <= 1.3)] = 'Neutral'
    
    return pcr, signal

# Create custom indicator using VectorBT factory
from vectorbt.indicators.factory import IndicatorFactory

# 1. Simple custom indicator - Rate of Change
ROC = IndicatorFactory(
    class_name='ROC',
    short_name='roc',
    input_names=['close'],
    param_names=['window'],
    output_names=['roc']
).from_apply_func(
    lambda close, window: (close / close.shift(window) - 1) * 100,
    window=10
)

# Use the custom indicator
roc = ROC.run(close, window=[5, 10, 20])
print(f"ROC values: {roc.roc.tail()}")

# 2. Complex custom indicator - Trend Strength
def trend_strength(close, fast_period=10, slow_period=30):
    """
    Custom trend strength indicator
    Combines multiple factors
    """
    # Calculate components
    fast_ma = close.rolling(window=fast_period).mean()
    slow_ma = close.rolling(window=slow_period).mean()
    
    # Trend direction
    trend = (fast_ma > slow_ma).astype(int) * 2 - 1  # 1 for up, -1 for down
    
    # Trend strength (distance between MAs)
    strength = abs(fast_ma - slow_ma) / slow_ma * 100
    
    # Momentum
    momentum = close.pct_change(periods=fast_period) * 100
    
    # Combined score
    score = trend * strength * (1 + abs(momentum) / 100)
    
    return score, trend, strength

TrendStrength = IndicatorFactory(
    class_name='TrendStrength',
    short_name='ts',
    input_names=['close'],
    param_names=['fast_period', 'slow_period'],
    output_names=['score', 'trend', 'strength']
).from_apply_func(
    trend_strength,
    fast_period=10,
    slow_period=30
)

# 3. Indian market specific - Delivery Percentage Indicator
def delivery_volume_indicator(volume, delivery_volume):
    """
    High delivery % indicates genuine interest
    """
    delivery_pct = (delivery_volume / volume) * 100
    
    # Signal generation
    high_delivery = delivery_pct > 60  # Above 60% is significant
    low_delivery = delivery_pct < 30   # Below 30% is speculative
    
    # Moving average of delivery %
    delivery_ma = delivery_pct.rolling(window=5).mean()
    
    # Trend in delivery
    increasing_delivery = delivery_ma > delivery_ma.shift(5)
    
    return delivery_pct, high_delivery, increasing_delivery

# 4. Custom composite indicator
def custom_composite(close, volume, high, low):
    """
    Combine multiple indicators into one signal
    """
    # Components
    rsi = vbt.RSI.run(close, window=14).rsi
    macd = vbt.MACD.run(close).macd
    bb = vbt.BBANDS.run(close)
    atr = vbt.ATR.run(high, low, close).atr
    
    # Normalize components (0 to 1)
    rsi_norm = rsi / 100
    macd_norm = (macd - macd.min()) / (macd.max() - macd.min())
    bb_pos = (close - bb.lower_band) / (bb.upper_band - bb.lower_band)
    atr_norm = (atr - atr.min()) / (atr.max() - atr.min())
    
    # Weighted composite
    weights = [0.3, 0.3, 0.2, 0.2]  # RSI, MACD, BB, ATR
    composite = (
        rsi_norm * weights[0] +
        macd_norm * weights[1] +
        bb_pos * weights[2] +
        (1 - atr_norm) * weights[3]  # Lower volatility is better
    )
    
    return composite

# Strategy 1: Triple Screen Trading System
def triple_screen_strategy(data):
    """
    Elder's Triple Screen Trading System
    Screen 1: Weekly trend (MACD)
    Screen 2: Daily oscillator (RSI)
    Screen 3: Intraday breakout
    """
    close = data.get('Close')
    high = data.get('High')
    low = data.get('Low')
    
    # Screen 1: Weekly MACD (resample to weekly)
    weekly_close = close.resample('W').last()
    weekly_macd = vbt.MACD.run(weekly_close)
    weekly_trend = weekly_macd.macd > weekly_macd.signal
    
    # Align weekly to daily
    weekly_trend_daily = weekly_trend.reindex(close.index, method='ffill')
    
    # Screen 2: Daily RSI
    daily_rsi = vbt.RSI.run(close, window=14)
    oversold = daily_rsi.rsi < 30
    overbought = daily_rsi.rsi > 70
    
    # Screen 3: Breakout
    breakout_up = close > high.shift(1)
    breakout_down = close < low.shift(1)
    
    # Combine signals
    buy_signal = weekly_trend_daily & oversold & breakout_up
    sell_signal = ~weekly_trend_daily & overbought & breakout_down
    
    return buy_signal, sell_signal

# Strategy 2: Confluence Strategy
def confluence_strategy(data, min_confirmations=3):
    """
    Enter only when multiple indicators agree
    """
    close = data.get('Close')
    high = data.get('High')
    low = data.get('Low')
    volume = data.get('Volume')
    
    # Calculate all indicators
    sma_50 = vbt.MA.run(close, window=50)
    sma_200 = vbt.MA.run(close, window=200)
    rsi = vbt.RSI.run(close, window=14)
    macd = vbt.MACD.run(close)
    bb = vbt.BBANDS.run(close)
    obv = vbt.OBV.run(close, volume)
    
    # Bullish conditions
    bull_conditions = pd.DataFrame({
        'trend': close > sma_200.ma,
        'ma_cross': sma_50.ma > sma_200.ma,
        'rsi': (rsi.rsi > 50) & (rsi.rsi < 70),
        'macd': macd.macd > macd.signal,
        'bb': close > bb.middle_band,
        'obv': obv.obv > obv.obv.rolling(20).mean()
    })
    
    # Count confirmations
    bull_score = bull_conditions.sum(axis=1)
    
    # Generate signals
    strong_buy = bull_score >= min_confirmations
    strong_sell = bull_score <= (6 - min_confirmations)
    
    return strong_buy, strong_sell, bull_score

# Strategy 3: Indian Market Combo
def indian_market_strategy(data):
    """
    Combine SuperTrend, VWAP, and CPR
    Popular combination in Indian markets
    """
    close = data.get('Close')
    high = data.get('High')
    low = data.get('Low')
    volume = data.get('Volume')
    
    # SuperTrend
    st, direction = supertrend(high, low, close, period=10, multiplier=3)
    
    # VWAP
    vwap = calculate_vwap(high, low, close, volume)
    
    # CPR
    pivot, tc, bc, cpr_width, narrow_cpr = calculate_cpr(high, low, close)
    
    # Combine conditions
    buy_conditions = (
        (direction == 1) &  # SuperTrend bullish
        (close > vwap) &    # Price above VWAP
        (close > pivot) &   # Price above pivot
        narrow_cpr          # Narrow CPR (trending day)
    )
    
    sell_conditions = (
        (direction == -1) &  # SuperTrend bearish
        (close < vwap) &     # Price below VWAP
        (close < pivot)      # Price below pivot
    )
    
    return buy_conditions, sell_conditions

# Backtest the confluence strategy
data = vbt.YFData.download("RELIANCE.NS", start="2023-01-01", end="2024-01-01")
buy_signals, sell_signals, score = confluence_strategy(data, min_confirmations=4)

portfolio = vbt.Portfolio.from_signals(
    data.get('Close'),
    buy_signals,
    sell_signals,
    init_cash=10_00_000,
    fees=0.0003
)

print(f"Confluence Strategy Return: {portfolio.total_return():.2%}")
print(f"Number of trades: {portfolio.trades.count()}")
print(f"Win rate: {portfolio.trades.win_rate():.2%}")

# Optimize indicator parameters
import numpy as np

# Test multiple parameter combinations
rsi_windows = np.arange(10, 30, 2)
rsi_oversold = np.arange(20, 40, 5)
rsi_overbought = np.arange(60, 80, 5)

# Calculate RSI for all windows
rsi_multi = vbt.RSI.run(close, window=rsi_windows)

# Test all combinations
results = []

for oversold in rsi_oversold:
    for overbought in rsi_overbought:
        if overbought <= oversold:
            continue
            
        entries = rsi_multi.rsi_crossed_below(oversold)
        exits = rsi_multi.rsi_crossed_above(overbought)
        
        pf = vbt.Portfolio.from_signals(
            close, entries, exits,
            init_cash=10_00_000,
            fees=0.0003
        )
        
        # Store results
        for window in rsi_windows:
            results.append({
                'window': window,
                'oversold': oversold,
                'overbought': overbought,
                'return': pf[window].total_return(),
                'sharpe': pf[window].sharpe_ratio(),
                'trades': pf[window].trades.count()
            })

# Find best parameters
results_df = pd.DataFrame(results)
best_by_return = results_df.nlargest(5, 'return')
best_by_sharpe = results_df.nlargest(5, 'sharpe')

print("Best by Return:")
print(best_by_return[['window', 'oversold', 'overbought', 'return']])
print("\nBest by Sharpe:")
print(best_by_sharpe[['window', 'oversold', 'overbought', 'sharpe']])