diff options
Diffstat (limited to 'services/strategy-engine/strategies/combined_strategy.py')
| -rw-r--r-- | services/strategy-engine/strategies/combined_strategy.py | 39 |
1 files changed, 34 insertions, 5 deletions
diff --git a/services/strategy-engine/strategies/combined_strategy.py b/services/strategy-engine/strategies/combined_strategy.py index be1cbed..907d9c5 100644 --- a/services/strategy-engine/strategies/combined_strategy.py +++ b/services/strategy-engine/strategies/combined_strategy.py @@ -20,6 +20,9 @@ class CombinedStrategy(BaseStrategy): self._strategies: list[tuple[BaseStrategy, float]] = [] # (strategy, weight) self._threshold: float = 0.5 self._quantity: Decimal = Decimal("0.01") + self._trade_history: dict[str, list[bool]] = {} # strategy_name -> [win, loss, ...] + self._adaptive_weights: bool = False + self._history_window: int = 20 # Last N signals to evaluate @property def warmup_period(self) -> int: @@ -30,6 +33,8 @@ class CombinedStrategy(BaseStrategy): def configure(self, params: dict) -> None: self._threshold = float(params.get("threshold", 0.5)) self._quantity = Decimal(str(params.get("quantity", "0.01"))) + self._adaptive_weights = bool(params.get("adaptive_weights", False)) + self._history_window = int(params.get("history_window", 20)) if self._threshold <= 0: raise ValueError(f"Threshold must be positive, got {self._threshold}") if self._quantity <= 0: @@ -41,6 +46,29 @@ class CombinedStrategy(BaseStrategy): raise ValueError(f"Weight must be positive, got {weight}") self._strategies.append((strategy, weight)) + def record_result(self, strategy_name: str, is_win: bool) -> None: + """Record a trade result for adaptive weighting.""" + if strategy_name not in self._trade_history: + self._trade_history[strategy_name] = [] + self._trade_history[strategy_name].append(is_win) + # Keep only last N results + if len(self._trade_history[strategy_name]) > self._history_window: + self._trade_history[strategy_name] = self._trade_history[strategy_name][-self._history_window:] + + def _get_adaptive_weight(self, strategy_name: str, base_weight: float) -> float: + """Get weight adjusted by recent performance.""" + if not self._adaptive_weights: + return base_weight + + history = self._trade_history.get(strategy_name, []) + if len(history) < 5: # Not enough data, use base weight + return base_weight + + win_rate = sum(1 for w in history if w) / len(history) + # Scale weight: 0.5x at 20% win rate, 1.0x at 50%, 1.5x at 80% + scale = 0.5 + win_rate # Range: 0.5 to 1.5 + return base_weight * scale + def reset(self) -> None: for strategy, _ in self._strategies: strategy.reset() @@ -49,7 +77,7 @@ class CombinedStrategy(BaseStrategy): if not self._strategies: return None - total_weight = sum(w for _, w in self._strategies) + total_weight = sum(self._get_adaptive_weight(s.name, w) for s, w in self._strategies) if total_weight == 0: return None @@ -59,12 +87,13 @@ class CombinedStrategy(BaseStrategy): for strategy, weight in self._strategies: signal = strategy.on_candle(candle) if signal is not None: + effective_weight = self._get_adaptive_weight(strategy.name, weight) if signal.side == OrderSide.BUY: - score += weight * signal.conviction - reasons.append(f"{strategy.name}:BUY({weight}*{signal.conviction:.2f})") + score += effective_weight * signal.conviction + reasons.append(f"{strategy.name}:BUY({effective_weight}*{signal.conviction:.2f})") elif signal.side == OrderSide.SELL: - score -= weight * signal.conviction - reasons.append(f"{strategy.name}:SELL({weight}*{signal.conviction:.2f})") + score -= effective_weight * signal.conviction + reasons.append(f"{strategy.name}:SELL({effective_weight}*{signal.conviction:.2f})") normalized = score / total_weight # Range: -1.0 to 1.0 |