QuantEngineByItz/tools/build_scenario_shock_matrix_v1.py

#!/usr/bin/env python3
"""SCENARIO_SHOCK_MATRIX_V1 — spec/formulas/domains/simulation.yaml.

Applies 5 deterministic stress multipliers to a base net-profit distribution to
produce per-scenario CE70/CVaR95. governance/todo/v8_9_p2_adoption_plan.yaml P2-A.

Hard rule (AGENTS.md): no base distribution -> all scenarios DATA_MISSING. Never
fabricate a shocked distribution from nothing.
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SIMULATION = ROOT / "Temp" / "forecast_simulation_engine_v1.json"
DEFAULT_DISTRIBUTION = ROOT / "Temp" / "net_profit_distribution_v1.json"
DEFAULT_OUT = ROOT / "Temp" / "scenario_shock_matrix_v1.json"

SCENARIO_DEFINITIONS = {
    "base_case": {"shock_multiplier": 1.0},
    "adverse_case": {"shock_multiplier": 1.5},
    "liquidity_drought_case": {"shock_multiplier": 1.3, "capacity_derate_pct": 40},
    "crisis_case": {"shock_multiplier": 2.0, "correlation_to_one": True},
    "fx_shock_case": {"shock_multiplier": 1.2, "applies_only_to": "foreign_assets"},
    "tax_cost_case": {"shock_multiplier": 1.0, "additional_cost_pct": 5},
}


def _load(path: Path) -> dict:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _quantile(sorted_values: list[float], q: float) -> float:
    if len(sorted_values) == 1:
        return sorted_values[0]
    pos = q * (len(sorted_values) - 1)
    lower = int(pos)
    upper = min(lower + 1, len(sorted_values) - 1)
    frac = pos - lower
    return sorted_values[lower] + (sorted_values[upper] - sorted_values[lower]) * frac


def _cvar95(sorted_values: list[float]) -> float:
    threshold_idx = max(1, int(len(sorted_values) * 0.05))
    tail = sorted_values[:threshold_idx]
    return sum(tail) / len(tail)


def apply_shock(distribution: list[float], scenario_id: str) -> list[float]:
    scenario = SCENARIO_DEFINITIONS[scenario_id]
    multiplier = scenario["shock_multiplier"]
    additional_cost_pct = scenario.get("additional_cost_pct", 0)
    shocked = []
    for v in distribution:
        value = v * multiplier if v < 0 else v / multiplier
        if additional_cost_pct:
            value -= abs(value) * (additional_cost_pct / 100.0)
        shocked.append(value)
    return shocked


def evaluate_scenario(distribution: list[float] | None, scenario_id: str) -> dict:
    if not distribution:
        return {
            "scenario_id": scenario_id,
            "scenario_ce70_krw": None,
            "scenario_cvar95_krw": None,
            "gate": "DATA_MISSING",
        }
    shocked = sorted(apply_shock(distribution, scenario_id))
    return {
        "scenario_id": scenario_id,
        "scenario_ce70_krw": _quantile(shocked, 0.30),
        "scenario_cvar95_krw": _cvar95(shocked),
        "gate": "PASS",
    }


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--distribution", default=str(DEFAULT_DISTRIBUTION))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    doc = _load(Path(args.distribution))
    distribution = doc.get("net_profit_distribution_after_tax_fee_slippage")
    distribution = [float(v) for v in distribution] if isinstance(distribution, list) and distribution else None

    scenario_results = [evaluate_scenario(distribution, scenario_id) for scenario_id in SCENARIO_DEFINITIONS]

    result = {
        "formula_id": "SCENARIO_SHOCK_MATRIX_V1",
        "gate": "PASS" if distribution else "DATA_MISSING",
        "scenario_results": scenario_results,
        "source_paths": [str(Path(args.distribution))],
    }
    out = Path(args.out)
    out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())