From aa1f65a8925c6c3ab35043928519f4edcf2e5dc6 Mon Sep 17 00:00:00 2001 From: Jeffrey Date: Mon, 13 Apr 2026 22:54:26 +0800 Subject: [PATCH] added testing units for price volume strategies --- freqtrade | 2 +- strategy/adaptive_trend/adaptive_trend.py | 5 +- tests/test_adaptive_trend_strategy.py | 448 ++++++++++++++++++ tests/test_beta_factors_strategy.py | 321 +++++++++++++ tests/test_polymarket_logical_arb_strategy.py | 367 ++++++++++++++ 5 files changed, 1140 insertions(+), 3 deletions(-) create mode 100644 tests/test_adaptive_trend_strategy.py create mode 100644 tests/test_beta_factors_strategy.py create mode 100644 tests/test_polymarket_logical_arb_strategy.py diff --git a/freqtrade b/freqtrade index ab093ff..8c7385d 160000 --- a/freqtrade +++ b/freqtrade @@ -1 +1 @@ -Subproject commit ab093ff0e1af445f0b8491ea1168c46e1a51b2c0 +Subproject commit 8c7385dc98ecab1359edd878afcff91fffdec683 diff --git a/strategy/adaptive_trend/adaptive_trend.py b/strategy/adaptive_trend/adaptive_trend.py index 2039a8f..db60ce7 100644 --- a/strategy/adaptive_trend/adaptive_trend.py +++ b/strategy/adaptive_trend/adaptive_trend.py @@ -264,9 +264,10 @@ def custom_stoploss( We store the stop price in trade custom data so it “trails”. Returns stoploss as a negative value (relative stop) as required by Freqtrade. """ - df = self.dp.get_analyzed_dataframe(pair, self.timeframe) + df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) + if df is None or df.empty: - return 1 # keep default / do nothing + return 1 last = df.iloc[-1] atr = last.get("atr", None) diff --git a/tests/test_adaptive_trend_strategy.py b/tests/test_adaptive_trend_strategy.py new file mode 100644 index 0000000..24bb494 --- /dev/null +++ b/tests/test_adaptive_trend_strategy.py @@ -0,0 +1,448 @@ +"""Tests for adaptive_trend strategy.""" + +# to test this file, run the command on bash: python -m pytest tests/test_adaptive_trend_strategy.py -q + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import numpy as np +import pandas as pd + +from strategy.adaptive_trend.adaptive_trend import adaptive_trend + +def _make_ohlcv(n: int = 300, start: str = "2025-01-01") -> pd.DataFrame: + """Create synthetic 4h OHLCV data.""" + np.random.seed(42) + dates = pd.date_range(start, periods=n, freq="4H", tz="UTC") + close = 100 + np.cumsum(np.random.randn(n) * 0.5) + + return pd.DataFrame( + { + "date": dates, + "open": close - 0.2, + "high": close + 0.5, + "low": close - 0.5, + "close": close, + "volume": np.random.randint(100, 10000, size=n).astype(float), + } + ) + + +def _make_market_cap_csvs(base_dir: Path): + """Create fake market-cap CSV files beside the strategy file.""" + snapped = pd.date_range("2025-01-01", periods=20, freq="D", tz="UTC") + + btc = pd.DataFrame( + { + "snapped_at": snapped, + "market_cap": np.linspace(1_000_000_000, 1_200_000_000, len(snapped)), + } + ) + eth = pd.DataFrame( + { + "snapped_at": snapped, + "market_cap": np.linspace(800_000_000, 900_000_000, len(snapped)), + } + ) + xrp = pd.DataFrame( + { + "snapped_at": snapped, + "market_cap": np.linspace(200_000_000, 250_000_000, len(snapped)), + } + ) + + btc.to_csv(base_dir / "btc-usd-max.csv", index=False) + eth.to_csv(base_dir / "eth-usd-max.csv", index=False) + xrp.to_csv(base_dir / "xrp-usd-max.csv", index=False) + + +class DummyDP: + """Minimal dataprovider stub for custom_stoploss.""" + + def __init__(self, analyzed_df: pd.DataFrame): + self.analyzed_df = analyzed_df + + def get_analyzed_dataframe(self, pair: str, timeframe: str): + return self.analyzed_df, None + + +class DummyTrade: + """Minimal trade stub for stoploss / exit tests.""" + + def __init__(self, is_short: bool = False, open_date_utc: datetime | None = None): + self.is_short = is_short + self.user_data = {} + self.open_date_utc = open_date_utc or datetime.now(timezone.utc) + + def get_custom_data(self, key: str): + return self.user_data.get(key) + + def set_custom_data(self, key: str, val): + self.user_data[key] = val + + +class TestAdaptiveTrendHelpers: + def test_base_symbol_extracts_base(self): + strat = adaptive_trend({}) + assert strat._base_symbol("ETH/USDT") == "ETH" + assert strat._base_symbol("btc/usdc") == "BTC" + + +class TestAdaptiveTrendMarketCap: + def test_load_market_cap_data_returns_dataframe(self, monkeypatch, tmp_path): + strat = adaptive_trend({}) + fake_strategy_file = tmp_path / "adaptive_trend.py" + fake_strategy_file.write_text("# dummy") + + _make_market_cap_csvs(tmp_path) + + monkeypatch.setattr(adaptive_trend, "__module__", adaptive_trend.__module__) + monkeypatch.setattr( + Path, + "resolve", + lambda self: fake_strategy_file, + ) + + result = strat.load_market_cap_data() + + assert result is not None + assert not result.empty + for col in ["date", "symbol", "marketCap", "rank", "total_count"]: + assert col in result.columns + + def test_merge_market_cap_adds_expected_columns(self): + strat = adaptive_trend({}) + df = _make_ohlcv(50) + + mcap = pd.DataFrame( + { + "date": pd.to_datetime(df["date"], utc=True).dt.floor("D").unique().repeat(3), + "symbol": ["BTC", "ETH", "XRP"] * len(pd.to_datetime(df["date"], utc=True).dt.floor("D").unique()), + "marketCap": [1000, 800, 200] * len(pd.to_datetime(df["date"], utc=True).dt.floor("D").unique()), + } + ) + mcap["rank"] = mcap.groupby("date")["marketCap"].rank(ascending=False, method="min") + mcap["total_count"] = mcap.groupby("date")["symbol"].transform("count") + + strat._mcap_df = mcap + result = strat._merge_market_cap(df.copy(), "BTC/USDT") + + for col in ["mcap_rank", "mcap_total", "allow_long_mcap", "allow_short_mcap"]: + assert col in result.columns + + assert set(result["allow_long_mcap"].unique()).issubset({0, 1}) + assert set(result["allow_short_mcap"].unique()).issubset({0, 1}) + + def test_merge_market_cap_defaults_to_allow_when_no_data(self): + strat = adaptive_trend({}) + strat._mcap_df = None + + # Force loader to return None + strat.load_market_cap_data = lambda: None + + df = _make_ohlcv(20) + result = strat._merge_market_cap(df.copy(), "BTC/USDT") + + assert (result["allow_long_mcap"] == 1).all() + assert (result["allow_short_mcap"] == 1).all() + + +class TestAdaptiveTrendIndicators: + def test_populate_indicators_adds_expected_columns(self, monkeypatch): + strat = adaptive_trend({}) + strat._mcap_df = None + strat.load_market_cap_data = lambda: None + + df = _make_ohlcv(300) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + expected_cols = [ + "mom", + "atr", + "ret", + "sr_long", + "sr_short", + "mcap_rank", + "mcap_total", + "allow_long_mcap", + "allow_short_mcap", + ] + + for col in expected_cols: + assert col in result.columns, f"Missing column: {col}" + + def test_momentum_has_nans_before_lookback(self): + strat = adaptive_trend({}) + strat._mcap_df = None + strat.load_market_cap_data = lambda: None + + df = _make_ohlcv(300) + L = int(strat.mom_lookback.value) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + assert result["mom"].iloc[:L].isna().all() + + def test_atr_is_numeric_after_warmup(self): + strat = adaptive_trend({}) + strat._mcap_df = None + strat.load_market_cap_data = lambda: None + + df = _make_ohlcv(300) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + valid = result["atr"].dropna() + assert not valid.empty + assert valid.dtype in [np.float64, np.float32] + + def test_sharpe_columns_are_numeric(self): + strat = adaptive_trend({}) + strat._mcap_df = None + strat.load_market_cap_data = lambda: None + + df = _make_ohlcv(300) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + assert result["sr_long"].dtype in [np.float64, np.float32] + assert result["sr_short"].dtype in [np.float64, np.float32] + + +class TestAdaptiveTrendEntries: + def test_populate_entry_trend_sets_enter_long(self): + strat = adaptive_trend({}) + + df = pd.DataFrame( + { + "volume": [1000, 1000, 1000], + "mom": [0.01, 0.05, 0.10], + "sr_long": [0.1, 0.5, 1.0], + "sr_short": [0.1, 0.1, 0.1], + "allow_long_mcap": [1, 1, 1], + "allow_short_mcap": [0, 0, 0], + } + ) + + result = strat.populate_entry_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + assert "enter_long" in result.columns + assert result["enter_long"].fillna(0).sum() >= 1 + + def test_populate_entry_trend_sets_enter_short_when_conditions_met(self): + strat = adaptive_trend({}) + + df = pd.DataFrame( + { + "volume": [1000, 1000, 1000], + "mom": [-0.01, -0.10, -0.20], + "sr_long": [0.1, 0.1, 0.1], + "sr_short": [0.5, 2.0, 2.5], + "allow_long_mcap": [0, 0, 0], + "allow_short_mcap": [1, 1, 1], + } + ) + + result = strat.populate_entry_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + assert "enter_short" in result.columns + assert result["enter_short"].fillna(0).sum() >= 1 + + def test_no_entry_when_volume_zero(self): + strat = adaptive_trend({}) + + df = pd.DataFrame( + { + "volume": [0, 0], + "mom": [0.10, -0.10], + "sr_long": [10, 10], + "sr_short": [10, 10], + "allow_long_mcap": [1, 1], + "allow_short_mcap": [1, 1], + } + ) + + result = strat.populate_entry_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + assert result.get("enter_long", pd.Series([0, 0])).fillna(0).sum() == 0 + assert result.get("enter_short", pd.Series([0, 0])).fillna(0).sum() == 0 + + +class TestAdaptiveTrendExits: + def test_populate_exit_trend_sets_zero_exits(self): + strat = adaptive_trend({}) + df = _make_ohlcv(20) + + result = strat.populate_exit_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + + assert "exit_long" in result.columns + assert "exit_short" in result.columns + assert (result["exit_long"] == 0).all() + assert (result["exit_short"] == 0).all() + + def test_custom_exit_returns_time_exit_after_hold_days(self): + strat = adaptive_trend({}) + trade = DummyTrade(open_date_utc=datetime.now(timezone.utc) - timedelta(days=61)) + + result = strat.custom_exit( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.02, + ) + + assert result == "time_exit" + + def test_custom_exit_returns_none_before_hold_days(self): + strat = adaptive_trend({}) + trade = DummyTrade(open_date_utc=datetime.now(timezone.utc) - timedelta(days=10)) + + result = strat.custom_exit( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.02, + ) + + assert result is None + + +class TestAdaptiveTrendCustomStoploss: + def test_custom_stoploss_returns_default_when_no_dataframe(self): + strat = adaptive_trend({}) + strat.dp = DummyDP(pd.DataFrame()) + + trade = DummyTrade(is_short=False) + result = strat.custom_stoploss( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.01, + ) + + assert result == 1 + + def test_custom_stoploss_long_sets_trailing_stop(self): + strat = adaptive_trend({}) + + analyzed = pd.DataFrame( + { + "atr": [1.5, 2.0], + } + ) + strat.dp = DummyDP(analyzed) + + trade = DummyTrade(is_short=False) + result = strat.custom_stoploss( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.01, + ) + + assert -0.99 <= result <= 0.0 + assert "atr_trail_stop" in trade.user_data + assert trade.user_data["atr_trail_stop"] < 100.0 + + def test_custom_stoploss_short_sets_trailing_stop(self): + strat = adaptive_trend({}) + + analyzed = pd.DataFrame( + { + "atr": [1.5, 2.0], + } + ) + strat.dp = DummyDP(analyzed) + + trade = DummyTrade(is_short=True) + result = strat.custom_stoploss( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.01, + ) + + assert -0.99 <= result <= 0.0 + assert "atr_trail_stop" in trade.user_data + assert trade.user_data["atr_trail_stop"] > 100.0 + + def test_custom_stoploss_long_only_trails_up(self): + strat = adaptive_trend({}) + strat.dp = DummyDP(pd.DataFrame({"atr": [2.0]})) + + trade = DummyTrade(is_short=False) + + first = strat.custom_stoploss( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.01, + ) + first_stop = trade.user_data["atr_trail_stop"] + + second = strat.custom_stoploss( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=95.0, + current_profit=-0.03, + ) + second_stop = trade.user_data["atr_trail_stop"] + + assert second_stop >= first_stop + assert -0.99 <= first <= 0.0 + assert -0.99 <= second <= 0.0 + + +class TestAdaptiveTrendStakeSizing: + def test_custom_stake_amount_long_uses_proposed_stake(self): + strat = adaptive_trend({}) + + stake = strat.custom_stake_amount( + pair="BTC/USDT", + current_time=datetime.now(timezone.utc), + current_rate=100.0, + proposed_stake=1000.0, + min_stake=100.0, + max_stake=5000.0, + leverage=1.0, + side="long", + ) + + assert 100.0 <= stake <= 5000.0 + assert stake == 1000.0 + + def test_custom_stake_amount_short_scales_down(self): + strat = adaptive_trend({}) + + stake = strat.custom_stake_amount( + pair="BTC/USDT", + current_time=datetime.now(timezone.utc), + current_rate=100.0, + proposed_stake=1000.0, + min_stake=100.0, + max_stake=5000.0, + leverage=1.0, + side="short", + ) + + assert 100.0 <= stake <= 5000.0 + assert stake < 1000.0 + + def test_custom_stake_amount_clamps_to_bounds(self): + strat = adaptive_trend({}) + + stake = strat.custom_stake_amount( + pair="BTC/USDT", + current_time=datetime.now(timezone.utc), + current_rate=100.0, + proposed_stake=50.0, + min_stake=100.0, + max_stake=5000.0, + leverage=1.0, + side="long", + ) + + assert stake == 100.0 \ No newline at end of file diff --git a/tests/test_beta_factors_strategy.py b/tests/test_beta_factors_strategy.py new file mode 100644 index 0000000..370d301 --- /dev/null +++ b/tests/test_beta_factors_strategy.py @@ -0,0 +1,321 @@ +"""Tests for beta_factors_model strategy.""" + +# to test this file, run the command on bash: python -m pytest tests/test_beta_factors_strategy.py -q + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import numpy as np +import pandas as pd + +from strategy.crypto_factors_regression.beta_factors_model import beta_factors_model + + +def _make_weekly_ohlcv(n: int = 20, start: str = "2025-01-05") -> pd.DataFrame: + """Create synthetic weekly OHLCV data.""" + np.random.seed(42) + dates = pd.date_range(start, periods=n, freq="7D", tz="UTC") + close = 100 + np.cumsum(np.random.randn(n) * 2.0) + + return pd.DataFrame( + { + "date": dates, + "open": close - 1.0, + "high": close + 2.0, + "low": close - 2.0, + "close": close, + "volume": np.random.randint(100, 10000, size=n).astype(float), + } + ) + + +def _make_marketcap_csv(tmp_path: Path) -> Path: + """Create fake Bitcoin market cap CSV.""" + dates = pd.date_range("2024-12-30", periods=40, freq="D", tz="UTC") + df = pd.DataFrame( + { + "timeClose": dates, + "marketCap": np.linspace(1_000_000_000, 1_500_000_000, len(dates)), + } + ) + path = tmp_path / "Bitcoin_marketcap.csv" + df.to_csv(path, sep=";", index=False) + return path + + +class DummyModel: + """Minimal sklearn-like model stub.""" + + def __init__(self, preds): + self.preds = np.asarray(preds) + + def predict(self, X): + n = len(X) + if self.preds.ndim == 0: + return np.full((n,), float(self.preds)) + if len(self.preds) >= n: + return self.preds[:n] + return np.resize(self.preds, n) + + +class DummyTrade: + """Minimal trade stub for custom_exit tests.""" + + def __init__(self, open_date_utc: datetime | None = None): + self.open_date_utc = open_date_utc or datetime.now(timezone.utc) + + +class TestBetaFactorsModelHelpers: + def test_get_model_caches_loaded_model(self, monkeypatch): + strat = beta_factors_model({}) + dummy_model = DummyModel([0.1]) + + calls = {"count": 0} + + def fake_load(path): + calls["count"] += 1 + return dummy_model + + monkeypatch.setattr("strategy.crypto_factors_regression.beta_factors_model.joblib.load", fake_load) + + model1 = strat.get_model() + model2 = strat.get_model() + + assert model1 is dummy_model + assert model2 is dummy_model + assert calls["count"] == 1 + + +class TestBetaFactorsModelMarketCap: + def test_load_market_cap_data_returns_dataframe(self, monkeypatch, tmp_path): + strat = beta_factors_model({}) + csv_path = _make_marketcap_csv(tmp_path) + + fake_strategy_file = tmp_path / "beta_factors_model.py" + fake_strategy_file.write_text("# dummy") + + class DummyResolvedPath(type(Path())): + pass + + monkeypatch.setattr( + "strategy.crypto_factors_regression.beta_factors_model.Path.resolve", + lambda self: fake_strategy_file, + ) + + result = strat.load_market_cap_data() + + assert result is not None + assert not result.empty + assert "timeClose" in result.columns + assert "marketCap" in result.columns + assert "marketCap_shifted" in result.columns + + def test_load_market_cap_data_is_cached(self, monkeypatch): + strat = beta_factors_model({}) + cached = pd.DataFrame({"x": [1]}) + strat.btc_cap = cached + + result = strat.load_market_cap_data() + + assert result is cached + + +class TestBetaFactorsModelIndicators: + def test_populate_indicators_adds_expected_columns(self): + strat = beta_factors_model({}) + + marketcap = pd.DataFrame( + { + "timeClose": pd.date_range("2025-01-01", periods=60, freq="D", tz="UTC").floor("D"), + "marketCap": np.linspace(1_000_000_000, 1_500_000_000, 60), + } + ) + marketcap["marketCap_shifted"] = marketcap["marketCap"].shift(1) + + strat.load_market_cap_data = lambda: marketcap + strat.get_model = lambda: DummyModel(np.full(20, 0.08)) + + df = _make_weekly_ohlcv(20) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + expected_cols = [ + "week_end", + "marketCap_shifted", + "ret", + "cmkt", + "cmom", + "csize", + "csize_cmkt", + "cmkt_2", + "cmom_3", + "pred_ret", + "ml_signal", + ] + + for col in expected_cols: + assert col in result.columns, f"Missing column: {col}" + + def test_pred_ret_is_numeric_and_aligned(self): + strat = beta_factors_model({}) + + marketcap = pd.DataFrame( + { + "timeClose": pd.date_range("2025-01-01", periods=60, freq="D", tz="UTC").floor("D"), + "marketCap": np.linspace(1_000_000_000, 1_500_000_000, 60), + } + ) + marketcap["marketCap_shifted"] = marketcap["marketCap"].shift(1) + + preds = np.linspace(-0.1, 0.1, 20) + strat.load_market_cap_data = lambda: marketcap + strat.get_model = lambda: DummyModel(preds) + + df = _make_weekly_ohlcv(20) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + assert len(result["pred_ret"]) == len(df) + assert result["pred_ret"].dtype in [np.float64, np.float32] + + def test_ml_signal_positive_negative_and_zero(self): + strat = beta_factors_model({}) + + marketcap = pd.DataFrame( + { + "timeClose": pd.date_range("2025-01-01", periods=60, freq="D", tz="UTC").floor("D"), + "marketCap": np.linspace(1_000_000_000, 1_500_000_000, 60), + } + ) + marketcap["marketCap_shifted"] = marketcap["marketCap"].shift(1) + + preds = np.array([0.10, -0.10, 0.00, 0.08, -0.08] * 4) + strat.load_market_cap_data = lambda: marketcap + strat.get_model = lambda: DummyModel(preds) + + df = _make_weekly_ohlcv(20) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + valid_signals = set(result["ml_signal"].unique()) + assert valid_signals.issubset({-1, 0, 1}) + + def test_ml_signal_forces_zero_when_features_contain_zero(self): + strat = beta_factors_model({}) + + marketcap = pd.DataFrame( + { + "timeClose": pd.date_range("2025-01-01", periods=20, freq="D", tz="UTC").floor("D"), + "marketCap": np.linspace(1_000_000_000, 1_100_000_000, 20), + } + ) + marketcap["marketCap_shifted"] = marketcap["marketCap"].shift(1) + + strat.load_market_cap_data = lambda: marketcap + strat.get_model = lambda: DummyModel(np.full(10, 0.5)) + + df = _make_weekly_ohlcv(10) + result = strat.populate_indicators(df.copy(), metadata={"pair": "BTC/USDT"}) + + # Early rows should become zero because rolling / pct-change features are zero-filled + assert (result["ml_signal"].iloc[:3] == 0).any() + + +class TestBetaFactorsModelEntriesAndExits: + def test_populate_entry_trend_sets_enter_long(self): + strat = beta_factors_model({}) + + df = pd.DataFrame( + { + "ml_signal": [0, 1, 1, -1], + "volume": [100, 100, 0, 100], + } + ) + + result = strat.populate_entry_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + + assert "enter_long" in result.columns + assert result["enter_long"].fillna(0).sum() == 1 + + def test_populate_exit_trend_sets_exit_long(self): + strat = beta_factors_model({}) + + df = pd.DataFrame( + { + "ml_signal": [0, -1, 1, -1], + "volume": [100, 100, 100, 0], + } + ) + + result = strat.populate_exit_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + + assert "exit_long" in result.columns + assert result["exit_long"].fillna(0).sum() == 1 + + def test_no_entry_when_volume_zero(self): + strat = beta_factors_model({}) + + df = pd.DataFrame( + { + "ml_signal": [1, 1], + "volume": [0, 0], + } + ) + + result = strat.populate_entry_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + assert result.get("enter_long", pd.Series([0, 0])).fillna(0).sum() == 0 + + def test_no_exit_when_volume_zero(self): + strat = beta_factors_model({}) + + df = pd.DataFrame( + { + "ml_signal": [-1, -1], + "volume": [0, 0], + } + ) + + result = strat.populate_exit_trend(df.copy(), metadata={"pair": "BTC/USDT"}) + assert result.get("exit_long", pd.Series([0, 0])).fillna(0).sum() == 0 + + +class TestBetaFactorsModelCustomExit: + def test_custom_exit_returns_time_exit_after_hold_days(self): + strat = beta_factors_model({}) + trade = DummyTrade(open_date_utc=datetime.now(timezone.utc) - timedelta(days=8)) + + result = strat.custom_exit( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.01, + ) + + assert result == "time_exit" + + def test_custom_exit_returns_none_before_hold_days(self): + strat = beta_factors_model({}) + trade = DummyTrade(open_date_utc=datetime.now(timezone.utc) - timedelta(days=3)) + + result = strat.custom_exit( + pair="BTC/USDT", + trade=trade, + current_time=datetime.now(timezone.utc), + current_rate=100.0, + current_profit=0.01, + ) + + assert result is None + + +class TestBetaFactorsModelParameters: + def test_strategy_static_attributes(self): + assert beta_factors_model.timeframe == "1w" + assert beta_factors_model.can_short is False + assert beta_factors_model.stoploss == -0.15 + assert beta_factors_model.startup_candle_count == 10 + + def test_threshold_parameters_exist(self): + strat = beta_factors_model({}) + assert hasattr(strat, "buy_threshold") + assert hasattr(strat, "sell_threshold") \ No newline at end of file diff --git a/tests/test_polymarket_logical_arb_strategy.py b/tests/test_polymarket_logical_arb_strategy.py new file mode 100644 index 0000000..1595a92 --- /dev/null +++ b/tests/test_polymarket_logical_arb_strategy.py @@ -0,0 +1,367 @@ +"""Tests for PolymarketLogicalArbStrategy.""" + +# to test this file, run the command on bash: python -m pytest tests/test_polymarket_logical_arb_strategy.py -q + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +from strategy.PolymarketLogicalArb.PolymarketLogicalArbStrategy import PolymarketLogicalArbStrategy + +def _make_pair_mapping_csv(tmp_path: Path) -> Path: + """Create a minimal pair mapping CSV for BTC threshold contracts.""" + pairs = [ + "WillThePriceOfBitcoinBeAbove900YES20260121/USDC", + "WillThePriceOfBitcoinBeAbove920YES20260121/USDC", + "WillThePriceOfBitcoinBeAbove960YES20260121/USDC", + ] + path = tmp_path / "freqtrade_pair_mapping.csv" + pd.DataFrame({"pair": pairs}).to_csv(path, index=False) + return path + + +def _make_strategy_ohlcv(n: int = 40, start: str = "2026-01-21") -> pd.DataFrame: + """Create a minimal OHLCV dataframe with deterministic 4h candles.""" + dates = pd.date_range(start, periods=n, freq="4H", tz="UTC") + close = np.linspace(0.30, 0.45, n) + + return pd.DataFrame( + { + "date": dates, + "open": close - 0.01, + "high": close + 0.02, + "low": close - 0.02, + "close": close, + "volume": np.full(n, 1000.0), + } + ) + + +class DummyDP: + """Minimal dataprovider stub for related pair lookups.""" + + def __init__(self, pair_to_df: dict[str, pd.DataFrame]): + self.pair_to_df = pair_to_df + + def get_pair_dataframe(self, pair: str, timeframe: str): + return self.pair_to_df.get(pair) + + +class TestPolymarketLogicalArbStrategyHelpers: + def test_normalize_raw_entry_to_pair_accepts_pair_format(self): + raw = "WillThePriceOfBitcoinBeAbove900YES20260121/USDC" + pair = PolymarketLogicalArbStrategy._normalize_raw_entry_to_pair(raw) + assert pair == raw + + def test_normalize_raw_entry_to_pair_converts_file_format(self): + raw = "WillThePriceOfBitcoinBeAbove900YES20260121_USDC.feather" + pair = PolymarketLogicalArbStrategy._normalize_raw_entry_to_pair(raw) + assert pair == "WillThePriceOfBitcoinBeAbove900YES20260121/USDC" + + def test_build_btc_relations_from_source(self, tmp_path): + csv_path = _make_pair_mapping_csv(tmp_path) + rels = PolymarketLogicalArbStrategy.build_btc_relations_from_source(str(csv_path)) + + rel_ids = {r["id"] for r in rels} + assert "btc_above_920_implies_above_900_20260121" in rel_ids + assert "btc_above_960_implies_above_900_20260121" in rel_ids + assert "btc_above_960_implies_above_920_20260121" in rel_ids + assert len(rels) == 3 + + def test_build_btc_relations_empty_source_raises(self, tmp_path): + path = tmp_path / "pairs.csv" + pd.DataFrame({"pair": ["ETH/USDC", "BTC/USDC"]}).to_csv(path, index=False) + + with pytest.raises(ValueError): + PolymarketLogicalArbStrategy.build_btc_relations_from_source(str(path)) + + +class TestPolymarketLogicalArbStrategyIndicators: + def setup_method(self): + PolymarketLogicalArbStrategy._PAIR_CACHE_BUILT = False + PolymarketLogicalArbStrategy.RELATIONSHIPS = [] + + def test_informative_pairs_returns_all_unique_pairs(self, tmp_path): + csv_path = _make_pair_mapping_csv(tmp_path) + + strategy = PolymarketLogicalArbStrategy({}) + strategy.PAIR_SOURCE_PATH = str(csv_path) + strategy._PAIR_CACHE_BUILT = False + strategy.RELATIONSHIPS = [] + + pairs = strategy.informative_pairs() + pair_names = {p[0] for p in pairs} + + assert "WillThePriceOfBitcoinBeAbove900YES20260121/USDC" in pair_names + assert "WillThePriceOfBitcoinBeAbove920YES20260121/USDC" in pair_names + assert "WillThePriceOfBitcoinBeAbove960YES20260121/USDC" in pair_names + assert all(tf == strategy.timeframe for _, tf in pairs) + + def test_populate_indicators_adds_expected_columns(self, tmp_path): + csv_path = _make_pair_mapping_csv(tmp_path) + + base = _make_strategy_ohlcv() + related_920 = base.copy() + related_960 = base.copy() + + related_920["close"] = np.linspace(0.20, 0.30, len(related_920)) + related_960["close"] = np.linspace(0.10, 0.20, len(related_960)) + + strategy = PolymarketLogicalArbStrategy({}) + strategy.PAIR_SOURCE_PATH = str(csv_path) + strategy.dp = DummyDP( + { + "WillThePriceOfBitcoinBeAbove920YES20260121/USDC": related_920, + "WillThePriceOfBitcoinBeAbove960YES20260121/USDC": related_960, + } + ) + strategy.DEBUG_PRINTS = False + strategy._PAIR_CACHE_BUILT = False + strategy.RELATIONSHIPS = [] + + result = strategy.populate_indicators( + base.copy(), + metadata={"pair": "WillThePriceOfBitcoinBeAbove900YES20260121/USDC"}, + ) + + expected_cols = [ + "logic_signal", + "logic_rel_id", + "logic_role", + "logic_best_gap", + "logic_best_gap_z", + "logic_subset_mom_1", + "logic_subset_mom_2", + "logic_superset_mom_1", + "logic_current_abs_ret_1", + "logic_subset_abs_ret_1", + "logic_superset_abs_ret_1", + "logic_rank_score", + "logic_rank_pct", + "bars_to_end", + "enough_history", + "contract_tradeable", + "dbg_reject_reason", + ] + + for col in expected_cols: + assert col in result.columns, f"Missing column: {col}" + + def test_contract_tradeable_is_binary(self, tmp_path): + csv_path = _make_pair_mapping_csv(tmp_path) + + base = _make_strategy_ohlcv() + base.loc[0, "close"] = 0.005 + base.loc[1, "close"] = 0.995 + + related_920 = _make_strategy_ohlcv() + related_960 = _make_strategy_ohlcv() + + strategy = PolymarketLogicalArbStrategy({}) + strategy.PAIR_SOURCE_PATH = str(csv_path) + strategy.dp = DummyDP( + { + "WillThePriceOfBitcoinBeAbove920YES20260121/USDC": related_920, + "WillThePriceOfBitcoinBeAbove960YES20260121/USDC": related_960, + } + ) + strategy.DEBUG_PRINTS = False + strategy._PAIR_CACHE_BUILT = False + strategy.RELATIONSHIPS = [] + + result = strategy.populate_indicators( + base.copy(), + metadata={"pair": "WillThePriceOfBitcoinBeAbove900YES20260121/USDC"}, + ) + + assert set(result["contract_tradeable"].unique()).issubset({0, 1}) + assert result.loc[0, "contract_tradeable"] == 0 + assert result.loc[1, "contract_tradeable"] == 0 + + +class TestPolymarketLogicalArbStrategySignals: + def setup_method(self): + PolymarketLogicalArbStrategy._PAIR_CACHE_BUILT = False + PolymarketLogicalArbStrategy.RELATIONSHIPS = [] + + def test_populate_entry_trend_adds_enter_columns(self): + strategy = PolymarketLogicalArbStrategy({}) + strategy.DEBUG_PRINTS = False + + n = 12 + df = pd.DataFrame( + { + "date": pd.date_range("2026-01-21", periods=n, freq="4H", tz="UTC"), + "close": np.full(n, 0.40), + "logic_role": ["superset"] * n, + "enough_history": [0] * 6 + [1] * 6, + "logic_rank_pct": [0.8, 0.7, 0.6, 0.5, 0.4, 0.35, 0.25, 0.09, 0.50, 0.20, 0.31, 0.05], + "bars_to_end": list(reversed(range(n))), + "contract_tradeable": [1] * n, + "logic_rel_id": ["btc_above_960_implies_above_900_20260121"] * n, + "logic_best_gap": np.linspace(0.05, 0.01, n), + "logic_best_gap_z": np.linspace(-0.5, -2.0, n), + } + ) + + result = strategy.populate_entry_trend( + df.copy(), + metadata={"pair": "WillThePriceOfBitcoinBeAbove900YES20260121/USDC"}, + ) + + assert "enter_long" in result.columns + assert "enter_tag" in result.columns + assert (result["enter_long"].fillna(0) >= 0).all() + assert result["enter_long"].fillna(0).sum() > 0 + + def test_strong_entry_tag_is_used_for_very_low_rank(self): + strategy = PolymarketLogicalArbStrategy({}) + strategy.DEBUG_PRINTS = False + + df = pd.DataFrame( + { + "date": pd.date_range("2026-01-21", periods=10, freq="4H", tz="UTC"), + "close": np.full(10, 0.40), + "logic_role": ["superset"] * 10, + "enough_history": [1] * 10, + "logic_rank_pct": [0.50, 0.40, 0.25, 0.09, 0.08, 0.30, 0.11, 0.10, 0.50, 0.70], + "bars_to_end": [9, 8, 7, 6, 5, 4, 3, 2, 1, 0], + "contract_tradeable": [1] * 10, + "logic_rel_id": ["btc_above_960_implies_above_900_20260121"] * 10, + "logic_best_gap": np.full(10, 0.03), + "logic_best_gap_z": np.full(10, -1.5), + } + ) + + result = strategy.populate_entry_trend( + df.copy(), + metadata={"pair": "WillThePriceOfBitcoinBeAbove900YES20260121/USDC"}, + ) + + strong_tags = result["enter_tag"].dropna().astype(str) + assert any(tag.startswith("logic_v2_strong:") for tag in strong_tags) + + def test_populate_exit_trend_adds_exit_long(self): + strategy = PolymarketLogicalArbStrategy({}) + + df = pd.DataFrame( + { + "date": pd.date_range("2026-01-21", periods=8, freq="4H", tz="UTC"), + "logic_rank_pct": [0.10, 0.20, 0.30, 0.50, 0.86, 0.40, 0.20, 0.10], + "bars_to_end": [7, 6, 5, 4, 3, 2, 1, 0], + } + ) + + result = strategy.populate_exit_trend( + df.copy(), + metadata={"pair": "WillThePriceOfBitcoinBeAbove900YES20260121/USDC"}, + ) + + assert "exit_long" in result.columns + assert result["exit_long"].fillna(0).sum() >= 2 + + def test_exit_triggered_by_high_rank_or_near_end(self): + strategy = PolymarketLogicalArbStrategy({}) + + df = pd.DataFrame( + { + "date": pd.date_range("2026-01-21", periods=6, freq="4H", tz="UTC"), + "logic_rank_pct": [0.10, 0.20, 0.86, 0.30, 0.20, 0.10], + "bars_to_end": [5, 4, 3, 2, 1, 0], + } + ) + + result = strategy.populate_exit_trend( + df.copy(), + metadata={"pair": "WillThePriceOfBitcoinBeAbove900YES20260121/USDC"}, + ) + + assert result.loc[2, "exit_long"] == 1 + assert result.loc[4, "exit_long"] == 1 + assert result.loc[5, "exit_long"] == 1 + + +class TestPolymarketLogicalArbStrategyEntryConfirmation: + def setup_method(self): + PolymarketLogicalArbStrategy._PAIR_CACHE_BUILT = False + PolymarketLogicalArbStrategy.RELATIONSHIPS = [] + + def test_confirm_trade_entry_accepts_valid_pair_and_price(self, tmp_path): + csv_path = _make_pair_mapping_csv(tmp_path) + + strategy = PolymarketLogicalArbStrategy({}) + strategy.PAIR_SOURCE_PATH = str(csv_path) + strategy._PAIR_CACHE_BUILT = False + strategy.RELATIONSHIPS = [] + + ok = strategy.confirm_trade_entry( + pair="WillThePriceOfBitcoinBeAbove900YES20260121/USDC", + order_type="limit", + amount=1000.0, + rate=0.45, + time_in_force="GTC", + current_time=datetime.now(timezone.utc), + entry_tag="logic_v2:test", + side="long", + ) + + assert ok is True + + def test_confirm_trade_entry_rejects_unknown_pair(self, tmp_path): + csv_path = _make_pair_mapping_csv(tmp_path) + + strategy = PolymarketLogicalArbStrategy({}) + strategy.PAIR_SOURCE_PATH = str(csv_path) + strategy._PAIR_CACHE_BUILT = False + strategy.RELATIONSHIPS = [] + + ok = strategy.confirm_trade_entry( + pair="BTC/USDT", + order_type="limit", + amount=1000.0, + rate=0.45, + time_in_force="GTC", + current_time=datetime.now(timezone.utc), + entry_tag="logic_v2:test", + side="long", + ) + + assert ok is False + + def test_confirm_trade_entry_rejects_bad_price(self, tmp_path): + csv_path = _make_pair_mapping_csv(tmp_path) + + strategy = PolymarketLogicalArbStrategy({}) + strategy.PAIR_SOURCE_PATH = str(csv_path) + strategy._PAIR_CACHE_BUILT = False + strategy.RELATIONSHIPS = [] + + low = strategy.confirm_trade_entry( + pair="WillThePriceOfBitcoinBeAbove900YES20260121/USDC", + order_type="limit", + amount=1000.0, + rate=0.001, + time_in_force="GTC", + current_time=datetime.now(timezone.utc), + entry_tag="logic_v2:test", + side="long", + ) + + high = strategy.confirm_trade_entry( + pair="WillThePriceOfBitcoinBeAbove900YES20260121/USDC", + order_type="limit", + amount=1000.0, + rate=1.10, + time_in_force="GTC", + current_time=datetime.now(timezone.utc), + entry_tag="logic_v2:test", + side="long", + ) + + assert low is False + assert high is False \ No newline at end of file