#!/usr/bin/env python3 """ESEMPIO SINTETICO GSO Radar: nessun evento, quota o risultato reale. Genera e controlla il dataset didattico della guida sulla qualita dei dati. Solo libreria standard Python; nessuna rete, casualita o dipendenza esterna. """ import argparse import csv import hashlib import json from collections import Counter from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from pathlib import Path DATASET_NAME = "gso-qualita-dati-SINTETICO-v1.csv" FIELDS = [ "dataset_kind", "event_id", "season", "competition", "market", "selection", "kickoff_utc", "decision_utc", "quote_published_utc", "quote_acquired_utc", "odds_decimal", "stat_source_last_event_utc", "stat_published_utc", "stat_acquired_utc", "mean_goals_previous_5", ] NOTICE = "DATI INTERAMENTE SINTETICI: non sono eventi, quote o risultati reali." def iso(value): return value.isoformat(timespec="seconds").replace("+00:00", "Z") def parse_utc(value): if not value: return None result = datetime.fromisoformat(value.replace("Z", "+00:00")) if result.tzinfo is None: raise ValueError("Un timestamp deve dichiarare il fuso orario") return result.astimezone(timezone.utc) def generate(csv_path): """Due stagioni separate, 500 eventi ciascuna; gli ID non sono cronologici.""" ranks = Counter() rows = [] for number in range(1, 1001): season = "A" if number <= 502 and number not in (29, 30) else "B" rank = ranks[season] ranks[season] += 1 year = 2024 if season == "A" else 2025 kickoff = datetime(year, 8, 1, 19, tzinfo=timezone.utc) + timedelta(days=rank // 10) decision = kickoff - timedelta(hours=1) late = 26 <= number <= 35 quote_acquired = decision + timedelta(minutes=2 if late else -2) quote_published = quote_acquired - timedelta(seconds=30) stat_missing = number <= 30 row = { "dataset_kind": "SYNTHETIC_EDUCATIONAL_ONLY", "event_id": f"M{number:04d}", "season": season, "competition": "COMPETIZIONE_FITTIZIA", "market": "1X2", "selection": "X", "kickoff_utc": iso(kickoff), "decision_utc": iso(decision), "quote_published_utc": iso(quote_published), "quote_acquired_utc": iso(quote_acquired), "odds_decimal": str(Decimal("3.00") + Decimal(number % 21) / 100), "stat_source_last_event_utc": "" if stat_missing else iso(kickoff - timedelta(days=3)), "stat_published_utc": "" if stat_missing else iso(decision - timedelta(hours=3)), "stat_acquired_utc": "" if stat_missing else iso(decision - timedelta(hours=2)), "mean_goals_previous_5": "" if stat_missing else str(Decimal(number % 16) / 5), } rows.append(row) # Copie interamente identiche: non aggiungiamo un import-row-id artificiale. rows.extend(dict(row) for row in rows[:12]) csv_path.parent.mkdir(parents=True, exist_ok=True) with csv_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=FIELDS, lineterminator="\n") writer.writeheader() writer.writerows(rows) def audit(csv_path, output_dir): with csv_path.open(encoding="utf-8", newline="") as handle: reader = csv.DictReader(handle) if reader.fieldnames != FIELDS: raise ValueError("Colonne inattese: usare il CSV didattico completo") imported = list(reader) seen_rows = set() unique = [] for row in imported: signature = tuple(row[field] for field in FIELDS) if signature not in seen_rows: seen_rows.add(signature) unique.append(row) ids = [row["event_id"] for row in unique] if len(ids) != len(set(ids)): raise ValueError("Stesso event_id con righe discordanti: non deduplicare per solo ID") expected_ids = {f"M{n:04d}" for n in range(1, 1001)} if set(ids) != expected_ids: raise ValueError("Copertura differente dagli eventi attesi M0001-M1000") if any(row["dataset_kind"] != "SYNTHETIC_EDUCATIONAL_ONLY" for row in unique): raise ValueError("Questo audit riproduce esclusivamente il dataset sintetico dichiarato") flagged = [] seasons = {season: {"events": 0, "missing_stat": 0} for season in ("A", "B")} for row in unique: kickoff = parse_utc(row["kickoff_utc"]) decision = parse_utc(row["decision_utc"]) published = parse_utc(row["quote_published_utc"]) acquired = parse_utc(row["quote_acquired_utc"]) stat_missing = not row["mean_goals_previous_5"].strip() if kickoff is None or decision is None or decision >= kickoff: raise ValueError(f"Decisione non pre-match o timestamp assente: {row['event_id']}") if published is None or acquired is None or published > acquired: raise ValueError(f"Timestamp quota assente o sequenza incoerente: {row['event_id']}") late_quote = max(published, acquired) > decision try: odds = Decimal(row["odds_decimal"]) if not odds.is_finite() or odds <= 1: raise ValueError("Quota non valida") if not stat_missing: stat = Decimal(row["mean_goals_previous_5"]) if not stat.is_finite() or stat < 0: raise ValueError("Statistica non valida") source_time = parse_utc(row["stat_source_last_event_utc"]) stat_published = parse_utc(row["stat_published_utc"]) stat_acquired = parse_utc(row["stat_acquired_utc"]) if any(value is None for value in (source_time, stat_published, stat_acquired)): raise ValueError("Provenienza temporale della statistica assente") if not source_time <= stat_published <= stat_acquired <= decision: raise ValueError("Statistica non disponibile prima della decisione") except (InvalidOperation, ValueError) as exc: raise ValueError(f"{row['event_id']}: {exc}") from exc if row["season"] not in seasons: raise ValueError("Stagione non prevista") seasons[row["season"]]["events"] += 1 seasons[row["season"]]["missing_stat"] += int(stat_missing) flagged.append({ **row, "missing_required_stat": int(stat_missing), "quote_after_decision": int(late_quote), "usable": int(not stat_missing and not late_quote), }) missing_ids = [r["event_id"] for r in flagged if r["missing_required_stat"]] late_ids = [r["event_id"] for r in flagged if r["quote_after_decision"]] overlap_ids = sorted(set(missing_ids) & set(late_ids)) excluded_ids = sorted(set(missing_ids) | set(late_ids)) used = sum(r["usable"] for r in flagged) for season in seasons.values(): season["missing_stat_percent"] = round(season["missing_stat"] / season["events"] * 100, 1) counts = { "expected_events": 1000, "imported_rows": len(imported), "additional_exact_duplicates": len(imported) - len(unique), "unique_events": len(unique), "missing_required_stat": len(missing_ids), "quote_after_decision": len(late_ids), "both_problems": len(overlap_ids), "distinct_excluded": len(excluded_ids), "usable_events": used, "initial_event_coverage_percent": len(unique) / 1000 * 100, "usable_percent_of_expected": used / 1000 * 100, "missing_stat_percent_overall": len(missing_ids) / len(unique) * 100, } expected_counts = { "expected_events": 1000, "imported_rows": 1012, "additional_exact_duplicates": 12, "unique_events": 1000, "missing_required_stat": 30, "quote_after_decision": 10, "both_problems": 5, "distinct_excluded": 35, "usable_events": 965, "initial_event_coverage_percent": 100.0, "usable_percent_of_expected": 96.5, "missing_stat_percent_overall": 3.0, } if counts != expected_counts: raise ValueError(f"Conteggi diversi da quelli pubblicati: {counts}") expected_seasons = { "A": {"events": 500, "missing_stat": 28, "missing_stat_percent": 5.6}, "B": {"events": 500, "missing_stat": 2, "missing_stat_percent": 0.4}, } if seasons != expected_seasons: raise ValueError(f"Tassi stagionali inattesi: {seasons}") if missing_ids != [f"M{n:04d}" for n in range(1, 31)]: raise ValueError("Identificativi con statistica mancante inattesi") if late_ids != [f"M{n:04d}" for n in range(26, 36)]: raise ValueError("Identificativi con quote tardive inattesi") season_a_times = [parse_utc(r["kickoff_utc"]) for r in unique if r["season"] == "A"] season_b_times = [parse_utc(r["kickoff_utc"]) for r in unique if r["season"] == "B"] if max(season_a_times) >= min(season_b_times): raise ValueError("Le stagioni sintetiche non sono separate nel tempo") summary = { "notice": NOTICE, "dataset_sha256": hashlib.sha256(csv_path.read_bytes()).hexdigest(), "counts": counts, "seasons": seasons, "missing_stat_event_ids": missing_ids, "late_quote_event_ids": late_ids, "overlap_event_ids": overlap_ids, "excluded_event_ids": excluded_ids, "checks": { "expected_counts_match": True, "season_rates_match": True, "temporal_order_checked": True, "duplicate_rule": "Elimina soltanto righe identiche in tutte le colonne; poi controlla unicita evento", "profit_or_predictive_accuracy_calculated": False, }, } output_dir.mkdir(parents=True, exist_ok=True) with (output_dir / "audit-qualita-dati-SINTETICO.csv").open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=FIELDS + ["missing_required_stat", "quote_after_decision", "usable"], lineterminator="\n") writer.writeheader() writer.writerows(flagged) (output_dir / "audit-qualita-dati-SINTETICO.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return summary def main(): parser = argparse.ArgumentParser(description=NOTICE) parser.add_argument("--generate", action="store_true", help="Rigenera deterministicamente il CSV sintetico") parser.add_argument("--csv", type=Path, help="CSV da controllare; default nella cartella dello script") parser.add_argument("--output-dir", type=Path, default=Path(__file__).resolve().parent) args = parser.parse_args() csv_path = args.csv or args.output_dir / DATASET_NAME if args.generate: generate(csv_path) summary = audit(csv_path, args.output_dir) print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()