diff --git a/CHANGELOG.md b/CHANGELOG.md index f0e0b7a..9ff932a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), ## [0.5.1] - 2026-08-29 +### 🍳 Cookbooks +- **LangGraph Financial Analyst**: Added a four-node yfinance + DeepSeek agent with durable WAL checkpoints, real post-memo SIGKILL recovery, independent external-call/token receipts, and scoped sub-millisecond fast-forward measurements. + ### 🔄 Liveness & Auto-Supervision - **Programmatic `@supervise` & `supervise()`**: Embedded parent-child process supervisor that catches `SIGKILL (137)`, OOM, and abnormal subprocess crashes, automatically relaunching until completion. - **CLI Watcher (`lil watch`)**: Real-time terminal watcher with ASCII progress indicators, exponential backoff, and jitter (`lil watch script.py`). diff --git a/README.md b/README.md index 5454ac7..ba2829f 100644 --- a/README.md +++ b/README.md @@ -181,11 +181,33 @@ Explore runnable self-contained examples in [`examples/`](examples/): | **CrewAI** | [**Durable Tools Example**](examples/crewai_durable_tools.py) | ✅ Ready | Multi-agent tool execution with step-level resumption and zero duplicate side-effects | | **LlamaIndex** | [**Durable Workflows Example**](examples/llamaindex_durable_workflow.py) | ✅ Ready | Event-driven `@step` pipeline with crash durability and sub-millisecond fast-forward | | **OpenAI Swarm** | [**Durable Handoff Example**](examples/swarm_durable_handoff.py) | ✅ Ready | Multi-agent context handoff with WAL v2 serialization | -| **LangGraph** | [**Issue #82: Financial Analyst Agent**](https://github.com/sdageltc/letitloop/issues/82) | 🤝 Contributor | 4-step `yfinance` + StateGraph equity analysis surviving simulated SIGKILL | +| **LangGraph** | [**Financial Analyst Agent**](examples/cookbooks/langgraph_financial_analyst.py) | ✅ Ready | 4-step `yfinance` + DeepSeek StateGraph with independently audited SIGKILL recovery | | **DSPy** | [**Issue #83: Prompt Optimizer Pipeline**](https://github.com/sdageltc/letitloop/issues/83) | 🤝 Contributor | Async `BootstrapFewShot` / Teleprompter tuning with zero lost progress | | **Playwright** | [**Issue #88: Web Scraping Agent**](https://github.com/sdageltc/letitloop/issues/88) | 🤝 Contributor | Multi-page browser scraper that checkpoints DOM items to skip scraped pages | | **Pydantic AI** | [**Issue #89: Pydantic AI Integration**](https://github.com/sdageltc/letitloop/issues/89) | 🤝 Contributor | Type-safe agent with tool-calling checkpointing and zero token waste | +Install and run the financial analyst without paid API calls: + +```bash +python -m pip install -e ".[financial-agent]" +python examples/cookbooks/langgraph_financial_analyst.py --ticker AAPL --offline +python examples/cookbooks/langgraph_financial_analyst.py --ticker AAPL --offline --demo +``` + +For a live investment memo, configure DeepSeek only through the environment: + +```bash +export DEEPSEEK_API_KEY="your-key" +python examples/cookbooks/langgraph_financial_analyst.py --ticker AAPL --model deepseek:deepseek-v4-flash --live +``` + +If a local Python installation has no default CA bundle, set `SSL_CERT_FILE="$(python -m certifi)"`. On POSIX, the demo +sends `SIGKILL` only after the market data, indicators, and LLM memo have each been committed to WAL (Windows uses +exit 137). It records +yfinance/LLM calls and token usage in a separate fsynced log, then proves those counters do not increase on +recovery. Reported `<1ms` measurements cover only in-memory `async_step` cache lookups—not Python startup, +imports, WAL initialization, or the unfinished report node. + --- ## 🛡️ GitHub Action CI Gate (v2) diff --git a/examples/cookbooks/langgraph_financial_analyst.py b/examples/cookbooks/langgraph_financial_analyst.py new file mode 100644 index 0000000..68d4259 --- /dev/null +++ b/examples/cookbooks/langgraph_financial_analyst.py @@ -0,0 +1,996 @@ +"""LangGraph Financial Analyst Cookbook — crash-proof analysis with @durable_async. + +Combines injectable market-data/LLM adapters with LangGraph StateGraph orchestration +and LetItLoop ``@durable_async`` WAL checkpoints. The adapters make the cookbook +safe to exercise offline in tests while later phases can supply yfinance and a real +LLM without changing the graph or recovery code. + +Architecture: + Fetch Market Data -> Calculate Indicators -> Generate Investment Memo -> Generate Report + (WAL step 1) (WAL step 2) (WAL step 3) (WAL step 4) + +On POSIX, the demo sends a real SIGKILL immediately after the investment memo's +durable step has been committed (Windows uses exit 137). An independent fsynced +call log proves that recovery does not repeat yfinance or LLM calls. Its <1ms +timings cover only in-memory ``async_step`` +cache lookups; process startup, imports, WAL loading, and unfinished work are excluded. + +Modes: + - offline (default): deterministic market data and memo; no external API calls + - live: yfinance market data and DeepSeek through ``orchestrator.llm.call_llm`` + +Usage: + python -m pip install -e ".[financial-agent]" + python examples/cookbooks/langgraph_financial_analyst.py --demo + python examples/cookbooks/langgraph_financial_analyst.py --ticker NVDA --offline + DEEPSEEK_API_KEY=... python examples/cookbooks/langgraph_financial_analyst.py --ticker AAPL --live + python examples/cookbooks/langgraph_financial_analyst.py --ticker AAPL --offline --kill-at 2 +""" + +from __future__ import annotations + +import argparse +import asyncio +import dataclasses +import hashlib +import json +import math +import os +import pathlib +import subprocess +import sys +import time +from datetime import date, timedelta +from typing import Any, Awaitable, Callable, Dict, Optional, TypedDict + +ROOT = pathlib.Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from orchestrator.decorators import async_step, durable_async # noqa: E402 +from orchestrator.llm import call_llm # noqa: E402 + +WAL_DIR_DEFAULT = str(ROOT / ".bench_wal" / "cookbooks" / "langgraph_financial_analyst") +GOAL_ID = "langgraph-financial-analyst" +DEFAULT_MODEL = "deepseek:deepseek-v4-flash" + +JsonDict = Dict[str, Any] +MarketDataFetcher = Callable[[str], Awaitable[JsonDict]] +LLMCaller = Callable[[JsonDict, JsonDict, str], Awaitable[JsonDict]] + + +class MarketDataError(RuntimeError): + """Raised when live market data cannot be fetched or normalised safely.""" + + +class FinancialAgentState(TypedDict, total=False): + """JSON/WAL-safe state shared by every LangGraph node.""" + + ticker: str + model: str + offline: bool + market_data: JsonDict + indicators: JsonDict + investment_memo: JsonDict + final_report: JsonDict + + +@dataclasses.dataclass(frozen=True) +class FinancialAgentDependencies: + """External side effects injected into the graph for live use or tests.""" + + market_data_fetcher: MarketDataFetcher + llm_caller: LLMCaller + + +# --- Step 1: Market Data Ingestion (live adapter comes in phase 2) --- + + +def _simulated_market_data(ticker: str) -> JsonDict: + """Return deterministic JSON-safe data for tests and the offline demo.""" + ticker = _normalise_ticker(ticker) + seed = int(hashlib.sha256(ticker.encode()).hexdigest()[:6], 16) + base = 100.0 + (seed % 150) + first_day = date(2025, 1, 1) + close_prices = [round(base + (index * 0.15) + (((index % 7) - 3) * 0.4), 2) for index in range(260)] + dates = [(first_day + timedelta(days=index)).isoformat() for index in range(len(close_prices))] + market_cap = int(base * 12_500_000_000) + return { + "ticker": ticker, + "source": "simulated", + "period": "1y", + "interval": "1d", + "currency": "USD", + "current_price": close_prices[-1], + "market_cap": market_cap, + "trailing_pe": 24.0, + "forward_pe": 21.5, + "revenue_growth": 0.08, + "profit_margin": 0.21, + "volume": 54_200_000, + "dates": dates, + "close_prices": close_prices, + # Compatibility aliases retained until the legacy report is replaced. + "prices": close_prices, + "market_cap_b": round(market_cap / 1_000_000_000, 2), + } + + +def _finite_float(value: Any) -> Optional[float]: + """Convert numpy/pandas scalars to a finite built-in float.""" + if value is None or isinstance(value, bool): + return None + try: + converted = float(value) + except (TypeError, ValueError, OverflowError): + return None + return converted if math.isfinite(converted) else None + + +def _finite_int(value: Any) -> Optional[int]: + converted = _finite_float(value) + return int(converted) if converted is not None else None + + +def _lookup(container: Any, *names: str) -> Any: + """Read a value from yfinance dict-like or attribute-style objects.""" + if container is None: + return None + for name in names: + try: + if isinstance(container, dict) and name in container: + return container[name] + return container[name] + except (KeyError, TypeError, AttributeError): + try: + value = getattr(container, name) + except (AttributeError, KeyError, TypeError): + continue + else: + return value + return None + + +def _date_to_iso(value: Any) -> str: + candidate = value.date() if hasattr(value, "date") and callable(value.date) else value + if hasattr(candidate, "isoformat") and callable(candidate.isoformat): + return str(candidate.isoformat()) + return str(candidate) + + +def _normalise_history(history: Any) -> tuple[list[str], list[float]]: + """Extract aligned dates/prices from a pandas-like yfinance history frame.""" + if history is None or bool(getattr(history, "empty", False)): + raise MarketDataError("yfinance returned no historical prices") + try: + close_series = history["Close"] + rows = close_series.items() + except (KeyError, TypeError, AttributeError) as exc: + raise MarketDataError("yfinance history is missing the Close column") from exc + + dates: list[str] = [] + close_prices: list[float] = [] + for index, raw_price in rows: + price = _finite_float(raw_price) + if price is None: + continue + dates.append(_date_to_iso(index)) + close_prices.append(price) + + if len(close_prices) < 50: + raise MarketDataError(f"At least 50 finite closing prices are required; yfinance returned {len(close_prices)}") + return dates, close_prices + + +def _fetch_market_data(ticker: str, *, yf_module: Any = None) -> JsonDict: + """Fetch one year of prices/fundamentals and return only JSON-safe values.""" + ticker = _normalise_ticker(ticker) + if yf_module is None: + try: + import yfinance as yf_module + except ImportError as exc: + raise MarketDataError("yfinance is required for live mode; install it with `pip install yfinance`") from exc + + try: + security = yf_module.Ticker(ticker) + history = security.history( + period="1y", + interval="1d", + auto_adjust=True, + actions=False, + timeout=15, + ) + except Exception as exc: + raise MarketDataError(f"Unable to fetch one year of history for {ticker}: {exc}") from exc + + dates, close_prices = _normalise_history(history) + + try: + info = getattr(security, "info", None) or {} + except Exception: + info = {} + try: + fast_info = getattr(security, "fast_info", None) or {} + except Exception: + fast_info = {} + + current_price = _finite_float( + _lookup(fast_info, "last_price", "lastPrice") + or _lookup(info, "currentPrice", "regularMarketPrice") + or close_prices[-1] + ) + market_cap = _finite_int(_lookup(fast_info, "market_cap", "marketCap") or _lookup(info, "marketCap")) + volume = _finite_int( + _lookup(fast_info, "last_volume", "lastVolume") or _lookup(info, "regularMarketVolume", "volume") + ) + currency = _lookup(info, "currency") or _lookup(fast_info, "currency") + currency = str(currency) if currency is not None else None + + return { + "ticker": ticker, + "source": "yfinance", + "period": "1y", + "interval": "1d", + "currency": currency, + "current_price": current_price or close_prices[-1], + "market_cap": market_cap, + "trailing_pe": _finite_float(_lookup(info, "trailingPE")), + "forward_pe": _finite_float(_lookup(info, "forwardPE")), + "revenue_growth": _finite_float(_lookup(info, "revenueGrowth")), + "profit_margin": _finite_float(_lookup(info, "profitMargins")), + "volume": volume, + "dates": dates, + "close_prices": close_prices, + # Compatibility aliases retained until the legacy report is replaced. + "prices": close_prices, + "market_cap_b": round(market_cap / 1_000_000_000, 2) if market_cap is not None else None, + } + + +async def _offline_market_data_fetcher(ticker: str) -> JsonDict: + """Async dependency used by tests and ``--offline`` runs.""" + return _simulated_market_data(ticker) + + +async def _default_market_data_fetcher(ticker: str) -> JsonDict: + """Run the synchronous market-data adapter without blocking the event loop.""" + return await asyncio.to_thread(_fetch_market_data, ticker) + + +# --- Step 2: Technical Signal Computation --- + + +def _ema(values: list[float], period: int) -> list[float]: + """Return an EMA series seeded with the first value.""" + if not values: + raise ValueError("EMA requires at least one price") + if period <= 0: + raise ValueError("EMA period must be positive") + multiplier = 2.0 / (period + 1) + output = [values[0]] + for value in values[1:]: + output.append(((value - output[-1]) * multiplier) + output[-1]) + return output + + +def _rsi(values: list[float], period: int = 14) -> float: + """Calculate Wilder's RSI, including stable all-up/down/flat handling.""" + if len(values) < period + 1: + raise ValueError(f"RSI{period} requires at least {period + 1} prices") + changes = [values[index] - values[index - 1] for index in range(1, len(values))] + gains = [max(change, 0.0) for change in changes] + losses = [max(-change, 0.0) for change in changes] + average_gain = sum(gains[:period]) / period + average_loss = sum(losses[:period]) / period + for gain, loss in zip(gains[period:], losses[period:]): + average_gain = ((average_gain * (period - 1)) + gain) / period + average_loss = ((average_loss * (period - 1)) + loss) / period + + if average_gain == 0.0 and average_loss == 0.0: + return 50.0 + if average_loss == 0.0: + return 100.0 + if average_gain == 0.0: + return 0.0 + relative_strength = average_gain / average_loss + return 100.0 - (100.0 / (1.0 + relative_strength)) + + +def _calculate_technical_signals(market_data: JsonDict) -> JsonDict: + """Compute SMA20/SMA50, Wilder RSI14, and MACD(12, 26, 9).""" + raw_prices = market_data.get("close_prices") or market_data.get("prices") or [] + prices = [price for raw in raw_prices if (price := _finite_float(raw)) is not None] + if len(prices) < 50: + raise ValueError(f"At least 50 finite closing prices are required; received {len(prices)}") + + sma20 = sum(prices[-20:]) / 20 + sma50 = sum(prices[-50:]) / 50 + rsi14 = _rsi(prices, period=14) + ema12 = _ema(prices, period=12) + ema26 = _ema(prices, period=26) + macd_series = [short - long for short, long in zip(ema12, ema26)] + signal_series = _ema(macd_series, period=9) + macd = macd_series[-1] + macd_signal = signal_series[-1] + macd_histogram = macd - macd_signal + returns = [ + (prices[index] - prices[index - 1]) / prices[index - 1] + for index in range(1, len(prices)) + if prices[index - 1] != 0 + ] + volatility = sum(abs(value) for value in returns) / len(returns) if returns else 0.0 + current_price = _finite_float(market_data.get("current_price")) or prices[-1] + trend = "BULLISH" if sma20 > sma50 and macd > macd_signal else "BEARISH" + + return { + "ticker": market_data["ticker"], + "current_price": round(current_price, 6), + "sma20": round(sma20, 6), + "sma50": round(sma50, 6), + "rsi14": round(rsi14, 6), + "macd": round(macd, 6), + "macd_signal": round(macd_signal, 6), + "macd_histogram": round(macd_histogram, 6), + "trend": trend, + "volatility": round(volatility, 6), + # Compatibility aliases retained for the phase-1 report contract. + "sma_short": round(sma20, 6), + "sma_long": round(sma50, 6), + "rsi_approx": round(rsi14, 6), + } + + +# --- Step 3: LLM / Reasoning Synthesis --- + + +def _synthesize_analyst_thesis(signals: JsonDict) -> JsonDict: + """Synthesize investment thesis, valuation targets, and key risk factors.""" + ticker = signals["ticker"] + trend = signals["trend"] + current_price = signals["current_price"] + + if trend == "BULLISH": + recommendation = "OVERWEIGHT" + target_price = round(current_price * 1.18, 2) + risk_level = "MODERATE" + rationale = f"{ticker} exhibits upward momentum with price trading above SMA long-term average." + else: + recommendation = "NEUTRAL" + target_price = round(current_price * 1.02, 2) + risk_level = "HIGH" + rationale = f"{ticker} displays consolidation pressure below moving average threshold." + + return { + "ticker": ticker, + "recommendation": recommendation, + "target_price": target_price, + "risk_level": risk_level, + "rationale": rationale, + } + + +async def _offline_llm_caller(market_data: JsonDict, indicators: JsonDict, model: str) -> JsonDict: + """Return a deterministic memo and simulated token usage without network calls.""" + memo = _synthesize_analyst_thesis(indicators) + text = f"""## Investment view +{memo["recommendation"]} on {market_data["ticker"]} with a target price of ${memo["target_price"]}. + +## Main reasons +{memo["rationale"]} + +## Risks +Risk level: {memo["risk_level"]}. This offline example does not include all company-specific risks. + +## Disclaimer +This educational example is not financial advice. +""".strip() + prompt = _build_investment_memo_prompt(market_data, indicators) + prompt_tokens = max(1, len(prompt) // 4) + completion_tokens = max(1, len(text) // 4) + memo.update( + { + "text": text, + "model": model, + "provider": "offline", + "usage": { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + }, + } + ) + return memo + + +def _build_investment_memo_prompt(market_data: JsonDict, indicators: JsonDict) -> str: + """Build a compact, inspectable prompt without sending the full price history.""" + fundamentals = { + "ticker": market_data["ticker"], + "currency": market_data.get("currency"), + "current_price": market_data.get("current_price"), + "market_cap": market_data.get("market_cap"), + "trailing_pe": market_data.get("trailing_pe"), + "forward_pe": market_data.get("forward_pe"), + "revenue_growth": market_data.get("revenue_growth"), + "profit_margin": market_data.get("profit_margin"), + } + technicals = { + "sma20": indicators["sma20"], + "sma50": indicators["sma50"], + "rsi14": indicators["rsi14"], + "macd": indicators["macd"], + "macd_signal": indicators["macd_signal"], + "macd_histogram": indicators["macd_histogram"], + } + return f"""Write a concise investment memorandum for {market_data["ticker"]}. + +Fundamentals and current price: +{json.dumps(fundamentals, indent=2, sort_keys=True)} + +Technical indicators: +{json.dumps(technicals, indent=2, sort_keys=True)} + +Use Markdown and include these clearly labelled sections: +1. Investment view (bullish, neutral, or bearish) +2. Main reasons grounded in the supplied fundamentals and indicators +3. Key risks and uncertainties +4. Disclaimer that this is educational analysis, not financial advice + +Do not invent missing fundamentals. State when a supplied value is unavailable. +""".strip() + + +def _normalise_llm_usage(usage: Any) -> JsonDict: + """Return the provider's token accounting as finite JSON-safe integers.""" + raw_usage = usage if isinstance(usage, dict) else {} + prompt_tokens = max(0, _finite_int(raw_usage.get("prompt_tokens")) or 0) + completion_tokens = max(0, _finite_int(raw_usage.get("completion_tokens")) or 0) + total_tokens = _finite_int(raw_usage.get("total_tokens")) + if total_tokens is None: + total_tokens = prompt_tokens + completion_tokens + return { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": max(0, total_tokens), + } + + +async def _live_llm_caller(market_data: JsonDict, indicators: JsonDict, model: str) -> JsonDict: + """Call the repository's synchronous provider adapter without blocking the graph.""" + prompt = _build_investment_memo_prompt(market_data, indicators) + try: + response = await asyncio.to_thread( + call_llm, + prompt, + model, + system=( + "You are a careful financial research assistant. Use only the supplied data, " + "separate observations from inference, and never present the memo as personalized advice." + ), + max_tokens=1_000, + temperature=0.2, + ) + except Exception as exc: + raise RuntimeError(f"LLM request failed for {market_data['ticker']} using {model}: {exc}") from exc + if not isinstance(response, dict): + raise RuntimeError("The LLM adapter returned a non-object response") + text = response.get("text") + if not isinstance(text, str) or not text.strip(): + raise RuntimeError("The LLM adapter returned an empty investment memo") + return { + "ticker": market_data["ticker"], + "text": text.strip(), + "model": str(response.get("model") or model), + "provider": str(response.get("provider") or model.partition(":")[0]), + "usage": _normalise_llm_usage(response.get("usage")), + } + + +# --- Step 4: Executive Briefing Generation --- + + +def _generate_final_report(thesis: JsonDict, signals: JsonDict) -> JsonDict: + """Produce formatted executive analyst artifact.""" + ticker = thesis["ticker"] + memo_text = str(thesis.get("text") or thesis.get("rationale") or "No investment memo was produced.") + if thesis.get("recommendation") and thesis.get("target_price") is not None: + rating_line = ( + f"**Rating:** {thesis['recommendation']} | **Target:** ${thesis['target_price']} " + f"| **Risk:** {thesis.get('risk_level', 'N/A')}" + ) + summary = f"{ticker} rated {thesis['recommendation']} (Target: ${thesis['target_price']})" + else: + rating_line = ( + f"**Memo model:** {thesis.get('model', 'unknown')} | **Provider:** {thesis.get('provider', 'unknown')}" + ) + summary = f"{ticker} investment memorandum generated" + report_md = f"""# Equity Research Note: {ticker} +{rating_line} + +## Key Technicals +- Spot Price: ${signals["current_price"]} +- Trend Signal: {signals["trend"]} (RSI14: {signals["rsi14"]}) +- SMA20 / SMA50: ${signals["sma20"]} / ${signals["sma50"]} +- MACD / Signal / Histogram: {signals["macd"]} / {signals["macd_signal"]} / {signals["macd_histogram"]} + +## Investment Thesis +{memo_text} +""".strip() + + return { + "ticker": ticker, + "status": "COMPLETED", + "summary": summary, + "markdown": report_md, + } + + +# --- Injectable node contracts / LangGraph builder --- + + +async def _fetch_market_data_node( + state: FinancialAgentState, dependencies: FinancialAgentDependencies +) -> FinancialAgentState: + return {"market_data": await dependencies.market_data_fetcher(state["ticker"])} + + +async def _compute_indicators_node( + state: FinancialAgentState, dependencies: FinancialAgentDependencies +) -> FinancialAgentState: + del dependencies + return {"indicators": _calculate_technical_signals(state["market_data"])} + + +async def _generate_investment_memo_node( + state: FinancialAgentState, dependencies: FinancialAgentDependencies +) -> FinancialAgentState: + memo = await dependencies.llm_caller(state["market_data"], state["indicators"], state["model"]) + return {"investment_memo": memo} + + +async def _generate_report_node( + state: FinancialAgentState, dependencies: FinancialAgentDependencies +) -> FinancialAgentState: + del dependencies + report = _generate_final_report(state["investment_memo"], state["indicators"]) + return {"final_report": report} + + +def _normalise_ticker(ticker: str) -> str: + normalised = ticker.upper().strip() + if not normalised or not all(char.isalnum() or char in {"-", "."} for char in normalised): + raise ValueError(f"Invalid ticker symbol: {ticker!r}") + return normalised + + +def _workflow_identity(ticker: str, model: str, offline: bool) -> str: + """Return a stable ticker-aware identity so unrelated runs never share cache.""" + ticker_slug = _normalise_ticker(ticker).lower().replace(".", "-") + mode = "offline" if offline else "live" + model_digest = hashlib.sha256(model.encode("utf-8")).hexdigest()[:8] + return f"{ticker_slug}-{mode}-{model_digest}" + + +def _workflow_wal_dir(wal_dir: str, ticker: str, model: str, offline: bool) -> str: + return str(pathlib.Path(wal_dir) / _workflow_identity(ticker, model, offline)) + + +def _append_call_event(call_log_path: str, event: JsonDict) -> None: + """Append and fsync one external-call receipt independent of the workflow WAL.""" + path = pathlib.Path(call_log_path) + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"timestamp": time.time(), **event} + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(payload, sort_keys=True) + "\n") + handle.flush() + os.fsync(handle.fileno()) + + +def _with_call_logging( + dependencies: FinancialAgentDependencies, call_log_path: Optional[str] +) -> FinancialAgentDependencies: + """Wrap successful external calls with receipts that survive SIGKILL.""" + if call_log_path is None: + return dependencies + + async def logged_market_fetcher(ticker: str) -> JsonDict: + result = await dependencies.market_data_fetcher(ticker) + _append_call_event(call_log_path, {"event": "market_fetch", "ticker": ticker}) + return result + + async def logged_llm_caller(market_data: JsonDict, indicators: JsonDict, model: str) -> JsonDict: + result = await dependencies.llm_caller(market_data, indicators, model) + _append_call_event( + call_log_path, + { + "event": "llm_call", + "ticker": market_data["ticker"], + "model": model, + "usage": _normalise_llm_usage(result.get("usage")), + }, + ) + return result + + return FinancialAgentDependencies( + market_data_fetcher=logged_market_fetcher, + llm_caller=logged_llm_caller, + ) + + +def _summarise_call_log(call_log_path: str) -> JsonDict: + """Summarise the independent call receipts used by the recovery proof.""" + summary = { + "market_fetch_calls": 0, + "llm_calls": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + } + path = pathlib.Path(call_log_path) + if not path.exists(): + return summary + + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + try: + event = json.loads(line) + except json.JSONDecodeError as exc: + raise RuntimeError(f"Invalid call log entry at line {line_number}: {exc}") from exc + if event.get("event") == "market_fetch": + summary["market_fetch_calls"] += 1 + elif event.get("event") == "llm_call": + summary["llm_calls"] += 1 + usage = _normalise_llm_usage(event.get("usage")) + summary["prompt_tokens"] += usage["prompt_tokens"] + summary["completion_tokens"] += usage["completion_tokens"] + summary["total_tokens"] += usage["total_tokens"] + return summary + + +def _resolve_dependencies( + *, + offline: bool, + market_data_fetcher: Optional[MarketDataFetcher], + llm_caller: Optional[LLMCaller], + call_log_path: Optional[str] = None, +) -> FinancialAgentDependencies: + dependencies = FinancialAgentDependencies( + market_data_fetcher=( + market_data_fetcher or (_offline_market_data_fetcher if offline else _default_market_data_fetcher) + ), + llm_caller=(llm_caller or (_offline_llm_caller if offline else _live_llm_caller)), + ) + return _with_call_logging(dependencies, call_log_path) + + +def _kill_after_step(kill_at: Optional[int], step_index: int) -> None: + if kill_at == step_index: + if os.name == "nt": + os._exit(137) + import signal + + os.kill(os.getpid(), signal.SIGKILL) + + +def _build_langgraph_pipeline( + dependencies: Optional[FinancialAgentDependencies] = None, + *, + kill_at: Optional[int] = None, +): + """Build the real four-node async StateGraph used by every workflow run.""" + try: + from langgraph.graph import END, START, StateGraph + except ImportError as exc: + raise RuntimeError( + "LangGraph is required for this cookbook; install it with `python -m pip install -e '.[financial-agent]'`." + ) from exc + + deps = dependencies or _resolve_dependencies(offline=True, market_data_fetcher=None, llm_caller=None) + workflow = StateGraph(FinancialAgentState) + + async def fetch_market_data(state: FinancialAgentState) -> FinancialAgentState: + update = await async_step("fetch_market_data", _fetch_market_data_node, state, deps) + _kill_after_step(kill_at, 0) + return update + + async def compute_indicators(state: FinancialAgentState) -> FinancialAgentState: + update = await async_step("compute_indicators", _compute_indicators_node, state, deps) + _kill_after_step(kill_at, 1) + return update + + async def generate_investment_memo(state: FinancialAgentState) -> FinancialAgentState: + update = await async_step("generate_investment_memo", _generate_investment_memo_node, state, deps) + _kill_after_step(kill_at, 2) + return update + + async def generate_report(state: FinancialAgentState) -> FinancialAgentState: + update = await async_step("generate_report", _generate_report_node, state, deps) + _kill_after_step(kill_at, 3) + return update + + workflow.add_node("fetch_market_data", fetch_market_data) + workflow.add_node("compute_indicators", compute_indicators) + workflow.add_node("generate_investment_memo", generate_investment_memo) + workflow.add_node("generate_report", generate_report) + workflow.add_edge(START, "fetch_market_data") + workflow.add_edge("fetch_market_data", "compute_indicators") + workflow.add_edge("compute_indicators", "generate_investment_memo") + workflow.add_edge("generate_investment_memo", "generate_report") + workflow.add_edge("generate_report", END) + return workflow.compile() + + +def _with_legacy_aliases(state: FinancialAgentState) -> JsonDict: + """Expose the new names while preserving the existing cookbook result keys.""" + result: JsonDict = dict(state) + result["signals"] = state["indicators"] + result["thesis"] = state["investment_memo"] + result["report"] = state["final_report"] + return result + + +async def run_financial_analyst_async( + ticker: str = "AAPL", + wal_dir: str = WAL_DIR_DEFAULT, + kill_at: Optional[int] = None, + *, + model: str = DEFAULT_MODEL, + offline: bool = True, + market_data_fetcher: Optional[MarketDataFetcher] = None, + llm_caller: Optional[LLMCaller] = None, + call_log_path: Optional[str] = None, +) -> JsonDict: + """Execute the injectable four-node workflow inside ``@durable_async``.""" + ticker = _normalise_ticker(ticker) + dependencies = _resolve_dependencies( + offline=offline, + market_data_fetcher=market_data_fetcher, + llm_caller=llm_caller, + call_log_path=call_log_path, + ) + identity = _workflow_identity(ticker, model, offline) + run_dir = _workflow_wal_dir(wal_dir, ticker, model, offline) + + @durable_async(goal_id=f"{GOAL_ID}:{identity}", wal_dir=run_dir) + async def _execute() -> JsonDict: + initial_state: FinancialAgentState = { + "ticker": ticker, + "model": model, + "offline": offline, + } + graph = _build_langgraph_pipeline(dependencies, kill_at=kill_at) + final_state = await graph.ainvoke(initial_state) + return _with_legacy_aliases(final_state) + + return await _execute() + + +def run_financial_analyst( + ticker: str = "AAPL", + wal_dir: str = WAL_DIR_DEFAULT, + kill_at: Optional[int] = None, + *, + model: str = DEFAULT_MODEL, + offline: bool = True, + market_data_fetcher: Optional[MarketDataFetcher] = None, + llm_caller: Optional[LLMCaller] = None, + call_log_path: Optional[str] = None, +) -> JsonDict: + """Synchronous CLI/backward-compatible wrapper around the async workflow.""" + return asyncio.run( + run_financial_analyst_async( + ticker=ticker, + wal_dir=wal_dir, + kill_at=kill_at, + model=model, + offline=offline, + market_data_fetcher=market_data_fetcher, + llm_caller=llm_caller, + call_log_path=call_log_path, + ) + ) + + +async def _measure_in_memory_fast_forwards( + *, + wal_dir: str, + ticker: str, + model: str, + offline: bool, + step_ids: tuple[str, ...], +) -> dict[str, float]: + """Measure only cached ``async_step`` lookups after WAL initialization.""" + ticker = _normalise_ticker(ticker) + identity = _workflow_identity(ticker, model, offline) + run_dir = _workflow_wal_dir(wal_dir, ticker, model, offline) + + @durable_async(goal_id=f"{GOAL_ID}:{identity}", wal_dir=run_dir) + async def _measure() -> dict[str, float]: + async def must_not_execute() -> None: + raise AssertionError("A measured fast-forward unexpectedly executed its underlying function") + + durations_ms: dict[str, float] = {} + for step_id in step_ids: + started_ns = time.perf_counter_ns() + await async_step(step_id, must_not_execute) + durations_ms[step_id] = (time.perf_counter_ns() - started_ns) / 1_000_000 + return durations_ms + + return await _measure() + + +def demo_sigkill_recovery( + wal_dir: str = WAL_DIR_DEFAULT, + ticker: str = "NVDA", + *, + model: str = DEFAULT_MODEL, + offline: bool = True, +) -> JsonDict: + """Prove committed calls survive POSIX SIGKILL (or Windows exit 137) without repetition.""" + import shutil + + ticker = _normalise_ticker(ticker) + run_dir = _workflow_wal_dir(wal_dir, ticker, model, offline) + call_log_path = str(pathlib.Path(run_dir) / "external_calls.jsonl") + committed_steps = ("fetch_market_data", "compute_indicators", "generate_investment_memo") + if os.path.exists(run_dir): + shutil.rmtree(run_dir, ignore_errors=True) + + print(f"[demo] WAL directory: {run_dir}") + print(f"[demo] Independent call log: {call_log_path}") + print(f"[demo] 1) Launching {ticker}; SIGKILL follows the committed investment-memo step...") + + proc = subprocess.Popen( + [ + sys.executable, + "-c", + f""" +import pathlib, sys +sys.path.insert(0, {str(ROOT)!r}) +from examples.cookbooks.langgraph_financial_analyst import run_financial_analyst +run_financial_analyst( + ticker={ticker!r}, wal_dir={wal_dir!r}, kill_at=2, + model={model!r}, offline={offline!r}, call_log_path={call_log_path!r} +) +""", + ], + env={**os.environ, "PYTHONUNBUFFERED": "1", "PYTHONPATH": str(ROOT)}, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + child_stdout, child_stderr = proc.communicate(timeout=120 if not offline else 30) + except subprocess.TimeoutExpired as exc: + proc.kill() + proc.communicate() + raise RuntimeError("Demo subprocess did not reach the post-memo SIGKILL checkpoint") from exc + + expected_exit_codes = {137} if os.name == "nt" else {-9, 137} + if proc.returncode not in expected_exit_codes: + details = (child_stderr or child_stdout).strip() + raise RuntimeError(f"Demo subprocess exited with {proc.returncode}, not SIGKILL; output: {details[-1000:]}") + print(f"[demo] Subprocess received SIGKILL (pid={proc.pid}, exit={proc.returncode})") + + from orchestrator.state import load_state + + checkpoint = load_state(str(pathlib.Path(run_dir) / "state.json"), journal_dir=run_dir) + committed_before_resume = set(checkpoint.data.get("step_outputs", {})) + assert committed_before_resume == set(committed_steps), committed_before_resume + print(f"[demo] WAL committed steps: {', '.join(sorted(committed_before_resume))}; report pending") + + calls_before_resume = _summarise_call_log(call_log_path) + assert calls_before_resume["market_fetch_calls"] == 1, calls_before_resume + assert calls_before_resume["llm_calls"] == 1, calls_before_resume + assert calls_before_resume["total_tokens"] > 0, calls_before_resume + print( + "[demo] Before recovery: " + f"market fetch calls={calls_before_resume['market_fetch_calls']}, " + f"LLM calls={calls_before_resume['llm_calls']}, " + f"tokens={calls_before_resume['total_tokens']}" + ) + + fast_forward_ms = asyncio.run( + _measure_in_memory_fast_forwards( + wal_dir=wal_dir, + ticker=ticker, + model=model, + offline=offline, + step_ids=committed_steps, + ) + ) + for step_id in committed_steps: + duration_ms = fast_forward_ms[step_id] + target = "PASS" if duration_ms < 1.0 else "above target on this run" + print(f"[demo] {step_id} in-memory fast-forward: {duration_ms:.3f}ms ({target})") + + print("[demo] 2) Resuming only the unfinished report node from WAL...") + t0 = time.perf_counter() + result = run_financial_analyst( + ticker=ticker, + wal_dir=wal_dir, + kill_at=None, + model=model, + offline=offline, + call_log_path=call_log_path, + ) + overall_resume_ms = (time.perf_counter() - t0) * 1000 + + print(f"[demo] Overall resume (WAL load + graph + report): {overall_resume_ms:.2f}ms") + print(f"[demo] Report: {result['report']['summary']}") + + assert "report" in result, "Pipeline failed to produce final report upon resume" + assert result["report"]["status"] == "COMPLETED" + calls_after_resume = _summarise_call_log(call_log_path) + assert calls_after_resume == calls_before_resume, { + "before": calls_before_resume, + "after": calls_after_resume, + } + print( + "[demo] After recovery: " + f"market fetch calls={calls_after_resume['market_fetch_calls']}, " + f"LLM calls={calls_after_resume['llm_calls']}, " + f"tokens={calls_after_resume['total_tokens']}" + ) + + print("[demo] 3) Validating a second fully cached run...") + result2 = run_financial_analyst( + ticker=ticker, + wal_dir=wal_dir, + kill_at=None, + model=model, + offline=offline, + call_log_path=call_log_path, + ) + assert result2 == result, "Resumed state mismatch across warm runs" + calls_after_warm_run = _summarise_call_log(call_log_path) + assert calls_after_warm_run == calls_before_resume + print( + "[demo] SUCCESS: zero re-fetching, 0 duplicate LLM calls, " + f"0 duplicate tokens (saved {calls_before_resume['total_tokens']} tokens per replay)." + ) + result["recovery_proof"] = { + "calls_before_resume": calls_before_resume, + "calls_after_resume": calls_after_resume, + "calls_after_warm_run": calls_after_warm_run, + "committed_before_resume": sorted(committed_before_resume), + "in_memory_fast_forward_ms": fast_forward_ms, + "overall_resume_ms": round(overall_resume_ms, 6), + } + return result + + +def main() -> None: + parser = argparse.ArgumentParser(description="LangGraph Financial Analyst Cookbook") + parser.add_argument("--ticker", default="AAPL", help="Ticker symbol (default: AAPL)") + parser.add_argument("--model", default=DEFAULT_MODEL, help="Provider-prefixed LLM model") + parser.add_argument("--wal-dir", default=WAL_DIR_DEFAULT, help="WAL checkpoint directory") + parser.add_argument( + "--kill-at", type=int, choices=range(4), default=None, help="Send SIGKILL after durable step 0-3" + ) + parser.add_argument("--demo", action="store_true", help="Run end-to-end SIGKILL recovery demonstration") + mode = parser.add_mutually_exclusive_group() + mode.add_argument("--offline", dest="offline", action="store_true", help="Use deterministic local adapters") + mode.add_argument("--live", dest="offline", action="store_false", help="Use injected live adapters") + parser.set_defaults(offline=True) + args = parser.parse_args() + + if args.demo: + demo_sigkill_recovery( + wal_dir=args.wal_dir, + ticker=args.ticker, + model=args.model, + offline=args.offline, + ) + else: + out = run_financial_analyst( + ticker=args.ticker, + wal_dir=args.wal_dir, + kill_at=args.kill_at, + model=args.model, + offline=args.offline, + ) + print(f"Completed: {out['report']['summary']}") + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index 24996c1..a497eba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,6 +65,12 @@ dev = [ "pre-commit>=3.0", "fastapi>=0.100.0", "httpx>=0.24.0", + "langgraph>=1.2,<2", +] +financial-agent = [ + "certifi>=2024.0.0", + "langgraph>=1.2,<2", + "yfinance>=1.0,<2", ] openai = [ "openai>=1.0.0", diff --git a/tests/test_cookbooks.py b/tests/test_cookbooks.py index 2cc955f..5b767d0 100644 --- a/tests/test_cookbooks.py +++ b/tests/test_cookbooks.py @@ -1,30 +1,482 @@ -"""tests/test_cookbooks.py — Smoke tests for framework cookbook recipes in demo/mock mode.""" +"""Unit and regression tests for framework cookbooks to guarantee zero bit-rot in CI.""" -import os -import subprocess -import sys -from pathlib import Path +from __future__ import annotations +import asyncio +import json +import pathlib +from datetime import date, timedelta -def test_dspy_durable_optimize_mock_run(tmp_path): - """Test DSPy prompt optimization cookbook with simulated mock LLM.""" - cookbook_path = Path(__file__).resolve().parent.parent / "examples" / "cookbooks" / "dspy_durable_optimize.py" +import pytest + + +@pytest.mark.fast +def test_langgraph_financial_analyst_run_and_resume(tmp_path: pathlib.Path) -> None: + """Verify LangGraph financial analyst runs first time and fast-forwards on resume.""" + from examples.cookbooks.langgraph_financial_analyst import ( + DEFAULT_MODEL, + _workflow_wal_dir, + run_financial_analyst, + ) + + wal_dir = str(tmp_path / "wal_langgraph_analyst") + + # 1. Initial full run + r1 = run_financial_analyst(ticker="AAPL", wal_dir=wal_dir, kill_at=None) + assert "report" in r1 + assert r1["report"]["status"] == "COMPLETED" + assert "signals" in r1 + assert r1["signals"]["trend"] in ["BULLISH", "BEARISH"] + assert "market_data" in r1 + assert r1["market_data"]["ticker"] == "AAPL" + + # 2. Resumed run (must be identical) + r2 = run_financial_analyst(ticker="AAPL", wal_dir=wal_dir, kill_at=None) + assert r2 == r1 + + # 3. WAL persistence check + run_dir = _workflow_wal_dir(wal_dir, "AAPL", DEFAULT_MODEL, True) + wal_file = pathlib.Path(run_dir) / "state.wal.jsonl" + assert wal_file.exists() + assert "LILWAL02:" in wal_file.read_text(encoding="utf-8") + + +@pytest.mark.fast +def test_langgraph_financial_analyst_async_dependency_injection(tmp_path: pathlib.Path) -> None: + """The async workflow accepts fake external adapters and caches their JSON-safe outputs.""" + from examples.cookbooks.langgraph_financial_analyst import ( + _summarise_call_log, + _workflow_wal_dir, + run_financial_analyst_async, + ) + from orchestrator.state import load_state + + calls = {"market": 0, "llm": 0} + + async def fake_market_data_fetcher(ticker: str): + calls["market"] += 1 + return { + "ticker": ticker, + "source": "fake", + "close_prices": [100.0 + index for index in range(60)], + "current_price": 159.0, + "market_cap_b": 100.0, + "volume": 1_000, + "timestamp": 0.0, + } + + async def fake_llm_caller(market_data, indicators, model): + calls["llm"] += 1 + return { + "ticker": market_data["ticker"], + "text": "Test investment memo", + "recommendation": "HOLD", + "target_price": indicators["current_price"], + "risk_level": "MODERATE", + "rationale": "Deterministic fake memo for dependency-injection testing.", + "model": model, + "provider": "fake", + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + } + + kwargs = { + "ticker": "AAPL", + "wal_dir": str(tmp_path / "wal_injected"), + "model": "deepseek:test-model", + "offline": False, + "market_data_fetcher": fake_market_data_fetcher, + "llm_caller": fake_llm_caller, + "call_log_path": str(tmp_path / "external_calls.jsonl"), + } + first = asyncio.run(run_financial_analyst_async(**kwargs)) + calls_after_first = _summarise_call_log(kwargs["call_log_path"]) + second = asyncio.run(run_financial_analyst_async(**kwargs)) + calls_after_second = _summarise_call_log(kwargs["call_log_path"]) + + assert first == second + assert calls == {"market": 1, "llm": 1} + assert ( + calls_after_first + == calls_after_second + == { + "market_fetch_calls": 1, + "llm_calls": 1, + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + } + ) + assert first["market_data"]["source"] == "fake" + assert first["investment_memo"]["provider"] == "fake" + assert first["final_report"]["status"] == "COMPLETED" + assert first["signals"] == first["indicators"] + assert first["thesis"] == first["investment_memo"] + assert first["report"] == first["final_report"] + json.dumps(first) + + run_dir = _workflow_wal_dir(kwargs["wal_dir"], "AAPL", kwargs["model"], False) + state = load_state(str(pathlib.Path(run_dir) / "state.json"), journal_dir=run_dir) + memo_step = state.data["step_outputs"]["generate_investment_memo"] + assert memo_step["investment_memo"]["text"] == "Test investment memo" + assert memo_step["investment_memo"]["usage"]["total_tokens"] == 15 + + +@pytest.mark.fast +def test_deepseek_llm_prompt_and_usage_without_network(monkeypatch: pytest.MonkeyPatch) -> None: + """The live node sends every required signal once and preserves provider usage.""" + from examples.cookbooks import langgraph_financial_analyst as cookbook + + captured = {"calls": 0} + + def fake_call_llm(prompt, model, **kwargs): + captured.update({"calls": captured["calls"] + 1, "prompt": prompt, "model": model, "kwargs": kwargs}) + return { + "text": "Test investment memo", + "model": model, + "provider": "deepseek", + "usage": {"prompt_tokens": 120, "completion_tokens": 30, "total_tokens": 150}, + } + + monkeypatch.setattr(cookbook, "call_llm", fake_call_llm) + market_data = { + "ticker": "AAPL", + "currency": "USD", + "current_price": 200.0, + "market_cap": 3_000_000_000_000, + "trailing_pe": 25.0, + "forward_pe": 22.0, + "revenue_growth": 0.08, + "profit_margin": 0.21, + } + indicators = { + "sma20": 198.0, + "sma50": 190.0, + "rsi14": 61.5, + "macd": 2.4, + "macd_signal": 2.0, + "macd_histogram": 0.4, + } + + result = asyncio.run(cookbook._live_llm_caller(market_data, indicators, "deepseek:deepseek-v4-flash")) + + assert captured["calls"] == 1 + assert captured["model"] == "deepseek:deepseek-v4-flash" + for required_text in ("AAPL", "sma20", "sma50", "rsi14", "macd", "macd_signal", "macd_histogram"): + assert required_text in captured["prompt"] + assert captured["kwargs"]["temperature"] == pytest.approx(0.2) + assert result == { + "ticker": "AAPL", + "text": "Test investment memo", + "model": "deepseek:deepseek-v4-flash", + "provider": "deepseek", + "usage": {"prompt_tokens": 120, "completion_tokens": 30, "total_tokens": 150}, + } + json.dumps(result) + + +@pytest.mark.fast +def test_deepseek_failure_names_ticker_model_and_provider_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Provider failures should be actionable without exposing prompt or credentials.""" + from examples.cookbooks import langgraph_financial_analyst as cookbook + + def failed_call_llm(prompt, model, **kwargs): + del prompt, model, kwargs + raise RuntimeError("provider unavailable") + + monkeypatch.setattr(cookbook, "call_llm", failed_call_llm) + market_data = {"ticker": "AAPL", "current_price": 200.0} + indicators = { + "sma20": 198.0, + "sma50": 190.0, + "rsi14": 61.5, + "macd": 2.4, + "macd_signal": 2.0, + "macd_histogram": 0.4, + } + + with pytest.raises( + RuntimeError, + match=r"LLM request failed for AAPL using deepseek:deepseek-v4-flash: provider unavailable", + ): + asyncio.run(cookbook._live_llm_caller(market_data, indicators, "deepseek:deepseek-v4-flash")) + + +@pytest.mark.fast +def test_financial_indicators_known_uptrend_values() -> None: + """SMA, Wilder RSI, and MACD should match deterministic reference values.""" + from examples.cookbooks.langgraph_financial_analyst import _calculate_technical_signals + + result = _calculate_technical_signals( + { + "ticker": "TEST", + "close_prices": [float(value) for value in range(1, 61)], + "current_price": 60.0, + } + ) + + assert result["sma20"] == pytest.approx(50.5) + assert result["sma50"] == pytest.approx(35.5) + assert result["rsi14"] == pytest.approx(100.0) + assert result["macd"] == pytest.approx(6.866964) + assert result["macd_signal"] == pytest.approx(6.804976) + assert result["macd_histogram"] == pytest.approx(0.061989) + assert result["trend"] == "BULLISH" + + +@pytest.mark.fast +def test_financial_indicators_flat_prices_and_invalid_input() -> None: + """Flat prices are neutral; empty or undersized histories fail clearly.""" + from examples.cookbooks.langgraph_financial_analyst import _calculate_technical_signals + + flat = _calculate_technical_signals({"ticker": "FLAT", "close_prices": [100.0] * 60, "current_price": 100.0}) + assert flat["sma20"] == pytest.approx(100.0) + assert flat["sma50"] == pytest.approx(100.0) + assert flat["rsi14"] == pytest.approx(50.0) + assert flat["macd"] == pytest.approx(0.0) + assert flat["macd_signal"] == pytest.approx(0.0) + assert flat["macd_histogram"] == pytest.approx(0.0) + + falling = _calculate_technical_signals( + {"ticker": "DOWN", "close_prices": [float(value) for value in range(60, 0, -1)], "current_price": 1.0} + ) + assert falling["rsi14"] == pytest.approx(0.0) + + with pytest.raises(ValueError, match="At least 50"): + _calculate_technical_signals({"ticker": "EMPTY", "close_prices": []}) + with pytest.raises(ValueError, match="At least 50"): + _calculate_technical_signals({"ticker": "SHORT", "close_prices": [100.0] * 49}) + + +@pytest.mark.fast +def test_yfinance_fetch_normalises_json_and_skips_non_finite_values() -> None: + """The live adapter accepts pandas-like objects without leaking their scalar types.""" + from examples.cookbooks.langgraph_financial_analyst import _fetch_market_data + + class NumpyLikeFloat: + def __init__(self, value: float): + self.value = float(value) + + def __float__(self): + return self.value + + class FakeCloseSeries: + def items(self): + values = [NumpyLikeFloat(100.0 + index) for index in range(60)] + values.extend([float("nan"), float("inf")]) + first_day = date(2025, 1, 1) + return [(first_day + timedelta(days=index), value) for index, value in enumerate(values)] + + class FakeHistory: + empty = False + + def __getitem__(self, column): + assert column == "Close" + return FakeCloseSeries() + + class FakeTicker: + def __init__(self, ticker): + assert ticker == "AAPL" + self.history_kwargs = None + self.info = { + "currency": "USD", + "trailingPE": float("nan"), + "forwardPE": NumpyLikeFloat(22.5), + "revenueGrowth": None, + "profitMargins": NumpyLikeFloat(0.21), + } + self.fast_info = { + "last_price": NumpyLikeFloat(159.5), + "market_cap": NumpyLikeFloat(3_000_000_000), + "last_volume": NumpyLikeFloat(42_000_000), + } + + def history(self, **kwargs): + self.history_kwargs = kwargs + return FakeHistory() + + class FakeYFinance: + last_ticker = None + + @classmethod + def Ticker(cls, ticker): + cls.last_ticker = FakeTicker(ticker) + return cls.last_ticker + + result = _fetch_market_data(" aapl ", yf_module=FakeYFinance) + + assert FakeYFinance.last_ticker.history_kwargs == { + "period": "1y", + "interval": "1d", + "auto_adjust": True, + "actions": False, + "timeout": 15, + } + assert result["ticker"] == "AAPL" + assert result["source"] == "yfinance" + assert result["currency"] == "USD" + assert result["current_price"] == pytest.approx(159.5) + assert result["market_cap"] == 3_000_000_000 + assert result["trailing_pe"] is None + assert result["forward_pe"] == pytest.approx(22.5) + assert result["revenue_growth"] is None + assert result["profit_margin"] == pytest.approx(0.21) + assert len(result["dates"]) == len(result["close_prices"]) == 60 + assert all(isinstance(price, float) for price in result["close_prices"]) + json.dumps(result) + + +@pytest.mark.fast +@pytest.mark.parametrize("failure", ["empty", "network"]) +def test_yfinance_fetch_reports_empty_history_and_network_errors(failure: str) -> None: + from examples.cookbooks.langgraph_financial_analyst import MarketDataError, _fetch_market_data + + class FakeTicker: + def history(self, **kwargs): + del kwargs + if failure == "network": + raise OSError("network unavailable") + return type("EmptyHistory", (), {"empty": True})() + + class FakeYFinance: + @staticmethod + def Ticker(ticker): + del ticker + return FakeTicker() + + expected = "Unable to fetch" if failure == "network" else "no historical prices" + with pytest.raises(MarketDataError, match=expected): + _fetch_market_data("AAPL", yf_module=FakeYFinance) + + +@pytest.mark.integration +def test_langgraph_financial_analyst_sigkill_recovery(tmp_path: pathlib.Path) -> None: + """A post-memo SIGKILL must not repeat market, LLM, or token consumption.""" + from examples.cookbooks.langgraph_financial_analyst import demo_sigkill_recovery + + wal_dir = str(tmp_path / "wal_langgraph_demo") + result = demo_sigkill_recovery(wal_dir=wal_dir, ticker="NVDA") + assert "report" in result + assert result["report"]["status"] == "COMPLETED" + assert result["ticker"] == "NVDA" + proof = result["recovery_proof"] + expected_calls = proof["calls_before_resume"] + assert proof["calls_after_resume"] == expected_calls + assert proof["calls_after_warm_run"] == expected_calls + assert expected_calls["market_fetch_calls"] == 1 + assert expected_calls["llm_calls"] == 1 + assert expected_calls["total_tokens"] > 0 + assert proof["committed_before_resume"] == [ + "compute_indicators", + "fetch_market_data", + "generate_investment_memo", + ] + assert set(proof["in_memory_fast_forward_ms"]) == { + "fetch_market_data", + "compute_indicators", + "generate_investment_memo", + } + # Functional correctness is strict (the trap callback must never execute); + # CI timing is intentionally generous to avoid failures from noisy runners. + assert all(duration_ms < 25.0 for duration_ms in proof["in_memory_fast_forward_ms"].values()) + + +@pytest.mark.fast +def test_financial_agent_tickers_use_distinct_wal_caches(tmp_path: pathlib.Path) -> None: + """AAPL checkpoints must never satisfy an NVDA workflow.""" + from examples.cookbooks.langgraph_financial_analyst import ( + _workflow_wal_dir, + run_financial_analyst_async, + ) + + calls = {"market": 0, "llm": 0} + + async def fake_market_data_fetcher(ticker: str): + calls["market"] += 1 + return { + "ticker": ticker, + "source": "fake", + "close_prices": [100.0 + index for index in range(60)], + "current_price": 159.0, + } + + async def fake_llm_caller(market_data, indicators, model): + del indicators + calls["llm"] += 1 + return { + "ticker": market_data["ticker"], + "text": f"Memo for {market_data['ticker']}", + "model": model, + "provider": "fake", + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}, + } + + wal_dir = str(tmp_path / "ticker_wal") + model = "deepseek:test-model" + common = { + "wal_dir": wal_dir, + "model": model, + "offline": False, + "market_data_fetcher": fake_market_data_fetcher, + "llm_caller": fake_llm_caller, + } + aapl = asyncio.run(run_financial_analyst_async(ticker="AAPL", **common)) + nvda = asyncio.run(run_financial_analyst_async(ticker="NVDA", **common)) + + assert aapl["ticker"] == aapl["market_data"]["ticker"] == "AAPL" + assert nvda["ticker"] == nvda["market_data"]["ticker"] == "NVDA" + assert calls == {"market": 2, "llm": 2} + assert _workflow_wal_dir(wal_dir, "AAPL", model, False) != _workflow_wal_dir(wal_dir, "NVDA", model, False) + + +@pytest.mark.integration +def test_dspy_durable_optimize_mock_run(tmp_path: pathlib.Path) -> None: + """Preserve the upstream subprocess smoke test for the completed DSPy cookbook.""" + import os + import subprocess + import sys + + cookbook_path = ( + pathlib.Path(__file__).resolve().parent.parent / "examples" / "cookbooks" / "dspy_durable_optimize.py" + ) assert cookbook_path.exists(), f"Cookbook not found: {cookbook_path}" - wal_dir = tmp_path / "dspy_wal" + wal_dir = tmp_path / "dspy_wal_subprocess" wal_dir.mkdir() - env = os.environ.copy() - env["PYTHONPATH"] = str(Path(__file__).resolve().parent.parent) + env["PYTHONPATH"] = str(pathlib.Path(__file__).resolve().parent.parent) env["LETITLOOP_WAL_DIR"] = str(wal_dir) env["DSPY_DEMO_MODE"] = "1" - res = subprocess.run( + result = subprocess.run( [sys.executable, str(cookbook_path), "--demo", "--wal-dir", str(wal_dir)], env=env, capture_output=True, text=True, timeout=30, ) - assert res.returncode == 0, f"Cookbook failed with returncode {res.returncode}:\n{res.stderr}" - assert "SUCCESS: DSPy prompt optimizer recovered" in res.stdout or "Optimization Result:" in res.stdout + assert result.returncode == 0, f"Cookbook failed with returncode {result.returncode}:\n{result.stderr}" + assert "SUCCESS: DSPy prompt optimizer recovered" in result.stdout or "Optimization Result:" in result.stdout + + +@pytest.mark.fast +def test_cookbook_module_builders() -> None: + """Verify the financial cookbook constructs the required graph topology.""" + from examples.cookbooks.langgraph_financial_analyst import _build_langgraph_pipeline + + lg_pipeline = _build_langgraph_pipeline() + graph = lg_pipeline.get_graph() + assert set(graph.nodes) == { + "__start__", + "fetch_market_data", + "compute_indicators", + "generate_investment_memo", + "generate_report", + "__end__", + } + assert {(edge.source, edge.target) for edge in graph.edges} == { + ("__start__", "fetch_market_data"), + ("fetch_market_data", "compute_indicators"), + ("compute_indicators", "generate_investment_memo"), + ("generate_investment_memo", "generate_report"), + ("generate_report", "__end__"), + }