root/trading_bot_v4
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add v11 test sweep system (256 combinations) with office hours scheduling

Browse Source 
Co-authored-by: mindesbunister <32161838+mindesbunister@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2025-12-06 19:15:54 +00:00
parent 67cc7598f2
commit eb0d41aed5
7 changed files with 1113 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
backtester/__pycache__/v11_moneyline_all_filters.cpython-312.pyc Normal file
View File

Binary file not shown.

321
backtester/v11_moneyline_all_filters.py Normal file
View File

@@ -0,0 +1,321 @@
"""
v11 "Money Line All Filters" indicator implementation for backtesting.
CRITICAL DIFFERENCE FROM v9:
- v11: ALL filters actually applied to signals (useQualityFilters toggle)
- v9 bug: Filters calculated but signals ignored them
Based on moneyline_v11_all_filters.pinescript lines 271-272:
finalLongSignal = buyReady and (not useQualityFilters or (longOk and adxOk and longBufferOk and longPositionOk and volumeOk and rsiLongOk))
finalShortSignal = sellReady and (not useQualityFilters or (shortOk and adxOk and shortBufferOk and shortPositionOk and volumeOk and rsiShortOk))
Test sweep parameters (8 params × 2 values = 256 combinations):
- flip_threshold: 0.5, 0.6
- adx_min: 18, 21
- long_pos_max: 75, 80
- short_pos_min: 20, 25
- vol_min: 0.8, 1.0
- entry_buffer_atr: 0.15, 0.20
- rsi_long_min: 35, 40
- rsi_short_max: 65, 70
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
import numpy as np
import pandas as pd
from backtester.math_utils import calculate_adx, calculate_atr, rma
Direction = Literal["long", "short"]
@dataclass
class MoneyLineV11Inputs:
"""v11 Money Line indicator parameters for test sweep."""
# Basic Money Line parameters (fixed for test)
confirm_bars: int = 0 # Immediate signals
cooldown_bars: int = 3 # Prevent overtrading
# ATR profile (fixed for test - 5-minute chart defaults)
atr_period: int = 12 # ATR calculation length
multiplier: float = 3.8 # ATR band multiplier
# Filter parameters (8 parameters being optimized)
flip_threshold: float = 0.5 # % price must move to flip (TEST: 0.5, 0.6)
adx_min: float = 21 # Minimum ADX for signal (TEST: 18, 21)
long_pos_max: float = 75 # Don't long above X% of range (TEST: 75, 80)
short_pos_min: float = 20 # Don't short below X% of range (TEST: 20, 25)
vol_min: float = 1.0 # Minimum volume ratio (TEST: 0.8, 1.0)
entry_buffer_atr: float = 0.20 # ATR buffer beyond line (TEST: 0.15, 0.20)
rsi_long_min: float = 35 # RSI minimum for longs (TEST: 35, 40)
rsi_short_max: float = 70 # RSI maximum for shorts (TEST: 65, 70)
# Fixed filter parameters (not being optimized in test)
adx_length: int = 16 # ADX calculation length
rsi_length: int = 14 # RSI calculation length
vol_max: float = 3.5 # Maximum volume ratio
rsi_long_max: float = 70 # RSI maximum for longs
rsi_short_min: float = 30 # RSI minimum for shorts
@dataclass
class MoneyLineV11Signal:
timestamp: pd.Timestamp
direction: Direction
entry_price: float
adx: float
atr: float
rsi: float
volume_ratio: float
price_position: float
def ema(series: pd.Series, length: int) -> pd.Series:
"""Exponential Moving Average."""
return series.ewm(span=length, adjust=False).mean()
def rolling_volume_ratio(volume: pd.Series, length: int = 20) -> pd.Series:
"""Volume ratio vs moving average."""
avg = volume.rolling(length).mean()
return volume / avg
def price_position(high: pd.Series, low: pd.Series, close: pd.Series, length: int = 100) -> pd.Series:
"""Price position in percentage of range (0-100)."""
highest = high.rolling(length).max()
lowest = low.rolling(length).min()
return 100.0 * (close - lowest) / (highest - lowest)
def rsi(series: pd.Series, length: int) -> pd.Series:
"""Relative Strength Index."""
delta = series.diff()
gain = np.where(delta > 0, delta, 0.0)
loss = np.where(delta < 0, -delta, 0.0)
avg_gain = rma(pd.Series(gain), length)
avg_loss = rma(pd.Series(loss), length)
rs = avg_gain / avg_loss.replace(0, np.nan)
rsi_series = 100 - (100 / (1 + rs))
return rsi_series.fillna(50.0)
def supertrend_v11(df: pd.DataFrame, atr_period: int, multiplier: float,
flip_threshold: float, confirm_bars: int) -> tuple[pd.Series, pd.Series]:
"""
Calculate v11 Money Line (Supertrend with flip threshold).
Returns:
(supertrend_line, trend): Line values and trend direction (1=bull, -1=bear)
"""
# Use chart prices (not Heikin Ashi for test)
high, low, close = df['high'], df['low'], df['close']
# Calculate ATR
tr = pd.concat([
high - low,
(high - close.shift(1)).abs(),
(low - close.shift(1)).abs()
], axis=1).max(axis=1)
atr = rma(tr, atr_period)
# Supertrend bands
src = (high + low) / 2
up = src - (multiplier * atr)
dn = src + (multiplier * atr)
# Initialize tracking arrays
up1 = up.copy()
dn1 = dn.copy()
trend = pd.Series(1, index=df.index) # Start bullish
tsl = up1.copy() # Trailing stop line
# Momentum tracking for anti-whipsaw
bull_momentum = pd.Series(0, index=df.index)
bear_momentum = pd.Series(0, index=df.index)
# Calculate flip threshold
threshold = flip_threshold / 100.0
for i in range(1, len(df)):
# Update bands
if close.iloc[i-1] > up1.iloc[i-1]:
up1.iloc[i] = max(up.iloc[i], up1.iloc[i-1])
else:
up1.iloc[i] = up.iloc[i]
if close.iloc[i-1] < dn1.iloc[i-1]:
dn1.iloc[i] = min(dn.iloc[i], dn1.iloc[i-1])
else:
dn1.iloc[i] = dn.iloc[i]
# Get previous trend and tsl
prev_trend = trend.iloc[i-1]
prev_tsl = tsl.iloc[i-1]
# Update TSL based on trend
if prev_trend == 1:
tsl.iloc[i] = max(up1.iloc[i], prev_tsl)
else:
tsl.iloc[i] = min(dn1.iloc[i], prev_tsl)
# Check for flip with threshold and momentum
threshold_amount = tsl.iloc[i] * threshold
if prev_trend == 1:
# Currently bullish - check for bearish flip
if close.iloc[i] < (tsl.iloc[i] - threshold_amount):
bear_momentum.iloc[i] = bear_momentum.iloc[i-1] + 1
bull_momentum.iloc[i] = 0
else:
bear_momentum.iloc[i] = 0
bull_momentum.iloc[i] = 0
# Flip after confirm_bars + 1 consecutive bearish bars
if bear_momentum.iloc[i] >= (confirm_bars + 1):
trend.iloc[i] = -1
else:
trend.iloc[i] = 1
else:
# Currently bearish - check for bullish flip
if close.iloc[i] > (tsl.iloc[i] + threshold_amount):
bull_momentum.iloc[i] = bull_momentum.iloc[i-1] + 1
bear_momentum.iloc[i] = 0
else:
bull_momentum.iloc[i] = 0
bear_momentum.iloc[i] = 0
# Flip after confirm_bars + 1 consecutive bullish bars
if bull_momentum.iloc[i] >= (confirm_bars + 1):
trend.iloc[i] = 1
else:
trend.iloc[i] = -1
return tsl, trend
def money_line_v11_signals(df: pd.DataFrame, inputs: Optional[MoneyLineV11Inputs] = None) -> list[MoneyLineV11Signal]:
"""
v11 "Money Line All Filters" signal generation.
CRITICAL: ALL filters applied to signals (this is what makes v11 different from v9 bug).
From pinescript lines 271-272:
finalLongSignal = buyReady and (longOk and adxOk and longBufferOk and longPositionOk and volumeOk and rsiLongOk)
finalShortSignal = sellReady and (shortOk and adxOk and shortBufferOk and shortPositionOk and volumeOk and rsiShortOk)
Filters applied:
- ADX minimum (trend strength)
- Entry buffer (price beyond line by X*ATR)
- Price position (don't chase extremes)
- Volume ratio (avoid dead/overheated)
- RSI boundaries (momentum confirmation)
"""
if inputs is None:
inputs = MoneyLineV11Inputs()
data = df.copy()
data = data.sort_index()
# Calculate Money Line
supertrend, trend = supertrend_v11(
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[MoneyLineV11Signal] = []
cooldown_remaining = 0
# Skip warmup period (200 bars for price position)
warmup_bars = 200
for idx in range(max(1, warmup_bars), len(data)):
row = data.iloc[idx]
prev = data.iloc[idx - 1]
# Detect trend flip (buyReady/sellReady in pinescript)
flip_long = prev.trend == -1 and row.trend == 1
flip_short = prev.trend == 1 and row.trend == -1
if cooldown_remaining > 0:
cooldown_remaining -= 1
continue
# V11 CRITICAL: Apply ALL filters (this is what was broken in v9)
# ADX filter (adxOk)
adx_ok = row.adx >= inputs.adx_min
# Volume filter (volumeOk)
volume_ok = inputs.vol_min <= row.volume_ratio <= inputs.vol_max
if flip_long:
# Entry buffer check (longBufferOk)
entry_buffer_ok = row.close > (row.supertrend + inputs.entry_buffer_atr * row.atr)
# Long filters
rsi_ok = inputs.rsi_long_min <= row.rsi <= inputs.rsi_long_max # rsiLongOk
pos_ok = row.price_position < inputs.long_pos_max # longPositionOk
# V11: ALL filters must pass (this is the fix from v9)
if adx_ok and volume_ok and rsi_ok and pos_ok and entry_buffer_ok:
signals.append(
MoneyLineV11Signal(
timestamp=row.name,
direction="long",
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
elif flip_short:
# Entry buffer check (shortBufferOk)
entry_buffer_ok = row.close < (row.supertrend - inputs.entry_buffer_atr * row.atr)
# Short filters
rsi_ok = inputs.rsi_short_min <= row.rsi <= inputs.rsi_short_max # rsiShortOk
pos_ok = row.price_position > inputs.short_pos_min # shortPositionOk
# V11: ALL filters must pass (this is the fix from v9)
if adx_ok and volume_ok and rsi_ok and pos_ok and entry_buffer_ok:
signals.append(
MoneyLineV11Signal(
timestamp=row.name,
direction="short",
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