backtest.py

import warnings
warnings.filterwarnings("ignore")

import matplotlib
matplotlib.use('Agg') 
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import os
import time
import random
import json

# Silence Pandas Future Warnings
pd.options.mode.chained_assignment = None
try: pd.set_option('future.no_silent_downcasting', True)
except: pass

# --- CONFIGURATION ---
START_DATE = "2010-01-01" 
INITIAL_CAPITAL = 1000000 
SIMULATION_TIME_MIN = 35 
CACHE_FILE = "fundamental_cache.json"

def get_all_csv_tickers():
    try:
        df = pd.read_csv("EQUITY_L.csv")
        df.columns = [c.strip() for c in df.columns]
        if 'SERIES' in df.columns: df = df[df['SERIES'] == 'EQ']
        tickers = [f"{x}.NS" for x in df['SYMBOL'].tolist()]
        return tickers
    except:
        return ["RELIANCE.NS", "TCS.NS", "INFY.NS", "HDFCBANK.NS"]

def fundamental_deep_scan(tickers):
    print(f"🔍 PHASE 1: Deep Fundamental Scan of {len(tickers)} stocks...")
    print("⏳ This will take 30-45 minutes. It will only happen ONCE and save to cache.")
    
    scored_stocks = []
    count = 0
    
    for ticker in tickers:
        count += 1
        if count % 50 == 0:
            print(f"   -> Scanned {count}/{len(tickers)} stocks...")
            
        try:
            stock = yf.Ticker(ticker)
            info = stock.info
            
            roe = info.get('returnOnEquity', 0) or 0
            pe = info.get('trailingPE', 0) or 1000
            growth = info.get('revenueGrowth', 0) or 0
            
            score = 0
            if roe > 0.15: score += 40
            if growth > 0.10: score += 30
            if 0 < pe < 60: score += 30
            
            # Keep companies with strong actual business fundamentals
            if score >= 40:
                scored_stocks.append({'ticker': ticker, 'score': score})
                
        except Exception:
            pass 
            
        # Delay to prevent IP Ban from Yahoo Finance
        time.sleep(random.uniform(0.1, 0.4))
        
    scored_stocks.sort(key=lambda x: x['score'], reverse=True)
    
    # We take the top 250 fundamentally strongest companies
    elite_tickers = [x['ticker'] for x in scored_stocks[:250]]
    
    with open(CACHE_FILE, 'w') as f:
        json.dump(elite_tickers, f)
        
    print(f"✅ Phase 1 Complete. Saved {len(elite_tickers)} Elite Stocks to cache.")
    return elite_tickers

def load_fundamental_universe():
    if os.path.exists(CACHE_FILE):
        print("📂 Loading Fundamentally Strong Universe from Cache...")
        with open(CACHE_FILE, 'r') as f:
            return json.load(f)
    else:
        all_tickers = get_all_csv_tickers()
        return fundamental_deep_scan(all_tickers)

def run_strategy_genome(data, genome):
    if data.empty: return -1.0, []
    
    nifty = data.get("^NSEI")
    gold = data.get("GC=F")
    if nifty is None or gold is None: return -1.0, []

    stock_cols = [c for c in data.columns if c not in ["^NSEI", "GC=F"]]
    stocks = data[stock_cols]
    
    lookback = int(genome['lookback'])
    top_n = int(genome['top_n'])
    rebalance_days = int(genome['rebalance'])
    stop_loss = float(genome['stop_loss'])
    trend_filter = int(genome['trend_filter']) 
    max_vol = float(genome['max_vol']) 
    
    momentum = stocks.pct_change(lookback)
    daily_returns = stocks.pct_change(1)
    volatility = daily_returns.rolling(lookback).std() * np.sqrt(252) 
    nifty_ma = nifty.rolling(trend_filter).mean()
    
    curve = [INITIAL_CAPITAL]
    curr_val = INITIAL_CAPITAL
    dates = stocks.index
    
    if len(dates) < 260: return -0.5, []
    sim_dates = [dates[252]]
    
    for i in range(252, len(dates)-1, rebalance_days):
        curr = dates[i]
        nxt = dates[min(i+rebalance_days, len(dates)-1)]
        
        try:
            is_bull = nifty.loc[curr] > nifty_ma.loc[curr]
            period_ret = 0.0
            
            if is_bull:
                if curr in momentum.index and curr in volatility.index:
                    scores = momentum.loc[curr]
                    vols = volatility.loc[curr]
                    
                    # MICRO-CAP UNLOCKED: Lowered to ₹10 to catch real fundamental turnarounds.
                    # (Since Phase 1 ensures they have >15% ROE, a ₹12 stock here is a true hidden gem, not a scam)
                    valid_prices = stocks.loc[curr] > 10.0
                    
                    # Must be fundamentally strong (by universe), positive momentum, low vol, NOT a fractional penny stock
                    valid_stocks = scores[(scores > 0) & (vols < max_vol) & valid_prices]
                    picks = valid_stocks.sort_values(ascending=False).head(top_n).index.tolist()
                    
                    if len(picks) > 0:
                        p1 = stocks.loc[curr, picks]
                        p2 = stocks.loc[nxt, picks]
                        stock_ret = ((p2 - p1) / p1).mean()
                        if pd.isna(stock_ret): stock_ret = 0.0
                        period_ret = stock_ret 
            else:
                g_ret = (gold.loc[nxt] - gold.loc[curr]) / gold.loc[curr]
                if pd.isna(g_ret): g_ret = 0.0
                period_ret = g_ret
            
            if period_ret < -stop_loss: period_ret = -stop_loss
                
            curr_val = curr_val * (1 + period_ret)
            if curr_val < 0: curr_val = 0
                
            curve.append(curr_val)
            sim_dates.append(nxt)
        except:
            continue
            
    if not curve: return -1.0, []
    
    final = curve[-1]
    years = (sim_dates[-1] - sim_dates[0]).days / 365.25
    cagr = (final / INITIAL_CAPITAL) ** (1/years) - 1 if years > 0 else 0
    
    return cagr, pd.Series(curve, index=sim_dates)

