diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d6e891..ba073d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ All notable changes to Investo are documented here. The format follows ## [Unreleased] ### Added +- **Five new MCP tools** (23 → 28): + - `technical_snapshot` — price/momentum context (50/200-day moving averages + golden/death cross, + RSI(14) with Wilder's smoothing, annualized volatility, 1-year max drawdown, beta vs the market + index, 52-week position). Explicitly framed as *context, not a trading signal*. Backed by a new + cached `sources/yahoo.get_history`. + - `dcf_sensitivity` — intrinsic value across a discount-rate × terminal-growth grid plus the + break-even growth implied by today's price. Fetches statements **once** and reuses them across + all 25 grid cells (get_financials is uncached). + - `compare_companies` — head-to-head across 2–6 arbitrary tickers (answers "compare KPIT with Tata + Elxsi and Tata Tech" directly), not limited to a curated group; share is named set-relative, not + "market share". + - `peer_group_directory` — lists the curated peer groups and members, so a client can see how + companies are grouped and why. + - `export_report` — renders a full analysis to an HTML/PDF file (the **only** non-read-only tool; + its LLM-supplied path is sandboxed to the export directory). - **PDF export** — `investo analyze --pdf [FILE]` renders the research note to PDF, with **no new required dependency**. It shells out to a system Chrome/Edge/Chromium/Brave if one is installed (the usual case), falls back to a Playwright-managed Chromium (`pip install 'investo[pdf]' && diff --git a/README.md b/README.md index bf65f52..2a63429 100644 --- a/README.md +++ b/README.md @@ -175,6 +175,11 @@ of Reliance?"* | `red_flags` | Automated deterioration warnings + overall risk level | | `investment_thesis` | Synthesized pros/cons, quality grade, valuation stance & one-line verdict | | `ai_signals` | Compact machine-readable digest (thesis, quality, confidence, ownership/growth signals, risk, valuation) | +| `technical_snapshot` | Price/momentum context: 50/200-DMA + golden/death cross, RSI, volatility, drawdown, beta, 52-week position (context, not a signal) | +| `dcf_sensitivity` | Intrinsic value across a discount-rate × terminal-growth grid + the growth implied by today's price | +| `compare_companies` | Head-to-head across 2–6 named tickers (not a curated group) | +| `peer_group_directory` | List the curated peer groups and their members | +| `export_report` | Render a full analysis to an HTML/PDF file (writes a file; path sandboxed) | | `analyze_company` | Everything above bundled into one report (with a confidence/provenance evidence layer) | | `get_sec_facts` | SEC EDGAR cross-check (US/ADR only) | diff --git a/manifest.json b/manifest.json index f6e4c4d..04047f3 100644 --- a/manifest.json +++ b/manifest.json @@ -73,7 +73,12 @@ { "name": "investment_thesis", "description": "Pros/cons, quality grade, valuation stance & verdict" }, { "name": "ai_signals", "description": "Compact machine-readable analysis digest" }, { "name": "get_sec_facts", "description": "SEC EDGAR cross-check (US/ADR)" }, - { "name": "provider_status", "description": "Active data providers + disclosure" } + { "name": "provider_status", "description": "Active data providers + disclosure" }, + { "name": "technical_snapshot", "description": "Price/momentum context: DMAs, RSI, volatility, drawdown, beta (not a signal)" }, + { "name": "dcf_sensitivity", "description": "Intrinsic value across a discount-rate x terminal-growth grid + implied break-even growth" }, + { "name": "compare_companies", "description": "Head-to-head across 2-6 named tickers (not a curated peer group)" }, + { "name": "peer_group_directory", "description": "List the curated peer groups and their members" }, + { "name": "export_report", "description": "Render a full analysis to an HTML or PDF file (writes a file)" } ], "compatibility": { "runtimes": { "python": ">=3.10" } diff --git a/src/investo/analysis/multi.py b/src/investo/analysis/multi.py new file mode 100644 index 0000000..b472fbe --- /dev/null +++ b/src/investo/analysis/multi.py @@ -0,0 +1,90 @@ +"""Head-to-head comparison across an arbitrary set of tickers. + +Unlike ``compare_peers``, which works from a curated group, this compares exactly the tickers the +caller names — answering "compare KPIT with Tata Elxsi and Tata Tech" directly. It reuses +``peers._peer_row`` (same currency normalisation) and the same concurrent fetch. + +It deliberately does **not** compute a "market share": revenue share within a set the user invented +is not market share, and calling it that would mislead. The share it does report is named +``revenue_share_of_set`` in prose and is explicitly set-relative. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor + +from ..models import MultiCompare, PeerRow, Provenance +from ..sources import data +from . import evidence as ev +from .peers import _peer_row + + +def compare_companies(symbols: list[str]) -> MultiCompare: + """Compare 2-6 tickers side by side (already-resolved exchange tickers).""" + # De-duplicate, preserving the caller's order. + seen: set[str] = set() + ordered: list[str] = [] + for s in symbols: + u = s.upper() + if u not in seen: + seen.add(u) + ordered.append(u) + + if len(ordered) < 2: + return MultiCompare(tickers=ordered, + note="Need at least two distinct tickers to compare.") + + base_ccy = data.get_info(ordered[0]).get("currency") or \ + data.get_info(ordered[0]).get("financialCurrency") + with ThreadPoolExecutor(max_workers=8) as pool: + fetched = list(pool.map(lambda sym: _peer_row(sym, base_ccy), ordered)) + rows: list[PeerRow] = [r for r in fetched if r is not None] + + if not rows: + return MultiCompare(tickers=ordered, note="None of the tickers resolved to usable data.") + + total_rev = sum(r.revenue_ttm for r in rows if r.revenue_ttm is not None) + if total_rev > 0: + for r in rows: + if r.revenue_ttm is not None: + r.market_share_proxy = r.revenue_ttm / total_rev # set-relative; see _summary + + present = sum(1 for r in rows if r.revenue_ttm is not None or r.market_cap is not None) + return MultiCompare( + tickers=ordered, + rows=rows, + summary=_summary(rows), + note="Revenue & market cap normalized to the first ticker's trading currency. " + "Shares are within this set, not market share.", + evidence=ev.build_meta( + sources=[Provenance(source=ev.SRC_YAHOO, detail="peer fundamentals")], + present=present, expected=len(rows), + notes=[f"Comparing {len(rows)} of {len(ordered)} requested tickers."]), + ) + + +def _summary(rows: list[PeerRow]) -> list[str]: + """A few grounded leaders across the set — no subject, so no rank-of-self.""" + out: list[str] = [] + + def leader(attr: str, label: str, fmt, reverse: bool = True) -> None: + vals = [(r, getattr(r, attr)) for r in rows if getattr(r, attr) is not None] + if not vals: + return + best = sorted(vals, key=lambda x: x[1], reverse=reverse)[0][0] + out.append(f"{label}: {best.name or best.ticker} ({fmt(getattr(best, attr))}).") + + leader("market_cap", "Largest", lambda v: _money(v)) + leader("net_margin", "Highest net margin", lambda v: f"{v:.1%}") + leader("revenue_growth_yoy", "Fastest growth", lambda v: f"{v:.1%}") + leader("pe", "Cheapest P/E", lambda v: f"{v:.1f}x", reverse=False) + leader("roe", "Best ROE", lambda v: f"{v:.1%}") + return out + + +def _money(value: float) -> str: + if abs(value) >= 1e9: + return f"{value / 1e9:,.1f}B" + if abs(value) >= 1e7: + return f"{value / 1e7:,.0f} Cr" + return f"{value:,.0f}" diff --git a/src/investo/analysis/peers.py b/src/investo/analysis/peers.py index 803ce62..4367eac 100644 --- a/src/investo/analysis/peers.py +++ b/src/investo/analysis/peers.py @@ -119,6 +119,25 @@ def get_peers(symbol: str) -> tuple[list[str], dict | None]: return res.peers, res.group +def peer_group_directory(): + """List the curated peer groups so a client can see how companies are grouped and why.""" + from ..models import PeerGroupDirectory, PeerGroupInfo + + groups = [ + PeerGroupInfo( + key=key, + label=g.get("label", key), + outlook=g.get("outlook"), + industry_cagr=g.get("industry_cagr"), + updated_at=g.get("updated_at"), + member_count=len(g.get("members", [])), + members=list(g.get("members", [])), + ) + for key, g in peer_groups().items() + ] + return PeerGroupDirectory(groups=groups, count=len(groups)) + + def _bounded(value: float | None, lo: float, hi: float) -> float | None: if value is None: return None diff --git a/src/investo/analysis/sensitivity.py b/src/investo/analysis/sensitivity.py new file mode 100644 index 0000000..2d0593f --- /dev/null +++ b/src/investo/analysis/sensitivity.py @@ -0,0 +1,119 @@ +"""DCF sensitivity: intrinsic value across a discount-rate x terminal-growth grid. + +A single DCF number hides how much it rests on two assumptions. This grid shows the spread, and +the implied break-even growth answers the more useful question: what future growth is the *market* +already paying for at today's price? + +Performance note that shapes the whole module: ``yahoo.get_financials`` is **not** cached (only +info and FX are). A 5x5 grid calls ``compute_dcf`` 25 times, so we fetch info/financials/ratios +**once** and pass them into every call — otherwise one tool invocation becomes 25 multi-second +statement fetches on the analyze critical path. ``test_sensitivity`` asserts the single fetch. +""" + +from __future__ import annotations + +from ..config import CONFIG +from ..models import DcfSensitivity, Provenance +from ..sources import data +from . import evidence as ev +from .dcf import compute_dcf +from .ratios import compute_ratios + +# Grid geometry: base +/- span, in steps. Kept small — the point is the shape, not a heat map. +_R_STEP = 0.015 +_R_SPAN = 2 # -> 5 columns: base +/- 3.0pp +_G_STEP = 0.01 +_G_SPAN = 2 # -> 5 rows: base +/- 2.0pp +_BISECT_ITERS = 40 +_BISECT_HI = 0.50 # no company grows FCF > 50%/yr forever; cap the break-even search here + + +def dcf_sensitivity(symbol: str, market: str = "IN") -> DcfSensitivity: + symbol = symbol.upper() + # Fetch once; reuse across all 25 grid cells and the break-even bisection. + info = data.get_info(symbol) + financials = data.get_financials(symbol) + ratios = compute_ratios(symbol, info=info, financials=financials) + + base = compute_dcf(symbol, info=info, financials=financials, ratios=ratios) + result = DcfSensitivity( + ticker=symbol, currency=base.currency, base=base, current_price=base.current_price, + ) + + if base.intrinsic_value_per_share is None: + result.note = base.note or "DCF not available, so no sensitivity grid." + result.evidence = ev.build_meta( + sources=[Provenance(source=ev.SRC_YAHOO, detail="statements")], + present=0, expected=1, reason="DCF not computable") + return result + + r0 = base.discount_rate if base.discount_rate is not None else \ + CONFIG.discount_rate_for_market(market) + g0 = base.terminal_growth if base.terminal_growth is not None else CONFIG.dcf_terminal_growth + + rates = [round(r0 + (i - _R_SPAN) * _R_STEP, 4) for i in range(2 * _R_SPAN + 1)] + growths = [round(g0 + (j - _G_SPAN) * _G_STEP, 4) for j in range(2 * _G_SPAN + 1)] + result.discount_rates = rates + result.terminal_growths = growths + + grid: list[list[float | None]] = [] + for g in growths: + row: list[float | None] = [] + for r in rates: + if r <= g: + row.append(None) # matches compute_dcf's own guard; undefined here + continue + cell = compute_dcf(symbol, info=info, financials=financials, ratios=ratios, + discount_rate=r, terminal_growth=g) + row.append(cell.intrinsic_value_per_share) + grid.append(row) + result.grid = grid + + result.implied_breakeven_growth = _breakeven_growth( + symbol, info, financials, ratios, base, g0) + + notes = ["Intrinsic value per share across discount rate (columns) x terminal growth (rows)."] + if result.implied_breakeven_growth is None: + notes.append("Break-even growth exceeds 50%/yr or is otherwise unreachable.") + result.evidence = ev.build_meta( + sources=[Provenance(source=ev.SRC_YAHOO, detail="statements"), + Provenance(source=ev.SRC_HEURISTIC, detail="two-stage DCF")], + present=1, expected=1, notes=notes, + ) + return result + + +def _breakeven_growth(symbol, info, financials, ratios, base, g0) -> float | None: + """The explicit-stage growth that makes intrinsic value equal today's price. + + Intrinsic value is monotonic in growth, so bisect rather than solve — there is no clean + closed form for a two-stage-plus-Gordon model. Returns None if the price can't be reached + within a sane growth ceiling. + """ + price = base.current_price + if price is None or price <= 0: + return None + + def intrinsic_at(g: float) -> float | None: + return compute_dcf(symbol, info=info, financials=financials, ratios=ratios, + growth_rate=g).intrinsic_value_per_share + + lo, hi = g0 + 0.0001, _BISECT_HI + v_lo, v_hi = intrinsic_at(lo), intrinsic_at(hi) + if v_lo is None or v_hi is None: + return None + # The price must sit between the two ends for a root to exist in [lo, hi]. + if not (min(v_lo, v_hi) <= price <= max(v_lo, v_hi)): + return None + + ascending = v_hi >= v_lo + for _ in range(_BISECT_ITERS): + mid = (lo + hi) / 2 + v = intrinsic_at(mid) + if v is None: + return None + if (v < price) == ascending: + lo = mid + else: + hi = mid + return round((lo + hi) / 2, 4) diff --git a/src/investo/analysis/technical.py b/src/investo/analysis/technical.py new file mode 100644 index 0000000..170ca4d --- /dev/null +++ b/src/investo/analysis/technical.py @@ -0,0 +1,279 @@ +"""Technical snapshot: the price/momentum backdrop a fundamentals analysis usually omits. + +Trend (50/200-day moving averages and their crossover), momentum (RSI), volatility, drawdown, +beta and where the price sits in its 52-week range. This is **context, not a signal** — the module +emits no buy/sell verdict, and the model's docstring says so, because a technical readout inside a +fundamentals tool is easy for a downstream LLM to over-read. + +The computation is a pure function of a price DataFrame, so it unit-tests offline against a +synthetic series with no network. +""" + +from __future__ import annotations + +import math +from typing import Any, Literal + +from ..models import Provenance, TechnicalSnapshot +from ..sources import data +from . import evidence as ev + +_RSI_PERIOD = 14 +_DMA_SHORT = 50 +_DMA_LONG = 200 +_TRADING_DAYS = 252 +_CROSS_LOOKBACK = 5 # a crossover within the last week still counts as "just happened" +_BENCHMARK = {"IN": "^NSEI", "US": "^GSPC"} + + +def technical_snapshot( + symbol: str, + *, + history: Any | None = None, + info: dict[str, Any] | None = None, + benchmark_history: Any | None = None, + market: str = "IN", +) -> TechnicalSnapshot: + """Compute the technical snapshot for ``symbol``. + + ``history`` / ``info`` / ``benchmark_history`` are injectable so this runs offline in tests; + when omitted they are fetched (one cached history call each). + """ + symbol = symbol.upper() + if history is None: + history = data.get_history(symbol, period="2y", interval="1d") + if info is None: + info = data.get_info(symbol) + + if history is None or getattr(history, "empty", True) or len(history) < 2: + return TechnicalSnapshot( + ticker=symbol, + currency=info.get("currency") if info else None, + note="No price history available for a technical snapshot.", + evidence=ev.build_meta(sources=[Provenance(source=ev.SRC_YAHOO)], + present=0, expected=7, + reason="no price history"), + ) + + close = [float(c) for c in history["Close"].tolist() if c is not None and not _isnan(c)] + snap = TechnicalSnapshot(ticker=symbol, currency=(info or {}).get("currency")) + snap.as_of = _last_date(history) + snap.price = close[-1] if close else None + + _fill_trend(snap, close) + snap.rsi_14 = _rsi(close, _RSI_PERIOD) + snap.annualized_volatility = _annualized_vol(close) + snap.max_drawdown_1y = _max_drawdown(close[-_TRADING_DAYS:]) + _fill_52w(snap, history) + _fill_beta(snap, history, info, benchmark_history, market) + + snap.observations = _observations(snap) + present = sum(v is not None for v in ( + snap.dma_50, snap.dma_200, snap.rsi_14, snap.annualized_volatility, + snap.max_drawdown_1y, snap.fifty_two_week_position, snap.beta)) + snap.evidence = ev.build_meta( + sources=[Provenance(source=ev.SRC_YAHOO, detail="daily OHLCV")], + present=present, expected=7, as_of=snap.as_of, + notes=["Technical context only — not a trading signal."], + ) + return snap + + +# -------------------------------------------------------------------------------------- +# Indicators +# -------------------------------------------------------------------------------------- +def _fill_trend(snap: TechnicalSnapshot, close: list[float]) -> None: + snap.dma_50 = _sma(close, _DMA_SHORT) + snap.dma_200 = _sma(close, _DMA_LONG) + price = close[-1] + if snap.dma_50 is not None: + snap.above_50dma = price >= snap.dma_50 + if snap.dma_200 is not None: + snap.above_200dma = price >= snap.dma_200 + snap.cross_signal = _cross_signal(close) + + +def _sma(values: list[float], window: int) -> float | None: + if len(values) < window: + return None + return sum(values[-window:]) / window + + +def _cross_signal(close: list[float]) -> Literal["golden", "death", "none"]: + """Golden/death cross if the 50-DMA crossed the 200-DMA within the last few sessions.""" + if len(close) < _DMA_LONG + _CROSS_LOOKBACK: + return "none" + for lag in range(_CROSS_LOOKBACK): + end = len(close) - lag + s_now = sum(close[end - _DMA_SHORT:end]) / _DMA_SHORT + l_now = sum(close[end - _DMA_LONG:end]) / _DMA_LONG + s_prev = sum(close[end - _DMA_SHORT - 1:end - 1]) / _DMA_SHORT + l_prev = sum(close[end - _DMA_LONG - 1:end - 1]) / _DMA_LONG + if s_prev <= l_prev and s_now > l_now: + return "golden" + if s_prev >= l_prev and s_now < l_now: + return "death" + return "none" + + +def _rsi(close: list[float], period: int) -> float | None: + """RSI with **Wilder's smoothing** — the standard, not a simple rolling mean. + + Seed with a simple average of the first ``period`` changes, then smooth: + ``avg = (prev*(period-1) + current) / period``. A run with no losses is RSI 100 (the + division-by-zero guard), which is the correct reading of an unbroken rally. + """ + if len(close) < period + 1: + return None + deltas = [close[i] - close[i - 1] for i in range(1, len(close))] + gains = [max(d, 0.0) for d in deltas] + losses = [-min(d, 0.0) for d in deltas] + + avg_gain = sum(gains[:period]) / period + avg_loss = sum(losses[:period]) / period + for i in range(period, len(deltas)): + avg_gain = (avg_gain * (period - 1) + gains[i]) / period + avg_loss = (avg_loss * (period - 1) + losses[i]) / period + + if avg_loss == 0: + return 100.0 + rs = avg_gain / avg_loss + return 100.0 - (100.0 / (1.0 + rs)) + + +def _annualized_vol(close: list[float]) -> float | None: + """Standard deviation of daily log returns, annualised by sqrt(252).""" + rets = [math.log(close[i] / close[i - 1]) + for i in range(1, len(close)) if close[i - 1] > 0 and close[i] > 0] + if len(rets) < 2: + return None + mean = sum(rets) / len(rets) + var = sum((r - mean) ** 2 for r in rets) / (len(rets) - 1) + return math.sqrt(var) * math.sqrt(_TRADING_DAYS) + + +def _max_drawdown(close: list[float]) -> float | None: + """Largest peak-to-trough decline over the window (a negative fraction).""" + if len(close) < 2: + return None + peak = close[0] + worst = 0.0 + for price in close: + peak = max(peak, price) + if peak > 0: + worst = min(worst, price / peak - 1.0) + return worst + + +def _fill_52w(snap: TechnicalSnapshot, history: Any) -> None: + window = history["Close"].tail(_TRADING_DAYS) + lo, hi = float(window.min()), float(window.max()) + if hi > lo and snap.price is not None: + snap.fifty_two_week_position = max(0.0, min(1.0, (snap.price - lo) / (hi - lo))) + + +def _fill_beta(snap: TechnicalSnapshot, history: Any, info: dict[str, Any] | None, + benchmark_history: Any | None, market: str) -> None: + """Beta vs the market index over aligned daily returns; fall back to Yahoo's stored beta.""" + bench = _BENCHMARK.get((market or "IN").upper(), "^NSEI") + if benchmark_history is None: + benchmark_history = data.get_history(bench, period="2y", interval="1d") + + beta = _beta_from_returns(history, benchmark_history) + if beta is not None: + snap.beta = beta + snap.beta_benchmark = bench + return + stored = (info or {}).get("beta") + if stored is not None: + try: + snap.beta = float(stored) + snap.beta_benchmark = "Yahoo (stored)" + except (TypeError, ValueError): + pass + + +def _beta_from_returns(stock: Any, bench: Any) -> float | None: + if bench is None or getattr(bench, "empty", True): + return None + # Inner-join on date so holidays/half-days don't misalign the two series. + joined = _join_returns(stock, bench) + if len(joined) < 30: + return None + xs = [b for _, b in joined] + ys = [s for s, _ in joined] + mx = sum(xs) / len(xs) + my = sum(ys) / len(ys) + cov = sum((x - mx) * (y - my) for x, y in zip(xs, ys, strict=True)) / len(xs) + var = sum((x - mx) ** 2 for x in xs) / len(xs) + return cov / var if var > 0 else None + + +def _join_returns(stock: Any, bench: Any) -> list[tuple[float, float]]: + s = _daily_returns(stock) + b = _daily_returns(bench) + common = sorted(set(s) & set(b)) + return [(s[d], b[d]) for d in common] + + +def _daily_returns(df: Any) -> dict[Any, float]: + closes = df["Close"] + out: dict[Any, float] = {} + prev = None + prev_idx = None + for idx, val in closes.items(): + v = float(val) + if prev is not None and prev > 0 and v > 0: + out[_day(idx)] = v / prev - 1.0 + prev, prev_idx = v, idx + _ = prev_idx + return out + + +def _day(idx: Any) -> Any: + return idx.date() if hasattr(idx, "date") else idx + + +# -------------------------------------------------------------------------------------- +# Prose +# -------------------------------------------------------------------------------------- +def _observations(s: TechnicalSnapshot) -> list[str]: + out: list[str] = [] + if s.above_50dma is not None and s.above_200dma is not None: + if s.above_50dma and s.above_200dma: + out.append("Trading above both the 50- and 200-day moving averages (uptrend).") + elif not s.above_50dma and not s.above_200dma: + out.append("Trading below both the 50- and 200-day moving averages (downtrend).") + else: + out.append("Mixed: above one moving average and below the other.") + if s.cross_signal == "golden": + out.append("Golden cross: the 50-DMA recently crossed above the 200-DMA.") + elif s.cross_signal == "death": + out.append("Death cross: the 50-DMA recently crossed below the 200-DMA.") + if s.rsi_14 is not None: + if s.rsi_14 >= 70: + out.append(f"RSI {s.rsi_14:.0f}: technically overbought.") + elif s.rsi_14 <= 30: + out.append(f"RSI {s.rsi_14:.0f}: technically oversold.") + else: + out.append(f"RSI {s.rsi_14:.0f}: neutral momentum.") + if s.max_drawdown_1y is not None and s.max_drawdown_1y <= -0.2: + out.append(f"Down {abs(s.max_drawdown_1y):.0%} from its 1-year peak at the worst point.") + if s.fifty_two_week_position is not None: + where = ("near its 52-week high" if s.fifty_two_week_position >= 0.8 + else "near its 52-week low" if s.fifty_two_week_position <= 0.2 + else "mid-range in its 52-week band") + out.append(f"Price is {where}.") + return out + + +def _isnan(x: float) -> bool: + return isinstance(x, float) and math.isnan(x) + + +def _last_date(history: Any) -> str | None: + try: + idx = history.index[-1] + return idx.date().isoformat() if hasattr(idx, "date") else str(idx) + except Exception: + return None diff --git a/src/investo/models.py b/src/investo/models.py index 796c001..4cf696f 100644 --- a/src/investo/models.py +++ b/src/investo/models.py @@ -543,6 +543,89 @@ class AiSignals(_Base): red_flags: list[str] = Field(default_factory=list) +# -------------------------------------------------------------------------------------- +# Analysis tools: export, technicals, DCF sensitivity, multi-company compare +# -------------------------------------------------------------------------------------- +class ExportResult(_Base): + path: str + format: Literal["pdf", "html"] + bytes: int = 0 + engine: str | None = None # e.g. "chrome.exe (headless)" | "playwright-chromium" + warnings: list[str] = Field(default_factory=list) + + +class TechnicalSnapshot(_Base): + """Price/momentum context — deliberately not a trading signal. + + Provides the technical backdrop (trend, volatility, drawdown) a fundamentals analysis usually + lacks. It carries no buy/sell verdict: the numbers are context for a human, not a + recommendation. + """ + + ticker: str + currency: str | None = None + as_of: str | None = None + price: float | None = None + dma_50: float | None = None + dma_200: float | None = None + above_50dma: bool | None = None + above_200dma: bool | None = None + cross_signal: Literal["golden", "death", "none"] | None = None + rsi_14: float | None = None + annualized_volatility: float | None = None + max_drawdown_1y: float | None = None + beta: float | None = None + beta_benchmark: str | None = None + fifty_two_week_position: float | None = None # 0..1 within the 52-week range + observations: list[str] = Field(default_factory=list) + evidence: EvidenceMeta | None = None + note: str | None = None + + +class DcfSensitivity(_Base): + """Intrinsic value across a discount-rate x terminal-growth grid, plus the growth the market + is implying at today's price. The exhibit every real DCF note carries.""" + + ticker: str + currency: str | None = None + base: DCFResult | None = None + discount_rates: list[float] = Field(default_factory=list) # column axis + terminal_growths: list[float] = Field(default_factory=list) # row axis + grid: list[list[float | None]] = Field(default_factory=list) # grid[tg_i][r_i]; None where r<=g + grid_metric: Literal["intrinsic_per_share", "margin_of_safety"] = "intrinsic_per_share" + implied_breakeven_growth: float | None = None + current_price: float | None = None + evidence: EvidenceMeta | None = None + note: str | None = None + + +class MultiCompare(_Base): + """A head-to-head across an arbitrary set of tickers (not a curated peer group).""" + + tickers: list[str] = Field(default_factory=list) + rows: list[PeerRow] = Field(default_factory=list) + summary: list[str] = Field(default_factory=list) + evidence: EvidenceMeta | None = None + note: str | None = None + + +class PeerGroupInfo(_Base): + key: str + label: str + outlook: str | None = None + industry_cagr: str | None = None + updated_at: str | None = None + member_count: int = 0 + members: list[str] = Field(default_factory=list) + + +class PeerGroupDirectory(_Base): + """The curated peer groups, so a client can see how companies are grouped and why.""" + + groups: list[PeerGroupInfo] = Field(default_factory=list) + count: int = 0 + + # -------------------------------------------------------------------------------------- # Signals / SWOT seeds / master report # -------------------------------------------------------------------------------------- diff --git a/src/investo/server.py b/src/investo/server.py index 68cbde7..c87eabf 100644 --- a/src/investo/server.py +++ b/src/investo/server.py @@ -3,10 +3,11 @@ Exposes the analysis toolkit to an AI client (Claude Code, Claude Desktop, Cursor). Run with ``python -m investo.server`` (stdio transport) or via the ``investo-mcp`` script. -Every tool is read-only and hits external data APIs, so all are annotated -``readOnlyHint=True, openWorldHint=True``. Tools return typed pydantic models, so FastMCP -emits an output schema and structured content the client can render; failures surface as MCP -``isError`` results via FastMCP's built-in handling. +Almost every tool is read-only and hits external data APIs, so those are annotated +``readOnlyHint=True, openWorldHint=True``. The one exception is ``export_report``, which writes a +file and is annotated ``readOnlyHint=False``; its output path is sandboxed to the export directory. +Tools return typed pydantic models, so FastMCP emits an output schema and structured content the +client can render; failures surface as MCP ``isError`` results via FastMCP's built-in handling. """ from __future__ import annotations @@ -26,6 +27,8 @@ BuffettChecklist, CompanyProfile, DCFResult, + DcfSensitivity, + ExportResult, Financials, FundamentalTrend, GrowthOutlook, @@ -33,8 +36,10 @@ InvestmentThesis, Management, MoatSignals, + MultiCompare, NewsFeed, PeerComparison, + PeerGroupDirectory, ProviderStatus, Ratios, RedFlagReport, @@ -44,6 +49,7 @@ SearchResult, SecFacts, ShareholdingPattern, + TechnicalSnapshot, ) from .resolve import resolve, resolve_ticker @@ -63,8 +69,11 @@ _log = logging.getLogger("investo.server") -# All tools are read-only data retrieval against external (open-world) APIs. +# Most tools are read-only data retrieval against external (open-world) APIs. _READ = ToolAnnotations(readOnlyHint=True, openWorldHint=True) +# export_report writes a file — it is the one non-read-only tool. +_WRITE = ToolAnnotations(readOnlyHint=False, openWorldHint=True, + destructiveHint=False, idempotentHint=False) # Validated enums / bounds -> surfaced in each tool's JSON input schema. Market = Literal["IN", "US"] @@ -93,6 +102,30 @@ def _symbol(ticker_or_name: str, market: Market = "IN") -> str: return resolved or s.upper() +def _safe_export_path(path: str | None, ext: str): + """Resolve an LLM-supplied export path, confined to the export directory. + + ``path`` comes from a tool call, so it is untrusted. Any path that resolves outside the base + directory — a ``..`` traversal, or an absolute path (on either OS) — is rejected outright + rather than silently clamped, so the behaviour is identical on Windows and POSIX. A plain + relative name, optionally with subdirectories, is allowed; the extension is forced to match + the requested format. + """ + import tempfile + from pathlib import Path + + from .config import CONFIG + + base = Path(CONFIG.export_dir or (Path(tempfile.gettempdir()) / "investo-exports")).resolve() + base.mkdir(parents=True, exist_ok=True) + + name = path or f"investo-export.{ext}" + candidate = (base / name).resolve() + if not candidate.is_relative_to(base): + raise ValueError("Export path must stay inside the export directory.") + return candidate.with_suffix(f".{ext}") + + # -------------------------------------------------------------------------------------- # Tools # -------------------------------------------------------------------------------------- @@ -298,6 +331,84 @@ def relative_metrics(ticker: str, market: Market = "IN") -> RelativeComparison: return relative_comparison(symbol, _compare(symbol), compute_ratios(symbol)) +@mcp.tool(title="Technical snapshot", annotations=_READ) +def technical_snapshot(ticker: str, market: Market = "IN") -> TechnicalSnapshot: + """Price/momentum context: 50/200-day moving averages and any golden/death cross, RSI(14), + annualized volatility, 1-year max drawdown, beta vs the market index, and where the price + sits in its 52-week range. This is *context, not a trading signal* — no buy/sell verdict. + """ + from .analysis.technical import technical_snapshot as _tech + return _tech(_symbol(ticker, market), market=market) + + +@mcp.tool(title="DCF sensitivity", annotations=_READ) +def dcf_sensitivity(ticker: str, market: Market = "IN") -> DcfSensitivity: + """Intrinsic value per share across a discount-rate x terminal-growth grid, plus the + break-even growth the market is implying at today's price. Shows how much the DCF rests on + its two key assumptions rather than presenting a single fragile number. + """ + from .analysis.sensitivity import dcf_sensitivity as _sens + return _sens(_symbol(ticker, market), market) + + +@mcp.tool(title="Compare companies", annotations=_READ) +def compare_companies( + tickers: Annotated[list[str], Field(min_length=2, max_length=6)], + market: Market = "IN", +) -> MultiCompare: + """Head-to-head comparison across 2-6 named tickers (not a curated peer group) — e.g. KPIT + vs Tata Elxsi vs Tata Technologies. Revenue and market cap are normalized to the first + ticker's trading currency; shares reported are within this set, not market share. + """ + from .analysis.multi import compare_companies as _multi + return _multi([_symbol(t, market) for t in tickers]) + + +@mcp.tool(title="Peer group directory", annotations=_READ) +def peer_group_directory() -> PeerGroupDirectory: + """List Investo's curated peer groups (label, outlook, industry CAGR, members) so a client + can see how companies are grouped and why a given company is compared to a given cohort. + """ + from .analysis.peers import peer_group_directory as _dir + return _dir() + + +@mcp.tool(title="Export report", annotations=_WRITE) +def export_report( + query: str, + format: Literal["pdf", "html"] = "pdf", + path: str | None = None, + market: Market = "IN", +) -> ExportResult: + """Render a full analysis to an HTML or PDF file and return where it was written. + + PDF uses a headless Chrome/Edge if one is installed, else a Playwright-managed Chromium; + if neither is available the HTML is written and an error explains how to enable PDF. The + output path is sandboxed to the export directory (INVESTO_EXPORT_DIR, else a temp dir) — + a caller cannot write outside it. + """ + from .analysis.report import analyze + from .export import PdfExportError, html_to_pdf, save_html + + out = _safe_export_path(path, format) + report = analyze(query, market) + + if format == "html": + written = save_html(report, out) + return ExportResult(path=str(written), format="html", + bytes=written.stat().st_size, engine="renderer") + + from .render import render_html + html = render_html(report) + out.with_suffix(".html").write_text(html, encoding="utf-8") # sidecar survives a PDF failure + try: + engine, warnings = html_to_pdf(html, out) + except PdfExportError as exc: + raise ValueError(f"{exc}") from exc + return ExportResult(path=str(out), format="pdf", bytes=out.stat().st_size, + engine=engine, warnings=warnings) + + @mcp.tool(title="Full analysis", annotations=_READ) async def analyze_company(query: str, market: Market = "IN", ctx: Context | None = None) -> AnalysisReport: """Full investment analysis for a company name or ticker. diff --git a/src/investo/sources/data.py b/src/investo/sources/data.py index 007f58b..e25d1ff 100644 --- a/src/investo/sources/data.py +++ b/src/investo/sources/data.py @@ -25,6 +25,7 @@ # Re-export the calls that are Yahoo-sourced regardless of keys. search = yahoo.search get_financials = yahoo.get_financials +get_history = yahoo.get_history fx_rate = yahoo.fx_rate get_esg_score = yahoo.get_esg_score get_growth_estimates = yahoo.get_growth_estimates diff --git a/src/investo/sources/yahoo.py b/src/investo/sources/yahoo.py index f905d1c..357e1c7 100644 --- a/src/investo/sources/yahoo.py +++ b/src/investo/sources/yahoo.py @@ -46,8 +46,10 @@ # calls within one analysis don't re-hit the network). _CACHE_TTL = 900.0 # 15 minutes for company info _FX_TTL = 3600.0 # 1 hour for FX rates +_HIST_TTL = 3600.0 # 1 hour for price history _MISSING = object() _INFO_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} +_HISTORY_CACHE: dict[str, tuple[float, Any]] = {} def _cache_get(cache: dict[str, tuple[float, Any]], key: str, ttl: float) -> Any: @@ -266,6 +268,28 @@ def get_financials(symbol: str, period: str = "annual") -> Financials: ) +def get_history(symbol: str, period: str = "2y", interval: str = "1d"): + """Return an OHLCV price-history DataFrame (auto-adjusted), or None. Never raises. + + Cached for an hour like the other Yahoo calls. A **copy** is handed out: a caller that adds + indicator columns must not mutate the cached frame the next caller will read. + """ + key = f"{symbol.upper()}|{period}|{interval}" + cached = _cache_get(_HISTORY_CACHE, key, _HIST_TTL) + if cached is not _MISSING: + return cached.copy() if cached is not None else None + ratelimit.wait("yahoo", CONFIG.yahoo_min_interval) + df = None + try: + raw = _ticker(symbol).history(period=period, interval=interval, auto_adjust=True) + if raw is not None and not raw.empty: + df = raw + except Exception: + df = None + _cache_set(_HISTORY_CACHE, key, df) + return df.copy() if df is not None else None + + def get_news_raw(symbol: str) -> list[dict[str, Any]]: """Return yfinance's raw news list for a symbol (may be empty).""" try: diff --git a/tests/test_multi_compare.py b/tests/test_multi_compare.py new file mode 100644 index 0000000..71488c8 --- /dev/null +++ b/tests/test_multi_compare.py @@ -0,0 +1,69 @@ +"""Multi-company compare tests (no network — stubbed _peer_row).""" + +from investo.analysis import multi +from investo.models import PeerRow + + +def _wire(monkeypatch, rows: dict[str, PeerRow | None]): + monkeypatch.setattr(multi.data, "get_info", lambda s: {"currency": "INR"}) + monkeypatch.setattr(multi, "_peer_row", lambda sym, ccy: rows.get(sym.upper())) + + +def _row(ticker, name, mcap=None, nm=None, growth=None, pe=None, roe=None, rev=None): + return PeerRow(ticker=ticker, name=name, market_cap=mcap, net_margin=nm, + revenue_growth_yoy=growth, pe=pe, roe=roe, revenue_ttm=rev) + + +def test_compares_the_named_tickers(monkeypatch): + _wire(monkeypatch, { + "KPITTECH.NS": _row("KPITTECH.NS", "KPIT", mcap=3.4e11, nm=0.10, growth=0.12, pe=23.7, + roe=0.197, rev=6e10), + "TATAELXSI.NS": _row("TATAELXSI.NS", "Tata Elxsi", mcap=3.6e11, nm=0.166, growth=0.02, + pe=33.5, roe=0.21, rev=4e10), + }) + mc = multi.compare_companies(["KPITTECH.NS", "TATAELXSI.NS"]) + assert [r.ticker for r in mc.rows] == ["KPITTECH.NS", "TATAELXSI.NS"] + assert mc.evidence is not None + + +def test_duplicates_are_collapsed_preserving_order(monkeypatch): + _wire(monkeypatch, {"A.NS": _row("A.NS", "A", rev=10), "B.NS": _row("B.NS", "B", rev=10)}) + mc = multi.compare_companies(["A.NS", "B.NS", "a.ns", "A.NS"]) + assert mc.tickers == ["A.NS", "B.NS"] + + +def test_unresolvable_rows_are_dropped(monkeypatch): + _wire(monkeypatch, {"A.NS": _row("A.NS", "A", rev=10), "GHOST.NS": None}) + mc = multi.compare_companies(["A.NS", "GHOST.NS"]) + assert [r.ticker for r in mc.rows] == ["A.NS"] + + +def test_fewer_than_two_distinct_tickers_is_a_note(monkeypatch): + _wire(monkeypatch, {"A.NS": _row("A.NS", "A")}) + mc = multi.compare_companies(["A.NS", "a.ns"]) + assert mc.rows == [] + assert "at least two" in mc.note + + +def test_set_relative_share_sums_to_one(monkeypatch): + _wire(monkeypatch, { + "A.NS": _row("A.NS", "A", rev=60.0), + "B.NS": _row("B.NS", "B", rev=40.0), + }) + mc = multi.compare_companies(["A.NS", "B.NS"]) + total = sum(r.market_share_proxy for r in mc.rows if r.market_share_proxy is not None) + assert abs(total - 1.0) < 1e-9 + # And the note is explicit that this is within-set, not market share. + assert "not market share" in mc.note + + +def test_summary_names_grounded_leaders(monkeypatch): + _wire(monkeypatch, { + "A.NS": _row("A.NS", "A", mcap=100, nm=0.2, growth=0.3, pe=10, roe=0.25, rev=50), + "B.NS": _row("B.NS", "B", mcap=200, nm=0.1, growth=0.1, pe=20, roe=0.15, rev=50), + }) + mc = multi.compare_companies(["A.NS", "B.NS"]) + joined = " ".join(mc.summary) + assert "Largest: B" in joined # bigger market cap + assert "Highest net margin: A" in joined + assert "Cheapest P/E: A" in joined diff --git a/tests/test_sensitivity.py b/tests/test_sensitivity.py new file mode 100644 index 0000000..b215bd2 --- /dev/null +++ b/tests/test_sensitivity.py @@ -0,0 +1,102 @@ +"""DCF-sensitivity tests (no network — stubbed data facade). + +The point of interest is not just the grid but the N+1 guard: a 5x5 grid calls compute_dcf 25 +times, and get_financials is uncached, so the module must fetch statements exactly once. +""" + + +from investo.analysis import sensitivity +from investo.models import DCFResult, Financials, Ratios + + +def _wire(monkeypatch, *, intrinsic=1000.0, price=1000.0): + """Stub the data facade and compute_dcf with a monotone, analytic surrogate. + + intrinsic rises with growth and falls with the discount rate — the real DCF's qualitative + behaviour — so grid/breakeven logic can be checked without the network. + """ + calls = {"financials": 0, "info": 0} + + monkeypatch.setattr(sensitivity.data, "get_info", + lambda s: (calls.__setitem__("info", calls["info"] + 1), + {"currency": "INR"})[1]) + monkeypatch.setattr(sensitivity.data, "get_financials", + lambda s, *a, **k: (calls.__setitem__("financials", + calls["financials"] + 1), + Financials(ticker=s))[1]) + monkeypatch.setattr(sensitivity, "compute_ratios", + lambda s, **k: Ratios(ticker=s)) + + def fake_dcf(symbol, *, info=None, financials=None, ratios=None, + discount_rate=None, terminal_growth=None, growth_rate=None, **kw): + r = discount_rate if discount_rate is not None else 0.12 + g = terminal_growth if terminal_growth is not None else 0.04 + gr = growth_rate if growth_rate is not None else 0.10 + # Monotone surrogate: up with growth, down with discount rate. + value = intrinsic * (1 + (gr - 0.10) * 5) * (1 + (g - 0.04) * 3) * (0.12 / r) + return DCFResult(ticker=symbol, currency="INR", discount_rate=r, terminal_growth=g, + growth_rate=gr, intrinsic_value_per_share=value, current_price=price) + + monkeypatch.setattr(sensitivity, "compute_dcf", fake_dcf) + return calls + + +def test_get_financials_is_fetched_exactly_once(monkeypatch): + # The N+1 guard: 25 grid cells + base + break-even bisection must reuse one statement fetch. + calls = _wire(monkeypatch) + sensitivity.dcf_sensitivity("KPITTECH.NS", "IN") + assert calls["financials"] == 1 + + +def test_grid_is_five_by_five_with_the_expected_axes(monkeypatch): + _wire(monkeypatch) + s = sensitivity.dcf_sensitivity("KPITTECH.NS", "IN") + assert len(s.discount_rates) == 5 + assert len(s.terminal_growths) == 5 + assert len(s.grid) == 5 and all(len(row) == 5 for row in s.grid) + + +def test_cells_where_rate_not_above_growth_are_none(monkeypatch): + _wire(monkeypatch) + s = sensitivity.dcf_sensitivity("KPITTECH.NS", "IN") + for gi, g in enumerate(s.terminal_growths): + for ri, r in enumerate(s.discount_rates): + if r <= g: + assert s.grid[gi][ri] is None + + +def test_intrinsic_is_monotonic_in_the_grid(monkeypatch): + _wire(monkeypatch) + s = sensitivity.dcf_sensitivity("KPITTECH.NS", "IN") + mid = len(s.discount_rates) // 2 + # Along a row, value falls as the discount rate rises. + row = [c for c in s.grid[mid] if c is not None] + assert row == sorted(row, reverse=True) + # Down a column, value rises as terminal growth rises. + col = [s.grid[gi][mid] for gi in range(len(s.terminal_growths)) + if s.grid[gi][mid] is not None] + assert col == sorted(col) + + +def test_breakeven_growth_recovers_the_price(monkeypatch): + # Price set equal to the base intrinsic -> break-even growth ~ the base growth (0.10). + _wire(monkeypatch, intrinsic=1000.0, price=1000.0) + s = sensitivity.dcf_sensitivity("KPITTECH.NS", "IN") + assert s.implied_breakeven_growth is not None + assert abs(s.implied_breakeven_growth - 0.10) < 0.01 + + +def test_no_dcf_yields_a_note_not_a_crash(monkeypatch): + _wire(monkeypatch) + monkeypatch.setattr(sensitivity, "compute_dcf", + lambda *a, **k: DCFResult(ticker="X", note="no FCF")) + s = sensitivity.dcf_sensitivity("X", "IN") + assert s.grid == [] + assert s.note is not None + + +def test_unreachable_breakeven_is_none(monkeypatch): + # Price far above any achievable intrinsic -> no break-even within the growth ceiling. + _wire(monkeypatch, intrinsic=1000.0, price=10_000_000.0) + s = sensitivity.dcf_sensitivity("X", "IN") + assert s.implied_breakeven_growth is None diff --git a/tests/test_server_tools.py b/tests/test_server_tools.py new file mode 100644 index 0000000..f05fc0a --- /dev/null +++ b/tests/test_server_tools.py @@ -0,0 +1,99 @@ +"""Server-tool registration and safety tests (no network). + +These don't invoke the tools (that would hit the network); they check the wiring — that the new +tools are registered, that the one file-writing tool is correctly marked non-read-only, that the +export path is sandboxed, and that the manifest lists exactly what the server registers. +""" + +import json +from pathlib import Path + +import pytest + +from investo import server + +_NEW_TOOLS = {"technical_snapshot", "dcf_sensitivity", "compare_companies", + "peer_group_directory", "export_report"} + + +def _tools() -> dict: + return server.mcp._tool_manager._tools + + +def test_all_new_tools_are_registered(): + names = set(_tools()) + assert _NEW_TOOLS <= names + assert len(names) == 28 # 23 existing + 5 new + + +def test_export_report_is_the_only_non_read_only_tool(): + for name, tool in _tools().items(): + ann = tool.annotations + read_only = ann.readOnlyHint if ann else None + if name == "export_report": + assert read_only is False, "a file-writing tool must not claim readOnlyHint=True" + else: + assert read_only is True, f"{name} should be read-only" + + +def test_new_tools_have_descriptions(): + tools = _tools() + for name in _NEW_TOOLS: + assert (tools[name].description or "").strip(), f"{name} has no description" + + +# -------------------------------------------------------------------------------------- +# The export path is LLM-controlled -> it must not escape the sandbox +# -------------------------------------------------------------------------------------- +@pytest.mark.parametrize("evil", [ + "../../etc/passwd", + "../../../Windows/System32/drivers/etc/hosts", + "sub/../../escape", + "/etc/shadow", # POSIX-absolute — rejected uniformly, on Windows too +]) +def test_export_path_escapes_are_rejected(evil): + with pytest.raises(ValueError, match="inside the export directory"): + server._safe_export_path(evil, "pdf") + + +def _export_dir(monkeypatch, tmp_path): + """`_safe_export_path` imports CONFIG from investo.config each call, so patch it there.""" + import dataclasses + + import investo.config as config + monkeypatch.setattr(config, "CONFIG", dataclasses.replace(config.CONFIG, + export_dir=str(tmp_path))) + + +def test_a_plain_name_lands_inside_the_sandbox(tmp_path, monkeypatch): + _export_dir(monkeypatch, tmp_path) + out = server._safe_export_path("kpit-report", "pdf") + assert out.parent == tmp_path.resolve() + assert out.name == "kpit-report.pdf" + + +def test_a_subdirectory_name_is_allowed(tmp_path, monkeypatch): + _export_dir(monkeypatch, tmp_path) + out = server._safe_export_path("reports/kpit", "html") + assert out.is_relative_to(tmp_path.resolve()) + assert out.name == "kpit.html" + + +def test_export_path_forces_the_requested_extension(tmp_path, monkeypatch): + _export_dir(monkeypatch, tmp_path) + assert server._safe_export_path("report.txt", "pdf").suffix == ".pdf" + assert server._safe_export_path("report", "html").suffix == ".html" + + +# -------------------------------------------------------------------------------------- +# Manifest parity — this repo is structurally prone to drift here +# -------------------------------------------------------------------------------------- +def test_manifest_lists_exactly_the_registered_tools(): + manifest = json.loads((Path(__file__).resolve().parent.parent / "manifest.json").read_text()) + manifest_names = {t["name"] for t in manifest["tools"]} + registered = set(_tools()) + assert manifest_names == registered, ( + f"manifest and server disagree; " + f"only in manifest: {manifest_names - registered}; " + f"only in server: {registered - manifest_names}" + ) diff --git a/tests/test_technical.py b/tests/test_technical.py new file mode 100644 index 0000000..1218e61 --- /dev/null +++ b/tests/test_technical.py @@ -0,0 +1,130 @@ +"""Technical-snapshot tests (no network — synthetic price frames only).""" + +import numpy as np +import pandas as pd + +from investo.analysis.technical import ( + _annualized_vol, + _cross_signal, + _max_drawdown, + _rsi, + technical_snapshot, +) + + +def _frame(values: list[float], start: str = "2025-01-01") -> pd.DataFrame: + idx = pd.date_range(start, periods=len(values), freq="D") + return pd.DataFrame({"Close": [float(v) for v in values]}, index=idx) + + +# -------------------------------------------------------------------------------------- +# RSI — Wilder's smoothing, not a simple mean +# -------------------------------------------------------------------------------------- +def test_rsi_of_a_pure_rally_is_100(): + # No losses in the window -> the divide-by-zero guard returns 100, the correct reading. + assert _rsi([float(i) for i in range(1, 40)], 14) == 100.0 + + +def test_rsi_of_a_pure_selloff_is_zero(): + assert _rsi([float(i) for i in range(40, 1, -1)], 14) == 0.0 + + +def test_rsi_matches_a_hand_computed_wilder_value(): + # A known 15-point series; the expected value is Wilder's RSI, which differs from a + # simple-moving-average RSI — this pins the smoothing, not just "some RSI". + prices = [44.34, 44.09, 44.15, 43.61, 44.33, 44.83, 45.10, 45.42, + 45.84, 46.08, 45.89, 46.03, 45.61, 46.28, 46.28] + rsi = _rsi(prices, 14) + assert rsi is not None + assert abs(rsi - 70.53) < 0.5 # canonical Wilder worked example + + +def test_rsi_needs_enough_points(): + assert _rsi([1.0, 2.0, 3.0], 14) is None + + +# -------------------------------------------------------------------------------------- +# Moving-average crossover +# -------------------------------------------------------------------------------------- +def test_golden_cross_detected_when_recent(): + # Long decline (50-DMA below 200-DMA) then a sharp late spike that yanks it above. + vals = list(np.linspace(160, 80, 250)) + [800.0, 800.0, 800.0] + assert _cross_signal(vals) == "golden" + + +def test_death_cross_detected_when_recent(): + vals = list(np.linspace(80, 160, 250)) + [5.0] * 12 + assert _cross_signal(vals) == "death" + + +def test_no_recent_cross_reads_none(): + # A steady uptrend: the 50-DMA has been above the 200-DMA for a long time, no recent cross. + assert _cross_signal([float(i) for i in range(1, 320)]) == "none" + + +def test_cross_needs_enough_history(): + assert _cross_signal([100.0] * 50) == "none" + + +# -------------------------------------------------------------------------------------- +# Volatility & drawdown +# -------------------------------------------------------------------------------------- +def test_annualized_vol_of_a_flat_series_is_zero(): + assert _annualized_vol([100.0] * 30) == 0.0 + + +def test_max_drawdown_of_a_v_shape(): + # 100 -> 50 -> 120: worst drawdown is -50%. + dd = _max_drawdown([100, 80, 50, 70, 120]) + assert abs(dd - (-0.5)) < 1e-9 + + +def test_max_drawdown_of_a_monotonic_rally_is_zero(): + assert _max_drawdown([10, 20, 30, 40]) == 0.0 + + +# -------------------------------------------------------------------------------------- +# The whole snapshot +# -------------------------------------------------------------------------------------- +def test_snapshot_from_a_synthetic_uptrend(): + vals = list(np.linspace(60, 140, 300)) + bench = _frame(list(np.linspace(100, 130, 300))) + snap = technical_snapshot("TEST.NS", history=_frame(vals), + info={"currency": "INR"}, benchmark_history=bench, market="IN") + assert snap.above_50dma is True and snap.above_200dma is True + assert snap.fifty_two_week_position == 1.0 # ends at the high + assert snap.rsi_14 == 100.0 # unbroken rally + assert snap.beta is not None and snap.beta_benchmark == "^NSEI" + assert snap.evidence.confidence is not None + assert any("moving average" in o for o in snap.observations) + + +def test_52_week_position_at_the_low_is_zero(): + vals = list(np.linspace(200, 100, 300)) # ends at the low + snap = technical_snapshot("TEST.NS", history=_frame(vals), info={"currency": "INR"}, + benchmark_history=_frame([100.0] * 300)) + assert snap.fifty_two_week_position == 0.0 + + +def test_beta_falls_back_to_yahoo_stored_when_no_benchmark(): + vals = list(np.linspace(100, 120, 60)) + snap = technical_snapshot("TEST.NS", history=_frame(vals), + info={"currency": "INR", "beta": 1.35}, + benchmark_history=_frame([])) # empty -> can't compute + assert snap.beta == 1.35 + assert snap.beta_benchmark == "Yahoo (stored)" + + +def test_no_history_returns_a_note_not_a_crash(): + snap = technical_snapshot("TEST.NS", history=None, info={"currency": "INR"}, + benchmark_history=_frame([])) + assert snap.note is not None + assert snap.rsi_14 is None + assert snap.evidence is not None + + +def test_snapshot_carries_the_context_not_a_signal_caveat(): + vals = list(np.linspace(60, 140, 300)) + snap = technical_snapshot("TEST.NS", history=_frame(vals), info={"currency": "INR"}, + benchmark_history=_frame([100.0] * 300)) + assert any("not a trading signal" in n for n in snap.evidence.notes)