From 369a77b5cffce6e054dadcbeadaf1bc4d132835e Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Tue, 19 May 2026 13:38:40 +0000 Subject: [PATCH] feat(strategy_pythagoras): implement candle_pattern, pythagorean_ratio, fractal_mirror + register in compiler --- .../multi_swarm_core/protocol/compiler.py | 25 ++++++ .../strategy_pythagoras/indicators.py | 70 ++++++++++++++++ .../tests/test_indicators.py | 82 +++++++++++++++++++ 3 files changed, 177 insertions(+) create mode 100644 src/strategy_pythagoras/strategy_pythagoras/indicators.py create mode 100644 src/strategy_pythagoras/tests/test_indicators.py diff --git a/src/multi_swarm_core/multi_swarm_core/protocol/compiler.py b/src/multi_swarm_core/multi_swarm_core/protocol/compiler.py index 19e09bc..c108240 100644 --- a/src/multi_swarm_core/multi_swarm_core/protocol/compiler.py +++ b/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]] = { diff --git a/src/strategy_pythagoras/strategy_pythagoras/indicators.py b/src/strategy_pythagoras/strategy_pythagoras/indicators.py new file mode 100644 index 0000000..e2e2a87 --- /dev/null +++ b/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") diff --git a/src/strategy_pythagoras/tests/test_indicators.py b/src/strategy_pythagoras/tests/test_indicators.py new file mode 100644 index 0000000..ac0b4ef --- /dev/null +++ b/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 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)