institutional_metrics.py

import math

import numpy as np
import pandas as pd

from governance_config import load_governance_config
from holding_period_policy import strict_calendar_exit_due


COST_POLICY = load_governance_config("cost_policy")
COST_STRESS_MULTIPLIERS = list(COST_POLICY.get("stress_multipliers") or [1.0, 1.5, 2.0])


def safe_float(value, default=0.0):
    try:
        if value is None or value == "":
            return float(default)
        value = float(value)
        if math.isnan(value) or math.isinf(value):
            return float(default)
        return value
    except Exception:
        return float(default)


def clean_signal(signal):
    signal = pd.Series(signal).replace([np.inf, -np.inf], np.nan).fillna(0.0)
    return signal.clip(-1, 1).round().astype(int)


def infer_bars_per_year(frame):
    if frame is None or "datetime" not in frame or len(frame) < 3:
        return 252.0
    times = pd.to_datetime(frame["datetime"], utc=True, errors="coerce").dropna()
    if len(times) < 3:
        return 252.0
    median_seconds = times.diff().dropna().dt.total_seconds().median()
    if not median_seconds or median_seconds <= 0:
        return 252.0
    if median_seconds >= 18 * 3600:
        return 252.0
    return min(365.0 * 24.0 * 3600.0 / float(median_seconds), 365.0 * 24.0)


def _trade_returns(position, equity, datetimes=None):
    trades = []
    entry_equity = None
    entry_idx = None
    current_pos = 0
    for idx, pos in enumerate(position):
        pos = int(pos)
        if current_pos == 0 and pos != 0:
            entry_equity = float(equity.iloc[idx])
            entry_idx = idx
        elif current_pos != 0 and pos != current_pos:
            if entry_equity and entry_equity > 0:
                trade = {"return": float(equity.iloc[idx] / entry_equity - 1.0), "holding_bars": idx - entry_idx}
                if datetimes is not None and entry_idx is not None:
                    delta = datetimes.iloc[idx] - datetimes.iloc[entry_idx]
                    trade["holding_days"] = safe_float(delta.total_seconds() / 86400.0)
                trades.append(trade)
            entry_equity = float(equity.iloc[idx]) if pos != 0 else None
            entry_idx = idx if pos != 0 else None
        current_pos = pos
    if current_pos != 0 and entry_equity and entry_equity > 0:
        idx = len(position) - 1
        trade = {"return": float(equity.iloc[-1] / entry_equity - 1.0), "holding_bars": idx - entry_idx}
        if datetimes is not None and entry_idx is not None:
            delta = datetimes.iloc[idx] - datetimes.iloc[entry_idx]
            trade["holding_days"] = safe_float(delta.total_seconds() / 86400.0)
        trades.append(trade)
    return trades


