Сенкс чел!
Я только что стырил твою стратегию, еще правда не проверил насколько хорошо это работает!
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable, Sequence
import backtrader as bt
@dataclass(frozen=True)
class SignalChoice:
period: int
value: float
score: float
class AdaptiveCompositeMomentum(bt.Indicator):
lines = ("momentum",)
params = dict(
price_period=5,
momentum_periods=(8, 13, 21, 34),
endpoint_window=4,
pattern_window=3,
normalize=True,
smooth_period=5,
)
plotinfo = dict(subplot=True)
plotlines = dict(momentum=dict(color="tab:blue"))
def __init__(self) -> None:
periods = tuple(int(p) for p in self.p.momentum_periods)
self._periods = periods
self._smoothed_price = bt.ind.SMA(self.data, period=int(self.p.price_period))
self.addminperiod(
int(self.p.price_period)
+ max(periods)
+ int(self.p.endpoint_window)
+ int(self.p.pattern_window)
+ int(self.p.smooth_period)
+ 2
)
self._raw_values: list[float] = []
def next(self) -> None:
current_slope = self._slope(0, int(self.p.pattern_window))
parts: list[float] = []
for period in self._periods:
endpoint_ago = self._best_endpoint(period, current_slope)
old_price = float(self._smoothed_price[-endpoint_ago])
cur_price = float(self._smoothed_price[0])
if self.p.normalize:
value = 0.0 if old_price == 0 else (cur_price - old_price) / old_price
else:
value = cur_price - old_price
parts.append(value)
raw = sum(parts) / len(parts)
self._raw_values.append(raw)
smooth_period = int(self.p.smooth_period)
if smooth_period <= 1 or len(self._raw_values) < smooth_period:
self.lines.momentum[0] = raw
else:
self.lines.momentum[0] = sum(self._raw_values[-smooth_period:]) / smooth_period
def _best_endpoint(self, period: int, current_slope: float) -> int:
pattern_window = int(self.p.pattern_window)
endpoint_window = int(self.p.endpoint_window)
best_ago = period
best_error = float("inf")
for ago in range(max(pattern_window + 1, period - endpoint_window), period + endpoint_window + 1):
historical_slope = self._slope(-ago, pattern_window)
error = abs(current_slope - historical_slope)
if error < best_error:
best_error = error
best_ago = ago
return best_ago
def _slope(self, index: int, window: int) -> float:
start = float(self._smoothed_price[index])
end = float(self._smoothed_price[index - window])
base = abs(end) if end else 1.0
return (start - end) / base


