Think like a quant.
Trade like a machine.

import yfinance as yf
import pandas as pd
import numpy as np

# ── STEP 1: GET DATA ───────────────────────────────────
# .NS suffix = NSE (National Stock Exchange India)
# auto_adjust=True → splits and dividends handled automatically
ticker     = 'RELIANCE.NS'
start_date = '2020-01-01'
end_date   = '2024-12-31'

data = yf.download(ticker, start=start_date, end=end_date, auto_adjust=True)
data = data[['Close']].copy()

# ── STEP 2: COMPUTE MOVING AVERAGES ───────────────────
# SMA(10) = average of last 10 days → reacts fast to price changes
# SMA(50) = average of last 50 days → shows the bigger, slower trend
# When fast > slow: uptrend. When fast < slow: downtrend.
short_window = 10
long_window  = 50

data['SMA_short'] = data['Close'].rolling(window=short_window).mean()
data['SMA_long']  = data['Close'].rolling(window=long_window).mean()

# ── STEP 3: GENERATE SIGNALS ──────────────────────────
# Signal = 1 when we SHOULD be in the market, 0 when we shouldn't
# Short MA above long MA = bullish = hold the stock
data['Signal'] = 0
data.loc[data['SMA_short'] > data['SMA_long'], 'Signal'] = 1

# Position = CHANGE in signal
# +1 = just crossed up → BUY today
# -1 = just crossed down → SELL today
#  0 = no change, hold current position
data['Position'] = data['Signal'].diff()

# ── STEP 4: CALCULATE RETURNS ─────────────────────────
# Daily return = today's % price change
# Strategy return = daily return ONLY on days we're in the market
# .shift(1) is CRITICAL: use yesterday's signal for today's return
# (In real life, you see the crossover after market close, 
#  so you can only act the NEXT day's open)
data['Daily_Return']    = data['Close'].pct_change()
data['Strategy_Return'] = data['Daily_Return'] * data['Signal'].shift(1)

# Cumulative returns: how ₹1 grows over time
data['Cum_Market']   = (1 + data['Daily_Return']).cumprod()
data['Cum_Strategy'] = (1 + data['Strategy_Return']).cumprod()

# ── STEP 5: COMPUTE METRICS ───────────────────────────
years = len(data) / 252   # 252 = avg trading days in a year

total_ret  = (data['Cum_Strategy'].iloc[-1] - 1) * 100
cagr       = (data['Cum_Strategy'].iloc[-1] ** (1/years) - 1) * 100
mkt_cagr   = (data['Cum_Market'].iloc[-1]   ** (1/years) - 1) * 100

# Sharpe: annualized excess return / annualized volatility
daily_avg    = data['Strategy_Return'].mean()
daily_std    = data['Strategy_Return'].std()
sharpe       = (daily_avg / daily_std) * (252 ** 0.5) if daily_std != 0 else 0

# Max drawdown: worst peak-to-trough loss
rolling_max  = data['Cum_Strategy'].cummax()
drawdown     = (data['Cum_Strategy'] - rolling_max) / rolling_max
max_dd       = drawdown.min() * 100

# Win rate
in_market    = data[data['Signal'].shift(1) == 1]['Strategy_Return']
win_rate     = (in_market > 0).sum() / len(in_market) * 100 if len(in_market) > 0 else 0
gross_profit = in_market[in_market > 0].sum()
gross_loss   = abs(in_market[in_market < 0].sum())
profit_factor = gross_profit / gross_loss if gross_loss != 0 else float('inf')
num_trades   = (data['Position'] == 1).sum()

print(f"CAGR:          {cagr:+.1f}%  (market: {mkt_cagr:+.1f}%)")
print(f"Total Return:  {total_ret:+.1f}%")
print(f"Sharpe Ratio:  {sharpe:.2f}  (>1 = good)")
print(f"Max Drawdown:  {max_dd:.1f}%")
print(f"Win Rate:      {win_rate:.1f}%")
print(f"Profit Factor: {profit_factor:.2f}")
print(f"Num Trades:    {num_trades}")

