Multi_Swarm_Coevolutive/docs/superpowers/plans/2026-05-11-temporal-features.md

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:

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:

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

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:

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):

_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:

def _eval_node(node: Node, df: pd.DataFrame) -> pd.Series:
    if isinstance(node, FeatureNode):
        return df[node.name]

With:

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

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:

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:

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

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:

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

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:

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

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:

Leaf - feature OHLCV:
  {{"kind": "feature", "name": "open|high|low|close|volume"}}

con:

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

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:

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

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.