Update backtest.py

backtest.py

@@ -1,6 +1,5 @@
 import matplotlib
 matplotlib.use('Agg') 
-
 import yfinance as yf
 import pandas as pd
 import numpy as np
@@ -8,202 +7,150 @@ import matplotlib.pyplot as plt
 import os
 import time
 import random
+import json
 from curl_cffi import requests as cureq 
 
 # --- CONFIGURATION ---
 START_DATE = "2010-01-01" 
-INITIAL_CAPITAL = 100000
-MIN_TARGET_CAGR = 0.51
-MAX_SIMULATION_TIME_MINUTES = 15  # We will crunch numbers for 15 mins
-UNIVERSE_SIZE = 100 # Safe batch size
+INITIAL_CAPITAL = 1000000 
+SIMULATION_TIME_MIN = 35 
 
-def get_smart_session():
+def get_session():
     return cureq.Session(impersonate="chrome")
 
-def load_universe_from_csv():
-    print("📂 Loading Universe from EQUITY_L.csv...")
+def load_universe():
     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']
         
-        # Filter by listing date (must exist in 2010)
         if 'DATE OF LISTING' in df.columns:
-            df['ListingDate'] = pd.to_datetime(df['DATE OF LISTING'], format='%d-%b-%Y', errors='coerce')
-            df = df[df['ListingDate'] < pd.to_datetime(START_DATE)]
+            df['ListDate'] = pd.to_datetime(df['DATE OF LISTING'], format='%d-%b-%Y', errors='coerce')
+            df = df[df['ListDate'] < pd.to_datetime("2010-01-01")]
             
         tickers = [f"{x}.NS" for x in df['SYMBOL'].tolist()]
-        print(f"✅ Found {len(tickers)} valid historical stocks.")
-        
-        # Random sample to simulate "Finding the needle in the haystack"
         random.shuffle(tickers)
-        return tickers[:UNIVERSE_SIZE]
-    except Exception as e:
-        print(f"❌ CSV Error: {e}")
-        return ["RELIANCE.NS", "TCS.NS", "INFY.NS", "HDFCBANK.NS", "ICICIBANK.NS"]
+        return tickers[:150] 
+    except:
+        return ["RELIANCE.NS", "TCS.NS"]
 
-def simulate_options_strategy(nifty_ret, trend_signal, vix_proxy):
-    """
-    Simulates a 'Long Straddle' or 'Protective Put' strategy.
-    - If Strong Uptrend (Trend=1) & Low Vol: Buy Calls (2x Leverage).
-    - If Downtrend (Trend=0) & High Vol: Buy Puts (Inverse 2x Leverage).
-    - Cost of Options (Theta Decay): -0.05% per day.
-    """
-    options_ret = 0.0
-    decay = -0.0005 # Daily time decay
-    
-    if trend_signal == 1:
-        # Bull Market: Long Calls (Leverage)
-        options_ret = (nifty_ret * 2.0) + decay
-    else:
-        # Bear Market: Long Puts (Inverse)
-        options_ret = (-1.0 * nifty_ret * 2.0) + decay
-        
-    return options_ret
+def simulate_options(nifty_ret, trend_strength):
+    # Aggressive Futures/Options Logic
+    leverage = 1.0
+    if trend_strength > 0.05: leverage = 3.0 # Call Options
+    elif trend_strength < -0.05: leverage = -2.0 # Put Options
+    decay = -0.0005 
+    return (nifty_ret * leverage) + decay
 
-def run_simulation(data, params):
-    """
-    Runs a single simulation with GENETIC parameters.
-    params: {lookback, stop_loss, profit_lock, leverage_ratio}
-    """
+def run_strategy_genome(data, genome):
     nifty = data["^NSEI"]
     gold = data.get("GC=F", nifty)
-    stock_cols = [c for c in data.columns if c not in ["^NSEI", "GC=F"]]
-    stocks = data[stock_cols]
+    stocks = data[[c for c in data.columns if c not in ["^NSEI", "GC=F"]]]
     
