#!/usr/bin/env python3 """ V11 Money Line - SHORTS ONLY Backtester PURPOSE: Optimize SHORT signal parameters WITHOUT touching LONG settings. This is a SEPARATE file from v11_moneyline_all_filters.py to avoid accidentally changing the proven LONG settings. Current LONG settings (DO NOT CHANGE - these are proven profitable): - ATR Period: 12, Multiplier: 3.8 - ADX Min: 15 (ADX Length: 17) - RSI Long: 56-69 - Long Position Max: 85% - Entry Buffer: -0.15 ATR (early entry) - Volume Filter: OFF SHORT settings to optimize (this is what we're sweeping): - RSI Short Min/Max (currently 30-70 - NEEDS OPTIMIZATION) - Short Position Min (currently 5% - way too low) - ADX Min for shorts (may differ from longs) - Entry Buffer for shorts - Volume filter for shorts Usage: python3 v11_shorts_only.py --sweep # Run parameter sweep for shorts python3 v11_shorts_only.py --baseline # Test with default short settings python3 v11_shorts_only.py --custom --rsi-min 40 --rsi-max 70 --pos-min 25 Created: Jan 6, 2026 Author: Trading Bot System """ from dataclasses import dataclass, field from typing import Optional, Callable, List, Tuple import pandas as pd import numpy as np import argparse import sys import csv from pathlib import Path from datetime import datetime import itertools import multiprocessing as mp #====================================================================== # TRADE CONFIG AND RESULTS (self-contained) #====================================================================== @dataclass class ShortTradeConfig: """Trading configuration for shorts simulation.""" position_size: float = 1000.0 take_profit_1_size_percent: float = 100.0 # TP1 closes 100% for shorts atr_multiplier_tp1: float = 1.45 / 0.43 * (1/2.0) # Targeting ~1.45% TP1 atr_multiplier_sl: float = 3.0 min_tp1_percent: float = 1.45 max_tp1_percent: float = 1.45 min_sl_percent: float = 0.8 max_sl_percent: float = 2.0 max_bars_per_trade: int = 200 # ~16 hours max @dataclass class ShortTradeResult: """Result of a single simulated short trade.""" entry_time: pd.Timestamp exit_time: pd.Timestamp entry_price: float exit_price: float realized_pnl: float profit_percent: float exit_reason: str bars_held: int tp1_hit: bool mae_percent: float mfe_percent: float adx_at_entry: float atr_at_entry: float rsi_at_entry: float price_position_at_entry: float @dataclass class ShortSimulationResult: """Result of a shorts simulation backtest.""" trades: List[ShortTradeResult] = field(default_factory=list) @property def total_trades(self) -> int: return len(self.trades) @property def winning_trades(self) -> int: return sum(1 for t in self.trades if t.realized_pnl > 0) @property def losing_trades(self) -> int: return sum(1 for t in self.trades if t.realized_pnl <= 0) @property def win_rate(self) -> float: if self.total_trades == 0: return 0.0 return self.winning_trades / self.total_trades @property def total_pnl(self) -> float: return sum(t.realized_pnl for t in self.trades) @property def average_pnl(self) -> float: if self.total_trades == 0: return 0.0 return self.total_pnl / self.total_trades @property def profit_factor(self) -> float: gross_profit = sum(t.realized_pnl for t in self.trades if t.realized_pnl > 0) gross_loss = abs(sum(t.realized_pnl for t in self.trades if t.realized_pnl <= 0)) if gross_loss == 0: return float('inf') if gross_profit > 0 else 0.0 return gross_profit / gross_loss @property def max_drawdown(self) -> float: if not self.trades: return 0.0 cumulative = 0.0 peak = 0.0 max_dd = 0.0 for t in self.trades: cumulative += t.realized_pnl if cumulative > peak: peak = cumulative dd = peak - cumulative if dd > max_dd: max_dd = dd return max_dd @dataclass class ShortOnlyInputs: """ Parameters for SHORT-ONLY signal generation. FIXED parameters (from proven LONG optimization - shared with indicator): - atr_period, multiplier: Supertrend calculation - adx_length: ADX calculation period - flip_threshold, confirm_bars: Trend flip confirmation SWEEP parameters (what we're optimizing for SHORTs): - adx_min: Minimum ADX for short entries (may differ from long) - rsi_short_min, rsi_short_max: RSI boundaries for shorts - short_pos_min: Minimum price position for shorts (avoid catching falling knives) - entry_buffer_atr: How far beyond line price must be (negative = early entry) - vol_min, vol_max: Volume ratio bounds """ # FIXED - Supertrend/ATR (proven settings, don't sweep) atr_period: int = 12 multiplier: float = 3.8 adx_length: int = 17 rsi_length: int = 14 flip_threshold: float = 0.0 # Percentage for flip confirmation confirm_bars: int = 1 # Bars needed to confirm flip cooldown_bars: int = 2 # Bars between signals # SWEEP - Short-specific filters (THESE are what we're optimizing) adx_min: float = 15.0 # Minimum ADX for trend strength rsi_short_min: float = 30.0 # RSI minimum for shorts (default 30) rsi_short_max: float = 70.0 # RSI maximum for shorts (default 70) short_pos_min: float = 5.0 # Price position minimum % (default 5 - too low!) entry_buffer_atr: float = -0.15 # ATR multiplier for entry buffer (negative = early) vol_min: float = 0.0 # Minimum volume ratio (0 = disabled) vol_max: float = 5.0 # Maximum volume ratio # Control use_quality_filters: bool = True # Apply all filters @dataclass class ShortSignal: """A single SHORT signal from the Money Line indicator.""" timestamp: pd.Timestamp entry_price: float adx: float atr: float rsi: float volume_ratio: float price_position: float direction: str = "short" # Always short def ema(series: pd.Series, period: int) -> pd.Series: """Exponential Moving Average.""" return series.ewm(span=period, adjust=False).mean() def rolling_volume_ratio(volume: pd.Series, period: int = 20) -> pd.Series: """Volume relative to 20-period SMA.""" sma = volume.rolling(window=period).mean() return volume / sma def price_position(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 100) -> pd.Series: """Price position within recent range (0-100%).""" highest = high.rolling(window=period).max() lowest = low.rolling(window=period).min() range_val = highest - lowest pos = np.where(range_val > 0, (close - lowest) / range_val * 100, 50.0) return pd.Series(pos, index=close.index) def rsi(close: pd.Series, period: int = 14) -> pd.Series: """Relative Strength Index.""" delta = close.diff() gain = delta.where(delta > 0, 0.0) loss = (-delta).where(delta < 0, 0.0) avg_gain = gain.ewm(alpha=1/period, min_periods=period, adjust=False).mean() avg_loss = loss.ewm(alpha=1/period, min_periods=period, adjust=False).mean() rs = avg_gain / avg_loss.replace(0, np.nan) return 100 - (100 / (1 + rs)) def calculate_atr(df: pd.DataFrame, period: int = 14) -> pd.Series: """Average True Range.""" high = df['high'] low = df['low'] close = df['close'] tr1 = high - low tr2 = abs(high - close.shift(1)) tr3 = abs(low - close.shift(1)) tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) return tr.rolling(window=period).mean() def calculate_adx(df: pd.DataFrame, period: int = 14) -> pd.Series: """Average Directional Index.""" high = df['high'] low = df['low'] close = df['close'] plus_dm = high.diff() minus_dm = -low.diff() plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0) minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0) tr = pd.concat([ high - low, abs(high - close.shift(1)), abs(low - close.shift(1)) ], axis=1).max(axis=1) atr = tr.rolling(window=period).mean() plus_di = 100 * (plus_dm.rolling(window=period).mean() / atr) minus_di = 100 * (minus_dm.rolling(window=period).mean() / atr) dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, np.nan) adx = dx.rolling(window=period).mean() return adx def supertrend_shorts( df: pd.DataFrame, atr_period: int = 12, multiplier: float = 3.8, flip_threshold: float = 0.0, confirm_bars: int = 1 ) -> Tuple[pd.Series, pd.Series]: """ Calculate SuperTrend and trend direction. Optimized numpy version for speed. Returns: tsl: Trailing stop line trend: 1 for bullish, -1 for bearish """ close = df['close'].values high = df['high'].values low = df['low'].values n = len(close) # Calculate ATR tr = np.maximum( high - low, np.maximum( np.abs(high - np.roll(close, 1)), np.abs(low - np.roll(close, 1)) ) ) tr[0] = high[0] - low[0] atr = pd.Series(tr).rolling(window=atr_period).mean().values src = (high + low) / 2 up = src - (multiplier * atr) dn = src + (multiplier * atr) up1 = up.copy() dn1 = dn.copy() trend = np.ones(n, dtype=int) tsl = up1.copy() bull_momentum = np.zeros(n, dtype=int) bear_momentum = np.zeros(n, dtype=int) threshold = flip_threshold / 100.0 confirm_threshold = confirm_bars + 1 for i in range(1, n): if close[i-1] > up1[i-1]: up1[i] = max(up[i], up1[i-1]) else: up1[i] = up[i] if close[i-1] < dn1[i-1]: dn1[i] = min(dn[i], dn1[i-1]) else: dn1[i] = dn[i] if trend[i-1] == 1: tsl[i] = max(up1[i], tsl[i-1]) else: tsl[i] = min(dn1[i], tsl[i-1]) threshold_amount = tsl[i] * threshold if trend[i-1] == 1: if close[i] < (tsl[i] - threshold_amount): bear_momentum[i] = bear_momentum[i-1] + 1 bull_momentum[i] = 0 else: bear_momentum[i] = 0 bull_momentum[i] = 0 trend[i] = -1 if bear_momentum[i] >= confirm_threshold else 1 else: if close[i] > (tsl[i] + threshold_amount): bull_momentum[i] = bull_momentum[i-1] + 1 bear_momentum[i] = 0 else: bull_momentum[i] = 0 bear_momentum[i] = 0 trend[i] = 1 if bull_momentum[i] >= confirm_threshold else -1 return pd.Series(tsl, index=df.index), pd.Series(trend, index=df.index) def generate_short_signals( df: pd.DataFrame, inputs: Optional[ShortOnlyInputs] = None ) -> Tuple[List[ShortSignal], pd.DataFrame]: """ Generate SHORT-ONLY signals from Money Line indicator. IGNORES all LONG signals completely - we only care about SHORTs here. """ if inputs is None: inputs = ShortOnlyInputs() data = df.sort_index().copy() # Calculate SuperTrend supertrend, trend = supertrend_shorts( data, inputs.atr_period, inputs.multiplier, inputs.flip_threshold, inputs.confirm_bars ) data['supertrend'] = supertrend data['trend'] = trend # Calculate indicators data["rsi"] = rsi(data["close"], inputs.rsi_length) data["atr"] = calculate_atr(data, inputs.atr_period) data["adx"] = calculate_adx(data, inputs.adx_length) data["volume_ratio"] = rolling_volume_ratio(data["volume"]) data["price_position"] = price_position(data["high"], data["low"], data["close"]) signals: List[ShortSignal] = [] cooldown_remaining = 0 warmup_bars = 200 for idx in range(max(1, warmup_bars), len(data)): row = data.iloc[idx] prev = data.iloc[idx - 1] # ONLY detect SHORT signals (trend flip from bullish to bearish) flip_short = prev.trend == 1 and row.trend == -1 if not flip_short: continue if cooldown_remaining > 0: cooldown_remaining -= 1 continue # Apply quality filters if enabled if not inputs.use_quality_filters: # No filters - just trend flip signals.append( ShortSignal( timestamp=row.name, entry_price=float(row.close), adx=float(row.adx), atr=float(row.atr), rsi=float(row.rsi), volume_ratio=float(row.volume_ratio), price_position=float(row.price_position), ) ) cooldown_remaining = inputs.cooldown_bars continue # === ALL FILTERS FOR SHORTS === # ADX filter (trend strength) if inputs.adx_min > 0: adx_ok = row.adx >= inputs.adx_min else: adx_ok = True # Volume filter if inputs.vol_min > 0: volume_ok = inputs.vol_min <= row.volume_ratio <= inputs.vol_max else: volume_ok = row.volume_ratio <= inputs.vol_max # Entry buffer check (price must be below line by X*ATR) if inputs.entry_buffer_atr != 0: # Negative entry_buffer_atr means early entry (price doesn't need to be far below) entry_buffer_ok = row.close < (row.supertrend - inputs.entry_buffer_atr * row.atr) else: entry_buffer_ok = True # RSI filter for shorts rsi_ok = inputs.rsi_short_min <= row.rsi <= inputs.rsi_short_max # Price position filter (avoid catching falling knives at bottom) if inputs.short_pos_min > 0: pos_ok = row.price_position > inputs.short_pos_min else: pos_ok = True # ALL filters must pass if adx_ok and volume_ok and rsi_ok and pos_ok and entry_buffer_ok: signals.append( ShortSignal( timestamp=row.name, entry_price=float(row.close), adx=float(row.adx), atr=float(row.atr), rsi=float(row.rsi), volume_ratio=float(row.volume_ratio), price_position=float(row.price_position), ) ) cooldown_remaining = inputs.cooldown_bars return signals, data def simulate_shorts_only( df: pd.DataFrame, symbol: str, inputs: Optional[ShortOnlyInputs] = None, config: Optional[TradeConfig] = None, ) -> SimulationResult: """ Run backtest simulation for SHORT signals only. """ if inputs is None: inputs = ShortOnlyInputs() if config is None: config = TradeConfig() data = df.sort_index().copy() index_positions = {ts: idx for idx, ts in enumerate(data.index)} signals, enriched = generate_short_signals(data, inputs) trades = [] next_available_index = 0 for signal in signals: start_idx = index_positions.get(signal.timestamp) if start_idx is None or start_idx < next_available_index: continue # Convert ShortSignal to format expected by simulator from backtester.v11_moneyline_all_filters import MoneyLineV11Signal v11_signal = MoneyLineV11Signal( timestamp=signal.timestamp, direction="short", entry_price=signal.entry_price, adx=signal.adx, atr=signal.atr, rsi=signal.rsi, volume_ratio=signal.volume_ratio, price_position=signal.price_position, ) trade = _simulate_trade(data, start_idx, v11_signal, symbol, config) if trade is None: continue trades.append(trade) next_available_index = trade._exit_index return SimulationResult(trades=trades) def run_short_sweep(df: pd.DataFrame, symbol: str = "SOLUSDT") -> pd.DataFrame: """ Run parameter sweep specifically for SHORT optimization. Sweep Parameters: - rsi_short_min: [30, 35, 40, 45] - rsi_short_max: [60, 65, 70, 75] - short_pos_min: [5, 15, 25, 35] - adx_min: [10, 15, 20, 25] - entry_buffer_atr: [-0.15, 0.0, 0.10, 0.20] Total combinations: 4 x 4 x 4 x 4 x 4 = 1,024 """ # Parameter grid for shorts rsi_short_min_values = [30, 35, 40, 45] rsi_short_max_values = [60, 65, 70, 75] short_pos_min_values = [5, 15, 25, 35] adx_min_values = [10, 15, 20, 25] entry_buffer_values = [-0.15, 0.0, 0.10, 0.20] results = [] total_combos = ( len(rsi_short_min_values) * len(rsi_short_max_values) * len(short_pos_min_values) * len(adx_min_values) * len(entry_buffer_values) ) print(f"šŸ” SHORT PARAMETER SWEEP") print(f" Total combinations: {total_combos}") print(f" Symbol: {symbol}") print(f" Data range: {df.index[0]} to {df.index[-1]}") print() combo_count = 0 for rsi_min in rsi_short_min_values: for rsi_max in rsi_short_max_values: # Skip invalid RSI ranges if rsi_min >= rsi_max: continue for pos_min in short_pos_min_values: for adx_min in adx_min_values: for entry_buffer in entry_buffer_values: combo_count += 1 inputs = ShortOnlyInputs( rsi_short_min=rsi_min, rsi_short_max=rsi_max, short_pos_min=pos_min, adx_min=adx_min, entry_buffer_atr=entry_buffer, ) config = TradeConfig( tp1_percent=1.45, # Same as longs tp1_size=100, # Full close at TP1 sl_percent=2.0, ) result = simulate_shorts_only(df, symbol, inputs, config) results.append({ 'rsi_short_min': rsi_min, 'rsi_short_max': rsi_max, 'short_pos_min': pos_min, 'adx_min': adx_min, 'entry_buffer_atr': entry_buffer, 'total_trades': result.total_trades, 'win_rate': round(result.win_rate * 100, 1), 'total_pnl': round(result.total_pnl, 2), 'avg_pnl': round(result.average_pnl, 2), 'profit_factor': round(result.profit_factor, 3) if result.profit_factor else 0, 'max_drawdown': round(result.max_drawdown, 2), }) if combo_count % 100 == 0: print(f" Progress: {combo_count}/{total_combos} ({100*combo_count/total_combos:.1f}%)") results_df = pd.DataFrame(results) # Sort by total PnL descending results_df = results_df.sort_values('total_pnl', ascending=False) return results_df def load_data(filepath: str = None) -> pd.DataFrame: """Load OHLCV data from CSV.""" if filepath is None: # Default data location filepath = Path(__file__).parent / "data" / "solusdt_5m.csv" print(f"šŸ“Š Loading data from: {filepath}") df = pd.read_csv(filepath) # Handle timestamp column if 'timestamp' in df.columns: df['timestamp'] = pd.to_datetime(df['timestamp']) df = df.set_index('timestamp') elif 'datetime' in df.columns: df['datetime'] = pd.to_datetime(df['datetime']) df = df.set_index('datetime') elif 'time' in df.columns: df['time'] = pd.to_datetime(df['time']) df = df.set_index('time') # Ensure required columns required = ['open', 'high', 'low', 'close', 'volume'] df.columns = df.columns.str.lower() for col in required: if col not in df.columns: raise ValueError(f"Missing required column: {col}") print(f" Rows: {len(df):,}") print(f" Date range: {df.index[0]} to {df.index[-1]}") return df def main(): parser = argparse.ArgumentParser(description='V11 Money Line SHORTS ONLY Backtester') parser.add_argument('--sweep', action='store_true', help='Run full parameter sweep for shorts') parser.add_argument('--baseline', action='store_true', help='Test with default short settings') parser.add_argument('--custom', action='store_true', help='Test with custom parameters') # Custom parameters parser.add_argument('--rsi-min', type=float, default=30.0, help='RSI Short Minimum (default: 30)') parser.add_argument('--rsi-max', type=float, default=70.0, help='RSI Short Maximum (default: 70)') parser.add_argument('--pos-min', type=float, default=5.0, help='Short Position Minimum %% (default: 5)') parser.add_argument('--adx-min', type=float, default=15.0, help='ADX Minimum (default: 15)') parser.add_argument('--entry-buffer', type=float, default=-0.15, help='Entry Buffer ATR (default: -0.15)') parser.add_argument('--data', type=str, default=None, help='Path to OHLCV CSV file') parser.add_argument('--output', type=str, default=None, help='Output CSV file for sweep results') args = parser.parse_args() # Load data df = load_data(args.data) print() print("=" * 60) print(" V11 MONEY LINE - SHORTS ONLY BACKTESTER") print("=" * 60) print() if args.sweep: print("šŸ” Running SHORT parameter sweep...") print() results = run_short_sweep(df) # Save results output_file = args.output or "sweep_v11_shorts_only.csv" results.to_csv(output_file, index=False) print(f"\nšŸ’¾ Results saved to: {output_file}") # Show top 10 results print("\n" + "=" * 60) print(" TOP 10 SHORT CONFIGURATIONS") print("=" * 60) print() top10 = results.head(10) for i, row in top10.iterrows(): print(f"#{results.index.get_loc(i)+1}:") print(f" RSI: {row['rsi_short_min']}-{row['rsi_short_max']}") print(f" Position Min: {row['short_pos_min']}%") print(f" ADX Min: {row['adx_min']}") print(f" Entry Buffer: {row['entry_buffer_atr']}") print(f" Results: {row['total_trades']} trades, {row['win_rate']}% WR, ${row['total_pnl']} PnL") print(f" PF: {row['profit_factor']}, Max DD: ${row['max_drawdown']}") print() elif args.baseline: print("šŸ“Š Testing BASELINE short settings...") print(" RSI: 30-70") print(" Position Min: 5%") print(" ADX Min: 15") print(" Entry Buffer: -0.15") print() inputs = ShortOnlyInputs() # Default settings config = TradeConfig(tp1_percent=1.45, tp1_size=100, sl_percent=2.0) result = simulate_shorts_only(df, "SOLUSDT", inputs, config) print("=" * 60) print(" BASELINE RESULTS") print("=" * 60) print(f" Total Trades: {result.total_trades}") print(f" Win Rate: {result.win_rate*100:.1f}%") print(f" Total PnL: ${result.total_pnl:.2f}") print(f" Avg PnL: ${result.average_pnl:.2f}") print(f" Profit Factor: {result.profit_factor:.3f}") print(f" Max Drawdown: ${result.max_drawdown:.2f}") elif args.custom: print("šŸ“Š Testing CUSTOM short settings...") print(f" RSI: {args.rsi_min}-{args.rsi_max}") print(f" Position Min: {args.pos_min}%") print(f" ADX Min: {args.adx_min}") print(f" Entry Buffer: {args.entry_buffer}") print() inputs = ShortOnlyInputs( rsi_short_min=args.rsi_min, rsi_short_max=args.rsi_max, short_pos_min=args.pos_min, adx_min=args.adx_min, entry_buffer_atr=args.entry_buffer, ) config = TradeConfig(tp1_percent=1.45, tp1_size=100, sl_percent=2.0) result = simulate_shorts_only(df, "SOLUSDT", inputs, config) print("=" * 60) print(" CUSTOM RESULTS") print("=" * 60) print(f" Total Trades: {result.total_trades}") print(f" Win Rate: {result.win_rate*100:.1f}%") print(f" Total PnL: ${result.total_pnl:.2f}") print(f" Avg PnL: ${result.average_pnl:.2f}") print(f" Profit Factor: {result.profit_factor:.3f}") print(f" Max Drawdown: ${result.max_drawdown:.2f}") else: parser.print_help() print("\nāŒ Please specify --sweep, --baseline, or --custom") if __name__ == "__main__": main()