**Evaluating trading strategies** I need to consider various strategies and features, like using the NIFTY benchmark and examining relative strength, trend momentum, and volume. I could combine a scoring system for breakouts versus pullbacks and include risk control with ATR-based trailing indicators. It's important to focus on intraday trading for cash equities, keeping in mind risk limits and transaction costs. The code should be efficient yet readable, implemented in a single runnable file using libraries like pandas and yfinance. I want to ensure the setup is valid for NSE intraday trading while considering operational hours and holidays.**Defining trading timelines** I'm outlining the trading hours: from 09:15 to 15:30, with the last bar closing at 15:25. For intraday trading, new trades can happen after the market opens, based on the 15 or 30-minute range. I'll use VWAP reset daily and apply a 20-bar relative volume adjusted for the time of day while excluding incomplete bars. The opening range for the first 15 or 30 minutes will guide my breakout entries, starting from the earliest 09:30 bar that finishes at 09:35. There is no universally “best” signal generator without specifying the stock universe, transaction costs, and risk limits—and validating it out of sample. Below is a **practical, auditable NSE research framework**, not Goldman Sachs research or a guaranteed-profit system. It supports: | Component | Implementation | |---|---| | **Swing trading** | Long-only; trend + NIFTY regime filter; 20-day breakout or EMA20 recovery | | **Intraday trading** | Long/short; 5-minute opening-range breakout + VWAP + trend confirmation | | **Signal quality** | Relative volume, momentum, ADX, liquidity, and relative-strength scoring | | **Risk management** | ATR stops, risk-based sizing, exposure/volume caps, gap-entry rejection | | **Backtesting** | Next-bar entries, fees, slippage, conservative stop/target handling | | **Data hygiene** | IST timestamps, regular-session filtering, exclusion of forming candles | **Important:** Yahoo Finance is used only as a convenient research adapter. Its intraday coverage and timeliness are unsuitable for dependable live execution. Replace it with broker/licensed NSE data before trading. ## 1. Install Python 3.10+: ```bash python -m pip install -U numpy pandas yfinance ``` ## 2. Save as `nse_signals.py` ```python from __future__ import annotations import argparse import logging from dataclasses import dataclass from pathlib import Path import numpy as np import pandas as pd import yfinance as yf IST = "Asia/Kolkata" OHLCV = ["Open", "High", "Low", "Close", "Volume"] # Regular NSE cash-session assumptions. OPEN_MINUTE = 9 * 60 + 15 OR_END_MINUTE = 9 * 60 + 30 LAST_ENTRY_MINUTE = 14 * 60 + 45 FORCE_EXIT_MINUTE = 15 * 60 + 15 CLOSE_MINUTE = 15 * 60 + 30 logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") log = logging.getLogger(__name__) @dataclass(frozen=True) class Config: capital: float = 1_000_000.0 # ₹10 lakh risk_fraction: float = 0.005 # 0.5% estimated risk per trade max_position_fraction: float = 0.20 max_volume_participation: float = 0.01 min_daily_turnover: float = 100_000_000.0 # Approx. ₹10 crore min_score: int = 80 # Illustrative per-side cost proxies, NOT an Indian tax schedule. # None selects 15 bps swing / 5 bps intraday. fee_bps: float | None = None slippage_bps: float = 5.0 max_entry_gap_atr: float = 0.50 swing_max_bars: int = 15 def strategy_parameters(mode: str) -> tuple[float, float]: """Return stop distance in ATRs, and gross reward/risk target.""" return (2.0, 2.5) if mode == "swing" else (1.5, 2.0) def fee_rate(mode: str, cfg: Config) -> float: bps = cfg.fee_bps if bps is None: bps = 15.0 if mode == "swing" else 5.0 return bps / 10_000.0 # ---------------------------------------------------------------------- # DATA ADAPTER # Replace this function for production, preserving its output contract: # OHLCV columns + bar_end; unique, sorted, IST-aware index. # Intraday timestamps must label the START of each 5-minute candle. # ---------------------------------------------------------------------- def load_bars( symbol: str, mode: str, asof: pd.Timestamp, ) -> pd.DataFrame: interval = "1d" if mode == "swing" else "5m" period = "2y" if mode == "swing" else "1mo" raw = yf.Ticker(symbol).history( period=period, interval=interval, auto_adjust=True, actions=False, ) if raw.empty: raise ValueError(f"No data returned for {symbol}/{mode}") x = raw.loc[:, OHLCV].copy() idx = pd.DatetimeIndex(x.index) x.index = ( idx.tz_localize(IST) if idx.tz is None else idx.tz_convert(IST) ) x = x[~x.index.duplicated(keep="last")].sort_index() x = x.apply(pd.to_numeric, errors="coerce").dropna() valid = ( (x[["Open", "High", "Low", "Close"]] > 0).all(axis=1) & (x["Volume"] >= 0) & (x["High"] >= x[["Open", "Close", "Low"]].max(axis=1)) & (x["Low"] <= x[["Open", "Close", "High"]].min(axis=1)) ) x = x.loc[valid] x = x.loc[x.index.dayofweek < 5] if mode == "intraday": minute = x.index.hour * 60 + x.index.minute x = x.loc[ (minute >= OPEN_MINUTE) & (minute < CLOSE_MINUTE) ].copy() x["bar_end"] = x.index + pd.Timedelta(minutes=5) else: x.index = x.index.normalize() x["bar_end"] = ( x.index + pd.Timedelta(hours=15, minutes=30) ) # Exclude candles whose scheduled regular-session end has not passed. # Special/Muhurat sessions require a proper exchange-calendar adapter. x = x.loc[x["bar_end"] <= asof].copy() if x.empty: raise ValueError(f"No completed candles for {symbol}/{mode}") return x # ---------------------------------------------------------------------- # INDICATORS # All calculations use the current completed bar and earlier observations. # ---------------------------------------------------------------------- def rma(s: pd.Series, n: int) -> pd.Series: """Wilder-style exponential smoothing.""" return s.ewm(alpha=1 / n, adjust=False, min_periods=n).mean() def indicators(bars: pd.DataFrame) -> pd.DataFrame: x = bars.copy() h, l, c = x["High"], x["Low"], x["Close"] for n in (9, 20, 50, 200): x[f"ema{n}"] = c.ewm( span=n, adjust=False, min_periods=n ).mean() previous_close = c.shift(1) tr = pd.concat( [ h - l, (h - previous_close).abs(), (l - previous_close).abs(), ], axis=1, ).max(axis=1) x["atr"] = rma(tr, 14) delta = c.diff() gain = rma(delta.clip(lower=0), 14) loss = rma((-delta).clip(lower=0), 14) rs = gain / loss.replace(0, np.nan) rsi = 100 - 100 / (1 + rs) rsi = rsi.mask((loss == 0) & (gain > 0), 100) rsi = rsi.mask((loss == 0) & (gain == 0), 50) x["rsi"] = rsi up_move = h.diff() down_move = -l.diff() plus_dm = up_move.where( (up_move > down_move) & (up_move > 0), 0.0 ) minus_dm = down_move.where( (down_move > up_move) & (down_move > 0), 0.0 ) plus_di = 100 * rma(plus_dm, 14) / x["atr"] minus_di = 100 * rma(minus_dm, 14) / x["atr"] dx = ( 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan) ) x["adx"] = rma(dx, 14) return x def points(*components: tuple[pd.Series, int]) -> pd.Series: """Transparent quality score; NOT a probability of success.""" return sum( weight * condition.fillna(False).astype(int) for condition, weight in components ) # ---------------------------------------------------------------------- # SIGNAL GENERATION # ---------------------------------------------------------------------- def generate_signals( stock: pd.DataFrame, benchmark: pd.DataFrame, mode: str, cfg: Config, ) -> pd.DataFrame: x = indicators(stock) # Exact alignment: missing benchmark observations fail the regime gate. # No forward-filling from stale benchmark candles. b = indicators(benchmark).reindex(x.index) c = x["Close"] atr = x["atr"] turnover = c * x["Volume"] atr_fraction = atr / c if mode == "swing": avg_volume = x["Volume"].shift(1).rolling(20).mean() x["rvol"] = x["Volume"] / avg_volume.replace(0, np.nan) liquid = ( turnover.shift(1).rolling(20).mean() >= cfg.min_daily_turnover ) trend = ( (c > x["ema20"]) & (x["ema20"] > x["ema50"]) & (x["ema50"] > x["ema200"]) & (x["ema50"] > x["ema50"].shift(5)) ) market_up = ( (b["Close"] > b["ema200"]) & (b["ema50"] > b["ema200"]) ) # Excludes the current candle from the breakout reference. prior_high = x["High"].shift(1).rolling(20).max() breakout = ( (c > prior_high) & (c.shift(1) <= prior_high.shift(1)) ) recovery = ( (c.shift(1) <= x["ema20"].shift(1)) & (c > x["ema20"]) & (c > x["High"].shift(1)) ) stock_return = c.pct_change(20, fill_method=None) index_return = b["Close"].pct_change(20, fill_method=None) relative_strength = stock_return > index_return x["score"] = points( (trend, 25), (market_up, 20), (x["rvol"] >= 1.25, 20), (relative_strength, 15), (x["rsi"].between(52, 72), 10), (x["adx"] >= 20, 10), ) eligible = ( trend & market_up & liquid & (x["rvol"] >= 1.0) & atr_fraction.between(0.005, 0.08) & ((c - x["ema20"]) / atr <= 3.0) ) fire = ( eligible & (breakout | recovery) & (x["score"] >= cfg.min_score) ) # Cash-equity swing strategy is deliberately long-only. x["side"] = fire.astype("int8") x["setup"] = np.select( [fire & breakout, fire & recovery], ["20D_BREAKOUT", "EMA20_RECOVERY"], default="", ) else: day = x.index.normalize() minute = pd.Series( x.index.hour * 60 + x.index.minute, index=x.index, ) # Approximate average turnover of prior completed sessions. session_turnover = turnover.groupby(day).sum() prior_session_turnover = ( session_turnover.shift(1) .rolling(10, min_periods=5) .mean() ) avg_turnover = pd.Series( day.map(prior_session_turnover), index=x.index, ) liquid = avg_turnover >= cfg.min_daily_turnover # Compare volume with the SAME 5-minute slot in prior sessions. # This avoids treating normal opening volume as unusual volume. expected_volume = x["Volume"].groupby(minute).transform( lambda s: s.shift(1).rolling( 10, min_periods=5 ).median() ) x["rvol"] = ( x["Volume"] / expected_volume.replace(0, np.nan) ) typical_price = ( x["High"] + x["Low"] + c ) / 3.0 cumulative_pv = ( typical_price * x["Volume"] ).groupby(day).cumsum() cumulative_volume = x["Volume"].groupby(day).cumsum() x["vwap"] = ( cumulative_pv / cumulative_volume.replace(0, np.nan) ) opening = minute.between( OPEN_MINUTE, OR_END_MINUTE - 1 ) # Cumulative construction avoids leaking future opening candles. x["or_high"] = ( x["High"].where(opening) .groupby(day).cummax() .groupby(day).ffill() ) x["or_low"] = ( x["Low"].where(opening) .groupby(day).cummin() .groupby(day).ffill() ) opening_count = opening.astype(int).groupby(day).cumsum() previous_close = c.groupby(day).shift(1) contiguous = x.index.to_series().diff().eq( pd.Timedelta(minutes=5) ) long_trigger = ( (c > x["or_high"]) & (previous_close <= x["or_high"]) ) short_trigger = ( (c < x["or_low"]) & (previous_close >= x["or_low"]) ) long_trend = ( (c > x["vwap"]) & (c > x["ema20"]) & (x["ema9"] > x["ema20"]) & (x["ema20"] > x["ema20"].shift(3)) ) short_trend = ( (c < x["vwap"]) & (c < x["ema20"]) & (x["ema9"] < x["ema20"]) & (x["ema20"] < x["ema20"].shift(3)) ) market_up = ( (b["Close"] > b["ema20"]) & (b["ema20"] > b["ema50"]) ) market_down = ( (b["Close"] < b["ema20"]) & (b["ema20"] < b["ema50"]) ) close_location = ( (c - x["Low"]) / (x["High"] - x["Low"]).replace(0, np.nan) ) long_score = points( (long_trend, 25), (market_up, 20), (x["rvol"] >= 1.30, 20), (x["adx"] >= 20, 15), (x["rsi"].between(52, 75), 10), (close_location >= 0.70, 10), ) short_score = points( (short_trend, 25), (market_down, 20), (x["rvol"] >= 1.30, 20), (x["adx"] >= 20, 15), (x["rsi"].between(25, 48), 10), (close_location <= 0.30, 10), ) eligible = ( minute.between( OR_END_MINUTE, LAST_ENTRY_MINUTE - 5 ) & (opening_count == 3) & contiguous & liquid & (x["rvol"] >= 1.0) & atr_fraction.between(0.0005, 0.025) & ((c - x["vwap"]).abs() / atr <= 2.5) ) long_fire = ( eligible & long_trigger & long_trend & market_up & (long_score >= cfg.min_score) ) short_fire = ( eligible & short_trigger & short_trend & market_down & (short_score >= cfg.min_score) ) # At most one qualifying signal per direction per session. long_fire = ( long_fire & (long_fire.groupby(day).cumsum() == 1) ) short_fire = ( short_fire & (short_fire.groupby(day).cumsum() == 1) ) x["side"] = np.select( [long_fire, short_fire], [1, -1], default=0 ).astype("int8") x["score"] = pd.concat( [long_score, short_score], axis=1 ).max(axis=1).astype(int) x["setup"] = np.select( [long_fire, short_fire], ["ORB_LONG", "ORB_SHORT"], default="", ) stop_atr, reward_risk = strategy_parameters(mode) distance = stop_atr * atr active = x["side"] != 0 # Indicative plans using signal close. # Actual target and quantity must be recalculated from the fill. x["stop"] = np.where( active, c - x["side"] * distance, np.nan ) x["target_ref"] = np.where( active, c + x["side"] * reward_risk * distance, np.nan, ) return x # ---------------------------------------------------------------------- # POSITION SIZING # Independent-trade sizing, NOT a portfolio allocator. # ---------------------------------------------------------------------- def position_size( equity: float, entry: float, stop: float, previous_volume: float, mode: str, cfg: Config, ) -> int: values = [equity, entry, stop, previous_volume] if not np.isfinite(values).all() or equity <= 0 or entry <= 0: return 0 distance = abs(entry - stop) if distance <= 0: return 0 # Approximate round-trip cost allowance for sizing. cost_buffer = 2 * entry * ( fee_rate(mode, cfg) + cfg.slippage_bps / 10_000.0 ) by_risk = ( equity * cfg.risk_fraction / (distance + cost_buffer) ) by_exposure = ( equity * cfg.max_position_fraction / entry ) by_volume = ( previous_volume * cfg.max_volume_participation ) return max(0, int(min(by_risk, by_exposure, by_volume))) # ---------------------------------------------------------------------- # SINGLE-SYMBOL OHLC BACKTEST # - Signal at completed bar; entry at NEXT bar's open. # - One position at a time. # - Stop first if both stop and target are touched inside a candle. # - Intraday flattening at the 15:15 bar OPEN. # - Swing time exit at the open after max holding bars. # - Open final positions remain marked to market, not force-liquidated. # ---------------------------------------------------------------------- def backtest( x: pd.DataFrame, mode: str, cfg: Config, ) -> tuple[pd.DataFrame, dict]: balance = cfg.capital position = None equity_curve = [balance] records = [] fee = fee_rate(mode, cfg) slip = cfg.slippage_bps / 10_000.0 _, reward_risk = strategy_parameters(mode) for i in range(1, len(x)): t = x.index[i] row = x.iloc[i] previous = x.iloc[i - 1] o, h, l, c = ( float(row[k]) for k in ["Open", "High", "Low", "Close"] ) minute = t.hour * 60 + t.minute if ( position is not None and mode == "intraday" and t - x.index[i - 1] != pd.Timedelta(minutes=5) ): raise ValueError( f"Missing intraday bars while holding at {t}; " "refusing to invent an execution path." ) if position is None and int(previous["side"]) != 0: side = int(previous["side"]) stop = float(previous["stop"]) allowed = True if mode == "intraday": allowed = ( t.normalize() == x.index[i - 1].normalize() and t - x.index[i - 1] == pd.Timedelta(minutes=5) and minute <= LAST_ENTRY_MINUTE ) allowed = allowed and ( abs(o - float(previous["Close"])) <= cfg.max_entry_gap_atr * float(previous["atr"]) ) entry = o * (1 + side * slip) unit_risk = side * (entry - stop) if allowed and side * (o - stop) > 0 and unit_risk > 0: qty = position_size( balance, entry, stop, float(previous["Volume"]), mode, cfg, ) if qty > 0: entry_fee = fee * entry * qty balance -= entry_fee position = { "side": side, "entry": entry, "stop": stop, "target": entry + side * reward_risk * unit_risk, "risk": unit_risk, "qty": qty, "entry_fee": entry_fee, "entry_bar": t, "held": 0, } if position is not None: p = position side = p["side"] stop, target = p["stop"], p["target"] exit_price = None reason = "" timed_exit = ( (mode == "intraday" and minute >= FORCE_EXIT_MINUTE) or ( mode == "swing" and p["held"] >= cfg.swing_max_bars ) ) # Opening gaps are resolved before the candle's later range. if side * (o - stop) <= 0: exit_price, reason = o, "GAP_STOP" elif side * (o - target) >= 0: # Conservative: no favorable opening-gap improvement. exit_price, reason = target, "GAP_TARGET" elif timed_exit: exit_price, reason = o, "TIME_EXIT" elif ( (side == 1 and l <= stop) or (side == -1 and h >= stop) ): exit_price, reason = stop, "STOP" elif ( (side == 1 and h >= target) or (side == -1 and l <= target) ): exit_price, reason = target, "TARGET" if exit_price is not None: exit_fill = exit_price * (1 - side * slip) gross_pnl = ( side * (exit_fill - p["entry"]) * p["qty"] ) exit_fee = fee * exit_fill * p["qty"] net_pnl = gross_pnl - p["entry_fee"] - exit_fee balance += gross_pnl - exit_fee records.append({ "entry_bar": p["entry_bar"], "exit_bar": t, "side": side, "qty": p["qty"], "entry": p["entry"], "exit": exit_fill, "pnl": net_pnl, "R": net_pnl / (p["qty"] * p["risk"]), "reason": reason, }) position = None else: position["held"] += 1 unrealized = 0.0 if position is not None: unrealized = ( position["side"] * (c - position["entry"]) * position["qty"] ) equity_curve.append(balance + unrealized) trade_columns = [ "entry_bar", "exit_bar", "side", "qty", "entry", "exit", "pnl", "R", "reason", ] trades = pd.DataFrame(records, columns=trade_columns) eq = pd.Series(equity_curve, dtype=float) winners = trades.loc[trades["pnl"] > 0, "pnl"].sum() losses = -trades.loc[trades["pnl"] < 0, "pnl"].sum() profit_factor = ( winners / losses if losses > 0 else (np.inf if winners > 0 else np.nan) ) summary = { "closed_trades": len(trades), "win_rate_pct": ( 100 * (trades["pnl"] > 0).mean() if len(trades) else np.nan ), "profit_factor": profit_factor, "average_R": trades["R"].mean(), "return_mtm_pct": 100 * (eq.iloc[-1] / cfg.capital - 1), "max_drawdown_pct": 100 * (1 - eq / eq.cummax()).max(), "open_position": position is not None, } return trades, summary # ---------------------------------------------------------------------- # LATEST SCAN / REPORTING # ---------------------------------------------------------------------- def latest_snapshot( x: pd.DataFrame, symbol: str, mode: str, cfg: Config, ) -> dict: row = x.iloc[-1] now = pd.Timestamp.now(tz=IST) signal_at = pd.Timestamp(row["bar_end"]) age_minutes = (now - signal_at).total_seconds() / 60 side = int(row["side"]) raw_signal = {1: "LONG", -1: "SHORT", 0: "WAIT"}[side] if mode == "intraday": minute = now.hour * 60 + now.minute timely = ( signal_at.date() == now.date() and 0 <= age_minutes <= 2 and OR_END_MINUTE <= minute <= LAST_ENTRY_MINUTE ) else: # A heuristic freshness check, not an exchange-calendar check. timely = 0 <= age_minutes <= 7 * 24 * 60 signal = raw_signal if side != 0 and not timely: signal = "WAIT_STALE" qty = 0 if side != 0 and timely: qty = position_size( cfg.capital, float(row["Close"]), float(row["stop"]), float(row["Volume"]), mode, cfg, ) if qty == 0: signal = "WAIT_SIZE" return { "symbol": symbol, "mode": mode, "signal_at": signal_at.isoformat(), "signal": signal, "raw_signal": raw_signal, "setup": row["setup"], "score": int(row["score"]), "entry_ref": float(row["Close"]), "stop": float(row["stop"]), "target_ref": float(row["target_ref"]), "qty_hint": qty, "price_risk_rs": ( qty * abs(float(row["Close"]) - float(row["stop"])) if qty else 0.0 ), "atr": float(row["atr"]), "rsi": float(row["rsi"]), "rvol": float(row["rvol"]), "age_minutes": round(age_minutes, 2), "data_status": ( "RESEARCH_ONLY" if timely else "STALE_OR_OFF_HOURS" ), } def main() -> None: parser = argparse.ArgumentParser( description="NSE swing + intraday research signal generator" ) parser.add_argument( "--symbols", nargs="+", default=[ "RELIANCE.NS", "TCS.NS", "HDFCBANK.NS", "ICICIBANK.NS", "INFY.NS", ], ) parser.add_argument( "--mode", choices=["swing", "intraday", "both"], default="both" ) parser.add_argument("--benchmark", default="^NSEI") parser.add_argument("--capital", type=float, default=1_000_000) parser.add_argument("--risk", type=float, default=0.005) parser.add_argument("--max-position", type=float, default=0.20) parser.add_argument("--min-score", type=int, default=80) parser.add_argument("--fee-bps", type=float, default=None) parser.add_argument("--slippage-bps", type=float, default=5.0) parser.add_argument("--backtest", action="store_true") parser.add_argument("--out", default="signal_output") args = parser.parse_args() if args.capital <= 0: parser.error("--capital must be positive") if not 0 < args.risk <= 0.02: parser.error("--risk must be > 0 and <= 0.02") if not 0 < args.max_position <= 1: parser.error("--max-position must be > 0 and <= 1") if not 0 <= args.min_score <= 100: parser.error("--min-score must be between 0 and 100") if args.slippage_bps < 0 or ( args.fee_bps is not None and args.fee_bps < 0 ): parser.error("Costs cannot be negative") cfg = Config( capital=args.capital, risk_fraction=args.risk, max_position_fraction=args.max_position, min_score=args.min_score, fee_bps=args.fee_bps, slippage_bps=args.slippage_bps, ) symbols = list(dict.fromkeys( s.upper() if "." in s else f"{s.upper()}.NS" for s in args.symbols )) modes = ( ["swing", "intraday"] if args.mode == "both" else [args.mode] ) output = Path(args.out) output.mkdir(parents=True, exist_ok=True) snapshots, histories, summaries, all_trades = [], [], [], [] asof = pd.Timestamp.now(tz=IST) for mode in modes: try: benchmark = load_bars(args.benchmark, mode, asof) except Exception as exc: log.error("Benchmark unavailable for %s: %s", mode, exc) continue for symbol in symbols: try: stock = load_bars(symbol, mode, asof) frame = generate_signals(stock, benchmark, mode, cfg) snapshots.append( latest_snapshot(frame, symbol, mode, cfg) ) history = frame.loc[ frame["side"] != 0, [ "bar_end", "side", "setup", "score", "Close", "stop", "target_ref", "atr", "rsi", "rvol", ], ].copy() history["symbol"] = symbol history["mode"] = mode histories.append(history.reset_index(drop=True)) except Exception as exc: log.warning("Skipping %s/%s: %s", symbol, mode, exc) continue if args.backtest: try: trades, summary = backtest(frame, mode, cfg) summary.update(symbol=symbol, mode=mode) summaries.append(summary) trades["symbol"] = symbol trades["mode"] = mode all_trades.append(trades) except Exception as exc: log.warning( "Backtest rejected for %s/%s: %s", symbol, mode, exc, ) if not snapshots: raise SystemExit("No usable data. Check symbols/feed availability.") latest = pd.DataFrame(snapshots).sort_values( ["mode", "score"], ascending=[True, False] ) latest.to_csv(output / "latest_scan.csv", index=False) pd.concat(histories, ignore_index=True).to_csv( output / "historical_signals.csv", index=False ) display = [ "symbol", "mode", "signal_at", "signal", "score", "entry_ref", "stop", "target_ref", "qty_hint", "data_status", ] print("\nLATEST COMPLETED-BAR SCAN") print(latest[display].round(2).to_string(index=False)) if args.backtest: summary_frame = pd.DataFrame(summaries) summary_frame.to_csv( output / "backtest_summary.csv", index=False ) trades_frame = ( pd.concat(all_trades, ignore_index=True) if all_trades else pd.DataFrame() ) trades_frame.to_csv(output / "trades.csv", index=False) print("\nINDEPENDENT SINGLE-SYMBOL BACKTESTS — NOT A PORTFOLIO") print(summary_frame.round(2).to_string(index=False)) print(f"\nFiles saved to: {output.resolve()}") if __name__ == "__main__": main() ``` ## 3. Run it ### Scan both strategies and run the historical simulation ```bash python nse_signals.py \ --symbols RELIANCE.NS TCS.NS HDFCBANK.NS ICICIBANK.NS INFY.NS \ --mode both \ --capital 1000000 \ --risk 0.005 \ --backtest ``` ### Intraday only ```bash python nse_signals.py \ --symbols RELIANCE.NS SBIN.NS ICICIBANK.NS LT.NS \ --mode intraday ``` ### Swing only, with smaller position risk ```bash python nse_signals.py \ --symbols RELIANCE.NS TCS.NS INFY.NS HDFCBANK.NS \ --mode swing \ --capital 500000 \ --risk 0.003 ``` ## 4. How to interpret the output - **`LONG` / `SHORT`**: A new setup occurred on the latest completed candle. - **`WAIT`**: No new setup. This is an entry scanner, **not a recommendation to close an existing position**. - **`WAIT_STALE`**: A setup exists, but the timestamp check rejects it. - **`entry_ref`**: Signal candle’s close—not a promised execution price. - **`stop` / `target_ref`**: Indicative price levels. Recalculate sizing and target from the actual fill. - **`qty_hint`**: Independent-trade sizing. It does **not** account for existing holdings or other simultaneous signals. - **`score = 80`** means 80 quality points, **not an 80% win probability**. `latest_scan.csv` contains only the latest-bar assessment. Older entry events are in `historical_signals.csv`; they should not be treated as current signals. ## Before using real money 1. **Validate the edge.** These parameters are starting assumptions, not optimized or verified profitable settings. One month of intraday Yahoo data is far too little for meaningful validation. Use longer clean data, chronological out-of-sample testing, and paper trading. 2. **Model actual Indian costs.** Replace the fee proxies with brokerage, STT, exchange/SEBI charges, GST, stamp duty, and applicable delivery/DP charges. Stress-test higher slippage. 3. **Add portfolio controls.** Enforce total open risk, available cash/margin, sector concentration, daily-loss limits, and duplicate-order prevention. Do not simply take every independently sized signal. 4. **Respect execution constraints.** Confirm intraday-short eligibility, broker square-off rules, tick sizes, holidays/special sessions, corporate actions, and price bands. Stops may not execute during gaps or circuits; the stated risk budget is **not a guaranteed maximum loss**. The backtest is intentionally simple and transparent. It is useful for rejecting weak ideas—not proof that a strategy will work live.

swinger