""" shell_volledig.py Verdieping van de Shell-meting (vgl. Bijlage G van "De Shell"): - 50.000 willekeurige deals (instelbaar), GEEN HCP-restrictie - ZOWEL NZ als OW worden gemeten, elk in hun beste (= langste) fit - Per partnership-observatie wordt vastgelegd: fitlengte, troefkleur, Shellgrootte (strenge definitie), totaal HCP, werkende HCP, DD-slagen in de eigen beste fit - Output: 1. Verdeling van Shellgroottes (alle observaties en alleen 8+-fits) 2. Per Shellgrootte: werkende HCP -> n, gemiddeld aantal slagen, sd 3. Voorspelkracht: correlatie en RMSE van 'werkende HCP' versus 'totaal HCP' als voorspeller van het aantal slagen 4. CSV-export van alle ruwe observaties (voor eigen analyse) Definities (identiek aan Bijlage G, strenge variant): - Troefkleur : langste fit van het partnership; bij gelijke lengte willekeurig (geen voorkeur hoog/laag). Ook fits van 7 of minder worden meegenomen (misfit-deals), maar apart herkenbaar via de kolom fitlen. - Uitgeschakelde zijkleur : minstens een van beide partners heeft daarin een singleton of renonce. Doubletons tellen NIET mee. - Shellgrootte : 40 - 10 per uitgeschakelde zijkleur. - Werkende HCP : HCP in actieve kleuren + 4 per Aas in een uitgeschakelde kleur. - DD-slagen : gemiddelde van de twee partners als leider in de gekozen troefkleur (double dummy, DDS-engine). Gebruik: python shell_volledig.py --deals 50000 --workers 8 --csv shell_volledig.csv Vereist: Python 3.13+, endplay (pip install endplay) """ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ """ import argparse import math import random import sys from collections import defaultdict from concurrent.futures import ProcessPoolExecutor from endplay.dealer import generate_deals from endplay.dds import calc_all_tables from endplay.types import Denom, Player # --------------------------------------------------------------------------- # Basisgereedschap # --------------------------------------------------------------------------- SUITS = [Denom.spades, Denom.hearts, Denom.diamonds, Denom.clubs] SUITNAMES = {Denom.spades: "S", Denom.hearts: "H", Denom.diamonds: "D", Denom.clubs: "C"} HCP_CHAR = {"A": 4, "K": 3, "Q": 2, "J": 1} SIDES = { "NZ": (Player.north, Player.south), "OW": (Player.east, Player.west), } def suit_string(hand, denom): """Kaarten van een hand in een kleur als string, bijv. 'AQT4'.""" if denom == Denom.spades: return str(hand.spades) if denom == Denom.hearts: return str(hand.hearts) if denom == Denom.diamonds: return str(hand.diamonds) return str(hand.clubs) def hcp_of(s): return sum(HCP_CHAR.get(ch, 0) for ch in s) def analyse_side(deal, table, p1, p2, rng): """Analyseer een partnership: kies beste fit, bepaal Shell en HCP's.""" h1, h2 = deal[p1], deal[p2] holdings = {d: (suit_string(h1, d), suit_string(h2, d)) for d in SUITS} lengths = {d: len(holdings[d][0]) + len(holdings[d][1]) for d in SUITS} fitlen = max(lengths.values()) candidates = [d for d in SUITS if lengths[d] == fitlen] trump = rng.choice(candidates) # Strenge definitie: zijkleur uitgeschakeld als minstens een van beide # partners daarin een singleton of renonce heeft. disabled = [d for d in SUITS if d != trump and (len(holdings[d][0]) <= 1 or len(holdings[d][1]) <= 1)] shell = 40 - 10 * len(disabled) total_hcp = sum(hcp_of(a) + hcp_of(b) for a, b in holdings.values()) working = 0 for d in SUITS: a, b = holdings[d] if d in disabled: working += 4 * (a.count("A") + b.count("A")) else: working += hcp_of(a) + hcp_of(b) tricks = (table[trump, p1] + table[trump, p2]) / 2.0 return { "fitlen": fitlen, "trump": SUITNAMES[trump], "shell": shell, "total_hcp": total_hcp, "working_hcp": working, "tricks": tricks, } # --------------------------------------------------------------------------- # Worker # --------------------------------------------------------------------------- BATCH = 32 # aantal deals per calc_all_tables-aanroep def worker(args): n_deals, seed = args rng = random.Random(seed) rows = [] done = 0 while done < n_deals: chunk = min(BATCH, n_deals - done) deals = list(generate_deals(produce=chunk)) tables = calc_all_tables(deals) for deal, table in zip(deals, tables): for side, (p1, p2) in SIDES.items(): rec = analyse_side(deal, table, p1, p2, rng) rec["side"] = side rows.append(rec) done += chunk return rows # --------------------------------------------------------------------------- # Statistiek-hulpjes (geen externe afhankelijkheden) # --------------------------------------------------------------------------- def mean(xs): return sum(xs) / len(xs) def stdev(xs): if len(xs) < 2: return 0.0 m = mean(xs) return math.sqrt(sum((x - m) ** 2 for x in xs) / (len(xs) - 1)) def pearson(xs, ys): n = len(xs) mx, my = mean(xs), mean(ys) sx = math.sqrt(sum((x - mx) ** 2 for x in xs)) sy = math.sqrt(sum((y - my) ** 2 for y in ys)) if sx == 0 or sy == 0: return float("nan") return sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / (sx * sy) def linfit_rmse(xs, ys): """RMSE van de beste lineaire voorspelling y = a + b*x.""" n = len(xs) mx, my = mean(xs), mean(ys) sxx = sum((x - mx) ** 2 for x in xs) if sxx == 0: return float("nan") b = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / sxx a = my - b * mx return math.sqrt(sum((y - (a + b * x)) ** 2 for x, y in zip(xs, ys)) / n) # --------------------------------------------------------------------------- # Rapportage # --------------------------------------------------------------------------- def report(rows, n_deals): n_obs = len(rows) print("=" * 72) print(f"SHELL-METING ZONDER HCP-FILTER | {n_deals} deals, " f"{n_obs} partnership-observaties (NZ en OW per deal)") print("=" * 72) # ---- 1. Verdeling Shellgroottes ------------------------------------- for label, subset in ( ("Alle observaties", rows), ("Alleen observaties met 8+-fit", [r for r in rows if r["fitlen"] >= 8]), ): print(f"\n--- Verdeling Shellgroottes: {label} (n={len(subset)}) ---") counts = defaultdict(int) for r in subset: counts[r["shell"]] += 1 for shell in sorted(counts, reverse=True): pct = 100.0 * counts[shell] / len(subset) print(f" Shell {shell:>2}: {counts[shell]:>7} ({pct:5.1f}%)") kleiner = sum(v for k, v in counts.items() if k < 40) print(f" Shell < 40 totaal: {100.0 * kleiner / len(subset):5.1f}%") # ---- 2. Werkende HCP -> slagen, per Shellgrootte --------------------- print("\n--- Werkende HCP versus DD-slagen in beste fit, per Shell ---") print(" (alleen cellen met n >= 30)") per_shell = defaultdict(list) for r in rows: per_shell[r["shell"]].append(r) for shell in sorted(per_shell, reverse=True): sub = per_shell[shell] print(f"\n Shellgrootte {shell} (n={len(sub)})") print(f" {'wHCP':>5} {'n':>7} {'gem.slagen':>11} {'sd':>6}") buckets = defaultdict(list) for r in sub: buckets[r["working_hcp"]].append(r["tricks"]) for w in sorted(buckets): xs = buckets[w] if len(xs) < 30: continue print(f" {w:>5} {len(xs):>7} {mean(xs):>11.2f} {stdev(xs):>6.2f}") # ---- 3. Voorspelkracht: werkende vs. totaal HCP ---------------------- print("\n--- Voorspelkracht voor het aantal slagen ---") print(f" {'groep':<28} {'n':>7} {'r(wHCP)':>8} {'r(tHCP)':>8} " f"{'RMSE(w)':>8} {'RMSE(t)':>8}") def line(label, subset): if len(subset) < 100: return t = [r["tricks"] for r in subset] w = [r["working_hcp"] for r in subset] h = [r["total_hcp"] for r in subset] print(f" {label:<28} {len(subset):>7} {pearson(w, t):>8.3f} " f"{pearson(h, t):>8.3f} {linfit_rmse(w, t):>8.3f} " f"{linfit_rmse(h, t):>8.3f}") line("Alle observaties", rows) line("Alleen 8+-fit", [r for r in rows if r["fitlen"] >= 8]) line("Misfit (fit <= 7)", [r for r in rows if r["fitlen"] <= 7]) for shell in (40, 30, 20, 10): line(f"Shell {shell} (8+-fit)", [r for r in rows if r["shell"] == shell and r["fitlen"] >= 8]) print("\n Lezing: r dichter bij 1 en lagere RMSE = betere voorspeller.") print(" Bij Shell 40 geldt wHCP = tHCP per definitie; gelijke r en RMSE") print(" op die regel is dus een ingebouwde consistentiecheck.") print(" Binnen een vaste Shellgrootte is r(wHCP) vs r(tHCP) de eerlijke") print(" vergelijking; over alle groepen heen toont de totaalregel het") print(" gecombineerde effect van Shell-indeling plus werkende punten.") def write_csv(rows, path): import csv with open(path, "w", newline="", encoding="utf-8") as f: wr = csv.DictWriter(f, fieldnames=[ "side", "fitlen", "trump", "shell", "total_hcp", "working_hcp", "tricks"]) wr.writeheader() wr.writerows(rows) print(f"\nRuwe data weggeschreven naar: {path}") # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(description="Shell-meting zonder HCP-filter") ap.add_argument("--deals", type=int, default=50000) ap.add_argument("--workers", type=int, default=8) ap.add_argument("--seed", type=int, default=None, help="seed voor tie-breaks bij gelijke fitlengtes") ap.add_argument("--csv", type=str, default=None, help="pad voor CSV-export van ruwe observaties") args = ap.parse_args() base_seed = args.seed if args.seed is not None else random.randrange(1 << 30) per_worker = [args.deals // args.workers] * args.workers per_worker[0] += args.deals - sum(per_worker) tasks = [(n, base_seed + i) for i, n in enumerate(per_worker) if n > 0] rows = [] if args.workers <= 1: for t in tasks: rows.extend(worker(t)) else: with ProcessPoolExecutor(max_workers=args.workers) as ex: for part in ex.map(worker, tasks): rows.extend(part) report(rows, args.deals) if args.csv: write_csv(rows, args.csv) if __name__ == "__main__": main()