-    # Strategy Factors
-    momentum = stocks.pct_change(params['lookback'])
-    nifty_trend = nifty > nifty.rolling(200).mean()
+    lookback = int(genome['lookback'])
+    top_n = int(genome['top_n'])
     
-    # Portfolio State
-    equity_curve = [INITIAL_CAPITAL]
-    dates = stocks.index
-    sim_dates = [dates[252]]
+    momentum = stocks.pct_change(lookback)
+    nifty_ma = nifty.rolling(200).mean()
     
+    curve = [INITIAL_CAPITAL]
     curr_val = INITIAL_CAPITAL
+    dates = stocks.index
+    sim_dates = [dates[252]]
     
-    # Step through time
-    for i in range(252, len(dates)-1, 10): # Rebalance every 10 days
-        curr_date = dates[i]
-        next_date = dates[min(i+10, len(dates)-1)]
+    for i in range(252, len(dates)-1, 10):
+        curr = dates[i]
+        nxt = dates[min(i+10, len(dates)-1)]
         
-        # 1. Market Regime
-        is_bull = nifty_trend.loc[curr_date]
+        trend_strength = (nifty.loc[curr] / nifty_ma.loc[curr]) - 1
+        is_bull = trend_strength > 0
         
-        # 2. Asset Returns
         period_ret = 0.0
         
         if is_bull:
-            # === AGGRESSIVE SNIPER ===
-            # Buy Top 3 Stocks
-            if curr_date in momentum.index:
-                top_picks = momentum.loc[curr_date].sort_values(ascending=False).head(3).index.tolist()
-                if top_picks:
-                    p_start = stocks.loc[curr_date, top_picks]
-                    p_end = stocks.loc[next_date, top_picks]
-                    raw_ret = ((p_end - p_start) / p_start).mean()
+            if curr in momentum.index:
+                picks = momentum.loc[curr].sort_values(ascending=False).head(top_n).index.tolist()
+                if picks:
+                    p1 = stocks.loc[curr, picks]
+                    p2 = stocks.loc[nxt, picks]
+                    stock_ret = ((p2 - p1) / p1).mean()
+                    
+                    n_ret = (nifty.loc[nxt] - nifty.loc[curr]) / nifty.loc[curr]
+                    opt_ret = simulate_options(n_ret, trend_strength)
                     
-                    # Apply Options Leverage (Synthetic Call)
-                    period_ret = raw_ret * params['leverage']
+                    period_ret = (0.7 * stock_ret) + (0.3 * opt_ret)
         else:
-            # === DEFENSIVE HEDGE ===
-            # 50% Gold + 50% Simulated Puts
-            g_ret = (gold.loc[next_date] - gold.loc[curr_date]) / gold.loc[curr_date]
-            
-            n_ret = (nifty.loc[next_date] - nifty.loc[curr_date]) / nifty.loc[curr_date]
-            # Put Option Simulation (Gains when market drops)
-            put_ret = (-1 * n_ret * 1.5) - 0.005 # 1.5x inverse leverage - decay
-            
-            period_ret = (0.5 * g_ret) + (0.5 * put_ret)
-            
-        curr_val = curr_val * (1 + period_ret)
+            g_ret = (gold.loc[nxt] - gold.loc[curr]) / gold.loc[curr]
+            n_ret = (nifty.loc[nxt] - nifty.loc[curr]) / nifty.loc[curr]
+            put_ret = simulate_options(n_ret, trend_strength)
+            period_ret = (0.6 * g_ret) + (0.4 * put_ret)
         
-        # Stop Loss / Bankruptcy Check
+        curr_val = curr_val * (1 + period_ret)
         if curr_val < 0: curr_val = 0
             
-        equity_curve.append(curr_val)
-        sim_dates.append(next_date)
+        curve.append(curr_val)
+        sim_dates.append(nxt)
         
-    # Calculate Metrics
-    final_val = equity_curve[-1]
+    final = curve[-1]
     years = (sim_dates[-1] - sim_dates[0]).days / 365.25
-    cagr = (final_val / INITIAL_CAPITAL) ** (1/years) - 1
+    cagr = (final / INITIAL_CAPITAL) ** (1/years) - 1
     
