# Feature temporali nella grammatica Hypothesis — 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:** Aggiungere quattro feature temporali (`hour`, `dow`, `is_weekend`, `minute_of_hour`) alla grammatica delle strategie Hypothesis come `FeatureNode`, universalmente accessibili a ogni genoma e usabili con i comparator esistenti. **Architecture:** Estensione puramente additiva. La whitelist `KNOWN_FEATURES` in `protocol/grammar.py` cresce da 5 a 9 nomi. Il dispatcher di `FeatureNode` in `protocol/compiler.py` acquisisce un branch prioritario che mappa i nomi temporali a serie derivate da `df.index` (DatetimeIndex UTC). Il prompt template di `agents/hypothesis.py` riceve due esempi few-shot. Nessuna modifica a parser, mutation/crossover, genome dataclass. **Tech Stack:** Python 3.13, pandas (DatetimeIndex), pytest. Esecuzione via `uv run`. Repository: `/home/adriano/Documenti/Git_XYZ/Multi_Swarm_Coevolutive`. **Spec di riferimento:** `docs/superpowers/specs/2026-05-11-temporal-features-design.md` --- ## File map | File | Tipo | Responsabilità | |------|------|----------------| | `src/multi_swarm/protocol/grammar.py` | Modify | Estendere `KNOWN_FEATURES` | | `src/multi_swarm/protocol/compiler.py` | Modify | Aggiungere `_TIME_FEATURE_FNS` + branch in `_eval_node` | | `src/multi_swarm/agents/hypothesis.py` | Modify | Estendere prompt template con sezione feature temporali + 2 esempi | | `tests/unit/test_protocol_validator.py` | Modify | +2 test (accept/reject) | | `tests/unit/test_protocol_compiler.py` | Modify | +5 test (4 feature + 1 integrazione) | --- ## Task 1: Grammar extension + validator tests **Files:** - Modify: `src/multi_swarm/protocol/grammar.py:21-23` - Modify: `tests/unit/test_protocol_validator.py` (append) - [ ] **Step 1.1: Write failing test — validator accepts temporal features** Append to `tests/unit/test_protocol_validator.py`: ```python def test_validator_accepts_temporal_features() -> None: for name in ("hour", "dow", "is_weekend", "minute_of_hour"): src = _wrap( { "op": "gt", "args": [ {"kind": "feature", "name": name}, {"kind": "literal", "value": 0}, ], } ) ast = parse_strategy(src) validate_strategy(ast) # no exception def test_validator_rejects_temporal_typo() -> None: src = _wrap( { "op": "gt", "args": [ {"kind": "feature", "name": "weekday"}, {"kind": "literal", "value": 0}, ], } ) ast = parse_strategy(src) with pytest.raises(ValidationError, match="unknown feature"): validate_strategy(ast) ``` - [ ] **Step 1.2: Run tests to verify they fail** Run: `uv run pytest tests/unit/test_protocol_validator.py::test_validator_accepts_temporal_features tests/unit/test_protocol_validator.py::test_validator_rejects_temporal_typo -v` Expected: First test FAILs with `ValidationError: unknown feature: hour`. Second test PASSes already (weekday is unknown today too). - [ ] **Step 1.3: Extend `KNOWN_FEATURES` whitelist** Edit `src/multi_swarm/protocol/grammar.py`, lines 21-23: ```python KNOWN_FEATURES: frozenset[str] = frozenset( {"open", "high", "low", "close", "volume", "hour", "dow", "is_weekend", "minute_of_hour"} ) ``` - [ ] **Step 1.4: Run tests to verify both pass** Run: `uv run pytest tests/unit/test_protocol_validator.py -v` Expected: All tests PASS (both new tests + all pre-existing ones). - [ ] **Step 1.5: Commit** ```bash git add src/multi_swarm/protocol/grammar.py tests/unit/test_protocol_validator.py git commit -m "feat(protocol): extend KNOWN_FEATURES with temporal feature names" ``` --- ## Task 2: Compiler — `hour` feature **Files:** - Modify: `src/multi_swarm/protocol/compiler.py:135-137` - Modify: `tests/unit/test_protocol_compiler.py` (append) - [ ] **Step 2.1: Write failing test for `hour`** Append to `tests/unit/test_protocol_compiler.py`: ```python def test_compile_hour_feature_returns_index_hour(ohlcv: pd.DataFrame) -> None: src = json.dumps( { "rules": [ { "condition": { "op": "gt", "args": [ {"kind": "feature", "name": "hour"}, {"kind": "literal", "value": -1}, ], }, "action": "entry-long", } ] } ) ast = parse_strategy(src) fn = compile_strategy(ast) signal = fn(ohlcv) # Tutte le righe hanno hour >= 0 > -1, quindi tutte entry-long assert (signal == Side.LONG).all() ``` - [ ] **Step 2.2: Run test to verify it fails** Run: `uv run pytest tests/unit/test_protocol_compiler.py::test_compile_hour_feature_returns_index_hour -v` Expected: FAIL with `KeyError: 'hour'` (df has no `hour` column, dispatcher falls into `df[name]`). - [ ] **Step 2.3: Add `_TIME_FEATURE_FNS` and dispatcher branch** Edit `src/multi_swarm/protocol/compiler.py`. Insert after line 108 (end of `INDICATOR_FNS`): ```python _TIME_FEATURE_FNS: dict[str, Callable[[pd.DatetimeIndex], pd.Series]] = { "hour": lambda idx: pd.Series(idx.hour, index=idx, dtype="int64"), "dow": lambda idx: pd.Series(idx.dayofweek, index=idx, dtype="int64"), "is_weekend": lambda idx: pd.Series((idx.dayofweek >= 5).astype("int64"), index=idx), "minute_of_hour": lambda idx: pd.Series(idx.minute, index=idx, dtype="int64"), } ``` Then modify `_eval_node` at line 135-137. Replace: ```python def _eval_node(node: Node, df: pd.DataFrame) -> pd.Series: if isinstance(node, FeatureNode): return df[node.name] ``` With: ```python def _eval_node(node: Node, df: pd.DataFrame) -> pd.Series: if isinstance(node, FeatureNode): if node.name in _TIME_FEATURE_FNS: return _TIME_FEATURE_FNS[node.name](df.index) return df[node.name] ``` - [ ] **Step 2.4: Run test to verify it passes** Run: `uv run pytest tests/unit/test_protocol_compiler.py::test_compile_hour_feature_returns_index_hour -v` Expected: PASS. - [ ] **Step 2.5: Commit** ```bash git add src/multi_swarm/protocol/compiler.py tests/unit/test_protocol_compiler.py git commit -m "feat(protocol): dispatcher temporal features (hour) in compiler" ``` --- ## Task 3: Compiler — `dow` and `is_weekend` tests **Files:** - Modify: `tests/unit/test_protocol_compiler.py` (append) Nessuna modifica al sorgente: `_TIME_FEATURE_FNS` definito in Task 2 contiene già le quattro funzioni. Questi test verificano semantica e copertura. - [ ] **Step 3.1: Add `dow` test** Append to `tests/unit/test_protocol_compiler.py`: ```python def test_compile_dow_feature_monday_is_zero(ohlcv: pd.DataFrame) -> None: # 2024-01-01 e' un lunedi -> dow=0; gating eq dow 0 deve dare LONG su monday only. src = json.dumps( { "rules": [ { "condition": { "op": "eq", "args": [ {"kind": "feature", "name": "dow"}, {"kind": "literal", "value": 0}, ], }, "action": "entry-long", } ] } ) ast = parse_strategy(src) fn = compile_strategy(ast) signal = fn(ohlcv) # ohlcv fixture: 200h da 2024-01-01 00:00 UTC -> primo lunedi e' bar 0..23 monday_hours = signal[(signal.index.dayofweek == 0)] other_hours = signal[(signal.index.dayofweek != 0)] assert (monday_hours == Side.LONG).all() assert (other_hours == Side.FLAT).all() ``` - [ ] **Step 3.2: Add `is_weekend` test** Append: ```python def test_compile_is_weekend_returns_zero_one(ohlcv: pd.DataFrame) -> None: src = json.dumps( { "rules": [ { "condition": { "op": "eq", "args": [ {"kind": "feature", "name": "is_weekend"}, {"kind": "literal", "value": 1}, ], }, "action": "entry-long", } ] } ) ast = parse_strategy(src) fn = compile_strategy(ast) signal = fn(ohlcv) weekend = signal[signal.index.dayofweek >= 5] weekdays = signal[signal.index.dayofweek < 5] assert (weekend == Side.LONG).all() assert (weekdays == Side.FLAT).all() ``` - [ ] **Step 3.3: Run both tests** Run: `uv run pytest tests/unit/test_protocol_compiler.py::test_compile_dow_feature_monday_is_zero tests/unit/test_protocol_compiler.py::test_compile_is_weekend_returns_zero_one -v` Expected: Both PASS. - [ ] **Step 3.4: Commit** ```bash git add tests/unit/test_protocol_compiler.py git commit -m "test(protocol): compiler semantica dow + is_weekend" ``` --- ## Task 4: Compiler — `minute_of_hour` test **Files:** - Modify: `tests/unit/test_protocol_compiler.py` (append) - [ ] **Step 4.1: Add `minute_of_hour` test** Append: ```python def test_compile_minute_of_hour_zero_on_1h_timeframe(ohlcv: pd.DataFrame) -> None: # Fixture ohlcv ha freq=1h, quindi tutti i minute_of_hour sono 0. # gating eq minute_of_hour 0 -> LONG su TUTTE le righe. src = json.dumps( { "rules": [ { "condition": { "op": "eq", "args": [ {"kind": "feature", "name": "minute_of_hour"}, {"kind": "literal", "value": 0}, ], }, "action": "entry-long", } ] } ) ast = parse_strategy(src) fn = compile_strategy(ast) signal = fn(ohlcv) assert (signal == Side.LONG).all() ``` - [ ] **Step 4.2: Run test** Run: `uv run pytest tests/unit/test_protocol_compiler.py::test_compile_minute_of_hour_zero_on_1h_timeframe -v` Expected: PASS. - [ ] **Step 4.3: Commit** ```bash git add tests/unit/test_protocol_compiler.py git commit -m "test(protocol): compiler semantica minute_of_hour su 1h" ``` --- ## Task 5: Compiler — integrazione con regola completa **Files:** - Modify: `tests/unit/test_protocol_compiler.py` (append) - [ ] **Step 5.1: Add integration test** Append: ```python def test_rule_with_temporal_gating_compiles_and_executes(ohlcv: pd.DataFrame) -> None: # Regola: entry-long se hour > 14 AND close > sma(20). # close in fixture e' lineare crescente, quindi close > sma(20) e' True dopo warmup. # entry-long deve apparire solo nelle bar con hour > 14. src = json.dumps( { "rules": [ { "condition": { "op": "and", "args": [ { "op": "gt", "args": [ {"kind": "feature", "name": "hour"}, {"kind": "literal", "value": 14}, ], }, { "op": "gt", "args": [ {"kind": "feature", "name": "close"}, {"kind": "indicator", "name": "sma", "params": [20]}, ], }, ], }, "action": "entry-long", } ] } ) ast = parse_strategy(src) fn = compile_strategy(ast) signal = fn(ohlcv) # Bar con hour <= 14: mai LONG (gating temporale blocca). morning = signal[signal.index.hour <= 14] assert (morning == Side.FLAT).all() # Bar con hour > 14 e dopo warmup sma (>=20 bar dall'inizio): LONG. afternoon_warm = signal[(signal.index.hour > 14) & (np.arange(len(signal)) >= 20)] assert (afternoon_warm == Side.LONG).all() ``` - [ ] **Step 5.2: Run test** Run: `uv run pytest tests/unit/test_protocol_compiler.py::test_rule_with_temporal_gating_compiles_and_executes -v` Expected: PASS. - [ ] **Step 5.3: Run full compiler + validator test suite to check regressions** Run: `uv run pytest tests/unit/test_protocol_compiler.py tests/unit/test_protocol_validator.py -v` Expected: All tests PASS (pre-existing + new). Nessun test rotto. - [ ] **Step 5.4: Commit** ```bash git add tests/unit/test_protocol_compiler.py git commit -m "test(protocol): integration test gating temporale + sma" ``` --- ## Task 6: Update Hypothesis prompt **Files:** - Modify: `src/multi_swarm/agents/hypothesis.py:84-85` - [ ] **Step 6.1: Edit prompt template** In `src/multi_swarm/agents/hypothesis.py`, alla riga 84-85 sostituire: ```python Leaf - feature OHLCV: {{"kind": "feature", "name": "open|high|low|close|volume"}} ``` con: ```python Leaf - feature OHLCV: {{"kind": "feature", "name": "open|high|low|close|volume"}} Leaf - feature TEMPORALI (sempre accessibili, UTC): {{"kind": "feature", "name": "hour"}} // range 0-23 {{"kind": "feature", "name": "dow"}} // range 0-6 (lun=0, dom=6) {{"kind": "feature", "name": "is_weekend"}} // 0 o 1 {{"kind": "feature", "name": "minute_of_hour"}} // range 0-59 Esempi di gating temporale: // Solo durante la sessione US (14:00-22:00 UTC) {{"op": "and", "args": [ {{"op": "gt", "args": [{{"kind": "feature", "name": "hour"}}, {{"kind": "literal", "value": 14}}]}}, {{"op": "lt", "args": [{{"kind": "feature", "name": "hour"}}, {{"kind": "literal", "value": 22}}]}} ]}} // Solo nel weekend (sab+dom) {{"op": "eq", "args": [{{"kind": "feature", "name": "is_weekend"}}, {{"kind": "literal", "value": 1}}]}} ``` - [ ] **Step 6.2: Run existing hypothesis tests to verify prompt format still valid** Run: `uv run pytest tests/unit/test_hypothesis_agent.py -v` Expected: All tests PASS. Il template `{feature_access}` continua a funzionare perché non lo abbiamo toccato. - [ ] **Step 6.3: Commit** ```bash git add src/multi_swarm/agents/hypothesis.py git commit -m "feat(hypothesis): aggiungi feature temporali al prompt con 2 esempi few-shot" ``` --- ## Task 7: Smoke run end-to-end **Files:** - Nessuna modifica al codice. Validazione che il loop intero giri con la grammatica estesa: carica OHLCV, genera 4 genomi, compila, backtesta, valuta DSR, applica Adversarial, persiste. - [ ] **Step 7.1: Run smoke script** Run: `uv run python -m scripts.smoke_run` Expected: completamento senza eccezioni, output finale contenente `Smoke run completed`. - [ ] **Step 7.2: Inspect at least one generated genome for temporal feature usage** Run: ```bash LATEST=$(sqlite3 runs.db "SELECT id FROM runs WHERE name LIKE 'smoke%' ORDER BY started_at DESC LIMIT 1;") sqlite3 runs.db "SELECT genome_id, substr(raw_text, 1, 600) FROM evaluations WHERE run_id='$LATEST' LIMIT 4;" ``` Expected output: 4 righe raw_text JSON. Almeno 1 dovrebbe contenere `"name": "hour"`, `"name": "dow"`, `"name": "is_weekend"`, o `"name": "minute_of_hour"`. Se 0/4 usano feature temporali, il prompt non è abbastanza eloquente — apri un follow-up per iterare il prompt (non bloccante per questa PR). - [ ] **Step 7.3: Push branch + open PR** ```bash git log --oneline -8 # verifica 6 commit dei Task 1-6 git push origin HEAD ``` Aprire PR con titolo `feat: feature temporali nella grammatica Hypothesis` referenziando lo spec. --- ## Self-review notes (autore del piano) - Tutti i 7 hard requirement dello spec (`grammar`, `compiler`, `prompt`, 4 feature, integration test, smoke, backward compat) sono coperti dai Task 1-7. - Nessun placeholder `TBD`/`TODO`. - Tipi consistenti: `_TIME_FEATURE_FNS` definito una volta in Task 2 e referenziato implicitamente dai tester nei Task 3-5 senza bisogno di re-definizione. - Test pre-esistenti non vengono toccati; il Task 5 include `pytest` sull'intera suite del protocollo come regression check. - Backward compat: `KNOWN_FEATURES` cresce, il branch OHLCV resta invariato → genomi vecchi restano validi senza migrazione DB.