backend/red_flags.py

# backend/red_flags.py
"""
Rule-based red flag detection. Zero LLM involvement — every flag is a
direct threshold check against extracted metrics, so this works even in
Mode C (no LLM available at all).

IMPORTANT — metric shapes coming out of metrics_extractor.py are NOT uniform:
  - metrics produced by find_metric_in_text() (revenue, deposits, net_income,
    etc.) are dicts: {"value": float, "confidence": "high"|"medium"|"low",
    "alternatives": [...], "needs_clarification": bool}
  - metrics produced by find_ratio_in_text() (gross_npa_pct, attrition,
    de_ratio, eps, etc.) are plain floats, or None if not found.

get_value() below normalizes both shapes into (value, confidence) so the
threshold checks don't need to know which extractor produced the number.

HONESTY NOTE ON SECTOR COVERAGE:
The original FinSight planning doc listed several thresholds that are NOT
implemented here because the relevant fields are not extracted anywhere in
metrics_extractor.py (PE ratio, ROE, offshore %, utilization %, FDA
rejections, promoter pledge %, auditor flags). Rather than invent numbers
or silently skip them, those are simply absent from the rule sets below.
Adding them is a metrics_extractor.py task first, not a red_flags.py one.

Where a doc-listed metric wasn't extractable as-is but a close substitute
WAS computable from two existing dict-metrics, that's called out explicitly
in the relevant evaluate_*_flags() function (e.g. PHARMA's R&D% is derived
from r_and_d / revenue; ENERGY's leverage check uses debt/total_assets as
a proxy for debt/equity, since equity isn't extracted anywhere).
"""

# ── shared helpers ──────────────────────────────────────────────


def get_value(metric):
    """
    Normalize the two metric shapes from metrics_extractor.py into a
    single (value, confidence) tuple.

    - dict shape (from find_metric_in_text): {"value": ..., "confidence": ...}
    - float shape (from find_ratio_in_text): just the number, confidence
      unknown so we default to "medium" — ratios don't carry a confidence
      score today, this is a known gap, not a guess we're hiding.
    - None: metric wasn't found at all.

    Returns (None, None) if there's nothing usable.
    """
    if metric is None:
        return None, None
    if isinstance(metric, dict):
        return metric.get("value"), metric.get("confidence")
    if isinstance(metric, (int, float)):
        return metric, "medium"
    return None, None


def _check_threshold(value, op, threshold):
    if value is None:
        return False
    if op == "gt":
        return value > threshold
    if op == "lt":
        return value < threshold
    raise ValueError(f"Unknown op: {op}")


def evaluate_ratio_flags(metrics: dict, rules: dict) -> list:
    """
    Generic threshold checker for a sector's flat {metric_key: rule} dict.
    Works for both dict-shaped and float-shaped metrics via get_value().
    """
    triggered = []
    for metric_key, rule in rules.items():
        raw = metrics.get(metric_key)
        value, confidence = get_value(raw)
        if value is None:
            continue
        if _check_threshold(value, rule["op"], rule["threshold"]):
            triggered.append({
                "flag": rule["flag"],
                "message": rule["message"].format(
                    value=round(value, 2), threshold=rule["threshold"]
                ),
                "metric": metric_key,
                "value": round(value, 2),
                "threshold": rule["threshold"],
                "confidence": confidence,
            })
    return triggered


def evaluate_yoy_decline(
    metrics_by_year: dict,
    year: str,
    metric_key: str,
    threshold_pct: float,
    flag_name: str,
    label: str
) -> list:
    """
    Generic YoY decline checker. Needs the prior year's value for
    metric_key in addition to the current year, so this takes the full
    get_company_metrics() output (all years), not just one year's slice.

    Returns a list with 0 or 1 flag dict.
    """
    try:
        prior_year = str(int(year) - 1)
    except ValueError:
        return []

    current = metrics_by_year.get(str(year), {})
    prior = metrics_by_year.get(prior_year)

    if not prior:
        return []  # no prior year on record — can't compute YoY, not a flag

    current_val, current_conf = get_value(current.get(metric_key))
    prior_val, _ = get_value(prior.get(metric_key))

    if current_val is None or prior_val is None or prior_val == 0:
        return []

    decline_pct = ((prior_val - current_val) / prior_val) * 100

    if decline_pct > threshold_pct:
        return [{
            "flag": flag_name,
            "message": (
                f"{label} declined {round(decline_pct, 1)}% YoY "
                f"({prior_year} -> {year}), exceeding the "
                f"{threshold_pct}% threshold"
            ),
            "metric": metric_key,
            "value": round(decline_pct, 2),
            "threshold": threshold_pct,
            "confidence": current_conf,
        }]

    return []


