gitadmin/dzentra_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

07.4.4.1.9.6.2 Momentum & Breakout Semantic Engine

Browse Source
This commit is contained in:
Sergey
2026-05-12 23:25:16 +03:00
parent 8b83055e6a
commit 4a34338041
10 changed files with 591 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
app/src/trading/strategies/trend.py
View File

@@ -8,6 +8,8 @@ from src.integrations.exchange.service import ExchangeService
from src.trading.market_analysis.models import (
MarketPhase,
MarketState,
MomentumState,
TrendDirection,
TrendQuality,
TrendStrength,
)
@@ -128,12 +130,24 @@ class TrendStrategy:
"market_phase_change_percent": market.phase_change_percent,
"market_phase_direction_consistency": market.payload.get("market_phase_direction_consistency"),
"market_phase_reason": market.phase_reason,
"momentum_state": market.momentum_state.value,
"momentum_direction": market.momentum_direction.value,
"momentum_change_percent": market.momentum_change_percent,
"momentum_strength": market.momentum_strength,
"breakout_level": market.breakout_level,
"breakout_distance_percent": market.breakout_distance_percent,
"breakout_reason": market.breakout_reason,
"market_trend_gap_percent": market.trend_gap_percent,
"market_trend_consistency": market.trend_consistency,
"runtime_window_ttl_seconds": self._window_ttl_seconds,
"runtime_window_size": len(prices),
}
breakout_signal = self._breakout_signal(market, base_payload)
if breakout_signal is not None:
return breakout_signal
if not market.is_trade_allowed:
return SignalResult(
signal=SignalType.HOLD,
@@ -286,6 +300,48 @@ class TrendStrategy:
},
)
def _breakout_signal(self, market, base_payload: dict) -> SignalResult | None:
momentum_state = getattr(market, "momentum_state", MomentumState.UNKNOWN)
momentum_direction = getattr(market, "momentum_direction", TrendDirection.UNKNOWN)
momentum_strength = float(getattr(market, "momentum_strength", 0.0) or 0.0)
if momentum_state == MomentumState.BREAKOUT_UP:
return SignalResult(
signal=SignalType.BUY,
reason="BREAKOUT_UP подтверждён momentum/breakout semantic layer.",
confidence=self._calculate_breakout_confidence(momentum_strength),
payload={
**base_payload,
"entry_block_reason": None,
"entry_block_message": None,
"breakout_signal": True,
"expected_direction": "BUY",
},
)
if momentum_state == MomentumState.BREAKOUT_DOWN:
return SignalResult(
signal=SignalType.SELL,
reason="BREAKOUT_DOWN подтверждён momentum/breakout semantic layer.",
confidence=self._calculate_breakout_confidence(momentum_strength),
payload={
**base_payload,
"entry_block_reason": None,
"entry_block_message": None,
"breakout_signal": True,
"expected_direction": "SELL",
},
)
return None
def _calculate_breakout_confidence(self, momentum_strength: float) -> float:
strength_score = min(1.0, max(0.0, momentum_strength) / 2)
confidence = 0.55 + (strength_score * 0.35)
return round(min(0.95, confidence), 2)
def _analysis_price(self, snapshot: dict[str, object]) -> float:
bid = self._safe_float(snapshot.get("bid_price"))
ask = self._safe_float(snapshot.get("ask_price"))