#!/usr/bin/env python3 """Offline comparison eligibility for a deliberately narrow normalized contract.""" import argparse from datetime import datetime, timedelta import json import math from pathlib import Path import re import sys VERSION = "1.0.0" MAX_BYTES = 65536 KNOWN = {"period": "REGULATION", "market": "TOTAL_GOALS", "settlement_rules": "fixture-soccer-regulation-v1"} PAIR_FIELDS = {"event_namespace": "EVENT_NAMESPACE", "event_id": "EVENT", "period": "PERIOD", "market": "MARKET", "line": "LINE", "settlement_rules": "SETTLEMENT_RULES", "in_play": "PHASE"} FIELDS = set(PAIR_FIELDS) | {"selections", "status", "complete_snapshot", "captured_at", "price_updated_at", "odds"} def number(value): try: return type(value) in (int, float) and math.isfinite(value) except OverflowError: return False def utc(value, label, reasons): if value is None: reasons.append(label + "_MISSING") return None try: if not isinstance(value, str) or not 19 <= len(value) <= 40 or "T" not in value: raise ValueError result = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: reasons.append(label + "_INVALID") return None if result.tzinfo is None: reasons.append(label + "_NAIVE") return None if result.utcoffset() != timedelta(0): reasons.append(label + "_NOT_UTC") return None return result def selections(value): return sorted(value) if isinstance(value, list) and all(isinstance(s, str) for s in value) else None def observation(value, side, now, max_age, reasons): local = [] if not isinstance(value, dict): reasons.append(side + ":OBSERVATION_INVALID") return None if set(value) - FIELDS: local.append("UNEXPECTED_FIELDS") for key in ("event_namespace", "event_id"): item = value.get(key) if not isinstance(item, str) or not 1 <= len(item) <= 128 or item.strip() != item: local.append(key.upper() + "_INVALID") for key, expected in KNOWN.items(): if value.get(key) != expected: local.append(key.upper() + "_UNKNOWN") line = value.get("line") if not number(line) or line != 2.5: local.append("LINE_UNSUPPORTED") if selections(value.get("selections")) != ["OVER", "UNDER"]: local.append("SELECTIONS_INVALID") if type(value.get("in_play")) is not bool: local.append("PHASE_UNKNOWN") status = value.get("status") if status not in ("OPEN", "SUSPENDED", "CLOSED"): local.append("STATUS_UNKNOWN") elif status != "OPEN": local.append("STATE_" + status) if value.get("complete_snapshot") is not True: local.append("INCOMPLETE_SNAPSHOT") odds = value.get("odds") if not isinstance(odds, dict) or set(odds) != {"OVER", "UNDER"}: local.append("ODDS_INCOMPLETE") elif not all(number(price) and price > 1 for price in odds.values()): local.append("DECIMAL_ODDS_INVALID") times = {} for key in ("captured_at", "price_updated_at"): stamp = times[key] = utc(value.get(key), key.upper(), local) if stamp is not None: age = (now - stamp).total_seconds() if age < 0: local.append(key.upper() + "_FUTURE") elif age > max_age: local.append(key.upper() + "_STALE_BY_POLICY") capture, price_time = times["captured_at"], times["price_updated_at"] if capture is not None and price_time is not None and price_time > capture: local.append("PRICE_AFTER_CAPTURE") reasons.extend(side + ":" + reason for reason in local) return capture def evaluate(document): if not isinstance(document, dict) or set(document) != {"now", "policy", "cases"}: raise ValueError("ENVELOPE_INVALID") errors = [] now = utc(document["now"], "NOW", errors) if errors: raise ValueError(errors[0]) policy = document["policy"] if (not isinstance(policy, dict) or set(policy) != {"max_age_seconds", "max_capture_skew_seconds"} or not all(number(v) and 0 <= v <= 86400 for v in policy.values())): raise ValueError("POLICY_INVALID") cases = document["cases"] if not isinstance(cases, list) or not 1 <= len(cases) <= 32: raise ValueError("CASES_INVALID") results, seen = [], set() for case in cases: if not isinstance(case, dict) or set(case) != {"id", "left", "right"}: raise ValueError("CASE_INVALID") name = case["id"] if not isinstance(name, str) or not re.fullmatch(r"[a-z0-9_-]{1,80}", name) or name in seen: raise ValueError("CASE_ID_INVALID") seen.add(name) reasons = [] left, right = case["left"], case["right"] captures = [observation(value, side, now, policy["max_age_seconds"], reasons) for value, side in ((left, "left"), (right, "right"))] if isinstance(left, dict) and isinstance(right, dict): for field, code in PAIR_FIELDS.items(): if field in left and field in right and left[field] != right[field]: reasons.append("pair:" + code + "_MISMATCH") if selections(left.get("selections")) != selections(right.get("selections")): reasons.append("pair:SELECTIONS_MISMATCH") if all(stamp is not None for stamp in captures): if abs((captures[0] - captures[1]).total_seconds()) > policy["max_capture_skew_seconds"]: reasons.append("pair:CAPTURE_SKEW_EXCEEDS_POLICY") results.append({"id": name, "eligible_for_comparison": not reasons, "reason_codes": reasons}) eligible = sum(result["eligible_for_comparison"] for result in results) return {"version": VERSION, "reference_time": document["now"], "policy": policy, "eligible_count": eligible, "held_count": len(results) - eligible, "results": results} def strict_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("DUPLICATE_JSON_KEY") result[key] = value return result def reject_constant(_value): raise ValueError("NONFINITE_JSON_NUMBER") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("input", type=Path, help="normalized JSON cases, at most 64 KiB") parser.add_argument("--version", action="version", version=VERSION) args = parser.parse_args() try: with args.input.open("rb") as source: raw = source.read(MAX_BYTES + 1) if len(raw) > MAX_BYTES: raise ValueError("INPUT_TOO_LARGE") doc = json.loads(raw, object_pairs_hook=strict_object, parse_constant=reject_constant) print(json.dumps(evaluate(doc), indent=2, allow_nan=False)) except (OSError, ValueError, RecursionError, OverflowError) as exc: code = str(exc) if type(exc) is ValueError and re.fullmatch(r"[A-Z_]+", str(exc)) else "INPUT_INVALID" print(json.dumps({"error": code}), file=sys.stderr) return 2 return 0 if __name__ == "__main__": sys.exit(main())