1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|
"""Combined strategy that aggregates signals from multiple sub-strategies."""
from decimal import Decimal
from shared.models import Candle, Signal, OrderSide
from strategies.base import BaseStrategy
class CombinedStrategy(BaseStrategy):
"""Combines multiple strategies using weighted signal voting.
Each sub-strategy votes BUY (+weight), SELL (-weight), or HOLD (0).
The combined signal fires when the weighted sum exceeds a threshold.
"""
name: str = "combined"
def __init__(self) -> None:
super().__init__()
self._strategies: list[tuple[BaseStrategy, float]] = [] # (strategy, weight)
self._threshold: float = 0.5
self._quantity: Decimal = Decimal("0.01")
@property
def warmup_period(self) -> int:
if not self._strategies:
return 0
return max(s.warmup_period for s, _ in self._strategies)
def configure(self, params: dict) -> None:
self._threshold = float(params.get("threshold", 0.5))
self._quantity = Decimal(str(params.get("quantity", "0.01")))
if self._threshold <= 0:
raise ValueError(f"Threshold must be positive, got {self._threshold}")
if self._quantity <= 0:
raise ValueError(f"Quantity must be positive, got {self._quantity}")
def add_strategy(self, strategy: BaseStrategy, weight: float = 1.0) -> None:
"""Add a sub-strategy with a weight."""
if weight <= 0:
raise ValueError(f"Weight must be positive, got {weight}")
self._strategies.append((strategy, weight))
def reset(self) -> None:
for strategy, _ in self._strategies:
strategy.reset()
def on_candle(self, candle: Candle) -> Signal | None:
if not self._strategies:
return None
total_weight = sum(w for _, w in self._strategies)
if total_weight == 0:
return None
score = 0.0
reasons = []
for strategy, weight in self._strategies:
signal = strategy.on_candle(candle)
if signal is not None:
if signal.side == OrderSide.BUY:
score += weight * signal.conviction
reasons.append(f"{strategy.name}:BUY({weight}*{signal.conviction:.2f})")
elif signal.side == OrderSide.SELL:
score -= weight * signal.conviction
reasons.append(f"{strategy.name}:SELL({weight}*{signal.conviction:.2f})")
normalized = score / total_weight # Range: -1.0 to 1.0
if normalized >= self._threshold:
return Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.BUY,
price=candle.close,
quantity=self._quantity,
reason=f"Combined score {normalized:.2f} >= {self._threshold} [{', '.join(reasons)}]",
)
elif normalized <= -self._threshold:
return Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.SELL,
price=candle.close,
quantity=self._quantity,
reason=f"Combined score {normalized:.2f} <= -{self._threshold} [{', '.join(reasons)}]",
)
return None
|
