Spaces:
Sleeping
Sleeping
| import os | |
| from typing import Any | |
| from langchain_core.tools import tool | |
| from agent.evidence import evidence_envelope, make_evidence_record, parse_evidence_envelope | |
| from storage import metrics_db, news_cache | |
| TOP_K_FILINGS = 5 | |
| TOP_K_TRANSCRIPTS = 4 | |
| _tavily_client: Any = None | |
| class _LazyVectorStore: | |
| """Keep heavy vector dependencies lazy while preserving a patchable seam.""" | |
| def search(*args, **kwargs): | |
| from storage import vector_store as implementation | |
| return implementation.search(*args, **kwargs) | |
| vector_store = _LazyVectorStore() | |
| def TavilyClient(*args, **kwargs): | |
| """Lazily construct Tavily's client; kept as a named dependency seam.""" | |
| from tavily import TavilyClient as implementation | |
| return implementation(*args, **kwargs) | |
| def _empty(tool_name: str, query: dict, message: str) -> str: | |
| return evidence_envelope(tool=tool_name, query=query, status="EMPTY", message=message) | |
| def _error(tool_name: str, query: dict, exc: Exception) -> str: | |
| return evidence_envelope( | |
| tool=tool_name, | |
| query=query, | |
| status="ERROR", | |
| message=f"{type(exc).__name__}: {exc}", | |
| error_code="TOOL_EXECUTION_ERROR", | |
| ) | |
| def _vector_record(result: dict, fallback_source: str): | |
| metadata = dict(result.get("metadata") or {}) | |
| source = metadata.get("source") or fallback_source | |
| metadata.setdefault( | |
| "chunk_context", | |
| f"Source: {source} | Section: {metadata.get('section', '')} | Date: {metadata.get('filing_date') or metadata.get('date') or ''}", | |
| ) | |
| return make_evidence_record( | |
| source=source, | |
| content=result.get("text", ""), | |
| document_id=str(metadata.get("document_id") or ":".join(filter(None, [ | |
| source, | |
| str(metadata.get("ticker", "")), | |
| str(metadata.get("period", "")), | |
| str(metadata.get("filing_date") or metadata.get("date") or ""), | |
| str(metadata.get("section", "")), | |
| ]))), | |
| chunk_id=str(metadata.get("chunk_id")) if metadata.get("chunk_id") else None, | |
| source_url=metadata.get("source_url"), | |
| as_of=metadata.get("filing_date") or metadata.get("date"), | |
| metadata=metadata, | |
| ) | |
| def _get_tavily_client() -> Any: | |
| global _tavily_client | |
| if _tavily_client is None: | |
| _tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"]) | |
| return _tavily_client | |
| def _fmt_billions(v) -> str: | |
| return f"${v / 1e9:.1f}B" if v is not None else "N/A" | |
| def _fmt_pct(v) -> str: | |
| return f"{v:.1%}" if v is not None else "N/A" | |
| def _fmt_millions(v) -> str: | |
| return f"${v / 1e6:.0f}M" if v is not None else "N/A" | |
| def _fmt_row(data: dict) -> str: | |
| yoy = f" (YoY: {data['revenue_yoy_pct']:+.1f}%)" if data.get("revenue_yoy_pct") is not None else "" | |
| lines = [ | |
| f" Filing: {data.get('form_type', '')} {data.get('period', '')} | Date: {data['filing_date']}", | |
| f" Revenue: {_fmt_billions(data['revenue'])}{yoy}", | |
| f" EPS (diluted): {data['eps']:.2f}" if data.get("eps") else " EPS: N/A", | |
| f" Gross Margin: {_fmt_pct(data['gross_margin'])}", | |
| f" Operating Margin: {_fmt_pct(data['operating_margin'])}", | |
| f" Free Cash Flow: {_fmt_billions(data['free_cash_flow'])}", | |
| ] | |
| # Valuation parameter inputs | |
| shares = data.get("shares_diluted") | |
| lines.append(f" Shares Outstanding (diluted): {shares / 1e6:.0f}M" if shares else " Shares Outstanding: N/A") | |
| lines.append(f" Effective Tax Rate: {_fmt_pct(data.get('effective_tax_rate'))}") | |
| lines.append(f" Interest Expense: {_fmt_millions(data.get('interest_expense'))}") | |
| lines.append(f" Total Debt (LT): {_fmt_billions(data.get('total_debt'))}") | |
| lines.append(f" Capex: {_fmt_millions(data.get('capex'))}") | |
| lines.append(f" Dividends Paid: {_fmt_millions(data.get('dividends_paid'))}") | |
| lines.append(f" Share Buybacks: {_fmt_millions(data.get('buybacks'))}") | |
| equity = data.get("stockholders_equity") | |
| debt = data.get("total_debt") | |
| de_ratio = f" (D/E: {debt / equity:.2f}x)" if (debt and equity and equity > 0) else "" | |
| lines.append(f" Stockholders' Equity: {_fmt_billions(equity)}{de_ratio}") | |
| if data.get("guidance_text"): | |
| lines.append(f" Guidance: {data['guidance_text']}") | |
| warnings = data.get("quality_warnings") or [] | |
| contexts = data.get("metric_contexts") or {} | |
| lineage = [] | |
| if isinstance(contexts, dict): | |
| for metric, context in sorted(contexts.items()): | |
| if not isinstance(context, dict): | |
| continue | |
| source = context.get("source") or "derived" | |
| selection = context.get("selection") or "unspecified" | |
| concept = context.get("concept") | |
| detail = f"{metric}={source}/{selection}" | |
| if concept: | |
| detail += f"/{concept}" | |
| lineage.append(detail) | |
| lines.extend([ | |
| f" Data Quality: {data.get('data_quality_status') or 'LEGACY_UNVERIFIED'}", | |
| f" Accession: {data.get('accession') or 'N/A'} | Report Date: {data.get('report_date') or 'N/A'}", | |
| f" Quality Warnings: {', '.join(str(item) for item in warnings) if warnings else 'none'}", | |
| f" Metric Lineage: {'; '.join(lineage) if lineage else 'not recorded'}", | |
| ]) | |
| return "\n".join(lines) | |
| def get_financial_metrics(ticker: str) -> str: | |
| """Retrieve structured financial metrics for a ticker across all ingested periods. | |
| Always call this first to anchor quantitative claims. Returns an evidence.v1 | |
| JSON envelope; cite a record by copying its `ref` into `evidence_ref`.""" | |
| query = {"ticker": ticker.upper()} | |
| try: | |
| rows = metrics_db.get_all_metrics(ticker) | |
| except Exception as exc: | |
| return _error("get_financial_metrics", query, exc) | |
| if not rows: | |
| return _empty( | |
| "get_financial_metrics", | |
| query, | |
| f"Ticker {ticker.upper()} not ingested. Run: python ingest.py {ticker.upper()}", | |
| ) | |
| usable_rows = [ | |
| row for row in rows | |
| if row.get("data_quality_status") in {"VERIFIED", "CHECK_REQUIRED"} | |
| ] | |
| if not usable_rows: | |
| statuses = sorted({ | |
| str(row.get("data_quality_status") or "LEGACY_UNVERIFIED") | |
| for row in rows | |
| }) | |
| return _empty( | |
| "get_financial_metrics", | |
| query, | |
| ( | |
| f"Metrics exist for {ticker.upper()}, but none carries SEC period lineage " | |
| f"(statuses: {', '.join(statuses)}). Run: python ingest.py " | |
| f"{ticker.upper()} --full to restore citable metrics." | |
| ), | |
| ) | |
| try: | |
| records = [] | |
| for row in usable_rows: | |
| content = f"Company: {row['company_name']} ({row['ticker']})\n{_fmt_row(row)}" | |
| period = str(row.get("period") or "unknown") | |
| filing_date = str(row.get("filing_date") or "") | |
| accession = str(row.get("accession") or "") | |
| records.append(make_evidence_record( | |
| source="metrics", | |
| content=content, | |
| document_id=( | |
| f"metrics:{ticker.upper()}:{accession}" | |
| if accession else f"metrics:{ticker.upper()}:{period}:{filing_date}" | |
| ), | |
| source_url=row.get("source_url") or None, | |
| as_of=filing_date or None, | |
| metadata={ | |
| "ticker": ticker.upper(), | |
| "company_name": row.get("company_name"), | |
| "period": period, | |
| "filing_date": filing_date, | |
| "form_type": row.get("form_type"), | |
| "period_basis": row.get("period_basis", "unknown"), | |
| "report_date": row.get("report_date"), | |
| "accession": row.get("accession"), | |
| "source_url": row.get("source_url"), | |
| "metric_contexts": row.get("metric_contexts") or {}, | |
| "quality_warnings": row.get("quality_warnings") or [], | |
| "data_quality_status": row.get("data_quality_status") or "LEGACY_UNVERIFIED", | |
| }, | |
| )) | |
| except Exception as exc: | |
| return _error("get_financial_metrics", query, exc) | |
| return evidence_envelope(tool="get_financial_metrics", query=query, records=records) | |
| def search_filing(query: str, ticker: str, since: str = "", period: str = "") -> str: | |
| """Search SEC filings (10-Q and 10-K) for relevant text using semantic search. | |
| Use for MD&A, risk factors, and outlook sections. | |
| Use `since` (YYYY-MM-DD) to restrict results to filings after that date. | |
| Use `period` (e.g. 'Q12024', 'FY2023') to fetch chunks from a specific past period for cross-year/cross-quarter comparison. | |
| Returns an evidence.v1 JSON envelope; copy the supporting record's `ref` exactly.""" | |
| request = {"query": query, "ticker": ticker.upper(), "since": since, "period": period} | |
| try: | |
| results = vector_store.search( | |
| "filings", query, ticker, n_results=TOP_K_FILINGS, | |
| min_filing_date=since if since else None, | |
| period=period if period else None, | |
| ) | |
| except Exception as exc: | |
| return _error("search_filing", request, exc) | |
| if not results: | |
| return _empty( | |
| "search_filing", | |
| request, | |
| f"No filing data for {ticker.upper()} matching query (period={period or 'any'}). Run: python ingest.py {ticker.upper()}", | |
| ) | |
| try: | |
| records = [_vector_record(result, "10-Q") for result in results] | |
| except Exception as exc: | |
| return _error("search_filing", request, exc) | |
| return evidence_envelope(tool="search_filing", query=request, records=records) | |
| def search_transcript(query: str, ticker: str, since: str = "", period: str = "") -> str: | |
| """Search earnings call transcripts for management commentary. | |
| Use for CEO/CFO tone, forward guidance, and analyst Q&A themes. | |
| Use `since` (YYYY-MM-DD) to restrict results to transcripts after that date. | |
| Use `period` (e.g. 'Q12024', 'FY2023') to fetch chunks from a specific past period for cross-year/cross-quarter comparison. | |
| Returns an evidence.v1 JSON envelope; copy the supporting record's `ref` exactly.""" | |
| request = {"query": query, "ticker": ticker.upper(), "since": since, "period": period} | |
| try: | |
| results = vector_store.search( | |
| "transcripts", query, ticker, n_results=TOP_K_TRANSCRIPTS, | |
| min_filing_date=since if since else None, | |
| period=period if period else None, | |
| ) | |
| except Exception as exc: | |
| return _error("search_transcript", request, exc) | |
| if not results: | |
| return _empty( | |
| "search_transcript", | |
| request, | |
| f"No transcript for {ticker.upper()} matching query (period={period or 'any'}). It may not have been ingested yet.", | |
| ) | |
| try: | |
| records = [_vector_record(result, "transcript") for result in results] | |
| except Exception as exc: | |
| return _error("search_transcript", request, exc) | |
| return evidence_envelope(tool="search_transcript", query=request, records=records) | |
| def _fmt_revision_signal(pct: float) -> str: | |
| if pct >= 1.0: | |
| return "positive momentum" | |
| if pct <= -1.0: | |
| return "negative momentum" | |
| return "stable" | |
| def get_analyst_expectations(ticker: str) -> str: | |
| """Fetch analyst consensus EPS / revenue estimates, 30-day estimate revision %, | |
| and post-earnings d1/d5/since-release stock price reaction. Use to compare reported | |
| actuals against market expectations and gauge market reception of the latest filing. | |
| Returns an evidence.v1 JSON envelope with explicit period/event alignment flags.""" | |
| from ingestion.analyst import fetch_analyst_estimates, fetch_price_reaction | |
| request = {"ticker": ticker.upper()} | |
| try: | |
| rows = metrics_db.get_all_metrics(ticker) | |
| latest_filing_date = rows[0]["filing_date"] if rows else None | |
| target_period = rows[0].get("period") if rows else None | |
| est, est_err = fetch_analyst_estimates(ticker, target_period=target_period) | |
| price, price_err = (fetch_price_reaction(ticker, latest_filing_date) | |
| if latest_filing_date else (None, "no filing date in metrics_db")) | |
| except Exception as exc: | |
| return _error("get_analyst_expectations", request, exc) | |
| if est is None and price is None: | |
| return _empty( | |
| "get_analyst_expectations", | |
| request, | |
| f"No analyst data available for {ticker.upper()}. Errors: {est_err}; {price_err}", | |
| ) | |
| # Use exact schema field names so the LLM can copy values verbatim without renaming. | |
| lines = [f"Analyst expectations for {ticker.upper()} (field names match MarketExpectations schema):"] | |
| if est: | |
| estimates_aligned = bool(est.get("period_aligned") and est.get("comparison_allowed")) | |
| eps = est.get("consensus_eps_est") if estimates_aligned else None | |
| rev = est.get("consensus_rev_est") if estimates_aligned else None | |
| rev30 = est.get("estimate_revision_30d_pct") if estimates_aligned else None | |
| lines.append(f" consensus_eps_est: {eps:.4f}" if eps is not None else " consensus_eps_est: null") | |
| lines.append( | |
| f" consensus_rev_est_bn: {rev / 1e9:.4f}" # already converted to billions | |
| if rev is not None else " consensus_rev_est_bn: null" | |
| ) | |
| lines.append( | |
| f" revision_30d_pct: {rev30:.4f} # signal: {_fmt_revision_signal(rev30)}" | |
| if rev30 is not None else " revision_30d_pct: null" | |
| ) | |
| lines.append(f" target_period: {est.get('target_period') or 'null'}") | |
| lines.append(f" provider_period_codes: {est.get('provider_period_codes') or {}}") | |
| lines.append(f" period_aligned: {str(bool(est.get('period_aligned'))).lower()}") | |
| lines.append(f" comparison_allowed: {str(bool(est.get('comparison_allowed'))).lower()}") | |
| lines.append(f" alignment_status: {est.get('alignment_status') or 'UNVERIFIED'}") | |
| else: | |
| lines.append(f" consensus_eps_est: null # unavailable: {est_err}") | |
| lines.append(" consensus_rev_est_bn: null") | |
| lines.append(" revision_30d_pct: null") | |
| if price: | |
| event_allowed = bool(price.get("event_aligned") and price.get("comparison_allowed")) | |
| d1 = price.get("d1_pct") if event_allowed else None | |
| d5 = price.get("d5_pct") if event_allowed else None | |
| since = price.get("since_release_pct") if event_allowed else None | |
| lines.append(f" d1_price_reaction_pct: {d1:.4f}" if d1 is not None else " d1_price_reaction_pct: null") | |
| lines.append(f" d5_price_reaction_pct: {d5:.4f}" if d5 is not None else " d5_price_reaction_pct: null") | |
| lines.append(f" since_release_price_reaction_pct: {since:.4f}" if since is not None else " since_release_price_reaction_pct: null") | |
| lines.append(f" event_date: {price.get('event_date') or 'null'}") | |
| lines.append(f" event_kind: {price.get('event_kind') or 'unknown'}") | |
| lines.append(f" event_timing: {price.get('event_timing') or 'unknown'}") | |
| lines.append(f" event_aligned: {str(bool(price.get('event_aligned'))).lower()}") | |
| lines.append(f" event_comparison_allowed: {str(bool(price.get('comparison_allowed'))).lower()}") | |
| lines.append(f" price_alignment_status: {price.get('alignment_status') or 'UNVERIFIED'}") | |
| else: | |
| lines.append(f" d1_price_reaction_pct: null # unavailable: {price_err}") | |
| lines.append(" d5_price_reaction_pct: null") | |
| lines.append(" since_release_price_reaction_pct: null") | |
| try: | |
| content = "\n".join(lines) | |
| record = make_evidence_record( | |
| source="analyst", | |
| content=content, | |
| document_id=f"analyst:{ticker.upper()}:{latest_filing_date or 'current'}", | |
| as_of=(est or {}).get("as_of") or latest_filing_date, | |
| metadata={ | |
| "ticker": ticker.upper(), | |
| "latest_filing_date": latest_filing_date, | |
| "target_period": target_period, | |
| "estimate_error": est_err, | |
| "price_error": price_err, | |
| }, | |
| ) | |
| except Exception as exc: | |
| return _error("get_analyst_expectations", request, exc) | |
| return evidence_envelope(tool="get_analyst_expectations", query=request, records=[record]) | |
| def search_news(query: str, ticker: str, days: int = 30) -> str: | |
| """Search for recent news about a company using Tavily. | |
| Use to find events after the most recent filing date. | |
| Args: | |
| query: The search query (e.g., "earnings", "acquisition") | |
| ticker: Stock ticker (e.g., "AAPL") — used to ground the search | |
| days: Number of days back to search (default: 30) | |
| Returns: | |
| An evidence.v1 JSON envelope (OK, EMPTY, or ERROR). Copy a supporting | |
| record's `ref` exactly when citing news.""" | |
| request = {"query": query, "ticker": ticker.upper(), "days": days} | |
| # Check cache first (keyed by original args, 1-hour TTL) | |
| cached = news_cache.get(ticker, query, days) | |
| if cached is not None: | |
| if parse_evidence_envelope(cached): | |
| return cached | |
| # Legacy cache payloads have no immutable locator or source-level | |
| # provenance. Never repackage them as newly verified evidence. | |
| return _empty( | |
| "search_news", | |
| request, | |
| "Legacy cached news omitted because its provenance cannot be verified.", | |
| ) | |
| try: | |
| # Look up company name from metrics | |
| metrics_rows = metrics_db.get_all_metrics(ticker) | |
| company_name = metrics_rows[0]["company_name"] if metrics_rows else ticker | |
| # Get the latest filing date for filtering | |
| latest_filing_date = metrics_rows[0]["filing_date"] if metrics_rows else None | |
| # Compose ticker-grounded query | |
| full_query = f"{ticker.upper()} {company_name} {query}" | |
| results = _get_tavily_client().search( | |
| query=full_query, | |
| max_results=5, | |
| search_depth="basic", | |
| days=days, | |
| topic="news" | |
| ).get("results", []) | |
| except Exception as exc: | |
| return _error("search_news", request, exc) | |
| # Filter results by filing date | |
| if results and latest_filing_date: | |
| results = [r for r in results if r.get("published_date", "") >= latest_filing_date] | |
| if not results: | |
| result_str = _empty("search_news", request, "No recent news found.") | |
| else: | |
| try: | |
| records = [] | |
| for r in results: | |
| published = str(r.get("published_date") or "unknown") | |
| url = str(r.get("url") or "") | |
| title = str(r.get("title") or "Untitled") | |
| content = f"Source: news | {published}\nTitle: {title}\nURL: {url}\n{r.get('content', '')}" | |
| records.append(make_evidence_record( | |
| source="news", | |
| content=content, | |
| document_id=url or f"news:{ticker.upper()}:{published}:{title}", | |
| source_url=url or None, | |
| as_of=published if published != "unknown" else None, | |
| metadata={ | |
| "ticker": ticker.upper(), | |
| "title": title, | |
| "published_date": published, | |
| }, | |
| )) | |
| except Exception as exc: | |
| return _error("search_news", request, exc) | |
| result_str = evidence_envelope(tool="search_news", query=request, records=records) | |
| news_cache.put(ticker, query, days, result_str) | |
| return result_str | |