# -*- coding: utf-8 -*- """tbs_straddle_backtest_multi.ipynb Automatically generated by Colab. Original file is located at https://colab.research.google.com/drive/1kjWQHMHnt-6eac_ET5xqVLtEHBId2Zl4 # Nifty Time-Based Short Straddle / Strangle — Complete Multi-Moneyness Backtest This notebook reproduces the **full** analysis set with two added features: * **Interactive moneyness** — choose **ATM / OTM1 / OTM2 / ITM1 / ITM2**. * **Minimum opening-straddle filter** — skip any expiry day whose ATM straddle at the opening time is below a threshold you pick (e.g. 75). **What it contains (each framing threaded with your moneyness + min-straddle choice):** 1. Opening-straddle **trend plot** across the years (to pick the floor). 2. **09:30 entry → 15:30 exit** (single entry): leaderboard, PnL by SL, realized-vs-implied RR, year-by-year tables, annual & quarterly PnL by SL. 3. **15-minute iterative entries → 15:30 exit** (Morning / Mid-day / Afternoon baskets): master leaderboard, average PnL by basket & SL, annual PnL by basket & SL, realized-vs-implied RR by basket. 4. **2-hour window baskets**: master leaderboard, average PnL by basket & SL, realized-vs-implied RR by basket. """ import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns import re from IPython.display import display pd.set_option('display.width', 240) pd.set_option('display.max_columns', 60) # ----------------------------- CONFIG ----------------------------- DATA_PATH = "nifty_minute_chain.parquet" OPEN_TIME = "09:30" # straddle gauge for the min-straddle filter (09:30 -> a 75 floor drops ~24 days) ENTRY_TIME = "09:30" # single-entry default EXIT_TIME = "15:30" # universal square-off FRICTION = 0.001 # 0.1% per leg on entry + exit premium # Strike offsets (points) from the ATM strike, per leg. # CE off >0 = call above spot (OTM) ; CE off <0 = call below spot (ITM) # PE off <0 = put below spot (OTM) ; PE off >0 = put above spot (ITM) MONEYNESS = { 'ATM': {'ce_off': 0, 'pe_off': 0}, 'OTM1': {'ce_off': +50, 'pe_off': -50}, 'OTM2': {'ce_off': +100, 'pe_off': -100}, 'ITM1': {'ce_off': -50, 'pe_off': +50}, 'ITM2': {'ce_off': -100, 'pe_off': +100}, } SL_LEVELS = list(range(10, 101, 10)) + [150, 200, 250, 300] SL_LABELS = [f"{s}%" for s in SL_LEVELS] ANNUALISE = np.sqrt(52) # weekly expiries -> 52 (matches the study methodology) print("Moneyness:", list(MONEYNESS.keys())) print("SL grid :", SL_LEVELS) chain = pd.read_parquet(DATA_PATH) chain['date_str'] = chain['date'].astype(str) chain['hhmm'] = chain['hhmm'].astype(str).str.strip() print(f"Rows: {chain.shape[0]:,} | Cols: {chain.shape[1]} | " f"{chain['date_str'].min()} -> {chain['date_str'].max()} | " f"expiry days: {chain.loc[chain['dte']==0,'date_str'].nunique()}") chain.head(3) """## Step 1 — Opening-straddle trend across the years Plotted **before** the backtest so you can see the volatility regime and choose a sensible minimum-straddle floor (dotted red line = example 75). """ d0 = chain[chain['dte'] == 0].copy() open_strad = (d0[d0['hhmm'] == OPEN_TIME].dropna(subset=['straddle']) .assign(dt=lambda x: pd.to_datetime(x['date_str'])) .sort_values('dt')[['dt','date_str','straddle','atm']].reset_index(drop=True)) open_strad['year'] = open_strad['dt'].dt.year print(f"Expiry days with a valid {OPEN_TIME} straddle: {len(open_strad)}\n") print(open_strad.groupby('year')['straddle'].agg(['count','min','median','mean','max']).round(1)) fig, ax = plt.subplots(figsize=(14, 6)) ax.plot(open_strad['dt'], open_strad['straddle'], marker='o', ms=3, lw=1, color='#16335c', label=f'Opening straddle @ {OPEN_TIME}') ax.plot(open_strad['dt'], open_strad['straddle'].rolling(8, min_periods=1).median(), color='#e07b39', lw=2, label='8-expiry rolling median') for yr, g in open_strad.groupby('year'): m = g['straddle'].mean() ax.hlines(m, g['dt'].min(), g['dt'].max(), color='green', ls='--', lw=1.2) ax.text(g['dt'].iloc[len(g)//2], m + 4, f"{yr} avg {m:.0f}", color='green', fontsize=8, ha='center') ax.axhline(75, color='red', ls=':', lw=1.5, label='Example floor = 75') ax.set_title('Nifty ATM Opening Straddle on Expiry Days'); ax.set_xlabel('Date') ax.set_ylabel('ATM straddle premium (points)'); ax.grid(alpha=0.3); ax.legend() plt.tight_layout(); plt.show() """## Step 2 — Choose configuration (moneyness + minimum straddle) The next cell prompts for the structure and the minimum opening straddle. It falls back to `ATM` / no-filter if run non-interactively; use the commented line to set them by hand. """ def ask_choice(): opts = list(MONEYNESS.keys()) try: raw = input(f"Structure to test {opts}: ").strip().upper() except Exception: raw = "" moneyness = raw if raw in MONEYNESS else "ATM" if raw and raw not in MONEYNESS: print(f" '{raw}' not recognised -> ATM") try: raw2 = input("Minimum opening straddle (e.g. 75, blank = none): ").strip() except Exception: raw2 = "" try: min_strad = float(raw2) if raw2 != "" else 0.0 except ValueError: print(f" '{raw2}' not a number -> no filter"); min_strad = 0.0 return moneyness, min_strad SELECTED_MONEYNESS, MIN_STRADDLE = ask_choice() # Manual override: # SELECTED_MONEYNESS, MIN_STRADDLE = "OTM1", 75 cfg = MONEYNESS[SELECTED_MONEYNESS] elig = int((open_strad['straddle'] >= MIN_STRADDLE).sum()) print(f"\nStructure : {SELECTED_MONEYNESS} (CE {cfg['ce_off']:+d}, PE {cfg['pe_off']:+d})") print(f"Min straddle : {MIN_STRADDLE}") print(f"Eligible days : {elig}/{len(open_strad)} (straddle >= {MIN_STRADDLE} at {OPEN_TIME})") """## Step 3 — Build per-day minute panel (`grouped`) Moneyness-independent: change the structure/threshold in Step 2 and just re-run the backtests; no need to rebuild this. """ work = chain[chain['dte'] == 0].copy() dates_with_open = work[work['hhmm'] == OPEN_TIME]['date_str'].unique() work = work[work['date_str'].isin(dates_with_open)].copy() work = work.dropna(subset=['atm', 'straddle']) work = work[(work['hhmm'] >= '09:15') & (work['hhmm'] <= '15:30')].copy() time_range = pd.date_range('09:15', '15:30', freq='min').strftime('%H:%M').tolist() grouped = {} for date_str, g in work.groupby('date_str'): rg = g.set_index('hhmm').reindex(time_range) fcols = [c for c in rg.columns if c.startswith(('ce_','pe_')) or c in ('atm','straddle')] rg[fcols] = rg[fcols].ffill() if not rg.empty: grouped[date_str] = rg print(f"Days in panel: {len(grouped)} (min-straddle filter applied inside the backtest)") # Fast numpy cache (rebuilt once; the backtest engine reads from this for speed). PANEL = {} for ds, df in grouped.items(): idx = {t: i for i, t in enumerate(df.index)} cols = {c: df[c].to_numpy(dtype='float64') for c in df.columns if c.startswith(('ce_', 'pe_')) or c == 'atm'} so = df.loc[OPEN_TIME, 'straddle'] if OPEN_TIME in idx else np.nan PANEL[ds] = {'idx': idx, 'atm': cols['atm'], 'cols': cols, 'open': float(so) if pd.notna(so) else np.nan} print(f"Numpy panel cached for {len(PANEL)} days.") """## Step 4 — Backtest engine (moneyness + min-straddle filter)""" _COLNAME = {} def get_col_name(opt_type, delta): k = (opt_type, delta) if k not in _COLNAME: _COLNAME[k] = (f'{opt_type}_atm' if delta == 0 else (f'{opt_type}_p{delta}' if delta > 0 else f'{opt_type}_m{abs(delta)}')) return _COLNAME[k] def run_backtest(grouped, sl_pct, moneyness='ATM', min_straddle=0.0, entry_time='09:30', exit_time='15:30'): """Fast panel-based engine. Sells the chosen structure at entry_time, squares off at exit_time with independent leg-wise SL, and skips any day whose opening straddle (at OPEN_TIME) is < min_straddle. Reads the precomputed PANEL for speed.""" ce_off = MONEYNESS[moneyness]['ce_off']; pe_off = MONEYNESS[moneyness]['pe_off'] ce_entry_col = get_col_name('ce', ce_off); pe_entry_col = get_col_name('pe', pe_off) all_trades = []; skipped = 0 for date_str, p in PANEL.items(): idx = p['idx'] if entry_time not in idx or exit_time not in idx: continue op = p['open'] if np.isnan(op) or op < min_straddle: skipped += 1; continue ei = idx[entry_time]; xi = idx[exit_time] atm = p['atm']; cols = p['cols'] ce_arr = cols.get(ce_entry_col); pe_arr = cols.get(pe_entry_col) if ce_arr is None or pe_arr is None: continue eav = atm[ei]; ece = ce_arr[ei]; epe = pe_arr[ei] if np.isnan(eav) or np.isnan(ece) or np.isnan(epe): continue entry_atm = int(eav); ce_strike = entry_atm + ce_off; pe_strike = entry_atm + pe_off ce_sl_t = ece * (1 + sl_pct); pe_sl_t = epe * (1 + sl_pct) ce_active = pe_active = True ce_exit = ece; pe_exit = epe; last_ce = ece; last_pe = epe ce_etime = pe_etime = exit_time; ce_reason = pe_reason = f'{exit_time} Sqoff' for i in range(ei + 1, xi + 1): if not ce_active and not pe_active: break cav = atm[i] if np.isnan(cav): continue curr_atm = int(cav) if ce_active: arr = cols.get(get_col_name('ce', int(round((ce_strike - curr_atm) / 50.0) * 50))) if arr is not None: v = arr[i] if not np.isnan(v): last_ce = v if v >= ce_sl_t: ce_active = False; ce_exit = v; ce_reason = 'SL Hit' if pe_active: arr = cols.get(get_col_name('pe', int(round((pe_strike - curr_atm) / 50.0) * 50))) if arr is not None: v = arr[i] if not np.isnan(v): last_pe = v if v >= pe_sl_t: pe_active = False; pe_exit = v; pe_reason = 'SL Hit' if ce_active: ce_exit = last_ce if pe_active: pe_exit = last_pe ce_cost = FRICTION * (ece + ce_exit); pe_cost = FRICTION * (epe + pe_exit) ce_pnl = (ece - ce_exit) - ce_cost; pe_pnl = (epe - pe_exit) - pe_cost all_trades.append({ 'Date': date_str, 'Config': moneyness, 'Open_Straddle': round(op, 2), 'Entry_ATM': entry_atm, 'CE_Strike': ce_strike, 'PE_Strike': pe_strike, 'Entry_CE': round(ece, 2), 'Entry_PE': round(epe, 2), 'Exit_CE': round(ce_exit, 2), 'Exit_PE': round(pe_exit, 2), 'CE_Exit_Reason': ce_reason, 'PE_Exit_Reason': pe_reason, 'CE_PnL_Net': round(ce_pnl, 2), 'PE_PnL_Net': round(pe_pnl, 2), 'Day_PnL': round(ce_pnl + pe_pnl, 2), }) run_backtest.last_skipped = skipped return pd.DataFrame(all_trades) def run_flexible_backtest(grouped_data, sl_pct, entry_time, exit_time, moneyness='ATM', min_straddle=0.0): return run_backtest(grouped_data, sl_pct, moneyness=moneyness, min_straddle=min_straddle, entry_time=entry_time, exit_time=exit_time) print("Engine ready (fast panel-based).") """### Shared metrics & plotting helpers (used by every framing)""" def calc_metrics(trades_df, sl_label, config_key=None): total = len(trades_df) base = {'SL Level': sl_label, 'Trades': total, 'Win Rate %': 0, 'Implied RR': 0, 'Avg Winner': 0, 'Avg Loser': 0, 'Realized RR': 0, 'RR Efficiency': 0, 'Sharpe': 0, 'Profit Factor': 0, 'Max DD': 0, 'Total PnL': 0} if config_key is not None: base = {'Config Key': config_key, **base} if total == 0: return base pnl = trades_df['Day_PnL'].values winners = pnl[pnl > 0]; losers = pnl[pnl < 0] win_rate = len(winners) / total * 100 avg_w = winners.mean() if len(winners) else 0 avg_l = losers.mean() if len(losers) else 0 realized_rr = avg_w / abs(avg_l) if avg_l != 0 else 0 sl_num = float(sl_label.replace('%', '')) / 100.0 implied_rr = 1.0 / sl_num if sl_num > 0 else 0 rr_eff = realized_rr / implied_rr if implied_rr > 0 else 0 sharpe = (pnl.mean() / pnl.std() * ANNUALISE) if pnl.std() > 0 else 0 cum = np.cumsum(pnl); peak = np.maximum.accumulate(cum); max_dd = (cum - peak).min() gw = winners.sum() if len(winners) else 0 gl = abs(losers.sum()) if len(losers) else 0.01 base.update({'Win Rate %': round(win_rate, 1), 'Implied RR': round(implied_rr, 2), 'Avg Winner': round(avg_w, 2), 'Avg Loser': round(avg_l, 2), 'Realized RR': round(realized_rr, 2), 'RR Efficiency': round(rr_eff, 2), 'Sharpe': round(sharpe, 2), 'Profit Factor': round(gw / gl, 2), 'Max DD': round(max_dd, 2), 'Total PnL': round(pnl.sum(), 2)}) return base def _sl_palette(labels): order = sorted(labels, key=lambda x: float(x.replace('%', ''))) base = sns.color_palette('viridis', len(order)) return order, {sl: ('red' if sl == '300%' else base[i]) for i, sl in enumerate(order)} def run_basket_set(time_configs): """Run the selected moneyness + filter across every (entry,exit) config x SL level.""" results = {} for c in time_configs: key = f"Basket: {c['basket']}, Entry: {c['entry']}, Exit: {c['exit']}" results[key] = {} for sl in SL_LEVELS: results[key][f"{sl}%"] = run_flexible_backtest( grouped, sl / 100.0, c['entry'], c['exit'], moneyness=SELECTED_MONEYNESS, min_straddle=MIN_STRADDLE) return results def build_master(all_flex): rows = [calc_metrics(t, sl, cfg_key) for cfg_key, d in all_flex.items() for sl, t in d.items()] df = pd.DataFrame(rows) df[['Basket', 'Entry Time', 'Exit Time']] = df['Config Key'].str.extract( r'Basket: ([\w-]+), Entry: (\d{2}:\d{2}), Exit: (\d{2}:\d{2})') return df BASKET_ORDER = ['Morning', 'Mid-day', 'Afternoon'] print("Helpers ready.") """# Framing 1 — Single 09:30 entry → 15:30 exit""" print(f"[{SELECTED_MONEYNESS} | min straddle {MIN_STRADDLE}] single entry {ENTRY_TIME} -> {EXIT_TIME}\n") all_results = {} for sl in SL_LEVELS: all_results[f"{sl}%"] = run_backtest(grouped, sl/100.0, SELECTED_MONEYNESS, MIN_STRADDLE, entry_time=ENTRY_TIME, exit_time=EXIT_TIME) print(f"Days skipped by filter: {run_backtest.last_skipped}") for lab, t in all_results.items(): print(f" SL {lab:>4}: {len(t):3d} trades | total PnL {t['Day_PnL'].sum():9.1f}") """### Leaderboard (single entry)""" leader_df = (pd.DataFrame([calc_metrics(t, lab) for lab, t in all_results.items()]) .sort_values('Sharpe', ascending=False).reset_index(drop=True)) print(f"LEADERBOARD | {SELECTED_MONEYNESS} | min straddle {MIN_STRADDLE} (by Sharpe)") display(leader_df) """### Total PnL by stop-loss level (single entry)""" totals = {k: v['Day_PnL'].sum() for k, v in all_results.items()} plt.figure(figsize=(11, 5)) bars = plt.bar(list(totals), list(totals.values()), color=['#b3261e' if x < 0 else '#2e7d32' for x in totals.values()]) plt.title(f"Total PnL by Stop-Loss — {SELECTED_MONEYNESS} | min straddle {MIN_STRADDLE}") plt.xlabel("Stop-Loss level"); plt.ylabel("Total PnL (points)") plt.axhline(0, color='k', lw=0.8); plt.grid(axis='y', alpha=0.3) for b, v in zip(bars, totals.values()): plt.text(b.get_x()+b.get_width()/2, v, f"{v:.0f}", ha='center', va='bottom' if v >= 0 else 'top', fontsize=8) plt.tight_layout(); plt.show() """### Realized vs Implied Reward-to-Risk (single entry)""" order, _ = _sl_palette(leader_df['SL Level']) rr = leader_df.set_index('SL Level').loc[order] plt.figure(figsize=(12, 5)) plt.plot(order, rr['Realized RR'], marker='o', label='Realized RR') plt.plot(order, rr['Implied RR'], marker='o', label='Implied RR') plt.title(f"Realized vs Implied RR Across Stop-Loss Levels — {SELECTED_MONEYNESS}") plt.xlabel("Stop-Loss level"); plt.ylabel("Reward-to-Risk ratio") plt.grid(alpha=0.3, ls='--'); plt.legend(); plt.tight_layout(); plt.show() """### Year-by-year metrics, and annual / quarterly PnL by SL (single entry)""" # Year-by-year metric tables yearly_rows = [] for sl_label, t in all_results.items(): tt = t.copy(); tt['Year'] = pd.to_datetime(tt['Date']).dt.year for yr, ydf in tt.groupby('Year'): m = calc_metrics(ydf, sl_label); m['Year'] = yr; yearly_rows.append(m) yearly_performance_df = (pd.DataFrame(yearly_rows) .sort_values(['SL Level', 'Year']).reset_index(drop=True)) yearly_performance_df = yearly_performance_df[['Year', 'SL Level'] + [c for c in yearly_performance_df.columns if c not in ('Year', 'SL Level')]] for yr in sorted(yearly_performance_df['Year'].unique()): print(f"--- Year {yr} ---") display(yearly_performance_df[yearly_performance_df['Year'] == yr].drop(columns='Year')) def plot_annual_returns_grouped(all_results): rows = [] for sl, t in all_results.items(): tt = t.copy(); tt['Year'] = pd.to_datetime(tt['Date']).dt.year s = tt.groupby('Year')['Day_PnL'].sum().reset_index(); s['SL Level'] = sl; rows.append(s) comb = pd.concat(rows) order, pal = _sl_palette(comb['SL Level'].unique()) plt.figure(figsize=(12, 7)) ax = sns.barplot(data=comb, x='Year', y='Day_PnL', hue='SL Level', hue_order=order, palette=pal) plt.title(f'Annual Returns by SL Level (Market Regime) — {SELECTED_MONEYNESS}') plt.ylabel('Total PnL (Points)'); plt.xlabel('Year'); plt.grid(axis='y', ls='--', alpha=0.6) plt.legend(title='SL Level', bbox_to_anchor=(1.02, 1), loc='upper left') for c in ax.containers: ax.bar_label(c, fmt='%.0f', fontsize=8, padding=2) plt.tight_layout(); plt.show() plot_annual_returns_grouped(all_results) def plot_yearly_pnl_by_sl(yearly_df): df = yearly_df.copy() order, pal = _sl_palette(df['SL Level'].unique()) df['SL Level'] = pd.Categorical(df['SL Level'], categories=order, ordered=True) g = sns.catplot(data=df, x='SL Level', y='Total PnL', hue='SL Level', hue_order=order, col='Year', kind='bar', col_wrap=2, height=4.2, aspect=1.3, palette=pal, sharey=False, legend=False) g.set_axis_labels("Stop-Loss level", "Total PnL (Points)"); g.set_titles("Year: {col_name}") g.fig.suptitle(f'Total PnL by SL Level (by Year) — {SELECTED_MONEYNESS}', y=1.02, fontsize=15) for ax in g.axes.flat: for c in ax.containers: ax.bar_label(c, fmt='%.0f', fontsize=7, padding=2) ax.grid(axis='y', ls='--', alpha=0.6); ax.tick_params(axis='x', rotation=45) plt.tight_layout(rect=[0, 0.03, 1, 0.98]); plt.show() plot_yearly_pnl_by_sl(yearly_performance_df) def plot_quarterly_returns(all_results): rows = [] for sl, t in all_results.items(): tt = t.copy(); tt['Date'] = pd.to_datetime(tt['Date']) tt['Year'] = tt['Date'].dt.year; tt['Quarter'] = tt['Date'].dt.quarter s = tt.groupby(['Year', 'Quarter'])['Day_PnL'].sum().reset_index(); s['SL Level'] = sl rows.append(s) comb = pd.concat(rows).reset_index(drop=True) comb['Quarter'] = pd.Categorical(comb['Quarter'], categories=[1, 2, 3, 4], ordered=True) order, pal = _sl_palette(comb['SL Level'].unique()) comb['SL Level'] = pd.Categorical(comb['SL Level'], categories=order, ordered=True) g = sns.catplot(data=comb, x='Quarter', y='Day_PnL', hue='SL Level', hue_order=order, col='Year', kind='bar', col_wrap=2, height=4.2, aspect=1.3, palette=pal, sharey=False, legend_out=True) g.set_axis_labels("Quarter", "Total PnL (Points)"); g.set_titles("Year: {col_name}") g.fig.suptitle(f'Quarterly Returns by SL Level (by Year) — {SELECTED_MONEYNESS}', y=1.02, fontsize=15) for ax in g.axes.flat: for c in ax.containers: ax.bar_label(c, fmt='%.0f', fontsize=7, padding=2) ax.grid(axis='y', ls='--', alpha=0.6) plt.tight_layout(rect=[0, 0.03, 1, 0.98]); plt.show() plot_quarterly_returns(all_results) """# Framing 2 — 15-minute iterative entries → 15:30 exit (time baskets) Morning, Mid-day and Afternoon windows; every entry exits at 15:30. Morning starts at `ENTRY_TIME` (09:30) so the trade is never opened before the straddle gauge used by the filter. """ def gen_entries(start, end, step=15): out, t, e = [], pd.to_datetime(start, format='%H:%M'), pd.to_datetime(end, format='%H:%M') while t <= e: out.append(t.strftime('%H:%M')); t += pd.Timedelta(minutes=step) return out time_configs = [] for e in gen_entries('09:30', '11:15'): time_configs.append({'entry': e, 'exit': '15:30', 'basket': 'Morning'}) for e in gen_entries('11:30', '13:15'): time_configs.append({'entry': e, 'exit': '15:30', 'basket': 'Mid-day'}) for e in gen_entries('13:30', '15:15'): time_configs.append({'entry': e, 'exit': '15:30', 'basket': 'Afternoon'}) print(f"15-min configs: {len(time_configs)} | total backtests: {len(time_configs)*len(SL_LEVELS)}") all_flexible_results = run_basket_set(time_configs) master_leader_df = build_master(all_flexible_results) print("\nMaster leaderboard (top by Sharpe):") display(master_leader_df.sort_values('Sharpe', ascending=False) .drop(columns='Config Key').head(10).reset_index(drop=True)) """### Average total PnL per basket & SL — 15-min (across all entry times)""" def plot_avg_pnl_by_basket_sl(master_df, suffix): agg = master_df.groupby(['Basket', 'SL Level'])['Total PnL'].mean().reset_index() order, pal = _sl_palette(agg['SL Level'].unique()) agg['Basket'] = pd.Categorical(agg['Basket'], categories=BASKET_ORDER, ordered=True) plt.figure(figsize=(14, 7)) ax = sns.barplot(data=agg, x='Basket', y='Total PnL', hue='SL Level', hue_order=order, palette=pal) plt.title(f'Average Total PnL per Basket and SL Level ({suffix}) — {SELECTED_MONEYNESS}') plt.xlabel('Time Basket'); plt.ylabel('Average Total PnL (Points)') plt.legend(title='SL Level', bbox_to_anchor=(1.02, 1), loc='upper left') plt.grid(axis='y', ls='--', alpha=0.6) for c in ax.containers: ax.bar_label(c, fmt='%.0f', fontsize=7, padding=2) plt.tight_layout(); plt.show() plot_avg_pnl_by_basket_sl(master_leader_df, 'Across All Entry Times, 15-min') """### Average annual PnL by basket & SL — 15-min (faceted by year)""" def plot_avg_pnl_by_basket_sl_yearly(all_flex, suffix): rows = [] for key, d in all_flex.items(): bm = re.search(r'Basket: ([\w-]+)', key); basket = bm.group(1) if bm else 'Unknown' for sl, t in d.items(): if t.empty: continue tt = t.copy(); tt['Year'] = pd.to_datetime(tt['Date']).dt.year s = tt.groupby('Year')['Day_PnL'].sum().reset_index() s['SL Level'] = sl; s['Basket'] = basket; rows.append(s) comb = pd.concat(rows) avg = (comb.groupby(['Year', 'Basket', 'SL Level'])['Day_PnL'].mean().reset_index() .rename(columns={'Day_PnL': 'Average Annual PnL'})) order, pal = _sl_palette(avg['SL Level'].unique()) avg['Basket'] = pd.Categorical(avg['Basket'], categories=BASKET_ORDER, ordered=True) g = sns.catplot(data=avg, x='Basket', y='Average Annual PnL', hue='SL Level', hue_order=order, col='Year', kind='bar', col_wrap=2, height=4.2, aspect=1.3, palette=pal, sharey=False, legend_out=True) g.set_axis_labels("Time Basket", "Average Annual PnL (Points)"); g.set_titles("Year: {col_name}") g.fig.suptitle(f'Average Annual PnL by Basket and SL Level ({suffix}) — {SELECTED_MONEYNESS}', y=1.02, fontsize=15) for ax in g.axes.flat: for c in ax.containers: ax.bar_label(c, fmt='%.0f', fontsize=6, padding=2) ax.grid(axis='y', ls='--', alpha=0.6) plt.tight_layout(rect=[0, 0.03, 1, 0.98]); plt.show() plot_avg_pnl_by_basket_sl_yearly(all_flexible_results, '15-min') """### Realized vs Implied RR per basket — 15-min""" def plot_rr_by_basket(master_df, suffix): df = master_df.copy(); df['SL Numeric'] = df['SL Level'].str.replace('%', '').astype(float) rr = (df.groupby(['Basket', 'SL Level', 'SL Numeric']) .agg(Average_Realized_RR=('Realized RR', 'mean'), Average_Implied_RR=('Implied RR', 'mean')).reset_index() .sort_values(['Basket', 'SL Numeric'])) rr['Basket'] = pd.Categorical(rr['Basket'], categories=BASKET_ORDER, ordered=True) melt = pd.melt(rr, id_vars=['Basket', 'SL Level', 'SL Numeric'], value_vars=['Average_Realized_RR', 'Average_Implied_RR'], var_name='RR Type', value_name='RR Value') g = sns.FacetGrid(melt, col='Basket', col_wrap=3, height=4.5, aspect=1.2, sharey=False, col_order=BASKET_ORDER) g.map_dataframe(sns.lineplot, x='SL Level', y='RR Value', hue='RR Type', marker='o') g.set_axis_labels("Stop-Loss level", "Reward-to-Risk ratio"); g.set_titles("Basket: {col_name}") g.add_legend(title='RR Type') g.fig.suptitle(f'Average Realized vs Implied RR per Basket ({suffix}) — {SELECTED_MONEYNESS}', y=1.02, fontsize=15) for ax in g.axes.flat: ax.tick_params(axis='x', rotation=45); ax.grid(True, ls='--', alpha=0.6) plt.tight_layout(rect=[0, 0.03, 1, 0.98]); plt.show() plot_rr_by_basket(master_leader_df, '15-min') """# Framing 3 — 2-hour window baskets Four staggered entries per basket collapsing to a single ~2-hour exit (Morning→11:30, Mid-day→13:30, Afternoon→15:30). """ time_configs_2hr = [] for e in ['09:16','09:20','09:25','09:30']: time_configs_2hr.append({'entry': e, 'exit': '11:30', 'basket': 'Morning'}) for e in ['11:16','11:20','11:25','11:30']: time_configs_2hr.append({'entry': e, 'exit': '13:30', 'basket': 'Mid-day'}) for e in ['13:16','13:20','13:25','13:30']: time_configs_2hr.append({'entry': e, 'exit': '15:30', 'basket': 'Afternoon'}) print(f"2-hour configs: {len(time_configs_2hr)} | total backtests: {len(time_configs_2hr)*len(SL_LEVELS)}") all_flexible_results_2hr = run_basket_set(time_configs_2hr) master_leader_df_2hr = build_master(all_flexible_results_2hr) print("\nAverage performance per 2-hour basket:") display(master_leader_df_2hr.groupby('Basket').agg( Average_Total_PnL=('Total PnL', 'mean'), Average_Sharpe=('Sharpe', 'mean'), Average_Win_Rate=('Win Rate %', 'mean'), Average_Realized_RR=('Realized RR', 'mean'), Average_Implied_RR=('Implied RR', 'mean')).reindex(BASKET_ORDER).round(2)) """### Average total PnL per basket & SL — 2-hour""" plot_avg_pnl_by_basket_sl(master_leader_df_2hr, '2-Hour Baskets') """### Realized vs Implied RR per basket — 2-hour""" plot_rr_by_basket(master_leader_df_2hr, '2-Hour Baskets') """--- ### Notes - **Filter** = trade a day only if its ATM straddle at `OPEN_TIME` (default 09:30) ≥ `MIN_STRADDLE`. It applies identically to all three framings (a day-level eligibility gate), so the day-set is consistent across them. Switch `OPEN_TIME` to `09:15` for the true open (a 75 floor then removes far fewer days). - **Re-running a different structure/threshold**: re-run Step 2, then the framing cells. `grouped` (Step 3) is moneyness-independent. - **Sharpe** uses √52 (weekly expiries) consistently. **Stop-loss** is a % of collected premium, so the same SL% is a much smaller spot move on ITM than OTM — not comparable across moneyness in spot terms. - Excluded the 5-minute / 1-minute density sweeps (not in your list); the three requested framings are all here. """