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

07.4.4.1.1 — Market State Human UI + HOLD Lifecycle Fix

Browse Source
This commit is contained in:
Sergey
2026-05-10 23:20:54 +03:00
parent 8024cd9d9a
commit ef7cec68cc
14 changed files with 1209 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

256
app/src/trading/market_analysis/service.py Normal file
View File

@@ -0,0 +1,256 @@
# 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,
MarketState,
TrendDirection,
VolatilityState,
)
class MarketAnalysisService:
_fast_ema_period = 20
_slow_ema_period = 50
_atr_period = 14
_rsi_period = 14
_min_candles = 60
_low_volatility_atr_percent = 0.05
_high_volatility_atr_percent = 1.8
_trend_gap_percent = 0.03
def analyze(
self,
symbol: str,
*,
interval: str = "5m",
limit: int = 200,
) -> MarketAnalysisResult:
try:
batch = ExchangeService().get_klines(
symbol=symbol,
interval=interval,
limit=limit,
)
except Exception as exc:
return self._unknown(
symbol=symbol,
interval=interval,
reason=f"Не удалось получить свечи: {exc}",
)
candles = batch.candles
closes = [item.close_price for item in candles]
if len(candles) < self._min_candles:
return self._unknown(
symbol=batch.symbol,
interval=interval,
reason="Недостаточно свечей для анализа рынка.",
candles_count=len(candles),
)
close_price = closes[-1] if closes else None
ema_fast = ema(closes, self._fast_ema_period)
ema_slow = ema(closes, self._slow_ema_period)
atr_value = atr(candles, self._atr_period)
rsi_value = rsi(closes, self._rsi_period)
if (
close_price is None
or close_price <= 0
or ema_fast is None
or ema_slow is None
or atr_value is None
):
return self._unknown(
symbol=batch.symbol,
interval=interval,
reason="Недостаточно данных для расчёта EMA / ATR.",
candles_count=len(candles),
)
atr_percent = (atr_value / close_price) * 100
volatility = self._classify_volatility(atr_percent)
trend = self._classify_trend(
ema_fast=ema_fast,
ema_slow=ema_slow,
)
state = self._classify_market_state(
trend=trend,
volatility=volatility,
)
is_trade_allowed = state in {
MarketState.TREND_UP,
MarketState.TREND_DOWN,
}
reason = self._reason(
state=state,
trend=trend,
volatility=volatility,
atr_percent=atr_percent,
rsi_value=rsi_value,
)
return MarketAnalysisResult(
symbol=batch.symbol,
interval=interval,
state=state,
trend=trend,
volatility=volatility,
close_price=close_price,
ema_fast=ema_fast,
ema_slow=ema_slow,
atr=atr_value,
atr_percent=atr_percent,
rsi=rsi_value,
candles_count=len(candles),
reason=reason,
is_trade_allowed=is_trade_allowed,
payload={
"symbol": batch.symbol,
"interval": interval,
"market_state": state.value,
"trend": trend.value,
"volatility": volatility.value,
"close_price": close_price,
"ema_fast_period": self._fast_ema_period,
"ema_slow_period": self._slow_ema_period,
"ema_fast": round(ema_fast, 8),
"ema_slow": round(ema_slow, 8),
"atr_period": self._atr_period,
"atr": round(atr_value, 8),
"atr_percent": round(atr_percent, 4),
"rsi_period": self._rsi_period,
"rsi": round(rsi_value, 2) if rsi_value is not None else None,
"candles_count": len(candles),
"is_trade_allowed": is_trade_allowed,
},
)
def _classify_trend(
self,
*,
ema_fast: float,
ema_slow: float,
) -> TrendDirection:
if ema_slow <= 0:
return TrendDirection.UNKNOWN
gap_percent = ((ema_fast - ema_slow) / ema_slow) * 100
if gap_percent >= self._trend_gap_percent:
return TrendDirection.UP
if gap_percent <= -self._trend_gap_percent:
return TrendDirection.DOWN
return TrendDirection.FLAT
def _classify_volatility(self, atr_percent: float) -> VolatilityState:
if atr_percent <= 0:
return VolatilityState.UNKNOWN
if atr_percent < self._low_volatility_atr_percent:
return VolatilityState.LOW
if atr_percent > self._high_volatility_atr_percent:
return VolatilityState.HIGH
return VolatilityState.NORMAL
def _classify_market_state(
self,
*,
trend: TrendDirection,
volatility: VolatilityState,
) -> MarketState:
if volatility == VolatilityState.HIGH:
return MarketState.HIGH_VOLATILITY
if volatility == VolatilityState.LOW:
return MarketState.LOW_VOLATILITY
if trend == TrendDirection.UP:
return MarketState.TREND_UP
if trend == TrendDirection.DOWN:
return MarketState.TREND_DOWN
if trend == TrendDirection.FLAT:
return MarketState.RANGE
return MarketState.UNKNOWN
def _reason(
self,
*,
state: MarketState,
trend: TrendDirection,
volatility: VolatilityState,
atr_percent: float,
rsi_value: float | None,
) -> str:
rsi_text = f", RSI={rsi_value:.2f}" if rsi_value is not None else ""
if state == MarketState.TREND_UP:
return f"Рынок в восходящем тренде. ATR={atr_percent:.2f}%{rsi_text}."
if state == MarketState.TREND_DOWN:
return f"Рынок в нисходящем тренде. ATR={atr_percent:.2f}%{rsi_text}."
if state == MarketState.RANGE:
return f"Рынок в боковике. Тренд не подтверждён. ATR={atr_percent:.2f}%{rsi_text}."
if state == MarketState.HIGH_VOLATILITY:
return f"Рынок слишком волатильный. ATR={atr_percent:.2f}%{rsi_text}."
if state == MarketState.LOW_VOLATILITY:
return f"Рынок слишком спокойный. ATR={atr_percent:.2f}%{rsi_text}."
return f"Состояние рынка не определено. Trend={trend}, volatility={volatility}."
def _unknown(
self,
*,
symbol: str,
interval: str,
reason: str,
candles_count: int = 0,
) -> MarketAnalysisResult:
return MarketAnalysisResult(
symbol=symbol,
interval=interval,
state=MarketState.UNKNOWN,
trend=TrendDirection.UNKNOWN,
volatility=VolatilityState.UNKNOWN,
close_price=None,
ema_fast=None,
ema_slow=None,
atr=None,
atr_percent=None,
rsi=None,
candles_count=candles_count,
reason=reason,
is_trade_allowed=False,
payload={
"symbol": symbol,
"interval": interval,
"market_state": MarketState.UNKNOWN.value,
"trend": TrendDirection.UNKNOWN.value,
"volatility": VolatilityState.UNKNOWN.value,
"candles_count": candles_count,
"is_trade_allowed": False,
"reason": reason,
},
)