def backtest_engine():
    print(f"⚙️ Initializing Phase 2: AI Genetic Backtest...")
    start_time = time.time()
    
    tickers = load_fundamental_universe()
    tickers += ["^NSEI", "GC=F"]
    
    try:
        print(f"🌍 Fetching 16-Year History for Elite Universe...")
        data = yf.download(tickers, start=START_DATE, progress=False, threads=True)
        if isinstance(data.columns, pd.MultiIndex):
            try: data = data['Close']
            except: pass 
            
        data = data.ffill().bfill().infer_objects(copy=False) 
        if data.empty: return None

        population = []
        for _ in range(30):
            population.append({
                'lookback': random.choice([10, 20, 30, 45, 60, 90]), 
                'top_n': random.choice([5, 6, 7, 8, 9, 10]), 
                'rebalance': random.choice([3, 5, 7, 10, 14]), 
                'stop_loss': random.choice([0.02, 0.04, 0.06, 0.08]), 
                'trend_filter': random.choice([30, 50, 100, 200]),
                'max_vol': random.choice([0.30, 0.40, 0.50, 0.60, 0.80]) 
            })
            
        best_cagr = -1.0
        best_curve = None
        stall_count = 0
        generation = 1
        
        while (time.time() - start_time) < (SIMULATION_TIME_MIN * 60):
            print(f"\n🧬 Gen {generation}: Testing 1.0x Portfolios (Strict Fundamentals + Price > ₹10)")
            results = []
            
            for genome in population:
                cagr, curve = run_strategy_genome(data, genome)
                results.append((cagr, curve, genome))
                
            results.sort(key=lambda x: x[0], reverse=True)
            
            if results:
                current_top_cagr = results[0][0]
                
                if current_top_cagr > best_cagr + 0.001: 
                    best_cagr = current_top_cagr
                    best_curve = results[0][1]
                    best_dna = results[0][2]
                    stall_count = 0
                else:
                    stall_count += 1
                
                print(f"   🏆 16-Year Average CAGR: {best_cagr*100:.1f}%")
                print(f"   🧬 DNA: {best_dna['top_n']} Stocks | Bal: {best_dna['rebalance']}d | Regime: {best_dna['trend_filter']}d | Vol Cap: {best_dna['max_vol']*100}%")
                
                survivors = [x[2] for x in results[:6]] 
                new_pop = list(survivors) 
                
                while len(new_pop) < 30:
                    p1 = random.choice(survivors)
                    p2 = random.choice(survivors)
                    
                    child = {
                        'lookback': p1['lookback'] if random.random() > 0.5 else p2['lookback'],
                        'top_n': p1['top_n'] if random.random() > 0.5 else p2['top_n'],
                        'rebalance': p1['rebalance'] if random.random() > 0.5 else p2['rebalance'],
                        'stop_loss': p1['stop_loss'] if random.random() > 0.5 else p2['stop_loss'],
                        'trend_filter': p1['trend_filter'] if random.random() > 0.5 else p2['trend_filter'],
                        'max_vol': p1['max_vol'] if random.random() > 0.5 else p2['max_vol']
                    }
                    
                    mutation_rate = 0.8 if stall_count >= 3 else 0.3
                    
                    if random.random() < mutation_rate: child['lookback'] = random.choice([10, 20, 30, 45, 60, 90])
                    if random.random() < mutation_rate: child['top_n'] = random.choice([5, 6, 7, 8, 9, 10])
                    if random.random() < mutation_rate: child['rebalance'] = random.choice([3, 5, 7, 10, 14])
                    if random.random() < mutation_rate: child['stop_loss'] = random.choice([0.02, 0.04, 0.06, 0.08])
                    if random.random() < mutation_rate: child['trend_filter'] = random.choice([30, 50, 100, 200])
                    if random.random() < mutation_rate: child['max_vol'] = random.choice([0.30, 0.40, 0.50, 0.60, 0.80])
                    
                    new_pop.append(child)
                
                if stall_count >= 4:
                    stall_count = 0
                    
                population = new_pop
                generation += 1
            
            time.sleep(1)

        output_file = "backtest_result.png"
        if os.path.exists(output_file): os.remove(output_file)
        
        if best_curve is not None:
            plt.figure(figsize=(12, 7))
            plt.plot(best_curve, label=f"Fundamentally Strong Strategy ({best_cagr*100:.1f}%)", color='blue', linewidth=2)
            
            nifty = data["^NSEI"]
            bench = (nifty.loc[best_curve.index] / nifty.loc[best_curve.index[0]]) * INITIAL_CAPITAL
            plt.plot(bench, label="Nifty 50 Index", color='gray', linestyle='--')
            
            plt.yscale('log')
            plt.title("Renaissance Engine: Quality Momentum (Zero-Leverage, 5-10 Stocks, ₹10+ Floor)")
            plt.ylabel("Portfolio Value (Log Scale)")
            plt.legend()
            plt.grid(True, alpha=0.3)
            plt.savefig(output_file)
            plt.close()
            return output_file
            
        return None
        
    except Exception as e:
        print(f"❌ Error: {e}")
        return None

if __name__ == "__main__":
    backtest_engine()