# strategy_pythagoras Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Creare un nuovo workspace member `strategy_pythagoras` parallelo a `strategy_crypto`, con 3 indicatori candle (`candle_pattern`, `pythagorean_ratio`, `fractal_mirror`), 7 stili cognitivi Pythagoras-aligned, fitness con bonus asset-invariance BTC↔ETH, paper-trading runner + dashboard NiceGUI, e 1 run GA short (`pythagoras-smoke-001`) + report cross-book `docs/analysis_first_run.md`. **Architecture:** Monorepo uv workspace; il genoma `HypothesisAgentGenome` del core non cambia, si aggiungono solo 3 indicatori al `multi_swarm_core.protocol` e si scrive un nuovo `prompts.json`. Backend + frontend portati (non importati) da `strategy_crypto`, DB isolato. Fitness callback registrata dal runner senza modificare il GA core. **Tech Stack:** Python 3.13, uv workspaces, pandas, numpy, pytest, NiceGUI 3.11, SQLite (via stdlib sqlite3), multi-swarm-core (GA/LLM/protocol). **Spec di riferimento:** `docs/superpowers/specs/2026-05-19-strategy-pythagoras-design.md` **Encoding param int (vincolo grammar, non era nella spec):** la grammar accetta solo `params: list[float]`. Quindi: - `candle_pattern(length, sym0, sym1, ...)` — `length ∈ [3,12]`, simboli `0=U`, `1=D`, `2=0(doji)` - `fractal_mirror(k, axis_int)` — `axis_int=0` → "h" (mirror temporale), `axis_int=1` → "v" (mirror prezzo) - `pythagorean_ratio(lookback)` — int 12-200 Il prompts.json documenterà questa encoding ai LLM agenti. --- ## File structure | Path | Responsibility | Action | |---|---|---| | `pyproject.toml` (root) | Workspace registration | Modify | | `src/strategy_pythagoras/pyproject.toml` | Package metadata, hatchling include | Create | | `src/strategy_pythagoras/README.md` | Package docs | Create | | `src/strategy_pythagoras/strategy_pythagoras/__init__.py` | Marker package | Create | | `src/strategy_pythagoras/strategy_pythagoras/indicators.py` | 3 indicatori candle (helper functions) | Create | | `src/strategy_pythagoras/strategy_pythagoras/prompts.json` | 7 stili cognitivi + schema v3.2 | Create | | `src/strategy_pythagoras/strategy_pythagoras/backend/__init__.py` | Re-export | Create | | `src/strategy_pythagoras/strategy_pythagoras/backend/schema.py` | DDL paper_trading_* + init_schema() | Create (port) | | `src/strategy_pythagoras/strategy_pythagoras/backend/portfolio.py` | Portfolio domain | Create (port) | | `src/strategy_pythagoras/strategy_pythagoras/backend/executor.py` | PaperExecutor | Create (port) | | `src/strategy_pythagoras/strategy_pythagoras/backend/persistence.py` | PaperRepository | Create (port) | | `src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py` | invariance bonus callback | Create | | `src/strategy_pythagoras/strategy_pythagoras/frontend/__init__.py` | Marker | Create | | `src/strategy_pythagoras/strategy_pythagoras/frontend/data.py` | Dual-DB reader + invariance metrics | Create (port + extend) | | `src/strategy_pythagoras/strategy_pythagoras/frontend/nicegui_app.py` | NiceGUI app, 4 tabs | Create (port + extend) | | `src/strategy_pythagoras/strategy_pythagoras/strategies/.gitkeep` | Empty dir tracker | Create | | `src/strategy_pythagoras/tests/__init__.py` | Test marker | Create | | `src/strategy_pythagoras/tests/test_indicators.py` | Unit-test 3 indicatori | Create | | `src/strategy_pythagoras/tests/test_prompts.py` | Schema/loading prompts.json | Create | | `src/multi_swarm_core/multi_swarm_core/protocol/grammar.py` | Extend `KNOWN_INDICATORS` | Modify | | `src/multi_swarm_core/multi_swarm_core/protocol/validator.py` | Extend `INDICATOR_ARITY` + params validation | Modify | | `src/multi_swarm_core/multi_swarm_core/protocol/compiler.py` | Register helper functions in `INDICATOR_FNS` | Modify | | `scripts/run_pythagoras_smoke.py` | GA smoke-test runner | Create | | `docs/analysis_first_run.md` | Cross-book report con i winner | Create (post-run) | --- ## Phase 0 — Workspace scaffolding ### Task 0.1: Crea skeleton package + registrazione workspace **Files:** - Create: `src/strategy_pythagoras/pyproject.toml` - Create: `src/strategy_pythagoras/README.md` - Create: `src/strategy_pythagoras/strategy_pythagoras/__init__.py` - Create: `src/strategy_pythagoras/strategy_pythagoras/backend/__init__.py` - Create: `src/strategy_pythagoras/strategy_pythagoras/frontend/__init__.py` - Create: `src/strategy_pythagoras/strategy_pythagoras/strategies/.gitkeep` - Create: `src/strategy_pythagoras/tests/__init__.py` - Modify: `pyproject.toml` (root) - [ ] **Step 1: Crea `src/strategy_pythagoras/pyproject.toml`** ```toml [project] name = "strategy-pythagoras" version = "0.1.0" description = "Strategy Pythagoras: candle-pattern GA su framework Pythagoras-Malanga, paper-trading runner + NiceGUI dashboard" authors = [{ name = "Adriano Dal Pastro", email = "adrianodalpastro@tielogic.com" }] requires-python = ">=3.13" dependencies = [ "multi-swarm-core", "nicegui>=3.11.1", "plotly>=5.24", "pandas>=2.2", "pyarrow>=18.0", ] [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel.force-include] "strategy_pythagoras/strategies" = "strategy_pythagoras/strategies" "strategy_pythagoras/prompts.json" = "strategy_pythagoras/prompts.json" ``` - [ ] **Step 2: Crea `src/strategy_pythagoras/README.md`** ```markdown # strategy_pythagoras Strategia di test crypto (BTC/ETH) basata sul framework Pythagoras-Malanga (frattali H-C + trasformata di Fourier + geometria Evideon). Workspace member del monorepo `multi_swarm_coevolutive`. ## Scope Esegue un **GA coevolutivo** con 7 stili cognitivi Pythagoras-aligned e 3 indicatori candle (`candle_pattern`, `pythagorean_ratio`, `fractal_mirror`). La fitness include un bonus di asset-invariance BTC↔ETH (±36 barre = ±3h su 5m TF, come da paper). ## Layout ``` strategy_pythagoras/ ├── backend/ paper trading (PaperExecutor, Portfolio, PaperRepository) ├── frontend/ NiceGUI dashboard (dual-DB + invariance metrics) ├── indicators.py 3 indicatori candle nuovi ├── fitness_invariance.py callback BTC↔ETH correlation ├── prompts.json 7 stili cognitivi (schema v3.2 Pythagoras-themed) └── strategies/ JSON winners freezati (post-GA) ``` ## Run GA smoke test ```bash uv run python scripts/run_pythagoras_smoke.py ``` ## Dashboard ```bash uv run python -m strategy_pythagoras.frontend.nicegui_app ``` In produzione: `https://swarm.tielogic.xyz/strategy_pythagoras_gui/`. ## DB schema Schema isolato in `state/strategy_pythagoras.db` (GA) e `state/strategy_pythagoras_paper.db` (paper trading). Env vars: - `STRATEGY_PYTHAGORAS_DB_PATH` - `STRATEGY_PYTHAGORAS_PAPER_DB_PATH` ## Reference - Spec: `docs/superpowers/specs/2026-05-19-strategy-pythagoras-design.md` - PDF sorgente: `Pythagoras/Pythagoras_Trading_Prediction_.pdf`, `Pythagoras/Libro_frattali.pdf` - Riassunti: `Pythagoras/Pythagoras_Trading_Prediction.summary.md`, `Pythagoras/Libro_frattali.summary.md` ``` - [ ] **Step 3: Crea i 4 `__init__.py` vuoti e il `.gitkeep`** ```bash touch /opt/docker/multi_swarm_coevolutive/src/strategy_pythagoras/strategy_pythagoras/__init__.py touch /opt/docker/multi_swarm_coevolutive/src/strategy_pythagoras/strategy_pythagoras/backend/__init__.py touch /opt/docker/multi_swarm_coevolutive/src/strategy_pythagoras/strategy_pythagoras/frontend/__init__.py touch /opt/docker/multi_swarm_coevolutive/src/strategy_pythagoras/tests/__init__.py mkdir -p /opt/docker/multi_swarm_coevolutive/src/strategy_pythagoras/strategy_pythagoras/strategies touch /opt/docker/multi_swarm_coevolutive/src/strategy_pythagoras/strategy_pythagoras/strategies/.gitkeep ``` - [ ] **Step 4: Modifica root `pyproject.toml`** Read `/opt/docker/multi_swarm_coevolutive/pyproject.toml` then update: - In `[project].dependencies`: aggiungi `"strategy-pythagoras",` dopo `"strategy-crypto",` - In `[tool.uv.workspace].members`: aggiungi `"src/strategy_pythagoras"` dopo `"src/strategy_crypto"` - In `[tool.uv.sources]`: aggiungi `strategy-pythagoras = { workspace = true }` sotto `strategy-crypto` - In `[tool.pytest.ini_options].testpaths`: aggiungi `"src/strategy_pythagoras/tests"` - [ ] **Step 5: Esegui `uv sync` per validare** ```bash cd /opt/docker/multi_swarm_coevolutive && uv sync ``` Expected: nessun errore, mostra `Installed: strategy-pythagoras==0.1.0 (workspace)`. - [ ] **Step 6: Verifica import del package** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c "import strategy_pythagoras; print('OK')" ``` Expected: stampa `OK`. - [ ] **Step 7: Commit** ```bash git add src/strategy_pythagoras pyproject.toml git commit -m "feat(strategy_pythagoras): scaffold workspace member + register in uv" ``` --- ## Phase 1 — Core extensions (indicatori) ### Task 1.1: Estendi grammar con i 3 nuovi indicatori **Files:** - Modify: `src/multi_swarm_core/multi_swarm_core/protocol/grammar.py` - [ ] **Step 1: Read current grammar** Read `src/multi_swarm_core/multi_swarm_core/protocol/grammar.py` to locate `KNOWN_INDICATORS`. - [ ] **Step 2: Estendi `KNOWN_INDICATORS`** Sostituisci la definizione di `KNOWN_INDICATORS` aggiungendo i 3 nuovi nomi: ```python KNOWN_INDICATORS: frozenset[str] = frozenset( {"sma", "sma_pct", "rsi", "atr", "atr_pct", "macd", "macd_pct", "realized_vol", "candle_pattern", "pythagorean_ratio", "fractal_mirror"} ) ``` - [ ] **Step 3: Verifica import** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c "from multi_swarm_core.protocol.grammar import KNOWN_INDICATORS; assert 'candle_pattern' in KNOWN_INDICATORS; assert 'pythagorean_ratio' in KNOWN_INDICATORS; assert 'fractal_mirror' in KNOWN_INDICATORS; print('OK')" ``` Expected: stampa `OK`. - [ ] **Step 4: Commit** ```bash git add src/multi_swarm_core/multi_swarm_core/protocol/grammar.py git commit -m "feat(protocol): register 3 Pythagoras indicators in KNOWN_INDICATORS" ``` --- ### Task 1.2: Estendi validator con `INDICATOR_ARITY` per i 3 indicatori **Files:** - Modify: `src/multi_swarm_core/multi_swarm_core/protocol/validator.py` - Test: `src/strategy_pythagoras/tests/test_validator_indicators.py` - [ ] **Step 1: Crea test che asserisce validazione corretta** Crea `src/strategy_pythagoras/tests/test_validator_indicators.py`: ```python """Validator accetta i 3 nuovi indicatori candle con arity corrette.""" from __future__ import annotations import pytest from multi_swarm_core.protocol.parser import parse_strategy from multi_swarm_core.protocol.validator import ValidationError, validate_strategy def _strategy_with_indicator(name: str, params: list[float]) -> dict: return { "rules": [ { "condition": { "op": "gt", "args": [ {"kind": "indicator", "name": name, "params": params}, {"kind": "literal", "value": 0.5}, ], }, "action": "entry-long", } ] } def test_candle_pattern_valid_min_arity() -> None: # length=3, syms=[U,D,U] s = parse_strategy(_strategy_with_indicator("candle_pattern", [3, 0, 1, 0])) validate_strategy(s) # no raise def test_candle_pattern_valid_max_arity() -> None: # length=12, 12 syms s = parse_strategy(_strategy_with_indicator("candle_pattern", [12] + [0] * 12)) validate_strategy(s) def test_candle_pattern_rejects_too_few_params() -> None: s = parse_strategy(_strategy_with_indicator("candle_pattern", [3, 0, 1])) # mancano simboli with pytest.raises(ValidationError): validate_strategy(s) def test_pythagorean_ratio_valid() -> None: s = parse_strategy(_strategy_with_indicator("pythagorean_ratio", [89])) validate_strategy(s) def test_pythagorean_ratio_rejects_zero_params() -> None: s = parse_strategy(_strategy_with_indicator("pythagorean_ratio", [])) with pytest.raises(ValidationError): validate_strategy(s) def test_fractal_mirror_valid_h() -> None: s = parse_strategy(_strategy_with_indicator("fractal_mirror", [12, 0])) validate_strategy(s) def test_fractal_mirror_valid_v() -> None: s = parse_strategy(_strategy_with_indicator("fractal_mirror", [12, 1])) validate_strategy(s) def test_fractal_mirror_rejects_one_param() -> None: s = parse_strategy(_strategy_with_indicator("fractal_mirror", [12])) with pytest.raises(ValidationError): validate_strategy(s) ``` - [ ] **Step 2: Run test — atteso FAIL (validator non conosce ancora i nomi)** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_validator_indicators.py -v ``` Expected: tutti i test FAILED con errore tipo "unknown indicator" o accettano arity errate. - [ ] **Step 3: Estendi `INDICATOR_ARITY` nel validator** Read `src/multi_swarm_core/multi_swarm_core/protocol/validator.py` e localizza il dict `INDICATOR_ARITY`. Aggiungi le tre righe: ```python INDICATOR_ARITY: dict[str, tuple[int, int]] = { "sma": (1, 1), "sma_pct": (1, 1), "rsi": (1, 1), "atr": (1, 1), "atr_pct": (1, 1), "realized_vol": (1, 1), "macd": (0, 3), "macd_pct": (0, 3), # Pythagoras indicators (params encoded as floats; see strategy_pythagoras docs) "candle_pattern": (4, 13), # [length, sym0, ..., sym_{length-1}] -> 1+length, length in [3,12] "pythagorean_ratio": (1, 1), # [lookback] "fractal_mirror": (2, 2), # [k, axis_int] axis_int in {0,1} } ``` - [ ] **Step 4: Aggiungi validazione semantica dei params per i 3 indicatori** Cerca in `validator.py` la funzione `_validate_indicator` (o equivalente). Aggiungi check post-arity per i 3 nuovi indicatori. Implementazione tipo: ```python def _validate_indicator(node: IndicatorNode) -> None: name = node.name if name not in KNOWN_INDICATORS: raise ValidationError(f"unknown indicator: {name}") n = len(node.params) lo, hi = INDICATOR_ARITY[name] if not (lo <= n <= hi): raise ValidationError(f"'{name}' arity: expected [{lo},{hi}], got {n}") # Pythagoras-specific param semantics if name == "candle_pattern": length = int(node.params[0]) if not (3 <= length <= 12): raise ValidationError(f"candle_pattern length must be in [3,12], got {length}") if n != 1 + length: raise ValidationError( f"candle_pattern: expected 1+length={1 + length} params, got {n}" ) for i, sym in enumerate(node.params[1:], start=1): sym_int = int(sym) if sym_int not in (0, 1, 2): raise ValidationError( f"candle_pattern sym[{i - 1}] must be 0(U)/1(D)/2(doji), got {sym}" ) elif name == "pythagorean_ratio": lookback = int(node.params[0]) if not (12 <= lookback <= 200): raise ValidationError(f"pythagorean_ratio lookback in [12,200], got {lookback}") elif name == "fractal_mirror": k = int(node.params[0]) if not (3 <= k <= 12): raise ValidationError(f"fractal_mirror k must be in [3,12], got {k}") axis_int = int(node.params[1]) if axis_int not in (0, 1): raise ValidationError(f"fractal_mirror axis must be 0(h)/1(v), got {axis_int}") ``` (Adatta l'edit preservando il codice esistente per gli altri indicatori. Se la funzione attuale è una validazione generica solo per arity, aggiungi i check Pythagoras come blocco aggiuntivo dopo il check arity.) - [ ] **Step 5: Run test — atteso PASS** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_validator_indicators.py -v ``` Expected: tutti i 7 test PASS. - [ ] **Step 6: Commit** ```bash git add src/multi_swarm_core/multi_swarm_core/protocol/validator.py src/strategy_pythagoras/tests/test_validator_indicators.py git commit -m "feat(protocol): validate arity + semantics of 3 Pythagoras indicators" ``` --- ### Task 1.3: Implementa `candle_pattern` helper + register in compiler **Files:** - Create: `src/strategy_pythagoras/strategy_pythagoras/indicators.py` - Modify: `src/multi_swarm_core/multi_swarm_core/protocol/compiler.py` - Test: `src/strategy_pythagoras/tests/test_indicators.py` - [ ] **Step 1: Crea test failing per `candle_pattern`** Crea `src/strategy_pythagoras/tests/test_indicators.py`: ```python """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_UDU(ohlcv_30: pd.DataFrame) -> None: # con open,close alternati U,D,U,D,U,D,... cercando "U,D,U" sulle ultime 3 # bisogna determinare se le ultime 3 candele sono U,D,U # close[-3]=127, open[-3]=128-(-1)=128 -> D # close[-2]=128, open[-2]=128-1=127 -> U # close[-1]=129, open[-1]=129+1=130 -> D? Verifico: # i pari (0,2,4,..28) tile[1.0,-1.0] -> open=close-1=U; i dispari -> open=close+1=D # n=30, ultime 3 sono idx 27 (D), 28 (U), 29 (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]) # Prima della 3a barra non c'e' storia sufficiente -> 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) out = candle_pattern(df, [3, 1, 0, 2]) # ..., D, U, doji 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]) # la prima barra non puo' dipendere da barre future 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: # mirror verticale: correlation(close[-k:], (max-close[-k:])) # Per serie monotona = -1 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]) # primi k-1 valori: nan o 0; in ogni caso non sollevare assert len(out) == len(ohlcv_30) ``` - [ ] **Step 2: Run test — atteso FAIL (import non risolto)** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_indicators.py -v ``` Expected: ImportError `strategy_pythagoras.indicators`. - [ ] **Step 3: Crea `src/strategy_pythagoras/strategy_pythagoras/indicators.py`** ```python """Indicatori candle Pythagoras. Vincoli grammar: ``IndicatorNode.params`` e' sempre ``list[float]``. Quindi: - candle_pattern: params = [length, sym0, sym1, ..., sym_{length-1}] length ∈ [3,12]; sym ∈ {0=U, 1=D, 2=doji} - pythagorean_ratio: params = [lookback] lookback ∈ [12,200] - fractal_mirror: params = [k, axis_int] k ∈ [3,12]; axis_int=0(h) 1(v) """ from __future__ import annotations import numpy as np import pandas as pd _DOJI_THRESHOLD = 0.001 # |close-open|/open < threshold -> doji 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") ``` - [ ] **Step 4: Run test — atteso PASS** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_indicators.py -v ``` Expected: tutti i test PASS. - [ ] **Step 5: Wire indicators into `multi_swarm_core.protocol.compiler.INDICATOR_FNS`** Read `src/multi_swarm_core/multi_swarm_core/protocol/compiler.py` e localizza il dict `INDICATOR_FNS`. Aggiungi questo import in cima al file: ```python from strategy_pythagoras.indicators import ( candle_pattern as _ind_candle_pattern, fractal_mirror as _ind_fractal_mirror, pythagorean_ratio as _ind_pythagorean_ratio, ) ``` Estendi `INDICATOR_FNS` con: ```python "candle_pattern": _ind_candle_pattern, "pythagorean_ratio": _ind_pythagorean_ratio, "fractal_mirror": _ind_fractal_mirror, ``` **Adattamento signature:** le helper esistenti tipo `_ind_sma(df, length)` ricevono un float, non un list. Verifica come `IndicatorNode.params` viene splatato. Se il dispatch passa `*params` come positional args, modifica le 3 nuove funzioni in `indicators.py` per accettare `*params`: ```python def candle_pattern(df: pd.DataFrame, *params: float) -> pd.Series: ... ``` (Decidi in base alla lettura del dispatch in `compiler.py`. Per le helper interne `_ind_*` esistenti, sembrano accettare i parametri esplosi: replica lo stesso pattern. Aggiorna anche i test in `test_indicators.py` per chiamare con `*[3,0,1,0]` o variabili splate.) - [ ] **Step 6: Run full pytest per assicurarsi che nulla si rompa** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/multi_swarm_core/tests/ src/strategy_pythagoras/tests/ -v -x ``` Expected: tutti i test verdi. - [ ] **Step 7: Commit** ```bash git add src/strategy_pythagoras/strategy_pythagoras/indicators.py src/strategy_pythagoras/tests/test_indicators.py src/multi_swarm_core/multi_swarm_core/protocol/compiler.py git commit -m "feat(strategy_pythagoras): implement candle_pattern, pythagorean_ratio, fractal_mirror + register in compiler" ``` --- ## Phase 2 — Prompts (7 stili cognitivi) ### Task 2.1: Scrivi `prompts.json` con schema v3.2 **Files:** - Create: `src/strategy_pythagoras/strategy_pythagoras/prompts.json` - Test: `src/strategy_pythagoras/tests/test_prompts.py` - [ ] **Step 1: Crea test che asserisce schema valido** `src/strategy_pythagoras/tests/test_prompts.py`: ```python """prompts.json valido, 7 stili Pythagoras-aligned, schema v3.2.""" from __future__ import annotations import json from importlib.resources import files import pytest EXPECTED_STYLES = { "pythagorean", "fractal_geometer", "fourier_analyst", "evideonic_projector", "candle_grammarian", "recurrence_theorist", "skeptic_quant", } @pytest.fixture def prompts() -> dict: p = files("strategy_pythagoras") / "prompts.json" return json.loads(p.read_text(encoding="utf-8")) def test_schema_version(prompts: dict) -> None: assert prompts["_schema"] == "3.2" def test_top_level_fields(prompts: dict) -> None: for k in ( "agent_role", "pattern_guidance", "instruction", "domain_warnings", "anti_patterns", "output_priorities", "styles", ): assert k in prompts, f"missing field: {k}" def test_styles_set(prompts: dict) -> None: assert set(prompts["styles"].keys()) == EXPECTED_STYLES def test_styles_have_directive_and_focus_metrics(prompts: dict) -> None: for name, style in prompts["styles"].items(): assert "directive" in style, f"{name} missing directive" assert "focus_metrics" in style, f"{name} missing focus_metrics" assert len(style["focus_metrics"]) == 4, f"{name} focus_metrics len != 4" def test_directives_ascii_safe(prompts: dict) -> None: for name, style in prompts["styles"].items(): directive = style["directive"] for ch in directive: assert ord(ch) <= 0x7F, f"{name} directive contains non-ASCII char {ch!r}" def test_directives_length(prompts: dict) -> None: for name, style in prompts["styles"].items(): n = len(style["directive"]) assert 800 <= n <= 950, f"{name} directive length {n} outside [800,950]" def test_directives_first_phrase(prompts: dict) -> None: for name, style in prompts["styles"].items(): assert style["directive"].startswith("Il mercato e "), \ f"{name} directive must start with 'Il mercato e '" def test_directives_end_archetype(prompts: dict) -> None: for name, style in prompts["styles"].items(): assert "Archetipo dominante:" in style["directive"][-200:], \ f"{name} missing Archetipo dominante in last 200 chars" ``` - [ ] **Step 2: Run test — atteso FAIL (file non esiste)** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_prompts.py -v ``` Expected: FileNotFoundError. - [ ] **Step 3: Scrivi `prompts.json`** Crea `src/strategy_pythagoras/strategy_pythagoras/prompts.json` come segue (le directive vanno mantenute ASCII-safe e di lunghezza 800-950 char ciascuna). ```json { "_comment": "Stili cognitivi del GA per strategy_pythagoras. Dominio: pattern frattali ricorrenti su mercati crypto secondo il framework Pythagoras-Malanga (Fourier + frattali H-C + Evideon).", "_schema": "3.2", "_changelog": "v1.0 - Initial release. Schema clonato da strategy_crypto v3.2 con contenuto Pythagoras-aligned.", "_focus_metrics_design": "Le focus_metrics sono ENFASI per la lente, non filtri. 4 per stile. Devono includere almeno 1 dei 3 indicatori Pythagoras (candle_pattern, pythagorean_ratio, fractal_mirror).", "_design_invariants": "(1) ASCII-safe; (2) Archetipo dominante: come ancora; (3) Lookback range esplicito; (4) Prima frase 'Il mercato e ...'; (5) Lunghezza directive 800-950 char.", "_param_encoding_note": "Vincolo grammar: tutti i params sono float. candle_pattern: [length, sym0, sym1, ...] (sym: 0=U up close>open, 1=D down close 2% sui literal (numerologia spuriosa); (3) fractal_mirror con k = lookback_window (tautologico, sempre prossimo a 1.0); (4) letterali con piu di 4 decimali (es. 1.6180339 -> usa 1.618); (5) piu di 4 condizioni in AND; (6) crossover tra indicatori dello stesso tipo con lookback vicini (chattering); (7) time_in_market > 80% (leveraged buy&hold camuffato).", "output_priorities": "Trade-off: (1) coerenza con la lente cognitiva: la strategia deve essere riconoscibile come emanata dal tuo stile (es. pythagorean usa ratios, candle_grammarian sequenze esplicite); (2) asset-invariance: segnali che attivano sia su BTC che su ETH entro 36 barre (e' il bonus della fitness); (3) selettivita: poche entry forti, alto SNR; (4) composizionalita: pattern lunghi come somma di corti; (5) robustezza vs random baseline (3-sigma richiesto dal skeptic_quant).", "styles": { "pythagorean": { "directive": "Il mercato e un'armonia di ratios sacri: la sezione aurea (phi=1.618 e 1/phi=0.618), la radice di due (1.414), pi mezzi (1.571) ed e mezzi (1.359) strutturano i livelli di supporto, resistenza e target. Usa pythagorean_ratio con lookback Fibonacci (89, 144) per misurare l'estensione del range recente e cerca punti dove esso e' prossimo a phi entro 0.5%: indica un swing maturo che tende all'inversione armonica. Combina con candle_pattern di 3-5 candele per timing. Evita di chiudere troppo presto: la potenza di un ratio sacro si manifesta quando il prezzo passa il livello e poi torna a testarlo. Mai usare letterali con piu di 4 decimali: l'armonia non e' nei decimali ma nella struttura. Lookback consigliato: 89-144. Archetipo dominante: musicista che ascolta gli accordi nascosti del mercato.", "focus_metrics": ["pythagorean_ratio", "candle_pattern", "sma_pct", "realized_vol"] }, "fractal_geometer": { "directive": "Il mercato e autosimilare: una sequenza di 3 candele si ripete dilatata su 6, 12 candele e oltre. La tua firma e' detectare la stessa struttura a scale diverse. Usa candle_pattern di lunghezza 3 per identificare l'unita atomica, poi cerca conferma con fractal_mirror su finestra 2x o 3x: se la correlation e' alta significa che la scala maggiore replica la struttura. Pythagorean_ratio con lookback grande (>100) misura l'ampiezza globale del frattale e i suoi target di proiezione. Atr_pct conferma che la volatilita corrente e coerente con la scala. Evita pattern di lunghezza > 6: i frattali complessi vanno composti, non vincolati esplicitamente. Mai usare fractal_mirror con k uguale al lookback_window. Lookback consigliato: 48-144. Archetipo dominante: geometra che misura la dimensione di Hausdorff del prezzo.", "focus_metrics": ["candle_pattern", "fractal_mirror", "atr_pct", "pythagorean_ratio"] }, "fourier_analyst": { "directive": "Il mercato e somma di seni: una funzione complessa scomposta in frequenze armoniche. Il tuo compito e' isolare le componenti ricorrenti. Usa sma_pct su 3 lookback diversi (es. 30, 60, 120) come proxy delle componenti a bassa frequenza; quando 2 su 3 si allineano nella stessa direzione, la frequenza dominante e' chiara. Realized_vol misura l'energia totale; un calo improvviso indica trasformazione di fase. Candle_pattern di 4-6 candele identifica l'inviluppo locale del segnale composito. Cerca pattern dove il presente e' la proiezione del passato dopo una traslazione temporale costante (Poincare). Evita oscillatori dello stesso tipo a lookback vicini: chattering. Lookback consigliato: 60-200. Archetipo dominante: ingegnere del segnale che ricostruisce la portante nascosta.", "focus_metrics": ["sma_pct", "realized_vol", "candle_pattern", "atr"] }, "evideonic_projector": { "directive": "Il mercato e il presente proiettato dal passato via due operazioni: mirror H (inversione temporale) e mirror V (inversione prezzo), poi scalato per phi, 1/phi, sqrt2. Il tuo segnale principale e' fractal_mirror su entrambi gli assi: quando l'h mirror su 6-12 candele e' altamente positivo (>0.7), il pattern recente e' un riflesso temporale di un pattern precedente — segnale di prosecuzione armonica; quando v mirror e' positivo, e' un riflesso direzionale — segnale di inversione. Usa pythagorean_ratio per fissare i target di proiezione (entry x phi). Candle_pattern serve da gate per filtrare il contesto. Evita target con ratio > 3 (zona di rumore, fuori scala phi). Lookback consigliato: 24-96. Archetipo dominante: proiettore evideonico che riflette ogni evento nei suoi quattro alter-ego.", "focus_metrics": ["fractal_mirror", "pythagorean_ratio", "candle_pattern", "sma_pct"] }, "candle_grammarian": { "directive": "Il mercato parla una lingua di tre simboli: U (close>open), D (close 7 (frasi troppo lunghe = overfitting). Lookback consigliato: 12-48. Archetipo dominante: grammatico che decifra la sintassi delle candele.", "focus_metrics": ["candle_pattern", "volume", "atr", "realized_vol"] }, "recurrence_theorist": { "directive": "Il mercato e regolato da Poincare: ogni configurazione torna prima o poi vicino a se stessa. Il tuo lavoro e' trovare il pattern di oggi che firo' anche ieri a distanza k. Usa candle_pattern su 5-8 candele come firma, poi fractal_mirror con axis=0 (h, mirror temporale) su lookback grande (100-200) per misurare quanto la struttura corrente assomigli a quella passata invertita: se >0.5, il ciclo e' nella fase di ripetizione. Pythagorean_ratio fornisce target di estensione. Sma_pct su lookback medio conferma il regime macro. Evita pattern troppo corti (3 candele): troppo poco distinti per ricorrenza affidabile. Cerca pattern lunghi e selettivi (poche entry). Lookback consigliato: 100-200. Archetipo dominante: archeologo del prezzo che cerca repliche di antichi pattern.", "focus_metrics": ["candle_pattern", "fractal_mirror", "pythagorean_ratio", "sma_pct"] }, "skeptic_quant": { "directive": "Il mercato e rumore con piccoli edge statistici. Tu sei l'anticorpo del sistema: non fidarti dei pattern senza significativita 3-sigma vs random. Usa realized_vol e atr_pct come gate di regime (entra solo in regimi di volatilita media, 30-70 percentile). Sma_pct conferma direzione del trend macro. Candle_pattern di 3-4 candele puo' essere usato, ma solo come conferma di un setup gia validato da volatilita e trend: mai come segnale primario. Vincolo duro: max 3 condizioni in AND (poche e robuste); literal con almeno il 20% di margine di sicurezza tra entry ed exit. Nessun pythagorean_ratio (e numerologia spuria) ne fractal_mirror (tautologici nella tua lente). Lookback consigliato: 60-150. Archetipo dominante: scettico quantitativo che falsifica prima di confermare.", "focus_metrics": ["realized_vol", "atr_pct", "sma_pct", "candle_pattern"] } } } ``` - [ ] **Step 4: Run test — atteso PASS** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_prompts.py -v ``` Expected: tutti i test PASS. Se un test sulla `directive length` fallisce per un singolo stile, edita quella directive aggiungendo/rimuovendo testo finché rientra in [800, 950]. Ripeti finché tutti i test passano. - [ ] **Step 5: Commit** ```bash git add src/strategy_pythagoras/strategy_pythagoras/prompts.json src/strategy_pythagoras/tests/test_prompts.py git commit -m "feat(strategy_pythagoras): prompts.json with 7 Pythagoras-aligned cognitive styles" ``` --- ## Phase 3 — Backend port (paper-trading) ### Task 3.1: Port `schema.py` (DB DDL) **Files:** - Create: `src/strategy_pythagoras/strategy_pythagoras/backend/schema.py` - [ ] **Step 1: Read source pattern** ```bash cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/backend/schema.py ``` - [ ] **Step 2: Crea `src/strategy_pythagoras/strategy_pythagoras/backend/schema.py` come copia esatta** Copia il contenuto, sostituendo l'env var di default da `STRATEGY_CRYPTO_DB_PATH` a `STRATEGY_PYTHAGORAS_PAPER_DB_PATH`, e il default path da `state/strategy_crypto.db` a `state/strategy_pythagoras_paper.db`. Mantieni invariati nome tabelle (`paper_trading_runs`, `paper_trading_positions`, ecc.). - [ ] **Step 3: Smoke test** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c " import os, tempfile os.environ['STRATEGY_PYTHAGORAS_PAPER_DB_PATH'] = tempfile.mktemp(suffix='.db') from strategy_pythagoras.backend.schema import init_schema init_schema() print('OK') " ``` Expected: stampa `OK` senza errori. - [ ] **Step 4: Commit** ```bash git add src/strategy_pythagoras/strategy_pythagoras/backend/schema.py git commit -m "feat(strategy_pythagoras): port DB schema from strategy_crypto" ``` --- ### Task 3.2: Port `portfolio.py`, `executor.py`, `persistence.py` **Files:** - Create: `src/strategy_pythagoras/strategy_pythagoras/backend/portfolio.py` - Create: `src/strategy_pythagoras/strategy_pythagoras/backend/executor.py` - Create: `src/strategy_pythagoras/strategy_pythagoras/backend/persistence.py` - [ ] **Step 1: Read source pattern** ```bash cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/backend/portfolio.py cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/backend/executor.py cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/backend/persistence.py ``` - [ ] **Step 2: Copia i 3 file in `src/strategy_pythagoras/strategy_pythagoras/backend/`** Per ciascun file: copia 1:1. Sostituisci tutti i riferimenti `strategy_crypto` → `strategy_pythagoras` (import path, type hint, eventuali commenti). NON sostituire i riferimenti a `multi_swarm_core` (resta invariato). - [ ] **Step 3: Aggiorna `backend/__init__.py`** Replica esattamente l'`__init__.py` di `strategy_crypto.backend` adattando i nomi del package. - [ ] **Step 4: Smoke test imports** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c " from strategy_pythagoras.backend import Portfolio, PaperExecutor, PaperRepository print('OK') " ``` (Sostituisci i simboli effettivamente esportati se diversi.) Expected: `OK`. - [ ] **Step 5: Commit** ```bash git add src/strategy_pythagoras/strategy_pythagoras/backend/ git commit -m "feat(strategy_pythagoras): port paper-trading backend (Portfolio, Executor, Repository)" ``` --- ## Phase 4 — Fitness callback (asset-invariance) ### Task 4.1: Implementa `fitness_invariance.py` + integra nel GA loop **Files:** - Create: `src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py` - Test: `src/strategy_pythagoras/tests/test_fitness_invariance.py` - [ ] **Step 1: Crea test failing** `src/strategy_pythagoras/tests/test_fitness_invariance.py`: ```python """Bonus invariance: pattern che firano simultaneamente su 2 asset entro tolleranza.""" from __future__ import annotations import numpy as np import pandas as pd import pytest from strategy_pythagoras.fitness_invariance import ( apply_invariance_bonus, corr_signal, ) def test_corr_signal_perfect_alignment() -> None: # Tutte le entry su BTC coincidono con entry su ETH allo stesso tick. 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: base = 1.0 bonus = 0.5 alpha = 0.3 result = apply_invariance_bonus(base, bonus, alpha) assert result == 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]) # nessuna entry -> definiamo corr_signal=0 (nessun bonus) assert corr_signal(entries_btc, entries_eth, tolerance_bars=0) == 0.0 ``` - [ ] **Step 2: Run test — atteso FAIL (modulo non esiste)** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_fitness_invariance.py -v ``` - [ ] **Step 3: Crea `fitness_invariance.py`** `src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py`: ```python """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. tolerance_bars: finestra di tolleranza in barre. Returns: ∈ [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: # cerca un t' in b_set t.c. |t' - ti| <= tolerance_bars # ti e' un index value; assumiamo index intero monotono (bar index) 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) ``` - [ ] **Step 4: Run test — atteso PASS** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/test_fitness_invariance.py -v ``` Expected: tutti i test PASS. - [ ] **Step 5: Identifica il punto di integrazione nel GA loop** ```bash grep -rn 'compute_fitness\|fitness_train\|fitness_oos' /opt/docker/multi_swarm_coevolutive/src/multi_swarm_core/multi_swarm_core/ga/ /opt/docker/multi_swarm_coevolutive/src/multi_swarm_core/multi_swarm_core/orchestrator/ 2>/dev/null | head -20 ``` Espandi la lettura del file che chiama `compute_fitness`. Verifica se il GA accetta una callback di post-processing per la fitness, o se la fitness e' iniettabile da fuori. Se NO (no hook esposto): documenta nel runner script (Phase 6.1) che lo smoke test fa un wrapping esterno alla fitness scoring: dopo che il GA produce winners, ricalcola fitness con bonus e ri-sorta. Questo e' il path piu pulito non-invasivo. - [ ] **Step 6: Commit** ```bash git add src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py src/strategy_pythagoras/tests/test_fitness_invariance.py git commit -m "feat(strategy_pythagoras): invariance bonus callback (BTC↔ETH corr_signal)" ``` --- ## Phase 5 — Frontend port ### Task 5.1: Port `frontend/data.py` con dual-DB reader **Files:** - Create: `src/strategy_pythagoras/strategy_pythagoras/frontend/data.py` - [ ] **Step 1: Read source** ```bash cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/frontend/data.py ``` - [ ] **Step 2: Copia + adatta path env vars** Crea `src/strategy_pythagoras/strategy_pythagoras/frontend/data.py` come copia con sostituzioni: - `STRATEGY_CRYPTO_DB_PATH` → `STRATEGY_PYTHAGORAS_PAPER_DB_PATH` - `state/strategy_crypto.db` → `state/strategy_pythagoras_paper.db` - `state/strategy_crypto_paper.db` (se citato) → `state/strategy_pythagoras_paper.db` - Default GA DB env: usa `STRATEGY_PYTHAGORAS_DB_PATH` (default `state/strategy_pythagoras.db`) - [ ] **Step 3: Aggiungi 2 query helper per le nuove tab della dashboard** In coda al file aggiungi: ```python def load_invariance_metrics(ga_db_path: str) -> "pd.DataFrame": """Per ogni winner ritorna (genome_id, sharpe_btc, sharpe_eth, invariance_score).""" import sqlite3 import pandas as pd con = sqlite3.connect(ga_db_path) try: # NB: lo schema effettivo dipende da come il runner salva le metric. # Schema atteso (vedi Phase 6.1, scripts/run_pythagoras_smoke.py): # table pythagoras_winners(genome_id TEXT, cognitive_style TEXT, # fitness REAL, sharpe_btc REAL, sharpe_eth REAL, # invariance_score REAL, rules_json TEXT) return pd.read_sql_query( "SELECT genome_id, cognitive_style, fitness, sharpe_btc, sharpe_eth, invariance_score " "FROM pythagoras_winners ORDER BY fitness DESC", con, ) finally: con.close() def load_candle_pattern_usage(ga_db_path: str) -> "pd.DataFrame": """Per ogni winner estrae le sequenze candle_pattern usate (per heatmap).""" import json import sqlite3 import pandas as pd con = sqlite3.connect(ga_db_path) try: df = pd.read_sql_query( "SELECT genome_id, cognitive_style, rules_json FROM pythagoras_winners", con, ) finally: con.close() records: list[dict] = [] for _, row in df.iterrows(): rules = json.loads(row["rules_json"]).get("rules", []) for r in rules: for ind_name, params in _walk_indicators(r["condition"]): if ind_name == "candle_pattern": length = int(params[0]) syms = [int(s) for s in params[1:1 + length]] seq_str = "".join({0: "U", 1: "D", 2: "0"}[s] for s in syms) records.append({ "genome_id": row["genome_id"], "cognitive_style": row["cognitive_style"], "sequence": seq_str, "length": length, }) return pd.DataFrame.from_records(records) def _walk_indicators(node: dict): if "op" in node: for a in node.get("args", []): yield from _walk_indicators(a) elif node.get("kind") == "indicator": yield node["name"], node["params"] ``` - [ ] **Step 4: Smoke import** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c "from strategy_pythagoras.frontend import data; print('OK')" ``` - [ ] **Step 5: Commit** ```bash git add src/strategy_pythagoras/strategy_pythagoras/frontend/data.py git commit -m "feat(strategy_pythagoras): port frontend data layer + invariance/candle helpers" ``` --- ### Task 5.2: Port `nicegui_app.py` con 4 tabs **Files:** - Create: `src/strategy_pythagoras/strategy_pythagoras/frontend/nicegui_app.py` - [ ] **Step 1: Read source** ```bash cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/frontend/nicegui_app.py ``` - [ ] **Step 2: Crea `nicegui_app.py` portando struttura base** Replica scheletro del file `strategy_crypto`. Sostituzioni: - import: `from strategy_pythagoras.frontend.data import ...` - root_path env: `DASHBOARD_ROOT_PATH` defaultato a `/strategy_pythagoras_gui` - Titolo pagina: "Strategy Pythagoras" Aggiungi 4 tab al posto di quelli di strategy_crypto: ```python # Pseudocode per la struttura tabs with ui.tabs() as tabs: t_genomes = ui.tab("Genomes") t_patterns = ui.tab("Patterns") t_ratios = ui.tab("Ratios") t_invariance = ui.tab("Invariance") with ui.tab_panels(tabs, value=t_genomes): with ui.tab_panel(t_genomes): # Table dei winners da load_invariance_metrics df = load_invariance_metrics(GA_DB_PATH) ui.table(rows=df.to_dict("records"), columns=[...]) with ui.tab_panel(t_patterns): # heatmap delle sequenze candle (count per stile) df_pat = load_candle_pattern_usage(GA_DB_PATH) # plotly heatmap: x=sequence, y=cognitive_style, z=count ... with ui.tab_panel(t_ratios): # histogram dei literal usati con pythagorean_ratio # bins centrati su [1.414, 1.571, 1.618, 2.0, 2.618, 3.1416] ... with ui.tab_panel(t_invariance): # scatter sharpe_btc vs sharpe_eth, color = invariance_score ... ``` L'implementazione completa di patterns/ratios/invariance tab dipende dai dati prodotti dal runner (Phase 6). Inizia con stub semplici (table dei dati grezzi). Plotly heatmap/scatter possono essere aggiunti dopo lo smoke test. - [ ] **Step 3: Smoke launch** ```bash cd /opt/docker/multi_swarm_coevolutive && timeout 5 uv run python -m strategy_pythagoras.frontend.nicegui_app || true ``` Expected: avvio senza eccezioni nei primi 5s (poi timeout — OK). - [ ] **Step 4: Commit** ```bash git add src/strategy_pythagoras/strategy_pythagoras/frontend/nicegui_app.py src/strategy_pythagoras/strategy_pythagoras/frontend/__init__.py git commit -m "feat(strategy_pythagoras): NiceGUI dashboard with 4 tabs (Genomes/Patterns/Ratios/Invariance)" ``` --- ## Phase 6 — Smoke GA run + analisi ### Task 6.1: Crea runner script `scripts/run_pythagoras_smoke.py` **Files:** - Create: `scripts/run_pythagoras_smoke.py` - [ ] **Step 1: Identifica entry point del GA core** ```bash grep -rn 'def run\|def main\|GA(' /opt/docker/multi_swarm_coevolutive/src/multi_swarm_core/multi_swarm_core/ga/ /opt/docker/multi_swarm_coevolutive/src/multi_swarm_core/multi_swarm_core/orchestrator/ 2>/dev/null | head -10 ls /opt/docker/multi_swarm_coevolutive/scripts/ ``` Read `scripts/run_paper_trading.py` come esempio se esiste; cerca un esempio simile (es. `scripts/run_ga.py`). - [ ] **Step 2: Scrivi `scripts/run_pythagoras_smoke.py`** ```python """Smoke test del GA per strategy_pythagoras. - Population 20, 5 generations - Asset: BTC 5m + ETH 5m (serie da strategy_crypto/series/) - Train: 2024-07 → 2024-12; Test: 2025-01 - Stili cognitivi: 7 da strategy_pythagoras/prompts.json - 3 nuovi indicatori (candle_pattern, pythagorean_ratio, fractal_mirror) registrati - Fitness post-processing: apply_invariance_bonus - Output: persiste winners in state/strategy_pythagoras.db (tabella pythagoras_winners) """ from __future__ import annotations import json import os import sqlite3 from pathlib import Path # Forza il caricamento dei nuovi indicatori prima del GA import strategy_pythagoras.indicators # noqa: F401 (registra side-effect via compiler.py) from strategy_pythagoras.fitness_invariance import ( apply_invariance_bonus, corr_signal, ) # Adatta gli import seguenti al vero entry point del core; placeholder esemplificativi: # from multi_swarm_core.orchestrator.run import run_ga # from multi_swarm_core.config import GaConfig ROOT = Path(__file__).resolve().parents[1] DB_PATH = Path(os.getenv("STRATEGY_PYTHAGORAS_DB_PATH", ROOT / "state" / "strategy_pythagoras.db")) PROMPTS_PATH = ROOT / "src" / "strategy_pythagoras" / "strategy_pythagoras" / "prompts.json" SERIES_DIR = ROOT / "src" / "strategy_crypto" / "series" RUN_NAME = "pythagoras-smoke-001" def init_winners_table(con: sqlite3.Connection) -> None: con.execute(""" CREATE TABLE IF NOT EXISTS pythagoras_winners ( genome_id TEXT PRIMARY KEY, cognitive_style TEXT, fitness REAL, sharpe_btc REAL, sharpe_eth REAL, invariance_score REAL, rules_json TEXT, generation INTEGER, run_name TEXT ) """) con.commit() def main() -> None: DB_PATH.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(DB_PATH) init_winners_table(con) # === BLOCCO DA ADATTARE ALL'API EFFETTIVA DEL CORE === # Pseudocodice di alto livello: leggi le firme reali da # multi_swarm_core.orchestrator e adattalo. L'idea e' (1) caricare 2 serie # (BTC, ETH 5m), (2) lanciare il GA con population=20, generations=5, # prompts da PROMPTS_PATH, (3) per ogni winner ricalcolare fitness # combinando sharpe_btc, sharpe_eth e bonus invariance, (4) persistere # in pythagoras_winners. # # raise NotImplementedError("Adatta gli import del core qui sopra; vedi orchestrator.run") print(f"Smoke test {RUN_NAME} pronto. Configura gli import del core e rilancia.") con.close() if __name__ == "__main__": main() ``` Lo script e' deliberatamente uno scheletro: l'esatto entry point del GA core va letto al momento dell'implementazione. L'implementer deve: 1. Aprire `src/multi_swarm_core/multi_swarm_core/orchestrator/*.py` e identificare la funzione che lancia un GA 2. Replicare il pattern di `scripts/run_paper_trading.py` (se esiste) o equivalente 3. Wirare l'output del GA al wrapping invariance + persistere in `pythagoras_winners` - [ ] **Step 3: Commit** ```bash git add scripts/run_pythagoras_smoke.py git commit -m "feat(strategy_pythagoras): scaffold smoke-test runner script (stub orchestrator wiring TODO at exec time)" ``` --- ### Task 6.2: Lancia smoke test (richiede consumo di token LLM) **STOP — CONFERMA UTENTE OBBLIGATORIA prima di proseguire.** Questo task consuma token reali via OpenRouter/Anthropic. Stima: ~20 genomi × 5 generazioni × 2 LLM call/genome ≈ 200 chiamate. Costo dipendente da `model_tier` distribuzione (70% C qwen, 30% B sonnet). - [ ] **Step 1: Chiedi conferma all'utente** > "Sto per lanciare il GA smoke run pythagoras-smoke-001 (pop=20, gen=5, BTC+ETH 5m). Costo stimato: ~200 LLM call (70% qwen, 30% sonnet). Procedo?" - [ ] **Step 2: Esegui (post-conferma)** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python scripts/run_pythagoras_smoke.py 2>&1 | tee state/pythagoras-smoke-001.log ``` Expected: 5 generazioni completate, almeno 1 winner persistito in `pythagoras_winners`. - [ ] **Step 3: Verifica persistenza** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c " import sqlite3 con = sqlite3.connect('state/strategy_pythagoras.db') rows = con.execute('SELECT cognitive_style, fitness, sharpe_btc, sharpe_eth, invariance_score FROM pythagoras_winners ORDER BY fitness DESC LIMIT 5').fetchall() for r in rows: print(r) " ``` Expected: top-5 winners stampati. - [ ] **Step 4: Verifica dashboard mostra winners** ```bash cd /opt/docker/multi_swarm_coevolutive && timeout 10 uv run python -m strategy_pythagoras.frontend.nicegui_app & sleep 3 curl -s http://localhost:8080/strategy_pythagoras_gui/ | head -50 | grep -i pythagoras || echo "MISSING" ``` (Adatta porta. Expected: trova "pythagoras" nel HTML.) - [ ] **Step 5: Commit** ```bash git add state/pythagoras-smoke-001.log git commit -m "chore(strategy_pythagoras): smoke run pythagoras-smoke-001 log" ``` --- ### Task 6.3: Scrivi report `docs/analysis_first_run.md` **Files:** - Create: `docs/analysis_first_run.md` - [ ] **Step 1: Estrai metriche dal DB** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c " import sqlite3, json con = sqlite3.connect('state/strategy_pythagoras.db') # Top-5 print('=== TOP-5 ===') for row in con.execute('SELECT genome_id, cognitive_style, fitness, sharpe_btc, sharpe_eth, invariance_score FROM pythagoras_winners ORDER BY fitness DESC LIMIT 5').fetchall(): print(row) # Conteggio sequenze candle_pattern per stile print('=== CANDLE PATTERNS ===') for row in con.execute('SELECT cognitive_style, rules_json FROM pythagoras_winners').fetchall(): style, rules_json = row rules = json.loads(rules_json).get('rules', []) # walk e dump def walk(n, out): if isinstance(n, dict): if n.get('name') == 'candle_pattern': out.append(n['params']) for v in n.values(): if isinstance(v, (dict, list)): walk(v, out) elif isinstance(n, list): for v in n: walk(v, out) cps = [] walk(rules, cps) for p in cps: length = int(p[0]); syms = ''.join({0:'U',1:'D',2:'0'}[int(s)] for s in p[1:1+length]) print(f' {style}: {syms}') # Histogram literal vicini a costanti universali print('=== RATIO LITERALS ===') for row in con.execute('SELECT rules_json FROM pythagoras_winners').fetchall(): rules = json.loads(row[0]).get('rules', []) def walk(n, out): if isinstance(n, dict): if n.get('op') in ('gt','lt','eq') and len(n.get('args',[])) == 2: a, b = n['args'] if a.get('name') == 'pythagorean_ratio' and b.get('kind') == 'literal': out.append(float(b['value'])) if b.get('name') == 'pythagorean_ratio' and a.get('kind') == 'literal': out.append(float(a['value'])) for v in n.values(): if isinstance(v, (dict, list)): walk(v, out) elif isinstance(n, list): for v in n: walk(v, out) rl = [] walk(rules, rl) for v in rl: print(f' {v:.4f}') " | tee state/pythagoras-smoke-001-stats.txt ``` - [ ] **Step 2: Scrivi `docs/analysis_first_run.md`** Crea il report con le 6 sezioni elencate nella spec §4, popolandole con i dati estratti allo Step 1. Struttura attesa (gli `{{placeholder}}` vanno sostituiti coi dati reali): ```markdown # strategy_pythagoras — Analisi del primo run **Run:** pythagoras-smoke-001 **Data:** {{YYYY-MM-DD}} **Setup:** population=20, generations=5, BTC+ETH 5m, train 2024-07/12, test 2025-01 ## 1. Sintesi numerica dei riassunti - [Pythagoras_Trading_Prediction.summary.md](../src/strategy_pythagoras/Pythagoras/Pythagoras_Trading_Prediction.summary.md) - [Libro_frattali.summary.md](../src/strategy_pythagoras/Pythagoras/Libro_frattali.summary.md) Numeri chiave dai due libri: phi=1.618, pi=3.1416, sqrt2=1.414, e=2.718; Solfeggio 396-417-528-639-741-852; 25 linee / 588 candele; tassonomia range 3-6-12-24-39-56; 57 pattern LONG nel Libro su spazio teorico 1080 = 3^3+3^4+3^5+3^6. ## 2. Top-5 winners | Rank | genome_id | cognitive_style | fitness | sharpe_BTC | sharpe_ETH | invariance_score | |---|---|---|---|---|---|---| | 1 | {{...}} | {{...}} | {{...}} | {{...}} | {{...}} | {{...}} | | ... | | | | | | | ## 3. Pattern frattali emersi Sequenze `candle_pattern` usate dai winners (dump): | cognitive_style | sequenza | length | |---|---|---| | {{...}} | {{...}} | {{...}} | **Overlap teorico:** dei 1080 pattern combinatori possibili {U,D,0}^3..6, i winner ne usano {{N}} distinti. Confronto con i 57 LONG del Libro dei Frattali: copertura non valutabile senza decoding visuale del Libro (esplicitamente fuori scope: "senza passare alle immagini"). ## 4. Ratios di prezzo emersi Distribuzione dei literal usati in confronti con `pythagorean_ratio`: | Bin | Range | Count | |---|---|---| | ≈ 1/phi | [0.59, 0.65] | {{...}} | | ≈ sqrt2 | [1.39, 1.44] | {{...}} | | ≈ phi | [1.59, 1.65] | {{...}} | | ≈ 2.0 | [1.95, 2.05] | {{...}} | | ≈ phi^2 | [2.58, 2.66] | {{...}} | | ≈ pi | [3.10, 3.20] | {{...}} | | altro | resto | {{...}} | ## 5. Cross-asset invariance Istogramma `invariance_score` sui top genomi: {{stat distribution}}. Pattern firanti su entrambi gli asset entro ±36 barre (±3h su 5m TF): {{N}} / {{tot}}. ## 6. Conclusione (onesta) - Winners con sharpe > 1.0 su test set: {{N}} su {{tot}} - Winners con invariance_score > 0.3: {{N}} - Lo skeptic_quant ha generato strategie significativamente diverse dagli altri stili: {{si/no}} - Cosa il framework Pythagoras predice e cosa NON regge al backtest: - **Tiene:** {{lista delle predizioni del paper che hanno qualche supporto empirico}} - **Non tiene:** {{lista delle predizioni senza supporto, es. numeri Solfeggio assenti dai winners}} Prossimi passi: - Estendere a Oro/Argento (asset citati dal paper) se invariance > 0.3 conferma - Estendere a range candele 12-56 se range 3-12 mostra edge stabile - Far girare runs piu lunghi (pop=50, gen=20) solo se questo smoke conferma trend positivo ``` - [ ] **Step 3: Commit** ```bash git add docs/analysis_first_run.md state/pythagoras-smoke-001-stats.txt git commit -m "docs(strategy_pythagoras): first-run analysis report" ``` --- ## Phase 7 — Verifica criteri di accettazione ### Task 7.1: Esegui tutti i criteri di §6 della spec - [ ] **Step 1: `uv sync` riesce** ```bash cd /opt/docker/multi_swarm_coevolutive && uv sync ``` - [ ] **Step 2: Import dei 3 indicatori** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run python -c "from strategy_pythagoras.indicators import candle_pattern, pythagorean_ratio, fractal_mirror; print('OK')" ``` - [ ] **Step 3: pytest tutta la suite Pythagoras + core** ```bash cd /opt/docker/multi_swarm_coevolutive && uv run pytest src/strategy_pythagoras/tests/ src/multi_swarm_core/tests/ -v ``` Expected: 100% verde. - [ ] **Step 4: GA short run completato** ```bash cd /opt/docker/multi_swarm_coevolutive && sqlite3 state/strategy_pythagoras.db "SELECT COUNT(*) FROM pythagoras_winners WHERE run_name='pythagoras-smoke-001'" ``` Expected: > 0. - [ ] **Step 5: Almeno 1 winner con fitness > 0 e stile Pythagoras** ```bash cd /opt/docker/multi_swarm_coevolutive && sqlite3 state/strategy_pythagoras.db "SELECT cognitive_style, fitness FROM pythagoras_winners WHERE fitness > 0 ORDER BY fitness DESC LIMIT 1" ``` - [ ] **Step 6: Dashboard avvia su subpath** ```bash cd /opt/docker/multi_swarm_coevolutive && timeout 5 uv run python -m strategy_pythagoras.frontend.nicegui_app & sleep 2 curl -fsS http://localhost:8080/strategy_pythagoras_gui/ > /dev/null && echo "OK" || echo "FAIL" kill %1 2>/dev/null ``` - [ ] **Step 7: `docs/analysis_first_run.md` esiste e contiene 6 sezioni** ```bash cd /opt/docker/multi_swarm_coevolutive && grep -c '^## ' docs/analysis_first_run.md ``` Expected: ≥ 6. - [ ] **Step 8: Final commit (se non già fatto)** ```bash cd /opt/docker/multi_swarm_coevolutive && git status ``` Se ci sono modifiche pendenti, commit con messaggio appropriato. --- ## Self-review **Spec coverage:** | Requisito spec | Task | |---|---| | §1 Layout package | Task 0.1 | | §1 Workspace registration | Task 0.1 | | §1 Persistence + env vars | Task 0.1, 3.1 | | §2 Genoma invariato | (no task: implicito; nessun cambio core ad hypothesis.py) | | §2 3 indicatori (`candle_pattern`, `pythagorean_ratio`, `fractal_mirror`) | Task 1.1, 1.2, 1.3 | | §2 Vincoli protettivi | Task 1.2 (validator), Task 2.1 (anti_patterns) | | §3 prompts.json 7 stili | Task 2.1 | | §3 ASCII-safe, archetipo, lookback range, lunghezza | Task 2.1 (test_prompts.py) | | §4 Fitness con bonus invariance | Task 4.1 | | §4 GA run short | Task 6.1, 6.2 | | §4 Deliverable analysis report | Task 6.3 | | §4 Dashboard 4 tabs | Task 5.1, 5.2 | | §6 Criteri accettazione | Task 7.1 | **Placeholder scan:** Tutti gli step contengono codice concreto o comandi specifici. I `{{placeholder}}` nel report `docs/analysis_first_run.md` sono esplicitamente sostituiti coi dati estratti nello Step 1 di Task 6.3. Le porzioni "BLOCCO DA ADATTARE ALL'API EFFETTIVA DEL CORE" in Task 6.1 sono inevitabili senza prima leggere `multi_swarm_core/orchestrator`; istruzioni esplicite su come adattare sono fornite. **Type consistency:** Indicator names sono `candle_pattern`, `pythagorean_ratio`, `fractal_mirror` ovunque. Encoding `axis_int=0` (h) / `1` (v) coerente nei test, indicators.py, validator. Env vars: `STRATEGY_PYTHAGORAS_DB_PATH` (GA), `STRATEGY_PYTHAGORAS_PAPER_DB_PATH` (paper), `GA_INVARIANCE_ALPHA`, `GA_INVARIANCE_TOLERANCE_BARS` coerenti tra fitness_invariance.py, schema.py, runner script. **Note esecuzione:** Task 6.2 (smoke run) richiede conferma utente esplicita (consumo token). Task 6.1 contiene uno scheletro: l'esatta API del core orchestrator va letta al momento. Tutto il resto è eseguibile in autonomia.