def evaluate_negative_value(metrics: dict, metric_key: str, flag_name: str, label: str) -> list:
    """Flags a metric that is present and below zero. No derivation,
    no threshold guessing — just a sign check on an already-extracted
    dict-shaped metric."""
    value, confidence = get_value(metrics.get(metric_key))
    if value is None or value >= 0:
        return []
    return [{
        "flag": flag_name,
        "message": f"{label} is negative ({round(value, 2)})",
        "metric": metric_key,
        "value": round(value, 2),
        "threshold": 0,
        "confidence": confidence,
    }]


def evaluate_derived_ratio(
    metrics: dict,
    numerator_key: str,
    denominator_key: str,
    op: str,
    threshold: float,
    flag_name: str,
    label: str,
    as_percent: bool = True
) -> list:
    """
    Computes numerator/denominator from two dict-shaped metrics and checks
    it against a threshold. Used where a doc-listed ratio isn't directly
    extracted but is computable from two values that ARE extracted
    (e.g. PHARMA R&D% = r_and_d / revenue).

    Confidence is the LOWER of the two input confidences — a derived
    number can't be more trustworthy than its weakest input.
    """
    num_val, num_conf = get_value(metrics.get(numerator_key))
    den_val, den_conf = get_value(metrics.get(denominator_key))

    if num_val is None or den_val is None or den_val == 0:
        return []

    ratio = (num_val / den_val) * (100 if as_percent else 1)

    if not _check_threshold(ratio, op, threshold):
        return []

    rank = {"high": 0, "medium": 1, "low": 2}
    confidence = max([num_conf, den_conf], key=lambda c: rank.get(c, 1))

    return [{
        "flag": flag_name,
        "message": f"{label} of {round(ratio, 2)}{'%' if as_percent else ''} "
                    f"{'exceeds' if op == 'gt' else 'is below'} the "
                    f"{threshold}{'%' if as_percent else ''} threshold",
        "metric": f"{numerator_key}/{denominator_key}",
        "value": round(ratio, 2),
        "threshold": threshold,
        "confidence": confidence,
    }]


# ── severity weights (used for risk_score) ──────────────────────

FLAG_SEVERITY = {
    "HIGH_GROSS_NPA": 30,
    "HIGH_NET_NPA": 30,
    "LOW_CAPITAL_ADEQUACY": 25,
    "LOW_CASA": 10,
    "DEPOSIT_DECLINE_YOY": 20,
    "NEGATIVE_PAT": 35,

    "HIGH_ATTRITION": 25,
    "REVENUE_DECLINE_YOY": 20,
    "NEGATIVE_NET_INCOME": 35,

    "LOW_RND_PCT": 15,

    "HIGH_LEVERAGE_ASSET_RATIO": 25,

    "HIGH_DEBT_EQUITY": 30,
}

DEFAULT_SEVERITY = 10


def compute_risk_score(flags: list) -> int:
    """Sum severity weights, capped at 100. Simple and auditable —
    no ML, no curve-fitting, just addition."""
    score = sum(FLAG_SEVERITY.get(f["flag"], DEFAULT_SEVERITY) for f in flags)
    return min(score, 100)


def overall_confidence(flags: list) -> str:
    """Lowest-confidence flag drives the overall confidence label —
    a risk_score is only as trustworthy as its weakest input."""
    if not flags:
        return "high"  # no flags triggered, nothing to be unsure about
    rank = {"high": 0, "medium": 1, "low": 2}
    worst = max(flags, key=lambda f: rank.get(f["confidence"], 1))
    return worst["confidence"] or "medium"


# ── sector rule sets ─────────────────────────────────────────────

BANK_RATIO_FLAGS = {
    "gross_npa_pct": {
        "op": "gt", "threshold": 5, "flag": "HIGH_GROSS_NPA",
        "message": "Gross NPA% of {value} exceeds the {threshold}% threshold"
    },
    "net_npa_pct": {
        "op": "gt", "threshold": 3, "flag": "HIGH_NET_NPA",
        "message": "Net NPA% of {value} exceeds the {threshold}% threshold"
    },
    "capital_adequacy": {
        "op": "lt", "threshold": 10, "flag": "LOW_CAPITAL_ADEQUACY",
        "message": "Capital adequacy of {value}% is below the {threshold}% threshold"
    },
    "casa_ratio": {
        "op": "lt", "threshold": 30, "flag": "LOW_CASA",
        "message": "CASA ratio of {value}% is below the {threshold}% threshold "
                   "(lower-cost deposit base is weak)"
    },
}

