summaryrefslogtreecommitdiff
path: root/services/strategy-engine/tests
diff options
context:
space:
mode:
Diffstat (limited to 'services/strategy-engine/tests')
-rw-r--r--services/strategy-engine/tests/test_bollinger_strategy.py42
-rw-r--r--services/strategy-engine/tests/test_ema_crossover_strategy.py60
-rw-r--r--services/strategy-engine/tests/test_macd_strategy.py4
3 files changed, 65 insertions, 41 deletions
diff --git a/services/strategy-engine/tests/test_bollinger_strategy.py b/services/strategy-engine/tests/test_bollinger_strategy.py
index 473d9b4..7761f2d 100644
--- a/services/strategy-engine/tests/test_bollinger_strategy.py
+++ b/services/strategy-engine/tests/test_bollinger_strategy.py
@@ -107,12 +107,14 @@ def test_bollinger_squeeze_detection():
"""Tight bandwidth → no signal during squeeze."""
# Use a strategy with a high squeeze threshold so constant prices trigger squeeze
s = BollingerStrategy()
- s.configure({
- "period": 5,
- "num_std": 2.0,
- "min_bandwidth": 0.0,
- "squeeze_threshold": 0.5, # Very high threshold to ensure squeeze triggers
- })
+ s.configure(
+ {
+ "period": 5,
+ "num_std": 2.0,
+ "min_bandwidth": 0.0,
+ "squeeze_threshold": 0.5, # Very high threshold to ensure squeeze triggers
+ }
+ )
# Feed identical prices → bandwidth = 0 (below any threshold)
for _ in range(6):
@@ -126,12 +128,14 @@ def test_bollinger_squeeze_detection():
def test_bollinger_squeeze_breakout_buy():
"""Squeeze ends with price above SMA → BUY signal."""
s = BollingerStrategy()
- s.configure({
- "period": 5,
- "num_std": 1.0,
- "min_bandwidth": 0.0,
- "squeeze_threshold": 0.01,
- })
+ s.configure(
+ {
+ "period": 5,
+ "num_std": 1.0,
+ "min_bandwidth": 0.0,
+ "squeeze_threshold": 0.01,
+ }
+ )
# Feed identical prices to create a squeeze (bandwidth = 0)
for _ in range(6):
@@ -149,12 +153,14 @@ def test_bollinger_squeeze_breakout_buy():
def test_bollinger_pct_b_conviction():
"""Signals near band extremes have higher conviction via %B."""
s = BollingerStrategy()
- s.configure({
- "period": 5,
- "num_std": 1.0,
- "min_bandwidth": 0.0,
- "squeeze_threshold": 0.0, # Disable squeeze for this test
- })
+ s.configure(
+ {
+ "period": 5,
+ "num_std": 1.0,
+ "min_bandwidth": 0.0,
+ "squeeze_threshold": 0.0, # Disable squeeze for this test
+ }
+ )
# Build up with stable prices
for _ in range(5):
diff --git a/services/strategy-engine/tests/test_ema_crossover_strategy.py b/services/strategy-engine/tests/test_ema_crossover_strategy.py
index 9e48478..67a20bf 100644
--- a/services/strategy-engine/tests/test_ema_crossover_strategy.py
+++ b/services/strategy-engine/tests/test_ema_crossover_strategy.py
@@ -21,9 +21,18 @@ def make_candle(close: float) -> Candle:
)
-def _make_strategy(short: int = 3, long: int = 6, pullback_enabled: bool = False) -> EmaCrossoverStrategy:
+def _make_strategy(
+ short: int = 3, long: int = 6, pullback_enabled: bool = False
+) -> EmaCrossoverStrategy:
s = EmaCrossoverStrategy()
- s.configure({"short_period": short, "long_period": long, "quantity": "0.01", "pullback_enabled": pullback_enabled})
+ s.configure(
+ {
+ "short_period": short,
+ "long_period": long,
+ "quantity": "0.01",
+ "pullback_enabled": pullback_enabled,
+ }
+ )
return s
@@ -103,13 +112,15 @@ def test_ema_reset_clears_state():
def test_ema_pullback_entry():
"""Crossover detected, then pullback to short EMA triggers signal."""
strategy = EmaCrossoverStrategy()
- strategy.configure({
- "short_period": 3,
- "long_period": 6,
- "quantity": "0.01",
- "pullback_enabled": True,
- "pullback_tolerance": 0.05, # 5% tolerance for test simplicity
- })
+ strategy.configure(
+ {
+ "short_period": 3,
+ "long_period": 6,
+ "quantity": "0.01",
+ "pullback_enabled": True,
+ "pullback_tolerance": 0.05, # 5% tolerance for test simplicity
+ }
+ )
# Declining prices so short EMA stays below long EMA
declining = [100, 98, 96, 94, 92, 90, 88, 86, 84, 82]
@@ -129,6 +140,7 @@ def test_ema_pullback_entry():
# The short EMA will be tracking recent prices; feed a price that pulls back
# toward it. We use a moderate price to get close to short EMA.
import pandas as pd
+
series = pd.Series(list(strategy._closes))
short_ema_val = series.ewm(span=3, adjust=False).mean().iloc[-1]
# Feed a candle at approximately the short EMA value
@@ -141,13 +153,15 @@ def test_ema_pullback_entry():
def test_ema_pullback_cancelled_on_reversal():
"""Crossover detected, then reversal cancels the pending signal."""
strategy = EmaCrossoverStrategy()
- strategy.configure({
- "short_period": 3,
- "long_period": 6,
- "quantity": "0.01",
- "pullback_enabled": True,
- "pullback_tolerance": 0.001, # Very tight tolerance — won't trigger easily
- })
+ strategy.configure(
+ {
+ "short_period": 3,
+ "long_period": 6,
+ "quantity": "0.01",
+ "pullback_enabled": True,
+ "pullback_tolerance": 0.001, # Very tight tolerance — won't trigger easily
+ }
+ )
# Declining prices
declining = [100, 98, 96, 94, 92, 90, 88, 86, 84, 82]
@@ -172,12 +186,14 @@ def test_ema_pullback_cancelled_on_reversal():
def test_ema_immediate_mode():
"""With pullback_enabled=False, original immediate entry works."""
strategy = EmaCrossoverStrategy()
- strategy.configure({
- "short_period": 3,
- "long_period": 6,
- "quantity": "0.01",
- "pullback_enabled": False,
- })
+ strategy.configure(
+ {
+ "short_period": 3,
+ "long_period": 6,
+ "quantity": "0.01",
+ "pullback_enabled": False,
+ }
+ )
# Declining prices so short EMA stays below long EMA
declining = [100, 98, 96, 94, 92, 90, 88, 86, 84, 82]
diff --git a/services/strategy-engine/tests/test_macd_strategy.py b/services/strategy-engine/tests/test_macd_strategy.py
index cd24ee0..17dd2cf 100644
--- a/services/strategy-engine/tests/test_macd_strategy.py
+++ b/services/strategy-engine/tests/test_macd_strategy.py
@@ -98,7 +98,9 @@ def test_macd_signal_line_crossover():
assert len(buy_signals) > 0, "Expected at least one BUY signal"
# Check that at least one is a signal-line crossover or histogram crossover
all_reasons = [sig.reason for sig in buy_signals]
- assert any("crossover" in r for r in all_reasons), f"Expected crossover signal, got: {all_reasons}"
+ assert any("crossover" in r for r in all_reasons), (
+ f"Expected crossover signal, got: {all_reasons}"
+ )
def test_macd_conviction_varies_with_distance():