767 lines
25 KiB
Python
767 lines
25 KiB
Python
#!/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()
|