File size: 5,971 Bytes
c7041fa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """Sensitivity analysis of BudgetOptimizer heuristic weights.
Scientifically tests whether the hand-set BUDGET_PRIORITY_DEFAULTS weights are
"load-bearing" for the system's user-facing decisions, by:
(A) Showing the tier-feasibility decision (check_budget_feasibility) is
INVARIANT to weight perturbation (it reads empirical per-destination
cost_tiers data, not the weights).
(B) Monte-Carlo perturbing the weights (each default x U[0.7, 1.3]) and
measuring how much optimize_with_priorities outputs move + whether the
recommended accommodation/food tiers flip.
Run: python -m scripts.sensitivity_analysis (from chatbot-ml-service/)
"""
import json
import os
import random
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.models.budget_optimizer.optimizer import BudgetOptimizer # noqa: E402
KB_PATH = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"app", "data", "knowledge_base", "destinations.json",
)
DEST_IDS = ["da-nang", "phu-quoc", "hoi-an"]
SCENARIOS = [
# (budget_vnd, days, adults, children, month)
(5_000_000, 3, 2, 1, 7),
(8_000_000, 4, 2, 0, 3),
(15_000_000, 5, 2, 2, 12),
(3_000_000, 3, 1, 0, None),
]
USER_PRIORITIES = {"accommodation": "high", "food": "low", "activities": "medium"}
N_DRAWS = 500
PERTURB_LOW, PERTURB_HIGH = 0.70, 1.30 # +/-30% per-category noise
def _perturbed_weights(base: dict, rng: random.Random) -> dict:
return {k: v * rng.uniform(PERTURB_LOW, PERTURB_HIGH) for k, v in base.items()}
def _max_rel_delta(a: dict, b: dict) -> float:
out = 0.0
for k in a:
denom = max(1.0, abs(a[k]))
out = max(out, abs(a[k] - b[k]) / denom)
return out
def main() -> None:
with open(KB_PATH, encoding="utf-8-sig") as f:
dests = json.load(f)["destinations"]
opt = BudgetOptimizer(dests)
base = dict(BudgetOptimizer.BUDGET_PRIORITY_DEFAULTS)
rng = random.Random(42)
print("=" * 78)
print("SENSITIVITY ANALYSIS - BudgetOptimizer heuristic weights")
print(f"base weights: {base}")
print(f"Monte-Carlo: {N_DRAWS} draws, each weight x U[{PERTURB_LOW},{PERTURB_HIGH}]")
print("=" * 78)
# ---- (A) tier-feasibility invariance ------------------------------------
print("\n[A] check_budget_feasibility() - core user-facing decision")
print(f"{'destination':<10}{'budget':>11}{'days':>5}{'pax':>5}"
f"{'tier(base)':>13}{'tier flips/' + str(N_DRAWS):>16}")
total_flips_A = 0
for did in DEST_IDS:
dest = opt.get_destination(did)
for budget, days, ad, ch, month in SCENARIOS:
base_res = opt.check_budget_feasibility(dest, budget, "VND", days, ad, ch, month)
base_tier = base_res.get("recommended_tier") or "infeasible"
flips = 0
for _ in range(N_DRAWS):
opt.BUDGET_PRIORITY_DEFAULTS = _perturbed_weights(base, rng)
r = opt.check_budget_feasibility(dest, budget, "VND", days, ad, ch, month)
if (r.get("recommended_tier") or "infeasible") != base_tier:
flips += 1
opt.BUDGET_PRIORITY_DEFAULTS = base
total_flips_A += flips
print(f"{did:<10}{budget:>11,}{days:>5}{ad + ch:>5}"
f"{base_tier:>13}{flips:>16}")
print(f" => tier-feasibility flips across all cases: {total_flips_A} "
f"(expected 0 - decision uses empirical cost_tiers, NOT weights)")
# ---- (B) optimize_with_priorities allocation sensitivity ----------------
print("\n[B] optimize_with_priorities() - advisory allocation (DOES use weights)")
print(f"user priorities: {USER_PRIORITIES}")
print(f"{'destination':<10}{'budget':>11}{'days':>5}"
f"{'accomTier flips':>16}{'foodTier flips':>16}{'maxRelDelta(mean)':>20}")
worst_delta = 0.0
total_flips_B = 0
for did in DEST_IDS:
dest = opt.get_destination(did)
for budget, days, ad, ch, month in SCENARIOS:
base_res = opt.optimize_with_priorities(
dest, budget, "VND", days, ad, ch, USER_PRIORITIES, month)
base_alloc = base_res["allocation"]
base_tiers = base_res["recommended_tiers"]
a_flip = fd_flip = 0
deltas = []
for _ in range(N_DRAWS):
opt.BUDGET_PRIORITY_DEFAULTS = _perturbed_weights(base, rng)
r = opt.optimize_with_priorities(
dest, budget, "VND", days, ad, ch, USER_PRIORITIES, month)
if r["recommended_tiers"]["accommodation"] != base_tiers["accommodation"]:
a_flip += 1
if r["recommended_tiers"]["food"] != base_tiers["food"]:
fd_flip += 1
deltas.append(_max_rel_delta(base_alloc, r["allocation"]))
opt.BUDGET_PRIORITY_DEFAULTS = base
mean_delta = sum(deltas) / len(deltas)
worst_delta = max(worst_delta, max(deltas))
total_flips_B += a_flip + fd_flip
print(f"{did:<10}{budget:>11,}{days:>5}"
f"{a_flip:>16}{fd_flip:>16}{mean_delta:>19.1%}")
print(f" => recommended-tier flips (accom+food): {total_flips_B} / "
f"{len(DEST_IDS) * len(SCENARIOS) * N_DRAWS * 2}")
print(f" => worst-case max allocation rel-delta under +/-30% weight noise: "
f"{worst_delta:.1%}")
print("\n" + "=" * 78)
print("VERDICT")
print(" (A) Core decision (which tier is affordable / feasible) is "
f"{'INVARIANT' if total_flips_A == 0 else 'SENSITIVE'} to weights")
print(" (B) Advisory allocation tier choices flip in "
f"{total_flips_B} cases; allocation amounts move <= ~weight noise,")
print(" bounded by normalization (Sum=1) -> weights are tunable")
print(" priors, not load-bearing constants.")
print("=" * 78)
if __name__ == "__main__":
main()
|