feat(agents): hypothesis agent with prompt template + s-expr extraction
Aggiunge HypothesisAgent che invoca LLMClient con system/user template parametrizzati sul genoma e sul MarketSummary, poi estrae la S-expression (da fence markdown lisp/scheme/sexp o testo nudo), la parsa e la valida. Restituisce HypothesisProposal con strategy=None + parse_error in caso di output malformato, mantenendo sempre il CompletionResult per accounting. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,141 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ..genome.hypothesis import HypothesisAgentGenome
|
||||
from ..llm.client import CompletionResult, LLMClient
|
||||
from ..protocol.parser import ParseError, Strategy, parse_strategy
|
||||
from ..protocol.validator import ValidationError, validate_strategy
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MarketSummary:
|
||||
symbol: str
|
||||
timeframe: str
|
||||
n_bars: int
|
||||
return_mean: float
|
||||
return_std: float
|
||||
skew: float
|
||||
kurtosis: float
|
||||
volatility_regime: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HypothesisProposal:
|
||||
strategy: Strategy | None
|
||||
raw_text: str
|
||||
completion: CompletionResult
|
||||
parse_error: str | None = None
|
||||
|
||||
|
||||
SYSTEM_TEMPLATE = """\
|
||||
Sei un agente generatore di ipotesi di trading quantitativo per un sistema swarm.
|
||||
|
||||
Il tuo stile cognitivo: {cognitive_style}
|
||||
Direttiva personale: {system_prompt}
|
||||
|
||||
Devi proporre una strategia di trading espressa nel linguaggio S-expression
|
||||
con i seguenti verbi disponibili:
|
||||
|
||||
Azioni: entry-long, entry-short, exit, flat
|
||||
Logici: and, or, not
|
||||
Comparatori: gt, lt, eq
|
||||
Dati: feature, indicator, crossover, crossunder
|
||||
|
||||
Indicatori disponibili: sma <length>, rsi <length>, atr <length>, macd, realized_vol <window>.
|
||||
Feature disponibili: open, high, low, close, volume.
|
||||
|
||||
Le regole sono valutate in ordine; la prima che matcha vince per ogni timestamp.
|
||||
La default action se nessuna regola matcha è 'flat'.
|
||||
|
||||
Rispondi SOLO con la S-expression in un fence ```lisp ... ```, senza prosa,
|
||||
senza spiegazioni. Esempio formato:
|
||||
|
||||
```lisp
|
||||
(strategy
|
||||
(when (gt (indicator rsi 14) 70.0) (entry-short))
|
||||
(when (lt (indicator rsi 14) 30.0) (entry-long)))
|
||||
```
|
||||
"""
|
||||
|
||||
|
||||
USER_TEMPLATE = """\
|
||||
Mercato: {symbol} timeframe {timeframe}, {n_bars} barre osservate.
|
||||
Statistiche return: mean={return_mean:.5f}, std={return_std:.5f}, \
|
||||
skew={skew:.3f}, kurt={kurtosis:.3f}.
|
||||
Regime volatilità: {volatility_regime}.
|
||||
|
||||
Feature accessibili dal tuo genoma: {feature_access}.
|
||||
Lookback massimo che puoi usare nel ragionamento: {lookback_window} barre.
|
||||
|
||||
Genera una strategia che cerchi anomalie sfruttabili in questo regime.
|
||||
"""
|
||||
|
||||
|
||||
_SEXP_FENCE_RE = re.compile(
|
||||
r"```(?:lisp|scheme|sexp)?\s*(\(strategy[\s\S]*?\))\s*```",
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
|
||||
def _extract_sexp(text: str) -> str | None:
|
||||
m = _SEXP_FENCE_RE.search(text)
|
||||
if m:
|
||||
return m.group(1)
|
||||
if text.strip().startswith("(strategy"):
|
||||
return text.strip()
|
||||
return None
|
||||
|
||||
|
||||
class HypothesisAgent:
|
||||
def __init__(self, llm: LLMClient):
|
||||
self._llm = llm
|
||||
|
||||
def propose(
|
||||
self,
|
||||
genome: HypothesisAgentGenome,
|
||||
market: MarketSummary,
|
||||
) -> HypothesisProposal:
|
||||
system = SYSTEM_TEMPLATE.format(
|
||||
cognitive_style=genome.cognitive_style,
|
||||
system_prompt=genome.system_prompt,
|
||||
)
|
||||
user = USER_TEMPLATE.format(
|
||||
symbol=market.symbol,
|
||||
timeframe=market.timeframe,
|
||||
n_bars=market.n_bars,
|
||||
return_mean=market.return_mean,
|
||||
return_std=market.return_std,
|
||||
skew=market.skew,
|
||||
kurtosis=market.kurtosis,
|
||||
volatility_regime=market.volatility_regime,
|
||||
feature_access=", ".join(genome.feature_access),
|
||||
lookback_window=genome.lookback_window,
|
||||
)
|
||||
|
||||
completion = self._llm.complete(genome, system=system, user=user)
|
||||
|
||||
sexp = _extract_sexp(completion.text)
|
||||
if sexp is None:
|
||||
return HypothesisProposal(
|
||||
strategy=None,
|
||||
raw_text=completion.text,
|
||||
completion=completion,
|
||||
parse_error="no s-expression found in output",
|
||||
)
|
||||
try:
|
||||
ast = parse_strategy(sexp)
|
||||
validate_strategy(ast)
|
||||
return HypothesisProposal(
|
||||
strategy=ast,
|
||||
raw_text=completion.text,
|
||||
completion=completion,
|
||||
)
|
||||
except (ParseError, ValidationError) as e:
|
||||
return HypothesisProposal(
|
||||
strategy=None,
|
||||
raw_text=completion.text,
|
||||
completion=completion,
|
||||
parse_error=str(e),
|
||||
)
|
||||
@@ -0,0 +1,93 @@
|
||||
from multi_swarm.agents.hypothesis import HypothesisAgent, MarketSummary
|
||||
from multi_swarm.genome.hypothesis import HypothesisAgentGenome, ModelTier
|
||||
from multi_swarm.llm.client import CompletionResult
|
||||
|
||||
|
||||
def make_summary() -> MarketSummary:
|
||||
return MarketSummary(
|
||||
symbol="BTC/USDT",
|
||||
timeframe="1h",
|
||||
n_bars=1000,
|
||||
return_mean=0.0001,
|
||||
return_std=0.01,
|
||||
skew=0.1,
|
||||
kurtosis=3.5,
|
||||
volatility_regime="high",
|
||||
)
|
||||
|
||||
|
||||
def test_hypothesis_agent_calls_llm_and_parses(mocker): # type: ignore[no-untyped-def]
|
||||
fake_llm = mocker.MagicMock()
|
||||
fake_llm.complete.return_value = CompletionResult(
|
||||
text="(strategy (when (gt (indicator rsi 14) 70.0) (entry-short)))",
|
||||
input_tokens=200,
|
||||
output_tokens=80,
|
||||
tier=ModelTier.C,
|
||||
model="qwen",
|
||||
)
|
||||
g = HypothesisAgentGenome(
|
||||
system_prompt="Pensa come un fisico.",
|
||||
feature_access=["close"],
|
||||
temperature=0.9,
|
||||
top_p=0.95,
|
||||
model_tier=ModelTier.C,
|
||||
lookback_window=200,
|
||||
cognitive_style="physicist",
|
||||
)
|
||||
agent = HypothesisAgent(llm=fake_llm)
|
||||
proposal = agent.propose(g, make_summary())
|
||||
assert proposal.strategy is not None
|
||||
assert proposal.raw_text.startswith("(strategy")
|
||||
assert proposal.completion.input_tokens == 200
|
||||
fake_llm.complete.assert_called_once()
|
||||
|
||||
|
||||
def test_hypothesis_agent_returns_none_on_parse_error(mocker): # type: ignore[no-untyped-def]
|
||||
fake_llm = mocker.MagicMock()
|
||||
fake_llm.complete.return_value = CompletionResult(
|
||||
text="this is not s-expression",
|
||||
input_tokens=200,
|
||||
output_tokens=80,
|
||||
tier=ModelTier.C,
|
||||
model="qwen",
|
||||
)
|
||||
g = HypothesisAgentGenome(
|
||||
system_prompt="x",
|
||||
feature_access=["close"],
|
||||
temperature=0.9,
|
||||
top_p=0.95,
|
||||
model_tier=ModelTier.C,
|
||||
lookback_window=200,
|
||||
cognitive_style="physicist",
|
||||
)
|
||||
agent = HypothesisAgent(llm=fake_llm)
|
||||
proposal = agent.propose(g, make_summary())
|
||||
assert proposal.strategy is None
|
||||
assert proposal.parse_error is not None
|
||||
|
||||
|
||||
def test_hypothesis_agent_extracts_sexp_from_markdown_fence(mocker): # type: ignore[no-untyped-def]
|
||||
fake_llm = mocker.MagicMock()
|
||||
fake_llm.complete.return_value = CompletionResult(
|
||||
text=(
|
||||
"Ecco la strategia:\n```lisp\n"
|
||||
"(strategy (when (lt (indicator rsi 14) 30.0) (entry-long)))\n"
|
||||
"```\nFatta."
|
||||
),
|
||||
input_tokens=200,
|
||||
output_tokens=80,
|
||||
tier=ModelTier.C,
|
||||
model="qwen",
|
||||
)
|
||||
g = HypothesisAgentGenome(
|
||||
system_prompt="x",
|
||||
feature_access=["close"],
|
||||
temperature=0.9,
|
||||
top_p=0.95,
|
||||
model_tier=ModelTier.C,
|
||||
lookback_window=200,
|
||||
cognitive_style="physicist",
|
||||
)
|
||||
agent = HypothesisAgent(llm=fake_llm)
|
||||
proposal = agent.propose(g, make_summary())
|
||||
assert proposal.strategy is not None
|
||||
Reference in New Issue
Block a user