Format code with ruff to pass CI format check
This commit is contained in:
@@ -172,7 +172,9 @@ class RiskEngine:
|
||||
if equity <= Decimal("0"):
|
||||
raise ValueError("Equity must be positive to allocate exposure")
|
||||
if risk_amount <= Decimal("0"):
|
||||
logger.debug("Risk amount is non-positive", strategy=strategy, risk_amount=str(risk_amount))
|
||||
logger.debug(
|
||||
"Risk amount is non-positive", strategy=strategy, risk_amount=str(risk_amount)
|
||||
)
|
||||
return False
|
||||
|
||||
strategy_pct = max_per_strategy_pct or self._max_per_strategy_pct
|
||||
|
||||
@@ -45,7 +45,9 @@ class SupertrendStrategy(Strategy):
|
||||
multiplier=float(self._multiplier),
|
||||
)
|
||||
|
||||
direction_col = next((col for col in supertrend.columns if col.startswith("SUPERTd_")), None)
|
||||
direction_col = next(
|
||||
(col for col in supertrend.columns if col.startswith("SUPERTd_")), None
|
||||
)
|
||||
if direction_col is None:
|
||||
logger.debug("Supertrend direction series missing", strategy=self.name)
|
||||
return None
|
||||
@@ -110,7 +112,11 @@ class SupertrendStrategy(Strategy):
|
||||
|
||||
def get_stop_loss(self, entry_price: Decimal, side: Side) -> Decimal:
|
||||
"""Use ATR buffer for Supertrend stop loss."""
|
||||
atr_buffer = self._last_atr if self._last_atr and self._last_atr > Decimal("0") else entry_price * Decimal("0.02")
|
||||
atr_buffer = (
|
||||
self._last_atr
|
||||
if self._last_atr and self._last_atr > Decimal("0")
|
||||
else entry_price * Decimal("0.02")
|
||||
)
|
||||
if side == Side.BUY:
|
||||
stop = entry_price - atr_buffer
|
||||
else:
|
||||
@@ -150,7 +156,14 @@ class SupertrendStrategy(Strategy):
|
||||
|
||||
@staticmethod
|
||||
def _trend_level(supertrend: pd.DataFrame) -> Decimal | None:
|
||||
trend_col = next((col for col in supertrend.columns if col.startswith("SUPERT_") and not col.startswith("SUPERTd_")), None)
|
||||
trend_col = next(
|
||||
(
|
||||
col
|
||||
for col in supertrend.columns
|
||||
if col.startswith("SUPERT_") and not col.startswith("SUPERTd_")
|
||||
),
|
||||
None,
|
||||
)
|
||||
if trend_col is None:
|
||||
return None
|
||||
return SupertrendStrategy._decimal_from_series_tail(supertrend[trend_col])
|
||||
|
||||
@@ -19,7 +19,9 @@ from tradefinder.data.streamer import (
|
||||
MarkPriceMessage,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.skip(reason="Async WebSocket tests have timing issues - streamer verified manually")
|
||||
pytestmark = pytest.mark.skip(
|
||||
reason="Async WebSocket tests have timing issues - streamer verified manually"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -126,10 +126,14 @@ class TestSupertrendStrategySignals:
|
||||
candles = _build_candle_sequence(base_timestamp, partial)
|
||||
assert strategy.generate_signal(candles) is None
|
||||
|
||||
def test_generate_signal_empty_candles_returns_none(self, default_strategy: SupertrendStrategy) -> None:
|
||||
def test_generate_signal_empty_candles_returns_none(
|
||||
self, default_strategy: SupertrendStrategy
|
||||
) -> None:
|
||||
assert default_strategy.generate_signal([]) is None
|
||||
|
||||
def test_generate_signal_none_input_raises_type_error(self, default_strategy: SupertrendStrategy) -> None:
|
||||
def test_generate_signal_none_input_raises_type_error(
|
||||
self, default_strategy: SupertrendStrategy
|
||||
) -> None:
|
||||
with pytest.raises(TypeError):
|
||||
default_strategy.generate_signal(None) # type: ignore[arg-type]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user