import json import math import pandas as pd import streamlit as st st.set_page_config(page_title="Index Scalping Paper Lab", page_icon="📈", layout="centered") st.title("Index Scalping — Paper Lab") st.warning("SIMULATION ONLY. This app has no broker connection and cannot place live orders.") st.write("Upload historical one-minute OHLCV data to test the EMA/VWAP pullback strategy.") with st.sidebar: st.header("Risk settings") starting_equity = st.number_input("Starting balance (£)", 50.0, 999.0, 500.0, 50.0) risk_pct = st.number_input("Risk per trade (%)", 0.05, 1.0, 0.25, 0.05) / 100 daily_loss_pct = st.number_input("Daily loss limit (%)", 0.25, 3.0, 1.0, 0.25) / 100 atr_mult = st.number_input("ATR stop multiple", 0.5, 3.0, 1.2, 0.1) reward_risk = st.number_input("Reward:risk target", 0.5, 3.0, 1.5, 0.25) spread = st.number_input("Assumed spread (index points)", 0.0, 100.0, 1.0, 0.25) max_spread = st.number_input("Maximum spread (points)", 0.0, 100.0, 2.0, 0.25) slippage = st.number_input("Slippage per side (points)", 0.0, 10.0, 0.25, 0.05) point_value = st.number_input("£ per index point per unit", 0.001, 10000.0, 1.0, 0.5) min_qty = st.number_input("Minimum quantity", 0.001, 1000.0, 1.0, 1.0) qty_step = st.number_input("Quantity increment", 0.001, 1000.0, 1.0, 1.0) max_qty = st.number_input("Maximum quantity cap", 0.001, 1000.0, 1.0, 1.0) max_hold = st.number_input("Maximum holding bars", 1, 120, 10, 1) allow_short = st.checkbox("Allow simulated short trades", True) session_start = st.text_input("Session start (data timezone)", "13:30") session_end = st.text_input("Session end (data timezone)", "20:00") upload = st.file_uploader("Upload CSV", type=["csv"]) st.caption("Required columns: timestamp, open, high, low, close, volume. Use one consistent timezone and data you are authorised to use.") def prepare(df): required = ["timestamp", "open", "high", "low", "close", "volume"] missing = set(required) - set(df.columns) if missing: raise ValueError("Missing columns: " + ", ".join(sorted(missing))) df = df[required].copy() df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce") for c in ["open", "high", "low", "close", "volume"]: df[c] = pd.to_numeric(df[c], errors="coerce") if df.isna().any().any(): raise ValueError("CSV has blank or invalid values.") df = df.sort_values("timestamp").drop_duplicates("timestamp").reset_index(drop=True) if len(df) < 100: st.info("For meaningful testing, upload substantially more than 100 bars; a short sample is only a format check.") if (df[["open","high","low","close"]] <= 0).any().any(): raise ValueError("Prices must be positive.") if (df["high"] < df[["open","close","low"]].max(axis=1)).any() or (df["low"] > df[["open","close","high"]].min(axis=1)).any(): raise ValueError("Invalid OHLC bar.") return df def backtest(raw): d = raw.copy() d["date"] = d.timestamp.dt.date d["ema_fast"] = d.close.ewm(span=9, adjust=False).mean() d["ema_slow"] = d.close.ewm(span=21, adjust=False).mean() # Resampled 5-minute bars are labelled at their close, so the trend is not used before completion. five = d.set_index("timestamp").resample("5min", label="right", closed="left").agg( {"open":"first","high":"max","low":"min","close":"last","volume":"sum"}).dropna(subset=["close"]).reset_index() five["trend_ema"] = five.close.ewm(span=21, adjust=False).mean() five["up"] = (five.close > five.trend_ema) & (five.trend_ema.diff() > 0) five["down"] = (five.close < five.trend_ema) & (five.trend_ema.diff() < 0) d = pd.merge_asof(d.sort_values("timestamp"), five[["timestamp","up","down"]].sort_values("timestamp"), on="timestamp", direction="backward") typical = (d.high + d.low + d.close) / 3 vol = d.volume.clip(lower=0) d["vwap"] = (typical * vol).groupby(d.date).cumsum() / vol.groupby(d.date).cumsum().replace(0, float("nan")) prev = d.close.shift(1) tr = pd.concat([d.high-d.low, (d.high-prev).abs(), (d.low-prev).abs()], axis=1).max(axis=1) d["atr"] = tr.ewm(alpha=1/14, adjust=False, min_periods=14).mean() d["long"] = d.up.fillna(False) & (d.ema_fast > d.ema_slow) & (d.low.shift(1) <= d.ema_fast.shift(1)) & (d.close > d.ema_fast) & (d.close > d.open) & (d.close > d.vwap) d["short"] = d.down.fillna(False) & (d.ema_fast < d.ema_slow) & (d.high.shift(1) >= d.ema_fast.shift(1)) & (d.close < d.ema_fast) & (d.close < d.open) & (d.close < d.vwap) equity = float(starting_equity) position = None trades = [] day0 = None day_start = equity losses_today = trades_today = 0 for i in range(1, len(d)): r = d.iloc[i] day = r.timestamp.date() if day != day0: day0, day_start, losses_today, trades_today = day, equity, 0, 0 if position: held = i - position["i"] exit_price, reason = None, None # Conservative: if both stop and target are touched in the same candle, assume stop first. if position["side"] == "long": if r.low <= position["stop"]: exit_price, reason = position["stop"], "stop" elif r.high >= position["target"]: exit_price, reason = position["target"], "target" else: if r.high >= position["stop"]: exit_price, reason = position["stop"], "stop" elif r.low <= position["target"]: exit_price, reason = position["target"], "target" if exit_price is None and held >= max_hold: exit_price, reason = r.close, "time_exit" if exit_price is not None: direction = 1 if position["side"] == "long" else -1 entry_exec = position["entry"] + direction * (spread/2 + slippage) exit_exec = exit_price - direction * (spread/2 + slippage) pnl = (exit_exec-entry_exec) * direction * position["qty"] * point_value equity += pnl losses_today += int(pnl < 0) trades.append({"entry_time":str(position["time"]), "exit_time":str(r.timestamp), "side":position["side"], "entry_price":position["entry"], "exit_price":exit_price, "qty":position["qty"], "stop":position["stop"], "target":position["target"], "reason":reason, "pnl_after_estimated_spread_slippage":round(pnl, 4), "equity_after":round(equity, 2), "bars_held":held}) position = None in_session = session_start <= r.timestamp.strftime("%H:%M") <= session_end daily_floor = day_start * (1 - daily_loss_pct) if position is None and in_session and equity > daily_floor and losses_today < 3 and trades_today < 8 and spread <= max_spread and pd.notna(r.atr) and r.atr > 0: side = "long" if bool(r.long) else ("short" if allow_short and bool(r.short) else None) if side: dist = float(r.atr) * atr_mult risk_cash = equity * risk_pct per_unit_risk = (dist + spread + 2*slippage) * point_value raw_qty = min(max_qty, risk_cash / per_unit_risk) if per_unit_risk > 0 else 0 qty = math.floor(raw_qty / qty_step) * qty_step if qty_step > 0 else 0 if qty >= min_qty: entry = float(r.close) if side == "long": stop, target = entry-dist, entry+dist*reward_risk else: stop, target = entry+dist, entry-dist*reward_risk position = {"side":side,"entry":entry,"time":r.timestamp,"i":i,"stop":stop,"target":target,"qty":qty} trades_today += 1 if position: r = d.iloc[-1] direction = 1 if position["side"] == "long" else -1 entry_exec = position["entry"] + direction*(spread/2+slippage) exit_exec = float(r.close) - direction*(spread/2+slippage) pnl = (exit_exec-entry_exec)*direction*position["qty"]*point_value equity += pnl trades.append({"entry_time":str(position["time"]),"exit_time":str(r.timestamp),"side":position["side"], "entry_price":position["entry"],"exit_price":float(r.close),"qty":position["qty"], "stop":position["stop"],"target":position["target"],"reason":"end_of_data", "pnl_after_estimated_spread_slippage":round(pnl,4),"equity_after":round(equity,2), "bars_held":len(d)-1-position["i"]}) t = pd.DataFrame(trades) if t.empty: return t, {"starting_equity":starting_equity,"ending_equity":round(equity,2),"net_pnl":0,"trades":0, "note":"No trades generated. Check session times, data and instrument settings."} p = t["pnl_after_estimated_spread_slippage"] wins, losses = p[p>0].sum(), p[p<0].sum() return t, {"starting_equity":starting_equity,"ending_equity":round(equity,2),"net_pnl":round(equity-starting_equity,2), "return_pct":round((equity/starting_equity-1)*100,2),"trades":len(t), "win_rate_pct":round((p>0).mean()*100,2), "profit_factor":round(wins/abs(losses),3) if losses < 0 else None, "note":"Simulated OHLC fills; real spreads, slippage, fees and execution can differ."} if upload: try: bars = prepare(pd.read_csv(upload)) st.success(f"Loaded {len(bars):,} bars.") if st.button("Run backtest", type="primary", use_container_width=True): if spread > max_spread: st.error("Assumed spread exceeds maximum allowed spread.") else: trades, metrics = backtest(bars) a,b,c = st.columns(3) a.metric("Net simulated P&L", f"£{metrics.get('net_pnl',0):.2f}") b.metric("Return", f"{metrics.get('return_pct',0):.2f}%") c.metric("Closed trades", str(metrics.get("trades",0))) st.json(metrics) if not trades.empty: st.dataframe(trades, use_container_width=True) st.download_button("Download trade log CSV", trades.to_csv(index=False), "paper_trades.csv", "text/csv") st.download_button("Download metrics JSON", json.dumps(metrics, indent=2), "metrics.json", "application/json") except Exception as e: st.error(str(e)) else: st.info("Upload a CSV to run the strategy. The app does not provide live market prices.") st.subheader("Strategy and safety") st.markdown("- EMA 9/21 pullback with completed 5-minute EMA trend filter\n- Session VWAP confirmation\n- ATR stop and configurable reward:risk target (default 1.5R)\n- Position sizing, quantity cap, spread filter, daily loss limit, and trade limits\n- No broker credentials, no API integration, no live order functionality")