diff --git a/src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py b/src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py new file mode 100644 index 0000000..b088ca4 --- /dev/null +++ b/src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py @@ -0,0 +1,50 @@ +"""Bonus di asset-invariance per la fitness del GA. + +corr_signal = frazione di entries su asset A che hanno corrispondenza su asset B +entro +/-tolerance_bars (default 36 = 3h su 5m TF, vedi paper Pythagoras p. 43). +""" +from __future__ import annotations + +import os + +import pandas as pd + +GA_INVARIANCE_ALPHA = float(os.getenv("GA_INVARIANCE_ALPHA", "0.3")) +GA_INVARIANCE_TOLERANCE_BARS = int(os.getenv("GA_INVARIANCE_TOLERANCE_BARS", "36")) + + +def corr_signal( + entries_a: pd.Series, + entries_b: pd.Series, + tolerance_bars: int = GA_INVARIANCE_TOLERANCE_BARS, +) -> float: + """Frazione di entries A con match in B entro +/-tolerance_bars. + + Args: + entries_a, entries_b: Series binarie {0,1} sullo stesso index temporale (interi). + tolerance_bars: finestra di tolleranza in barre. + + Returns: + In [0, 1]. 0 se entries_a non ha alcuna entry o nessun match. + """ + a_idx = entries_a[entries_a > 0].index.tolist() + b_idx = entries_b[entries_b > 0].index.tolist() + if not a_idx or not b_idx: + return 0.0 + b_set = set(b_idx) + matched = 0 + for ti in a_idx: + for delta in range(-tolerance_bars, tolerance_bars + 1): + if (ti + delta) in b_set: + matched += 1 + break + return matched / len(a_idx) + + +def apply_invariance_bonus( + base_fitness: float, + invariance_score: float, + alpha: float = GA_INVARIANCE_ALPHA, +) -> float: + """``fitness * (1 + alpha * invariance_score)``.""" + return base_fitness * (1.0 + alpha * invariance_score) diff --git a/src/strategy_pythagoras/tests/test_fitness_invariance.py b/src/strategy_pythagoras/tests/test_fitness_invariance.py new file mode 100644 index 0000000..cee9bac --- /dev/null +++ b/src/strategy_pythagoras/tests/test_fitness_invariance.py @@ -0,0 +1,43 @@ +"""Bonus invariance: pattern che firano simultaneamente su 2 asset entro tolleranza.""" +from __future__ import annotations + +import pandas as pd +import pytest + +from strategy_pythagoras.fitness_invariance import ( + apply_invariance_bonus, + corr_signal, +) + + +def test_corr_signal_perfect_alignment() -> None: + entries_btc = pd.Series([0, 1, 0, 1, 0], index=[0, 1, 2, 3, 4]) + entries_eth = pd.Series([0, 1, 0, 1, 0], index=[0, 1, 2, 3, 4]) + assert corr_signal(entries_btc, entries_eth, tolerance_bars=0) == pytest.approx(1.0) + + +def test_corr_signal_no_overlap() -> None: + entries_btc = pd.Series([0, 1, 0, 0, 0], index=[0, 1, 2, 3, 4]) + entries_eth = pd.Series([0, 0, 0, 0, 1], index=[0, 1, 2, 3, 4]) + assert corr_signal(entries_btc, entries_eth, tolerance_bars=0) == pytest.approx(0.0) + + +def test_corr_signal_within_tolerance() -> None: + # entry su BTC a t=1, su ETH a t=3, tolerance=2 -> match + entries_btc = pd.Series([0, 1, 0, 0, 0], index=[0, 1, 2, 3, 4]) + entries_eth = pd.Series([0, 0, 0, 1, 0], index=[0, 1, 2, 3, 4]) + assert corr_signal(entries_btc, entries_eth, tolerance_bars=2) == pytest.approx(1.0) + + +def test_apply_invariance_bonus_increases_fitness() -> None: + assert apply_invariance_bonus(1.0, 0.5, 0.3) == pytest.approx(1.0 * (1.0 + 0.3 * 0.5)) + + +def test_apply_invariance_bonus_alpha_zero() -> None: + assert apply_invariance_bonus(1.0, 0.7, 0.0) == pytest.approx(1.0) + + +def test_corr_signal_zero_entries() -> None: + entries_btc = pd.Series([0, 0, 0], index=[0, 1, 2]) + entries_eth = pd.Series([0, 0, 0], index=[0, 1, 2]) + assert corr_signal(entries_btc, entries_eth, tolerance_bars=0) == 0.0