"""FP&A planning and variance control tower.

Portfolio case study using synthetic data.  The workflow demonstrates how an
analyst can move from a monthly ledger to a governed P&L, a variance queue and
an explicit reforecast scenario.  It deliberately keeps definitions, controls
and limitations visible rather than presenting a black-box forecast.
"""

from __future__ import annotations

import argparse
import json
import sqlite3
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable

import numpy as np
import pandas as pd


REQUIRED_COLUMNS = {
    "month",
    "business_unit",
    "cost_centre",
    "function",
    "account_code",
    "account",
    "account_type",
    "budget_gbp",
    "actual_gbp",
    "committed_gbp",
    "headcount_fte",
}


@dataclass(frozen=True)
class Scenario:
    name: str
    revenue_growth_pct: float = 0.0
    direct_cost_inflation_pct: float = 0.0
    discretionary_cost_reduction_pct: float = 0.0
    south_recovery_pct: float = 0.0
    cash_conversion_pct: float = 0.86


@dataclass(frozen=True)
class ControlTotals:
    row_count: int
    months: int
    cost_centres: int
    budget_gbp: float
    actual_gbp: float
    committed_gbp: float
    missing_values: int
    duplicate_keys: int


P_AND_L_SQL = """
WITH signed_ledger AS (
    SELECT
        month,
        business_unit,
        cost_centre,
        function,
        account,
        account_type,
        CASE WHEN account_type = 'Revenue' THEN budget_gbp ELSE -budget_gbp END AS signed_budget,
        CASE WHEN account_type = 'Revenue' THEN actual_gbp ELSE -actual_gbp END AS signed_actual,
        CASE WHEN account_type = 'Revenue' THEN 0 ELSE -committed_gbp END AS signed_commitment
    FROM ledger
), unit_month AS (
    SELECT
        month,
        business_unit,
        SUM(signed_budget) AS budget_ebitda_gbp,
        SUM(signed_actual) AS actual_ebitda_gbp,
        SUM(signed_actual + signed_commitment) AS exposed_ebitda_gbp,
        SUM(CASE WHEN account_type = 'Revenue' THEN signed_budget ELSE 0 END) AS budget_revenue_gbp,
        SUM(CASE WHEN account_type = 'Revenue' THEN signed_actual ELSE 0 END) AS actual_revenue_gbp,
        -SUM(CASE WHEN account_type = 'Direct cost' THEN signed_actual ELSE 0 END) AS direct_cost_gbp,
        -SUM(CASE WHEN account_type = 'Operating expense' THEN signed_actual ELSE 0 END) AS opex_gbp
    FROM signed_ledger
    GROUP BY month, business_unit
)
SELECT
    month,
    business_unit,
    ROUND(budget_revenue_gbp, 2) AS budget_revenue_gbp,
    ROUND(actual_revenue_gbp, 2) AS actual_revenue_gbp,
    ROUND(direct_cost_gbp, 2) AS direct_cost_gbp,
    ROUND(opex_gbp, 2) AS opex_gbp,
    ROUND(budget_ebitda_gbp, 2) AS budget_ebitda_gbp,
    ROUND(actual_ebitda_gbp, 2) AS actual_ebitda_gbp,
    ROUND(actual_ebitda_gbp - budget_ebitda_gbp, 2) AS ebitda_variance_gbp,
    ROUND(100.0 * (actual_ebitda_gbp - budget_ebitda_gbp)
          / NULLIF(ABS(budget_ebitda_gbp), 0), 2) AS ebitda_variance_pct,
    ROUND(exposed_ebitda_gbp, 2) AS exposed_ebitda_gbp
FROM unit_month
ORDER BY month, business_unit;
"""