def _simulate_execution(
    position,
    frame,
    max_hold_bars=None,
    stop_loss_pct=None,
    take_profit_pct=None,
    execution_timing="next_open",
    max_holding_period=None,
    holding_period_unit="trading_sessions",
    calendar_deadline_policy="last_executable_session_before_deadline",
):
    """Apply next-open execution and return end-of-bar positions plus strategy returns.

    ``override`` contains already signed strategy returns.  This is important on
    signal exits and reversals: the old position owns the previous-close to open
    gap, while the new position (if any) owns only the open to close move.
    """
    closes = pd.to_numeric(frame["close"], errors="coerce").astype(float).to_numpy()
    opens = pd.to_numeric(frame.get("open", frame["close"]), errors="coerce").astype(float).to_numpy()
    highs = pd.to_numeric(frame["high"], errors="coerce").astype(float).to_numpy() if "high" in frame else closes
    lows = pd.to_numeric(frame["low"], errors="coerce").astype(float).to_numpy() if "low" in frame else closes
    raw = position.to_numpy(dtype=int, copy=True)
    n = len(raw)
    eff = np.zeros(n, dtype=int)
    override = np.zeros(n, dtype=float)
    turnover = np.zeros(n, dtype=float)
    held = 0
    entry_ref = 0.0
    entry_t = -1
    actual_holding_days = []
    block_value = None
    max_hold = int(max_holding_period if max_holding_period is not None else (max_hold_bars or 0))
    holding_period_unit = str(holding_period_unit or "trading_sessions")
    if holding_period_unit not in {"trading_sessions", "calendar_days"}:
        raise ValueError(f"unsupported_holding_period_unit:{holding_period_unit}")
    if calendar_deadline_policy != "last_executable_session_before_deadline":
        raise ValueError(f"unsupported_calendar_deadline_policy:{calendar_deadline_policy}")
    if execution_timing not in {"next_open", "next_executable"}:
        raise ValueError(f"unsupported_execution_timing:{execution_timing}")
    datetimes = pd.to_datetime(frame.get("datetime"), utc=True, errors="coerce") if "datetime" in frame else None
    sl = float(stop_loss_pct) if stop_loss_pct else 0.0
    tp = float(take_profit_pct) if take_profit_pct else 0.0
    for t in range(n):
        val = int(raw[t])
        if block_value is not None and val != block_value:
            block_value = None

        changed_at_open = val != held
        overnight_return = 0.0
        if changed_at_open:
            if held != 0 and t > 0 and closes[t - 1] > 0:
                if datetimes is not None and entry_t >= 0 and pd.notna(datetimes.iloc[t]) and pd.notna(datetimes.iloc[entry_t]):
                    actual_holding_days.append(
                        max((datetimes.iloc[t] - datetimes.iloc[entry_t]).total_seconds() / 86400.0, 0.0)
                    )
                overnight_return = held * (opens[t] / closes[t - 1] - 1.0)
                turnover[t] += abs(held)
            held = 0
            entry_t = -1
            if val != 0 and block_value is None and t > 0:
                held = val
                entry_t = t
                entry_ref = opens[t]
                turnover[t] += abs(held)
        elif held == 0 and val != 0 and block_value is None and t > 0:
            held = val
            entry_t = t
            entry_ref = opens[t]
            turnover[t] += abs(held)

        if held == 0:
            override[t] = overnight_return
            eff[t] = 0
            continue
        if not (entry_ref > 0):
            held = 0
            override[t] = overnight_return
            continue
        base = opens[t] if changed_at_open or t == entry_t else closes[t - 1]
        fill = closes[t]
        active_position = held
        exit_now = False
        if held > 0:
            stop_price = entry_ref * (1.0 + sl)
            target_price = entry_ref * (1.0 + tp)
            if sl < 0 and lows[t] <= stop_price:
                fill = opens[t] if opens[t] <= stop_price else stop_price
                exit_now = True
            elif tp > 0 and highs[t] >= target_price:
                fill = opens[t] if opens[t] >= target_price else target_price
                exit_now = True
        else:
            stop_price = entry_ref * (1.0 - sl)
            target_price = entry_ref * (1.0 - tp)
            if sl < 0 and highs[t] >= stop_price:
                fill = opens[t] if opens[t] >= stop_price else stop_price
                exit_now = True
            elif tp > 0 and lows[t] <= target_price:
                fill = opens[t] if opens[t] <= target_price else target_price
                exit_now = True
        intraday_return = held * (fill / base - 1.0) if base > 0 else 0.0
        override[t] = (1.0 + overnight_return) * (1.0 + intraday_return) - 1.0
        if not exit_now and max_hold > 0:
            if holding_period_unit == "trading_sessions":
                exit_now = (t - entry_t + 1) >= max_hold
            elif datetimes is not None:
                exit_now = strict_calendar_exit_due(datetimes, entry_t, t, max_hold)
        if exit_now:
            if datetimes is not None and entry_t >= 0 and pd.notna(datetimes.iloc[t]) and pd.notna(datetimes.iloc[entry_t]):
                actual_holding_days.append(
                    max((datetimes.iloc[t] - datetimes.iloc[entry_t]).total_seconds() / 86400.0, 0.0)
                )
            block_value = held
            turnover[t] += abs(held)
            held = 0
            entry_t = -1
        eff[t] = active_position if exit_now else held
    if held != 0 and entry_t >= 0 and datetimes is not None and n > 0:
        if pd.notna(datetimes.iloc[-1]) and pd.notna(datetimes.iloc[entry_t]):
            actual_holding_days.append(
                max((datetimes.iloc[-1] - datetimes.iloc[entry_t]).total_seconds() / 86400.0, 0.0)
            )
    index = position.index
    override_series = pd.Series(override, index=index)
    override_series.attrs["turnover_series"] = pd.Series(turnover, index=index)
    override_series.attrs["actual_holding_days"] = actual_holding_days
    return pd.Series(eff, index=index), override_series


