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

Stage 07.4.3 — trend strategy, signal aggregation and journal UX improvements

Browse Source
This commit is contained in:
Sergey
2026-05-01 11:43:26 +03:00
parent 80f29443d4
commit ec8e53c416
8 changed files with 510 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/src/telegram/handlers/auto.py
View File

@@ -76,7 +76,9 @@ def _build_auto_text() -> str:
f"Риск: {risk}\n"
f"PnL: {state.pnl_usd:.2f} USD\n"
f"Последний анализ: {state.last_check_at or '—'}\n"
f"Сигнал: {_signal_label(state.last_signal)}"
f"Сигнал: {_signal_label(state.last_signal)} · {state.last_signal_repeat_count} подряд\n"
f"Уверенность: {state.last_signal_confidence:.2f}\n"
f"Причина: {state.last_signal_reason or '—'}"
)

126
app/src/telegram/handlers/journal_ui.py
View File

@@ -21,6 +21,8 @@ LEVEL_ICONS = {
}
EVENT_TITLES = {
"auto_signal_generated": "Сигнал автоторговли",
"auto_signal_summary": "Итог серии сигналов",
"app_start": "Запуск приложения",
"system_open_alert": "Система загружена с предупреждениями",
"system_open_requested": "Открытие системы",
@@ -165,16 +167,40 @@ def render_clear_confirm(
return "\n".join(lines)
def _normalize_datetime(value: str) -> str:
def _parse_local_datetime(value: str) -> datetime | None:
try:
settings = load_settings()
dt = datetime.fromisoformat(value)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
dt = dt.astimezone(ZoneInfo(settings.tz))
return dt.strftime("%Y-%m-%d %H:%M:%S")
return dt.astimezone(ZoneInfo(settings.tz))
except Exception:
return value
return None
def _date_group_label(dt: datetime | None) -> str:
if dt is None:
return "Без даты"
settings = load_settings()
today = datetime.now(ZoneInfo(settings.tz)).date()
if dt.date() == today:
return "Сегодня"
if (today - dt.date()).days == 1:
return "Вчера"
return dt.strftime("%Y-%m-%d")
def _time_label(dt: datetime | None, raw_value: str) -> str:
if dt is None:
return raw_value
return dt.strftime("%H:%M:%S")
def _event_title(event_type: str) -> str:
@@ -189,29 +215,101 @@ def _humanize_message(message: str) -> str:
return message
def _payload(event: dict) -> dict:
payload = event.get("payload")
return payload if isinstance(payload, dict) else {}
def _render_auto_signal(event: dict, created_time: str) -> list[str]:
payload = _payload(event)
signal = str(payload.get("signal", "HOLD")).upper()
strategy = str(payload.get("strategy", "AUTO")).upper()
symbol = str(payload.get("symbol", "—"))
reason = str(payload.get("reason", ""))
confidence = float(payload.get("confidence", 0.0) or 0.0)
repeat_count = int(payload.get("repeat_count", 1) or 1)
is_strong_signal = bool(payload.get("is_strong_signal", False))
is_aggregated = bool(payload.get("is_aggregated", False))
signal_icon = {
"BUY": "🟢",
"SELL": "🔴",
"HOLD": "🟡",
}.get(signal, "•")
prefix = ""
if is_strong_signal:
prefix += "📈 "
if is_aggregated:
prefix += "🧠 "
lines = [
f"{prefix}{signal_icon} <b>AUTO · {signal}</b>",
f"{created_time} · {strategy} · {symbol}",
]
if is_aggregated:
lines.append(f"{repeat_count} {signal} подряд")
if confidence > 0:
lines.append(f"Уверенность: {confidence:.2f}")
if reason:
lines.append(f"Причина: {reason}")
return lines
def _render_default_event(event: dict, created_time: str) -> list[str]:
level = str(event.get("level", "INFO")).upper()
icon = LEVEL_ICONS.get(level, "•")
title = _event_title(str(event.get("event_type", "")))
message = _humanize_message(str(event.get("message", "")))
lines = [
f"{icon} <b>{level}</b> · {title}",
f"{created_time}",
]
if message:
lines.append(message)
return lines
def render(events, page, total_pages):
lines = [
"<b>📒 Журнал</b>",
"",
"<b>МОНИТОРИНГ</b>",
"",
"<b>Последние события:</b>",
"",
]
if not events:
lines.append("Событий пока нет.")
return "\n".join(lines)
for event in events:
level = str(event.get("level", "INFO")).upper()
icon = LEVEL_ICONS.get(level, "•")
title = _event_title(str(event.get("event_type", "")))
created_at = _normalize_datetime(str(event.get("created_at", "")))
message = _humanize_message(str(event.get("message", "")))
current_group = None
for event in events:
raw_created_at = str(event.get("created_at", ""))
dt = _parse_local_datetime(raw_created_at)
group_label = _date_group_label(dt)
created_time = _time_label(dt, raw_created_at)
if group_label != current_group:
current_group = group_label
lines.append(f"<b>{group_label}</b>")
lines.append("")
event_type = str(event.get("event_type", ""))
if event_type in {"auto_signal_generated", "auto_signal_summary"}:
lines.extend(_render_auto_signal(event, created_time))
else:
lines.extend(_render_default_event(event, created_time))
lines.append(f"{icon} [ <b>{level}</b> ] <b>{title}</b>")
lines.append(f"• {created_at}")
lines.append("")
return "\n".join(lines).rstrip()

3
app/src/trading/auto/runner.py
View File

@@ -105,9 +105,6 @@ class AutoTradeRunner:
# обновить live-экран Telegram
@classmethod
async def _refresh_screen(cls) -> None:
if cls._current_screen != "auto":
return
if not all(
[
cls._bot,

149
app/src/trading/auto/service.py
View File

@@ -7,6 +7,7 @@ from datetime import datetime
from src.core.config import load_settings
from src.trading.auto.state import AutoTradeState
from src.trading.journal.service import JournalService
from src.trading.strategies.base import BaseStrategy, StrategyContext
from src.trading.strategies.registry import StrategyRegistry
@@ -15,6 +16,9 @@ class AutoTradeService:
_state = AutoTradeState()
_loop_task: asyncio.Task | None = None
_loop_interval_seconds = 5
_last_signal_key: str | None = None
_last_signal_value: str | None = None
_same_signal_count = 0
# получить текущее состояние автоторговли
def get_state(self) -> AutoTradeState:
@@ -108,6 +112,9 @@ class AutoTradeService:
def set_strategy(self, strategy: str) -> AutoTradeState:
state = self.get_state()
state.strategy = strategy.strip().upper()
self._last_signal_key = None
self._last_signal_value = None
self._same_signal_count = 0
return state
# установить риск
@@ -131,6 +138,135 @@ class AutoTradeService:
state = self.get_state()
return StrategyRegistry.get(state.strategy)
# записать новый сигнал и итог предыдущей серии при смене сигнала
def _log_signal_if_changed(
self,
*,
strategy_name: str,
state: AutoTradeState,
signal: str,
reason: str,
confidence: float,
payload: dict | None,
) -> None:
signal_key = f"{state.status}:{state.symbol}:{strategy_name}:{signal}:{reason}"
previous_signal = self._last_signal_value
previous_count = self._same_signal_count
is_same_signal = signal_key == self._last_signal_key
if is_same_signal:
self._same_signal_count += 1
return
if previous_signal is not None:
if previous_count > 1:
self._log_signal_summary(
strategy_name=strategy_name,
state=state,
previous_signal=previous_signal,
previous_count=previous_count,
next_signal=signal,
)
else:
self._log_signal_event(
strategy_name=strategy_name,
state=state,
signal=previous_signal,
reason=f"{previous_signal} завершился без серии.",
confidence=0.0,
payload={
"previous_signal": previous_signal,
"next_signal": signal,
},
)
self._last_signal_key = signal_key
self._last_signal_value = signal
self._same_signal_count = 1
# Новый сигнал не пишем сразу.
# Он попадёт в журнал при следующей смене сигнала:
# либо как одиночный сигнал, либо как серия.
# записать сам сигнал в журнал
def _log_signal_event(
self,
*,
strategy_name: str,
state: AutoTradeState,
signal: str,
reason: str,
confidence: float,
payload: dict | None,
) -> None:
emoji_map = {
"BUY": "🟢",
"SELL": "🔴",
"HOLD": "🟡",
}
emoji = emoji_map.get(signal, "•")
try:
JournalService().log_ui_info(
event_type="auto_signal_generated",
message=f"{emoji} Сигнал автоторговли {signal}: {reason}",
screen="auto",
action="run_cycle",
payload={
"strategy": strategy_name,
"status": state.status,
"symbol": state.symbol,
"signal": signal,
"confidence": confidence,
"reason": reason,
"repeat_count": 1,
"is_strong_signal": confidence > 0.7,
"is_aggregated": False,
"payload": payload or {},
},
)
except Exception:
pass
# записать итог серии одинаковых сигналов при смене сигнала
def _log_signal_summary(
self,
*,
strategy_name: str,
state: AutoTradeState,
previous_signal: str,
previous_count: int,
next_signal: str,
) -> None:
emoji_map = {
"BUY": "🟢",
"SELL": "🔴",
"HOLD": "🟡",
}
emoji = emoji_map.get(previous_signal, "•")
try:
JournalService().log_ui_info(
event_type="auto_signal_summary",
message=(
f"{emoji} {previous_count} {previous_signal} подряд "
f"до смены на {next_signal}"
),
screen="auto",
action="signal_summary",
payload={
"strategy": strategy_name,
"status": state.status,
"symbol": state.symbol,
"signal": previous_signal,
"next_signal": next_signal,
"repeat_count": previous_count,
"is_aggregated": True,
},
)
except Exception:
pass
# выполнить один цикл анализа рынка
def run_cycle(self) -> AutoTradeState:
state = self.get_state()
@@ -145,4 +281,17 @@ class AutoTradeService:
state.last_check_at = datetime.now().strftime("%H:%M:%S")
state.last_signal = result.signal.value
self._log_signal_if_changed(
strategy_name=strategy.name,
state=state,
signal=result.signal.value,
reason=result.reason,
confidence=result.confidence,
payload=result.payload,
)
state.last_signal_repeat_count = self._same_signal_count
state.last_signal_confidence = result.confidence
state.last_signal_reason = result.reason
return state

11
app/src/trading/auto/state.py
View File

@@ -26,4 +26,13 @@ class AutoTradeState:
last_check_at: str | None = None
# последний сигнал стратегии
last_signal: str | None = None
last_signal: str | None = None
# количество одинаковых сигналов подряд
last_signal_repeat_count: int = 0
# уверенность последнего сигнала от 0.0 до 1.0
last_signal_confidence: float = 0.0
# причина последнего сигнала
last_signal_reason: str | None = None

3
app/src/trading/strategies/registry.py
View File

@@ -4,13 +4,14 @@ from __future__ import annotations
from src.trading.strategies.base import BaseStrategy
from src.trading.strategies.hold import HoldStrategy
from src.trading.strategies.trend import TrendStrategy
class StrategyRegistry:
# доступные стратегии
_strategies: dict[str, BaseStrategy] = {
"HOLD": HoldStrategy(),
"TREND": HoldStrategy(),
"TREND": TrendStrategy(),
"GRID": HoldStrategy(),
"SCALP": HoldStrategy(),
}

91
app/src/trading/strategies/trend.py Normal file
View File

@@ -0,0 +1,91 @@
# app/src/trading/strategies/trend.py
from __future__ import annotations
from src.integrations.exchange.service import ExchangeService
from src.trading.strategies.base import StrategyContext
from src.trading.strategies.signals import SignalResult, SignalType
class TrendStrategy:
name = "TREND"
_last_prices: dict[str, float] = {}
_threshold_percent = 0.02
# анализ простого тренда по изменению цены
def analyze(self, context: StrategyContext) -> SignalResult:
try:
ticker = ExchangeService().get_price(context.symbol)
except Exception as exc:
return SignalResult(
signal=SignalType.HOLD,
reason="Не удалось получить рыночную цену. Безопасный HOLD.",
confidence=0.0,
payload={
"strategy": self.name,
"symbol": context.symbol,
"error": str(exc),
},
)
symbol = ticker.symbol
current_price = ticker.price
previous_price = self._last_prices.get(symbol)
self._last_prices[symbol] = current_price
if previous_price is None or previous_price <= 0:
return SignalResult(
signal=SignalType.HOLD,
reason="Недостаточно данных для определения тренда.",
confidence=0.0,
payload={
"strategy": self.name,
"symbol": symbol,
"price": current_price,
},
)
change_percent = ((current_price - previous_price) / previous_price) * 100
if change_percent >= self._threshold_percent:
return SignalResult(
signal=SignalType.BUY,
reason="Цена растёт выше порога тренда.",
confidence=min(1.0, abs(change_percent) / self._threshold_percent),
payload={
"strategy": self.name,
"symbol": symbol,
"previous_price": previous_price,
"current_price": current_price,
"change_percent": round(change_percent, 5),
},
)
if change_percent <= -self._threshold_percent:
return SignalResult(
signal=SignalType.SELL,
reason="Цена падает ниже порога тренда.",
confidence=min(1.0, abs(change_percent) / self._threshold_percent),
payload={
"strategy": self.name,
"symbol": symbol,
"previous_price": previous_price,
"current_price": current_price,
"change_percent": round(change_percent, 5),
},
)
return SignalResult(
signal=SignalType.HOLD,
reason="Изменение цены ниже порога тренда.",
confidence=0.0,
payload={
"strategy": self.name,
"symbol": symbol,
"previous_price": previous_price,
"current_price": current_price,
"change_percent": round(change_percent, 5),
},
)