analysis/signals.py

"""analysis/signals.py — shared signal types for the Analyst Edge layer.

These Pydantic models carry deterministically-computed evidence (verbatim
before/after text, counts, deltas) from the analysis modules to the LangGraph
agent and synthesis node. The LLM explains; the code supplies the figures.
"""
from __future__ import annotations

from typing import Literal, Optional
from pydantic import BaseModel, ConfigDict, Field


class QuarterDelta(BaseModel):
    """A verbatim text change detected between two consecutive filing periods."""
    model_config = ConfigDict(extra="ignore")

    kind: Literal[
        "risk_added",
        "risk_removed",
        "risk_reworded",
        "guidance_language_shift",
        "term_frequency",
        "kpi_dropped",
        # transcript drift kinds (analysis/tone_drift.py)
        "tone_trend",
        "topic_arc",
        "recurring_evasion",
        "topic_fade",
    ] = Field(description="Type of delta detected.")

    period_from: str = Field(description="Prior filing period, e.g. 'Q42025'.")
    period_to: str = Field(description="Current filing period, e.g. 'Q12026'.")

    before_text: str = Field(
        default="",
        description="Verbatim fragment from the prior period. Empty for risk_added.",
    )
    after_text: str = Field(
        default="",
        description="Verbatim fragment from the current period. Empty for risk_removed.",
    )

    computed_metric: str = Field(
        default="",
        description="A computed summary, e.g. '2→8 occurrences (+300%)' for term_frequency.",
    )

    source: Literal["10-K", "10-Q", "transcript"] = Field(
        default="10-Q",
        description="Filing type the delta was detected in.",
    )

    significance: Literal["HIGH", "MEDIUM", "LOW"] = Field(
        default="MEDIUM",
        description="Computed significance: HIGH for new risks or large frequency swings, etc.",
    )

    term: str = Field(
        default="",
        description="The term or risk label being tracked (for term_frequency / kpi_dropped).",
    )