remove: V10 momentum system - backtest proved it adds no value

- Removed v10 TradingView indicator (moneyline_v10_momentum_dots.pinescript)
- Removed v10 penalty system from signal-quality.ts (-30/-25 point penalties)
- Removed backtest result files (sweep_*.csv)
- Updated copilot-instructions.md to remove v10 references
- Simplified direction-specific quality thresholds (LONG 90+, SHORT 80+)

Rationale:
- 1,944 parameter combinations tested in backtest
- All top results IDENTICAL (568 trades, $498 P&L, 61.09% WR)
- Momentum parameters had ZERO impact on trade selection
- Profit factor 1.027 too low (barely profitable after fees)
- Max drawdown -$1,270 vs +$498 profit = terrible risk-reward
- v10 penalties were blocking good trades (bug: applied to wrong positions)

Keeping v9 as production system - simpler, proven, effective.
This commit is contained in:
mindesbunister
2025-11-28 22:35:32 +01:00
parent 4fb6a45fab
commit 5f7702469e
5869 changed files with 1116282 additions and 116 deletions

1
backtester/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Lightweight local backtesting utilities for Money Line indicators."""

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

70
backtester/cli.py Normal file
View File

@@ -0,0 +1,70 @@
from __future__ import annotations
import argparse
import dataclasses
from pathlib import Path
from typing import Optional
import pandas as pd
from backtester.data_loader import load_csv
from backtester.indicators.money_line import MoneyLineInputs
from backtester.simulator import TradeConfig, simulate_money_line
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run Money Line backtests against CSV OHLCV data.")
parser.add_argument("--csv", type=Path, required=True, help="Path to CSV file with timestamp, open, high, low, close, volume columns")
parser.add_argument("--symbol", type=str, required=True, help="Symbol label (e.g., SOL-PERP)")
parser.add_argument("--timeframe", type=str, default="5", help="Timeframe label for reference (default: 5)")
parser.add_argument("--start", type=str, default=None, help="Optional ISO timestamp for start filter (e.g., 2024-01-01)")
parser.add_argument("--end", type=str, default=None, help="Optional ISO timestamp for end filter")
parser.add_argument("--position-size", type=float, default=1000.0, dest="position_size", help="Notional size in USD for each simulated trade")
parser.add_argument("--max-bars", type=int, default=None, help="Maximum bars to hold a trade before force exit")
parser.add_argument("--export-trades", type=Path, default=None, help="Optional path to write detailed trade results as CSV")
return parser.parse_args()
def main() -> None:
args = parse_args()
data_slice = load_csv(
path=args.csv,
symbol=args.symbol,
timeframe=args.timeframe,
start=args.start,
end=args.end,
)
df = data_slice.data
if not isinstance(df.index, pd.DatetimeIndex):
df = df.copy()
df.index = pd.to_datetime(df.index)
config = TradeConfig(position_size=args.position_size, max_bars_per_trade=args.max_bars)
inputs = MoneyLineInputs()
result = simulate_money_line(df=df, symbol=data_slice.symbol, inputs=inputs, config=config)
print("=== Backtest Summary ===")
print(f"Symbol: {data_slice.symbol}")
print(f"Rows processed: {len(df)}")
print(f"Trades: {len(result.trades)}")
print(f"Total PnL: ${result.total_pnl:,.2f}")
print(f"Average PnL: ${result.average_pnl:,.2f}")
print(f"Win rate: {result.win_rate * 100:.2f}%")
print(f"Max drawdown: ${result.max_drawdown:,.2f}")
if args.export_trades:
trades_df = pd.DataFrame(
[
{k: v for k, v in dataclasses.asdict(trade).items() if k != "_exit_index"}
for trade in result.trades
]
)
trades_df.to_csv(args.export_trades, index=False)
print(f"Detailed trades exported to {args.export_trades}")
if __name__ == "__main__":
main()

38
backtester/data_loader.py Normal file
View File

@@ -0,0 +1,38 @@
"""Utilities for loading OHLCV data for local backtesting."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
import pandas as pd
@dataclass
class DataSlice:
symbol: str
timeframe: str
data: pd.DataFrame
def load_csv(path: Path, symbol: str, timeframe: str, start: Optional[str] = None, end: Optional[str] = None) -> DataSlice:
if not path.exists():
raise FileNotFoundError(f"Missing data file: {path}")
df = pd.read_csv(path, parse_dates=["timestamp"])
df = df.sort_values("timestamp").reset_index(drop=True)
if start:
df = df[df["timestamp"] >= pd.Timestamp(start)]
if end:
df = df[df["timestamp"] <= pd.Timestamp(end)]
if df.empty:
raise ValueError("No rows remain after applying date filters")
expected_cols = {"timestamp", "open", "high", "low", "close", "volume"}
missing = expected_cols.difference(df.columns)
if missing:
raise ValueError(f"Missing columns in {path}: {sorted(missing)}")
df = df.set_index("timestamp")
return DataSlice(symbol=symbol.upper(), timeframe=timeframe, data=df)

