Adriano
44eb6436c1
Sostituisce la grammatica S-expression con uno schema JSON stretto. La grammatica S-expression falliva il parsing nel 64% delle generazioni del modello Qwen3-235B sul run reale; JSON e' nativo per gli LLM moderni e si parsa con json.loads. Cambiamenti principali: - grammar.py: costanti rinominate LOGICAL_OPS / COMPARATOR_OPS / CROSSOVER_OPS / ACTION_VALUES / KIND_VALUES. - parser.py: nuovo AST a dataclass tipizzato (OpNode, IndicatorNode, FeatureNode, LiteralNode, Rule, Strategy); parse_strategy ora consuma JSON tramite json.loads. - validator.py: walk dispatchato per tipo (isinstance) invece di pattern-matching su 'kind'; arity check su operatori e indicator. - compiler.py: traversal del nuovo AST tipizzato, dispatch per isinstance; logica indicator/feature/literal invariata. - hypothesis.py: prompt SYSTEM riscritto con esempi JSON e vincoli espliciti su no-nesting; estrazione via fence ```json``` + fallback brace-balanced. - __init__.py: re-export pubblico delle entita' del protocollo. - Tutti i test (parser, validator, compiler, hypothesis_agent, falsification, adversarial, e2e, smoke_run) migrati a JSON. - Rimossa dipendenza sexpdata da pyproject.toml + uv.lock. Test: 135 passed (era 122; aggiunti casi parser/validator). ruff + mypy strict clean. Smoke run end-to-end OK. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
import json
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from multi_swarm.agents.adversarial import AdversarialAgent, AdversarialReport, Severity
|
|
from multi_swarm.protocol.parser import parse_strategy
|
|
|
|
|
|
@pytest.fixture
|
|
def ohlcv() -> pd.DataFrame:
|
|
idx = pd.date_range("2024-01-01", periods=500, freq="1h", tz="UTC")
|
|
close = 100 + np.cumsum(np.random.RandomState(0).normal(0.0, 1.0, 500))
|
|
return pd.DataFrame(
|
|
{
|
|
"open": close,
|
|
"high": close + 0.5,
|
|
"low": close - 0.5,
|
|
"close": close,
|
|
"volume": 1.0,
|
|
},
|
|
index=idx,
|
|
)
|
|
|
|
|
|
def test_degenerate_always_long_flagged(ohlcv: pd.DataFrame) -> None:
|
|
src = json.dumps(
|
|
{
|
|
"rules": [
|
|
{
|
|
"condition": {
|
|
"op": "gt",
|
|
"args": [
|
|
{"kind": "feature", "name": "close"},
|
|
{"kind": "literal", "value": -1e9},
|
|
],
|
|
},
|
|
"action": "entry-long",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
assert isinstance(report, AdversarialReport)
|
|
assert any(f.name == "degenerate" and f.severity == Severity.HIGH for f in report.findings)
|
|
|
|
|
|
def test_no_findings_on_reasonable_strategy(ohlcv: pd.DataFrame) -> None:
|
|
src = json.dumps(
|
|
{
|
|
"rules": [
|
|
{
|
|
"condition": {
|
|
"op": "gt",
|
|
"args": [
|
|
{"kind": "indicator", "name": "rsi", "params": [14]},
|
|
{"kind": "literal", "value": 70.0},
|
|
],
|
|
},
|
|
"action": "entry-short",
|
|
},
|
|
{
|
|
"condition": {
|
|
"op": "lt",
|
|
"args": [
|
|
{"kind": "indicator", "name": "rsi", "params": [14]},
|
|
{"kind": "literal", "value": 30.0},
|
|
],
|
|
},
|
|
"action": "entry-long",
|
|
},
|
|
]
|
|
}
|
|
)
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
high_findings = [f for f in report.findings if f.severity == Severity.HIGH]
|
|
assert len(high_findings) == 0
|
|
|
|
|
|
def test_zero_trade_strategy_flagged(ohlcv: pd.DataFrame) -> None:
|
|
src = json.dumps(
|
|
{
|
|
"rules": [
|
|
{
|
|
"condition": {
|
|
"op": "gt",
|
|
"args": [
|
|
{"kind": "feature", "name": "close"},
|
|
{"kind": "literal", "value": 1e9},
|
|
],
|
|
},
|
|
"action": "entry-long",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
assert any(f.name == "no_trades" for f in report.findings)
|