#!/usr/bin/env python3 # -*- coding: utf-8 -*- """双色球「预测挑战」验证器 —— 用于客观判定一个提交的选号策略是否真的超过随机。 背景：本项目已有结论是「不存在能超过随机的策略」。本验证器的作用是**给这个结论一个可被推翻的机会**： 任何人（人或智能体）提交一个策略，就用本脚本在**严格样本外**的前提下判定它。 提交格式（一个 .py 文件，必须只依赖标准库）： def predict(history): '''history: 严格早于本期的历史开奖，按时间升序，元素为 dict： {'issue','r1'..'r6','blue','date','sales','pool', ...} 返回：(6 个红球的集合/列表, 蓝球整数)''' ... return [1,2,3,4,5,6], 7 用法： python 预测挑战_验证器.py 提交文件.py [--declared-rules N] 判定规则（与《双色球理性决策手册》第 9 节「复活条款」一致）： 1) 训练/测试切分：默认以 --split 日期（默认 2023-01-01）为界，测试段只用严格历史； 2) 统计量：测试段平均红球命中数； 3) 零假设：每期命中数服从 Hypergeometric(33,6,6)，均值 36/33 = 1.0909； 4) 多重比较：调用方必须声明本次一共试了多少条规则（--declared-rules，默认 1）， 门槛用 Bonferroni 等价口径 z = Phi^{-1}(1 - 0.05/(2N))； 5) 复活门槛：效应量需 ≥ +35%（即平均命中 ≥ 1.0909 × 1.35 = 1.4727）。 只有同时满足「统计显著」且「效应量 ≥ +35%」，才判定为**推翻现有结论**。 """ from __future__ import annotations import argparse import importlib.util import math import sys from pathlib import Path HERE = Path(__file__).resolve().parent DEFAULT_DATA = HERE / "_C3_evidence" / "ssq_history_parsed.csv" MEAN_RANDOM = 36.0 / 33.0 # 1.0909... VARIANCE_RANDOM = 0.753099 # Hypergeom(33,6,6) 的方差 MIN_EFFECT = 1.35 # 复活条款：效应量 ≥ +35% def norm_cdf(x: float) -> float: return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0))) def norm_ppf(p: float) -> float: # 只用标准库的近似：Acklam 反函数 a = [-3.969683028665376e+01, 2.209460984245205e+02, -2.759285104469687e+02, 1.383577518672690e+02, -3.066479806614716e+01, 2.506628277459239e+00] b = [-5.447609879822406e+01, 1.615858368580409e+02, -1.556989798598866e+02, 6.680131188771972e+01, -1.328068155288572e+01] c = [-7.784894002430293e-03, -3.223964580411365e-01, -2.400758277161838e+00, -2.549732539343734e+00, 4.374664141464968e+00, 2.938163982698783e+00] d = [7.784695709041462e-03, 3.224671290700398e-01, 2.445134137142996e+00, 3.754408661907416e+00] plow, phigh = 0.02425, 1 - 0.02425 if p < plow: q = math.sqrt(-2 * math.log(p)) return (((((c[0]*q+c[1])*q+c[2])*q+c[3])*q+c[4])*q+c[5]) / ((((d[0]*q+d[1])*q+d[2])*q+d[3])*q+1) if p > phigh: q = math.sqrt(-2 * math.log(1 - p)) return -(((((c[0]*q+c[1])*q+c[2])*q+c[3])*q+c[4])*q+c[5]) / ((((d[0]*q+d[1])*q+d[2])*q+d[3])*q+1) q = p - 0.5 r = q * q return (((((a[0]*r+a[1])*r+a[2])*r+a[3])*r+a[4])*r+a[5])*q / (((((b[0]*r+b[1])*r+b[2])*r+b[3])*r+b[4])*r+1) def load_history(path: Path) -> list[dict]: import csv rows = [] with path.open(encoding="utf-8-sig", newline="") as f: for r in csv.DictReader(f): try: rows.append({ "issue": r["期号"], "reds": [int(r[f"r{i}"]) for i in range(1, 7)], "blue": int(r["蓝球"]), "date": r["开奖日期"], "sales": int((r.get("总投注额") or "0").replace(",", "") or 0), "pool": int((r.get("奖池") or "0").replace(",", "") or 0), "first_n": int(r.get("一等奖注数") or 0), "first_prize": int((r.get("一等奖单注奖金") or "0").replace(",", "") or 0), "second_n": int(r.get("二等奖注数") or 0), "second_prize": int((r.get("二等奖单注奖金") or "0").replace(",", "") or 0), }) except Exception: continue rows.sort(key=lambda x: x["date"]) return rows def load_strategy(path: Path): spec = importlib.util.spec_from_file_location("submission", str(path)) if spec is None or spec.loader is None: raise RuntimeError("无法加载提交文件") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) fn = getattr(mod, "predict", None) if not callable(fn): raise RuntimeError("提交文件里没有定义 predict(history)") return fn def main() -> int: try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") except Exception: pass ap = argparse.ArgumentParser() ap.add_argument("submission", help="提交的策略文件（含 predict(history)）") ap.add_argument("--split", default="2023-01-01", help="测试段起始日期（默认 2023-01-01）") ap.add_argument("--declared-rules", type=int, default=1, help="本次一共试了多少条规则（用于多重比较）") ap.add_argument("--data", default=str(DEFAULT_DATA), help="历史数据 CSV 路径") args = ap.parse_args() rows = load_history(Path(args.data)) if len(rows) < 100: print("历史数据不足") return 2 predict = load_strategy(Path(args.submission)) split_idx = next((i for i, r in enumerate(rows) if r["date"] >= args.split), None) if split_idx is None or split_idx < 100: print(f"切分点无效：{args.split}") return 2 test = rows[split_idx:] print(f"数据 {len(rows)} 期｜训练 {split_idx} 期（至 {rows[split_idx-1]['date']}）" f"｜测试 {len(test)} 期（{test[0]['date']} 起）") hits = [] blue_hits = 0 for i in range(split_idx, len(rows)): history = rows[:i] # 严格历史，不含本期 actual = rows[i] try: pred = predict(list(history)) reds, blue = pred[0], int(pred[1]) except Exception as exc: print(f"第 {actual['issue']} 期调用 predict 出错：{exc}") return 3 h = len(set(int(x) for x in reds) & set(actual["reds"])) hits.append(h) if blue == actual["blue"]: blue_hits += 1 n = len(hits) mean_hit = sum(hits) / n se = math.sqrt(VARIANCE_RANDOM / n) z = (mean_hit - MEAN_RANDOM) / se p = 2 * (1 - norm_cdf(abs(z))) M = max(1, args.declared_rules) z_gate = abs(norm_ppf(0.05 / (2 * M))) effect = mean_hit / MEAN_RANDOM - 1 blue_rate = blue_hits / n print(f"红球平均命中：{mean_hit:.4f}（随机基准 {MEAN_RANDOM:.4f}）") print(f"蓝球命中率：{blue_rate:.4f}（随机基准 {1/16:.4f}）") print(f"z = {z:.3f}　p = {p:.4g}　多重比较门槛 |z| ≥ {z_gate:.3f}（按 {M} 条规则）") print(f"效应量：{effect*100:+.1f}%（复活条款要求 ≥ +{(MIN_EFFECT-1)*100:.0f}%）") print("-" * 60) sig = abs(z) >= z_gate and p < 0.05 strong = effect >= (MIN_EFFECT - 1) if sig and strong and z > 0: print("✅ 判定：**推翻现有结论** —— 该策略在样本外显著且效应量达标，值得全量复核。") elif sig and z > 0: print("⚠️ 判定：统计上显著，但效应量未达 +35% 门槛；按复活条款不足以重开预测方向。") elif z > 0: print("❌ 判定：未达显著（很可能只是样本波动）。") else: print("❌ 判定：表现不优于随机。") return 0 if __name__ == "__main__": raise SystemExit(main())