View File

@@ -0,0 +1,176 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
try: # Python 3.8+ has Literal in typing, otherwise fall back to typing_extensions
from typing import Literal
except ImportError: # pragma: no cover - compatibility path for Python 3.7
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 MoneyLineInputs:
atr_length: int = 14
adx_length: int = 14
rsi_length: int = 14
ma_fast_length: int = 50
ma_slow_length: int = 200
ma_gap_threshold: float = 0.35
flip_threshold_percent: float = 0.6
cooldown_bars: int = 3
momentum_spacing: int = 4
momentum_cooldown: int = 3
momentum_min_adx: float = 23.0
momentum_min_volume_ratio: float = 1.0
momentum_long_max_pos: float = 70.0
momentum_short_min_pos: float = 30.0
@dataclass
class MoneyLineSignal:
timestamp: pd.Timestamp
direction: Direction
entry_price: float
adx: float
atr: float
rsi: float
volume_ratio: float
price_position: float
signal_type: Literal["primary", "momentum"]
def ema(series: pd.Series, length: int) -> pd.Series:
return series.ewm(span=length, adjust=False).mean()
def rolling_volume_ratio(volume: pd.Series, length: int = 20) -> pd.Series:
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:
highest = high.rolling(length).max()
lowest = low.rolling(length).min()
return 100.0 * (close - lowest) / (highest - lowest)
def money_line_signals(df: pd.DataFrame, inputs: Optional[MoneyLineInputs] = None) -> list[MoneyLineSignal]:
if inputs is None:
inputs = MoneyLineInputs()
data = df.copy()
data = data.sort_index()
data["ema_fast"] = ema(data["close"], inputs.ma_fast_length)
data["ema_slow"] = ema(data["close"], inputs.ma_slow_length)
data["rsi"] = rsi(data["close"], inputs.rsi_length)
data["atr"] = calculate_atr(data, inputs.atr_length)
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"])
ma_gap = 100.0 * (data["ema_fast"] - data["ema_slow"]) / data["close"]
ma_gap_score = np.tanh(ma_gap / inputs.ma_gap_threshold)
signals: list[MoneyLineSignal] = []
last_direction: Optional[Direction] = None
cooldown_remaining = 0
momentum_cooldown = 0
for idx in range(1, len(data)):
row = data.iloc[idx]
prev = data.iloc[idx - 1]
close = row.close
fast = row.ema_fast
slow = row.ema_slow
gap_score = ma_gap_score.iloc[idx]
flip_up = prev.close <= prev.ema_fast and close > fast
flip_down = prev.close >= prev.ema_fast and close < fast
direction: Optional[Direction] = None
if flip_up and gap_score > inputs.flip_threshold_percent / 100.0:
direction = "long"
elif flip_down and gap_score < -inputs.flip_threshold_percent / 100.0:
direction = "short"
if direction and cooldown_remaining == 0:
signals.append(
MoneyLineSignal(
timestamp=row.name,
direction=direction,
entry_price=float(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),
signal_type="primary",
)
)
last_direction = direction
cooldown_remaining = inputs.cooldown_bars
momentum_cooldown = inputs.momentum_cooldown
else:
cooldown_remaining = max(0, cooldown_remaining - 1)
momentum_cooldown = max(0, momentum_cooldown - 1)
if (
last_direction
and momentum_cooldown == 0
and row.adx >= inputs.momentum_min_adx
and row.volume_ratio >= inputs.momentum_min_volume_ratio
):
pos = row.price_position
if last_direction == "long" and pos <= inputs.momentum_long_max_pos:
signals.append(
MoneyLineSignal(
timestamp=row.name,
direction="long",
entry_price=float(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),
signal_type="momentum",
)
)
momentum_cooldown = inputs.momentum_spacing
elif last_direction == "short" and pos >= inputs.momentum_short_min_pos:
signals.append(
MoneyLineSignal(
timestamp=row.name,
direction="short",
entry_price=float(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),
signal_type="momentum",
)
)
momentum_cooldown = inputs.momentum_spacing
return signals
def rsi(series: pd.Series, length: int) -> pd.Series:
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)

