Build Folio, one pipeline stage at a time.

uv init folio && cd folio
mkdir -p src/folio/{providers,corpus,route,retrieve,compute,synthesize,eval,receipts}
mkdir -p tests data/{corpus,index,cache} scripts
uv add pydantic pydantic-settings numpy python-dateutil
uv add langchain langchain-community langchain-anthropic langchain-openai
uv add anthropic openai                 # only the provider(s) you'll wire
uv add --dev pytest

from functools import lru_cache
from typing import Literal
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    llm: Literal["anthropic", "openai", "fake"] = "fake"    # fake => ZERO keys
    embedder: Literal["local", "openai", "fake"] = "local"

    anthropic_chat_model: str = "claude-sonnet-4-5"         # the default brain
    openai_chat_model: str = "gpt-4.1-mini"                 # the alternate
    local_embed_model: str = "sentence-transformers/all-MiniLM-L6-v2"

    anthropic_api_key: str = ""
    openai_api_key: str = ""

    # retrieval knobs
    retrieval_top_k: int = 20      # hybrid returns this many
    rerank_top_k: int = 6          # this many survive the reranker

    # the hard guardrail: staleness horizon (days) before data must be re-fetched
    freshness_days: int = 3

    index_dir: str = "./data/index"
    corpus_dir: str = "./data/corpus"

    model_config = SettingsConfigDict(env_file=".env", extra="ignore", case_sensitive=False)


@lru_cache(maxsize=1)
def get_settings() -> "Settings":
    return Settings()

from typing import Protocol, runtime_checkable, Any
import numpy as np


@runtime_checkable
class Chatter(Protocol):
    def answer(self, prompt: str, *, system: str = "", tools: list | None = None) -> Any: ...


@runtime_checkable
class Embedder(Protocol):
    dim: int
    def embed(self, texts: list[str]) -> np.ndarray: ...     # (n, dim) float32, unit-norm

# src/folio/receipts/base.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone


@dataclass
class Provenance:
    """Where a fact came from. Attached to EVERY number and EVERY claim."""
    doc_id: str                       # "TCS_concall_Q1FY26.pdf"
    locator: str                      # "p.7 · CFO remarks" or a table cell ref
    as_of: datetime                   # when this datum was true / fetched
    kind: Literal["filing", "price", "screener", "news", "computed"] = "filing"  # type: ignore

    def is_stale(self, horizon_days: int) -> bool:
        return (datetime.now(timezone.utc) - self.as_of).days > horizon_days

    def render(self) -> str:
        return f"[{self.doc_id} · {self.locator} · as of {self.as_of:%Y-%m-%d}]"


@dataclass
class Receipt:
    """Base for every stage's trace. Subclasses add stage-specific fields."""
    stage: str
    started: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    notes: list[str] = field(default_factory=list)

    def render(self) -> str:                # each stage overrides for a richer view
        return f"── {self.stage} ──\n" + "\n".join(f"  · {n}" for n in self.notes)

Prices / quotes    nsepython · jugaad-data · yfinance (".NS" suffix)   unofficial, rate-limited
Fundamentals       Screener.in (exportable) · NSE/BSE announcements     retail gold standard
Concalls / AR      company IR pages, exchange sites                      PDFs — your wheelhouse
IPOs               SEBI (DRHP/RHP), exchanges, Chittorgarh               Chittorgarh = retail-friendly
GMP                grey-market aggregators                               UNOFFICIAL — caveat always
Mutual funds       AMFI (official NAV history) · factsheets              deferred to v4

# src/folio/corpus/types.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class SourceDoc:
    doc_id: str                       # stable id, used as the citation key
    ticker: str                       # "TCS", "INFY" (NSE symbol)
    doc_type: str                     # "concall" | "annual_report" | "drhp" | "news" | "fundamentals"
    text: str
    as_of: datetime                   # publication / fetch date — the freshness anchor
    fiscal_period: str = ""           # "Q1FY26", "FY25" — parsed, not guessed
    url: str = ""
    meta: dict = field(default_factory=dict)

# src/folio/corpus/freshness.py
from datetime import datetime, timezone
from folio.corpus.types import SourceDoc

class StaleDataError(RuntimeError):
    """Raised when a datum that must be current (a price) is older than the horizon."""

def assert_fresh(doc: SourceDoc, horizon_days: int) -> None:
    age = (datetime.now(timezone.utc) - doc.as_of).days
    if doc.doc_type in {"price", "fundamentals"} and age > horizon_days:
        raise StaleDataError(
            f"{doc.ticker} {doc.doc_type} is {age}d old (horizon {horizon_days}d). "
            f"Re-fetch before quoting it as current.")

# src/folio/route/plan.py
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field

Intent = Literal["single_stock", "ipo", "portfolio", "mutual_fund", "news", "glossary"]

