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

build 051: switch HTF analysis to canonical candles

Browse Source
This commit is contained in:
Sergey
2026-07-16 08:24:53 +03:00
parent c8d33f8baa
commit 0b2187dac4
3 changed files with 1180 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
app/src/trading/market_analysis/htf.py
View File

@@ -2,6 +2,8 @@
from __future__ import annotations
from math import isfinite
from src.core.numbers import safe_float
from src.core.types import JsonDict
from src.integrations.exchange.service import ExchangeService
@@ -52,7 +54,7 @@ def htf_volatility_context(
}
try:
batch = ExchangeService().get_klines(
candles = ExchangeService().get_candles(
symbol=symbol,
interval=service._htf_interval,
limit=service._htf_limit,
@@ -67,10 +69,7 @@ def htf_volatility_context(
"htf_reason": f"HTF_KLINES_ERROR: {exc}",
}
candles = batch.candles
closes = [item.close_price for item in candles]
if len(candles) < service._min_candles or not closes:
if len(candles) < service._min_candles:
return {
"htf_interval": service._htf_interval,
"htf_atr_percent": None,
@@ -80,10 +79,15 @@ def htf_volatility_context(
"htf_reason": "HTF_NOT_ENOUGH_CANDLES",
}
close_price = safe_float(closes[-1])
close_price = safe_float(candles[-1].close_price)
atr_value = atr(candles, service._atr_period)
if close_price is None or close_price <= 0 or atr_value is None:
if (
close_price is None
or not isfinite(close_price)
or close_price <= 0
or atr_value is None
):
return {
"htf_interval": service._htf_interval,
"htf_atr_percent": None,
@@ -150,7 +154,7 @@ def htf_trend_context(
}
try:
batch = ExchangeService().get_klines(
candles = ExchangeService().get_candles(
symbol=symbol,
interval=service._htf_interval,
limit=service._htf_limit,
@@ -158,12 +162,19 @@ def htf_trend_context(
except Exception as exc:
return _htf_unknown_context(f"HTF_KLINES_ERROR: {exc}")
candles = batch.candles
closes = [item.close_price for item in candles]
if len(candles) < service._min_candles:
return _htf_unknown_context("HTF_NOT_ENOUGH_CANDLES")
closes: list[float] = []
for candle in candles:
close_value = safe_float(candle.close_price)
if close_value is None or not isfinite(close_value):
return _htf_unknown_context("HTF_INDICATORS_UNAVAILABLE")
closes.append(close_value)
close_price = closes[-1] if closes else None
ema_fast = ema(closes, service._fast_ema_period)
ema_slow = ema(closes, service._slow_ema_period)
@@ -369,7 +380,7 @@ def safe_volatility_state(value: object) -> VolatilityState | None:
return VolatilityState(str(value))
except Exception:
return None
def _htf_unknown_context(reason: str) -> JsonDict:
return {
@@ -484,4 +495,4 @@ def _htf_confirmation_score(
if trend_efficiency is not None:
score += (trend_efficiency - 0.3) * 0.15
return max(0.0, min(1.0, score))
return max(0.0, min(1.0, score))

516
app/tests/unit/trading/market_analysis/test_htf_candles.py Normal file
View File

@@ -0,0 +1,516 @@
# app/tests/unit/trading/market_analysis/test_htf_candles.py
from __future__ import annotations
from datetime import datetime, timezone
from decimal import Decimal
from types import SimpleNamespace
import pytest
import src.trading.market_analysis.htf as module
from src.market_data.acquisition.models.candle import Candle
from src.trading.market_analysis.models import (
MarketPhase,
MarketState,
TrendDirection,
TrendQuality,
TrendStrength,
VolatilityState,
)
def _service() -> SimpleNamespace:
return SimpleNamespace(
_htf_interval="1h",
_htf_limit=120,
_min_candles=60,
_atr_period=14,
_atr_baseline_window=60,
_low_volatility_atr_percent=0.05,
_high_volatility_atr_percent=1.8,
_fast_ema_period=20,
_slow_ema_period=50,
_ema_fast_slope_window=5,
_ema_slow_slope_window=8,
_trend_consistency_window=20,
_candle_noise_window=20,
_min_clean_body_ratio=0.35,
_price_position_window=20,
_min_clean_candle_score=0.55,
_min_price_position_score=0.55,
)
def _candle(
*,
index: int = 0,
close_price: str = "100",
) -> Candle:
return Candle(
symbol="BTC/USD_LEVERAGE",
interval="1h",
open_time=datetime.fromtimestamp(
1_750_000_000 + index * 3600,
tz=timezone.utc,
),
open_price=Decimal(str(99 + index)),
high_price=Decimal(str(101 + index)),
low_price=Decimal(str(98 + index)),
close_price=Decimal(close_price),
volume=Decimal("10"),
source="test",
)
def _candles(count: int = 60) -> tuple[Candle, ...]:
return tuple(
_candle(
index=index,
close_price=str(100 + index),
)
for index in range(count)
)
def test_htf_volatility_skips_same_interval(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class ForbiddenExchangeService:
def __init__(self) -> None:
raise AssertionError("ExchangeService must not be created.")
monkeypatch.setattr(
module,
"ExchangeService",
ForbiddenExchangeService,
)
result = module.htf_volatility_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="1h",
)
assert result["htf_reason"] == "HTF_SKIPPED_SAME_INTERVAL"
assert result["htf_interval"] == "1h"
def test_htf_volatility_uses_get_candles_with_exact_arguments(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[dict[str, object]] = []
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
calls.append(
{
"symbol": symbol,
"interval": interval,
"limit": limit,
}
)
return _candles(10)
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
result = module.htf_volatility_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
)
assert calls == [
{
"symbol": "BTC/USD_LEVERAGE",
"interval": "1h",
"limit": 120,
}
]
assert result["htf_reason"] == "HTF_NOT_ENOUGH_CANDLES"
def test_htf_volatility_preserves_legacy_error_reason(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
raise RuntimeError("candles unavailable")
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
result = module.htf_volatility_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
)
assert result["htf_reason"] == (
"HTF_KLINES_ERROR: candles unavailable"
)
def test_htf_volatility_rejects_non_finite_close(
monkeypatch: pytest.MonkeyPatch,
) -> None:
candles = list(_candles())
candles[-1] = _candle(index=59, close_price="NaN")
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
return tuple(candles)
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
monkeypatch.setattr(module, "atr", lambda candles, period: 1.0)
result = module.htf_volatility_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
)
assert result["htf_reason"] == "HTF_ATR_UNAVAILABLE"
def test_htf_volatility_returns_existing_success_payload(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
return _candles()
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
monkeypatch.setattr(module, "atr", lambda candles, period: 2.0)
monkeypatch.setattr(
module,
"atr_percent_baseline",
lambda **kwargs: 1.0,
)
monkeypatch.setattr(
module,
"classify_volatility",
lambda **kwargs: VolatilityState.HIGH,
)
result = module.htf_volatility_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
)
expected_atr_percent = (2.0 / 159.0) * 100
assert result == {
"htf_interval": "1h",
"htf_atr_percent": round(expected_atr_percent, 4),
"htf_atr_percent_baseline": 1.0,
"htf_volatility_ratio": round(expected_atr_percent, 4),
"htf_volatility": VolatilityState.HIGH.value,
"htf_reason": "HTF_OK",
}
def test_htf_trend_skips_same_interval(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class ForbiddenExchangeService:
def __init__(self) -> None:
raise AssertionError("ExchangeService must not be created.")
monkeypatch.setattr(
module,
"ExchangeService",
ForbiddenExchangeService,
)
result = module.htf_trend_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="1h",
local_state=MarketState.TREND_UP,
local_trend=TrendDirection.UP,
)
assert result["htf_reason"] == "HTF_SKIPPED_SAME_INTERVAL"
assert result["htf_market_state"] == MarketState.TREND_UP.value
assert result["htf_trend"] == TrendDirection.UP.value
def test_htf_trend_uses_get_candles_with_exact_arguments(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[dict[str, object]] = []
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
calls.append(
{
"symbol": symbol,
"interval": interval,
"limit": limit,
}
)
return _candles(10)
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
result = module.htf_trend_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
local_state=MarketState.TREND_UP,
local_trend=TrendDirection.UP,
)
assert calls == [
{
"symbol": "BTC/USD_LEVERAGE",
"interval": "1h",
"limit": 120,
}
]
assert result["htf_reason"] == "HTF_NOT_ENOUGH_CANDLES"
def test_htf_trend_preserves_legacy_error_reason(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
raise RuntimeError("candles unavailable")
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
result = module.htf_trend_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
local_state=MarketState.TREND_UP,
local_trend=TrendDirection.UP,
)
assert result["htf_reason"] == (
"HTF_KLINES_ERROR: candles unavailable"
)
def test_htf_trend_rejects_non_finite_close(
monkeypatch: pytest.MonkeyPatch,
) -> None:
candles = list(_candles())
candles[10] = _candle(index=10, close_price="NaN")
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
return tuple(candles)
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
result = module.htf_trend_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
local_state=MarketState.TREND_UP,
local_trend=TrendDirection.UP,
)
assert result["htf_reason"] == "HTF_INDICATORS_UNAVAILABLE"
def test_htf_trend_converts_decimal_closes_to_float(
monkeypatch: pytest.MonkeyPatch,
) -> None:
ema_calls: list[list[float]] = []
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
return _candles()
def fake_ema(
values: list[float],
period: int,
) -> None:
ema_calls.append(values)
return None
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
monkeypatch.setattr(module, "ema", fake_ema)
monkeypatch.setattr(module, "atr", lambda candles, period: 1.0)
result = module.htf_trend_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
local_state=MarketState.TREND_UP,
local_trend=TrendDirection.UP,
)
assert result["htf_reason"] == "HTF_INDICATORS_UNAVAILABLE"
assert len(ema_calls) == 2
assert all(
isinstance(value, float)
for values in ema_calls
for value in values
)
assert ema_calls[0][0] == 100.0
assert ema_calls[0][-1] == 159.0
def test_htf_trend_returns_existing_success_payload(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class FakeExchangeService:
def get_candles(
self,
symbol: str,
*,
interval: str,
limit: int,
) -> tuple[Candle, ...]:
return _candles()
monkeypatch.setattr(module, "ExchangeService", FakeExchangeService)
monkeypatch.setattr(module, "ema", lambda values, period: 150.0)
monkeypatch.setattr(module, "atr", lambda candles, period: 2.0)
monkeypatch.setattr(
module,
"adaptive_threshold",
lambda **kwargs: 0.1,
)
monkeypatch.setattr(
module,
"ema_slope_percent",
lambda **kwargs: 0.2,
)
monkeypatch.setattr(
module,
"classify_trend",
lambda **kwargs: TrendDirection.UP,
)
monkeypatch.setattr(
module,
"trend_gap_percent_value",
lambda **kwargs: 1.0,
)
monkeypatch.setattr(
module,
"classify_trend_strength",
lambda **kwargs: TrendStrength.STRONG,
)
monkeypatch.setattr(
module,
"trend_consistency",
lambda **kwargs: 0.8,
)
monkeypatch.setattr(
module,
"trend_efficiency",
lambda **kwargs: 0.7,
)
monkeypatch.setattr(
module,
"calculate_ema_distance_atr_ratio",
lambda **kwargs: 1.5,
)
monkeypatch.setattr(
module,
"calculate_candle_noise_score",
lambda *args, **kwargs: 0.8,
)
monkeypatch.setattr(
module,
"calculate_price_position_score",
lambda **kwargs: 0.9,
)
monkeypatch.setattr(
module,
"classify_trend_quality",
lambda **kwargs: TrendQuality.CLEAN,
)
monkeypatch.setattr(
module,
"_htf_market_phase",
lambda **kwargs: MarketPhase.IMPULSE,
)
monkeypatch.setattr(
module,
"_htf_market_state",
lambda **kwargs: MarketState.TREND_UP,
)
monkeypatch.setattr(
module,
"_htf_alignment",
lambda **kwargs: "ALIGNED",
)
monkeypatch.setattr(
module,
"_htf_confirmation_score",
lambda **kwargs: 0.9,
)
result = module.htf_trend_context(
_service(),
symbol="BTC/USD_LEVERAGE",
base_interval="5m",
local_state=MarketState.TREND_UP,
local_trend=TrendDirection.UP,
)
assert result == {
"htf_market_state": MarketState.TREND_UP.value,
"htf_trend": TrendDirection.UP.value,
"htf_trend_strength": TrendStrength.STRONG.value,
"htf_trend_quality": TrendQuality.CLEAN.value,
"htf_market_phase": MarketPhase.IMPULSE.value,
"htf_alignment": "ALIGNED",
"htf_confirmation_score": 0.9,
"htf_reason": "HTF_1h:TREND_UP:UP:ALIGNED",
}