46
backtester/math_utils.py Normal file
View File

@@ -0,0 +1,46 @@
from __future__ import annotations
import numpy as np
import pandas as pd
def rma(series: pd.Series, length: int) -> pd.Series:
alpha = 1.0 / length
result = series.astype(float).copy()
for i in range(1, len(series)):
prev = result.iat[i - 1]
curr = series.iat[i]
result.iat[i] = alpha * curr + (1 - alpha) * prev
return result
def calculate_atr(df: pd.DataFrame, length: int) -> pd.Series:
high, low, close = df["high"], df["low"], df["close"]
tr = pd.concat([
(high - low),
(high - close.shift(1)).abs(),
(low - close.shift(1)).abs(),
], axis=1).max(axis=1)
tr.iloc[0] = (high.iloc[0] - low.iloc[0])
return rma(tr, length)
def calculate_adx(df: pd.DataFrame, length: int) -> pd.Series:
high, low, close = df["high"], df["low"], df["close"]
up_move = high.diff()
down_move = -low.diff()
plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0.0)
minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0.0)
tr = pd.concat([
(high - low),
(high - close.shift(1)).abs(),
(low - close.shift(1)).abs(),
], axis=1).max(axis=1)
tr.iloc[0] = (high.iloc[0] - low.iloc[0])
atr = rma(tr, length)
plus_di = 100.0 * rma(pd.Series(plus_dm), length) / atr
minus_di = 100.0 * rma(pd.Series(minus_dm), length) / atr
dx = 100.0 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan)
dx = dx.fillna(0.0)
return rma(dx, length)

View File

@@ -0,0 +1,2 @@
pandas>=2.2
numpy>=1.26

387
backtester/simulator.py Normal file
View File

