feat(strategy_pythagoras): invariance bonus callback (BTC<->ETH corr_signal)

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).
This commit is contained in:
Adriano Dal Pastro
2026-05-19 13:58:39 +00:00
parent af68bc44b4
commit b8bf0c186c
2 changed files with 93 additions and 0 deletions
@@ -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)
@@ -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