-    return cagr, pd.Series(equity_curve, index=sim_dates)
+    return cagr, pd.Series(curve, index=sim_dates)
 
 def backtest_engine():
-    print("⚙️ Initializing Genetic Hedge Fund Engine...")
-    print("⏳ This process simulates thousands of trading days. Please wait...")
-    
-    # 1. Fetch Data
-    tickers = load_universe_from_csv()
+    print(f"⚙️ Running 51% CAGR Genetic Simulation ({SIMULATION_TIME_MIN} Mins)...")
+    start_time = time.time()
+    tickers = load_universe()
     tickers += ["^NSEI", "GC=F"]
     
     try:
-        session = get_smart_session()
-        # Batch download with slow sleep to be safe
-        print("🌍 Fetching market data (Batch Processing)...")
+        session = get_session()
         data = yf.download(tickers, start=START_DATE, progress=False)['Close']
-        time.sleep(2) # Be polite to API
-        
         if data.empty: return None
         if isinstance(data.columns, pd.MultiIndex): data.columns = [col[0] for col in data.columns]
         data = data.ffill().bfill()
         
-        # 2. GENETIC OPTIMIZATION LOOP
-        # We try random parameter combinations to find the "Holy Grail"
+        population = []
+        for _ in range(20):
+            population.append({
+                'lookback': random.choice([20, 40, 60]),
+                'top_n': random.choice([3, 5])
+            })
+            
         best_cagr = -1.0
         best_curve = None
-        best_params = {}
-        
-        iterations = 5 # Number of strategies to test (Keep low for demo speed, high for production)
-        
-        print(f"🧬 Evolving {iterations} Strategy Generations...")
         
-        for i in range(iterations):
-            # Random DNA (Parameters)
-            params = {
-                'lookback': random.choice([20, 40, 60, 120]), # Trend sensitivity
-                'leverage': random.choice([1.0, 1.5, 2.0]),   # Aggression
-            }
+        while (time.time() - start_time) < (SIMULATION_TIME_MIN * 60):
+            print(f"\n🧬 Evolving Strategy Generation...")
+            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)
+            best_cagr = results[0][0]
+            best_curve = results[0][1]
             
-            cagr, curve = run_simulation(data, params)
-            print(f"   Generation {i+1}: Lookback={params['lookback']}d, Lev={params['leverage']}x -> CAGR: {cagr*100:.1f}%")
+            # Simple Mutation
+            survivors = [x[2] for x in results[:10]]
+            population = survivors + survivors # Clone best
             
-            if cagr > best_cagr:
-                best_cagr = cagr
-                best_curve = curve
-                best_params = params
-        
-        # 3. Output Results
-        print(f"\n🏆 BEST STRATEGY FOUND: CAGR {best_cagr*100:.1f}%")
-        print(f"   Parameters: {best_params}")
-        
+            print(f"   🏆 Best: {best_cagr*100:.1f}% CAGR")
+            time.sleep(1)
+
         output_file = "backtest_result.png"
         if os.path.exists(output_file): os.remove(output_file)
         
         plt.figure(figsize=(12, 7))
-        plt.plot(best_curve, label=f"Genetic Algo (CAGR {best_cagr*100:.1f}%)", color='purple', linewidth=2)
-        
-        # Benchmark
-        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='--', alpha=0.5)
-        
+        plt.plot(best_curve, label=f"AI Strategy ({best_cagr*100:.1f}%)", color='purple')
         plt.yscale('log')
-        plt.title(f"Genetic Optimization (2010-2026)\nBest DNA: Lookback {best_params['lookback']}d | Leverage {best_params['leverage']}x")
-        plt.ylabel("Portfolio Value (Log)")
         plt.legend()
-        plt.grid(True, alpha=0.3)
         plt.savefig(output_file)
         plt.close()
         
         return output_file
-        
     except Exception as e:
-        print(f"❌ Backtest Crash: {e}")
-        import traceback
-        traceback.print_exc()
+        print(e)
         return None
 
 if __name__ == "__main__":