fix: Add TP1 TP2 SL prices to Telegram trade notifications
This commit is contained in:
mindesbunister
766
backtester/v11_shorts_only.py
Normal file
766
backtester/v11_shorts_only.py
Normal file
@@ -0,0 +1,766 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user