#!/usr/bin/env python3 """Offline fixed-panel exercise; Python 3.9+, standard library only. Synthetic comparable regional quotes in one currency and price basis. No fetching, imputation, automatic repricing or provider benchmark. Weights are editorial policy inputs, not measured sales or population shares. Observed/stale/missing are supplied evidence states, not verified by this script. """ from collections import Counter from copy import deepcopy from decimal import Decimal as D import json WEIGHTS = {k: D('0.25') for k in 'ABCD'} BASE = {'A': '80', 'B': '100', 'C': '120', 'D': '200'} def decimal(value): if isinstance(value, bool): raise ValueError('Boolean is not a price or weight') n = D(str(value)) if not n.is_finite(): raise ValueError('Finite values required') return n def evaluate(rows, weights): weights = {k: decimal(v) for k, v in weights.items()} if not weights or any(w <= 0 for w in weights.values()) or sum(weights.values()) != 1: raise ValueError('Positive fixed weights must sum to one') keys = [r['region'] for r in rows] if len(set(keys)) != len(keys) or set(keys) != set(weights): raise ValueError('Exactly one row per planned region; retain missing rows') base = {}; current = {}; omitted = [] for r in rows: k = r['region']; b = decimal(r['base']) if b <= 0: raise ValueError('Positive baseline needed for a price relative') base[k] = b state = r['state'] if state not in ('observed', 'missing', 'stale'): raise ValueError('Declare observed, missing or stale') if state != 'observed': omitted.append(k) continue if r['current'] is None: raise ValueError('Observed needs a price; use missing state for unknown') price = decimal(r['current']) if price < 0: raise ValueError('Negative quote outside this exercise') current[k] = price # Verified zero is a value, not a missing observation. coverage = sum((weights[k] for k in current), D(0)) contribution = sum((weights[k] * (current[k] / base[k] - 1) for k in current), D(0)) subset_change = contribution / coverage if coverage else None full_change = contribution if coverage == 1 else None naive_change = ((sum(current.values()) / len(current)) / (sum(base.values()) / len(base)) - 1) if current else None return { 'planned_regions': len(weights), 'observed_regions': len(current), 'count_coverage': D(len(current)) / len(weights), 'weight_coverage': coverage, 'omitted_regions': sorted(omitted), 'known_contribution': contribution, 'observed_subset_change': subset_change, 'full_panel_change': full_change, 'naive_unpaired_mean_change': naive_change, 'decision': 'REVIEW_FULL_PANEL' if coverage == 1 else 'HOLD_FULL_PANEL', } def bounds(report, low, high): """Explicit hypothetical bounds, not imputation or a confidence interval.""" low, high = decimal(low), decimal(high) if low < -1 or high < low: raise ValueError('Invalid hypothetical relative-change bounds') missing_weight = 1 - report['weight_coverage'] return [report['known_contribution'] + missing_weight * low, report['known_contribution'] + missing_weight * high] def scenarios(): complete = [{'region': k, 'base': b, 'current': str(D(b) * D('1.1')), 'state': 'observed'} for k, b in BASE.items()] high_missing = deepcopy(complete); high_missing[3].update(current=None, state='missing') low_missing = deepcopy(complete); low_missing[0].update(current=None, state='missing') recovered_fall = deepcopy(complete); recovered_fall[3]['current'] = '120' verified_zero = deepcopy(complete); verified_zero[3]['current'] = '0' stale_high = deepcopy(complete); stale_high[3]['state'] = 'stale' return [('complete_rise', complete), ('expensive_region_missing', high_missing), ('cheapest_region_missing', low_missing), ('alternative_recovery_fall', recovered_fall), ('confirmed_zero_quote', verified_zero), ('stale_high_quote', stale_high)] def run(): items = scenarios(); reports = {name: evaluate(rows, WEIGHTS) for name, rows in items} expected = { 'complete_rise': (D(1), D('.1'), D('.1'), D('.1')), 'expensive_region_missing': (D('.75'), D('.1'), None, D('-.12')), 'cheapest_region_missing': (D('.75'), D('.1'), None, D('.232')), 'alternative_recovery_fall': (D(1), D('-.025'), D('-.025'), D('-.1')), 'confirmed_zero_quote': (D(1), D('-.175'), D('-.175'), D('-.34')), 'stale_high_quote': (D('.75'), D('.1'), None, D('-.12')), } for name, values in expected.items(): got = reports[name] if tuple(got[k] for k in ('weight_coverage', 'observed_subset_change', 'full_panel_change', 'naive_unpaired_mean_change')) != values: raise AssertionError(name) interval = bounds(reports['expensive_region_missing'], '-.4', '.1') if interval != [D('-.025'), D('.1')]: raise AssertionError('Sensitivity bounds') # Counterexamples: reject duplicate sampling, silent new regions, bad # baselines and bad weights; no observations must remain unknown. tests = [] duplicate = deepcopy(items[0][1]); duplicate.append(deepcopy(duplicate[0])) tests.append((duplicate, WEIGHTS)) new_region = deepcopy(items[0][1]); new_region[-1]['region'] = 'E' tests.append((new_region, WEIGHTS)) zero_base = deepcopy(items[0][1]); zero_base[0]['base'] = '0' tests.append((zero_base, WEIGHTS)) tests.append((items[0][1], {k: D('.2') for k in 'ABCD'})) for rows, weights in tests: try: evaluate(rows, weights) except ValueError: pass else: raise AssertionError('Invalid panel accepted') empty = deepcopy(items[0][1]) for row in empty: row.update(current=None, state='missing') empty_result = evaluate(empty, WEIGHTS) if empty_result['full_panel_change'] is not None or empty_result['observed_subset_change'] is not None or empty_result['decision'] != 'HOLD_FULL_PANEL': raise AssertionError('Empty panel became a price signal') return {'mode': 'offline_synthetic_fixed_panel', 'network_requests': 0, 'scenarios': reports, 'decision_counts': dict(Counter(x['decision'] for x in reports.values())), 'hypothetical_missing_change_bounds': ['-0.4', '0.1'], 'full_panel_sensitivity_bounds': interval, 'counterexamples_passed': 5, 'limits': 'Declared evidence states and equal editorial weights; no real prices, provider measurements, market representativeness or automatic business action.'} if __name__ == '__main__': print(json.dumps(run(), indent=2, default=str))