# Compute regime indicators first
data['SMA200']   = data['Close'].rolling(200).mean()
data['BB_width']  = (data['BB_upper'] - data['BB_lower']) / data['SMA20']
# (assumes you've computed ADX, BB_upper, BB_lower, SMA20 already)

def get_regime(row):
    if row['ADX_14'] > 25 and row['Close'] > row['SMA200']:
        return 'trending_up'
    elif row['ADX_14'] > 25 and row['Close'] < row['SMA200']:
        return 'trending_down'   # stay flat or short
    elif row['BB_width'] < 0.05:
        return 'squeeze'           # breakout strategy
    else:
        return 'ranging'           # mean reversion strategy

data['regime'] = data.apply(get_regime, axis=1)

# Apply different logic per regime
data['signal'] = 0
data.loc[(data['regime'] == 'trending_up')  & (data['SMA_short'] > data['SMA_long']), 'signal'] = 1
data.loc[(data['regime'] == 'ranging')       & (data['RSI_14'] < 30), 'signal'] = 1

def walk_forward_test(data, is_months=18, oos_months=6):
    """
    Rolls a window: train on is_months, test on oos_months,
    advance by oos_months, repeat. Returns stitched OOS results.
    More rigorous than a single IS/OOS split.
    """
    results = []
    data['date'] = pd.to_datetime(data.index)
    window_start = data['date'].iloc[0]

    while True:
        is_end  = window_start + pd.DateOffset(months=is_months)
        oos_end = is_end       + pd.DateOffset(months=oos_months)
        if oos_end > data['date'].iloc[-1]: break

        is_data  = data[(data['date'] >= window_start) & (data['date'] < is_end)]
        oos_data = data[(data['date'] >= is_end)       & (data['date'] < oos_end)]

        best_params = optimize_strategy(is_data)   # your param search
        oos_result  = run_strategy(oos_data, best_params)
        results.append(oos_result)
        window_start += pd.DateOffset(months=oos_months)

    return pd.concat(results)

import pandas as pd
import numpy as np
from statsmodels.tsa.stattools import coint

def compute_indicators(df):
    """
    Computes moving averages, ATR, RSI, MACD, and Bollinger Bands with standard definitions.
    """
    h = df['High']; l = df['Low']; c = df['Close']
    data = df.copy()

    # Moving Averages
    for n in [10, 20, 50, 200]:
        data[f'SMA_{n}'] = c.rolling(n).mean()
        data[f'EMA_{n}'] = c.ewm(span=n, adjust=False).mean()

    # ATR (Average True Range)
    prev_c = c.shift(1)
    tr     = pd.concat([h-l, (h-prev_c).abs(), (l-prev_c).abs()], axis=1).max(axis=1)
    data['ATR_14'] = tr.ewm(alpha=1/14, adjust=False).mean()

    # RSI (Wilder smoothing via ewm)
    delta = c.diff()
    gain  = delta.clip(lower=0).ewm(alpha=1/14, adjust=False).mean()
    loss  = (-delta.clip(upper=0)).ewm(alpha=1/14, adjust=False).mean()
    data['RSI_14'] = 100 - (100 / (1 + gain / loss))

    # MACD
    ema12 = c.ewm(span=12, adjust=False).mean()
    ema26 = c.ewm(span=26, adjust=False).mean()
    data['MACD']        = ema12 - ema26
    data['MACD_signal'] = data['MACD'].ewm(span=9, adjust=False).mean()
    data['MACD_hist']   = data['MACD'] - data['MACD_signal']

    # Bollinger Bands
    sma20 = c.rolling(20).mean()
    std20 = c.rolling(20).std()
    data['BB_upper']  = sma20 + 2 * std20
    data['BB_middle'] = sma20
    data['BB_lower']  = sma20 - 2 * std20
    data['BB_width']  = (data['BB_upper'] - data['BB_lower']) / sma20

    # ADX
    up   = h - h.shift(1)
    down = l.shift(1) - l
    pdm  = up.where((up > down) & (up > 0), 0)
    ndm  = down.where((down > up) & (down > 0), 0)
    pdi  = 100 * pdm.ewm(alpha=1/14, adjust=False).mean() / data['ATR_14']
    ndi  = 100 * ndm.ewm(alpha=1/14, adjust=False).mean() / data['ATR_14']
    dx   = 100 * (pdi - ndi).abs() / (pdi + ndi)
    data['ADX_14'] = dx.ewm(alpha=1/14, adjust=False).mean()

    return data