class ResearchPlan(BaseModel):
    intent: Intent
    tickers: list[str] = Field(default_factory=list)     # ["TCS"] — NSE symbols
    ipo_name: str = ""                                    # for intent="ipo"
    needs: list[Literal["concall", "annual_report", "drhp",
                        "fundamentals", "price", "news"]] = Field(default_factory=list)
    horizon: Literal["latest_quarter", "3yr", "since_purchase"] = "latest_quarter"
    reason: str = ""                                      # why this route — for the receipt

# src/folio/route/switchboard.py
from folio.route.plan import ResearchPlan
from folio.providers.base import Chatter

_SYS = """You are a router for an Indian-market research assistant. Classify the query
into exactly one intent and list the documents needed. You NEVER answer the question,
NEVER give buy/sell opinions — you only produce a plan. Fiscal years are April–March."""

def route(query: str, chatter: Chatter) -> ResearchPlan:
    # LangChain structured output: the model is forced into the ResearchPlan schema
    plan = chatter.answer(query, system=_SYS, tools=None)   # returns a validated ResearchPlan
    return plan   # invalid JSON / bad enum → pydantic raises here, at the front door

# src/folio/retrieve/hybrid.py
from rank_bm25 import BM25Okapi        # or LangChain's BM25Retriever
# FAISS store: faiss.IndexFlatIP(dim) over unit-norm embeddings (cosine == inner product);
# Embedder from the M0 provider seam. Both built plainly in src/folio/retrieve/.

def rrf(rankings: list[list[str]], k: int = 60) -> dict[str, float]:
    """Reciprocal Rank Fusion — fuse on RANK POSITION, not raw score.
    A FAISS cosine of 0.71 and a BM25 score of 14.2 aren't comparable,
    but 'ranked 2nd in each list' is."""
    scores: dict[str, float] = {}
    for ranking in rankings:
        for rank, chunk_id in enumerate(ranking):
            scores[chunk_id] = scores.get(chunk_id, 0.0) + 1.0 / (k + rank + 1)
    return scores

def hybrid_search(query, bm25, faiss_store, embedder, top_k):
    lexical = bm25.rank(query)                     # list[chunk_id], best first
    dense   = faiss_store.search(embedder.embed([query])[0], top_k)  # list[(chunk, score)]
    fused   = rrf([lexical, [c.chunk_id for c, _ in dense]])
    ranked  = sorted(fused, key=fused.get, reverse=True)[:top_k]
    return ranked

# src/folio/receipts/funnel.py
from dataclasses import dataclass, field
from folio.receipts.base import Receipt

@dataclass
class FunnelReceipt(Receipt):
    hybrid_in: int = 0          # chunks from M3
    reranked: int = 0           # after cross-encoder
    kept: int = 0               # after top-k cut
    chars_before: int = 0
    chars_after: int = 0        # after compression
    per_chunk: list[dict] = field(default_factory=list)  # {id, hybrid_rank, ce_score, kept}

    def render(self) -> str:
        pct = 100 * self.chars_after // max(self.chars_before, 1)
        head = (f"── refine ──  {self.hybrid_in} → rerank → keep {self.kept}   "
                f"context {self.chars_before}→{self.chars_after} chars ({pct}%)")
        rows = "\n".join(
            f"  {'✓' if c['kept'] else '·'} {c['id'][:12]}  "
            f"hy#{c['hybrid_rank']:<2} ce={c['ce_score']:+.2f}" for c in self.per_chunk)
        return head + "\n" + rows

# src/folio/compute/lineitems.py
from pydantic import BaseModel
from folio.receipts.base import Provenance

class LineItem(BaseModel):
    name: str            # "revenue", "net_profit", "total_debt", "shareholder_equity"
    value_inr: float     # RAW rupees — never crore/lakh internally
    period: str          # "Q1FY26", "FY25"
    prov: Provenance     # exactly where in the filing this number was read from

# The LLM's ONLY numeric job: extract, with the page/table locator, into LineItem[].
# It is explicitly instructed NOT to compute, sum, or infer any value.

# src/folio/compute/ratios.py
from dataclasses import dataclass
from datetime import datetime, timezone
from folio.compute.lineitems import LineItem
from folio.receipts.base import Provenance

@dataclass
class ComputedFigure:
    name: str
    value: float
    unit: str                 # "%", "x", "₹ cr"
    inputs: list[Provenance]  # the line items this was computed FROM
    formula: str              # "net_profit / revenue * 100" — human-auditable

    def prov(self) -> Provenance:
        return Provenance(doc_id="computed", locator=self.formula,
                          as_of=datetime.now(timezone.utc), kind="computed")

def net_margin(rev: LineItem, pat: LineItem) -> ComputedFigure:
    return ComputedFigure("net_margin", round(pat.value_inr / rev.value_inr * 100, 2),
                          "%", [rev.prov, pat.prov], "net_profit / revenue * 100")

def debt_to_equity(debt: LineItem, equity: LineItem) -> ComputedFigure:
    return ComputedFigure("debt_to_equity", round(debt.value_inr / equity.value_inr, 2),
                          "x", [debt.prov, equity.prov], "total_debt / shareholder_equity")