IT_RATIO_FLAGS = {
    "attrition": {
        "op": "gt", "threshold": 25, "flag": "HIGH_ATTRITION",
        "message": "Attrition rate of {value}% exceeds the {threshold}% threshold"
    },
}

MANUFACTURING_RATIO_FLAGS = {
    "de_ratio": {
        "op": "gt", "threshold": 2, "flag": "HIGH_DEBT_EQUITY",
        "message": "Debt/Equity ratio of {value} exceeds the {threshold} threshold"
    },
}


def evaluate_bank_flags(metrics_by_year: dict, year: str) -> list:
    metrics = metrics_by_year.get(str(year), {})
    flags = evaluate_ratio_flags(metrics, BANK_RATIO_FLAGS)
    flags += evaluate_yoy_decline(
        metrics_by_year, year, "deposits", 10, "DEPOSIT_DECLINE_YOY", "Deposits"
    )
    flags += evaluate_negative_value(
        metrics, "profit_after_tax", "NEGATIVE_PAT", "Profit after tax"
    )
    return flags


def evaluate_it_flags(metrics_by_year: dict, year: str) -> list:
    metrics = metrics_by_year.get(str(year), {})
    flags = evaluate_ratio_flags(metrics, IT_RATIO_FLAGS)
    flags += evaluate_yoy_decline(
        metrics_by_year, year, "revenue", 5, "REVENUE_DECLINE_YOY", "Revenue"
    )
    flags += evaluate_negative_value(
        metrics, "net_income", "NEGATIVE_NET_INCOME", "Net income"
    )
    return flags


def evaluate_pharma_flags(metrics_by_year: dict, year: str) -> list:
    metrics = metrics_by_year.get(str(year), {})
    # R&D% isn't directly extracted (only the absolute r_and_d figure is) —
    # derived here from r_and_d / revenue. Doc's threshold was "<12%".
    flags = evaluate_derived_ratio(
        metrics, "r_and_d", "revenue", "lt", 12, "LOW_RND_PCT", "R&D spend"
    )
    flags += evaluate_yoy_decline(
        metrics_by_year, year, "revenue", 5, "REVENUE_DECLINE_YOY", "Revenue"
    )
    flags += evaluate_negative_value(
        metrics, "net_income", "NEGATIVE_NET_INCOME", "Net income"
    )
    return flags


def evaluate_energy_flags(metrics_by_year: dict, year: str) -> list:
    metrics = metrics_by_year.get(str(year), {})
    # True debt/equity isn't computable — equity isn't extracted anywhere
    # for ENERGY. debt/total_assets is used as the closest honest proxy
    # for leverage risk, not a substitute claimed to be the same thing.
    flags = evaluate_derived_ratio(
        metrics, "debt", "total_assets", "gt", 50,
        "HIGH_LEVERAGE_ASSET_RATIO", "Debt-to-assets", as_percent=True
    )
    flags += evaluate_yoy_decline(
        metrics_by_year, year, "revenue", 5, "REVENUE_DECLINE_YOY", "Revenue"
    )
    flags += evaluate_negative_value(
        metrics, "net_income", "NEGATIVE_NET_INCOME", "Net income"
    )
    return flags


def evaluate_manufacturing_flags(metrics_by_year: dict, year: str) -> list:
    metrics = metrics_by_year.get(str(year), {})
    flags = evaluate_ratio_flags(metrics, MANUFACTURING_RATIO_FLAGS)
    flags += evaluate_yoy_decline(
        metrics_by_year, year, "revenue", 5, "REVENUE_DECLINE_YOY", "Revenue"
    )
    flags += evaluate_negative_value(
        metrics, "net_income", "NEGATIVE_NET_INCOME", "Net income"
    )
    return flags


