14f476dd09
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).
1607 lines
66 KiB
Markdown
1607 lines
66 KiB
Markdown
# 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: <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**
|
||
|
||
```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.
|