VARIANCE_QUEUE_SQL = """
WITH account_variance AS (
    SELECT
        business_unit,
        cost_centre,
        function,
        account,
        account_type,
        SUM(budget_gbp) AS budget_gbp,
        SUM(actual_gbp) AS actual_gbp,
        SUM(committed_gbp) AS committed_gbp,
        CASE
            WHEN account_type = 'Revenue' THEN SUM(actual_gbp - budget_gbp)
            ELSE SUM(budget_gbp - actual_gbp - committed_gbp)
        END AS favourable_variance_gbp
    FROM ledger
    GROUP BY business_unit, cost_centre, function, account, account_type
), prioritised AS (
    SELECT
        *,
        ABS(favourable_variance_gbp) AS absolute_variance_gbp,
        100.0 * favourable_variance_gbp / NULLIF(ABS(budget_gbp), 0) AS variance_pct,
        DENSE_RANK() OVER (ORDER BY ABS(favourable_variance_gbp) DESC) AS materiality_rank,
        SUM(ABS(favourable_variance_gbp)) OVER () AS total_absolute_variance
    FROM account_variance
)
SELECT
    business_unit,
    cost_centre,
    function,
    account,
    account_type,
    ROUND(budget_gbp, 2) AS budget_gbp,
    ROUND(actual_gbp, 2) AS actual_gbp,
    ROUND(committed_gbp, 2) AS committed_gbp,
    ROUND(favourable_variance_gbp, 2) AS favourable_variance_gbp,
    ROUND(variance_pct, 2) AS variance_pct,
    materiality_rank,
    CASE
        WHEN favourable_variance_gbp < 0 AND ABS(variance_pct) >= 12 THEN 'Escalate'
        WHEN favourable_variance_gbp < 0 AND ABS(variance_pct) >= 6 THEN 'Explain and recover'
        WHEN favourable_variance_gbp >= 0 AND ABS(variance_pct) >= 10 THEN 'Validate upside'
        ELSE 'Monitor'
    END AS action
FROM prioritised
WHERE absolute_variance_gbp >= total_absolute_variance * 0.006
ORDER BY absolute_variance_gbp DESC;
"""


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path, help="Synthetic monthly ledger CSV")
    parser.add_argument("--output-dir", type=Path, default=Path("fpa_outputs"))
    return parser.parse_args()


def load_ledger(path: Path) -> pd.DataFrame:
    frame = pd.read_csv(path)
    missing_columns = REQUIRED_COLUMNS.difference(frame.columns)
    if missing_columns:
        raise ValueError(f"Missing required fields: {sorted(missing_columns)}")
    frame["month"] = pd.to_datetime(frame["month"], errors="raise")
    numeric = ["budget_gbp", "actual_gbp", "committed_gbp", "headcount_fte"]
    frame[numeric] = frame[numeric].apply(pd.to_numeric, errors="raise")
    if (frame[["budget_gbp", "actual_gbp", "committed_gbp"]] < 0).any().any():
        raise ValueError("Ledger values must be stored as unsigned amounts")
    if not frame["account_type"].isin({"Revenue", "Direct cost", "Operating expense"}).all():
        raise ValueError("Unexpected account type")
    return frame.sort_values(["month", "business_unit", "cost_centre", "account_code"])


def control_totals(frame: pd.DataFrame) -> ControlTotals:
    key = ["month", "cost_centre", "account_code", "scenario"]
    return ControlTotals(
        row_count=len(frame),
        months=frame["month"].nunique(),
        cost_centres=frame["cost_centre"].nunique(),
        budget_gbp=round(float(frame["budget_gbp"].sum()), 2),
        actual_gbp=round(float(frame["actual_gbp"].sum()), 2),
        committed_gbp=round(float(frame["committed_gbp"].sum()), 2),
        missing_values=int(frame[list(REQUIRED_COLUMNS)].isna().sum().sum()),
        duplicate_keys=int(frame.duplicated(key).sum()),
    )


def build_database(frame: pd.DataFrame) -> sqlite3.Connection:
    database = sqlite3.connect(":memory:")
    prepared = frame.copy()
    prepared["month"] = prepared["month"].dt.strftime("%Y-%m-%d")
    prepared.to_sql("ledger", database, index=False, if_exists="replace")
    database.executescript(
        """
        CREATE INDEX idx_ledger_month ON ledger(month);
        CREATE INDEX idx_ledger_unit ON ledger(business_unit);
        CREATE INDEX idx_ledger_cost_centre ON ledger(cost_centre);
        CREATE INDEX idx_ledger_account_type ON ledger(account_type);
        """
    )
    return database


def query_frame(database: sqlite3.Connection, sql: str) -> pd.DataFrame:
    return pd.read_sql_query(sql, database)


