From b8bf0c186c20692c051c8b0dd28c735e0bae7553 Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Tue, 19 May 2026 13:58:39 +0000 Subject: [PATCH] feat(strategy_pythagoras): invariance bonus callback (BTC<->ETH corr_signal) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit corr_signal: frazione di entries A con match in B entro +/-tolerance_bars (default 36 barre = 3h su 5m TF, env GA_INVARIANCE_TOLERANCE_BARS). apply_invariance_bonus: fitness * (1 + alpha * invariance_score), alpha=0.3 (env GA_INVARIANCE_ALPHA). Spec plan Pythagoras ยง4. GA integration approach (Step 5 findings): - multi_swarm_core.ga.fitness.compute_fitness e compute_combined_fitness NON espongono callback/hook per post-processing della fitness. - orchestrator.run.run_phase1(...) -> str ritorna solo il run_id; le evaluations (incl. fitness scalare) vengono persistite via repo.save_evaluation dentro al loop GA (run.py:264-277). - I winner sono recuperati dopo il run con repo.list_evaluations(run_id) e ri-ordinati per fitness (vedi pattern run.py:302-310 per WFA re-eval). - Pattern (b) confermato: il runner Task 6.1 chiamera' run_phase1 due volte (BTC, ETH), poi per ogni coppia di evaluations matchera' i genome_id, calcolera' corr_signal sulle entries dei rispettivi backtests e applichera' apply_invariance_bonus per ri-rankare esternamente i winner. Nessuna modifica a multi_swarm_core necessaria in questo task. Tests: 6/6 PASS (perfect alignment, no overlap, within tolerance, bonus formula, alpha=0, zero entries). --- .../strategy_pythagoras/fitness_invariance.py | 50 +++++++++++++++++++ .../tests/test_fitness_invariance.py | 43 ++++++++++++++++ 2 files changed, 93 insertions(+) create mode 100644 src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py create mode 100644 src/strategy_pythagoras/tests/test_fitness_invariance.py 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