def evaluate_general_flags(metrics_by_year: dict, year: str) -> list:
    metrics = metrics_by_year.get(str(year), {})
    # GENERAL has no leverage/profitability ratio extracted (no de_ratio,
    # no ROE, no PE) — eps alone isn't threshold-able without a share
    # price or prior-year eps to compare against, so it's left out rather
    # than guessing a cutoff. Only revenue trend + profitability sign
    # checks are implemented here.
    flags = evaluate_yoy_decline(
        metrics_by_year, year, "revenue", 5, "REVENUE_DECLINE_YOY", "Revenue"
    )
    flags += evaluate_negative_value(
        metrics, "net_income", "NEGATIVE_NET_INCOME", "Net income"
    )
    return flags


SECTOR_EVALUATORS = {
    "BANK": evaluate_bank_flags,
    "IT": evaluate_it_flags,
    "PHARMA": evaluate_pharma_flags,
    "ENERGY": evaluate_energy_flags,
    "MANUFACTURING": evaluate_manufacturing_flags,
    "GENERAL": evaluate_general_flags,
}


# ── main entry point ──────────────────────────────────────────────

def evaluate_red_flags(graph, company: str, year: str, sector: str = "GENERAL") -> dict:
    """
    Main entry point. `graph` is a FinancialGraph instance (or anything
    exposing get_company_metrics(company) -> {year: metrics_dict}).
    """
    evaluator = SECTOR_EVALUATORS.get(sector)

    if evaluator is None:
        return {
            "company": company,
            "year": year,
            "sector": sector,
            "flags_triggered": [],
            "risk_score": None,
            "confidence": None,
            "error": f"Red flag rules for sector '{sector}' not implemented yet"
        }

    metrics_by_year = graph.get_company_metrics(company)

    if str(year) not in metrics_by_year:
        return {
            "company": company,
            "year": year,
            "sector": sector,
            "flags_triggered": [],
            "risk_score": None,
            "confidence": None,
            "error": f"No filing found for {company} in {year}"
        }

    flags = evaluator(metrics_by_year, str(year))

    return {
        "company": company,
        "year": year,
        "sector": sector,
        "flags_triggered": flags,
        "risk_score": compute_risk_score(flags),
        "confidence": overall_confidence(flags),
    }


if __name__ == "__main__":
    class FakeGraph:
        def __init__(self, data):
            self._data = data

        def get_company_metrics(self, company):
            return self._data.get(company, {})

    fake_data = {
        "HDFC Bank": {
            "2023": {
                "deposits": {"value": 1_900_000_00_00_000, "confidence": "high"},
                "gross_npa_pct": 1.3, "net_npa_pct": 0.4,
                "casa_ratio": 44.0, "capital_adequacy": 18.9,
            },
            "2024": {
                "profit_after_tax": {"value": 608_120_00_00_000, "confidence": "high"},
                "deposits": {"value": 1_500_000_00_00_000, "confidence": "high"},
                "gross_npa_pct": 6.2, "net_npa_pct": 0.33,
                "casa_ratio": 28.0, "capital_adequacy": 19.3,
            },
        },
        "Infosys": {
            "2023": {"revenue": {"value": 1_500_000_000_000, "confidence": "high"}},
            "2024": {
                "revenue": {"value": 1_300_000_000_000, "confidence": "high"},
                "net_income": {"value": -50_000_000, "confidence": "medium"},
                "attrition": 27.5,
            },
        },
        "SunPharma": {
            "2024": {
                "revenue": {"value": 500_000_000_000, "confidence": "high"},
                "r_and_d": {"value": 30_000_000_000, "confidence": "high"},
                "net_income": {"value": 60_000_000_000, "confidence": "high"},
            },
        },
        "TataSteel": {
            "2024": {
                "revenue": {"value": 800_000_000_000, "confidence": "high"},
                "net_income": {"value": 10_000_000_000, "confidence": "high"},
                "de_ratio": 2.8,
            },
        },
    }

    fg = FakeGraph(fake_data)

    for company, year, sector in [
        ("HDFC Bank", "2024", "BANK"),
        ("Infosys", "2024", "IT"),
        ("SunPharma", "2024", "PHARMA"),
        ("TataSteel", "2024", "MANUFACTURING"),
        ("HDFC Bank", "2024", "ENERGY"),
    ]:
        result = evaluate_red_flags(fg, company, year, sector=sector)
        print(f"\n{company} ({sector}, {year})")
        print(f"  risk_score={result['risk_score']} confidence={result['confidence']}")
        if result.get("error"):
            print(f"  error: {result['error']}")
        for f in result["flags_triggered"]:
            print(f"  [{f['flag']}] {f['message']} (confidence={f['confidence']})")