Risk Management and Compliance

# Generated from ch16_regulatory_capital/01_regulatory_monitoring_tool.py for in-browser execution.
# Agno agent wiring is removed so the tool runs as a plain function;
# the full version is on GitHub. Do not edit: regenerate with
# scripts/build_demos.py.
# regulatory_monitoring_tool.py
# Book reference: Chapter 16, "Architecture"
# Repo notes:
#   - _retrieve_from_source lives in support.py (synthetic, offline).
#   - The printed listing's header pins agno==1.0.6 / google-genai==0.8.0 /
#     pinecone-client==4.1.2; the repo pins agno 2.x per the root
#     pyproject.toml. See ERRATA in the root README.
from datetime import datetime, timezone

import os
import sys

from support import _retrieve_from_source

# Source-of-truth registry — owned by Group Risk.
# Authority verification is structural, not runtime.
SOURCE_OF_TRUTH = {
    "iasb": {
        "url_pattern": "https://www.ifrs.org/news-and-events/news/",
        "publication_id_format": "IFRS-YYYY-NN",
        "review_cadence_days": 14,
    },
    "eiopa": {
        "url_pattern": "https://www.eiopa.europa.eu/publications_en",
        "publication_id_format": "EIOPA-BoS-YY-NNN",
        "review_cadence_days": 7,
    },
    "irdai": {
        "url_pattern": "https://irdai.gov.in/circulars",
        "publication_id_format": "IRDAI/REG/CIR/NNN/YYYY-YY",
        "review_cadence_days": 7,
    },
    # ... mas, naic, pra, iais entries follow the same shape
}


def fetch_regulatory_publications(
    source_id: str,
    cycle_window_days: int = 7,
) -> dict:
    """Return new publications from the named source within the cycle window.
    Authority verification is encoded at the tool layer.
    Verbatim retrievals truncated to <=14 words per copyright discipline.
    """
    if source_id not in SOURCE_OF_TRUTH:                        # authority drift guard
        return {"status": "rejected", "reason": "source_not_in_authority_registry"}

    source = SOURCE_OF_TRUTH[source_id]
    retrieved_at = datetime.now(timezone.utc).isoformat()       # staleness marker
    publications = _retrieve_from_source(source, cycle_window_days)

    for pub in publications:                                    # copyright discipline
        if len(pub["verbatim_excerpt"].split()) > 14:
            pub["paraphrase_required"] = True
            pub["verbatim_excerpt"] = " ".join(
                pub["verbatim_excerpt"].split()[:14]
            ) + " ..."

    return {
        "status": "ok",
        "source_id": source_id,
        "retrieved_at": retrieved_at,
        "publications": publications,
        "diagnostic_surface": {
            "source_authority": source["url_pattern"],
            "publication_id_format": source["publication_id_format"],
            "review_cadence_days": source["review_cadence_days"],
        },
    }


if __name__ == "__main__":
    # Exercise the tool directly: an in-registry source and the
    # authority-drift rejection path.
    print(fetch_regulatory_publications("eiopa", cycle_window_days=7))
    print(fetch_regulatory_publications("some_blog", cycle_window_days=7))

# Generated from ch16_regulatory_capital/02_capital_impact_tool.py for in-browser execution.
# Agno agent wiring is removed so the tool runs as a plain function;
# the full version is on GitHub. Do not edit: regenerate with
# scripts/build_demos.py.
# capital_impact_tool.py
# Book reference: Chapter 16, "Architecture"
# Repo note: _load_capital_snapshot and _attribute_impact_by_business_line
# live in support.py; snapshots ship in data/capital_snapshots.json.
import os
import sys

from support import _attribute_impact_by_business_line, _load_capital_snapshot


