#!/usr/bin/env python3 """Test exact parameters worker is using""" import pandas as pd import sys sys.path.insert(0, 'backtester') from v11_moneyline_all_filters import money_line_v11_signals, MoneyLineV11Inputs # Load dataset print("Loading data...") df = pd.read_csv('data/solusdt_5m.csv') df['timestamp'] = pd.to_datetime(df['timestamp']) df = df.set_index('timestamp') print(f"āœ“ Loaded {len(df)} bars") # Test EXACT worker configuration (most permissive combination) config = { 'flip_threshold': 0.4, # Most permissive 'adx_min': 0, # Filter DISABLED 'long_pos_max': 100, # Most permissive 'short_pos_min': 0, # Filter DISABLED 'vol_min': 0.0, # Filter DISABLED 'entry_buffer_atr': 0.0, # Filter DISABLED 'rsi_long_min': 25, # Permissive 'rsi_short_max': 80, # Permissive } print("\nTest 1: Worker parameters WITH use_quality_filters=True") print(f"Config: {config}") inputs = MoneyLineV11Inputs( use_quality_filters=True, flip_threshold=config['flip_threshold'], adx_min=config['adx_min'], long_pos_max=config['long_pos_max'], short_pos_min=config['short_pos_min'], vol_min=config['vol_min'], entry_buffer_atr=config['entry_buffer_atr'], rsi_long_min=config['rsi_long_min'], rsi_long_max=70, rsi_short_min=30, rsi_short_max=config['rsi_short_max'], ) print(f"use_quality_filters: {inputs.use_quality_filters}") print(f"RSI bounds: long [{inputs.rsi_long_min}, {inputs.rsi_long_max}], short [{inputs.rsi_short_min}, {inputs.rsi_short_max}]") signals = money_line_v11_signals(df, inputs) print(f"RESULT: {len(signals)} signals") if len(signals) == 0: print("\nšŸ”“ ZERO SIGNALS - Same as worker!") print("\nTest 2: Try WITHOUT quality filters") inputs2 = MoneyLineV11Inputs( use_quality_filters=False, # Bypass all filters flip_threshold=0.4, ) signals2 = money_line_v11_signals(df, inputs2) print(f"RESULT: {len(signals2)} signals") if len(signals2) > 0: print("\nāœ“ Filter bypass works - problem is WITH filters") print("Debugging which filter is blocking...") # Check each filter component print("\n First 10 rows of data:") print(df[['open', 'high', 'low', 'close', 'volume']].head(10)) else: print(f"\nāœ… SUCCESS: {len(signals)} signals generated!") print(f" First 3 signals:") for s in signals[:3]: print(f" {s.timestamp} {s.direction} @ ${s.entry_price:.2f}")