Base per l'esecuzione del sub-project strategy_pythagoras: - docs/superpowers/specs/2026-05-19-strategy-pythagoras-design.md (spec) - docs/superpowers/plans/2026-05-19-strategy-pythagoras.md (14 task, 72 step TDD) - src/strategy_pythagoras/Pythagoras/*.summary.md (riassunti numerici dei 2 PDF) - src/strategy_pythagoras/Pythagoras/_extracted/*.txt (estrazione testo grezzo via pypdf) I PDF stessi non vengono committati (vedi .gitignore).
66 KiB
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], simboli0=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
[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
# 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
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_PATHSTRATEGY_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]: aggiungistrategy-pythagoras = { workspace = true }sottostrategy-crypto -
In
[tool.pytest.ini_options].testpaths: aggiungi"src/strategy_pythagoras/tests" -
Step 5: Esegui
uv syncper validare
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
cd /opt/docker/multi_swarm_coevolutive && uv run python -c "import strategy_pythagoras; print('OK')"
Expected: stampa OK.
- Step 7: Commit
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:
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
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
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:
"""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)
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_ARITYnel validator
Read src/multi_swarm_core/multi_swarm_core/protocol/validator.py e localizza il dict INDICATOR_ARITY. Aggiungi le tre righe:
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:
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
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
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:
"""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)
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
"""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
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:
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:
"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:
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
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
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:
"""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)
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).
{
"_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: <metafora> 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<open, 2=doji). pythagorean_ratio: [lookback]. fractal_mirror: [k, axis_int] (0=h temporale, 1=v prezzo).",
"agent_role": "Sei un agente generatore di ipotesi di trading quantitativo per un sistema swarm coevolutivo che cerca pattern frattali ricorrenti sui mercati crypto secondo il framework Pythagoras-Malanga (frattali H-C, trasformata di Fourier, geometria Evideon). Sei parte di una popolazione che esplora collettivamente lo spazio dei pattern: la diversita delle ipotesi e un asset critico. Preferisci esplorare territori meno ovvi rispetto a quelli che la tua lente cognitiva renderebbe predicibili. La strategia che produci deve essere riconoscibile come emanata dal tuo stile.",
"pattern_guidance": "Forme da cercare:\n - Sequenze candle 3-12 lunghezza in alfabeto {U,D,doji} via candle_pattern\n - Ratios di prezzo vicini a 1.618 (phi), 0.618 (1/phi), 1.414 (sqrt2), 1.571 (pi/2), 2.718 (e) via pythagorean_ratio\n - Mirror H/V come proiezioni via fractal_mirror\n - Pattern composti: lunghi come concatenazione di corti (Pattern 13 = Pattern 3 + Pattern 11 nel paper)\n - Cicli ricorrenti via Poincare: stesso pattern firato a distanze regolari\nCerca pattern che si REPLICANO su BTC e ETH al ~stesso timestamp (entro 36 barre = 3h su 5m TF).",
"instruction": "Genera una strategia che cerchi pattern frattali ricorrenti coerenti col framework Pythagoras-Malanga, riconoscibile come emanata dal tuo stile.",
"domain_warnings": "Crypto trada 24/7 senza CME gap. I numeri sacri (Solfeggio 396-417-528-639-741-852 Hz, 137.0359, 121.449, 161.86) sono prior teorici, NON leggi: usali come scale candidate per literal, non come dogma. Il paper Pythagoras e' esplicitamente non-falsificabile (cita 'consapevolezza del trader' come jolly): il backtest e' l'unico arbitro. La tolleranza ±3h del paper su 5m TF = ±36 barre per il bonus di asset-invariance.",
"anti_patterns": "Evita: (1) sequenza candle_pattern con piu di 7 simboli vincolati (overfitting); (2) pythagorean_ratio con tolleranza > 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<open), 0 (doji). Ogni 'parola' di 3-12 lettere e' una frase. Il tuo dominio e' la sintassi diretta: usa candle_pattern come segnale principale, con sequenze di lunghezza 3-5 per la decisione di entry. Cerca pattern dove l'ultima candela e' U dopo una serie 'DD' o '0D' (reversal classico) o pattern 'UU0U' che indicano accumulazione con pausa. Volume conferma la 'pronuncia' del pattern: alto volume = pattern ben articolato. Atr modula la dimensione di stop e target. Combina al massimo due candle_pattern indipendenti in AND, mai di piu. Mai pattern di lunghezza > 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
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
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
cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/backend/schema.py
- Step 2: Crea
src/strategy_pythagoras/strategy_pythagoras/backend/schema.pycome 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
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
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
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
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
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:
"""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)
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:
"""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
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
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
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
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(defaultstate/strategy_pythagoras.db) -
Step 3: Aggiungi 2 query helper per le nuove tab della dashboard
In coda al file aggiungi:
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
cd /opt/docker/multi_swarm_coevolutive && uv run python -c "from strategy_pythagoras.frontend import data; print('OK')"
- Step 5: Commit
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
cat /opt/docker/multi_swarm_coevolutive/src/strategy_crypto/strategy_crypto/frontend/nicegui_app.py
- Step 2: Crea
nicegui_app.pyportando struttura base
Replica scheletro del file strategy_crypto. Sostituzioni:
- import:
from strategy_pythagoras.frontend.data import ... - root_path env:
DASHBOARD_ROOT_PATHdefaultato a/strategy_pythagoras_gui - Titolo pagina: "Strategy Pythagoras"
Aggiungi 4 tab al posto di quelli di strategy_crypto:
# 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
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
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
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
"""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:
- Aprire
src/multi_swarm_core/multi_swarm_core/orchestrator/*.pye identificare la funzione che lancia un GA - Replicare il pattern di
scripts/run_paper_trading.py(se esiste) o equivalente - Wirare l'output del GA al wrapping invariance + persistere in
pythagoras_winners
- Step 3: Commit
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)
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
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
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
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
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):
# 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
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 syncriesce
cd /opt/docker/multi_swarm_coevolutive && uv sync
- Step 2: Import dei 3 indicatori
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
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
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
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
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.mdesiste e contiene 6 sezioni
cd /opt/docker/multi_swarm_coevolutive && grep -c '^## ' docs/analysis_first_run.md
Expected: ≥ 6.
- Step 8: Final commit (se non già fatto)
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.