# P/E, ROCE, 3-yr revenue/PAT CAGR, margin trend — all the same shape:
# pure functions of LineItems, each returning a ComputedFigure with formula + input provenance.

# src/folio/synthesize/brief.py
from pydantic import BaseModel
from folio.receipts.base import Provenance

class Claim(BaseModel):
    text: str                 # a single, plain-language statement
    prov: list[Provenance]    # ≥1 source. A Claim with no provenance is INVALID by construction.

class NumberedClaim(Claim):
    figure: float
    unit: str                 # this claim asserts a COMPUTED number → must trace to a ComputedFigure

class DecisionBrief(BaseModel):
    ticker: str
    as_of: str
    business: Claim                       # how they actually make money, plainly
    financials: list[NumberedClaim]       # 3-yr revenue / PAT / margin — computed, not recalled
    valuation: list[NumberedClaim]        # P/E, ROCE, debt vs own history & sector
    bull_case: list[Claim]
    bear_case: list[Claim]
    what_changed: list[Claim]             # delta vs last quarter
    flags: list[Claim]                    # margin compression, promoter pledge, mgmt change, RPTs, thesis-breaking news
    # NOTE: there is deliberately NO recommendation field. You make the call.

# src/folio/synthesize/agent.py  (LangGraph ReAct — sketch)
TOOLS = [
    retrieve_chunks,      # M3+M4: hybrid + crucible over the corpus
    extract_line_items,   # M5: LLM reads a filing section into LineItem[]
    compute_ratio,        # M5: deterministic ratio functions (the model CANNOT do the math)
    get_price,            # M1/M5: read-only, freshness-gated
    search_news,          # cross-cutting: recency-filtered, source-quality-gated
]

# ReAct: the model proposes a tool call → runtime executes → observation returns →
# repeat until it has enough EVIDENCE to fill the DecisionBrief schema. It stops when
# every required field has ≥1 sourced Claim, not when it "feels done".

# src/folio/eval/numbers_audit.py
from dataclasses import dataclass
from folio.synthesize.brief import DecisionBrief, NumberedClaim

@dataclass
class AuditResult:
    total_numbers: int
    orphans: list[str]          # numbers with no source / no computed lineage
    stale: list[str]            # numbers whose provenance is past the freshness horizon
    @property
    def passed(self) -> bool:
        return not self.orphans and not self.stale

def audit(brief: DecisionBrief, horizon_days: int) -> AuditResult:
    orphans, stale, n = [], [], 0
    for claim in _all_numbered_claims(brief):        # walk financials + valuation
        n += 1
        if not claim.prov:                            # should be impossible — schema forbids — but verify
            orphans.append(claim.text); continue
        if all(p.is_stale(horizon_days) and p.kind in {"price", "fundamentals"} for p in claim.prov):
            stale.append(claim.text)
    return AuditResult(n, orphans, stale)

class IPOBrief(BaseModel):
    company: str
    as_of: str
    business: Claim                       # how the business makes money
    financials: list[NumberedClaim]       # 3-yr revenue / PAT trend — computed
    use_of_proceeds: list[Claim]          # what they'll do with the money
    key_risks: list[Claim]                # distilled from the legally-required risk-factors section
    shareholding: list[Claim]             # promoter / shareholding pattern
    valuation_vs_peers: list[NumberedClaim]  # P/E vs listed comparables
    red_flags: list[Claim]                # RPTs, litigation, declining margins
    gmp: Claim | None = None              # ONLY with the hard caveat; never a thesis basis

class HoldingBrief(BaseModel):
    ticker: str
    cost_basis: float
    current_price: NumberedClaim          # freshness-gated; caveated if stale
    since_you_bought: list[Claim]         # results, news, events since purchase date
    valuation_snapshot: list[NumberedClaim]  # vs its own history and sector
    thesis_check: Claim                   # is the ORIGINAL reason you bought it still intact?
    flags: list[Claim]                    # fixed set: results miss, pledge increase, mgmt change, sector headwind
    # again: no verdict.

                         ┌─────────────────┐
        query ──────────▶│   SUPERVISOR    │  plans, dispatches, assembles the brief
                         └───────┬─────────┘
              ┌──────────────────┼──────────────────┐
              ▼                  ▼                  ▼
        ┌───────────┐    ┌───────────┐    ┌───────────┐
        │ BULL      │    │ BEAR      │    │ FLAGS     │   each: own ReAct loop + tools
        │ analyst   │    │ analyst   │    │ analyst   │        mandate: make the STRONGEST
        └─────┬─────┘    └─────┬─────┘    └─────┬─────┘        case for its side, sourced
              └────────────────┼────────────────┘
                               ▼
                     ┌───────────────────┐
                     │ SHARED SCRATCHPAD │  evidence pool: computed figures + cited chunks,
                     └───────────────────┘  written once, read by all — no duplicate retrieval