134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import pandas as pd
|
|
|
|
from web_core.constants import TREND_BEAR, TREND_BULL, TREND_NEUTRAL
|
|
from web_core.strategy import classify_bars, detect_trends
|
|
|
|
|
|
def make_df(rows: list[dict[str, float]]) -> pd.DataFrame:
|
|
idx = pd.date_range("2025-01-01", periods=len(rows), freq="D")
|
|
return pd.DataFrame(rows, index=idx)
|
|
|
|
|
|
def test_classify_bars_real_bull_real_bear_and_fake() -> None:
|
|
df = make_df(
|
|
[
|
|
{"Open": 100, "High": 110, "Low": 95, "Close": 105, "Volume": 1000},
|
|
{"Open": 106, "High": 111, "Low": 102, "Close": 112, "Volume": 1000},
|
|
{"Open": 111, "High": 112, "Low": 100, "Close": 99, "Volume": 1000},
|
|
{"Open": 100, "High": 103, "Low": 97, "Close": 101, "Volume": 1000},
|
|
]
|
|
)
|
|
|
|
out = classify_bars(
|
|
df,
|
|
use_body_range=False,
|
|
volume_filter_enabled=False,
|
|
volume_sma_window=20,
|
|
volume_multiplier=1.0,
|
|
)
|
|
|
|
assert list(out["classification"]) == ["unclassified", "real_bull", "real_bear", "fake"]
|
|
|
|
|
|
def test_classify_bars_with_body_range_ignores_wicks() -> None:
|
|
df = make_df(
|
|
[
|
|
{"Open": 100, "High": 130, "Low": 90, "Close": 105, "Volume": 1000},
|
|
{"Open": 106, "High": 109, "Low": 104, "Close": 108, "Volume": 1000},
|
|
]
|
|
)
|
|
|
|
wick_based = classify_bars(
|
|
df,
|
|
use_body_range=False,
|
|
volume_filter_enabled=False,
|
|
volume_sma_window=20,
|
|
volume_multiplier=1.0,
|
|
)
|
|
body_based = classify_bars(
|
|
df,
|
|
use_body_range=True,
|
|
volume_filter_enabled=False,
|
|
volume_sma_window=20,
|
|
volume_multiplier=1.0,
|
|
)
|
|
|
|
assert wick_based.iloc[1]["classification"] == "fake"
|
|
assert body_based.iloc[1]["classification"] == "real_bull"
|
|
|
|
|
|
def test_detect_trends_fake_bars_do_not_break_real_bar_sequence() -> None:
|
|
df = make_df(
|
|
[
|
|
{"Open": 100, "High": 110, "Low": 95, "Close": 105, "Volume": 1000},
|
|
{"Open": 106, "High": 112, "Low": 104, "Close": 113, "Volume": 1000},
|
|
{"Open": 113, "High": 115, "Low": 108, "Close": 112, "Volume": 1000},
|
|
{"Open": 112, "High": 120, "Low": 111, "Close": 121, "Volume": 1000},
|
|
]
|
|
)
|
|
|
|
classified = classify_bars(
|
|
df,
|
|
use_body_range=False,
|
|
volume_filter_enabled=False,
|
|
volume_sma_window=20,
|
|
volume_multiplier=1.0,
|
|
)
|
|
trended, events = detect_trends(classified)
|
|
|
|
assert list(classified["classification"]) == ["unclassified", "real_bull", "fake", "real_bull"]
|
|
assert trended.iloc[-1]["trend_state"] == TREND_BULL
|
|
assert any(e.event == "Bullish trend started" for e in events)
|
|
|
|
|
|
def test_detect_trends_reversal_requires_two_opposite_real_bars() -> None:
|
|
df = make_df(
|
|
[
|
|
{"Open": 100, "High": 110, "Low": 95, "Close": 105, "Volume": 1000},
|
|
{"Open": 106, "High": 112, "Low": 104, "Close": 113, "Volume": 1000},
|
|
{"Open": 113, "High": 120, "Low": 111, "Close": 121, "Volume": 1000},
|
|
{"Open": 120, "High": 121, "Low": 109, "Close": 108, "Volume": 1000},
|
|
{"Open": 108, "High": 109, "Low": 95, "Close": 94, "Volume": 1000},
|
|
]
|
|
)
|
|
|
|
classified = classify_bars(
|
|
df,
|
|
use_body_range=False,
|
|
volume_filter_enabled=False,
|
|
volume_sma_window=20,
|
|
volume_multiplier=1.0,
|
|
)
|
|
trended, events = detect_trends(classified)
|
|
|
|
assert trended.iloc[2]["trend_state"] == TREND_BULL
|
|
assert trended.iloc[3]["trend_state"] == TREND_BULL
|
|
assert trended.iloc[4]["trend_state"] == TREND_BEAR
|
|
assert any("Bearish reversal confirmed" in e.event for e in events)
|
|
|
|
|
|
def test_detect_trends_remains_neutral_without_two_consecutive_real_bars() -> None:
|
|
df = make_df(
|
|
[
|
|
{"Open": 100, "High": 110, "Low": 95, "Close": 105, "Volume": 1000},
|
|
{"Open": 105, "High": 108, "Low": 102, "Close": 106, "Volume": 1000},
|
|
{"Open": 106, "High": 116, "Low": 100, "Close": 117, "Volume": 1000},
|
|
{"Open": 117, "High": 118, "Low": 111, "Close": 115, "Volume": 1000},
|
|
{"Open": 115, "High": 116, "Low": 104, "Close": 103, "Volume": 1000},
|
|
]
|
|
)
|
|
|
|
classified = classify_bars(
|
|
df,
|
|
use_body_range=False,
|
|
volume_filter_enabled=False,
|
|
volume_sma_window=20,
|
|
volume_multiplier=1.0,
|
|
)
|
|
trended, events = detect_trends(classified)
|
|
|
|
assert trended.iloc[-1]["trend_state"] == TREND_NEUTRAL
|
|
assert len(events) == 0
|