def backtest_return_series(
    frame,
    signal,
    fee_bps=4.0,
    max_hold_bars=None,
    stop_loss_pct=None,
    take_profit_pct=None,
    execution_timing="next_open",
    max_holding_period=None,
    holding_period_unit="trading_sessions",
    calendar_deadline_policy="last_executable_session_before_deadline",
):
    """Return the executable net-return path used by ``backtest_metrics``.

    Cross-sectional validation needs a synchronized, equal-weight portfolio path;
    returning the path from the same execution model prevents a second, subtly
    different implementation from changing the test assumptions.
    """
    signal = clean_signal(signal).reindex(frame.index).fillna(0).astype(int)
    position = signal.shift(1).fillna(0).astype(int)
    position, gross_returns = _simulate_execution(
        position,
        frame,
        max_hold_bars=max_hold_bars,
        stop_loss_pct=stop_loss_pct,
        take_profit_pct=take_profit_pct,
        execution_timing=execution_timing,
        max_holding_period=max_holding_period,
        holding_period_unit=holding_period_unit,
        calendar_deadline_policy=calendar_deadline_policy,
    )
    turnover_series = gross_returns.attrs["turnover_series"]
    net_returns = gross_returns - turnover_series * (float(fee_bps) / 10000.0)
    return net_returns.astype(float)