@@ -0,0 +1,387 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Callable, List, Optional
import numpy as np
import pandas as pd
from backtester.indicators.money_line import (
Direction,
MoneyLineInputs,
MoneyLineSignal,
money_line_signals,
)
QualityFilter = Callable[[MoneyLineSignal], bool]
@dataclass
class TradeConfig:
position_size: float = 1000.0
take_profit_1_size_percent: float = 60.0
atr_multiplier_tp1: float = 2.0
atr_multiplier_tp2: float = 4.0
atr_multiplier_sl: float = 3.0
min_tp1_percent: float = 0.5
max_tp1_percent: float = 1.5
min_tp2_percent: float = 1.0
max_tp2_percent: float = 3.0
min_sl_percent: float = 0.8
max_sl_percent: float = 2.0
fallback_tp1_percent: float = 0.8
fallback_tp2_percent: float = 1.7
fallback_sl_percent: float = 1.3
trailing_atr_multiplier: float = 1.5
trailing_min_percent: float = 0.25
trailing_max_percent: float = 0.9
max_bars_per_trade: Optional[int] = None
@dataclass
class SimulatedTrade:
symbol: str
direction: Direction
signal_type: str
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
tp2_hit: bool
trailing_active: bool
mae_percent: float
mfe_percent: float
quality_score: Optional[float] = None
adx_at_entry: Optional[float] = None
atr_at_entry: Optional[float] = None
runner_size: float = 0.0
tp1_size: float = 0.0
_exit_index: int = field(repr=False, default=-1)
@dataclass
class SimulationResult:
trades: List[SimulatedTrade]
@property
def total_pnl(self) -> float:
return sum(t.realized_pnl for t in self.trades)
@property
def win_rate(self) -> float:
wins = sum(1 for t in self.trades if t.realized_pnl > 0)
return 0.0 if not self.trades else wins / len(self.trades)
@property
def average_pnl(self) -> float:
return 0.0 if not self.trades else self.total_pnl / len(self.trades)
@property
def max_drawdown(self) -> float:
equity = 0.0
peak = 0.0
max_dd = 0.0
for trade in self.trades:
equity += trade.realized_pnl
peak = max(peak, equity)
max_dd = min(max_dd, equity - peak)
return max_dd
def simulate_money_line(
df: pd.DataFrame,
symbol: str,
inputs: Optional[MoneyLineInputs] = None,
config: Optional[TradeConfig] = None,
quality_filter: Optional[QualityFilter] = None,
) -> SimulationResult:
if inputs is None:
inputs = MoneyLineInputs()
if config is None:
config = TradeConfig()
if quality_filter is None:
quality_filter = lambda _: True # type: ignore
data = df.sort_index().copy()
index_positions = {ts: idx for idx, ts in enumerate(data.index)}
signals = money_line_signals(data, inputs)
trades: List[SimulatedTrade] = []
next_available_index = 0
for signal in signals:
if signal.timestamp not in index_positions:
continue
start_idx = index_positions[signal.timestamp]
if start_idx < next_available_index:
continue
if not quality_filter(signal):
continue
trade = _simulate_trade(data, start_idx, signal, symbol, config)
if trade is None:
continue
trades.append(trade)
next_available_index = trade._exit_index
return SimulationResult(trades=trades)
def _simulate_trade(
data: pd.DataFrame,
start_idx: int,
signal: MoneyLineSignal,
symbol: str,
config: TradeConfig,
) -> Optional[SimulatedTrade]:
if start_idx >= len(data) - 1:
return None
entry_price = float(signal.entry_price)
if not np.isfinite(entry_price) or entry_price <= 0:
return None
tp1_percent = _percent_from_atr(
signal.atr,
entry_price,
config.atr_multiplier_tp1,
config.min_tp1_percent,
config.max_tp1_percent,
config.fallback_tp1_percent,
)
tp2_percent = _percent_from_atr(
signal.atr,
entry_price,
config.atr_multiplier_tp2,
config.min_tp2_percent,
config.max_tp2_percent,
config.fallback_tp2_percent,
)
sl_percent = _percent_from_atr(
signal.atr,
entry_price,
config.atr_multiplier_sl,
config.min_sl_percent,
config.max_sl_percent,
config.fallback_sl_percent,
)
direction = signal.direction
tp1_price = _target_price(entry_price, tp1_percent, direction)
tp2_price = _target_price(entry_price, tp2_percent, direction)
stop_price = _stop_price(entry_price, sl_percent, direction)
tp1_fraction = config.take_profit_1_size_percent / 100.0
tp1_fraction = np.clip(tp1_fraction, 0.0, 1.0)
tp1_size = config.position_size * tp1_fraction
runner_size = config.position_size - tp1_size
tp1_hit = False
tp2_hit = False
trailing_active = False
remaining_size = config.position_size
realized_pnl = 0.0
exit_reason = "TIME"
exit_price = entry_price
exit_idx = start_idx
bars_held = 0
mae = 0.0
mfe = 0.0
runner_stop_percent = _runner_stop_offset(signal.adx)
runner_stop_price = _stop_price(entry_price, runner_stop_percent, direction)
trailing_stop_price = runner_stop_price
favorable_price = entry_price
max_bars = config.max_bars_per_trade or len(data)
for idx in range(start_idx + 1, len(data)):
bar = data.iloc[idx]
bar_high = float(bar.high)
bar_low = float(bar.low)
bars_held += 1
mae = min(mae, _profit_percent(bar_low if direction == "long" else bar_high, entry_price, direction))
mfe = max(mfe, _profit_percent(bar_high if direction == "long" else bar_low, entry_price, direction))
if not tp1_hit:
if _stop_hit(bar_low, bar_high, stop_price, direction):
realized_pnl += config.position_size * _profit_percent(stop_price, entry_price, direction) / 100.0
exit_reason = "SL"
exit_price = stop_price
exit_idx = idx
break
if _target_hit(bar_low, bar_high, tp1_price, direction):
tp1_hit = True
if tp1_size > 0:
realized_pnl += tp1_size * _profit_percent(tp1_price, entry_price, direction) / 100.0
remaining_size -= tp1_size
exit_reason = "TP1"
runner_stop_price = _stop_price(entry_price, runner_stop_percent, direction)
trailing_stop_price = runner_stop_price
favorable_price = entry_price
# Continue evaluating same bar for runner logic
else:
if remaining_size <= 0:
exit_reason = "TP1"
exit_price = tp1_price
exit_idx = idx
break
if _stop_hit(bar_low, bar_high, runner_stop_price, direction):
realized_pnl += remaining_size * _profit_percent(runner_stop_price, entry_price, direction) / 100.0
exit_reason = "BREAKEVEN" if runner_stop_percent == 0 else "RUNNER_SL"
exit_price = runner_stop_price
exit_idx = idx
break
if (not tp2_hit) and _target_hit(bar_low, bar_high, tp2_price, direction):
tp2_hit = True
trailing_active = True
exit_reason = "TP2"
favorable_price = _update_favorable_price(favorable_price, bar_high, bar_low, direction)
trailing_stop_price = _compute_trailing_stop(
favorable_price,
entry_price,
signal.atr,
config,
direction,
)
if trailing_active:
favorable_price = _update_favorable_price(favorable_price, bar_high, bar_low, direction)
trailing_stop_price = _compute_trailing_stop(
favorable_price,
entry_price,
signal.atr,
config,
direction,
)
if _stop_hit(bar_low, bar_high, trailing_stop_price, direction):
realized_pnl += remaining_size * _profit_percent(trailing_stop_price, entry_price, direction) / 100.0
exit_reason = "TRAILING_SL"
exit_price = trailing_stop_price
exit_idx = idx
break
if bars_held >= max_bars:
exit_price = float(bar.close)
realized_pnl += remaining_size * _profit_percent(exit_price, entry_price, direction) / 100.0
exit_reason = "MAX_TIME"
exit_idx = idx
break
else:
final_bar = data.iloc[-1]
exit_price = float(final_bar.close)
realized_pnl += remaining_size * _profit_percent(exit_price, entry_price, direction) / 100.0
exit_reason = "END"
exit_idx = len(data) - 1
exit_time = data.index[exit_idx]
profit_percent = realized_pnl / config.position_size * 100 if config.position_size else 0.0
return SimulatedTrade(
symbol=symbol,
direction=direction,
signal_type=signal.signal_type,
entry_time=signal.timestamp,
exit_time=exit_time,
entry_price=entry_price,
exit_price=exit_price,
realized_pnl=realized_pnl,
profit_percent=profit_percent,
exit_reason=exit_reason,
bars_held=bars_held,
tp1_hit=tp1_hit,
tp2_hit=tp2_hit,
trailing_active=trailing_active,
mae_percent=mae,
mfe_percent=mfe,
quality_score=None,
adx_at_entry=signal.adx,
atr_at_entry=signal.atr,
runner_size=runner_size,
tp1_size=tp1_size,
_exit_index=exit_idx,
)
def _percent_from_atr(
atr_value: float,
price: float,
multiplier: float,
min_percent: float,
max_percent: float,
fallback: float,
) -> float:
if price <= 0:
return fallback
atr_percent = (atr_value / price) * 100 if price else 0.0
if atr_percent == 0:
return fallback
percent = atr_percent * multiplier
return float(np.clip(percent, min_percent, max_percent))
def _target_price(entry: float, percent: float, direction: Direction) -> float:
if direction == "long":
return entry * (1 + percent / 100.0)
return entry * (1 - percent / 100.0)
def _stop_price(entry: float, percent: float, direction: Direction) -> float:
if direction == "long":
return entry * (1 - percent / 100.0)
return entry * (1 + percent / 100.0)
def _stop_hit(bar_low: float, bar_high: float, stop_price: float, direction: Direction) -> bool:
if direction == "long":
return bar_low <= stop_price
return bar_high >= stop_price
def _target_hit(bar_low: float, bar_high: float, target_price: float, direction: Direction) -> bool:
if direction == "long":
return bar_high >= target_price
return bar_low <= target_price
def _profit_percent(price: float, entry: float, direction: Direction) -> float:
if entry == 0:
return 0.0
if direction == "long":
return (price - entry) / entry * 100.0
return (entry - price) / entry * 100.0
def _runner_stop_offset(adx: float) -> float:
if adx < 20:
return 0.0
if adx < 25:
return 0.3
return 0.55
def _update_favorable_price(current: float, bar_high: float, bar_low: float, direction: Direction) -> float:
if direction == "long":
return max(current, bar_high)
return min(current, bar_low)
def _compute_trailing_stop(
favorable_price: float,
entry_price: float,
atr_value: float,
config: TradeConfig,
direction: Direction,
) -> float:
atr_percent = (atr_value / entry_price) * 100 if entry_price else 0.0
trail_percent = atr_percent * config.trailing_atr_multiplier
trail_percent = float(np.clip(trail_percent, config.trailing_min_percent, config.trailing_max_percent))
if direction == "long":
return favorable_price * (1 - trail_percent / 100.0)
return favorable_price * (1 + trail_percent / 100.0)