def apply_scenario(frame: pd.DataFrame, scenario: Scenario) -> pd.DataFrame:
    model = frame.copy()
    model["scenario_actual_gbp"] = model["actual_gbp"]

    revenue = model["account_type"].eq("Revenue")
    direct_cost = model["account_type"].eq("Direct cost")
    discretionary = model["account"].isin(
        {"Travel and site costs", "Professional services", "Sales and marketing", "Other operating cost"}
    )
    south_revenue = revenue & model["business_unit"].eq("South Infrastructure")

    model.loc[revenue, "scenario_actual_gbp"] *= 1 + scenario.revenue_growth_pct / 100
    model.loc[direct_cost, "scenario_actual_gbp"] *= 1 + scenario.direct_cost_inflation_pct / 100
    model.loc[discretionary, "scenario_actual_gbp"] *= 1 - scenario.discretionary_cost_reduction_pct / 100
    model.loc[south_revenue, "scenario_actual_gbp"] *= 1 + scenario.south_recovery_pct / 100
    return model


def scenario_summary(frame: pd.DataFrame, scenario: Scenario) -> dict:
    model = apply_scenario(frame, scenario)
    revenue = model.loc[model["account_type"].eq("Revenue"), "scenario_actual_gbp"].sum()
    costs = model.loc[~model["account_type"].eq("Revenue"), "scenario_actual_gbp"].sum()
    ebitda = revenue - costs
    base_revenue = frame.loc[frame["account_type"].eq("Revenue"), "actual_gbp"].sum()
    base_costs = frame.loc[~frame["account_type"].eq("Revenue"), "actual_gbp"].sum()
    base_ebitda = base_revenue - base_costs
    return {
        **asdict(scenario),
        "revenue_gbp": round(float(revenue), 2),
        "costs_gbp": round(float(costs), 2),
        "ebitda_gbp": round(float(ebitda), 2),
        "ebitda_change_gbp": round(float(ebitda - base_ebitda), 2),
        "estimated_operating_cash_gbp": round(float(ebitda * scenario.cash_conversion_pct), 2),
    }


def concentration_metrics(queue: pd.DataFrame) -> dict:
    adverse = queue.loc[queue["favourable_variance_gbp"] < 0].copy()
    adverse["absolute"] = adverse["favourable_variance_gbp"].abs()
    total = adverse["absolute"].sum()
    top_five = adverse.nlargest(5, "absolute")["absolute"].sum()
    return {
        "adverse_items": int(len(adverse)),
        "adverse_variance_gbp": round(float(total), 2),
        "top_five_share_pct": round(float(100 * top_five / total), 1) if total else 0,
        "escalations": int(queue["action"].eq("Escalate").sum()),
    }


def save_outputs(
    output_dir: Path,
    monthly: pd.DataFrame,
    queue: pd.DataFrame,
    summary: dict,
) -> None:
    output_dir.mkdir(parents=True, exist_ok=True)
    monthly.to_csv(output_dir / "monthly_business_unit_pnl.csv", index=False)
    queue.to_csv(output_dir / "variance_action_queue.csv", index=False)
    (output_dir / "executive_summary.json").write_text(
        json.dumps(summary, indent=2), encoding="utf-8"
    )


def main() -> None:
    args = parse_args()
    ledger = load_ledger(args.input)
    controls = control_totals(ledger)
    if controls.missing_values or controls.duplicate_keys:
        raise ValueError(f"Control failure: {asdict(controls)}")

    database = build_database(ledger)
    monthly = query_frame(database, P_AND_L_SQL)
    queue = query_frame(database, VARIANCE_QUEUE_SQL)

    scenarios: Iterable[Scenario] = (
        Scenario(name="Base case"),
        Scenario(
            name="Management recovery",
            revenue_growth_pct=1.5,
            direct_cost_inflation_pct=1.0,
            discretionary_cost_reduction_pct=8.0,
            south_recovery_pct=4.0,
        ),
        Scenario(
            name="Downside",
            revenue_growth_pct=-4.0,
            direct_cost_inflation_pct=5.0,
            south_recovery_pct=-2.0,
            cash_conversion_pct=0.78,
        ),
    )
    summary = {
        "controls": asdict(controls),
        "variance_concentration": concentration_metrics(queue),
        "scenarios": [scenario_summary(ledger, scenario) for scenario in scenarios],
        "latest_month": monthly[monthly["month"].eq(monthly["month"].max())].to_dict("records"),
    }
    save_outputs(args.output_dir, monthly, queue, summary)
    print(json.dumps(summary, indent=2))


if __name__ == "__main__":
    main()