stonks-oracle/services/intelligence_pipeline_v3/parsing/normalizer.py

"""Normalization rules for financial text values.

Converts literal financial text into normalized numeric values:
- "$94.9 billion" → 94_900_000_000.0
- "25 basis points" → 0.25 (percentage points)
- "$1.52 per share" → 1.52
- "4%" → 4.0
- "$123.45" → 123.45
- "€99" → 99.0
"""

from __future__ import annotations

import re

# Multiplier suffixes for money normalization
_MULTIPLIERS: dict[str, float] = {
    "trillion": 1_000_000_000_000.0,
    "billion": 1_000_000_000.0,
    "million": 1_000_000.0,
    "thousand": 1_000.0,
    "k": 1_000.0,
    "m": 1_000_000.0,
    "b": 1_000_000_000.0,
    "t": 1_000_000_000_000.0,
    "bn": 1_000_000_000.0,
    "mn": 1_000_000.0,
    "tn": 1_000_000_000_000.0,
}

_NUMBER_RE = re.compile(r"[+-]?\d[\d,]*\.?\d*")


def _extract_number(text: str) -> float | None:
    """Extract the first numeric value from text, stripping commas."""
    match = _NUMBER_RE.search(text)
    if not match:
        return None
    num_str = match.group().replace(",", "")
    try:
        return float(num_str)
    except ValueError:
        return None


def _find_multiplier(text: str) -> float:
    """Find a magnitude multiplier in text (billion, million, etc.)."""
    lower = text.lower()
    for suffix, mult in _MULTIPLIERS.items():
        # Match word boundaries for short suffixes to avoid false positives
        if len(suffix) <= 2:
            if re.search(rf"\b{suffix}\b", lower):
                return mult
        else:
            if suffix in lower:
                return mult
    return 1.0


def normalize_money(text: str) -> float | None:
    """Normalize a money expression to its numeric value.

    Examples:
        "$94.9 billion" → 94_900_000_000.0
        "$1.52 per share" → 1.52
        "€123.45" → 123.45
        "$1,234" → 1234.0
    """
    number = _extract_number(text)
    if number is None:
        return None

    # Check for "per share" — don't apply multiplier
    lower = text.lower()
    if "per share" in lower:
        return number

    multiplier = _find_multiplier(text)
    return number * multiplier


def normalize_percentage(text: str) -> float | None:
    """Normalize a percentage to its numeric value.

    Examples:
        "4%" → 4.0
        "-2.5%" → -2.5
        "+1.2 percent" → 1.2
    """
    return _extract_number(text)


def normalize_basis_points(text: str) -> float | None:
    """Normalize basis points to percentage points.

    Examples:
        "25 basis points" → 0.25
        "50bps" → 0.50
        "100 bps" → 1.0
    """
    number = _extract_number(text)
    if number is None:
        return None
    return number / 100.0


def normalize_range(text: str) -> tuple[float | None, float | None]:
    """Normalize a range expression to (low, high).

    Examples:
        "$10-$12" → (10.0, 12.0)
        "$1.50 to $2.00" → (1.50, 2.00)
    """
    numbers = _NUMBER_RE.findall(text)
    if len(numbers) < 2:
        return (None, None)
    try:
        low = float(numbers[0].replace(",", ""))
        high = float(numbers[1].replace(",", ""))
        return (low, high)
    except ValueError:
        return (None, None)


def normalize_value(candidate_type: str, text: str) -> float | None:
    """Normalize a candidate value based on its type.

    Returns the normalized numeric value, or None if not applicable.
    For ranges, returns the midpoint.
    """
    if candidate_type == "money" or candidate_type == "currency":
        return normalize_money(text)
    elif candidate_type == "percentage":
        return normalize_percentage(text)
    elif candidate_type == "basis_points":
        return normalize_basis_points(text)
    elif candidate_type == "eps":
        return normalize_money(text)
    elif candidate_type == "revenue":
        return normalize_money(text)
    elif candidate_type == "range":
        low, high = normalize_range(text)
        if low is not None and high is not None:
            return (low + high) / 2.0
        return low
    else:
        return None