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

07.4.4.1.9 Adaptive Market Diagnostics Layer

Browse Source
This commit is contained in:
Sergey
2026-05-11 22:09:16 +03:00
parent 9ba1297c46
commit 5325ea3855
10 changed files with 500 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
app/src/trading/market_analysis/service.py
View File

@@ -1,13 +1,14 @@
# app/src/trading/market_analysis/service.py
from __future__ import annotations
from src.integrations.exchange.service import ExchangeService
from src.trading.market_analysis.indicators import atr, ema, rsi
from src.trading.market_analysis.models import (
MarketAnalysisResult,
MarketPhase,
MarketState,
TrendDirection,
TrendQuality,
TrendStrength,
VolatilityState,
)
@@ -24,6 +25,7 @@ class MarketAnalysisService:
_high_volatility_atr_percent = 1.8
_trend_gap_percent = 0.03
_trend_consistency_window = 20
def analyze(
self,
@@ -77,12 +79,29 @@ class MarketAnalysisService:
)
atr_percent = (atr_value / close_price) * 100
trend_gap_percent = self._trend_gap_percent_value(
ema_fast=ema_fast,
ema_slow=ema_slow,
)
volatility = self._classify_volatility(atr_percent)
trend = self._classify_trend(
ema_fast=ema_fast,
ema_slow=ema_slow,
)
trend_strength = self._classify_trend_strength(trend_gap_percent)
trend_consistency = self._trend_consistency(
closes=closes,
trend=trend,
)
trend_quality = self._classify_trend_quality(trend_consistency)
market_phase = self._classify_market_phase(
trend=trend,
volatility=volatility,
trend_strength=trend_strength,
trend_quality=trend_quality,
rsi_value=rsi_value,
)
state = self._classify_market_state(
trend=trend,
@@ -123,6 +142,15 @@ class MarketAnalysisService:
"market_state": state.value,
"trend": trend.value,
"volatility": volatility.value,
"market_trend_strength": trend_strength.value,
"market_trend_quality": trend_quality.value,
"market_phase": market_phase.value,
"market_trend_gap_percent": round(trend_gap_percent, 5)
if trend_gap_percent is not None
else None,
"market_trend_consistency": round(trend_consistency, 3)
if trend_consistency is not None
else None,
"close_price": close_price,
"ema_fast_period": self._fast_ema_period,
"ema_slow_period": self._slow_ema_period,
@@ -136,18 +164,37 @@ class MarketAnalysisService:
"candles_count": len(candles),
"is_trade_allowed": is_trade_allowed,
},
trend_strength=trend_strength,
trend_quality=trend_quality,
market_phase=market_phase,
trend_gap_percent=trend_gap_percent,
trend_consistency=trend_consistency,
)
def _trend_gap_percent_value(
self,
*,
ema_fast: float,
ema_slow: float,
) -> float | None:
if ema_slow <= 0:
return None
return ((ema_fast - ema_slow) / ema_slow) * 100
def _classify_trend(
self,
*,
ema_fast: float,
ema_slow: float,
) -> TrendDirection:
if ema_slow <= 0:
return TrendDirection.UNKNOWN
gap_percent = self._trend_gap_percent_value(
ema_fast=ema_fast,
ema_slow=ema_slow,
)
gap_percent = ((ema_fast - ema_slow) / ema_slow) * 100
if gap_percent is None:
return TrendDirection.UNKNOWN
if gap_percent >= self._trend_gap_percent:
return TrendDirection.UP
@@ -157,6 +204,99 @@ class MarketAnalysisService:
return TrendDirection.FLAT
def _classify_trend_strength(
self,
trend_gap_percent: float | None,
) -> TrendStrength:
if trend_gap_percent is None:
return TrendStrength.UNKNOWN
gap = abs(trend_gap_percent)
if gap < 0.08:
return TrendStrength.WEAK
if gap < 0.25:
return TrendStrength.NORMAL
return TrendStrength.STRONG
def _trend_consistency(
self,
*,
closes: list[float],
trend: TrendDirection,
) -> float | None:
if len(closes) < 2:
return None
window = closes[-self._trend_consistency_window :]
if len(window) < 2:
return None
up_moves = 0
down_moves = 0
for previous_price, current_price in zip(window, window[1:]):
if current_price > previous_price:
up_moves += 1
elif current_price < previous_price:
down_moves += 1
total_moves = max(1, len(window) - 1)
if trend == TrendDirection.UP:
return up_moves / total_moves
if trend == TrendDirection.DOWN:
return down_moves / total_moves
return None
def _classify_trend_quality(
self,
trend_consistency: float | None,
) -> TrendQuality:
if trend_consistency is None:
return TrendQuality.UNKNOWN
if trend_consistency >= 0.6:
return TrendQuality.CLEAN
return TrendQuality.NOISY
def _classify_market_phase(
self,
*,
trend: TrendDirection,
volatility: VolatilityState,
trend_strength: TrendStrength,
trend_quality: TrendQuality,
rsi_value: float | None,
) -> MarketPhase:
if volatility == VolatilityState.LOW:
return MarketPhase.SQUEEZE
if trend == TrendDirection.FLAT:
return MarketPhase.RANGE
if trend not in {TrendDirection.UP, TrendDirection.DOWN}:
return MarketPhase.UNKNOWN
if trend_strength == TrendStrength.WEAK:
return MarketPhase.RANGE
if trend_quality == TrendQuality.NOISY:
return MarketPhase.PULLBACK
if trend == TrendDirection.UP and rsi_value is not None and rsi_value < 45:
return MarketPhase.PULLBACK
if trend == TrendDirection.DOWN and rsi_value is not None and rsi_value > 55:
return MarketPhase.PULLBACK
return MarketPhase.IMPULSE
def _classify_volatility(self, atr_percent: float) -> VolatilityState:
if atr_percent <= 0:
return VolatilityState.UNKNOWN
@@ -249,8 +389,18 @@ class MarketAnalysisService:
"market_state": MarketState.UNKNOWN.value,
"trend": TrendDirection.UNKNOWN.value,
"volatility": VolatilityState.UNKNOWN.value,
"market_trend_strength": TrendStrength.UNKNOWN.value,
"market_trend_quality": TrendQuality.UNKNOWN.value,
"market_phase": MarketPhase.UNKNOWN.value,
"market_trend_gap_percent": None,
"market_trend_consistency": None,
"candles_count": candles_count,
"is_trade_allowed": False,
"reason": reason,
},
trend_strength=TrendStrength.UNKNOWN,
trend_quality=TrendQuality.UNKNOWN,
market_phase=MarketPhase.UNKNOWN,
trend_gap_percent=None,
trend_consistency=None,
)