diff --git a/README.md b/README.md index de393a2..9ba6dd2 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ [![CI](https://github.com/OnePunchMonk/AgentQuant/actions/workflows/ci.yml/badge.svg)](https://github.com/OnePunchMonk/AgentQuant/actions) ![Python](https://img.shields.io/badge/python-3.10%2B-blue) -![Tests](https://img.shields.io/badge/tests-55%20passed-brightgreen) +![Tests](https://img.shields.io/badge/tests-63%20passed-brightgreen) --- @@ -18,6 +18,8 @@ AgentQuant is a regime-adaptive research platform that runs a real **ReAct agent 4. **Reflects** on results and retries if Sharpe is below the configured threshold (up to `max_iterations` times). 5. **Stores** the best result to SQLite memory so future runs can recall what worked in similar regimes. +Every completed run now emits a screenshot-friendly **regime card** and a transparent candidate table with pass/watch/reject verdicts, Sharpe, Calmar, Sortino, max drawdown, and bootstrapped Sharpe p5. + --- ## Platform Preview @@ -54,6 +56,20 @@ analyze ──► hypothesize ──► backtest ──► reflect store → SQLite memory ``` +### Multi-Agent Swarm + +The optional swarm mode runs the same research loop through specialized agents: + +```mermaid +flowchart LR + M["Memory Agent
learned patterns"] --> R["Regime Analyst
market context"] + R --> S["Strategy Specialists
momentum, mean reversion, volatility"] + S --> C["Critic Agent
reject invalid or duplicate candidates"] + C --> B["Backtest Coordinator
multi-window validation"] + B --> M + B --> O["Regime card + comparison table"] +``` + ### Key Components | Module | What it does | @@ -63,7 +79,11 @@ analyze ──► hypothesize ──► backtest ──► reflect | `src/agent/base_planner.py` | `BasePlanner` ABC with Gemini / OpenAI / Fallback | | `src/agent/context_builder.py` | `RegimeContext` dataclass with VIX percentile, multi-horizon momentum | | `src/agent/parameter_grid.py` | Canonical grids per strategy; regime-aware prior selection | +| `src/agent/memory_layer.py` | Agentic memory layer that turns SQLite history into strategy patterns | +| `src/agent/reporting.py` | Regime card, comparison table, and pass/watch/reject verdicts | +| `src/agent/trace.py` | Live trace event stream for the ReAct loop | | `src/agent/strategy_memory.py` | SQLite cross-session memory | +| `src/agent/swarm/` | Memory Agent, Regime Analyst, Specialists, Critic, and Backtest Coordinator | | `src/research/alpha_store.py` | SQLite memory for accepted, watchlisted, and rejected alpha candidates | | `src/research/nla_memory.py` | Explicit NLA-style narrative memory and `nla-gemma4` JSONL ingestion | | `src/research/workspace.py` | Experiment registry, robustness summaries, and research memo generation | @@ -79,23 +99,36 @@ analyze ──► hypothesize ──► backtest ──► reflect --- -### Experimental Agent Swarm Branch +### Visible Agent Loop + +Run with a live terminal trace to watch the agent move through hypothesis, backtest, reflection, retry, and memory storage: + +```bash +agentquant run --ticker SPY --trace +``` + +Run the multi-agent architecture from main: -The `agent-swarm-method` branch explores a multi-agent version of AgentQuant. It decomposes the research loop into specialized agents: +```bash +agentquant run --ticker SPY --swarm --strategies momentum mean_reversion volatility +``` -- **Memory Agent** retrieves and stores strategy patterns across runs. -- **Regime Analyst** builds the market/macro context used by downstream agents. -- **Strategy Specialists** generate proposals for momentum, mean reversion, volatility, and trend-following approaches. -- **Critic Agent** pre-screens proposals before expensive backtests. -- **Backtest Coordinator** validates approved proposals across multiple time windows and ranks by robustness. +Browse accumulated strategy memory: -That branch is intentionally experimental and sits alongside the main ReAct pipeline. To inspect it: +```bash +agentquant memory +agentquant memory --regime LowVol-Bull --patterns +agentquant memory --export markdown +``` + +Render the latest stored one-page regime card: ```bash -git checkout agent-swarm-method -pytest tests/test_swarm.py -v +agentquant regime-card ``` +The Colab quick demo is in `notebooks/agentquant_colab_spy.ipynb`. It runs a full SPY loop in three cells and works with or without a Gemini API key. + --- ## Quick Start @@ -120,7 +153,13 @@ cp .env.example .env # 5. Run the agent python -m src.agent.runner -# 6. Run the dashboard +# Or use the CLI +agentquant run --ticker SPY --trace + +# 6. Browse memory +agentquant memory --patterns + +# 7. Run the dashboard python run_app.py ``` @@ -135,7 +174,7 @@ pip install -e ".[dev]" pytest tests/ -v ``` -**55 tests passing** across: +**63 tests passing** across: - `test_config.py` — Pydantic validation - `test_data_ingest.py` — live ticker fetch and cache range coverage - `test_metrics.py` — Sharpe, drawdown, Calmar, Sortino @@ -147,6 +186,9 @@ pytest tests/ -v - `test_alpha_store.py` — alpha memory persistence and retrieval - `test_nla_memory.py` — explicit NLA memory and JSONL ingestion - `test_research_workspace.py` — experiment registry summaries and memos +- `test_memory_layer.py` — agentic memory pattern extraction and markdown export +- `test_reporting_cli.py` — regime card, verdicts, and CLI parsing +- `test_swarm.py` — synthetic-data smoke tests for the multi-agent swarm --- @@ -159,9 +201,13 @@ AgentQuant/ │ │ ├── agent_graph.py # ReAct agent loop (analyze→hypothesize→backtest→reflect→store) │ │ ├── base_planner.py # LLM abstraction: Gemini / OpenAI / Fallback │ │ ├── context_builder.py # RegimeContext dataclass + builder +│ │ ├── memory_layer.py # Agentic memory pattern extraction │ │ ├── parameter_grid.py # Canonical parameter grids per strategy │ │ ├── proposal_generator.py # LLM → Grid → Random fallback chain +│ │ ├── reporting.py # Regime card + comparison table renderers │ │ ├── strategy_memory.py # SQLite cross-session memory +│ │ ├── swarm/ # Multi-agent Memory/Regime/Critic/Backtest agents +│ │ ├── trace.py # Live trace events │ │ ├── tools.py # Tool-calling interface for LangGraph │ │ └── runner.py # Main entry point │ ├── data/ @@ -193,7 +239,7 @@ AgentQuant/ ├── experiments/ │ ├── results_store.py # SQLite experiment tracking │ └── walk_forward.py # Walk-forward validation -├── tests/ # 55 tests +├── tests/ # 63 tests ├── docs/ # Documentation ├── config.yaml # Project configuration ├── .env.example # Environment template diff --git a/notebooks/agentquant_colab_spy.ipynb b/notebooks/agentquant_colab_spy.ipynb new file mode 100644 index 0000000..f98924c --- /dev/null +++ b/notebooks/agentquant_colab_spy.ipynb @@ -0,0 +1,72 @@ +{ + "nbformat": 4, + "nbformat_minor": 5, + "metadata": { + "colab": { + "name": "AgentQuant SPY demo" + }, + "kernelspec": { + "name": "python3", + "display_name": "Python 3" + }, + "language_info": { + "name": "python" + } + }, + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# AgentQuant SPY demo\n", + "\n", + "Three-cell Colab path: install AgentQuant, optionally set a Gemini key, and run a full SPY agent loop with a trace, regime card, and candidate comparison table." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "!pip -q install \"git+https://github.com/OnePunchMonk/agentquant.git\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import os\n", + "from getpass import getpass\n", + "\n", + "key = getpass(\"Gemini API key (leave blank to use grid-search fallback): \")\n", + "if key:\n", + " os.environ[\"GOOGLE_API_KEY\"] = key\n", + "else:\n", + " os.environ.pop(\"GOOGLE_API_KEY\", None)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from src.agent.agent_graph import run_agent\n", + "from src.agent.reporting import render_comparison_table, render_regime_card\n", + "from src.agent.trace import TraceRecorder\n", + "from src.data.ingest import fetch_ohlcv_data\n", + "\n", + "data = fetch_ohlcv_data(ticker=\"SPY\")\n", + "data.update(fetch_ohlcv_data(ticker=\"^VIX\"))\n", + "state = run_agent(data, strategy_type=\"momentum\", asset=\"SPY\", max_iterations=2, trace=TraceRecorder(live=True))\n", + "\n", + "print(render_regime_card(state))\n", + "print()\n", + "print(render_comparison_table(state.get(\"results\", [])))" + ] + } + ] +} diff --git a/pyproject.toml b/pyproject.toml index 38cfc76..75dd4e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ dependencies = [ "matplotlib>=3.8", "pyarrow>=16.0", "tabulate>=0.9", + "rich>=13.7", "statsmodels>=0.14", "requests>=2.31", ] @@ -54,6 +55,7 @@ dev = [ [project.scripts] run-agent = "src.agent.runner:main" +agentquant = "src.cli:main" [tool.pytest.ini_options] pythonpath = ["."] @@ -63,6 +65,8 @@ addopts = "-v --tb=short" [tool.ruff] line-length = 100 target-version = "py310" + +[tool.ruff.lint] select = ["E", "F", "W", "I"] ignore = ["E501"] diff --git a/src/agent/agent_graph.py b/src/agent/agent_graph.py index 0ee13b8..7590b28 100644 --- a/src/agent/agent_graph.py +++ b/src/agent/agent_graph.py @@ -18,6 +18,7 @@ from src.agent.context_builder import RegimeContext, build_context from src.agent.proposal_generator import Proposal, ProposalGenerator from src.agent.strategy_memory import PastResult, StrategyMemory +from src.agent.trace import TraceRecorder, emit_trace from src.research.alpha_store import AlphaStore from src.research.nla_memory import NLAMemoryStore from src.utils.config import config @@ -40,6 +41,7 @@ class AgentState(TypedDict, total=False): should_continue: bool memory_context: str run_log: List[str] + trace: Optional[TraceRecorder] def analyze_node(state: AgentState) -> AgentState: @@ -71,6 +73,13 @@ def analyze_node(state: AgentState) -> AgentState: state["memory_context"] = f"{memory_ctx}\n\n{alpha_ctx}\n\n{nla_ctx}" state["run_log"] = state.get("run_log", []) state["run_log"].append(f"Regime: {regime_label} (confidence: {context.regime_confidence:.0%})") + emit_trace( + state.get("trace"), + "analyze", + f"Regime {regime_label} detected at {context.regime_confidence:.0%} confidence.", + regime=regime_label, + confidence=context.regime_confidence, + ) logger.info("Regime: %s, Confidence: %.0f%%", regime_label, context.regime_confidence * 100) return state @@ -97,6 +106,14 @@ def hypothesize_node(state: AgentState) -> AgentState: f"Iteration {iteration}: Generated {len(proposals)} proposals " f"(methods: {[p.generation_method for p in proposals]})" ) + emit_trace( + state.get("trace"), + "hypothesize", + f"Iteration {iteration}: generated {len(proposals)} candidate strategies.", + iteration=iteration, + methods=[p.generation_method for p in proposals], + proposals=[p.params for p in proposals], + ) for i, p in enumerate(proposals): logger.info(" Proposal %d: %s (confidence=%.2f, method=%s)", @@ -121,8 +138,12 @@ def backtest_node(state: AgentState) -> AgentState: metrics = bt_result["metrics"] results.append({ "proposal_idx": i, + "strategy_type": strategy_type, "params": proposal.params, "sharpe": metrics.get("sharpe_ratio", 0.0), + "calmar": metrics.get("calmar", 0.0), + "sortino": metrics.get("sortino", 0.0), + "bootstrap_sharpe_p5": metrics.get("bootstrap_sharpe_p5", 0.0), "total_return": metrics.get("total_return", 0.0), "max_drawdown": metrics.get("max_drawdown", 0.0), "num_trades": metrics.get("num_trades", 0), @@ -144,11 +165,22 @@ def backtest_node(state: AgentState) -> AgentState: f"Best: Sharpe={best['sharpe']:.2f}, Return={best['total_return']:.1%}, " f"Params={best['params']}" ) + emit_trace( + state.get("trace"), + "backtest", + ( + f"Best candidate Sharpe={best['sharpe']:.2f}, " + f"Calmar={best.get('calmar', 0):.2f}, p5={best.get('bootstrap_sharpe_p5', 0):.2f}." + ), + best=best, + results=results, + ) logger.info("Best result: Sharpe=%.2f, Return=%.1f%%, Params=%s", best["sharpe"], best["total_return"] * 100, best["params"]) else: state["best_result"] = None state["run_log"].append("No valid backtest results.") + emit_trace(state.get("trace"), "backtest", "No valid candidate backtests completed.") logger.warning("No valid backtest results produced.") return state @@ -166,6 +198,7 @@ def reflect_node(state: AgentState) -> AgentState: if best is None: state["should_continue"] = iteration < max_iter state["run_log"].append(f"Reflect: No results. {'Retrying...' if state['should_continue'] else 'Stopping.'}") + emit_trace(state.get("trace"), "reflect", state["run_log"][-1]) return state sharpe = best.get("sharpe", 0.0) @@ -175,18 +208,21 @@ def reflect_node(state: AgentState) -> AgentState: state["run_log"].append( f"Reflect: Sharpe {sharpe:.2f} >= threshold {min_sharpe:.2f}. ACCEPTING." ) + emit_trace(state.get("trace"), "reflect", state["run_log"][-1], accepted=True) logger.info("Result accepted: Sharpe %.2f >= %.2f", sharpe, min_sharpe) elif iteration >= max_iter: state["should_continue"] = False state["run_log"].append( f"Reflect: Sharpe {sharpe:.2f} < {min_sharpe:.2f} but max iterations reached. Accepting best available." ) + emit_trace(state.get("trace"), "reflect", state["run_log"][-1], accepted=True) logger.info("Max iterations reached. Accepting best: Sharpe %.2f", sharpe) else: state["should_continue"] = True state["run_log"].append( f"Reflect: Sharpe {sharpe:.2f} < {min_sharpe:.2f}. Retrying (iteration {iteration}/{max_iter})." ) + emit_trace(state.get("trace"), "reflect", state["run_log"][-1], accepted=False) logger.info("Result below threshold. Will retry. (iteration %d/%d)", iteration, max_iter) return state @@ -199,6 +235,7 @@ def store_node(state: AgentState) -> AgentState: best = state.get("best_result") if best is None: state["run_log"].append("Store: Nothing to persist.") + emit_trace(state.get("trace"), "store", "No accepted result to persist.") return state context = state.get("context") @@ -223,6 +260,9 @@ def store_node(state: AgentState) -> AgentState: params=best["params"], metrics={ "sharpe_ratio": best.get("sharpe", 0.0), + "calmar": best.get("calmar", 0.0), + "sortino": best.get("sortino", 0.0), + "bootstrap_sharpe_p5": best.get("bootstrap_sharpe_p5", 0.0), "total_return": best.get("total_return", 0.0), "max_drawdown": best.get("max_drawdown", 0.0), "num_trades": best.get("num_trades", 0), @@ -239,6 +279,9 @@ def store_node(state: AgentState) -> AgentState: params=best["params"], metrics={ "sharpe_ratio": best.get("sharpe", 0.0), + "calmar": best.get("calmar", 0.0), + "sortino": best.get("sortino", 0.0), + "bootstrap_sharpe_p5": best.get("bootstrap_sharpe_p5", 0.0), "total_return": best.get("total_return", 0.0), "max_drawdown": best.get("max_drawdown", 0.0), "num_trades": best.get("num_trades", 0), @@ -250,6 +293,7 @@ def store_node(state: AgentState) -> AgentState: state["run_log"].append( f"Store: Persisted result {run_id}, alpha {alpha.alpha_id}, NLA note {nla.record_id}." ) + emit_trace(state.get("trace"), "store", state["run_log"][-1], run_id=run_id) logger.info("Persisted result %s, alpha %s, NLA note %s.", run_id, alpha.alpha_id, nla.record_id) return state @@ -259,6 +303,7 @@ def run_agent( strategy_type: str = "momentum", asset: str = None, max_iterations: int = None, + trace: Optional[TraceRecorder] = None, ) -> AgentState: """ Run the full agent loop: analyze → hypothesize → backtest → reflect → (loop or store). @@ -280,6 +325,7 @@ def run_agent( "should_continue": True, "memory_context": "", "run_log": [], + "trace": trace, } # Step 1: Analyze (once) diff --git a/src/agent/context_builder.py b/src/agent/context_builder.py index acc3308..16688ce 100644 --- a/src/agent/context_builder.py +++ b/src/agent/context_builder.py @@ -34,6 +34,7 @@ class RegimeContext: rsi_14: float = 50.0 price_vs_sma200: float = 0.0 regime_confidence: float = 0.5 + memory_context: str = "" alpha_memory_context: str = "" nla_memory_context: str = "" @@ -55,6 +56,8 @@ def to_prompt_string(self) -> str: f" RSI (14): {self.rsi_14:.1f}\n" f" Drawdown from peak: {self.drawdown_from_peak * 100:.1f}%\n" ) + if self.memory_context: + context += f"\n{self.memory_context}\n" if self.alpha_memory_context: context += f"\n{self.alpha_memory_context}\n" if self.nla_memory_context: diff --git a/src/agent/memory_layer.py b/src/agent/memory_layer.py new file mode 100644 index 0000000..308756c --- /dev/null +++ b/src/agent/memory_layer.py @@ -0,0 +1,203 @@ +""" +Agentic memory layer for cross-run strategy learning. + +This module turns persisted strategy rows into operational patterns that can be +shown in the CLI, injected into prompts, and used by the swarm memory agent. +""" + +import json +from collections import defaultdict +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from src.agent.strategy_memory import PastResult, StrategyMemory + +PATTERN_MIN_SAMPLES = 2 +GOOD_SHARPE = 0.5 +BAD_SHARPE = 0.0 + + +@dataclass +class StrategyPattern: + """A compact statement about what memory has learned.""" + + scope: str + verdict: str + evidence: str + sample_size: int + avg_sharpe: float + best_sharpe: float + params: Dict[str, Any] = field(default_factory=dict) + + def to_sentence(self) -> str: + param_text = f" params={self.params}" if self.params else "" + return ( + f"{self.verdict}: {self.scope} | n={self.sample_size}, " + f"avg_sharpe={self.avg_sharpe:.2f}, best={self.best_sharpe:.2f} | " + f"{self.evidence}{param_text}" + ) + + +class AgenticMemoryLayer: + """Pattern extractor and browser over StrategyMemory.""" + + def __init__(self, memory: Optional[StrategyMemory] = None): + self.memory = memory or StrategyMemory() + + def recent_runs( + self, + regime: str = "", + strategy_type: str = "", + limit: int = 25, + ) -> List[PastResult]: + return self.memory.list_runs(regime=regime, strategy_type=strategy_type, limit=limit) + + def summary_rows( + self, + regime: str = "", + strategy_type: str = "", + limit: int = 500, + ) -> List[Dict[str, Any]]: + return self.memory.summarize(regime=regime, strategy_type=strategy_type, limit=limit) + + def extract_patterns( + self, + regime: str = "", + strategy_types: Optional[List[str]] = None, + limit: int = 200, + ) -> List[StrategyPattern]: + if strategy_types: + runs: List[PastResult] = [] + for strategy in strategy_types: + runs.extend( + self.memory.list_runs( + regime=regime, + strategy_type=strategy, + limit=limit, + order_by="timestamp", + ) + ) + else: + runs = self.memory.list_runs(regime=regime, limit=limit) + + patterns: List[StrategyPattern] = [] + by_scope: Dict[tuple, List[PastResult]] = defaultdict(list) + for run in runs: + by_scope[(run.regime, run.strategy_type)].append(run) + + for (scope_regime, strategy), history in sorted(by_scope.items()): + if not history: + continue + sharpes = [h.sharpe for h in history] + avg_sharpe = sum(sharpes) / len(sharpes) + best = max(history, key=lambda h: h.sharpe) + worst = min(history, key=lambda h: h.sharpe) + verdict = self._verdict(avg_sharpe, best.sharpe, worst.sharpe) + patterns.append( + StrategyPattern( + scope=f"{strategy} in {scope_regime}", + verdict=verdict, + evidence=( + f"{sum(1 for h in history if h.sharpe >= GOOD_SHARPE)} good runs, " + f"{sum(1 for h in history if h.sharpe <= BAD_SHARPE)} rejected-or-poor runs" + ), + sample_size=len(history), + avg_sharpe=avg_sharpe, + best_sharpe=best.sharpe, + params=self._decode_params(best.params), + ) + ) + patterns.extend(self._parameter_patterns(scope_regime, strategy, history)) + + return patterns + + def to_prompt_context( + self, + regime: str, + strategy_types: Optional[List[str]] = None, + limit: int = 200, + ) -> str: + patterns = self.extract_patterns(regime=regime, strategy_types=strategy_types, limit=limit) + if not patterns: + return f"AGENTIC MEMORY: no learned patterns yet for {regime}." + lines = [f"AGENTIC MEMORY PATTERNS FOR {regime}:"] + lines.extend(f"- {pattern.to_sentence()}" for pattern in patterns[:12]) + return "\n".join(lines) + + def render_markdown( + self, + regime: str = "", + strategy_type: str = "", + limit: int = 25, + ) -> str: + runs = self.recent_runs(regime=regime, strategy_type=strategy_type, limit=limit) + return self.memory.export_markdown(runs) + + def _parameter_patterns( + self, + regime: str, + strategy: str, + history: List[PastResult], + ) -> List[StrategyPattern]: + if len(history) < PATTERN_MIN_SAMPLES: + return [] + + buckets: Dict[str, List[PastResult]] = defaultdict(list) + for run in history: + params = self._decode_params(run.params) + bucket = self._bucket_params(strategy, params) + if bucket: + buckets[bucket].append(run) + + patterns = [] + for bucket, bucket_runs in buckets.items(): + if len(bucket_runs) < PATTERN_MIN_SAMPLES: + continue + sharpes = [run.sharpe for run in bucket_runs] + avg_sharpe = sum(sharpes) / len(sharpes) + best = max(bucket_runs, key=lambda h: h.sharpe) + patterns.append( + StrategyPattern( + scope=f"{strategy} {bucket} in {regime}", + verdict=self._verdict(avg_sharpe, best.sharpe, min(sharpes)), + evidence="parameter cluster repeated across stored runs", + sample_size=len(bucket_runs), + avg_sharpe=avg_sharpe, + best_sharpe=best.sharpe, + params=self._decode_params(best.params), + ) + ) + return patterns + + @staticmethod + def _decode_params(raw: str) -> Dict[str, Any]: + try: + decoded = json.loads(raw) + return decoded if isinstance(decoded, dict) else {} + except Exception: + return {} + + @staticmethod + def _bucket_params(strategy: str, params: Dict[str, Any]) -> str: + if strategy == "momentum" and "slow_window" in params: + slow = int(params["slow_window"]) + start = (slow // 30) * 30 + return f"slow_window={start}-{start + 29}" + if strategy == "mean_reversion" and "window" in params: + window = int(params["window"]) + start = (window // 10) * 10 + return f"window={start}-{start + 9}" + if strategy == "volatility" and "vol_threshold" in params: + threshold = round(float(params["vol_threshold"]), 2) + return f"vol_threshold~{threshold:.2f}" + return "" + + @staticmethod + def _verdict(avg_sharpe: float, best_sharpe: float, worst_sharpe: float) -> str: + if avg_sharpe >= GOOD_SHARPE: + return "worked" + if best_sharpe >= GOOD_SHARPE and worst_sharpe <= BAD_SHARPE: + return "regime-sensitive" + if avg_sharpe <= BAD_SHARPE: + return "avoid" + return "mixed" diff --git a/src/agent/reporting.py b/src/agent/reporting.py new file mode 100644 index 0000000..466ba18 --- /dev/null +++ b/src/agent/reporting.py @@ -0,0 +1,98 @@ +"""Console reporting helpers for agent runs.""" + +from typing import Any, Dict, Iterable, List, Optional + +from src.utils.config import config + + +def _fmt_pct(value: float) -> str: + return f"{value * 100:.1f}%" + + +def verdict_for_metrics(metrics: Dict[str, Any], min_sharpe: Optional[float] = None) -> str: + """Return a concise pass/reject verdict for a candidate.""" + threshold = config.agent.min_acceptable_sharpe if min_sharpe is None else min_sharpe + sharpe = float(metrics.get("sharpe", metrics.get("sharpe_ratio", 0.0)) or 0.0) + bootstrap_p5 = float(metrics.get("bootstrap_sharpe_p5", sharpe) or 0.0) + max_drawdown = float(metrics.get("max_drawdown", 0.0) or 0.0) + if sharpe >= threshold and bootstrap_p5 >= 0 and max_drawdown <= config.agent.risk.max_drawdown: + return "passed" + if sharpe >= threshold: + return "watch" + return "rejected" + + +def result_rows(results: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Normalize agent result dicts for display.""" + rows = [] + for idx, result in enumerate(results, start=1): + metrics = { + "sharpe": result.get("sharpe", result.get("sharpe_ratio", 0.0)), + "calmar": result.get("calmar", 0.0), + "sortino": result.get("sortino", 0.0), + "max_drawdown": result.get("max_drawdown", 0.0), + "bootstrap_sharpe_p5": result.get("bootstrap_sharpe_p5", 0.0), + } + rows.append({ + "Rank": idx, + "Verdict": verdict_for_metrics(metrics), + "Strategy": result.get("strategy_type", ""), + "Params": result.get("params", {}), + "Sharpe": round(float(metrics["sharpe"] or 0.0), 3), + "Calmar": round(float(metrics["calmar"] or 0.0), 3), + "Sortino": round(float(metrics["sortino"] or 0.0), 3), + "Max DD": _fmt_pct(float(metrics["max_drawdown"] or 0.0)), + "Boot p5": round(float(metrics["bootstrap_sharpe_p5"] or 0.0), 3), + "Return": _fmt_pct(float(result.get("total_return", 0.0) or 0.0)), + "Method": result.get("generation_method", ""), + }) + return rows + + +def render_comparison_table(results: Iterable[Dict[str, Any]]) -> str: + """Render a screenshot-friendly markdown table.""" + rows = result_rows(results) + if not rows: + return "No candidate backtest results." + headers = ["Rank", "Verdict", "Strategy", "Sharpe", "Calmar", "Sortino", "Max DD", "Boot p5", "Return", "Method", "Params"] + lines = [ + "| " + " | ".join(headers) + " |", + "| " + " | ".join("---" for _ in headers) + " |", + ] + for row in rows: + lines.append( + "| {Rank} | {Verdict} | {Strategy} | {Sharpe:.3f} | {Calmar:.3f} | " + "{Sortino:.3f} | {Max DD} | {Boot p5:.3f} | {Return} | {Method} | `{Params}` |".format(**row) + ) + return "\n".join(lines) + + +def render_regime_card(state: Dict[str, Any]) -> str: + """Render a one-page summary of the completed agent run.""" + context = state.get("context") or state.get("regime_context") + best = state.get("best_result") or {} + regime_label = getattr(context, "regime_label", state.get("regime_label", "Unknown")) + confidence = getattr(context, "regime_confidence", state.get("regime_confidence", 0.0)) + strategy = best.get("strategy_type", state.get("strategy_type", "")) + params = best.get("params", {}) + reasoning = best.get("reasoning", "") or best.get("thesis", "") or "No explicit reasoning captured." + + lines = [ + "# AgentQuant Regime Card", + "", + f"Regime: {regime_label} ({confidence:.0%} confidence)", + f"Top strategy: {strategy} {params}", + "", + "| Metric | Value |", + "|---|---:|", + f"| Verdict | {verdict_for_metrics(best)} |", + f"| Sharpe | {float(best.get('sharpe', best.get('mean_sharpe', 0.0)) or 0.0):.3f} |", + f"| Calmar | {float(best.get('calmar', 0.0) or 0.0):.3f} |", + f"| Sortino | {float(best.get('sortino', 0.0) or 0.0):.3f} |", + f"| Max drawdown | {_fmt_pct(float(best.get('max_drawdown', 0.0) or 0.0))} |", + f"| Bootstrap Sharpe p5 | {float(best.get('bootstrap_sharpe_p5', 0.0) or 0.0):.3f} |", + "", + "Why this fits:", + reasoning, + ] + return "\n".join(lines) diff --git a/src/agent/runner.py b/src/agent/runner.py index 5348707..be9b183 100644 --- a/src/agent/runner.py +++ b/src/agent/runner.py @@ -7,12 +7,12 @@ """ import logging -import os import pandas as pd from dotenv import load_dotenv from src.agent.agent_graph import run_agent +from src.agent.reporting import render_comparison_table, render_regime_card from src.data.ingest import fetch_ohlcv_data from src.utils.config import config from src.utils.logging import setup_logging @@ -86,6 +86,11 @@ def main(): except Exception: print(df[cols].to_string(index=False)) + print() + print(render_regime_card(state)) + print() + print(render_comparison_table(results)) + logger.info("Agent run finished.") diff --git a/src/agent/strategy_memory.py b/src/agent/strategy_memory.py index e0a8943..4801f1c 100644 --- a/src/agent/strategy_memory.py +++ b/src/agent/strategy_memory.py @@ -10,10 +10,10 @@ import logging import sqlite3 import uuid -from dataclasses import asdict, dataclass -from datetime import datetime +from dataclasses import dataclass +from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Sequence from src.utils.config import config @@ -67,7 +67,7 @@ def store(self, result: PastResult) -> str: if not result.run_id: result.run_id = str(uuid.uuid4())[:8] if not result.timestamp: - result.timestamp = datetime.utcnow().isoformat() + result.timestamp = datetime.now(timezone.utc).isoformat() with sqlite3.connect(self.db_path) as conn: conn.execute( @@ -110,6 +110,128 @@ def recall( return [PastResult(**dict(row)) for row in rows] + def list_runs( + self, + regime: str = "", + strategy_type: str = "", + limit: int = 25, + order_by: str = "timestamp", + descending: bool = True, + ) -> List[PastResult]: + """List stored runs with optional filters.""" + allowed_order = { + "timestamp", + "sharpe", + "total_return", + "max_drawdown", + "confidence", + "regime", + "strategy_type", + } + if order_by not in allowed_order: + order_by = "timestamp" + + query = "SELECT * FROM strategy_runs WHERE 1=1" + params: list = [] + if regime: + query += " AND regime = ?" + params.append(regime) + if strategy_type: + query += " AND strategy_type = ?" + params.append(strategy_type) + direction = "DESC" if descending else "ASC" + query += f" ORDER BY {order_by} {direction} LIMIT ?" + params.append(limit) + + with sqlite3.connect(self.db_path) as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute(query, params).fetchall() + return [PastResult(**dict(row)) for row in rows] + + def query_regime( + self, + regime: str, + strategy_type: str = "", + limit: int = 100, + ) -> List[PastResult]: + """Compatibility helper for agentic memory consumers.""" + return self.list_runs(regime=regime, strategy_type=strategy_type, limit=limit) + + def query_all(self, strategy_type: str = "", limit: int = 200) -> List[PastResult]: + """Return recent runs across all regimes.""" + return self.list_runs(strategy_type=strategy_type, limit=limit) + + def summarize( + self, + regime: str = "", + strategy_type: str = "", + limit: int = 500, + ) -> List[Dict[str, Any]]: + """Aggregate memory by regime and strategy.""" + query = """ + SELECT + regime, + strategy_type, + COUNT(*) AS runs, + AVG(sharpe) AS avg_sharpe, + MAX(sharpe) AS best_sharpe, + AVG(total_return) AS avg_return, + AVG(max_drawdown) AS avg_drawdown, + MAX(timestamp) AS last_seen + FROM ( + SELECT * FROM strategy_runs + WHERE (? = '' OR regime = ?) + AND (? = '' OR strategy_type = ?) + ORDER BY timestamp DESC + LIMIT ? + ) + GROUP BY regime, strategy_type + ORDER BY avg_sharpe DESC, runs DESC + """ + params = (regime, regime, strategy_type, strategy_type, limit) + with sqlite3.connect(self.db_path) as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute(query, params).fetchall() + return [dict(row) for row in rows] + + def export_markdown( + self, + runs: Sequence[PastResult], + title: str = "AgentQuant Strategy Memory", + ) -> str: + """Render selected runs as screenshot-friendly markdown.""" + lines = [f"# {title}", ""] + if not runs: + lines.append("No strategy memory records found.") + return "\n".join(lines) + + lines.extend([ + "| Timestamp | Regime | Strategy | Sharpe | Return | Max DD | Params | Reasoning |", + "|---|---|---:|---:|---:|---:|---|---|", + ]) + for run in runs: + try: + params = json.loads(run.params) + except Exception: + params = run.params + reasoning = (run.reasoning or "").replace("\n", " ").strip() + if len(reasoning) > 120: + reasoning = reasoning[:117] + "..." + lines.append( + "| {timestamp} | {regime} | {strategy} | {sharpe:.3f} | " + "{ret:.1%} | {dd:.1%} | `{params}` | {reasoning} |".format( + timestamp=run.timestamp[:10], + regime=run.regime, + strategy=run.strategy_type, + sharpe=run.sharpe, + ret=run.total_return, + dd=run.max_drawdown, + params=params, + reasoning=reasoning, + ) + ) + return "\n".join(lines) + def to_prompt_context(self, regime: str, strategy_type: str = "", n: int = 5) -> str: """Format past results as context for LLM prompt.""" results = self.recall(regime, strategy_type, n) diff --git a/src/agent/swarm/__init__.py b/src/agent/swarm/__init__.py new file mode 100644 index 0000000..c1f898b --- /dev/null +++ b/src/agent/swarm/__init__.py @@ -0,0 +1,6 @@ +"""Multi-agent research swarm for AgentQuant.""" + +from src.agent.swarm.orchestrator import SwarmOrchestrator, run_swarm +from src.agent.swarm.state import SwarmResult, SwarmState + +__all__ = ["SwarmOrchestrator", "SwarmResult", "SwarmState", "run_swarm"] diff --git a/src/agent/swarm/coordinator.py b/src/agent/swarm/coordinator.py new file mode 100644 index 0000000..06d7cdc --- /dev/null +++ b/src/agent/swarm/coordinator.py @@ -0,0 +1,217 @@ +"""Backtest coordinator agent for multi-window validation.""" + +import logging +from collections import defaultdict +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any, Dict, List, Tuple + +import numpy as np +import pandas as pd + +from src.agent.proposal_generator import Proposal +from src.agent.swarm.state import SwarmState +from src.backtest.metrics import PerformanceMetrics +from src.backtest.runner import run_backtest +from src.utils.config import config + +logger = logging.getLogger(__name__) + +N_WINDOWS = 4 +MIN_WINDOW_BARS = 126 +WARMUP_BARS = 252 + + +def _build_windows( + ohlcv_data: Dict[str, pd.DataFrame], + ref_asset: str, + n_windows: int = N_WINDOWS, +) -> List[Tuple[str, pd.Timestamp, pd.Timestamp]]: + if ref_asset not in ohlcv_data or ohlcv_data[ref_asset].empty: + return [] + + idx = ohlcv_data[ref_asset].index + usable = idx[WARMUP_BARS:] + if len(usable) < MIN_WINDOW_BARS: + return [] + if len(usable) < n_windows * MIN_WINDOW_BARS: + return [("full_period", usable[0], idx[-1])] + + window_size = len(usable) // n_windows + windows = [] + for i in range(n_windows): + start = usable[i * window_size] + end = usable[min((i + 1) * window_size - 1, len(usable) - 1)] + windows.append((f"W{i + 1}_{start:%Y%m}_{end:%Y%m}", start, end)) + return windows + + +def _slice_data( + ohlcv_data: Dict[str, pd.DataFrame], + test_start: pd.Timestamp, + warmup_bars: int, +) -> Dict[str, pd.DataFrame]: + sliced = {} + for ticker, df in ohlcv_data.items(): + if df.empty: + continue + pos = df.index.searchsorted(test_start) + sliced[ticker] = df.iloc[max(0, pos - warmup_bars):] + return sliced + + +def _backtest_one_window( + proposal_key: str, + proposal: Proposal, + strategy_type: str, + ohlcv_data: Dict[str, pd.DataFrame], + assets: List[str], + window_label: str, + test_start: pd.Timestamp, + test_end: pd.Timestamp, +) -> Dict[str, Any]: + try: + result = run_backtest( + _slice_data(ohlcv_data, test_start, WARMUP_BARS), + assets, + strategy_type, + proposal.params, + eval_start=test_start, + ) + if not result or result.get("equity_curve") is None: + return {"proposal_key": proposal_key, "window_label": window_label, "error": "No result"} + + equity = result["equity_curve"] + equity_test = equity.loc[(equity.index >= test_start) & (equity.index <= test_end)] + metrics = ( + PerformanceMetrics.from_equity(equity_test, bootstrap=True) + if len(equity_test) > 5 + else result["metrics"] + ) + return { + "proposal_key": proposal_key, + "params": proposal.params, + "strategy_type": strategy_type, + "window_label": window_label, + "test_start": str(test_start.date()), + "test_end": str(test_end.date()), + "sharpe": metrics.get("sharpe", metrics.get("sharpe_ratio", 0.0)), + "total_return": metrics.get("total_return", 0.0), + "max_drawdown": metrics.get("max_drawdown", 0.0), + "calmar": metrics.get("calmar", 0.0), + "sortino": metrics.get("sortino", 0.0), + "bootstrap_sharpe_p5": metrics.get("bootstrap_sharpe_p5", 0.0), + "error": None, + } + except Exception as exc: + logger.debug("Window backtest failed for %s/%s: %s", proposal_key, window_label, exc) + return { + "proposal_key": proposal_key, + "window_label": window_label, + "sharpe": 0.0, + "total_return": 0.0, + "max_drawdown": 0.0, + "error": str(exc), + } + + +def _aggregate_windows(window_results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + by_proposal: Dict[str, List[Dict[str, Any]]] = defaultdict(list) + for result in window_results: + if result.get("error") is None: + by_proposal[result["proposal_key"]].append(result) + + ranking = [] + for proposal_key, rows in by_proposal.items(): + if not rows: + continue + sharpes = np.array([float(row.get("sharpe", 0.0) or 0.0) for row in rows]) + returns = np.array([float(row.get("total_return", 0.0) or 0.0) for row in rows]) + drawdowns = np.array([float(row.get("max_drawdown", 0.0) or 0.0) for row in rows]) + calmar = np.array([float(row.get("calmar", 0.0) or 0.0) for row in rows]) + sortino = np.array([float(row.get("sortino", 0.0) or 0.0) for row in rows]) + boot = np.array([float(row.get("bootstrap_sharpe_p5", 0.0) or 0.0) for row in rows]) + ranking.append( + { + "proposal_key": proposal_key, + "params": rows[0].get("params", {}), + "strategy_type": rows[0].get("strategy_type", ""), + "mean_sharpe": round(float(np.mean(sharpes)), 4), + "min_sharpe": round(float(np.min(sharpes)), 4), + "sharpe_std": round(float(np.std(sharpes)), 4), + "robustness_score": round(float(np.mean(sharpes) - np.std(sharpes)), 4), + "mean_return": round(float(np.mean(returns)), 4), + "worst_drawdown": round(float(np.max(drawdowns)), 4), + "calmar": round(float(np.mean(calmar)), 4), + "sortino": round(float(np.mean(sortino)), 4), + "bootstrap_sharpe_p5": round(float(np.min(boot)), 4), + "n_windows": len(rows), + } + ) + ranking.sort(key=lambda item: item["robustness_score"], reverse=True) + return ranking + + +def run_backtest_coordinator(state: SwarmState) -> SwarmState: + """Run all approved proposals across multiple time windows.""" + approved = state.get("approved_proposals", []) + assets = state.get("assets", [config.reference_asset]) + ohlcv_data = state["ohlcv_data"] + ref_asset = assets[0] if assets else config.reference_asset + windows = _build_windows(ohlcv_data, ref_asset) + + if not approved or not windows: + state["window_results"] = [] + state["final_ranking"] = [] + state["best_result"] = None + state.setdefault("run_log", []).append("[Coordinator] No approved proposals or windows to test.") + return state + + proposal_pairs: List[Tuple[str, Proposal, str]] = [] + for strategy_type, proposals in state.get("specialist_proposals", {}).items(): + for proposal in proposals: + if proposal in approved: + key = f"{strategy_type}:{tuple(sorted(proposal.params.items()))}" + proposal_pairs.append((key, proposal, strategy_type)) + + seen = set() + unique_pairs = [] + for key, proposal, strategy_type in proposal_pairs: + if key in seen: + continue + seen.add(key) + unique_pairs.append((key, proposal, strategy_type)) + + results: List[Dict[str, Any]] = [] + max_workers = min(max(len(unique_pairs) * len(windows), 1), 8) + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = [] + for proposal_key, proposal, strategy_type in unique_pairs: + for label, start, end in windows: + futures.append( + executor.submit( + _backtest_one_window, + proposal_key, + proposal, + strategy_type, + ohlcv_data, + assets, + label, + start, + end, + ) + ) + for future in as_completed(futures): + results.append(future.result()) + + ranking = _aggregate_windows(results) + best = ranking[0] if ranking else None + state["window_results"] = results + state["final_ranking"] = ranking + state["best_result"] = best + best_text = f"{best['robustness_score']:.3f}" if best else "N/A" + state.setdefault("run_log", []).append( + f"[Coordinator] {len(unique_pairs)} proposals x {len(windows)} windows -> " + f"{len(results)} results; best robustness={best_text}." + ) + logger.info(state["run_log"][-1]) + return state diff --git a/src/agent/swarm/critic_agent.py b/src/agent/swarm/critic_agent.py new file mode 100644 index 0000000..ee25079 --- /dev/null +++ b/src/agent/swarm/critic_agent.py @@ -0,0 +1,89 @@ +"""Critic agent that validates and de-duplicates strategy proposals.""" + +import logging +from typing import Dict, List, Tuple + +from src.agent.proposal_generator import Proposal, ProposalValidator +from src.agent.swarm.state import SwarmState + +logger = logging.getLogger(__name__) + + +class CriticAgent: + """Screens proposals before expensive multi-window backtests.""" + + def __init__(self): + self.validator = ProposalValidator() + + def review( + self, + proposals: List[Proposal], + strategy_type: str, + context, + ) -> Tuple[List[Proposal], List[Dict[str, str]]]: + approved: List[Proposal] = [] + rejected: List[Dict[str, str]] = [] + seen = set() + + for proposal in proposals: + key = (strategy_type, tuple(sorted(proposal.params.items()))) + if key in seen: + rejected.append({"proposal": str(proposal.params), "reason": "Duplicate proposal."}) + continue + seen.add(key) + + valid = self.validator.validate( + { + **proposal.params, + "confidence": proposal.confidence, + "reasoning": proposal.reasoning, + "regime_characteristic_used": proposal.regime_characteristic_used, + }, + strategy_type, + ) + if strategy_type == "trend_following": + sw = proposal.params.get("short_window", 0) + mw = proposal.params.get("medium_window", 0) + lw = proposal.params.get("long_window", 0) + valid = proposal if 0 < sw < mw < lw else None + + if valid is None: + rejected.append({"proposal": str(proposal.params), "reason": "Invalid parameters."}) + continue + + proposal.confidence = self._risk_adjusted_confidence(proposal, context) + approved.append(proposal) + + return approved, rejected + + @staticmethod + def _risk_adjusted_confidence(proposal: Proposal, context) -> float: + confidence = float(proposal.confidence or 0.5) + regime = getattr(context, "regime_label", "").lower() + if "crisis" in regime and proposal.params.get("slow_window", 0) > 100: + confidence *= 0.75 + if "lowvol" in regime and proposal.params.get("slow_window", 0) >= 100: + confidence = min(1.0, confidence + 0.1) + return max(0.0, min(1.0, confidence)) + + +def run_critic_agent(state: SwarmState) -> SwarmState: + """Review proposals from all specialists.""" + critic = CriticAgent() + context = state.get("regime_context") + approved: List[Proposal] = [] + rejections: List[Dict[str, str]] = [] + + for strategy_type, proposals in state.get("specialist_proposals", {}).items(): + ok, bad = critic.review(proposals, strategy_type, context) + approved.extend(ok) + rejections.extend(bad) + + state["approved_proposals"] = approved + state["rejected_count"] = len(rejections) + state["rejection_log"] = rejections + state.setdefault("run_log", []).append( + f"[Critic] Approved {len(approved)} proposals, rejected {len(rejections)}." + ) + logger.info(state["run_log"][-1]) + return state diff --git a/src/agent/swarm/memory_agent.py b/src/agent/swarm/memory_agent.py new file mode 100644 index 0000000..40d78f5 --- /dev/null +++ b/src/agent/swarm/memory_agent.py @@ -0,0 +1,86 @@ +"""Memory agent for cross-run strategy learning.""" + +import json +import logging +from typing import Any, Dict, List, Optional + +from src.agent.memory_layer import AgenticMemoryLayer +from src.agent.strategy_memory import PastResult, StrategyMemory +from src.agent.swarm.state import SwarmState + +logger = logging.getLogger(__name__) + + +class MemoryAgent: + """Extracts learned patterns and stores swarm outcomes.""" + + def __init__(self, memory: Optional[StrategyMemory] = None): + self.layer = AgenticMemoryLayer(memory=memory) + + def retrieve_patterns(self, regime_label: str, strategy_types: List[str]) -> List[str]: + return [ + pattern.to_sentence() + for pattern in self.layer.extract_patterns( + regime=regime_label, + strategy_types=strategy_types, + limit=200, + ) + ] + + def to_context_string(self, regime_label: str, strategy_types: List[str]) -> str: + return self.layer.to_prompt_context(regime=regime_label, strategy_types=strategy_types) + + def store_swarm_results( + self, + final_ranking: List[Dict[str, Any]], + regime_label: str, + ) -> List[str]: + run_ids = [] + for item in final_ranking[:5]: + result = PastResult( + regime=regime_label, + strategy_type=item.get("strategy_type", ""), + params=json.dumps(item.get("params", {})), + sharpe=float(item.get("mean_sharpe", 0.0) or 0.0), + total_return=float(item.get("mean_return", 0.0) or 0.0), + max_drawdown=float(item.get("worst_drawdown", 0.0) or 0.0), + confidence=float(item.get("robustness_score", 0.0) or 0.0), + generation_method="swarm", + reasoning=( + "Multi-agent swarm result. " + f"mean_sharpe={item.get('mean_sharpe', 0):.2f}, " + f"std={item.get('sharpe_std', 0):.2f}, " + f"min={item.get('min_sharpe', 0):.2f}." + ), + ) + run_ids.append(self.layer.memory.store(result)) + return run_ids + + +def run_memory_agent(state: SwarmState) -> SwarmState: + """Retrieve memory before specialists and persist rankings after backtests.""" + context = state.get("regime_context") + if context is None: + return state + + regime_label = context.regime_label + strategy_types = state.get("strategy_types", ["momentum"]) + agent = MemoryAgent() + + if not state.get("memory_context"): + patterns = agent.retrieve_patterns(regime_label, strategy_types) + state["memory_patterns"] = patterns + state["memory_context"] = agent.to_context_string(regime_label, strategy_types) + context.memory_context = state["memory_context"] + state.setdefault("run_log", []).append( + f"[Memory Agent] Retrieved {len(patterns)} learned patterns for {regime_label}." + ) + + if state.get("final_ranking"): + run_ids = agent.store_swarm_results(state["final_ranking"], regime_label) + state.setdefault("run_log", []).append( + f"[Memory Agent] Stored {len(run_ids)} swarm results." + ) + logger.info("Stored swarm memory run IDs: %s", run_ids) + + return state diff --git a/src/agent/swarm/orchestrator.py b/src/agent/swarm/orchestrator.py new file mode 100644 index 0000000..3897594 --- /dev/null +++ b/src/agent/swarm/orchestrator.py @@ -0,0 +1,122 @@ +"""Top-level multi-agent swarm orchestrator.""" + +import logging +import time +from typing import Dict, List, Optional + +import pandas as pd + +from src.agent.swarm.coordinator import run_backtest_coordinator +from src.agent.swarm.critic_agent import run_critic_agent +from src.agent.swarm.memory_agent import run_memory_agent +from src.agent.swarm.regime_analyst import run_regime_analyst +from src.agent.swarm.specialist_agents import run_strategy_specialists +from src.agent.swarm.state import SwarmResult, SwarmState +from src.strategies.strategy_registry import STRATEGY_REGISTRY +from src.utils.config import config + +logger = logging.getLogger(__name__) + +DEFAULT_STRATEGY_TYPES = ["momentum", "mean_reversion", "volatility", "trend_following"] + + +class SwarmOrchestrator: + """Runs the multi-agent research workflow behind an explicit flag.""" + + def __init__( + self, + strategy_types: Optional[List[str]] = None, + min_approved_proposals: int = 2, + ): + self.strategy_types = [ + strategy + for strategy in (strategy_types or DEFAULT_STRATEGY_TYPES) + if strategy in STRATEGY_REGISTRY + ] + self.min_approved_proposals = min_approved_proposals + + def run( + self, + ohlcv_data: Dict[str, pd.DataFrame], + assets: Optional[List[str]] = None, + ) -> SwarmResult: + start = time.perf_counter() + state: SwarmState = { + "ohlcv_data": ohlcv_data, + "assets": assets or [config.reference_asset], + "strategy_types": self.strategy_types, + "run_log": [], + } + + state = run_regime_analyst(state) + state = run_memory_agent(state) + state = run_strategy_specialists(state) + generated = len(state.get("all_proposals", [])) + state = run_critic_agent(state) + + approved = len(state.get("approved_proposals", [])) + rejected = state.get("rejected_count", 0) + if approved < self.min_approved_proposals: + proposals = sorted( + state.get("all_proposals", []), + key=lambda proposal: proposal.confidence, + reverse=True, + ) + state["approved_proposals"] = proposals[: self.min_approved_proposals] + approved = len(state["approved_proposals"]) + state.setdefault("run_log", []).append( + f"[Orchestrator] Re-admitted top proposals to reach {approved} approved candidates." + ) + + state = run_backtest_coordinator(state) + state = run_memory_agent(state) + state.setdefault("run_log", []).append( + f"[Orchestrator] Completed swarm in {time.perf_counter() - start:.1f}s." + ) + return self._build_result(state, generated, approved, rejected) + + @staticmethod + def _build_result( + state: SwarmState, + generated: int, + approved: int, + rejected: int, + ) -> SwarmResult: + best = state.get("best_result") or {} + context = state.get("regime_context") + methods: Dict[str, int] = {} + for proposal in state.get("all_proposals", []): + methods[proposal.generation_method] = methods.get(proposal.generation_method, 0) + 1 + + window_labels = { + row.get("window_label") + for row in state.get("window_results", []) + if row.get("window_label") + } + return SwarmResult( + best_params=best.get("params", {}), + best_strategy_type=best.get("strategy_type", ""), + mean_sharpe=float(best.get("mean_sharpe", 0.0) or 0.0), + min_sharpe=float(best.get("min_sharpe", 0.0) or 0.0), + sharpe_std=float(best.get("sharpe_std", 0.0) or 0.0), + robustness_score=float(best.get("robustness_score", 0.0) or 0.0), + total_proposals_generated=generated, + proposals_approved=approved, + proposals_rejected=rejected, + n_windows_tested=len(window_labels), + regime_label=getattr(context, "regime_label", "Unknown"), + regime_confidence=getattr(context, "regime_confidence", 0.0), + regime_narrative=state.get("regime_narrative", ""), + generation_methods=methods, + run_log=state.get("run_log", []), + full_ranking=state.get("final_ranking", []), + memory_patterns=state.get("memory_patterns", []), + ) + + +def run_swarm( + ohlcv_data: Dict[str, pd.DataFrame], + assets: Optional[List[str]] = None, + strategy_types: Optional[List[str]] = None, +) -> SwarmResult: + return SwarmOrchestrator(strategy_types=strategy_types).run(ohlcv_data, assets=assets) diff --git a/src/agent/swarm/regime_analyst.py b/src/agent/swarm/regime_analyst.py new file mode 100644 index 0000000..3ed6c63 --- /dev/null +++ b/src/agent/swarm/regime_analyst.py @@ -0,0 +1,86 @@ +"""Regime analyst agent for the multi-agent swarm.""" + +import logging +from typing import Dict + +import pandas as pd + +from src.agent.context_builder import RegimeContext, build_context +from src.agent.swarm.state import SwarmState +from src.features.engine import compute_features +from src.features.regime import detect_regime_full +from src.utils.config import config + +logger = logging.getLogger(__name__) + + +def _build_narrative(signals, context: RegimeContext) -> str: + parts = [] + vol_text = { + "crisis": "CRISIS VOL: volatility sits in the upper tail; prefer capital preservation and shorter confirmation windows.", + "high": "HIGH VOL: elevated volatility; require stricter drawdown controls and avoid fragile momentum settings.", + "mid": "MID VOL: balanced conditions; compare momentum and mean reversion directly.", + "low": "LOW VOL: calm conditions; trend following and long-horizon momentum deserve priority.", + } + parts.append(vol_text.get(signals.vol_regime, f"Volatility regime: {signals.vol_regime}.")) + + momentum_pct = signals.momentum_63d * 100 + if signals.momentum_63d > 0.03: + parts.append(f"TREND: positive 3M momentum ({momentum_pct:.1f}%).") + elif signals.momentum_63d < -0.03: + parts.append(f"TREND: negative 3M momentum ({momentum_pct:.1f}%).") + else: + parts.append(f"TREND: neutral 3M momentum ({momentum_pct:.1f}%).") + + parts.append( + f"VIX: {signals.vix_level:.1f}, trailing percentile {signals.vix_percentile_252d:.0f}, " + f"regime confidence {signals.regime_confidence:.0%}." + ) + parts.append( + "Momentum alignment: " + f"1M={context.momentum_21d * 100:.1f}%, " + f"3M={context.momentum_63d * 100:.1f}%, " + f"12M={context.momentum_252d * 100:.1f}%." + ) + return "\n".join(parts) + + +def _get_macro_context(ohlcv_data: Dict[str, pd.DataFrame]) -> str: + lines = [] + if "TLT" in ohlcv_data: + try: + tlt_return = ohlcv_data["TLT"]["Close"].pct_change(63).iloc[-1] + lines.append(f"Long-bond 3M return proxy: {tlt_return * 100:.1f}%.") + except Exception: + pass + if "HYG" in ohlcv_data and "LQD" in ohlcv_data: + try: + hyg = ohlcv_data["HYG"]["Close"].iloc[-1] / ohlcv_data["HYG"]["Close"].iloc[-63] + lqd = ohlcv_data["LQD"]["Close"].iloc[-1] / ohlcv_data["LQD"]["Close"].iloc[-63] + lines.append(f"Credit stress proxy (LQD-HYG 3M spread): {lqd - hyg:.3f}.") + except Exception: + pass + return "\n".join(lines) if lines else "Macro context unavailable from current universe." + + +def run_regime_analyst(state: SwarmState) -> SwarmState: + """Build the market context consumed by all downstream agents.""" + ohlcv_data = state["ohlcv_data"] + asset = state.get("assets", [config.reference_asset])[0] + features_df = compute_features(ohlcv_data, asset, config.vix_ticker) + signals = detect_regime_full(features_df) + context = build_context(features_df) + context.regime_label = signals.regime_label + context.regime_confidence = signals.regime_confidence + + narrative = _build_narrative(signals, context) + state["features_df"] = features_df + state["regime_context"] = context + state["regime_narrative"] = narrative + state["macro_summary"] = _get_macro_context(ohlcv_data) + state.setdefault("run_log", []).append( + f"[Regime Analyst] {signals.regime_label} | VIX={signals.vix_level:.1f} | " + f"confidence={signals.regime_confidence:.0%}" + ) + logger.info(state["run_log"][-1]) + return state diff --git a/src/agent/swarm/specialist_agents.py b/src/agent/swarm/specialist_agents.py new file mode 100644 index 0000000..900f7f6 --- /dev/null +++ b/src/agent/swarm/specialist_agents.py @@ -0,0 +1,62 @@ +"""Strategy specialist agents for the swarm.""" + +import logging +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Dict, List + +from src.agent.context_builder import RegimeContext +from src.agent.proposal_generator import Proposal, ProposalGenerator +from src.agent.swarm.state import SwarmState + +logger = logging.getLogger(__name__) + + +class StrategySpecialist: + """Generates proposals for one strategy family.""" + + def __init__(self, strategy_type: str): + self.strategy_type = strategy_type + self.generator = ProposalGenerator() + + def generate(self, context: RegimeContext, n: int = 3) -> List[Proposal]: + proposals = self.generator.generate( + context=context, + n_proposals=n, + strategy_type=self.strategy_type, + ) + for proposal in proposals: + proposal.generation_method = f"{self.strategy_type}:{proposal.generation_method}" + if not proposal.reasoning: + proposal.reasoning = f"{self.strategy_type} specialist proposal for {context.regime_label}." + return proposals + + +def run_strategy_specialists(state: SwarmState) -> SwarmState: + """Run configured strategy specialists in parallel.""" + context = state.get("regime_context") + if context is None: + raise ValueError("regime_context is required before running specialists") + + strategy_types = state.get("strategy_types", ["momentum"]) + specialist_proposals: Dict[str, List[Proposal]] = {} + all_proposals: List[Proposal] = [] + + max_workers = min(len(strategy_types), 4) or 1 + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = { + executor.submit(StrategySpecialist(strategy_type).generate, context, 3): strategy_type + for strategy_type in strategy_types + } + for future in as_completed(futures): + strategy_type = futures[future] + proposals = future.result() + specialist_proposals[strategy_type] = proposals + all_proposals.extend(proposals) + + state["specialist_proposals"] = specialist_proposals + state["all_proposals"] = all_proposals + state.setdefault("run_log", []).append( + f"[Specialists] Generated {len(all_proposals)} proposals from {len(strategy_types)} specialists." + ) + logger.info(state["run_log"][-1]) + return state diff --git a/src/agent/swarm/state.py b/src/agent/swarm/state.py new file mode 100644 index 0000000..dd4759e --- /dev/null +++ b/src/agent/swarm/state.py @@ -0,0 +1,94 @@ +"""Shared state and result objects for the AgentQuant swarm.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, TypedDict + +import pandas as pd + +from src.agent.context_builder import RegimeContext +from src.agent.proposal_generator import Proposal + + +class SwarmState(TypedDict, total=False): + """State passed between specialized swarm agents.""" + + ohlcv_data: Dict[str, pd.DataFrame] + assets: List[str] + strategy_types: List[str] + features_df: pd.DataFrame + regime_context: Optional[RegimeContext] + regime_narrative: str + macro_summary: str + specialist_proposals: Dict[str, List[Proposal]] + all_proposals: List[Proposal] + approved_proposals: List[Proposal] + rejected_count: int + rejection_log: List[Dict[str, str]] + window_results: List[Dict[str, Any]] + final_ranking: List[Dict[str, Any]] + best_result: Optional[Dict[str, Any]] + memory_patterns: List[str] + memory_context: str + run_log: List[str] + + +@dataclass +class SwarmResult: + """Structured result returned by SwarmOrchestrator.""" + + best_params: Dict[str, Any] = field(default_factory=dict) + best_strategy_type: str = "" + mean_sharpe: float = 0.0 + min_sharpe: float = 0.0 + sharpe_std: float = 0.0 + robustness_score: float = 0.0 + total_proposals_generated: int = 0 + proposals_approved: int = 0 + proposals_rejected: int = 0 + n_windows_tested: int = 0 + regime_label: str = "Unknown" + regime_confidence: float = 0.0 + regime_narrative: str = "" + generation_methods: Dict[str, int] = field(default_factory=dict) + run_log: List[str] = field(default_factory=list) + full_ranking: List[Dict[str, Any]] = field(default_factory=list) + memory_patterns: List[str] = field(default_factory=list) + + def summary(self) -> str: + return "\n".join( + [ + "=== Swarm Result ===", + f"Regime: {self.regime_label} (confidence={self.regime_confidence:.0%})", + f"Best strategy: {self.best_strategy_type} {self.best_params}", + ( + "Sharpe mean/min/std: " + f"{self.mean_sharpe:.3f} / {self.min_sharpe:.3f} / {self.sharpe_std:.3f}" + ), + f"Robustness: {self.robustness_score:.3f}", + f"Windows tested: {self.n_windows_tested}", + ( + "Proposals: " + f"{self.total_proposals_generated} generated, " + f"{self.proposals_approved} approved, " + f"{self.proposals_rejected} rejected" + ), + f"Methods: {self.generation_methods}", + ] + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "best_params": self.best_params, + "best_strategy_type": self.best_strategy_type, + "mean_sharpe": self.mean_sharpe, + "min_sharpe": self.min_sharpe, + "sharpe_std": self.sharpe_std, + "robustness_score": self.robustness_score, + "total_proposals": self.total_proposals_generated, + "proposals_approved": self.proposals_approved, + "proposals_rejected": self.proposals_rejected, + "n_windows": self.n_windows_tested, + "regime_label": self.regime_label, + "regime_confidence": self.regime_confidence, + "generation_methods": self.generation_methods, + } diff --git a/src/agent/trace.py b/src/agent/trace.py new file mode 100644 index 0000000..013b0d0 --- /dev/null +++ b/src/agent/trace.py @@ -0,0 +1,52 @@ +"""Live trace events for the AgentQuant ReAct loop.""" + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional + + +@dataclass +class TraceEvent: + """One visible step in the agent loop.""" + + stage: str + message: str + payload: Dict[str, Any] = field(default_factory=dict) + timestamp: str = field(default_factory=lambda: datetime.utcnow().strftime("%H:%M:%S")) + + +class TraceRecorder: + """Collect trace events and optionally print them live.""" + + def __init__(self, live: bool = False): + self.live = live + self.events: List[TraceEvent] = [] + + def emit(self, stage: str, message: str, **payload: Any) -> None: + event = TraceEvent(stage=stage, message=message, payload=payload) + self.events.append(event) + if self.live: + self._print_event(event) + + def _print_event(self, event: TraceEvent) -> None: + try: + from rich.console import Console + except Exception: + print(f"[{event.timestamp}] {event.stage}: {event.message}") + return + + console = Console() + color = { + "analyze": "cyan", + "hypothesize": "magenta", + "backtest": "yellow", + "reflect": "blue", + "store": "green", + "swarm": "green", + }.get(event.stage, "white") + console.print(f"[dim]{event.timestamp}[/dim] [{color}]{event.stage.upper()}[/{color}] {event.message}") + + +def emit_trace(trace: Optional[TraceRecorder], stage: str, message: str, **payload: Any) -> None: + if trace is not None: + trace.emit(stage, message, **payload) diff --git a/src/app/streamlit_app.py b/src/app/streamlit_app.py index b861daf..7b18f66 100644 --- a/src/app/streamlit_app.py +++ b/src/app/streamlit_app.py @@ -27,6 +27,7 @@ from src.agent.context_builder import build_context from src.agent.parameter_grid import ParameterGrid from src.agent.proposal_generator import ProposalGenerator +from src.agent.reporting import verdict_for_metrics from src.backtest.runner import run_backtest from src.data.ingest import fetch_ohlcv_data from src.features.engine import compute_features @@ -564,9 +565,16 @@ def main(): if st.session_state.strategies else "", "Params": str(p.params), "Sharpe": round(m.get("sharpe_ratio", 0), 3), + "Verdict": verdict_for_metrics({ + "sharpe": m.get("sharpe_ratio", 0), + "bootstrap_sharpe_p5": m.get("bootstrap_sharpe_p5", 0), + "max_drawdown": m.get("max_drawdown", 0), + }), "Return": f"{m.get('total_return', 0) * 100:.1f}%", "Max DD": f"{m.get('max_drawdown', 0) * 100:.1f}%", "Calmar": round(m.get("calmar", 0), 3), + "Sortino": round(m.get("sortino", 0), 3), + "Boot p5": round(m.get("bootstrap_sharpe_p5", 0), 3), "Trades": m.get("num_trades", 0), "Method": data["proposal"].generation_method, }) diff --git a/src/backtest/runner.py b/src/backtest/runner.py index d209403..6b0bc08 100644 --- a/src/backtest/runner.py +++ b/src/backtest/runner.py @@ -10,7 +10,6 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List, Optional -import numpy as np import pandas as pd from src.backtest.metrics import PerformanceMetrics @@ -205,7 +204,7 @@ def run_backtest( idx = combined.index.intersection(eq.index) combined = combined.loc[idx] + eq.loc[idx] - combined_metrics = PerformanceMetrics.from_equity(combined) + combined_metrics = PerformanceMetrics.from_equity(combined, bootstrap=True) combined_metrics["sharpe_ratio"] = combined_metrics.pop("sharpe", 0.0) combined_metrics["num_trades"] = sum(r["metrics"]["num_trades"] for r in asset_results) @@ -254,4 +253,4 @@ def _run_one(proposal: Dict) -> Dict: results.append(future.result()) results.sort(key=lambda x: x["metrics"].get("sharpe_ratio", -999), reverse=True) - return results \ No newline at end of file + return results diff --git a/src/cli.py b/src/cli.py new file mode 100644 index 0000000..f8aa8bc --- /dev/null +++ b/src/cli.py @@ -0,0 +1,179 @@ +"""Command line interface for AgentQuant.""" + +import argparse +from typing import Any, Dict, List + +import pandas as pd + +from src.agent.agent_graph import run_agent +from src.agent.memory_layer import AgenticMemoryLayer +from src.agent.reporting import render_comparison_table, render_regime_card +from src.agent.swarm import run_swarm +from src.agent.trace import TraceRecorder +from src.data.ingest import fetch_ohlcv_data +from src.utils.config import config +from src.utils.logging import setup_logging + + +def _print_table(rows: List[Dict[str, Any]]) -> None: + if not rows: + print("No records found.") + return + df = pd.DataFrame(rows) + try: + print(df.to_markdown(index=False, floatfmt=".3f")) + except Exception: + print(df.to_string(index=False)) + + +def _memory_command(args: argparse.Namespace) -> int: + layer = AgenticMemoryLayer() + if args.patterns: + patterns = layer.extract_patterns( + regime=args.regime or "", + strategy_types=[args.strategy] if args.strategy else None, + limit=args.limit, + ) + if not patterns: + print("No memory patterns found.") + return 0 + for pattern in patterns: + print(f"- {pattern.to_sentence()}") + return 0 + + if args.export == "markdown": + print( + layer.render_markdown( + regime=args.regime or "", + strategy_type=args.strategy or "", + limit=args.limit, + ) + ) + return 0 + + rows = [] + for row in layer.summary_rows(args.regime or "", args.strategy or "", limit=args.limit): + rows.append({ + "Regime": row["regime"], + "Strategy": row["strategy_type"], + "Runs": row["runs"], + "Avg Sharpe": row["avg_sharpe"], + "Best Sharpe": row["best_sharpe"], + "Avg Return": row["avg_return"], + "Avg DD": row["avg_drawdown"], + "Last Seen": str(row["last_seen"])[:10], + }) + _print_table(rows) + return 0 + + +def _run_command(args: argparse.Namespace) -> int: + setup_logging(config.log_level) + ohlcv_data = fetch_ohlcv_data( + ticker=args.ticker if args.ticker else None, + start_date=args.start, + end_date=args.end, + ) + if args.ticker and config.vix_ticker not in ohlcv_data: + ohlcv_data.update(fetch_ohlcv_data(ticker=config.vix_ticker, start_date=args.start, end_date=args.end)) + + assets = [args.ticker or config.reference_asset] + trace = TraceRecorder(live=args.trace) + + if args.swarm: + result = run_swarm( + ohlcv_data=ohlcv_data, + assets=assets, + strategy_types=args.strategies or None, + ) + print(result.summary()) + if result.full_ranking: + normalized = [ + { + **row, + "sharpe": row.get("mean_sharpe", 0.0), + "total_return": row.get("mean_return", 0.0), + "max_drawdown": row.get("worst_drawdown", 0.0), + "generation_method": "swarm", + } + for row in result.full_ranking + ] + print() + print(render_comparison_table(normalized)) + return 0 + + strategy_type = args.strategy or (config.strategies[0].name if config.strategies else "momentum") + state = run_agent( + ohlcv_data=ohlcv_data, + strategy_type=strategy_type, + asset=assets[0], + max_iterations=args.max_iterations, + trace=trace if args.trace else None, + ) + print(render_regime_card(state)) + print() + print(render_comparison_table(state.get("results", []))) + return 0 + + +def _regime_card_command(args: argparse.Namespace) -> int: + layer = AgenticMemoryLayer() + runs = layer.recent_runs(limit=1) + if not runs: + print("No completed run found in strategy memory.") + return 0 + latest = runs[0] + card_state = { + "regime_label": latest.regime, + "regime_confidence": latest.confidence, + "strategy_type": latest.strategy_type, + "best_result": { + "strategy_type": latest.strategy_type, + "params": latest.params, + "sharpe": latest.sharpe, + "total_return": latest.total_return, + "max_drawdown": latest.max_drawdown, + "reasoning": latest.reasoning, + }, + } + print(render_regime_card(card_state)) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="agentquant", description="AgentQuant research CLI") + subparsers = parser.add_subparsers(dest="command", required=True) + + run_parser = subparsers.add_parser("run", help="Run the research agent") + run_parser.add_argument("--ticker", default="", help="Single ticker to research, default SPY") + run_parser.add_argument("--start", default=None, help="Start date, YYYY-MM-DD") + run_parser.add_argument("--end", default=None, help="End date, YYYY-MM-DD") + run_parser.add_argument("--strategy", default="", help="Single-agent strategy type") + run_parser.add_argument("--strategies", nargs="*", help="Swarm strategy specialists") + run_parser.add_argument("--max-iterations", type=int, default=None) + run_parser.add_argument("--trace", action="store_true", help="Show live hypothesis/backtest/reflection trace") + run_parser.add_argument("--swarm", action="store_true", help="Run the multi-agent swarm") + run_parser.set_defaults(func=_run_command) + + memory_parser = subparsers.add_parser("memory", help="Browse agentic strategy memory") + memory_parser.add_argument("--regime", default="") + memory_parser.add_argument("--strategy", default="") + memory_parser.add_argument("--limit", type=int, default=25) + memory_parser.add_argument("--patterns", action="store_true", help="Show learned strategy patterns") + memory_parser.add_argument("--export", choices=["table", "markdown"], default="table") + memory_parser.set_defaults(func=_memory_command) + + card_parser = subparsers.add_parser("regime-card", help="Render the latest stored regime card") + card_parser.set_defaults(func=_regime_card_command) + + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_memory_layer.py b/tests/test_memory_layer.py new file mode 100644 index 0000000..ea6d6bf --- /dev/null +++ b/tests/test_memory_layer.py @@ -0,0 +1,58 @@ +import json + +from src.agent.memory_layer import AgenticMemoryLayer +from src.agent.strategy_memory import PastResult, StrategyMemory + + +def test_agentic_memory_extracts_strategy_patterns(tmp_path): + memory = StrategyMemory(db_path=str(tmp_path / "memory.db")) + memory.store( + PastResult( + regime="LowVol-Bull", + strategy_type="momentum", + params=json.dumps({"fast_window": 20, "slow_window": 100}), + sharpe=0.8, + total_return=0.12, + max_drawdown=0.08, + ) + ) + memory.store( + PastResult( + regime="LowVol-Bull", + strategy_type="momentum", + params=json.dumps({"fast_window": 50, "slow_window": 150}), + sharpe=0.6, + total_return=0.10, + max_drawdown=0.10, + ) + ) + + patterns = AgenticMemoryLayer(memory).extract_patterns( + regime="LowVol-Bull", + strategy_types=["momentum"], + ) + + assert patterns + assert any(pattern.verdict == "worked" for pattern in patterns) + assert "LowVol-Bull" in AgenticMemoryLayer(memory).to_prompt_context("LowVol-Bull", ["momentum"]) + + +def test_memory_markdown_export_contains_metrics(tmp_path): + memory = StrategyMemory(db_path=str(tmp_path / "memory.db")) + memory.store( + PastResult( + regime="HighVol-Bear", + strategy_type="mean_reversion", + params=json.dumps({"window": 20, "num_std": 2.0}), + sharpe=0.4, + total_return=0.05, + max_drawdown=0.04, + reasoning="Worked during volatile chop.", + ) + ) + + markdown = AgenticMemoryLayer(memory).render_markdown(limit=5) + + assert "AgentQuant Strategy Memory" in markdown + assert "HighVol-Bear" in markdown + assert "0.400" in markdown diff --git a/tests/test_reporting_cli.py b/tests/test_reporting_cli.py new file mode 100644 index 0000000..bb86f13 --- /dev/null +++ b/tests/test_reporting_cli.py @@ -0,0 +1,37 @@ +from src.agent.reporting import render_comparison_table, render_regime_card, verdict_for_metrics +from src.cli import build_parser + + +def test_verdict_rejects_weak_sharpe(): + assert verdict_for_metrics({"sharpe": -0.1, "max_drawdown": 0.05}) == "rejected" + + +def test_regime_card_and_comparison_table_render(): + state = { + "regime_label": "LowVol-Bull", + "regime_confidence": 0.8, + "strategy_type": "momentum", + "best_result": { + "strategy_type": "momentum", + "params": {"fast_window": 20, "slow_window": 100}, + "sharpe": 0.9, + "calmar": 1.2, + "sortino": 1.4, + "max_drawdown": 0.08, + "bootstrap_sharpe_p5": 0.2, + "reasoning": "Trend strength supports longer momentum windows.", + }, + } + card = render_regime_card(state) + table = render_comparison_table([state["best_result"]]) + + assert "AgentQuant Regime Card" in card + assert "LowVol-Bull" in card + assert "Boot p5" in table + + +def test_cli_parses_memory_patterns(): + parser = build_parser() + args = parser.parse_args(["memory", "--patterns", "--regime", "LowVol-Bull"]) + assert args.command == "memory" + assert args.patterns is True diff --git a/tests/test_swarm.py b/tests/test_swarm.py new file mode 100644 index 0000000..f2f04fa --- /dev/null +++ b/tests/test_swarm.py @@ -0,0 +1,70 @@ +import numpy as np +import pandas as pd + +from src.agent.proposal_generator import Proposal + + +def _make_ohlcv(n: int = 620, seed: int = 0) -> pd.DataFrame: + rng = np.random.default_rng(seed) + close = 100 * np.cumprod(1 + rng.normal(0.0005, 0.01, n)) + idx = pd.date_range("2020-01-01", periods=n) + return pd.DataFrame( + { + "Open": close, + "High": close * 1.005, + "Low": close * 0.995, + "Close": close, + "Volume": 1_000_000, + }, + index=idx, + ) + + +def _make_data(n: int = 620) -> dict: + return { + "SPY": _make_ohlcv(n, seed=0), + "^VIX": pd.DataFrame( + {"Close": np.random.default_rng(1).uniform(15, 35, n)}, + index=pd.date_range("2020-01-01", periods=n), + ), + } + + +def test_regime_analyst_produces_narrative(): + from src.agent.swarm.regime_analyst import run_regime_analyst + + state = {"ohlcv_data": _make_data(), "assets": ["SPY"], "strategy_types": ["momentum"], "run_log": []} + result = run_regime_analyst(state) + + assert result["regime_context"].regime_label + assert "VIX" in result["regime_narrative"] + + +def test_critic_rejects_duplicate_proposals(): + from src.agent.context_builder import RegimeContext + from src.agent.swarm.critic_agent import CriticAgent + + critic = CriticAgent() + proposal = Proposal(params={"fast_window": 10, "slow_window": 30}, confidence=0.5) + approved, rejected = critic.review( + [proposal, Proposal(params={"fast_window": 10, "slow_window": 30}, confidence=0.4)], + "momentum", + RegimeContext(regime_label="LowVol-Bull"), + ) + + assert len(approved) == 1 + assert len(rejected) == 1 + + +def test_swarm_runs_synthetic_smoke(monkeypatch, tmp_path): + monkeypatch.setenv("GOOGLE_API_KEY", "") + monkeypatch.chdir(tmp_path) + + from src.agent.swarm.orchestrator import SwarmOrchestrator + from src.agent.swarm.state import SwarmResult + + result = SwarmOrchestrator(strategy_types=["momentum"]).run(_make_data(), assets=["SPY"]) + + assert isinstance(result, SwarmResult) + assert result.regime_label + assert result.total_proposals_generated > 0