"""
Pure-Python Portfolio Engine for EMDS QGIS Port

This module replaces the Windows-only LPA/Flex/VisiRule-based portfolio
processor (executeLPAPortfolio.exe) with a portable, pure-Python implementation
of the greedy-knapsack allocation algorithm that was encoded in portfolio.pl,
process.pl, and dataloader.pl in the ArcGIS Pro EMDS plugin.

Unlike NetWeaverBridge and CDPBridge, no .NET subprocess is needed -- the
portfolio optimization is mathematically simple (greedy knapsack with an
optional multi-silo budget constraint) and therefore runs in-process and
cross-platform (Windows, Linux, macOS).

The public entry point is the ``PortfolioBridge`` class, whose
``process_data`` method takes a pandas DataFrame of candidate projects and
returns a DataFrame with the standard EMDS portfolio output columns:

    ItemId, Cost, Benefit, BudgetCode,
    InPortfolioSort, PortfolioSort,
    CumCost, CumBenefit, ActSort

Algorithm summary (matches portfolio.pl / process.pl):

    1. Compute per-row cost_benefit = Benefit * 10000 / Cost
       (0 if Cost == 0 -- the "priceless" case).
    2. Sort projects by the chosen strategy and direction:
         - cost        => sort on Cost
         - benefit     => sort on Benefit
         - prioritization / prioritization_cost
                       => sort on the prioritization column (benefit-like)
         - costbenefit => partition into
             ZCB  (Cost=0, Benefit>0),  sorted by Benefit
             ZCZB (Cost=0, Benefit=0),  unsorted
             CZB  (Cost>0, Benefit=0),  sorted by Cost
             CB   (Cost>0, Benefit>0),  sorted by cost_benefit ratio
           and concatenate in the order
             down:  [ZCB, CB, ZCZB, CZB]
             up:    [CZB, ZCZB, CB, ZCB]
       Direction semantics (from process.pl sort/4):
           down = high to low  (descending)
           up   = low  to high (ascending)
    3. Walk the sorted list left to right; for each candidate, if its cost
       (split across silos per funding_percent) fits in all three remaining
       silo budgets, pick it. Otherwise skip it. Items that cannot be
       afforded are dropped from further consideration -- this mirrors the
       Prolog ``append(_, [Project|Rest], Projects)`` + ``!`` semantics,
       where the skipped prefix is discarded.
    4. Output: for selected items (in sort order) ActSort = 1..N and
       CumCost/CumBenefit are running totals; then unselected items follow
       with ActSort = N+1..M, with CumCost/CumBenefit continuing.

Author: EMDS QGIS Port POC
Date: 2026-04-16
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional

import pandas as pd

try:
    from emds_qgis_plugin import debug_logger
except Exception:  # pragma: no cover -- plugin package not on sys.path
    debug_logger = None


# --------------------------------------------------------------------------- #
# Strategy / direction constants                                               #
# --------------------------------------------------------------------------- #

STRATEGY_COST = "cost"
STRATEGY_BENEFIT = "benefit"
STRATEGY_COSTBENEFIT = "costbenefit"
STRATEGY_PRIORITIZATION = "prioritization"
STRATEGY_PRIORITIZATION_COST = "prioritization_cost"

_VALID_STRATEGIES = {
    STRATEGY_COST,
    STRATEGY_BENEFIT,
    STRATEGY_COSTBENEFIT,
    STRATEGY_PRIORITIZATION,
    STRATEGY_PRIORITIZATION_COST,
}

DIRECTION_UP = "up"        # ascending (low to high) -- process.pl sort/4
DIRECTION_DOWN = "down"    # descending (high to low)
_VALID_DIRECTIONS = {DIRECTION_UP, DIRECTION_DOWN}


# --------------------------------------------------------------------------- #
# Parameter dataclasses                                                        #
# --------------------------------------------------------------------------- #

@dataclass
class FundingPercent:
    """Three-silo funding percentage for a BudgetCode.

    Mirrors the LPA ``funding_percent(Code, A, B, C)`` dynamic fact. A, B,
    C sum to 100 for the canonical case; the algorithm does not validate
    this because the original Prolog didn't either.
    """

    code: str
    percent_a: float
    percent_b: float
    percent_c: float


@dataclass
class PortfolioReport:
    """Summary statistics emitted alongside the result DataFrame.

    Mirrors the ``data_report(Left, Total, Count, Efficacy)`` fact written
    by portfolio.pl. LeftA/LeftB/LeftC capture the leftover in each silo.
    """

    total_spent: float = 0.0
    total_budget: float = 0.0
    budget_left: float = 0.0
    left_silo_a: float = 0.0
    left_silo_b: float = 0.0
    left_silo_c: float = 0.0
    selected_count: int = 0
    efficacy: float = 0.0           # sum of benefit across selected items
    rows_considered: int = 0
    processing_time_ms: int = 0

    def to_dict(self) -> Dict[str, float]:
        return {
            "total_spent": float(self.total_spent),
            "total_budget": float(self.total_budget),
            "budget_left": float(self.budget_left),
            "left_silo_a": float(self.left_silo_a),
            "left_silo_b": float(self.left_silo_b),
            "left_silo_c": float(self.left_silo_c),
            "selected_count": int(self.selected_count),
            "efficacy": float(self.efficacy),
            "rows_considered": int(self.rows_considered),
            "processing_time_ms": int(self.processing_time_ms),
        }


@dataclass
class PortfolioParameters:
    """Everything the engine needs to run a portfolio allocation.

    Attribute naming tracks the ArcGIS Pro wizard / C# EmdsPortfolioProcessor
    so that the QGIS wizard can hand values straight through.
    """

    unique_field: str                       # e.g. "objectid" -- becomes ItemId
    cost_field: str
    benefit_field: str
    strategy: str                           # one of _VALID_STRATEGIES
    direction: str = DIRECTION_DOWN         # "up" | "down"
    overall_budget: float = 0.0
    prioritization_field: str = ""
    funding_category_field: str = ""
    silo_budgets: Optional[List[float]] = None     # [budget_a, budget_b, budget_c]
    funding_percents: List[FundingPercent] = field(default_factory=list)
    portfolio_name: str = ""
    portfolio_description: str = ""
    benefit_units: str = ""
    cost_units: str = ""


# --------------------------------------------------------------------------- #
# Core engine                                                                  #
# --------------------------------------------------------------------------- #

def _log(level: str, message: str, context: Optional[Dict] = None) -> None:
    if debug_logger is None:
        return
    fn = getattr(debug_logger, level, None)
    if callable(fn):
        if context is None:
            fn(message)
        else:
            fn(message, context)


def _normalize_strategy(strategy: str) -> str:
    key = (strategy or "").strip().lower()
    # Accept a few friendly spellings from the wizard.
    aliases = {
        "cost": STRATEGY_COST,
        "benefit": STRATEGY_BENEFIT,
        "benefit_cost_ratio": STRATEGY_COSTBENEFIT,
        "benefitcostratio": STRATEGY_COSTBENEFIT,
        "costbenefit": STRATEGY_COSTBENEFIT,
        "cost_benefit": STRATEGY_COSTBENEFIT,
        "prioritization": STRATEGY_PRIORITIZATION,
        "prioritization_cost": STRATEGY_PRIORITIZATION_COST,
        "prioritizationcost": STRATEGY_PRIORITIZATION_COST,
        "prioritization_cost_ratio": STRATEGY_PRIORITIZATION_COST,
    }
    if key in _VALID_STRATEGIES:
        return key
    resolved = aliases.get(key)
    if resolved is None:
        raise ValueError(
            f"Unknown portfolio strategy '{strategy}'. "
            f"Expected one of: {sorted(_VALID_STRATEGIES)}"
        )
    return resolved


def _normalize_direction(direction: str) -> str:
    key = (direction or "").strip().lower()
    aliases = {
        "up": DIRECTION_UP,
        "ascending": DIRECTION_UP,
        "asc": DIRECTION_UP,
        "true": DIRECTION_UP,       # legacy isAscending=true
        "down": DIRECTION_DOWN,
        "descending": DIRECTION_DOWN,
        "desc": DIRECTION_DOWN,
        "false": DIRECTION_DOWN,    # legacy isAscending=false
    }
    resolved = aliases.get(key, key)
    if resolved not in _VALID_DIRECTIONS:
        raise ValueError(
            f"Unknown portfolio direction '{direction}'. "
            f"Expected 'up' / 'down' (or 'ascending' / 'descending')."
        )
    return resolved


def _lookup_funding_percent(
    code: str,
    funding_map: Dict[str, FundingPercent],
) -> FundingPercent:
    """Mirror the LPA ``funding_percentage/4`` default: fall back to (100,0,0)."""
    if not code:
        return FundingPercent(code="", percent_a=100.0, percent_b=0.0, percent_c=0.0)
    entry = funding_map.get(code)
    if entry is not None:
        return entry
    # Case-insensitive fallback, because BudgetCode values can be noisy.
    lower = code.lower()
    for key, value in funding_map.items():
        if key.lower() == lower:
            return value
    return FundingPercent(code=code, percent_a=100.0, percent_b=0.0, percent_c=0.0)


def _compute_sort_key(
    *,
    strategy: str,
    cost: float,
    benefit: float,
    prioritization: float,
) -> float:
    """Return the single-axis numeric sort key for non-costbenefit modes."""
    if strategy == STRATEGY_COST:
        return cost
    if strategy == STRATEGY_BENEFIT:
        return benefit
    if strategy == STRATEGY_PRIORITIZATION:
        # prioritization mode uses the prioritization column as the benefit axis
        return prioritization
    if strategy == STRATEGY_PRIORITIZATION_COST:
        # prioritization_cost mode: same axis as prioritization;
        # the cost-benefit handling takes the costbenefit branch below.
        return prioritization
    raise ValueError(f"_compute_sort_key unexpectedly called with strategy={strategy}")


def _safe_float(value, default: float = 0.0) -> float:
    """Coerce engine inputs to float, treating None/NaN/blank as ``default``."""
    if value is None:
        return default
    try:
        fval = float(value)
    except (TypeError, ValueError):
        return default
    if fval != fval:  # NaN check
        return default
    return fval


def _partition_costbenefit(rows: List[dict]) -> tuple[List[dict], List[dict], List[dict], List[dict]]:
    """Four-way partition used by the costbenefit strategy (process.pl)."""
    zcb: List[dict] = []   # Cost=0, Benefit>0
    zczb: List[dict] = []  # Cost=0, Benefit=0
    czb: List[dict] = []   # Cost>0, Benefit=0
    cb: List[dict] = []    # Cost>0, Benefit>0
    for row in rows:
        cost = row["_cost"]
        benefit = row["_benefit"]
        if cost == 0 and benefit > 0:
            zcb.append(row)
        elif cost == 0 and benefit == 0:
            zczb.append(row)
        elif cost > 0 and benefit == 0:
            czb.append(row)
        else:  # cost > 0 and benefit > 0 (negative values fall here too)
            cb.append(row)
    return zcb, zczb, czb, cb


def _sort_rows(rows: List[dict], key: str, direction: str) -> List[dict]:
    """Stable sort; direction 'up'=ascending, 'down'=descending."""
    reverse = direction == DIRECTION_DOWN
    return sorted(rows, key=lambda r: r[key], reverse=reverse)


def _sort_for_strategy(
    *,
    strategy: str,
    direction: str,
    rows: List[dict],
) -> List[dict]:
    """Return ``rows`` sorted per the LPA strategy and direction.

    Non-costbenefit modes use a single-axis sort on ``_sort_key``.
    Costbenefit mode does the 4-way partition from process.pl.
    """
    if strategy != STRATEGY_COSTBENEFIT and strategy != STRATEGY_PRIORITIZATION_COST:
        return _sort_rows(rows, "_sort_key", direction)

    # costbenefit and prioritization_cost both use the 4-way partition;
    # the partition axes are driven by _cost / _benefit which, for
    # prioritization_cost, are cost and prioritization respectively (the
    # caller populates _benefit with the prioritization value in that case).
    zcb, zczb, czb, cb = _partition_costbenefit(rows)
    if direction == DIRECTION_DOWN:
        zcb_sorted = _sort_rows(zcb, "_benefit", DIRECTION_DOWN)
        czb_sorted = _sort_rows(czb, "_cost", DIRECTION_UP)
        cb_sorted = _sort_rows(cb, "_cost_benefit", DIRECTION_DOWN)
        return zcb_sorted + cb_sorted + zczb + czb_sorted
    # direction up
    zcb_sorted = _sort_rows(zcb, "_benefit", DIRECTION_UP)
    czb_sorted = _sort_rows(czb, "_cost", DIRECTION_DOWN)
    cb_sorted = _sort_rows(cb, "_cost_benefit", DIRECTION_UP)
    return czb_sorted + zczb + cb_sorted + zcb_sorted


def _greedy_allocate(
    *,
    sorted_rows: List[dict],
    overall_budget: float,
    silo_budgets: Optional[List[float]],
    funding_map: Dict[str, FundingPercent],
    progress_callback: Optional[Callable[[int, int, int], None]] = None,
) -> None:
    """Walk ``sorted_rows`` left-to-right marking selected items.

    Mutates each row to set ``_selected`` (0 or 1) and
    ``_portfolio_sort`` (1-based position in the sorted iteration).

    Allocation rule (from portfolio.pl):
        * If silo budgets are in effect (any of budget_a/b/c > 0), an item
          can only be picked if all three cumulative silo spends stay
          within their respective silo budgets.
        * Otherwise the overall_budget bound alone applies.
        * Items that don't fit are skipped (InPortfolioSort stays 0). The
          Prolog variant *also* dropped skipped items from the remaining
          search list; the net behaviour on this ordered walk is identical
          because we never revisit an item.
    """

    use_silos = silo_budgets is not None and any(b > 0 for b in silo_budgets)
    budget_a, budget_b, budget_c = (silo_budgets or [0.0, 0.0, 0.0])
    spent_a = spent_b = spent_c = 0.0
    spent_total = 0.0
    total = len(sorted_rows)

    for index, row in enumerate(sorted_rows, start=1):
        row["_portfolio_sort"] = index
        row["_selected"] = 0

        cost = row["_cost"]
        fits = True

        if use_silos:
            fp = _lookup_funding_percent(row["_budget_code"], funding_map)
            cost_a = cost * (fp.percent_a / 100.0)
            cost_b = cost * (fp.percent_b / 100.0)
            cost_c = cost * (fp.percent_c / 100.0)
            if budget_a > 0 and spent_a + cost_a > budget_a:
                fits = False
            if fits and budget_b > 0 and spent_b + cost_b > budget_b:
                fits = False
            if fits and budget_c > 0 and spent_c + cost_c > budget_c:
                fits = False
            if fits:
                spent_a += cost_a
                spent_b += cost_b
                spent_c += cost_c
                spent_total += cost
                row["_selected"] = 1
        else:
            if overall_budget > 0 and spent_total + cost > overall_budget:
                fits = False
            if fits:
                spent_total += cost
                row["_selected"] = 1

        if progress_callback and total:
            percentage = int(index * 100 / total)
            try:
                progress_callback(index, total, percentage)
            except Exception:
                # Progress callbacks must never break the allocation loop.
                pass


def _build_report(
    *,
    rows: List[dict],
    params: PortfolioParameters,
    elapsed_ms: int,
) -> PortfolioReport:
    report = PortfolioReport(
        rows_considered=len(rows),
        processing_time_ms=elapsed_ms,
    )

    use_silos = params.silo_budgets is not None and any(
        b > 0 for b in params.silo_budgets
    )
    budget_a, budget_b, budget_c = (params.silo_budgets or [0.0, 0.0, 0.0])

    funding_map = {fp.code: fp for fp in params.funding_percents}

    spent_total = 0.0
    spent_a = spent_b = spent_c = 0.0
    efficacy = 0.0
    count = 0
    for row in rows:
        if row.get("_selected") != 1:
            continue
        count += 1
        cost = row["_cost"]
        benefit = row["_benefit"]
        spent_total += cost
        efficacy += benefit
        if use_silos:
            fp = _lookup_funding_percent(row["_budget_code"], funding_map)
            spent_a += cost * (fp.percent_a / 100.0)
            spent_b += cost * (fp.percent_b / 100.0)
            spent_c += cost * (fp.percent_c / 100.0)

    report.total_spent = spent_total
    report.selected_count = count
    report.efficacy = efficacy
    if use_silos:
        report.total_budget = budget_a + budget_b + budget_c
        report.budget_left = report.total_budget - spent_total
        report.left_silo_a = max(0.0, budget_a - spent_a)
        report.left_silo_b = max(0.0, budget_b - spent_b)
        report.left_silo_c = max(0.0, budget_c - spent_c)
    else:
        report.total_budget = params.overall_budget
        report.budget_left = max(0.0, params.overall_budget - spent_total)
    return report


def _build_result_dataframe(
    *,
    rows: List[dict],
) -> pd.DataFrame:
    """Project-order rows become the output DataFrame in PortfolioSort order.

    Emits ``ActSort``, ``CumCost``, ``CumBenefit`` columns with the same
    semantics as the C# Program.cs post-processing:
        - Selected items (in PortfolioSort order) get ActSort = 1..N and
          running cumulative totals.
        - Then unselected items (in PortfolioSort order) get ActSort =
          N+1..M with cumulative totals continuing from where the selected
          tail left off.
    """
    # Ensure rows are in PortfolioSort order (they already are -- this is
    # defensive for callers passing an out-of-order list).
    rows_sorted = sorted(rows, key=lambda r: r["_portfolio_sort"])

    cum_cost = 0.0
    cum_benefit = 0.0
    counter = 1
    # First pass: selected items.
    for row in rows_sorted:
        if row["_selected"] == 1:
            cum_cost += row["_cost"]
            cum_benefit += row["_benefit"]
            row["_cum_cost"] = cum_cost
            row["_cum_benefit"] = cum_benefit
            row["_act_sort"] = counter
            counter += 1
    # Second pass: unselected items.
    for row in rows_sorted:
        if row["_selected"] == 0:
            cum_cost += row["_cost"]
            cum_benefit += row["_benefit"]
            row["_cum_cost"] = cum_cost
            row["_cum_benefit"] = cum_benefit
            row["_act_sort"] = counter
            counter += 1

    # Build DataFrame in the canonical EMDS column order.
    columns = [
        "ItemId",
        "Cost",
        "Benefit",
        "BudgetCode",
        "InPortfolioSort",
        "PortfolioSort",
        "CumCost",
        "CumBenefit",
        "ActSort",
    ]
    data = {name: [] for name in columns}
    for row in rows_sorted:
        data["ItemId"].append(row["_item_id"])
        data["Cost"].append(row["_cost"])
        data["Benefit"].append(row["_benefit"])
        data["BudgetCode"].append(row["_budget_code"])
        data["InPortfolioSort"].append(int(row["_selected"]))
        data["PortfolioSort"].append(int(row["_portfolio_sort"]))
        data["CumCost"].append(float(row["_cum_cost"]))
        data["CumBenefit"].append(float(row["_cum_benefit"]))
        data["ActSort"].append(int(row["_act_sort"]))

    return pd.DataFrame(data, columns=columns)


def run_portfolio(
    data: pd.DataFrame,
    params: PortfolioParameters,
    *,
    progress_callback: Optional[Callable[[int, int, int], None]] = None,
) -> tuple[pd.DataFrame, PortfolioReport]:
    """Run the portfolio allocation and return (result_df, report).

    ``data`` may contain any number of extra columns; only the ones named
    in ``params`` are read. The returned DataFrame contains ONLY the
    canonical EMDS portfolio output columns -- the caller is expected to
    merge the result back onto the original geometries / attributes by
    joining on ``ItemId`` (which equals the value of params.unique_field).
    """

    if data is None:
        raise ValueError("Portfolio engine requires a non-None input DataFrame")
    if not isinstance(data, pd.DataFrame):
        raise TypeError("Portfolio engine input must be a pandas DataFrame")

    strategy = _normalize_strategy(params.strategy)
    direction = _normalize_direction(params.direction)

    _log("debug", "Portfolio engine start", {
        "rows": int(len(data.index)),
        "columns": list(data.columns),
        "strategy": strategy,
        "direction": direction,
        "overall_budget": params.overall_budget,
        "silo_budgets": params.silo_budgets,
        "unique_field": params.unique_field,
        "cost_field": params.cost_field,
        "benefit_field": params.benefit_field,
        "prioritization_field": params.prioritization_field,
        "funding_category_field": params.funding_category_field,
        "funding_percents": len(params.funding_percents),
    })

    start = time.time()

    # Resolve field names with a case-insensitive fallback so the wizard
    # doesn't have to match exact casing from a GeoPackage table.
    def _resolve_column(name: str, *, required: bool) -> str:
        if not name:
            if required:
                raise ValueError("Required field name was empty")
            return ""
        if name in data.columns:
            return name
        lower_map = {str(col).lower(): str(col) for col in data.columns}
        resolved = lower_map.get(name.lower())
        if resolved is not None:
            return resolved
        if required:
            raise ValueError(
                f"Input data is missing required column '{name}'. "
                f"Available columns: {list(data.columns)}"
            )
        return ""

    unique_col = _resolve_column(params.unique_field, required=True)
    cost_col = _resolve_column(params.cost_field, required=True)

    uses_prioritization = strategy in (
        STRATEGY_PRIORITIZATION,
        STRATEGY_PRIORITIZATION_COST,
    )
    if uses_prioritization:
        prio_col = _resolve_column(params.prioritization_field, required=True)
        benefit_col = prio_col                     # prioritization takes the benefit slot
    else:
        benefit_col = _resolve_column(params.benefit_field, required=True)
        prio_col = _resolve_column(params.prioritization_field, required=False)

    funding_col = _resolve_column(params.funding_category_field, required=False)

    funding_map = {fp.code: fp for fp in params.funding_percents}

    # Build a list of row-dicts with only the values we care about, so the
    # sort / walk loop touches small homogeneous objects. Also guards
    # against NaN / string-cost upstream problems.
    rows: List[dict] = []
    for idx, row in data.iterrows():
        cost = _safe_float(row[cost_col])
        benefit = _safe_float(row[benefit_col])
        prioritization = _safe_float(row[prio_col]) if prio_col else 0.0
        budget_code = str(row[funding_col]) if funding_col else "Budget1"

        cost_benefit = 0.0 if cost == 0 else (benefit * 10000.0 / cost)

        rows.append({
            "_item_id": row[unique_col],
            "_cost": cost,
            "_benefit": benefit,
            "_prioritization": prioritization,
            "_budget_code": budget_code or "Budget1",
            "_cost_benefit": cost_benefit,
            "_sort_key": _compute_sort_key(
                strategy=strategy,
                cost=cost,
                benefit=benefit,
                prioritization=prioritization,
            ) if strategy != STRATEGY_COSTBENEFIT else cost_benefit,
        })

    sorted_rows = _sort_for_strategy(
        strategy=strategy,
        direction=direction,
        rows=rows,
    )

    _greedy_allocate(
        sorted_rows=sorted_rows,
        overall_budget=params.overall_budget,
        silo_budgets=params.silo_budgets,
        funding_map=funding_map,
        progress_callback=progress_callback,
    )

    result_df = _build_result_dataframe(rows=sorted_rows)
    elapsed_ms = int((time.time() - start) * 1000)
    report = _build_report(rows=sorted_rows, params=params, elapsed_ms=elapsed_ms)

    _log("debug", "Portfolio engine complete", {
        "selected_count": report.selected_count,
        "total_spent": report.total_spent,
        "budget_left": report.budget_left,
        "efficacy": report.efficacy,
        "processing_time_ms": report.processing_time_ms,
    })

    return result_df, report


# --------------------------------------------------------------------------- #
# Bridge wrapper -- API-compatible (in spirit) with NetWeaverBridge/CDPBridge  #
# --------------------------------------------------------------------------- #

class PortfolioBridge:
    """In-process Portfolio engine with a bridge-shaped public API.

    NetWeaverBridge and CDPBridge wrap a .NET CLI subprocess because the
    underlying engines are .NET-only. The portfolio algorithm is pure
    math, so this bridge is a thin wrapper around ``run_portfolio``.
    Keeping the same ``process_data(data, parameters, progress_callback)``
    signature lets the executor treat all three engines uniformly.
    """

    def __init__(self, engine_exe_path: Optional[str] = None):
        """``engine_exe_path`` is accepted for API symmetry but ignored."""
        self.engine_name = "Portfolio"
        self.engine_exe_path = engine_exe_path  # unused; kept for parity
        self.last_report: Optional[PortfolioReport] = None

    # ------------------------------------------------------------------ #
    # Public API                                                         #
    # ------------------------------------------------------------------ #

    def process_data(
        self,
        data: pd.DataFrame,
        *,
        parameters: Optional[Dict] = None,
        progress_callback: Optional[Callable[[int, int, int], None]] = None,
    ) -> pd.DataFrame:
        """Run the portfolio allocation and return the output DataFrame.

        ``parameters`` is the dict produced by the wizard, conforming to
        ``PortfolioParameters`` field names (plus tolerant aliases like
        ``isAscending``, ``overallBudget``, etc.).
        """

        params = self._parameters_from_dict(parameters or {})
        result_df, report = run_portfolio(
            data,
            params,
            progress_callback=progress_callback,
        )
        self.last_report = report
        return result_df

    # ------------------------------------------------------------------ #
    # Parameter coercion from the wizard's dict                          #
    # ------------------------------------------------------------------ #

    @staticmethod
    def _parameters_from_dict(parameters: Dict) -> PortfolioParameters:
        """Parse a loose wizard/executor dict into a PortfolioParameters."""

        def _get(*keys, default=None):
            for key in keys:
                if key in parameters and parameters[key] is not None:
                    return parameters[key]
            return default

        strategy = _get("strategy", "portfolio_strategy", "mode", default=STRATEGY_COST)
        direction = _get(
            "direction", "sort_direction", "portfolio_direction",
            default=None,
        )
        if direction is None:
            is_ascending = _get("isAscending", "is_ascending")
            if is_ascending is not None:
                direction = DIRECTION_UP if bool(is_ascending) else DIRECTION_DOWN
            else:
                direction = DIRECTION_DOWN

        overall_budget = _safe_float(_get("overall_budget", "overallBudget", "budget"), 0.0)

        silo_raw = _get("silo_budgets", "silo_budget", "silos")
        silo_budgets: Optional[List[float]] = None
        if isinstance(silo_raw, (list, tuple)) and len(silo_raw) >= 1:
            silo_budgets = [
                _safe_float(silo_raw[0], 0.0),
                _safe_float(silo_raw[1], 0.0) if len(silo_raw) > 1 else 0.0,
                _safe_float(silo_raw[2], 0.0) if len(silo_raw) > 2 else 0.0,
            ]
        elif isinstance(silo_raw, dict):
            silo_budgets = [
                _safe_float(silo_raw.get("a") or silo_raw.get("budget_a") or silo_raw.get("budget1")),
                _safe_float(silo_raw.get("b") or silo_raw.get("budget_b") or silo_raw.get("budget2")),
                _safe_float(silo_raw.get("c") or silo_raw.get("budget_c") or silo_raw.get("budget3")),
            ]

        funding_percents_raw = _get(
            "funding_percents", "funding_percentages", "funding_table",
            default=[],
        )
        funding_percents: List[FundingPercent] = []
        for item in funding_percents_raw or []:
            if isinstance(item, FundingPercent):
                funding_percents.append(item)
                continue
            if isinstance(item, dict):
                funding_percents.append(FundingPercent(
                    code=str(item.get("code", "")),
                    percent_a=_safe_float(
                        item.get("percent_a") or item.get("budget1") or item.get("a"),
                        0.0,
                    ),
                    percent_b=_safe_float(
                        item.get("percent_b") or item.get("budget2") or item.get("b"),
                        0.0,
                    ),
                    percent_c=_safe_float(
                        item.get("percent_c") or item.get("budget3") or item.get("c"),
                        0.0,
                    ),
                ))
            elif isinstance(item, (list, tuple)) and len(item) >= 4:
                funding_percents.append(FundingPercent(
                    code=str(item[0]),
                    percent_a=_safe_float(item[1]),
                    percent_b=_safe_float(item[2]),
                    percent_c=_safe_float(item[3]),
                ))

        return PortfolioParameters(
            unique_field=str(_get("unique_field", "uniqueFieldName", "unique_field_name") or ""),
            cost_field=str(_get("cost_field", "costFieldName", "cost_field_name") or ""),
            benefit_field=str(_get("benefit_field", "benefitsFieldName", "benefit_field_name") or ""),
            strategy=str(strategy),
            direction=str(direction),
            overall_budget=overall_budget,
            prioritization_field=str(_get(
                "prioritization_field",
                "prioritizationFieldName",
                "priortizationFieldName",
                default="",
            ) or ""),
            funding_category_field=str(_get(
                "funding_category_field",
                "fundingCategoryFieldName",
                default="",
            ) or ""),
            silo_budgets=silo_budgets,
            funding_percents=funding_percents,
            portfolio_name=str(_get("portfolio_name", "portfolioName", default="") or ""),
            portfolio_description=str(_get("portfolio_description", "portfolioDescription", default="") or ""),
            benefit_units=str(_get("benefit_units", "benefitsUnits", default="") or ""),
            cost_units=str(_get("cost_units", "costUnits", default="") or ""),
        )


# --------------------------------------------------------------------------- #
# Smoke test                                                                   #
# --------------------------------------------------------------------------- #

if __name__ == "__main__":  # pragma: no cover -- manual sanity check
    print("EMDS Portfolio Engine - in-process pure-Python replacement")
    print("=" * 60)
    df = pd.DataFrame({
        "objectid": [1, 2, 3, 4, 5, 6],
        "cost_usd": [10, 20, 30, 40, 5, 0],
        "benefit_score": [5, 25, 30, 50, 3, 12],
        "fund_code": ["Budget1"] * 6,
    })
    params = PortfolioParameters(
        unique_field="objectid",
        cost_field="cost_usd",
        benefit_field="benefit_score",
        strategy=STRATEGY_COSTBENEFIT,
        direction=DIRECTION_DOWN,
        overall_budget=60.0,
    )
    result, report = run_portfolio(df, params)
    print(result.to_string(index=False))
    print()
    print("Report:", report.to_dict())
