feat(strategy_pythagoras): implement candle_pattern, pythagorean_ratio, fractal_mirror + register in compiler

src/multi_swarm_core/multi_swarm_core/protocol/compiler.py

@@ -25,6 +25,12 @@ from typing import Any
 import numpy as np
 import pandas as pd  # type: ignore[import-untyped]
 
+from strategy_pythagoras.indicators import (
+    candle_pattern as _public_candle_pattern,
+    fractal_mirror as _public_fractal_mirror,
+    pythagorean_ratio as _public_pythagorean_ratio,
+)
+
 from ..backtest.orders import Side
 from .parser import (
     FeatureNode,
@@ -126,6 +132,22 @@ def _ind_macd(
     return macd_line - signal_line
 
 
+def _ind_candle_pattern(df: pd.DataFrame, *params: float) -> pd.Series:
+    # Adapter: il dispatch in _eval_node fa ``fn(df, *node.params)``, ma la
+    # public API in strategy_pythagoras.indicators accetta ``params: list[float]``
+    # come singolo argomento. Re-pack qui per mantenere indicators.py testabile
+    # in isolamento.
+    return _public_candle_pattern(df, list(params))
+
+
+def _ind_pythagorean_ratio(df: pd.DataFrame, lookback: float) -> pd.Series:
+    return _public_pythagorean_ratio(df, [lookback])
+
+
+def _ind_fractal_mirror(df: pd.DataFrame, k: float, axis_int: float) -> pd.Series:
+    return _public_fractal_mirror(df, [k, axis_int])
+
+
 # Annotated as ``dict[str, Any]`` deliberately: each indicator has its own
 # arity and parameter names, so a single ``Callable`` signature would be a
 # lie. Dispatch happens in :func:`_eval_node`, which validates the verb name
@@ -139,6 +161,9 @@ INDICATOR_FNS: dict[str, Any] = {
     "realized_vol": _ind_realized_vol,
     "macd": _ind_macd,
     "macd_pct": _ind_macd_pct,
+    "candle_pattern": _ind_candle_pattern,
+    "pythagorean_ratio": _ind_pythagorean_ratio,
+    "fractal_mirror": _ind_fractal_mirror,
 }
 
 _TIME_FEATURE_FNS: dict[str, Callable[[pd.DatetimeIndex], pd.Series]] = {

src/strategy_pythagoras/strategy_pythagoras/indicators.py

@@ -0,0 +1,70 @@
+"""Indicatori candle Pythagoras.
+
+Vincoli grammar: ``IndicatorNode.params`` e' sempre ``list[float]``. Quindi:
+- candle_pattern: params = [length, sym0, sym1, ..., sym_{length-1}]
+  length in [3,12]; sym in {0=U, 1=D, 2=doji}
+- pythagorean_ratio: params = [lookback]  lookback in [12,200]
+- fractal_mirror: params = [k, axis_int]  k in [3,12]; axis_int=0(h) 1(v)
+"""
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+_DOJI_THRESHOLD = 0.001
+
+
+def _symbol_series(df: pd.DataFrame) -> pd.Series:
+    """Mappa ogni candela in {0=U, 1=D, 2=doji}."""
+    close = df["close"]
+    open_ = df["open"]
+    rel = (close - open_).abs() / open_.replace(0, np.nan)
+    sym = np.where(close > open_, 0, np.where(close < open_, 1, 2))
+    sym = np.where(rel.values < _DOJI_THRESHOLD, 2, sym)
+    return pd.Series(sym, index=df.index, dtype="int64")
+
+
+def candle_pattern(df: pd.DataFrame, params: list[float]) -> pd.Series:
+    """1.0 se le ultime ``length`` candele matchano la sequenza, 0.0 altrimenti."""
+    length = int(params[0])
+    target = [int(s) for s in params[1:1 + length]]
+    syms = _symbol_series(df)
+    out = pd.Series(0.0, index=df.index, dtype="float64")
+    if len(syms) < length:
+        return out
+    arr = syms.values
+    target_arr = np.array(target, dtype=arr.dtype)
+    for i in range(length - 1, len(arr)):
+        if np.array_equal(arr[i - length + 1: i + 1], target_arr):
+            out.iat[i] = 1.0
+    return out
+
+
+def pythagorean_ratio(df: pd.DataFrame, params: list[float]) -> pd.Series:
+    """``max(close[-lookback:]) / min(close[-lookback:])`` rolling."""
+    lookback = int(params[0])
+    close = df["close"]
+    hi = close.rolling(lookback, min_periods=1).max()
+    lo = close.rolling(lookback, min_periods=1).min().replace(0, np.nan)
+    return (hi / lo).fillna(1.0)
+
+
+def fractal_mirror(df: pd.DataFrame, params: list[float]) -> pd.Series:
+    """Correlation tra close[-k:] e suo mirror su asse axis."""
+    k = int(params[0])
+    axis_int = int(params[1])
+    close = df["close"].values
+    out = np.full(len(close), 0.0)
+    for i in range(k - 1, len(close)):
+        window = close[i - k + 1: i + 1]
+        if axis_int == 0:  # h: mirror temporale
+            mirror = window[::-1]
+        else:              # v: mirror prezzo
+            mirror = window.max() - (window - window.min())
+        std_w = window.std()
+        std_m = mirror.std()
+        if std_w < 1e-12 or std_m < 1e-12:
+            out[i] = 0.0
+        else:
+            out[i] = float(np.corrcoef(window, mirror)[0, 1])
+    return pd.Series(out, index=df.index, dtype="float64")

src/strategy_pythagoras/tests/test_indicators.py

@@ -0,0 +1,82 @@
+"""Unit-test dei 3 indicatori Pythagoras."""
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from strategy_pythagoras.indicators import (
+    candle_pattern,
+    fractal_mirror,
+    pythagorean_ratio,
+)
+
+
+@pytest.fixture
+def ohlcv_30() -> pd.DataFrame:
+    """30 candele sintetiche: pattern alternato U,D,U,D,..."""
+    n = 30
+    close = np.array([100.0 + i for i in range(n)])  # monotone
+    open_ = close - np.tile([1.0, -1.0], n // 2)     # U,D,U,D,...
+    return pd.DataFrame({"open": open_, "high": close + 0.5, "low": open_ - 0.5, "close": close, "volume": [1.0] * n})
+
+
+# -- candle_pattern -----------------------------------------------------------
+
+def test_candle_pattern_matches_recent(ohlcv_30: pd.DataFrame) -> None:
+    # Verifica che il symbol mapping U=0,D=1,doji=2 lavori correttamente.
+    # Con la fixture: indices pari sono U (close>open), dispari sono D (close<open).
+    # ultime 3 candele (idx 27,28,29) -> D,U,D
+    out = candle_pattern(ohlcv_30, [3, 1, 0, 1])  # D, U, D
+    assert out.iloc[-1] == 1.0
+    out2 = candle_pattern(ohlcv_30, [3, 0, 0, 0])  # U, U, U: no match
+    assert out2.iloc[-1] == 0.0
+
+
+def test_candle_pattern_zero_for_short_history(ohlcv_30: pd.DataFrame) -> None:
+    out = candle_pattern(ohlcv_30, [3, 0, 0, 0])
+    assert out.iloc[0] == 0.0
+    assert out.iloc[1] == 0.0
+
+
+def test_candle_pattern_doji_symbol(ohlcv_30: pd.DataFrame) -> None:
+    df = ohlcv_30.copy()
+    # forza la candela [-1] doji: |close-open|/open < 0.001
+    df.loc[df.index[-1], "open"] = df["close"].iloc[-1] * (1 - 1e-5)
+    # ultime 3 dopo modifica: D (idx 27), U (idx 28), doji (idx 29)
+    out = candle_pattern(df, [3, 1, 0, 2])
+    assert out.iloc[-1] == 1.0
+
+
+# -- pythagorean_ratio --------------------------------------------------------
+
+def test_pythagorean_ratio_basic(ohlcv_30: pd.DataFrame) -> None:
+    out = pythagorean_ratio(ohlcv_30, [12])
+    # ultimi 12 close: 118..129 -> max/min = 129/118
+    expected = 129.0 / 118.0
+    assert abs(out.iloc[-1] - expected) < 1e-9
+
+
+def test_pythagorean_ratio_no_lookahead(ohlcv_30: pd.DataFrame) -> None:
+    out = pythagorean_ratio(ohlcv_30, [12])
+    out0 = pythagorean_ratio(ohlcv_30.iloc[:13], [12])
+    assert abs(out.iloc[12] - out0.iloc[-1]) < 1e-9
+
+
+# -- fractal_mirror -----------------------------------------------------------
+
+def test_fractal_mirror_h_pattern_inverted(ohlcv_30: pd.DataFrame) -> None:
+    # mirror temporale: correlazione tra close[-k:] e close[-k:][::-1]
+    # Per close monotono, correlation(seq, seq_reversed) = -1
+    out = fractal_mirror(ohlcv_30, [6, 0])  # axis=0 (h)
+    assert out.iloc[-1] < -0.99
+
+
+def test_fractal_mirror_v_axis(ohlcv_30: pd.DataFrame) -> None:
+    out = fractal_mirror(ohlcv_30, [6, 1])
+    assert out.iloc[-1] < -0.99
+
+
+def test_fractal_mirror_clamps_initial(ohlcv_30: pd.DataFrame) -> None:
+    out = fractal_mirror(ohlcv_30, [6, 0])
+    assert len(out) == len(ohlcv_30)