#!/usr/bin/env python3 """Read-only receipt checks. Standard library only; no network or ledger writes.""" import argparse import calendar from datetime import date, datetime, timezone from decimal import Decimal import json from pathlib import Path import re import sys from urllib.parse import urlparse VERSION = '1.0' STATES = {'pending', 'met', 'missed', 'mixed', 'unresolved', 'not_operationally_defined', 'condition_not_met', 'attribution_unresolved'} RESOLVED = {'met', 'missed', 'mixed'} MEASURES = {'fulfillment', 'signed_error', 'absolute_error', 'relative_absolute_error', 'threshold_shortfall', 'days_late', 'brier_loss'} def number(value): return type(value) in (int, float, Decimal) and Decimal(str(value)).is_finite() def reject_constant(value): raise ValueError('Non-finite JSON number: ' + value) def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError('Duplicate JSON key: ' + key) result[key] = value return result def load_receipt(path): # Never write or execute receipt content, formulas, URLs or copy locations. return json.loads(Path(path).read_text(encoding='utf-8'), parse_float=Decimal, parse_constant=reject_constant, object_pairs_hook=unique_object) def date_bounds(value): """Validate ISO dates and return UTC bounds; preserve year/month uncertainty.""" if not isinstance(value, str): raise ValueError('Expected ISO date or timestamp') if re.fullmatch(r'\d{4}', value): year = int(value) return datetime(year, 1, 1, tzinfo=timezone.utc), datetime(year, 12, 31, 23, 59, 59, tzinfo=timezone.utc) if re.fullmatch(r'\d{4}-\d{2}', value): year, month = map(int, value.split('-')) last = calendar.monthrange(year, month)[1] return datetime(year, month, 1, tzinfo=timezone.utc), datetime(year, month, last, 23, 59, 59, tzinfo=timezone.utc) if re.fullmatch(r'\d{4}-\d{2}-\d{2}', value): day = date.fromisoformat(value) return datetime.combine(day, datetime.min.time(), timezone.utc), datetime.combine(day, datetime.max.time(), timezone.utc) if not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})', value): raise ValueError('Use ISO dates or timestamps with a timezone') stamp = datetime.fromisoformat(value.replace('Z', '+00:00')).astimezone(timezone.utc) return stamp, stamp def text(value): return isinstance(value, str) and bool(value.strip()) class Validator: def __init__(self, receipt): self.r = receipt self.errors = [] self.incomplete = [] self.warnings = [] def error(self, path, message): self.errors.append({'path': path, 'message': message}) def missing(self, path): self.incomplete.append({'path': path, 'message': 'Required for a complete assessment'}) def shape(self, value, path, strings='', objects='', arrays='', booleans='', numbers='', optional=''): error_count = len(self.errors) if not isinstance(value, dict): self.error(path, 'Expected an object') return False groups = {'string': strings.split(), 'object': objects.split(), 'array': arrays.split(), 'boolean': booleans.split(), 'number': numbers.split()} allowed = set(optional.split()) | {k for fields in groups.values() for k in fields} for key in value.keys() - allowed: self.error(path + '.' + key, 'Unknown field for receipt template 1.0') for kind, fields in groups.items(): for key in fields: p = path + '.' + key if key not in value: self.error(p, 'Missing field') continue item = value[key] if item is None and kind in {'string', 'boolean', 'number'}: continue valid = {'string': lambda: isinstance(item, str), 'object': lambda: isinstance(item, dict), 'array': lambda: isinstance(item, list), 'boolean': lambda: type(item) is bool, 'number': lambda: number(item)}[kind]() if not valid: self.error(p, 'Expected ' + kind + (' or null' if kind in {'string', 'boolean', 'number'} else '')) return len(self.errors) == error_count def refs(self, values, path, known): if isinstance(values, list): if any(not text(v) for v in values): self.error(path, 'Evidence references must be nonempty string IDs') else: if len(values) != len(set(values)): self.error(path, 'Duplicate evidence reference') for value in values: if value not in known: self.error(path, 'Unknown evidence ID: ' + value) def validate(self): r = self.r if not self.shape(r, '$', strings='receipt_template_version methodology_version methodology_url receipt_kind publication_status assessment_id assessment_date', objects='forecast resolution_contract assessment aggregate_treatment review revision historical_preservation', arrays='evidence observations calculations', optional='example_notice'): return self.report() if r.get('receipt_template_version') != VERSION or r.get('methodology_version') != VERSION: self.error('$', 'Only receipt template 1.0 and methodology 1.0 are supported') kind = r.get('receipt_kind') status = r.get('publication_status') if kind not in {'blank_template', 'hypothetical_example', 'assessment'}: self.error('$.receipt_kind', 'Unsupported receipt kind') if status not in {'draft', 'example_only', 'published'}: self.error('$.publication_status', 'Unsupported publication status') if kind == 'blank_template' and status != 'draft': self.error('$.publication_status', 'Blank templates must remain drafts') if kind == 'hypothetical_example' and (status != 'example_only' or not text(r.get('example_notice'))): self.error('$', 'Examples require example_only status and an explicit notice') if kind != 'hypothetical_example' and ('example_notice' in r or status == 'example_only'): self.error('$', 'Example-only fields require hypothetical_example kind') f, c, a, g, review, rev, history = [r.get(k, {}) for k in ('forecast','resolution_contract','assessment','aggregate_treatment','review','revision','historical_preservation')] self.shape(f, '$.forecast', strings='record_id forecast_family_id forecaster exact_quote statement_date date_precision primary_source_url archived_source_url source_locator', arrays='qualifiers_and_conditions') self.shape(c, '$.resolution_contract', strings='contract_id contract_version registered_at frozen_at frozen_copy_location designation knowledge_timing_note event_or_quantity population success_rule allowed_human_assistance acceptable_primary_evidence review_publication_cutoff', objects='measurement deadline explicit_probability', arrays='express_conditions independent_subclaims') self.shape(a, '$.assessment', strings='state as_of reason', arrays='evidence_ids condition_findings subclaim_findings limitations_and_gaps') self.shape(g, '$.aggregate_treatment', strings='prospective_aggregate_inclusion reason existing_historical_eligibility_action', booleans='initial_forecast_in_family designated_family_representative') self.shape(review, '$.review', strings='assessor assessed_at second_reviewer reviewed_at disagreement_note', booleans='reproduced_from_receipt material_interpretation_disagreement') self.shape(rev, '$.revision', strings='previous_receipt_id reason', numbers='receipt_revision', arrays='changes', optional='previous_assessment') self.shape(history, '$.historical_preservation', strings='scope', arrays='legacy_fields_to_write') if self.errors: return self.report() measurement, deadline, probability = c['measurement'], c['deadline'], c['explicit_probability'] self.shape(measurement, '$.resolution_contract.measurement', strings='metric denominator units sampling_method') self.shape(deadline, '$.resolution_contract.deadline', strings='start end timezone precision') self.shape(probability, '$.resolution_contract.explicit_probability', strings='quoted_expression source_locator', numbers='value_0_to_1') if self.errors: return self.report() if history['legacy_fields_to_write'] != [] or g['existing_historical_eligibility_action'] != 'no_change': self.error('$.historical_preservation', 'Legacy writes or eligibility changes are prohibited') if g['prospective_aggregate_inclusion'] not in {'not_evaluated', 'excluded', 'included'}: self.error('$.aggregate_treatment.prospective_aggregate_inclusion', 'Unsupported inclusion value') state = a['state'] if state is not None and state not in STATES: self.error('$.assessment.state', 'Unsupported assessment state') if f['date_precision'] not in {None, 'exact_day', 'approximate', 'unknown'}: self.error('$.forecast.date_precision', 'Unsupported date precision') if c['designation'] not in {None, 'prospective', 'retrospective'}: self.error('$.resolution_contract.designation', 'Use prospective or retrospective') if deadline.get('precision') not in {None, 'exact_day', 'month', 'year', 'interval', 'approximate', 'unknown'}: self.error('$.resolution_contract.deadline.precision', 'Unsupported deadline precision') p = probability.get('value_0_to_1') if p is not None and number(p) and not 0 <= p <= 1: self.error('$.resolution_contract.explicit_probability.value_0_to_1', 'Probability must be between 0 and 1') evidence_ids = set() for i, ev in enumerate(r['evidence']): path = '$.evidence[' + str(i) + ']' if not self.shape(ev, path, strings='evidence_id primary_source_url archived_source_url publisher_or_author source_type publication_date observation_start observation_end exact_passage locator supports', arrays='limitations'): continue ident = ev.get('evidence_id') if not text(ident) or ident in evidence_ids: self.error(path + '.evidence_id', 'Missing or duplicate evidence ID') else: evidence_ids.add(ident) observation_ids = set() for i, obs in enumerate(r['observations']): path = '$.observations[' + str(i) + ']' if not self.shape(obs, path, strings='observation_id metric units population_measured observation_start observation_end', numbers='value', objects='uncertainty', arrays='evidence_ids limitations'): continue ident = obs.get('observation_id') if not text(ident) or ident in observation_ids: self.error(path + '.observation_id', 'Missing or duplicate observation ID') else: observation_ids.add(ident) self.shape(obs['uncertainty'], path + '.uncertainty', strings='description', numbers='lower upper') lo, hi = obs['uncertainty'].get('lower'), obs['uncertainty'].get('upper') if number(lo) and number(hi) and lo > hi: self.error(path + '.uncertainty', 'Lower bound exceeds upper bound') self.refs(obs.get('evidence_ids'), path + '.evidence_ids', evidence_ids) self.refs(a['evidence_ids'], '$.assessment.evidence_ids', evidence_ids) subcontracts = c['independent_subclaims'] findings = a['subclaim_findings'] sub_ids, found_ids = set(), set() for i, sub in enumerate(subcontracts): path = '$.resolution_contract.independent_subclaims[' + str(i) + ']' if self.shape(sub, path, strings='subclaim_id exact_statement success_rule'): ident = sub.get('subclaim_id') if not text(ident) or ident in sub_ids: self.error(path, 'Missing or duplicate subclaim ID') else: sub_ids.add(ident) for i, finding in enumerate(findings): path = '$.assessment.subclaim_findings[' + str(i) + ']' if self.shape(finding, path, strings='subclaim_id state reason', arrays='evidence_ids'): ident = finding.get('subclaim_id') if ident not in sub_ids or ident in found_ids: self.error(path, 'Unknown or duplicate subclaim finding') else: found_ids.add(ident) if finding.get('state') not in STATES: self.error(path + '.state', 'Unsupported subclaim state') self.refs(finding.get('evidence_ids'), path + '.evidence_ids', evidence_ids) if state == 'mixed' and (not subcontracts or found_ids != sub_ids or not {'met','missed'} <= {s.get('state') for s in findings if isinstance(s, dict)}): self.error('$.assessment.state', 'Mixed requires separately contracted met and missed subclaims') for i, condition in enumerate(c['express_conditions']): if not text(condition): self.error('$.resolution_contract.express_conditions['+str(i)+']', 'Preserve condition as nonempty exact text') if self.errors: return self.report() condition_texts = [] for i, finding in enumerate(a['condition_findings']): path = '$.assessment.condition_findings['+str(i)+']' if self.shape(finding, path, strings='condition state reason', arrays='evidence_ids'): condition_texts.append(finding.get('condition')) if finding.get('condition') not in c['express_conditions'] or finding.get('state') not in {'met','not_met','unknown'}: self.error(path, 'Unknown condition or unsupported condition state') self.refs(finding.get('evidence_ids'), path+'.evidence_ids', evidence_ids) if len(condition_texts) != len(set(condition_texts)): self.error('$.assessment.condition_findings', 'Duplicate condition finding') if self.errors: return self.report() if state in RESOLVED and (set(condition_texts) != set(c['express_conditions']) or any(x.get('state') != 'met' for x in a['condition_findings'])): self.error('$.assessment.condition_findings', 'Resolved forecasts require all stated prerequisites met') if state == 'condition_not_met' and not any(x.get('state') == 'not_met' for x in a['condition_findings']): self.error('$.assessment.state', 'condition_not_met requires an unmet prerequisite finding') for i, calc in enumerate(r['calculations']): self.calculation(calc, i, evidence_ids) if kind == 'blank_template' and (state is not None or r['calculations'] or r['evidence'] or r['observations']): self.error('$', 'Blank template cannot contain an assessment, observations, evidence or calculations') if kind == 'hypothetical_example': if g['prospective_aggregate_inclusion'] != 'excluded' or f['record_id'] is not None: self.error('$', 'Examples must be excluded and must not assess a canonical record') self.warnings.append('Fictional example only; never eligible as a real assessment.') if g['prospective_aggregate_inclusion'] == 'included': if kind != 'assessment' or state not in RESOLVED or c['designation'] != 'prospective' or g['initial_forecast_in_family'] is not True or g['designated_family_representative'] is not True: self.error('$.aggregate_treatment', 'Included receipts require a resolved prospective initial family representative') if not r['calculations'] or all(x.get('result_unrounded') is None and x.get('exact_fraction') is None for x in r['calculations'] if isinstance(x,dict)): self.error('$.aggregate_treatment', 'Included receipts require a reproducible numeric result') n = rev['receipt_revision'] if 'previous_assessment' not in rev: self.error('$.revision.previous_assessment', 'Missing field') if type(n) is not int or n < 1: self.error('$.revision.receipt_revision', 'Expected a positive integer') if n == 1 and (rev['previous_receipt_id'] is not None or rev.get('previous_assessment') is not None or rev['changes']): self.error('$.revision', 'Initial receipts cannot overwrite a previous assessment') if number(n) and n > 1 and (not text(rev['previous_receipt_id']) or rev.get('previous_assessment') is None or not rev['changes'] or not text(rev['reason'])): self.error('$.revision', 'Revisions require prior receipt, prior assessment, explicit changes and reason') for i, change in enumerate(rev['changes']): path = '$.revision.changes['+str(i)+']' if self.shape(change, path, strings='field reason', arrays='evidence_ids', optional='old_value new_value'): if 'old_value' not in change or 'new_value' not in change: self.error(path, 'Retain both old_value and new_value') self.refs(change.get('evidence_ids'), path+'.evidence_ids', evidence_ids) self.dates_and_urls(r) self.completeness() if status == 'published' or g['prospective_aggregate_inclusion'] == 'included': for item in self.incomplete: self.error(item['path'], 'Published/included receipt is incomplete') return self.report() def calculation(self, calc, i, known): path = '$.calculations['+str(i)+']' if not self.shape(calc, path, strings='measure formula units withheld_reason', objects='inputs', numbers='result_unrounded', arrays='evidence_ids', optional='exact_fraction'): return measure, inputs, result = calc.get('measure'), calc.get('inputs', {}), calc.get('result_unrounded') self.refs(calc.get('evidence_ids'), path+'.evidence_ids', known) fraction = calc.get('exact_fraction') if measure not in MEASURES: self.error(path+'.measure', 'Unsupported measure; arbitrary editorial scores are prohibited') return if fraction is not None: if not isinstance(fraction,dict) or set(fraction) != {'numerator','denominator'} or any(type(fraction.get(k)) is not int for k in ('numerator','denominator')) or fraction['denominator'] <= 0: self.error(path+'.exact_fraction', 'Expected integer numerator and positive integer denominator') return scored = result is not None or fraction is not None if not scored: if not text(calc.get('withheld_reason')): self.error(path+'.withheld_reason', 'A null result requires a reason') return if calc.get('withheld_reason') is not None: self.error(path, 'Withheld calculations cannot contain numeric results') state = self.r['assessment']['state'] if state not in RESOLVED or self.r['review']['material_interpretation_disagreement'] is True: self.error(path, 'Unresolved outcomes or material disagreements require null results') if not calc.get('evidence_ids'): self.error(path+'.evidence_ids', 'A numeric result requires evidence references') try: def native(key): value = inputs[key] if not number(value): raise ValueError('Input '+key+' must be a finite JSON number, not text or boolean') return Decimal(str(value)) if measure == 'brier_loss': expected_keys = {'p','y'} p, y = native('p'), native('y') explicit = self.r['resolution_contract']['explicit_probability'] if not 0 <= p <= 1 or y not in (0,1) or state not in {'met','missed'}: raise ValueError('Brier requires probability in [0,1] and a resolved binary outcome') if y != (1 if state == 'met' else 0) or explicit.get('value_0_to_1') != p: raise ValueError('Brier inputs disagree with the assessment or quoted probability') quote = explicit.get('quoted_expression') if not text(quote) or quote not in (self.r['forecast'].get('exact_quote') or '') or not text(explicit.get('source_locator')): raise ValueError('Brier requires the explicit probability quotation and its locator') tokens = re.findall(r'\d+(?:\.\d+)?', quote) if len(tokens) != 1: raise ValueError('Probability ranges or ambiguous quotations cannot supply a point probability') quoted_p = Decimal(tokens[0]) / (100 if '%' in quote or re.search(r'\bpercent\b',quote,re.I) else 1) if quoted_p != p: raise ValueError('Quoted probability does not match p') expected = (p-y)**2 elif measure == 'fulfillment': subs = self.r['resolution_contract']['independent_subclaims'] findings = self.r['assessment']['subclaim_findings'] if subs: expected_keys = {'met_count','total_count'} m, total = native('met_count'), native('total_count') if len(findings) != len(subs) or any(s.get('state') not in {'met','missed'} for s in findings): raise ValueError('All independent subclaims must resolve before compound fulfillment') if total != len(subs) or m != sum(s.get('state') == 'met' for s in findings): raise ValueError('Counts must preserve every contracted subclaim') derived = 'met' if m == total else 'missed' if m == 0 else 'mixed' if state != derived: raise ValueError('Compound assessment state disagrees with subclaim results') expected = m / total else: expected_keys = {'y'} y = native('y') if state not in {'met','missed'} or y != (1 if state == 'met' else 0): raise ValueError('Indivisible event fulfillment is exactly 0 or 1') expected = y elif measure == 'days_late': expected_keys = {'achievement_date','deadline'} if any(not re.fullmatch(r'\d{4}-\d{2}-\d{2}', str(inputs[k])) for k in expected_keys): raise ValueError('days_late uses exact ISO calendar dates; uncertain dates need a withheld result') end = self.r['resolution_contract']['deadline'].get('end') or '' if not end.startswith(inputs['deadline']) or self.r['resolution_contract']['deadline'].get('precision') != 'exact_day': raise ValueError('Calculation deadline must match the exact-day contract deadline') expected = Decimal((date.fromisoformat(inputs['achievement_date'])-date.fromisoformat(inputs['deadline'])).days) if expected < 0 or (expected > 0 and state != 'missed'): raise ValueError('A late achievement requires a missed deadline') elif measure == 'threshold_shortfall': expected_keys = {'threshold','observed_value','comparison'} t, o = native('threshold'), native('observed_value') if inputs['comparison'] != 'at_least': raise ValueError('This validator supports at_least shortfalls; other directions require manual review') expected = max(Decimal(0),t-o) if (o >= t) != (state == 'met') or state not in {'met','missed'}: raise ValueError('Threshold outcome disagrees with the assessment') else: expected_keys = {'predicted_value','observed_value'} predicted, observed = native('predicted_value'), native('observed_value') expected = predicted-observed if measure != 'signed_error': expected = abs(expected) if measure == 'relative_absolute_error': if observed == 0: raise ValueError('Relative error is undefined when observed_value is zero') expected /= abs(observed) if set(inputs) != expected_keys: raise ValueError('Unexpected or missing calculation inputs') if measure != 'fulfillment': observations = self.r['observations'] matches = [o for o in observations if isinstance(o,dict) and set(o.get('evidence_ids',[])) & set(calc['evidence_ids'])] if measure == 'days_late': matches = [o for o in matches if str(o.get('observation_start','')).startswith(inputs['achievement_date']) or str(o.get('observation_end','')).startswith(inputs['achievement_date'])] else: observed_input = inputs['y'] if measure == 'brier_loss' else inputs['observed_value'] matches = [o for o in matches if o.get('value') == observed_input and o.get('metric') == self.r['resolution_contract']['measurement']['metric']] for o in matches: uncertainty = o.get('uncertainty',{}) lo, hi = uncertainty.get('lower'), uncertainty.get('upper') if number(lo) and number(hi) and lo != hi: raise ValueError('Uncertain measurements require an interval/manual review, not a scalar result') if not matches: raise ValueError('Calculation outcome input lacks a matching cited observation') end = self.r['resolution_contract']['deadline'].get('end') if end and measure != 'days_late': for obs in matches: if obs.get('observation_start') and date_bounds(obs['observation_start'])[0] > date_bounds(end)[1]: raise ValueError('Later observations cannot substitute for the deadline outcome') for actual in ([Decimal(str(result))] if number(result) else []) + ([Decimal(fraction['numerator'])/fraction['denominator']] if fraction is not None else []): if abs(actual-expected) > Decimal('1e-12')*max(Decimal(1),abs(expected)): raise ValueError('Result does not reproduce from the supplied inputs') except (KeyError, ValueError, TypeError, ArithmeticError) as exc: self.error(path, str(exc)) def dates_and_urls(self, value, path='$'): if isinstance(value, dict): for key, item in value.items(): child = path+'.'+key if item is not None and (key.endswith('_date') or key in {'registered_at','frozen_at','review_publication_cutoff','as_of','assessed_at','reviewed_at','observation_start','observation_end'} or path.endswith('.deadline') and key in {'start','end'}): try: date_bounds(item) except (ValueError, TypeError, OverflowError) as exc: self.error(child, str(exc)) if item is not None and (key.endswith('_url') or key == 'methodology_url'): try: parsed = urlparse(item) if parsed.scheme not in {'http','https'} or not parsed.hostname or parsed.username or parsed.password: raise ValueError('Expected an HTTP(S) URL without credentials') host = parsed.hostname.lower() if self.r.get('receipt_kind') != 'hypothetical_example' and (host in {'localhost','example.com','example.org','example.net'} or host.endswith('.invalid') or host in {'invalid','127.0.0.1','::1'}): raise ValueError('Placeholder URLs cannot support a real assessment') except (ValueError, TypeError) as exc: self.error(child, str(exc)) self.dates_and_urls(item, child) for start, end in [('observation_start','observation_end'),('registered_at','frozen_at'),('assessed_at','reviewed_at'),('start','end') if path.endswith('.deadline') else ('unused','unused')]: if value.get(start) and value.get(end): try: if date_bounds(value[start])[0] > date_bounds(value[end])[1]: self.error(path, start+' is later than '+end) except (ValueError,TypeError,OverflowError): pass elif isinstance(value,list): for i,item in enumerate(value): self.dates_and_urls(item,path+'['+str(i)+']') def completeness(self): r = self.r required = { '$': 'assessment_id assessment_date', '$.forecast': 'forecast_family_id forecaster exact_quote statement_date date_precision primary_source_url source_locator', '$.resolution_contract': 'contract_id contract_version registered_at frozen_at frozen_copy_location designation knowledge_timing_note event_or_quantity population success_rule allowed_human_assistance acceptable_primary_evidence review_publication_cutoff', '$.resolution_contract.measurement': 'metric denominator units sampling_method', '$.resolution_contract.deadline': 'timezone precision', '$.assessment': 'state as_of reason', '$.aggregate_treatment': 'reason', '$.review': 'assessor assessed_at second_reviewer reviewed_at', '$.historical_preservation': 'scope' } def get(path): obj = r for key in path.split('.')[1:]: obj = obj.get(key,{}) return obj for path, fields in required.items(): obj = get(path) for field in fields.split(): if not text(obj.get(field)): self.missing(path+'.'+field) review = r['review'] if review.get('reproduced_from_receipt') is not True: self.missing('$.review.reproduced_from_receipt') if type(review.get('material_interpretation_disagreement')) is not bool: self.missing('$.review.material_interpretation_disagreement') if review.get('material_interpretation_disagreement') is True and not text(review.get('disagreement_note')): self.missing('$.review.disagreement_note') if review.get('assessor') and review.get('second_reviewer') == review.get('assessor'): self.error('$.review', 'Second reviewer must be distinct from the assessor') state = r['assessment']['state'] if state in RESOLVED and not r['assessment']['evidence_ids']: self.missing('$.assessment.evidence_ids') if state in RESOLVED and not r['calculations']: self.missing('$.calculations') if state in RESOLVED and not r['resolution_contract']['deadline'].get('end'): self.missing('$.resolution_contract.deadline.end') for i, ev in enumerate(r['evidence']): if isinstance(ev,dict): for field in 'primary_source_url publisher_or_author source_type publication_date exact_passage locator supports'.split(): if not text(ev.get(field)): self.missing('$.evidence['+str(i)+'].'+field) c = r['resolution_contract'] if state in {'missed','mixed'} and r['assessment'].get('as_of') and c['deadline'].get('end'): try: if date_bounds(r['assessment']['as_of'])[1] < date_bounds(c['deadline']['end'])[1]: self.error('$.assessment.as_of', 'The contracted window is still open; a missed deadline is not established') except (ValueError,TypeError,OverflowError): pass if c.get('frozen_at') and c.get('designation') == 'prospective': for ev in r['evidence']: if isinstance(ev,dict) and ev.get('observation_start'): try: if date_bounds(c['frozen_at'])[0] >= date_bounds(ev['observation_start'])[0]: self.error('$.resolution_contract.designation', 'Recorded outcome observations predate the prospective freeze; review designation') except (ValueError,TypeError,OverflowError): pass cutoff = c.get('review_publication_cutoff') if cutoff: for i, ev in enumerate(r['evidence']): if isinstance(ev,dict) and ev.get('publication_date'): try: if date_bounds(ev['publication_date'])[0] > date_bounds(cutoff)[1]: self.error('$.evidence['+str(i)+'].publication_date', 'Evidence exceeds the contracted publication cutoff') except (ValueError,TypeError,OverflowError): pass def report(self): kind = self.r.get('receipt_kind') if isinstance(self.r,dict) else None return {'valid': not self.errors, 'complete_for_human_review': not self.errors and not self.incomplete and kind == 'assessment', 'receipt_kind': kind, 'errors': self.errors, 'incomplete_fields': self.incomplete, 'warnings': self.warnings + ['No source retrieval, factual verification, interpretation approval or publication authorization is performed.'], 'ledger_writes': 0, 'network_requests': 0} def validate_receipt(receipt): return Validator(receipt).validate() def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('files', nargs='+', help='Receipt JSON files to read') parser.add_argument('--require-complete', action='store_true', help='Fail incomplete drafts, templates and fictional examples') args = parser.parse_args(argv) reports = [] included = {} for path in args.files: try: receipt = load_receipt(path) report = validate_receipt(receipt) if report['valid'] and receipt.get('aggregate_treatment',{}).get('prospective_aggregate_inclusion') == 'included': family = receipt.get('forecast',{}).get('forecast_family_id') if text(family): if family in included: report['errors'].append({'path':'$.forecast.forecast_family_id','message':'Duplicate included family in supplied files: '+family}) report['valid'] = report['complete_for_human_review'] = False included[family] = path except (OSError, ValueError, TypeError, RecursionError) as exc: report = {'valid':False,'complete_for_human_review':False,'errors':[{'path':'$','message':str(exc)}],'ledger_writes':0,'network_requests':0} report['file'] = str(path) reports.append(report) passed = all(r['valid'] and (not args.require_complete or r['complete_for_human_review']) for r in reports) print(json.dumps({'validator_version':VERSION,'passed':passed,'require_complete':args.require_complete,'receipts':reports},indent=2)) return 0 if passed else 1 if __name__ == '__main__': sys.exit(main())