def backtest_metrics(
    frame,
    signal,
    fee_bps=4.0,
    max_hold_bars=None,
    stop_loss_pct=None,
    take_profit_pct=None,
    include_return_series=False,
    execution_timing="next_open",
    max_holding_period=None,
    holding_period_unit="trading_sessions",
    calendar_deadline_policy="last_executable_session_before_deadline",
    dynamic_cost_components=None,
):
    signal = clean_signal(signal).reindex(frame.index).fillna(0).astype(int)
    position = signal.shift(1).fillna(0).astype(int)
    position, gross_returns = _simulate_execution(
        position,
        frame,
        max_hold_bars=max_hold_bars,
        stop_loss_pct=stop_loss_pct,
        take_profit_pct=take_profit_pct,
        execution_timing=execution_timing,
        max_holding_period=max_holding_period,
        holding_period_unit=holding_period_unit,
        calendar_deadline_policy=calendar_deadline_policy,
    )
    turnover_series = gross_returns.attrs["turnover_series"]
    required_costs = [name for name, required in (COST_POLICY.get("required_components") or {}).items() if required]
    dynamic_cost_inputs_present = bool(dynamic_cost_components)
    dynamic_series = []
    if dynamic_cost_inputs_present:
        for name in required_costs:
            value = dynamic_cost_components.get(name)
            if isinstance(value, pd.Series):
                series = pd.to_numeric(value.reindex(frame.index), errors="coerce")
            elif isinstance(value, (list, tuple, np.ndarray)) and len(value) == len(frame):
                series = pd.Series(value, index=frame.index, dtype=float)
            else:
                series = None
            if series is None or series.isna().any() or (series < 0).any():
                dynamic_series = []
                break
            dynamic_series.append(series.astype(float))
    dynamic_costs_complete = len(dynamic_series) == len(required_costs) and bool(required_costs)
    applied_cost_series = sum(dynamic_series, pd.Series(0.0, index=frame.index)) if dynamic_costs_complete else (
        turnover_series * (float(fee_bps) / 10000.0)
    )
    net_returns = gross_returns - applied_cost_series
    stress = [float(value) for value in COST_STRESS_MULTIPLIERS]
    while len(stress) < 3:
        stress.append(stress[-1] if stress else 1.0)
    equity = (1.0 + net_returns).cumprod()
    drawdown = equity / equity.cummax() - 1.0
    bars_per_year = infer_bars_per_year(frame)

    datetimes = None
    if "datetime" in frame:
        datetimes = pd.to_datetime(frame["datetime"], utc=True, errors="coerce")
    trades = _trade_returns(position, equity, datetimes)
    trade_array = np.array([item["return"] for item in trades], dtype=float)
    holding_bars = np.array([item.get("holding_bars", 0.0) for item in trades], dtype=float)
    actual_holding_days = gross_returns.attrs.get("actual_holding_days") or []
    holding_days = np.array(actual_holding_days, dtype=float) if actual_holding_days else np.array(
        [item.get("holding_days", 0.0) for item in trades if "holding_days" in item], dtype=float
    )

    periods = max(len(net_returns) - 1, 1)
    final_equity = safe_float(equity.iloc[-1], 1.0)
    total_roi = final_equity - 1.0
    if final_equity > 0:
        annualized_return = final_equity ** (bars_per_year / periods) - 1.0
    else:
        annualized_return = -1.0

    volatility = safe_float(net_returns.std(ddof=0), 0.0)
    sharpe = safe_float((net_returns.mean() / volatility) * math.sqrt(bars_per_year)) if volatility > 0 else 0.0
    downside = net_returns[net_returns < 0]
    downside_vol = safe_float(downside.std(ddof=0), 0.0)
    sortino = safe_float((net_returns.mean() / downside_vol) * math.sqrt(bars_per_year)) if downside_vol > 0 else 0.0

    wins = trade_array[trade_array > 0]
    losses = trade_array[trade_array < 0]
    gross_profit = safe_float(wins.sum(), 0.0)
    gross_loss = abs(safe_float(losses.sum(), 0.0))
    if gross_loss > 0:
        profit_factor = gross_profit / gross_loss
    elif gross_profit > 0:
        profit_factor = 10.0
    else:
        profit_factor = 0.0

    max_drawdown = safe_float(drawdown.min(), 0.0)
    calmar = safe_float(annualized_return / abs(max_drawdown)) if max_drawdown < 0 else 0.0
    exposure = safe_float((position.abs() > 0).mean(), 0.0)
    turnover = safe_float(turnover_series.sum(), 0.0)
    avg_trade_roi = safe_float(trade_array.mean(), 0.0) if len(trade_array) else 0.0
    win_rate = safe_float((trade_array > 0).mean(), 0.0) if len(trade_array) else 0.0

    result = {
        "total_roi": total_roi,
        "annualized_return": safe_float(annualized_return),
        "avg_trade_roi": avg_trade_roi,
        "win_rate": win_rate,
        "profit_factor": safe_float(profit_factor),
        "trades": int(len(trade_array)),
        "max_drawdown": max_drawdown,
        "sharpe": sharpe,
        "sortino": sortino,
        "calmar": calmar,
        "final_equity": final_equity,
        "last_signal": int(signal.iloc[-1]),
        "turnover": turnover,
        "avg_turnover": safe_float(turnover_series.mean(), 0.0),
        "exposure": exposure,
        "avg_holding_bars": safe_float(holding_bars.mean(), 0.0) if len(holding_bars) else 0.0,
        "avg_holding_days": safe_float(holding_days.mean(), 0.0) if len(holding_days) else 0.0,
        "max_holding_bars": safe_float(holding_bars.max(), 0.0) if len(holding_bars) else 0.0,
        "max_holding_days": safe_float(holding_days.max(), 0.0) if len(holding_days) else 0.0,
        "same_day_ratio": safe_float((holding_bars <= 1).mean(), 0.0) if len(holding_bars) else 0.0,
        "within_5d_ratio": safe_float((holding_days <= 5.0).mean(), 0.0) if len(holding_days) else (
            safe_float((holding_bars <= 5).mean(), 0.0) if len(holding_bars) else 0.0
        ),
        "bars_per_year": safe_float(bars_per_year),
        "gross_profit": gross_profit,
        "gross_loss": gross_loss,
        "execution_timing": execution_timing,
        "holding_period_unit": holding_period_unit,
        "calendar_deadline_policy": calendar_deadline_policy,
        "execution_research_status": (
            "DYNAMIC_COST_SERIES_APPLIED"
            if dynamic_costs_complete
            else "RESEARCH_MORE_DYNAMIC_COST_SERIES_INVALID"
            if dynamic_cost_inputs_present
            else str((COST_POLICY.get("legacy_reference") or {}).get("conclusion_cap_when_only_fixed_bps_exists", "RESEARCH_MORE"))
        ),
        "dynamic_cost_series_applied": dynamic_costs_complete,
        "cost_stress_1x_total_roi": safe_float((1.0 + gross_returns - applied_cost_series * stress[0]).prod() - 1.0),
        "cost_stress_1_5x_total_roi": safe_float((1.0 + gross_returns - applied_cost_series * stress[1]).prod() - 1.0),
        "cost_stress_2x_total_roi": safe_float((1.0 + gross_returns - applied_cost_series * stress[2]).prod() - 1.0),
    }
    if include_return_series:
        result["net_return_series"] = net_returns.astype(float)
    return result