core — Graph, Data & LLM Infrastructure

🎯 Why This Module Exists

Without core infrastructure, every agent would need to independently manage LLM connections, data fetching, state management, and CLI parsing. By centralizing these concerns, the project achieves consistency and extensibility. The graph/ submodule defines AgentState — a TypedDict with three annotated fields (messages, data, metadata) that serves as the universal contract all agents read from and write to. The llm/ submodule abstracts away provider differences through a ModelProvider enum and LLMModel Pydantic class, so switching from OpenAI to Groq or Anthropic is a one-line change in configuration. The tools/ submodule provides a single financial data gateway with automatic rate-limit handling and retry logic. The data/ submodule handles Pydantic models and an in-memory cache with intelligent merge-on-write deduplication. This separation means adding a new agent requires zero infrastructure changes — just import AgentState, call get_model(), and use the API tools.

🚶 Step-by-Step Narrated Walkthrough

Follow a single agent requesting AAPL price data and generating an LLM analysis as it flows through the core infrastructure:

1. Graph Entry — The StateGraph dispatches the agent function with an AgentState containing messages, data (tickers, portfolio, date range), and metadata (model config).
2. State Read — The agent reads state["data"]["tickers"] to discover it needs to analyze AAPL. The Annotated[dict, merge_dicts] reducer ensures parallel agents merge cleanly.
3. Price Fetch — Agent calls get_prices("AAPL", start_date, end_date) in tools/api.py. The function builds a cache key: "AAPL_2025-01-01_2025-12-31".
4. Cache Lookup — Cache.get_prices(cache_key) checks the in-memory _prices_cache dict. On a hit, it returns immediately — no HTTP call needed.
5. API Request — On a cache miss, _make_api_request() sends a GET to api.financialdatasets.ai with the X-API-KEY header. If the API returns HTTP 429, it retries with linear backoff (60s, 90s, 120s).
6. Response Parsing — The raw JSON is validated through the PriceResponse Pydantic model, producing a list of Price objects.
7. Cache Write — Cache.set_prices() merges new data with any existing entries using the time field as a deduplication key — O(n) merge with O(1) key lookup.
8. LLM Model Resolution — Agent calls get_model(model_name, model_provider). The function matches the ModelProvider enum and instantiates the correct LangChain chat class (e.g., ChatOpenAI).
9. LLM Invocation — The agent sends a structured prompt to the resolved LLM. LLMModel.has_json_mode() determines whether to request JSON output directly or parse free-text.
10. State Write — The agent writes its analysis to state["data"]["analyst_signals"]. The merge_dicts reducer merges this with signals from other parallel agents.
11. Debug Output — show_agent_reasoning(output, agent_name) serializes the result to formatted JSON for console debugging, handling Pandas objects, custom classes, and raw strings.

Cache Merge-on-Write Deduplication

The Problem: Multiple agents may request overlapping date ranges for the same ticker. If AAPL data for Jan-Jun is cached and an agent requests Mar-Sep, the new fetch returns Mar-Sep data that partially overlaps with the cache. Without deduplication, downstream agents would see duplicate data points that distort moving averages, volatility calculations, and other time-series analyses.

The Approach: Cache._merge_data() builds a set of existing keys (O(n) construction, O(1) lookup per item), then appends only items whose key field is not already present. The key field varies by data type: time for prices, report_period for financials, filing_date for insider trades, and date for company news. The method is generic — it accepts a key_field parameter, so adding a new data type requires no changes to the merge logic.

Why This Algorithm: A naive append would cause duplicate entries that skew analysis. A full replacement would lose earlier data outside the new range. The merge approach preserves all unique data points with O(n+m) time complexity where n is existing items and m is new items.

flowchart TD
    A["New API data arrives\n(m items)"] --> B["Build existing_keys set\nfrom cached data (n items)"]
    B --> C{"For each new item:\nkey in existing_keys?"}
    C -->|"Yes"| D["Skip\n(already cached)"]
    C -->|"No"| E["Append to merged list"]
    D --> F["Return merged list\n(no duplicates)"]
    E --> F

Hot Paths

The most performance-sensitive code paths in this module:

• Cache lookup (get_prices, get_financial_metrics): O(1) dict lookup by cache key. This is the fast path that avoids network I/O entirely.
• Cache merge (_merge_data): O(n+m) where n = existing items, m = new items. The set construction dominates; iteration is linear. Called once per API response.
• get_model() dispatch: O(1) provider enum match. Instantiates LangChain client objects which cache underlying HTTP sessions.

Resource Usage

• Memory: Cache stores raw dicts (not Pydantic objects) to minimize memory. A typical run with 5 tickers and 1 year of daily prices uses approximately 2-5 MB of cache memory.
• Network I/O: API calls to financialdatasets.ai are the primary bottleneck (200-800ms per call). Cache hits eliminate this entirely. Rate-limited retries add 60-120s delays.
• LLM Latency: Model instantiation is fast (~10ms); inference latency depends on the provider (1-30s per call). This is external to core infrastructure but dominates end-to-end time.