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

build 048: switch market analysis consumers to canonical Candle model

Browse Source
This commit is contained in:
Sergey
2026-07-15 18:55:17 +03:00
parent 77e87c0504
commit 3a253d89a9
7 changed files with 1478 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
app/src/trading/market_analysis/indicators/volatility.py
View File

@@ -3,59 +3,90 @@
from __future__ import annotations
from collections.abc import Sequence
from math import isfinite
from src.core.numbers import safe_float
from src.core.types import NumericLike
from src.integrations.exchange.models import Kline
from src.market_data.acquisition.models.candle import Candle
from src.trading.market_analysis.models import VolatilityState
def atr(candles: list[Kline], period: int = 14) -> float | None:
def atr(
candles: Sequence[Candle],
period: int = 14,
) -> float | None:
if period <= 0 or len(candles) < period + 1:
return None
true_ranges: list[float] = []
for previous, current in zip(candles, candles[1:]):
high_low = current.high_price - current.low_price
high_close = abs(current.high_price - previous.close_price)
low_close = abs(current.low_price - previous.close_price)
previous_close = safe_float(previous.close_price)
current_high = safe_float(current.high_price)
current_low = safe_float(current.low_price)
true_ranges.append(max(high_low, high_close, low_close))
if (
previous_close is None
or current_high is None
or current_low is None
or not isfinite(previous_close)
or not isfinite(current_high)
or not isfinite(current_low)
):
continue
high_low = current_high - current_low
high_close = abs(current_high - previous_close)
low_close = abs(current_low - previous_close)
true_ranges.append(
max(
high_low,
high_close,
low_close,
)
)
if len(true_ranges) < period:
return None
recent = true_ranges[-period:]
return sum(recent) / period
def atr_percent_baseline(
*,
candles: Sequence[Kline],
candles: Sequence[Candle],
close_price: float,
atr_period: int,
atr_baseline_window: int,
) -> float | None:
if close_price <= 0:
if not isfinite(close_price) or close_price <= 0:
return None
values: list[float] = []
window: list[Kline] = list(candles[-atr_baseline_window:])
window = list(candles[-atr_baseline_window:])
for index in range(atr_period, len(window) + 1):
part: list[Kline] = window[:index]
atr_value = atr(list(part), atr_period)
part = window[:index]
atr_value = atr(part, atr_period)
if atr_value is None:
if atr_value is None or not isfinite(atr_value):
continue
close = getattr(part[-1], "close_price", None)
close = safe_float(part[-1].close_price)
if close is None or close <= 0:
if (
close is None
or not isfinite(close)
or close <= 0
):
continue
values.append((atr_value / close) * 100)
values.append(
(atr_value / close) * 100
)
if not values:
return None
@@ -66,7 +97,10 @@ def atr_percent_baseline(
if len(values) % 2 == 1:
return values[middle]
return (values[middle - 1] + values[middle]) / 2
return (
values[middle - 1]
+ values[middle]
) / 2
def adaptive_threshold(
@@ -79,10 +113,20 @@ def adaptive_threshold(
multiplier_value = safe_float(multiplier)
minimum_value = safe_float(minimum) or 0.0
if atr_value is None or atr_value <= 0 or multiplier_value is None:
return minimum_value
if (
atr_value is None
or multiplier_value is None
or not isfinite(atr_value)
or not isfinite(multiplier_value)
or not isfinite(minimum_value)
or atr_value <= 0
):
return minimum_value if isfinite(minimum_value) else 0.0
return max(minimum_value, atr_value * multiplier_value)
return max(
minimum_value,
atr_value * multiplier_value,
)
def classify_volatility(
@@ -95,22 +139,50 @@ def classify_volatility(
) -> VolatilityState:
atr_value = safe_float(atr_percent)
if atr_value is None or atr_value <= 0:
if (
atr_value is None
or not isfinite(atr_value)
or atr_value <= 0
):
return VolatilityState.UNKNOWN
local_ratio = safe_float(volatility_ratio)
htf_ratio = safe_float(htf_volatility_ratio)
if local_ratio is not None and not isfinite(local_ratio):
local_ratio = None
if htf_ratio is not None and not isfinite(htf_ratio):
htf_ratio = None
if htf_ratio is not None:
if htf_ratio > 1.8 and (local_ratio is None or local_ratio > 1.1):
if (
htf_ratio > 1.8
and (
local_ratio is None
or local_ratio > 1.1
)
):
return VolatilityState.HIGH
if htf_ratio < 0.55 and (local_ratio is None or local_ratio < 0.85):
if (
htf_ratio < 0.55
and (
local_ratio is None
or local_ratio < 0.85
)
):
return VolatilityState.LOW
if local_ratio is None:
low_value = safe_float(low_volatility_atr_percent) or 0.05
high_value = safe_float(high_volatility_atr_percent) or 1.8
low_value = safe_float(low_volatility_atr_percent)
high_value = safe_float(high_volatility_atr_percent)
if low_value is None or not isfinite(low_value):
low_value = 0.05
if high_value is None or not isfinite(high_value):
high_value = 1.8
if atr_value < low_value:
return VolatilityState.LOW

18
app/src/trading/market_analysis/quality.py
View File

@@ -3,13 +3,15 @@
from __future__ import annotations
from collections.abc import Sequence
from math import isfinite
from src.integrations.exchange.models import Kline
from src.core.numbers import safe_float
from src.market_data.acquisition.models.candle import Candle
from src.trading.market_analysis.models import TrendDirection
def candle_noise_score(
candles: Sequence[Kline],
candles: Sequence[Candle],
*,
candle_noise_window: int,
min_clean_body_ratio: float,
@@ -23,16 +25,20 @@ def candle_noise_score(
total_count = 0
for candle in window:
high = getattr(candle, "high_price", None)
low = getattr(candle, "low_price", None)
open_price = getattr(candle, "open_price", None)
close_price = getattr(candle, "close_price", None)
high = safe_float(candle.high_price)
low = safe_float(candle.low_price)
open_price = safe_float(candle.open_price)
close_price = safe_float(candle.close_price)
if (
high is None
or low is None
or open_price is None
or close_price is None
or not isfinite(high)
or not isfinite(low)
or not isfinite(open_price)
or not isfinite(close_price)
or high <= low
):
continue

6
app/src/trading/market_analysis/structure.py
View File

@@ -6,7 +6,7 @@ from collections.abc import Sequence
from src.core.numbers import safe_float
from src.core.types import NumericLike
from src.integrations.exchange.models import Kline
from src.market_data.acquisition.models.candle import Candle
from src.trading.market_analysis.models import MarketStructure
@@ -47,12 +47,12 @@ def structure_params(
return window, left, right
# определить структуру рынка по swing high / swing low:
# Определить структуру рынка по swing high / swing low:
# HH/HL = восходящая структура
# LH/LL = нисходящая структура
# MIXED = противоречивая структура
def market_structure(
candles: Sequence[Kline],
candles: Sequence[Candle],
*,
atr_percent: NumericLike | None = None,
candle_noise_score: NumericLike | None = None,

136
app/tests/unit/trading/market_analysis/indicators/test_volatility_candle.py Normal file
View File

@@ -0,0 +1,136 @@
# app/tests/unit/trading/market_analysis/indicators/test_volatility_candle.py
from __future__ import annotations
from datetime import datetime, timezone
from decimal import Decimal
from src.market_data.acquisition.models.candle import Candle
def _candle(
*,
index: int = 0,
open_price: str = "100",
high_price: str = "110",
low_price: str = "90",
close_price: str = "105",
volume: str = "10",
) -> Candle:
return Candle(
symbol="BTC/USD_LEVERAGE",
interval="1m",
open_time=datetime.fromtimestamp(
1_750_000_000 + index * 60,
tz=timezone.utc,
),
open_price=Decimal(open_price),
high_price=Decimal(high_price),
low_price=Decimal(low_price),
close_price=Decimal(close_price),
volume=Decimal(volume),
source="test",
)
from src.trading.market_analysis.indicators.volatility import (
atr,
atr_percent_baseline,
)
def test_atr_accepts_canonical_candles_with_decimal_values() -> None:
candles = [
_candle(
index=0,
high_price="105",
low_price="95",
close_price="100",
),
_candle(
index=1,
high_price="110",
low_price="95",
close_price="105",
),
_candle(
index=2,
high_price="112",
low_price="100",
close_price="108",
),
]
result = atr(candles, period=2)
assert result == 13.5
assert isinstance(result, float)
def test_atr_returns_none_when_candles_are_insufficient() -> None:
candles = [
_candle(index=0),
_candle(index=1),
]
assert atr(candles, period=2) is None
def test_atr_returns_none_for_non_positive_period() -> None:
candles = [
_candle(index=0),
_candle(index=1),
]
assert atr(candles, period=0) is None
def test_atr_skips_pair_with_non_finite_decimal_value() -> None:
candles = [
_candle(index=0, close_price="100"),
_candle(index=1, high_price="NaN", low_price="95"),
_candle(index=2, high_price="112", low_price="100"),
]
assert atr(candles, period=2) is None
def test_atr_percent_baseline_accepts_decimal_candles() -> None:
candles = [
_candle(
index=index,
open_price=str(100 + index),
high_price=str(105 + index),
low_price=str(95 + index),
close_price=str(101 + index),
)
for index in range(8)
]
result = atr_percent_baseline(
candles=candles,
close_price=108.0,
atr_period=2,
atr_baseline_window=8,
)
assert result is not None
assert isinstance(result, float)
assert result > 0
def test_atr_percent_baseline_returns_none_for_invalid_close_price() -> None:
candles = [
_candle(index=index)
for index in range(5)
]
assert (
atr_percent_baseline(
candles=candles,
close_price=0.0,
atr_period=2,
atr_baseline_window=5,
)
is None
)

154
app/tests/unit/trading/market_analysis/test_quality_candle.py Normal file
View File

@@ -0,0 +1,154 @@
# app/tests/unit/trading/market_analysis/test_quality_candle.py
from __future__ import annotations
from datetime import datetime, timezone
from decimal import Decimal
from src.market_data.acquisition.models.candle import Candle
def _candle(
*,
index: int = 0,
open_price: str = "100",
high_price: str = "110",
low_price: str = "90",
close_price: str = "105",
volume: str = "10",
) -> Candle:
return Candle(
symbol="BTC/USD_LEVERAGE",
interval="1m",
open_time=datetime.fromtimestamp(
1_750_000_000 + index * 60,
tz=timezone.utc,
),
open_price=Decimal(open_price),
high_price=Decimal(high_price),
low_price=Decimal(low_price),
close_price=Decimal(close_price),
volume=Decimal(volume),
source="test",
)
from src.trading.market_analysis.quality import candle_noise_score
def test_candle_noise_score_accepts_decimal_candles() -> None:
candles = [
_candle(
index=0,
open_price="91",
high_price="100",
low_price="90",
close_price="99",
),
_candle(
index=1,
open_price="94",
high_price="100",
low_price="90",
close_price="96",
),
]
result = candle_noise_score(
candles,
candle_noise_window=2,
min_clean_body_ratio=0.5,
)
assert result == 0.5
assert isinstance(result, float)
def test_candle_noise_score_returns_none_for_empty_sequence() -> None:
assert (
candle_noise_score(
[],
candle_noise_window=10,
min_clean_body_ratio=0.5,
)
is None
)
def test_candle_noise_score_skips_zero_range_candle() -> None:
candles = [
_candle(
index=0,
open_price="100",
high_price="100",
low_price="100",
close_price="100",
),
_candle(
index=1,
open_price="91",
high_price="100",
low_price="90",
close_price="99",
),
]
result = candle_noise_score(
candles,
candle_noise_window=2,
min_clean_body_ratio=0.5,
)
assert result == 1.0
def test_candle_noise_score_skips_non_finite_decimal_value() -> None:
candles = [
_candle(
index=0,
high_price="NaN",
),
]
assert (
candle_noise_score(
candles,
candle_noise_window=1,
min_clean_body_ratio=0.5,
)
is None
)
def test_candle_noise_score_uses_requested_tail_window() -> None:
candles = [
_candle(
index=0,
open_price="91",
high_price="100",
low_price="90",
close_price="99",
),
_candle(
index=1,
open_price="94",
high_price="100",
low_price="90",
close_price="96",
),
_candle(
index=2,
open_price="92",
high_price="100",
low_price="90",
close_price="98",
),
]
result = candle_noise_score(
candles,
candle_noise_window=2,
min_clean_body_ratio=0.5,
)
assert result == 0.5

170
app/tests/unit/trading/market_analysis/test_structure_candle.py Normal file
View File

@@ -0,0 +1,170 @@
# app/tests/unit/trading/market_analysis/test_structure_candle.py
from __future__ import annotations
from datetime import datetime, timezone
from decimal import Decimal
from src.market_data.acquisition.models.candle import Candle
def _candle(
*,
index: int = 0,
open_price: str = "100",
high_price: str = "110",
low_price: str = "90",
close_price: str = "105",
volume: str = "10",
) -> Candle:
return Candle(
symbol="BTC/USD_LEVERAGE",
interval="1m",
open_time=datetime.fromtimestamp(
1_750_000_000 + index * 60,
tz=timezone.utc,
),
open_price=Decimal(open_price),
high_price=Decimal(high_price),
low_price=Decimal(low_price),
close_price=Decimal(close_price),
volume=Decimal(volume),
source="test",
)
from src.trading.market_analysis.models import MarketStructure
from src.trading.market_analysis.structure import market_structure
def _structure_candles(
*,
highs: list[float],
lows: list[float],
) -> list[Candle]:
candles: list[Candle] = []
for index, (high, low) in enumerate(zip(highs, lows)):
midpoint = (high + low) / 2
candles.append(
_candle(
index=index,
open_price=str(midpoint),
high_price=str(high),
low_price=str(low),
close_price=str(midpoint),
)
)
return candles
def test_market_structure_detects_higher_highs_and_higher_lows() -> None:
candles = _structure_candles(
highs=[2, 4, 3, 5, 4, 6, 5, 7, 6, 5],
lows=[1, 0, 1, 0.5, 1.5, 1, 2, 1.5, 2.5, 2],
)
result, reason = market_structure(
candles,
atr_percent=0.30,
structure_window=10,
structure_swing_left=1,
structure_swing_right=1,
)
assert result == MarketStructure.HH_HL
assert reason.startswith("HIGHER_HIGH_HIGHER_LOW:")
def test_market_structure_detects_lower_highs_and_lower_lows() -> None:
candles = _structure_candles(
highs=[
8.0,
10.0,
9.0,
9.5,
8.5,
9.0,
8.0,
8.5,
7.5,
7.0,
],
lows=[
7.0,
6.0,
7.0,
5.5,
6.5,
5.0,
6.0,
4.5,
5.5,
5.0,
],
)
result, reason = market_structure(
candles,
atr_percent=0.30,
structure_window=10,
structure_swing_left=1,
structure_swing_right=1,
)
assert result == MarketStructure.LH_LL
assert reason.startswith("LOWER_HIGH_LOWER_LOW:")
def test_market_structure_detects_mixed_structure() -> None:
candles = _structure_candles(
highs=[2, 4, 3, 5, 4, 6, 5, 7, 6, 5],
lows=[5, 4, 5, 3, 4, 2, 3, 1, 2, 1.5],
)
result, reason = market_structure(
candles,
atr_percent=0.30,
structure_window=10,
structure_swing_left=1,
structure_swing_right=1,
)
assert result == MarketStructure.MIXED
assert reason.startswith("MIXED_MARKET_STRUCTURE:")
def test_market_structure_returns_unknown_when_candles_are_insufficient() -> None:
candles = [
_candle(index=index)
for index in range(5)
]
result, reason = market_structure(
candles,
structure_window=10,
structure_swing_left=1,
structure_swing_right=1,
)
assert result == MarketStructure.UNKNOWN
assert reason == "STRUCTURE_NOT_ENOUGH_CANDLES"
def test_market_structure_accepts_decimal_candle_values() -> None:
candles = _structure_candles(
highs=[2, 4, 3, 5, 4, 6, 5, 7, 6, 5],
lows=[1, 0, 1, 0.5, 1.5, 1, 2, 1.5, 2.5, 2],
)
result, _ = market_structure(
candles,
atr_percent=0.30,
structure_window=10,
structure_swing_left=1,
structure_swing_right=1,
)
assert result == MarketStructure.HH_HL