def assess_capital_impact(
    publication_id: str,
    affected_business_lines: list[str],
    capital_model_snapshot_id: str,
) -> dict:
    """Read the named capital model snapshot; return the SCR module
    breakdown and the impact attribution per affected business line.
    Read-only against snapshot_id; never mutates live state.
    """
    snapshot = _load_capital_snapshot(capital_model_snapshot_id)   # read-only

    # Discrimination surface: which modules drove which movement.
    module_breakdown = {
        "market_risk":               snapshot["scr_market_risk_usd_m"],
        "life_underwriting_risk":    snapshot["scr_life_uw_risk_usd_m"],
        "non_life_underwriting_risk":snapshot["scr_nonlife_uw_risk_usd_m"],
        "health_underwriting_risk":  snapshot["scr_health_uw_risk_usd_m"],
        "default_risk":              snapshot["scr_default_risk_usd_m"],
        "operational_risk":          snapshot["scr_operational_risk_usd_m"],
    }

    impact_by_module = _attribute_impact_by_business_line(
        snapshot, affected_business_lines, publication_id
    )

    return {
        "status": "ok",
        "snapshot_id": capital_model_snapshot_id,
        "snapshot_close_date": snapshot["close_date"],
        "scr_module_breakdown_usd_m": module_breakdown,
        "impact_attribution_by_module_usd_m": impact_by_module,
        "diagnostic_surface": {
            "snapshot_parameter_versions": snapshot["parameter_versions"],
            "prior_cycle_precedent": snapshot["prior_cycle_close_date"],
        },
    }


if __name__ == "__main__":
    # Exercise the tool directly against the shipped snapshot file.
    print(assess_capital_impact(
        publication_id="EIOPA-BoS-25-142",
        affected_business_lines=["motor_india", "commercial_property"],
        capital_model_snapshot_id="SNAP-FY2025-Q2",
    ))

# Generated from ch16_regulatory_capital/03_orsa_drafting_tool.py for in-browser execution.
# Agno agent wiring is removed so the tool runs as a plain function;
# the full version is on GitHub. Do not edit: regenerate with
# scripts/build_demos.py.
# orsa_drafting_tool.py
# Book reference: Chapter 16, "Architecture"
# Repo note: _generate_paragraph lives in support.py.
import os
import sys

from support import _generate_paragraph, _non_life_movement


def draft_risk_profile_section(
    impact_assessment: dict,            # typed input from assess_capital_impact
    prior_cycle_orsa_excerpt_id: str,   # supervisor-audience precedent
    reasoning_trace: list[dict],        # typed reasoning trace
) -> dict:
    """Structured commentary tool. Supervisor-audience citations:
    every quantitative claim back-traces to capital model output identifier,
    parameter version, and prior-period precedent for material change assertions.
    """
    paragraph = _generate_paragraph(impact_assessment, prior_cycle_orsa_excerpt_id)
    # The cited figure is derived from the same computation as the
    # paragraph, so the audit record and the drafted text cannot diverge.
    _, nl_pct = _non_life_movement(impact_assessment)

    citations = [
        {
            "claim": f"non-life underwriting risk increased {nl_pct:.1f} percent",
            "capital_model_output_id": impact_assessment["snapshot_id"],
            "parameter_version": impact_assessment["diagnostic_surface"]
                                                ["snapshot_parameter_versions"]["non_life"],
            "prior_period_precedent_id": prior_cycle_orsa_excerpt_id,
            "material_change_threshold_breached": True,
        },
        # ... one entry per quantitative claim in the paragraph
    ]

    return {
        "status": "ok",
        "draft_paragraph": paragraph,
        "citations": citations,
        "reasoning_trace_ref": [step["step_id"] for step in reasoning_trace],
    }


if __name__ == "__main__":
    # Exercise the tool directly with an assessment from the sibling tool.
    import importlib
    assess = importlib.import_module("02_capital_impact_tool").assess_capital_impact
    impact = assess(
        publication_id="EIOPA-BoS-25-142",
        affected_business_lines=["motor_india", "commercial_property"],
        capital_model_snapshot_id="SNAP-FY2025-Q2",
    )
    result = draft_risk_profile_section(
        impact_assessment=impact,
        prior_cycle_orsa_excerpt_id="ORSA-FY2025-Q1-RP-04",
        reasoning_trace=[{"step_id": "capital_impact_1"}],
    )
    print(f"status: {result['status']}")
    print(f"draft_paragraph: {result['draft_paragraph']}")
    print(f"citations: {result['citations']}")
    print(f"reasoning_trace_ref: {result['reasoning_trace_ref']}")