def backtest_metrics(returns, risk_free=0.07):
    """
    Computes CAGR, Sharpe Ratio, Sortino Ratio, Calmar, Max Drawdown, and Profit Factor.
    """
    r        = returns.dropna()
    cum      = (1 + r).cumprod()
    n_years  = len(r) / 252
    daily_rf = risk_free / 252

    total_return = cum.iloc[-1] - 1
    cagr         = cum.iloc[-1] ** (1 / n_years) - 1
    vol_ann      = r.std() * np.sqrt(252)

    # Downside deviation (for Sortino)
    downside = r[r < daily_rf]
    sortino_vol = downside.std() * np.sqrt(252) if len(downside) > 0 else 1e-9

    # Drawdown
    rolling_max = cum.cummax()
    drawdown    = (cum - rolling_max) / rolling_max
    max_dd      = drawdown.min()
    dd_dur      = (drawdown < 0).groupby((drawdown < 0 != (drawdown < 0).shift()).cumsum()).sum().max()

    excess = cagr - risk_free
    sharpe  = excess / vol_ann      if vol_ann != 0      else 0
    sortino = excess / sortino_vol  if sortino_vol != 0  else 0
    calmar  = cagr / abs(max_dd)    if max_dd != 0        else 0

    in_mkt       = r[r != 0]
    win_rate     = (in_mkt > 0).mean()
    gross_profit = in_mkt[in_mkt > 0].sum()
    gross_loss   = abs(in_mkt[in_mkt < 0].sum())
    pf           = gross_profit / gross_loss if gross_loss != 0 else np.inf
    expectancy   = win_rate * in_mkt[in_mkt > 0].mean() + (1 - win_rate) * in_mkt[in_mkt < 0].mean()

    return {
        'total_return_pct': round(total_return * 100, 2), 
        'cagr_pct': round(cagr * 100, 2),
        'sharpe': round(sharpe, 3),   
        'sortino': round(sortino, 3), 
        'calmar': round(calmar, 3),
        'max_dd_pct': round(max_dd * 100, 2), 
        'max_dd_days': int(dd_dur),
        'win_rate_pct': round(win_rate * 100, 2), 
        'profit_factor': round(pf, 3),
        'expectancy_pct': round(expectancy * 100, 4)
    }

def pairs_analysis(price_a, price_b, lookback=60):
    """
    Runs OLS regression and tests for Granger Cointegration.
    """
    import statsmodels.api as sm
    score, pvalue, _ = coint(price_a, price_b)
    
    if pvalue > 0.05:
        print("⚠ Not cointegrated — do not trade this pair")
        return None

    result      = sm.OLS(price_a, sm.add_constant(price_b)).fit()
    hedge_ratio = result.params[price_b.name]
    spread      = price_a - hedge_ratio * price_b

    mu     = spread.rolling(lookback).mean()
    sd     = spread.rolling(lookback).std() # standard deviation
    zscore = (spread - mu) / sd

    signal = pd.Series(0, index=zscore.index)
    signal[zscore < -2.0] =  1   # long spread
    signal[zscore >  2.0] = -1   # short spread
    signal[(zscore > -0.5) & (zscore < 0.5)] = 0  # exit zone
    signal = signal.ffill()

    return {'spread': spread, 'zscore': zscore, 'signal': signal, 'hedge_ratio': hedge_ratio}