"""FinIR-Intent baseline: a deterministic, rule-based NL -> FinIR Intent compiler. This is the constrained structured-output baseline for the first milestone (see docs/huggingface-intent-handoff.md in the FinIR repo). It is intentionally *not* a trained model: a small, fully offline, dependency-free pattern matcher that emits exactly the canonical envelope defined by ``finir.intent`` -- the same JSON Schema and Python types the core runtime already validates and executes. Design boundary (do not blur this): * This module only *interprets* natural language into the canonical envelope. * It never performs financial arithmetic beyond parsing a number out of text. * It never invents a quantity for vague language -- that maps to ``ambiguous``. * It never guesses which of several conflicting operations on the same target the user meant -- that also maps to ``ambiguous`` rather than silently picking one. * Target names, units and the operation vocabulary are exactly the ones in ``finir.intent.schema`` / ``schemas/finir-intent-v1.schema.json``. No new fields, no alias resolution inside the contract -- alias resolution happens here, before the envelope is emitted, exactly as ``docs/intent-contract.md`` requires. Supported phrasing is deliberately narrow and documented in MODEL_CARD.md. Natural language outside these patterns correctly falls back to ``ambiguous`` (a target was named but no parseable quantity was found) rather than being guessed. """ from __future__ import annotations import re from typing import Any from finir.intent import SCHEMA_VERSION, IntentCompiler # --------------------------------------------------------------------------- targets # Canonical target -> the finance "kind" used to decide default unit/currency # handling. This mirrors 1:1 the input types declared in # finir_intent.reference_model.build_reference_model(), so every "valid" envelope # this baseline emits is also executable end-to-end against that model. _TARGET_KIND: dict[str, str] = { "revenue": "money", "cogs": "money", "opex": "money", "payment_terms": "days", "accounts_payable": "money", "inventory": "money", "capex": "money", "debt": "money", "interest_rate": "percentage", "cash": "money", "price": "money", "volume": "quantity", } # NL synonym -> canonical target (model input node name). This alias table lives # entirely in the interpretation layer -- the runtime resolves no aliases # (docs/intent-contract.md "Targets"). _TARGET_ALIASES: dict[str, str] = { "revenue": "revenue", "sales": "revenue", "top line": "revenue", "cogs": "cogs", "cost of goods sold": "cogs", "cost of goods": "cogs", "cost of sales": "cogs", "supplier cost": "cogs", "supplier costs": "cogs", "opex": "opex", "operating expenses": "opex", "operating expense": "opex", "operating costs": "opex", "payment terms": "payment_terms", "customer payment terms": "payment_terms", "receivable terms": "payment_terms", "credit terms": "payment_terms", "accounts payable": "accounts_payable", "trade payables": "accounts_payable", "payables balance": "accounts_payable", "amount owed to suppliers": "accounts_payable", "supplier invoices": "accounts_payable", "inventory levels": "inventory", "inventory": "inventory", "stock levels": "inventory", "stock": "inventory", "capital expenditure": "capex", "capital expenditures": "capex", "capital spending": "capex", "capex": "capex", "borrowings": "debt", "loan balance": "debt", "total debt": "debt", "debt": "debt", "interest rate": "interest_rate", "cost of debt": "interest_rate", "borrowing rate": "interest_rate", "cash balance": "cash", "cash on hand": "cash", "cash position": "cash", "cash": "cash", "unit price": "price", "selling price": "price", "sale price": "price", "price": "price", "sales volume": "volume", "unit volume": "volume", "units sold": "volume", "volume": "volume", } _SORTED_ALIASES = sorted(_TARGET_ALIASES, key=len, reverse=True) _UP_WORDS = ( "increase", "increases", "raise", "raises", "grow", "grows", "rise", "rises", "up", "higher", "extend", "boost", "boosts", ) _DOWN_WORDS = ( "decrease", "decreases", "reduce", "reduces", "cut", "cuts", "trim", "trims", "lower", "drop", "drops", "down", "fall", "falls", ) # Word-boundary matching is required here: naive substring checks on these short # words false-positive constantly in ordinary English (e.g. "up" inside "supplier" # or "group", "down" inside "downside", "cut" inside "circuit"). _UP_RE = re.compile(r"\b(?:" + "|".join(_UP_WORDS) + r")\b") _DOWN_RE = re.compile(r"\b(?:" + "|".join(_DOWN_WORDS) + r")\b") # Requests that are clear but cannot be expressed as a FinIR model mutation -- # distinct from "ambiguous" (right domain, missing quantity). See # docs/intent-contract.md section 7 / docs/huggingface-intent-handoff.md section 7. _UNSUPPORTED_WORDS = ( "acquire", "acquires", "acquiring", "acquired", "acquisition", "acquisitions", "merge", "merges", "merging", "merged", "merger", "mergers", "hire", "hires", "hiring", "hired", "fire staff", "fire employees", "fired staff", "fired employees", "lay off", "laying off", "laid off", "layoff", "layoffs", "ipo", "go public", "going public", "buy back", "buying back", "bought back", "buyback", "buybacks", "share buyback", "litigation", "lawsuit", "lawsuits", "sue", "sues", "sued", "suing", "bankrupt", "bankruptcy", "restructure the board", "new ceo", ) # Word-boundary matching: a naive substring check false-positives on ordinary # English containing these as a fragment (e.g. "merge" inside "emergency", "sue" # inside "issue", "fire" inside "fired up about revenue growth" is fine but "fire" # bare must not match inside e.g. "firewall" or "fireside"). _UNSUPPORTED_RE = re.compile(r"\b(?:" + "|".join(re.escape(w) for w in _UNSUPPORTED_WORDS) + r")\b") # Multi-word unsupported phrases where a number or a couple of filler words can sit # between the trigger words, so a flat word list can't catch them (e.g. "fire 100 # employees", "take the company public"). Anchored to headcount/listing nouns so # they don't false-positive on ordinary English ("fire up the sales pipeline" has no # employees/staff/... noun, so it does not match). _UNSUPPORTED_PATTERNS = ( re.compile( r"\bfir(?:e|es|ed|ing)\b(?:\s+\w+){0,3}\s+(?:employees|workers|staff|people|headcount)\b" ), re.compile(r"\b(?:take|takes|taking|took)\b(?:\s+\w+){0,3}\s+public\b"), re.compile(r"\bgo(?:es|ing)?\s+public\b"), re.compile(r"\bpublic\s+(?:offering|listing)\b"), re.compile(r"\blist(?:s|ed|ing)?\s+on\s+the\s+(?:stock\s+)?exchange\b"), re.compile(r"\b(?:cut|cuts|reduce|reduces|slash|slashes)\s+headcount\b"), ) # Spelled-out cardinal numbers (deterministic, exact parsing of an unambiguous # number phrase -- not a guess). Kept small and targeted: only the words needed to # resolve a spelled-out number, never a general vocabulary. _NUM_WORDS: dict[str, int] = { "zero": 0, "one": 1, "two": 2, "three": 3, "four": 4, "five": 5, "six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10, "eleven": 11, "twelve": 12, "thirteen": 13, "fourteen": 14, "fifteen": 15, "sixteen": 16, "seventeen": 17, "eighteen": 18, "nineteen": 19, "twenty": 20, "thirty": 30, "forty": 40, "fifty": 50, "sixty": 60, "seventy": 70, "eighty": 80, "ninety": 90, } _NUM_SCALES: dict[str, int] = { "hundred": 100, "thousand": 1_000, "million": 1_000_000, "billion": 1_000_000_000, } _NUM_WORD_ALT = "|".join(sorted({*_NUM_WORDS, *_NUM_SCALES}, key=len, reverse=True)) _PCT_RE = re.compile(r"([+-]?\d+(?:\.\d+)?)\s*(?:%|\bpercent\b)") # Only a phrase built entirely from recognized number/scale words is captured -- # deliberately not a generic "any words before percent" grab, which would swallow # unrelated leading words (e.g. "up" in "bump ... up five percent"). _SPELLED_PCT_RE = re.compile( rf"\b((?:{_NUM_WORD_ALT})(?:[ -](?:{_NUM_WORD_ALT}))*)\s+percent\b", re.I ) _SPELLED_MONEY_RE = re.compile( rf"^((?:{_NUM_WORD_ALT})(?:[ -](?:{_NUM_WORD_ALT}))*)\s*(rand|rands|dollars?|zar|usd)?\s*\.?\s*$", re.I, ) _DAYS_RANGE_RE = re.compile(r"\b(\d+)\s*(?:to|->|→)\s*(\d+)\s*days?\b") _DAYS_TO_RE = re.compile(r"\bto\s+(\d+)\s*days?\b") _BY_RE = re.compile(r"\bby\s+(.+)$") _TO_RE = re.compile(r"\bto\s+(.+)$") _MONEY_AMOUNT_RE = re.compile(r"(r|zar|usd|\$)?\s*([\d]+(?:\.\d+)?)\s*(zar|usd)?", re.I) _RANGE_RE = re.compile( r"\b(?:range|sweep|grid|scan|explore)\b.*?\b(?:from|between)\b\s*(?:r|zar|usd|\$)?\s*([\d.]+)" r"\s*(?:zar|usd)?\s*\b(?:to|and)\b\s*(?:r|zar|usd|\$)?\s*([\d.]+)\s*(?:zar|usd)?.*?\b(\d+)\s*(?:steps|points)\b", re.I | re.S, ) _SCENARIO_SPLIT_RE = re.compile(r"([A-Za-z][A-Za-z ]{0,24}?)\s+scenario\s*:\s*", re.I) _NO_CHANGE_BODY = { "no change", "no changes", "none", "base", "unchanged", "the base case", "base case", } def _strip_thousands(text: str) -> str: """Remove thousand-separator commas from numbers (``5,000,000`` -> ``5000000``).""" return re.sub(r"(?<=\d),(?=\d{3}\b)", "", text) def _amount(raw: str) -> float: return float(re.sub(r"[^\d.]", "", raw)) def _currency(*fragments: str | None) -> str | None: for frag in fragments: if not frag: continue low = frag.strip().lower() if low in ("r", "zar", "rand", "rands"): return "ZAR" if low in ("$", "usd", "dollar", "dollars"): return "USD" return None def _words_to_number(phrase: str) -> float | None: """Convert a spelled-out cardinal number phrase (e.g. 'five million') to a float. Deterministic, exact parsing of an unambiguous number -- not a guess. Returns None if any token isn't a recognized number/scale word. """ tokens = phrase.strip().lower().replace("-", " ").split() if not tokens: return None total = 0 current = 0 for t in tokens: if t == "and": continue if t in _NUM_WORDS: current += _NUM_WORDS[t] elif t in _NUM_SCALES: scale = _NUM_SCALES[t] if scale == 100: current = (current or 1) * scale else: total += (current or 1) * scale current = 0 else: return None return float(total + current) def _direction_sign(low: str) -> float: up = bool(_UP_RE.search(low)) down = bool(_DOWN_RE.search(low)) return -1.0 if (down and not up) else 1.0 def _find_target(low: str) -> str | None: for alias in _SORTED_ALIASES: if re.search(r"\b" + re.escape(alias) + r"\b", low): return _TARGET_ALIASES[alias] return None def _split_clauses(text: str) -> list[str]: protected = _strip_thousands(text) parts = re.split(r"\s*;\s*|\s*,\s*(?=[A-Za-z])|\s+\band\b\s+", protected) clauses = [] for p in parts: p = re.sub(r"^(and|then|also)\s+", "", p.strip(), flags=re.I).strip(" .") if p: clauses.append(p) return clauses def _parse_clause(clause: str) -> tuple[dict[str, Any] | None, str | None]: """Parse one clause. Returns (operation, note); note is set only on failure.""" low = clause.lower() target = _find_target(low) if target is None: return None, None kind = _TARGET_KIND.get(target, "money") m = _DAYS_RANGE_RE.search(low) if m: return { "operation": "set", "target": target, "value": _amount(m.group(2)), "unit": "days", }, None m = _DAYS_TO_RE.search(low) if m: return { "operation": "set", "target": target, "value": _amount(m.group(1)), "unit": "days", }, None # "set/reduce/... TARGET to N%" is an absolute set to that percentage level # (new = value), not a relative_change -- must be checked before the generic # relative-change percentage rule below. m_to = _TO_RE.search(low) if m_to: pct = _PCT_RE.match(m_to.group(1).strip()) if pct and kind == "percentage": return ( { "operation": "set", "target": target, "value": _amount(pct.group(1)) / 100.0, "unit": "percentage", }, None, ) if pct: # "set/change to N%" (e.g. "set opex to 45%") is # not a coherent instruction for this target's type -- refuse rather # than silently reinterpreting "to" as "by" (relative_change), which # would invent a meaning the user did not state. return None, "target_no_value" m = _PCT_RE.search(low) if m: value = _direction_sign(low) * _amount(m.group(1)) / 100.0 return {"operation": "relative_change", "target": target, "value": value}, None m = _SPELLED_PCT_RE.search(low) if m: if m_to and _SPELLED_PCT_RE.match(m_to.group(1).strip()): # Same "set X to " guard as the digit case above -- a # spelled-out number must not bypass it. return None, "target_no_value" num = _words_to_number(m.group(1)) if num is not None: value = _direction_sign(low) * num / 100.0 return {"operation": "relative_change", "target": target, "value": value}, None m = _BY_RE.search(low) if m and kind == "money": amt_match = _MONEY_AMOUNT_RE.search(m.group(1)) if amt_match and amt_match.group(2): amt = _amount(amt_match.group(2)) sign = -1.0 if _direction_sign(low) < 0 else 1.0 op: dict[str, Any] = { "operation": "absolute_change", "target": target, "value": sign * amt, } ccy = _currency(amt_match.group(1), amt_match.group(3)) if ccy: op["currency"] = ccy return op, None spelled = _SPELLED_MONEY_RE.match(m.group(1).strip()) if spelled: num = _words_to_number(spelled.group(1)) if num is not None: sign = -1.0 if _direction_sign(low) < 0 else 1.0 op = {"operation": "absolute_change", "target": target, "value": sign * num} ccy = _currency(spelled.group(2)) if ccy: op["currency"] = ccy return op, None if m_to and kind in ("money", "days"): amt_match = _MONEY_AMOUNT_RE.search(m_to.group(1)) if amt_match and amt_match.group(2): amt = _amount(amt_match.group(2)) op = {"operation": "set", "target": target, "value": amt} if kind == "money": ccy = _currency(amt_match.group(1), amt_match.group(3)) if ccy: op["currency"] = ccy else: # days, without the word "days" already matched above op["unit"] = "days" return op, None # kind == "quantity": a plain numeric "set ... to N" is left unhandled (unit # intentionally never guessed for quantity targets -- see MODEL_CARD.md "known # limitations": the runtime's semantic validator only accepts unit == 'scalar' # or no unit for Quantity-typed targets today). return None, "target_no_value" def _envelope( status: str, *, operations: list[dict[str, Any]] | None = None, reason: str | None = None ) -> dict[str, Any]: env: dict[str, Any] = { "schema_version": SCHEMA_VERSION, "status": status, "operations": operations or [], } if reason is not None: env["reason"] = reason return env def _is_unsupported(low: str) -> bool: if _UNSUPPORTED_RE.search(low): return True return any(pat.search(low) for pat in _UNSUPPORTED_PATTERNS) def _try_parse_scenarios(text: str) -> dict[str, Any] | None: parts = _SCENARIO_SPLIT_RE.split(text) if len(parts) < 3: return None preamble = parts[0].strip(" .") if preamble and len(preamble.split()) > 8: return None scenarios: list[dict[str, Any]] = [] for i in range(1, len(parts), 2): name = re.sub(r"\s+", "_", parts[i].strip().lower()) body = parts[i + 1].strip(" .") if i + 1 < len(parts) else "" if not body or body.lower() in _NO_CHANGE_BODY: scenarios.append({"name": name, "operations": []}) continue ops: list[dict[str, Any]] = [] unparsed = False for clause in _split_clauses(body): op, _note = _parse_clause(clause) if op is not None: if op["operation"] == "range": return None # range is never valid inside a scenario ops.append(op) else: unparsed = True if unparsed and not ops: return None # can't confidently build this scenario -> fall through to ambiguous targets = [o["target"] for o in ops] if len(targets) != len(set(targets)): return None scenarios.append({"name": name, "operations": ops}) if len(scenarios) < 2: return None return {"schema_version": SCHEMA_VERSION, "status": "valid", "scenarios": scenarios} def compile_intent(text: str) -> dict[str, Any]: """Compile one natural-language financial instruction into a canonical envelope. Always returns a dict that structurally validates against ``finir.intent.json_schema()`` (verify with ``finir.intent.FinIRIntent.from_obj`` before executing). """ scenario_env = _try_parse_scenarios(text) if scenario_env is not None: return scenario_env low = text.lower() if _is_unsupported(low): return _envelope( "unsupported", reason=f"request cannot be represented as a FinIR model mutation: {text!r}", ) range_match = _RANGE_RE.search(_strip_thousands(low)) if range_match: target = _find_target(low) if target is not None: return _envelope( "valid", operations=[ { "operation": "range", "target": target, "min": _amount(range_match.group(1)), "max": _amount(range_match.group(2)), "steps": int(range_match.group(3)), } ], ) ops: list[dict[str, Any]] = [] saw_target_without_value = False for clause in _split_clauses(text): op, note = _parse_clause(clause) if op is not None: ops.append(op) elif note == "target_no_value": saw_target_without_value = True if not ops: if saw_target_without_value: return _envelope( "ambiguous", reason="a financial target was identified but no quantitative change was specified", ) return _envelope( "ambiguous", reason="no financial target or quantitative change was identified" ) targets = [o["target"] for o in ops] if len(targets) != len(set(targets)): return _envelope( "ambiguous", reason="conflicting operations on the same target were requested; cannot resolve unambiguously", ) return _envelope("valid", operations=ops) class BaselineIntentCompiler(IntentCompiler): """The FinIR-Intent baseline, implementing the core :class:`IntentCompiler` seam. Deterministic and fully offline -- no external LLM/API dependency. Emits exactly the canonical envelope; validation and execution are left entirely to ``finir.intent`` / ``FinancialModel.apply_intent`` (never duplicated here). """ def compile(self, text: str) -> dict[str, Any]: return compile_intent(text)